Add OAS 3.1 jsonSchemaDialect-aware schema meta-validation

Author: p1c2u

openapi_spec_validator/validation/keywords.py

@@ -3,6 +3,7 @@
 from collections.abc import Iterator
 from collections.abc import Mapping
 from collections.abc import Sequence
+from functools import partial
 from typing import TYPE_CHECKING
 from typing import Any
 from typing import cast
@@ -11,11 +12,13 @@
 from jsonschema.exceptions import SchemaError
 from jsonschema.exceptions import ValidationError
 from jsonschema.protocols import Validator
+from jsonschema.validators import validator_for
 from jsonschema_path.paths import SchemaPath
 from openapi_schema_validator import oas30_format_checker
 from openapi_schema_validator import oas31_format_checker
 from openapi_schema_validator.validators import OAS30Validator
 from openapi_schema_validator.validators import OAS31Validator
+from openapi_schema_validator.validators import check_openapi_schema
 
 from openapi_spec_validator.validation.exceptions import (
     DuplicateOperationIDError,
@@ -34,6 +37,8 @@
         KeywordValidatorRegistry,
     )
 
+OAS31_BASE_DIALECT_URI = "https://spec.openapis.org/oas/3.1/dialect/base"
+
 
 class KeywordValidator:
     def __init__(self, registry: "KeywordValidatorRegistry"):
@@ -101,6 +106,32 @@ def _collect_properties(self, schema: SchemaPath) -> set[str]:
 
         return props
 
