#!/usr/bin/env python3 """ Generate in-toto link files, linking git tag → orig tarball. For each buildinfo already ingested in astra.db: - Finds the buildinfo file on disk via the local buildinfos index - Parses the orig tarball name and sha256 from the buildinfo file - Finds the matching git tag from the local clone using version normalization - Writes a .link.json with material = git_tag (+ commit hash), product = orig_tarball The material key uses the remote URL of the local clone (stripped of scheme/credentials/.git) so it matches the gitcommit artifacts produced by `astra ingest git`. Usage: python3 scripts/gen_intoto_links.py \\ --astra-db /data/astra.db \\ --buildinfos-dir /data/buildinfos \\ --repos-dir /repos \\ --out /data/links/ Then ingest: for f in /data/links/*.link.json; do astra ingest intoto --linker debtrace "$f" done """ import argparse import json import os import re import signal import sqlite3 import subprocess import sys signal.signal(signal.SIGPIPE, signal.SIG_DFL) ORIG_RE = re.compile( r'^\s+([0-9a-f]{64})\s+\d+\s+(\S+\.orig\.tar\.\w+)\s*$' ) _AMD64_SUFFIXES = { "amd64", "amd64-buildd", "amd64+buildd", "amd64+fix", "amd64+workaround", "buildd-amd64", "amd64source", } _ALL_SUFFIXES = {"all", "all-buildd"} _ACCEPTED_SUFFIXES = _AMD64_SUFFIXES | _ALL_SUFFIXES _BINNMU_RE = re.compile(r"\+b\d+$") _BPO_RE = re.compile(r"~bpo\d+\+\d+$") # --- tag matching (same logic as ingest_repos.py) --- def strip_epoch(version: str) -> str: idx = version.find(":") return version[idx + 1:] if idx != -1 else version def strip_binnmu(version: str) -> str: return _BINNMU_RE.sub("", version) def strip_bpo(version: str) -> str: return _BPO_RE.sub("", version) def upstream_from_debian_version(version: str | None) -> str | None: if not version: return None v = strip_epoch(version) if "-" in v: return v.rsplit("-", 1)[0] return v def version_candidates(*versions: str | None) -> list[str]: candidates: list[str] = [] for version in versions: if not version: continue v = strip_epoch(version.strip()) variants = [v, strip_binnmu(v), strip_bpo(v), strip_bpo(strip_binnmu(v))] for var in variants: if var and var not in candidates: candidates.append(var) return candidates def trailing_separator_variants(value: str) -> list[str]: variants: list[str] = [] v = value.strip() if v.startswith("refs/tags/"): v = v[len("refs/tags/"):] while v.endswith("^{}") or v.endswith("{}"): v = v[:-3] if v.endswith("^{}") else v[:-2] v = v.strip() if v: variants.append(v) if "/" in v: last = v.rsplit("/", 1)[-1] if last and last not in variants: variants.append(last) for sep in ("+", "~", "_", "-"): if sep in v: base = v.split(sep, 1)[0] if base and base not in variants: variants.append(base) return variants def separator_prefix_match(a: str, b: str) -> bool: if not a or not b: return False if a == b: return True if a.startswith(b): return len(a) == len(b) or a[len(b)] in "+-._~/" if b.startswith(a): return len(b) == len(a) or b[len(a)] in "+-._~/" return False def find_matching_tag(tags: list[str], candidates: list[str]) -> tuple[str, str] | tuple[None, None]: for candidate in candidates: candidate_variants = trailing_separator_variants(candidate) hits: list[str] = [] for tag in tags: tag_variants = trailing_separator_variants(tag) matched = any( cv == tv or cv in tv or tv in cv or separator_prefix_match(cv, tv) for cv in candidate_variants for tv in tag_variants ) if matched: hits.append(tag) if hits: hits.sort(key=lambda x: (len(x), x)) return hits[0], candidate return None, None # --- git helpers --- def get_remote_tags(repo_path: str) -> tuple[list[str], dict[str, str]]: """Returns (tag_names, {tag_name: commit_hash}). For annotated tags the commit hash comes from the ^{} dereferenced line (the actual commit, not the tag object). For lightweight tags it's the same line. """ try: cp = subprocess.run( ["git", "-C", repo_path, "ls-remote", "--tags"], capture_output=True, text=True, timeout=120, ) except subprocess.TimeoutExpired: print(f" [timeout] ls-remote timed out for {repo_path}", file=sys.stderr) return [], {} if cp.returncode != 0: return [], {} raw: dict[str, str] = {} # tag_name -> commit (lightweight or tag object) deref: dict[str, str] = {} # tag_name -> commit (annotated tag dereferenced) for line in cp.stdout.splitlines(): parts = line.strip().split() if len(parts) != 2: continue commit, ref = parts if ref.endswith("^{}"): tag_name = ref.removeprefix("refs/tags/")[:-3] deref[tag_name] = commit else: tag_name = ref.removeprefix("refs/tags/") if tag_name: raw[tag_name] = commit tag_to_commit = {**raw, **deref} # deref wins for annotated tags return list(raw.keys()), tag_to_commit def get_remote_url(clone_path: str) -> str | None: result = subprocess.run( ["git", "-C", clone_path, "remote", "get-url", "origin"], capture_output=True, text=True, timeout=10, ) if result.returncode != 0: return None url = result.stdout.strip() if not url: return None from urllib.parse import urlparse, urlunparse parsed = urlparse(url) if parsed.username or parsed.password: url = urlunparse(parsed._replace(netloc=parsed.hostname + (f":{parsed.port}" if parsed.port else ""))) return url or None # --- buildinfo index (same logic as ingest_from_debtrace.py) --- def build_index(buildinfos_dir: str) -> dict[tuple[str, str], str]: index: dict[tuple[str, str], str] = {} priority: dict[tuple[str, str], int] = {} for root, dirs, files in os.walk(buildinfos_dir): dirs.sort() for fname in files: if not fname.endswith(".buildinfo"): continue base = fname[:-len(".buildinfo")] parts = base.split("_", 2) if len(parts) != 3: continue source, version, arch_suffix = parts if arch_suffix not in _ACCEPTED_SUFFIXES: continue prio = 0 if arch_suffix in _AMD64_SUFFIXES else 1 key = (source, version) if key not in index or prio < priority.get(key, 999): index[key] = os.path.join(root, fname) priority[key] = prio return index def find_buildinfo(index: dict, source_name: str, version: str) -> str | None: ver = version.split(":", 1)[1] if ":" in version else version return index.get((source_name, ver)) # --- buildinfo / link helpers --- def parse_orig_tarball(buildinfo_path: str) -> tuple[str, str] | tuple[None, None]: in_checksums = False try: with open(buildinfo_path, errors="replace") as f: for line in f: if line.startswith("Checksums-Sha256:"): in_checksums = True continue if in_checksums: if line.startswith((" ", "\t")): m = ORIG_RE.match(line) if m: return m.group(2), m.group(1) else: break except OSError as e: print(f" [warn] cannot read {buildinfo_path}: {e}", file=sys.stderr) return None, None def repo_to_path(url: str) -> str: path = url.removeprefix("https://").removeprefix("http://") if path.endswith(".git"): path = path[:-4] return path def make_link(source_name: str, git_tag: str, commit_hash: str, tarball_name: str, tarball_sha256: str, remote_url: str) -> dict: material_key = f"{repo_to_path(remote_url)}@{git_tag}" return { "signed": { "_type": "link", "name": source_name, "materials": { material_key: {"sha256": commit_hash}, }, "products": { tarball_name: {"sha256": tarball_sha256}, }, "command": [], "environment": {}, }, "signatures": [], } def safe_filename(source_name: str, version: str) -> str: return f"{source_name}_{version.replace(':', '+')}.link.json" _BUILDINFO_RE = re.compile(r'^artifact:buildinfo:deb/([^@]+)@([^?]+)') # --- DB query --- def fetch_rows(astra_db: str) -> list[tuple[str, str]]: """Return (source_name, version) for every buildinfo ingested in astra.db.""" con = sqlite3.connect(astra_db) cur = con.cursor() cur.execute("SELECT DISTINCT astra_id FROM artifacts WHERE kind = 'buildinfo'") results = [] for (astra_id,) in cur: m = _BUILDINFO_RE.match(astra_id) if m: results.append((m.group(1), m.group(2))) con.close() return results def main(): ap = argparse.ArgumentParser() ap.add_argument("--astra-db", default="/data/astra.db", help="path to astra.db") ap.add_argument("--buildinfos-dir", required=True, help="path to buildinfos root") ap.add_argument("--repos-dir", required=True, help="path to local git clones (one subdir per source)") ap.add_argument("--out", required=True, help="output directory for .link.json files") ap.add_argument("--source", help="filter to one source package") args = ap.parse_args() os.makedirs(args.out, exist_ok=True) print("[index] scanning buildinfos directory...", file=sys.stderr) index = build_index(args.buildinfos_dir) print(f"[index] {len(index)} buildinfos indexed", file=sys.stderr) rows = fetch_rows(args.astra_db) if args.source: rows = [(s, v) for s, v in rows if s == args.source] print(f"[astra] {len(rows)} buildinfo artifacts", file=sys.stderr) # Group by source_name so we call ls-remote once per repo by_source: dict[str, list[str]] = {} for source_name, version in rows: by_source.setdefault(source_name, []).append(version) written = skipped = no_clone = no_tag = 0 for source_name, versions in by_source.items(): clone_path = os.path.join(args.repos_dir, source_name) if not os.path.isdir(clone_path): no_clone += len(versions) continue url = get_remote_url(clone_path) if url is None: no_clone += len(versions) continue tags, tag_to_commit = get_remote_tags(clone_path) for version in versions: candidates = version_candidates(version, upstream_from_debian_version(version)) tag, _ = find_matching_tag(tags, candidates) if tag is None: print(f" [no-tag] {source_name} {version}", file=sys.stderr) no_tag += 1 continue commit_hash = tag_to_commit.get(tag, "") buildinfo_path = find_buildinfo(index, source_name, version) if buildinfo_path is None: skipped += 1 continue tarball_name, tarball_sha256 = parse_orig_tarball(buildinfo_path) if tarball_name is None: print(f" [no-tarball] {source_name} {version}", file=sys.stderr) skipped += 1 continue link = make_link(source_name, tag, commit_hash, tarball_name, tarball_sha256, url) fpath = os.path.join(args.out, safe_filename(source_name, version)) with open(fpath, "w") as f: json.dump(link, f, indent=2) written += 1 print( f"Done: {written} written, {skipped} no tarball, {no_tag} no tag matched, {no_clone} no clone", file=sys.stderr, ) if __name__ == "__main__": main()