From 92744a33cdf28af1dba112e701ca275ee62d3bf8 Mon Sep 17 00:00:00 2001 From: Dex Date: Tue, 14 Jul 2026 18:11:08 +0100 Subject: [PATCH] fix(cli): resolve modular OpenAPI references (#7586) Fixes #7585 --- cli/openbb_cli/dispatchers/openapi_schema.py | 263 ++++++++++++++++++- cli/tests/test_dispatchers_openapi_schema.py | 226 ++++++++++++++++ 2 files changed, 479 insertions(+), 10 deletions(-) diff --git a/cli/openbb_cli/dispatchers/openapi_schema.py b/cli/openbb_cli/dispatchers/openapi_schema.py index dc58928f8f5..726f4b14e10 100644 --- a/cli/openbb_cli/dispatchers/openapi_schema.py +++ b/cli/openbb_cli/dispatchers/openapi_schema.py @@ -6,6 +6,7 @@ import argparse import json import re from typing import Any +from urllib.parse import unquote, urldefrag, urljoin, urlsplit import httpx @@ -258,14 +259,19 @@ def _resolve_schema(schema: dict[str, Any]) -> tuple[type, list[Any], bool]: choices: list[Any] = [] is_list = False for s in non_null: - if s.get("type") == "array": + schema_type = s.get("type") + if schema_type == "array" or ( + isinstance(schema_type, list) and "array" in schema_type + ): is_list = True for v in s.get("enum", []): if v not in choices: choices.append(v) return (str, choices, is_list) - if schema.get("type") == "array": + schema_type = schema.get("type", "string") + schema_types = schema_type if isinstance(schema_type, list) else [schema_type] + if "array" in schema_types: items = schema.get("items", {}) or {} py_type, item_choices, _ = _resolve_schema(items) return (py_type, item_choices, True) @@ -273,17 +279,22 @@ def _resolve_schema(schema: dict[str, Any]) -> tuple[type, list[Any], bool]: if "const" in schema: return (type(schema["const"]), [schema["const"]], False) - py_type = _TYPE_MAP.get(schema.get("type", "string"), str) + scalar_type = next( + (item for item in schema_types if item not in {"array", "null"}), "string" + ) + py_type = _TYPE_MAP.get(scalar_type, str) return (py_type, list(schema.get("enum", [])), False) def _is_json_arg(schema: dict[str, Any]) -> bool: """Whether a request-body field must be supplied as a raw JSON value.""" - if schema.get("type") == "object" or "properties" in schema: + schema_type = schema.get("type") + schema_types = schema_type if isinstance(schema_type, list) else [schema_type] + if "object" in schema_types or "properties" in schema: return True if schema.get("additionalProperties"): return True - if schema.get("type") == "array": + if "array" in schema_types: items = schema.get("items") return isinstance(items, dict) and _is_json_arg(items) return any( @@ -304,8 +315,28 @@ def request_body_parameters( body_schema: dict[str, Any] | None, ) -> list[dict[str, Any]]: """Flatten a dereferenced request-body schema into OpenAPI parameter objects.""" - if not isinstance(body_schema, dict) or body_schema.get("type") != "object": + if not isinstance(body_schema, dict): return [] + body_type = body_schema.get("type") + body_types = body_type if isinstance(body_type, list) else [body_type] + if "object" not in body_types: + object_members = [ + member + for member in body_schema.get("anyOf", []) + if isinstance(member, dict) + and ( + "object" + in ( + member.get("type") + if isinstance(member.get("type"), list) + else [member.get("type")] + ) + or "properties" in member + ) + ] + if len(object_members) != 1: + return [] + body_schema = {**body_schema, **object_members[0]} required = set(body_schema.get("required") or []) out: list[dict[str, Any]] = [] for name, prop in (body_schema.get("properties") or {}).items(): @@ -615,6 +646,195 @@ def _extract_embedded_spec(html: str) -> dict[str, Any] | None: return None +def _resolve_json_pointer(document: Any, fragment: str) -> Any: + """Resolve a URI-fragment JSON pointer within one OpenAPI document.""" + if not fragment: + return document + pointer = unquote(fragment) + if not pointer.startswith("/"): + raise ValueError(f"Unsupported OpenAPI reference fragment: #{fragment}") + node = document + for raw_part in pointer[1:].split("/"): + part = raw_part.replace("~1", "/").replace("~0", "~") + if isinstance(node, dict): + node = node[part] + elif isinstance(node, list): + node = node[int(part)] + else: + raise ValueError(f"Invalid OpenAPI reference fragment: #{fragment}") + return node + + +def _url_origin(url: str) -> tuple[str, str, int | None]: + """Return a normalized HTTP(S) origin for reference security checks.""" + parsed = urlsplit(url) + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError(f"Unsupported OpenAPI reference URL: {url}") + default_port = 443 if parsed.scheme == "https" else 80 + return parsed.scheme, parsed.hostname.lower(), parsed.port or default_port + + +def _bundle_external_refs( + document: dict[str, Any], + source_url: str, + *, + timeout: float, + headers: dict[str, str], +) -> dict[str, Any]: + """Inline same-origin external refs while preserving root-local refs.""" + max_documents = 512 + max_document_bytes = 8 * 1024 * 1024 + read_chunk_bytes = 64 * 1024 + max_depth = 128 + root_url, _ = urldefrag(source_url) + root_origin = _url_origin(root_url) + documents: dict[str, dict[str, Any]] = {root_url: document} + + def load_document(url: str) -> dict[str, Any]: + cached = documents.get(url) + if cached is not None: + return cached + if _url_origin(url) != root_origin: + raise ValueError(f"External OpenAPI reference must be same-origin: {url}") + if len(documents) >= max_documents: + raise ValueError("OpenAPI reference graph exceeds the document limit") + with httpx.stream( + "GET", + url, + timeout=timeout, + follow_redirects=False, + headers=headers, + ) as response: + if response.is_redirect: + raise ValueError( + "Redirects are not allowed for external OpenAPI references" + ) + response.raise_for_status() + content_length = response.headers.get("content-length") + if content_length and int(content_length) > max_document_bytes: + raise ValueError("External OpenAPI reference exceeds the size limit") + content = bytearray() + for chunk in response.iter_bytes(chunk_size=read_chunk_bytes): + if len(chunk) > max_document_bytes - len(content): + raise ValueError( + "External OpenAPI reference exceeds the size limit" + ) + content.extend(chunk) + text = bytes(content).decode(response.encoding or "utf-8") + content_type = response.headers.get("content-type", "") + parsed = _ensure_openapi_dict( + _parse_spec_text( + text, + content_type=content_type, + ), + url, + ) + documents[url] = parsed + return parsed + + def normalize_schema_type(value: dict[str, Any]) -> dict[str, Any]: + """Convert OpenAPI 3.1 type arrays for the existing code generator.""" + schema_types = value.get("type") + if not isinstance(schema_types, list): + return value + normalized = {key: item for key, item in value.items() if key != "type"} + normalized["anyOf"] = [{"type": item} for item in schema_types] + return normalized + + def visit( + node: Any, + current_url: str, + current_document: dict[str, Any], + *, + imported: bool, + seen: frozenset[tuple[str, str]], + depth: int, + ) -> Any: + if depth > max_depth: + raise ValueError("OpenAPI reference graph exceeds the depth limit") + if isinstance(node, dict): + if "$id" in node: + raise ValueError("OpenAPI $id resource identifiers are not supported") + ref = node.get("$ref") + if isinstance(ref, str) and (imported or not ref.startswith("#")): + target_url, fragment = urldefrag(urljoin(current_url, ref)) + if fragment and not fragment.startswith("/"): + raise ValueError("OpenAPI reference anchors are not supported") + marker = (target_url, fragment) + if marker in seen: + return {"$ref": f"{target_url}#{fragment}"} + target_document = ( + current_document + if target_url == current_url + else load_document(target_url) + ) + target = _resolve_json_pointer(target_document, fragment) + resolved = visit( + target, + target_url, + target_document, + imported=True, + seen=seen | {marker}, + depth=depth + 1, + ) + unsupported_siblings = set(node) - {"$ref", "summary", "description"} + if unsupported_siblings: + raise ValueError( + "OpenAPI reference siblings other than summary and description " + "are not supported" + ) + siblings = { + key: visit( + value, + current_url, + current_document, + imported=imported, + seen=seen, + depth=depth + 1, + ) + for key, value in node.items() + if key != "$ref" + } + if isinstance(resolved, dict): + return normalize_schema_type({**resolved, **siblings}) + return resolved + return normalize_schema_type( + { + key: visit( + value, + current_url, + current_document, + imported=imported, + seen=seen, + depth=depth + 1, + ) + for key, value in node.items() + } + ) + if isinstance(node, list): + return [ + visit( + value, + current_url, + current_document, + imported=imported, + seen=seen, + depth=depth + 1, + ) + for value in node + ] + return node + + return visit( + document, + root_url, + document, + imported=False, + seen=frozenset(), + depth=0, + ) + + def fetch_openapi( base_url: str, *, @@ -643,25 +863,37 @@ def fetch_openapi( ) if response.status_code < 400: try: - return _ensure_openapi_dict( + parsed = _ensure_openapi_dict( _parse_spec_text( response.text, content_type=response.headers.get("content-type", ""), ), full_url, ) + return _bundle_external_refs( + parsed, + str(getattr(response, "url", full_url)), + timeout=timeout, + headers=merged_headers, + ) except (json.JSONDecodeError, ValueError): pass if explicit_path: response.raise_for_status() - return _ensure_openapi_dict( + parsed = _ensure_openapi_dict( _parse_spec_text( response.text, content_type=response.headers.get("content-type", ""), ), full_url, ) + return _bundle_external_refs( + parsed, + str(getattr(response, "url", full_url)), + timeout=timeout, + headers=merged_headers, + ) landing_url = base_url.rstrip("/") + "/" landing = httpx.get( @@ -674,16 +906,27 @@ def fetch_openapi( landing.raise_for_status() embedded = _extract_embedded_spec(landing.text) if embedded is not None: - return embedded + return _bundle_external_refs( + embedded, + str(getattr(landing, "url", landing_url)), + timeout=timeout, + headers=merged_headers, + ) response.raise_for_status() - return _ensure_openapi_dict( + parsed = _ensure_openapi_dict( _parse_spec_text( response.text, content_type=response.headers.get("content-type", ""), ), full_url, ) + return _bundle_external_refs( + parsed, + str(getattr(response, "url", full_url)), + timeout=timeout, + headers=merged_headers, + ) def _ensure_openapi_dict(parsed: Any, source_url: str) -> dict[str, Any]: diff --git a/cli/tests/test_dispatchers_openapi_schema.py b/cli/tests/test_dispatchers_openapi_schema.py index a1b9c6c8dbc..cf1daea6a90 100644 --- a/cli/tests/test_dispatchers_openapi_schema.py +++ b/cli/tests/test_dispatchers_openapi_schema.py @@ -13,6 +13,7 @@ import argparse import pytest from openbb_cli.dispatchers.openapi_schema import ( + _bundle_external_refs, _is_json_arg, _provider_choices, _resolve_schema, @@ -1458,3 +1459,228 @@ def test_fetch_openapi_rejects_non_dict_body(monkeypatch): monkeypatch.setattr(openapi_schema.httpx, "get", lambda *a, **k: _Resp()) with pytest.raises(ValueError, match="not an OpenAPI document"): openapi_schema.fetch_openapi("http://h", path="/swagger/v1/swagger.json") + + +def test_bundle_external_refs_resolves_nested_same_origin_documents(monkeypatch): + from openbb_cli.dispatchers import openapi_schema + + documents = { + "https://api.example/spec/paths/items.json": ( + '{"post":{"requestBody":{"content":{"application/json":' + '{"schema":{"$ref":"../schemas/body.json"}}}}}}' + ), + "https://api.example/spec/schemas/body.json": ( + '{"type":["object","null"],"properties":{"symbol":{"type":"string"}}}' + ), + } + + class _Response: + status_code = 200 + headers = {"content-type": "application/json"} + is_redirect = False + encoding = "utf-8" + + def __init__(self, url): + self.url = url + self.text = documents[url] + self.content = self.text.encode() + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def iter_bytes(self, *, chunk_size): + assert chunk_size == 64 * 1024 + yield self.content + + def raise_for_status(self): + return None + + monkeypatch.setattr( + openapi_schema.httpx, + "stream", + lambda _method, url, **_kwargs: _Response(url), + ) + bundled = _bundle_external_refs( + {"openapi": "3.1.0", "paths": {"/items": {"$ref": "paths/items.json"}}}, + "https://api.example/spec/openapi.json", + timeout=1, + headers={}, + ) + + schema = bundled["paths"]["/items"]["post"]["requestBody"]["content"][ + "application/json" + ]["schema"] + assert schema["anyOf"] == [{"type": "object"}, {"type": "null"}] + assert request_body_parameters(schema)[0]["name"] == "symbol" + + +def test_bundle_external_refs_rejects_redirects(monkeypatch): + from openbb_cli.dispatchers import openapi_schema + + class _Redirect: + status_code = 302 + headers = {"location": "http://169.254.169.254/latest/meta-data"} + text = "" + content = b"" + is_redirect = True + encoding = "utf-8" + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + monkeypatch.setattr( + openapi_schema.httpx, "stream", lambda *_args, **_kwargs: _Redirect() + ) + + with pytest.raises(ValueError, match="Redirects are not allowed"): + _bundle_external_refs( + {"openapi": "3.1.0", "paths": {"/x": {"$ref": "x.json"}}}, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + +def test_bundle_external_refs_rejects_oversized_content_length(monkeypatch): + from openbb_cli.dispatchers import openapi_schema + + class _Oversized: + status_code = 200 + headers = {"content-length": str(8 * 1024 * 1024 + 1)} + is_redirect = False + encoding = "utf-8" + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def raise_for_status(self): + return None + + def iter_bytes(self): + raise AssertionError("body must not be buffered") + + monkeypatch.setattr( + openapi_schema.httpx, "stream", lambda *_args, **_kwargs: _Oversized() + ) + with pytest.raises(ValueError, match="size limit"): + _bundle_external_refs( + {"openapi": "3.1.0", "paths": {"/x": {"$ref": "x.json"}}}, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + +def test_bundle_external_refs_rejects_oversized_stream(monkeypatch): + from openbb_cli.dispatchers import openapi_schema + + class _Oversized: + status_code = 200 + headers = {} + is_redirect = False + encoding = "utf-8" + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def raise_for_status(self): + return None + + def iter_bytes(self, *, chunk_size): + assert chunk_size == 64 * 1024 + yield b"x" * (8 * 1024 * 1024 + 1) + + monkeypatch.setattr( + openapi_schema.httpx, "stream", lambda *_args, **_kwargs: _Oversized() + ) + with pytest.raises(ValueError, match="size limit"): + _bundle_external_refs( + {"openapi": "3.1.0", "paths": {"/x": {"$ref": "x.json"}}}, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + +def test_bundle_external_refs_rejects_cross_origin_urls(): + with pytest.raises(ValueError, match="same-origin"): + _bundle_external_refs( + { + "openapi": "3.1.0", + "paths": {"/x": {"$ref": "https://other.example/x.json"}}, + }, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + +def test_bundle_external_refs_rejects_resource_identifiers(): + with pytest.raises(ValueError, match="resource identifiers"): + _bundle_external_refs( + {"openapi": "3.1.0", "$id": "nested.json", "paths": {}}, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + +def test_bundle_external_refs_preserves_non_nullable_type_unions(): + bundled = _bundle_external_refs( + {"openapi": "3.1.0", "type": ["integer", "number"], "paths": {}}, + "https://api.example/openapi.json", + timeout=1, + headers={}, + ) + + assert bundled["anyOf"] == [{"type": "integer"}, {"type": "number"}] + + +def test_bundle_external_refs_rejects_schema_ref_siblings(monkeypatch): + from openbb_cli.dispatchers import openapi_schema + + class _Response: + status_code = 200 + headers = {"content-type": "application/json"} + is_redirect = False + encoding = "utf-8" + content = b'{"type":"integer"}' + + def __enter__(self): + return self + + def __exit__(self, *_args): + return None + + def iter_bytes(self, *, chunk_size): + assert chunk_size == 64 * 1024 + yield self.content + + def raise_for_status(self): + return None + + monkeypatch.setattr( + openapi_schema.httpx, "stream", lambda *_args, **_kwargs: _Response() + ) + with pytest.raises(ValueError, match="reference siblings"): + _bundle_external_refs( + { + "openapi": "3.1.0", + "paths": {"/x": {"$ref": "x.json", "minimum": 1}}, + }, + "https://api.example/openapi.json", + timeout=1, + headers={}, + )