Merge pull request #289 from ThreatConnect-Inc/develop

Release 3.0.7
-   APP-3859 - [API] Enhancements for ThreatConnect 7.x

release_notes.md

@@ -1,5 +1,9 @@
 # Release Notes
 
+### 3.0.7
+
+-   APP-3859 - [API] Enhancements for ThreatConnect 7.x
+
 ### 3.0.6
 
 -   APP-3857 - [CLI] Updated spectool for general improvements and fixes

tcex/__metadata__.py

@@ -5,5 +5,5 @@
 __license__ = 'Apache License, Version 2'
 __package_name__ = 'tcex'
 __url__ = 'https://github.com/ThreatConnect-Inc/tcex'
-__version__ = '3.0.6'
+__version__ = '3.0.7'
 __download_url__ = f'https://github.com/ThreatConnect-Inc/tcex/tarball/{__version__}'

tcex/api/tc/ti_transform/model/transform_model.py

@@ -141,7 +141,7 @@ class GroupTransformModel(TiTransformModel, extra=Extra.forbid):
 
     name: MetadataTransformModel = Field(..., description='')
     # campaign
-    first_seen: Optional[MetadataTransformModel] = Field(None, description='')
+    first_seen: Optional[DatetimeTransformModel] = Field(None, description='')
     # document
     malware: Optional[MetadataTransformModel] = Field(None, description='')
     password: Optional[MetadataTransformModel] = Field(None, description='')
@@ -150,7 +150,7 @@ class GroupTransformModel(TiTransformModel, extra=Extra.forbid):
     score: Optional[MetadataTransformModel] = Field(None, description='')
     to_addr: Optional[MetadataTransformModel] = Field(None, description='')
     # event, incident
-    event_date: Optional[MetadataTransformModel] = Field(None, description='')
+    event_date: Optional[DatetimeTransformModel] = Field(None, description='')
     status: Optional[MetadataTransformModel] = Field(None, description='')
     # report
     publish_date: Optional[DatetimeTransformModel] = Field(None, description='')

tcex/api/tc/utils/threat_intel_utils.py

