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 0a18ff7..743c8b7 100644 --- a/apps/python-app/libs/cssc_graph/cssc_graph/queries.py +++ b/apps/python-app/libs/cssc_graph/cssc_graph/queries.py @@ -380,6 +380,178 @@ def show(store: GraphStore, ref: str) -> dict[str, Any] | None: } +# -- provenance timeline ------------------------------------------------------ + +_OCC_ROW = ( + "{n}.key AS key, {n}.registry AS registry, {n}.repository AS repository, " + "{n}.digest AS digest, {n}.ref AS ref, {n}.deletedAt AS deletedAt, " + "{n}.deleteReason AS deleteReason" +) + + +def _family_basename(repository: str) -> str: + """The trailing image name of a repository (``.../golden/python`` -> ``python``).""" + + return repository.rsplit("/", 1)[-1] + + +def _provenance_state(repository: str, deleted_at: Any) -> str: + if deleted_at: + return "deleted" + if "quarantine/" in repository or repository.endswith("/quarantine"): + return "present-quarantine" + if "golden/" in repository or repository.endswith("/golden"): + return "present-golden" + return "present" + + +def _image_node(row: dict[str, Any]) -> dict[str, Any]: + return { + "key": row["key"], + "registry": row["registry"], + "repository": row["repository"], + "digest": row["digest"], + "ref": row["ref"], + "deletedAt": row.get("deletedAt"), + "deleteReason": row.get("deleteReason"), + "nodeType": "image", + "state": _provenance_state(row["repository"], row.get("deletedAt")), + } + + +def provenance(store: GraphStore, family: str) -> dict[str, Any]: + """Per-family provenance timeline: typed nodes and dated edges. + + Scopes to one repository *family* — the trailing image name, e.g. ``python`` + for both ``.../quarantine/python`` and ``.../golden/python``. Returns + ``{nodes, edges}`` where each node carries a ``nodeType`` (``tag-root`` | + ``image`` | ``referrer``) and a ``state`` (``upstream`` | + ``present-quarantine`` | ``present-golden`` | ``present`` | ``deleted``), and + each edge a ``type`` (``imported`` | ``promoted`` | ``built`` | ``attests``) + with a ``date``. + + The upstream tag root is synthesised from ``ArtifactMirrored`` (the mirror + source), ``sha256-*`` fallback tags are skipped, and referrers are labelled + by their raw ``artifactType``. + """ + + all_occ = store.query( + "MATCH (o:Occurrence) RETURN " + _OCC_ROW.format(n="o") + ) + family_occ = [o for o in all_occ if _family_basename(o["repository"]) == family] + + nodes: dict[str, dict[str, Any]] = {} + edges: list[dict[str, Any]] = [] + seen_edges: set[tuple[Any, ...]] = set() + + def add_edge(kind: str, frm: str, to: str, date: Any, **extra: Any) -> None: + ekey = (kind, frm, to, date, extra.get("artifactType"), extra.get("platform")) + if ekey in seen_edges: + return + seen_edges.add(ekey) + edge = {"type": kind, "from": frm, "to": to, "date": date} + edge.update({k: v for k, v in extra.items() if v}) + edges.append(edge) + + # 1. imported: synthesise the upstream tag root from each mirror source. + upstream_keys: set[str] = set() + for occ in family_occ: + for r in store.query( + "MATCH (q:Occurrence {key: $k})-[e:MIRRORED_FROM]->(up:Occurrence) " + "RETURN up.key AS key, up.ref AS ref, up.digest AS digest, " + "e.tag AS tag, e.recordedAt AS recordedAt", + {"k": occ["key"]}, + ): + upstream_keys.add(r["key"]) + tag = r.get("tag") or "" + if tag.startswith("sha256-"): # skip OCI referrer fallback tags + continue + root_key = f"{r['ref']}:{tag}" if tag else r["ref"] + nodes.setdefault( + root_key, + { + "key": root_key, + "ref": r["ref"], + "digest": r.get("digest"), + "tag": tag, + "nodeType": "tag-root", + "state": "upstream", + }, + ) + add_edge("imported", root_key, occ["key"], r.get("recordedAt"), tag=tag) + + # 2. image nodes, excluding pure upstream sources folded into tag roots. + for occ in family_occ: + if occ["key"] in upstream_keys: + continue + nodes.setdefault(occ["key"], _image_node(occ)) + image_keys = [k for k, n in nodes.items() if n["nodeType"] == "image"] + + # 3. promoted: quarantine -> golden (the edge is stored golden -> quarantine). + for key in image_keys: + for r in store.query( + "MATCH (g:Occurrence {key: $k})-[e:PROMOTED_FROM]->(q:Occurrence) " + "RETURN " + _OCC_ROW.format(n="q") + ", e.tag AS tag, e.recordedAt AS recordedAt", + {"k": key}, + ): + nodes.setdefault(r["key"], _image_node(r)) + add_edge("promoted", r["key"], key, r.get("recordedAt"), tag=r.get("tag")) + + # 4. built: always base -> image, whether the family occ is the image or the base. + for key in image_keys: + for r in store.query( + "MATCH (i:Occurrence {key: $k})-[e:BUILT_FROM]->(b:Occurrence) " + "RETURN " + _OCC_ROW.format(n="b") + ", e.tag AS tag, e.recordedAt AS recordedAt", + {"k": key}, + ): + nodes.setdefault(r["key"], _image_node(r)) + add_edge("built", r["key"], key, r.get("recordedAt"), tag=r.get("tag")) + for r in store.query( + "MATCH (i:Occurrence)-[e:BUILT_FROM]->(b:Occurrence {key: $k}) " + "RETURN " + _OCC_ROW.format(n="i") + ", e.tag AS tag, e.recordedAt AS recordedAt", + {"k": key}, + ): + nodes.setdefault(r["key"], _image_node(r)) + add_edge("built", key, r["key"], r.get("recordedAt"), tag=r.get("tag")) + + # 5. attests: referrers -> image, directly and rolled up from platform children. + for key in image_keys: + subjects = [(key, None)] + [ + (child["key"], _platform_label(child)) for child in _platform_children(store, key) + ] + for subj_key, platform in subjects: + for r in store.query( + "MATCH (ref:Occurrence)-[e:REFERS_TO]->(s:Occurrence {key: $k}) " + "RETURN " + _OCC_ROW.format(n="ref") + ", e.artifactType AS artifactType, " + "e.observedAt AS observedAt ORDER BY e.artifactType, ref.key", + {"k": subj_key}, + ): + # The source of a REFERS_TO is always a referrer artifact, even + # when it shares the family repository, so type it as such. + nodes[r["key"]] = { + "key": r["key"], + "registry": r["registry"], + "repository": r["repository"], + "digest": r["digest"], + "ref": r["ref"], + "deletedAt": r.get("deletedAt"), + "deleteReason": r.get("deleteReason"), + "nodeType": "referrer", + "state": "deleted" if r.get("deletedAt") else "present", + "artifactType": r.get("artifactType"), + } + add_edge( + "attests", + r["key"], + key, + r.get("observedAt"), + artifactType=r.get("artifactType"), + platform=platform, + ) + + return {"nodes": list(nodes.values()), "edges": edges} + + # -- export ------------------------------------------------------------------- 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 b446777..12d69d1 100644 --- a/apps/python-app/libs/cssc_graph/tests/test_queries.py +++ b/apps/python-app/libs/cssc_graph/tests/test_queries.py @@ -204,3 +204,39 @@ def test_cli_referrers_no_rollup(tmp_path: Path): assert result.exit_code == 0, result.output # --no-rollup keeps child referrers off the index, so no platform-tagged edge. assert "{linux/amd64}" not in result.output + + +def test_provenance_family_has_typed_nodes_and_dated_edges(store): + sub = queries.provenance(store, "python") + node_types = {n["nodeType"] for n in sub["nodes"]} + assert {"tag-root", "image", "referrer"} <= node_types + edge_kinds = {e["type"] for e in sub["edges"]} + assert {"imported", "promoted", "built", "attests"} <= edge_kinds + # every edge is dated + assert all("date" in e for e in sub["edges"]) + + +def test_provenance_states_and_upstream_tag_root(store): + sub = queries.provenance(store, "python") + by_state: dict[str, list[dict]] = {} + for n in sub["nodes"]: + by_state.setdefault(n["state"], []).append(n) + assert by_state.get("present-quarantine"), "expected a quarantine image node" + assert by_state.get("present-golden"), "expected a golden image node" + # The upstream source is folded into a synthetic tag-root, not an image node. + roots = [n for n in sub["nodes"] if n["nodeType"] == "tag-root"] + assert any(r["ref"] == "docker.io/library/python" for r in roots) + image_refs = {n["ref"] for n in sub["nodes"] if n["nodeType"] == "image"} + assert "docker.io/library/python" not in image_refs + + +def test_provenance_referrer_labelled_by_artifact_type(store): + sub = queries.provenance(store, "python") + attests = [e for e in sub["edges"] if e["type"] == "attests"] + assert any(e.get("artifactType") == "application/vnd.in-toto+json" for e in attests) + + +def test_provenance_unknown_family_is_empty(store): + sub = queries.provenance(store, "does-not-exist") + assert sub == {"nodes": [], "edges": []} +