Limit OpenAPI 3.0 discovery to reachable reference scopes

Author: p1c2u

openapi_schema_validator/_references.py

@@ -4,6 +4,8 @@
 from typing import Mapping
 from typing import Sequence
 from typing import cast
+from urllib.parse import urldefrag
+from urllib.parse import urljoin
 from urllib.parse import urlsplit
 
 from jsonschema.exceptions import SchemaError
@@ -33,19 +35,26 @@ def _checked_oas30_scope_id(contents: Any) -> str | None:
     return DRAFT4.id_of(contents)
 
 
-_SCOPED_DRAFT4_NAME = "openapi-3.0-scoped-draft4"
 _SCHEMA_MAP_KEYS = {
     "definitions",
     "dependencies",
     "patternProperties",
     "properties",
 }
 _SCHEMA_ARRAY_KEYS = {"allOf", "anyOf", "oneOf"}
+_OAS30_APPLICATORS = _SCHEMA_ARRAY_KEYS | {
+    "additionalProperties",
+    "items",
+    "not",
+    "properties",
+}
 
 
 def _uses_draft4(resource: Resource[Any]) -> bool:
     specification = resource._specification
-    return specification is DRAFT4 or specification.name == _SCOPED_DRAFT4_NAME
+    return specification is DRAFT4 or isinstance(
+        specification, _OAS30DiscoverySpecification
+    )
 
 
 def _discoverable_id(contents: Any) -> str:
@@ -73,43 +82,68 @@ def _discoverable_subresources(
         and (key not in _SCHEMA_ARRAY_KEYS or isinstance(value, (list, tuple)))
     }
     for child in DRAFT4.subresources_of(shaped):
-        if not isinstance(child, Mapping):
-            continue
-        if "$schema" in child:
-            if "$ref" in child:
-                child = dict(child)
-                del child["$schema"]
-            elif not isinstance(child["$schema"], str):
-                continue
-        yield child
+        if isinstance(child, Mapping) and (
+            "$ref" in child
+            or "$schema" not in child
+            or isinstance(child["$schema"], str)
+        ):
+            yield child
 
 
-def _scoped_resource(
-    contents: Any, *, id_already_applied: bool
-) -> Resource[Any]:
-    # Crawling a resource registered at its resolved URI must not apply
-    # its relative id again. Ignore malformed metadata during discovery.
-    def id_of(schema: Any) -> str | None:
+class _ForeignDialectDiscovery(Exception):
+    """Preflight cannot establish scope through another dialect's crawl."""
+
+
+class _OAS30DiscoverySpecification:
+    """Keep Draft 4 discovery guarded without changing schema contents."""
+
+    name = "openapi-3.0-discovery"
+    subresources_of = staticmethod(_discoverable_subresources)
+    maybe_in_subresource = staticmethod(DRAFT4.maybe_in_subresource)
+
+    def __init__(
+        self, contents: Any, id_already_applied: bool, preflight: bool
+    ) -> None:
+        self._scoped_contents = contents if id_already_applied else None
+        self.preflight = preflight
+
+    def id_of(self, schema: Any) -> str | None:
         identifier = _discoverable_id(schema)
         if (
-            (id_already_applied and schema is contents)
+            schema is self._scoped_contents
             or not identifier
             or identifier.startswith("#")
             or "$ref" in schema
         ):
             return None
         return identifier
 