@@ -205,19 +205,29 @@ def resolve_variables(self, inputs: List[str]) -> List[str]:
         """
         resolved_inputs = []
         for input_ in inputs:
+            # handle null inputs
             if not input_:
                 resolved_inputs.append(None)
                 continue
-            if input_.strip() not in self.resolvable_variables:
+
+            # clean up input
+            input_ = input_.strip()
+
+            # handle unknown input types
+            if input_ not in self.resolvable_variables:
                 resolved_inputs.append(input_)
                 continue
-            input_ = input_.strip()
+
+            # special handling of group types (no API request required)
             if input_ == '${GROUP_TYPES}':
                 for type_ in self.group_types:
                     resolved_inputs.append(type_)
                 continue
 
+            # get variable settings
             resolvable_variable_details = self.resolvable_variables[input_]
+
+            # make API call to retrieve variable data
             r = self.session_tc.get(
                 resolvable_variable_details.get('url'), params={'resultLimit': 10_000}
             )
@@ -226,16 +236,6 @@ def resolve_variables(self, inputs: List[str]) -> List[str]:
                 raise RuntimeError(f'Could not retrieve {input_} from ThreatConnect API.')
 
             json_ = r.json()
-            # No TQL filter to filter out API users during REST call so have to do it manually here.
-            if input_ in ['${API_USERS}', '${USERS}']:
-                temp_data = []
-                for item in json_.get('data', []):
-                    if item.get('role') == 'Api User' and input_ == '${API_USERS}':
-                        temp_data.append(item)
-                    elif item.get('role') != 'Api User' and input_ == '${USERS}':
-                        temp_data.append(item)
-                json_['data'] = temp_data
-
             for item in jmespath.search(resolvable_variable_details.get('jmspath'), json_):
                 resolved_inputs.append(str(item))
 

tcex/api/tc/v3/_gen/_gen.py

@@ -23,6 +23,12 @@
 utils = Utils()
 
 
+def log_server():
+    """Log server."""
+    _api_server = os.getenv('TC_API_PATH')
+    typer.secho(f'Using server {_api_server}', fg=typer.colors.BRIGHT_MAGENTA)
+
+
 class GenerateArgs(GenerateArgsABC):
     """Generate Models for TC API Types"""
 
@@ -263,6 +269,7 @@ def all(  # pylint: disable=redefined-builtin
     ),
 ):
     """Generate Args."""
+    log_server()
     gen_type = gen_type.value.lower()
     for type_ in ObjectTypes:
         type_ = utils.snake_string(type_.value)
@@ -285,6 +292,7 @@ def args(
     ),
 ):
     """Generate Args."""
+    log_server()
     gen_args(utils.snake_string(type_.value), indent_blocks)
 
 
@@ -295,6 +303,7 @@ def code(
     ),
 ):
     """Generate Args."""
+    log_server()
     tv = utils.snake_string(type_.value)
     gen_filter(tv)
     gen_model(tv)
@@ -308,6 +317,7 @@ def filter(  # pylint: disable=redefined-builtin
     ),
 ):
     """Generate the filter code."""
+    log_server()
     gen_filter(utils.snake_string(type_.value))
 
 
@@ -318,6 +328,7 @@ def model(
     ),
 ):
     """Generate the model"""
+    log_server()
     gen_model(utils.snake_string(type_.value))
 
 
@@ -328,6 +339,7 @@ def object(  # pylint: disable=redefined-builtin
     ),
 ):
     """Generate the filter class"""
+    log_server()
     gen_object(utils.snake_string(type_.value))
 
 

tcex/api/tc/v3/_gen/_gen_abc.py

@@ -172,6 +172,9 @@ def _prop_contents_updated(self) -> dict:
         # remove unused fields, if any
         self._prop_content_remove_unused(_properties)
 
+        # update "bad" data
+        self._prop_content_update(_properties)
+
         # critical fix for breaking API change
         if self.type_ in [
             'case_attributes',
@@ -315,6 +318,10 @@ def _prop_content_fix_types(self, _properties: dict):
 
     def _prop_content_remove_unused(self, properties: dict):
         """Remove unused fields from properties."""
+        if self.type_ in ['attribute_types']:
+            if 'owner' in properties:
+                del properties['owner']
+
         if self.type_ in [
             'attribute_types',
             'case_attributes',
@@ -334,6 +341,53 @@ def _prop_content_remove_unused(self, properties: dict):
                 if field in properties:
                     del properties[field]
 
+        if self.type_ in ['users']:
+            unused_fields = [
+                'customTqlTimeout',
+                'disabled',
+                'firstName',
+                'jobFunction',
+                'jobRole',
+                'lastLogin',
+                'lastName',
+                'lastPasswordChange',
+                'locked',
+                'logoutIntervalMinutes',
+                'owner',
+                'ownerRoles',
+                'password',
+                'passwordResetRequired',
+                'pseudonym',
+                'systemRole',
+                'termsAccepted',
+                'termsAcceptedDate',
+                'tqlTimeout',
+                'twoFactorResetRequired',
+                'uiTheme',
+            ]
+            for field in unused_fields:
+                if field in properties:
+                    del properties[field]
+
+        if self.type_ in ['victims']:
+            unused_fields = [
+                'ownerId',  # Core Issue: should not be listed for this type
+            ]
+            for field in unused_fields:
+                if field in properties:
+                    del properties[field]
+
+    def _prop_content_update(self, properties: dict):
+        """Update "bad" data in properties."""
+        if self.type_ in ['groups']:
+            # fixed fields that are missing readOnly property
+            properties['downVoteCount']['readOnly'] = True
+            properties['upVoteCount']['readOnly'] = True
+
+        if self.type_ in ['victims']:
+            # ownerName is readOnly, but readOnly is not defined in response from OPTIONS endpoint
+            properties['ownerName']['readOnly'] = True
+
     @property
     def _prop_models(self) -> List[PropertyModel]:
         """Return a list of PropertyModel objects."""

tcex/api/tc/v3/_gen/_gen_object_abc.py

@@ -191,8 +191,7 @@ def filter(self) -> 'ArtifactFilter':
                 f'''{self.i2}self._request(''',
                 f'''{self.i3}method='GET',''',
                 f'''{self.i3}url=f\'\'\'{{self.url('GET')}}/download\'\'\',''',
-                f'''{self.i3}# headers={{'content-type': 'application/octet-stream'}},''',
-                f'''{self.i3}headers=None,''',
+                f'''{self.i3}headers={{'Accept': 'application/octet-stream'}},''',
                 f'''{self.i3}params=params,''',
                 f'''{self.i2})''',
                 f'''{self.i2}return self.request.content''',
@@ -203,7 +202,7 @@ def filter(self) -> 'ArtifactFilter':
                 f'''{self.i3}method='GET',''',
                 f'''{self.i3}body=None,''',
                 f'''{self.i3}url=f\'\'\'{{self.url('GET')}}/pdf\'\'\',''',
-                f'''{self.i3}headers=None,''',
+                f'''{self.i3}headers={{'Accept': 'application/octet-stream'}},''',
                 f'''{self.i3}params=params,''',
                 f'''{self.i2})''',
                 '',

tcex/api/tc/v3/attribute_types/attribute_type.py

@@ -53,7 +53,17 @@ def filter(self) -> 'AttributeTypeFilter':
 
 
 class AttributeType(ObjectABC):
-    """AttributeTypes Object."""
+    """AttributeTypes Object.
+
+    Args:
+        allow_markdown (bool, kwargs): Flag that enables markdown feature in the attribute value
+            field.
+        description (str, kwargs): The description of the attribute type.
+        error_message (str, kwargs): The error message displayed.
+        max_size (int, kwargs): The maximum size of the attribute value.
+        name (str, kwargs): The name of the attribute type.
+        validation_rule (object, kwargs): The validation rule that governs the attribute value.
+    """
 
     def __init__(self, **kwargs):
         """Initialize class properties."""

tcex/api/tc/v3/attribute_types/attribute_type_model.py

@@ -65,12 +65,54 @@ class AttributeTypeModel(
     _shared_type = PrivateAttr(False)
     _staged = PrivateAttr(False)
 
+    allow_markdown: bool = Field(
+        None,
+        description='Flag that enables markdown feature in the attribute value field.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='allowMarkdown',
+    )
+    description: Optional[str] = Field(
+        None,
+        description='The description of the attribute type.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='description',
+    )
+    error_message: Optional[str] = Field(
+        None,
+        description='The error message displayed.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='errorMessage',
+    )
     id: Optional[int] = Field(
         None,
         description='The ID of the item.',
         read_only=True,
         title='id',
     )
+    max_size: Optional[int] = Field(
+        None,
+        description='The maximum size of the attribute value.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='maxSize',
+    )
+    name: Optional[str] = Field(
+        None,
+        description='The name of the attribute type.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='name',
+    )
+    validation_rule: Optional[dict] = Field(
+        None,
+        description='The validation rule that governs the attribute value.',
+        methods=['POST', 'PUT'],
+        read_only=False,
+        title='validationRule',
+    )
 
 
 # add forward references

tcex/api/tc/v3/case_attributes/case_attribute.py

@@ -1,6 +1,6 @@
 """CaseAttribute / CaseAttributes Object"""
 # standard library
-from typing import Union
+from typing import TYPE_CHECKING, Iterator, Union
 
 # first-party
 from tcex.api.tc.v3.api_endpoints import ApiEndpoints
@@ -11,6 +11,11 @@
 )
 from tcex.api.tc.v3.object_abc import ObjectABC
 from tcex.api.tc.v3.object_collection_abc import ObjectCollectionABC
+from tcex.api.tc.v3.security_labels.security_label_model import SecurityLabelModel
+
+if TYPE_CHECKING:  # pragma: no cover
+    # first-party
+    from tcex.api.tc.v3.security_labels.security_label import SecurityLabel
 
 
 class CaseAttributes(ObjectCollectionABC):
@@ -59,9 +64,13 @@ class CaseAttribute(ObjectABC):
         case_id (int, kwargs): Case associated with attribute.
         default (bool, kwargs): A flag indicating that this is the default attribute of its type
             within the object. Only applies to certain attribute and data types.
+        pinned (bool, kwargs): A flag indicating that the attribute has been noted for importance.
+        security_labels (SecurityLabels, kwargs): A list of Security Labels corresponding to the
+            Intel item (NOTE: Setting this parameter will replace any existing tag(s) with
+            the one(s) specified).
         source (str, kwargs): The attribute source.
         type (str, kwargs): The attribute type.
-        value (str, kwargs): Attribute value.
+        value (str, kwargs): The attribute value.
     """
 
     def __init__(self, **kwargs):
