Repository: oasis-open/cti-stix-validator (python)

Issue:
Support additional schemas (custom extensions)
We should add an option to allow custom additional schemas. The validator should then ensure the STIX input validates against both the official specification (via the [OASIS schemas](https://github.com/oasis-open/cti-stix2-json-schemas)) and these additional schemas.

Pull request description:
Additional schemas
Fixes #43. Note that to use a schema for a custom observable object type or custom observable extension, you have to have a schema for `observed-data`, but it can be an empty object.

Patch:
diff --git a/stix2validator/scripts/stix2_validator.py b/stix2validator/scripts/stix2_validator.py
index 0efd33d..73c3d0e 100644
--- a/stix2validator/scripts/stix2_validator.py
+++ b/stix2validator/scripts/stix2_validator.py
@@ -158,8 +158,9 @@ def _get_arg_parser(is_script=True):
         "-s",
         "--schemas",
         dest="schema_dir",
-        help="Schema directory. If not provided, the STIX schemas bundled "
-             "with this script will be used."
+        help="Custom schema directory. If provided, input will be validated "
+             "against these schemas in addition to the STIX schemas bundled "
+             "with this script."
     )
 
     # Output options
diff --git a/stix2validator/util.py b/stix2validator/util.py
index 87bbb04..4da0be5 100644
--- a/stix2validator/util.py
+++ b/stix2validator/util.py
@@ -1,5 +1,4 @@
 from collections import Iterable
-import os
 
 
 class ValidationOptions(object):
@@ -73,11 +72,6 @@ class ValidationOptions(object):
             self.refresh_cache = refresh_cache
             self.clear_cache = clear_cache
 
-        # If no schema directory given, use default bundled with this package
-        if not self.schema_dir:
-            self.schema_dir = os.path.abspath(os.path.dirname(__file__) +
-                                              '/schemas/')
-
         # Convert string of comma-separated checks to a list,
         # and convert check code numbers to names
         if self.disabled:
diff --git a/stix2validator/validator.py b/stix2validator/validator.py
index a54b053..19cddcb 100644
--- a/stix2validator/validator.py
+++ b/stix2validator/validator.py
@@ -16,6 +16,8 @@ from .errors import (NoJSONFileFoundError, SchemaError, SchemaInvalidError,
                      ValidationError, pretty_error)
 from .util import ValidationOptions
 
+DEFAULT_SCHEMA_DIR = os.path.abspath(os.path.dirname(__file__) + '/schemas/')
+
 
 def _is_iterable_non_string(val):
     return hasattr(val, "__iter__") and not isinstance(val, string_types)
@@ -486,7 +488,9 @@ def find_schema(schema_dir, obj_type):
     Return the file path of the first match it finds.
     """
     schema_filename = obj_type + '.json'
-    for root, dirnames, filenames in os.walk(schema_dir):
+
+    # If no schema directory given, use default bundled with this package
+    for root, dirnames, filenames in os.walk(schema_dir or DEFAULT_SCHEMA_DIR):
         if schema_filename in filenames:
             return os.path.join(root, schema_filename)
 
@@ -511,30 +515,36 @@ def load_schema(schema_path):
     return schema
 
 
-def _schema_validate(sdo, options):
-    """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
-    iterated to trigger the actual generation.
+def _get_error_generator(type, obj, schema_dir=None, default='core'):
+    """Get a generator for validating against the schema for the given object type.
 
-    Do not call this function directly; use validate_instance() instead, as it
-    calls this one. This function does not perform any custom checks.
+    Args:
+        type (str): The object type to find the schema for.
+        obj: The object to be validated.
+        schema_dir (str): The path in which to search for schemas.
+        default (str): If the schema for the given type cannot be found, use
+            the one with this name instead.
+
+    Returns:
+        A generator for errors found when validating the object against the
+        appropriate schema.
     """
     try:
-        sdo_schema_path = find_schema(options.schema_dir, sdo['type'])
-        sdo_schema = load_schema(sdo_schema_path)
+        schema_path = find_schema(schema_dir, type)
+        schema = load_schema(schema_path)
     except (KeyError, TypeError):
         # Assume a custom object with no schema
         try:
-            sdo_schema_path = find_schema(options.schema_dir, 'core')
-            sdo_schema = load_schema(sdo_schema_path)
+            schema_path = find_schema(schema_dir, default)
+            schema = load_schema(schema_path)
         except (KeyError, TypeError):
-            raise SchemaInvalidError("Cannot locate a schema for the "
-                                     "object's type, nor the base "
-                                     "schema (core.json).")
+            raise SchemaInvalidError("Cannot locate a schema for the object's "
+                                     "type, nor the base schema ({}.json).".format(default))
 
-    if sdo['type'] == 'observed-data':
-        # Validate against schemas for specific object types later
-        sdo_schema['allOf'][1]['properties']['objects'] = {
+    if type == 'observed-data' and schema_dir is None:
+        # Validate against schemas for specific observed data object types later.
+        # If schema_dir is None the schema is custom and won't need to be modified.
+        schema['allOf'][1]['properties']['objects'] = {
             "objects": {
                 "type": "object",
                 "minProperties": 1
@@ -542,13 +552,29 @@ def _schema_validate(sdo, options):
         }
 
     # Don't use custom validator; only check schemas, no additional checks
-    sdo_validator = load_validator(sdo_schema_path, sdo_schema)
+    validator = load_validator(schema_path, schema)
     try:
-        sdo_errors = sdo_validator.iter_errors(sdo)
+        error_gen = validator.iter_errors(obj)
     except schema_exceptions.RefResolutionError:
         raise SchemaInvalidError('Invalid JSON schema: a JSON '
                                  'reference failed to resolve')
 
+    return error_gen
+
+
+def _schema_validate(sdo, options):
+    """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
+    iterated to trigger the actual generation.
+
+    This function first creates generators for the built-in schemas, then adds
+    generators for additional schemas from the options, if specified.
+
+    Do not call this function directly; use validate_instance() instead, as it
+    calls this one. This function does not perform any custom checks.
+    """
+    error_gens = []
+
     if 'id' in sdo:
         try:
             error_prefix = sdo['id'] + ": "
@@ -556,32 +582,33 @@ def _schema_validate(sdo, options):
             error_prefix = 'unidentifiable object: '
     else:
         error_prefix = ''
-    error_gens = [(sdo_errors, error_prefix)]
+
+    # Get validator for built-in schema
+    base_sdo_errors = _get_error_generator(sdo['type'], sdo)
+    error_gens.append((base_sdo_errors, error_prefix))
+
+    # Get validator for any user-supplied schema
+    if options.schema_dir:
+        custom_sdo_errors = _get_error_generator(sdo['type'], sdo, options.schema_dir)
+        error_gens.append((custom_sdo_errors, error_prefix))
 
     # Validate each cyber observable object separately
     if sdo['type'] == 'observed-data' and 'objects' in sdo:
         for key, obj in iteritems(sdo['objects']):
-            try:
-                obj_schema_path = find_schema(options.schema_dir, obj['type'])
-                obj_schema = load_schema(obj_schema_path)
-            except (KeyError, TypeError):
-                # Assume a custom object with no schema
-                try:
-                    obj_schema_path = find_schema(options.schema_dir,
-                                                  'cyber-observable-core')
-                    obj_schema = load_schema(obj_schema_path)
-                except (KeyError, TypeError):
-                    raise SchemaInvalidError("Cannot locate a schema for the "
-                                             "object's type, nor the base "
-                                             "schema (core.json).")
-
-            obj_validator = load_validator(obj_schema_path, obj_schema)
-            try:
-                obj_errors = obj_validator.iter_errors(obj)
-            except schema_exceptions.RefResolutionError:
-                raise SchemaInvalidError('Invalid JSON schema: a JSON '
-                                         'reference failed to resolve')
-            error_gens.append((obj_errors,
+            # Get validator for built-in schemas
+            base_obs_errors = _get_error_generator(obj['type'],
+                                                   obj,
+                                                   None,
+                                                   'cyber-observable-core')
+            error_gens.append((base_obs_errors,
+                               error_prefix + 'object \'' + key + '\': '))
+
+            # Get validator for any user-supplied schema
+            custom_obs_errors = _get_error_generator(obj['type'],
+                                                     obj,
+                                                     options.schema_dir,
+                                                     'cyber-observable-core')
+            error_gens.append((custom_obs_errors,
                                error_prefix + 'object \'' + key + '\': '))
 
     return error_gens


Test patch:
diff --git a/stix2validator/test/__init__.py b/stix2validator/test/__init__.py
index d4613e1..7db8f4c 100644
--- a/stix2validator/test/__init__.py
+++ b/stix2validator/test/__init__.py
@@ -3,11 +3,10 @@ import unittest
 
 from .. import ValidationOptions, validate_parsed_json
 
-SCHEMA_DIR = os.path.abspath(os.path.dirname(__file__) + "../../schemas")
-
 
 class ValidatorTest(unittest.TestCase):
-    options = ValidationOptions(schema_dir=SCHEMA_DIR, strict=True)
+    options = ValidationOptions(strict=True)
+    custom_schemas = os.path.abspath(os.path.dirname(__file__) + "/test_schemas")
 
     def check_ignore(self, instance, code):
         """Test that the given instance is valid if the given check is ignored.
@@ -28,10 +27,9 @@ class ValidatorTest(unittest.TestCase):
                     ValidationOptions constructor.
         """
         if 'strict' in kwargs:
-            options = ValidationOptions(schema_dir=SCHEMA_DIR, **kwargs)
+            options = ValidationOptions(**kwargs)
         else:
-            options = ValidationOptions(schema_dir=SCHEMA_DIR, strict=True,
-                                        **kwargs)
+            options = ValidationOptions(strict=True, **kwargs)
         results = validate_parsed_json(instance, options)
         self.assertTrue(results.is_valid)
 
@@ -45,9 +43,8 @@ class ValidatorTest(unittest.TestCase):
                     ValidationOptions constructor.
         """
         if 'strict' in kwargs:
-            options = ValidationOptions(schema_dir=SCHEMA_DIR, **kwargs)
+            options = ValidationOptions(**kwargs)
         else:
-            options = ValidationOptions(schema_dir=SCHEMA_DIR, strict=True,
-                                        **kwargs)
+            options = ValidationOptions(strict=True, **kwargs)
         results = validate_parsed_json(instance, options)
         self.assertEqual(results.is_valid, False)
diff --git a/stix2validator/test/indicator_tests.py b/stix2validator/test/indicator_tests.py
index 7377d26..5830ac3 100644
--- a/stix2validator/test/indicator_tests.py
+++ b/stix2validator/test/indicator_tests.py
@@ -159,3 +159,32 @@ class IndicatorTestCases(ValidatorTest):
         indicator = copy.deepcopy(self.valid_indicator)
         indicator['pattern'] = "[windows-registry-key:key LIKE 'HKEY_LOCAL_MACHINE\\\\Foo\\\\Bar%']"
         self.assertTrueWithOptions(indicator)
+
+    def test_additional_schemas(self):
+        indicator = copy.deepcopy(self.valid_indicator)
+        self.assertFalseWithOptions(indicator, schema_dir=self.custom_schemas)
+
+        indicator['name'] = "Foobar"
+        self.assertTrueWithOptions(indicator, schema_dir=self.custom_schemas)
+
+    def test_additional_schemas_custom_type(self):
+        # no schema exists for this type
+        new_obj = {
+            "type": "x-type",
+            "id": "x-type--353ed279-5f4f-4a79-bffc-b2e2ed08ea1f",
+            "created": "2016-04-06T20:03:48.000Z",
+            "modified": "2016-04-06T20:03:48.000Z",
+            "property1": 10,
+            "property2": "fizzbuzz"
+        }
+        self.assertFalseWithOptions(new_obj, schema_dir=self.custom_schemas)
+
+        # properties are wrong types (str vs int)
+        new_obj['type'] = 'x-new-type'
+        new_obj['id'] = 'x-new-type--353ed279-5f4f-4a79-bffc-b2e2ed08ea1f'
+        self.assertFalseWithOptions(new_obj, schema_dir=self.custom_schemas)
+
+        # now it's valid
+        new_obj['property1'] = 'fizzbuzz'
+        new_obj['property2'] = 10
+        self.assertTrueWithOptions(new_obj, schema_dir=self.custom_schemas)
diff --git a/stix2validator/test/observed_data_tests.py b/stix2validator/test/observed_data_tests.py
index bb2a934..1a24632 100644
--- a/stix2validator/test/observed_data_tests.py
+++ b/stix2validator/test/observed_data_tests.py
@@ -785,3 +785,24 @@ class ObservedDataTestCases(ValidatorTest):
             }
         }
         self.assertFalseWithOptions(observed_data)
+
+    def test_additional_schemas_custom_observable(self):
+        observed_data = copy.deepcopy(self.valid_observed_data)
+        observed_data['objects']['2'] = {
+            "type": "x-new-observable",
+            "foo": 100
+        }
+        self.assertFalseWithOptions(observed_data, schema_dir=self.custom_schemas)
+
+        observed_data['objects']['2']['foo'] = 'something'
+        self.assertTrueWithOptions(observed_data, schema_dir=self.custom_schemas)
+
+    def test_additional_schemas_custom_extension(self):
+        observed_data = copy.deepcopy(self.valid_observed_data)
+        observed_data['objects']['1']['extensions'] = {'x-example-com-foobar-ext': {
+            "bar_value": "something",
+        }}
+        self.assertFalseWithOptions(observed_data, schema_dir=self.custom_schemas)
+
+        observed_data['objects']['1']['extensions']['x-example-com-foobar-ext']['foo_value'] = 'something else'
+        self.assertTrueWithOptions(observed_data, schema_dir=self.custom_schemas)
diff --git a/stix2validator/test/test_schemas/file.json b/stix2validator/test/test_schemas/file.json
new file mode 100644
index 0000000..0cca60e
--- /dev/null
+++ b/stix2validator/test/test_schemas/file.json
@@ -0,0 +1,20 @@
+{
+    "properties": {
+        "extensions": {
+            "patternProperties": {
+                "^x-example-com-foobar-ext$": {
+                    "type": "object",
+                    "properties": {
+                        "foo_value": {
+                            "type": "string"
+                        },
+                        "bar_value": {
+                            "type": "string"
+                        }
+                    },
+                    "required": ["foo_value"]
+                }
+            }
+        }
+    }
+}
diff --git a/stix2validator/test/test_schemas/indicator.json b/stix2validator/test/test_schemas/indicator.json
new file mode 100644
index 0000000..dac2297
--- /dev/null
+++ b/stix2validator/test/test_schemas/indicator.json
@@ -0,0 +1,8 @@
+{
+    "properties": {
+        "name": {
+            "type": "string",
+            "pattern": "^Foobar$"
+        }
+    }
+}
diff --git a/stix2validator/test/test_schemas/observed-data.json b/stix2validator/test/test_schemas/observed-data.json
new file mode 100644
index 0000000..2c63c08
--- /dev/null
+++ b/stix2validator/test/test_schemas/observed-data.json
@@ -0,0 +1,2 @@
+{
+}
diff --git a/stix2validator/test/test_schemas/x-new-observable.json b/stix2validator/test/test_schemas/x-new-observable.json
new file mode 100644
index 0000000..4f2b36e
--- /dev/null
+++ b/stix2validator/test/test_schemas/x-new-observable.json
@@ -0,0 +1,7 @@
+{
+    "properties": {
+        "foo": {
+            "type": "string"
+        }
+    }
+}
diff --git a/stix2validator/test/test_schemas/x-new-type.json b/stix2validator/test/test_schemas/x-new-type.json
new file mode 100644
index 0000000..df609c7
--- /dev/null
+++ b/stix2validator/test/test_schemas/x-new-type.json
@@ -0,0 +1,10 @@
+{
+    "properties": {
+        "property1": {
+            "type": "string"
+        },
+        "property2": {
+            "type": "integer"
+        }
+    }
+}