diff --git a/src/govoplan_ops/backend/infrastructure.py b/src/govoplan_ops/backend/infrastructure.py index 9458a88..87a7060 100644 --- a/src/govoplan_ops/backend/infrastructure.py +++ b/src/govoplan_ops/backend/infrastructure.py @@ -1,171 +1,18 @@ from __future__ import annotations -import json -import os from pathlib import Path -from typing import Any, Mapping - -MAX_CAPABILITY_DOCUMENT_BYTES = 256 * 1024 -CAPABILITY_STATES = { - "configured", - "available_unconfigured", - "externally_supplied", - "unavailable", -} +from govoplan_core.core.infrastructure_capabilities import ( + deployment_capability_status as _deployment_capability_status, +) def deployment_capability_status( path: Path | None = None, ) -> dict[str, object]: - configured_path = path - if configured_path is None: - raw_path = os.getenv("GOVOPLAN_DEPLOYMENT_CAPABILITIES_PATH", "").strip() - if not raw_path: - return _unavailable(configured=False, error=None) - configured_path = Path(raw_path) - try: - document = _read_document(configured_path) - return { - "configured": True, - "available": True, - "schema_version": 1, - "installation_id": document["installation_id"], - "profile": document["profile"], - "capabilities": document["capabilities"], - "post_install_tasks": document["post_install_tasks"], - "error": None, - } - except (OSError, ValueError, json.JSONDecodeError) as exc: - return _unavailable(configured=True, error=str(exc)) + """Project Core's validated non-secret deployment receipt into Ops.""" + + return _deployment_capability_status(path) -def _read_document(path: Path) -> dict[str, object]: - if path.is_symlink() or not path.is_file(): - raise ValueError("Deployment capability receipt is not a regular file.") - size = path.stat().st_size - if size > MAX_CAPABILITY_DOCUMENT_BYTES: - raise ValueError("Deployment capability receipt exceeds 256 KiB.") - raw = path.read_bytes() - if len(raw) != size: - raise ValueError("Deployment capability receipt changed while being read.") - payload = json.loads(raw.decode("utf-8")) - if not isinstance(payload, Mapping) or payload.get("schema_version") != 1: - raise ValueError("Deployment capability receipt has an unsupported schema.") - installation_id = _required_text(payload, "installation_id", maximum=100) - profile = _required_text(payload, "profile", maximum=100) - raw_capabilities = payload.get("capabilities") - if not isinstance(raw_capabilities, list) or len(raw_capabilities) > 100: - raise ValueError("Deployment capability receipt has invalid capabilities.") - capabilities = [_capability(item) for item in raw_capabilities] - ids = [str(item["id"]) for item in capabilities] - if len(ids) != len(set(ids)): - raise ValueError("Deployment capability receipt repeats a capability id.") - raw_tasks = payload.get("post_install_tasks", []) - if not isinstance(raw_tasks, list) or len(raw_tasks) > 100: - raise ValueError("Deployment capability receipt has invalid post-install tasks.") - tasks = [_task(item) for item in raw_tasks] - return { - "installation_id": installation_id, - "profile": profile, - "capabilities": capabilities, - "post_install_tasks": tasks, - } - - -def _capability(value: object) -> dict[str, object]: - if not isinstance(value, Mapping): - raise ValueError("Deployment capability entries must be objects.") - state = _required_text(value, "state", maximum=40) - if state not in CAPABILITY_STATES: - raise ValueError(f"Deployment capability state is unsupported: {state!r}.") - endpoint = value.get("endpoint", {}) - if not isinstance(endpoint, Mapping) or len(endpoint) > 10: - raise ValueError("Deployment capability endpoint metadata is invalid.") - normalized_endpoint: dict[str, object] = {} - for key, raw in endpoint.items(): - if not isinstance(key, str) or len(key) > 50: - raise ValueError("Deployment capability endpoint key is invalid.") - if isinstance(raw, bool) or raw is None: - normalized_endpoint[key] = raw - elif isinstance(raw, int): - normalized_endpoint[key] = raw - elif isinstance(raw, str) and len(raw) <= 500: - normalized_endpoint[key] = raw - else: - raise ValueError("Deployment capability endpoint value is invalid.") - secret_refs = _string_list(value.get("secret_refs"), maximum_items=30) - if any(not item.startswith("env:") for item in secret_refs): - raise ValueError("Deployment capability secrets must use environment references.") - return { - "id": _required_text(value, "id", maximum=120), - "label": _required_text(value, "label", maximum=200), - "state": state, - "source": _required_text(value, "source", maximum=120), - "detail": _required_text(value, "detail", maximum=1000), - "endpoint": normalized_endpoint, - "secret_refs": secret_refs, - "dependent_modules": _string_list( - value.get("dependent_modules"), - maximum_items=100, - ), - } - - -def _task(value: object) -> dict[str, object]: - if not isinstance(value, Mapping): - raise ValueError("Deployment post-install task entries must be objects.") - return { - "id": _required_text(value, "id", maximum=120), - "resume_key": _required_text(value, "resume_key", maximum=240), - "capability_id": _required_text(value, "capability_id", maximum=120), - "state": _required_text(value, "state", maximum=40), - "owner_module": _required_text(value, "owner_module", maximum=120), - "summary": _required_text(value, "summary", maximum=1000), - "required_inputs": _string_list( - value.get("required_inputs"), - maximum_items=30, - ), - "secret_boundary": _required_text( - value, - "secret_boundary", - maximum=120, - ), - } - - -def _required_text( - value: Mapping[str, Any], - key: str, - *, - maximum: int, -) -> str: - raw = value.get(key) - text = str(raw).strip() if raw is not None else "" - if not text or len(text) > maximum: - raise ValueError(f"Deployment capability field {key!r} is invalid.") - return text - - -def _string_list(value: object, *, maximum_items: int) -> list[str]: - if not isinstance(value, list) or len(value) > maximum_items: - raise ValueError("Deployment capability list field is invalid.") - result: list[str] = [] - for item in value: - if not isinstance(item, str) or not item.strip() or len(item) > 500: - raise ValueError("Deployment capability list item is invalid.") - result.append(item.strip()) - return result - - -def _unavailable(*, configured: bool, error: str | None) -> dict[str, object]: - return { - "configured": configured, - "available": False, - "schema_version": None, - "installation_id": None, - "profile": None, - "capabilities": [], - "post_install_tasks": [], - "error": error, - } +__all__ = ["deployment_capability_status"]