From 1f07534ab878d11629def6af11b661bb17faa8d1 Mon Sep 17 00:00:00 2001 From: Sergey Antropoff Date: Sun, 12 Jul 2026 23:16:56 +0300 Subject: [PATCH] feat: parse API Viewer source artifacts --- app/contracts/__init__.py | 1 + app/contracts/source.py | 151 +++++++++++++++++++++++++++++ docs/api-viewer-research.md | 11 +++ tests/unit/test_contract_source.py | 57 +++++++++++ 4 files changed, 220 insertions(+) create mode 100644 app/contracts/__init__.py create mode 100644 app/contracts/source.py create mode 100644 tests/unit/test_contract_source.py diff --git a/app/contracts/__init__.py b/app/contracts/__init__.py new file mode 100644 index 0000000..3af4470 --- /dev/null +++ b/app/contracts/__init__.py @@ -0,0 +1 @@ +"""Authoritative API contract ingestion and normalization.""" diff --git a/app/contracts/source.py b/app/contracts/source.py new file mode 100644 index 0000000..40732b1 --- /dev/null +++ b/app/contracts/source.py @@ -0,0 +1,151 @@ +"""Safe source adapters for Proxmox API Viewer artifacts.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Protocol, cast + + +class SourceError(ValueError): + """Raised when an API source cannot be parsed safely.""" + + +@dataclass(frozen=True, slots=True) +class ParseWarning: + """A recoverable variation found in a source artifact.""" + + code: str + path: str + message: str + + +@dataclass(frozen=True, slots=True) +class ParsedSource: + """Parsed source tree with non-fatal diagnostics.""" + + nodes: tuple[dict[str, Any], ...] + warnings: tuple[ParseWarning, ...] = () + + +class SourceImporter(Protocol): + """Asynchronous boundary for obtaining source artifact bytes.""" + + async def load(self) -> bytes: ... + + +@dataclass(frozen=True, slots=True) +class LocalFileImporter: + """Load an artifact from a caller-selected local path.""" + + path: Path + + async def load(self) -> bytes: + return self.path.read_bytes() + + +class ApiViewerParser: + """Extract the JSON-compatible ``apiSchema`` value without executing JS.""" + + declaration = b"const apiSchema" + known_node_fields = frozenset({"children", "info", "leaf", "path", "text"}) + + def parse(self, raw: bytes) -> ParsedSource: + if not raw.strip(): + raise SourceError("source artifact is empty") + + payload = self._extract_payload(raw) + try: + decoded = json.loads(payload) + except (UnicodeDecodeError, json.JSONDecodeError) as exc: + raise SourceError(f"invalid apiSchema JSON: {exc}") from exc + + if isinstance(decoded, Mapping): + raw_nodes = [decoded] + elif isinstance(decoded, list): + raw_nodes = decoded + else: + raise SourceError("apiSchema must be an object or array of objects") + + nodes: list[dict[str, Any]] = [] + warnings: list[ParseWarning] = [] + for index, value in enumerate(raw_nodes): + if not isinstance(value, Mapping): + raise SourceError(f"apiSchema node /{index} must be an object") + node = cast(dict[str, Any], dict(value)) + nodes.append(node) + self._inspect_node(node, f"/{index}", warnings) + return ParsedSource(tuple(nodes), tuple(warnings)) + + def _extract_payload(self, raw: bytes) -> bytes: + stripped = raw.strip() + if stripped.startswith((b"[", b"{")): + return stripped + + declaration_at = raw.find(self.declaration) + if declaration_at < 0: + raise SourceError("apiSchema declaration was not found") + equals_at = raw.find(b"=", declaration_at + len(self.declaration)) + if equals_at < 0: + raise SourceError("apiSchema declaration has no assignment") + + start = self._next_non_space(raw, equals_at + 1) + if start >= len(raw) or raw[start] not in b"[{": + raise SourceError("apiSchema assignment must start with an array or object") + end = self._matching_end(raw, start) + return raw[start : end + 1] + + @staticmethod + def _next_non_space(raw: bytes, start: int) -> int: + while start < len(raw) and raw[start] in b" \t\r\n": + start += 1 + return start + + @staticmethod + def _matching_end(raw: bytes, start: int) -> int: + opening = raw[start] + closing = ord("]") if opening == ord("[") else ord("}") + depth = 0 + in_string = False + escaped = False + for index in range(start, len(raw)): + byte = raw[index] + if in_string: + if escaped: + escaped = False + elif byte == ord("\\"): + escaped = True + elif byte == ord('"'): + in_string = False + continue + if byte == ord('"'): + in_string = True + elif byte == opening: + depth += 1 + elif byte == closing: + depth -= 1 + if depth == 0: + return index + raise SourceError("apiSchema assignment is truncated") + + def _inspect_node( + self, node: Mapping[str, Any], path: str, warnings: list[ParseWarning] + ) -> None: + for field in sorted(node.keys() - self.known_node_fields): + warnings.append( + ParseWarning("unknown-node-field", f"{path}/{field}", "field was preserved") + ) + children = node.get("children", []) + if not isinstance(children, list): + warnings.append( + ParseWarning("invalid-children", f"{path}/children", "expected an array") + ) + return + for index, child in enumerate(children): + child_path = f"{path}/children/{index}" + if isinstance(child, Mapping): + self._inspect_node(child, child_path, warnings) + else: + warnings.append(ParseWarning("invalid-child", child_path, "expected an object")) diff --git a/docs/api-viewer-research.md b/docs/api-viewer-research.md index af1f179..bc89470 100644 --- a/docs/api-viewer-research.md +++ b/docs/api-viewer-research.md @@ -57,3 +57,14 @@ The fixture is not a complete snapshot and must never be used to claim broad Proxmox compatibility. Full imports should preserve the immutable raw `apidoc.js`, response metadata, retrieval timestamp, and checksum outside the small test-fixture path. + +## Parser boundary + +`app.contracts.source.ApiViewerParser` accepts either the saved JSON sample or +the official JavaScript wrapper. It locates the exact `const apiSchema` +assignment, scans the balanced JSON value while respecting escaped strings, and +decodes only that value; no downloaded JavaScript is evaluated. Recoverable +tree variations produce structured warnings and unknown node fields remain in +the parsed dictionaries. `SourceImporter` and `LocalFileImporter` keep artifact +retrieval separate from parsing so later remote imports can enforce their own +network policy. diff --git a/tests/unit/test_contract_source.py b/tests/unit/test_contract_source.py new file mode 100644 index 0000000..edd1338 --- /dev/null +++ b/tests/unit/test_contract_source.py @@ -0,0 +1,57 @@ +"""Tests for safe API Viewer source parsing.""" + +import json +from pathlib import Path + +import pytest + +from app.contracts.source import ApiViewerParser, LocalFileImporter, SourceError + +FIXTURE = Path(__file__).parents[1] / "fixtures" / "api-viewer" / "pve-9.2.3-version.json" + + +def test_parse_saved_json_fixture() -> None: + parsed = ApiViewerParser().parse(FIXTURE.read_bytes()) + + assert parsed.nodes[0]["path"] == "/version" + assert parsed.warnings == () + + +def test_extract_api_schema_without_executing_trailing_javascript() -> None: + raw = b'const apiSchema = [{"path":"/x]y","leaf":1}]; throw new Error("no");' + + parsed = ApiViewerParser().parse(raw) + + assert parsed.nodes[0]["path"] == "/x]y" + + +@pytest.mark.parametrize( + "raw, message", + [ + (b"", "empty"), + (b"const other = [];", "not found"), + (b"const apiSchema = [", "truncated"), + (b"const apiSchema = [}];", "invalid"), + (b"42", "not found"), + ], +) +def test_reject_malformed_sources(raw: bytes, message: str) -> None: + with pytest.raises(SourceError, match=message): + ApiViewerParser().parse(raw) + + +def test_preserve_unknown_fields_and_warn() -> None: + raw = json.dumps([{"path": "/version", "future": {"enabled": True}}]).encode() + + parsed = ApiViewerParser().parse(raw) + + assert parsed.nodes[0]["future"] == {"enabled": True} + assert parsed.warnings[0].code == "unknown-node-field" + assert parsed.warnings[0].path == "/0/future" + + +async def test_local_file_importer(tmp_path: Path) -> None: + artifact = tmp_path / "api.json" + artifact.write_bytes(b"[]") + + assert await LocalFileImporter(artifact).load() == b"[]"