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

R-1303: the nightly security watch — security_watch.py, the shared pin list + upstream gate, 35 tests, the CI job (03:40 UTC)
gates / gates (push) Successful in 3m53s
Details

Browse Source
This commit is contained in:
admin committed 2026-10-26 23:20:06 +01:00
1 parent 97ae7086c0
commit 1f62d62915
7 files changed
+1495
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.gitea/workflows/security-watch.yml
+125
View File
@@ -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 <monitoring@felhom.eu>", "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
REUSE.md
+6
View File
@@ -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 |
hub/Dockerfile
+2
View File
@@ -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
scripts/CHANGELOG.md
+16
View File
@@ -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`,
scripts/security_watch.py
+760
View File
@@ -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] <programs, box first> …"; 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=<file> [--catalog=<dir>] [--controller=<dir>] [--seed] [--no-mail]
[--test-mail] [--today=YYYY-MM-DD] [--json=<results file>]
Exit: 0 the run finished (whatever it found) · 1 a mail could not be sent · 2 the list of pins is unreadable.
"""
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 <monitoring@felhom.eu>"
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-<n> (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"<entry>(.*?)</entry>", text, re.S):
link = re.search(r'<link[^>]*href="([^"]+)"', e)
tag = link.group(1).rsplit("/", 1)[-1] if link else ""
title = html.unescape(re.sub(r"<[^>]+>", "", (re.search(r"<title>(.*?)</title>", e, re.S) or [""] * 2)[1]))
content = html.unescape(html.unescape((re.search(r"<content[^>]*>(.*?)</content>", 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"<item>(.*?)</item>", text, re.S):
def tag(t):
m = re.search(r"<%s[^>]*>(.*?)</%s>" % (t, t), it, re.S)
return html.unescape(re.sub(r"<!\[CDATA\[|\]\]>", "", 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:]))
scripts/security_watch_inventory.py
+287
View File
@@ -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/<app>/docker-compose.yml `image:` lines, and the `upstream:` block of the same template's
`.felhom.yml` — one line per image, keyed by the image without its tag.
* box controller/internal/infra/infra.go `*Image` consts, and controller/internal/infra/upstream.txt beside it.
* builds every `FROM` in controller/Dockerfile, controller/infra-images/samba/Dockerfile and felhom.eu
hub/Dockerfile, and `ARG RESTIC_VERSION`, each with a `# upstream: …` comment on the line ABOVE it.
THE SOURCE GRAMMAR (space-separated, at least one; NEW-APP-CHECKLIST.md 1.10 points here):
github:OWNER/REPO the repo's published security advisories AND its release notes (release feed)
advisories:OWNER/REPO advisories published in ANOTHER repo (Nextcloud: nextcloud/security-advisories)
osv:ECOSYSTEM/NAME an OSV.dev package at our version (today only Go/stdlib, for the Go builders)
feed:NAME a vendor feed the watch knows by name (security_watch.FEEDS: cloudflare-blog)
vulnnotes:NAME a vendor's release-notes file whose security lines carry a marker (cloudflared: VULN-)
self our own build — its code is scanned by security_scan_gate.py, not watched here
gap:<why> no machine-readable source exists; the reason is listed in every weekly mail
A pin with NO line refuses the push (`--gate`). A gap is allowed — it is a stated hole, not a silent one.
USAGE
python3 scripts/security_watch_inventory.py --list [--json] # the generated list (siblings beside this repo)
python3 scripts/security_watch_inventory.py --gate=catalog <catalog root>
python3 scripts/security_watch_inventory.py --gate=controller <controller repo root>
python3 scripts/security_watch_inventory.py --gate=hub <felhom.eu root>
Exit (gate): 0 every pin has an upstream · 1 convicted · 2 a file nobody can read.
Decoys: catalog `scripts/test_gate_decoys.py` and controller `controller/scripts/test_gate_decoys.py` (COVERS "upstream").
stdlib only — the CI runners carry python3 and git and nothing else (no PyYAML).
"""
import json
import os
import re
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
FELHOM = os.path.dirname(HERE)
WORKSPACE = os.environ.get("FELHOM_WORKSPACE") or os.path.dirname(FELHOM)
KINDS = ("github", "advisories", "osv", "feed", "vulnnotes", "self", "gap")
KNOWN_FEEDS = ("cloudflare-blog",)
KNOWN_VULNNOTES = ("cloudflared",)
REPO_RX = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9._-]+$")
IMAGE_RX = re.compile(r"^\s*image:\s*[\"']?([^\"'\s#]+)")
FROM_RX = re.compile(r"^\s*FROM\s+(?:--platform=\S+\s+)?(\S+)", re.I)
UPSTREAM_COMMENT_RX = re.compile(r"^\s*#\s*upstream:\s*(.+?)\s*$")
class InventoryError(Exception):
"""A file nobody can read — the gate exits 2, the watch counts it as NOT CHECKED."""
def parse_sources(text):
"""'github:a/b gap:…' -> ([(kind, arg)], [problems]). A gap's reason runs to the end of the line."""
text = (text or "").strip().strip('"').strip("'").strip()
out, problems = [], []
if not text:
return out, ["no source"]
rest = text
while rest:
m = re.match(r"(\w+)(?::(\S*))?\s*", rest)
if not m:
problems.append("unreadable source %r" % rest)
break
kind, arg = m.group(1), m.group(2) or ""
if kind == "gap":
reason = rest[len("gap:"):].strip() if rest.startswith("gap:") else ""
if len(reason) < 10:
problems.append("gap without a reason (at least a sentence: why no source exists)")
out.append(("gap", reason))
break
rest = rest[m.end():]
if kind not in KINDS:
problems.append("unknown source kind %r" % kind)
elif kind in ("github", "advisories") and not REPO_RX.match(arg):
problems.append("%s needs OWNER/REPO, got %r" % (kind, arg))
elif kind == "osv" and not re.match(r"^[A-Za-z.:]+/\S+$", arg):
problems.append("osv needs ECOSYSTEM/NAME, got %r" % arg)
elif kind == "feed" and arg not in KNOWN_FEEDS:
problems.append("unknown feed %r (known: %s)" % (arg, ", ".join(KNOWN_FEEDS)))
elif kind == "vulnnotes" and arg not in KNOWN_VULNNOTES:
problems.append("unknown vulnnotes %r (known: %s)" % (arg, ", ".join(KNOWN_VULNNOTES)))
elif kind == "self" and arg:
problems.append("self takes no argument")
out.append((kind, arg))
return out, problems
def split_ref(ref):
"""'ghcr.io/a/b:v1.2@sha256:…' -> ('ghcr.io/a/b', 'v1.2'). A ref with no tag reads as 'latest'."""
ref = ref.split("@", 1)[0]
head, _, last = ref.rpartition("/")
if ":" in last:
name, tag = last.split(":", 1)
return (head + "/" + name) if head else name, tag
return ref, "latest"
def _read(path):
try:
with open(path, encoding="utf-8") as f:
return f.read()
except OSError as e:
raise InventoryError("cannot read %s: %s" % (path, e))
def felhom_upstream_block(text):
"""The `upstream:` mapping of a .felhom.yml: {image: source text}. A line reader — only top-level `upstream:` and
its two-space-indented `key: value` lines count; a comment or another key's nested block never does."""
out, inside = {}, False
for line in text.splitlines():
if re.match(r"^upstream:\s*(#.*)?$", line):
inside = True
continue
if not inside:
continue
if not line.strip() or line.lstrip().startswith("#"):
continue
if not line.startswith(" "):
break # the next top-level key ends the block
m = re.match(r"^\s+([^\s:#][^\s#]*?):\s+(.+?)\s*$", line)
if m:
out[m.group(1)] = m.group(2)
return out
def catalog_items(root):
items = []
tdir = os.path.join(root, "templates")
if not os.path.isdir(tdir):
raise InventoryError("no templates/ under %s" % root)
for app in sorted(os.listdir(tdir)):
compose = os.path.join(tdir, app, "docker-compose.yml")
if not os.path.isfile(compose):
continue
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:]))
scripts/test_security_watch.py
+299
View File
@@ -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"<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"])
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())