Repository: oasis-open/cti-stix-validator
Pull request #78 (2018-11-05 19:58:46)

Addressing Issue #72: Enforce referenced objects in relationship objects
Added a command-line feature (--enforce_refs) which users to get a warning message when not all SDOs referenced by SROs are contained in the same bundle. 


Patch:
diff --git a/stix2validator/musts.py b/stix2validator/musts.py
index e0d4bae..f30bd16 100644
--- a/stix2validator/musts.py
+++ b/stix2validator/musts.py
@@ -133,11 +133,11 @@ def marking_selector_syntax(instance):
                     try:
                         idx = int(index_match.group(1))
                         obj = obj[idx]
-                    except IndexError as e:
+                    except IndexError:
                         yield JSONError("'%s' is not a valid selector because"
                                         " %s is not a valid index."
                                         % (selector, idx), instance['id'])
-                    except KeyError as e:
+                    except KeyError:
                         yield JSONError("'%s' is not a valid selector because"
                                         " '%s' is not a list."
                                         % (selector, prev_segmt), instance['id'])
@@ -148,7 +148,7 @@ def marking_selector_syntax(instance):
                         yield JSONError("'%s' is not a valid selector because"
                                         " %s is not a property."
                                         % (selector, e), instance['id'])
-                    except TypeError as e:
+                    except TypeError:
                         yield JSONError("'%s' is not a valid selector because"
                                         " '%s' is not a property."
                                         % (selector, segmt), instance['id'])
diff --git a/stix2validator/shoulds.py b/stix2validator/shoulds.py
index 1f81ab5..6605d2f 100644
--- a/stix2validator/shoulds.py
+++ b/stix2validator/shoulds.py
@@ -1033,6 +1033,33 @@ def extref_hashes(instance):
                                  % (src), instance['id'], 'extref-hashes')
 
 
+def enforce_relationship_refs(instance):
+    """Ensures that all SDOs being referenced by the SRO are contained
+    within the same bundle"""
+    if instance['type'] != 'bundle' or 'objects' not in instance:
+        return
+
+    rel_references = set()
+
+    """Find and store all ids"""
+    for obj in instance['objects']:
+        if obj['type'] != 'relationship':
+            rel_references.add(obj['id'])
+
+    """Check if id has been encountered"""
+    for obj in instance['objects']:
+        if obj['type'] == 'relationship':
+            if obj['source_ref'] not in rel_references:
+                yield JSONError("Relationship object %s makes reference to %s "
+                                "Which is not found in current bundle "
+                                % (obj['id'], obj['source_ref']), 'enforce-relationship-refs')
+
+            if obj['target_ref'] not in rel_references:
+                yield JSONError("Relationship object %s makes reference to %s "
+                                "Which is not found in current bundle "
+                                % (obj['id'], obj['target_ref']), 'enforce-relationship-refs')
+
+
 def duplicate_ids(instance):
     """Ensure objects with duplicate IDs have different `modified` timestamps.
     """
@@ -1143,6 +1170,7 @@ CHECKS = {
     'marking-definition-type': vocab_marking_definition,
     'relationship-types': relationships_strict,
     'duplicate-ids': duplicate_ids,
+    'enforce_relationship_refs': enforce_relationship_refs,
     'all-vocabs': [
         vocab_attack_motivation,
         vocab_attack_resource_level,
@@ -1198,6 +1226,10 @@ def list_shoulds(options):
     """Construct the list of 'SHOULD' validators to be run by the validator.
     """
     validator_list = []
+    # --enforce_refs
+    # enable checking references in bundles if option selected
+    if options.enforce_refs is True:
+        validator_list.append(CHECKS['enforce_relationship_refs'])
 
     # Default: enable all
     if not options.disabled and not options.enabled:
diff --git a/stix2validator/util.py b/stix2validator/util.py
index 98eb2bd..10e7d86 100644
--- a/stix2validator/util.py
+++ b/stix2validator/util.py
@@ -251,6 +251,15 @@ def parse_args(cmd_args, is_script=False):
         help="Clear the cache of external source values after validation."
     )
 
+    parser.add_argument(
+        "--enforce_refs",
+        dest="enforce_refs",
+        action="store_true",
+        default=False,
+        help="Ensures that all SDOs being referenced by the SRO are contained "
+             "within the same bundle."
+    )
+
     args = parser.parse_args(cmd_args)
 
     if not is_script:
@@ -292,13 +301,15 @@ class ValidationOptions(object):
             validation.
         clear_cache: Specifies that the cache of values from external sources
             should be cleared after validation.
+        enforce_refs:Ensures that all SDOs being referenced by the SRO are
+            contained within the same bundle
 
     """
     def __init__(self, cmd_args=None, verbose=False, silent=False,
                  files=None, recursive=False, schema_dir=None,
                  disabled="", enabled="", strict=False,
                  strict_types=False, strict_properties=False, no_cache=False,
-                 refresh_cache=False, clear_cache=False):
+                 refresh_cache=False, clear_cache=False, enforce_refs=False):
 
         if cmd_args is not None:
             self.verbose = cmd_args.verbose
@@ -314,6 +325,7 @@ class ValidationOptions(object):
             self.no_cache = cmd_args.no_cache
             self.refresh_cache = cmd_args.refresh_cache
             self.clear_cache = cmd_args.clear_cache
+            self.enforce_refs = cmd_args.enforce_refs
         else:
             # input options
             self.files = files
@@ -328,6 +340,7 @@ class ValidationOptions(object):
             self.strict_properties = strict_properties
             self.disabled = disabled
             self.enabled = enabled
+            self.enforce_refs = enforce_refs
 
             # cache options
             self.no_cache = no_cache


Test patch:
diff --git a/stix2validator/test/relationship_tests.py b/stix2validator/test/relationship_tests.py
index 76e20e8..1ad821f 100644
--- a/stix2validator/test/relationship_tests.py
+++ b/stix2validator/test/relationship_tests.py
@@ -97,3 +97,37 @@ class RelationshipTestCases(ValidatorTest):
         relationship = copy.deepcopy(self.valid_relationship)
         del relationship['relationship_type']
         self.assertFalseWithOptions(relationship)
+
+    def test_enforce_refs(self):
+        invalid_bundle = u"""
+        {
+          "type": "bundle",
+          "id": "bundle--44af6c39-c09b-49c5-9de2-394224b04982",
+          "spec_version": "2.0",
+          "objects": [
+            {
+          "type": "malware",
+          "id": "malware--fdd60b30-b67c-11e3-b0b9-f01faf20d111",
+          "created": "2014-02-20T09:16:08.989Z",
+          "modified": "2014-02-20T09:16:08.989Z",
+          "name": "Poison Ivy",
+          "labels": [
+            "remote-access-trojan"
+          ]
+        },
+        {
+          "type": "relationship",
+          "id": "relationship--f191e70e-1736-47c3-b0f9-fdfe01387eb1",
+          "created": "2014-02-20T09:16:08.989Z",
+          "modified": "2014-02-20T09:16:08.989Z",
+          "relationship_type": "indicates",
+          "source_ref": "indicator--a932fcc6-e032-176c-126f-cb970a5a1adf",
+          "target_ref": "malware--fdd60b30-b67c-11e3-b0b9-f01faf20d111"
+        }
+          ]
+        }
+        """
+        self.options.enforce_refs = True
+        results = validate_string(invalid_bundle, self.options)
+        self.options.enforce_refs = False
+        self.assertTrue(len(results.errors) >= 1)