diff --git a/.gitea/workflows/security-watch.yml b/.gitea/workflows/security-watch.yml new file mode 100644 index 00000000..3b39e106 --- /dev/null +++ b/.gitea/workflows/security-watch.yml @@ -0,0 +1,125 @@ +# security-watch — every night, read the security notices of everything Felhom ships, and mail the operator +# (admin@felhom.eu, never a household) what affects a pinned version (R-1303). The logic, its classes and its mails: +# scripts/security_watch.py; the list of pins: scripts/security_watch_inventory.py; why these sources: +# documentation/audits/security-watch-2026-10-26/SPIKE.md. +# +# THE MEMORY is documentation/security-watch/state.json on `main`, committed back by this job with its own Actions +# token. Why there and not elsewhere (2026-10-26): the two runners each have their OWN disk (act-runner-data, +# act-runner-2-data — a job lands on either), so a runner file would be two memories; Actions cache needs a node +# `uses:` step, which this python-and-git runner cannot run; anything on the DooPlex host is a host change. A file in +# the repo is one memory, readable by a person, and its history says when each advisory was first mailed. +# If the push of the memory fails, the job fails and the alarm below mails — the next night may then mail an advisory +# twice, which is the safe direction. +# +# Manual run (`workflow_dispatch`, mode): +# run a normal night, now +# seed record every advisory open now as known and mail nothing new (the first run; prints what it WOULD mail) +# dry no mail, no memory change — read the result table in the log +# test-mail send ONE clearly marked test advisory to the operator; touches nothing else +name: security-watch +on: + schedule: + - cron: "40 3 * * *" # 03:40 UTC: after the 03:20 gates run + workflow_dispatch: + inputs: + mode: + description: "run | seed | dry | test-mail" + default: "dry" + +jobs: + watch: + runs-on: felhom-gates + timeout-minutes: 20 + permissions: + contents: write + steps: + - name: Fetch main (latest, for the memory) and the two sibling repos + run: | + echo "WATCH_STARTED_AT=$(date +%s)" >> "$GITHUB_ENV" + base=http://gitea.gitea-system.svc.cluster.local:3000/admin + git init -q . && git remote add origin $base/felhom.eu.git + git fetch -q --depth 1 origin main && git checkout -q -B main FETCH_HEAD + echo "felhom.eu main at $(git rev-parse --short=12 HEAD)" + for r in felhom-controller app-catalog-felhom.eu; do + git init -q ../$r && git -C ../$r remote add origin $base/$r.git + git -C ../$r fetch -q --depth 1 origin main && git -C ../$r checkout -q FETCH_HEAD + echo "$r at $(git -C ../$r rev-parse --short=12 HEAD)" + done + + - name: Run the watch + env: + RESEND_API_KEY: ${{ secrets.RESEND_API_KEY }} + GH_ADVISORY_TOKEN: ${{ secrets.GH_ADVISORY_TOKEN }} + MODE: ${{ github.event.inputs.mode }} + run: | + set -u + mkdir -p documentation/security-watch + mode="${MODE:-run}" + [ "${GITHUB_EVENT_NAME:-}" = "schedule" ] && mode=run + echo "mode: $mode" + case "$mode" in + test-mail) python3 scripts/security_watch.py --test-mail; exit $? ;; + dry) extra="--no-mail" ; state=/tmp/state-dry.json + cp documentation/security-watch/state.json "$state" 2>/dev/null || true ;; + seed) extra="--seed" ; state=documentation/security-watch/state.json ;; + run) extra="" ; state=documentation/security-watch/state.json ;; + *) echo "unknown mode $mode"; exit 2 ;; + esac + python3 scripts/security_watch.py --state="$state" --catalog=../app-catalog-felhom.eu \ + --controller=../felhom-controller --json=/tmp/watch-results.json $extra + + - name: Commit the memory (only when it changed) + if: github.event.inputs.mode != 'dry' && github.event.inputs.mode != 'test-mail' + env: + PUSH_TOKEN: ${{ secrets.GITEA_TOKEN }} + run: | + set -u + f=documentation/security-watch/state.json + if [ ! -f "$f" ]; then + echo "no memory file was written - nothing to commit"; exit 0 + fi + if git ls-files --error-unmatch "$f" >/dev/null 2>&1 && git diff --quiet -- "$f"; then + echo "memory unchanged - nothing to commit"; exit 0 + fi + cp "$f" /tmp/state.json + auth=$(python3 -c 'import base64,os;print(base64.b64encode(("x:"+os.environ["PUSH_TOKEN"]).encode()).decode())') + for attempt in 1 2 3; do + git fetch -q --depth 1 origin main && git checkout -q -B main FETCH_HEAD + mkdir -p "$(dirname "$f")" && cp /tmp/state.json "$f" + git add "$f" + git -c user.name="felhom security watch" -c user.email=monitoring@felhom.eu \ + commit -q -m "security-watch: memory of $(date -u +%F) (R-1303). Docs only." || { echo "nothing to commit"; exit 0; } + if git -c http.extraHeader="Authorization: Basic $auth" push -q origin HEAD:main; then + echo "memory pushed: $(git rev-parse --short=12 HEAD)"; exit 0 + fi + echo "push refused (attempt $attempt) - re-fetching main" + sleep 5 + done + echo "MEMORY NOT SAVED after 3 attempts"; exit 1 + + - name: Alarm on failure + if: failure() + env: + RESEND_API_KEY: ${{ secrets.RESEND_API_KEY }} + run: | + python3 - <<'PY' + import json, os, sys, time, urllib.request + key = os.environ.get("RESEND_API_KEY", "") + if not key: + sys.exit("ALARM FAILED: RESEND_API_KEY is empty") + run = os.environ.get("GITHUB_RUN_ID", "?") + link = "https://gitea.dooplex.hu/%s/actions/runs/%s" % (os.environ.get("GITHUB_REPOSITORY", "?"), run) + started = os.environ.get("WATCH_STARTED_AT", "") + elapsed = "%d s" % (int(time.time()) - int(started)) if started.isdigit() else "unknown" + print("ALARM-LINK " + link) + body = json.dumps({"from": "Felhom CI ", "to": ["admin@felhom.eu"], + "subject": "[felhom CI] security watch FAILED", + "text": ("The nightly security watch did not finish (R-1303).\n\nRun : %s\nElapsed : %s\n\n" + "Until it runs again, silence from the watch does NOT mean that nothing affects us.\n" + "The run log names the step: the watch itself, a mail it could not send, or its memory " + "it could not save.\n") % (link, elapsed)}).encode() + req = urllib.request.Request("https://api.resend.com/emails", data=body, method="POST", headers={ + "Authorization": "Bearer %s" % key, "Content-Type": "application/json", "User-Agent": "felhom-ci/1.0"}) + with urllib.request.urlopen(req, timeout=30) as r: + print("RESEND-ACCEPTED id=%s" % json.load(r)["id"]) + PY diff --git a/REUSE.md b/REUSE.md index 224e570e..4b2427b2 100644 --- a/REUSE.md +++ b/REUSE.md @@ -159,6 +159,12 @@ | `scheduleDaily` | hub/cmd/hub/main.go (~L449) | `(ctx, name, "HH:MM", fn, logger)` | Daily jobs in Europe/Budapest (prune etc.) | Blocking — run as goroutine. `parseHM` returns 0,0 (midnight) on bad input. | | `hu_grep.py` | scripts/hu_grep.py | `PATTERN --anchor ASCII [--negative TEXT] PATH…` | ANY search for Hungarian (accented) text in files — prints file:line hits, and refuses to report a zero unless an ASCII anchor hits and a negative control misses (R-364) | Reads bytes in Python, never via a shell; refuses a pattern that arrived transformed (octal escapes, U+FFFD). Exit 0 found · 1 tested zero · 2 refused | +### The list of everything we ship, with its upstream (R-1303, scripts/) +- `scripts/security_watch_inventory.py` — THE reader of every pin (catalog `image:` + `.felhom.yml` `upstream:`, + controller `infra.go` + `upstream.txt`, every Dockerfile `FROM` with its `# upstream:` comment) and of the source + grammar. The nightly `scripts/security_watch.py` and the catalog's and controller's `upstream` gates all import or call + it — never copy its parser. + ## 2. Canonical patterns (copy structure from THE named file) | Pattern | Canonical file | Key traits | diff --git a/hub/Dockerfile b/hub/Dockerfile index 42c98a3b..67aa5212 100644 --- a/hub/Dockerfile +++ b/hub/Dockerfile @@ -1,4 +1,5 @@ # R-972: a supported, patched Go and the current Alpine, pinned by version and digest (no moving tag). +# upstream: osv:Go/stdlib FROM golang:1.27.3-alpine3.24@sha256:b8c9b82affbbb82a020bbcf998dad2ba9516944ac7e883524642a481a9505e9c AS builder WORKDIR /app @@ -19,6 +20,7 @@ RUN go mod tidy && \ -ldflags "-s -w -X main.Version=${VERSION} -X main.BuildTime=${BUILD_TIME}" \ -o /felhom-hub ./cmd/hub/ +# upstream: gap:Alpine's package fixes (Samba among them) are in its secdb, which the watch does not read; BEHIND on the point release is the signal FROM alpine:3.24.2@sha256:294b683cb724975bec92580e1e685676bd4b50bda910ddb8c51d4cabeaec77e6 RUN apk add --no-cache ca-certificates tzdata diff --git a/scripts/CHANGELOG.md b/scripts/CHANGELOG.md index 44f89cc4..3c7597d5 100644 --- a/scripts/CHANGELOG.md +++ b/scripts/CHANGELOG.md @@ -1,3 +1,19 @@ +## 2026-10-26 (night) — the nightly security watch (R-1303) + +- **`security_watch.py`** (new): every night (CI `.gitea/workflows/security-watch.yml`, 03:40 UTC) it reads GitHub's + per-repo security advisories (with `GH_ADVISORY_TOKEN` when present; without it the apps' lists in a 2-night + rotation), the GitHub release feeds, the Cloudflare blog and cloudflared's `VULN-` notes, OSV for the Go builders and + the registry tags, and sorts every pin into ADVISORY / NOT CHECKED / GAP / FLOATING / BEHIND / CURRENT. Mails (operator + only): a new advisory that night (a reminder after 7 days), BEHIND weekly on Monday, a source down two nights. Its + memory is `documentation/security-watch/state.json`, committed back by the job. Spike: + `documentation/audits/security-watch-2026-10-26/SPIKE.md`. +- **`security_watch_inventory.py`** (new): the one reader of every pin and its upstream, and the `--gate=` the catalog + and the controller run (`upstream`). +- **`test_security_watch.py`** (new, 35 checks, no network): the classes with fake answers (today's cloudflared blog + item as a recorded answer), one mail per advisory, the 7-day reminder, a source down one night vs two, the weekly, the + seed, the inventory gate. Red-proved four ways (dedupe, streak, the 30-day window, a swallowed source error). +- `hub/Dockerfile`: a `# upstream:` comment above each `FROM`. + ## 2026-10-26 (evening) — CI: heavy suites by path, no second full gate run, a nightly run (R-1297 A); a decoy hole closed - **`script_tests_gate.py` `PATH_FILTERED`:** on a CI branch push with a known file list, `test_gate_decoys.py`, diff --git a/scripts/security_watch.py b/scripts/security_watch.py new file mode 100644 index 00000000..cf9740c8 --- /dev/null +++ b/scripts/security_watch.py @@ -0,0 +1,760 @@ +#!/usr/bin/env python3 +"""security_watch.py — the nightly watch for security notices on everything Felhom ships (R-1303). + +WHY. Cloudflare's eMudhra notice (R-1300) reached us by mail; our own check (check-infra-pins.py, monthly, by hand) would +have said only „a newer version exists", up to four weeks late. This job reads, every night, the places each program +publishes its security notices, and sorts every pin into exactly one class: + + ADVISORY a notice affects the version we pin (with the fixed version when the notice gives one) + NOT CHECKED a source this pin needs failed tonight — printed and counted, never a silent pass + GAP the pin has no machine-readable source (a stated `gap:` line); BEHIND is still shown + FLOATING the tag names no exact version (`redis:7-alpine`), so no range can be matched; BEHIND still shown + BEHIND a newer version exists in the same channel, and no notice + CURRENT none of the above + +WHAT IT READS (the spike that chose these: documentation/audits/security-watch-2026-10-26/SPIKE.md): + github:O/R GitHub's per-repo security advisories (API; GH_ADVISORY_TOKEN when present) + the release feed + (github.com/O/R/releases.atom — no API quota) for notes newer than our pin that name a CVE/GHSA/ + vulnerability/security fix + advisories:O/R the per-repo advisories of ANOTHER repo (Nextcloud) + osv:Go/stdlib OSV.dev for our Go builder version + 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) +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): + * ADVISORY: one mail that night with every NEW advisory (and a reminder for each still open after 7 days); + subject „[Felhom security] …"; each entry says what, our version, the fixed version, the + severity, the link, and „Next: the security-fix lane (B23)". + * BEHIND: one summary on Monday (or when the last one is older than 8 days), box programs first, then apps; GAP, + FLOATING and tonight's NOT CHECKED are listed in it too. + * NOT CHECKED: when the same source failed two nights in a row — once per streak. + +MEMORY: a JSON state file (`--state`), committed back to felhom.eu by the CI job (documentation/security-watch/ +state.json; the reasons in the workflow). A missing state file means: seed (record tonight's advisories as known, +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= [--catalog=] [--controller=] [--seed] [--no-mail] + [--test-mail] [--today=YYYY-MM-DD] [--json=] +Exit: 0 the run finished (whatever it found) · 1 a mail could not be sent · 2 the list of pins is unreadable. +""" +import concurrent.futures +import datetime +import email.utils +import html +import json +import os +import re +import sys +import urllib.error +import urllib.request +import zlib + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import security_watch_inventory as inv # noqa: E402 + +UA = "felhom-security-watch/1.0 (+https://felhom.eu)" +MAIL_TO = "admin@felhom.eu" +MAIL_FROM = "Felhom security watch " +REMIND_DAYS = 7 +FEED_DAYS = 30 + +FEEDS = { + # The only automated way to Cloudflare's eMudhra notice (SPIKE.md, P1): the blog carries full text. + "cloudflare-blog": {"url": "https://blog.cloudflare.com/rss/", "keywords": ("cloudflared",)}, +} +VULNNOTES = { + # Cloudflare marks security fixes in cloudflared's notes with VULN- (2026.10.0, 2026.10.4). + "cloudflared": {"url": "https://raw.githubusercontent.com/cloudflare/cloudflared/master/RELEASE_NOTES", + "marker": "VULN-"}, +} +STRONG = re.compile(r"\bCVE-\d{4}-\d{3,}|\bGHSA(?:-[0-9a-z]{4}){3}\b|vulnerab|security (?:fix|issue|update|release|" + r"advisory|patch|vulnerability)|\[security\]", re.I) +FEED_WORDS = re.compile(r"vulnerab|compromis|\bCVE-\d{4}|security (?:update|fix|advisory|issue|notice)|exploit", re.I) + + +class SourceError(Exception): + pass + + +# ── network (the tests replace Net) ────────────────────────────────────────────────────────────────────────────────── +class Net: + def __init__(self, gh_token=""): + self.gh_token = gh_token + self.requests = {} + + def _count(self, url): + host = url.split("/")[2] + self.requests[host] = self.requests.get(host, 0) + 1 + + def get(self, url, headers=None, timeout=30): + self._count(url) + h = {"User-Agent": UA} + h.update(headers or {}) + try: + with urllib.request.urlopen(urllib.request.Request(url, headers=h), timeout=timeout) as r: + return r.status, dict(r.headers), r.read() + except urllib.error.HTTPError as e: + return e.code, dict(e.headers or {}), e.read() if hasattr(e, "read") else b"" + except Exception as e: # noqa: BLE001 — every transport failure is a NOT CHECKED, named + raise SourceError("%s: %s" % (url.split("?")[0], e)) + + def post_json(self, url, body, timeout=30): + self._count(url) + req = urllib.request.Request(url, data=json.dumps(body).encode(), method="POST", + headers={"User-Agent": UA, "Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.load(r) + except Exception as e: # noqa: BLE001 + raise SourceError("%s: %s" % (url, e)) + + +# ── versions ───────────────────────────────────────────────────────────────────────────────────────────────────────── +def version_of(tag): + m = re.match(r"^v?(\d+(?:\.\d+)*)", tag or "") + return tuple(int(x) for x in m.group(1).split(".")) if m else None + + +def vcmp(a, b): + n = max(len(a), len(b)) + a, b = a + (0,) * (n - len(a)), b + (0,) * (n - len(b)) + return (a > b) - (a < b) + + +PRERELEASE = re.compile(r"(?:alpha|beta|rc|dev|nightly|preview|pre|ea)[.\d-]*$|-(?:a|b)\d", re.I) + + +def is_floating(tag): + """A tag with fewer than three numbers (`redis:7.4-alpine`, `postgres:16-…`) takes the line's newest at install — + there is no exact version to match a notice's range against.""" + v = version_of(tag) + return v is None or len(v) < 3 + + +def same_channel(ours, other): + """Same major — except calendar versions (cloudflared 2026.x), whose first number is a year.""" + return ours[0] >= 1000 or ours[0] == other[0] + + +def range_affects(rng, ours, package=""): + """GHSA's free-text range ('>= v3.0.0, <= v3.7.15', '≤ 1.36.0', 'v3.2.2', 'server <= 3.0.3') -> True/False/None. + A clause of ONLY lower bounds with more than one of them is Nextcloud's per-branch shape ('>= 32.0.0, >= 33.0.0') + and reads as OR; every other clause is AND. A bare version means 'up to and including'.""" + if rng is None: + return None + pm = re.search(r"/v(\d+)$", package or "") + if pm and int(pm.group(1)) > 1 and int(pm.group(1)) != ours[0]: + return False + rng = rng.replace("≤", "<=").replace("≥", ">=").replace("＝", "=") + any_clause = False + for clause in rng.split("||"): + cons = re.findall(r"(<=|>=|==|<|>|=)?\s*v?(\d+(?:\.\d+)+|\d+)(?![\d.])", clause) + cons = [(op, v) for op, v in cons if op or "." in v] + if not cons: + continue + any_clause = True + res = [] + for op, v in cons: + c = vcmp(ours, tuple(int(x) for x in v.split("."))) + res.append({"<=": c <= 0, ">=": c >= 0, "<": c < 0, ">": c > 0, "=": c == 0, "==": c == 0, "": c <= 0}[op]) + lower_only = len(cons) > 1 and all(op in (">=", ">") for op, _ in cons) + if (any(res) if lower_only else all(res)): + return True + return False if any_clause else None + + +def patched_for(text, ours): + """The fixed version that applies to OUR branch: the lowest named on our major; else the lowest named above us.""" + vs = [tuple(int(x) for x in v.split(".")) for v in re.findall(r"v?(\d+(?:\.\d+)+)", text or "")] + if not vs: + return None + same = [v for v in vs if v[0] == ours[0]] + if same: + return min(same) + if all(v[0] < ours[0] for v in vs) and ours[0] < 1000: + return max(vs) # every fix is on an older branch: ours is past all of them + return min(vs) + + +# A repo whose advisory list covers more than the program we run: only these package names count. +PACKAGE_ONLY = {"nextcloud/security-advisories": re.compile(r"^server$", re.I)} + + +def ghsa_affects(adv, ours, repo=""): + """-> (affected True/False/None, fixed text). None = the range is unreadable (listed, never dropped).""" + verdicts, fixed = [], [] + only = PACKAGE_ONLY.get(repo) + for vul in adv.get("vulnerabilities") or []: + pkg = (vul.get("package") or {}).get("name") or "" + if only and not only.search(pkg.strip()): + continue + r = range_affects(vul.get("vulnerable_version_range"), ours, pkg) + p = patched_for(vul.get("patched_versions"), ours) + if p is not None and vcmp(ours, p) >= 0 and r is not False: + r = False + elif r is None and p is not None: + r = vcmp(ours, p) < 0 + verdicts.append(r) + if r: + pv = (vul.get("patched_versions") or "").strip() + fixed.append(pv if pv and pv.lower() not in ("none", "tbd", "n/a") else "no fixed version named yet") + if any(v is True for v in verdicts): + return True, ", ".join(sorted(set(fixed))) + if not verdicts or all(v is False for v in verdicts): + return False if verdicts else (None if not only else False), "" if verdicts else "the notice names no version range" + return None, "the range is unreadable — check by hand" + + +# ── sources ────────────────────────────────────────────────────────────────────────────────────────────────────────── +class Sources: + def __init__(self, net, today, rotate): + self.net, self.today, self.rotate = net, today, rotate + self._adv, self._rel, self._feed, self._notes, self._osv, self._tags = {}, {}, {}, {}, {}, {} + self.rate_limited = False + + def gh_advisories(self, repo): + if repo in self._adv: + return self._adv[repo] + out = [] + if self.rate_limited: + raise SourceError("GitHub API rate limit reached earlier tonight") + url = "https://api.github.com/repos/%s/security-advisories?per_page=100&state=published" % repo + headers = {"Accept": "application/vnd.github+json"} + if self.net.gh_token: + headers["Authorization"] = "Bearer %s" % self.net.gh_token + for _ in range(5): + st, h, body = self.net.get(url, headers) + if st in (403, 429) and str(h.get("X-RateLimit-Remaining", h.get("x-ratelimit-remaining", ""))) == "0": + self.rate_limited = True + raise SourceError("GitHub API rate limit (%s)" % st) + if st != 200: + raise SourceError("GitHub advisories for %s: HTTP %s" % (repo, st)) + out += json.loads(body or b"[]") + m = re.search(r'<([^>]+)>;\s*rel="next"', h.get("Link", h.get("link", "")) or "") + if not m: + break + url = m.group(1) + self._adv[repo] = out + return out + + def release_feed(self, repo): + if repo in self._rel: + return self._rel[repo] + st, _h, body = self.net.get("https://github.com/%s/releases.atom" % repo) + if st != 200: + raise SourceError("release feed of %s: HTTP %s" % (repo, st)) + text = body.decode("utf-8", "replace") + entries = [] + for e in re.findall(r"(.*?)", text, re.S): + link = re.search(r']*href="([^"]+)"', e) + tag = link.group(1).rsplit("/", 1)[-1] if link else "" + title = html.unescape(re.sub(r"<[^>]+>", "", (re.search(r"(.*?)", e, re.S) or [""] * 2)[1])) + content = html.unescape(html.unescape((re.search(r"]*>(.*?)", e, re.S) or [""] * 2)[1])) + content = re.sub(r"<[^>]+>", " ", content) + entries.append({"tag": tag, "title": title.strip(), "content": content, + "link": link.group(1) if link else "https://github.com/%s/releases" % repo}) + self._rel[repo] = entries + return entries + + def feed(self, name): + if name in self._feed: + return self._feed[name] + st, _h, body = self.net.get(FEEDS[name]["url"]) + if st != 200: + raise SourceError("feed %s: HTTP %s" % (name, st)) + text = body.decode("utf-8", "replace") + items = [] + for it in re.findall(r"(.*?)", text, re.S): + def tag(t): + m = re.search(r"<%s[^>]*>(.*?)" % (t, t), it, re.S) + return html.unescape(re.sub(r"", "", m.group(1))) if m else "" + items.append({"title": tag("title").strip(), "link": tag("link").strip(), + "guid": tag("guid").strip() or tag("link").strip(), "date": tag("pubDate").strip(), + "text": tag("title") + " " + tag("description") + " " + tag("content:encoded")}) + self._feed[name] = items + return items + + def vulnnotes(self, name): + if name in self._notes: + return self._notes[name] + st, _h, body = self.net.get(VULNNOTES[name]["url"]) + if st != 200: + raise SourceError("release notes %s: HTTP %s" % (name, st)) + secs, cur = {}, None + for line in body.decode("utf-8", "replace").splitlines(): + if re.match(r"^\d+(?:\.\d+)+\s*$", line): + cur = line.strip() + secs[cur] = [] + elif cur and line.strip(): + secs[cur].append(line.strip()) + self._notes[name] = secs + return secs + + def osv(self, eco_name, version): + key = (eco_name, version) + if key in self._osv: + return self._osv[key] + eco, name = eco_name.split("/", 1) + d = self.net.post_json("https://api.osv.dev/v1/query", + {"package": {"ecosystem": eco, "name": name}, "version": version}) + self._osv[key] = d.get("vulns", []) + return self._osv[key] + + def _registry_tags(self, path): + st, _h, body = self.net.get("https://auth.docker.io/token?service=registry.docker.io&scope=repository:%s:pull" % path) + if st != 200: + raise SourceError("Docker registry token for %s: HTTP %s" % (path, st)) + tok = json.loads(body).get("token", "") + st, _h, body = self.net.get("https://registry-1.docker.io/v2/%s/tags/list?n=10000" % path, + {"Authorization": "Bearer %s" % tok}) + if st != 200: + raise SourceError("Docker registry tags of %s: HTTP %s" % (path, st)) + return json.loads(body).get("tags") or [] + + def tags(self, image): + if image in self._tags: + return self._tags[image] + reg, path = registry_of(image) + names = [] + if reg == "docker.io": + url = "https://hub.docker.com/v2/repositories/%s/tags?page_size=100&ordering=last_updated" % path + for _ in range(3): + st, _h, body = self.net.get(url) + if st == 404 and not names: + names = self._registry_tags(path) # the web API misses some repos the registry serves (plant-it) + break + if st != 200: + raise SourceError("Docker Hub tags of %s: HTTP %s" % (path, st)) + d = json.loads(body) + names += [t["name"] for t in d.get("results", [])] + url = d.get("next") + if not url: + break + elif reg == "ghcr.io": + st, _h, body = self.net.get("https://ghcr.io/token?scope=repository:%s:pull" % path) + 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 [] + else: + names = None # a registry the watch does not read (our own Gitea, GitLab): stated, not a failure + self._tags[image] = names + return names + + +def registry_of(image): + first = image.split("/", 1)[0] + if "." in first or ":" in first: + reg, path = first, image.split("/", 1)[1] if "/" in image else "" + else: + reg, path = "docker.io", image + if reg == "lscr.io": + reg = "docker.io" + if reg == "docker.io" and "/" not in path: + path = "library/" + path + return reg, path + + +# ── one pin ────────────────────────────────────────────────────────────────────────────────────────────────────────── +def newest_in_channel(tag, names): + ours = version_of(tag) + if not ours or not names: + return None, None + shape = re.compile("^" + re.sub(r"\d+", r"\\d+", re.escape(tag)) + "$") + best, newer_major = None, None + for n in names: + if not shape.match(n): + continue + v = version_of(n) + if not v or len(v) != len(ours): + continue + if same_channel(ours, v): + if vcmp(v, ours) > 0 and (best is None or vcmp(v, version_of(best)) > 0): + best = n + elif v[0] > ours[0] and (newer_major is None or vcmp(v, version_of(newer_major)) > 0): + newer_major = n + return best, newer_major + + +def check_item(it, src, today): + """-> the item with `class`, `advisories`, `not_checked`, `behind`, `notes` filled in.""" + ours = version_of(it["tag"]) + floating = is_floating(it["tag"]) + advs, nc, notes = [], [], [] + kinds = [k for k, _ in it.get("sources") or []] + for kind, arg in it.get("sources") or []: + try: + if kind in ("github", "advisories"): + if src.rotate and it["group"] == "app" and (zlib.crc32(arg.encode()) + today.toordinal()) % 2: + notes.append("advisories of %s: read tomorrow (no-key rotation)" % arg) + elif floating: + notes.append("advisories of %s: not matched (floating tag)" % arg) + else: + for a in src.gh_advisories(arg): + if a.get("withdrawn_at"): + continue + hit, fixed = ghsa_affects(a, ours, arg) + if hit is False: + continue + advs.append({"id": a.get("ghsa_id") or a.get("cve_id") or "?", "source": "GitHub advisory", + "severity": a.get("severity") or "", "summary": a.get("summary") or "", + "fixed": fixed, "link": a.get("html_url") or "", + "unsure": hit is None, "published": (a.get("published_at") or "")[:10]}) + if kind == "github" and not floating: + for e in src.release_feed(arg): + v = version_of(e["tag"]) + if not v or PRERELEASE.search(e["tag"]) or vcmp(v, ours) <= 0 or not same_channel(ours, v): + continue + m = STRONG.search(e["content"]) or STRONG.search(e["title"]) + if m: + line = next((ln.strip() for ln in re.split(r"[\n.]", e["content"]) if STRONG.search(ln)), m.group(0)) + advs.append({"id": "release:%s@%s" % (arg, e["tag"]), "source": "release notes", + "severity": "", "summary": line[:200], "fixed": e["tag"], "link": e["link"], + "unsure": False, "published": ""}) + elif kind == "osv" and not floating: + ver = ".".join(str(x) for x in ours) + for vuln in src.osv(arg, ver): + fx = [] + for a in vuln.get("affected", []): + for r in a.get("ranges", []): + fx += [e["fixed"] for e in r.get("events", []) if "fixed" in e] + advs.append({"id": vuln.get("id", "?"), "source": "OSV (%s)" % arg, "severity": "", + "summary": vuln.get("summary") or vuln.get("details", "")[:160], + "fixed": ", ".join(sorted(set(fx))) or "?", "unsure": False, + "link": "https://osv.dev/vulnerability/%s" % vuln.get("id", ""), "published": ""}) + elif kind == "feed": + cut = today - datetime.timedelta(days=FEED_DAYS) + for item in src.feed(arg): + try: + d = email.utils.parsedate_to_datetime(item["date"]).date() + except Exception: # noqa: BLE001 — an undated item is judged, never dropped + d = today + low = item["text"].lower() + if d < cut or not any(k in low for k in FEEDS[arg]["keywords"]) or not FEED_WORDS.search(item["text"]): + continue + advs.append({"id": "feed:%s:%s" % (arg, item["guid"]), "source": "vendor notice (%s)" % arg, + "severity": "", "summary": item["title"], "fixed": "read the notice", "unsure": True, + "link": item["link"], "published": str(d), "no_reminder": True}) + elif kind == "vulnnotes" and not floating: + for ver, lines in src.vulnnotes(arg).items(): + v = version_of(ver) + if v and vcmp(v, ours) > 0: + for ln in lines: + if VULNNOTES[arg]["marker"] in ln: + vid = re.search(r"%s\S+" % re.escape(VULNNOTES[arg]["marker"]), ln) + advs.append({"id": "vulnnotes:%s@%s:%s" % (arg, ver, vid.group(0).rstrip(":") if vid else ""), + "source": "release notes (%s)" % arg, "severity": "", "summary": ln[:200], + "fixed": ver, "unsure": False, "published": "", + "link": VULNNOTES[arg]["url"]}) + except SourceError as e: + nc.append({"source": source_name(kind, arg), "why": str(e)}) + behind, newer_major = None, None + if it.get("no_registry"): + gh = next((a for k, a in it.get("sources") or [] if k == "github"), None) + try: + names = [e["tag"].lstrip("v") for e in src.release_feed(gh)] if gh else None + if names is None: + notes.append("no registry and no release feed to compare with") + else: + behind, newer_major = newest_in_channel(it["tag"].lstrip("v"), names) + except SourceError as e: + nc.append({"source": "github", "why": str(e)}) + elif "self" not in kinds: + try: + names = src.tags(it["image"]) + if names is None: + notes.append("registry not read (%s)" % registry_of(it["image"])[0]) + else: + behind, newer_major = newest_in_channel(it["tag"], names) + except SourceError as e: + 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) + it.update({"advisories": advs, "not_checked": nc, "behind": behind, "notes": notes}) + if it.get("problems"): + nc.append({"source": "inventory", "why": "; ".join(it["problems"])}) + if advs: + it["class"] = "ADVISORY" + elif nc: + it["class"] = "NOT CHECKED" + elif "gap" in kinds and not [k for k in kinds if k != "gap"]: + it["class"] = "GAP" + elif "self" in kinds and len(kinds) == 1: + it["class"] = "SELF" + elif floating: + it["class"] = "FLOATING" + elif behind: + it["class"] = "BEHIND" + else: + it["class"] = "CURRENT" + return it + + +def source_name(kind, arg): + return {"github": "github", "advisories": "github", "osv": "osv", "feed": "feed:" + arg, + "vulnnotes": "vulnnotes:" + arg}.get(kind, kind) + + +# ── memory and mails (pure: the tests drive this directly) ────────────────────────────────────────────────────────── +def adv_key(it, a): + return "%s|%s" % (it["image"], a["id"]) + + +def label(it): + return "%s %s (%s)" % (it["image"], it["tag"], it["owner"]) + + +def plan(items, state, today, seed=False): + """-> (mails, new_state, would_have_mailed). A mail is {"kind","subject","text"}.""" + st = json.loads(json.dumps(state or {})) + st.setdefault("version", 1) + known = st.setdefault("advisories", {}) + fails = st.setdefault("failures", {}) + iso = today.isoformat() + mails, would = [], [] + open_now, checked_images = {}, set() + for it in items: + if not it.get("not_checked"): + checked_images.add(it["image"]) + for a in it.get("advisories", []): + open_now[adv_key(it, a)] = (it, a) + new, remind = [], [] + for k, (it, a) in sorted(open_now.items()): + if k not in known: + new.append((it, a)) + known[k] = {"first": iso, "last_mailed": iso, "program": label(it), "summary": a["summary"][:160]} + elif not a.get("no_reminder") and (today - datetime.date.fromisoformat(known[k]["last_mailed"])).days >= REMIND_DAYS: + remind.append((it, a)) + known[k]["last_mailed"] = iso + for k in list(known): + if k not in open_now and k.split("|", 1)[0] in checked_images: + del known[k] # closed: its pin was checked tonight and the notice no longer applies + if new or remind: + m = advisory_mail(new, remind, today) + if seed: + would.append(m) + else: + mails.append(m) + failed_now = {} + for it in items: + for f in it.get("not_checked", []): + failed_now.setdefault(f["source"], []).append("%s — %s" % (label(it), f["why"])) + streak_mail = [] + for s in sorted(set(fails) | set(failed_now)): + rec = fails.setdefault(s, {"streak": 0, "mailed": False}) + if s in failed_now: + rec["streak"] += 1 + if rec["streak"] >= 2 and not rec["mailed"]: + streak_mail.append((s, rec["streak"], failed_now[s])) + rec["mailed"] = True + else: + del fails[s] + if streak_mail and not seed: + mails.append(not_checked_mail(streak_mail, today)) + last_weekly = st.get("weekly_last") + due = today.weekday() == 0 and last_weekly != iso + if last_weekly and (today - datetime.date.fromisoformat(last_weekly)).days >= 8: + due = True + if seed: + st["weekly_last"] = iso + st["seeded_at"] = iso + elif due: + mails.append(weekly_mail(items, today)) + st["weekly_last"] = iso + return mails, st, would + + +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)"), + " Our version: %s (%s; %s)" % (it["tag"], it["owner"], it["where"]), + " Fixed in: %s" % (a["fixed"] or "?"), + " Severity: %s" % (a["severity"] or "not given"), + " 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", + ""] + + +def advisory_mail(new, remind, today): + allp = new + remind + box = [it for it, _ in allp if it["group"] != "app"] + names = [] + for it, _ in sorted(allp, key=lambda p: (p[0]["group"] == "app", p[0]["image"])): + n = it["image"].rsplit("/", 1)[-1] + if n not in names: + names.append(n) + subject = "[Felhom security] %s — %d new advisor%s%s" % ( + ", ".join(names[:4]) + (" …" if len(names) > 4 else ""), len(new), "y" if len(new) == 1 else "ies", + ", %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"])): + lines += _adv_lines(it, a) + for it, a in sorted(remind, key=lambda p: (p[0]["group"] == "app", 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)} + + +def not_checked_mail(streaks, today): + lines = ["The security watch could not check these sources for two nights or more. Silence from the watch does", + 'not mean "safe" while this lasts.', ""] + for s, n, what in streaks: + lines.append("%s — %d nights in a row:" % (s, n)) + lines += [" " + w for w in what[:20]] + ([" … %d more" % (len(what) - 20)] if len(what) > 20 else []) + [""] + lines.append("Night of %s. The run log names every failure." % today) + return {"kind": "not_checked", "subject": "[Felhom security] the watch could not check %s" % ", ".join(s for s, _, _ in streaks), + "text": "\n".join(lines)} + + +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")] + 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"]) + for it in items for a in it.get("advisories", [])})) + lines += block("No machine-readable source (GAP) — not watched for notices:", + sorted({"%s — %s" % (it["image"], next((v for k, v in it.get("sources", []) if k == "gap"), "")) + for it in items if it.get("class") == "GAP"})) + lines += block("Floating tags (no exact version to match a notice against):", + sorted({"%s:%s (%s)" % (it["image"], it["tag"], it["owner"]) for it in items if it.get("class") == "FLOATING"})) + lines += block("NOT CHECKED tonight:", ["%s — %s" % (label(it), "; ".join(f["why"] for f in it["not_checked"])) + for it in items if it.get("not_checked")]) + counts = {} + for it in items: + counts[it.get("class", "?")] = counts.get(it.get("class", "?"), 0) + 1 + lines.append("Counts: " + ", ".join("%s %d" % kv for kv in sorted(counts.items()))) + return {"kind": "weekly", "subject": "[Felhom security] weekly: %d behind, %d advisories open" % ( + sum(1 for it in items if it.get("behind")), sum(len(it.get("advisories", [])) for it in items)), + "text": "\n".join(lines)} + + +def send(mail, key, test=False): + body = json.dumps({"from": MAIL_FROM, "to": [MAIL_TO], "subject": mail["subject"], "text": mail["text"]}).encode() + req = urllib.request.Request("https://api.resend.com/emails", data=body, method="POST", headers={ + "Authorization": "Bearer %s" % key, "Content-Type": "application/json", + # Cloudflare in front of api.resend.com refuses urllib's default agent (403 error 1010, measured 2026-08-02). + "User-Agent": "felhom-ci/1.0"}) + with urllib.request.urlopen(req, timeout=30) as r: + return json.load(r)["id"] + + +# ── the run ────────────────────────────────────────────────────────────────────────────────────────────────────────── +def run(items, net, today, rotate): + src = Sources(net, today, rotate) + # 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"] + for it in box: + check_item(it, src, today) + with concurrent.futures.ThreadPoolExecutor(max_workers=6) as ex: + list(ex.map(lambda it: check_item(it, src, today), apps)) + return items + + +def table(items): + order = {"ADVISORY": 0, "NOT CHECKED": 1, "BEHIND": 2, "GAP": 3, "FLOATING": 4, "SELF": 5, "CURRENT": 6} + out = [] + for it in sorted(items, key=lambda i: (order.get(i["class"], 9), i["group"] == "app", i["image"])): + extra = "" + if it["class"] == "ADVISORY": + extra = "; ".join("%s (fixed %s)" % (a["id"], a["fixed"]) for a in it["advisories"][:4]) + if len(it["advisories"]) > 4: + extra += "; … %d more" % (len(it["advisories"]) - 4) + elif it["class"] == "NOT CHECKED": + extra = "; ".join(f["why"] for f in it["not_checked"])[:200] + elif it.get("behind"): + extra = "newest " + it["behind"] + out.append("%-11s %-4s %-46s %-22s %s" % (it["class"], it["group"][:4], it["image"][:46], it["tag"][:22], extra)) + return out + + +def main(argv): + opts = dict(a[2:].split("=", 1) if "=" in a else (a[2:], "1") for a in argv if a.startswith("--")) + today = datetime.date.fromisoformat(opts["today"]) if "today" in opts else datetime.datetime.utcnow().date() + key = os.environ.get("RESEND_API_KEY", "") + if "test-mail" in opts: + it = {"image": "TEST-program", "tag": "1.0.0", "owner": "a test entry", "where": "no file", "group": "box"} + a = {"id": "TEST-0000", "summary": "THIS IS A TEST — not a real advisory (R-1303 live proof)", "fixed": "1.0.1", + "severity": "none (test)", "source": "test", "link": "https://felhom.eu", "unsure": False} + m = advisory_mail([(it, a)], [], today) + m["subject"] = "[Felhom security] TEST — not a real advisory: " + m["subject"].split("] ", 1)[1] + if not key: + print("TEST MAIL NOT SENT: RESEND_API_KEY is empty") + return 1 + print("TEST-MAIL RESEND-ACCEPTED id=%s subject=%s" % (send(m, key), m["subject"])) + return 0 + try: + items, problems = inv.full_inventory(catalog=opts.get("catalog"), controller=opts.get("controller")) + except inv.InventoryError as e: + print("THE LIST OF PINS IS UNREADABLE: %s" % e) + 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"), + "PRESENT — every repo's advisories tonight" if token else "ABSENT — apps' advisories in a 2-night rotation")) + run(items, net, today, rotate=not token) + for line in table(items): + print(line) + counts = {} + for it in items: + counts[it["class"]] = counts.get(it["class"], 0) + 1 + print("COUNTS " + " ".join("%s=%d" % (k.replace(" ", "_"), v) for k, v in sorted(counts.items()))) + print("REQUESTS " + " ".join("%s=%d" % kv for kv in sorted(net.requests.items()))) + state_path = opts.get("state") + state, missing = {}, True + if state_path and os.path.isfile(state_path): + 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) + 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.", ""] + for w in would: + lines += [w["subject"], w["text"], ""] + mails.append({"kind": "state_missing", "subject": "[Felhom security] the watch's memory was missing — re-seeded", + "text": "\n".join(lines) or "(no open advisory)"}) + for w in would: + print("WOULD HAVE MAILED (seed): %s" % w["subject"]) + print(w["text"]) + if "json" in opts: + json.dump(items, open(opts["json"], "w", encoding="utf-8"), indent=1, ensure_ascii=False, default=str) + rc = 0 + for m in mails: + if "no-mail" in opts: + print("MAIL (not sent, --no-mail): %s" % m["subject"]) + continue + if not key: + print("MAIL NOT SENT — RESEND_API_KEY is empty: %s" % m["subject"]) + rc = 1 + continue + try: + print("MAIL RESEND-ACCEPTED id=%s kind=%s subject=%s" % (send(m, key), m["kind"], m["subject"])) + except Exception as e: # noqa: BLE001 + print("MAIL FAILED (%s): %s" % (m["subject"], e)) + rc = 1 + if state_path and rc == 0: + with open(state_path + ".tmp", "w", encoding="utf-8") as f: + json.dump(new_state, f, indent=1, sort_keys=True, ensure_ascii=False) + f.write("\n") + os.replace(state_path + ".tmp", state_path) + print("STATE written: %s (%d known advisories)" % (state_path, len(new_state.get("advisories", {})))) + return rc + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/scripts/security_watch_inventory.py b/scripts/security_watch_inventory.py new file mode 100644 index 00000000..f54545f9 --- /dev/null +++ b/scripts/security_watch_inventory.py @@ -0,0 +1,287 @@ +#!/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 + ups = felhom_upstream_block(_read(os.path.join(tdir, app, ".felhom.yml"))) \ + if os.path.isfile(os.path.join(tdir, app, ".felhom.yml")) else {} + 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, + "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:])) diff --git a/scripts/test_security_watch.py b/scripts/test_security_watch.py new file mode 100644 index 00000000..760dbc51 --- /dev/null +++ b/scripts/test_security_watch.py @@ -0,0 +1,299 @@ +#!/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"") +# 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"]) + 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))) + + +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())