+    def _get_schema_checker(
+        self, schema: SchemaPath, schema_value: Any
+    ) -> Callable[[Any], None]:
+        return cast(
+            Callable[[Any], None],
+            getattr(
+                self.default_validator.value_validator_cls,
+                "check_schema",
+            ),
+        )
+
+    def _validate_schema_meta(
+        self, schema: SchemaPath, schema_value: Any
+    ) -> OpenAPIValidationError | None:
+        try:
+            schema_checker = self._get_schema_checker(schema, schema_value)
+        except ValueError as exc:
+            return OpenAPIValidationError(str(exc))
+        try:
+            schema_checker(schema_value)
+        except (SchemaError, ValidationError) as err:
+            return cast(
+                OpenAPIValidationError, OpenAPIValidationError.create_from(err)
+            )
+        return None
+
     def __call__(
         self,
         schema: SchemaPath,
@@ -114,23 +145,17 @@ def __call__(
             )
             return
 
+        schema_id = id(schema_value)
         if not meta_checked:
             assert self.meta_checked_schema_ids is not None
-            schema_id = id(schema_value)
             if schema_id not in self.meta_checked_schema_ids:
-                try:
-                    schema_check = getattr(
-                        self.default_validator.value_validator_cls,
-                        "check_schema",
-                    )
-                    schema_check(schema_value)
-                except (SchemaError, ValidationError) as err:
-                    yield OpenAPIValidationError.create_from(err)
-                    return
                 self.meta_checked_schema_ids.append(schema_id)
+                err = self._validate_schema_meta(schema, schema_value)
+                if err is not None:
+                    yield err
+                    return
 
         assert self.visited_schema_ids is not None
-        schema_id = id(schema_value)
         if schema_id in self.visited_schema_ids:
             return
         self.visited_schema_ids.append(schema_id)
@@ -218,6 +243,54 @@ def __call__(
                 yield from self.default_validator(schema, default_value)
 
 
+class OpenAPIV31SchemaValidator(SchemaValidator):
+    default_jsonschema_dialect_id = OAS31_BASE_DIALECT_URI
+
+    def _get_schema_checker(
+        self, schema: SchemaPath, schema_value: Any
+    ) -> Callable[[Any], None]:
+        if isinstance(schema_value, Mapping):
+            schema_to_check = dict(schema_value)
+            if "$schema" in schema_to_check:
+                dialect_source = schema_to_check
+            else:
+                jsonschema_dialect_id = self._get_jsonschema_dialect_id(schema)
+                dialect_source = {"$schema": jsonschema_dialect_id}
+                schema_to_check = {
+                    **schema_to_check,
+                    "$schema": jsonschema_dialect_id,
+                }
+        else:
+            jsonschema_dialect_id = self._get_jsonschema_dialect_id(schema)
+            dialect_source = {"$schema": jsonschema_dialect_id}
+            schema_to_check = schema_value
+
+        validator_cls = validator_for(
+            dialect_source,
+            default=cast(Any, None),
+        )
+        if validator_cls is None:
+            raise ValueError(
+                f"Unknown JSON Schema dialect: {dialect_source['$schema']!r}"
+            )
+        return partial(
+            check_openapi_schema,
+            validator_cls,
+            format_checker=oas31_format_checker,
+        )
+
+    def _get_jsonschema_dialect_id(self, schema: SchemaPath) -> str:
+        schema_root = self._get_schema_root(schema)
+        try:
+            return (schema_root // "jsonSchemaDialect").read_str()
+        except KeyError:
+            return self.default_jsonschema_dialect_id
+
+    def _get_schema_root(self, schema: SchemaPath) -> SchemaPath:
+        # jsonschema-path currently has no public API for root traversal.
+        return schema._clone_with_parts(())
+
+
 class SchemasValidator(KeywordValidator):
     @property
     def schema_validator(self) -> SchemaValidator:

openapi_spec_validator/validation/validators.py

@@ -144,7 +144,7 @@ class OpenAPIV31SpecValidator(SpecValidator):
         "path": keywords.PathValidator,
         "response": keywords.OpenAPIV3ResponseValidator,
         "responses": keywords.ResponsesValidator,
-        "schema": keywords.SchemaValidator,
+        "schema": keywords.OpenAPIV31SchemaValidator,
         "schemas": keywords.SchemasValidator,
     }
     root_keywords = ["paths", "components"]

poetry.lock

@@ -25,7 +25,7 @@ classifiers = [
 ]
 dependencies = [
     "jsonschema >=4.24.0,<4.25.0",
-    "openapi-schema-validator >=0.7.0,<0.8.0",
+    "openapi-schema-validator >=0.7.2,<0.8.0",
     "jsonschema-path >=0.4.2,<0.5.0",
     "lazy-object-proxy >=1.7.1,<2.0",
 ]

pyproject.toml

@@ -0,0 +1,89 @@
+from openapi_spec_validator import OpenAPIV31SpecValidator
+from openapi_spec_validator.validation.exceptions import OpenAPIValidationError
+
+
+def make_spec(
+    component_schema: dict[str, object] | bool,
+    json_schema_dialect: str | None = None,
+) -> dict[str, object]:
+    spec: dict[str, object] = {
+        "openapi": "3.1.0",
+        "info": {
+            "title": "Test API",
+            "version": "0.0.1",
+        },
+        "paths": {},
+        "components": {
+            "schemas": {
+                "Component": component_schema,
+            },
+        },
+    }
+    if json_schema_dialect is not None:
+        spec["jsonSchemaDialect"] = json_schema_dialect
+    return spec
+
+
+def test_root_json_schema_dialect_is_honored():
+    spec = make_spec(
+        {"type": "object"},
+        json_schema_dialect="https://json-schema.org/draft/2019-09/schema",
+    )
+
+    errors = list(OpenAPIV31SpecValidator(spec).iter_errors())
+    assert errors == []
+
+
+def test_schema_dialect_overrides_root_json_schema_dialect():
+    root_dialect = "https://json-schema.org/draft/2019-09/schema"
+    schema_dialect = "https://json-schema.org/draft/2020-12/schema"
+    spec = make_spec(
+        {
+            "$schema": schema_dialect,
+            "type": "object",
+        },
+        json_schema_dialect=root_dialect,
+    )
+
+    errors = list(OpenAPIV31SpecValidator(spec).iter_errors())
+
+    assert errors == []
+
+
+def test_unknown_dialect_raises_error():
+    spec = make_spec(
+        {"type": "object"},
+        json_schema_dialect="https://example.com/custom",
+    )
+
+    errors = list(OpenAPIV31SpecValidator(spec).iter_errors())
+
+    assert len(errors) == 1
+    assert isinstance(errors[0], OpenAPIValidationError)
+    assert "Unknown JSON Schema dialect" in errors[0].message
+
+
+def test_meta_check_error_stops_further_schema_traversal():
+    spec = make_spec(
+        {
+            "type": 1,
+            "required": ["missing_property"],
+        },
+        json_schema_dialect="https://json-schema.org/draft/2020-12/schema",
+    )
+
+    errors = list(OpenAPIV31SpecValidator(spec).iter_errors())
+
+    assert len(errors) == 1
+    assert "is not valid under any of the given schemas" in errors[0].message
+
+
+def test_boolean_schema_uses_root_json_schema_dialect():
+    spec = make_spec(
+        True,
+        json_schema_dialect="https://json-schema.org/draft/2019-09/schema",
+    )
+
+    errors = list(OpenAPIV31SpecValidator(spec).iter_errors())
+
+    assert errors == []