server_automation

1#!/usr/bin/env python3
2"""Report newer upstream tags for the image pins in group_vars/all/images.yml.
3
4Read-only: nothing is ever changed. Used by check-image-updates.yml, which
5publishes the result to Home Assistant.
6
7For every pinned image the registry's tags are listed, and only tags of the
8SAME SHAPE as the pin are considered: same prefix (v, ubuntu-, ...) and same
9suffix pattern (-alpine, -cuda, -rpi, -lsNNN, ...). That keeps variants apart
10and drops pre-releases (an -rc1 tag has a different shape than the pin). The
11best candidate must also exist for every architecture listed in `arch`.
12
13Usage:
14  check_image_updates.py --images group_vars/all/images.yml \
15      [--running running.json] --out result.json
16running.json: {"<host>": ["repo:tag", ...]} - enables the drift check.
17"""
18import argparse
19import concurrent.futures as cf
20import datetime as dt
21import json
22import re
23import sys
24import time
25import urllib.error
26import urllib.request
27
28import yaml
29
30ACCEPT = ", ".join([
31    "application/vnd.oci.image.index.v1+json",
32    "application/vnd.docker.distribution.manifest.list.v2+json",
33    "application/vnd.docker.distribution.manifest.v2+json",
34    "application/vnd.oci.image.manifest.v1+json",
35])
36HEX = re.compile(r"(?<![0-9a-z])[0-9a-f]{7,40}(?![0-9a-z])")
37VERSION = re.compile(r"^(?P<prefix>[A-Za-z_-]*?)(?P<ver>\d+(?:\.\d+)*)(?P<suffix>.*)$")
38TIMEOUT = 30
39
40
41# ---------------------------------------------------------------- registry --
42def registry_of(image):
43    """docker.io/library/x -> (registry API host, repository path)."""
44    host, _, repo = image.partition("/")
45    if host == "docker.io":
46        return "registry-1.docker.io", repo
47    if host == "lscr.io":                      # lscr.io is a front for ghcr.io
48        return "ghcr.io", repo
49    return host, repo
50
51
52class Registry:
53    """Minimal anonymous Registry v2 client with token auth from the 401."""
54
55    def __init__(self, host, repo):
56        self.base = f"https://{host}/v2/{repo}"
57        self.headers = {}
58
59    def _open(self, url, method="GET", accept=None, attempt=0):
60        headers = dict(self.headers)
61        if accept:
62            headers["Accept"] = accept
63        req = urllib.request.Request(url, headers=headers, method=method)
64        try:
65            return urllib.request.urlopen(req, timeout=TIMEOUT)
66        except urllib.error.HTTPError as err:
67            # Rate limited (ghcr.io, Docker Hub): back off and retry
68            if err.code == 429 and attempt < 4:
69                wait = err.headers.get("Retry-After", "")
70                time.sleep(min(int(wait), 60) if wait.isdigit() else 5 * 2 ** attempt)
71                return self._open(url, method, accept, attempt + 1)
72            challenge = err.headers.get("WWW-Authenticate", "")
73            if err.code != 401 or "Authorization" in self.headers or not challenge.startswith("Bearer"):
74                raise
75            params = dict(re.findall(r'(\w+)="([^"]*)"', challenge))
76            query = "&".join(f"{k}={params[k]}" for k in ("service", "scope") if k in params)
77            token = json.load(urllib.request.urlopen(f"{params['realm']}?{query}", timeout=TIMEOUT))
78            self.headers["Authorization"] = "Bearer " + (token.get("token") or token.get("access_token"))
79            return self._open(url, method, accept, attempt)
80
81    def tags(self):
82        out, url = [], f"{self.base}/tags/list?n=10000"
83        while url:
84            resp = self._open(url)
85            out += json.load(resp).get("tags") or []
86            link = resp.headers.get("Link")
87            nxt = re.search(r"<([^>]+)>", link) if link else None
88            url = (self.base.split("/v2/")[0] + nxt.group(1)) if nxt else None
89        return out
90
91    def platforms(self, tag):
92        """Architectures a tag provides (one manifest GET)."""
93        doc = json.load(self._open(f"{self.base}/manifests/{tag}", accept=ACCEPT))
94        if "manifests" in doc:
95            return {m["platform"]["architecture"] for m in doc["manifests"]
96                    if m.get("platform", {}).get("os") not in (None, "unknown")}
97        return None                              # single-arch image: unknown here
98
99
100# ------------------------------------------------------------- versioning --
101def parse(tag):
102    """'v1.5.3-ls322' -> ('v', (1,5,3), '-ls#', (322,)).  None if unversioned."""
103    m = VERSION.match(HEX.sub("H", tag))
104    if not m:
105        return None
106    suffix = m.group("suffix")
107    nums = tuple(int(n) for n in re.findall(r"\d+", suffix))
108    return (m.group("prefix"), tuple(int(p) for p in m.group("ver").split(".")),
109            re.sub(r"\d+", "#", suffix), nums)
110
111
112def kind_of(old, new):
113    if new[0] != old[0]:
114        return "major"
115    if len(new) > 1 and len(old) > 1 and new[1] != old[1]:
116        return "minor"
117    return "patch"
118
119
120def check(key, entry):
121    tag = entry["tag"]
122    pinned_digest = "@sha256:" in tag
123    result = {"key": key, "service": entry.get("service", key), "tier": entry.get("tier", "main"),
124              "image": entry["image"], "current": entry.get("version") or tag.split("@")[0],
125              "latest": None, "latest_same_major": None, "kind": None, "note": ""}
126    # Digest pins compare their human version ("v3.40.0+12 (075cbd5)") against
127    # plain release tags: drop the build/commit decoration.
128    reference = re.split(r"[ +]", entry.get("version", ""))[0] if pinned_digest else tag
129    cur = parse(reference)
130    if cur is None:
131        result["note"] = "pin has no version to compare"
132        return result
133    reg = Registry(*registry_of(entry["image"]))
134    candidates = []
135    for t in reg.tags():
136        p = parse(t)
137        if p and p[0] == cur[0] and p[2] == cur[2] and (p[1], p[3]) > (cur[1], cur[3]):
138            candidates.append(((p[1], p[3]), t, p))
139    if not candidates:
140        if pinned_digest:
141            result["note"] = "pinned by digest; upstream has no newer version tag"
142        return result
143    candidates.sort(reverse=True)
144    needed = set(entry.get("arch") or [])
145
146    def first_available(cands):
147        for _, t, p in cands[:8]:                # newest first, bounded lookups
148            plats = reg.platforms(t)
149            if plats is None or needed <= plats:
150                return t, p
151        return None, None
152
153    best, best_p = first_available(candidates)
154    same_major = [c for c in candidates if c[2][1][0] == cur[1][0]]
155    best_sm, _ = first_available(same_major) if same_major else (None, None)
156    if best:
157        # A git-hash "version" (e.g. linuxserver's e7c7ce2d-ls262) carries no
158        # semantics - only that a newer build exists.
159        kind = "build" if "H" in cur[0] else kind_of(cur[1], best_p[1])
160        result.update(latest=best, kind=kind,
161                      latest_same_major=best_sm if best_sm and best_sm != best else None)
162    else:
163        result["note"] = f"newer tags exist but none for {', '.join(sorted(needed))}"
164    return result
165
166
167# ------------------------------------------------------------------ drift --
168def normalize(ref):
169    """Compare image references the way docker displays them."""
170    return ref.replace("docker.io/library/", "").replace("docker.io/", "")
171
172
173def drift(images, running):
174    """Running containers whose repo is pinned for that host but whose tag is not."""
175    pins = {}
176    for key, e in images.items():
177        repo = normalize(e["image"])
178        for host in ([e.get("host")] if e.get("host") not in (None, "all") else ["*"]):
179            pins.setdefault((host, repo), set()).add(e["tag"])
180    out = []
181    for host, refs in running.items():
182        for ref in refs:
183            ref = normalize(ref)
184            repo, _, tag = ref.partition("@")[0].rpartition(":")
185            if not repo or "/" not in ref and ":" not in ref:
186                continue
187            digest = ref.partition("@")[2]
188            wanted = pins.get((host, repo), set()) | pins.get(("*", repo), set())
189            if not wanted:
190                continue
191            matches = any(tag == w.split("@")[0] and (not digest or w.endswith(digest)) for w in wanted)
192            if not matches:
193                out.append({"host": host, "image": repo, "running": ref.split("/")[-1],
194                            "pinned": sorted(w.split("@")[0] for w in wanted)})
195    return out
196
197
198def main():
199    ap = argparse.ArgumentParser()
200    ap.add_argument("--images", required=True)
201    ap.add_argument("--running")
202    ap.add_argument("--out", required=True)
203    args = ap.parse_args()
204
205    images = yaml.safe_load(open(args.images))["images"]
206    results, errors = [], []
207    with cf.ThreadPoolExecutor(6) as pool:
208        futures = {pool.submit(check, k, e): k for k, e in images.items()}
209        for fut in cf.as_completed(futures):
210            key = futures[fut]
211            try:
212                results.append(fut.result())
213            except Exception as exc:              # one registry must not sink the report
214                errors.append({"key": key, "error": f"{type(exc).__name__}: {exc}"[:200]})
215    results.sort(key=lambda r: (r["service"], r["key"]))
216    running = json.load(open(args.running)) if args.running else {}
217    report = {
218        "checked": dt.datetime.now(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
219        "images": len(images),
220        "updates": [r for r in results if r["latest"] and r["tier"] == "main"],
221        "aux_updates": [r for r in results if r["latest"] and r["tier"] != "main"],
222        "unchecked": [{"key": r["key"], "note": r["note"]} for r in results if r["note"]],
223        "drift": drift(images, running),
224        "errors": errors,
225    }
226    json.dump(report, open(args.out, "w"), indent=1)
227    print(f"{len(report['updates'])} main updates, {len(report['aux_updates'])} aux updates, "
228          f"{len(report['drift'])} drifted, {len(errors)} errors")
229    return 0
230
231
232if __name__ == "__main__":
233    sys.exit(main())
234