Refactor/CIAC 832/Process Indicators as SCOs

Packs/CommonScripts/.secrets-ignore

@@ -109,4 +109,6 @@ PrivateIPRangeList
 http://www.testhxxp.com
 http://www.meow.com
 https://www.testhxxp.com
-https://www.meow.com
+https://www.meow.com
+701393b3b8e6ae6e70effcda7598a8cf92d0adb1aaeb5aa91c73004519644801
+uuid.uuid5(SCO_DET_ID_NAMESPACE

Packs/CommonScripts/ReleaseNotes/1_11_71.md

@@ -0,0 +1,8 @@
+
+#### Scripts
+
+##### StixCreator
+
+- Added a flag for creating SCO indicators.
+- Updated the process of generating IDs for SDO indicators such that a given indicator will have the same ID every run. This applies both when clicking on the button "Export (STIX)", and when running the script manually.
+- Updated the Docker image to: *demisto/py3-tools:1.0.0.57669*.

Packs/CommonScripts/Scripts/StixCreator/StixCreator.py

@@ -6,9 +6,9 @@
 
 ''' IMPORTS '''
 import json
-
+import uuid
 from stix2 import Bundle, ExternalReference, Indicator, Vulnerability
-from stix2 import AttackPattern, Campaign, Malware, Infrastructure, IntrusionSet, Report, ThreatActor
+from stix2 import AttackPattern, Campaign, Malware, Infrastructure, IntrusionSet, Report, ThreatActor  # TODO, ASN
 from stix2 import Tool, CourseOfAction
 from typing import Any, Callable
 
@@ -23,10 +23,11 @@
     "ipv6cidr": "ipv6-addr:value",
     "url": "url:value",
     "email": "email-message:sender_ref.value",
-    "username": "user-account:account_login",
+    "account": "user-account:account_login",
     "domain": "domain-name:value",
-    "hostname": "domain-name:value",
-    "registry key": "windows-registry-key:key"
+    "host": "domain-name:value",
+    "registry key": "windows-registry-key:key",
+    "asn": "autonomous-system:name"
 }
 
 SDOs: dict[str, Callable] = {  # pragma: no cover
@@ -42,6 +43,46 @@
     "course of action": CourseOfAction
 }
 
+SCO_DET_ID_NAMESPACE = uuid.UUID('00abedb4-aa42-466c-9c01-fed23315a9b7')
+PAWN_UUID = uuid.uuid5(uuid.NAMESPACE_URL, 'https://www.paloaltonetworks.com')
+
+XSOAR_TYPES_TO_STIX_SCO = {   # pragma: no cover
+    FeedIndicatorType.CIDR: 'ipv4-addr',
+    FeedIndicatorType.DomainGlob: 'domain-name',
+    FeedIndicatorType.IPv6: 'ipv6-addr',
+    FeedIndicatorType.IPv6CIDR: 'ipv6-addr',
+    FeedIndicatorType.Account: 'user-account',
+    FeedIndicatorType.Domain: 'domain-name',
+    FeedIndicatorType.Email: 'email-addr',
+    FeedIndicatorType.IP: 'ipv4-addr',
+    FeedIndicatorType.Registry: 'windows-registry-key',
+    FeedIndicatorType.File: 'file',
+    FeedIndicatorType.URL: 'url',
+    FeedIndicatorType.Software: 'software',
+    FeedIndicatorType.AS: 'autonomous-system',
+}
+
+XSOAR_TYPES_TO_STIX_SDO = {  # pragma: no cover
+    ThreatIntel.ObjectsNames.ATTACK_PATTERN: 'attack-pattern',
+    ThreatIntel.ObjectsNames.CAMPAIGN: 'campaign',
+    ThreatIntel.ObjectsNames.COURSE_OF_ACTION: 'course-of-action',
+    ThreatIntel.ObjectsNames.INFRASTRUCTURE: 'infrastructure',
+    ThreatIntel.ObjectsNames.INTRUSION_SET: 'intrusion-set',
+    ThreatIntel.ObjectsNames.REPORT: 'report',
+    ThreatIntel.ObjectsNames.THREAT_ACTOR: 'threat-actor',
+    ThreatIntel.ObjectsNames.TOOL: 'tool',
+    ThreatIntel.ObjectsNames.MALWARE: 'malware',
+    FeedIndicatorType.CVE: 'vulnerability',
+}
+
+
+HASH_TYPE_TO_STIX_HASH_TYPE = {  # pragma: no cover
+    'md5': 'MD5',
+    'sha1': 'SHA-1',
+    'sha256': 'SHA-256',
+    'sha512': 'SHA-512',
+}
+
 
 def hash_type(value: str) -> str:  # pragma: no cover
     length = len(value)
