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_63.md

@@ -0,0 +1,9 @@
+
+#### Scripts
+
+##### StixCreator
+
+
+- Added an option to enter a flag for creating SCO indicators.
+- Updated the process of generating IDs for SDO indicators.
+- Updated the Docker image to: *demisto/py3-tools:1.0.0.55284*.

Packs/CommonScripts/Scripts/StixCreator/StixCreator.py

@@ -6,7 +6,7 @@
 
 ''' 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 Tool, CourseOfAction
@@ -23,9 +23,9 @@
     "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",
+    "host": "domain-name:value",
     "registry key": "windows-registry-key:key"
 }
 
@@ -42,6 +42,38 @@
     "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 = {
+    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: 'asn',
+}
+
+XSOAR_TYPES_TO_STIX_SDO = {
+    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: 'instruction-set',
+    ThreatIntel.ObjectsNames.REPORT: 'report',
+    ThreatIntel.ObjectsNames.THREAT_ACTOR: 'threat-actor',
+    ThreatIntel.ObjectsNames.TOOL: 'tool',
+    ThreatIntel.ObjectsNames.MALWARE: 'malware',
+    FeedIndicatorType.CVE: 'vulnerability',
+}
+
 
 def hash_type(value: str) -> str:  # pragma: no cover
     length = len(value)
@@ -68,10 +100,58 @@ 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.
+    """
+    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.
+    """
+    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 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):
@@ -83,6 +163,7 @@ def main():
         except:     # noqa: E722
             return_error('indicators argument is invalid json object')
 
+    demisto.debug(f'StixCreator {demisto.args()=}\n{user_args=}\n{all_args=}')
     indicators = []
 
     for indicator_fields in all_args:
@@ -95,75 +176,88 @@ def main():
         else:
             value = all_args[indicator_fields].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(all_args[indicator_fields], stix_type, value)
+            indicator = {
+                "type": stix_type,
+                "spec_version": "2.1",
+                "value": value,
+                "id": stix_id
+            }
+            indicators.append(indicator)
+        else:
+            demisto_score = all_args[indicator_fields].get('score', '').lower()
 
-        if demisto_score in ["bad", "malicious"]:
-            kwargs["score"] = "High"
+            if demisto_score in ["bad", "malicious"]:
+                kwargs["score"] = "High"
 
-        elif demisto_score == "suspicious":
-            kwargs["score"] = "Medium"
+            elif demisto_score == "suspicious":
+                kwargs["score"] = "Medium"
 
-        elif demisto_score in ["good", "benign"]:
-            kwargs["score"] = "None"
+            elif demisto_score in ["good", "benign"]:
+                kwargs["score"] = "None"
 
-        else:
-            kwargs["score"] = "Not Specified"
+            else:
+                kwargs["score"] = "Not Specified"
 
-        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', '')
+            stix_type = XSOAR_TYPES_TO_STIX_SDO.get(demisto_indicator_type, 'indicator')
+            stix_id = create_sdo_stix_uuid(all_args[indicator_fields], stix_type, value)
+            kwargs["id"] = stix_id
 
-        kwargs = {k: v for k, v in kwargs.items() if v}  # Removing keys with empty strings
+            kwargs["created"] = dateparser.parse(all_args[indicator_fields].get('timestamp', ''))
+            kwargs["modified"] = dateparser.parse(all_args[indicator_fields].get('lastSeen', f'{kwargs["created"]}'))
+            kwargs["labels"] = [demisto_indicator_type.lower()]
+            kwargs["description"] = all_args[indicator_fields].get('description', '')
 
-        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 = {k: v for k, v in kwargs.items() if v}  # Removing keys with empty strings
 
-            indicators.append(indicator)
+            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)
 
-        except KeyError:
+            except KeyError:
 
-            demisto.debug(f"{demisto_indicator_type} isn't an SCO checking other IOC types")
+                demisto.debug(f"{demisto_indicator_type} isn't a SCO, checking other IOC types")
 
-            try:
-                indicator_type = demisto_indicator_type.lower()
+                try:
+                    indicator_type = demisto_indicator_type.lower()
 
-                if indicator_type == 'cve':
-                    kwargs["external_references"] = [ExternalReference(source_name="cve", external_id=value)]
+                    if indicator_type == 'cve':
+                        kwargs["external_references"] = [ExternalReference(source_name="cve", external_id=value)]
 
-                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)]
+                    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)]
 
-                    except KeyError:
-                        pass
+                        except KeyError:
+                            pass
 
-                elif indicator_type == 'malware':
+                    elif indicator_type == 'malware':
 
-                    kwargs['is_family'] = argToBoolean(all_args[indicator_fields].get('ismalwarefamily', '').lower())
+                        kwargs['is_family'] = argToBoolean(all_args[indicator_fields].get('ismalwarefamily', 'False').lower())
 
-                indicator = SDOs[indicator_type](
-                    name=value,
-                    **kwargs
-                )
+                    indicator = SDOs[indicator_type](
+                        name=value,
+                        **kwargs
+                    )
 
-                indicators.append(indicator)
+                    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 +269,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.55284
 subtype: python3
 runas: DBotWeakRole
 tests: