#!/usr/bin/env python3 """image_runtime.py — which runtime parts are INSIDE an image, read from its registry without a pull (R-1327). WHY. HTTP2Shell (CVE-2026-111497, 2026-10-21) is a bug in the Brotli decoder that Node.js and Python's `brotli` / `brotlicffi` carry. An app's own security notices never mention it — the app did nothing wrong — so the nightly watch (security_watch.py, R-1303) read every app notice and still could not see that Home Assistant and RomM on the test boxes ran the bad code. This module reads what the image itself carries: 1. the image CONFIG (one small JSON blob): `NODE_VERSION`, `PYTHON_VERSION` in its Env; 2. only where that does not answer, the LAYERS, streamed top-down and never stored: the Alpine package database (`lib/apk/db/installed`), the Debian one (`var/lib/dpkg/status`), Node's `include/node/node_version.h`, and the `brotli-*` / `brotlicffi-*` `.dist-info` folders of Python. The topmost layer that holds a file wins (a later layer replaces an earlier one's copy). ZSTD LAYERS (R-1327, 2026-10-28). Home Assistant and other images ship zstd-compressed layers, and Python before 3.14 has no zstd. The decoder is the `zstandard` package (0.25.0), pinned by the sha256 PyPI publishes for each wheel: an importable `zstandard` first, else the pinned wheel from the cache (`~/.cache/felhom-wheels`; fetched only when FELHOM_WHEEL_FETCH=1 — the CI workflows set it), else the `zstd` program. Not the CI runner image: that image is built in the homelab repo and a change there is a DooPlex change. The wheel's hash is checked every time it is opened. What it cannot read is said, never guessed: a registry it has no token route for, a zstd layer where no decoder exists, a layer that does not decode, a layer over the byte budget — each makes the result `unchecked` with its reason, and the watch prints NOT CHECKED. USAGE (also a library: `read_image(ref)` -> dict) python3 scripts/image_runtime.py IMAGE[:TAG] ... [--budget-mb=N] [--json] """ import atexit import hashlib import json import os import platform import re import shutil import subprocess import sys import tempfile import zipfile import tarfile import urllib.error import urllib.request UA = "felhom-image-runtime/1.0 (+https://felhom.eu)" ACCEPT = ", ".join([ "application/vnd.oci.image.index.v1+json", "application/vnd.docker.distribution.manifest.list.v2+json", "application/vnd.oci.image.manifest.v1+json", "application/vnd.docker.distribution.manifest.v2+json"]) DEFAULT_BUDGET = int(__import__("os").environ.get("IMAGE_RUNTIME_BUDGET_MB", "1500")) * 1024 * 1024 # ^ bytes of compressed layers one image may stream before the rest is unchecked (IMAGE_RUNTIME_BUDGET_MB raises it for a by-hand read) NODE_H = re.compile(r"(^|/)usr/(local/)?include/node/node_version\.h$") PYDIST = re.compile(r"site-packages/(brotli|Brotli|brotlicffi)-([0-9][0-9A-Za-z.\-+]*)\.dist-info/(METADATA|RECORD)$") class Unreadable(Exception): pass ZSTANDARD_VERSION = "0.25.0" # (Python, libc) -> (wheel URL, sha256) — the digests PyPI publishes for the 0.25.0 files (released 2025-09-14), read # 2026-10-28. The CI runner is Alpine (musl) with Python 3.12; DooPlex is Debian (glibc) with Python 3.13. _PYPI = "https://files.pythonhosted.org/packages/" ZSTANDARD_WHEELS = { ("3.12", "musl"): (_PYPI + "fc/5f/75aafd4b9d11b5407b641b8e41a57864097663699f23e9ad4dbb91dc6bfe/" "zstandard-0.25.0-cp312-cp312-musllinux_1_2_x86_64.whl", "474d2596a2dbc241a556e965fb76002c1ce655445e4e3bf38e5477d413165ffa"), ("3.13", "musl"): (_PYPI + "b2/e5/fbd822d5c6f427cf158316d012c5a12f233473c2f9c5fe5ab1ae5d21f3d8/" "zstandard-0.25.0-cp313-cp313-musllinux_1_2_x86_64.whl", "4f187a0bb61b35119d1926aee039524d1f93aaf38a9916b8c4b78ac8514a0aaf"), ("3.12", "glibc"): (_PYPI + "53/60/7be26e610767316c028a2cbedb9a3beabdbe33e2182c373f71a1c0b88f36/" "zstandard-0.25.0-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", "5a56ba0db2d244117ed744dfa8f6f5b366e14148e00de44723413b2f3938a902"), ("3.13", "glibc"): (_PYPI + "63/4b/e3678b4e776db00f9f7b2fe58e547e8928ef32727d7a1ff01dea010f3f13/" "zstandard-0.25.0-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", "8e735494da3db08694d26480f1493ad2cf86e99bdd53e8e9771b2752a5c0246a"), } _ZSTANDARD = {} # the loaded module, or the reason there is none — decided once per process def _libc(): if platform.libc_ver()[0] == "glibc": return "glibc" import glob return "musl" if glob.glob("/lib/ld-musl-*") else "unknown" def _wheel_cache(): return os.path.join(os.environ.get("XDG_CACHE_HOME") or os.path.join(os.path.expanduser("~"), ".cache"), "felhom-wheels") def zstandard_module(fetch=None, wheels=None, cache=None, try_import=True): """(module, how) or (None, why). An importable `zstandard` first; else the pinned wheel for this Python and libc, from the cache or — only when `fetch` (default: FELHOM_WHEEL_FETCH=1) — from PyPI; its sha256 is checked before every use, and a wheel with any other hash is never opened. Pinned by test_image_runtime.py (wheel_*).""" if try_import: try: import zstandard return zstandard, "zstandard %s (installed)" % getattr(zstandard, "__version__", "?") except ImportError: pass fetch = os.environ.get("FELHOM_WHEEL_FETCH") == "1" if fetch is None else fetch key = ("%d.%d" % sys.version_info[:2], _libc()) pin = (wheels if wheels is not None else ZSTANDARD_WHEELS).get(key) if not pin: return None, "no zstd decoder: no pinned zstandard wheel for Python %s / %s" % key url, want = pin cache = cache or _wheel_cache() path = os.path.join(cache, url.rsplit("/", 1)[1]) data = open(path, "rb").read() if os.path.isfile(path) else None if data is None or hashlib.sha256(data).hexdigest() != want: if not fetch: return None, "no zstd decoder: the pinned zstandard wheel is not in %s (FELHOM_WHEEL_FETCH=1 fetches it)" % cache try: data = urllib.request.urlopen(urllib.request.Request(url, headers={"User-Agent": UA}), timeout=60).read() except Exception as e: # noqa: BLE001 — network, DNS, TLS: the zstd layers are NOT CHECKED, said out loud return None, "no zstd decoder: the pinned zstandard wheel could not be fetched (%s)" % str(e)[:100] if hashlib.sha256(data).hexdigest() != want: return None, "no zstd decoder: the fetched zstandard wheel's sha256 is not the pinned one — not used" os.makedirs(cache, exist_ok=True) with open(path + ".new", "wb") as f: f.write(data) os.replace(path + ".new", path) into = tempfile.mkdtemp(prefix="felhom-zstandard-") atexit.register(shutil.rmtree, into, True) import io with zipfile.ZipFile(io.BytesIO(data)) as z: # from the bytes just hashed, never from the file again z.extractall(into) sys.path.insert(0, into) try: import importlib sys.modules.pop("zstandard", None) mod = importlib.import_module("zstandard") except Exception as e: # noqa: BLE001 return None, "no zstd decoder: the pinned zstandard wheel did not load (%s)" % str(e)[:100] finally: sys.path.remove(into) return mod, "zstandard %s (pinned wheel, sha256 checked)" % getattr(mod, "__version__", "?") def zstd_decoder(): """How this process decodes a zstd layer: ('module', mod) | ('program', 'zstd') | (None, why). Decided once; zstd_decoder_how() says it in words for the log.""" if "d" not in _ZSTANDARD: mod, how = zstandard_module() if mod: _ZSTANDARD["d"], _ZSTANDARD["how"] = ("module", mod), how elif shutil.which("zstd"): _ZSTANDARD["d"], _ZSTANDARD["how"] = ("program", "zstd"), "the zstd program" else: _ZSTANDARD["d"], _ZSTANDARD["how"] = (None, how), "NONE — " + how return _ZSTANDARD["d"] def zstd_decoder_how(): zstd_decoder() return _ZSTANDARD["how"] def split_ref(ref): """'rommapp/romm:5.4.1' -> (registry, path, tag). lscr.io is Docker Hub's mirror (the watch's rule).""" tag = "latest" name = ref if "@" in name: name, tag = name.split("@", 1) elif re.search(r":[^/]+$", name): name, tag = name.rsplit(":", 1) first = name.split("/", 1)[0] if "." in first or ":" in first: reg, path = first, name.split("/", 1)[1] else: reg, path = "docker.io", name if reg in ("lscr.io", "index.docker.io", "registry-1.docker.io"): reg = "docker.io" if reg == "docker.io" and "/" not in path: path = "library/" + path return reg, path, tag RETRY_TRIES = 3 # a registry 429 is waited out (its Retry-After, capped, else 10/30 s) and tried again; only the RETRY_BACKOFF_S = (10, 30) # last try's 429 is an answer (ghcr.io refused blob reads in back-to-back dry runs, 2026-10-28) def _get(url, headers=None, stream=False): import time for i in range(RETRY_TRIES): req = urllib.request.Request(url, headers=dict({"User-Agent": UA}, **(headers or {}))) try: r = urllib.request.urlopen(req, timeout=60) except urllib.error.HTTPError as e: if e.code == 429 and i < RETRY_TRIES - 1: ra = str((e.headers or {}).get("Retry-After", "")) time.sleep(min(int(ra), 120) if ra.isdigit() else RETRY_BACKOFF_S[min(i, len(RETRY_BACKOFF_S) - 1)]) continue return e.code, e.headers, b"" if stream: return r.status, r.headers, r return r.status, r.headers, r.read() class Registry: BASES = {"docker.io": "https://registry-1.docker.io", "ghcr.io": "https://ghcr.io", "quay.io": "https://quay.io"} def __init__(self, reg, path): if reg not in self.BASES: raise Unreadable("registry %s: no anonymous token route" % reg) self.base, self.path = self.BASES[reg], path hdr = {} if reg == "docker.io": url = "https://auth.docker.io/token?service=registry.docker.io&scope=repository:%s:pull" % path # Optional Docker Hub login (a by-hand read: a login has its own pull limit; CI stays anonymous and fills the cache itself, R-1327). # The NAMES of the variables are read here; the values are never printed. import base64 import os u, p = os.environ.get("DOCKERHUB_USER", ""), os.environ.get("DOCKERHUB_TOKEN", "") if u and p: hdr = {"Authorization": "Basic " + base64.b64encode(("%s:%s" % (u, p)).encode()).decode()} elif reg == "quay.io": url = "https://quay.io/v2/auth?service=quay.io&scope=repository:%s:pull" % path else: url = "https://ghcr.io/token?scope=repository:%s:pull" % path st, _h, body = _get(url, hdr) if st == 401 and hdr: st, _h, body = _get(url) # a login that is not Docker Hub's: anonymous, as before if st != 200: raise Unreadable("%s token for %s: HTTP %s" % (reg, path, st)) self.auth = {"Authorization": "Bearer %s" % json.loads(body).get("token", "")} def manifest(self, ref): st, h, body = _get("%s/v2/%s/manifests/%s" % (self.base, self.path, ref), dict(self.auth, Accept=ACCEPT)) if st != 200: raise Unreadable("manifest %s:%s: HTTP %s" % (self.path, ref, st)) return json.loads(body), h.get("Docker-Content-Digest", "") def blob(self, digest, stream=False): st, _h, body = _get("%s/v2/%s/blobs/%s" % (self.base, self.path, digest), self.auth, stream=stream) if st != 200: raise Unreadable("blob %s: HTTP %s" % (digest[:19], st)) return body def _platform_manifest(reg, ref): m, digest = reg.manifest(ref) if "manifests" in m: # an index: take linux/amd64 (every Felhom box is amd64) pick = [x for x in m["manifests"] if (x.get("platform") or {}).get("os") == "linux" and (x.get("platform") or {}).get("architecture") == "amd64"] if not pick: raise Unreadable("no linux/amd64 image in the index") digest = pick[0]["digest"] m, _ = reg.manifest(digest) return m, digest def parse_apk(text): out, name = {}, None for line in text.splitlines(): if line.startswith("P:"): name = line[2:] elif line.startswith("V:") and name: out[name] = line[2:] return out def parse_dpkg(text): out = {} for block in text.split("\n\n"): p = re.search(r"^Package: (\S+)", block, re.M) v = re.search(r"^Version: (\S+)", block, re.M) if p and v: out[p.group(1)] = v.group(1) return out def parse_node_h(text): parts = [re.search(r"#define NODE_%s_VERSION (\d+)" % k, text) for k in ("MAJOR", "MINOR", "PATCH")] return ".".join(p.group(1) for p in parts) if all(parts) else None def scan_layers(reg, layers, budget=DEFAULT_BUDGET, decoder=None): """Stream layers top-down. Returns (found, notes): found holds apk/dpkg/node_h/pydist; the first (topmost) wins. `decoder` is zstd_decoder()'s answer (a seam for the tests); a zstd layer with none is a NOTE, never skipped quietly.""" found, notes, spent = {}, [], 0 for layer in reversed(layers): mt, size = layer.get("mediaType", ""), int(layer.get("size") or 0) zstd = "zstd" in mt if zstd: kind, dec = decoder or zstd_decoder() if not kind: notes.append("zstd layer %s: %s" % (layer["digest"][:19], dec)) continue if spent + size > budget: notes.append("byte budget reached before layer %s (%d MB)" % (layer["digest"][:19], size >> 20)) break spent += size resp = reg.blob(layer["digest"], stream=True) proc = None try: if zstd and kind == "module": # the pinned zstandard: a streaming reader, nothing on disk fileobj, mode = dec.ZstdDecompressor().stream_reader(resp, read_across_frames=True), "r|" elif zstd: # the zstd program decompresses the stream (a pipe, nothing on disk) proc = subprocess.Popen(["zstd", "-dc"], stdin=subprocess.PIPE, stdout=subprocess.PIPE) import threading def feed(src=resp, dst=proc.stdin): try: for chunk in iter(lambda: src.read(1 << 20), b""): dst.write(chunk) except (OSError, ValueError): pass finally: try: dst.close() except OSError: pass threading.Thread(target=feed, daemon=True).start() fileobj, mode = proc.stdout, "r|" else: fileobj = resp mode = "r|" if "tar" in mt and "gzip" not in mt and "gz" not in mt else "r|gz" with tarfile.open(fileobj=fileobj, mode=mode) as tf: for m in tf: n = m.name.lstrip("./") if n == "lib/apk/db/installed" and "apk" not in found and m.isfile(): found["apk"] = parse_apk(tf.extractfile(m).read().decode("utf-8", "replace")) elif n == "var/lib/dpkg/status" and "dpkg" not in found and m.isfile(): found["dpkg"] = parse_dpkg(tf.extractfile(m).read().decode("utf-8", "replace")) elif NODE_H.search(n) and "node_h" not in found and m.isfile(): found["node_h"] = parse_node_h(tf.extractfile(m).read().decode("utf-8", "replace")) else: pm = PYDIST.search(n) if pm: key = pm.group(1).lower() found.setdefault("pydist", {}).setdefault(key, pm.group(2)) except Exception as e: # noqa: BLE001 — tar, zlib, zstd (ZstdError), I/O: the layer is NOT CHECKED, never a pass notes.append("layer %s unreadable: %s" % (layer["digest"][:19], type(e).__name__)) finally: if proc: proc.kill() proc.wait() try: resp.close() except Exception: pass return found, notes def _head(url, headers): req = urllib.request.Request(url, method="HEAD", headers=dict({"User-Agent": UA}, **headers)) try: r = urllib.request.urlopen(req, timeout=60) return r.status, r.headers except urllib.error.HTTPError as e: return e.code, e.headers def resolve_digest(ref): """The digest the TAG points at now (the index's, for a multi-arch image) — a HEAD request, which Docker Hub does not count against its anonymous pull limit (a GET of every manifest every night hit HTTP 429 on the first run, 2026-10-27). The cache is keyed by it: a changed tag is a new key. Raises Unreadable.""" r, path, tag = split_ref(ref) reg = Registry(r, path) st, h = _head("%s/v2/%s/manifests/%s" % (reg.base, path, tag), dict(reg.auth, Accept=ACCEPT)) if st != 200 or not h.get("Docker-Content-Digest"): raise Unreadable("manifest %s:%s: HTTP %s" % (path, tag, st)) return h["Docker-Content-Digest"] def read_image(ref, budget=DEFAULT_BUDGET, layers=True): """{'ref', 'digest', 'env': {...}, 'node', 'python', 'brotli', 'brotlicffi', 'apk', 'dpkg', 'unchecked': [...]}. `node` comes from NODE_VERSION, else node_version.h, else the apk/dpkg `nodejs` package. `unchecked` lists every reason a part could not be read — an empty list means everything this module looks for was looked for.""" out = {"ref": ref, "unchecked": []} try: r, path, tag = split_ref(ref) reg = Registry(r, path) m, digest = _platform_manifest(reg, tag) out["digest"] = digest cfg = json.loads(reg.blob(m["config"]["digest"])) env = dict(e.split("=", 1) for e in ((cfg.get("config") or {}).get("Env") or []) if "=" in e) out["env"] = {k: env[k] for k in ("NODE_VERSION", "PYTHON_VERSION") if k in env} out["node"] = env.get("NODE_VERSION", "").lstrip("v") or None out["python"] = env.get("PYTHON_VERSION") or None if layers: found, notes = scan_layers(reg, m.get("layers") or [], budget) out["unchecked"] += notes apk, dpkg = found.get("apk") or {}, found.get("dpkg") or {} out["apk"] = {k: v for k, v in apk.items() if re.match(r"^(nodejs|nodejs-current|brotli|brotli-libs|python3)$", k)} out["dpkg"] = {k: v for k, v in dpkg.items() if re.match(r"^(nodejs|libbrotli1|python3(\.\d+)?)$", k)} if not out["node"]: out["node"] = found.get("node_h") or (apk.get("nodejs") or apk.get("nodejs-current") or "").split("-r")[0] \ or (dpkg.get("nodejs") or "").split("-")[0].split("+")[0].split("~")[0] or None py = found.get("pydist") or {} out["brotli"], out["brotlicffi"] = py.get("brotli"), py.get("brotlicffi") except Unreadable as e: out["unchecked"].append(str(e)) except (urllib.error.URLError, OSError, ValueError, KeyError) as e: out["unchecked"].append("%s: %s" % (type(e).__name__, str(e)[:120])) return out def main(argv): budget = DEFAULT_BUDGET refs, as_json = [], False for a in argv[1:]: if a.startswith("--budget-mb="): budget = int(a.split("=", 1)[1]) * 1024 * 1024 elif a == "--json": as_json = True else: refs.append(a) res = [read_image(r, budget) for r in refs] if as_json: print(json.dumps(res, indent=1)) else: for x in res: print("%-55s node=%-9s brotli=%-7s brotlicffi=%-8s apk=%s dpkg=%s%s" % ( x["ref"], x.get("node"), x.get("brotli"), x.get("brotlicffi"), x.get("apk", {}), x.get("dpkg", {}), (" UNCHECKED: " + "; ".join(x["unchecked"])) if x["unchecked"] else "")) return 0 if __name__ == "__main__": sys.exit(main(sys.argv))