"""Governed scaffolding for rapp-* managed workload packages.""" from __future__ import annotations import argparse import re import shutil import subprocess from pathlib import Path from typing import Any PACKAGE_TYPES = ( "helm-managed-platform-service", "manifest-managed-platform-service", "knative-managed-service", "grouped-composition", ) RAILS = ("rail-kubernetes", "rail-knative") CLASSIFICATIONS = ("public", "internal", "confidential", "restricted") CRITICALITIES = ("low", "medium", "high", "critical") COMPUTE_REEFS = ("reef-railiance",) SLUG = re.compile(r"^[a-z0-9]+(-[a-z0-9]+)*$") REEF_SLUG = re.compile(r"^reef-[a-z0-9]+(-[a-z0-9]+)*$") DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$") EXPOSE_RE = re.compile(r"^EXPOSE\s+(\d+)", re.MULTILINE) USER_RE = re.compile(r"^USER\s+(\d+)", re.MULTILINE) HEALTH_PATH_RE = re.compile(r'path:\s*["\']?(/[A-Za-z0-9._/-]+)') IMAGE_PIN_RE = re.compile( r"(forgejo\.coulomb\.social/coulomb/[a-z0-9-]+@)sha256:[0-9a-f]{64}" ) UNSET_DIGEST = "sha256:" + ("0" * 64) _FAMILY_VALIDATOR = Path.home() / "railiance-master" / "tools" / "validate-family-declarations.py" def _refuse(message: str) -> dict[str, Any]: return {"ok": False, "error": message} def _write(path: Path, content: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(content if content.endswith("\n") else content + "\n") def _underscore(app: str) -> str: return app.replace("-", "_") def init( path: Path, *, app: str, ownership_repo: str, rail: str = "rail-kubernetes", classification: str = "confidential", criticality: str = "high", package_type: str = "manifest-managed-platform-service", purpose: str | None = None, force: bool = False, ) -> dict[str, Any]: check = _check_identity(app, ownership_repo, rail, package_type, classification, criticality) if check: return check rapp_id = f"rapp-{app}" dest = path.expanduser().resolve() dest.mkdir(parents=True, exist_ok=True) declaration = dest / "declarations" / "rapp.yaml" if declaration.is_file() and not force: return _refuse(f"{declaration} already exists; refuse to overwrite a live contract") purpose_text = purpose or f"Package and operate the {app} service on Railiance." files = { ".repo-classification.yaml": ( "repo_classification:\n" ' standard: Repo Classification Standard\n' ' version: "1.0"\n' " classified_by: repo-manager\n" " category: project\n" " domain: infotech\n" ), "README.md": f"# {rapp_id}\n\nManaged runtime package for `{app}`.\n", "INTENT.md": ( f"# Intent\n\nProvide the managed Railiance runtime package for `{app}`.\n" ), "SCOPE.md": ( "# Scope\n\n## In scope\n\n- packaging, smoke, rollback, and the family declaration\n\n" "## Out of scope\n\n- application domain ownership\n- credential values\n" ), "AGENTS.md": ( f"# Repository agent guide\n\nWorkplans use the prefix " f"`{rapp_id.upper()}-WP-`.\n" ), "CLAUDE.md": f"# {rapp_id}\n\n@SCOPE.md\n@AGENTS.md\n", "Makefile": ( "SHELL := /bin/bash\n\n.PHONY: check\n\ncheck:\n" "\tpython3 -m unittest discover -s tests -v\n" ), ".gitignore": ".rendered/\n__pycache__/\n", "declarations/rapp.yaml": _declaration( rapp_id=rapp_id, app=app, ownership_repo=ownership_repo, rail=rail, classification=classification, criticality=criticality, package_type=package_type, purpose=purpose_text, ), f"workplans/{rapp_id.upper()}-WP-0001-bootstrap.md": ( f"---\nid: {rapp_id.upper()}-WP-0001\ntype: workplan\n" f'title: "Bootstrap {rapp_id}"\nstatus: proposed\n' f"flavor: planning\ndepends_on: []\n---\n\n" f"# Bootstrap {rapp_id}\n" ), } written = [] for rel, content in files.items(): target = dest / rel if target.is_file() and not force and rel not in {"declarations/rapp.yaml"}: continue _write(target, content) written.append(rel) (dest / "workplans" / "archived").mkdir(exist_ok=True) return {"ok": True, "path": str(dest), "rapp_id": rapp_id, "written": written} def _check_identity( app: str, ownership_repo: str, rail: str, package_type: str, classification: str, criticality: str, ) -> dict[str, Any] | None: if not SLUG.match(app) or app.startswith("rapp-"): return _refuse("app must be a workload slug without the rapp- prefix") if rail not in RAILS: return _refuse(f"unknown rail {rail!r}; allowed: {', '.join(RAILS)}") if package_type not in PACKAGE_TYPES: return _refuse(f"unknown package_type {package_type!r}") if classification not in CLASSIFICATIONS: return _refuse(f"unknown classification {classification!r}") if criticality not in CRITICALITIES: return _refuse(f"unknown criticality {criticality!r}") if not SLUG.match(ownership_repo) or ownership_repo.startswith("rapp-"): return _refuse("ownership_repo must be an app or layer slug, not the rapp itself") return None def _declaration( *, rapp_id: str, app: str, ownership_repo: str, rail: str, classification: str, criticality: str, package_type: str, purpose: str, ) -> str: return ( "kind: managed-workload-package\n" "repo_family: rapp\n" f"rapp_id: {rapp_id}\n" f"repo: {rapp_id}\n" f"ownership_repo: {ownership_repo}\n" "contract_version: 1.0.0\n" "readiness_state: draft\n" "workload_identity:\n" f" name: {app}\n" f" package_type: {package_type}\n" f"data_classification: {classification}\n" f"criticality: {criticality}\n" f"primary_rail: {rail}\n" "supported_rails:\n" f" - {rail}\n" "bound_reefs: []\n" "runtime_dependencies:\n" " - kubernetes-api\n" " - openbao-database-secrets-engine\n" "composition:\n" f" purpose: {purpose}\n" " member_repos:\n" f" - repo: {rapp_id}\n" " role: managed runtime package\n" " deployables:\n" f" - {app}\n" "rollout_contract:\n" " default_mode: kubectl-server-side-apply\n" "smoke_contract:\n" " required:\n" " - healthz-ok\n" "rollback_contract:\n" " order:\n" " - previous-immutable-image-digest\n" " - apply-reviewed-git-revision\n" ) def inspect_app(from_app: Path, app: str) -> dict[str, Any]: root = from_app.expanduser().resolve() container = "" for name in ("Containerfile", "Dockerfile"): candidate = root / name if candidate.is_file(): container = candidate.read_text() break expose = EXPOSE_RE.search(container) user = USER_RE.search(container) port = int(expose.group(1)) if expose else 8080 uid = int(user.group(1)) if user else 10001 deploy_dir = root / "deploy" deploy_files = sorted(deploy_dir.glob("*.yaml")) if deploy_dir.is_dir() else [] health = "/healthz" ready = "/readyz" blob = container for path in deploy_files: blob += "\n" + path.read_text() for src in (root / "src").rglob("*.py") if (root / "src").is_dir() else []: try: blob += "\n" + src.read_text() except OSError: continue if len(blob) > 400_000: break paths = HEALTH_PATH_RE.findall(blob) if "/health" in paths and "/healthz" not in paths: health = "/health" ready = "/health" if "/healthz" in paths: health = "/healthz" if "/readyz" in paths: ready = "/readyz" digest = UNSET_DIGEST pin = IMAGE_PIN_RE.search(blob) if pin: digest = "sha256:" + pin.group(0).rsplit("sha256:", 1)[1] return { "app": app, "root": str(root), "port": port, "uid": uid, "health_path": health, "ready_path": ready, "image_repository": f"forgejo.coulomb.social/coulomb/{app}", "image_digest": digest, "deploy_files": [str(p) for p in deploy_files], "has_chart": (root / "helm").is_dir() or (root / "charts").is_dir(), "has_image_workflow": (root / ".forgejo" / "workflows" / "image.yaml").is_file(), } def skeleton( path: Path, *, from_app: Path, app: str | None = None, package_type: str = "manifest-managed-platform-service", force: bool = False, dedicated_postgres: bool = False, ) -> dict[str, Any]: dest = path.expanduser().resolve() app_name = app or dest.name.removeprefix("rapp-") if package_type == "helm-managed-platform-service": return _refuse("Helm skeleton is not generated; pass an existing chart or use manifests") if package_type not in PACKAGE_TYPES: return _refuse(f"unknown package_type {package_type!r}") facts = inspect_app(from_app, app_name) written: list[str] = [] notes: list[str] = [] if facts["deploy_files"]: manifests = dest / "manifests" if manifests.exists() and any(manifests.glob("*.yaml")) and not force: return _refuse(f"{manifests} already has manifests; pass --force to replace") manifests.mkdir(parents=True, exist_ok=True) for src in facts["deploy_files"]: target = manifests / Path(src).name shutil.copy2(src, target) written.append(str(target.relative_to(dest))) notes.append(f"absorbed {len(facts['deploy_files'])} file(s) from {from_app}/deploy") else: runtime = dest / "manifests" / "runtime.yaml" if runtime.is_file() and not force: return _refuse(f"{runtime} already exists; pass --force to replace") _write(runtime, _runtime_manifest(app_name, facts)) written.append("manifests/runtime.yaml") notes.append("generated Deployment/Service/ServiceAccount/NetworkPolicy") if dedicated_postgres: notes.append("dedicated Cluster not generated; pass reviewed CNPG YAML by hand") makefile = dest / "Makefile" if not makefile.is_file() or force or makefile.read_text().count("\n") < 12: _write(makefile, _makefile(app_name, facts, dest.name)) written.append("Makefile") render = dest / "tools" / "render.py" if not render.is_file() or force: _write(render, _render_py(app_name)) render.chmod(0o755) written.append("tools/render.py") verify = dest / "tools" / "verify_live.sh" if not verify.is_file() or force: _write(verify, _verify_sh(app_name, facts)) verify.chmod(0o755) written.append("tools/verify_live.sh") test = dest / "tests" / "test_packaging.py" if not test.is_file() or force: _write(test, _packaging_test(app_name)) written.append("tests/test_packaging.py") return {"ok": True, "path": str(dest), "facts": facts, "written": written, "notes": notes} def ensure_image_workflow(from_app: Path, app: str) -> dict[str, Any]: root = from_app.expanduser().resolve() if not root.is_dir(): return _refuse(f"app checkout missing: {root}") target = root / ".forgejo" / "workflows" / "image.yaml" if target.is_file(): return {"ok": True, "path": str(target), "written": False, "note": "already present"} _write(target, _image_workflow(app)) return { "ok": True, "path": str(target), "written": True, "note": "first deploy waits on the CI digest; no workstation build", } def draft_postgres_consumer( path: Path, *, app: str, family_root: Path | None = None, ) -> dict[str, Any]: dest = path.expanduser().resolve() app_name = app or dest.name.removeprefix("rapp-") slug = _underscore(app_name) draft = ( "apiVersion: rapp-postgres.railiance.io/v1alpha1\n" "kind: PostgresConsumer\n" "metadata:\n" f" name: {app_name}\n" "spec:\n" f" database: {slug}\n" f" schema: {slug}\n" f" costAttributionKey: platform:{app_name}\n" f" clientNamespaces: [{app_name}]\n" " roles:\n" f" owner: {slug}_owner\n" f" migration: {slug}_migrate\n" f" runtime: {slug}_app\n" " tenantKeyingRequired: true\n" " # Draft only. Do not apply from this package. Review in rapp-postgres.\n" ) handoff = dest / "handoffs" / "postgres-consumer.yaml" search = (family_root or dest.parent).resolve() live = search / "rapp-postgres" / "consumers" / f"{app_name}.yaml" notes = [] if live.is_file(): _write( dest / "handoffs" / "README.md", f"# Handoffs\n\nPostgres consumer already lives at `{live}`.\n" "This package does not apply it.\n", ) notes.append(f"existing consumer left in place: {live}") return {"ok": True, "path": str(live), "written": False, "notes": notes} _write(handoff, draft) notes.append(str(handoff)) if (search / "rapp-postgres" / "consumers").is_dir(): target = search / "rapp-postgres" / "consumers" / f"{app_name}.yaml" if not target.is_file(): _write(target, draft) notes.append(str(target)) return {"ok": True, "path": str(handoff), "written": True, "notes": notes} def wrap( path: Path, *, app: str, ownership_repo: str, from_app: Path, rail: str = "rail-kubernetes", classification: str = "confidential", criticality: str = "high", package_type: str = "manifest-managed-platform-service", purpose: str | None = None, force: bool = False, dedicated_postgres: bool = False, family_root: Path | None = None, ) -> dict[str, Any]: dest = path.expanduser().resolve() created = init( dest, app=app, ownership_repo=ownership_repo, rail=rail, classification=classification, criticality=criticality, package_type=package_type, purpose=purpose, force=force, ) if not created.get("ok"): return created skel = skeleton( dest, from_app=from_app, app=app, package_type=package_type, force=force, dedicated_postgres=dedicated_postgres, ) if not skel.get("ok"): return skel image = ensure_image_workflow(from_app, app) if not image.get("ok"): return image consumer = {} if dedicated_postgres else draft_postgres_consumer( dest, app=app, family_root=family_root or dest.parent ) if consumer and not consumer.get("ok"): return consumer checked = validate(dest, family_root=family_root or dest.parent) return { "ok": True, "path": str(dest), "init": created, "skeleton": skel, "image_workflow": image, "postgres_consumer": consumer, "validate": checked, "placed": False, "applied": False, } def place(path: Path, *, reef: str, family_root: Path | None = None) -> dict[str, Any]: if not REEF_SLUG.match(reef): return _refuse(f"unknown reef {reef!r}") if reef == "reef-storage": return _refuse("reef-storage hosts no rail; it is a consumed capability") dest = path.expanduser().resolve() declaration = dest / "declarations" / "rapp.yaml" if not declaration.is_file(): return _refuse(f"missing {declaration}") search = (family_root or dest.parent).resolve() if reef not in COMPUTE_REEFS and not (search / reef).is_dir(): return _refuse(f"reef {reef!r} is not a known compute reef") text = declaration.read_text() if re.search(r"^bound_reefs:\n - ", text, re.MULTILINE): text = re.sub( r"^bound_reefs:\n(?: - .+\n)+", f"bound_reefs:\n - {reef}\n", text, flags=re.MULTILINE, ) else: text = re.sub( r"^bound_reefs:\s*\[\]\s*$", f"bound_reefs:\n - {reef}", text, flags=re.MULTILINE, ) if "exposure:" in text and "posture: public" in text: return _refuse("place does not grant public exposure; edit exposure separately") declaration.write_text(text) return {"ok": True, "path": str(dest), "bound_reefs": [reef]} def validate(path: Path, *, family_root: Path | None = None) -> dict[str, Any]: dest = path.expanduser().resolve() declaration = dest / "declarations" / "rapp.yaml" if not declaration.is_file(): return _refuse(f"missing {declaration}") missing = [ name for name in ( "README.md", "INTENT.md", "SCOPE.md", "AGENTS.md", ".repo-classification.yaml", ) if not (dest / name).is_file() ] if missing: return _refuse(f"missing bootstrap files: {', '.join(missing)}") if _FAMILY_VALIDATOR.is_file(): import tempfile search_root = (family_root or dest.parent).resolve() with tempfile.TemporaryDirectory(prefix="rmgr-rapp-") as tmp: tmp_root = Path(tmp) (tmp_root / dest.name).symlink_to(dest) for sibling in ("rail-kubernetes", "rail-knative", "reef-railiance"): candidate = search_root / sibling if candidate.is_dir(): (tmp_root / sibling).symlink_to(candidate) proc = subprocess.run( ["python3", str(_FAMILY_VALIDATOR), "--root", str(tmp_root)], check=False, capture_output=True, text=True, ) return { "ok": proc.returncode == 0, "path": str(dest), "validator": str(_FAMILY_VALIDATOR), "stdout": proc.stdout.strip(), "stderr": proc.stderr.strip(), "exit_code": proc.returncode, } text = declaration.read_text() required = ( "kind: managed-workload-package", "repo_family: rapp", "ownership_repo:", "primary_rail:", "composition:", ) absent = [item for item in required if item not in text] if absent: return _refuse(f"declaration missing {absent}") return {"ok": True, "path": str(dest), "validator": None} def pin_image(path: Path, digest: str) -> dict[str, Any]: if not DIGEST.fullmatch(digest): return _refuse("digest must be sha256:<64 lowercase hex>") dest = path.expanduser().resolve() rewritten = 0 for rel in ("manifests", "bindings", "declarations"): root = dest / rel if not root.exists(): continue files = [root] if root.is_file() else list(root.rglob("*")) for file in files: if not file.is_file(): continue text = file.read_text() updated, n = IMAGE_PIN_RE.subn(rf"\g<1>{digest}", text) if n: file.write_text(updated) rewritten += n if rewritten == 0: return _refuse("no forgejo.coulomb.social digest pin found") makefile = dest / "Makefile" if makefile.is_file(): makefile.write_text( re.sub(r"sha256:[0-9a-f]{64}", digest, makefile.read_text(), count=1) ) return {"ok": True, "path": str(dest), "digest": digest, "rewritten": rewritten} def _runtime_manifest(app: str, facts: dict[str, Any]) -> str: port = facts["port"] uid = facts["uid"] health = facts["health_path"] ready = facts["ready_path"] image = f"{facts['image_repository']}@{facts['image_digest']}" return f"""apiVersion: v1 kind: Namespace metadata: name: {app} --- apiVersion: apps/v1 kind: Deployment metadata: name: {app} namespace: {app} labels: app.kubernetes.io/name: {app} spec: replicas: 1 selector: matchLabels: app.kubernetes.io/name: {app} template: metadata: labels: app.kubernetes.io/name: {app} spec: automountServiceAccountToken: false serviceAccountName: {app} securityContext: runAsNonRoot: true runAsUser: {uid} seccompProfile: type: RuntimeDefault containers: - name: {app} image: {image} imagePullPolicy: IfNotPresent ports: - name: http containerPort: {port} securityContext: allowPrivilegeEscalation: false capabilities: drop: ["ALL"] readOnlyRootFilesystem: true readinessProbe: httpGet: path: {ready} port: http livenessProbe: httpGet: path: {health} port: http --- apiVersion: v1 kind: Service metadata: name: {app} namespace: {app} spec: selector: app.kubernetes.io/name: {app} ports: - name: http port: {port} targetPort: http --- apiVersion: v1 kind: ServiceAccount metadata: name: {app} namespace: {app} automountServiceAccountToken: false --- apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: {app}-default-deny namespace: {app} spec: podSelector: {{}} policyTypes: [Ingress, Egress] --- apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: {app}-runtime namespace: {app} spec: podSelector: matchLabels: app.kubernetes.io/name: {app} policyTypes: [Ingress, Egress] ingress: - from: - namespaceSelector: matchLabels: kubernetes.io/metadata.name: kube-system ports: - {{protocol: TCP, port: {port}}} egress: - to: - namespaceSelector: matchLabels: kubernetes.io/metadata.name: kube-system ports: - {{protocol: UDP, port: 53}} - {{protocol: TCP, port: 53}} """ def _makefile(app: str, facts: dict[str, Any], field_manager: str) -> str: digest = facts["image_digest"] repo = facts["image_repository"] return f"""SHELL := /bin/bash TARGET ?= railiance01 NAMESPACE := {app} DEPLOYMENT := {app} IMAGE_REPOSITORY := {repo} IMAGE_DIGEST ?= {digest} DIGEST ?= RENDERED := .rendered MANIFESTS := $(wildcard manifests/*.yaml) .PHONY: check test render validate-inputs server-dry-run deploy status verify-live rollback check: test test: python3 -m unittest discover -s tests -v validate-inputs: @echo "$(IMAGE_DIGEST)" | grep -Eq '^sha256:[0-9a-f]{{64}}$$' \\ || (echo "IMAGE_DIGEST must be sha256:<64 hex>" >&2; exit 2) render: validate-inputs @rm -rf $(RENDERED) @mkdir -p $(RENDERED) IMAGE_REPOSITORY=$(IMAGE_REPOSITORY) IMAGE_DIGEST=$(IMAGE_DIGEST) \\ python3 tools/render.py $(MANIFESTS) --out $(RENDERED) server-dry-run: render {{ for f in $(RENDERED)/*.yaml; do echo '---'; cat "$$f"; done; }} \\ | ssh -o BatchMode=yes $(TARGET) \\ kubectl apply --server-side --force-conflicts --dry-run=server -f - deploy: render {{ for f in $(RENDERED)/*.yaml; do echo '---'; cat "$$f"; done; }} \\ | ssh -o BatchMode=yes $(TARGET) \\ kubectl apply --server-side --force-conflicts --field-manager={field_manager} -f - ssh -o BatchMode=yes $(TARGET) \\ kubectl -n $(NAMESPACE) rollout status deployment/$(DEPLOYMENT) --timeout=180s status: ssh -o BatchMode=yes $(TARGET) kubectl -n $(NAMESPACE) get deploy,pods,svc verify-live: validate-inputs EXPECTED_IMAGE="$(IMAGE_REPOSITORY)@$(IMAGE_DIGEST)" \\ TARGET=$(TARGET) NAMESPACE=$(NAMESPACE) ./tools/verify_live.sh rollback: @test -n "$(DIGEST)" || (echo "DIGEST=sha256:<64 hex> is required" >&2; exit 2) $(MAKE) deploy IMAGE_DIGEST=$(DIGEST) """ def _render_py(app: str) -> str: return f'''#!/usr/bin/env python3 """Rewrite digest pins into rendered manifests.""" from __future__ import annotations import argparse import os import re import sys from pathlib import Path IMAGE_LINE = re.compile( r"(image:\\s+)(forgejo\\.coulomb\\.social/coulomb/{app})(@sha256:[0-9a-f]{{64}})?" ) def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("manifests", nargs="+", type=Path) parser.add_argument("--out", required=True, type=Path) args = parser.parse_args() repo = os.environ.get("IMAGE_REPOSITORY", "forgejo.coulomb.social/coulomb/{app}") digest = os.environ.get("IMAGE_DIGEST", "") if not re.fullmatch(r"sha256:[0-9a-f]{{64}}", digest): print("IMAGE_DIGEST must be sha256:<64 lowercase hex>", file=sys.stderr) return 2 replacement = rf"\\1{{repo}}@{{digest}}" args.out.mkdir(parents=True, exist_ok=True) for src in args.manifests: (args.out / src.name).write_text(IMAGE_LINE.sub(replacement, src.read_text())) return 0 if __name__ == "__main__": raise SystemExit(main()) ''' def _verify_sh(app: str, facts: dict[str, Any]) -> str: port = facts["port"] health = facts["health_path"] ready = facts["ready_path"] return f"""#!/usr/bin/env bash set -euo pipefail TARGET="${{TARGET:-railiance01}}" NAMESPACE="${{NAMESPACE:-{app}}}" DEPLOYMENT="${{DEPLOYMENT:-{app}}}" EXPECTED_IMAGE="${{EXPECTED_IMAGE:?EXPECTED_IMAGE is required}}" remote() {{ ssh -o BatchMode=yes "$TARGET" "$1" }} live_image="$(remote "kubectl -n ${{NAMESPACE}} get deploy ${{DEPLOYMENT}} -o jsonpath='{{.spec.template.spec.containers[0].image}}'")" if [[ "$live_image" != "$EXPECTED_IMAGE" ]]; then echo "digest mismatch live=$live_image expected=$EXPECTED_IMAGE" >&2 exit 1 fi remote "kubectl -n ${{NAMESPACE}} rollout status deployment/${{DEPLOYMENT}} --timeout=60s >/dev/null" health="$(remote "kubectl -n ${{NAMESPACE}} exec deploy/${{DEPLOYMENT}} -- python3 -c \\"import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:{port}{health}').status)\\"")" ready="$(remote "kubectl -n ${{NAMESPACE}} exec deploy/${{DEPLOYMENT}} -- python3 -c \\"import urllib.request; print(urllib.request.urlopen('http://127.0.0.1:{port}{ready}').status)\\"")" [[ "$health" == "200" ]] [[ "$ready" == "200" ]] python3 - "$live_image" "$health" "$ready" <<'PY' import json, sys print(json.dumps({{ "health_ok": sys.argv[2] == "200", "ready_ok": sys.argv[3] == "200", "live_image": sys.argv[1], "secret_values_observed": False, }}, sort_keys=True)) PY """ def _packaging_test(app: str) -> str: return f'''import pathlib import re import unittest ROOT = pathlib.Path(__file__).resolve().parents[1] class PackagingTests(unittest.TestCase): def test_declaration_names_the_workload(self): decl = (ROOT / "declarations" / "rapp.yaml").read_text() self.assertIn("ownership_repo:", decl) self.assertIn("name: {app}", decl.split("workload_identity:", 1)[1][:200]) self.assertNotIn("posture: public", decl) def test_package_has_no_floating_tag(self): texts = [] for folder in ("manifests", "declarations"): root = ROOT / folder if not root.exists(): continue for path in root.rglob("*"): if path.is_file(): texts.append(path.read_text()) blob = "\\n".join(texts) self.assertNotRegex(blob, re.compile(r"image:\\s+[^\\n]+:(latest|main)\\b")) self.assertNotIn("sk-", blob) ''' def _image_workflow(app: str) -> str: return f"""name: Build and Publish Container Image # Images are built by CI from a tarball of the pushed commit, never from # a workstation working tree. on: push: branches: - main paths: - ".forgejo/workflows/image.yaml" - "Containerfile" - "src/**" - "pyproject.toml" - "README.md" - "LICENSE" workflow_dispatch: env: REGISTRY: forgejo.coulomb.social IMAGE_NAME: coulomb/{app} DOCKER_HOST: tcp://127.0.0.1:2375 jobs: build-and-push: runs-on: container-build steps: - name: Build and push image env: REGISTRY_USER: ${{{{ secrets.REGISTRY_USER }}}} REGISTRY_TOKEN: ${{{{ secrets.REGISTRY_TOKEN }}}} run: | set -eu REF="${{GITHUB_SHA:-main}}" SHORT="${{REF:0:7}}" mkdir -p buildctx "${{HOME}}/bin" wget -qO /tmp/repo.tar.gz \\ "https://forgejo.coulomb.social/${{GITHUB_REPOSITORY}}/archive/${{SHORT}}.tar.gz" tar xzf /tmp/repo.tar.gz -C buildctx --strip-components=1 wget -qO- https://download.docker.com/linux/static/stable/x86_64/docker-27.3.1.tgz \\ | tar xz --strip-components=1 -C "${{HOME}}/bin" docker/docker export PATH="${{HOME}}/bin:${{PATH}}" echo "${{REGISTRY_TOKEN}}" | docker login "${{REGISTRY}}" -u "${{REGISTRY_USER}}" --password-stdin IMAGE="${{REGISTRY}}/${{IMAGE_NAME}}" docker build -f buildctx/Containerfile -t "${{IMAGE}}:latest" -t "${{IMAGE}}:main-${{SHORT}}" buildctx docker push "${{IMAGE}}:latest" docker push "${{IMAGE}}:main-${{SHORT}}" echo "pushed ${{IMAGE}}:latest and ${{IMAGE}}:main-${{SHORT}}" - name: Report immutable digest run: | set -eu export PATH="${{HOME}}/bin:${{PATH}}" IMAGE="${{REGISTRY}}/${{IMAGE_NAME}}" SHORT="${{GITHUB_SHA:0:7}}" docker inspect --format='{{{{index .RepoDigests 0}}}}' "${{IMAGE}}:main-${{SHORT}}" """ def add_rapp_parser(sub: argparse._SubParsersAction) -> None: rapp = sub.add_parser("rapp", help="Scaffold or validate a rapp-* package") rapp_sub = rapp.add_subparsers(dest="rapp_command", required=True) p_init = rapp_sub.add_parser("init", help="Create a draft rapp-* baseline") p_init.add_argument("--path", required=True, help="Destination checkout") p_init.add_argument("--app", required=True, help="Workload name (no rapp- prefix)") p_init.add_argument("--ownership-repo", required=True) p_init.add_argument("--rail", default="rail-kubernetes", choices=RAILS) p_init.add_argument("--classification", default="confidential", choices=CLASSIFICATIONS) p_init.add_argument("--criticality", default="high", choices=CRITICALITIES) p_init.add_argument("--package-type", default="manifest-managed-platform-service", choices=PACKAGE_TYPES) p_init.add_argument("--purpose", default=None) p_init.add_argument("--force", action="store_true") p_skel = rapp_sub.add_parser("skeleton", help="Generate or absorb runtime manifests") p_skel.add_argument("--path", required=True) p_skel.add_argument("--from-app", required=True) p_skel.add_argument("--app", default=None) p_skel.add_argument("--package-type", default="manifest-managed-platform-service", choices=PACKAGE_TYPES) p_skel.add_argument("--force", action="store_true") p_skel.add_argument("--dedicated-postgres", action="store_true") p_wrap = rapp_sub.add_parser("wrap", help="init + skeleton + image + consumer + validate") p_wrap.add_argument("--path", required=True) p_wrap.add_argument("--app", required=True) p_wrap.add_argument("--ownership-repo", required=True) p_wrap.add_argument("--from-app", required=True) p_wrap.add_argument("--rail", default="rail-kubernetes", choices=RAILS) p_wrap.add_argument("--classification", default="confidential", choices=CLASSIFICATIONS) p_wrap.add_argument("--criticality", default="high", choices=CRITICALITIES) p_wrap.add_argument("--package-type", default="manifest-managed-platform-service", choices=PACKAGE_TYPES) p_wrap.add_argument("--purpose", default=None) p_wrap.add_argument("--family-root", default=None) p_wrap.add_argument("--force", action="store_true") p_wrap.add_argument("--dedicated-postgres", action="store_true") p_place = rapp_sub.add_parser("place", help="Set bound_reefs only") p_place.add_argument("--path", required=True) p_place.add_argument("--reef", default="reef-railiance") p_place.add_argument("--family-root", default=None) p_val = rapp_sub.add_parser("validate", help="Validate a rapp-* checkout") p_val.add_argument("--path", required=True) p_val.add_argument("--family-root", default=None) p_pin = rapp_sub.add_parser("pin-image", help="Rewrite digest pins") p_pin.add_argument("--path", required=True) p_pin.add_argument("--digest", required=True)