@@ -68,10 +109,114 @@ def guess_indicator_type(type_: str, val: str) -> str:
     return (auto_detect_indicator_type(val) or type_).lower()
 
 
+def create_sco_stix_uuid(xsoar_indicator: dict, stix_type: Optional[str], value: str) -> str:
+    """
+    Create uuid for SCO objects.
+    Args:
+        xsoar_indicator: dict - The XSOAR representation of the indicator.
+        stix_type: Optional[str] - The indicator type according to STIX.
+        value: str - The value of the indicator.
+    Returns:
+        The uuid that represents the indicator according to STIX.
+    """
+    if stixid := xsoar_indicator.get('CustomFields', {}).get('stixid'):
+        return stixid
+    if stix_type == 'user-account':
+        account_type = xsoar_indicator.get('CustomFields', {}).get('accounttype')
+        user_id = xsoar_indicator.get('CustomFields', {}).get('userid')
+        unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE,
+                               f'{{"account_login":"{value}","account_type":"{account_type}","user_id":"{user_id}"}}')
+    elif stix_type == 'windows-registry-key':
+        unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"key":"{value}"}}')
+    elif stix_type == 'file':
+        if 'md5' == get_hash_type(value):
+            unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"MD5":"{value}"}}}}')
+        elif 'sha1' == get_hash_type(value):
+            unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-1":"{value}"}}}}')
+        elif 'sha256' == get_hash_type(value):
+            unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-256":"{value}"}}}}')
+        elif 'sha512' == get_hash_type(value):
+            unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"hashes":{{"SHA-512":"{value}"}}}}')
+        else:
+            unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"value":"{value}"}}')
+    else:
+        unique_id = uuid.uuid5(SCO_DET_ID_NAMESPACE, f'{{"value":"{value}"}}')
+
+    return f'{stix_type}--{unique_id}'
+
+
+def create_sdo_stix_uuid(xsoar_indicator: dict, stix_type: Optional[str], value: str) -> str:
+    """
+    Create uuid for SDO objects.
+    Args:
+        xsoar_indicator: dict - The XSOAR representation of the indicator.
+        stix_type: Optional[str] - The indicator type according to STIX.
+        value: str - The value of the indicator.
+    Returns:
+        The uuid that represents the indicator according to STIX.
+    """
+    if stixid := xsoar_indicator.get('CustomFields', {}).get('stixid'):
+        return stixid
+    if stix_type == 'attack-pattern':
+        if mitre_id := xsoar_indicator.get('CustomFields', {}).get('mitreid'):
+            unique_id = uuid.uuid5(PAWN_UUID, f'{stix_type}:{mitre_id}')
+        else:
+            unique_id = uuid.uuid5(PAWN_UUID, f'{stix_type}:{value}')
+    else:
+        unique_id = uuid.uuid5(PAWN_UUID, f'{stix_type}:{value}')
+
+    return f'{stix_type}--{unique_id}'
+
+
+def add_file_fields_to_indicator(xsoar_indicator: Dict, value: str) -> Dict:
+    """
+    Create the hashes dictionary for the indicator object.
+    Args:
+        xsoar_indicator: Dict - The XSOAR representation of the indicator.
+        value: str - The value of the indicator.
+    Returns:
+        The dictionary with the file hashes.
+    """
+    hashes_dict = {}
+    for hash_kind in ['md5', 'sha1', 'sha256', 'sha512']:
+        if get_hash_type(value) == hash_kind:
+            hashes_dict[HASH_TYPE_TO_STIX_HASH_TYPE.get(hash_kind)] = value
+        elif hash_kind in xsoar_indicator:
+            hashes_dict[HASH_TYPE_TO_STIX_HASH_TYPE.get(hash_kind)] = xsoar_indicator.get(hash_kind, '')
+    return hashes_dict
+
+
+def create_stix_sco_indicator(stix_id: Optional[str], stix_type: Optional[str], value: str, xsoar_indicator: Dict) -> Dict:
+    """
+    Create stix sco indicator object.
+    Args:
+        stix_id: Optional[str] - The stix id of the indicator.
+        stix_type: str - the stix type of the indicator.
+        xsoar_indicator: Dict - The XSOAR representation of the indicator.
+        value: str - The value of the indicator.
+    Returns:
+        The Dictionary representing the stix indicator.
+    """
+    stix_indicator: Dict[str, Any] = {
+        "type": stix_type,
+        "spec_version": "2.1",
+        "id": stix_id
+    }
+    if stix_type == 'file':
+        stix_indicator['hashes'] = add_file_fields_to_indicator(xsoar_indicator, value)
+    elif stix_type == 'autonomous-system':
+        stix_indicator['number'] = value
+        stix_indicator['name'] = xsoar_indicator.get('name', '')
+    else:
+        stix_indicator['value'] = value
+    return stix_indicator
+
+
 def main():
 
     user_args = demisto.args().get('indicators', 'Unknown')
     doubleBackslash = demisto.args().get('doubleBackslash', True)
