789 lines
40 KiB
Python
789 lines
40 KiB
Python
#!/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."""
|
||
if it.get("abandoned"):
|
||
it.update({"advisories": [], "not_checked": [], "behind": None, "class": "GAP",
|
||
"notes": ["abandoned upstream (lifecycle: abandoned) — no fix will ever come; not looked up"]})
|
||
if it.get("problems"):
|
||
it["not_checked"].append({"source": "inventory", "why": "; ".join(it["problems"])})
|
||
it["class"] = "NOT CHECKED"
|
||
return it
|
||
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)
|
||
it["adv_unread"] = True
|
||
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)})
|
||
it["adv_unread"] = True
|
||
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)
|
||
# A seed that could not read some pins (no key: rotation; or a source down) leaves them PENDING: their first real
|
||
# read is recorded silently too, so the first nights after a seed are no mail storm. A pin added later is never
|
||
# pending, so its advisories mail normally.
|
||
pending = set(st.get("seed_pending", []))
|
||
if seed:
|
||
pending = {it["image"] for it in items if it.get("adv_unread")}
|
||
else:
|
||
for it in items:
|
||
if it["image"] in pending and not it.get("adv_unread"):
|
||
for a in it.get("advisories", []):
|
||
k = adv_key(it, a)
|
||
if k not in known:
|
||
known[k] = {"first": iso, "last_mailed": iso, "program": label(it), "summary": a["summary"][:160],
|
||
"silent": "first read after the seed"}
|
||
would.append({"kind": "advisory", "subject": "(silent, first read after the seed) %s %s" % (
|
||
it["image"], a["id"]), "text": ""})
|
||
pending.discard(it["image"])
|
||
st["seed_pending"] = sorted(pending)
|
||
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"),
|
||
"; ".join(it.get("notes", []))))
|
||
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:]))
|