#!/usr/bin/env python3
"""test_security_watch.py — the nightly security watch (R-1303) against FAKE source answers. No network.
What is pinned, each by its consequence (the mail or the class), not by its mechanism:
W1 a GitHub advisory whose range covers our pin -> ADVISORY with the fixed version; the same advisory at the fixed
version -> not ADVISORY (the negative control); Nextcloud's per-branch shape; an unreadable range is LISTED
W2 today's cloudflared case, as a RECORDED answer: the Cloudflare blog item of 2026-10-20 -> ADVISORY for cloudflared;
older than 30 days -> nothing; cloudflared's VULN- release-note marker -> ADVISORY with the fixed version
W3 BEHIND (a newer tag in the same channel, same shape), CURRENT, FLOATING, GAP, a beta never counts as a fix
W4 a source that fails -> NOT CHECKED, printed and counted — never CURRENT
W5 one advisory mails ONCE; again after 7 days while open; a vendor notice never repeats; a closed one is forgotten
W6 a source down one night -> no mail; two nights -> one mail; a third night -> no second mail; back up -> reset
W7 the weekly summary on Monday only (or when the last one is 8+ days old); a seed mails nothing and says what it
would have mailed
W8 the inventory: an image with no upstream line refuses (the gate), a gap needs a reason, the grammar
W9 a GitHub Actions advisory is not listed; an empty fix with a range ending at X reads „after X", not „no fix"
W10 the RUNTIME inside an image (R-1327, HTTP2Shell CVE-2026-111497): Node 24.18.1 -> ADVISORY from „runtime" with
the fixed 24.22.1; 24.22.1 -> none (negative control); Python Brotli 1.2.0 -> ADVISORY, 1.2.1 -> none; Node 20 (no
fix on its line) -> ADVISORY; a layer read error -> NOT CHECKED; a registry with no route -> a note, not a
failure; the result is cached by digest (a second night reads nothing); main() wires the reader
W12 the DooPlex (homelab) group (R-1328): read from its list file; a DooPlex advisory names the homelab session, never
the catalog's lane; a DooPlex image in a private registry is NOT CHECKED with the reason; its own weekly block
W11 ghcr.io tag lists are PAGED (Link: rel="next"): every page is read, so a newer tag on page 2 is seen
Red-proofs (2026-10-26): W5 with the `k not in known` test removed (every night mails) -> W5 fails; W6 with the streak
threshold at 1 -> W6 fails; W2 with FEED_DAYS check removed -> W2's old-item case fails; W4 with SourceError swallowed
-> W4 fails. W9 (2026-10-27, R-1310) was seen failing before its two rules existed (2 of 4 checks red, the two
controls green).
"""
import datetime
import json
import os
import shutil
import subprocess
import sys
import tempfile
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import security_watch as w # noqa: E402
import security_watch_inventory as inv # noqa: E402
fails = []
ran = 0
def check(name, cond, detail=""):
global ran
ran += 1
if cond:
print(" ok %s" % name)
else:
fails.append("%s %s" % (name, detail))
print(" XX %s %s" % (name, detail))
class FakeNet:
"""URL prefix -> (status, headers, body) | Exception. POSTs (OSV) answer from `posts`."""
def __init__(self, routes, posts=None):
self.routes, self.posts, self.gh_token, self.requests, self.seen = routes, posts or {}, "", {}, []
def get(self, url, headers=None, timeout=30):
self.seen.append(url)
for pfx, ans in self.routes.items():
if url.startswith(pfx):
if isinstance(ans, Exception):
raise w.SourceError(str(ans))
st, h, body = ans
return st, h, body if isinstance(body, bytes) else json.dumps(body).encode()
raise w.SourceError("no route for %s" % url)
def post_json(self, url, body, timeout=30):
key = (body["package"]["ecosystem"] + "/" + body["package"]["name"], body["version"])
if key not in self.posts:
raise w.SourceError("no OSV answer for %r" % (key,))
return self.posts[key]
def item(image, tag, sources, group="box", owner="test"):
srcs, problems = inv.parse_sources(sources)
return {"group": group, "owner": owner, "image": image, "tag": tag, "where": "test", "sources": srcs,
"problems": problems}
ADV_TRAEFIK = [{"ghsa_id": "GHSA-fh26-gfpp-7xxx", "severity": "high", "summary": "snicheck skips the check",
"html_url": "https://github.com/traefik/traefik/security/advisories/GHSA-fh26-gfpp-7xxx",
"published_at": "2026-10-07T00:00:00Z", "withdrawn_at": None,
"vulnerabilities": [{"package": {"name": "github.com/traefik/traefik/v2"},
"vulnerable_version_range": "<= v2.11.57", "patched_versions": "v2.11.58"},
{"package": {"name": "github.com/traefik/traefik/v3"},
"vulnerable_version_range": ">= v3.0.0, <= v3.7.13", "patched_versions": "v3.7.14"}]}]
TAGS = lambda *n: (200, {}, {"results": [{"name": x} for x in n], "next": None}) # noqa: E731
EMPTY_FEED = (200, {}, b"")
# The RECORDED answer for today's case (blog.cloudflare.com/rss/, item of 2026-10-20, trimmed).
CF_BLOG = (200, {}, ("<![CDATA[8 eMudhra certificates are compromised. Here's what you need to "
"know]]>https://blog.cloudflare.com/emudhra-certificate-compromise/"
"01M5DK5J81T0VC8YTBR2FKD8A3Tue, 20 Oct 2026 09:00:00 GMT"
"Eight eMudhra intermediate certificates are compromised ... "
"Cloudflare Tunnel (cloudflared): update to 2026.10.2 ...]]>"
"Something elsehttps://blog.cloudflare.com/x/x"
"Tue, 20 Oct 2026 09:00:00 GMTWorkers news, no tunnel."
"").encode())
CF_NOTES = (200, {}, b"2026.10.4\n- 2026-10-23 VULN-176088: Allowlist debug endpoints on the metrics server\n\n"
b"2026.10.3\n- 2026-10-21 chore: bump debian-images base image\n\n"
b"2026.10.2\n- 2026-10-20 Bump go to 1.27.3 and bump distroless image\n")
TODAY = datetime.date(2026, 10, 26) # a Monday
def run_items(items, routes, posts=None, rotate=False, today=TODAY):
w.run(items, FakeNet(routes, posts), today, rotate)
return items
def t_classes():
print("-- W1-W4: the classes")
gh = "https://api.github.com/repos/"
routes = {
gh + "traefik/traefik/": (200, {}, ADV_TRAEFIK),
"https://github.com/traefik/traefik/releases.atom": EMPTY_FEED,
"https://hub.docker.com/v2/repositories/library/traefik/": TAGS("v3.7.13", "v3.7.16", "v3.8.0-rc.1", "v2.11.60"),
gh + "gtsteffaniak/filebrowser/": (200, {}, []),
"https://github.com/gtsteffaniak/filebrowser/releases.atom": (200, {}, (
"2.0.0-beta"
"fixes CVE-2026-1111").encode()),
"https://hub.docker.com/v2/repositories/gtstef/filebrowser/": TAGS("1.5.6-stable", "1.5.7-stable", "1.6.0-beta"),
gh + "cloudflare/cloudflared/": (200, {}, []),
"https://github.com/cloudflare/cloudflared/releases.atom": EMPTY_FEED,
"https://hub.docker.com/v2/repositories/cloudflare/cloudflared/": TAGS("2026.9.3", "2026.10.3", "2026.10.4"),
"https://blog.cloudflare.com/rss/": CF_BLOG,
"https://raw.githubusercontent.com/cloudflare/cloudflared/": CF_NOTES,
gh + "down/repo/": w.SourceError("timed out"),
"https://hub.docker.com/v2/repositories/library/redis/": TAGS("7-alpine", "7.4.6-alpine"),
"https://hub.docker.com/v2/repositories/library/postgres/": TAGS("16-alpine", "17-alpine"),
gh + "nextcloud/security-advisories/": (200, {}, [
{"ghsa_id": "GHSA-nc01", "summary": "server", "vulnerabilities": [
{"package": {"name": "Server"}, "vulnerable_version_range": ">= 32.0.0, >= 33.0.0, >= 34.0.0",
"patched_versions": "32.0.12, 33.0.6, 34.0.1"}]},
{"ghsa_id": "GHSA-nc02", "summary": "the Deck app", "vulnerabilities": [
{"package": {"name": "Deck"}, "vulnerable_version_range": "< 99.0.0", "patched_versions": "99.0.0"}]}]),
gh + "nextcloud/server/": (200, {}, []),
"https://github.com/nextcloud/server/releases.atom": EMPTY_FEED,
"https://hub.docker.com/v2/repositories/library/nextcloud/": TAGS("34.0.0-apache", "34.0.4-apache"),
gh + "odd/one/": (200, {}, [{"ghsa_id": "GHSA-odd1", "summary": "odd", "vulnerabilities": [
{"package": {"name": "x"}, "vulnerable_version_range": "all versions", "patched_versions": None}]}]),
"https://github.com/odd/one/releases.atom": EMPTY_FEED,
"https://hub.docker.com/v2/repositories/odd/one/": TAGS("1.0.0"),
}
posts = {("Go/stdlib", "1.27.2"): {"vulns": [{"id": "GO-2026-6809", "summary": "net/http", "affected": [
{"ranges": [{"events": [{"introduced": "0"}, {"fixed": "1.27.3"}]}]}]}]}, ("Go/stdlib", "1.27.3"): {}}
tr13 = item("traefik", "v3.7.13", "github:traefik/traefik")
tr16 = item("traefik", "v3.7.16", "github:traefik/traefik")
fb = item("gtstef/filebrowser", "1.5.6-stable", "github:gtsteffaniak/filebrowser")
cf = item("cloudflare/cloudflared", "2026.10.3", "github:cloudflare/cloudflared feed:cloudflare-blog vulnnotes:cloudflared")
down = item("down/one", "1.0.0", "github:down/repo")
red = item("redis", "7-alpine", "github:redis/redis", group="app")
pg = item("postgres", "16-alpine", "gap:PostgreSQL announces on postgresql.org only", group="app")
nc4 = item("nextcloud", "34.0.4-apache", "github:nextcloud/server advisories:nextcloud/security-advisories", group="app")
nc0 = item("nextcloud", "34.0.0-apache", "github:nextcloud/server advisories:nextcloud/security-advisories", group="app")
odd = item("odd/one", "1.0.0", "github:odd/one", group="app")
go2 = item("golang", "1.27.2-trixie", "osv:Go/stdlib", group="build")
go3 = item("golang", "1.27.3-trixie", "osv:Go/stdlib", group="build")
for x in (go2, go3):
x["no_registry"] = False
routes["https://hub.docker.com/v2/repositories/library/golang/"] = TAGS("1.27.2-trixie", "1.27.3-trixie")
items = run_items([tr13, tr16, fb, cf, down, red, pg, nc4, nc0, odd, go2, go3], routes, posts)
check("W1 traefik v3.7.13 -> ADVISORY, fixed v3.7.14", tr13["class"] == "ADVISORY" and
tr13["advisories"][0]["fixed"] == "v3.7.14", tr13.get("advisories"))
check("W1 negative: traefik v3.7.16 -> CURRENT (the rc is no fix, v2 is another channel)",
tr16["class"] == "CURRENT", (tr16["class"], tr16.get("behind")))
check("W1 Nextcloud 34.0.4 -> no Server advisory (fixed 34.0.1); the Deck app's advisory is not ours",
nc4["class"] == "CURRENT", (nc4["class"], nc4["advisories"]))
check("W1 Nextcloud 34.0.0 -> ADVISORY GHSA-nc01", nc0["class"] == "ADVISORY" and
[a["id"] for a in nc0["advisories"]] == ["GHSA-nc01"], nc0["advisories"])
check("W1 an unreadable range is LISTED (unsure), never dropped", odd["class"] == "ADVISORY" and
odd["advisories"][0]["unsure"], odd.get("advisories"))
check("W1 OSV: Go 1.27.2 -> GO-2026-6809 fixed 1.27.3; Go 1.27.3 -> none",
go2["class"] == "ADVISORY" and go2["advisories"][0]["fixed"] == "1.27.3" and go3["class"] == "CURRENT",
(go2["class"], go3["class"]))
ids = [a["id"] for a in cf["advisories"]]
check("W2 the recorded Cloudflare blog item -> ADVISORY for cloudflared (and only that item)",
"feed:cloudflare-blog:01M5DK5J81T0VC8YTBR2FKD8A3" in ids and not any(i.endswith(":x") for i in ids), ids)
check("W2 the VULN- marker in 2026.10.4 -> ADVISORY fixed 2026.10.4; 2026.10.2's plain bump is not one",
any(a["fixed"] == "2026.10.4" and "VULN-176088" in a["id"] for a in cf["advisories"]) and
not any("2026.10.2" in a["id"] for a in cf["advisories"]), ids)
cf_old = item("cloudflare/cloudflared", "2026.9.3", "feed:cloudflare-blog")
run_items([cf_old], routes, today=datetime.date(2026, 12, 30))
check("W2 the same blog item 71 days later -> nothing (30-day window)", not cf_old["advisories"], cf_old["advisories"])
cf_9 = item("cloudflare/cloudflared", "2026.9.3", "feed:cloudflare-blog")
run_items([cf_9], routes)
check("W2 positive control at the pin it hit: cloudflared 2026.9.3 on 2026-10-26 -> ADVISORY", cf_9["class"] == "ADVISORY")
check("W3 filebrowser 1.5.6-stable -> BEHIND 1.5.7-stable (the beta is no channel, its CVE no fix)",
fb["class"] == "BEHIND" and fb["behind"] == "1.5.7-stable", (fb["class"], fb.get("behind"), fb["advisories"]))
check("W3 redis 7-alpine -> FLOATING", red["class"] == "FLOATING", red["class"])
check("W3 postgres gap -> GAP (BEHIND still computed: no newer 16-…)", pg["class"] == "GAP", pg["class"])
ab = item("msdeluise/plant-it", "0.10.0", "github:MDeLuise/plant-it", group="app")
ab["abandoned"] = True
seen_before = len(FakeNet({}).seen)
run_items([ab], {})
check("W3 an abandoned template -> GAP, nothing looked up (no route needed)", ab["class"] == "GAP" and
not ab["not_checked"] and seen_before == 0, (ab["class"], ab.get("not_checked")))
check("W4 a source that fails -> NOT CHECKED, with the reason", down["class"] == "NOT CHECKED" and
"timed out" in down["not_checked"][0]["why"], down.get("not_checked"))
lines = w.table(items)
check("W4 the table prints the NOT CHECKED line", any(l.startswith("NOT CHECKED") and "timed out" in l for l in lines))
def adv_item(image, aid, group="box", no_reminder=False):
it = {"image": image, "tag": "1.0.0", "owner": "t", "where": "t", "group": group, "class": "ADVISORY",
"not_checked": [], "advisories": [{"id": aid, "summary": "s", "fixed": "1.0.1", "severity": "high",
"source": "GitHub advisory", "link": "l", "unsure": False}]}
if no_reminder:
it["advisories"][0]["no_reminder"] = True
return it
def clean(image):
return {"image": image, "tag": "1.0.0", "owner": "t", "where": "t", "group": "app", "class": "CURRENT",
"not_checked": [], "advisories": []}
def down(image, source="github"):
return {"image": image, "tag": "1.0.0", "owner": "t", "where": "t", "group": "app", "class": "NOT CHECKED",
"advisories": [], "not_checked": [{"source": source, "why": "HTTP 503"}]}
def kinds(mails, only=None):
return [m["kind"] for m in mails if only is None or m["kind"] == only]
def t_memory():
print("-- W5-W7: memory and mails")
d = lambda n: TODAY + datetime.timedelta(days=n) # noqa: E731 (TODAY is a Monday)
st = {"weekly_last": TODAY.isoformat()}
m1, st, _ = w.plan([adv_item("traefik", "GHSA-1"), clean("x")], st, d(1))
check("W5 a new advisory mails that night", kinds(m1) == ["advisory"] and "[Felhom security] traefik" in m1[0]["subject"],
kinds(m1))
check("W5 the mail carries our version, the fixed version, the lane", all(s in m1[0]["text"] for s in
("Our version: 1.0.0", "Fixed in: 1.0.1", "security-fix lane (B23)")))
m2, st, _ = w.plan([adv_item("traefik", "GHSA-1")], st, d(2))
check("W5 the same advisory the next night mails NOTHING", m2 == [], kinds(m2))
m3, st, _ = w.plan([adv_item("traefik", "GHSA-1")], st, d(8))
adv3 = [m for m in m3 if m["kind"] == "advisory"]
check("W5 still open 7 days later -> one reminder", len(adv3) == 1 and "REMINDER" in adv3[0]["text"], kinds(m3))
m4, st, _ = w.plan([adv_item("traefik", "GHSA-1")], st, d(9))
check("W5 and not again the next night", kinds(m4, "advisory") == [], kinds(m4))
m5, st, _ = w.plan([adv_item("cloudflared", "feed:x", no_reminder=True)], st, d(10))
m6, st, _ = w.plan([adv_item("cloudflared", "feed:x", no_reminder=True)], st, d(20))
check("W5 a vendor notice mails once and never repeats", kinds(m5, "advisory") == ["advisory"] and
kinds(m6, "advisory") == [], (kinds(m5), kinds(m6)))
_, st, _ = w.plan([clean("traefik")], st, d(21))
check("W5 a pin checked clean forgets its closed advisory", "traefik|GHSA-1" not in st["advisories"], st["advisories"])
m7, st, _ = w.plan([adv_item("box/prog", "GHSA-9"), adv_item("app/prog", "GHSA-8", group="app")], st, d(22))
a7 = [m for m in m7 if m["kind"] == "advisory"]
check("W5 box programs first in the subject", a7 and a7[0]["subject"].startswith("[Felhom security] prog"),
a7[0]["subject"] if a7 else "")
print("-- W6: a source down")
st = {"weekly_last": TODAY.isoformat()}
n1, st, _ = w.plan([down("a")], st, d(1))
n2, st, _ = w.plan([down("a")], st, d(2))
n3, st, _ = w.plan([down("a")], st, d(3))
n4, st, _ = w.plan([clean("a")], st, d(4))
n5, st, _ = w.plan([down("a")], st, d(5))
check("W6 one night down -> no mail", n1 == [], kinds(n1))
check("W6 two nights -> one mail naming the source", kinds(n2) == ["not_checked"] and "github" in n2[0]["subject"], kinds(n2))
check("W6 a third night -> no second mail", n3 == [], kinds(n3))
check("W6 back up -> the streak resets (one more night down -> no mail)", n4 == [] and n5 == [] and
st["failures"]["github"]["streak"] == 1, st.get("failures"))
print("-- W7: weekly and seed")
wk, st2, _ = w.plan([clean("a")], {"weekly_last": "2026-10-19"}, TODAY)
check("W7 Monday -> the weekly summary", kinds(wk) == ["weekly"], kinds(wk))
wk2, _, _ = w.plan([clean("a")], st2, TODAY)
check("W7 not twice the same Monday", wk2 == [], kinds(wk2))
wk3, _, _ = w.plan([clean("a")], {"weekly_last": "2026-10-19"}, d(1))
check("W7 a Tuesday whose last weekly is 8 days old -> the weekly (a missed Monday)", kinds(wk3) == ["weekly"], kinds(wk3))
sm, sst, would = w.plan([adv_item("traefik", "GHSA-1"), down("b")], {}, d(1), seed=True)
check("W7 a seed mails nothing, records the advisory, and says what it would have mailed",
sm == [] and "traefik|GHSA-1" in sst["advisories"] and kinds(would) == ["advisory"], (kinds(sm), kinds(would)))
unread = clean("immich")
unread["adv_unread"] = True
_, sst, _ = w.plan([adv_item("traefik", "GHSA-1"), unread], {}, d(1), seed=True)
check("W7 a pin the seed could not read is PENDING", sst["seed_pending"] == ["immich"], sst.get("seed_pending"))
first = adv_item("immich", "GHSA-im1", group="app")
newapp = adv_item("newapp", "GHSA-na1", group="app")
pm, sst, _ = w.plan([adv_item("traefik", "GHSA-1"), first, newapp], sst, d(2))
check("W7 its first real read is recorded SILENTLY; a pin added later still mails",
"immich|GHSA-im1" in sst["advisories"] and sst["seed_pending"] == [] and
[m for m in pm if m["kind"] == "advisory" and "GHSA-im1" in m["text"]] == [] and
any(m["kind"] == "advisory" and "GHSA-na1" in m["text"] for m in pm), (kinds(pm), sst.get("seed_pending")))
def t_inventory():
print("-- W8: the inventory and its gate")
_, p = inv.parse_sources("gap:")
check("W8 a gap with no reason is a problem", p and "reason" in p[0], p)
s, p = inv.parse_sources('"github:a/b advisories:c/d"')
check("W8 two sources parse", s == [("github", "a/b"), ("advisories", "c/d")] and not p, (s, p))
_, p = inv.parse_sources("gitlab:a/b")
check("W8 an unknown kind is a problem", p and "unknown" in p[0], p)
ws = tempfile.mkdtemp(prefix="sw-inv-")
try:
d = os.path.join(ws, "templates", "demo")
os.makedirs(d)
open(os.path.join(d, "docker-compose.yml"), "w").write("services:\n a:\n image: demo/a:1.0.0\n b:\n image: demo/b:2.0.0\n")
open(os.path.join(d, ".felhom.yml"), "w").write('display_name: "x"\nupstream:\n demo/a: "github:demo/a"\n # demo/b: "github:demo/b"\n')
r = subprocess.run([sys.executable, os.path.join(HERE, "security_watch_inventory.py"), "--gate=catalog", ws],
capture_output=True, text=True)
check("W8 a decoy pin whose line is only a comment REFUSES (rc 1, names demo/b)",
r.returncode == 1 and "demo/b" in r.stdout, (r.returncode, r.stdout[-300:]))
finally:
shutil.rmtree(ws, ignore_errors=True)
def t_ranges():
"""W9 (R-1310, measured 2026-10-27): two shapes a hand read of all 38 flagged pins found wrong."""
print("-- W9: notices that are not about the product; an empty fix with a closed range")
ci = {"vulnerabilities": [{"package": {"ecosystem": "actions", "name": "home-assistant/actions/helpers/version"},
"vulnerable_version_range": "< September 5, 2023", "patched_versions": "September 5, 2023"}]}
a, _ = w.ghsa_affects(ci, w.version_of("2026.9.4"), "home-assistant/core")
check("W9 a GitHub Actions advisory (the project's CI) is not about the image -> not listed", a is False, a)
prod = {"vulnerabilities": [{"package": {"ecosystem": "go", "name": "code.vikunja.io/api"},
"vulnerable_version_range": ">= 2.3.0, <= 2.6.0", "patched_versions": ""}]}
a, fixed = w.ghsa_affects(prod, w.version_of("2.6.0"), "go-vikunja/vikunja")
check("W9 an empty fix with a range ending at 2.6.0 -> ADVISORY, fixed after 2.6.0 (not 'no fixed version')",
a is True and "after 2.6.0" in fixed and "no fixed version" not in fixed, (a, fixed))
a, fixed = w.ghsa_affects(prod, w.version_of("2.7.0"), "go-vikunja/vikunja")
check("W9 negative: the same notice at 2.7.0 -> not listed", a is False, (a, fixed))
open_ = {"vulnerabilities": [{"package": {"ecosystem": "go", "name": "x"},
"vulnerable_version_range": ">= 2.3.0", "patched_versions": ""}]}
a, fixed = w.ghsa_affects(open_, w.version_of("2.6.0"), "o/r")
check("W9 control: an open range with no fix still says 'no fixed version named yet'",
a is True and fixed == "no fixed version named yet", (a, fixed))
def t_runtime():
print("-- W10: the runtime inside an image")
reads = []
def reader_for(answers):
def read(ref):
reads.append(ref)
return dict(answers[ref], ref=ref)
return read
answers = {
"ex/node-old:1.0.0": {"digest": "sha256:a", "node": "24.18.1", "unchecked": []},
"ex/node-new:1.0.0": {"digest": "sha256:b", "node": "24.22.1", "unchecked": []},
"ex/py-old:1.0.0": {"digest": "sha256:c", "brotli": "1.2.0", "unchecked": []},
"ex/py-new:1.0.0": {"digest": "sha256:d", "brotli": "1.2.1", "unchecked": []},
"ex/node-eol:1.0.0": {"digest": "sha256:e", "node": "20.19.5", "unchecked": []},
"ex/broken:1.0.0": {"digest": "sha256:f", "unchecked": ["layer sha256:1234 unreadable: ReadError"]},
"ex/private:1.0.0": {"unchecked": ["registry gitea.dooplex.hu: no anonymous token route"]},
}
items = [item(r.split(":")[0], "1.0.0", "self", group="app") for r in answers]
for it in items:
it["sources"] = []
routes = {"https://hub.docker.com/v2/repositories/": TAGS("1.0.0")}
w.run(items, FakeNet(routes), TODAY, False, runtime=reader_for(answers))
by = {it["image"]: it for it in items}
a = by["ex/node-old"]["advisories"]
check("W10 Node 24.18.1 -> ADVISORY from runtime, CVE-2026-111497, fixed 24.22.1",
by["ex/node-old"]["class"] == "ADVISORY" and a and a[0]["source"] == "runtime" and
a[0]["id"] == "CVE-2026-111497" and "24.22.1" in a[0]["fixed"], (by["ex/node-old"]["class"], a))
check("W10 negative control: Node 24.22.1 -> no advisory", by["ex/node-new"]["class"] == "CURRENT",
(by["ex/node-new"]["class"], by["ex/node-new"]["advisories"]))
check("W10 Python Brotli 1.2.0 -> ADVISORY, fixed 1.2.1", by["ex/py-old"]["class"] == "ADVISORY" and
"1.2.1" in by["ex/py-old"]["advisories"][0]["fixed"], by["ex/py-old"]["advisories"])
check("W10 negative control: Brotli 1.2.1 -> no advisory", by["ex/py-new"]["class"] == "CURRENT", by["ex/py-new"]["class"])
check("W10 Node 20 (no fixed release on its line) -> ADVISORY naming the end of life",
by["ex/node-eol"]["class"] == "ADVISORY" and "end of life" in by["ex/node-eol"]["advisories"][0]["summary"],
by["ex/node-eol"]["advisories"])
check("W10 a layer read error -> NOT CHECKED with source runtime", by["ex/broken"]["class"] == "NOT CHECKED" and
by["ex/broken"]["not_checked"][0]["source"] == "runtime", (by["ex/broken"]["class"], by["ex/broken"]["not_checked"]))
check("W10 a registry with no route -> a note, not a failure", by["ex/private"]["class"] == "CURRENT" and
any("runtime" in n for n in by["ex/private"]["notes"]), (by["ex/private"]["class"], by["ex/private"]["notes"]))
# the cache: a second night with the same digests reads no layer
cache, resolved, full = {}, [], []
def resolve(ref):
resolved.append(ref)
return answers[ref].get("digest")
def read_full(ref):
full.append(ref)
return dict(answers[ref], ref=ref)
r = w.cached_runtime_reader(cache, resolve=resolve, read=read_full)
r("ex/node-old:1.0.0")
r("ex/node-old:1.0.0")
check("W10 the result is cached by digest: one full read for two nights", full == ["ex/node-old:1.0.0"] and
len(resolved) == 2 and "sha256:a" in cache, (full, resolved, list(cache)))
r("ex/broken:1.0.0")
check("W10 an UNCHECKED result is not cached (tomorrow tries again)", "sha256:f" not in cache, list(cache))
src = open(os.path.join(HERE, "security_watch.py"), encoding="utf-8").read()
check("W10 main() hands the reader to run() (off only with --no-runtime)",
"rreader = None if \"no-runtime\" in opts else cached_runtime_reader(rcache)" in src
and "runtime=rreader)" in src)
check("W10 the reader counts what it read and from the cache (the log's RUNTIME line)",
r.stats["read"] == 2 and r.stats["cached"] == 1 and r.stats["not_checked"] == 1 and "RUNTIME read=" in src,
r.stats)
def t_ghcr_pages():
print("-- W11: ghcr.io tags are paged")
routes = {
"https://ghcr.io/token": (200, {}, {"token": "t"}),
"https://ghcr.io/v2/ex/app/tags/list?n=10000": (200, {"Link": '; rel="next"'},
{"tags": ["0.9.0", "1.0.0"]}),
"https://ghcr.io/v2/ex/app/tags/list?last=1.0.0": (200, {}, {"tags": ["1.0.1", "1.1.0"]}),
}
src = w.Sources(FakeNet(routes), TODAY, False)
names = src.tags("ghcr.io/ex/app")
check("W11 both pages are read (1.1.0 is on page 2)", names == ["0.9.0", "1.0.0", "1.0.1", "1.1.0"], names)
def t_dooplex():
print("-- W12: the DooPlex group")
d = tempfile.mkdtemp()
try:
f = os.path.join(d, "dooplex-workloads.txt")
open(f, "w").write("# header\nrommapp/romm:5.4.0\tromm-system/romm.yaml\ngitea.dooplex.hu/admin/jarr:latest\tjarr/x.yaml\n")
items = inv.dooplex_items(f, catalog_sources={"rommapp/romm": "github:rommapp/romm"})
by = {i["image"]: i for i in items}
check("W12 the list file is read, group dooplex, the catalog's sources reused",
len(items) == 2 and by["rommapp/romm"]["group"] == "dooplex" and
by["rommapp/romm"]["sources_text"] == "github:rommapp/romm", items)
check("W12 an image the catalog does not ship is a stated gap (runtime + registry only)",
(by["gitea.dooplex.hu/admin/jarr"]["sources_text"] or "").startswith("gap:"), by["gitea.dooplex.hu/admin/jarr"])
finally:
shutil.rmtree(d)
it = {"image": "rommapp/romm", "tag": "5.4.0", "owner": "romm-system/romm.yaml", "where": "dooplex", "group": "dooplex"}
a = {"id": "CVE-2026-111497", "summary": "Node.js 24.18.1 inside the image", "fixed": "24.22.1", "severity": "critical",
"source": "runtime", "link": "x", "unsure": False}
m = w.advisory_mail([(it, a)], [], TODAY)
check("W12 a DooPlex advisory names the homelab session and not the lane", "homelab" in m["text"] and
"security-fix lane" not in m["text"].split("rommapp/romm")[1], m["text"][-400:])
priv = [item("gitea.dooplex.hu/admin/jarr", "latest", "gap:DooPlex workload", group="dooplex")]
priv[0]["sources"] = [("gap", "DooPlex workload")]
w.run(priv, FakeNet({}), TODAY, False, runtime=lambda ref: {"ref": ref, "unchecked": [
"registry gitea.dooplex.hu: no anonymous token route"]})
check("W12 a DooPlex image in a private registry -> NOT CHECKED with the reason", priv[0]["class"] == "NOT CHECKED" and
"gitea.dooplex.hu" in priv[0]["not_checked"][0]["why"], (priv[0]["class"], priv[0]["not_checked"]))
wk = w.weekly_mail([dict(it, behind="5.4.1", **{"class": "BEHIND"})], TODAY)
check("W12 the weekly mail has its own DooPlex block", "DooPlex (homelab)" in wk["text"], wk["text"][:300])
def main():
t_ranges()
t_classes()
t_memory()
t_inventory()
for t in (t_runtime, t_ghcr_pages, t_dooplex):
try:
t()
except Exception as e: # noqa: BLE001 — a missing feature is a red check, not a crash
check("%s ran" % t.__name__, False, "%s: %s" % (type(e).__name__, e))
print()
if fails:
for f in fails:
print("FAIL: %s" % f)
return 1
print("security_watch tests OK — %d checks" % ran)
return 0
sys.exit(main())