+    is_sco = argToBoolean(demisto.args().get('sco_flag', False))
     all_args = {}
 
     if isinstance(user_args, dict):
@@ -88,82 +233,92 @@ def main():
     for indicator_fields in all_args:
         kwargs: dict[str, Any] = {"allow_custom": True}
 
-        demisto_indicator_type = all_args[indicator_fields].get('indicator_type', 'Unknown')
+        xsoar_indicator = all_args[indicator_fields]
+        demisto_indicator_type = xsoar_indicator.get('indicator_type', 'Unknown')
 
         if doubleBackslash:
-            value = all_args[indicator_fields].get('value', '').replace('\\', r'\\')
+            value = xsoar_indicator.get('value', '').replace('\\', r'\\')
         else:
-            value = all_args[indicator_fields].get('value', '')
+            value = xsoar_indicator.get('value', '')
 
-        demisto_score = all_args[indicator_fields].get('score', '').lower()
+        if demisto_indicator_type in XSOAR_TYPES_TO_STIX_SCO and is_sco:
+            stix_type = XSOAR_TYPES_TO_STIX_SCO.get(demisto_indicator_type)
+            stix_id = create_sco_stix_uuid(xsoar_indicator, stix_type, value)
+            stix_indicator = create_stix_sco_indicator(stix_id, stix_type, value, xsoar_indicator)
+            indicators.append(stix_indicator)
 
-        if demisto_score in ["bad", "malicious"]:
-            kwargs["score"] = "High"
+        else:
+            demisto_score = xsoar_indicator.get('score', '').lower()
 
-        elif demisto_score == "suspicious":
-            kwargs["score"] = "Medium"
+            if demisto_score in ["bad", "malicious"]:
+                kwargs["score"] = "High"
 
-        elif demisto_score in ["good", "benign"]:
-            kwargs["score"] = "None"
+            elif demisto_score == "suspicious":
+                kwargs["score"] = "Medium"
 
-        else:
-            kwargs["score"] = "Not Specified"
+            elif demisto_score in ["good", "benign"]:
+                kwargs["score"] = "None"
 
-        kwargs["created"] = dateparser.parse(all_args[indicator_fields].get('timestamp', ''))
-        kwargs["modified"] = dateparser.parse(all_args[indicator_fields].get('lastSeen', f'{kwargs["created"]}'))
-        kwargs["id"] = all_args[indicator_fields].get('stixid', '')
-        kwargs["labels"] = [demisto_indicator_type.lower()]
-        kwargs["description"] = all_args[indicator_fields].get('description', '')
+            else:
+                kwargs["score"] = "Not Specified"
 
-        kwargs = {k: v for k, v in kwargs.items() if v}  # Removing keys with empty strings
+            stix_type = XSOAR_TYPES_TO_STIX_SDO.get(demisto_indicator_type, 'indicator')
+            stix_id = create_sdo_stix_uuid(xsoar_indicator, stix_type, value)
+            kwargs["id"] = stix_id
 
-        try:
-            indicator_type = demisto_indicator_type.lower().replace("-", "")
-            if indicator_type == 'file':
-                indicator_type = hash_type(value)
-            if indicator_type not in SCOs and indicator_type not in SDOs:
-                indicator_type = guess_indicator_type(indicator_type, value)
-            indicator = Indicator(pattern=f"[{SCOs[indicator_type]} = '{value}']",
-                                  pattern_type='stix',
-                                  **kwargs)
+            kwargs["created"] = dateparser.parse(xsoar_indicator.get('timestamp', ''))
+            kwargs["modified"] = dateparser.parse(xsoar_indicator.get('lastSeen', f'{kwargs["created"]}'))
+            kwargs["labels"] = [demisto_indicator_type.lower()]
+            kwargs["description"] = xsoar_indicator.get('description', '')
 
