511 lines
17 KiB
Python
511 lines
17 KiB
Python
import argparse
|
|
import base64
|
|
import datetime
|
|
import difflib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import subprocess_tee
|
|
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):
|
|
# tee to stderr - stdout leaks into GITHUB_OUTPUT $() captures
|
|
return subprocess_tee.run(("nix", "--extra-experimental-features", "nix-command flakes", "-L", *args), check=True, text=True, stdout=sys.stderr).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")
|
|
try:
|
|
with urllib.request.urlopen(req) as resp:
|
|
raw = resp.read()
|
|
except urllib.error.HTTPError as e:
|
|
sys.exit(f"gitea {method} {path}: {e.code} {e.reason}\n{e.read().decode()}")
|
|
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.isfile(f"{args.dir}/{name}") and os.path.getsize(f"{args.dir}/{name}") > 0]
|
|
sys.exit(1 if any(codes) else 0)
|
|
|
|
|
|
def has_findings(path):
|
|
return os.path.isfile(path) and os.path.getsize(path) > 0
|
|
|
|
|
|
def fmt_rdjsonl(path):
|
|
lines = []
|
|
with open(path) as f:
|
|
for raw in f:
|
|
rec = json.loads(raw)
|
|
loc = rec["location"]
|
|
r = loc.get("range", {}).get("start")
|
|
pos = f"{loc['path']}:{r['line']}:{r['column']}" if r else loc["path"]
|
|
lines.append(f"- `{pos}`: {rec['message']}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def fmt_errfmt(path):
|
|
lines = []
|
|
with open(path) as f:
|
|
for raw in f:
|
|
parts = raw.strip().split(">", 1)
|
|
if len(parts) == 2:
|
|
lines.append(f"- `{parts[0]}:{parts[1].split(':', 3)[-1].strip()}`" if ":" in parts[1] else f"- `{raw.strip()}`")
|
|
elif raw.strip():
|
|
lines.append(f"- `{raw.strip()}`")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def lint_summary(args):
|
|
sections = []
|
|
d = args.dir
|
|
|
|
if has_findings(f"{d}/nixfmt.diff"):
|
|
with open(f"{d}/nixfmt.diff") as f:
|
|
sections.append(f"## nixfmt\n\n```diff\n{f.read().rstrip()}\n```")
|
|
|
|
if has_findings(f"{d}/statix.errfmt"):
|
|
sections.append(f"## statix\n\n{fmt_errfmt(f'{d}/statix.errfmt')}")
|
|
|
|
if has_findings(f"{d}/deadnix.rdjsonl"):
|
|
sections.append(f"## deadnix\n\n{fmt_rdjsonl(f'{d}/deadnix.rdjsonl')}")
|
|
|
|
if has_findings(f"{d}/cycles.rdjsonl"):
|
|
sections.append(f"## systemd ordering cycles\n\n{fmt_rdjsonl(f'{d}/cycles.rdjsonl')}")
|
|
|
|
if sections:
|
|
print("\n\n".join(["# Lint findings"] + sections))
|
|
sys.exit(1)
|
|
else:
|
|
print("No lint findings.")
|
|
sys.exit(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("lint-summary", lint_summary, "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()
|