SPS · log · files · refs · tags · releases · people · languages · GitHub · file
tools/audit-recipes
23596 bytes at 2909919 · GitHub
#!/usr/bin/env python3
"""Audit SPS recipe trees for defects that break an install.
src(1) already validates recipe syntax, same-repository uniqueness, and the
selected dependency graph when it builds an index. This tool covers what an
index build cannot: cross-tree consistency, copy-paste mistakes, and whether
every declared source still downloads and still matches its recorded SHA-256.
A recipe whose upstream tarball moved, was re-rolled, or silently changed is
the single most common way a source distribution stops installing. Nothing
notices until a user runs sget and mkpkg aborts on a checksum. Run this in CI
and before a release.
Offline checks (default):
python3 tools/audit-recipes --trees /usr/src/sps/core:/usr/src/sps/extra
Online checks (downloads every source, streams it through sha256, keeps
nothing on disk):
python3 tools/audit-recipes --trees ... --net --jobs 16
Exit status is 0 when no finding is at or above --fail-level.
"""
from __future__ import annotations
import argparse
import concurrent.futures
import hashlib
import json
import os
import re
import ssl
import sys
import urllib.error
import urllib.request
from pathlib import Path
# Do NOT send a browser User-Agent. SourceForge and several other hosts serve
# an HTML interstitial to browser agents and the real archive to curl/wget, so
# a browser UA turns every SourceForge recipe into a bogus hash mismatch.
UA = "curl/8.5.0"
# Keys accepted by the SPS 1.0 recipe parser. Anything else is an error.
SINGLE_KEYS = {"name", "version", "release", "arch", "description"}
MULTI_KEYS = {
"source", "hash", "depend", "builddep", "optional", "conflict",
"prepare", "configure", "build", "install",
}
ALL_KEYS = SINGLE_KEYS | MULTI_KEYS
NAME_RE = re.compile(r"^[0-9A-Za-z][0-9A-Za-z+_.-]*$")
VERSION_RE = re.compile(r"^[0-9A-Za-z][0-9A-Za-z+_.~-]*$")
HASH_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
CONSTRAINT_RE = re.compile(r"^([0-9A-Za-z][0-9A-Za-z+_.-]*)(?:(<=|>=|=|<|>)(.+))?$")
# Severity ordering, low to high.
LEVELS = ["info", "warn", "error"]
def level_at_least(level: str, floor: str) -> bool:
return LEVELS.index(level) >= LEVELS.index(floor)
class Finding:
__slots__ = ("level", "code", "package", "path", "detail")
def __init__(self, level: str, code: str, package: str, path: Path, detail: str):
self.level = level
self.code = code
self.package = package
self.path = path
self.detail = detail
def as_dict(self) -> dict:
return {
"level": self.level,
"code": self.code,
"package": self.package,
"path": str(self.path),
"detail": self.detail,
}
def __str__(self) -> str:
head = f"{self.level:<5} {self.code:<22} {self.package or '-'}"
if self.detail:
return f"{head}\n {self.detail}\n {self.path}"
return f"{head}\n {self.path}"
def parse_recipe(path: Path) -> dict:
"""Parse one recipe the way lib/recipe.awk does.
Returns single-valued keys as strings and repeatable keys as lists. Keys
the parser does not know are collected in "unknown" so the audit can
report them rather than silently dropping them.
"""
rec: dict = {
"path": path,
"name": "",
"version": "",
"release": "",
"arch": "",
"description": "",
"unknown": [],
"duplicate_single": [],
"malformed": [],
}
for key in MULTI_KEYS:
rec[key] = []
try:
raw = path.read_text(encoding="utf-8", errors="replace")
except OSError as exc:
rec["malformed"].append(f"unreadable: {exc}")
return rec
rec["raw"] = raw
pending = ""
for physical in raw.splitlines():
line = pending + physical if pending else physical
pending = ""
if line.endswith("\\"):
pending = line[:-1]
continue
stripped = line.strip()
if not stripped or stripped.startswith("#"):
continue
parts = stripped.split(None, 1)
key = parts[0]
value = parts[1].strip() if len(parts) > 1 else ""
if key not in ALL_KEYS:
rec["unknown"].append(key)
continue
if not value:
rec["malformed"].append(f"empty value for '{key}'")
continue
if key in SINGLE_KEYS:
if rec[key]:
rec["duplicate_single"].append(key)
rec[key] = value
else:
rec[key].append(value)
if pending:
rec["malformed"].append("recipe ends with a line continuation")
return rec
def expand(value: str, rec: dict) -> str:
"""Expand the four variables the recipe parser defines."""
out = value
for var, key in (("name", "name"), ("version", "version"),
("release", "release"), ("arch", "arch")):
out = out.replace("${" + var + "}", rec.get(key) or "")
return out
def split_specs(records: list[str]) -> list[str]:
"""A depend/builddep record may hold several whitespace/comma specs."""
out: list[str] = []
for record in records:
for tok in re.split(r"[,\s]+", record.strip()):
if tok:
out.append(tok)
return out
def collect(trees: list[Path]) -> list[dict]:
recipes: list[dict] = []
for tree in trees:
for path in sorted(tree.rglob("recipe")):
if not path.is_file():
continue
rec = parse_recipe(path)
rec["tree"] = tree
recipes.append(rec)
return recipes
def static_checks(recipes: list[dict], universe: list[dict] | None = None) -> list[Finding]:
"""Check `recipes`, resolving dependencies against `universe`.
With --only the caller audits a handful of packages but their dependencies
still live in the rest of the tree, so the name universe must stay whole or
every external dependency looks missing.
"""
if universe is None:
universe = recipes
findings: list[Finding] = []
def add(level, code, rec, detail=""):
findings.append(Finding(level, code, rec.get("name") or "", rec["path"], detail))
by_name: dict[str, list[dict]] = {}
for rec in universe:
if rec["name"]:
by_name.setdefault(rec["name"], []).append(rec)
hash_owner: dict[str, str] = {}
url_owner: dict[str, str] = {}
for rec in recipes:
path = rec["path"]
pkgdir = path.parent.name
name = rec["name"]
for problem in rec["malformed"]:
add("error", "MALFORMED", rec, problem)
for key in rec["unknown"]:
add("error", "UNKNOWN-KEY", rec, f"'{key}' is not a recipe key")
for key in rec["duplicate_single"]:
add("error", "DUPLICATE-KEY", rec, f"'{key}' may appear only once")
# Required metadata.
if not name:
add("error", "MISSING-NAME", rec)
elif not NAME_RE.match(name):
add("error", "BAD-NAME", rec, name)
if not rec["version"]:
add("error", "MISSING-VERSION", rec)
elif not VERSION_RE.match(rec["version"]):
add("error", "BAD-VERSION", rec, rec["version"])
if not rec["release"]:
add("error", "MISSING-RELEASE", rec)
elif not rec["release"].isdigit():
add("warn", "NONNUMERIC-RELEASE", rec, rec["release"])
if not rec["install"]:
add("error", "NO-INSTALL-PHASE", rec)
if not rec["description"]:
add("warn", "NO-DESCRIPTION", rec)
if rec["arch"] and rec["arch"] == "noarch":
add("info", "NOARCH-SPELLING", rec, "official recipes use 'any'")
# The package directory should be the package name. src(1) does not
# require it (categories are not identity) but a mismatch means
# `cd extra/libs/foo` does not get you package foo, and the upgrade
# tooling keys off directory names.
if name and name != pkgdir:
add("warn", "NAME-DIR-MISMATCH", rec, f"name '{name}' in directory '{pkgdir}'")
# Sources and hashes pair up by declaration order.
nsrc, nhash = len(rec["source"]), len(rec["hash"])
if nsrc != nhash:
add("error", "SOURCE-HASH-COUNT", rec,
f"{nsrc} source record(s) but {nhash} hash record(s)")
for digest in rec["hash"]:
if not HASH_RE.match(digest):
add("error", "BAD-HASH-FORMAT", rec, digest)
continue
body = digest[7:]
if len(set(body)) <= 2:
add("error", "PLACEHOLDER-HASH", rec, digest)
for index, source in enumerate(rec["source"]):
url = expand(source, rec)
if re.search(r"\$\{[a-z]+\}", url):
add("error", "UNEXPANDED-VARIABLE", rec, source)
if any(c.isspace() for c in url):
add("error", "SOURCE-WHITESPACE", rec, source)
remote = url.startswith(("http://", "https://", "ftp://"))
if url.startswith("http://"):
add("warn", "PLAINTEXT-SOURCE", rec, url)
if remote:
# A remote URL that never mentions the version will silently
# keep serving an old tarball after a version bump, or (worse)
# serve a moving target that breaks the recorded hash.
if rec["version"] and rec["version"] not in url and \
"${version}" not in source:
add("warn", "URL-WITHOUT-VERSION", rec, url)
if url in url_owner and url_owner[url] != name:
add("info", "SHARED-SOURCE-URL", rec,
f"{url} is also used by {url_owner[url]}")
else:
url_owner[url] = name
elif url.startswith("file://"):
if not url.startswith("file:///"):
add("error", "RELATIVE-FILE-URL", rec, url)
else:
# Bare relative path, resolved from the recipe directory.
if url.startswith("/"):
add("warn", "ABSOLUTE-LOCAL-SOURCE", rec, url)
elif not (path.parent / url).exists():
add("error", "MISSING-LOCAL-SOURCE", rec, url)
if index < len(rec["hash"]) and HASH_RE.match(rec["hash"][index]):
body = rec["hash"][index][7:]
owner = f"{name}:{url}"
if body in hash_owner and hash_owner[body] != owner:
prior = hash_owner[body]
# Identical URL in two packages legitimately shares a
# digest (the kernel recipes all build the same tarball).
if prior.split(":", 1)[1] != url:
add("error", "HASH-REUSED-DIFFERENT-SOURCE", rec,
f"{body} is also the hash of {prior}")
else:
hash_owner.setdefault(body, owner)
if rec["source"] and not rec["hash"]:
add("error", "SOURCE-WITHOUT-HASH", rec)
# Dependency hygiene.
deps = split_specs(rec["depend"])
bdeps = split_specs(rec["builddep"])
for group, label in ((deps, "depend"), (bdeps, "builddep"),
(split_specs(rec["optional"]), "optional"),
(split_specs(rec["conflict"]), "conflict")):
for spec in group:
m = CONSTRAINT_RE.match(spec)
if not m:
add("error", "BAD-DEPENDENCY-SPEC", rec, f"{label} {spec}")
continue
if m.group(1) == name:
add("error", "SELF-DEPENDENCY", rec, f"{label} {spec}")
seen_dep: set[str] = set()
for spec in deps:
base = spec.split("<")[0].split(">")[0].split("=")[0]
if base in seen_dep:
add("warn", "DUPLICATE-DEPEND", rec, base)
seen_dep.add(base)
seen_build: set[str] = set()
for spec in bdeps:
base = spec.split("<")[0].split(">")[0].split("=")[0]
if base in seen_build:
add("warn", "DUPLICATE-BUILDDEP", rec, base)
seen_build.add(base)
if base in seen_dep:
# A runtime dependency is always installed before the build,
# so naming it again as a build dependency is redundant.
add("info", "DEPEND-AND-BUILDDEP", rec, base)
for spec in split_specs(rec["conflict"]):
base = spec.split("<")[0].split(">")[0].split("=")[0]
if base in seen_dep:
add("error", "DEPEND-AND-CONFLICT", rec, base)
# A phase that writes outside $PKG corrupts the build host.
for phase in ("prepare", "configure", "build", "install"):
for command in rec[phase]:
# Only a leading command word counts. The sudo package's own
# --with-vardir=/var/db/sudo is not privilege escalation.
if re.search(r"(?:^|[;&|]\s*|\$\(\s*)(?:sudo|doas)\s", command):
add("error", "PHASE-USES-SUDO", rec, f"{phase} {command}")
if re.search(r"(^|\s)(make|ninja|meson|cmake)\s+install(\s|$)", command) \
and "$PKG" not in command and "DESTDIR" not in command:
add("error", "INSTALL-OUTSIDE-PKG", rec, f"{phase} {command}")
if re.search(r"--libdir[= ]/?usr/lib(?![0-9])", command) or \
re.search(r"--libdir[= ]lib(?![0-9])(\s|$)", command):
add("warn", "LIBDIR-NOT-LIB64", rec, f"{phase} {command}")
# SPS builds CPython with --with-platlibdir=lib64, so a module staged
# under /usr/lib/pythonX.Y is somewhere the interpreter never looks.
# Judge the recipe as a whole: mentioning that path is fine when the
# recipe also relocates it into lib64.
phase_text = " ".join(
command for phase in ("prepare", "configure", "build", "install")
for command in rec[phase])
if "/usr/lib/python" in phase_text and "lib64/python" not in phase_text:
add("error", "PYTHON-SITE-PACKAGES-LIB", rec,
"stages a module under /usr/lib/pythonX.Y and never moves it "
"to /usr/lib64")
# Cross-tree duplicate names. src(1) only rejects duplicates inside one
# repository; the same name in core and extra resolves by priority and
# quietly shadows one of them.
selected = {id(rec) for rec in recipes}
for name, group in sorted(by_name.items()):
if len(group) < 2:
continue
trees = {str(r["tree"]) for r in group}
level = "error" if len(trees) == 1 else "warn"
code = "DUPLICATE-IN-REPOSITORY" if len(trees) == 1 else "SHADOWED-ACROSS-REPOSITORIES"
for rec in group:
if id(rec) not in selected:
continue
others = [str(r["path"]) for r in group if r is not rec]
findings.append(Finding(level, code, name, rec["path"],
"also defined at " + ", ".join(others)))
findings.extend(graph_checks(recipes, by_name))
return findings
def graph_checks(recipes: list[dict], by_name: dict[str, list[dict]]) -> list[Finding]:
"""Missing dependencies and dependency cycles across the whole tree."""
findings: list[Finding] = []
known = set(by_name)
runtime: dict[str, set[str]] = {}
for rec in recipes:
name = rec["name"]
if not name:
continue
edges: set[str] = set()
for label, records in (("depend", rec["depend"]), ("builddep", rec["builddep"])):
for spec in split_specs(records):
base = CONSTRAINT_RE.match(spec)
if not base:
continue
dep = base.group(1)
if dep not in known:
findings.append(Finding(
"error", "MISSING-DEPENDENCY", name, rec["path"],
f"{label} {dep} is not provided by any recipe in the tree"))
elif label == "depend":
edges.add(dep)
runtime.setdefault(name, set()).update(edges)
# Iterative DFS so a deep tree cannot blow the recursion limit.
WHITE, GREY, BLACK = 0, 1, 2
colour = dict.fromkeys(runtime, WHITE)
reported: set[tuple[str, ...]] = set()
for root in sorted(runtime):
if colour[root] != WHITE:
continue
stack = [(root, iter(sorted(runtime.get(root, ()))))]
path = [root]
colour[root] = GREY
while stack:
node, children = stack[-1]
advanced = False
for child in children:
if child not in colour:
continue
if colour[child] == GREY:
cycle = path[path.index(child):] + [child]
key = tuple(sorted(set(cycle)))
if key not in reported:
reported.add(key)
rec = by_name[child][0]
findings.append(Finding(
"error", "DEPENDENCY-CYCLE", child, rec["path"],
" -> ".join(cycle)))
continue
if colour[child] == WHITE:
colour[child] = GREY
path.append(child)
stack.append((child, iter(sorted(runtime.get(child, ())))))
advanced = True
break
if not advanced:
colour[node] = BLACK
stack.pop()
if path:
path.pop()
return findings
def fetch_sha256(url: str, timeout: int) -> tuple[str, int]:
"""Stream a URL through sha256 without keeping it on disk."""
req = urllib.request.Request(url, headers={"User-Agent": UA, "Accept": "*/*"})
ctx = ssl.create_default_context()
digest = hashlib.sha256()
total = 0
with urllib.request.urlopen(req, timeout=timeout, context=ctx) as resp:
while True:
chunk = resp.read(1 << 20)
if not chunk:
break
digest.update(chunk)
total += len(chunk)
return digest.hexdigest(), total
def net_checks(recipes: list[dict], jobs: int, timeout: int,
progress: bool) -> list[Finding]:
tasks: list[tuple[dict, str, str]] = []
for rec in recipes:
for index, source in enumerate(rec["source"]):
if index >= len(rec["hash"]):
continue
url = expand(source, rec)
if not url.startswith(("http://", "https://", "ftp://")):
continue
want = rec["hash"][index]
if not HASH_RE.match(want):
continue
tasks.append((rec, url, want[7:]))
findings: list[Finding] = []
done = 0
total = len(tasks)
def run(task):
rec, url, want = task
try:
got, size = fetch_sha256(url, timeout)
except (urllib.error.HTTPError, urllib.error.URLError, OSError,
ssl.SSLError, ValueError) as exc:
return Finding("error", "SOURCE-UNREACHABLE", rec["name"], rec["path"],
f"{url}\n {exc}")
if got != want:
return Finding("error", "HASH-MISMATCH", rec["name"], rec["path"],
f"{url}\n recorded sha256:{want}\n"
f" actual sha256:{got} ({size} bytes)")
return None
with concurrent.futures.ThreadPoolExecutor(max_workers=jobs) as pool:
for result in pool.map(run, tasks):
done += 1
if progress and (done % 25 == 0 or done == total):
print(f" verified {done}/{total} sources", file=sys.stderr, flush=True)
if result is not None:
findings.append(result)
return findings
def default_trees() -> list[Path]:
env = os.environ.get("SPS_RECIPE_TREES")
if env:
return [Path(p) for p in env.split(":") if p]
candidates = [Path("/usr/src/sps/core"), Path("/usr/src/sps/extra")]
return [p for p in candidates if p.is_dir()]
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(
prog="audit-recipes",
description="Audit SPS recipe trees for install-breaking defects.")
ap.add_argument("--trees", help="colon-separated recipe tree roots")
ap.add_argument("--only", help="comma-separated package names to audit")
ap.add_argument("--net", action="store_true",
help="download every source and verify its SHA-256")
ap.add_argument("--jobs", type=int, default=12, help="parallel downloads")
ap.add_argument("--timeout", type=int, default=600,
help="per-source download timeout in seconds")
ap.add_argument("--fail-level", choices=LEVELS, default="error",
help="lowest severity that makes this command fail")
ap.add_argument("--json", action="store_true", help="machine-readable output")
ap.add_argument("--quiet", action="store_true", help="only print findings")
args = ap.parse_args(argv)
trees = [Path(p) for p in args.trees.split(":") if p] if args.trees else default_trees()
if not trees:
print("audit-recipes: no recipe trees; pass --trees or set SPS_RECIPE_TREES",
file=sys.stderr)
return 2
for tree in trees:
if not tree.is_dir():
print(f"audit-recipes: not a directory: {tree}", file=sys.stderr)
return 2
universe = collect(trees)
recipes = universe
if args.only:
wanted = {n.strip() for n in args.only.split(",") if n.strip()}
recipes = [r for r in universe if r["name"] in wanted]
unknown = wanted - {r["name"] for r in recipes}
for name in sorted(unknown):
print(f"audit-recipes: no recipe named {name}", file=sys.stderr)
if not recipes:
print("audit-recipes: no recipes found", file=sys.stderr)
return 2
if not args.quiet:
print(f"audit-recipes: {len(recipes)} recipes in "
f"{', '.join(str(t) for t in trees)}", file=sys.stderr)
findings = static_checks(recipes, universe)
if args.net:
if not args.quiet:
print("audit-recipes: verifying sources over the network",
file=sys.stderr)
findings.extend(net_checks(recipes, args.jobs, args.timeout,
not args.quiet))
findings.sort(key=lambda f: (-LEVELS.index(f.level), f.code, f.package))
if args.json:
json.dump({"recipes": len(recipes),
"findings": [f.as_dict() for f in findings]},
sys.stdout, indent=2)
sys.stdout.write("\n")
else:
for finding in findings:
print(finding)
counts = {level: 0 for level in LEVELS}
for finding in findings:
counts[finding.level] += 1
if not args.quiet or findings:
print(f"\n{len(recipes)} recipes: "
f"{counts['error']} error, {counts['warn']} warn, "
f"{counts['info']} info", file=sys.stderr)
return 1 if any(level_at_least(f.level, args.fail_level) for f in findings) else 0
if __name__ == "__main__":
sys.exit(main())