-            indicators.append(indicator)
+            kwargs = {k: v for k, v in kwargs.items() if v}  # Removing keys with empty strings
 
-        except KeyError:
+            try:
+                indicator_type = demisto_indicator_type.lower().replace("-", "")
+                if indicator_type == 'file':
+                    indicator_type = hash_type(value)
+                if indicator_type not in SCOs and indicator_type not in SDOs:
+                    indicator_type = guess_indicator_type(indicator_type, value)
+                indicator = Indicator(pattern=f"[{SCOs[indicator_type]} = '{value}']",
+                                      pattern_type='stix',
+                                      **kwargs)
+                indicators.append(indicator)
 
-            demisto.debug(f"{demisto_indicator_type} isn't an SCO checking other IOC types")
+            except KeyError:
 
-            try:
-                indicator_type = demisto_indicator_type.lower()
+                demisto.debug(f"{demisto_indicator_type} isn't a SCO, checking other IOC types")
 
-                if indicator_type == 'cve':
-                    kwargs["external_references"] = [ExternalReference(source_name="cve", external_id=value)]
+                try:
+                    indicator_type = demisto_indicator_type.lower()
 
-                elif indicator_type == "attack pattern":
-                    try:
-                        mitreid = all_args[indicator_fields].get('mitreid', '')
-                        if mitreid:
-                            kwargs["external_references"] = [ExternalReference(source_name="mitre", external_id=mitreid)]
+                    if indicator_type == 'cve':
+                        kwargs["external_references"] = [ExternalReference(source_name="cve", external_id=value)]
 
-                    except KeyError:
-                        pass
+                    elif indicator_type == "attack pattern":
+                        try:
+                            mitreid = xsoar_indicator.get('mitreid', '')
+                            if mitreid:
+                                kwargs["external_references"] = [ExternalReference(source_name="mitre", external_id=mitreid)]
 
-                elif indicator_type == 'malware':
+                        except KeyError:
+                            pass
 
-                    kwargs['is_family'] = argToBoolean(all_args[indicator_fields].get('ismalwarefamily', '').lower())
+                    elif indicator_type == 'malware':
 
-                indicator = SDOs[indicator_type](
-                    name=value,
-                    **kwargs
-                )
+                        kwargs['is_family'] = argToBoolean(xsoar_indicator.get('ismalwarefamily', 'False').lower())
 
-                indicators.append(indicator)
+                    indicator = SDOs[indicator_type](
+                        name=value,
+                        **kwargs
+                    )
+
+                    indicators.append(indicator)
 
-            except (KeyError, TypeError) as e:
-                demisto.info(
-                    "Indicator type: {}, with the value: {} is not STIX compatible".format(demisto_indicator_type, value))
-                demisto.info("Export failure excpetion: {}".format(e))
-                continue
+                except (KeyError, TypeError) as e:
+                    demisto.info(
+                        "Indicator type: {}, with the value: {} is not STIX compatible".format(demisto_indicator_type, value))
+                    demisto.info("Export failure excpetion: {}".format(e))
+                    continue
 
     if len(indicators) > 1:
         bundle = Bundle(indicators, allow_custom=True)
@@ -175,12 +330,14 @@ def main():
                               raw_response=str(bundle)))
 
     elif len(indicators) == 1:
+        bundle = Bundle(indicators, allow_custom=True)
+        bundle_obj = bundle.get('objects', [])[0]
         context = {
-            'StixExportedIndicators(val.pattern && val.pattern == obj.pattern)': json.loads(str(indicators[0]))
+            'StixExportedIndicators(val.pattern && val.pattern == obj.pattern)': json.loads(str(bundle_obj))
         }
         res = (CommandResults(readable_output="",
                               outputs=context,
-                              raw_response=str(indicators[0])))
+                              raw_response=str(bundle_obj)))
     else:
         context = {
             'StixExportedIndicators': {}

Packs/CommonScripts/Scripts/StixCreator/StixCreator.yml

@@ -39,7 +39,7 @@ outputs:
   type: date
 scripttarget: 0
 runonce: false
-dockerimage: demisto/py3-tools:1.0.0.49703
+dockerimage: demisto/py3-tools:1.0.0.57669
 subtype: python3
 runas: DBotWeakRole
 tests: