Check local reference targets in OAS 3.0 schema checks

Reference siblings are ignored by the OAS 3.0 schema check, so schemas reachable only through a local reference (next to a root $ref, or under a non-schema key such as components) were no longer checked. Follow reachable JSON pointer references from the root and check targets the structural check did not cover.

References that depend on runtime resolution context are left to instance validation: external and anchor references, references inside a subschema with its own id, targets declaring another dialect, and shortcut calls with a legacy resolver or a root $schema.

Author: p1c2u

docs/validation.rst

@@ -161,6 +161,13 @@ In order to validate OpenAPI 3.0 schema, import and use ``OAS30Validator`` inste
 
    validate({"name": "John", "age": None}, schema, cls=OAS30Validator)
 
+In OpenAPI 3.0 an object with ``$ref`` is a Reference Object, and any other
+keys next to ``$ref`` are ignored, both when checking the schema and when
+validating instances. A ``$schema`` next to ``$ref`` does not switch the
+dialect either. Schema checks still follow local JSON pointer references
+(``#/...``) and check the schemas they point to, wherever those are stored in
+the document.
+
 Default dialect resolution
 --------------------------
 

openapi_schema_validator/_references.py

@@ -0,0 +1,150 @@
+"""Local reference targets for OpenAPI 3.0 schema checks.
+
+In OpenAPI 3.0 an object with ``$ref`` is a Reference Object and its other
+keys are ignored, so the structural schema check does not descend into
+them. A local reference can still point at a schema the structural check
+never visited (for example a key next to a root ``$ref``, or a non-schema
+key such as ``components``). This module finds those targets so they can
+be checked with the same metaschema.
+
+Only JSON pointer references (``#`` and ``#/...``) resolved against the
+root document are followed. References that depend on runtime resolution
+context are left to instance validation: external and anchor references,
+references inside a subschema that sets its own ``id``, and schemas that
+declare a different dialect with ``$schema``.
+"""
+
+from typing import Any
+from typing import Iterator
+from typing import Mapping
+from urllib.parse import unquote
+
+from jsonschema.validators import validator_for
+
+# Draft 4 keywords whose values are a schema, a list of schemas, or a
+# mapping of schemas.
+_SCHEMA_KEYWORDS = ("additionalItems", "additionalProperties", "items", "not")
+_SCHEMA_LIST_KEYWORDS = ("allOf", "anyOf", "items", "oneOf")
+_SCHEMA_MAP_KEYWORDS = ("dependencies", "patternProperties", "properties")
+
+
+def declares_other_dialect(schema: Any) -> bool:
+    """Whether ``schema`` switches jsonschema to another validator class.
+
+    A ``$schema`` next to a ``$ref`` is an ignored sibling and does not
+    count.
+    """
+    return (
+        isinstance(schema, Mapping)
+        and "$ref" not in schema
+        and isinstance(schema.get("$schema"), str)
+        and validator_for(schema, default=None) is not None
+    )
+
+
+def _children(
+    schema: Mapping[str, Any], *, include_definitions: bool
+) -> Iterator[Mapping[str, Any]]:
+    for keyword in _SCHEMA_KEYWORDS:
+        value = schema.get(keyword)
+        if isinstance(value, Mapping):
+            yield value
+    for keyword in _SCHEMA_LIST_KEYWORDS:
+        value = schema.get(keyword)
+        if isinstance(value, list):
+            yield from (item for item in value if isinstance(item, Mapping))
+    map_keywords: tuple[str, ...] = _SCHEMA_MAP_KEYWORDS
+    if include_definitions:
+        map_keywords = (*map_keywords, "definitions")
+    for keyword in map_keywords:
+        value = schema.get(keyword)
+        if isinstance(value, Mapping):
+            yield from (
+                item for item in value.values() if isinstance(item, Mapping)
+            )
+
+
+def _mark_covered(schema: Mapping[str, Any], covered: set[int]) -> None:
+    # Mirror what the structural check visits: every subschema, including
+    # definitions, but nothing inside a Reference Object.
+    pending = [schema]
+    while pending:
+        current = pending.pop()
+        if id(current) in covered:
+            continue
+        covered.add(id(current))
+        if "$ref" not in current:
+            pending.extend(_children(current, include_definitions=True))
+
+
+def _sets_scope(schema: Any) -> bool:
+    return (
+        isinstance(schema, Mapping)
+        and "$ref" not in schema
+        and isinstance(schema.get("id"), str)
+    )
+
+
+def _resolve_pointer(
+    document: Mapping[str, Any], ref: str
+) -> tuple[Any, bool]:
+    """Return the pointer target and whether an ``id`` on the way to it
+    changes its resolution scope."""
+    target: Any = document
+    scoped = False
+    if ref == "#":
+        return target, scoped
+    for token in unquote(ref[2:]).split("/"):
+        token = token.replace("~1", "/").replace("~0", "~")
+        if isinstance(target, list):
+            target = target[int(token)]
+        else:
+            target = target[token]
+        scoped = scoped or _sets_scope(target)
+    return target, scoped
+
+
+def iter_unchecked_reference_targets(
+    schema: Mapping[str, Any],
+) -> Iterator[Mapping[str, Any]]:
+    """Yield reachable local reference targets the structural check missed.
+
+    Each target is yielded once. Targets inside a yielded target are
+    treated as covered by that target's check.
+    """
+    covered: set[int] = set()
+    _mark_covered(schema, covered)
+
+    pending: list[tuple[Mapping[str, Any], bool]] = [(schema, False)]
+    visited: set[tuple[int, bool]] = set()
+    while pending:
+        current, scoped = pending.pop()
+        key = (id(current), scoped)
+        if key in visited:
+            continue
+        visited.add(key)
+
+        if "$ref" not in current:
+            if current is not schema and declares_other_dialect(current):
+                continue
+            # Definitions become reachable through references only.
+            for child in _children(current, include_definitions=False):
+                pending.append((child, scoped or _sets_scope(child)))
+            continue
+
+        ref = current["$ref"]
+        if scoped or not isinstance(ref, str):
+            continue
+        if ref != "#" and not ref.startswith("#/"):
+            continue
+        try:
+            target, target_scoped = _resolve_pointer(schema, ref)
+        except (LookupError, TypeError, ValueError):
+            # Unresolvable references are reported during validation.
+            continue
+        if not isinstance(target, Mapping) or declares_other_dialect(target):
+            continue
+        if id(target) not in covered:
+            _mark_covered(target, covered)
+            yield target
+        pending.append((target, target_scoped))

openapi_schema_validator/shortcuts.py

@@ -11,11 +11,13 @@
 from openapi_schema_validator._caches import ValidatorCache
 from openapi_schema_validator._dialects import OAS31_BASE_DIALECT_ID
 from openapi_schema_validator._dialects import OAS32_BASE_DIALECT_ID
+from openapi_schema_validator._references import declares_other_dialect
 from openapi_schema_validator.validators import OAS32Validator
 from openapi_schema_validator.validators import (
     build_enforce_properties_required_validator,
 )
 from openapi_schema_validator.validators import check_openapi_schema
+from openapi_schema_validator.validators import supports_reference_check
 
 _LOCAL_ONLY_REGISTRY = Registry()
 _VALIDATOR_CACHE = ValidatorCache()
@@ -24,6 +26,8 @@
 def _check_schema(
     cls: type[Validator],
     schema: dict[str, Any],
+    *,
+    legacy_resolver: bool,
 ) -> None:
     meta_schema = getattr(cls, "META_SCHEMA", None)
     # jsonschema's default check_schema path does not accept a custom
@@ -34,6 +38,14 @@ def _check_schema(
         OAS32_BASE_DIALECT_ID,
     ):
         check_openapi_schema(cls, schema)
+    elif supports_reference_check(cls):
+        # Local references are resolved against ``schema`` only when it is
+        # the reference root: a legacy resolver can carry another document,
+        # and a root ``$schema`` hands validation to that dialect.
+        check_references = not (
+            legacy_resolver or declares_other_dialect(schema)
+        )
+        cls.check_schema(schema, check_references=check_references)
     else:
         cls.check_schema(schema)
 
@@ -92,6 +104,10 @@ def validate(
     if not allow_remote_references:
         validator_kwargs.setdefault("registry", _LOCAL_ONLY_REGISTRY)
 
+    legacy_resolver = validator_kwargs.get("resolver") is not None or bool(
+        args and args[0] is not None
+    )
+
     key = _VALIDATOR_CACHE.build_key(
         schema=schema_dict,
         cls=cls,
@@ -104,7 +120,7 @@ def validate(
 
     if cached is None:
         if check_schema:
-            _check_schema(cls, schema_dict)
+            _check_schema(cls, schema_dict, legacy_resolver=legacy_resolver)
 
         validator = cls(schema_dict, *args, **validator_kwargs)
         cached = _VALIDATOR_CACHE.set(
@@ -113,7 +129,7 @@ def validate(
             schema_checked=check_schema,
         )
     elif check_schema and not cached.schema_checked:
-        _check_schema(cls, schema_dict)
+        _check_schema(cls, schema_dict, legacy_resolver=legacy_resolver)
         _VALIDATOR_CACHE.mark_schema_checked(key)
     else:
         _VALIDATOR_CACHE.touch(key)

openapi_schema_validator/validators.py

@@ -24,6 +24,9 @@
 from openapi_schema_validator._dialects import OAS32_BASE_DIALECT_ID
 from openapi_schema_validator._dialects import OAS32_BASE_DIALECT_METASCHEMA
 from openapi_schema_validator._dialects import register_openapi_dialect
+from openapi_schema_validator._references import (
+    iter_unchecked_reference_targets,
+)
 from openapi_schema_validator._specifications import (
     REGISTRY as OPENAPI_SPECIFICATIONS,
 )
@@ -73,6 +76,8 @@ def _check_oas30_schema(
     cls: Any,
     schema: Any,
     format_checker: Any = _CHECK_SCHEMA_UNSET,
+    *,
+    check_references: bool = True,
 ) -> None:
     if format_checker is _CHECK_SCHEMA_UNSET:
         format_checker = cls.FORMAT_CHECKER
@@ -83,15 +88,31 @@ def _check_oas30_schema(
         registry=OPENAPI_SPECIFICATIONS,
     )
     _raise_first_schema_error(validator, schema)
+    if not check_references or not isinstance(schema, Mapping):
+        return
+    # Reference siblings are ignored above, so check local reference
+    # targets the structural check did not reach.
+    for target in iter_unchecked_reference_targets(schema):
+        _raise_first_schema_error(validator, target)
 
 
 def _check_oas30_variant_schema(
     cls: Any,
     schema: Any,
     format_checker: Any = Draft4Validator.FORMAT_CHECKER,
+    *,
+    check_references: bool = True,
 ) -> None:
     # These variants previously inherited jsonschema's Draft 4 format checks.
-    _check_oas30_schema(cls, schema, format_checker)
+    _check_oas30_schema(
+        cls, schema, format_checker, check_references=check_references
+    )
+
+
+def supports_reference_check(cls: Any) -> bool:
+    """Whether ``cls.check_schema`` accepts ``check_references``."""
+    check = getattr(cls.check_schema, "__func__", None)
+    return check in (_check_oas30_schema, _check_oas30_variant_schema)
 
 
 def _raise_first_schema_error(validator: Validator, schema: Any) -> None: