Repository: oasis-open/cti-stix-validator
Pull request #191 (2022-04-01 15:05:19)

interop validation additions


Patch:
diff --git a/setup.py b/setup.py
index d5b17a5..7914274 100644
--- a/setup.py
+++ b/setup.py
@@ -26,6 +26,7 @@ install_requires = [
     'requests_cache',
     'simplejson',
     'stix2-patterns>=0.4.1',
+    'attrs<22.0,>21.2',
 ]
 
 setup(
diff --git a/stix2validator/errors.py b/stix2validator/errors.py
index 02be1fd..183a37f 100644
--- a/stix2validator/errors.py
+++ b/stix2validator/errors.py
@@ -130,6 +130,8 @@ def pretty_error(error, verbose=False):
     elif error.validator == 'pattern' and 'where_sighted_refs' in error.schema_path:
         msg = "'where_sighted_refs' must refer to Identity Objects"
 
+    elif error.validator == 'maxContains' and 'marking-definition' in error.schema['description']:
+        msg = "More than one tlp marking found in marking-definitions. Only one tlp marking is allowed."
     # Reword empty array errors
     elif type(error.instance) is list and len(error.instance) == 0:
         msg = re.sub(r"\[\] is not valid .+$", 'empty arrays are not allowed',
diff --git a/stix2validator/schemas-2.1 b/stix2validator/schemas-2.1
index 5353a26..9d86da9 160000
--- a/stix2validator/schemas-2.1
+++ b/stix2validator/schemas-2.1
@@ -1,1 +1,1 @@
-Subproject commit 5353a26a5078d7967a1bb4abb8406a75bffd5dff
+Subproject commit 9d86da9b3a5c5416fa4fbd92c76d891662a77401
diff --git a/stix2validator/util.py b/stix2validator/util.py
index efe260b..f78e10a 100644
--- a/stix2validator/util.py
+++ b/stix2validator/util.py
@@ -165,6 +165,7 @@ def parse_args(cmd_args, is_script=False):
         Instance of ``ValidationOptions``
 
     """
+
     parser = argparse.ArgumentParser(
         description=__doc__,
         formatter_class=NewlinesHelpFormatter,
@@ -298,6 +299,14 @@ def parse_args(cmd_args, is_script=False):
         help="Clear the cache of external source values after validation."
     )
 
+    parser.add_argument(
+        "--interop",
+        dest="interop",
+        action="store_true",
+        default=False,
+        help="Run validator with interop validation settings"
+    )
+
     parser.add_argument(
         "--enforce-refs",
         dest="enforce_refs",
@@ -359,7 +368,7 @@ class ValidationOptions(object):
                  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, enforce_refs=False):
+                 refresh_cache=False, clear_cache=False, enforce_refs=False, interop=False):
 
         if cmd_args is not None:
             self.version = cmd_args.version
@@ -377,6 +386,7 @@ class ValidationOptions(object):
             self.refresh_cache = cmd_args.refresh_cache
             self.clear_cache = cmd_args.clear_cache
             self.enforce_refs = cmd_args.enforce_refs
+            self.interop = cmd_args.interop
         else:
             # input options
             self.version = version
@@ -399,6 +409,7 @@ class ValidationOptions(object):
             self.refresh_cache = refresh_cache
             self.clear_cache = clear_cache
 
+            self.interop = interop
         # Set the output level (e.g., quiet vs. verbose)
         if self.silent and self.verbose:
             raise ValueError('Error: Output can either be silent or verbose, but not both.')
diff --git a/stix2validator/v21/interop.py b/stix2validator/v21/interop.py
new file mode 100644
index 0000000..ae2b045
--- /dev/null
+++ b/stix2validator/v21/interop.py
@@ -0,0 +1,34 @@
+from .errors import JSONError
+
+
+def interop_created_by_ref(instance):
+    """Ensures that all SDOs reference a Producer"""
+    if instance['type'] != 'bundle' or 'objects' not in instance:
+        return
+
+    rel_references = set()
+    bad_references = set()
+
+    """Find and store all producer ids and general identity ids
+    use the bad references to inform on nature of error"""
+    for obj in instance['objects']:
+        if obj['type'] == 'identity':
+            if 'created_by_ref' in obj:
+                rel_references.add(obj['id'])
+            else:
+                bad_references.add(obj['id'])
+
+    """Check if id is present in producers or normal identity SDOs"""
+    for obj in instance['objects']:
+        if obj['type'] != 'identity':
+            if 'created_by_ref' not in obj:
+                yield JSONError("created_by_ref is not present in Object", obj['id'])
+            else:
+                if obj['created_by_ref'] in bad_references:
+                    yield JSONError("references %s as a producer "
+                                    "but the identity is missing the property created_by_ref"
+                                    % (obj['created_by_ref']), obj['id'])
+                elif obj['created_by_ref'] not in rel_references:
+                    yield JSONError("created_by_ref has value %s "
+                                    "which is not found in bundle"
+                                    % (obj['created_by_ref']), obj['id'])
diff --git a/stix2validator/validator.py b/stix2validator/validator.py
index a603de2..6bc533b 100644
--- a/stix2validator/validator.py
+++ b/stix2validator/validator.py
@@ -8,7 +8,8 @@ import os
 import re
 import sys
 
-from jsonschema import Draft7Validator, RefResolver, draft7_format_checker
+from jsonschema import (Draft202012Validator, RefResolver,
+                        draft202012_format_checker)
 from jsonschema import exceptions as schema_exceptions
 from jsonschema.validators import extend
 import simplejson as json
@@ -20,6 +21,7 @@ from .util import (DEFAULT_VER, ValidationOptions, clear_requests_cache,
                    init_requests_cache)
 from .v20 import musts as musts20
 from .v20 import shoulds as shoulds20
+from .v21 import interop
 from .v21 import musts as musts21
 from .v21 import shoulds as shoulds21
 
@@ -524,14 +526,14 @@ def ref_store(validator, ref, instance, schema):
         except FileNotFoundError:
             pass
 
-    return Draft7Validator.VALIDATORS['$ref'](validator, ref, instance, schema)
+    return Draft202012Validator.VALIDATORS['$ref'](validator, ref, instance, schema)
 
 
-STIXValidator = extend(Draft7Validator, {'$ref': ref_store})
+STIXValidator = extend(Draft202012Validator, {'$ref': ref_store})
 
 
 # Built-in checker only ensures emails contain an '@'; we want a more robust check
-@draft7_format_checker.checks('email')
+@draft202012_format_checker.checks('email')
 def is_email(instance):
     if not isinstance(instance, str):
         return True
@@ -546,7 +548,7 @@ def load_validator(schema_path, schema):
         schema: A Python object representation of the same schema.
 
     Returns:
-        An instance of Draft7Validator.
+        An instance of Draft202012Validator.
 
     """
     global SCHEMA_STORE
@@ -563,7 +565,7 @@ def load_validator(schema_path, schema):
         resolver.store[schema_id] = schema
     # RefResolver creates a new store internally; persist it so we can use the same mappings every time
     SCHEMA_STORE = resolver.store
-    validator = STIXValidator(schema, resolver=resolver, format_checker=draft7_format_checker)
+    validator = STIXValidator(schema, resolver=resolver, format_checker=draft202012_format_checker)
     return validator
 
 
@@ -623,7 +625,7 @@ def _get_error_generator(type, obj, schema_dir=None, version=DEFAULT_VER, defaul
     if schema_dir is None:
         default_path = True
         schema_dir = os.path.abspath(os.path.dirname(__file__) + '/schemas-'
-                                     + version + '/')
+                                     + version + '/schemas/')
 
     try:
         schema_path = find_schema(schema_dir, type)
@@ -695,6 +697,28 @@ def _get_shoulds(options):
         return shoulds21.list_shoulds(options)
 
 
+def find_version(obj, options, bundle_version=None, error_prefix=''):
+    if options.version:
+        version = options.version
+    elif options.version is None and 'spec_version' in obj:
+        version = obj['spec_version']
+    else:
+        version = DEFAULT_VER
+
+    if bundle_version == '2.0':
+        version = bundle_version
+
+    # Allow 2.0 objects in 2.1+ bundles (2.1 SCOs don't have 'created')
+    _20_in_21_bundle = (bundle_version == '2.1' and 'spec_version' not in obj and
+                        'created' in obj)
+    if _20_in_21_bundle:
+        version = '2.0'
+        output.info("%sno spec_version so treated as a 2.0 object in a 2.1 bundle."
+                    % error_prefix)
+
+    return version
+
+
 def _schema_validate(obj, options, bundle_version=None):
     """Set up validation of a single STIX object against its type's schema.
     This does no actual validation; it just returns generators which must be
@@ -715,7 +739,6 @@ def _schema_validate(obj, options, bundle_version=None):
             spec_version property.
     """
     error_gens = []
-
     if 'id' in obj:
         try:
             error_prefix = obj['id'] + ": "
@@ -724,23 +747,7 @@ def _schema_validate(obj, options, bundle_version=None):
     else:
         error_prefix = ''
 
-    if options.version:
-        version = options.version
-    elif options.version is None and 'spec_version' in obj:
-        version = obj['spec_version']
-    else:
-        version = DEFAULT_VER
-
-    if bundle_version == '2.0':
-        version = bundle_version
-
-    # Allow 2.0 objects in 2.1+ bundles (2.1 SCOs don't have 'created')
-    _20_in_21_bundle = (bundle_version == '2.1' and 'spec_version' not in obj and
-                        'created' in obj)
-    if _20_in_21_bundle:
-        version = '2.0'
-        output.info("%sno spec_version so treated as a 2.0 object in a 2.1 bundle."
-                    % error_prefix)
+    version = find_version(obj, options, bundle_version, error_prefix)
 
     options.set_check_codes(version)
 
@@ -756,6 +763,14 @@ def _schema_validate(obj, options, bundle_version=None):
     if base_sdo_errors:
         error_gens.append((base_sdo_errors, error_prefix))
 
+    # Get validator for interop schemas if needed
+    if options.interop and version != '2.0':
+        interop_schema_dir = os.path.abspath(os.path.dirname(__file__) + '/schemas-'
+                                             + version + '/interop/')
+        interop_errors = _get_error_generator(obj['type'], obj, interop_schema_dir, version, default=core_schema)
+        if interop_errors:
+            error_gens.append((interop_errors, error_prefix))
+
     # Get validator for any user-supplied schema
     if options.schema_dir:
         custom_sdo_errors = _get_error_generator(obj['type'], obj, options.schema_dir, version, default=core_schema)
@@ -861,6 +876,9 @@ def validate_instance(instance, options=None):
 
     # Custom validation
     must_checks = _get_musts(options)
+    version = find_version(instance, options)
+    if options.interop is True and version != '2.0':
+        must_checks.append(interop.interop_created_by_ref)
     should_checks = _get_shoulds(options)
     output.info("Running the following additional checks: %s."
                 % ", ".join(x.__name__ for x in chain(must_checks, should_checks)))


Test patch:
diff --git a/stix2validator/test/v21/interop_tests.py b/stix2validator/test/v21/interop_tests.py
new file mode 100644
index 0000000..ed44536
--- /dev/null
+++ b/stix2validator/test/v21/interop_tests.py
@@ -0,0 +1,105 @@
+from . import ValidatorTest
+from ... import ValidationOptions, validate_string
+
+OBJECT_MARKING_FAILURE = u"""
+{
+    "type": "campaign",
+    "spec_version": "2.1",
+    "id": "campaign--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061",
+    "created": "2016-05-12T08:17:27.000Z",
+    "modified": "2016-05-12T08:17:27.000Z",
+    "created_by_ref": "identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+    "name": "Spear Phishing",
+    "external_references": [
+        {
+            "source_name": "capec",
+            "external_id": "CAPEC-163"
+        }
+    ],
+    "kill_chain_phases":
+      [
+        {
+        "kill_chain_name": "example-kill-chain",
+        "phase_name": "lateral-movement"
+        }
+      ],
+      "object_marking_refs": [
+        "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed",
+        "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed"
+      ]
+  }
+  """
+
+
+OBJECT_MARKINGS_SUCCESS = u"""
+{
+    "type": "campaign",
+    "spec_version": "2.1",
+    "id": "campaign--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061",
+    "created": "2016-05-12T08:17:27.000Z",
+    "modified": "2016-05-12T08:17:27.000Z",
+    "created_by_ref": "identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+    "name": "Spear Phishing",
+    "external_references": [
+        {
+            "source_name": "capec",
+            "external_id": "CAPEC-163"
+        }
+    ],
+    "kill_chain_phases":
+      [
+        {
+        "kill_chain_name": "example-kill-chain",
+        "phase_name": "lateral-movement"
+        }
+      ],
+      "object_marking_refs": [
+        "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ed",
+        "marking-definition--5e57c739-391a-4eb3-b6be-7d15ca92d5ef"
+      ]
+  }
+  """
+
+NON_INTEROP_STIX = u"""
+{
+    "type": "campaign",
+    "spec_version": "2.1",
+    "id": "campaign--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061",
+    "created": "2016-05-12T08:17:27.000Z",
+    "modified": "2016-05-12T08:17:27.000Z",
+    "name": "Spear Phishing"
+}
+"""
+INTEROP_STIX = u"""
+{
+    "type": "campaign",
+    "spec_version": "2.1",
+    "id": "campaign--0c7b5b88-8ff7-4a4d-aa9d-feb398cd0061",
+    "created": "2016-05-12T08:17:27.000Z",
+    "modified": "2016-05-12T08:17:27.000Z",
+    "created_by_ref": "identity--f431f809-377b-45e0-aa1c-6a4751cae5ff",
+    "name": "Spear Phishing"
+}
+"""
+
+
+class InteropTestCases(ValidatorTest):
+    def test_obj_markings_invalid(self):
+        options = ValidationOptions(interop=True, version="2.1")
+        results = validate_string(OBJECT_MARKING_FAILURE, options)
+        self.assertFalse(results.is_valid)
+
+    def test_obj_markings_valid(self):
+        options = ValidationOptions(interop=True, version="2.1")
+        results = validate_string(OBJECT_MARKINGS_SUCCESS, options)
+        self.assertTrue(results.is_valid)
+
+    def test_interop_sro_specifics_Compliant(self):
+        options = ValidationOptions(interop=True, version="2.1")
+        results = validate_string(INTEROP_STIX, options)
+        self.assertTrue(results.is_valid)
+
+    def test_interop_sro_specifics_Non_Compliant(self):
+        options = ValidationOptions(interop=True, version="2.1")
+        results = validate_string(NON_INTEROP_STIX, options)
+        self.assertFalse(results.is_valid)