diff --git a/.github/workflows/validate-graph.yml b/.github/workflows/validate-graph.yml new file mode 100644 index 0000000..353719b --- /dev/null +++ b/.github/workflows/validate-graph.yml @@ -0,0 +1,34 @@ +name: validate / graph + +on: + pull_request: + paths: + - "apps/python-app/libs/cssc_graph/**" + - "apps/python-app/services/graph-service/**" + - "supply-chain-graph/**" + - ".github/workflows/validate-graph.yml" + workflow_dispatch: + +permissions: + contents: read + +jobs: + graph: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@08c6903cd8c0fde910a37f88322edcfb5dd907a8 # v5.0.0 + - uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6 + with: + python-version: "3.12" + - name: Install declared graph test dependencies + run: | + python -m pip install -e 'apps/python-app/libs/cssc_graph[test]' \ + -e 'apps/python-app/services/graph-service[test]' + - name: Require the real graph engine and run library and HTTP tests + run: | + python -c 'import ladybug; print("Native LadybugDB engine available")' + python -m pytest --import-mode=importlib -q \ + apps/python-app/libs/cssc_graph/tests \ + apps/python-app/services/graph-service/tests + - name: Validate committed example schemas + run: cssc-graph validate supply-chain-graph diff --git a/apps/python-app/libs/cssc_graph/README.md b/apps/python-app/libs/cssc_graph/README.md index 725403c..6f11cb5 100644 --- a/apps/python-app/libs/cssc_graph/README.md +++ b/apps/python-app/libs/cssc_graph/README.md @@ -25,6 +25,14 @@ cssc-graph id supply-chain-graph/examples/artifact-mirrored.yaml `validate` exits non-zero on any error, so it fits pre-commit and CI. +For current indexing, query and integrity behavior, see the +[query guide](../../../../docs/guides/observability/querying-the-supply-chain-graph.md). +`index` emits unresolved-reference warnings by default; +`validate --strict-references` and `index --strict-references` reject missing +independent occurrence evidence. Deployment observations are available through +`deployments --ref`, and `tag-history --format json` includes explicit +observation-derived intervals and ambiguity. + ## Concepts - **Envelope** — the common wrapper on every record (`schemaVersion`, `kind`, diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/cli.py b/apps/python-app/libs/cssc_graph/cssc_graph/cli.py index 702faf2..7ebba73 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/cli.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/cli.py @@ -57,13 +57,19 @@ def cli() -> None: default="text", help="Output format.", ) -def validate(root: Path, schema_dir: Path | None, output_format: str) -> None: - """Validate every record under ROOT against the JSON Schemas.""" +@click.option("--strict-references", is_flag=True, help="Also reject references without independent input evidence.") +def validate(root: Path, schema_dir: Path | None, output_format: str, strict_references: bool) -> None: + """Validate schemas, optionally requiring all occurrence references to resolve.""" resolved_schema_dir = schema_dir or default_schema_dir(root) try: files = discover_record_files(root, resolved_schema_dir) diagnostics = validate_data(root, resolved_schema_dir) + if strict_references and not diagnostics: + from .records import iter_records + from .references import reference_diagnostics + + diagnostics = reference_diagnostics(iter_records(root, resolved_schema_dir), strict=True) except (FileNotFoundError, ValueError) as exc: raise click.ClickException(str(exc)) @@ -116,12 +122,18 @@ def validate(root: Path, schema_dir: Path | None, output_format: str) -> None: is_flag=True, help="Delete any existing database first for a clean rebuild.", ) -def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool) -> None: - """Validate ROOT, then index its records into a LadybugDB graph.""" +@click.option("--strict-references", is_flag=True, help="Reject unresolved references before changing the database.") +def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool, strict_references: bool) -> None: + """Validate and index ROOT; default reference warnings preserve permissive behavior.""" resolved_schema_dir = schema_dir or default_schema_dir(root) try: diagnostics = validate_data(root, resolved_schema_dir) + if strict_references and not diagnostics: + from .records import iter_records + from .references import reference_diagnostics + + diagnostics = reference_diagnostics(iter_records(root, resolved_schema_dir), strict=True) except (FileNotFoundError, ValueError) as exc: raise click.ClickException(str(exc)) if diagnostics: @@ -135,10 +147,15 @@ def index(root: Path, database: Path, schema_dir: Path | None, rebuild: bool) -> if rebuild: GraphStore.destroy(database) - with GraphStore(database) as store: - store.init_schema() - stats = index_data(store, root, resolved_schema_dir) + try: + with GraphStore(database) as store: + store.init_schema() + stats = index_data(store, root, resolved_schema_dir) + except ValueError as exc: + raise click.ClickException(str(exc)) from exc + for diagnostic in stats.diagnostics: + click.echo(diagnostic.format(root), err=True) click.echo(f"Indexed {stats.records} record(s) into {database}.") for kind, count in sorted(stats.by_kind.items()): click.echo(f" {kind}: {count}") @@ -168,9 +185,13 @@ def _format_option(func): def _open_store(database: Path): + """Open a compatible graph for queries, with actionable legacy-rebuild errors.""" from .graph import GraphStore - return GraphStore(database).connect() + try: + return GraphStore(database).connect() + except ValueError as exc: + raise click.ClickException(str(exc)) from exc def _emit_subgraph(subgraph: dict, output_format: str) -> None: @@ -215,7 +236,7 @@ def path(database: Path, digest: str | None, ref: str | None, depth: int, output @click.option("--tag", required=True) @_format_option def tag_history(database: Path, repo: str, tag: str, output_format: str) -> None: - """List every digest a tag pointed to over time, chronologically.""" + """List tag observations and evidence-derived intervals, with ambiguity explicit.""" from . import queries @@ -229,9 +250,38 @@ def tag_history(database: Path, repo: str, tag: str, output_format: str) -> None return for entry in history: click.echo(f" {entry['observedAt']} {entry['digest']} {entry.get('runUrl', '')}".rstrip()) + click.echo( + f" observed interval: [{entry['validFrom'] or 'unknown'}, " + f"{entry['validUntil'] or 'unknown'}); ambiguous={entry['ambiguous']}" + ) click.echo(f"{len(history)} observation(s).") +@cli.command(name="deployments") +@_database_option +@click.option("--ref", required=True, help="The deployed registry/repository[@digest|:tag].") +@_format_option +def deployments(database: Path, ref: str, output_format: str) -> None: + """List distinct deployment observations in chronological order.""" + from . import queries + + store = _open_store(database) + try: + history = queries.deployment_history(store, ref) + finally: + store.close() + if output_format == "json": + click.echo(json.dumps(history, indent=2)) + else: + for entry in history: + click.echo( + f"{entry['recordedAt']} {entry['cluster']}/{entry['namespace']} " + f"{entry['occurrence']} chart={entry['chart']}:{entry['chartVersion']} " + f"event={entry['eventId']}" + ) + click.echo(f"{len(history)} deployment observation(s).") + + @cli.command() @_database_option @click.option("--ref", required=True, help="The built image (registry/repository[@digest|:tag]).") diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/graph.py b/apps/python-app/libs/cssc_graph/cssc_graph/graph.py index efce088..d031428 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/graph.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/graph.py @@ -40,7 +40,7 @@ "CREATE REL TABLE IF NOT EXISTS POINTED_TO(" "FROM Tag TO Occurrence, observedAt STRING, digest STRING, runUrl STRING)", "CREATE REL TABLE IF NOT EXISTS RUNS(" - "FROM Deployment TO Occurrence, chart STRING, chartVersion STRING, " + "FROM Deployment TO Occurrence, eventId STRING, sourceJson STRING, chart STRING, chartVersion STRING, " "runUrl STRING, recordedAt STRING)", "CREATE REL TABLE IF NOT EXISTS REFERS_TO(" "FROM Occurrence TO Occurrence, artifactType STRING, observedAt STRING, " @@ -62,13 +62,32 @@ def __init__(self, database_path: str | Path) -> None: # -- lifecycle ---------------------------------------------------------- def connect(self) -> "GraphStore": + """Open the database and reject incompatible history schemas without altering them.""" import ladybug as lb # lazy: only the graph commands need the native dep self._db = lb.Database(str(self._path)) self._conn = lb.Connection(self._db) + try: + self.check_history_schema() + except ValueError: + self.close() + raise return self + def check_history_schema(self) -> None: + """Read-only compatibility guard shared by query opens and schema initialization.""" + existing = self.query("CALL show_tables() RETURN name") + if any(row["name"] == "RUNS" for row in existing): + columns = self.query("CALL table_info('RUNS') RETURN name") + if not {"eventId", "sourceJson"}.issubset({row["name"] for row in columns}): + raise ValueError( + "graph schema predates deployment event history; rebuild this derived " + "database with cssc-graph index --rebuild (source files are unchanged)" + ) + def init_schema(self) -> None: + """Create a fresh schema or reject an old deployment table without data loss.""" + self.check_history_schema() for ddl in SCHEMA_DDL: self.execute(ddl) diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/indexer.py b/apps/python-app/libs/cssc_graph/cssc_graph/indexer.py index 6946590..0dc3d82 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/indexer.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/indexer.py @@ -2,26 +2,31 @@ from __future__ import annotations +import hashlib from dataclasses import dataclass, field from pathlib import Path from typing import Any, Mapping from .graph import GraphStore from .identity import ( + canonical_json, make_occurrence_key, make_ref, split_ref, tag_key, ) from .records import iter_records +from .references import UnresolvedReferences, reference_diagnostics +from .validate import Diagnostic @dataclass class IndexStats: - """Counts produced by an indexing run.""" + """Indexed record counts and non-fatal reference diagnostics for one input set.""" records: int = 0 by_kind: dict[str, int] = field(default_factory=dict) + diagnostics: list[Diagnostic] = field(default_factory=list) def bump(self, kind: str) -> None: self.records += 1 @@ -175,17 +180,23 @@ def _index_artifact_built(self, record: Mapping[str, Any]) -> None: ) def _index_artifact_deployed(self, record: Mapping[str, Any]) -> None: + """Preserve each full deployment observation; replaying it is idempotent.""" image_key = self._merge_occurrence_from(record["image"]) env = record.get("environment", {}) dep_key = self._merge_deployment(env.get("cluster", ""), env.get("namespace", "")) chart = record.get("chart", {}) + event_id = "sha256:" + hashlib.sha256( + canonical_json({key: value for key, value in record.items() if key != "id"}).encode("utf-8") + ).hexdigest() self._store.execute( "MATCH (d:Deployment {key: $dep}), (o:Occurrence {key: $image}) " - "MERGE (d)-[r:RUNS]->(o) " - "SET r.chart = $chart, r.chartVersion = $chartVersion, " + "MERGE (d)-[r:RUNS {eventId: $eventId}]->(o) " + "SET r.sourceJson = $sourceJson, r.chart = $chart, r.chartVersion = $chartVersion, " "r.runUrl = $runUrl, r.recordedAt = $recordedAt", { "dep": dep_key, + "eventId": event_id, + "sourceJson": canonical_json(record.get("source", {})), "image": image_key, "chart": chart.get("name", ""), "chartVersion": chart.get("version", ""), @@ -300,7 +311,8 @@ def _run_url(record: Mapping[str, Any]) -> str: def _snake(kind: str) -> str: - out = [] + """Map a record kind to its per-kind indexer method suffix.""" + out: list[str] = [] for ch in kind: if ch.isupper() and out: out.append("_") @@ -308,8 +320,21 @@ def _snake(kind: str) -> str: return "".join(out) -def index_data(store: GraphStore, root: Path, schema_dir: Path | None = None) -> IndexStats: - """Index every record under *root* into *store*.""" +def index_data( + store: GraphStore, root: Path, schema_dir: Path | None = None, + *, strict_references: bool = False, +) -> IndexStats: + """Index a validated input set, returning reference diagnostics. + Strict unresolved references raise before any record writes. Declarations + anywhere in this input set resolve references regardless of file order. + """ + + records = list(iter_records(root, schema_dir)) + diagnostics = reference_diagnostics(records, strict=strict_references) + if strict_references and diagnostics: + raise UnresolvedReferences(diagnostics) indexer = Indexer(store) - return indexer.index_records(data for _, data in iter_records(root, schema_dir)) + stats = indexer.index_records(data for _, data in records) + stats.diagnostics = diagnostics + return stats diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py index 8c93084..4168dd1 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py @@ -6,10 +6,14 @@ from __future__ import annotations +import json +from itertools import groupby from typing import Any from .graph import GraphStore from .identity import split_ref, tag_key +from .timestamps import instant as _instant +from .timestamps import utc as _utc # Relationship types that form an artifact's supply-chain path. PATH_RELS: tuple[str, ...] = ("MIRRORED_FROM", "PROMOTED_FROM", "BUILT_FROM") @@ -66,12 +70,9 @@ def resolve_seed(store: GraphStore, *, digest: str | None = None, ref: str | Non def _latest_tag_occurrence(store: GraphStore, registry: str, repository: str, tag: str) -> str | None: - rows = store.query( - "MATCH (t:Tag {key: $tk})-[e:POINTED_TO]->(o:Occurrence) " - "RETURN o.key AS key ORDER BY e.observedAt DESC LIMIT 1", - {"tk": tag_key(registry, repository, tag)}, - ) - return rows[0]["key"] if rows else None + """Resolve the latest real instant only when the observed target is unambiguous.""" + rows = tag_history(store, f"{registry}/{repository}", tag) + return rows[-1]["occurrence"] if rows and not rows[-1]["ambiguous"] else None def _neighbors(store: GraphStore, key: str, rel: str, outgoing: bool) -> list[dict[str, Any]]: @@ -305,13 +306,63 @@ def index_of(store: GraphStore, child_key: str) -> str | None: def tag_history(store: GraphStore, ref: str, tag: str) -> list[dict[str, Any]]: + """Return original observations plus inferred, UTC-normalized target windows. + + Consecutive unambiguous observations of one digest share a window. A + conflicting observation breaks inference; an unknown end is never "now". + Bounds describe observations, not proof of continuous registry validity. + """ registry, repository = split_ref(ref) - return store.query( + rows = store.query( "MATCH (t:Tag {key: $tk})-[e:POINTED_TO]->(o:Occurrence) " "RETURN e.observedAt AS observedAt, o.digest AS digest, o.key AS occurrence, " "e.runUrl AS runUrl ORDER BY e.observedAt", {"tk": tag_key(registry, repository, tag)}, ) + rows.sort(key=lambda row: ( + _instant(row["observedAt"]), row["digest"], row["runUrl"] or "", row["observedAt"], + )) + groups = [list(group) for _, group in groupby(rows, key=lambda row: _instant(row["observedAt"]))] + start = 0 + while start < len(groups): + group = groups[start] + digest = group[0]["digest"] + if len({row["digest"] for row in group}) > 1: + for row in group: + row.update(validFrom=None, validUntil=None, ambiguous=True, intervalBasis="observations") + start += 1 + continue + end = start + 1 + while end < len(groups) and {row["digest"] for row in groups[end]} == {digest}: + end += 1 + until = None + if end < len(groups) and len({row["digest"] for row in groups[end]}) == 1: + until = _utc(groups[end][0]["observedAt"]) + for window in groups[start:end]: + for row in window: + row.update( + validFrom=_utc(group[0]["observedAt"]), validUntil=until, + ambiguous=False, intervalBasis="observations", + ) + start = end + return rows + + +def deployment_history(store: GraphStore, ref: str) -> list[dict[str, Any]]: + """Fetch resolved deployment history in one projection, then order exact instants.""" + keys = resolve_seed(store, ref=ref) + if not keys: + return [] + rows = store.query( + "MATCH (d:Deployment)-[e:RUNS]->(o:Occurrence) WHERE o.key IN $keys " + "RETURN d.cluster AS cluster, d.namespace AS namespace, o.key AS occurrence, " + "e.eventId AS eventId, e.sourceJson AS sourceJson, e.chart AS chart, e.chartVersion AS chartVersion, " + "e.recordedAt AS recordedAt, e.runUrl AS runUrl", + {"keys": keys}, + ) + for row in rows: + row["source"] = json.loads(row.pop("sourceJson")) + return sorted(rows, key=lambda row: (_instant(row["recordedAt"]), row["eventId"])) def find( @@ -347,6 +398,7 @@ def find( def show(store: GraphStore, ref: str) -> dict[str, Any] | None: + """Inspect one occurrence, including tag intervals and immutable deployment observations.""" seeds = resolve_seed(store, ref=ref) if not seeds: return None @@ -364,16 +416,24 @@ def show(store: GraphStore, ref: str) -> dict[str, Any] | None: "RETURN art.digest AS digest, art.mediaType AS mediaType, art.artifactType AS artifactType", {"k": key}, ) - tags = store.query( + tag_names = store.query( "MATCH (t:Tag)-[e:POINTED_TO]->(o:Occurrence {key: $k}) " - "RETURN t.tag AS tag, e.observedAt AS observedAt ORDER BY e.observedAt", + "RETURN DISTINCT t.tag AS tag ORDER BY tag", {"k": key}, ) + tags = [ + {"tag": row["tag"], **observation} + for row in tag_names + for observation in tag_history(store, occ["ref"], row["tag"]) + if observation["occurrence"] == key + ] + tags.sort(key=lambda row: (_instant(row["observedAt"]), row["tag"], row["digest"])) return { "occurrence": occ, "artifact": artifact[0] if artifact else None, "annotations": annotations, "tags": tags, + "deployments": deployment_history(store, key), "referrers": referrers(store, ref=key)["edges"], "deleted": bool(occ.get("deletedAt")), "path": traverse(store, [key], PATH_RELS, direction="both", max_depth=2), diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/references.py b/apps/python-app/libs/cssc_graph/cssc_graph/references.py new file mode 100644 index 0000000..9ab6028 --- /dev/null +++ b/apps/python-app/libs/cssc_graph/cssc_graph/references.py @@ -0,0 +1,111 @@ +"""Diagnose references without independent occurrence evidence in the input set.""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping +from pathlib import Path +from typing import Any + +from .identity import make_occurrence_key, occurrence_key, split_ref +from .validate import Diagnostic + + +class UnresolvedReferences(ValueError): + """Reject strict indexing while preserving the individual file diagnostics.""" + + def __init__(self, diagnostics: list[Diagnostic]) -> None: + self.diagnostics = diagnostics + super().__init__( + f"{len(diagnostics)} unresolved occurrence reference(s); nothing indexed." + ) + + +def _facts(record: Mapping[str, Any]) -> tuple[set[str], list[tuple[str, str]]]: + """Separate direct observations from reference-only positions of a validated record.""" + kind = record["kind"] + declared: set[str] = set() + references: list[tuple[str, str]] = [] + if kind == "ArtifactObserved": + declared.add(occurrence_key(record["occurrence"])) + elif kind == "ArtifactMirrored": + # A successful copy directly observes both the upstream and acquired image. + declared.update(occurrence_key(record[side]) for side in ("from", "to")) + elif kind == "ArtifactPromoted": + declared.add(occurrence_key(record["to"])) + references.append(("$.from", occurrence_key(record["from"]))) + elif kind == "ArtifactBuilt": + declared.add(occurrence_key(record["image"])) + base = record["base"] + registry, repository = split_ref(base["name"]) + references.append( + ("$.base", make_occurrence_key(registry, repository, base["digest"])) + ) + elif kind == "ArtifactDeployed": + references.append(("$.image", occurrence_key(record["image"]))) + elif kind in { + "TagObserved", + "ArtifactDeleted", + "ReferrerObserved", + "ImageIndexObserved", + }: + occurrence = record["occurrence"] + registry, repository = occurrence["registry"], occurrence["repository"] + if kind == "TagObserved": + declared.add(make_occurrence_key(registry, repository, record["digest"])) + elif kind == "ArtifactDeleted": + references.append( + ( + "$.digest", + make_occurrence_key(registry, repository, record["digest"]), + ) + ) + elif kind == "ReferrerObserved": + declared.add( + make_occurrence_key(registry, repository, record["referrer"]["digest"]) + ) + references.append( + ( + "$.subject", + make_occurrence_key( + registry, repository, record["subject"]["digest"] + ), + ) + ) + else: + declared.add( + make_occurrence_key(registry, repository, record["index"]["digest"]) + ) + declared.update( + make_occurrence_key(registry, repository, child["digest"]) + for child in record["platforms"] + ) + return declared, references + + +def reference_diagnostics( + records: Iterable[tuple[Path, Mapping[str, Any]]], + *, + strict: bool = False, +) -> list[Diagnostic]: + """Check all records together, not file order, without mutating a graph. + + Missing evidence in this input set is a warning by default, not proof that + the remote artifact never existed. Strict mode returns error diagnostics. + """ + declared: set[str] = set() + references: list[tuple[Path, str, str]] = [] + for path, record in records: + observations, targets = _facts(record) + declared.update(observations) + references.extend((path, location, key) for location, key in targets) + return [ + Diagnostic( + path, + location, + f"unresolved occurrence {key}: add an observation/acquisition/build record " + "or include the missing evidence in the data root", + severity="error" if strict else "warning", + ) + for path, location, key in references + if key not in declared + ] diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/schema.py b/apps/python-app/libs/cssc_graph/cssc_graph/schema.py index 848f613..0ce5b4b 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/schema.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/schema.py @@ -10,8 +10,19 @@ from jsonschema import Draft202012Validator, FormatChecker from referencing import Registry, Resource +from .timestamps import instant + SCHEMA_GLOB = "*.schema.json" SOURCES_SCHEMA_TITLE = "Sources" +FORMAT_CHECKER = FormatChecker() + + +@FORMAT_CHECKER.checks("date-time", raises=(ValueError, OverflowError)) +def _is_date_time(value: object) -> bool: + """Enforce the query parser's timestamp contract even without optional format extras.""" + if isinstance(value, str): + instant(value) + return True @dataclass(frozen=True) @@ -23,10 +34,11 @@ class SchemaBundle: sources_schema: dict[str, Any] | None def validator_for(self, schema: dict[str, Any]) -> Draft202012Validator: + """Validate references and formats using the same timestamp syntax as graph queries.""" return Draft202012Validator( schema, registry=self.registry, - format_checker=FormatChecker(), + format_checker=FORMAT_CHECKER, ) diff --git a/apps/python-app/libs/cssc_graph/cssc_graph/timestamps.py b/apps/python-app/libs/cssc_graph/cssc_graph/timestamps.py new file mode 100644 index 0000000..b9d4bbb --- /dev/null +++ b/apps/python-app/libs/cssc_graph/cssc_graph/timestamps.py @@ -0,0 +1,49 @@ +"""Shared RFC 3339 validation and exact ordering for recorded graph evidence.""" + +from __future__ import annotations + +import re +from calendar import monthrange +from datetime import UTC, datetime +from fractions import Fraction + + +def _parts(value: str) -> tuple[datetime, bool, str]: + """Parse RFC 3339, retaining leap-second and exact fractional evidence.""" + match = re.fullmatch( + r"(\d{4}-\d{2}-\d{2}[Tt]\d{2}:\d{2}:\d{2})(?:\.(\d+))?([Zz]|[+-](?:[01]\d|2[0-3]):[0-5]\d)", + value, + ) + if match is None: + raise ValueError( + f"observation timestamp requires an RFC 3339 timezone: {value!r}" + ) + whole, fraction, zone = match.groups() + leap_second = whole.endswith(":60") + # Position leap seconds after every fraction of :59, not at the next :00. + parseable = whole[:-2] + "59" if leap_second else whole + timestamp = datetime.fromisoformat(parseable.upper() + zone.upper()).astimezone(UTC) + if leap_second and ( + timestamp.hour != 23 + or timestamp.minute != 59 + or timestamp.day != monthrange(timestamp.year, timestamp.month)[1] + ): + raise ValueError("leap-second notation is only valid at the end of a UTC month") + return timestamp, leap_second, fraction or "" + + +def instant(value: str) -> tuple[datetime, bool, Fraction]: + """Order real instants exactly, including schema-valid sub-microsecond fractions.""" + timestamp, leap_second, fraction = _parts(value) + return timestamp, leap_second, Fraction(int(fraction or "0"), 10 ** len(fraction)) + + +def utc(value: str) -> str: + """Normalize a timestamp to UTC while retaining its exact fractional instant.""" + timestamp, leap_second, fraction = _parts(value) + digits = fraction.rstrip("0") + suffix = f".{digits}" if digits else "" + whole = timestamp.isoformat(timespec="seconds").removesuffix("+00:00") + if leap_second: + whole = whole[:-2] + "60" + return whole + suffix + "Z" diff --git a/apps/python-app/libs/cssc_graph/tests/test_integrity.py b/apps/python-app/libs/cssc_graph/tests/test_integrity.py new file mode 100644 index 0000000..ef872aa --- /dev/null +++ b/apps/python-app/libs/cssc_graph/tests/test_integrity.py @@ -0,0 +1,605 @@ +"""Regression coverage for reference integrity and evidence-preserving history.""" + +from __future__ import annotations + +import json +from copy import deepcopy +from pathlib import Path +from unittest.mock import patch + +import pytest +from click.testing import CliRunner + +pytest.importorskip("ladybug") + +from cssc_graph import queries +from cssc_graph.cli import cli +from cssc_graph.graph import GraphStore +from cssc_graph.indexer import Indexer, index_data +from cssc_graph.references import UnresolvedReferences, reference_diagnostics + +SCHEMAS = Path(__file__).resolve().parents[5] / "supply-chain-graph/schema" +DIGEST = "sha256:" + "a" * 64 +OTHER = "sha256:" + "b" * 64 +REF = "ghcr.io/example/app" +OCCURRENCE = {"registry": "ghcr.io", "repository": "example/app", "digest": DIGEST} +SOURCE = {"type": "human", "workflow": "integrity fixture"} + + +def observed(digest: str = DIGEST) -> dict: + """Declare one independently observed artifact occurrence.""" + return { + "schemaVersion": 1, + "kind": "ArtifactObserved", + "recordedAt": "2026-10-28T00:00:00Z", + "source": SOURCE, + "artifact": {"digest": digest}, + "occurrence": {**OCCURRENCE, "digest": digest}, + } + + +def deployed(at: str = "2026-10-28T01:00:00Z", version: str = "1") -> dict: + """Make a deployment observation with independently variable event evidence.""" + return { + "schemaVersion": 1, + "kind": "ArtifactDeployed", + "recordedAt": at, + "source": SOURCE, + "image": OCCURRENCE, + "environment": {"cluster": "test", "namespace": "demo"}, + "chart": {"name": "app", "version": version}, + } + + +def tag(at: str, digest: str = DIGEST) -> dict: + """Observe a tag target at a timestamp, preserving offsets for ordering tests.""" + return { + "schemaVersion": 1, + "kind": "TagObserved", + "recordedAt": at, + "observedAt": at, + "source": SOURCE, + "occurrence": {"registry": "ghcr.io", "repository": "example/app"}, + "tag": "stable", + "digest": digest, + } + + +@pytest.fixture +def store(tmp_path: Path): + """Own a real ephemeral graph for each integrity regression.""" + with GraphStore(tmp_path / "db") as database: + database.init_schema() + yield database + + +def test_reference_diagnostics_are_independent_of_file_order(): + """A later declaration resolves a reference without order-dependent warnings.""" + records = [(Path("deploy.json"), deployed()), (Path("observed.json"), observed())] + assert ( + reference_diagnostics(records) + == reference_diagnostics(list(reversed(records))) + == [] + ) + missing = reference_diagnostics(records[:1]) + assert len(missing) == 1 + assert missing[0].file == Path("deploy.json") and missing[0].location == "$.image" + assert f"{REF}@{DIGEST}" in missing[0].message + assert missing[0].severity == "warning" + + +def test_strict_reference_check_writes_nothing(store: GraphStore, tmp_path: Path): + """Strict indexing rejects dangling references before writing any graph facts.""" + (tmp_path / "deploy.json").write_text(json.dumps(deployed())) + with pytest.raises(UnresolvedReferences): + index_data(store, tmp_path, SCHEMAS, strict_references=True) + assert store.scalar("MATCH (o:Occurrence) RETURN count(o)") == 0 + + +def test_permissive_indexing_reports_but_preserves_missing_occurrence( + store: GraphStore, tmp_path: Path +): + """The legacy permissive path still indexes, but returns actionable warnings.""" + (tmp_path / "deploy.json").write_text(json.dumps(deployed())) + stats = index_data(store, tmp_path, SCHEMAS) + assert len(stats.diagnostics) == 1 + assert store.scalar("MATCH (o:Occurrence) RETURN count(o)") == 1 + + +def test_cli_strict_does_not_destroy_an_existing_database( + store: GraphStore, tmp_path: Path +): + """An invalid strict rebuild must not erase a previously indexed database.""" + Indexer(store).index_record(observed()) + store.close() + root = tmp_path / "input" + root.mkdir() + (root / "deploy.json").write_text(json.dumps(deployed())) + result = CliRunner().invoke( + cli, + [ + "index", + str(root), + "-d", + str(tmp_path / "db"), + "--schema-dir", + str(SCHEMAS), + "--rebuild", + "--strict-references", + ], + ) + assert result.exit_code == 1 and "deploy.json" in result.output + with GraphStore(tmp_path / "db") as existing: + assert existing.scalar("MATCH (o:Occurrence) RETURN count(o)") == 1 + + +def test_deployment_events_are_append_only_and_idempotent(store: GraphStore): + """Different observations survive even when they deploy the same image.""" + indexer = Indexer(store) + records = [deployed(), deployed("2026-10-28T02:00:00Z"), deployed(version="2")] + for record in records + list(reversed(records)): + indexer.index_record(record) + history = queries.deployment_history(store, REF) + assert len(history) == 3 + assert len({row["eventId"] for row in history}) == 3 + assert {row["chartVersion"] for row in history} == {"1", "2"} + assert len(queries.show(store, f"{REF}@{DIGEST}")["deployments"]) == 3 + + +def test_deployment_evidence_changes_are_not_overwritten(store: GraphStore): + """Distinct source evidence at the same timestamp remains independently inspectable.""" + first = deployed() + second = deepcopy(first) + second["source"] = {**SOURCE, "workflow": "another observation"} + Indexer(store).index_records([first, second, first]) + history = queries.deployment_history(store, REF) + assert len(history) == 2 + assert {row["source"]["workflow"] for row in history} == { + "integrity fixture", + "another observation", + } + + +def test_intervals_merge_repeated_targets_and_sort_real_instants(store: GraphStore): + """Offset-aware order and repeated observations define a single observed window.""" + Indexer(store).index_records( + [ + tag("2026-10-28T03:00:00+02:00", OTHER), + tag("2026-10-28T00:00:00Z"), + tag("2026-10-28T00:30:00Z"), + ] + ) + history = queries.tag_history(store, REF, "stable") + assert [row["digest"] for row in history] == [DIGEST, DIGEST, OTHER] + assert history[0]["validFrom"] == history[1]["validFrom"] == "2026-10-28T00:00:00Z" + assert ( + history[0]["validUntil"] == history[1]["validUntil"] == "2026-10-28T01:00:00Z" + ) + assert history[-1]["validUntil"] is None + + +def test_conflicting_tag_observations_do_not_fabricate_intervals_or_current_target( + store: GraphStore, +): + """Equivalent timestamps with different digests are ambiguous regardless of file order.""" + Indexer(store).index_records( + [ + tag("2026-10-28T01:00:00+01:00", OTHER), + tag("2026-10-28T00:00:00Z"), + ] + ) + history = queries.tag_history(store, REF, "stable") + assert len(history) == 2 + assert all( + row["ambiguous"] and row["validFrom"] is None and row["validUntil"] is None + for row in history + ) + assert queries.resolve_seed(store, ref=f"{REF}:stable") == [] + + +def test_ambiguous_next_observation_does_not_assert_a_known_end(store: GraphStore): + """Conflicting future evidence leaves the preceding interval's upper bound unknown.""" + Indexer(store).index_records( + [ + tag("2026-10-28T00:00:00Z"), + tag("2026-10-28T01:00:00Z"), + tag("2026-10-28T02:00:00+01:00", OTHER), + ] + ) + assert queries.tag_history(store, REF, "stable")[0]["validUntil"] is None + + +@pytest.mark.parametrize( + "kind,position", + [ + ("ArtifactBuilt", "$.base"), + ("ArtifactPromoted", "$.from"), + ("ReferrerObserved", "$.subject"), + ("ArtifactDeleted", "$.digest"), + ], +) +def test_all_reference_positions_have_actionable_locations(kind: str, position: str): + """Every reference-only record position reports the missing occurrence.""" + record = { + "kind": kind, + "image": {**OCCURRENCE, "digest": OTHER}, + "base": {"name": REF, "digest": DIGEST}, + "from": OCCURRENCE, + "to": {**OCCURRENCE, "digest": OTHER}, + "occurrence": {"registry": "ghcr.io", "repository": "example/app"}, + "digest": DIGEST, + "subject": {"digest": DIGEST}, + "referrer": {"digest": OTHER}, + } + diagnostics = reference_diagnostics([(Path("event.json"), record)], strict=True) + assert len(diagnostics) == 1 + assert diagnostics[0].severity == "error" + assert diagnostics[0].location == position + assert f"{REF}@{DIGEST}" in diagnostics[0].message + + +def test_tag_platform_and_mirror_observations_can_resolve_references(): + """Direct registry discovery/acquisition is evidence, not a dangling reference itself.""" + records = [ + { + "kind": "ArtifactMirrored", + "from": OCCURRENCE, + "to": {**OCCURRENCE, "digest": OTHER}, + }, + { + "kind": "ImageIndexObserved", + "occurrence": OCCURRENCE, + "index": {"digest": DIGEST}, + "platforms": [{"digest": OTHER}], + }, + tag("2026-10-28T00:00:00Z"), + deployed(), + ] + assert ( + reference_diagnostics( + [(Path(str(index)), record) for index, record in enumerate(records)] + ) + == [] + ) + + +def test_references_are_repository_scoped_not_just_digest_scoped(): + """The same content observed in a different repository cannot resolve a reference.""" + other_repo = observed() + other_repo["occurrence"] = {**OCCURRENCE, "repository": "example/different"} + assert ( + len(reference_diagnostics([(Path("o"), other_repo), (Path("d"), deployed())])) + == 1 + ) + + +def test_cli_warning_and_json_strict_validation(tmp_path: Path): + """Permissive indexing is visibly warned; strict validate returns structured errors.""" + root = tmp_path / "input" + root.mkdir() + (root / "deploy.json").write_text(json.dumps(deployed())) + runner = CliRunner() + warning = runner.invoke( + cli, + [ + "index", + str(root), + "-d", + str(tmp_path / "db"), + "--schema-dir", + str(SCHEMAS), + ], + ) + assert ( + warning.exit_code == 0 + and "warning:" in warning.output + and "$.image" in warning.output + ) + strict = runner.invoke( + cli, + [ + "validate", + str(root), + "--schema-dir", + str(SCHEMAS), + "--strict-references", + "--format", + "json", + ], + ) + assert strict.exit_code == 1 + body = json.loads(strict.output) + assert body["errorCount"] == 1 and body["diagnostics"][0]["severity"] == "error" + + +def test_cli_deployment_history_and_tag_json(tmp_path: Path): + """CLI commands expose the same additive history/interval fields as shared queries.""" + database = tmp_path / "db" + with GraphStore(database) as graph: + graph.init_schema() + Indexer(graph).index_records( + [deployed(), deployed(version="2"), tag("2026-10-28T00:00:00Z")] + ) + runner = CliRunner() + deployments = runner.invoke( + cli, ["deployments", "--ref", REF, "-d", str(database), "--format", "json"] + ) + assert deployments.exit_code == 0 + assert len(json.loads(deployments.output)) == 2 + tags = runner.invoke( + cli, + [ + "tag-history", + "--repo", + REF, + "--tag", + "stable", + "-d", + str(database), + "--format", + "json", + ], + ) + assert tags.exit_code == 0 + assert json.loads(tags.output)[0]["validFrom"] == "2026-10-28T00:00:00Z" + + +def test_old_schema_is_rejected_with_rebuild_guidance(tmp_path: Path): + """Old derived databases are not silently migrated or stripped of deployment facts.""" + with GraphStore(tmp_path / "old") as graph: + graph.execute("CREATE NODE TABLE Occurrence(key STRING, PRIMARY KEY(key))") + graph.execute("CREATE NODE TABLE Deployment(key STRING, PRIMARY KEY(key))") + graph.execute( + "CREATE REL TABLE RUNS(FROM Deployment TO Occurrence, recordedAt STRING)" + ) + with pytest.raises(ValueError, match="index --rebuild"): + graph.init_schema() + assert not any( + row["name"] == "eventId" + for row in graph.query("CALL table_info('RUNS') RETURN name") + ) + + +@pytest.mark.parametrize("command", ["show", "deployments", "index"]) +def test_legacy_read_commands_return_rebuild_guidance(tmp_path: Path, command: str): + """Query and index CLI opens turn legacy-schema rejection into actionable Click errors.""" + database = tmp_path / "old" + with GraphStore(database) as graph: + graph.execute("CREATE NODE TABLE Occurrence(key STRING, PRIMARY KEY(key))") + graph.execute("CREATE NODE TABLE Deployment(key STRING, PRIMARY KEY(key))") + graph.execute( + "CREATE REL TABLE RUNS(FROM Deployment TO Occurrence, recordedAt STRING)" + ) + if command == "index": + root = tmp_path / "input" + root.mkdir() + (root / "observed.json").write_text(json.dumps(observed())) + args = ["index", str(root), "--schema-dir", str(SCHEMAS), "-d", str(database)] + else: + args = [command, "--ref", f"{REF}@{DIGEST}", "-d", str(database)] + result = CliRunner().invoke(cli, args) + assert result.exit_code == 1 + assert "cssc-graph index --rebuild" in result.output + assert "Binder exception" not in result.output + assert result.exception is not None and isinstance(result.exception, SystemExit) + + +def test_tag_target_can_revert_and_intervals_stay_separate(store: GraphStore): + """Returning to an old digest creates a new window, not a fabricated continuous one.""" + Indexer(store).index_records( + [ + tag("2026-10-28T00:00:00Z"), + tag("2026-10-28T01:00:00Z", OTHER), + tag("2026-10-28T02:00:00Z"), + ] + ) + history = queries.tag_history(store, REF, "stable") + assert history[0]["validUntil"] == history[1]["validFrom"] == "2026-10-28T01:00:00Z" + assert history[-1]["validFrom"] == "2026-10-28T02:00:00Z" + shown = queries.show(store, f"{REF}@{DIGEST}") + assert len(shown["tags"]) == 2 + assert shown["tags"][0]["validFrom"] != shown["tags"][1]["validFrom"] + + +def test_deployment_history_sorts_offsets_and_is_environment_scoped(store: GraphStore): + """Event ordering uses real instants while keeping environment identity attached.""" + later = deployed("2026-10-28T00:30:00Z") + earlier = deployed("2026-10-28T01:00:00+01:00") + later["environment"] = {"cluster": "another", "namespace": "demo"} + Indexer(store).index_records([later, earlier]) + history = queries.deployment_history(store, REF) + assert [row["cluster"] for row in history] == ["test", "another"] + + +def test_repository_deployment_history_has_no_per_occurrence_query_loop( + store: GraphStore, +): + """A repository with many occurrences needs only seed resolution and one history query.""" + indexer = Indexer(store) + for number in range(10): + record = deployed() + record["image"] = {**OCCURRENCE, "digest": f"sha256:{number:064x}"} + indexer.index_record(record) + with patch.object(store, "query", wraps=store.query) as query: + history = queries.deployment_history(store, REF) + assert len(history) == 10 + assert query.call_count == 2 + + +def test_equivalent_offsets_with_one_target_are_not_ambiguous(store: GraphStore): + """Multiple spellings of the same instant are compatible when their digest agrees.""" + Indexer(store).index_records( + [ + tag("2026-10-28T01:00:00+01:00"), + tag("2026-10-28T00:00:00Z"), + ] + ) + rows = queries.tag_history(store, REF, "stable") + assert len(rows) == 2 and not any(row["ambiguous"] for row in rows) + assert queries.resolve_seed(store, ref=f"{REF}:stable") == [f"{REF}@{DIGEST}"] + + +def test_later_unambiguous_tag_observation_resolves_after_a_conflict(store: GraphStore): + """A past conflict remains visible without blocking an independently known later target.""" + Indexer(store).index_records( + [ + tag("2026-10-28T00:00:00Z"), + tag("2026-10-28T00:00:00Z", OTHER), + tag("2026-10-28T01:00:00Z", OTHER), + ] + ) + rows = queries.tag_history(store, REF, "stable") + assert rows[-1]["validFrom"] == "2026-10-28T01:00:00Z" + assert queries.resolve_seed(store, ref=f"{REF}:stable") == [f"{REF}@{OTHER}"] + + +def test_strict_success_and_unknown_query_shapes(store: GraphStore, tmp_path: Path): + """Strict indexing succeeds with complete evidence, and unknown histories remain empty.""" + root = tmp_path / "input" + root.mkdir() + for name, record in (("a-deploy", deployed()), ("z-observed", observed())): + (root / f"{name}.json").write_text(json.dumps(record)) + stats = index_data(store, root, SCHEMAS, strict_references=True) + assert stats.records == 2 and stats.diagnostics == [] + assert queries.tag_history(store, REF, "absent") == [] + assert queries.deployment_history(store, "ghcr.io/example/absent") == [] + + +def test_show_keeps_tag_observations_chronological_across_tag_names(store: GraphStore): + """Artifact details retain chronological ordering rather than grouping by tag name.""" + earlier = {**tag("2026-10-28T00:00:00Z"), "tag": "z-first"} + later = {**tag("2026-10-28T01:00:00Z"), "tag": "a-second"} + Indexer(store).index_records([later, earlier]) + assert [row["tag"] for row in queries.show(store, f"{REF}@{DIGEST}")["tags"]] == [ + "z-first", + "a-second", + ] + + +def test_schema_valid_lowercase_utc_timestamp_is_supported(store: GraphStore): + """RFC 3339 permits lowercase t/z even though datetime rejects lowercase z.""" + Indexer(store).index_record(tag("2026-10-28t00:00:00z")) + assert ( + queries.tag_history(store, REF, "stable")[0]["validFrom"] + == "2026-10-28T00:00:00Z" + ) + + +@pytest.mark.parametrize( + "field,value", + [ + ("recordedAt", "not-a-timestamp"), + ("observedAt", "2026-10-28T00:00:00"), + ("observedAt", "2026-02-30T00:00:00Z"), + ("observedAt", "2026-10-28T00:00:00+99:00"), + ("observedAt", "2026-10-28T01:00:00+00:60"), + ("observedAt", "2026-10-28T01:00:00-00:99"), + ("observedAt", "2026-10-28T01:00:00+23:60"), + ("observedAt", "2026-10-28T01:00:00+24:00"), + ("observedAt", "2016-12-31T12:00:60Z"), + ("observedAt", "2016-12-31T23:58:60Z"), + ("observedAt", "2016-12-30T23:59:60Z"), + ("observedAt", "2016-12-31T23:59:60+01:00"), + ], +) +def test_malformed_timestamps_fail_validation_before_indexing( + tmp_path: Path, field: str, value: str +): + """Schema checks must not silently skip date-time without optional dependency extras.""" + from cssc_graph.validate import validate_data + + record = tag("2026-10-28T00:00:00Z") + record[field] = value + (tmp_path / "tag.json").write_text(json.dumps(record)) + diagnostics = validate_data(tmp_path, SCHEMAS) + assert any( + item.location == f"$.{field}" and "date-time" in item.message + for item in diagnostics + ) + result = CliRunner().invoke( + cli, + [ + "index", + str(tmp_path), + "--schema-dir", + str(SCHEMAS), + "-d", + str(tmp_path / "db"), + ], + ) + assert result.exit_code == 1 and "nothing indexed" in result.output + assert not (tmp_path / "db").exists() + + +@pytest.mark.parametrize( + "timestamp", + [ + "2026-10-28t00:00:00z", + "2026-10-28T00:00:00.00000000001Z", + "2016-12-31T23:59:60Z", + "2017-01-01T00:59:60+01:00", + ], +) +def test_schema_and_query_timestamp_contracts_agree(tmp_path: Path, timestamp: str): + """The shared format checker accepts every supported exact query timestamp spelling.""" + from cssc_graph.validate import validate_data + + (tmp_path / "tag.json").write_text(json.dumps(tag(timestamp))) + assert validate_data(tmp_path, SCHEMAS) == [] + + +@pytest.mark.parametrize("offset", ["+00:00", "-00:00", "+23:59", "-23:59"]) +def test_valid_offset_boundaries_remain_accepted(offset: str): + """Reject malformed minutes/hours without narrowing the valid RFC 3339 offset range.""" + from cssc_graph.schema import FORMAT_CHECKER + from cssc_graph.timestamps import utc + + timestamp = f"2026-10-28T12:00:00{offset}" + assert FORMAT_CHECKER.conforms(timestamp, "date-time") + assert utc(timestamp).endswith("Z") + + +def test_schema_valid_leap_second_stays_between_adjacent_seconds(store: GraphStore): + """A leap second is neither rejected nor collapsed into the next minute.""" + Indexer(store).index_records( + [ + tag("2017-01-01T00:00:00Z"), + tag("2016-12-31T23:59:60Z", OTHER), + tag("2016-12-31T23:59:59.999999999Z"), + ] + ) + rows = queries.tag_history(store, REF, "stable") + assert [row["digest"] for row in rows] == [DIGEST, OTHER, DIGEST] + assert rows[0]["validUntil"] == rows[1]["validFrom"] == "2016-12-31T23:59:60Z" + assert rows[1]["validUntil"] == "2017-01-01T00:00:00Z" + + +def test_deployments_preserve_sub_microsecond_order(store: GraphStore): + """Deployment ordering follows exact instants rather than truncated event-ID ties.""" + Indexer(store).index_records( + [ + deployed("2026-10-28T00:00:00.0000002Z", version="later"), + deployed("2026-10-28t00:00:00.0000001z", version="earlier"), + ] + ) + assert [row["chartVersion"] for row in queries.deployment_history(store, REF)] == [ + "earlier", + "later", + ] + + +def test_sub_microsecond_instants_do_not_become_a_false_conflict(store: GraphStore): + """Datetime truncation must not collapse two distinct schema-valid observations.""" + Indexer(store).index_records( + [ + tag("2026-10-28T00:00:00.0000002Z", OTHER), + tag("2026-10-28T01:00:00.0000001+01:00"), + ] + ) + rows = queries.tag_history(store, REF, "stable") + assert not any(row["ambiguous"] for row in rows) + assert rows[0]["validFrom"] == "2026-10-28T00:00:00.0000001Z" + assert ( + rows[0]["validUntil"] == rows[1]["validFrom"] == "2026-10-28T00:00:00.0000002Z" + ) + assert queries.resolve_seed(store, ref=f"{REF}:stable") == [f"{REF}@{OTHER}"] diff --git a/apps/python-app/services/graph-service/src/graph_service/app.py b/apps/python-app/services/graph-service/src/graph_service/app.py index e834bbe..6c942e6 100644 --- a/apps/python-app/services/graph-service/src/graph_service/app.py +++ b/apps/python-app/services/graph-service/src/graph_service/app.py @@ -149,6 +149,12 @@ def tag_history(ref: str = Query(...), tag: str = Query(...)) -> dict[str, Any]: with reading() as store: return {"observations": queries.tag_history(store, ref, tag)} + @app.get("/artifacts/deployments") + def artifact_deployments(ref: str = Query(...)) -> dict[str, Any]: + """Return append-only deployment observations for an artifact or repository.""" + with reading() as store: + return {"deployments": queries.deployment_history(store, ref)} + @app.get("/search") def search( annotation: str | None = Query(None, description="name=value"), diff --git a/apps/python-app/services/graph-service/src/graph_service/config.py b/apps/python-app/services/graph-service/src/graph_service/config.py index d76a708..84422d3 100644 --- a/apps/python-app/services/graph-service/src/graph_service/config.py +++ b/apps/python-app/services/graph-service/src/graph_service/config.py @@ -20,6 +20,7 @@ class GraphSettings: schema_dir: Path | None rebuild_on_startup: bool max_depth: int + strict_references: bool = False def graph_settings() -> GraphSettings: @@ -34,6 +35,8 @@ def graph_settings() -> GraphSettings: * ``REBUILD_ON_STARTUP`` — rebuild the database from files on start (default ``true``; the single writer owns an ephemeral graph). * ``MAX_DEPTH`` — hard cap applied to every traversal depth (default ``6``). + * ``STRICT_REFERENCES`` — reject missing independent occurrence evidence + instead of indexing with visible warnings (default ``false``). """ schema_env = os.environ.get("SCHEMA_DIR") @@ -43,4 +46,5 @@ def graph_settings() -> GraphSettings: schema_dir=Path(schema_env) if schema_env else None, rebuild_on_startup=_flag(os.environ.get("REBUILD_ON_STARTUP", "true")), max_depth=int(os.environ.get("MAX_DEPTH", "6")), + strict_references=_flag(os.environ.get("STRICT_REFERENCES", "false")), ) diff --git a/apps/python-app/services/graph-service/src/graph_service/indexing.py b/apps/python-app/services/graph-service/src/graph_service/indexing.py index 16a5bfc..a82f41d 100644 --- a/apps/python-app/services/graph-service/src/graph_service/indexing.py +++ b/apps/python-app/services/graph-service/src/graph_service/indexing.py @@ -16,6 +16,8 @@ from cssc_graph.graph import GraphStore from cssc_graph.indexer import index_data +from cssc_graph.records import iter_records +from cssc_graph.references import reference_diagnostics from cssc_graph.validate import default_schema_dir, validate_data from .config import GraphSettings @@ -37,9 +39,11 @@ class GraphNotReady(RuntimeError): @dataclass class IndexInfo: + """Current indexed counts and visible non-fatal reference diagnostics.""" records: int = 0 by_kind: dict[str, int] = field(default_factory=dict) root: str = "" + diagnostics: list[str] = field(default_factory=list) class GraphIndex: @@ -70,11 +74,22 @@ def reading(self) -> Iterator[GraphStore]: yield self._store def rebuild(self) -> IndexInfo: - """(Re)build the database from the data root; single writer via a lock.""" + """Validate before replacing derived state; strict failures leave it not ready.""" settings = self._settings with self._lock: self.ready = False + root = settings.data_root + schema_dir = settings.schema_dir or default_schema_dir(root) + if root.exists(): + diagnostics = validate_data(root, schema_dir) + if not diagnostics and settings.strict_references: + diagnostics = reference_diagnostics(iter_records(root, schema_dir), strict=True) + if diagnostics: + messages = [d.format(root) for d in diagnostics] + for message in messages: + logger.error("%s", message) + raise IndexingError(messages) if self._store is not None: self._store.close() self._store = None @@ -84,15 +99,15 @@ def rebuild(self) -> IndexInfo: store.init_schema() info = IndexInfo(root=str(settings.data_root)) - root = settings.data_root if root.exists(): - schema_dir = settings.schema_dir or default_schema_dir(root) - diagnostics = validate_data(root, schema_dir) - if diagnostics: - store.close() - raise IndexingError([d.format(root) for d in diagnostics]) stats = index_data(store, root, schema_dir) - info = IndexInfo(records=stats.records, by_kind=dict(stats.by_kind), root=str(root)) + warnings = [d.format(root) for d in stats.diagnostics] + for warning in warnings: + logger.warning("%s", warning) + info = IndexInfo( + records=stats.records, by_kind=dict(stats.by_kind), root=str(root), + diagnostics=warnings, + ) else: logger.warning("data root %s does not exist; serving an empty graph", root) @@ -110,4 +125,8 @@ def close(self) -> None: self.ready = False def as_dict(self) -> dict[str, Any]: - return {"records": self.info.records, "byKind": self.info.by_kind, "root": self.info.root} + """Expose counts and reference warnings through readiness and rebuild responses.""" + return { + "records": self.info.records, "byKind": self.info.by_kind, "root": self.info.root, + "diagnostics": self.info.diagnostics, + } diff --git a/apps/python-app/services/graph-service/tests/test_app.py b/apps/python-app/services/graph-service/tests/test_app.py index fdd0f89..b57b1f7 100644 --- a/apps/python-app/services/graph-service/tests/test_app.py +++ b/apps/python-app/services/graph-service/tests/test_app.py @@ -9,6 +9,8 @@ from __future__ import annotations import json +from copy import deepcopy +from dataclasses import replace from pathlib import Path import pytest @@ -16,7 +18,6 @@ pytest.importorskip("ladybug") from fastapi.testclient import TestClient # noqa: E402 - from graph_service.app import create_app # noqa: E402 from graph_service.config import GraphSettings # noqa: E402 @@ -274,3 +275,90 @@ def test_not_ready_when_data_invalid(tmp_path: Path) -> None: assert broken.get("/healthz").status_code == 200 assert broken.get("/readyz").status_code == 503 assert broken.get("/artifacts/path", params={"ref": f"{APP_REF}@{DIGEST_APP}"}).status_code == 503 + + +def test_unresolved_references_are_visible_in_permissive_readiness(client: TestClient) -> None: + """Default indexing remains ready but exposes missing base-image evidence.""" + body = client.get("/readyz").json() + assert any("unresolved occurrence" in item and "$.base" in item for item in body["diagnostics"]) + + +def test_strict_references_fail_readiness_and_rebuild(tmp_path: Path, caplog: pytest.LogCaptureFixture) -> None: + """Strict service indexing reports missing evidence instead of serving a partial graph.""" + root = _write_data_root(tmp_path, RECORDS) + settings = replace(_settings(tmp_path, root), strict_references=True) + with TestClient(create_app(settings)) as strict: + assert strict.get("/healthz").status_code == 200 + assert strict.get("/readyz").status_code == 503 + response = strict.post("/index/rebuild") + assert response.status_code == 422 + assert any("$.base" in diagnostic for diagnostic in response.json()["detail"]) + assert "events/artifact-built/1.json" in caplog.text + assert "$.base" in caplog.text and DIGEST_UP in caplog.text + assert "add an observation/acquisition/build record" in caplog.text + + +def test_strict_references_resolve_later_input_evidence(tmp_path: Path) -> None: + """A later source observation resolves an earlier base reference in strict mode.""" + upstream = { + "schemaVersion": 1, "kind": "ArtifactObserved", "recordedAt": "2026-08-10T00:00:00Z", + "source": SOURCE, + "artifact": {"digest": DIGEST_UP}, + "occurrence": {"registry": "docker.io", "repository": "library/python", "digest": DIGEST_UP}, + } + root = _write_data_root(tmp_path, [*RECORDS, ("artifact-observed", upstream)]) + with TestClient(create_app(replace(_settings(tmp_path, root), strict_references=True))) as strict: + response = strict.get("/readyz") + assert response.status_code == 200 and response.json()["diagnostics"] == [] + + +def test_http_history_and_show_preserve_deployment_events(tmp_path: Path) -> None: + """The HTTP query and artifact details share append-only deployment history.""" + first = next(record for kind, record in RECORDS if kind == "artifact-deployed") + second = deepcopy(first) + second["recordedAt"] = "2026-08-11T00:00:00Z" + second["chart"]["version"] = "0.1.3" + second["source"] = {**SOURCE, "runAttempt": "2", "issueUrl": "https://github.com/example/repo/issues/1"} + root = _write_data_root(tmp_path, [*RECORDS, ("artifact-deployed", second)]) + with TestClient(create_app(_settings(tmp_path, root))) as client: + rows = client.get("/artifacts/deployments", params={"ref": APP_REF}).json()["deployments"] + assert len(rows) == 2 and rows[0]["eventId"] != rows[1]["eventId"] + assert rows[1]["source"] == second["source"] + shown = client.get("/artifacts/show", params={"ref": f"{APP_REF}@{DIGEST_APP}"}).json() + assert shown["deployments"] == rows + assert client.post("/index/rebuild").status_code == 200 + assert client.get("/artifacts/deployments", params={"ref": APP_REF}).json()["deployments"] == rows + + +def test_http_tag_intervals_and_ambiguous_resolution(tmp_path: Path) -> None: + """Conflicting observations return explicit unknown bounds and no arbitrary current target.""" + first = next(record for kind, record in RECORDS if kind == "tag-observed") + second = {**first, "digest": DIGEST_GOLDEN} + root = _write_data_root(tmp_path, [*RECORDS, ("tag-observed", second)]) + with TestClient(create_app(_settings(tmp_path, root))) as client: + rows = client.get("/repositories/tags/history", params={"ref": APP_REF, "tag": "0.1.0"}).json()["observations"] + assert len(rows) == 2 and all(row["ambiguous"] for row in rows) + assert all(row["validFrom"] is None for row in rows) + assert client.get("/artifacts/resolve", params={"ref": f"{APP_REF}:0.1.0"}).json()["occurrences"] == [] + + +def test_strict_reference_environment_configuration(monkeypatch) -> None: + """Operators can opt into strict references without changing default callers.""" + from graph_service.config import graph_settings + + monkeypatch.delenv("STRICT_REFERENCES", raising=False) + assert graph_settings().strict_references is False + monkeypatch.setenv("STRICT_REFERENCES", "true") + assert graph_settings().strict_references is True + + +def test_invalid_timestamp_is_rejected_before_serving_queries(tmp_path: Path) -> None: + """Malformed dates fail startup schema checks rather than later read endpoints.""" + records = deepcopy(RECORDS) + records[0][1]["recordedAt"] = "not-a-timestamp" + root = _write_data_root(tmp_path, records) + with TestClient(create_app(_settings(tmp_path, root))) as client: + assert client.get("/readyz").status_code == 503 + response = client.post("/index/rebuild") + assert response.status_code == 422 + assert any("$.recordedAt" in item and "date-time" in item for item in response.json()["detail"]) diff --git a/docs/architecture/observability/supply-chain-graph.md b/docs/architecture/observability/supply-chain-graph.md index 5996d42..2cb3c00 100644 --- a/docs/architecture/observability/supply-chain-graph.md +++ b/docs/architecture/observability/supply-chain-graph.md @@ -1,6 +1,8 @@ # Supply-chain graph: file-backed indexing -> **Status:** Proposed (design under review). No code has been implemented yet. +> **Status:** Core graph Phases 1–6 are implemented. The integrity behavior below +> covers [#232](https://github.com/toddysm/cssc-framework/issues/232); inventory +> and other follow-ups remain separately tracked. > Requirements: [#168](https://github.com/toddysm/cssc-framework/issues/168). > Storage-engine context: [#145](https://github.com/toddysm/cssc-framework/issues/145) > (Kùzu is archived; this design uses **LadybugDB** — its maintained successor, @@ -458,9 +460,29 @@ graph never asserts an exact introducing layer the data cannot prove. ### Validation Every record is validated against a JSON Schema in `schema/` before indexing. -The indexer rejects unknown `kind`s, malformed identity, and unresolved -references with actionable diagnostics (which file, which field, why) rather than -silently producing a partial graph. +Unknown `kind`s and malformed identity always fail validation. A second, +file-order-independent pass reports references without independent occurrence +evidence anywhere in the input set. Default indexing retains permissive +placeholder behavior with visible warnings; `--strict-references` and service +`STRICT_REFERENCES=true` reject these cases before indexing. A declaration is a +direct artifact/tag/platform observation or a successful acquire/build output, +not merely a deployment, deletion, base or referrer-subject reference. + +`RUNS` relationships are keyed by a full deployment-observation `eventId`, +including recorded time and source evidence. Distinct deployments do not +overwrite earlier observations; identical replays stay idempotent. The complete +source payload is retained as `RUNS.sourceJson` and returned as `source`. Shared +queries, `cssc-graph deployments`, `GET /artifacts/deployments`, and artifact +details expose the retained evidence. Old derived databases lacking this +property require an explicit rebuild; source schemas remain unchanged. + +Tag-history queries order timezone-aware instants and add UTC `validFrom` and +`validUntil` observation-derived bounds. Repeated consecutive targets share a +window. Simultaneous conflicting targets have unknown bounds and an explicit +ambiguity flag; a latest conflict does not resolve to an arbitrary current +target. No inferred window asserts continuous registry validity between +observations. See the [query guide](../../guides/observability/querying-the-supply-chain-graph.md) +for exact diagnostics, configuration, commands and output semantics. ## Indexing model diff --git a/docs/guides/observability/querying-the-supply-chain-graph.md b/docs/guides/observability/querying-the-supply-chain-graph.md index bea20e1..82eeda7 100644 --- a/docs/guides/observability/querying-the-supply-chain-graph.md +++ b/docs/guides/observability/querying-the-supply-chain-graph.md @@ -123,6 +123,69 @@ Indexed 7 record(s) into /tmp/demo-db. > git-ignored). The examples use `-d /tmp/demo-db`; set it once as `DB=/tmp/demo-db` > to shorten the commands. Every query command accepts `--format text|json`. +### Reference integrity, history, and upgrading an existing database + +Schema validation always rejects malformed records. Indexing additionally checks +references against the **entire input set**, not file order. It never contacts a +remote registry to fill a missing record. +Date-time validation uses the same exact RFC 3339 parser as the queries, +regardless of whether optional `jsonschema` format-checking extras are installed. +Malformed timestamps fail the build with file/field diagnostics, not later +queries with a server error. + +Direct observations declare occurrences: `ArtifactObserved`, both sides of a +successful mirror, promotion destinations, built images, tag observations, +referrer artifacts, and an observed image index plus its discovered platforms. +References that need independent evidence are promotion sources, build bases, +deployed images, deleted artifacts, and referrer subjects. A record in another +repository with the same digest does not resolve a missing occurrence. + +The default remains permissive: missing independent evidence produces a visible +warning while preserving the previous placeholder-node behavior: + +```text +warning: events/deploy.json: $.image: unresolved occurrence ghcr.io/example/app@sha256:...: add an observation/acquisition/build record or include the missing evidence in the data root +Indexed 1 record(s) into /tmp/demo-db. +``` + +For authoring or environments that require a complete local evidence set: + +```bash +cssc-graph validate supply-chain-graph --strict-references --format json +cssc-graph index supply-chain-graph -d /tmp/demo-db --rebuild --strict-references +``` + +Strict failures exit non-zero with file, field, target and remediation +diagnostics. They occur **before** indexing or deleting an existing database: + +```text +error: events/deploy.json: $.image: unresolved occurrence ghcr.io/example/app@sha256:...: add an observation/acquisition/build record or include the missing evidence in the data root +Error: 1 validation error(s); nothing indexed. +``` + +For `graph-service`, set `STRICT_REFERENCES=true` (Helm: +`graph-service.config.STRICT_REFERENCES: "true"`). The default is `"false"`; +reference warnings are logged and listed in `/readyz` and `/index/rebuild` +responses under `diagnostics`. A strict failure leaves liveness healthy but +readiness returns **503**, queries return **503**, and `POST /index/rebuild` +returns **422** with the actionable diagnostic list. +Startup logs also include each file, field, target and remediation diagnostic, +so operators do not need to request a rebuild to discover the missing evidence. + +Deployment history now uses additive `RUNS.eventId` and `RUNS.sourceJson` +properties. A database made with older code is rejected by both index and +query opens with explicit rebuild guidance; no +automatic migration or silent history deletion is performed. Rebuild the +**derived** database from committed source records: + +```bash +cssc-graph index supply-chain-graph -d /tmp/demo-db --rebuild +``` + +The service already rebuilds its derived database on startup. The source record +schemas are unchanged. Historical observations already lost from the source +files cannot be reconstructed by this change. + ### Answering the common questions Each question below is one of the required query outcomes from the requirements. @@ -158,6 +221,26 @@ cssc-graph cypher -d "$DB" 'MATCH (o:Occurrence) WHERE o.digest = "'"$D"'" RETUR are stored as append-only observations, never a mutable edge, so history is preserved). `show` resolves the current/indexed target. +JSON observations retain `observedAt`, `digest`, `occurrence`, and `runUrl`, and +add `validFrom`, `validUntil`, `ambiguous`, and +`intervalBasis: "observations"`. Bounds are UTC-normalized and ordered by real +instants, not timestamp text, preserving fractional precision beyond microseconds +and accepting lowercase RFC 3339 `t`/`z` and leap-second notation. +Leap markers must fall at the end of a UTC month after applying the offset. +This validates timestamp structure, not the truth of a recorded event or an +external authority's leap-second announcement schedule. +Consecutive observations of the same target share +one inferred window; a later, different target closes that window. Returning to +an earlier digest starts a new window. + +These are **observation-derived windows**, not proof of continuous tag validity +or the exact moment a registry tag changed. The last window's end is `null`, +not the current time. Conflicting digests at the same instant have +`ambiguous: true` and both bounds `null`; they also leave the preceding upper +bound unknown. A latest conflicting observation resolves to no current target, +rather than arbitrarily selecting a digest. Original observations and their +evidence remain inspectable. + ```bash cssc-graph tag-history --repo ghcr.io/toddysm/golden/python --tag 3.14-slim -d "$DB" cssc-graph show --ref ghcr.io/toddysm/golden/python:3.14-slim -d "$DB" @@ -171,6 +254,27 @@ occurrence: ghcr.io/toddysm/golden/python@sha256:2222…2222 tag: 3.14-slim @ 2026-08-09T13:10:05Z ``` +#### Deployment observations for an image + +`deployments` returns every retained observation with its environment, chart, +timestamp, run link, complete `source` evidence, and stable `eventId`. `show --format json` includes the same +history in `deployments` and the inferred intervals in `tags`. + +```bash +cssc-graph deployments \ + --ref ghcr.io/toddysm/apps/cssc-dashboard/packages-service -d "$DB" --format json +``` + +Two observations of the same image in the same environment remain distinct when +their timestamps, chart or source evidence differ. Replaying an identical +record does not duplicate it. The deployment `eventId` hashes the complete record +except its optional `id`, including time and source evidence; it is deliberately +different from the semantic content ID used for persisted source facts. +Repository history is fetched in one graph projection after seed resolution, +not one database call per occurrence. Queries retain the service's read guard +to protect against concurrent rebuilds; cost still scales with returned history, +and this endpoint is not a paginated or streaming inventory API. + #### 3. Given a built image — direct and transitive base images `bases` follows `BUILT_FROM` up the chain (and each base's own path is reachable @@ -345,6 +449,7 @@ parameters. `depth` is clamped to the service's `MAX_DEPTH`. | Full supply-chain path | `path` | `GET /artifacts/path?ref=…` (or `?digest=…`) | | Current tag target | `show --ref repo:tag` | `GET /artifacts/show?ref=repo:tag` | | Tag history | `tag-history` | `GET /repositories/tags/history?ref=…&tag=…` | +| Deployment observations | `deployments --ref` | `GET /artifacts/deployments?ref=…` | | Direct + transitive bases | `bases` | `GET /artifacts/bases?ref=…` | | Downstream derived images | `derived` | `GET /artifacts/derived?base=…` | | Find by annotation/type/repo | `find` | `GET /search?annotation=…` / `?type=…` / `?ref=…` |