317 lines
19 KiB
Python
317 lines
19 KiB
Python
#!/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
|
|
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.
|
|
"""
|
|
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"<feed></feed>")
|
|
# The RECORDED answer for today's case (blog.cloudflare.com/rss/, item of 2026-10-20, trimmed).
|
|
CF_BLOG = (200, {}, ("<rss><channel><item><title><![CDATA[8 eMudhra certificates are compromised. Here's what you need to "
|
|
"know]]></title><link>https://blog.cloudflare.com/emudhra-certificate-compromise/</link>"
|
|
"<guid isPermaLink=\"false\">01M5DK5J81T0VC8YTBR2FKD8A3</guid><pubDate>Tue, 20 Oct 2026 09:00:00 GMT"
|
|
"</pubDate><content:encoded><![CDATA[<p>Eight eMudhra intermediate certificates are compromised ... "
|
|
"Cloudflare Tunnel (cloudflared): update to 2026.10.2 ...</p>]]></content:encoded></item>"
|
|
"<item><title>Something else</title><link>https://blog.cloudflare.com/x/</link><guid>x</guid>"
|
|
"<pubDate>Tue, 20 Oct 2026 09:00:00 GMT</pubDate><description>Workers news, no tunnel.</description>"
|
|
"</item></channel></rss>").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, {}, (
|
|
"<feed><entry><title>2.0.0-beta</title><link href=\"https://github.com/x/releases/tag/v1.5.8-beta\"/>"
|
|
"<content>fixes CVE-2026-1111</content></entry></feed>").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 main():
|
|
t_classes()
|
|
t_memory()
|
|
t_inventory()
|
|
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())
|