From 9c99037da0ae9b5b681ad995957f2067a233ffb3 Mon Sep 17 00:00:00 2001 From: Shaggi Date: Thu, 30 Jul 2026 00:35:09 +0300 Subject: [PATCH 1/4] feat: add pinned Graphify POC adapter --- README.md | 7 + docs/graphify-poc.md | 89 +++ .../analyzer/graphify_adapter.py | 543 ++++++++++++++++++ tests/fixtures/graphify_0_9_30_graph.json | 83 +++ tests/unit/test_graphify_adapter.py | 283 +++++++++ 5 files changed, 1005 insertions(+) create mode 100644 docs/graphify-poc.md create mode 100644 src/fastapi_endpoint_detector/analyzer/graphify_adapter.py create mode 100644 tests/fixtures/graphify_0_9_30_graph.json create mode 100644 tests/unit/test_graphify_adapter.py diff --git a/README.md b/README.md index 7cc181f..fc480ac 100644 --- a/README.md +++ b/README.md @@ -177,6 +177,13 @@ escape, mutation, control flow, or dynamic registration as conditional evidence; no bootstrap/helper names are guessed. Dynamic behavior remains conservatively unresolved. +The optional Graphify experiment is not a configured analyzer backend. A private +POC adapter can explicitly run the separately installed, pinned +`graphifyy==0.9.30` command in code-only mode and validate immutable +baseline/target graph snapshots. It never runs by default, never uses semantic +LLM/server features, and cannot promote Graphify communities or similarity into +blast-radius evidence. See [the Graphify POC boundary](docs/graphify-poc.md). + ### `list` - List Endpoints List all FastAPI endpoints discovered in the application. diff --git a/docs/graphify-poc.md b/docs/graphify-poc.md new file mode 100644 index 0000000..f713940 --- /dev/null +++ b/docs/graphify-poc.md @@ -0,0 +1,89 @@ +# Graphify code-graph POC boundary + +Issue #110 evaluates Graphify as an optional generic code-graph provider. This +foundation is deliberately **not** connected to the default CLI or analyzer. +Mypy remains the only default semantic backend. Importing this project never +imports or installs Graphify, and missing Graphify tooling never causes fallback +or changes ordinary analysis. + +## Pinned, explicit execution + +The only attested tool is the separately installed PyPI package +`graphifyy==0.9.30`, whose console command reports exactly `graphify 0.9.30`. +Installation is an operator-controlled prerequisite outside the analyzed +checkout. `GraphifyRunner` takes the absolute executable path explicitly and +invokes this fixed argv shape without a shell: + +```text +graphify extract \ + --code-only --no-cluster --force --out +``` + +The runner: + +- rejects every other tool version; +- strips API keys and proxy variables from the child environment; +- passes `--code-only`, so document/media semantic extraction is disabled; +- never passes backend, model, MCP/server, wiki, watch, hook, global-graph, or + database flags; +- requires a fresh `baseline/` or `target/` directory outside the analyzed + project and never overwrites it; +- uses a private HOME/config directory inside that snapshot directory; +- validates `graphify-out/graph.json` before publishing a deterministic + `snapshot-receipt.json`. + +This is process isolation, not a security sandbox. The Graphify executable still +reads source on the host. Do not use the POC on untrusted source until it is +placed behind the hardened runtime boundary. + +## Supported `graph.json` contract + +The adapter version is `1`, bound to Graphify 0.9.30's NetworkX node-link JSON: + +| Level | Required contract | +|---|---| +| document | exactly `directed`, `multigraph`, `graph`, `nodes`, `links`, `hyperedges`, and optional `built_at_commit` | +| node | unique string `id`, string `label`, `file_type: code`, project-confined `source_file`, optional one-based `source_location` | +| link | existing `source` and `target` IDs, string `relation`, `confidence` in `EXTRACTED`, `INFERRED`, `AMBIGUOUS`, optional source span | +| semantic data | `hyperedges` must be empty | + +Duplicate JSON members, non-finite numbers, invalid UTF-8, schema drift, dangling +edges, duplicate node IDs, non-code nodes, paths outside the project, malformed +or reversed ranges, oversized graphs, and unexpected fields fail closed. +`source_location` accepts `L12`, `12`, or an inclusive range such as `L12-L18`. + +The adapter retains Graphify extraction strength separately from detector +confidence. It marks only source-backed `calls`, `imports`, `imports_from`, +`inherits`, `references`, and `re_exports` links as eligible for a future +traversal. `contains`, communities, similarity, natural-language relations, and +unknown relation types cannot become blast-radius evidence. + +## Immutable snapshot receipt + +The adapter reads one bounded byte snapshot, hashes those exact bytes with +SHA-256, then validates them. The exclusive receipt records: + +- side (`baseline` or `target`); +- Graphify package version and adapter schema version; +- exact graph SHA-256; +- node and edge counts. + +Baseline and target always occupy distinct fresh directories. A caller can pass +an expected SHA-256 when reopening a snapshot; a mismatch is an explicit error. + +## Current decision and remaining gates + +**Decision: BUILD the isolated adapter foundation; do not ADOPT the backend.** + +Before an ADOPT or HYBRID decision, a later tranche must still: + +1. verify Graphify 0.9.30 on controlled Python fixtures for aliases, methods, + inheritance, imports/re-exports, deleted source, and cross-file calls; +2. overlay secure FastAPI handler/DI identity without modifying `graph.json`; +3. calibrate EXTRACTED/INFERRED/AMBIGUOUS edges against HIGH/MEDIUM/LOW policy; +4. run target and baseline corpus comparisons and report candidate gain, false + positives, failures/abstentions, graph size, latency, and peak RSS; +5. prove no regression relative to mypy and record ADOPT, HYBRID, or STOP. + +No community, proximity, semantic label, or natural-language query result may +satisfy these gates. diff --git a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py new file mode 100644 index 0000000..7c68b93 --- /dev/null +++ b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py @@ -0,0 +1,543 @@ +"""Strict, opt-in adapter for pinned Graphify code-only snapshots. + +This module is deliberately not wired into the default analyzer. It provides a +bounded POC boundary that can invoke a separately installed, pinned Graphify +binary in code-only mode and adapt its immutable ``graph.json`` bytes into a +small source/provenance IR. It never installs tools, starts servers, queries a +semantic backend, or treats Graphify confidence as detector confidence. +""" + +from __future__ import annotations + +import hashlib +import json +import math +import os +import re +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Literal, cast + +from fastapi_endpoint_detector.strict_data import DuplicateKeyError, load_json_unique + +GRAPHIFY_PACKAGE_VERSION = "0.9.30" +GRAPHIFY_GRAPH_SCHEMA_VERSION = 1 +GRAPHIFY_OUTPUT_DIRECTORY = "graphify-out" +MAX_GRAPH_BYTES = 64 * 1024 * 1024 +MAX_GRAPH_NODES = 250_000 +MAX_GRAPH_EDGES = 1_000_000 +MAX_TEXT_LENGTH = 16_384 + +GraphSide = Literal["baseline", "target"] +GraphifyStrength = Literal["EXTRACTED", "INFERRED", "AMBIGUOUS"] + +_TRAVERSABLE_RELATIONS = frozenset( + {"calls", "imports", "imports_from", "inherits", "references", "re_exports"} +) +_TOP_LEVEL_KEYS = frozenset( + {"directed", "multigraph", "graph", "nodes", "links", "hyperedges", "built_at_commit"} +) +_NODE_KEYS = frozenset( + { + "id", + "label", + "file_type", + "source_file", + "source_location", + "confidence", + "confidence_score", + "community", + "community_name", + "norm_label", + "type", + "kind", + "metadata", + "origin_file", + "scope_id", + "scope_kind", + "target_file", + "target_fqn", + "package", + "namespace", + } +) +_EDGE_KEYS = frozenset( + { + "source", + "target", + "relation", + "confidence", + "confidence_score", + "source_file", + "source_location", + "weight", + "context", + "metadata", + "key", + "target_file", + "target_fqn", + "origin_file", + } +) +_LOCATION = re.compile(r"^(?:L)?(?P[1-9][0-9]*)(?:-(?:L)?(?P[1-9][0-9]*))?$") +_GIT_OID = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$") + + +class GraphifyAdapterError(RuntimeError): + """Raised when the explicit Graphify POC cannot produce trustworthy evidence.""" + + +@dataclass(frozen=True) +class GraphifySourceSpan: + """One project-relative, one-based inclusive source span.""" + + file_path: Path + start_line: int + end_line: int + + +@dataclass(frozen=True) +class GraphifyNode: + """A normalized code node from the pinned Graphify schema.""" + + node_id: str + label: str + span: GraphifySourceSpan | None + extractor_strength: GraphifyStrength | None + + +@dataclass(frozen=True) +class GraphifyEdge: + """A directed Graphify edge with extractor provenance kept separate.""" + + source_id: str + target_id: str + relation: str + extractor_strength: GraphifyStrength + span: GraphifySourceSpan | None + + @property + def traversable(self) -> bool: + """Whether the relation is eligible for later evidence-bearing traversal.""" + return self.relation in _TRAVERSABLE_RELATIONS and self.span is not None + + +@dataclass(frozen=True) +class GraphifySnapshot: + """One immutable byte snapshot adapted from a pinned Graphify graph.""" + + side: GraphSide + graph_sha256: str + graph_schema_version: int + graphify_version: str + built_at_commit: str | None + nodes: tuple[GraphifyNode, ...] + edges: tuple[GraphifyEdge, ...] + + +@dataclass(frozen=True) +class GraphifySnapshotReceipt: + """Durable receipt written beside a successfully validated graph snapshot.""" + + side: GraphSide + graph_sha256: str + graph_schema_version: int + graphify_version: str + node_count: int + edge_count: int + + def as_json(self) -> str: + """Return deterministic strict JSON for exclusive publication.""" + return ( + json.dumps( + { + "edge_count": self.edge_count, + "graph_schema_version": self.graph_schema_version, + "graph_sha256": self.graph_sha256, + "graphify_version": self.graphify_version, + "node_count": self.node_count, + "side": self.side, + }, + allow_nan=False, + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +def _bounded_string(value: object, location: str, *, allow_empty: bool = False) -> str: + if not isinstance(value, str) or (not allow_empty and not value): + raise GraphifyAdapterError(f"{location} must be a non-empty string") + if len(value) > MAX_TEXT_LENGTH or "\x00" in value: + raise GraphifyAdapterError(f"{location} is invalid or too large") + return value + + +def _finite_number(value: object, location: str) -> float: + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise GraphifyAdapterError(f"{location} must be a finite number") + result = float(value) + if not math.isfinite(result): + raise GraphifyAdapterError(f"{location} must be finite") + return result + + +def _validate_json_numbers(value: object) -> None: + pending = [value] + while pending: + current = pending.pop() + if isinstance(current, float) and not math.isfinite(current): + raise GraphifyAdapterError("graph.json contains a non-finite number") + if isinstance(current, dict): + pending.extend(current.values()) + elif isinstance(current, list): + pending.extend(current) + + +def _strict_json(raw: bytes, source: Path) -> dict[str, object]: + try: + text = raw.decode("utf-8") + value = load_json_unique(text) + except (UnicodeError, json.JSONDecodeError, DuplicateKeyError) as error: + raise GraphifyAdapterError(f"invalid strict JSON in {source}: {error}") from error + _validate_json_numbers(value) + if not isinstance(value, dict): + raise GraphifyAdapterError("graph.json must contain an object") + if set(value) != _TOP_LEVEL_KEYS and set(value) != _TOP_LEVEL_KEYS - {"built_at_commit"}: + extra = sorted(set(value) - _TOP_LEVEL_KEYS) + missing = sorted((_TOP_LEVEL_KEYS - {"built_at_commit"}) - set(value)) + raise GraphifyAdapterError( + f"unsupported graph.json top-level schema; extra={extra}, missing={missing}" + ) + return cast("dict[str, object]", value) + + +def _relative_source(project_root: Path, value: object, location: str) -> Path | None: + source = _bounded_string(value, location, allow_empty=True) + if not source: + return None + supplied = Path(source) + try: + absolute = ( + supplied.resolve(strict=False) + if supplied.is_absolute() + else (project_root / supplied).resolve(strict=False) + ) + relative = absolute.relative_to(project_root) + except ValueError as error: + raise GraphifyAdapterError(f"{location} escapes the project root: {source!r}") from error + if not relative.parts or ".." in relative.parts: + raise GraphifyAdapterError(f"{location} is not project relative: {source!r}") + if not absolute.is_file(): + raise GraphifyAdapterError(f"{location} does not identify a project file: {source!r}") + return relative + + +def _source_span( + project_root: Path, + source_value: object, + location_value: object, + location: str, +) -> GraphifySourceSpan | None: + file_path = _relative_source(project_root, source_value, f"{location}.source_file") + if location_value in {None, ""}: + return None + if file_path is None: + raise GraphifyAdapterError(f"{location} has a source location without a source file") + raw_location = _bounded_string(location_value, f"{location}.source_location") + match = _LOCATION.fullmatch(raw_location) + if match is None: + raise GraphifyAdapterError(f"{location}.source_location is unsupported: {raw_location!r}") + start_line = int(match.group("start")) + end_line = int(match.group("end") or start_line) + if end_line < start_line: + raise GraphifyAdapterError(f"{location}.source_location has a reversed range") + return GraphifySourceSpan(file_path, start_line, end_line) + + +def _strength(value: object, location: str, *, optional: bool = False) -> GraphifyStrength | None: + if value is None and optional: + return None + if value not in {"EXTRACTED", "INFERRED", "AMBIGUOUS"}: + raise GraphifyAdapterError(f"{location} has unsupported extractor confidence {value!r}") + return cast("GraphifyStrength", value) + + +def _validate_common_optional_fields(item: dict[str, object], location: str) -> None: + if "confidence_score" in item: + score = _finite_number(item["confidence_score"], f"{location}.confidence_score") + if not 0.0 <= score <= 1.0: + raise GraphifyAdapterError(f"{location}.confidence_score must be between 0 and 1") + if "weight" in item: + weight = _finite_number(item["weight"], f"{location}.weight") + if weight < 0.0: + raise GraphifyAdapterError(f"{location}.weight must be non-negative") + for field in ("context", "community_name", "norm_label", "type", "kind"): + if field in item and item[field] is not None: + _bounded_string(item[field], f"{location}.{field}", allow_empty=True) + community = item.get("community") + if community is not None and (type(community) is not int or community < 0): + raise GraphifyAdapterError(f"{location}.community must be a non-negative integer or null") + key = item.get("key") + if key is not None and (isinstance(key, bool) or not isinstance(key, (int, str))): + raise GraphifyAdapterError(f"{location}.key must be an integer or string") + if "metadata" in item and not isinstance(item["metadata"], dict): + raise GraphifyAdapterError(f"{location}.metadata must be an object") + + +def _read_graph_payload( + graph_path: Path, expected_sha256: str | None +) -> tuple[str, dict[str, object]]: + try: + size = graph_path.stat().st_size + if size > MAX_GRAPH_BYTES: + raise GraphifyAdapterError(f"graph.json exceeds {MAX_GRAPH_BYTES} bytes") + raw = graph_path.read_bytes() + except OSError as error: + raise GraphifyAdapterError( + f"cannot read Graphify snapshot {graph_path}: {error}" + ) from error + if len(raw) != size: + raise GraphifyAdapterError("graph.json changed while it was being read") + graph_sha256 = hashlib.sha256(raw).hexdigest() + if expected_sha256 is not None and graph_sha256 != expected_sha256: + raise GraphifyAdapterError( + f"Graphify snapshot hash mismatch: expected {expected_sha256}, got {graph_sha256}" + ) + return graph_sha256, _strict_json(raw, graph_path) + + +def _payload_collections( + payload: dict[str, object], +) -> tuple[list[object], list[object], str | None]: + if type(payload["directed"]) is not bool or type(payload["multigraph"]) is not bool: + raise GraphifyAdapterError("graph.json directed and multigraph fields must be booleans") + if not isinstance(payload["graph"], dict): + raise GraphifyAdapterError("graph.json graph field must be an object") + hyperedges = payload["hyperedges"] + if not isinstance(hyperedges, list) or hyperedges: + raise GraphifyAdapterError("code-only Graphify snapshots must have no semantic hyperedges") + raw_nodes = payload["nodes"] + raw_edges = payload["links"] + if not isinstance(raw_nodes, list) or not isinstance(raw_edges, list): + raise GraphifyAdapterError("graph.json nodes and links must be lists") + if len(raw_nodes) > MAX_GRAPH_NODES or len(raw_edges) > MAX_GRAPH_EDGES: + raise GraphifyAdapterError("graph.json exceeds the bounded node or edge limit") + built_at_commit = payload.get("built_at_commit") + if built_at_commit is not None: + built_at_commit = _bounded_string(built_at_commit, "built_at_commit") + if _GIT_OID.fullmatch(built_at_commit) is None: + raise GraphifyAdapterError("built_at_commit must be a lowercase full Git OID") + return raw_nodes, raw_edges, built_at_commit + + +def _adapt_nodes( + raw_nodes: list[object], project_root: Path +) -> tuple[tuple[GraphifyNode, ...], set[str]]: + nodes: list[GraphifyNode] = [] + node_ids: set[str] = set() + for index, raw_node in enumerate(raw_nodes): + location = f"nodes[{index}]" + if not isinstance(raw_node, dict): + raise GraphifyAdapterError(f"{location} must be an object") + node = cast("dict[str, object]", raw_node) + extra = set(node) - _NODE_KEYS + if extra: + raise GraphifyAdapterError(f"{location} has unsupported fields: {sorted(extra)}") + node_id = _bounded_string(node.get("id"), f"{location}.id") + if node_id in node_ids: + raise GraphifyAdapterError(f"duplicate Graphify node id: {node_id}") + node_ids.add(node_id) + if node.get("file_type") != "code": + raise GraphifyAdapterError(f"{location} is not a code-only node") + label = _bounded_string(node.get("label"), f"{location}.label") + span = _source_span( + project_root, + node.get("source_file", ""), + node.get("source_location"), + location, + ) + strength = _strength(node.get("confidence"), f"{location}.confidence", optional=True) + _validate_common_optional_fields(node, location) + nodes.append(GraphifyNode(node_id, label, span, strength)) + return tuple(nodes), node_ids + + +def _adapt_edges( + raw_edges: list[object], project_root: Path, node_ids: set[str] +) -> tuple[GraphifyEdge, ...]: + edges: list[GraphifyEdge] = [] + for index, raw_edge in enumerate(raw_edges): + location = f"links[{index}]" + if not isinstance(raw_edge, dict): + raise GraphifyAdapterError(f"{location} must be an object") + edge = cast("dict[str, object]", raw_edge) + extra = set(edge) - _EDGE_KEYS + if extra: + raise GraphifyAdapterError(f"{location} has unsupported fields: {sorted(extra)}") + source_id = _bounded_string(edge.get("source"), f"{location}.source") + target_id = _bounded_string(edge.get("target"), f"{location}.target") + if source_id not in node_ids or target_id not in node_ids: + raise GraphifyAdapterError(f"{location} references an unknown node") + relation = _bounded_string(edge.get("relation"), f"{location}.relation") + strength = _strength(edge.get("confidence"), f"{location}.confidence") + assert strength is not None + span = _source_span( + project_root, + edge.get("source_file", ""), + edge.get("source_location"), + location, + ) + _validate_common_optional_fields(edge, location) + edges.append(GraphifyEdge(source_id, target_id, relation, strength, span)) + return tuple(edges) + + +def load_graphify_snapshot( + graph_path: Path, + *, + project_root: Path, + side: GraphSide, + graphify_version: str = GRAPHIFY_PACKAGE_VERSION, + expected_sha256: str | None = None, +) -> GraphifySnapshot: + """Read one immutable graph.json byte snapshot and strictly adapt its evidence.""" + if side not in {"baseline", "target"}: + raise GraphifyAdapterError(f"unsupported snapshot side: {side!r}") + if graphify_version != GRAPHIFY_PACKAGE_VERSION: + raise GraphifyAdapterError( + f"unsupported Graphify version {graphify_version!r}; " + f"expected {GRAPHIFY_PACKAGE_VERSION}" + ) + graph_sha256, payload = _read_graph_payload(graph_path, expected_sha256) + raw_nodes, raw_edges, built_at_commit = _payload_collections(payload) + root = project_root.resolve(strict=True) + nodes, node_ids = _adapt_nodes(raw_nodes, root) + edges = _adapt_edges(raw_edges, root, node_ids) + return GraphifySnapshot( + side=side, + graph_sha256=graph_sha256, + graph_schema_version=GRAPHIFY_GRAPH_SCHEMA_VERSION, + graphify_version=graphify_version, + built_at_commit=built_at_commit, + nodes=nodes, + edges=edges, + ) + + +class GraphifyRunner: + """Explicit pinned Graphify runner; construction never occurs in default analysis.""" + + def __init__(self, project_root: Path, executable: Path, *, timeout: float = 300.0): + self.project_root = project_root.resolve(strict=True) + self.executable = executable.resolve(strict=True) + if not self.project_root.is_dir(): + raise GraphifyAdapterError(f"Graphify project root is not a directory: {project_root}") + if not self.executable.is_file(): + raise GraphifyAdapterError(f"Graphify executable is not a file: {executable}") + if timeout <= 0: + raise GraphifyAdapterError("Graphify timeout must be positive") + self.timeout = timeout + + @staticmethod + def _environment(home: Path) -> dict[str, str]: + allowed = ("PATH", "SYSTEMROOT", "WINDIR", "TMPDIR", "TEMP", "TMP", "LANG", "LC_ALL") + environment = {name: os.environ[name] for name in allowed if name in os.environ} + environment.update( + { + "GRAPHIFY_OUT": GRAPHIFY_OUTPUT_DIRECTORY, + "HOME": str(home), + "XDG_CONFIG_HOME": str(home / ".config"), + } + ) + return environment + + def _run(self, args: list[str], *, cwd: Path, home: Path) -> subprocess.CompletedProcess[str]: + try: + result = subprocess.run( + args, + cwd=cwd, + env=self._environment(home), + check=False, + capture_output=True, + text=True, + encoding="utf-8", + errors="strict", + timeout=self.timeout, + ) + except (OSError, subprocess.TimeoutExpired, UnicodeError) as error: + raise GraphifyAdapterError(f"Graphify command failed to execute: {error}") from error + if result.returncode: + detail = result.stderr.strip() or result.stdout.strip() or "no diagnostic output" + raise GraphifyAdapterError(f"Graphify command failed ({result.returncode}): {detail}") + return result + + def validate_tool(self, *, cwd: Path, home: Path) -> None: + """Require the exact separately installed POC version before extraction.""" + result = self._run([str(self.executable), "--version"], cwd=cwd, home=home) + if result.stdout.strip() != f"graphify {GRAPHIFY_PACKAGE_VERSION}": + raise GraphifyAdapterError( + f"unsupported Graphify version {result.stdout.strip()!r}; " + f"expected graphify {GRAPHIFY_PACKAGE_VERSION}" + ) + + def extract_snapshot(self, side: GraphSide, output_root: Path) -> GraphifySnapshot: + """Run one explicit code-only extraction into a fresh side-qualified directory.""" + if side not in {"baseline", "target"}: + raise GraphifyAdapterError(f"unsupported snapshot side: {side!r}") + root = output_root.expanduser().resolve(strict=False) + if root == self.project_root or root.is_relative_to(self.project_root): + raise GraphifyAdapterError("Graphify output must be outside the analyzed project") + side_directory = root / side + if side_directory.exists() or side_directory.is_symlink(): + raise GraphifyAdapterError( + f"Graphify snapshot directory already exists: {side_directory}" + ) + root.mkdir(parents=True, exist_ok=True) + side_directory.mkdir(mode=0o700) + home = side_directory / "home" + home.mkdir(mode=0o700) + self.validate_tool(cwd=side_directory, home=home) + command = [ + str(self.executable), + "extract", + str(self.project_root), + "--code-only", + "--no-cluster", + "--force", + "--out", + str(side_directory), + ] + self._run(command, cwd=side_directory, home=home) + graph_path = side_directory / GRAPHIFY_OUTPUT_DIRECTORY / "graph.json" + try: + graph_path.resolve(strict=True).relative_to(side_directory.resolve(strict=True)) + except (OSError, ValueError) as error: + raise GraphifyAdapterError("Graphify did not produce a confined graph.json") from error + snapshot = load_graphify_snapshot( + graph_path, + project_root=self.project_root, + side=side, + graphify_version=GRAPHIFY_PACKAGE_VERSION, + ) + receipt = GraphifySnapshotReceipt( + side=snapshot.side, + graph_sha256=snapshot.graph_sha256, + graph_schema_version=snapshot.graph_schema_version, + graphify_version=snapshot.graphify_version, + node_count=len(snapshot.nodes), + edge_count=len(snapshot.edges), + ) + receipt_path = side_directory / "snapshot-receipt.json" + try: + with receipt_path.open("x", encoding="utf-8") as handle: + handle.write(receipt.as_json()) + handle.flush() + os.fsync(handle.fileno()) + except OSError as error: + raise GraphifyAdapterError( + f"cannot publish Graphify snapshot receipt: {error}" + ) from error + return snapshot diff --git a/tests/fixtures/graphify_0_9_30_graph.json b/tests/fixtures/graphify_0_9_30_graph.json new file mode 100644 index 0000000..0cb50cc --- /dev/null +++ b/tests/fixtures/graphify_0_9_30_graph.json @@ -0,0 +1,83 @@ +{ + "built_at_commit": "1111111111111111111111111111111111111111", + "directed": false, + "graph": {}, + "hyperedges": [], + "links": [ + { + "confidence": "EXTRACTED", + "confidence_score": 1.0, + "context": "definition", + "relation": "contains", + "source": "app_file", + "source_file": "app.py", + "source_location": "L2", + "target": "handler", + "weight": 1.0 + }, + { + "confidence": "EXTRACTED", + "confidence_score": 1.0, + "context": "call", + "relation": "calls", + "source": "handler", + "source_file": "app.py", + "source_location": "L3", + "target": "helper", + "weight": 1.0 + }, + { + "confidence": "INFERRED", + "confidence_score": 0.8, + "context": "import", + "relation": "imports", + "source": "app_file", + "source_file": "app.py", + "source_location": "L1-L1", + "target": "helper", + "weight": 1.0 + }, + { + "confidence": "AMBIGUOUS", + "confidence_score": 0.5, + "context": "similarity", + "relation": "related_to", + "source": "handler", + "source_file": "app.py", + "source_location": "L2", + "target": "helper", + "weight": 0.5 + } + ], + "multigraph": true, + "nodes": [ + { + "community": null, + "file_type": "code", + "id": "app_file", + "label": "app.py", + "norm_label": "app.py", + "source_file": "app.py", + "source_location": null + }, + { + "community": null, + "confidence": "EXTRACTED", + "file_type": "code", + "id": "handler", + "label": "handler", + "norm_label": "handler", + "source_file": "app.py", + "source_location": "L2-L3" + }, + { + "community": null, + "file_type": "code", + "id": "helper", + "label": "helper", + "norm_label": "helper", + "source_file": "helpers.py", + "source_location": "L1" + } + ] +} diff --git a/tests/unit/test_graphify_adapter.py b/tests/unit/test_graphify_adapter.py new file mode 100644 index 0000000..59ce8cb --- /dev/null +++ b/tests/unit/test_graphify_adapter.py @@ -0,0 +1,283 @@ +"""Offline contract tests for the opt-in pinned Graphify POC boundary.""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import subprocess +from pathlib import Path +from typing import Any +from unittest.mock import patch + +import pytest + +from fastapi_endpoint_detector.analyzer.graphify_adapter import ( + GRAPHIFY_GRAPH_SCHEMA_VERSION, + GRAPHIFY_PACKAGE_VERSION, + GraphifyAdapterError, + GraphifyRunner, + GraphifySourceSpan, + load_graphify_snapshot, +) + +FIXTURE = Path(__file__).parents[1] / "fixtures" / "graphify_0_9_30_graph.json" + + +def _project(tmp_path: Path) -> Path: + project = tmp_path / "project" + project.mkdir() + (project / "app.py").write_text( + "from helpers import helper\n\ndef handler():\n return helper()\n", encoding="utf-8" + ) + (project / "helpers.py").write_text("def helper():\n return 1\n", encoding="utf-8") + return project + + +def _write_payload(path: Path, payload: dict[str, Any]) -> None: + path.write_text(json.dumps(payload, allow_nan=False, sort_keys=True), encoding="utf-8") + + +def _completed( + stdout: str = "", *, code: int = 0, stderr: str = "" +) -> subprocess.CompletedProcess[str]: + return subprocess.CompletedProcess([], code, stdout, stderr) + + +def test_loads_pinned_fixture_with_separate_provenance_and_direction(tmp_path: Path) -> None: + project = _project(tmp_path) + snapshot = load_graphify_snapshot(FIXTURE, project_root=project, side="target") + + assert snapshot.side == "target" + assert snapshot.graphify_version == GRAPHIFY_PACKAGE_VERSION + assert snapshot.graph_schema_version == GRAPHIFY_GRAPH_SCHEMA_VERSION + assert snapshot.graph_sha256 == hashlib.sha256(FIXTURE.read_bytes()).hexdigest() + assert snapshot.built_at_commit == "1" * 40 + assert snapshot.nodes[1].span == GraphifySourceSpan(Path("app.py"), 2, 3) + assert snapshot.nodes[1].extractor_strength == "EXTRACTED" + assert snapshot.edges[1].source_id == "handler" + assert snapshot.edges[1].target_id == "helper" + assert snapshot.edges[1].traversable is True + assert snapshot.edges[2].extractor_strength == "INFERRED" + assert snapshot.edges[2].traversable is True + assert snapshot.edges[3].extractor_strength == "AMBIGUOUS" + assert snapshot.edges[3].traversable is False + + +def test_snapshot_hash_is_checked_against_the_same_byte_snapshot(tmp_path: Path) -> None: + project = _project(tmp_path) + expected = hashlib.sha256(FIXTURE.read_bytes()).hexdigest() + + snapshot = load_graphify_snapshot( + FIXTURE, + project_root=project, + side="baseline", + expected_sha256=expected, + ) + assert snapshot.graph_sha256 == expected + + with pytest.raises(GraphifyAdapterError, match="hash mismatch"): + load_graphify_snapshot( + FIXTURE, + project_root=project, + side="baseline", + expected_sha256="0" * 64, + ) + with pytest.raises(GraphifyAdapterError, match="unsupported Graphify version"): + load_graphify_snapshot( + FIXTURE, + project_root=project, + side="baseline", + graphify_version="0.9.29", + ) + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + (lambda value: value.update({"schema_version": 2}), "top-level schema"), + (lambda value: value["hyperedges"].append({"semantic": True}), "hyperedges"), + (lambda value: value["nodes"].append(dict(value["nodes"][0])), "duplicate"), + (lambda value: value["links"][0].update({"target": "missing"}), "unknown node"), + (lambda value: value["nodes"][0].update({"file_type": "document"}), "code-only"), + (lambda value: value["nodes"][0].update({"source_file": "missing.py"}), "project file"), + (lambda value: value["nodes"][1].update({"source_location": "L9-L2"}), "reversed"), + (lambda value: value["links"][0].update({"confidence_score": 2.0}), "between 0 and 1"), + (lambda value: value.update({"built_at_commit": "ABC"}), "full Git OID"), + (lambda value: value["links"][0].update({"unexpected": "drift"}), "unsupported fields"), + ], + ids=[ + "schema-drift", + "semantic-hyperedge", + "duplicate-node", + "dangling-edge", + "non-code-node", + "missing-source", + "reversed-span", + "invalid-confidence-score", + "invalid-commit", + "edge-schema-drift", + ], +) +def test_strict_schema_rejects_unsupported_or_ambiguous_graphs( + tmp_path: Path, + mutation: Any, + message: str, +) -> None: + project = _project(tmp_path) + payload = json.loads(FIXTURE.read_text(encoding="utf-8")) + mutation(payload) + graph = tmp_path / "graph.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match=message): + load_graphify_snapshot(graph, project_root=project, side="target") + + +def test_strict_json_rejects_duplicate_members_and_non_finite_numbers(tmp_path: Path) -> None: + project = _project(tmp_path) + duplicate = tmp_path / "duplicate.json" + duplicate.write_text('{"directed":false,"directed":true}', encoding="utf-8") + with pytest.raises(GraphifyAdapterError, match="duplicate"): + load_graphify_snapshot(duplicate, project_root=project, side="target") + + non_finite = tmp_path / "non-finite.json" + non_finite.write_text( + FIXTURE.read_text(encoding="utf-8").replace('"weight": 0.5', '"weight": NaN'), + encoding="utf-8", + ) + with pytest.raises(GraphifyAdapterError, match="non-finite"): + load_graphify_snapshot(non_finite, project_root=project, side="target") + + +def test_source_paths_must_remain_inside_the_analyzed_project(tmp_path: Path) -> None: + project = _project(tmp_path) + payload = json.loads(FIXTURE.read_text(encoding="utf-8")) + payload["nodes"][0]["source_file"] = "../outside.py" + graph = tmp_path / "escape.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match="escapes the project root"): + load_graphify_snapshot(graph, project_root=project, side="target") + + +def test_runner_invokes_only_pinned_code_only_pipeline_and_writes_receipt( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + executable = tmp_path / "graphify" + executable.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8") + executable.chmod(0o700) + output_root = tmp_path / "snapshots" + calls: list[tuple[list[str], dict[str, object]]] = [] + + def fake_run(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: + calls.append((args, kwargs)) + if args[-1] == "--version": + return _completed(f"graphify {GRAPHIFY_PACKAGE_VERSION}\n") + graph = output_root / "target" / "graphify-out" / "graph.json" + graph.parent.mkdir(parents=True) + shutil.copyfile(FIXTURE, graph) + return _completed("code-only graph written\n") + + with ( + patch.dict( + os.environ, + { + "GEMINI_API_KEY": "must-not-pass", + "GOOGLE_API_KEY": "must-not-pass", + "HTTP_PROXY": "must-not-pass", + }, + ), + patch("subprocess.run", side_effect=fake_run), + ): + snapshot = GraphifyRunner(project, executable, timeout=12).extract_snapshot( + "target", output_root + ) + + assert snapshot.graph_sha256 == hashlib.sha256(FIXTURE.read_bytes()).hexdigest() + assert len(calls) == 2 + extract_args, extract_kwargs = calls[1] + assert extract_args == [ + str(executable.resolve()), + "extract", + str(project.resolve()), + "--code-only", + "--no-cluster", + "--force", + "--out", + str((output_root / "target").resolve()), + ] + assert extract_kwargs.get("shell", False) is False + environment = extract_kwargs["env"] + assert isinstance(environment, dict) + assert "GEMINI_API_KEY" not in environment + assert "GOOGLE_API_KEY" not in environment + assert "HTTP_PROXY" not in environment + assert "NO_PROXY" not in environment + assert all( + forbidden not in extract_args + for forbidden in ("--backend", "--mcp", "--wiki", "--watch", "--global") + ) + receipt_path = output_root / "target" / "snapshot-receipt.json" + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) + assert receipt == { + "edge_count": 4, + "graph_schema_version": GRAPHIFY_GRAPH_SCHEMA_VERSION, + "graph_sha256": snapshot.graph_sha256, + "graphify_version": GRAPHIFY_PACKAGE_VERSION, + "node_count": 3, + "side": "target", + } + + +def test_runner_is_no_clobber_and_rejects_project_local_output_before_execution( + tmp_path: Path, +) -> None: + project = _project(tmp_path) + executable = tmp_path / "graphify" + executable.write_text("tool", encoding="utf-8") + executable.chmod(0o700) + runner = GraphifyRunner(project, executable) + + with ( + patch("subprocess.run") as run, + pytest.raises(GraphifyAdapterError, match="outside"), + ): + runner.extract_snapshot("target", project / "snapshots") + run.assert_not_called() + + output_root = tmp_path / "snapshots" + (output_root / "baseline").mkdir(parents=True) + with ( + patch("subprocess.run") as run, + pytest.raises(GraphifyAdapterError, match="already exists"), + ): + runner.extract_snapshot("baseline", output_root) + run.assert_not_called() + + +def test_runner_rejects_unpinned_version_and_command_failure(tmp_path: Path) -> None: + project = _project(tmp_path) + executable = tmp_path / "graphify" + executable.write_text("tool", encoding="utf-8") + executable.chmod(0o700) + runner = GraphifyRunner(project, executable) + + with ( + patch("subprocess.run", return_value=_completed("graphify 9.9.9\n")), + pytest.raises(GraphifyAdapterError, match="unsupported Graphify version"), + ): + runner.extract_snapshot("target", tmp_path / "version-output") + + def failed(args: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + if args[-1] == "--version": + return _completed(f"graphify {GRAPHIFY_PACKAGE_VERSION}\n") + return _completed(code=7, stderr="structural extraction failed") + + with ( + patch("subprocess.run", side_effect=failed), + pytest.raises(GraphifyAdapterError, match="structural extraction failed"), + ): + runner.extract_snapshot("baseline", tmp_path / "failure-output") From db567ac5812dfa2f24fd4d69b744e309ba59ece0 Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 01:21:22 +0300 Subject: [PATCH 2/4] fix: harden offline Graphify snapshot import --- README.md | 13 +- docs/graphify-poc.md | 160 +++-- .../analyzer/graphify_adapter.py | 612 ++++++++++-------- tests/fixtures/graphify_0_9_30_graph.json | 2 +- tests/unit/test_graphify_adapter.py | 325 ++++++---- 5 files changed, 646 insertions(+), 466 deletions(-) diff --git a/README.md b/README.md index fc480ac..4d81ca8 100644 --- a/README.md +++ b/README.md @@ -177,12 +177,13 @@ escape, mutation, control flow, or dynamic registration as conditional evidence; no bootstrap/helper names are guessed. Dynamic behavior remains conservatively unresolved. -The optional Graphify experiment is not a configured analyzer backend. A private -POC adapter can explicitly run the separately installed, pinned -`graphifyy==0.9.30` command in code-only mode and validate immutable -baseline/target graph snapshots. It never runs by default, never uses semantic -LLM/server features, and cannot promote Graphify communities or similarity into -blast-radius evidence. See [the Graphify POC boundary](docs/graphify-poc.md). +The optional Graphify experiment is not a configured analyzer backend. Its private, +offline-only POC adapter validates operator-supplied `graph.json` snapshots against +pinned expected `graphifyy==0.9.30` metadata; it does not install or execute +Graphify. Isolated no-network execution is deferred to the trusted #101 sandbox +gate. It never runs by default, makes no LLM/server/network request, and cannot +promote Graphify communities or similarity into blast-radius evidence. See +[the Graphify POC boundary](docs/graphify-poc.md). ### `list` - List Endpoints diff --git a/docs/graphify-poc.md b/docs/graphify-poc.md index f713940..348d4b6 100644 --- a/docs/graphify-poc.md +++ b/docs/graphify-poc.md @@ -3,87 +3,117 @@ Issue #110 evaluates Graphify as an optional generic code-graph provider. This foundation is deliberately **not** connected to the default CLI or analyzer. Mypy remains the only default semantic backend. Importing this project never -imports or installs Graphify, and missing Graphify tooling never causes fallback -or changes ordinary analysis. - -## Pinned, explicit execution - -The only attested tool is the separately installed PyPI package -`graphifyy==0.9.30`, whose console command reports exactly `graphify 0.9.30`. -Installation is an operator-controlled prerequisite outside the analyzed -checkout. `GraphifyRunner` takes the absolute executable path explicitly and -invokes this fixed argv shape without a shell: - -```text -graphify extract \ - --code-only --no-cluster --force --out -``` - -The runner: - -- rejects every other tool version; -- strips API keys and proxy variables from the child environment; -- passes `--code-only`, so document/media semantic extraction is disabled; -- never passes backend, model, MCP/server, wiki, watch, hook, global-graph, or - database flags; -- requires a fresh `baseline/` or `target/` directory outside the analyzed - project and never overwrites it; -- uses a private HOME/config directory inside that snapshot directory; -- validates `graphify-out/graph.json` before publishing a deterministic - `snapshot-receipt.json`. - -This is process isolation, not a security sandbox. The Graphify executable still -reads source on the host. Do not use the POC on untrusted source until it is -placed behind the hardened runtime boundary. +imports, installs, or executes Graphify, and missing Graphify tooling never +causes fallback or changes ordinary analysis. -## Supported `graph.json` contract +## Offline-only foundation -The adapter version is `1`, bound to Graphify 0.9.30's NetworkX node-link JSON: +The adapter accepts only an operator-supplied `graph.json`. It performs bounded, +execution-free validation and can exclusively create a deterministic import +receipt. There is no extraction launcher or ordinary host subprocess path. +Graphify execution, including enforced no-network operation and a read-only +source mount, is deferred to the trusted hardened sandbox gate tracked by +[#101](https://github.com/shaggitza/test_ast_fastapi/issues/101). Do not run +Graphify against a checkout through an ad hoc host subprocess. -| Level | Required contract | +The expected producer metadata is pinned as: + +| Item | Expected value | |---|---| -| document | exactly `directed`, `multigraph`, `graph`, `nodes`, `links`, `hyperedges`, and optional `built_at_commit` | -| node | unique string `id`, string `label`, `file_type: code`, project-confined `source_file`, optional one-based `source_location` | -| link | existing `source` and `target` IDs, string `relation`, `confidence` in `EXTRACTED`, `INFERRED`, `AMBIGUOUS`, optional source span | -| semantic data | `hyperedges` must be empty | +| PyPI distribution | `graphifyy` | +| distribution version | `0.9.30` | +| console command | `graphify` | +| exact version output | `graphify 0.9.30` | +| adapter schema | `1` | + +These are **expected provenance metadata**, not proof that an imported file was +created by that executable. The future sandbox gate must attest the installed +artifact and invocation before passing its output to this importer. The adapter +does not install the package or call the command to obtain self-reported +metadata. + +## Supported `graph.json` contract -Duplicate JSON members, non-finite numbers, invalid UTF-8, schema drift, dangling -edges, duplicate node IDs, non-code nodes, paths outside the project, malformed -or reversed ranges, oversized graphs, and unexpected fields fail closed. -`source_location` accepts `L12`, `12`, or an inclusive range such as `L12-L18`. +The offline adapter is bound to the frozen, directed NetworkX node-link shape: -The adapter retains Graphify extraction strength separately from detector -confidence. It marks only source-backed `calls`, `imports`, `imports_from`, -`inherits`, `references`, and `re_exports` links as eligible for a future -traversal. `contains`, communities, similarity, natural-language relations, and -unknown relation types cannot become blast-radius evidence. +| Level | Required contract | +|---|---| +| document | exactly required `directed`, `multigraph`, `graph`, `nodes`, `links`, `hyperedges`, plus optional `built_at_commit` | +| graph mode | `directed: true`, `multigraph: true`, empty `graph`, empty `hyperedges` | +| node | unique string `id`, string `label`, `file_type: code`, non-empty project-confined regular `source_file`, optional bounded `source_location` | +| link | existing `source` and `target` IDs, an explicitly oriented relation, `confidence` in `EXTRACTED`, `INFERRED`, `AMBIGUOUS`, optional bounded source occurrence | -## Immutable snapshot receipt +The attested relation orientation is always the JSON `source` ID to the JSON +`target` ID: -The adapter reads one bounded byte snapshot, hashes those exact bytes with -SHA-256, then validates them. The exclusive receipt records: +| Relation | Orientation | +|---|---| +| `calls` | caller to callee | +| `imports`, `imports_from` | importer to imported symbol/module | +| `inherits` | subclass to base | +| `references` | referencer to referenced symbol | +| `re_exports` | exporter to exported symbol | +| `contains` | container to contained node | +| `related_to` | symmetric; never traversal evidence | + +Relations without a pinned orientation fail closed. Only source-backed `calls`, +`imports`, `imports_from`, `inherits`, `references`, and `re_exports` may be +eligible for future traversal. `contains`, similarity, community data, and +natural-language relations cannot become blast-radius evidence. + +All allowlisted fields are validated. Former unchecked fields such as nested +`metadata`, origin/target hints, scopes, package, and namespace are rejected. +Duplicate JSON members, non-finite numbers, invalid UTF-8, schema drift, +dangling edges, duplicate node IDs, non-code nodes, paths outside the project, +malformed/reversed/out-of-file ranges, oversized files, non-regular files, and +mutation during import fail closed. `source_location` accepts `L12`, `12`, or an +inclusive range such as `L12-L18`. + +## Bounded exact-byte provenance and receipt + +`graph.json` is opened once, required to be a regular file, and read through +that descriptor with a `MAX_GRAPH_BYTES + 1` bound. Descriptor/path identity, +size, and timestamp checks detect replacement or mutation during the read. The +SHA-256 is computed over those exact bytes. + +Each referenced source file is similarly confined, opened as a regular file, +and read at most once with a `MAX_SOURCE_BYTES + 1` bound. Node and edge line +occurrences must fit those exact bytes. Source SHA-256 values are retained on +nodes and spans, and all source snapshots are checked again before import +returns. + +`import_graphify_snapshot()` exclusively creates a receipt containing: - side (`baseline` or `target`); -- Graphify package version and adapter schema version; -- exact graph SHA-256; -- node and edge counts. +- graph SHA-256 and adapter schema version; +- pinned **expected** Graphify package/version/command/version-output metadata; +- attested directed/multigraph values; +- node and edge counts; +- explicit `offline-only` import mode. -Baseline and target always occupy distinct fresh directories. A caller can pass -an expected SHA-256 when reopening a snapshot; a mismatch is an explicit error. +A caller may require an expected lowercase SHA-256 when importing. Graph, +project, source, and receipt path failures are normalized to +`GraphifyAdapterError`. ## Current decision and remaining gates -**Decision: BUILD the isolated adapter foundation; do not ADOPT the backend.** +**Decision: BUILD the offline adapter foundation; do not ADOPT or invoke the +backend. Keep #110 open.** -Before an ADOPT or HYBRID decision, a later tranche must still: +Before an ADOPT or HYBRID decision, later work must still: -1. verify Graphify 0.9.30 on controlled Python fixtures for aliases, methods, - inheritance, imports/re-exports, deleted source, and cross-file calls; -2. overlay secure FastAPI handler/DI identity without modifying `graph.json`; -3. calibrate EXTRACTED/INFERRED/AMBIGUOUS edges against HIGH/MEDIUM/LOW policy; -4. run target and baseline corpus comparisons and report candidate gain, false +1. use the trusted #101 sandbox gate to enforce no network, a read-only source + mount, resource bounds, and pinned executable/package identity; +2. verify real Graphify 0.9.30 artifacts on controlled Python fixtures for + direction, ranges, aliases, methods, inheritance, imports/re-exports, + deleted source, and cross-file calls; +3. overlay secure FastAPI handler/DI identity without modifying `graph.json`; +4. calibrate EXTRACTED/INFERRED/AMBIGUOUS edges against HIGH/MEDIUM/LOW policy; +5. run target and baseline corpus comparisons and report candidate gain, false positives, failures/abstentions, graph size, latency, and peak RSS; -5. prove no regression relative to mypy and record ADOPT, HYBRID, or STOP. +6. prove no regression relative to mypy and record ADOPT, HYBRID, or STOP. No community, proximity, semantic label, or natural-language query result may -satisfy these gates. +satisfy these gates. This foundation makes no LLM, server, or network request; +its stronger no-network execution guarantee remains deferred because it does +not execute Graphify at all. diff --git a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py index 7c68b93..bdb649a 100644 --- a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py +++ b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py @@ -1,10 +1,9 @@ -"""Strict, opt-in adapter for pinned Graphify code-only snapshots. +"""Strict, offline-only adapter for pinned Graphify code-graph snapshots. -This module is deliberately not wired into the default analyzer. It provides a -bounded POC boundary that can invoke a separately installed, pinned Graphify -binary in code-only mode and adapt its immutable ``graph.json`` bytes into a -small source/provenance IR. It never installs tools, starts servers, queries a -semantic backend, or treats Graphify confidence as detector confidence. +This module is deliberately not wired into the default analyzer. It never +installs or executes Graphify. It only imports an operator-supplied ``graph.json`` +whose expected producer metadata and schema are pinned below. Execution remains +deferred to the trusted sandbox gate tracked by issue #101. """ from __future__ import annotations @@ -14,87 +13,91 @@ import math import os import re -import subprocess +import stat from dataclasses import dataclass from pathlib import Path from typing import Literal, cast from fastapi_endpoint_detector.strict_data import DuplicateKeyError, load_json_unique +GRAPHIFY_PACKAGE_NAME = "graphifyy" GRAPHIFY_PACKAGE_VERSION = "0.9.30" +GRAPHIFY_COMMAND_NAME = "graphify" +GRAPHIFY_EXPECTED_VERSION_OUTPUT = "graphify 0.9.30" GRAPHIFY_GRAPH_SCHEMA_VERSION = 1 -GRAPHIFY_OUTPUT_DIRECTORY = "graphify-out" +GRAPHIFY_EXPECTED_DIRECTED = True +GRAPHIFY_EXPECTED_MULTIGRAPH = True MAX_GRAPH_BYTES = 64 * 1024 * 1024 +MAX_SOURCE_BYTES = 16 * 1024 * 1024 MAX_GRAPH_NODES = 250_000 MAX_GRAPH_EDGES = 1_000_000 MAX_TEXT_LENGTH = 16_384 GraphSide = Literal["baseline", "target"] GraphifyStrength = Literal["EXTRACTED", "INFERRED", "AMBIGUOUS"] - +GraphifyRelationOrientation = Literal[ + "caller-to-callee", + "importer-to-imported", + "subclass-to-base", + "referencer-to-referenced", + "exporter-to-exported", + "container-to-contained", + "symmetric", +] + +_RELATION_ORIENTATIONS: dict[str, GraphifyRelationOrientation] = { + "calls": "caller-to-callee", + "imports": "importer-to-imported", + "imports_from": "importer-to-imported", + "inherits": "subclass-to-base", + "references": "referencer-to-referenced", + "re_exports": "exporter-to-exported", + "contains": "container-to-contained", + "related_to": "symmetric", +} _TRAVERSABLE_RELATIONS = frozenset( {"calls", "imports", "imports_from", "inherits", "references", "re_exports"} ) _TOP_LEVEL_KEYS = frozenset( {"directed", "multigraph", "graph", "nodes", "links", "hyperedges", "built_at_commit"} ) -_NODE_KEYS = frozenset( - { - "id", - "label", - "file_type", - "source_file", - "source_location", - "confidence", - "confidence_score", - "community", - "community_name", - "norm_label", - "type", - "kind", - "metadata", - "origin_file", - "scope_id", - "scope_kind", - "target_file", - "target_fqn", - "package", - "namespace", - } -) -_EDGE_KEYS = frozenset( - { - "source", - "target", - "relation", - "confidence", - "confidence_score", - "source_file", - "source_location", - "weight", - "context", - "metadata", - "key", - "target_file", - "target_fqn", - "origin_file", - } -) +_NODE_REQUIRED_KEYS = frozenset({"id", "label", "file_type", "source_file"}) +_NODE_KEYS = _NODE_REQUIRED_KEYS | { + "source_location", + "confidence", + "confidence_score", + "community", + "community_name", + "norm_label", + "type", + "kind", +} +_EDGE_REQUIRED_KEYS = frozenset({"source", "target", "relation", "confidence"}) +_EDGE_KEYS = _EDGE_REQUIRED_KEYS | { + "confidence_score", + "source_file", + "source_location", + "weight", + "context", + "key", +} _LOCATION = re.compile(r"^(?:L)?(?P[1-9][0-9]*)(?:-(?:L)?(?P[1-9][0-9]*))?$") _GIT_OID = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$") +_SHA256 = re.compile(r"^[0-9a-f]{64}$") class GraphifyAdapterError(RuntimeError): - """Raised when the explicit Graphify POC cannot produce trustworthy evidence.""" + """Raised when an offline Graphify snapshot cannot be trusted.""" @dataclass(frozen=True) class GraphifySourceSpan: - """One project-relative, one-based inclusive source span.""" + """One source occurrence tied to the exact bounded source bytes checked.""" file_path: Path start_line: int end_line: int + source_sha256: str @dataclass(frozen=True) @@ -103,34 +106,42 @@ class GraphifyNode: node_id: str label: str + source_file: Path + source_sha256: str span: GraphifySourceSpan | None extractor_strength: GraphifyStrength | None @dataclass(frozen=True) class GraphifyEdge: - """A directed Graphify edge with extractor provenance kept separate.""" + """A relation with its pinned source-to-target orientation.""" source_id: str target_id: str relation: str + orientation: GraphifyRelationOrientation extractor_strength: GraphifyStrength span: GraphifySourceSpan | None @property def traversable(self) -> bool: - """Whether the relation is eligible for later evidence-bearing traversal.""" + """Whether this source-backed relation may support future traversal.""" return self.relation in _TRAVERSABLE_RELATIONS and self.span is not None @dataclass(frozen=True) class GraphifySnapshot: - """One immutable byte snapshot adapted from a pinned Graphify graph.""" + """One immutable graph byte snapshot adapted without executing its producer.""" side: GraphSide graph_sha256: str graph_schema_version: int - graphify_version: str + expected_graphify_package: str + expected_graphify_version: str + expected_graphify_command: str + expected_version_output: str + directed: bool + multigraph: bool built_at_commit: str | None nodes: tuple[GraphifyNode, ...] edges: tuple[GraphifyEdge, ...] @@ -138,12 +149,17 @@ class GraphifySnapshot: @dataclass(frozen=True) class GraphifySnapshotReceipt: - """Durable receipt written beside a successfully validated graph snapshot.""" + """Durable receipt for a successfully validated offline import.""" side: GraphSide graph_sha256: str graph_schema_version: int - graphify_version: str + expected_graphify_package: str + expected_graphify_version: str + expected_graphify_command: str + expected_version_output: str + directed: bool + multigraph: bool node_count: int edge_count: int @@ -152,10 +168,16 @@ def as_json(self) -> str: return ( json.dumps( { + "directed": self.directed, "edge_count": self.edge_count, + "expected_graphify_command": self.expected_graphify_command, + "expected_graphify_package": self.expected_graphify_package, + "expected_graphify_version": self.expected_graphify_version, + "expected_version_output": self.expected_version_output, "graph_schema_version": self.graph_schema_version, "graph_sha256": self.graph_sha256, - "graphify_version": self.graphify_version, + "import_mode": "offline-only", + "multigraph": self.multigraph, "node_count": self.node_count, "side": self.side, }, @@ -167,6 +189,15 @@ def as_json(self) -> str: ) +@dataclass(frozen=True) +class _FileSnapshot: + path: Path + relative_path: Path | None + sha256: str + line_count: int + signature: tuple[int, int, int, int, int] + + def _bounded_string(value: object, location: str, *, allow_empty: bool = False) -> str: if not isinstance(value, str) or (not allow_empty and not value): raise GraphifyAdapterError(f"{location} must be a non-empty string") @@ -184,6 +215,43 @@ def _finite_number(value: object, location: str) -> float: return result +def _signature(value: os.stat_result) -> tuple[int, int, int, int, int]: + return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns, value.st_ctime_ns) + + +def _read_regular_file(path: Path, limit: int, description: str) -> tuple[bytes, _FileSnapshot]: + """Read at most limit+1 bytes through one descriptor and detect path mutation.""" + try: + with path.open("rb") as handle: + before = os.fstat(handle.fileno()) + if not stat.S_ISREG(before.st_mode): + raise GraphifyAdapterError(f"{description} is not a regular file: {path}") + raw = handle.read(limit + 1) + after = os.fstat(handle.fileno()) + current = path.stat() + except GraphifyAdapterError: + raise + except (OSError, RuntimeError) as error: + raise GraphifyAdapterError(f"cannot read {description} {path}: {error}") from error + + before_signature = _signature(before) + if _signature(after) != before_signature or _signature(current) != before_signature: + raise GraphifyAdapterError(f"{description} changed while it was being read") + if len(raw) > limit: + raise GraphifyAdapterError(f"{description} exceeds {limit} bytes") + if len(raw) != before.st_size: + raise GraphifyAdapterError(f"{description} changed while it was being read") + line_count = raw.count(b"\n") + (1 if raw and not raw.endswith(b"\n") else 0) + snapshot = _FileSnapshot( + path=path, + relative_path=None, + sha256=hashlib.sha256(raw).hexdigest(), + line_count=line_count, + signature=before_signature, + ) + return raw, snapshot + + def _validate_json_numbers(value: object) -> None: pending = [value] while pending: @@ -205,47 +273,76 @@ def _strict_json(raw: bytes, source: Path) -> dict[str, object]: _validate_json_numbers(value) if not isinstance(value, dict): raise GraphifyAdapterError("graph.json must contain an object") - if set(value) != _TOP_LEVEL_KEYS and set(value) != _TOP_LEVEL_KEYS - {"built_at_commit"}: + required = _TOP_LEVEL_KEYS - {"built_at_commit"} + if not required.issubset(value) or not set(value).issubset(_TOP_LEVEL_KEYS): extra = sorted(set(value) - _TOP_LEVEL_KEYS) - missing = sorted((_TOP_LEVEL_KEYS - {"built_at_commit"}) - set(value)) + missing = sorted(required - set(value)) raise GraphifyAdapterError( f"unsupported graph.json top-level schema; extra={extra}, missing={missing}" ) return cast("dict[str, object]", value) -def _relative_source(project_root: Path, value: object, location: str) -> Path | None: - source = _bounded_string(value, location, allow_empty=True) - if not source: - return None - supplied = Path(source) - try: - absolute = ( - supplied.resolve(strict=False) - if supplied.is_absolute() - else (project_root / supplied).resolve(strict=False) +class _SourceRegistry: + def __init__(self, project_root: Path): + self.project_root = project_root + self._snapshots: dict[Path, _FileSnapshot] = {} + + def read(self, value: object, location: str) -> _FileSnapshot: + source = _bounded_string(value, location) + supplied = Path(source) + try: + absolute = ( + supplied.resolve(strict=True) + if supplied.is_absolute() + else (self.project_root / supplied).resolve(strict=True) + ) + relative = absolute.relative_to(self.project_root) + except (OSError, RuntimeError, ValueError) as error: + raise GraphifyAdapterError( + f"{location} does not identify a confined project file: {source!r}" + ) from error + if not relative.parts or ".." in relative.parts: + raise GraphifyAdapterError(f"{location} is not project relative: {source!r}") + cached = self._snapshots.get(relative) + if cached is not None: + return cached + _raw, snapshot = _read_regular_file(absolute, MAX_SOURCE_BYTES, "source file") + snapshot = _FileSnapshot( + path=absolute, + relative_path=relative, + sha256=snapshot.sha256, + line_count=snapshot.line_count, + signature=snapshot.signature, ) - relative = absolute.relative_to(project_root) - except ValueError as error: - raise GraphifyAdapterError(f"{location} escapes the project root: {source!r}") from error - if not relative.parts or ".." in relative.parts: - raise GraphifyAdapterError(f"{location} is not project relative: {source!r}") - if not absolute.is_file(): - raise GraphifyAdapterError(f"{location} does not identify a project file: {source!r}") - return relative + self._snapshots[relative] = snapshot + return snapshot + + def verify_unchanged(self) -> None: + for snapshot in self._snapshots.values(): + try: + current_path = snapshot.path.resolve(strict=True) + current = snapshot.path.stat() + current_path.relative_to(self.project_root) + except (OSError, RuntimeError, ValueError) as error: + raise GraphifyAdapterError( + f"source file changed during import: {snapshot.relative_path}" + ) from error + if current_path != snapshot.path or _signature(current) != snapshot.signature: + raise GraphifyAdapterError( + f"source file changed during import: {snapshot.relative_path}" + ) def _source_span( - project_root: Path, + registry: _SourceRegistry, source_value: object, location_value: object, location: str, -) -> GraphifySourceSpan | None: - file_path = _relative_source(project_root, source_value, f"{location}.source_file") - if location_value in {None, ""}: - return None - if file_path is None: - raise GraphifyAdapterError(f"{location} has a source location without a source file") +) -> tuple[_FileSnapshot, GraphifySourceSpan | None]: + source = registry.read(source_value, f"{location}.source_file") + if location_value is None: + return source, None raw_location = _bounded_string(location_value, f"{location}.source_location") match = _LOCATION.fullmatch(raw_location) if match is None: @@ -254,7 +351,31 @@ def _source_span( end_line = int(match.group("end") or start_line) if end_line < start_line: raise GraphifyAdapterError(f"{location}.source_location has a reversed range") - return GraphifySourceSpan(file_path, start_line, end_line) + if end_line > source.line_count: + raise GraphifyAdapterError( + f"{location}.source_location exceeds the exact source bytes " + f"({source.line_count} lines)" + ) + assert source.relative_path is not None + return source, GraphifySourceSpan( + source.relative_path, + start_line, + end_line, + source.sha256, + ) + + +def _optional_edge_span( + registry: _SourceRegistry, edge: dict[str, object], location: str +) -> GraphifySourceSpan | None: + has_source = "source_file" in edge + location_value = edge.get("source_location") + if not has_source: + if location_value is not None: + raise GraphifyAdapterError(f"{location} has a source location without a source file") + return None + _source, span = _source_span(registry, edge["source_file"], location_value, location) + return span def _strength(value: object, location: str, *, optional: bool = False) -> GraphifyStrength | None: @@ -265,7 +386,7 @@ def _strength(value: object, location: str, *, optional: bool = False) -> Graphi return cast("GraphifyStrength", value) -def _validate_common_optional_fields(item: dict[str, object], location: str) -> None: +def _validate_optional_fields(item: dict[str, object], location: str) -> None: if "confidence_score" in item: score = _finite_number(item["confidence_score"], f"{location}.confidence_score") if not 0.0 <= score <= 1.0: @@ -275,47 +396,48 @@ def _validate_common_optional_fields(item: dict[str, object], location: str) -> if weight < 0.0: raise GraphifyAdapterError(f"{location}.weight must be non-negative") for field in ("context", "community_name", "norm_label", "type", "kind"): - if field in item and item[field] is not None: + if field in item: _bounded_string(item[field], f"{location}.{field}", allow_empty=True) - community = item.get("community") - if community is not None and (type(community) is not int or community < 0): - raise GraphifyAdapterError(f"{location}.community must be a non-negative integer or null") - key = item.get("key") - if key is not None and (isinstance(key, bool) or not isinstance(key, (int, str))): - raise GraphifyAdapterError(f"{location}.key must be an integer or string") - if "metadata" in item and not isinstance(item["metadata"], dict): - raise GraphifyAdapterError(f"{location}.metadata must be an object") + if "community" in item: + community = item["community"] + if community is not None and (type(community) is not int or community < 0): + raise GraphifyAdapterError( + f"{location}.community must be a non-negative integer or null" + ) + if "key" in item: + key = item["key"] + if isinstance(key, bool) or not isinstance(key, (int, str)): + raise GraphifyAdapterError(f"{location}.key must be an integer or string") + if isinstance(key, str): + _bounded_string(key, f"{location}.key", allow_empty=True) def _read_graph_payload( graph_path: Path, expected_sha256: str | None ) -> tuple[str, dict[str, object]]: - try: - size = graph_path.stat().st_size - if size > MAX_GRAPH_BYTES: - raise GraphifyAdapterError(f"graph.json exceeds {MAX_GRAPH_BYTES} bytes") - raw = graph_path.read_bytes() - except OSError as error: - raise GraphifyAdapterError( - f"cannot read Graphify snapshot {graph_path}: {error}" - ) from error - if len(raw) != size: - raise GraphifyAdapterError("graph.json changed while it was being read") - graph_sha256 = hashlib.sha256(raw).hexdigest() - if expected_sha256 is not None and graph_sha256 != expected_sha256: + if expected_sha256 is not None and _SHA256.fullmatch(expected_sha256) is None: + raise GraphifyAdapterError("expected_sha256 must be a lowercase SHA-256 digest") + raw, snapshot = _read_regular_file(graph_path, MAX_GRAPH_BYTES, "Graphify snapshot") + if expected_sha256 is not None and snapshot.sha256 != expected_sha256: raise GraphifyAdapterError( - f"Graphify snapshot hash mismatch: expected {expected_sha256}, got {graph_sha256}" + f"Graphify snapshot hash mismatch: expected {expected_sha256}, got {snapshot.sha256}" ) - return graph_sha256, _strict_json(raw, graph_path) + return snapshot.sha256, _strict_json(raw, graph_path) def _payload_collections( payload: dict[str, object], ) -> tuple[list[object], list[object], str | None]: - if type(payload["directed"]) is not bool or type(payload["multigraph"]) is not bool: - raise GraphifyAdapterError("graph.json directed and multigraph fields must be booleans") - if not isinstance(payload["graph"], dict): - raise GraphifyAdapterError("graph.json graph field must be an object") + if payload["directed"] is not GRAPHIFY_EXPECTED_DIRECTED: + raise GraphifyAdapterError( + f"graph.json directed must be attested value {GRAPHIFY_EXPECTED_DIRECTED}" + ) + if payload["multigraph"] is not GRAPHIFY_EXPECTED_MULTIGRAPH: + raise GraphifyAdapterError( + f"graph.json multigraph must be attested value {GRAPHIFY_EXPECTED_MULTIGRAPH}" + ) + if payload["graph"] != {}: + raise GraphifyAdapterError("graph.json graph field must be an empty object") hyperedges = payload["hyperedges"] if not isinstance(hyperedges, list) or hyperedges: raise GraphifyAdapterError("code-only Graphify snapshots must have no semantic hyperedges") @@ -334,7 +456,7 @@ def _payload_collections( def _adapt_nodes( - raw_nodes: list[object], project_root: Path + raw_nodes: list[object], registry: _SourceRegistry ) -> tuple[tuple[GraphifyNode, ...], set[str]]: nodes: list[GraphifyNode] = [] node_ids: set[str] = set() @@ -344,29 +466,43 @@ def _adapt_nodes( raise GraphifyAdapterError(f"{location} must be an object") node = cast("dict[str, object]", raw_node) extra = set(node) - _NODE_KEYS - if extra: - raise GraphifyAdapterError(f"{location} has unsupported fields: {sorted(extra)}") - node_id = _bounded_string(node.get("id"), f"{location}.id") + missing = _NODE_REQUIRED_KEYS - set(node) + if extra or missing: + raise GraphifyAdapterError( + f"{location} has unsupported schema; " + f"extra={sorted(extra)}, missing={sorted(missing)}" + ) + node_id = _bounded_string(node["id"], f"{location}.id") if node_id in node_ids: raise GraphifyAdapterError(f"duplicate Graphify node id: {node_id}") node_ids.add(node_id) - if node.get("file_type") != "code": + if node["file_type"] != "code": raise GraphifyAdapterError(f"{location} is not a code-only node") - label = _bounded_string(node.get("label"), f"{location}.label") - span = _source_span( - project_root, - node.get("source_file", ""), + label = _bounded_string(node["label"], f"{location}.label") + source, span = _source_span( + registry, + node["source_file"], node.get("source_location"), location, ) strength = _strength(node.get("confidence"), f"{location}.confidence", optional=True) - _validate_common_optional_fields(node, location) - nodes.append(GraphifyNode(node_id, label, span, strength)) + _validate_optional_fields(node, location) + assert source.relative_path is not None + nodes.append( + GraphifyNode( + node_id, + label, + source.relative_path, + source.sha256, + span, + strength, + ) + ) return tuple(nodes), node_ids def _adapt_edges( - raw_edges: list[object], project_root: Path, node_ids: set[str] + raw_edges: list[object], registry: _SourceRegistry, node_ids: set[str] ) -> tuple[GraphifyEdge, ...]: edges: list[GraphifyEdge] = [] for index, raw_edge in enumerate(raw_edges): @@ -375,169 +511,115 @@ def _adapt_edges( raise GraphifyAdapterError(f"{location} must be an object") edge = cast("dict[str, object]", raw_edge) extra = set(edge) - _EDGE_KEYS - if extra: - raise GraphifyAdapterError(f"{location} has unsupported fields: {sorted(extra)}") - source_id = _bounded_string(edge.get("source"), f"{location}.source") - target_id = _bounded_string(edge.get("target"), f"{location}.target") + missing = _EDGE_REQUIRED_KEYS - set(edge) + if extra or missing: + raise GraphifyAdapterError( + f"{location} has unsupported schema; " + f"extra={sorted(extra)}, missing={sorted(missing)}" + ) + source_id = _bounded_string(edge["source"], f"{location}.source") + target_id = _bounded_string(edge["target"], f"{location}.target") if source_id not in node_ids or target_id not in node_ids: raise GraphifyAdapterError(f"{location} references an unknown node") - relation = _bounded_string(edge.get("relation"), f"{location}.relation") - strength = _strength(edge.get("confidence"), f"{location}.confidence") + relation = _bounded_string(edge["relation"], f"{location}.relation") + orientation = _RELATION_ORIENTATIONS.get(relation) + if orientation is None: + raise GraphifyAdapterError( + f"{location}.relation has no attested source-to-target orientation: {relation!r}" + ) + strength = _strength(edge["confidence"], f"{location}.confidence") assert strength is not None - span = _source_span( - project_root, - edge.get("source_file", ""), - edge.get("source_location"), - location, + span = _optional_edge_span(registry, edge, location) + _validate_optional_fields(edge, location) + edges.append( + GraphifyEdge(source_id, target_id, relation, orientation, strength, span) ) - _validate_common_optional_fields(edge, location) - edges.append(GraphifyEdge(source_id, target_id, relation, strength, span)) return tuple(edges) +def _resolve_project_root(project_root: Path) -> Path: + try: + root = project_root.resolve(strict=True) + except (OSError, RuntimeError) as error: + raise GraphifyAdapterError( + f"Graphify project root does not exist: {project_root}" + ) from error + if not root.is_dir(): + raise GraphifyAdapterError(f"Graphify project root is not a directory: {project_root}") + return root + + def load_graphify_snapshot( graph_path: Path, *, project_root: Path, side: GraphSide, - graphify_version: str = GRAPHIFY_PACKAGE_VERSION, expected_sha256: str | None = None, ) -> GraphifySnapshot: - """Read one immutable graph.json byte snapshot and strictly adapt its evidence.""" + """Read and validate one offline graph snapshot; never execute Graphify.""" if side not in {"baseline", "target"}: raise GraphifyAdapterError(f"unsupported snapshot side: {side!r}") - if graphify_version != GRAPHIFY_PACKAGE_VERSION: - raise GraphifyAdapterError( - f"unsupported Graphify version {graphify_version!r}; " - f"expected {GRAPHIFY_PACKAGE_VERSION}" - ) + root = _resolve_project_root(project_root) graph_sha256, payload = _read_graph_payload(graph_path, expected_sha256) raw_nodes, raw_edges, built_at_commit = _payload_collections(payload) - root = project_root.resolve(strict=True) - nodes, node_ids = _adapt_nodes(raw_nodes, root) - edges = _adapt_edges(raw_edges, root, node_ids) + registry = _SourceRegistry(root) + nodes, node_ids = _adapt_nodes(raw_nodes, registry) + edges = _adapt_edges(raw_edges, registry, node_ids) + registry.verify_unchanged() return GraphifySnapshot( side=side, graph_sha256=graph_sha256, graph_schema_version=GRAPHIFY_GRAPH_SCHEMA_VERSION, - graphify_version=graphify_version, + expected_graphify_package=GRAPHIFY_PACKAGE_NAME, + expected_graphify_version=GRAPHIFY_PACKAGE_VERSION, + expected_graphify_command=GRAPHIFY_COMMAND_NAME, + expected_version_output=GRAPHIFY_EXPECTED_VERSION_OUTPUT, + directed=GRAPHIFY_EXPECTED_DIRECTED, + multigraph=GRAPHIFY_EXPECTED_MULTIGRAPH, built_at_commit=built_at_commit, nodes=nodes, edges=edges, ) -class GraphifyRunner: - """Explicit pinned Graphify runner; construction never occurs in default analysis.""" - - def __init__(self, project_root: Path, executable: Path, *, timeout: float = 300.0): - self.project_root = project_root.resolve(strict=True) - self.executable = executable.resolve(strict=True) - if not self.project_root.is_dir(): - raise GraphifyAdapterError(f"Graphify project root is not a directory: {project_root}") - if not self.executable.is_file(): - raise GraphifyAdapterError(f"Graphify executable is not a file: {executable}") - if timeout <= 0: - raise GraphifyAdapterError("Graphify timeout must be positive") - self.timeout = timeout - - @staticmethod - def _environment(home: Path) -> dict[str, str]: - allowed = ("PATH", "SYSTEMROOT", "WINDIR", "TMPDIR", "TEMP", "TMP", "LANG", "LC_ALL") - environment = {name: os.environ[name] for name in allowed if name in os.environ} - environment.update( - { - "GRAPHIFY_OUT": GRAPHIFY_OUTPUT_DIRECTORY, - "HOME": str(home), - "XDG_CONFIG_HOME": str(home / ".config"), - } - ) - return environment +def _receipt_for(snapshot: GraphifySnapshot) -> GraphifySnapshotReceipt: + return GraphifySnapshotReceipt( + side=snapshot.side, + graph_sha256=snapshot.graph_sha256, + graph_schema_version=snapshot.graph_schema_version, + expected_graphify_package=snapshot.expected_graphify_package, + expected_graphify_version=snapshot.expected_graphify_version, + expected_graphify_command=snapshot.expected_graphify_command, + expected_version_output=snapshot.expected_version_output, + directed=snapshot.directed, + multigraph=snapshot.multigraph, + node_count=len(snapshot.nodes), + edge_count=len(snapshot.edges), + ) - def _run(self, args: list[str], *, cwd: Path, home: Path) -> subprocess.CompletedProcess[str]: - try: - result = subprocess.run( - args, - cwd=cwd, - env=self._environment(home), - check=False, - capture_output=True, - text=True, - encoding="utf-8", - errors="strict", - timeout=self.timeout, - ) - except (OSError, subprocess.TimeoutExpired, UnicodeError) as error: - raise GraphifyAdapterError(f"Graphify command failed to execute: {error}") from error - if result.returncode: - detail = result.stderr.strip() or result.stdout.strip() or "no diagnostic output" - raise GraphifyAdapterError(f"Graphify command failed ({result.returncode}): {detail}") - return result - - def validate_tool(self, *, cwd: Path, home: Path) -> None: - """Require the exact separately installed POC version before extraction.""" - result = self._run([str(self.executable), "--version"], cwd=cwd, home=home) - if result.stdout.strip() != f"graphify {GRAPHIFY_PACKAGE_VERSION}": - raise GraphifyAdapterError( - f"unsupported Graphify version {result.stdout.strip()!r}; " - f"expected graphify {GRAPHIFY_PACKAGE_VERSION}" - ) - def extract_snapshot(self, side: GraphSide, output_root: Path) -> GraphifySnapshot: - """Run one explicit code-only extraction into a fresh side-qualified directory.""" - if side not in {"baseline", "target"}: - raise GraphifyAdapterError(f"unsupported snapshot side: {side!r}") - root = output_root.expanduser().resolve(strict=False) - if root == self.project_root or root.is_relative_to(self.project_root): - raise GraphifyAdapterError("Graphify output must be outside the analyzed project") - side_directory = root / side - if side_directory.exists() or side_directory.is_symlink(): - raise GraphifyAdapterError( - f"Graphify snapshot directory already exists: {side_directory}" - ) - root.mkdir(parents=True, exist_ok=True) - side_directory.mkdir(mode=0o700) - home = side_directory / "home" - home.mkdir(mode=0o700) - self.validate_tool(cwd=side_directory, home=home) - command = [ - str(self.executable), - "extract", - str(self.project_root), - "--code-only", - "--no-cluster", - "--force", - "--out", - str(side_directory), - ] - self._run(command, cwd=side_directory, home=home) - graph_path = side_directory / GRAPHIFY_OUTPUT_DIRECTORY / "graph.json" - try: - graph_path.resolve(strict=True).relative_to(side_directory.resolve(strict=True)) - except (OSError, ValueError) as error: - raise GraphifyAdapterError("Graphify did not produce a confined graph.json") from error - snapshot = load_graphify_snapshot( - graph_path, - project_root=self.project_root, - side=side, - graphify_version=GRAPHIFY_PACKAGE_VERSION, - ) - receipt = GraphifySnapshotReceipt( - side=snapshot.side, - graph_sha256=snapshot.graph_sha256, - graph_schema_version=snapshot.graph_schema_version, - graphify_version=snapshot.graphify_version, - node_count=len(snapshot.nodes), - edge_count=len(snapshot.edges), - ) - receipt_path = side_directory / "snapshot-receipt.json" - try: - with receipt_path.open("x", encoding="utf-8") as handle: - handle.write(receipt.as_json()) - handle.flush() - os.fsync(handle.fileno()) - except OSError as error: - raise GraphifyAdapterError( - f"cannot publish Graphify snapshot receipt: {error}" - ) from error - return snapshot +def import_graphify_snapshot( + graph_path: Path, + *, + project_root: Path, + side: GraphSide, + receipt_path: Path, + expected_sha256: str | None = None, +) -> GraphifySnapshot: + """Validate an offline snapshot and exclusively publish its import receipt.""" + snapshot = load_graphify_snapshot( + graph_path, + project_root=project_root, + side=side, + expected_sha256=expected_sha256, + ) + try: + with receipt_path.open("x", encoding="utf-8") as handle: + handle.write(_receipt_for(snapshot).as_json()) + handle.flush() + os.fsync(handle.fileno()) + except (OSError, RuntimeError) as error: + raise GraphifyAdapterError( + f"cannot publish Graphify snapshot receipt {receipt_path}: {error}" + ) from error + return snapshot diff --git a/tests/fixtures/graphify_0_9_30_graph.json b/tests/fixtures/graphify_0_9_30_graph.json index 0cb50cc..7daa8a5 100644 --- a/tests/fixtures/graphify_0_9_30_graph.json +++ b/tests/fixtures/graphify_0_9_30_graph.json @@ -1,6 +1,6 @@ { "built_at_commit": "1111111111111111111111111111111111111111", - "directed": false, + "directed": true, "graph": {}, "hyperedges": [], "links": [ diff --git a/tests/unit/test_graphify_adapter.py b/tests/unit/test_graphify_adapter.py index 59ce8cb..1e9c7d6 100644 --- a/tests/unit/test_graphify_adapter.py +++ b/tests/unit/test_graphify_adapter.py @@ -1,24 +1,27 @@ -"""Offline contract tests for the opt-in pinned Graphify POC boundary.""" +"""Offline contract tests for the pinned Graphify graph import boundary.""" from __future__ import annotations import hashlib import json import os -import shutil -import subprocess from pathlib import Path +from types import SimpleNamespace from typing import Any from unittest.mock import patch import pytest +from fastapi_endpoint_detector.analyzer import graphify_adapter from fastapi_endpoint_detector.analyzer.graphify_adapter import ( + GRAPHIFY_EXPECTED_DIRECTED, + GRAPHIFY_EXPECTED_MULTIGRAPH, GRAPHIFY_GRAPH_SCHEMA_VERSION, + GRAPHIFY_PACKAGE_NAME, GRAPHIFY_PACKAGE_VERSION, GraphifyAdapterError, - GraphifyRunner, GraphifySourceSpan, + import_graphify_snapshot, load_graphify_snapshot, ) @@ -35,32 +38,43 @@ def _project(tmp_path: Path) -> Path: return project -def _write_payload(path: Path, payload: dict[str, Any]) -> None: - path.write_text(json.dumps(payload, allow_nan=False, sort_keys=True), encoding="utf-8") +def _payload() -> dict[str, Any]: + return json.loads(FIXTURE.read_text(encoding="utf-8")) -def _completed( - stdout: str = "", *, code: int = 0, stderr: str = "" -) -> subprocess.CompletedProcess[str]: - return subprocess.CompletedProcess([], code, stdout, stderr) +def _write_payload(path: Path, payload: dict[str, Any]) -> None: + path.write_text(json.dumps(payload, allow_nan=False, sort_keys=True), encoding="utf-8") -def test_loads_pinned_fixture_with_separate_provenance_and_direction(tmp_path: Path) -> None: +def test_loads_pinned_fixture_with_exact_source_provenance_and_orientation( + tmp_path: Path, +) -> None: project = _project(tmp_path) snapshot = load_graphify_snapshot(FIXTURE, project_root=project, side="target") + app_hash = hashlib.sha256((project / "app.py").read_bytes()).hexdigest() assert snapshot.side == "target" - assert snapshot.graphify_version == GRAPHIFY_PACKAGE_VERSION + assert snapshot.expected_graphify_package == GRAPHIFY_PACKAGE_NAME + assert snapshot.expected_graphify_version == GRAPHIFY_PACKAGE_VERSION + assert snapshot.expected_graphify_command == "graphify" + assert snapshot.expected_version_output == "graphify 0.9.30" assert snapshot.graph_schema_version == GRAPHIFY_GRAPH_SCHEMA_VERSION assert snapshot.graph_sha256 == hashlib.sha256(FIXTURE.read_bytes()).hexdigest() + assert snapshot.directed is GRAPHIFY_EXPECTED_DIRECTED + assert snapshot.multigraph is GRAPHIFY_EXPECTED_MULTIGRAPH assert snapshot.built_at_commit == "1" * 40 - assert snapshot.nodes[1].span == GraphifySourceSpan(Path("app.py"), 2, 3) + assert snapshot.nodes[1].source_file == Path("app.py") + assert snapshot.nodes[1].source_sha256 == app_hash + assert snapshot.nodes[1].span == GraphifySourceSpan(Path("app.py"), 2, 3, app_hash) assert snapshot.nodes[1].extractor_strength == "EXTRACTED" assert snapshot.edges[1].source_id == "handler" assert snapshot.edges[1].target_id == "helper" + assert snapshot.edges[1].orientation == "caller-to-callee" assert snapshot.edges[1].traversable is True + assert snapshot.edges[2].orientation == "importer-to-imported" assert snapshot.edges[2].extractor_strength == "INFERRED" assert snapshot.edges[2].traversable is True + assert snapshot.edges[3].orientation == "symmetric" assert snapshot.edges[3].extractor_strength == "AMBIGUOUS" assert snapshot.edges[3].traversable is False @@ -84,12 +98,12 @@ def test_snapshot_hash_is_checked_against_the_same_byte_snapshot(tmp_path: Path) side="baseline", expected_sha256="0" * 64, ) - with pytest.raises(GraphifyAdapterError, match="unsupported Graphify version"): + with pytest.raises(GraphifyAdapterError, match="lowercase SHA-256"): load_graphify_snapshot( FIXTURE, project_root=project, side="baseline", - graphify_version="0.9.29", + expected_sha256="not-a-hash", ) @@ -101,11 +115,13 @@ def test_snapshot_hash_is_checked_against_the_same_byte_snapshot(tmp_path: Path) (lambda value: value["nodes"].append(dict(value["nodes"][0])), "duplicate"), (lambda value: value["links"][0].update({"target": "missing"}), "unknown node"), (lambda value: value["nodes"][0].update({"file_type": "document"}), "code-only"), - (lambda value: value["nodes"][0].update({"source_file": "missing.py"}), "project file"), (lambda value: value["nodes"][1].update({"source_location": "L9-L2"}), "reversed"), (lambda value: value["links"][0].update({"confidence_score": 2.0}), "between 0 and 1"), (lambda value: value.update({"built_at_commit": "ABC"}), "full Git OID"), - (lambda value: value["links"][0].update({"unexpected": "drift"}), "unsupported fields"), + (lambda value: value["links"][0].update({"unexpected": "drift"}), "unsupported schema"), + (lambda value: value.update({"graph": {"name": "unchecked"}}), "empty object"), + (lambda value: value["nodes"][0].update({"metadata": {}}), "unsupported schema"), + (lambda value: value["links"][0].update({"context": []}), "context"), ], ids=[ "schema-drift", @@ -113,20 +129,22 @@ def test_snapshot_hash_is_checked_against_the_same_byte_snapshot(tmp_path: Path) "duplicate-node", "dangling-edge", "non-code-node", - "missing-source", "reversed-span", "invalid-confidence-score", "invalid-commit", "edge-schema-drift", + "nonempty-graph-metadata", + "removed-unvalidated-field", + "malformed-known-field", ], ) -def test_strict_schema_rejects_unsupported_or_ambiguous_graphs( +def test_strict_schema_rejects_unsupported_or_malformed_graphs( tmp_path: Path, mutation: Any, message: str, ) -> None: project = _project(tmp_path) - payload = json.loads(FIXTURE.read_text(encoding="utf-8")) + payload = _payload() mutation(payload) graph = tmp_path / "graph.json" _write_payload(graph, payload) @@ -135,10 +153,72 @@ def test_strict_schema_rejects_unsupported_or_ambiguous_graphs( load_graphify_snapshot(graph, project_root=project, side="target") +@pytest.mark.parametrize("source_value", [None, ""]) +def test_nodes_require_nonempty_source_paths(tmp_path: Path, source_value: object) -> None: + project = _project(tmp_path) + payload = _payload() + if source_value is None: + del payload["nodes"][0]["source_file"] + else: + payload["nodes"][0]["source_file"] = source_value + graph = tmp_path / "missing-source.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match=r"source_file|missing"): + load_graphify_snapshot(graph, project_root=project, side="target") + + +@pytest.mark.parametrize( + "mutate", + [ + lambda value: value["nodes"][1].update({"source_location": "L999999"}), + lambda value: value["links"][1].update({"source_location": "L999999"}), + ], + ids=["node-occurrence", "edge-occurrence"], +) +def test_occurrences_cannot_exceed_exact_source_bytes(tmp_path: Path, mutate: Any) -> None: + project = _project(tmp_path) + payload = _payload() + mutate(payload) + graph = tmp_path / "out-of-range.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match="exceeds the exact source bytes"): + load_graphify_snapshot(graph, project_root=project, side="target") + + +@pytest.mark.parametrize( + ("field", "value"), + [("directed", False), ("multigraph", False), ("directed", 1)], +) +def test_requires_attested_graph_direction_values( + tmp_path: Path, field: str, value: object +) -> None: + project = _project(tmp_path) + payload = _payload() + payload[field] = value + graph = tmp_path / "direction.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match=field): + load_graphify_snapshot(graph, project_root=project, side="target") + + +def test_rejects_relations_without_attested_orientation(tmp_path: Path) -> None: + project = _project(tmp_path) + payload = _payload() + payload["links"][1]["relation"] = "unknown_direction" + graph = tmp_path / "orientation.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError, match="no attested source-to-target orientation"): + load_graphify_snapshot(graph, project_root=project, side="target") + + def test_strict_json_rejects_duplicate_members_and_non_finite_numbers(tmp_path: Path) -> None: project = _project(tmp_path) duplicate = tmp_path / "duplicate.json" - duplicate.write_text('{"directed":false,"directed":true}', encoding="utf-8") + duplicate.write_text('{"directed":true,"directed":false}', encoding="utf-8") with pytest.raises(GraphifyAdapterError, match="duplicate"): load_graphify_snapshot(duplicate, project_root=project, side="target") @@ -153,131 +233,118 @@ def test_strict_json_rejects_duplicate_members_and_non_finite_numbers(tmp_path: def test_source_paths_must_remain_inside_the_analyzed_project(tmp_path: Path) -> None: project = _project(tmp_path) - payload = json.loads(FIXTURE.read_text(encoding="utf-8")) + payload = _payload() payload["nodes"][0]["source_file"] = "../outside.py" graph = tmp_path / "escape.json" _write_payload(graph, payload) - with pytest.raises(GraphifyAdapterError, match="escapes the project root"): + with pytest.raises(GraphifyAdapterError, match="confined project file"): load_graphify_snapshot(graph, project_root=project, side="target") -def test_runner_invokes_only_pinned_code_only_pipeline_and_writes_receipt( - tmp_path: Path, +def test_graph_and_source_reads_enforce_exact_and_over_limit_boundaries( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: project = _project(tmp_path) - executable = tmp_path / "graphify" - executable.write_text("#!/bin/sh\nexit 1\n", encoding="utf-8") - executable.chmod(0o700) - output_root = tmp_path / "snapshots" - calls: list[tuple[list[str], dict[str, object]]] = [] - - def fake_run(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]: - calls.append((args, kwargs)) - if args[-1] == "--version": - return _completed(f"graphify {GRAPHIFY_PACKAGE_VERSION}\n") - graph = output_root / "target" / "graphify-out" / "graph.json" - graph.parent.mkdir(parents=True) - shutil.copyfile(FIXTURE, graph) - return _completed("code-only graph written\n") - - with ( - patch.dict( - os.environ, - { - "GEMINI_API_KEY": "must-not-pass", - "GOOGLE_API_KEY": "must-not-pass", - "HTTP_PROXY": "must-not-pass", - }, - ), - patch("subprocess.run", side_effect=fake_run), - ): - snapshot = GraphifyRunner(project, executable, timeout=12).extract_snapshot( - "target", output_root + graph_size = len(FIXTURE.read_bytes()) + source_size = max(len(path.read_bytes()) for path in project.glob("*.py")) + + monkeypatch.setattr(graphify_adapter, "MAX_GRAPH_BYTES", graph_size) + monkeypatch.setattr(graphify_adapter, "MAX_SOURCE_BYTES", source_size) + load_graphify_snapshot(FIXTURE, project_root=project, side="target") + + monkeypatch.setattr(graphify_adapter, "MAX_GRAPH_BYTES", graph_size - 1) + with pytest.raises(GraphifyAdapterError, match="Graphify snapshot exceeds"): + load_graphify_snapshot(FIXTURE, project_root=project, side="target") + + monkeypatch.setattr(graphify_adapter, "MAX_GRAPH_BYTES", graph_size) + monkeypatch.setattr(graphify_adapter, "MAX_SOURCE_BYTES", source_size - 1) + with pytest.raises(GraphifyAdapterError, match="source file exceeds"): + load_graphify_snapshot(FIXTURE, project_root=project, side="target") + + +def test_graph_read_rejects_non_regular_and_mutating_files( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + project = _project(tmp_path) + with pytest.raises(GraphifyAdapterError, match="not a regular file"): + load_graphify_snapshot(Path(os.devnull), project_root=project, side="target") + + real_fstat = os.fstat + calls = 0 + + def changing_fstat(fd: int) -> os.stat_result | SimpleNamespace: + nonlocal calls + calls += 1 + result = real_fstat(fd) + if calls == 2: + return SimpleNamespace( + st_mode=result.st_mode, + st_dev=result.st_dev, + st_ino=result.st_ino, + st_size=result.st_size, + st_mtime_ns=result.st_mtime_ns + 1, + st_ctime_ns=result.st_ctime_ns, + ) + return result + + monkeypatch.setattr(graphify_adapter.os, "fstat", changing_fstat) + with pytest.raises(GraphifyAdapterError, match="changed while it was being read"): + load_graphify_snapshot(FIXTURE, project_root=project, side="target") + + +def test_missing_graph_project_and_source_failures_are_adapter_errors(tmp_path: Path) -> None: + project = _project(tmp_path) + with pytest.raises(GraphifyAdapterError, match="cannot read Graphify snapshot"): + load_graphify_snapshot(tmp_path / "missing.json", project_root=project, side="target") + with pytest.raises(GraphifyAdapterError, match="project root does not exist"): + load_graphify_snapshot( + FIXTURE, project_root=tmp_path / "missing-project", side="target" ) - assert snapshot.graph_sha256 == hashlib.sha256(FIXTURE.read_bytes()).hexdigest() - assert len(calls) == 2 - extract_args, extract_kwargs = calls[1] - assert extract_args == [ - str(executable.resolve()), - "extract", - str(project.resolve()), - "--code-only", - "--no-cluster", - "--force", - "--out", - str((output_root / "target").resolve()), - ] - assert extract_kwargs.get("shell", False) is False - environment = extract_kwargs["env"] - assert isinstance(environment, dict) - assert "GEMINI_API_KEY" not in environment - assert "GOOGLE_API_KEY" not in environment - assert "HTTP_PROXY" not in environment - assert "NO_PROXY" not in environment - assert all( - forbidden not in extract_args - for forbidden in ("--backend", "--mcp", "--wiki", "--watch", "--global") - ) - receipt_path = output_root / "target" / "snapshot-receipt.json" + payload = _payload() + payload["nodes"][0]["source_file"] = "missing.py" + graph = tmp_path / "missing-source-file.json" + _write_payload(graph, payload) + with pytest.raises(GraphifyAdapterError, match="confined project file"): + load_graphify_snapshot(graph, project_root=project, side="target") + + +def test_offline_import_writes_pinned_receipt_without_subprocess(tmp_path: Path) -> None: + project = _project(tmp_path) + receipt_path = tmp_path / "snapshot-receipt.json" + + with patch("subprocess.run") as run: + snapshot = import_graphify_snapshot( + FIXTURE, + project_root=project, + side="target", + receipt_path=receipt_path, + expected_sha256=hashlib.sha256(FIXTURE.read_bytes()).hexdigest(), + ) + run.assert_not_called() + assert not hasattr(graphify_adapter, "GraphifyRunner") + receipt = json.loads(receipt_path.read_text(encoding="utf-8")) assert receipt == { + "directed": True, "edge_count": 4, + "expected_graphify_command": "graphify", + "expected_graphify_package": "graphifyy", + "expected_graphify_version": "0.9.30", + "expected_version_output": "graphify 0.9.30", "graph_schema_version": GRAPHIFY_GRAPH_SCHEMA_VERSION, "graph_sha256": snapshot.graph_sha256, - "graphify_version": GRAPHIFY_PACKAGE_VERSION, + "import_mode": "offline-only", + "multigraph": True, "node_count": 3, "side": "target", } - -def test_runner_is_no_clobber_and_rejects_project_local_output_before_execution( - tmp_path: Path, -) -> None: - project = _project(tmp_path) - executable = tmp_path / "graphify" - executable.write_text("tool", encoding="utf-8") - executable.chmod(0o700) - runner = GraphifyRunner(project, executable) - - with ( - patch("subprocess.run") as run, - pytest.raises(GraphifyAdapterError, match="outside"), - ): - runner.extract_snapshot("target", project / "snapshots") - run.assert_not_called() - - output_root = tmp_path / "snapshots" - (output_root / "baseline").mkdir(parents=True) - with ( - patch("subprocess.run") as run, - pytest.raises(GraphifyAdapterError, match="already exists"), - ): - runner.extract_snapshot("baseline", output_root) - run.assert_not_called() - - -def test_runner_rejects_unpinned_version_and_command_failure(tmp_path: Path) -> None: - project = _project(tmp_path) - executable = tmp_path / "graphify" - executable.write_text("tool", encoding="utf-8") - executable.chmod(0o700) - runner = GraphifyRunner(project, executable) - - with ( - patch("subprocess.run", return_value=_completed("graphify 9.9.9\n")), - pytest.raises(GraphifyAdapterError, match="unsupported Graphify version"), - ): - runner.extract_snapshot("target", tmp_path / "version-output") - - def failed(args: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: - if args[-1] == "--version": - return _completed(f"graphify {GRAPHIFY_PACKAGE_VERSION}\n") - return _completed(code=7, stderr="structural extraction failed") - - with ( - patch("subprocess.run", side_effect=failed), - pytest.raises(GraphifyAdapterError, match="structural extraction failed"), - ): - runner.extract_snapshot("baseline", tmp_path / "failure-output") + with pytest.raises(GraphifyAdapterError, match="cannot publish"): + import_graphify_snapshot( + FIXTURE, + project_root=project, + side="target", + receipt_path=receipt_path, + ) From f95ee6c9ae7b8f2f6c051d757ac11051f9f9fa5f Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 01:31:07 +0300 Subject: [PATCH 3/4] fix: bound offline Graphify file imports --- .../analyzer/graphify_adapter.py | 41 +++++--- tests/unit/test_graphify_adapter.py | 97 ++++++++++++++++++- 2 files changed, 126 insertions(+), 12 deletions(-) diff --git a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py index bdb649a..18b0f66 100644 --- a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py +++ b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py @@ -18,7 +18,7 @@ from pathlib import Path from typing import Literal, cast -from fastapi_endpoint_detector.strict_data import DuplicateKeyError, load_json_unique +from fastapi_endpoint_detector.strict_data import load_json_unique GRAPHIFY_PACKAGE_NAME = "graphifyy" GRAPHIFY_PACKAGE_VERSION = "0.9.30" @@ -209,7 +209,10 @@ def _bounded_string(value: object, location: str, *, allow_empty: bool = False) def _finite_number(value: object, location: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): raise GraphifyAdapterError(f"{location} must be a finite number") - result = float(value) + try: + result = float(value) + except (TypeError, ValueError, OverflowError) as error: + raise GraphifyAdapterError(f"{location} must be a finite number") from error if not math.isfinite(result): raise GraphifyAdapterError(f"{location} must be finite") return result @@ -221,18 +224,29 @@ def _signature(value: os.stat_result) -> tuple[int, int, int, int, int]: def _read_regular_file(path: Path, limit: int, description: str) -> tuple[bytes, _FileSnapshot]: """Read at most limit+1 bytes through one descriptor and detect path mutation.""" + flags = os.O_RDONLY + for supported_flag in ("O_BINARY", "O_CLOEXEC", "O_NOFOLLOW", "O_NONBLOCK"): + flags |= getattr(os, supported_flag, 0) + + descriptor: int | None = None try: - with path.open("rb") as handle: - before = os.fstat(handle.fileno()) - if not stat.S_ISREG(before.st_mode): - raise GraphifyAdapterError(f"{description} is not a regular file: {path}") + descriptor = os.open(path, flags) + before = os.fstat(descriptor) + if not stat.S_ISREG(before.st_mode): + raise GraphifyAdapterError(f"{description} is not a regular file: {path}") + handle = os.fdopen(descriptor, "rb") + descriptor = None + with handle: raw = handle.read(limit + 1) after = os.fstat(handle.fileno()) - current = path.stat() + current = path.stat(follow_symlinks=False) except GraphifyAdapterError: raise except (OSError, RuntimeError) as error: raise GraphifyAdapterError(f"cannot read {description} {path}: {error}") from error + finally: + if descriptor is not None: + os.close(descriptor) before_signature = _signature(before) if _signature(after) != before_signature or _signature(current) != before_signature: @@ -268,7 +282,7 @@ def _strict_json(raw: bytes, source: Path) -> dict[str, object]: try: text = raw.decode("utf-8") value = load_json_unique(text) - except (UnicodeError, json.JSONDecodeError, DuplicateKeyError) as error: + except (UnicodeError, TypeError, ValueError, OverflowError) as error: raise GraphifyAdapterError(f"invalid strict JSON in {source}: {error}") from error _validate_json_numbers(value) if not isinstance(value, dict): @@ -347,8 +361,13 @@ def _source_span( match = _LOCATION.fullmatch(raw_location) if match is None: raise GraphifyAdapterError(f"{location}.source_location is unsupported: {raw_location!r}") - start_line = int(match.group("start")) - end_line = int(match.group("end") or start_line) + try: + start_line = int(match.group("start")) + end_line = int(match.group("end") or start_line) + except (TypeError, ValueError, OverflowError) as error: + raise GraphifyAdapterError( + f"{location}.source_location is unsupported: {raw_location!r}" + ) from error if end_line < start_line: raise GraphifyAdapterError(f"{location}.source_location has a reversed range") if end_line > source.line_count: @@ -381,7 +400,7 @@ def _optional_edge_span( def _strength(value: object, location: str, *, optional: bool = False) -> GraphifyStrength | None: if value is None and optional: return None - if value not in {"EXTRACTED", "INFERRED", "AMBIGUOUS"}: + if not isinstance(value, str) or value not in {"EXTRACTED", "INFERRED", "AMBIGUOUS"}: raise GraphifyAdapterError(f"{location} has unsupported extractor confidence {value!r}") return cast("GraphifyStrength", value) diff --git a/tests/unit/test_graphify_adapter.py b/tests/unit/test_graphify_adapter.py index 1e9c7d6..a9586a4 100644 --- a/tests/unit/test_graphify_adapter.py +++ b/tests/unit/test_graphify_adapter.py @@ -5,9 +5,10 @@ import hashlib import json import os +import threading from pathlib import Path from types import SimpleNamespace -from typing import Any +from typing import TYPE_CHECKING, Any from unittest.mock import patch import pytest @@ -25,6 +26,10 @@ load_graphify_snapshot, ) +if TYPE_CHECKING: + from collections.abc import Callable + + FIXTURE = Path(__file__).parents[1] / "fixtures" / "graphify_0_9_30_graph.json" @@ -46,6 +51,34 @@ def _write_payload(path: Path, payload: dict[str, Any]) -> None: path.write_text(json.dumps(payload, allow_nan=False, sort_keys=True), encoding="utf-8") +def _assert_fifo_rejected_without_writer(fifo: Path, load: Callable[[], object]) -> None: + outcomes: list[object] = [] + + def invoke() -> None: + try: + outcomes.append(load()) + except Exception as error: + outcomes.append(error) + + thread = threading.Thread(target=invoke, daemon=True) + thread.start() + thread.join(timeout=1.0) + if thread.is_alive(): + try: + writer = os.open(fifo, os.O_WRONLY | getattr(os, "O_NONBLOCK", 0)) + except OSError: + pass + else: + os.close(writer) + thread.join(timeout=1.0) + pytest.fail("FIFO read blocked while waiting for a writer") + + assert len(outcomes) == 1 + error = outcomes[0] + assert isinstance(error, GraphifyAdapterError) + assert "not a regular file" in str(error) + + def test_loads_pinned_fixture_with_exact_source_provenance_and_orientation( tmp_path: Path, ) -> None: @@ -153,6 +186,49 @@ def test_strict_schema_rejects_unsupported_or_malformed_graphs( load_graphify_snapshot(graph, project_root=project, side="target") +@pytest.mark.parametrize( + ("collection", "field", "value"), + [ + ("nodes", "confidence", []), + ("links", "confidence", []), + ("nodes", "confidence_score", 10**400), + ("links", "weight", 10**400), + ], + ids=[ + "node-confidence-list", + "edge-confidence-list", + "huge-node-score", + "huge-edge-weight", + ], +) +def test_malformed_confidence_and_huge_numbers_are_adapter_errors( + tmp_path: Path, + collection: str, + field: str, + value: object, +) -> None: + project = _project(tmp_path) + payload = _payload() + payload[collection][0][field] = value + graph = tmp_path / "malformed-known-field.json" + _write_payload(graph, payload) + + with pytest.raises(GraphifyAdapterError): + load_graphify_snapshot(graph, project_root=project, side="target") + + +def test_parser_value_errors_are_adapter_errors(tmp_path: Path) -> None: + project = _project(tmp_path) + graph = tmp_path / "huge-json-number.json" + graph.write_text( + FIXTURE.read_text(encoding="utf-8").replace('"weight": 0.5', '"weight": ' + "9" * 5000), + encoding="utf-8", + ) + + with pytest.raises(GraphifyAdapterError): + load_graphify_snapshot(graph, project_root=project, side="target") + + @pytest.mark.parametrize("source_value", [None, ""]) def test_nodes_require_nonempty_source_paths(tmp_path: Path, source_value: object) -> None: project = _project(tmp_path) @@ -263,6 +339,25 @@ def test_graph_and_source_reads_enforce_exact_and_over_limit_boundaries( load_graphify_snapshot(FIXTURE, project_root=project, side="target") +@pytest.mark.skipif(not hasattr(os, "mkfifo"), reason="FIFOs are unavailable") +@pytest.mark.parametrize("fifo_input", ["graph", "source"]) +def test_graph_and_source_fifos_fail_without_a_writer(tmp_path: Path, fifo_input: str) -> None: + project = _project(tmp_path) + graph = tmp_path / "graph.json" + if fifo_input == "graph": + fifo = graph + else: + fifo = project / "app.py" + fifo.unlink() + _write_payload(graph, _payload()) + os.mkfifo(fifo) + + _assert_fifo_rejected_without_writer( + fifo, + lambda: load_graphify_snapshot(graph, project_root=project, side="target"), + ) + + def test_graph_read_rejects_non_regular_and_mutating_files( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: From 034fa39fc0d01619353280fd033cd48ce4160d94 Mon Sep 17 00:00:00 2001 From: shaggitza Date: Thu, 30 Jul 2026 01:43:45 +0300 Subject: [PATCH 4/4] style: format Graphify adapter files --- src/fastapi_endpoint_detector/analyzer/graphify_adapter.py | 7 ++----- tests/unit/test_graphify_adapter.py | 4 +--- 2 files changed, 3 insertions(+), 8 deletions(-) diff --git a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py index 18b0f66..6093fcc 100644 --- a/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py +++ b/src/fastapi_endpoint_detector/analyzer/graphify_adapter.py @@ -372,8 +372,7 @@ def _source_span( raise GraphifyAdapterError(f"{location}.source_location has a reversed range") if end_line > source.line_count: raise GraphifyAdapterError( - f"{location}.source_location exceeds the exact source bytes " - f"({source.line_count} lines)" + f"{location}.source_location exceeds the exact source bytes ({source.line_count} lines)" ) assert source.relative_path is not None return source, GraphifySourceSpan( @@ -550,9 +549,7 @@ def _adapt_edges( assert strength is not None span = _optional_edge_span(registry, edge, location) _validate_optional_fields(edge, location) - edges.append( - GraphifyEdge(source_id, target_id, relation, orientation, strength, span) - ) + edges.append(GraphifyEdge(source_id, target_id, relation, orientation, strength, span)) return tuple(edges) diff --git a/tests/unit/test_graphify_adapter.py b/tests/unit/test_graphify_adapter.py index a9586a4..6e64e0e 100644 --- a/tests/unit/test_graphify_adapter.py +++ b/tests/unit/test_graphify_adapter.py @@ -393,9 +393,7 @@ def test_missing_graph_project_and_source_failures_are_adapter_errors(tmp_path: with pytest.raises(GraphifyAdapterError, match="cannot read Graphify snapshot"): load_graphify_snapshot(tmp_path / "missing.json", project_root=project, side="target") with pytest.raises(GraphifyAdapterError, match="project root does not exist"): - load_graphify_snapshot( - FIXTURE, project_root=tmp_path / "missing-project", side="target" - ) + load_graphify_snapshot(FIXTURE, project_root=tmp_path / "missing-project", side="target") payload = _payload() payload["nodes"][0]["source_file"] = "missing.py"