diff --git a/scripts/forgejo_package_prune.py b/scripts/forgejo_package_prune.py index fc968b3..146d53b 100644 --- a/scripts/forgejo_package_prune.py +++ b/scripts/forgejo_package_prune.py @@ -7,6 +7,8 @@ import argparse import json import os import re +import shutil +import subprocess import sys import time import urllib.error @@ -107,6 +109,47 @@ def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]: return protected +def collect_live_cluster_versions( + *, kubectl: str = "kubectl", timeout: float = 60.0 +) -> tuple[set[tuple[str, str, str]], list[str]]: + """Protect image tags currently running in the cluster (best-effort). + + Enumerates all pod container images across namespaces via kubectl and + protects any `forgejo.coulomb.social/coulomb/:`. This closes the + gap where a live deployment pins a tag not declared in Helm values (e.g. + CI-deployed apps). Failures (no kubectl, no cluster access) return an empty + set with a note — pruning a reachable registry must not hard-depend on + cluster access, but the note surfaces reduced protection coverage. + """ + protected: set[tuple[str, str, str]] = set() + if shutil.which(kubectl) is None: + return protected, ["live-tag protection skipped: kubectl not found"] + jsonpath = ( + "{range .items[*]}" + "{range .spec.containers[*]}{.image}{'\\n'}{end}" + "{range .spec.initContainers[*]}{.image}{'\\n'}{end}" + "{end}" + ) + try: + result = subprocess.run( + [kubectl, "get", "pods", "--all-namespaces", "-o", f"jsonpath={jsonpath}"], + capture_output=True, + text=True, + timeout=timeout, + check=True, + ) + except Exception as exc: # noqa: BLE001 + return protected, [f"live-tag protection skipped: kubectl query failed ({exc})"] + for line in result.stdout.splitlines(): + image = line.strip() + if not image: + continue + match = FORGEJO_IMAGE_RE.match(image) + if match and match.group("tag"): + protected.add(("container", match.group("name"), match.group("tag"))) + return protected, [] + + def _api_request( method: str, url: str, @@ -338,6 +381,13 @@ def main(argv: list[str] | None = None) -> int: parser.add_argument("--apps-root", type=Path, default=DEFAULT_APPS_ROOT) parser.add_argument("--apply", action="store_true") parser.add_argument("--json", action="store_true", help="Emit JSON summary on stdout") + parser.add_argument( + "--no-protect-live", + dest="protect_live", + action="store_false", + help="Skip protecting image tags currently running in the cluster (kubectl)", + ) + parser.set_defaults(protect_live=True) args = parser.parse_args(argv) apply = bool(args.apply) @@ -345,6 +395,13 @@ def main(argv: list[str] | None = None) -> int: package_types = [part.strip() for part in args.types.split(",") if part.strip()] token = load_token() protected = collect_protected_versions(args.apps_root.expanduser()) + protect_notes: list[str] = [] + if args.protect_live: + live, protect_notes = collect_live_cluster_versions() + protected |= live + print(f"Protected live cluster tags: {len(live)}", file=sys.stderr) + for note in protect_notes: + print(f" WARN: {note}", file=sys.stderr) print(f"Forgejo package prune — owner={args.owner} keep={args.max_versions}", file=sys.stderr) print(f"Protected production tags: {len(protected)}", file=sys.stderr) @@ -402,6 +459,8 @@ def main(argv: list[str] | None = None) -> int: apply=apply, deleted=deleted, ) + summary["live_protection"] = args.protect_live + summary["protection_notes"] = protect_notes if args.json or not sys.stdout.isatty(): print(json.dumps(summary, indent=2)) diff --git a/tests/test_forgejo_package_prune.py b/tests/test_forgejo_package_prune.py index 49dbb0c..d3c0ce3 100644 --- a/tests/test_forgejo_package_prune.py +++ b/tests/test_forgejo_package_prune.py @@ -64,5 +64,35 @@ image: self.assertEqual(protected_plans, {("vergabe-teilnahme", "old-prod")}) + def test_collect_live_cluster_versions_protects_forgejo_tags(self) -> None: + images = "\n".join([ + "forgejo.coulomb.social/coulomb/state-hub:f2e042a", + "forgejo.coulomb.social/coulomb/vergabe-teilnahme:064d295", + "registry.k8s.io/pause:3.9", # unrelated registry → ignored + "gitea.coulomb.social/coulomb/old-app:x", # non-forgejo → ignored + "", + ]) + with mock.patch.object(prune.shutil, "which", return_value="/usr/bin/kubectl"), \ + mock.patch.object( + prune.subprocess, "run", + return_value=mock.Mock(stdout=images, returncode=0), + ): + protected, notes = prune.collect_live_cluster_versions() + self.assertEqual(notes, []) + self.assertEqual( + protected, + { + ("container", "state-hub", "f2e042a"), + ("container", "vergabe-teilnahme", "064d295"), + }, + ) + + def test_collect_live_cluster_versions_skips_without_kubectl(self) -> None: + with mock.patch.object(prune.shutil, "which", return_value=None): + protected, notes = prune.collect_live_cluster_versions() + self.assertEqual(protected, set()) + self.assertTrue(notes and "kubectl not found" in notes[0]) + + if __name__ == "__main__": unittest.main() \ No newline at end of file