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 6654c04..9f5afb9 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/cli.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/cli.py @@ -181,9 +181,10 @@ def _emit_subgraph(subgraph: dict, output_format: str) -> None: click.echo(f"{len(subgraph['nodes'])} node(s), {len(subgraph['edges'])} edge(s):") for edge in subgraph["edges"]: tag = f" ({edge['tag']})" if edge.get("tag") else "" + atype = f" [{edge['artifactType']}]" if edge.get("artifactType") else "" frm = refs.get(edge["from"], edge["from"]) to = refs.get(edge["to"], edge["to"]) - click.echo(f" {frm} --{edge['type']}--> {to}{tag}") + click.echo(f" {frm} --{edge['type']}--> {to}{tag}{atype}") @cli.command() @@ -266,6 +267,27 @@ def derived(database: Path, base: str, depth: int, output_format: str) -> None: _emit_subgraph(subgraph, output_format) +@cli.command() +@_database_option +@click.option("--digest", help="Seed by artifact digest (sha256:...).") +@click.option("--ref", help="Seed by registry/repository[@digest|:tag].") +@click.option("--depth", default=3, show_default=True, help="Max referrer depth (referrers-of-referrers).") +@_format_option +def referrers(database: Path, digest: str | None, ref: str | None, depth: int, output_format: str) -> None: + """List the referrer artifacts (SBOM/provenance/VEX/signatures) attached to an image.""" + + if not digest and not ref: + raise click.UsageError("provide --digest or --ref") + from . import queries + + store = _open_store(database) + try: + subgraph = queries.referrers(store, digest=digest, ref=ref, depth=depth) + finally: + store.close() + _emit_subgraph(subgraph, output_format) + + @cli.command() @_database_option @click.option("--annotation", help="Filter by annotation name=value.") @@ -319,6 +341,9 @@ def show(database: Path, ref: str, output_format: str) -> None: return occ = data["occurrence"] click.echo(f"occurrence: {occ['key']}") + if data.get("deleted"): + reason = occ.get("deleteReason") or "unknown" + click.echo(f" deleted: {occ.get('deletedAt')} (reason: {reason})") if data["artifact"]: art = data["artifact"] click.echo(f" type: {art.get('artifactType') or art.get('mediaType') or '-'}") @@ -326,6 +351,8 @@ def show(database: Path, ref: str, output_format: str) -> None: click.echo(f" annotation: {ann['name']}={ann['value']}") for tag in data["tags"]: click.echo(f" tag: {tag['tag']} @ {tag['observedAt']}") + for r in data.get("referrers", []): + click.echo(f" referrer: {r.get('artifactType') or '-'} ({r['from']})") @cli.command() 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 5c8f681..d2ca08a 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py @@ -17,7 +17,8 @@ _OCC_RETURN = ( "n.key AS key, n.registry AS registry, n.repository AS repository, " - "n.digest AS digest, n.ref AS ref" + "n.digest AS digest, n.ref AS ref, n.deletedAt AS deletedAt, " + "n.deleteReason AS deleteReason" ) @@ -25,7 +26,8 @@ def get_occurrence(store: GraphStore, key: str) -> dict[str, Any] | None: rows = store.query( "MATCH (o:Occurrence {key: $k}) " "RETURN o.key AS key, o.registry AS registry, o.repository AS repository, " - "o.digest AS digest, o.ref AS ref", + "o.digest AS digest, o.ref AS ref, o.deletedAt AS deletedAt, " + "o.deleteReason AS deleteReason", {"k": key}, ) return rows[0] if rows else None @@ -136,6 +138,8 @@ def _record_node(nodes: dict[str, dict[str, Any]], row: dict[str, Any]) -> None: "repository": row["repository"], "digest": row["digest"], "ref": row["ref"], + "deletedAt": row.get("deletedAt"), + "deleteReason": row.get("deleteReason"), } @@ -168,6 +172,72 @@ def derived(store: GraphStore, base: str, depth: int = 10) -> dict[str, Any]: return traverse(store, seeds, ("BUILT_FROM",), direction="in", max_depth=depth) +def referrers( + store: GraphStore, + *, + digest: str | None = None, + ref: str | None = None, + depth: int = 3, +) -> dict[str, Any]: + """Referrer artifacts of a subject occurrence. + + Returns ``{nodes, edges}`` where each ``REFERS_TO`` edge carries the + ``artifactType``. Follows referrers-of-referrers up to *depth* levels (a + signature on an SBOM, etc.); ``depth=0`` returns just the subject, matching + :func:`traverse`. + """ + + seeds = resolve_seed(store, digest=digest, ref=ref) + nodes: dict[str, dict[str, Any]] = {} + edges: list[dict[str, Any]] = [] + # Keyed like the indexer's REFERS_TO merge so distinct observations and + # artifact types between the same pair are all kept, not collapsed. + seen_edges: set[tuple[str, str, str | None, str | None]] = set() + for seed in seeds: + occ = get_occurrence(store, seed) + if occ: + nodes[seed] = occ + + visited: set[str] = set() + frontier = list(dict.fromkeys(seeds)) + for _ in range(depth): + nxt: list[str] = [] + for key in frontier: + if key in visited: + continue + visited.add(key) + rows = store.query( + "MATCH (r:Occurrence)-[e:REFERS_TO]->(s:Occurrence {key: $k}) " + "RETURN r.key AS key, r.registry AS registry, r.repository AS repository, " + "r.digest AS digest, r.ref AS ref, r.deletedAt AS deletedAt, " + "r.deleteReason AS deleteReason, e.artifactType AS artifactType, " + "e.observedAt AS observedAt ORDER BY e.artifactType, r.key", + {"k": key}, + ) + for row in rows: + _record_node(nodes, row) + atype = row.get("artifactType") + observed = row.get("observedAt") + ekey = (row["key"], key, atype, observed) + if ekey not in seen_edges: + seen_edges.add(ekey) + edges.append( + { + "type": "REFERS_TO", + "from": row["key"], + "to": key, + "artifactType": atype, + "observedAt": observed, + } + ) + nxt.append(row["key"]) + frontier = [k for k in nxt if k not in visited] + if not frontier: + break + + return {"nodes": list(nodes.values()), "edges": edges} + + def tag_history(store: GraphStore, ref: str, tag: str) -> list[dict[str, Any]]: registry, repository = split_ref(ref) return store.query( @@ -238,6 +308,8 @@ def show(store: GraphStore, ref: str) -> dict[str, Any] | None: "artifact": artifact[0] if artifact else None, "annotations": annotations, "tags": tags, + "referrers": referrers(store, ref=key)["edges"], + "deleted": bool(occ.get("deletedAt")), "path": traverse(store, [key], PATH_RELS, direction="both", max_depth=2), } @@ -245,10 +317,22 @@ def show(store: GraphStore, ref: str) -> dict[str, Any] | None: # -- export ------------------------------------------------------------------- +def _node_label(node: dict[str, Any]) -> str: + """A label that disambiguates occurrences of the same repository by digest.""" + ref = node.get("ref") or node["key"] + digest = node.get("digest") or "" + short = digest[7:19] if digest.startswith("sha256:") else digest[:12] + label = f"{ref}@{short}" if short else str(ref) + if node.get("deletedAt"): + label += " (deleted)" + return label + + def to_cytoscape(subgraph: dict[str, Any]) -> dict[str, Any]: - elements = [{"data": {"id": n["key"], "label": n.get("ref", n["key"]), **n}} for n in subgraph["nodes"]] + elements = [{"data": {"id": n["key"], "label": _node_label(n), **n}} for n in subgraph["nodes"]] for i, e in enumerate(subgraph["edges"]): - elements.append({"data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": e["type"], **e}}) + label = e.get("artifactType") or e["type"] + elements.append({"data": {"id": f"e{i}", "source": e["from"], "target": e["to"], "label": label, **e}}) return {"elements": elements} @@ -261,10 +345,11 @@ def to_mermaid(subgraph: dict[str, Any]) -> str: lines = ["flowchart LR"] ids = {n["key"]: f"n{i}" for i, n in enumerate(subgraph["nodes"])} for n in subgraph["nodes"]: - label = _mermaid_label(n.get("ref", n["key"])) + label = _mermaid_label(_node_label(n)) lines.append(f' {ids[n["key"]]}["{label}"]') for e in subgraph["edges"]: frm, to = ids.get(e["from"]), ids.get(e["to"]) if frm and to: - lines.append(f' {frm} -->|{e["type"]}| {to}') + rel = _mermaid_label(e.get("artifactType") or e["type"]) + lines.append(f' {frm} -->|{rel}| {to}') return "\n".join(lines) diff --git a/apps/python-app/libs/cssc_graph/tests/test_queries.py b/apps/python-app/libs/cssc_graph/tests/test_queries.py index 4ea8cb0..c7da7ee 100644 --- a/apps/python-app/libs/cssc_graph/tests/test_queries.py +++ b/apps/python-app/libs/cssc_graph/tests/test_queries.py @@ -94,3 +94,54 @@ def test_traverse_includes_seed_at_depth_zero(store): def test_mermaid_label_escaping(): assert queries._mermaid_label('a"b\nc') == "a#quot;b c" + + +DIGEST3 = "sha256:" + "3" * 64 +QUAR_REF = "ghcr.io/toddysm/quarantine/python" + + +def test_referrers_returns_edge_with_artifact_type(store): + sub = queries.referrers(store, ref=f"{GOLDEN_REF}@{DIGEST2}") + assert "REFERS_TO" in {e["type"] for e in sub["edges"]} + assert "application/vnd.in-toto+json" in {e.get("artifactType") for e in sub["edges"]} + + +def test_referrers_depth_zero_returns_only_subject(store): + sub = queries.referrers(store, ref=f"{GOLDEN_REF}@{DIGEST2}", depth=0) + assert sub["edges"] == [] + assert {n["key"] for n in sub["nodes"]} == {f"{GOLDEN_REF}@{DIGEST2}"} + + +def test_show_includes_referrers(store): + data = queries.show(store, f"{GOLDEN_REF}@{DIGEST2}") + assert data is not None + assert data["deleted"] is False + assert any(r.get("artifactType") == "application/vnd.in-toto+json" for r in data["referrers"]) + + +def test_show_marks_deleted_occurrence(store): + data = queries.show(store, f"{QUAR_REF}@{DIGEST3}") + assert data is not None + assert data["deleted"] is True + assert data["occurrence"]["deleteReason"] == "promoted" + + +def test_node_label_disambiguates_by_digest_and_marks_deleted(): + assert queries._node_label({"ref": GOLDEN_REF, "digest": DIGEST2}) == f"{GOLDEN_REF}@222222222222" + lbl = queries._node_label({"ref": QUAR_REF, "digest": DIGEST3, "deletedAt": "2026-08-09T13:10:10Z"}) + assert lbl.endswith("(deleted)") + + +def test_cli_referrers(tmp_path: Path): + from click.testing import CliRunner + + from cssc_graph.cli import cli + + db = tmp_path / "db" + with GraphStore(db) as gs: + gs.init_schema() + index_data(gs, EXAMPLES, SCHEMA_DIR) + result = CliRunner().invoke(cli, ["referrers", "-d", str(db), "--ref", f"{GOLDEN_REF}@{DIGEST2}"]) + assert result.exit_code == 0, result.output + assert "REFERS_TO" in result.output + assert "application/vnd.in-toto+json" in result.output