initial commit

ci.py

@@ -0,0 +1,451 @@
+import argparse
+import base64
+import datetime
+import difflib
+import json
+import os
+import re
+import subprocess
+import sys
+import urllib.request
+from collections import Counter
+from pathlib import Path
+
+
+def run(*args, **kw):
+    return subprocess.run(args, text=True, **kw)
+
+
+def nix(*args):
+    return run("nix", "--extra-experimental-features", "nix-command flakes", *args, check=True, capture_output=True).stdout
+
+
+def git(*args, check=True):
+    return run("git", *args, check=check, capture_output=True)
+
+
+def gitea(method, path, body=None):
+    url = f'{os.environ["API"]}/repos/{os.environ["REPO"]}{path}'
+    data = json.dumps(body).encode() if body is not None else None
+    req = urllib.request.Request(url, data=data, method=method)
+    req.add_header("Authorization", f'token {os.environ["GH_TOKEN"]}')
+    if data is not None:
+        req.add_header("Content-Type", "application/json")
+    with urllib.request.urlopen(req) as resp:
+        raw = resp.read()
+    return json.loads(raw) if raw else None
+
+
+def find_open_issue(label):
+    issues = gitea("GET", f"/issues?state=open&labels={label}")
+    return issues[0] if issues else None
+
+
+def build_hosts(ref):
+    hosts = json.loads(nix("eval", "--json", f"{ref}#nixosConfigurations", "--apply", "builtins.attrNames"))
+    return {
+        host: nix("build", f"{ref}#nixosConfigurations.{host}.config.system.build.toplevel",
+                  "--no-link", "--print-out-paths").strip()
+        for host in hosts
+    }
+
+
+def diff(args):
+    old, new = build_hosts(args.old), build_hosts(args.new)
+
+    def section(host):
+        if host not in old:
+            return f"## {host}\n\n_new host_"
+        if host not in new:
+            return f"## {host}\n\n_host removed_"
+        rendered = run("dix", "--color", "never", old[host], new[host], capture_output=True).stdout
+        return f"## {host}\n\n```\n{rendered}```"
+
+    parts = ([args.title] if args.title else []) + [section(h) for h in sorted(set(old) | set(new))]
+    print("\n\n".join(parts))
+
+
+# meta.toplevel is the closure path dix already covers; nftables.ruleset/tables
+# are duplicated by nftables.rendered (the assembled ruleset we keep and diff).
+# Pruned so inspect-diff reports only source config topology, once.
+INSPECT_PRUNE = [
+    ("meta", "toplevel"),
+    ("nftables", "ruleset"),
+    ("nftables", "tables"),
+]
+
+
+def compact(value):
+    return json.dumps(value, sort_keys=True)
+
+
+def scalar(value):
+    if value is None:
+        return "null"
+    if isinstance(value, bool):
+        return "true" if value else "false"
+    return str(value)
+
+
+def fmt(value, indent=0):
+    pad = "  " * indent
+    lines = []
+    if isinstance(value, dict):
+        for key in sorted(value):
+            child = value[key]
+            if isinstance(child, (dict, list)) and child:
+                lines.append(f"{pad}{key}:")
+                lines += fmt(child, indent + 1)
+            else:
+                lines.append(f"{pad}{key}: {scalar(child)}")
+    elif isinstance(value, list):
+        for item in value:
+            if isinstance(item, (dict, list)) and item:
+                body = fmt(item)
+                lines.append(f"{pad}- {body[0]}")
+                lines += [f"{pad}  {line}" for line in body[1:]]
+            else:
+                lines.append(f"{pad}- {scalar(item)}")
+    else:
+        lines.append(f"{pad}{scalar(value)}")
+    return lines
+
+
+STORE_HASH = re.compile(r"/nix/store/[a-z0-9]{32}-")
+
+
+def strip_store(value):
+    # closure hashes churn on any rebuild and are dix's job; drop them so exec
+    # commands / paths read as "<name>/bin/x" and diff only on real changes
+    if isinstance(value, dict):
+        return {k: strip_store(v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [strip_store(v) for v in value]
+    if isinstance(value, str):
+        return STORE_HASH.sub("", value)
+    return value
+
+
+def inspect_hosts(ref):
+    try:
+        names = json.loads(nix("eval", "--json", f"{ref}#inspect", "--apply", "builtins.attrNames"))
+    except subprocess.CalledProcessError as e:
+        if "does not provide attribute" in e.stderr:
+            return None
+        raise RuntimeError(f"inspect eval failed for {ref}:\n{e.stderr}") from e
+    return {
+        host: nix("build", f"{ref}#inspect.{host}",
+                  "--no-link", "--print-out-paths").strip()
+        for host in names
+    }
+
+
+def read_inspect(path):
+    with open(path) as f:
+        return json.load(f)
+
+
+def inspect_data(ref):
+    paths = inspect_hosts(ref)
+    if paths is None:
+        return None
+    reports = {host: read_inspect(path) for host, path in paths.items()}
+    for report in reports.values():
+        for parent, key in INSPECT_PRUNE:
+            report.get(parent, {}).pop(key, None)
+    return {host: strip_store(report) for host, report in reports.items()}
+
+
+def tree_changes(old, new, path=()):
+    if old == new:
+        return
+    if isinstance(old, dict) and isinstance(new, dict):
+        for key in sorted(old.keys() | new.keys()):
+            if key not in old:
+                yield path + (key,), "add", new[key]
+            elif key not in new:
+                yield path + (key,), "del", old[key]
+            else:
+                yield from tree_changes(old[key], new[key], path + (key,))
+    elif isinstance(old, list) and isinstance(new, list):
+        # order-independent: a reordered list is not a change
+        okeys, nkeys = list(map(compact, old)), list(map(compact, new))
+        values = dict(zip(okeys, old)) | dict(zip(nkeys, new))
+        for key in (Counter(okeys) - Counter(nkeys)).elements():
+            yield path, "delitem", values[key]
+        for key in (Counter(nkeys) - Counter(okeys)).elements():
+            yield path, "additem", values[key]
+    else:
+        yield path, "change", (old, new)
+
+
+def render_change(path, kind, payload):
+    label = ".".join(map(str, path[1:])) or path[0]
+    if kind in ("add", "del", "additem", "delitem"):
+        sign = "+" if kind in ("add", "additem") else "-"
+        value = [payload] if kind in ("additem", "delitem") else payload
+        if isinstance(value, (dict, list)) and value:
+            return [f"{sign} {label}:"] + [f"{sign} {line}" for line in fmt(value, 1)]
+        return [f"{sign} {label}: {scalar(value)}"]
+    old, new = payload
+    if isinstance(old, str) and isinstance(new, str) and "\n" in old + new:
+        body = difflib.unified_diff(old.splitlines(), new.splitlines(), lineterm="")
+        return [f"  {label}:"] + [ln for ln in body if not ln.startswith(("---", "+++"))]
+    if isinstance(old, (dict, list)) or isinstance(new, (dict, list)):
+        return render_change(path, "del", old) + render_change(path, "add", new)
+    return [f"- {label}: {scalar(old)}", f"+ {label}: {scalar(new)}"]
+
+
+def host_diff(old, new):
+    lines = []
+    attr = None
+    for path, kind, payload in tree_changes(old, new):
+        if path[0] != attr:
+            attr = path[0]
+            lines.append(f"@@ {attr} @@")
+        lines += render_change(path, kind, payload)
+    return "```diff\n" + "\n".join(lines) + "\n```" if lines else None
+
+
+def host_added(report):
+    return "```diff\n" + "\n".join(f"+ {line}" for line in fmt(report)) + "\n```"
+
+
+def inspect_diff(args):
+    old, new = inspect_data(args.old), inspect_data(args.new)
+    if old is None or new is None:
+        return
+
+    def block(host):
+        if host not in old:
+            return f"## {host} (new)\n\n{host_added(new[host])}"
+        if host not in new:
+            return f"## {host}\n\n_host removed_"
+        body = host_diff(old[host], new[host])
+        return f"## {host}\n\n{body}" if body else None
+
+    blocks = [b for b in map(block, sorted(set(old) | set(new))) if b]
+    if blocks:
+        print("\n\n".join(([args.title] if args.title else []) + blocks))
+
+
+def format_vulnix(vulns):
+    rows = []
+    for v in vulns:
+        for cve in v["affected_by"]:
+            score = v.get("cvssv3_basescore", {}).get(cve)
+            desc = v.get("description", {}).get(cve, "")
+            if len(desc) > 120:
+                desc = desc[:117] + "..."
+            rows.append((v["pname"], v["version"], cve, score, desc))
+    rows.sort(key=lambda r: -(r[3] or 0))
+
+    lines = [
+        "| Package | Version | CVE | CVSS | Description |",
+        "|---------|---------|-----|------|-------------|",
+    ]
+    for pname, version, cve, score, desc in rows:
+        score_str = f"{score:.1f}" if score else "?"
+        cve_link = f"[{cve}](https://nvd.nist.gov/vuln/detail/{cve})"
+        lines.append(f"| {pname} | {version} | {cve_link} | {score_str} | {desc} |")
+    return "\n".join(lines)
+
+
+def vulnix(args):
+    paths = list(build_hosts(args.ref).values())
+    result = run("vulnix", "-j", *paths, capture_output=True)
+    if result.returncode not in (0, 2):
+        print(result.stderr, file=sys.stderr)
+        sys.exit(result.returncode)
+    vulns = json.loads(result.stdout or "[]")
+    if vulns:
+        cve_count = sum(len(v["affected_by"]) for v in vulns)
+        date = datetime.date.today().isoformat()
+        with open(args.body_out, "w") as f:
+            f.write(f"vulnix scan ({date}): **{cve_count}** CVEs across **{len(vulns)}** packages.\n\n")
+            f.write(format_vulnix(vulns))
+            f.write("\n")
+    print(len(vulns))
+
+
+def find_cycles(edges):
+    adj = {}
+    for e in edges:
+        if e["kind"] == "after":
+            adj.setdefault(e["from"], []).append(e["to"])
+
+    visited, on_path = set(), set()
+    path = []
+    cycles = []
+
+    def dfs(node):
+        visited.add(node)
+        path.append(node)
+        on_path.add(node)
+        for nb in adj.get(node, []):
+            if nb in on_path:
+                i = path.index(nb)
+                cycles.append(path[i:] + [nb])
+            elif nb not in visited:
+                dfs(nb)
+        path.pop()
+        on_path.discard(node)
+
+    for node in adj:
+        if node not in visited:
+            dfs(node)
+    return cycles
+
+
+def lint(args):
+    out = os.path.abspath(args.out)
+    target = os.path.abspath(args.target)
+    os.makedirs(out, exist_ok=True)
+    os.chdir(target)
+
+    files = [f for f in git("ls-files", "-z", "*.nix").stdout.split("\0") if f]
+    if files:
+        run("nixfmt", *files)
+    with open(f"{out}/nixfmt.diff", "w") as f:
+        f.write(git("diff").stdout)
+    git("checkout", "--", ".")
+
+    with open(f"{out}/statix.errfmt", "w") as f:
+        f.write(run("statix", "check", ".", "-o", "errfmt", capture_output=True).stdout)
+
+    dead = run("deadnix", "-o", "json", ".", capture_output=True).stdout
+    with open(f"{out}/deadnix.rdjsonl", "w") as f:
+        for line in filter(str.strip, dead.splitlines()):
+            report = json.loads(line)
+            for r in report["results"]:
+                loc = {"path": report["file"]}
+                if "line" in r:
+                    loc["range"] = {
+                        "start": {"line": r["line"], "column": r["column"]},
+                        "end": {"line": r["line"], "column": r["endColumn"]},
+                    }
+                rec = {
+                    "message": r["message"],
+                    "location": loc,
+                    "severity": "ERROR",
+                }
+                json.dump(rec, f)
+                f.write("\n")
+
+    paths = inspect_hosts(target)
+    with open(f"{out}/cycles.rdjsonl", "w") as f:
+        for host, path in (paths or {}).items():
+            report = read_inspect(path)
+            for cycle in find_cycles(report["dependencies"]["edges"]):
+                chain = " → ".join(u.split(":", 1)[1] for u in cycle)
+                rec = {
+                    "message": f"[{host}] systemd ordering cycle: {chain}",
+                    "location": {"path": "flake.nix"},
+                    "severity": "ERROR",
+                }
+                json.dump(rec, f)
+                f.write("\n")
+
+
+LINTERS = [
+    ("nixfmt.diff", ["-f=diff", "-f.diff.strip=1", "-name=nixfmt", "-level=error"]),
+    ("statix.errfmt", ["-efm=%f>%l:%c:%*[A-Z]:%*[0-9]:%m", "-name=statix", "-level=error"]),
+    ("deadnix.rdjsonl", ["-f=rdjsonl", "-name=deadnix"]),
+    ("cycles.rdjsonl", ["-f=rdjsonl", "-name=systemd-cycles", "-filter-mode=nofilter"]),
+]
+
+
+def lint_report(args):
+    def reviewdog(path, flags):
+        with open(path) as f:
+            return run("reviewdog", *flags, f"-reporter={args.reporter}", "-fail-level=error", stdin=f).returncode
+
+    codes = [reviewdog(f"{args.dir}/{name}", flags)
+             for name, flags in LINTERS
+             if os.path.getsize(f"{args.dir}/{name}") > 0]
+    sys.exit(1 if any(codes) else 0)
+
+
+def post_comment(args):
+    body = args.marker + "\n" + Path(args.body_file).read_text()
+    pr = os.environ["PR"]
+    comments = gitea("GET", f"/issues/{pr}/comments")
+    existing = next((c for c in comments if c["body"].startswith(args.marker)), None)
+    if existing:
+        gitea("PATCH", f"/issues/comments/{existing['id']}", {"body": body})
+    else:
+        gitea("POST", f"/issues/{pr}/comments", {"body": body})
+
+
+def upsert_issue(args):
+    body = Path(args.body_file).read_text()
+    issue = find_open_issue(args.label)
+    if issue:
+        gitea("PATCH", f"/issues/{issue['number']}", {"body": body})
+    else:
+        gitea("POST", "/issues", {"title": args.title, "body": body, "labels": [label_id(args.label)]})
+
+
+def close_issue(args):
+    issue = find_open_issue(args.label)
+    if issue:
+        gitea("PATCH", f"/issues/{issue['number']}", {"state": "closed"})
+
+
+def label_id(name):
+    match = next((x for x in gitea("GET", "/labels") if x["name"] == name), None)
+    return match["id"] if match else gitea("POST", "/labels", {"name": name, "color": "#ededed"})["id"]
+
+
+def open_pr(args):
+    if git("diff", "--quiet", check=False).returncode == 0:
+        print("no changes; nothing to do")
+        return
+    git("switch", "-C", args.branch)
+    git("add", "-A")
+    git("-c", "user.name=gitea-actions[bot]", "-c", "user.email=gitea-actions@noreply.local",
+        "commit", "-m", args.title)
+
+    host = os.environ["API"].removesuffix("/api/v1")
+    auth = base64.b64encode(f'token:{os.environ["GH_TOKEN"]}'.encode()).decode()
+    git("-c", f"http.extraHeader=Authorization: Basic {auth}",
+        "push", "-f", f'{host}/{os.environ["REPO"]}.git', f"HEAD:{args.branch}")
+
+    owner = os.environ["REPO"].split("/")[0]
+    if gitea("GET", f"/pulls?state=open&head={owner}:{args.branch}"):
+        return
+    labels = [label_id(args.label)] if args.label else []
+    gitea("POST", "/pulls",
+          {"title": args.title, "head": args.branch, "base": args.base,
+           "body": Path(args.body_file).read_text(), "labels": labels})
+
+
+def main():
+    parser = argparse.ArgumentParser(prog="ci")
+    sub = parser.add_subparsers(required=True)
+
+    def command(name, fn, *positional):
+        p = sub.add_parser(name)
+        for arg in positional:
+            if arg.endswith("?"):
+                p.add_argument(arg[:-1], nargs="?")
+            else:
+                p.add_argument(arg)
+        p.set_defaults(fn=fn)
+
+    command("diff", diff, "old", "new", "title?")
+    command("inspect-diff", inspect_diff, "old", "new", "title?")
+    command("vulnix", vulnix, "ref", "body_out")
+    command("lint", lint, "target", "out")
+    command("lint-report", lint_report, "reporter", "dir")
+    command("post-comment", post_comment, "marker", "body_file")
+    command("upsert-issue", upsert_issue, "label", "title", "body_file")
+    command("close-issue", close_issue, "label")
+    command("open-pr", open_pr, "branch", "base", "title", "body_file", "label?")
+
+    args = parser.parse_args()
+    args.fn(args)
+
+
+if __name__ == "__main__":
+    main()