@@ -95,3 +104,23 @@ def model(self, data: Union['CaseAttributeModel', dict]):
             self._model = type(self.model)(**data)
         else:
             raise RuntimeError(f'Invalid data type: {type(data)} provided.')
+
+    @property
+    def security_labels(self) -> Iterator['SecurityLabel']:
+        """Yield SecurityLabel from SecurityLabels."""
+        # first-party
+        from tcex.api.tc.v3.security_labels.security_label import SecurityLabels
+
+        yield from self._iterate_over_sublist(SecurityLabels)
+
+    def stage_security_label(self, data: Union[dict, 'ObjectABC', 'SecurityLabelModel']):
+        """Stage security_label on the object."""
+        if isinstance(data, ObjectABC):
+            data = data.model
+        elif isinstance(data, dict):
+            data = SecurityLabelModel(**data)
+
+        if not isinstance(data, SecurityLabelModel):
+            raise RuntimeError('Invalid type passed in to stage_security_label')
+        data._staged = True
+        self.model.security_labels.data.append(data)

tcex/api/tc/v3/case_attributes/case_attribute_filter.py

@@ -121,6 +121,15 @@ def owner_name(self, operator: Enum, owner_name: str):
         """
         self._tql.add_filter('ownerName', operator, owner_name, TqlType.STRING)
 
+    def pinned(self, operator: Enum, pinned: bool):
+        """Filter Pinned based on **pinned** keyword.
+
+        Args:
+            operator: The operator enum for the filter.
+            pinned: Whether or not the attribute is pinned with importance.
+        """
+        self._tql.add_filter('pinned', operator, pinned, TqlType.BOOLEAN)
+
     def source(self, operator: Enum, source: str):
         """Filter Source based on **source** keyword.