#!/usr/bin/env python3 """security_watch_inventory.py — ONE list of every program Felhom ships, each with its pinned version and the place its security notices are published (R-1303). Read by the nightly watch (security_watch.py) and by the catalog's and the controller's push gates; never copied (the reuse_refs_check.py rule). WHERE THE FACTS LIVE (each kept by a person, beside the pin it describes): * catalog templates//docker-compose.yml `image:` lines, and the `upstream:` block of the same template's `.felhom.yml` — one line per image, keyed by the image without its tag. * box controller/internal/infra/infra.go `*Image` consts, and controller/internal/infra/upstream.txt beside it. * builds every `FROM` in controller/Dockerfile, controller/infra-images/samba/Dockerfile and felhom.eu hub/Dockerfile, and `ARG RESTIC_VERSION`, each with a `# upstream: …` comment on the line ABOVE it. THE SOURCE GRAMMAR (space-separated, at least one; NEW-APP-CHECKLIST.md 1.10 points here): github:OWNER/REPO the repo's published security advisories AND its release notes (release feed) advisories:OWNER/REPO advisories published in ANOTHER repo (Nextcloud: nextcloud/security-advisories) osv:ECOSYSTEM/NAME an OSV.dev package at our version (today only Go/stdlib, for the Go builders) feed:NAME a vendor feed the watch knows by name (security_watch.FEEDS: cloudflare-blog) vulnnotes:NAME a vendor's release-notes file whose security lines carry a marker (cloudflared: VULN-) self our own build — its code is scanned by security_scan_gate.py, not watched here gap: no machine-readable source exists; the reason is listed in every weekly mail A pin with NO line refuses the push (`--gate`). A gap is allowed — it is a stated hole, not a silent one. USAGE python3 scripts/security_watch_inventory.py --list [--json] # the generated list (siblings beside this repo) python3 scripts/security_watch_inventory.py --gate=catalog python3 scripts/security_watch_inventory.py --gate=controller python3 scripts/security_watch_inventory.py --gate=hub Exit (gate): 0 every pin has an upstream · 1 convicted · 2 a file nobody can read. Decoys: catalog `scripts/test_gate_decoys.py` and controller `controller/scripts/test_gate_decoys.py` (COVERS "upstream"). stdlib only — the CI runners carry python3 and git and nothing else (no PyYAML). """ import json import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) FELHOM = os.path.dirname(HERE) WORKSPACE = os.environ.get("FELHOM_WORKSPACE") or os.path.dirname(FELHOM) KINDS = ("github", "advisories", "osv", "feed", "vulnnotes", "self", "gap") KNOWN_FEEDS = ("cloudflare-blog",) KNOWN_VULNNOTES = ("cloudflared",) REPO_RX = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9._-]+$") IMAGE_RX = re.compile(r"^\s*image:\s*[\"']?([^\"'\s#]+)") FROM_RX = re.compile(r"^\s*FROM\s+(?:--platform=\S+\s+)?(\S+)", re.I) UPSTREAM_COMMENT_RX = re.compile(r"^\s*#\s*upstream:\s*(.+?)\s*$") class InventoryError(Exception): """A file nobody can read — the gate exits 2, the watch counts it as NOT CHECKED.""" def parse_sources(text): """'github:a/b gap:…' -> ([(kind, arg)], [problems]). A gap's reason runs to the end of the line.""" text = (text or "").strip().strip('"').strip("'").strip() out, problems = [], [] if not text: return out, ["no source"] rest = text while rest: m = re.match(r"(\w+)(?::(\S*))?\s*", rest) if not m: problems.append("unreadable source %r" % rest) break kind, arg = m.group(1), m.group(2) or "" if kind == "gap": reason = rest[len("gap:"):].strip() if rest.startswith("gap:") else "" if len(reason) < 10: problems.append("gap without a reason (at least a sentence: why no source exists)") out.append(("gap", reason)) break rest = rest[m.end():] if kind not in KINDS: problems.append("unknown source kind %r" % kind) elif kind in ("github", "advisories") and not REPO_RX.match(arg): problems.append("%s needs OWNER/REPO, got %r" % (kind, arg)) elif kind == "osv" and not re.match(r"^[A-Za-z.:]+/\S+$", arg): problems.append("osv needs ECOSYSTEM/NAME, got %r" % arg) elif kind == "feed" and arg not in KNOWN_FEEDS: problems.append("unknown feed %r (known: %s)" % (arg, ", ".join(KNOWN_FEEDS))) elif kind == "vulnnotes" and arg not in KNOWN_VULNNOTES: problems.append("unknown vulnnotes %r (known: %s)" % (arg, ", ".join(KNOWN_VULNNOTES))) elif kind == "self" and arg: problems.append("self takes no argument") out.append((kind, arg)) return out, problems def split_ref(ref): """'ghcr.io/a/b:v1.2@sha256:…' -> ('ghcr.io/a/b', 'v1.2'). A ref with no tag reads as 'latest'.""" ref = ref.split("@", 1)[0] head, _, last = ref.rpartition("/") if ":" in last: name, tag = last.split(":", 1) return (head + "/" + name) if head else name, tag return ref, "latest" def _read(path): try: with open(path, encoding="utf-8") as f: return f.read() except OSError as e: raise InventoryError("cannot read %s: %s" % (path, e)) def felhom_upstream_block(text): """The `upstream:` mapping of a .felhom.yml: {image: source text}. A line reader — only top-level `upstream:` and its two-space-indented `key: value` lines count; a comment or another key's nested block never does.""" out, inside = {}, False for line in text.splitlines(): if re.match(r"^upstream:\s*(#.*)?$", line): inside = True continue if not inside: continue if not line.strip() or line.lstrip().startswith("#"): continue if not line.startswith(" "): break # the next top-level key ends the block m = re.match(r"^\s+([^\s:#][^\s#]*?):\s+(.+?)\s*$", line) if m: out[m.group(1)] = m.group(2) return out def catalog_items(root): items = [] tdir = os.path.join(root, "templates") if not os.path.isdir(tdir): raise InventoryError("no templates/ under %s" % root) for app in sorted(os.listdir(tdir)): compose = os.path.join(tdir, app, "docker-compose.yml") if not os.path.isfile(compose): continue meta = _read(os.path.join(tdir, app, ".felhom.yml")) if os.path.isfile(os.path.join(tdir, app, ".felhom.yml")) else "" ups = felhom_upstream_block(meta) # `lifecycle: abandoned` — still running on boxes that had it, never updated again (plant-it): listed, not looked up. abandoned = bool(re.search(r"^lifecycle:\s*[\"']?abandoned", meta, re.M)) seen = set() for line in _read(compose).splitlines(): m = IMAGE_RX.match(line) if not m: continue image, tag = split_ref(m.group(1)) if (image, tag) in seen: continue seen.add((image, tag)) items.append({"group": "app", "owner": app, "image": image, "tag": tag, "abandoned": abandoned, "where": "templates/%s/.felhom.yml upstream: %s" % (app, image), "sources_text": ups.get(image)}) return items def infra_table(path): """upstream.txt beside infra.go: `ConstName source…` lines; # comments and blank lines ignored.""" out = {} for line in _read(path).splitlines(): s = line.split("#", 1)[0].strip() if not line.lstrip().startswith("#") else "" if not s: continue name, _, rest = s.partition(" ") out[name] = rest.strip() return out def dockerfile_items(path, owner, group="build"): """Every FROM (and ARG RESTIC_VERSION) with the `# upstream:` comment on the line above it.""" items, prev = [], "" stages = set() for line in _read(path).splitlines(): m = FROM_RX.match(line) if m: ref = m.group(1) st = re.search(r"\s+AS\s+(\S+)\s*$", line, re.I) if ref.lower() not in stages and ref != "scratch": image, tag = split_ref(ref) c = UPSTREAM_COMMENT_RX.match(prev) items.append({"group": group, "owner": owner, "image": image, "tag": tag, "where": "%s FROM %s" % (os.path.relpath(path, WORKSPACE), image), "sources_text": c.group(1) if c else None}) if st: stages.add(st.group(1).lower()) a = re.match(r"^\s*ARG\s+RESTIC_VERSION=(\S+)", line) if a: c = UPSTREAM_COMMENT_RX.match(prev) items.append({"group": group, "owner": owner, "image": "restic", "tag": a.group(1), "no_registry": True, "where": "%s ARG RESTIC_VERSION" % os.path.relpath(path, WORKSPACE), "sources_text": c.group(1) if c else None}) if line.strip(): prev = line return items def controller_items(repo): ctrl = os.path.join(repo, "controller") src = _read(os.path.join(ctrl, "internal", "infra", "infra.go")) table = infra_table(os.path.join(ctrl, "internal", "infra", "upstream.txt")) items = [] for const, ref in re.findall(r"^\s*(\w+Image)\s*=\s*\"([^\"]+)\"", src, re.M): image, tag = split_ref(ref) items.append({"group": "box", "owner": "controller infra", "image": image, "tag": tag, "const": const, "where": "controller/internal/infra/upstream.txt %s" % const, "sources_text": table.get(const)}) if not items: raise InventoryError("no *Image consts in infra.go — the reader is broken, not the pins") items += dockerfile_items(os.path.join(ctrl, "Dockerfile"), "controller image") samba = os.path.join(ctrl, "infra-images", "samba", "Dockerfile") if os.path.isfile(samba): items += dockerfile_items(samba, "felhom-samba image") return items def hub_items(felhom_root): return dockerfile_items(os.path.join(felhom_root, "hub", "Dockerfile"), "hub image") def judge(items): """Attach parsed sources and problems; returns the problems list (empty = every pin has an upstream).""" problems = [] for it in items: if it.get("sources_text") is None: it["sources"], it["problems"] = [], ["no upstream line"] else: it["sources"], it["problems"] = parse_sources(it["sources_text"]) for p in it["problems"]: problems.append("%s (%s:%s): %s" % (it["where"], it["image"], it["tag"], p)) return problems def full_inventory(felhom_root=FELHOM, catalog=None, controller=None): catalog = catalog or os.path.join(WORKSPACE, "app-catalog-felhom.eu") controller = controller or os.path.join(WORKSPACE, "felhom-controller") items = controller_items(controller) + hub_items(felhom_root) + catalog_items(catalog) problems = judge(items) return items, problems def gate(mode, root): try: items = {"catalog": catalog_items, "controller": controller_items, "hub": hub_items}[mode](root) except InventoryError as e: print("upstream gate INCONCLUSIVE: %s" % e) return 2 problems = judge(items) for p in problems: print(" FAIL %s" % p) if problems: print("upstream gate FAILED: %d of %d pins have no usable upstream line (security watch, felhom.eu R-1303). " "Grammar: felhom.eu/scripts/security_watch_inventory.py; NEW-APP-CHECKLIST.md 1.10." % (len(problems), len(items))) return 1 gaps = sum(1 for it in items if any(k == "gap" for k, _ in it["sources"])) print("upstream gate OK: %d pins, every one names its upstream (%d stated gaps)" % (len(items), gaps)) return 0 def main(argv): args = [a for a in argv if not a.startswith("--")] flags = [a for a in argv if a.startswith("--")] g = [f.split("=", 1)[1] for f in flags if f.startswith("--gate=")] if g: if g[0] not in ("catalog", "controller", "hub") or len(args) != 1: print(__doc__) return 2 return gate(g[0], args[0]) if "--list" in flags: try: items, problems = full_inventory() except InventoryError as e: print("inventory unreadable: %s" % e) return 2 if "--json" in flags: print(json.dumps(items, indent=1, ensure_ascii=False)) else: for it in items: print("%-6s %-22s %-48s %-24s %s" % (it["group"], it["owner"][:22], it["image"][:48], it["tag"][:24], it.get("sources_text") or "-- MISSING --")) print("%d pins (%d box/build, %d app); %d problems" % ( len(items), sum(1 for i in items if i["group"] != "app"), sum(1 for i in items if i["group"] == "app"), len(problems))) return 1 if problems else 0 print(__doc__) return 2 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))