admin/felhom.eu
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

security watch sees runtimes (R-1327) and DooPlex (R-1328); HTTP2Shell rows R-1329–R-1332; R-1326 closed
gates / gates (push) Successful in 3m27s
Details

Browse Source 
- 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).
This commit is contained in:
admin committed 2026-10-27 23:23:02 +01:00
1 parent 503692a1da
commit f5a4995da5
21 files changed
+3498 -29
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

scripts/security_watch.py
+134 -20
View File
@@ -21,6 +21,10 @@ WHAT IT READS (the spike that chose these: documentation/audits/security-watch-2
feed:NAME a vendor feed (FEEDS below): an item in the last 30 days naming the program and a security word
vulnnotes:NAME a vendor release-notes file whose security lines carry a marker (VULNNOTES below)
registry tags Docker Hub / ghcr.io / lscr.io, for BEHIND (same channel = same tag shape, same major)
runtime the image ITSELF (R-1327): Node.js and Python Brotli/brotlicffi inside it, from its registry config
and, where needed, its layers (image_runtime.py), matched against security_watch_runtimes.json —
a bug in the runtime under an app (HTTP2Shell, CVE-2026-111497) never reaches the app's own notices.
Cached by image digest in the state file (`runtime_cache`), so a night reads only changed images.
The list of pins: security_watch_inventory.py (ONE reader, shared with the catalog's and the controller's gates).
MAILS (operator only, admin@felhom.eu, the CI's Resend path — no household ever gets one):
@@ -36,7 +40,7 @@ state.json; the reasons in the workflow). A missing state file means: seed (reco
mail nothing new) AND send one mail saying the memory was missing, with the open list — never a silent re-seed.
USAGE
python3 scripts/security_watch.py --state=<file> [--catalog=<dir>] [--controller=<dir>] [--seed] [--no-mail]
python3 scripts/security_watch.py --state=<file> [--catalog=<dir>] [--controller=<dir>] [--seed] [--no-mail] [--no-runtime]
[--test-mail] [--today=YYYY-MM-DD] [--json=<results file>]
Exit: 0 the run finished (whatever it found) · 1 a mail could not be sent · 2 the list of pins is unreadable.
"""
@@ -54,6 +58,7 @@ import zlib
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import security_watch_inventory as inv # noqa: E402
import image_runtime # noqa: E402
UA = "felhom-security-watch/1.0 (+https://felhom.eu)"
MAIL_TO = "admin@felhom.eu"
@@ -219,6 +224,78 @@ def ghsa_affects(adv, ours, repo=""):
# ── sources ──────────────────────────────────────────────────────────────────────────────────────────────────────────
RUNTIME_TABLE = json.load(open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "security_watch_runtimes.json"),
encoding="utf-8"))["rows"]
def _vt(v):
return tuple(int(x) for x in re.findall(r"\d+", re.split(r"[-+~]", v or "", maxsplit=1)[0]))
def runtime_part(res, part):
if ":" in part:
db, name = part.split(":", 1)
return (res.get(db) or {}).get(name)
return res.get(part)
def runtime_advisories(res, group):
"""The runtime table's verdicts for one image read (image_runtime.read_image's dict). Pure."""
out = []
for row in RUNTIME_TABLE:
if row.get("groups") and group not in row["groups"]:
continue
ver = runtime_part(res, row["part"])
if not ver:
continue
v = _vt(ver)
if not v or v < _vt(row.get("from", "0")):
continue
n = row.get("line_parts", 1)
line = ".".join(str(x) for x in v[:n])
lines = sorted(row["fixed"], key=_vt)
fixed = row["fixed"].get(line)
if fixed:
if v >= _vt(fixed):
continue
why, fx = "", fixed
elif _vt(line) > _vt(lines[-1]):
continue # newer than the notice's lines
else:
why, fx = " — no fixed release on the %s line (end of life): move to a supported line" % line, \
", ".join(row["fixed"][k] for k in lines)
out.append({"id": row["id"], "source": "runtime", "severity": row.get("severity", ""),
"summary": "%s %s inside the image: %s%s" % (row["label"], ver, row["summary"], why),
"fixed": fx, "link": row.get("link", ""), "unsure": False, "published": row.get("published", "")})
return out
def cached_runtime_reader(cache, resolve=None, read=None):
"""A reader for run(): resolves the image's digest (cheap), answers from `cache` when known, else reads it and
caches a COMPLETE result only (an unchecked one is tried again the next night). `cache` lives in the state file."""
resolve = resolve or image_runtime.resolve_digest
read = read or image_runtime.read_image
def reader(ref):
try:
digest = resolve(ref)
except image_runtime.Unreadable as e:
return {"ref": ref, "unchecked": [str(e)]}
except Exception as e: # noqa: BLE001 — a network error is a NOT CHECKED, never a crash of the whole watch
return {"ref": ref, "unchecked": ["%s: %s" % (type(e).__name__, str(e)[:120])]}
if digest and digest in cache:
hit = dict(cache[digest], ref=ref)
hit["_used"] = True
cache[digest]["_seen"] = True
return hit
res = read(ref)
if digest and not res.get("unchecked"):
cache[digest] = {k: v for k, v in res.items() if k != "ref"}
cache[digest]["_seen"] = True
return res
return reader
class Sources:
def __init__(self, net, today, rotate):
self.net, self.today, self.rotate = net, today, rotate
@@ -348,11 +425,16 @@ class Sources:
if st != 200:
raise SourceError("ghcr.io token for %s: HTTP %s" % (path, st))
tok = json.loads(body).get("token", "")
st, _h, body = self.net.get("https://ghcr.io/v2/%s/tags/list?n=10000" % path,
{"Authorization": "Bearer %s" % tok})
if st != 200:
raise SourceError("ghcr.io tags of %s: HTTP %s" % (path, st))
names = json.loads(body).get("tags") or []
url = "https://ghcr.io/v2/%s/tags/list?n=10000" % path
for _ in range(50): # ghcr.io PAGES its list (Link: rel="next") — page 1 alone missed every newer tag (R-1327)
st, h, body = self.net.get(url, {"Authorization": "Bearer %s" % tok})
if st != 200:
raise SourceError("ghcr.io tags of %s: HTTP %s" % (path, st))
names += json.loads(body).get("tags") or []
nxt = re.search(r'<([^>]+)>;\s*rel="next"', (h or {}).get("Link", "") if hasattr(h, "get") else "")
if not nxt:
break
url = "https://ghcr.io" + nxt.group(1) if nxt.group(1).startswith("/") else nxt.group(1)
else:
names = None # a registry the watch does not read (our own Gitea, GitLab): stated, not a failure
self._tags[image] = names
@@ -365,7 +447,7 @@ def registry_of(image):
reg, path = first, image.split("/", 1)[1] if "/" in image else ""
else:
reg, path = "docker.io", image
if reg == "lscr.io":
if reg in ("lscr.io", "index.docker.io", "registry-1.docker.io"):
reg = "docker.io"
if reg == "docker.io" and "/" not in path:
path = "library/" + path
@@ -496,6 +578,15 @@ def check_item(it, src, today):
nc.append({"source": "registry:%s" % registry_of(it["image"])[0], "why": str(e)})
if newer_major:
notes.append("a newer major exists: %s" % newer_major)
if getattr(src, "runtime", None) and not it.get("no_registry") and it.get("image") and it.get("tag"):
res = src.runtime("%s:%s" % (it["image"], it["tag"]))
unchecked = res.get("unchecked") or []
no_route = [u for u in unchecked if u.startswith("registry ") and "no anonymous token route" in u]
if no_route and it["group"] != "dooplex":
notes.append("runtime: %s" % no_route[0])
elif unchecked:
nc.append({"source": "runtime", "why": "; ".join(unchecked)[:300]})
advs += runtime_advisories(res, it["group"])
it.update({"advisories": advs, "not_checked": nc, "behind": behind, "notes": notes})
if it.get("problems"):
nc.append({"source": "inventory", "why": "; ".join(it["problems"])})
@@ -608,6 +699,13 @@ def plan(items, state, today, seed=False):
return mails, st, would
GROUP_ORDER = {"box": 0, "build": 0, "app": 1, "dooplex": 2}
def _grp(it):
return GROUP_ORDER.get(it["group"], 1)
def _adv_lines(it, a, reminder=False):
return ["%s%s — %s" % ("REMINDER (still open): " if reminder else "", it["image"], a["id"]),
" What: %s" % (a["summary"] or "(no summary)"),
@@ -617,15 +715,17 @@ def _adv_lines(it, a, reminder=False):
" Source: %s%s" % (a["source"], " — the range could not be read; check by hand" if a.get("unsure") and
a["source"] == "GitHub advisory" else ""),
" Link: %s" % (a["link"] or "-"),
" Next: the security-fix lane (B23) — capability map B23; for a box program, a controller release",
" Next: " + ("the homelab session updates this DooPlex workload (Felhom changes nothing on DooPlex; the "
"list: documentation/security-watch/dooplex-workloads.txt)" if it["group"] == "dooplex" else
"the security-fix lane (B23) — capability map B23; for a box program, a controller release"),
""]
def advisory_mail(new, remind, today):
allp = new + remind
box = [it for it, _ in allp if it["group"] != "app"]
box = [it for it, _ in allp if _grp(it) == 0]
names = []
for it, _ in sorted(allp, key=lambda p: (p[0]["group"] == "app", p[0]["image"])):
for it, _ in sorted(allp, key=lambda p: (_grp(p[0]), p[0]["image"])):
n = it["image"].rsplit("/", 1)[-1]
if n not in names:
names.append(n)
@@ -634,9 +734,9 @@ def advisory_mail(new, remind, today):
", %d reminder%s" % (len(remind), "" if len(remind) == 1 else "s") if remind else "")
lines = ["The nightly security watch (felhom.eu R-1303) found a notice that affects a version we ship.", "",
"Box programs first%s." % ("" if box else " (none tonight)"), ""]
for it, a in sorted(new, key=lambda p: (p[0]["group"] == "app", p[0]["image"])):
for it, a in sorted(new, key=lambda p: (_grp(p[0]), p[0]["image"])):
lines += _adv_lines(it, a)
for it, a in sorted(remind, key=lambda p: (p[0]["group"] == "app", p[0]["image"])):
for it, a in sorted(remind, key=lambda p: (_grp(p[0]), p[0]["image"])):
lines += _adv_lines(it, a, reminder=True)
lines += ["A notice is mailed once, and again every %d days while it stays open. Night of %s." % (REMIND_DAYS, today)]
return {"kind": "advisory", "subject": subject, "text": "\n".join(lines)}
@@ -657,8 +757,9 @@ def weekly_mail(items, today):
def block(title, rows):
return ([title] + [" " + r for r in rows] + [""]) if rows else []
lines = ["The weekly summary of the security watch (felhom.eu R-1303): newer versions with no security notice.", ""]
for grp, name in (("box", "Box programs and builds"), ("app", "Catalog apps")):
sel = [it for it in items if (it["group"] == "app") == (grp == "app")]
for grp, name in ((0, "Box programs and builds"), (1, "Catalog apps"),
(2, "DooPlex (homelab) — a list for the homelab session; Felhom changes nothing there")):
sel = [it for it in items if _grp(it) == grp]
lines += block(name + " — BEHIND:", ["%s %s -> %s (%s)" % (it["image"], it["tag"], it["behind"], it["owner"])
for it in sel if it.get("behind")])
lines += block("Open advisories (mailed already):", sorted({"%s %s — %s" % (it["image"], it["tag"], a["id"])
@@ -691,11 +792,14 @@ def send(mail, key, test=False):
# ── the run ──────────────────────────────────────────────────────────────────────────────────────────────────────────
def run(items, net, today, rotate):
def run(items, net, today, rotate, runtime=None):
"""`runtime`: a reader (cached_runtime_reader) for the images' own runtime parts; None = not looked at (the tests
of W1–W9). main() always hands one."""
src = Sources(net, today, rotate)
src.runtime = runtime
# Box programs first, one by one (they share sources with nothing); apps in a small pool.
box = [it for it in items if it["group"] != "app"]
apps = [it for it in items if it["group"] == "app"]
box = [it for it in items if _grp(it) == 0]
apps = [it for it in items if _grp(it) != 0] # catalog apps and the DooPlex list share the pool
for it in box:
check_item(it, src, today)
with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex:
@@ -742,10 +846,17 @@ def main(argv):
return 2
token = os.environ.get("GH_ADVISORY_TOKEN", "")
net = Net(token)
print("security watch %s: %d pins (%d box/build, %d app); GitHub key %s" % (
today, len(items), sum(1 for i in items if i["group"] != "app"), sum(1 for i in items if i["group"] == "app"),
print("security watch %s: %d pins (%d box/build, %d app, %d DooPlex); GitHub key %s" % (
today, len(items), sum(1 for i in items if _grp(i) == 0), sum(1 for i in items if i["group"] == "app"),
sum(1 for i in items if i["group"] == "dooplex"),
"PRESENT — every repo's advisories tonight" if token else "ABSENT — apps' advisories in a 2-night rotation"))
run(items, net, today, rotate=not token)
state_path = opts.get("state")
prior = {}
if state_path and os.path.isfile(state_path):
prior = json.load(open(state_path, encoding="utf-8"))
rcache = {k: dict(v, _seen=False) for k, v in (prior.get("runtime_cache") or {}).items()}
run(items, net, today, rotate=not token,
runtime=None if "no-runtime" in opts else cached_runtime_reader(rcache))
for line in table(items):
print(line)
counts = {}
@@ -759,6 +870,9 @@ def main(argv):
state, missing = json.load(open(state_path, encoding="utf-8")), False
seed = "seed" in opts or missing
mails, new_state, would = plan(items, state, today, seed=seed)
# the runtime cache rides the same state file; only digests seen tonight are kept (it cannot grow without bound)
new_state["runtime_cache"] = {k: {x: y for x, y in v.items() if x != "_seen"} for k, v in rcache.items()
if v.get("_seen")}
if missing and "seed" not in opts:
lines = ["The watch's memory (%s) was MISSING tonight, so it re-seeded: every advisory open tonight is now" % state_path,
"recorded as known, and nothing was mailed as new. Read the list below — it is what is open.", ""]