export_image_dependencies.py

#!/usr/bin/env python3
"""Export manifest-cube components related to one image name.

Standard library only. The output describes relationships recorded by the
database; it does not infer a Python/package dependency graph.
"""

import argparse
import csv
import sqlite3
import sys
from pathlib import Path


FIELDS = [
    "build_id", "image_name", "image_version", "stream_name", "stream_module",
    "manifest_file", "sbom_created_at", "canonical_purl", "pull_spec",
    "build_component_id", "component_id", "type", "name", "component_version", "arch", "dev",
    "source_rpm_name", "purl", "attribution", "origin_image_id",
    "origin_image_name", "origin_image_digest", "origin_image_purl", "origin_ancestor_id",
    "origin_relationship", "origin_chain_depth", "relationship_basis",
    "dependency_scope", "scope_basis",
    "image_source_repositories", "image_source_hashes",
    "component_source_repositories", "component_source_hashes",
]


def source_values(sources, key):
    return "; ".join(sorted({item[key] for item in sources if item[key]}))


def sources_for_build(db, build_id):
    rows = db.execute(
        """
        SELECT s.source, s.hash
        FROM build_sources bs JOIN sources s ON s.source_id = bs.source_id
        WHERE bs.build_id = ? ORDER BY s.source, s.hash
        """, (build_id,)
    )
    return [dict(row) for row in rows]


def ancestors_for_build(db, build_id):
    rows = db.execute(
        """
        SELECT ba.ancestor_id, ba.relationship, ba.chain_depth, ai.image_id, ai.image_name,
               ai.image_digest, ai.image_purl, ai.source_module
        FROM build_ancestors ba JOIN ancestor_images ai ON ai.image_id = ba.image_id
        WHERE ba.build_id = ? ORDER BY ba.chain_depth, ba.relationship, ai.image_name
        """, (build_id,)
    )
    return [dict(row) for row in rows]


def sources_for_components(db, component_ids):
    result = {}
    # Keep below SQLite's usual maximum number of bound parameters.
    for start in range(0, len(component_ids), 900):
        batch = component_ids[start:start + 900]
        marks = ",".join("?" for _ in batch)
        rows = db.execute(
            f"""
            SELECT sc.component_id, s.source, s.hash
            FROM source_components sc JOIN sources s ON s.source_id = sc.source_id
            WHERE sc.component_id IN ({marks})
            ORDER BY sc.component_id, s.source, s.hash
            """, batch
        )
        for row in rows:
            result.setdefault(row["component_id"], []).append(
                {"source": row["source"], "hash": row["hash"]}
            )
    return result


def find_builds(db, image_name, image_version=None):
    query = """
        SELECT b.*, st.stream_name, st.module AS stream_module
        FROM builds b LEFT JOIN streams st ON st.stream_id = b.stream_id
        WHERE b.name = ? OR b.name LIKE '%/' || ?
    """
    args = [image_name, image_name]
    if image_version:
        query += " AND b.version = ?"
        args.append(image_version)
    query += " ORDER BY b.build_id"
    return list(db.execute(query, args))