-    specification: Specification[Any] = cast(Any, Specification)(
-        name=_SCOPED_DRAFT4_NAME,
-        id_of=id_of,
-        subresources_of=_discoverable_subresources,
-        anchors_in=lambda _, schema: (
-            DRAFT4.anchors_in(schema)
-            if _discoverable_id(schema).startswith("#")
-            else ()
-        ),
-        maybe_in_subresource=DRAFT4.maybe_in_subresource,
+    def anchors_in(self, schema: Any) -> Any:
+        if _discoverable_id(schema).startswith("#"):
+            return DRAFT4.anchors_in(schema)
+        return ()
+
+    def detect(self, contents: Mapping[str, Any]) -> Any:
+        if "$ref" in contents:
+            return self
+        detected = DRAFT4.detect(contents)
+        if detected is DRAFT4:
+            return self
+        if self.preflight:
+            raise _ForeignDialectDiscovery
+        return detected
+
+    def create_resource(self, contents: Any) -> Resource[Any]:
+        # Resource.from_contents calls detect() and then create_resource().
+        create_resource = cast(Callable[[Any, Any], Resource[Any]], Resource)
+        return create_resource(contents, self)
+
+
+def _scoped_resource(
+    contents: Any, *, id_already_applied: bool, preflight: bool = False
+) -> Resource[Any]:
+    specification = _OAS30DiscoverySpecification(
+        contents, id_already_applied, preflight
     )
     return specification.create_resource(contents)
 
@@ -120,6 +154,7 @@ def prepare_oas30_resolver(
     schema: Any,
     *,
     id_already_applied: bool = False,
+    preflight: bool = False,
 ) -> Any:
     uri = resolver._base_uri
     resource = scoped._registry.get(uri)
@@ -134,7 +169,10 @@ def prepare_oas30_resolver(
     # Existing resources may be registered at retrieval aliases, even if
     # the alias equals the raw relative id. Only owned scopes suppress it.
     registry = scoped._registry.with_resource(
-        uri, _scoped_resource(schema, id_already_applied=id_already_applied)
+        uri,
+        _scoped_resource(
+            schema, id_already_applied=id_already_applied, preflight=preflight
+        ),
     )
     return scoped._evolve(base_uri=scoped._base_uri, registry=registry)
 
@@ -146,8 +184,17 @@ def _register_scope(resolver: Any, scoped: Any, contents: Any) -> Any:
 
     # Register the reached scope so local lookups can use it without
     # discovering resources in ignored siblings.
+    parent = resolver._registry.get(resolver._base_uri)
+    specification = parent._specification if parent is not None else None
+    preflight = (
+        isinstance(specification, _OAS30DiscoverySpecification)
+        and specification.preflight
+    )
     registry = scoped._registry.with_resource(
-        uri, _scoped_resource(contents, id_already_applied=True)
+        uri,
+        _scoped_resource(
+            contents, id_already_applied=True, preflight=preflight
+        ),
     )
     return scoped._evolve(base_uri=uri, registry=registry)
 
@@ -199,7 +246,10 @@ def _enter_subresource(resolver: Any, schema: Mapping[str, Any]) -> Any:
 )
 
 
-def _resolve_pointer(resolver: Any, ref: str) -> Any:
+def _resolve_reference(resolver: Any, ref: str) -> Any:
+    if not ref.startswith("#/"):
+        return resolver.lookup(ref)
+
     document = resolver.lookup("#")
     resource = document.resolver._registry[document.resolver._base_uri]
     if _uses_draft4(resource):
@@ -209,6 +259,40 @@ def _resolve_pointer(resolver: Any, ref: str) -> Any:
     return resource.pointer(ref[1:], document.resolver)
 
 
+def _preflight_discovery(
+    resolver: Any,
+    ref: str,
+    cache: dict[int, tuple[Registry[Any], Registry[Any] | None]],
+) -> Any:
+    registry = resolver._registry
+    if ref.startswith("#/") or not registry._uncrawled:
+        return resolver
+
+    base_uri = resolver._base_uri
+    if ref.startswith("#"):
+        uri, fragment = base_uri, ref[1:]
+    else:
+        uri, fragment = urldefrag(urljoin(base_uri, ref))
+    if uri in registry and (
+        not fragment
+        or fragment.startswith("/")
+        or (uri, fragment) in registry._anchors
+    ):
+        return resolver
+
+    # An unresolved lookup drops its crawled registry. Retain discovery so
+    # multiple unresolved refs do not each scan the same document.
+    identity = id(registry)
+    if identity not in cache:
+        try:
+            crawled = registry.crawl()
+        except _ForeignDialectDiscovery:
+            crawled = None
+        cache[identity] = (registry, crawled)
+    _, crawled = cache[identity]
+    return None if crawled is None else crawled.resolver(base_uri)
+
+
 def declares_other_schema_dialect(schema: Any) -> bool:
     # OAS 3.0 validators are not registered with jsonschema. A recognized
     # dialect declaration changes the runtime class unless it is a ref sibling.
@@ -249,7 +333,7 @@ def check_oas30_local_references(
     check_schema: Callable[[Any], None],
     registry: Registry[Any] | None,
 ) -> None:
-    # A target can be stored under an ignored sibling. Follow local pointers
+    # A target can be stored under an ignored sibling. Follow local refs
     # separately from structural validation, using the untouched document.
     # Use an isolated registry for the walk. The caller's registry is only
     # consulted for ownership of already-registered scope URIs.
@@ -258,46 +342,70 @@ def check_oas30_local_references(
         DRAFT4.create_resource(schema)
     )
     resolver = prepare_oas30_resolver(
-        resolver, resolver, schema, id_already_applied=True
+        resolver, resolver, schema, id_already_applied=True, preflight=True
     )
     root_uri = resolver._base_uri
-    pending = [(schema, resolver)]
-    visited: set[tuple[int, str]] = set()
+    pending: list[
+        tuple[Mapping[str, Any], Any, tuple[Mapping[str, Any], ...]]
+    ] = [(schema, resolver, ())]
+    visited: set[tuple[int, str, tuple[int, ...]]] = set()
+    discovery: dict[int, tuple[Registry[Any], Registry[Any] | None]] = {}
     checked: set[int] = set()
     _mark_checked_schemas(schema, checked)
 
     while pending:
-        current, resolver = pending.pop()
-        key = (id(current), resolver._base_uri)
+        current, resolver, scopes = pending.pop()
+        key = (id(current), resolver._base_uri, tuple(map(id, scopes)))
         if key in visited:
             continue
         visited.add(key)
 
         if "$ref" in current:
             ref = current["$ref"]
-            # A caller-owned scope wins at runtime and is checked by its owner.
-            if not ref.startswith("#/") or (
-                resolver._base_uri != root_uri
-                and registry is not None
-                and resolver._base_uri in registry
-            ):
+            # Registry discovery for anchors and URI refs can choose a
+            # caller-owned resource. Only check those refs in isolation.
+            if registry and not ref.startswith("#/"):
                 continue
             try:
-                target = _resolve_pointer(resolver, ref)
-            except (Unresolvable, TypeError, ValueError):
+                for scope in scopes:
+                    resolver = _enter_subresource(resolver, scope)
+                # A caller-owned scope wins at runtime.
+                if (
+                    resolver._base_uri != root_uri
+                    and registry is not None
+                    and resolver._base_uri in registry
+                ):
+                    continue
+                resolver = _preflight_discovery(resolver, ref, discovery)
+                if resolver is None:
+                    continue
+                target = _resolve_reference(resolver, ref)
+            except (
+                Unresolvable,
+                TypeError,
+                ValueError,
+                _ForeignDialectDiscovery,
+            ):
                 # Resolution errors belong to instance validation.
                 continue
             if declares_other_schema_dialect(target.contents):
                 continue
             if id(target.contents) not in checked:
                 check_schema(target.contents)
                 _mark_checked_schemas(target.contents, checked)
-            pending.append((target.contents, target.resolver))
+            pending.append((target.contents, target.resolver, ()))
         else:
-            pending.extend(
-                (child, _enter_subresource(resolver, child))
-                for child in _oas30_schema_children(current)
-            )
+            # Definitions become reachable through refs, not by being present.
+            applicators = {
+                key: value
+                for key, value in current.items()
+                if key in _OAS30_APPLICATORS
+            }
+            for child in _oas30_schema_children(applicators):
+                child_scopes = (
+                    scopes if DRAFT4.id_of(child) is None else (*scopes, child)
+                )
+                pending.append((child, resolver, child_scopes))
 
 
 def validate_oas30_reference(
@@ -313,10 +421,7 @@ def validate_oas30_reference(
         validator._resolver, validator._resolver, validator.schema
     )
     try:
-        if ref.startswith("#/"):
-            target = _resolve_pointer(resolver, ref)
-        else:
-            target = resolver.lookup(ref)
+        target = _resolve_reference(resolver, ref)
     except Unresolvable as error:
         raise _WrappedReferencingError(error) from error
     return cast(