180 lines
11 KiB
Python
180 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""test_image_runtime.py — the image reader's parsers (R-1327) and the DooPlex list maker (R-1328). No network.
|
|
|
|
Pinned by consequence: an Alpine and a Debian package database read back the packages the runtime table matches; Node's
|
|
version header gives the version; an image reference splits the way the watch's registry_of does (lscr.io is Docker
|
|
Hub; a bare name is library/); a tar layer streamed top-down returns the TOPMOST copy of a file and the Python
|
|
`.dist-info` versions; the DooPlex list keeps one line per image:tag with every file that names it, and skips templated
|
|
and variable images. Red-proof: a scan that keeps the FIRST (bottom) copy instead of the topmost fails `topmost`.
|
|
|
|
ZSTD (R-1327, 2026-10-28): a zstd layer above a gzip one is read with NO `zstd` program on the PATH (the CI runner has
|
|
none) — through the pinned `zstandard` wheel; a zstd layer that does not decode, and one with no decoder at all, are
|
|
NOTES (the watch's NOT CHECKED), never a pass; the wheel loader opens a wheel only when its sha256 is the pinned one.
|
|
Where no decoder can be had (no wheel in the cache, FELHOM_WHEEL_FETCH unset) the read check says NOT CHECKED out loud
|
|
and is not counted; with FELHOM_WHEEL_FETCH=1 (CI sets it) a missing decoder FAILS. Red-proof: the 2026-10-27 module
|
|
(the `zstd` program or nothing) fails `zstd: a zstd layer is read with no zstd program`.
|
|
"""
|
|
import base64
|
|
import hashlib
|
|
import io
|
|
import os
|
|
import sys
|
|
import tarfile
|
|
import tempfile
|
|
import zipfile
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
import image_runtime as ir # noqa: E402
|
|
import dooplex_inventory as di # noqa: E402
|
|
|
|
fails, ran = [], 0
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
global ran
|
|
ran += 1
|
|
print(" %s %s%s" % ("ok" if cond else "XX", name, "" if cond else " %s" % (detail,)))
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
def tar_bytes(files):
|
|
buf = io.BytesIO()
|
|
with tarfile.open(fileobj=buf, mode="w:gz") as tf:
|
|
for name, data in files.items():
|
|
ti = tarfile.TarInfo(name)
|
|
ti.size = len(data)
|
|
tf.addfile(ti, io.BytesIO(data))
|
|
return buf.getvalue()
|
|
|
|
|
|
class FakeReg:
|
|
def __init__(self, blobs):
|
|
self.blobs = blobs
|
|
|
|
def blob(self, digest, stream=False):
|
|
return io.BytesIO(self.blobs[digest])
|
|
|
|
|
|
# A tar holding an Alpine package database (nodejs 24.22.1, brotli-libs 1.2.1), compressed by `zstd -19` (2026-10-28).
|
|
ZSTD_LAYER = base64.b64decode(
|
|
"KLUv/QRoLQMAsgUSEJB9hC/xJSHf5yhLWm0eYqY+iotURKotdmoB8FiUYcw4RjGQJXy2quX/jpvO2z94QaiU9/dP/+/npPcHKW0610jcI/M24iI"
|
|
"NCiAwMT3M5Vx0jvfWN08dUPKASwQMGmAZDXf7YaaA")
|
|
ZSTD_MT = "application/vnd.oci.image.layer.v1.tar+zstd"
|
|
GZ_MT = "application/vnd.oci.image.layer.v1.tar+gzip"
|
|
|
|
|
|
def zstd_checks():
|
|
gz = tar_bytes({"usr/local/lib/python3.12/site-packages/brotli-1.2.1.dist-info/RECORD": b"x",
|
|
"lib/apk/db/installed": b"P:nodejs\nV:20.0.0-r0\n"})
|
|
layers = [{"digest": "sha256:gz", "mediaType": GZ_MT, "size": len(gz)},
|
|
{"digest": "sha256:zst", "mediaType": ZSTD_MT, "size": len(ZSTD_LAYER)}]
|
|
reg = FakeReg({"sha256:gz": gz, "sha256:zst": ZSTD_LAYER})
|
|
path = os.environ.get("PATH", "")
|
|
os.environ["PATH"] = tempfile.mkdtemp() # no `zstd` program: the CI runner's case
|
|
ir._ZSTANDARD.clear() # the decoder is decided once per process: decide it HERE
|
|
try:
|
|
found, notes = ir.scan_layers(reg, layers)
|
|
finally:
|
|
os.environ["PATH"] = path
|
|
ir._ZSTANDARD.clear()
|
|
no_decoder = [n for n in notes if "no zstd decoder" in n]
|
|
if no_decoder and os.environ.get("FELHOM_WHEEL_FETCH") != "1":
|
|
print(" NOT CHECKED zstd: a zstd layer is read with no zstd program — %s" % no_decoder[0])
|
|
else:
|
|
check("zstd: a zstd layer is read with no zstd program (the top layer's apk db wins)",
|
|
(found.get("apk") or {}).get("nodejs") == "24.22.1-r0" and not notes, (found.get("apk"), notes))
|
|
check("zstd: the gzip layer under it is still read", (found.get("pydist") or {}).get("brotli") == "1.2.1",
|
|
found.get("pydist"))
|
|
found, notes = ir.scan_layers(reg, [{"digest": "sha256:zst", "mediaType": ZSTD_MT, "size": 9}],
|
|
decoder=(None, "no zstd decoder: test"))
|
|
check("zstd: no decoder is a NOTE (NOT CHECKED), never a silent pass", notes and "no zstd decoder" in notes[0]
|
|
and not found, (found, notes))
|
|
mod, _how = ir.zstandard_module()
|
|
if mod:
|
|
bad = ZSTD_LAYER[:40] + b"\x00" * 30 + ZSTD_LAYER[70:]
|
|
found, notes = ir.scan_layers(FakeReg({"sha256:bad": bad}), [{"digest": "sha256:bad", "mediaType": ZSTD_MT,
|
|
"size": len(bad)}], decoder=("module", mod))
|
|
check("zstd: a layer that does not decode is a NOTE (NOT CHECKED), never a pass",
|
|
any("unreadable" in n for n in notes), (found, notes))
|
|
found, notes = ir.scan_layers(FakeReg({"sha256:t": ZSTD_LAYER[:60]}), [{"digest": "sha256:t", "mediaType": ZSTD_MT,
|
|
"size": 60}], decoder=("module", mod))
|
|
check("zstd: a cut-off layer is a NOTE (NOT CHECKED), never a pass", any("unreadable" in n for n in notes),
|
|
(found, notes))
|
|
|
|
|
|
def wheel_checks():
|
|
"""The loader opens a wheel only when its sha256 is the pinned one. A fake `zstandard` wheel, served by file://."""
|
|
d = tempfile.mkdtemp()
|
|
whl = os.path.join(d, "zstandard-0.0-fake.whl")
|
|
with zipfile.ZipFile(whl, "w") as z:
|
|
z.writestr("zstandard/__init__.py", "__version__ = 'fake-for-test'\n")
|
|
good = hashlib.sha256(open(whl, "rb").read()).hexdigest()
|
|
key = ("%d.%d" % sys.version_info[:2], ir._libc())
|
|
url = "file://" + whl
|
|
mod, how = ir.zstandard_module(fetch=True, wheels={key: (url, good)}, cache=os.path.join(d, "c1"), try_import=False)
|
|
check("wheel: the pinned wheel is fetched, hash-checked and loaded", mod is not None and
|
|
getattr(mod, "__version__", "") == "fake-for-test" and "sha256 checked" in how, how)
|
|
mod, how = ir.zstandard_module(fetch=True, wheels={key: (url, "0" * 64)}, cache=os.path.join(d, "c2"), try_import=False)
|
|
check("wheel: a wheel whose sha256 is not the pinned one is never loaded", mod is None and "not the pinned" in how, how)
|
|
c3 = os.path.join(d, "c3")
|
|
os.makedirs(c3)
|
|
open(os.path.join(c3, os.path.basename(whl)), "wb").write(b"changed")
|
|
mod, how = ir.zstandard_module(fetch=False, wheels={key: (url, good)}, cache=c3, try_import=False)
|
|
check("wheel: a changed wheel in the cache is not loaded, and without FETCH nothing is fetched", mod is None and
|
|
"not in" in how, how)
|
|
mod, how = ir.zstandard_module(fetch=True, wheels={}, cache=c3, try_import=False)
|
|
check("wheel: no pin for this Python is a reason, never a guess", mod is None and "no pinned" in how, how)
|
|
sys.modules.pop("zstandard", None)
|
|
|
|
|
|
def main():
|
|
apk = ir.parse_apk("P:musl\nV:1.2.5-r10\n\nP:nodejs\nV:24.18.1-r0\n\nP:brotli-libs\nV:1.2.0-r1\n")
|
|
check("apk database -> nodejs and brotli-libs", apk.get("nodejs") == "24.18.1-r0" and apk.get("brotli-libs") == "1.2.0-r1", apk)
|
|
dpkg = ir.parse_dpkg("Package: libbrotli1\nStatus: install ok installed\nVersion: 1.0.9-2+b6\n\nPackage: nodejs\nVersion: 20.20.0-1nodesource1\n")
|
|
check("dpkg database -> libbrotli1 and nodejs", dpkg == {"libbrotli1": "1.0.9-2+b6", "nodejs": "20.20.0-1nodesource1"}, dpkg)
|
|
h = "#define NODE_MAJOR_VERSION 24\n#define NODE_MINOR_VERSION 22\n#define NODE_PATCH_VERSION 1\n"
|
|
check("node_version.h -> 24.22.1", ir.parse_node_h(h) == "24.22.1", ir.parse_node_h(h))
|
|
check("split_ref: lscr.io is Docker Hub, tag kept", ir.split_ref("lscr.io/linuxserver/qbittorrent:5.3.0") ==
|
|
("docker.io", "linuxserver/qbittorrent", "5.3.0"))
|
|
check("split_ref: a bare name is library/, no tag = latest", ir.split_ref("redis") == ("docker.io", "library/redis", "latest"))
|
|
check("split_ref: index.docker.io is Docker Hub", ir.split_ref("index.docker.io/plexinc/pms-docker:1.43") ==
|
|
("docker.io", "plexinc/pms-docker", "1.43"))
|
|
check("split_ref: ghcr.io with a port-free path", ir.split_ref("ghcr.io/home-assistant/home-assistant:2026.9.4") ==
|
|
("ghcr.io", "home-assistant/home-assistant", "2026.9.4"))
|
|
# two layers: the bottom one has an OLD apk db and brotli 1.1.0; the top one replaces the db and adds brotli 1.2.1
|
|
bottom = tar_bytes({"lib/apk/db/installed": b"P:nodejs\nV:24.18.1-r0\n",
|
|
"usr/lib/python3/site-packages/Brotli-1.1.0.dist-info/METADATA": b"x"})
|
|
top = tar_bytes({"lib/apk/db/installed": b"P:nodejs\nV:24.22.1-r0\n",
|
|
"usr/local/lib/python3.12/site-packages/brotli-1.2.1.dist-info/RECORD": b"x"})
|
|
layers = [{"digest": "sha256:bottom", "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "size": len(bottom)},
|
|
{"digest": "sha256:top", "mediaType": "application/vnd.oci.image.layer.v1.tar+gzip", "size": len(top)}]
|
|
found, notes = ir.scan_layers(FakeReg({"sha256:bottom": bottom, "sha256:top": top}), layers)
|
|
check("topmost: the top layer's apk db wins", (found.get("apk") or {}).get("nodejs") == "24.22.1-r0", found.get("apk"))
|
|
check("topmost: the top layer's Brotli dist-info wins", (found.get("pydist") or {}).get("brotli") == "1.2.1", found.get("pydist"))
|
|
check("no note for readable gzip layers", notes == [], notes)
|
|
found, notes = ir.scan_layers(FakeReg({}), [{"digest": "sha256:z", "mediaType": "application/vnd.oci.image.layer.v1.tar+zstd",
|
|
"size": 10}], budget=0)
|
|
check("a layer over the byte budget is a NOTE, never a silent pass", notes and ("budget" in notes[0] or "zstd" in notes[0]), notes)
|
|
zstd_checks()
|
|
wheel_checks()
|
|
d = tempfile.mkdtemp()
|
|
open(os.path.join(d, "a.yaml"), "w").write("spec:\n containers:\n - image: rommapp/romm:5.4.0\n name: x\n"
|
|
" - image: '{{ .Values.image }}'\n - image: $IMG\n")
|
|
open(os.path.join(d, "b.yml"), "w").write("image: rommapp/romm:5.4.0\nimage: busybox\n")
|
|
got = di.collect(d)
|
|
check("dooplex: one line per image:tag with every file", got.get("rommapp/romm:5.4.0") == {"a.yaml", "b.yml"}, got)
|
|
check("dooplex: a bare image becomes :latest; templated and variable images are skipped",
|
|
set(got) == {"rommapp/romm:5.4.0", "busybox:latest"}, set(got))
|
|
print()
|
|
if fails:
|
|
print("image_runtime tests FAILED: %s" % ", ".join(fails))
|
|
return 1
|
|
print("image_runtime tests OK — %d checks" % ran)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|