from __future__ import annotations import re import time from datetime import datetime from typing import Any from urllib.parse import quote, urlsplit, urlunsplit import httpx from ..connectors.portainer import PortainerConnector from ..core.config import settings from .controller_artifact_helper_spec import ( _IMAGE_ID, _REVISION, CONTROLLER_VERSION, HELPER_LABEL, HELPER_NAME, helper_container_spec, ) from .controller_artifact_helper_spec import ( SOURCE_REPOSITORY as SOURCE_REPOSITORY, ) from .deployment_broker import DeploymentBrokerClient GITEA_OWNER = "Jens" GITEA_REPOSITORY = "Unraid-AppOps-Gateway" REQUIRED_PUSH_CONTEXT = "Managed validation / full (push)" CONTROLLER_CONTAINER = "gitea-deploy-control" _HELPER_TIMEOUT_SECONDS = 240 _TIMESTAMP = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z$") _MAX_HELPER_LOG_CHARS = 4000 _MAX_HELPER_LOG_LINES = 200 _MAX_HELPER_LOG_WIRE_BYTES = _MAX_HELPER_LOG_CHARS + (8 * _MAX_HELPER_LOG_LINES) _HELPER_DIAGNOSTIC_PREFIX = "APPOPS_CONTROLLER_PROMOTION_DIAGNOSTIC" _HELPER_DIAGNOSTIC_CODES = frozenset({"helper_failed"}) _HELPER_DIAGNOSTIC_PHASES = frozenset( { "shell_setup", "tmp_root", "workspace_create", "workspace_validate", "tool_git", "tool_docker", "tool_compose", "tool_python", "tool_tar", "tool_curl", "source_commit", "source_tree", "source_archive", "archive_permissions", "archive_digest", "upgrade_script", "upgrade_permissions", "upgrade", "cleanup", } ) _HELPER_DIAGNOSTIC_LINE = re.compile( rf"^{_HELPER_DIAGNOSTIC_PREFIX} code=([a-z_]+) phase=([a-z_]+) " r"exitCode=(0|[1-9][0-9]{0,2})$" ) def helper_failure_diagnostics(exit_code: Any) -> dict[str, Any]: safe_exit_code = exit_code if type(exit_code) is int and 0 <= exit_code <= 255 else None return {"code": "helper_failed", "phase": "unknown", "exitCode": safe_exit_code} def safe_helper_diagnostics(value: Any) -> dict[str, Any] | None: if not isinstance(value, dict) or set(value) != {"code", "phase", "exitCode"}: return None code = value.get("code") phase = value.get("phase") exit_code = value.get("exitCode") if not isinstance(code, str) or code not in _HELPER_DIAGNOSTIC_CODES: return None if not isinstance(phase, str) or ( phase != "unknown" and phase not in _HELPER_DIAGNOSTIC_PHASES ): return None if exit_code is not None and ( type(exit_code) is not int or not 0 <= exit_code <= 255 ): return None return {"code": code, "phase": phase, "exitCode": exit_code} def parse_helper_diagnostics(log_text: Any, exit_code: Any) -> dict[str, Any]: fallback = helper_failure_diagnostics(exit_code) if ( fallback["exitCode"] is None or not isinstance(log_text, str) or len(log_text) > _MAX_HELPER_LOG_CHARS ): return fallback candidates = [ line for line in log_text.splitlines() if line.startswith(_HELPER_DIAGNOSTIC_PREFIX) ] if len(candidates) != 1: return fallback match = _HELPER_DIAGNOSTIC_LINE.fullmatch(candidates[0]) if match is None: return fallback code, phase, marker_exit_code = match.groups() parsed_exit_code = int(marker_exit_code) if ( code not in _HELPER_DIAGNOSTIC_CODES or phase not in _HELPER_DIAGNOSTIC_PHASES or parsed_exit_code != fallback["exitCode"] ): return fallback return {"code": code, "phase": phase, "exitCode": parsed_exit_code} def _decode_docker_log_stream(content: Any) -> str | None: if ( not isinstance(content, bytes) or not content or len(content) > _MAX_HELPER_LOG_WIRE_BYTES ): return None offset = 0 frame_count = 0 payload_size = 0 payloads: list[bytes] = [] while offset < len(content): if frame_count >= _MAX_HELPER_LOG_LINES or len(content) - offset < 8: return None header = content[offset : offset + 8] stream_type = header[0] if stream_type not in {1, 2} or header[1:4] != b"\x00\x00\x00": return None frame_size = int.from_bytes(header[4:8], "big") if frame_size == 0 or frame_size > _MAX_HELPER_LOG_CHARS - payload_size: return None payload_start = offset + 8 payload_end = payload_start + frame_size if payload_end > len(content): return None payloads.append(content[payload_start:payload_end]) payload_size += frame_size frame_count += 1 offset = payload_end try: return b"".join(payloads).decode("utf-8") except UnicodeDecodeError: return None class ControllerArtifactPromotionError(RuntimeError): pass class ControllerArtifactPromoter: def __init__( self, *, portainer: PortainerConnector | None = None, broker: DeploymentBrokerClient | None = None, ) -> None: self.portainer = portainer or PortainerConnector() self.broker = broker or DeploymentBrokerClient() @staticmethod def _gitea_base_url() -> str: raw = str(settings.apis.get("GITEA_URL") or "").strip().rstrip("/") parsed = urlsplit(raw) if parsed.scheme not in {"http", "https"} or not parsed.netloc or parsed.username: raise ControllerArtifactPromotionError("controller_promoter_gitea_url_invalid") return urlunsplit((parsed.scheme, parsed.netloc, parsed.path.rstrip("/"), "", "")) @staticmethod def _gitea_token() -> str: token = str(settings.apis.get("GITEA_ADMIN_TOKEN") or "").strip() if not token: raise ControllerArtifactPromotionError("controller_promoter_gitea_token_missing") return token @staticmethod def _status_key(item: dict[str, Any]) -> tuple[datetime, int] | None: raw = str(item.get("updated_at") or item.get("created_at") or "") if not _TIMESTAMP.fullmatch(raw): return None try: parsed = datetime.fromisoformat(raw.replace("Z", "+00:00")) except ValueError: return None identifier = item.get("id") return parsed, identifier if isinstance(identifier, int) and not isinstance(identifier, bool) else -1 def _validated_main_revision(self) -> str: branch = "main" base = self._gitea_base_url() headers = { "Authorization": f"token {self._gitea_token()}", "Accept": "application/json", "User-Agent": "itworx-controller-artifact-promoter/1", } timeout = max(2.0, min(20.0, float(settings.apis.get("GITEA_TIMEOUT_SECONDS") or 8))) verify = bool(settings.apis.get("GITEA_VERIFY_TLS", settings.verify_tls)) owner = quote(GITEA_OWNER, safe="") repository = quote(GITEA_REPOSITORY, safe="") try: with httpx.Client(timeout=timeout, verify=verify, follow_redirects=False) as client: branch_response = client.get( f"{base}/api/v1/repos/{owner}/{repository}/branches/{branch}", headers=headers, ) if branch_response.status_code != 200: raise ControllerArtifactPromotionError("controller_promoter_main_unavailable") branch_payload = branch_response.json() commit = branch_payload.get("commit") if isinstance(branch_payload, dict) else None revision = str(commit.get("id") or "") if isinstance(commit, dict) else "" if ( not isinstance(branch_payload, dict) or branch_payload.get("name") != branch or branch_payload.get("protected") is not True or not _REVISION.fullmatch(revision) ): raise ControllerArtifactPromotionError("controller_promoter_main_not_protected") status_response = client.get( f"{base}/api/v1/repos/{owner}/{repository}/statuses/{revision}", params={"limit": 100, "page": 1}, headers=headers, ) if status_response.status_code != 200: raise ControllerArtifactPromotionError("controller_promoter_status_unavailable") statuses = status_response.json() except ControllerArtifactPromotionError: raise except (httpx.RequestError, ValueError) as exc: raise ControllerArtifactPromotionError( "controller_promoter_gitea_unavailable" ) from exc if not isinstance(statuses, list) or len(statuses) > 100: raise ControllerArtifactPromotionError("controller_promoter_status_invalid") latest: tuple[tuple[datetime, int], str] | None = None for item in statuses: if not isinstance(item, dict) or item.get("context") != REQUIRED_PUSH_CONTEXT: continue state = str(item.get("status") or "") if state not in {"pending", "success", "failure", "error", "warning", "neutral", "cancelled"}: continue key = self._status_key(item) if key is not None and (latest is None or key > latest[0]): latest = (key, state) if latest is None or latest[1] != "success": raise ControllerArtifactPromotionError("controller_promoter_main_not_validated") return revision def _portainer_endpoint(self) -> str: config = self.portainer._config() if config is None or not config.endpoint_id: raise ControllerArtifactPromotionError("controller_promoter_portainer_unavailable") return config.endpoint_id def _controller_image_id(self) -> str: listed = self.portainer.list_containers() containers = listed.get("containers") if isinstance(listed, dict) else None candidate = next( ( item for item in containers or [] if isinstance(item, dict) and item.get("name") == CONTROLLER_CONTAINER ), None, ) if candidate is None or candidate.get("state") != "running": raise ControllerArtifactPromotionError("controller_promoter_controller_unavailable") identifier = str(candidate.get("id") or "") endpoint = self._portainer_endpoint() try: response = self.portainer._request( "GET", f"/api/endpoints/{endpoint}/docker/containers/{quote(identifier, safe='')}/json", ) payload = response.json() except (RuntimeError, ValueError) as exc: raise ControllerArtifactPromotionError( "controller_promoter_controller_inspect_failed" ) from exc image_id = str(payload.get("Image") or "") if isinstance(payload, dict) else "" if not _IMAGE_ID.fullmatch(image_id): raise ControllerArtifactPromotionError("controller_promoter_controller_image_unpinned") return image_id def _existing_helper(self) -> dict[str, Any] | None: listed = self.portainer.list_containers() containers = listed.get("containers") if isinstance(listed, dict) else None return next( ( item for item in containers or [] if isinstance(item, dict) and item.get("name") == HELPER_NAME ), None, ) def _remove_helper(self, identifier: str) -> None: endpoint = self._portainer_endpoint() self.portainer._request( "DELETE", f"/api/endpoints/{endpoint}/docker/containers/{quote(identifier, safe='')}", params={"v": "1", "force": "0"}, ) def _run_helper(self, *, revision: str, image_id: str) -> dict[str, Any]: endpoint = self._portainer_endpoint() existing = self._existing_helper() if existing is not None: identifier = str(existing.get("id") or "") if existing.get("state") == "running": raise ControllerArtifactPromotionError("controller_promoter_already_running") if identifier: try: inspect = self.portainer._request( "GET", f"/api/endpoints/{endpoint}/docker/containers/{quote(identifier, safe='')}/json", ).json() except (RuntimeError, ValueError) as exc: raise ControllerArtifactPromotionError( "controller_promoter_helper_inspect_failed" ) from exc labels = ((inspect.get("Config") or {}).get("Labels") or {}) if isinstance(inspect, dict) else {} if not isinstance(labels, dict) or labels.get(HELPER_LABEL) != "true": raise ControllerArtifactPromotionError("controller_promoter_name_conflict") self._remove_helper(identifier) spec = helper_container_spec(revision=revision, image_id=image_id) try: created_response = self.portainer._request( "POST", f"/api/endpoints/{endpoint}/docker/containers/create", params={"name": HELPER_NAME}, json=spec, ) created = created_response.json() helper_id = str(created.get("Id") or "") if isinstance(created, dict) else "" if not helper_id: raise ControllerArtifactPromotionError("controller_promoter_create_failed") self.portainer._request( "POST", f"/api/endpoints/{endpoint}/docker/containers/{quote(helper_id, safe='')}/start", ) except ControllerArtifactPromotionError: raise except (RuntimeError, ValueError) as exc: raise ControllerArtifactPromotionError("controller_promoter_start_failed") from exc deadline = time.monotonic() + _HELPER_TIMEOUT_SECONDS state: dict[str, Any] = {} while time.monotonic() < deadline: try: payload = self.portainer._request( "GET", f"/api/endpoints/{endpoint}/docker/containers/{quote(helper_id, safe='')}/json", ).json() except (RuntimeError, ValueError) as exc: raise ControllerArtifactPromotionError( "controller_promoter_helper_inspect_failed" ) from exc state = payload.get("State") if isinstance(payload, dict) and isinstance(payload.get("State"), dict) else {} if state.get("Running") is False: break time.sleep(2) else: return { "status": "outcome_unknown", "revision": revision, "reason": "controller_promoter_timeout", "helperId": helper_id, } exit_code = state.get("ExitCode") if type(exit_code) is not int or exit_code != 0: try: logs_response = self.portainer._request( "GET", f"/api/endpoints/{endpoint}/docker/containers/{quote(helper_id, safe='')}/logs", params={ "stdout": 1, "stderr": 1, "tail": _MAX_HELPER_LOG_LINES, }, ) log_text = ( _decode_docker_log_stream(logs_response.content) if isinstance(logs_response, httpx.Response) and logs_response.status_code == 200 else None ) except (httpx.HTTPError, RuntimeError, ValueError): log_text = None diagnostics = parse_helper_diagnostics( log_text, exit_code, ) try: self._remove_helper(helper_id) except RuntimeError: pass return { "status": "failed", "revision": revision, "reason": "controller_promoter_helper_failed", "diagnostics": diagnostics, } try: self._remove_helper(helper_id) except RuntimeError: pass return {"status": "promoted", "revision": revision} def run(self) -> dict[str, Any]: revision = self._validated_main_revision() current = self.broker.controller_health() artifact = current.get("artifact") if isinstance(current, dict) else None if ( isinstance(artifact, dict) and artifact.get("sourceRevision") == revision and artifact.get("controllerVersion") == CONTROLLER_VERSION ): return { "status": "already_applied", "revision": revision, "controllerVersion": CONTROLLER_VERSION, } result = self._run_helper(revision=revision, image_id=self._controller_image_id()) if result.get("status") != "promoted": return result verified = self.broker.controller_health() artifact = verified.get("artifact") if isinstance(verified, dict) else None if ( not isinstance(artifact, dict) or artifact.get("sourceRevision") != revision or artifact.get("controllerVersion") != CONTROLLER_VERSION ): return { "status": "outcome_unknown", "revision": revision, "reason": "controller_promoter_postcheck_failed", } return { "status": "promoted", "revision": revision, "controllerVersion": CONTROLLER_VERSION, "artifactSha256": artifact.get("artifactSha256"), }