Stix 2.1 ids

stix_shifter_utils/stix_translation/src/json_to_stix/id_contributing_properties.json

@@ -0,0 +1,20 @@
+{
+    "artifact": [[ "MD5", "SHA-1", "SHA-256", "SHA-512" ], "payload_bin"],
+    "autonomous-system": ["number"],
+    "directory": ["path"],
+    "domain-name": ["value"],
+    "email-addr": ["value"],
+    "email-message": ["from_ref", "subject", "body"],
+    "file": [[ "MD5", "SHA-1", "SHA-256", "SHA-512" ], "name", "extensions"],
+    "ipv4-addr": ["value"],
+    "ipv6-addr": ["value"],
+    "mac-addr": ["value"],
+    "mutex": ["name"],
+    "network-traffic": ["start", "src_ref", "dst_ref", "src_port", "dst_port", "protocols"],
+    "process": [],
+    "software": ["name", "cpe", "vendor", "version"],
+    "url": ["value"],
+    "user-account": ["account_type", "user_id", "account_login"],
+    "windows-registry-key": ["key", "values"],
+    "x509-certificate": [[ "MD5", "SHA-1", "SHA-256", "SHA-512" ], "serial_number"]
+}

stix_shifter_utils/stix_translation/src/json_to_stix/json_to_stix_translator.py

@@ -350,18 +350,34 @@ def _transform(self, object_map, observation, ds_map, ds_key, obj):
     # STIX 2.1 helper methods
     def _generate_and_apply_deterministic_id(self, object_id_map, cybox_objects):
         # Generates ID based on common namespace and SCO properties (omitting id and spec_version)
-        # TODO: References may need to be include as part of the ID generation
+        # TODO: Handle references when part of ID contributing properties
+
+        with open("stix_shifter_utils/stix_translation/src/json_to_stix/id_contributing_properties.json", 'r') as f:
+            contributing_properties_definitions =  json.load(f)
+
         for key, cybox in cybox_objects.items():
-            cybox_type = ""
-            # set id mapping key to original id
             object_id_map[key] = ""
             cybox_properties = {}
-            for property, value in cybox.items():
-                if property == "type":
-                    cybox_type = value
-                if not (property == "id" or re.match(".*_ref$", property)):
-                    cybox_properties[property] = value
-            unique_id = cybox_type + "--" + str(uuid.uuid5(namespace=uuid.UUID(UUID5_NAMESPACE), name=json.dumps(cybox_properties)))
+            cybox_type = cybox.get("type")
+            contributing_properties = contributing_properties_definitions.get(cybox_type)
+
+            if contributing_properties:
+                for contr_prop in contributing_properties:
+                    if type(contr_prop) is list: # list of hash types
+                        for hashtype in contr_prop:
+                            hash_prop = "hashes.{}".format(hashtype)
+                            if hash_prop in cybox:
+                                cybox_properties[hash_prop] = cybox[hash_prop]
+                                break
+                    elif contr_prop in cybox and not re.match(".*_ref$", contr_prop): # chicken and egg problem with refs
+                        cybox_properties[contr_prop] = cybox[contr_prop] 
+                if cybox_properties:
+                    unique_id = cybox_type + "--" + str(uuid.uuid5(namespace=uuid.UUID(UUID5_NAMESPACE), name=json.dumps(cybox_properties)))
+                else:
+                    self.logger.error("STIX object '{}' needs at least one of the following properties to generate ID {}".format(cybox_type, contributing_properties))
+            else: # STIX process or custom object used UUID4 for identifier
+                unique_id = "{}--{}".format(cybox_type, str(uuid.uuid4()))
+
             # set id mapping value to new id
             object_id_map[key] = unique_id
             # replace old id with new

tests/stix_translation/test_results_translation.py

@@ -31,6 +31,16 @@
             "file_path": "C:/some/path/"
         }
 
