admin
f5a4995da5
gates / gates (push) Successful in 3m27s
- scripts/image_runtime.py: Node / Python Brotli / brotlicffi / apk / dpkg read from an image's registry config and layers (no pull; HEAD for the nightly digest; zstd via the zstd program where present; quay.io, index.docker.io). - security_watch.py: a „runtime" source (security_watch_runtimes.json), cached by digest in state.json (154 images, filled on DooPlex tonight); ghcr.io tag lists are paged; the DooPlex group (own block, homelab next-step). - dooplex_inventory.py + documentation/security-watch/dooplex-workloads.txt (89 images, homelab-manifests 5aeec8d). - Tests W10–W12 (58 checks) + test_image_runtime.py (13), each red-proved. - audits/http2shell-2026-10-27: the Home Assistant lane walk (proven), HOMELAB-LIST.md (19 to update), the run log. - 08 §11 design line; 09 §3 decision 276 (HA monthly = same main version; decided by CC unattended).
315 lines
14 KiB
Python
315 lines
14 KiB
Python
#!/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/<app>/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:<why> 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 <catalog root>
|
|
python3 scripts/security_watch_inventory.py --gate=controller <controller repo root>
|
|
python3 scripts/security_watch_inventory.py --gate=hub <felhom.eu root>
|
|
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
|
|
|
|
|
|
DOOPLEX_LIST = os.path.join(FELHOM, "documentation", "security-watch", "dooplex-workloads.txt")
|
|
|
|
|
|
def dooplex_items(path=DOOPLEX_LIST, catalog_sources=None):
|
|
"""The DooPlex (homelab) workloads (R-1328) — a LIST from scripts/dooplex_inventory.py, group „dooplex": read and
|
|
mailed to the operator only; nothing on DooPlex is changed from here. An image the catalog also ships reuses the
|
|
catalog's upstream line; any other is a stated gap (the watch still reads its registry tags and its runtime)."""
|
|
if not os.path.isfile(path):
|
|
return []
|
|
items = []
|
|
for ln in _read(path).splitlines():
|
|
if not ln.strip() or ln.startswith("#"):
|
|
continue
|
|
ref, _, where = ln.partition("\t")
|
|
image, tag = split_ref(ref.strip())
|
|
items.append({"group": "dooplex", "owner": where.strip()[:80] or "DooPlex", "image": image, "tag": tag,
|
|
"where": "DooPlex (homelab)", "sources_text": (catalog_sources or {}).get(image) or
|
|
"gap:DooPlex workload — no notice source mapped; registry and runtime only"})
|
|
return items
|
|
|
|
|
|
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")
|
|
apps = catalog_items(catalog)
|
|
items = controller_items(controller) + hub_items(felhom_root) + apps
|
|
# the catalog's upstream lines, keyed by image, for the DooPlex images that are the same program
|
|
cat = {it["image"]: it["sources_text"] for it in apps if it.get("sources_text")}
|
|
items += dooplex_items(catalog_sources=cat)
|
|
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 DooPlex); %d problems" % (
|
|
len(items), sum(1 for i in items if i["group"] not in ("app", "dooplex")),
|
|
sum(1 for i in items if i["group"] == "app"), sum(1 for i in items if i["group"] == "dooplex"), len(problems)))
|
|
return 1 if problems else 0
|
|
print(__doc__)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|