diff --git a/Dockerfile b/Dockerfile index 3b2787a..39d9156 100644 --- a/Dockerfile +++ b/Dockerfile @@ -14,7 +14,7 @@ COPY .forgejo/workflows/image.yaml ./.forgejo/workflows/image.yaml COPY schemas/ ./schemas/ COPY k8s/ ./k8s/ COPY scripts/render_gitops.py scripts/check_gitops_promotion.py ./scripts/ -RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py +RUN uv sync --frozen --extra dev && uv run --no-sync python scripts/render_gitops.py --check --verify-frontend --verify-platform && uv run --no-sync pytest -q -p no:cacheprovider tests/test_frontend_patterns.py tests/test_instruction_evaluation.py tests/test_admin_sync_api.py tests/test_gitops_release.py # Stage 2 — runtime image FROM python:3.12-slim@sha256:78387bc3881b8273120a12ebe6c1ab22b018ccc2c9adf565ae1ac9b536e184ea AS runtime diff --git a/k8s/gitops/platform-source.json b/k8s/gitops/platform-source.json new file mode 100644 index 0000000..1752ca8 --- /dev/null +++ b/k8s/gitops/platform-source.json @@ -0,0 +1,7 @@ +{ + "schema_version": 1, + "repository": "coulomb/railiance-platform", + "revision": "743def17bec7f32600c8340b8e6b520430b88897", + "path": "scripts/forgejo_package_prune.py", + "sha256": "fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1" +} diff --git a/k8s/gitops/runtime.yaml b/k8s/gitops/runtime.yaml index 8597c49..aec6cd6 100644 --- a/k8s/gitops/runtime.yaml +++ b/k8s/gitops/runtime.yaml @@ -1183,6 +1183,293 @@ data: evidence: [] gaps: - "Add explicit ops inventory probes and evidence events." + forgejo_package_prune.py: "#!/usr/bin/env python3\n\"\"\"Forgejo package retention\ + \ prune \u2014 keep newest N versions per package.\"\"\"\n\nfrom __future__ import\ + \ annotations\n\nimport argparse\nimport json\nimport os\nimport re\nimport shutil\n\ + import subprocess\nimport sys\nimport time\nimport urllib.error\nimport urllib.parse\n\ + import urllib.request\nfrom dataclasses import dataclass\nfrom datetime import\ + \ datetime\nfrom pathlib import Path\nfrom typing import Any\n\nDEFAULT_BASE =\ + \ \"https://forgejo.coulomb.social\"\nDEFAULT_OWNER = \"coulomb\"\nDEFAULT_TYPES\ + \ = (\"container\", \"pypi\", \"npm\", \"generic\")\nDEFAULT_MAX_VERSIONS = 3\n\ + DEFAULT_APPS_ROOT = Path.home() / \"railiance-apps\"\nDEFAULT_FORGEJO_ADMIN_BAO_PATH\ + \ = \"platform/workloads/forgejo/forgejo-admin\"\nDEFAULT_FORGEJO_ADMIN_BAO_FIELD\ + \ = \"API_TOKEN\"\nLEGACY_FORGEJO_TOKEN_FILE = Path(\"/tmp/forgejo-tegwick-api-token\"\ + )\nFORGEJO_IMAGE_RE = re.compile(\n r\"^forgejo\\.coulomb\\.social/(?:coulomb/)?(?P[^:/]+)(?::(?P[^/\\\ + s]+))?$\",\n re.IGNORECASE,\n)\n\n\ndef protect_image(image: str, protected:\ + \ set[tuple[str, str, str]]) -> str | None:\n \"\"\"Digest references conservatively\ + \ protect all versions of their package.\n\n Package APIs do not prove which\ + \ tags or child manifests share a live digest.\n Retaining the whole package\ + \ avoids deleting live/rollback content through an\n alias. The additive inventory\ + \ intentionally keeps this protection until an\n owner explicitly retires the\ + \ reference.\n \"\"\"\n ref, separator, digest = image.partition(\"@\")\n\ + \ match = FORGEJO_IMAGE_RE.fullmatch(ref)\n if not match:\n if image.lower().startswith(\"\ + forgejo.coulomb.social/\"):\n return \"unrecognized Forgejo image reference\"\ + \n return None\n name = match.group(\"name\")\n if separator:\n \ + \ if not re.fullmatch(r\"sha256:[0-9a-f]{64}\", digest):\n return\ + \ \"invalid Forgejo image digest\"\n protected.add((\"container\", name,\ + \ \"*\"))\n else:\n protected.add((\"container\", name, match.group(\"\ + tag\") or \"latest\"))\n return None\n\n\n@dataclass(frozen=True)\nclass VersionRef:\n\ + \ package_type: str\n name: str\n version: str\n\n def key(self) ->\ + \ tuple[str, str, str]:\n return (self.package_type, self.name, self.version)\n\ + \n\n@dataclass(frozen=True)\nclass DeletePlan:\n package_type: str\n name:\ + \ str\n version: str\n created_at: str\n protected: bool\n reason:\ + \ str\n\n\ndef _parse_created_at(value: str | None) -> datetime:\n if not value:\n\ + \ return datetime.min\n try:\n return datetime.fromisoformat(value.replace(\"\ + Z\", \"+00:00\"))\n except ValueError:\n return datetime.min\n\n\ndef\ + \ collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]:\n\ + \ protected: set[tuple[str, str, str]] = set()\n if not apps_root.is_dir():\n\ + \ return protected\n\n patterns = [\n apps_root / \"helm\" /\ + \ \"*-values.yaml\",\n apps_root / \"charts\" / \"*\" / \"values.yaml\"\ + ,\n ]\n paths: list[Path] = []\n for pattern in patterns:\n paths.extend(sorted(pattern.parent.glob(pattern.name)))\n\ + \n try:\n import yaml # type: ignore\n except ImportError:\n \ + \ yaml = None\n\n for path in paths:\n text = path.read_text(encoding=\"\ + utf-8\")\n if yaml is not None:\n try:\n data\ + \ = yaml.safe_load(text) or {}\n except Exception:\n \ + \ data = {}\n image = data.get(\"image\") if isinstance(data, dict)\ + \ else None\n if isinstance(image, dict):\n repo = str(image.get(\"\ + repository\") or \"\").strip()\n tag = str(image.get(\"tag\") or\ + \ \"\").strip()\n if repo and tag:\n match =\ + \ FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match(\n \ + \ f\"{repo}:{tag}\"\n )\n if match:\n\ + \ name = match.group(\"name\")\n \ + \ protected.add((\"container\", name, tag))\n continue\n\n \ + \ repo_match = re.search(\n r\"repository:\\s*forgejo\\.coulomb\\.social/coulomb/([^\\\ + s]+)\",\n text,\n re.IGNORECASE,\n )\n tag_match\ + \ = re.search(r'^\\s*tag:\\s*\"?([^\"\\s#]+)\"?\\s*$', text, re.MULTILINE)\n \ + \ if repo_match and tag_match:\n protected.add((\"container\"\ + , repo_match.group(1), tag_match.group(1)))\n\n return protected\n\n\ndef collect_live_images_from_files(\n\ + \ paths: list[Path],\n) -> tuple[set[tuple[str, str, str]], list[str]]:\n \ + \ \"\"\"Protect image tags listed in exported live-image files.\n\n Each\ + \ file holds one image ref per line (`kubectl get pods ... jsonpath`\n output\ + \ from another cluster). This closes the multi-cluster gap\n (ACTIVITY-WP-0020-T07):\ + \ the prune host's kubectl only sees its own\n cluster, so every other production\ + \ cluster exports its live images to a\n file that is merged here. Unavailable\ + \ or empty exports produce notes;\n main refuses apply when any requested export\ + \ cannot provide coverage.\n \"\"\"\n protected: set[tuple[str, str, str]]\ + \ = set()\n notes: list[str] = []\n for raw_path in paths:\n path\ + \ = raw_path.expanduser()\n if not path.is_file():\n notes.append(f\"\ + live-images file missing: {path}\")\n continue\n try:\n \ + \ lines = path.read_text(encoding=\"utf-8\").splitlines()\n except\ + \ (OSError, UnicodeError):\n notes.append(f\"live-images file unreadable:\ + \ {path}\")\n continue\n has_images = False\n for line\ + \ in lines:\n image = line.strip()\n if not image or image.startswith(\"\ + #\"):\n continue\n has_images = True\n error\ + \ = protect_image(image, protected)\n if error:\n notes.append(f\"\ + {error} in live-images file: {path}\")\n if not has_images:\n \ + \ notes.append(f\"live-images file empty: {path}\")\n return protected, notes\n\ + \n\ndef collect_live_cluster_versions(\n *, kubectl: str = \"kubectl\", timeout:\ + \ float = 60.0\n) -> tuple[set[tuple[str, str, str]], list[str]]:\n \"\"\"\ + Protect image tags currently running in the cluster (best-effort).\n\n Enumerates\ + \ all pod container images across namespaces via kubectl and\n protects any\ + \ `forgejo.coulomb.social/coulomb/:`. This closes the\n gap where\ + \ a live deployment pins a tag not declared in Helm values (e.g.\n CI-deployed\ + \ apps). Failures (no kubectl, no cluster access) return an empty\n set with\ + \ a note \u2014 pruning a reachable registry must not hard-depend on\n cluster\ + \ access, but the note surfaces reduced protection coverage.\n \"\"\"\n \ + \ protected: set[tuple[str, str, str]] = set()\n if shutil.which(kubectl) is\ + \ None:\n return protected, [\"live-tag protection skipped: kubectl not\ + \ found\"]\n jsonpath = (\n \"{range .items[*]}\"\n \"{range\ + \ .spec.containers[*]}{.image}{'\\\\n'}{end}\"\n \"{range .spec.initContainers[*]}{.image}{'\\\ + \\n'}{end}\"\n \"{end}\"\n )\n try:\n result = subprocess.run(\n\ + \ [kubectl, \"get\", \"pods\", \"--all-namespaces\", \"-o\", f\"jsonpath={jsonpath}\"\ + ],\n capture_output=True,\n text=True,\n timeout=timeout,\n\ + \ check=True,\n )\n except Exception as exc: # noqa: BLE001\n\ + \ return protected, [f\"live-tag protection skipped: kubectl query failed\ + \ ({exc})\"]\n notes: list[str] = []\n for line in result.stdout.splitlines():\n\ + \ image = line.strip()\n if not image:\n continue\n \ + \ error = protect_image(image, protected)\n if error:\n \ + \ notes.append(error)\n if not result.stdout.strip():\n notes.append(\"\ + live cluster image inventory empty\")\n return protected, notes\n\n\ndef _api_request(\n\ + \ method: str,\n url: str,\n token: str,\n *,\n timeout: float\ + \ = 60.0,\n retries: int = 2,\n) -> Any:\n req = urllib.request.Request(\n\ + \ url,\n method=method,\n headers={\n \"Authorization\"\ + : f\"token {token}\",\n \"Accept\": \"application/json\",\n \ + \ },\n )\n last_exc: Exception | None = None\n for attempt in range(retries\ + \ + 1):\n try:\n with urllib.request.urlopen(req, timeout=timeout)\ + \ as resp:\n body = resp.read().decode(\"utf-8\")\n \ + \ return json.loads(body) if body else None\n except urllib.error.HTTPError:\n\ + \ raise # 4xx/5xx are real responses \u2014 surface them, do not retry\n\ + \ except (urllib.error.URLError, TimeoutError, OSError) as exc:\n \ + \ # Transient: connection reset, read timeout, TLS handshake timeout.\n\ + \ last_exc = exc\n if attempt < retries:\n \ + \ time.sleep(2 * (attempt + 1))\n continue\n raise\n\ + \ if last_exc: # pragma: no cover - defensive\n raise last_exc\n\n\n\ + def list_packages(\n base_url: str,\n token: str,\n owner: str,\n \ + \ package_type: str,\n) -> list[dict[str, Any]]:\n owner_q = urllib.parse.quote(owner)\n\ + \ items: list[dict[str, Any]] = []\n page = 1\n page_size = 50\n while\ + \ True:\n query = urllib.parse.urlencode(\n {\"limit\": page_size,\ + \ \"page\": page, \"type\": package_type}\n )\n url = f\"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}\"\ + \n payload = _api_request(\"GET\", url, token)\n if not isinstance(payload,\ + \ list):\n raise ValueError(\"invalid package inventory response\"\ + )\n batch = payload\n if not batch:\n break\n \ + \ items.extend(batch)\n if len(batch) < page_size:\n break\n\ + \ page += 1\n return items\n\n\ndef delete_version(\n base_url: str,\n\ + \ token: str,\n owner: str,\n package_type: str,\n name: str,\n \ + \ version: str,\n *,\n dry_run: bool,\n) -> None:\n owner_q = urllib.parse.quote(owner)\n\ + \ type_q = urllib.parse.quote(package_type)\n name_q = urllib.parse.quote(name,\ + \ safe=\"\")\n version_q = urllib.parse.quote(version, safe=\"\")\n path\ + \ = f\"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}\"\n if dry_run:\n\ + \ return\n url = f\"{base_url.rstrip('/')}{path}\"\n _api_request(\"\ + DELETE\", url, token)\n\n\ndef build_delete_plans(\n *,\n base_url: str,\n\ + \ token: str,\n owner: str,\n package_types: list[str],\n max_versions:\ + \ int,\n protected: set[tuple[str, str, str]],\n) -> tuple[list[DeletePlan],\ + \ list[str]]:\n plans: list[DeletePlan] = []\n errors: list[str] = []\n\n\ + \ for package_type in package_types:\n try:\n packages =\ + \ list_packages(base_url, token, owner, package_type)\n except urllib.error.HTTPError\ + \ as exc:\n errors.append(f\"list {package_type}: HTTP {exc.code}\"\ + )\n continue\n except Exception as exc: # noqa: BLE001\n \ + \ errors.append(f\"list {package_type}: {exc}\")\n continue\n\ + \n # Forgejo's package list endpoint returns one entry per (name, version).\n\ + \ # Group by package name; each group is that package's version set \u2014\ + \ there\n # is no separate per-package \"/versions\" endpoint.\n \ + \ by_name: dict[str, list[dict[str, Any]]] = {}\n for package in packages:\n\ + \ name = str(package.get(\"name\") or package.get(\"package_name\"\ + ) or \"\")\n if not name:\n continue\n by_name.setdefault(name,\ + \ []).append(package)\n\n for name, versions in by_name.items():\n \ + \ sorted_versions = sorted(\n versions,\n \ + \ key=lambda item: _parse_created_at(str(item.get(\"created_at\") or \"\")),\n\ + \ reverse=True,\n )\n keep = {\n \ + \ str(item.get(\"version\") or \"\")\n for item in sorted_versions[:max_versions]\n\ + \ if str(item.get(\"version\") or \"\")\n }\n \ + \ for item in sorted_versions[max_versions:]:\n version =\ + \ str(item.get(\"version\") or \"\")\n if not version or version\ + \ in keep:\n continue\n key = (package_type,\ + \ name, version)\n digest_protected = (package_type, name, \"*\"\ + ) in protected\n is_protected = key in protected or digest_protected\n\ + \ plans.append(\n DeletePlan(\n \ + \ package_type=package_type,\n name=name,\n\ + \ version=version,\n created_at=str(item.get(\"\ + created_at\") or \"\"),\n protected=is_protected,\n \ + \ reason=(\"protected_digest_package\" if digest_protected\ + \ else\n \"protected_production_tag\" if is_protected\ + \ else\n \"beyond_retention_depth\"),\n \ + \ )\n )\n return plans, errors\n\n\ndef _read_token_file(path:\ + \ Path) -> str:\n return path.read_text(encoding=\"utf-8\").strip()\n\n\ndef\ + \ _truthy_env(name: str) -> bool:\n return os.environ.get(name, \"\").strip().lower()\ + \ in {\"1\", \"true\", \"yes\", \"on\"}\n\n\ndef _load_token_from_file_env() ->\ + \ str | None:\n for env_name in (\"FORGEJO_TOKEN_FILE\", \"FORGEJO_ADMIN_TOKEN_FILE\"\ + ):\n raw_path = os.environ.get(env_name, \"\").strip()\n if not\ + \ raw_path:\n continue\n path = Path(raw_path).expanduser()\n\ + \ if not path.is_file():\n raise SystemExit(f\"ERROR: {env_name}\ + \ points to a missing file: {path}\")\n token = _read_token_file(path)\n\ + \ if token:\n return token\n raise SystemExit(f\"ERROR:\ + \ {env_name} points to an empty file: {path}\")\n return None\n\n\ndef _load_token_from_openbao()\ + \ -> tuple[str | None, str | None]:\n bao_bin = os.environ.get(\"FORGEJO_ADMIN_BAO_CLI\"\ + , \"bao\").strip() or \"bao\"\n bao_path = (\n os.environ.get(\"FORGEJO_ADMIN_BAO_PATH\"\ + , DEFAULT_FORGEJO_ADMIN_BAO_PATH).strip()\n or DEFAULT_FORGEJO_ADMIN_BAO_PATH\n\ + \ )\n bao_field = (\n os.environ.get(\"FORGEJO_ADMIN_BAO_FIELD\"\ + , DEFAULT_FORGEJO_ADMIN_BAO_FIELD).strip()\n or DEFAULT_FORGEJO_ADMIN_BAO_FIELD\n\ + \ )\n if shutil.which(bao_bin) is None:\n return None, f\"{bao_bin}\ + \ CLI not found\"\n try:\n result = subprocess.run(\n [bao_bin,\ + \ \"kv\", \"get\", f\"-field={bao_field}\", bao_path],\n capture_output=True,\n\ + \ text=True,\n check=True,\n )\n except subprocess.CalledProcessError\ + \ as exc:\n detail = exc.stderr.strip() or exc.stdout.strip() or f\"exit\ + \ {exc.returncode}\"\n return None, f\"{bao_bin} kv get failed: {detail}\"\ + \n except OSError as exc:\n return None, f\"{bao_bin} invocation failed:\ + \ {exc}\"\n token = result.stdout.strip()\n if not token:\n return\ + \ None, f\"{bao_bin} kv get returned an empty {bao_field} field\"\n return\ + \ token, None\n\n\ndef _token_help_message(bao_error: str | None) -> str:\n \ + \ lines = [\n \"ERROR: Forgejo API token required (read:package + write:package).\"\ + ,\n \" Primary: bao login -method=oidc -path=netkingdom role=forgejo-admin-workload-kv-read\"\ + ,\n f\" Default lane: {DEFAULT_FORGEJO_ADMIN_BAO_PATH} field {DEFAULT_FORGEJO_ADMIN_BAO_FIELD}\"\ + ,\n \" Override lane via FORGEJO_ADMIN_BAO_PATH / FORGEJO_ADMIN_BAO_FIELD\ + \ if needed.\",\n \" Break-glass: set FORGEJO_TOKEN / FORGEJO_ADMIN_TOKEN,\ + \ or set FORGEJO_TOKEN_FILE / FORGEJO_ADMIN_TOKEN_FILE.\",\n \" Legacy\ + \ /tmp fallback stays opt-in only via FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1.\"\ + ,\n \" See: railiance-platform/docs/forgejo-package-prune.md\",\n ]\n\ + \ if bao_error:\n lines.insert(3, f\" OpenBao lookup failed: {bao_error}\"\ + )\n return \"\\n\".join(lines)\n\n\ndef load_token() -> str:\n for env_name\ + \ in (\"FORGEJO_TOKEN\", \"FORGEJO_ADMIN_TOKEN\"):\n token = os.environ.get(env_name,\ + \ \"\").strip()\n if token:\n return token\n token = _load_token_from_file_env()\n\ + \ if token:\n return token\n token, bao_error = _load_token_from_openbao()\n\ + \ if token:\n return token\n\n if _truthy_env(\"FORGEJO_ALLOW_LEGACY_FILE_FALLBACK\"\ + ):\n if LEGACY_FORGEJO_TOKEN_FILE.is_file():\n token = _read_token_file(LEGACY_FORGEJO_TOKEN_FILE)\n\ + \ if token:\n return token\n suffix = f\"\ + legacy file {LEGACY_FORGEJO_TOKEN_FILE} is empty\"\n else:\n \ + \ suffix = f\"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is missing\"\n bao_error\ + \ = f\"{bao_error}; {suffix}\" if bao_error else suffix\n\n raise SystemExit(_token_help_message(bao_error))\n\ + \n\ndef emit_summary(\n *,\n owner: str,\n max_versions: int,\n package_types:\ + \ list[str],\n protected: set[tuple[str, str, str]],\n plans: list[DeletePlan],\n\ + \ errors: list[str],\n apply: bool,\n deleted: list[DeletePlan],\n) ->\ + \ dict[str, Any]:\n would_delete = [p for p in plans if not p.protected]\n\ + \ skipped_protected = [p for p in plans if p.protected]\n return {\n \ + \ \"kind\": \"forgejo_package_prune\",\n \"owner\": owner,\n \ + \ \"max_versions\": max_versions,\n \"package_types\": package_types,\n\ + \ \"protected_count\": len(protected),\n \"candidate_count\": len(would_delete),\n\ + \ \"skipped_protected_count\": len(skipped_protected),\n \"deleted_count\"\ + : len(deleted),\n \"apply\": apply,\n \"would_delete\": [\n \ + \ {\n \"type\": p.package_type,\n \"name\"\ + : p.name,\n \"version\": p.version,\n \"created_at\"\ + : p.created_at,\n }\n for p in would_delete\n ],\n\ + \ \"skipped_protected\": [\n {\n \"type\": p.package_type,\n\ + \ \"name\": p.name,\n \"version\": p.version,\n\ + \ \"reason\": p.reason,\n }\n for p in skipped_protected\n\ + \ ],\n \"deleted\": [\n {\n \"type\":\ + \ p.package_type,\n \"name\": p.name,\n \"version\"\ + : p.version,\n }\n for p in deleted\n ],\n \ + \ \"errors\": errors,\n }\n\n\ndef main(argv: list[str] | None = None) ->\ + \ int:\n parser = argparse.ArgumentParser(description=\"Prune old Forgejo package\ + \ versions\")\n parser.add_argument(\"--owner\", default=DEFAULT_OWNER)\n \ + \ parser.add_argument(\"--base-url\", default=os.environ.get(\"FORGEJO_BASE_URL\"\ + , DEFAULT_BASE))\n parser.add_argument(\"--max-versions\", type=int, default=DEFAULT_MAX_VERSIONS)\n\ + \ parser.add_argument(\n \"--types\",\n default=\",\".join(DEFAULT_TYPES),\n\ + \ help=\"Comma-separated package types\",\n )\n parser.add_argument(\"\ + --apps-root\", type=Path, default=DEFAULT_APPS_ROOT)\n parser.add_argument(\"\ + --apply\", action=\"store_true\")\n parser.add_argument(\"--json\", action=\"\ + store_true\", help=\"Emit JSON summary on stdout\")\n parser.add_argument(\n\ + \ \"--no-protect-live\",\n dest=\"protect_live\",\n action=\"\ + store_false\",\n help=\"Skip protecting image tags currently running in\ + \ the cluster (kubectl)\",\n )\n parser.set_defaults(protect_live=True)\n\ + \ parser.add_argument(\n \"--live-images-file\",\n dest=\"live_images_files\"\ + ,\n type=Path,\n action=\"append\",\n default=[],\n \ + \ help=(\n \"File with one image ref per line, exported from another\ + \ \"\n \"production cluster; repeatable. Tags matching the Forgejo\ + \ \"\n \"registry are protected in addition to the local kubectl scan.\"\ + \n ),\n )\n args = parser.parse_args(argv)\n\n apply = bool(args.apply)\n\ + \ dry_run = not apply\n package_types = [part.strip() for part in args.types.split(\"\ + ,\") if part.strip()]\n file_live, file_notes = collect_live_images_from_files(args.live_images_files)\n\ + \ if apply and file_notes:\n for note in file_notes:\n print(f\"\ + \ ERROR: {note}\", file=sys.stderr)\n print(\"Refusing apply: requested\ + \ live-image inventory is unavailable or empty\", file=sys.stderr)\n return\ + \ 2\n token = load_token()\n protected = collect_protected_versions(args.apps_root.expanduser())\n\ + \ protect_notes: list[str] = []\n if args.protect_live:\n live, protect_notes\ + \ = collect_live_cluster_versions()\n protected |= live\n print(f\"\ + Protected live cluster tags: {len(live)}\", file=sys.stderr)\n for note\ + \ in protect_notes:\n print(f\" WARN: {note}\", file=sys.stderr)\n\ + \ if args.live_images_files:\n protected |= file_live\n protect_notes.extend(file_notes)\n\ + \ print(f\"Protected exported live tags: {len(file_live)}\", file=sys.stderr)\n\ + \ for note in file_notes:\n print(f\" WARN: {note}\", file=sys.stderr)\n\ + \n print(f\"Forgejo package prune \u2014 owner={args.owner} keep={args.max_versions}\"\ + , file=sys.stderr)\n print(f\"Protected production tags: {len(protected)}\"\ + , file=sys.stderr)\n\n plans, errors = build_delete_plans(\n base_url=args.base_url,\n\ + \ token=token,\n owner=args.owner,\n package_types=package_types,\n\ + \ max_versions=max(1, args.max_versions),\n protected=protected,\n\ + \ )\n\n # Never partially prune after an incomplete package or requested\ + \ cluster scan.\n if apply and (errors or protect_notes):\n print(\"\ + Refusing apply: incomplete inventory/protection coverage\", file=sys.stderr)\n\ + \ return 2\n\n deleted: list[DeletePlan] = []\n for plan in plans:\n\ + \ if plan.protected:\n print(\n f\" skip protected\ + \ {plan.package_type}/{plan.name}:{plan.version}\",\n file=sys.stderr,\n\ + \ )\n continue\n if dry_run:\n print(\n\ + \ f\" would delete {plan.package_type}/{plan.name}:{plan.version}\"\ + ,\n file=sys.stderr,\n )\n continue\n \ + \ try:\n delete_version(\n args.base_url,\n \ + \ token,\n args.owner,\n plan.package_type,\n\ + \ plan.name,\n plan.version,\n dry_run=False,\n\ + \ )\n deleted.append(plan)\n print(\n \ + \ f\" deleted {plan.package_type}/{plan.name}:{plan.version}\"\ + ,\n file=sys.stderr,\n )\n except urllib.error.HTTPError\ + \ as exc:\n errors.append(f\"delete {plan.package_type}/{plan.name}:{plan.version}:\ + \ HTTP {exc.code}\")\n except Exception as exc: # noqa: BLE001\n \ + \ errors.append(f\"delete {plan.package_type}/{plan.name}:{plan.version}:\ + \ {exc}\")\n\n summary = emit_summary(\n owner=args.owner,\n \ + \ max_versions=args.max_versions,\n package_types=package_types,\n \ + \ protected=protected,\n plans=plans,\n errors=errors,\n \ + \ apply=apply,\n deleted=deleted,\n )\n summary[\"live_protection\"\ + ] = args.protect_live\n summary[\"protection_notes\"] = protect_notes\n\n \ + \ if args.json or not sys.stdout.isatty():\n print(json.dumps(summary,\ + \ indent=2))\n else:\n print(json.dumps(summary, indent=2))\n\n return\ + \ 1 if errors else 0\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main())\n" kind: ConfigMap metadata: name: actcore-ops-service-inventory @@ -1343,6 +1630,7 @@ spec: template: metadata: annotations: + activity-core/retention-sha256: fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1 kubectl.kubernetes.io/restartedAt: '2026-09-05T20:48:57+02:00' labels: app.kubernetes.io/name: actcore-worker @@ -1397,6 +1685,10 @@ spec: - mountPath: /var/custodian/runtime/prompts name: custodian-runtime readOnly: true + - mountPath: /opt/railiance-platform/scripts/forgejo_package_prune.py + name: ops-service-inventory + subPath: forgejo_package_prune.py + readOnly: true - mountPath: /opt/railiance-platform name: railiance-platform readOnly: true diff --git a/k8s/railiance/20-runtime.yaml b/k8s/railiance/20-runtime.yaml index 86349dd..8a7c4a6 100644 --- a/k8s/railiance/20-runtime.yaml +++ b/k8s/railiance/20-runtime.yaml @@ -1079,6 +1079,655 @@ data: evidence: [] gaps: - "Add explicit ops inventory probes and evidence events." + forgejo_package_prune.py: | + #!/usr/bin/env python3 + """Forgejo package retention prune — keep newest N versions per package.""" + + from __future__ import annotations + + import argparse + import json + import os + import re + import shutil + import subprocess + import sys + import time + import urllib.error + import urllib.parse + import urllib.request + from dataclasses import dataclass + from datetime import datetime + from pathlib import Path + from typing import Any + + DEFAULT_BASE = "https://forgejo.coulomb.social" + DEFAULT_OWNER = "coulomb" + DEFAULT_TYPES = ("container", "pypi", "npm", "generic") + DEFAULT_MAX_VERSIONS = 3 + DEFAULT_APPS_ROOT = Path.home() / "railiance-apps" + DEFAULT_FORGEJO_ADMIN_BAO_PATH = "platform/workloads/forgejo/forgejo-admin" + DEFAULT_FORGEJO_ADMIN_BAO_FIELD = "API_TOKEN" + LEGACY_FORGEJO_TOKEN_FILE = Path("/tmp/forgejo-tegwick-api-token") + FORGEJO_IMAGE_RE = re.compile( + r"^forgejo\.coulomb\.social/(?:coulomb/)?(?P[^:/]+)(?::(?P[^/\s]+))?$", + re.IGNORECASE, + ) + + + def protect_image(image: str, protected: set[tuple[str, str, str]]) -> str | None: + """Digest references conservatively protect all versions of their package. + + Package APIs do not prove which tags or child manifests share a live digest. + Retaining the whole package avoids deleting live/rollback content through an + alias. The additive inventory intentionally keeps this protection until an + owner explicitly retires the reference. + """ + ref, separator, digest = image.partition("@") + match = FORGEJO_IMAGE_RE.fullmatch(ref) + if not match: + if image.lower().startswith("forgejo.coulomb.social/"): + return "unrecognized Forgejo image reference" + return None + name = match.group("name") + if separator: + if not re.fullmatch(r"sha256:[0-9a-f]{64}", digest): + return "invalid Forgejo image digest" + protected.add(("container", name, "*")) + else: + protected.add(("container", name, match.group("tag") or "latest")) + return None + + + @dataclass(frozen=True) + class VersionRef: + package_type: str + name: str + version: str + + def key(self) -> tuple[str, str, str]: + return (self.package_type, self.name, self.version) + + + @dataclass(frozen=True) + class DeletePlan: + package_type: str + name: str + version: str + created_at: str + protected: bool + reason: str + + + def _parse_created_at(value: str | None) -> datetime: + if not value: + return datetime.min + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return datetime.min + + + def collect_protected_versions(apps_root: Path) -> set[tuple[str, str, str]]: + protected: set[tuple[str, str, str]] = set() + if not apps_root.is_dir(): + return protected + + patterns = [ + apps_root / "helm" / "*-values.yaml", + apps_root / "charts" / "*" / "values.yaml", + ] + paths: list[Path] = [] + for pattern in patterns: + paths.extend(sorted(pattern.parent.glob(pattern.name))) + + try: + import yaml # type: ignore + except ImportError: + yaml = None + + for path in paths: + text = path.read_text(encoding="utf-8") + if yaml is not None: + try: + data = yaml.safe_load(text) or {} + except Exception: + data = {} + image = data.get("image") if isinstance(data, dict) else None + if isinstance(image, dict): + repo = str(image.get("repository") or "").strip() + tag = str(image.get("tag") or "").strip() + if repo and tag: + match = FORGEJO_IMAGE_RE.match(repo) or FORGEJO_IMAGE_RE.match( + f"{repo}:{tag}" + ) + if match: + name = match.group("name") + protected.add(("container", name, tag)) + continue + + repo_match = re.search( + r"repository:\s*forgejo\.coulomb\.social/coulomb/([^\s]+)", + text, + re.IGNORECASE, + ) + tag_match = re.search(r'^\s*tag:\s*"?([^"\s#]+)"?\s*$', text, re.MULTILINE) + if repo_match and tag_match: + protected.add(("container", repo_match.group(1), tag_match.group(1))) + + return protected + + + def collect_live_images_from_files( + paths: list[Path], + ) -> tuple[set[tuple[str, str, str]], list[str]]: + """Protect image tags listed in exported live-image files. + + Each file holds one image ref per line (`kubectl get pods ... jsonpath` + output from another cluster). This closes the multi-cluster gap + (ACTIVITY-WP-0020-T07): the prune host's kubectl only sees its own + cluster, so every other production cluster exports its live images to a + file that is merged here. Unavailable or empty exports produce notes; + main refuses apply when any requested export cannot provide coverage. + """ + protected: set[tuple[str, str, str]] = set() + notes: list[str] = [] + for raw_path in paths: + path = raw_path.expanduser() + if not path.is_file(): + notes.append(f"live-images file missing: {path}") + continue + try: + lines = path.read_text(encoding="utf-8").splitlines() + except (OSError, UnicodeError): + notes.append(f"live-images file unreadable: {path}") + continue + has_images = False + for line in lines: + image = line.strip() + if not image or image.startswith("#"): + continue + has_images = True + error = protect_image(image, protected) + if error: + notes.append(f"{error} in live-images file: {path}") + if not has_images: + notes.append(f"live-images file empty: {path}") + return protected, notes + + + 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})"] + notes: list[str] = [] + for line in result.stdout.splitlines(): + image = line.strip() + if not image: + continue + error = protect_image(image, protected) + if error: + notes.append(error) + if not result.stdout.strip(): + notes.append("live cluster image inventory empty") + return protected, notes + + + def _api_request( + method: str, + url: str, + token: str, + *, + timeout: float = 60.0, + retries: int = 2, + ) -> Any: + req = urllib.request.Request( + url, + method=method, + headers={ + "Authorization": f"token {token}", + "Accept": "application/json", + }, + ) + last_exc: Exception | None = None + for attempt in range(retries + 1): + try: + with urllib.request.urlopen(req, timeout=timeout) as resp: + body = resp.read().decode("utf-8") + return json.loads(body) if body else None + except urllib.error.HTTPError: + raise # 4xx/5xx are real responses — surface them, do not retry + except (urllib.error.URLError, TimeoutError, OSError) as exc: + # Transient: connection reset, read timeout, TLS handshake timeout. + last_exc = exc + if attempt < retries: + time.sleep(2 * (attempt + 1)) + continue + raise + if last_exc: # pragma: no cover - defensive + raise last_exc + + + def list_packages( + base_url: str, + token: str, + owner: str, + package_type: str, + ) -> list[dict[str, Any]]: + owner_q = urllib.parse.quote(owner) + items: list[dict[str, Any]] = [] + page = 1 + page_size = 50 + while True: + query = urllib.parse.urlencode( + {"limit": page_size, "page": page, "type": package_type} + ) + url = f"{base_url.rstrip('/')}/api/v1/packages/{owner_q}?{query}" + payload = _api_request("GET", url, token) + if not isinstance(payload, list): + raise ValueError("invalid package inventory response") + batch = payload + if not batch: + break + items.extend(batch) + if len(batch) < page_size: + break + page += 1 + return items + + + def delete_version( + base_url: str, + token: str, + owner: str, + package_type: str, + name: str, + version: str, + *, + dry_run: bool, + ) -> None: + owner_q = urllib.parse.quote(owner) + type_q = urllib.parse.quote(package_type) + name_q = urllib.parse.quote(name, safe="") + version_q = urllib.parse.quote(version, safe="") + path = f"/api/v1/packages/{owner_q}/{type_q}/{name_q}/{version_q}" + if dry_run: + return + url = f"{base_url.rstrip('/')}{path}" + _api_request("DELETE", url, token) + + + def build_delete_plans( + *, + base_url: str, + token: str, + owner: str, + package_types: list[str], + max_versions: int, + protected: set[tuple[str, str, str]], + ) -> tuple[list[DeletePlan], list[str]]: + plans: list[DeletePlan] = [] + errors: list[str] = [] + + for package_type in package_types: + try: + packages = list_packages(base_url, token, owner, package_type) + except urllib.error.HTTPError as exc: + errors.append(f"list {package_type}: HTTP {exc.code}") + continue + except Exception as exc: # noqa: BLE001 + errors.append(f"list {package_type}: {exc}") + continue + + # Forgejo's package list endpoint returns one entry per (name, version). + # Group by package name; each group is that package's version set — there + # is no separate per-package "/versions" endpoint. + by_name: dict[str, list[dict[str, Any]]] = {} + for package in packages: + name = str(package.get("name") or package.get("package_name") or "") + if not name: + continue + by_name.setdefault(name, []).append(package) + + for name, versions in by_name.items(): + sorted_versions = sorted( + versions, + key=lambda item: _parse_created_at(str(item.get("created_at") or "")), + reverse=True, + ) + keep = { + str(item.get("version") or "") + for item in sorted_versions[:max_versions] + if str(item.get("version") or "") + } + for item in sorted_versions[max_versions:]: + version = str(item.get("version") or "") + if not version or version in keep: + continue + key = (package_type, name, version) + digest_protected = (package_type, name, "*") in protected + is_protected = key in protected or digest_protected + plans.append( + DeletePlan( + package_type=package_type, + name=name, + version=version, + created_at=str(item.get("created_at") or ""), + protected=is_protected, + reason=("protected_digest_package" if digest_protected else + "protected_production_tag" if is_protected else + "beyond_retention_depth"), + ) + ) + return plans, errors + + + def _read_token_file(path: Path) -> str: + return path.read_text(encoding="utf-8").strip() + + + def _truthy_env(name: str) -> bool: + return os.environ.get(name, "").strip().lower() in {"1", "true", "yes", "on"} + + + def _load_token_from_file_env() -> str | None: + for env_name in ("FORGEJO_TOKEN_FILE", "FORGEJO_ADMIN_TOKEN_FILE"): + raw_path = os.environ.get(env_name, "").strip() + if not raw_path: + continue + path = Path(raw_path).expanduser() + if not path.is_file(): + raise SystemExit(f"ERROR: {env_name} points to a missing file: {path}") + token = _read_token_file(path) + if token: + return token + raise SystemExit(f"ERROR: {env_name} points to an empty file: {path}") + return None + + + def _load_token_from_openbao() -> tuple[str | None, str | None]: + bao_bin = os.environ.get("FORGEJO_ADMIN_BAO_CLI", "bao").strip() or "bao" + bao_path = ( + os.environ.get("FORGEJO_ADMIN_BAO_PATH", DEFAULT_FORGEJO_ADMIN_BAO_PATH).strip() + or DEFAULT_FORGEJO_ADMIN_BAO_PATH + ) + bao_field = ( + os.environ.get("FORGEJO_ADMIN_BAO_FIELD", DEFAULT_FORGEJO_ADMIN_BAO_FIELD).strip() + or DEFAULT_FORGEJO_ADMIN_BAO_FIELD + ) + if shutil.which(bao_bin) is None: + return None, f"{bao_bin} CLI not found" + try: + result = subprocess.run( + [bao_bin, "kv", "get", f"-field={bao_field}", bao_path], + capture_output=True, + text=True, + check=True, + ) + except subprocess.CalledProcessError as exc: + detail = exc.stderr.strip() or exc.stdout.strip() or f"exit {exc.returncode}" + return None, f"{bao_bin} kv get failed: {detail}" + except OSError as exc: + return None, f"{bao_bin} invocation failed: {exc}" + token = result.stdout.strip() + if not token: + return None, f"{bao_bin} kv get returned an empty {bao_field} field" + return token, None + + + def _token_help_message(bao_error: str | None) -> str: + lines = [ + "ERROR: Forgejo API token required (read:package + write:package).", + " Primary: bao login -method=oidc -path=netkingdom role=forgejo-admin-workload-kv-read", + f" Default lane: {DEFAULT_FORGEJO_ADMIN_BAO_PATH} field {DEFAULT_FORGEJO_ADMIN_BAO_FIELD}", + " Override lane via FORGEJO_ADMIN_BAO_PATH / FORGEJO_ADMIN_BAO_FIELD if needed.", + " Break-glass: set FORGEJO_TOKEN / FORGEJO_ADMIN_TOKEN, or set FORGEJO_TOKEN_FILE / FORGEJO_ADMIN_TOKEN_FILE.", + " Legacy /tmp fallback stays opt-in only via FORGEJO_ALLOW_LEGACY_FILE_FALLBACK=1.", + " See: railiance-platform/docs/forgejo-package-prune.md", + ] + if bao_error: + lines.insert(3, f" OpenBao lookup failed: {bao_error}") + return "\n".join(lines) + + + def load_token() -> str: + for env_name in ("FORGEJO_TOKEN", "FORGEJO_ADMIN_TOKEN"): + token = os.environ.get(env_name, "").strip() + if token: + return token + token = _load_token_from_file_env() + if token: + return token + token, bao_error = _load_token_from_openbao() + if token: + return token + + if _truthy_env("FORGEJO_ALLOW_LEGACY_FILE_FALLBACK"): + if LEGACY_FORGEJO_TOKEN_FILE.is_file(): + token = _read_token_file(LEGACY_FORGEJO_TOKEN_FILE) + if token: + return token + suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is empty" + else: + suffix = f"legacy file {LEGACY_FORGEJO_TOKEN_FILE} is missing" + bao_error = f"{bao_error}; {suffix}" if bao_error else suffix + + raise SystemExit(_token_help_message(bao_error)) + + + def emit_summary( + *, + owner: str, + max_versions: int, + package_types: list[str], + protected: set[tuple[str, str, str]], + plans: list[DeletePlan], + errors: list[str], + apply: bool, + deleted: list[DeletePlan], + ) -> dict[str, Any]: + would_delete = [p for p in plans if not p.protected] + skipped_protected = [p for p in plans if p.protected] + return { + "kind": "forgejo_package_prune", + "owner": owner, + "max_versions": max_versions, + "package_types": package_types, + "protected_count": len(protected), + "candidate_count": len(would_delete), + "skipped_protected_count": len(skipped_protected), + "deleted_count": len(deleted), + "apply": apply, + "would_delete": [ + { + "type": p.package_type, + "name": p.name, + "version": p.version, + "created_at": p.created_at, + } + for p in would_delete + ], + "skipped_protected": [ + { + "type": p.package_type, + "name": p.name, + "version": p.version, + "reason": p.reason, + } + for p in skipped_protected + ], + "deleted": [ + { + "type": p.package_type, + "name": p.name, + "version": p.version, + } + for p in deleted + ], + "errors": errors, + } + + + def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Prune old Forgejo package versions") + parser.add_argument("--owner", default=DEFAULT_OWNER) + parser.add_argument("--base-url", default=os.environ.get("FORGEJO_BASE_URL", DEFAULT_BASE)) + parser.add_argument("--max-versions", type=int, default=DEFAULT_MAX_VERSIONS) + parser.add_argument( + "--types", + default=",".join(DEFAULT_TYPES), + help="Comma-separated package types", + ) + 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) + parser.add_argument( + "--live-images-file", + dest="live_images_files", + type=Path, + action="append", + default=[], + help=( + "File with one image ref per line, exported from another " + "production cluster; repeatable. Tags matching the Forgejo " + "registry are protected in addition to the local kubectl scan." + ), + ) + args = parser.parse_args(argv) + + apply = bool(args.apply) + dry_run = not apply + package_types = [part.strip() for part in args.types.split(",") if part.strip()] + file_live, file_notes = collect_live_images_from_files(args.live_images_files) + if apply and file_notes: + for note in file_notes: + print(f" ERROR: {note}", file=sys.stderr) + print("Refusing apply: requested live-image inventory is unavailable or empty", file=sys.stderr) + return 2 + 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) + if args.live_images_files: + protected |= file_live + protect_notes.extend(file_notes) + print(f"Protected exported live tags: {len(file_live)}", file=sys.stderr) + for note in file_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) + + plans, errors = build_delete_plans( + base_url=args.base_url, + token=token, + owner=args.owner, + package_types=package_types, + max_versions=max(1, args.max_versions), + protected=protected, + ) + + # Never partially prune after an incomplete package or requested cluster scan. + if apply and (errors or protect_notes): + print("Refusing apply: incomplete inventory/protection coverage", file=sys.stderr) + return 2 + + deleted: list[DeletePlan] = [] + for plan in plans: + if plan.protected: + print( + f" skip protected {plan.package_type}/{plan.name}:{plan.version}", + file=sys.stderr, + ) + continue + if dry_run: + print( + f" would delete {plan.package_type}/{plan.name}:{plan.version}", + file=sys.stderr, + ) + continue + try: + delete_version( + args.base_url, + token, + args.owner, + plan.package_type, + plan.name, + plan.version, + dry_run=False, + ) + deleted.append(plan) + print( + f" deleted {plan.package_type}/{plan.name}:{plan.version}", + file=sys.stderr, + ) + except urllib.error.HTTPError as exc: + errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: HTTP {exc.code}") + except Exception as exc: # noqa: BLE001 + errors.append(f"delete {plan.package_type}/{plan.name}:{plan.version}: {exc}") + + summary = emit_summary( + owner=args.owner, + max_versions=args.max_versions, + package_types=package_types, + protected=protected, + plans=plans, + errors=errors, + 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)) + else: + print(json.dumps(summary, indent=2)) + + return 1 if errors else 0 + + + if __name__ == "__main__": + raise SystemExit(main()) kind: ConfigMap metadata: name: actcore-ops-service-inventory @@ -1528,6 +2177,7 @@ spec: template: metadata: annotations: + activity-core/retention-sha256: fa3d4cc8cdcc14ecce509cf6865c6e4726d722f84814b97d40686d68d6ee5bb1 kubectl.kubernetes.io/restartedAt: '2026-09-05T20:48:57+02:00' labels: app.kubernetes.io/name: actcore-worker @@ -1582,6 +2232,10 @@ spec: - mountPath: /var/custodian/runtime/prompts name: custodian-runtime readOnly: true + - mountPath: /opt/railiance-platform/scripts/forgejo_package_prune.py + name: ops-service-inventory + subPath: forgejo_package_prune.py + readOnly: true - mountPath: /opt/railiance-platform name: railiance-platform readOnly: true diff --git a/scripts/render_gitops.py b/scripts/render_gitops.py index 9a312bd..5295ed2 100644 --- a/scripts/render_gitops.py +++ b/scripts/render_gitops.py @@ -40,11 +40,31 @@ def verify_frontend(source_dir=None): if len(body)>32768 or hashlib.sha256(body).hexdigest()!=digest or cm['data'][name].encode()!=body: raise ValueError('frontend definition projection mismatch: '+name) +def verify_platform(source_file=None): + pin=json.loads((ROOT/'k8s/gitops/platform-source.json').read_text()) + if (pin.get('schema_version')!=1 or pin.get('repository')!='coulomb/railiance-platform' + or pin.get('path')!='scripts/forgejo_package_prune.py' + or not re.fullmatch('[0-9a-f]{40}',pin['revision'])): + raise ValueError('invalid platform source pin') + if source_file is None: + url=f"https://forgejo.coulomb.social/coulomb/railiance-platform/raw/commit/{pin['revision']}/{pin['path']}" + with urllib.request.urlopen(url,timeout=10) as response: body=response.read(131073) + else: body=Path(source_file).read_bytes() + docs=list(yaml.safe_load_all(render())) + cm=next(d for d in docs if d['metadata']['name']=='actcore-ops-service-inventory') + worker=next(d for d in docs if d['metadata']['name']=='actcore-worker') + if (len(body)>131072 or hashlib.sha256(body).hexdigest()!=pin['sha256'] + or cm['data']['forgejo_package_prune.py'].encode()!=body + or worker['spec']['template']['metadata']['annotations'].get('activity-core/retention-sha256')!=pin['sha256']): + raise ValueError('platform tool projection mismatch') + if __name__=='__main__': - parser=argparse.ArgumentParser(); parser.add_argument('--check',action='store_true'); parser.add_argument('--verify-frontend',action='store_true'); args=parser.parse_args() + parser=argparse.ArgumentParser(); parser.add_argument('--check',action='store_true'); parser.add_argument('--verify-frontend',action='store_true'); parser.add_argument('--verify-platform',action='store_true'); args=parser.parse_args() path=ROOT/'k8s/gitops/runtime.yaml'; text=render() if args.check: if path.read_text()!=text: raise SystemExit('GitOps projection stale; run scripts/render_gitops.py') else: path.write_text(text) if args.verify_frontend: verify_frontend() + + if args.verify_platform: verify_platform() diff --git a/tests/test_gitops_release.py b/tests/test_gitops_release.py index c6ce51e..7770a26 100644 --- a/tests/test_gitops_release.py +++ b/tests/test_gitops_release.py @@ -72,3 +72,16 @@ def test_frontend_projection_checks_content_and_pin(tmp_path): renderer.verify_frontend(tmp_path) (tmp_path/'frontend-patterns-daily.md').write_text('tampered') with pytest.raises(ValueError): renderer.verify_frontend(tmp_path) + + +def test_retention_projection_is_pinned_and_mounted(tmp_path): + renderer=load('render_gitops') + docs=list(yaml.safe_load_all(renderer.render())) + cm=next(d for d in docs if d['metadata']['name']=='actcore-ops-service-inventory') + source=tmp_path/'prune.py';source.write_text(cm['data']['forgejo_package_prune.py']) + renderer.verify_platform(source) + worker=next(d for d in docs if d['metadata']['name']=='actcore-worker') + mounts=worker['spec']['template']['spec']['containers'][0]['volumeMounts'] + assert any(m.get('subPath')=='forgejo_package_prune.py' and m['readOnly'] is True for m in mounts) + source.write_text('tampered') + with pytest.raises(ValueError):renderer.verify_platform(source)