+CYBOX_ID = {
+            "source-ipv4-addr": "ipv4-addr--0b6a89e3-e345-51b7-a8ee-aaff7ebf2df5", 
+            "dest-ipv4-addr": "ipv4-addr--cb8e152d-60f0-596a-81e4-a22cc4a7f063", 
+            "url": "url--8265905f-c609-52e3-ae52-6681bcd6086d", 
+            "user-account": "user-account--3cd7ffc9-89f7-5b58-948c-117ec9b3e22a", 
+            "network-traffic": "network-traffic--2ec70516-29b5-59f3-9743-3b93e97db6d8",
+            "file": "file--243f1b5f-0391-501c-bed0-17e9f204f1d2",
+            "directory": "directory--9ce39e76-d59e-5db2-8f0e-2001f689ea9d"
+        }
+
 OPTIONS = {}
 
 
@@ -182,26 +192,31 @@ def test_STIX_2_1_cybox_observables(self):
         assert "src_port" in network_traffic_object and network_traffic_object['src_port'] == 3000
         assert "dst_port" in network_traffic_object and network_traffic_object['dst_port'] == 2000
         assert "protocols" in network_traffic_object and network_traffic_object['protocols'] == ['tcp'] 
+        assert "id" in network_traffic_object and network_traffic_object['id'] == CYBOX_ID["network-traffic"]
 
         # destination ipv4-addr
         destination_ipv4_object = TestTransform.get_first_cybox_of_id_stix_2_1(result_bundle_objects, network_traffic_object["dst_ref"])
         assert "type" in destination_ipv4_object and destination_ipv4_object['type'] == 'ipv4-addr'
         assert "value" in destination_ipv4_object and destination_ipv4_object['value'] == DATA["dest_ipaddr"]
+        assert "id" in destination_ipv4_object and destination_ipv4_object['id'] == CYBOX_ID["dest-ipv4-addr"]
 
         # source ipv4-addr
         source_ipv4_object = TestTransform.get_first_cybox_of_id_stix_2_1(result_bundle_objects, network_traffic_object["src_ref"])
         assert "type" in source_ipv4_object and source_ipv4_object['type'] == 'ipv4-addr'
         assert "value" in source_ipv4_object and source_ipv4_object['value'] == DATA["source_ipaddr"]
+        assert "id" in source_ipv4_object and source_ipv4_object['id'] == CYBOX_ID["source-ipv4-addr"]
 
         # url
         url_object = TestTransform.get_first_cybox_of_type_stix_2_1(result_bundle_objects, 'url')
         assert url_object, 'url object type not found'
         assert "value" in url_object and url_object['value'] == DATA['url']
+        assert "id" in url_object and url_object['id'] == CYBOX_ID["url"]
 
         # user-account
         user_account_object = TestTransform.get_first_cybox_of_type_stix_2_1(result_bundle_objects, 'user-account')
         assert  user_account_object, 'user-account object type not found'
         assert  "user_id" in user_account_object and user_account_object['user_id'] == DATA['username']
+        assert "id" in user_account_object and user_account_object['id'] == CYBOX_ID["user-account"]
 
         # file
         file_object = TestTransform.get_first_cybox_of_type_stix_2_1(result_bundle_objects, 'file')
@@ -212,10 +227,12 @@ def test_STIX_2_1_cybox_observables(self):
         assert "MD5" in hashes and hashes["MD5"] == DATA["md5hash"]
         assert "SHA-256" in hashes and hashes["SHA-256"] == DATA["sha256hash"] 
         assert "parent_directory_ref" in file_object
+        assert "id" in file_object and file_object['id'] == CYBOX_ID["file"]
 
         # directory
         directory_object = TestTransform.get_first_cybox_of_id_stix_2_1(result_bundle_objects, file_object["parent_directory_ref"])
         assert directory_object, 'directory object type not found'
         assert "path" in directory_object and directory_object["path"] == DATA["file_path"]
+        assert "id" in directory_object and directory_object['id'] == CYBOX_ID["directory"]