#!/usr/bin/env python3 """Measure upstream drift for app-catalog-felhom.eu image pins. Read-only, DooPlex-only. Reuses the auth-handling approach of scripts/check-image-resolvable.py (per-registry anonymous bearer-token flow against the Docker Registry v2 API), but instead of `docker manifest inspect` per-ref, lists the full upstream tag catalogue so we can compare the pinned tag against the newest one available, and classify the step as same-major (minor/patch) or major. """ import json import re import sys import time from pathlib import Path import requests SCRATCH = Path("/tmp/claude-1000/-mnt-5-hdd-felhom-eu-git/5f83c0fb-b1bf-404e-b99e-2e3b28a31615/scratchpad") SESSION = requests.Session() SESSION.headers.update({"User-Agent": "felhom-catalog-drift-audit/1.0 (read-only measurement)"}) def get_bearer_token(www_auth: str) -> str: assert www_auth.startswith("Bearer ") parts = www_auth[len("Bearer "):].split(",") d = {} for p in parts: k, v = p.split("=", 1) d[k] = v.strip('"') r = SESSION.get(d["realm"], params={"service": d.get("service"), "scope": d.get("scope")}, timeout=20) r.raise_for_status() return r.json()["token"] def tags_list_all(host: str, repo: str, max_pages=1000): """Full tag list via Docker Registry v2 API, following RFC5988 Link pagination.""" url = f"https://{host}/v2/{repo}/tags/list" params = {"n": 1000} r = SESSION.get(url, params=params, timeout=30) headers = {} if r.status_code == 401: tok = get_bearer_token(r.headers["WWW-Authenticate"]) headers = {"Authorization": f"Bearer {tok}"} r = SESSION.get(url, params=params, headers=headers, timeout=30) r.raise_for_status() j = r.json() tags = list(j.get("tags") or []) link = r.headers.get("Link") pages = 1 while link and pages < max_pages: m = re.search(r'<([^>]+)>;\s*rel="next"', link) if not m: break nexturl = m.group(1) if nexturl.startswith("/"): nexturl = f"https://{host}{nexturl}" r = SESSION.get(nexturl, headers=headers, timeout=30) r.raise_for_status() j = r.json() tags.extend(j.get("tags") or []) link = r.headers.get("Link") pages += 1 return tags def manifest_digest(host: str, repo: str, ref: str): """Resolve one ref's manifest digest (for floating-tag drift: 'has the tag moved'). Read-only.""" url = f"https://{host}/v2/{repo}/manifests/{ref}" accept = ("application/vnd.docker.distribution.manifest.v2+json," "application/vnd.docker.distribution.manifest.list.v2+json," "application/vnd.oci.image.manifest.v1+json," "application/vnd.oci.image.index.v1+json") r = SESSION.head(url, headers={"Accept": accept}, timeout=30) if r.status_code == 401: tok = get_bearer_token(r.headers["WWW-Authenticate"]) r = SESSION.head(url, headers={"Accept": accept, "Authorization": f"Bearer {tok}"}, timeout=30) if r.status_code != 200: return None, f"HTTP {r.status_code}" return r.headers.get("Docker-Content-Digest"), None # host resolution per repo prefix, mirroring how these refs are actually pulled def resolve_host_repo(ref_repo: str): if ref_repo.startswith("ghcr.io/"): return "ghcr.io", ref_repo[len("ghcr.io/"):] if ref_repo.startswith("lscr.io/"): return "lscr.io", ref_repo[len("lscr.io/"):] if ref_repo.startswith("registry.gitlab.com/"): return "registry.gitlab.com", ref_repo[len("registry.gitlab.com/"):] if ref_repo.startswith("gitea.dooplex.hu/"): return "gitea.dooplex.hu", ref_repo[len("gitea.dooplex.hu/"):] if ref_repo.startswith("quay.io/"): return "quay.io", ref_repo[len("quay.io/"):] # Docker Hub if "/" not in ref_repo: return "registry-1.docker.io", f"library/{ref_repo}" return "registry-1.docker.io", ref_repo VERSION_CORE_RE = re.compile(r'(\d+(?:\.\d+){1,3})') UNSTABLE_MARKERS = ("rc", "beta", "alpha", "dev", "nightly", "canary", "edge", "preview", "snapshot", "-pr", "test") def split_tag(tag: str): """(prefix, core_version_tuple, suffix) — first maximal dotted-numeric run of len>=2 is the core.""" m = VERSION_CORE_RE.search(tag) if not m: return None prefix = tag[:m.start()] core = tuple(int(x) for x in m.group(1).split(".")) suffix = tag[m.end():] return prefix, core, suffix def is_unstable(tag: str) -> bool: low = tag.lower() return any(mk in low for mk in UNSTABLE_MARKERS) def newest_matching(all_tags, current_tag): """Find the newest tag sharing the current tag's prefix/suffix 'shape', by core version tuple.""" cur = split_tag(current_tag) if not cur: return None, "current tag has no parseable version core" cur_prefix, cur_core, cur_suffix = cur # suffix "shape": if suffix contains a long hex-looking token, treat it as a wildcard hash def suffix_shape(suf): if re.search(r'[0-9a-f]{7,40}', suf.lower()) and re.search(r'[a-f]', suf.lower()): return re.sub(r'[0-9a-f]{7,40}', '', suf.lower()) return suf cur_shape = suffix_shape(cur_suffix) best = None best_core = None for t in all_tags: if t == current_tag: continue if is_unstable(t): continue parsed = split_tag(t) if not parsed: continue p, core, suf = parsed if p != cur_prefix: continue if suffix_shape(suf) != cur_shape: continue if len(core) != len(cur_core): continue # Some registries (linuxserver.io/lscr.io in particular) publish EXTRA tags for a # same-app-version base-image rebuild, shaped like the release tag plus a YYYYMMDD-ish # trailing component (e.g. bookstack 26.05.2 coexists with 26.05.20260608). That is not a # newer application release — it is a rebuild of an existing one — so any component that # looks like a calendar date (>= 20000, comfortably above any real release counter we saw # across all 58 pinned-semver images, all of which stayed under 10000) is excluded as a # candidate "newest" rather than misread as a huge version jump. if any(c >= 20000 for c in core): continue if best_core is None or core > best_core: best_core = core best = t if best is None: return None, "no matching newer tag found (same prefix/suffix shape)" if best_core <= cur_core: return None, "up to date (no tag newer than current within the same shape)" return (best, best_core, cur_core), None def main(): pins = json.load(open(SCRATCH / "pins.json")) results = {} for i, (ref, p) in enumerate(sorted(pins.items())): repo = p["repo"] tag = p["tag"] host, hrepo = resolve_host_repo(repo) entry = {"ref": ref, "repo": repo, "tag": tag, "kind": p["kind"], "host": host, "hrepo": hrepo} if repo.startswith("gitea.dooplex.hu"): entry["status"] = "internal-not-upstream" results[ref] = entry print(f"[{i+1}/{len(pins)}] SKIP internal: {ref}") continue try: if p["kind"] == "floating": dig, err = manifest_digest(host, hrepo, tag) if err: entry["status"] = "error" entry["error"] = err else: entry["status"] = "ok" entry["current_digest"] = dig print(f"[{i+1}/{len(pins)}] FLOAT {ref} -> {entry.get('current_digest') or entry.get('error')}") else: tags = tags_list_all(host, hrepo) entry["n_upstream_tags"] = len(tags) res, err = newest_matching(tags, tag) if err: entry["status"] = "no-newer-or-unparsed" entry["detail"] = err else: best, best_core, cur_core = res entry["status"] = "behind" entry["newest_tag"] = best entry["newest_core"] = list(best_core) entry["current_core"] = list(cur_core) entry["major_step"] = best_core[0] != cur_core[0] print(f"[{i+1}/{len(pins)}] SEMVER {ref} ({len(tags)} tags) -> {entry.get('newest_tag') or entry.get('detail')}") except requests.HTTPError as e: entry["status"] = "error" entry["error"] = f"HTTP error: {e}" print(f"[{i+1}/{len(pins)}] ERROR {ref}: {e}") except Exception as e: entry["status"] = "error" entry["error"] = f"{type(e).__name__}: {e}" print(f"[{i+1}/{len(pins)}] ERROR {ref}: {e}") results[ref] = entry time.sleep(0.15) json.dump(results, open(SCRATCH / "results.json", "w"), indent=2) print("\nwrote", SCRATCH / "results.json") if __name__ == "__main__": main()