def export(db, builds, output):
    count = 0
    with output.open("w", newline="", encoding="utf-8") as csv_file:
        writer = csv.DictWriter(csv_file, fieldnames=FIELDS)
        writer.writeheader()

        for build in builds:
            ancestors = ancestors_for_build(db, build["build_id"])
            ancestors_by_id = {item["image_id"]: item for item in ancestors}
            image_sources = sources_for_build(db, build["build_id"])
            components = list(db.execute(
                """
                SELECT bc.build_component_id, bc.origin_image_id, c.*
                FROM build_components bc JOIN components c
                  ON c.component_id = bc.component_id
                WHERE bc.build_id = ?
                ORDER BY c.type, c.name, c.version, c.component_id
                """, (build["build_id"],)
            ))
            component_sources = sources_for_components(
                db, [row["component_id"] for row in components]
            )

            for component in components:
                origin = ancestors_by_id.get(component["origin_image_id"])
                if origin:
                    attribution = "ancestor-linked"
                    basis = (
                        f"linked through ancestor image {origin['image_name']} "
                        f"({origin['relationship']}, chain depth {origin['chain_depth']})"
                    )
                    if origin["relationship"] == "builder":
                        scope = "build"
                        scope_basis = "inherited from a builder ancestor image"
                    elif origin["relationship"] == "parent":
                        scope = "runtime"
                        scope_basis = "inherited from a parent ancestor image"
                    else:
                        scope = "undetermined"
                        scope_basis = (
                            "ancestor relationship is recorded, but its build/runtime "
                            "scope is not defined"
                        )
                else:
                    attribution = "direct"
                    basis = "directly linked through build_components"
                    if component["dev"]:
                        scope = "build"
                        scope_basis = "component dev flag is set"
                    elif component["type"] == "github":
                        scope = "build"
                        scope_basis = "GitHub component is recorded as build metadata"
                    else:
                        scope = "undetermined"
                        scope_basis = (
                            "direct component membership is recorded, but the database "
                            "does not define whether it is build-time or runtime"
                        )

                writer.writerow({
                    "build_id": build["build_id"],
                    "image_name": build["name"],
                    "image_version": build["version"],
                    "stream_name": build["stream_name"],
                    "stream_module": build["stream_module"],
                    "manifest_file": build["manifest_file"],
                    "sbom_created_at": build["sbom_created_at"],
                    "canonical_purl": build["canonical_purl"],
                    "pull_spec": build["pull_spec"],
                    "build_component_id": component["build_component_id"],
                    "component_id": component["component_id"],
                    "type": component["type"],
                    "name": component["name"],
                    "component_version": component["version"],
                    "arch": component["arch"],
                    "dev": component["dev"],
                    "source_rpm_name": component["source_rpm_name"],
                    "purl": component["purl"],
                    "attribution": attribution,
                    "origin_image_id": component["origin_image_id"],
                    "origin_image_name": origin["image_name"] if origin else None,
                    "origin_image_digest": origin["image_digest"] if origin else None,
                    "origin_image_purl": origin["image_purl"] if origin else None,
                    "origin_ancestor_id": origin["ancestor_id"] if origin else None,
                    "origin_relationship": origin["relationship"] if origin else None,
                    "origin_chain_depth": origin["chain_depth"] if origin else None,
                    "relationship_basis": basis,
                    "dependency_scope": scope,
                    "scope_basis": scope_basis,
                    "image_source_repositories": source_values(image_sources, "source"),
                    "image_source_hashes": source_values(image_sources, "hash"),
                    "component_source_repositories": source_values(
                        component_sources.get(component["component_id"], []), "source"
                    ),
                    "component_source_hashes": source_values(
                        component_sources.get(component["component_id"], []), "hash"
                    ),
                })
                count += 1
    return count


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("database", type=Path)
    parser.add_argument("image_name")
    parser.add_argument("-o", "--output", type=Path)
    parser.add_argument("--build-version")
    args = parser.parse_args()

    if not args.database.is_file():
        print(f"database not found: {args.database}", file=sys.stderr)
        return 2
    if args.output:
        output = args.output
    else:
        image_basename = args.image_name.rsplit("/", 1)[-1]
        version_suffix = f"-{args.build_version}" if args.build_version else ""
        output = Path(f"{image_basename}{version_suffix}-dependencies.csv")

    db = sqlite3.connect(f"file:{args.database}?mode=ro", uri=True)
    db.row_factory = sqlite3.Row
    try:
        builds = find_builds(db, args.image_name, args.build_version)
        if not builds:
            print(f"no image builds matched: {args.image_name}", file=sys.stderr)
            return 1
        count = export(db, builds, output)
    finally:
        db.close()

    print(f"wrote {count} component records from {len(builds)} build(s) to {output}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())