413 lines
19 KiB
Python
413 lines
19 KiB
Python
#!/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
|
|
|
|
|
|
def _get(url, headers=None, stream=False):
|
|
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:
|
|
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))
|