floweyandFloweynt
95cd7d0457
Co-authored-by: Floweynt <50057682+Floweynt@users.noreply.github.com> Reviewed-on: #8
528 lines
18 KiB
Python
528 lines
18 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.error
|
||
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)
|
||
|
||
|
||
class GiteaError(Exception):
|
||
pass
|
||
|
||
|
||
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:
|
||
try:
|
||
detail = json.loads(e.read()).get("message", e.reason)
|
||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||
detail = e.reason
|
||
raise GiteaError(f"{method} {path}: {e.code} {detail}") from None
|
||
return json.loads(raw) if raw else None
|
||
|
||
|
||
def find_open_issue(label):
|
||
issues = gitea("GET", f"/issues?state=open&type=issues&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, "").replace("\n", " ").replace("|", "∣")
|
||
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()
|
||
push = git("-c", f"http.extraHeader=Authorization: Basic {auth}",
|
||
"push", "-f", f'{host}/{os.environ["REPO"]}.git', f"HEAD:{args.branch}",
|
||
check=False)
|
||
if push.returncode != 0:
|
||
raise RuntimeError(f"git push failed:\n{push.stderr}")
|
||
|
||
pulls = gitea("GET", f"/pulls?state=open&base_branch={args.base}")
|
||
existing = next((pr for pr in pulls if pr["head"]["ref"] == args.branch), None)
|
||
if existing:
|
||
print(f"PR #{existing['number']} already open, pushed update")
|
||
return
|
||
labels = [label_id(args.label)] if args.label else []
|
||
pr = gitea("POST", "/pulls",
|
||
{"title": args.title, "head": args.branch, "base": args.base,
|
||
"body": Path(args.body_file).read_text(), "labels": labels})
|
||
if not pr or "number" not in pr:
|
||
raise GiteaError(f"POST /pulls: unexpected response: {json.dumps(pr)}")
|
||
print(f"opened PR #{pr['number']}")
|
||
|
||
|
||
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()
|