From f00aa65a73eb78374350379d1acfca5e1ce67828 Mon Sep 17 00:00:00 2001 From: Andraxion Date: Wed, 29 Jul 2026 14:52:38 -0400 Subject: [PATCH] Bound adapter assemblies and extraction caches --- src/docforge/adapter_contract.py | 49 +++++++++++++++ src/docforge/incremental.py | 71 ++++++++++++++++++--- tests/test_adapter_contract.py | 18 ++++++ tests/test_incremental_cache.py | 102 +++++++++++++++++++++++++++++++ 4 files changed, 232 insertions(+), 8 deletions(-) create mode 100644 tests/test_incremental_cache.py diff --git a/src/docforge/adapter_contract.py b/src/docforge/adapter_contract.py index 1b396b8..c0829bb 100644 --- a/src/docforge/adapter_contract.py +++ b/src/docforge/adapter_contract.py @@ -21,6 +21,8 @@ from .adapter_validation import ( from .config_validation import ID_PATTERN from .errors import DocForgeError from .incremental import ( + MAX_EXTRACTION_CACHE_BYTES, + MAX_EXTRACTION_CACHE_SOURCES, CachedSource, ExtractionCache, affected_sources, @@ -40,6 +42,10 @@ from .models import ( ) from .telemetry import increment, stage +MAX_ADAPTER_EDGE_MULTIPLIER = 32 +MAX_ADAPTER_LOGIC_NODE_MULTIPLIER = 32 +MAX_ADAPTER_LOGIC_EDGE_MULTIPLIER = 64 + @dataclass(frozen=True) class AdapterNode: @@ -464,6 +470,9 @@ class AdapterProject: canonical_sources, captured, ) + self._enforce_assembly_limits( + AdapterAssembly(projection=projection, logic=self._last_logic) + ) return ProjectSnapshot( descriptor=self.descriptor, nodes=projection.core_nodes(), @@ -628,11 +637,14 @@ class AdapterProject: validate_manifest(manifest) if manifest.identity() != self._identity: raise DocForgeError("adapter_changed", "Adapter identity changed during the operation") + self._enforce_manifest_limits(manifest) cache = load_extraction_cache( self._cache_path, project_id=manifest.project_id, adapter_id=manifest.adapter_id, adapter_version=manifest.adapter_version, + max_bytes=MAX_EXTRACTION_CACHE_BYTES, + max_sources=MAX_EXTRACTION_CACHE_SOURCES, ) cached = {source.source_id: source for source in cache.sources} if cache else {} current = {source.source_id: source for source in manifest.sources} @@ -737,6 +749,7 @@ class AdapterProject: "Incremental assembly changed the manifest-bound project identity", ) _validate_adapter_assembly(assembly) + self._enforce_assembly_limits(assembly) stable = loader.load_manifest() validate_manifest(stable) if stable != manifest: @@ -752,6 +765,8 @@ class AdapterProject: adapter_version=manifest.adapter_version, sources=tuple(cache_records), ), + max_bytes=MAX_EXTRACTION_CACHE_BYTES, + max_sources=MAX_EXTRACTION_CACHE_SOURCES, ) self._last_logic = logic_projections self._last_build_report = { @@ -766,6 +781,40 @@ class AdapterProject: } return assembly + def _enforce_manifest_limits(self, manifest: AdapterManifest) -> None: + if len(manifest.sources) > MAX_EXTRACTION_CACHE_SOURCES: + raise DocForgeError( + "adapter_limit", + "Adapter manifest exceeds the bounded source limit", + maximum=MAX_EXTRACTION_CACHE_SOURCES, + actual=len(manifest.sources), + ) + + def _enforce_assembly_limits(self, assembly: AdapterAssembly) -> None: + maximum_nodes = self.descriptor.limits.max_nodes + maximum_edges = maximum_nodes * MAX_ADAPTER_EDGE_MULTIPLIER + maximum_logic_nodes = maximum_nodes * MAX_ADAPTER_LOGIC_NODE_MULTIPLIER + maximum_logic_edges = maximum_nodes * MAX_ADAPTER_LOGIC_EDGE_MULTIPLIER + actual_nodes = len(assembly.projection.nodes) + actual_edges = len(assembly.projection.edges) + actual_logic_nodes = sum(len(projection.nodes) for projection in assembly.logic) + actual_logic_edges = sum(len(projection.edges) for projection in assembly.logic) + limits = ( + ("nodes", actual_nodes, maximum_nodes), + ("edges", actual_edges, maximum_edges), + ("logic_nodes", actual_logic_nodes, maximum_logic_nodes), + ("logic_edges", actual_logic_edges, maximum_logic_edges), + ) + for label, actual, maximum in limits: + if actual > maximum: + raise DocForgeError( + "adapter_limit", + f"Adapter assembly exceeds the configured {label.replace('_', ' ')} limit", + kind=label, + maximum=maximum, + actual=actual, + ) + def canonical_source_paths(self) -> tuple[Path, ...]: """Adapters validate their own source sets before producing a projection.""" diff --git a/src/docforge/incremental.py b/src/docforge/incremental.py index 776111d..e728b12 100644 --- a/src/docforge/incremental.py +++ b/src/docforge/incremental.py @@ -4,6 +4,7 @@ from __future__ import annotations import json import os +import stat import tempfile from dataclasses import dataclass from pathlib import Path @@ -12,6 +13,8 @@ from typing import Any, cast from .errors import DocForgeError EXTRACTION_CACHE_SCHEMA_VERSION = 1 +MAX_EXTRACTION_CACHE_BYTES = 64_000_000 +MAX_EXTRACTION_CACHE_SOURCES = 10_000 @dataclass(frozen=True) @@ -42,13 +45,39 @@ def load_extraction_cache( project_id: str, adapter_id: str, adapter_version: str, + max_bytes: int = MAX_EXTRACTION_CACHE_BYTES, + max_sources: int = MAX_EXTRACTION_CACHE_SOURCES, ) -> ExtractionCache | None: """Read a cache generation, treating malformed or incompatible data as a miss.""" - if not path.is_file() or path.is_symlink(): + if max_bytes < 1 or max_sources < 1: + raise ValueError("Extraction cache limits must be positive") + try: + descriptor = os.open( + path, + os.O_RDONLY + | getattr(os, "O_CLOEXEC", 0) + | getattr(os, "O_NOFOLLOW", 0) + | getattr(os, "O_NONBLOCK", 0), + ) + except OSError: return None try: - raw = json.loads(path.read_text(encoding="utf-8")) + with os.fdopen(descriptor, "rb") as handle: + before = os.fstat(handle.fileno()) + if not stat.S_ISREG(before.st_mode) or before.st_size > max_bytes: + return None + encoded = handle.read(max_bytes + 1) + after = os.fstat(handle.fileno()) + if ( + len(encoded) > max_bytes + or before.st_dev != after.st_dev + or before.st_ino != after.st_ino + or before.st_size != after.st_size + or before.st_mtime_ns != after.st_mtime_ns + ): + return None + raw = json.loads(encoded.decode("utf-8")) if not isinstance(raw, dict): return None document = cast(dict[str, Any], raw) @@ -62,8 +91,11 @@ def load_extraction_cache( raw_sources = document.get("sources") if not isinstance(raw_sources, list): return None + source_items = cast(list[object], raw_sources) + if len(source_items) > max_sources: + return None sources: list[CachedSource] = [] - for raw_source in cast(list[object], raw_sources): + for raw_source in source_items: if not isinstance(raw_source, dict): return None item = cast(dict[str, object], raw_source) @@ -110,9 +142,24 @@ def load_extraction_cache( return None -def write_extraction_cache(path: Path, cache: ExtractionCache) -> None: +def write_extraction_cache( + path: Path, + cache: ExtractionCache, + *, + max_bytes: int = MAX_EXTRACTION_CACHE_BYTES, + max_sources: int = MAX_EXTRACTION_CACHE_SOURCES, +) -> None: """Atomically publish one validated extraction-cache generation.""" + if max_bytes < 1 or max_sources < 1: + raise ValueError("Extraction cache limits must be positive") + if len(cache.sources) > max_sources: + raise DocForgeError( + "cache_limit", + "Incremental extraction cache exceeds its source limit", + maximum=max_sources, + actual=len(cache.sources), + ) path.parent.mkdir(parents=True, exist_ok=True) document = { "schema_version": EXTRACTION_CACHE_SCHEMA_VERSION, @@ -131,17 +178,25 @@ def write_extraction_cache(path: Path, cache: ExtractionCache) -> None: for source in cache.sources ], } + encoded = ( + json.dumps(document, sort_keys=True, separators=(",", ":")).encode("utf-8") + b"\n" + ) + if len(encoded) > max_bytes: + raise DocForgeError( + "cache_limit", + "Incremental extraction cache exceeds its byte limit", + maximum=max_bytes, + actual=len(encoded), + ) with tempfile.NamedTemporaryFile( - mode="w", - encoding="utf-8", + mode="wb", prefix="extractions-", suffix=".json", dir=path.parent, delete=False, ) as descriptor: temporary = Path(descriptor.name) - json.dump(document, descriptor, sort_keys=True, separators=(",", ":")) - descriptor.write("\n") + descriptor.write(encoded) descriptor.flush() os.fsync(descriptor.fileno()) try: diff --git a/tests/test_adapter_contract.py b/tests/test_adapter_contract.py index 2d5bc3f..ee1b183 100644 --- a/tests/test_adapter_contract.py +++ b/tests/test_adapter_contract.py @@ -39,6 +39,7 @@ from docforge.mcp_server import ( ) from docforge.models import ( Edge, + Limits, LogicEdge, LogicNode, LogicProjection, @@ -341,6 +342,23 @@ class AdapterContractTests(unittest.TestCase): self.assertEqual("Workflow", index.get_node("guide.workflow")["node"]["title"]) self.assertEqual(projection.identity(), projection.identity()) + def test_adapter_project_enforces_graph_bounds_before_publication(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory).resolve() + project = AdapterProject( + Loader(self.projection(root)), + cache_root=root / ".cache" / "bounded", + settings=AdapterProjectSettings(limits=Limits(max_nodes=1)), + ) + + with self.assertRaises(DocForgeError) as captured: + project.load() + + self.assertEqual("adapter_limit", captured.exception.code) + self.assertEqual("nodes", captured.exception.details["kind"]) + self.assertEqual(1, captured.exception.details["maximum"]) + self.assertEqual(2, captured.exception.details["actual"]) + def test_projection_rejects_unsorted_metadata_graph_and_identity_changes(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory).resolve() diff --git a/tests/test_incremental_cache.py b/tests/test_incremental_cache.py new file mode 100644 index 0000000..cfd2a9d --- /dev/null +++ b/tests/test_incremental_cache.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import hashlib +import os +import tempfile +import unittest +from pathlib import Path + +from docforge.errors import DocForgeError +from docforge.incremental import ( + CachedSource, + ExtractionCache, + load_extraction_cache, + write_extraction_cache, +) + + +class ExtractionCacheBoundsTests(unittest.TestCase): + @staticmethod + def cache(*, payload: str = "ok", count: int = 1) -> ExtractionCache: + sources = tuple( + CachedSource( + source_id=f"source-{index}", + source_path=f"src/{index}.txt", + fingerprint=hashlib.sha256(str(index).encode()).hexdigest(), + extractor_version="fixture@1", + dependencies=(), + payload={"content": payload}, + ) + for index in range(count) + ) + return ExtractionCache("fixture", "fixture-adapter", "1", sources) + + def test_bounded_round_trip_and_oversized_read_is_a_cache_miss(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "extractions.json" + write_extraction_cache(path, self.cache(), max_bytes=1_000, max_sources=2) + loaded = load_extraction_cache( + path, + project_id="fixture", + adapter_id="fixture-adapter", + adapter_version="1", + max_bytes=1_000, + max_sources=2, + ) + self.assertEqual(self.cache(), loaded) + + with path.open("ab") as handle: + handle.write(b" " * 1_000) + self.assertIsNone( + load_extraction_cache( + path, + project_id="fixture", + adapter_id="fixture-adapter", + adapter_version="1", + max_bytes=1_000, + max_sources=2, + ) + ) + + def test_write_rejects_byte_and_source_limits_without_replacing_target(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "extractions.json" + path.write_bytes(b"preserved\n") + for cache, maximum_bytes, maximum_sources in ( + (self.cache(payload="x" * 500), 100, 2), + (self.cache(count=2), 1_000, 1), + ): + with self.assertRaises(DocForgeError) as captured: + write_extraction_cache( + path, + cache, + max_bytes=maximum_bytes, + max_sources=maximum_sources, + ) + self.assertEqual("cache_limit", captured.exception.code) + self.assertEqual(b"preserved\n", path.read_bytes()) + + def test_symlink_and_non_regular_targets_are_cache_misses(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + real = root / "real.json" + write_extraction_cache(real, self.cache(), max_bytes=1_000, max_sources=2) + linked = root / "linked.json" + linked.symlink_to(real) + fifo = root / "fifo" + os.mkfifo(fifo) + for path in (linked, fifo): + self.assertIsNone( + load_extraction_cache( + path, + project_id="fixture", + adapter_id="fixture-adapter", + adapter_version="1", + max_bytes=1_000, + max_sources=2, + ) + ) + + +if __name__ == "__main__": + unittest.main()