haproxy_agent.py

"""An agent for HaProxy that takes care of most of the authentication logic of AppAPI. Python 3.12 required."""

# SPDX-FileCopyrightText: 2025 Nextcloud GmbH and Nextcloud contributors
# SPDX-License-Identifier: AGPL-3.0-or-later

import asyncio
import collections
import hashlib
import hmac
import io
import ipaddress
import json
import logging
import os
import posixpath
import re
import socket
import ssl
import tarfile
import time
from base64 import b64encode
from enum import IntEnum
from ipaddress import IPv4Address, IPv6Address, ip_address
from typing import Any, Literal, Self
from urllib.parse import quote, unquote

import aiohttp
from aiohttp import web
from haproxyspoa.payloads.ack import AckPayload
from haproxyspoa.spoa_server import SpoaServer
from pydantic import (
    BaseModel,
    Field,
    ValidationError,
    computed_field,
    field_validator,
    model_validator,
)

# Reported by `/info`; AppAPI compares it against its minimum supported HaRP version.
# Keep it in sync with the release tag and use a full `x.y.z` string: a bare float like `0.4`
# is read as "0.4.0", which made every release look older than it is.
HARP_VERSION = "0.4.7"

APPID_PATTERN = re.compile(r"(?:^|/)exapps/([^/]+)")
# Matches `.` and `..` path segments in any literal/encoded form: separators can be `/`, %2F or `\` (a slash for WHATWG
# URL parsers), the dots `.` or %2E (any case), and the segment may go on with `;` or %3B (Tomcat drops parameters),
# or end at %3F or %23 (a wrapper ExApp that decodes the path cuts it there), as in AppAPI's PHP proxy.
DOT_SEGMENT_PATTERN = re.compile(r"(?:^|[/\\]|%2[fF])(?:\.|%2[eE]){1,2}(?:[/;\\]|%2[fF]|%3[bBfF]|%23|$)")
# An escaped RFC 3986 unreserved character (letters, digits, `-._~`): it means the same escaped or not.
UNRESERVED_ESCAPE_PATTERN = re.compile(r"%(?:[46][1-9A-Fa-f]|[57][0-9Aa]|3[0-9]|2[DdEe]|5[Ff]|7[Ee])")
LONE_PERCENT_PATTERN = re.compile(r"%(?![0-9A-Fa-f]{2})")
# The reply to HAProxy carries the forwarded path and must fit one SPOE frame (16380 bytes).
MAX_FORWARDED_PATH_LENGTH = 14 * 1024
CONTROL_CHARACTER_PATTERN = re.compile(r"[\x00-\x1f\x7f]")
LIFECYCLE_ENDPOINTS = ("heartbeat", "init", "enabled")
SHARED_KEY = os.environ.get("HP_SHARED_KEY")
# ExApp containers get this token instead of the shared key (see `with_exapp_frp_token`). It is accepted only for
# the FRP login, so an ExApp cannot use the control API. Deployed containers carry it: the derivation must not change.
EXAPP_FRP_TOKEN = (
    hmac.new(SHARED_KEY.encode("utf-8", "surrogatepass"), b"harp-frp-exapp-v1", hashlib.sha256).hexdigest()
    if SHARED_KEY
    else ""
)
NC_INSTANCE_URL = os.environ.get("NC_INSTANCE_URL")
SPOA_ADDRESS = os.environ.get("HP_SPOA_ADDRESS", "127.0.0.1:9600")
SPOA_HOST, SPOA_PORT = SPOA_ADDRESS.rsplit(":", 1)
SPOA_PORT = int(SPOA_PORT)
# Kubernetes environment variables
K8S_ENABLED = os.environ.get("HP_K8S_ENABLED", "false").lower() in {"1", "true", "yes"}
K8S_NAMESPACE = os.environ.get("HP_K8S_NAMESPACE", "nextcloud-exapps")
K8S_API_SERVER = os.environ.get("HP_K8S_API_SERVER")  # e.g. https://kubernetes.default.svc
K8S_CA_FILE = os.environ.get("HP_K8S_CA_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt")
K8S_TOKEN_ENV = os.environ.get("HP_K8S_BEARER_TOKEN")  # static token from env (never rotated)
K8S_TOKEN_FILE = os.environ.get("HP_K8S_BEARER_TOKEN_FILE", "/var/run/secrets/kubernetes.io/serviceaccount/token")
K8S_VERIFY_SSL = os.environ.get("HP_K8S_VERIFY_SSL", "true").lower() != "false"
K8S_STORAGE_CLASS = os.environ.get("HP_K8S_STORAGE_CLASS", "")
K8S_DEFAULT_STORAGE_SIZE = os.environ.get("HP_K8S_DEFAULT_STORAGE_SIZE", "10Gi")
K8S_HOST_ALIASES_RAW = os.environ.get("HP_K8S_HOST_ALIASES", "")  # "hostname:ip,hostname2:ip2"


def _in_cluster_api_server() -> str:
    """The URL of the API server whose address Kubernetes sets in every pod; an IPv6 address needs brackets."""
    host = os.environ["KUBERNETES_SERVICE_HOST"]
    port = os.environ.get("KUBERNETES_SERVICE_PORT", "443")
    return f"https://[{host}]:{port}" if ":" in host else f"https://{host}:{port}"


if not K8S_API_SERVER and os.environ.get("KUBERNETES_SERVICE_HOST"):
    K8S_API_SERVER = _in_cluster_api_server()

K8S_HTTP_TIMEOUT = aiohttp.ClientTimeout(total=60.0)
# `/info` probes the API server with its own short timeout: AppAPI waits 5 s for `/info` in its daemon checks, so a
# probe that inherits K8S_HTTP_TIMEOUT makes an unreachable API server look like an unreachable HaRP.
K8S_PROBE_TIMEOUT = aiohttp.ClientTimeout(total=3.0)
_k8s_session: aiohttp.ClientSession | None = None
K8S_NAME_MAX_LENGTH = 63
# Set up the logging configuration
LOG_LEVEL = os.environ.get("HP_LOG_LEVEL", "INFO").upper()
logging.basicConfig(level=LOG_LEVEL)
LOGGER = logging.getLogger(__name__)
LOGGER.setLevel(level=LOG_LEVEL)
logging.getLogger("haproxyspoa").setLevel(level=LOG_LEVEL)
logging.getLogger("aiohttp").setLevel(level=LOG_LEVEL)

NC_REQ_URL = NC_INSTANCE_URL.removesuffix("/").removesuffix("/index.php")
EX_APP_URL = f"{NC_REQ_URL}/index.php/apps/app_api/harp/exapp-meta"
USER_INFO_URL = f"{NC_REQ_URL}/index.php/apps/app_api/harp/user-info"
# Client headers the user-info lookup leaves out, as it does every content-* header
EXCLUDE_HEADERS_USER_INFO = {
    "accept-encoding",
    "connection",
    "expect",
    "host",
    "keep-alive",
    "proxy-authenticate",
    "proxy-authorization",
    "proxy-connection",
    "range",
    "te",
    "trailer",
    "transfer-encoding",
    "upgrade",
}
# aiohttp refuses to send a header that contains one of these
UNSENDABLE_HEADER_CHARS = re.compile(r"[\x00-\x08\x0a-\x1f\x7f]")

# Keep total below the SPOE `timeout processing` (30s); aiohttp's default is 5 minutes.
NC_HTTP_TIMEOUT = aiohttp.ClientTimeout(total=25.0, connect=5.0)
_nc_session: aiohttp.ClientSession | None = None

SPOA_AGENT = SpoaServer()
DOCKER_API_HOST = "127.0.0.1"
# The ports the FRP server accepts Docker Engine tunnels on (see `allowPorts` in start.sh); HP_DOCKER_ENGINE_PORTS
# can add others for setups where HaRP shares its network namespace with the Docker Engine or a socket proxy.
DEFAULT_DOCKER_ENGINE_PORTS = "24000-24099"
# Headers whose values are never written to the log.
REDACTED_LOG_HEADERS = frozenset({"harp-shared-key", "authorization-app-api", "authorization", "cookie"})


def _parse_trusted_proxies(value: str) -> list[ipaddress.IPv4Network | ipaddress.IPv6Network]:
    if not value:
        return []
    networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
    invalid: list[str] = []
    masked: list[tuple[str, ipaddress.IPv4Network | ipaddress.IPv6Network]] = []
    dequoted: list[tuple[str, str]] = []
    for raw_item in value.split(","):
        # Quotes reach us literally when the variable is set via an env-file or a compose
        # `environment:` list entry, so strip them before parsing.
        item = raw_item.strip()
        stripped = item.strip("'\"\u201c\u201d\u2018\u2019").strip()
        if not stripped:
            continue
        try:
            networks.append(ipaddress.ip_network(stripped))
            if stripped != item:
                dequoted.append((item, stripped))
            continue
        except ValueError:
            pass
        try:
            network = ipaddress.ip_network(stripped, strict=False)
        except ValueError:
            invalid.append(item)
            continue
        if stripped != item:
            dequoted.append((item, stripped))
        masked.append((stripped, network))
        networks.append(network)
    # The rescues below run before the trust summary; at the default "warning" log level they are
    # the only boot-time signal that a previously rejected value now enables header-based detection.
    for item, stripped in dequoted[:5]:
        LOGGER.warning("HP_TRUSTED_PROXY_IPS: quotes stripped from %r, using %r.", item, stripped)
    if len(dequoted) > 5:
        LOGGER.warning("HP_TRUSTED_PROXY_IPS: quotes stripped from %d more entries.", len(dequoted) - 5)
    # Same leniency as Nextcloud's own `trusted_proxies` handling: mask the host bits.
    for item, network in masked[:5]:
        LOGGER.warning("HP_TRUSTED_PROXY_IPS: interpreting %r as %r.", item, str(network))
    if len(masked) > 5:
        LOGGER.warning("HP_TRUSTED_PROXY_IPS: %d more entries with host bits set were masked.", len(masked) - 5)
    for bad in invalid[:5]:
        LOGGER.error("Invalid entry in HP_TRUSTED_PROXY_IPS ignored: %r is not an IPv4 or IPv6 network.", bad)
    if len(invalid) > 5:
        LOGGER.error("HP_TRUSTED_PROXY_IPS: %d more invalid entries ignored.", len(invalid) - 5)
    if networks:
        LOGGER.info("Trusting reverse proxies for client IP detection: %s", [str(p) for p in networks])
    else:
        LOGGER.error(
            "No valid entries in HP_TRUSTED_PROXY_IPS (%r). Client IP detection from headers is disabled. "
            "The X-Forwarded-For and X-Real-IP headers will not be respected. "
            "This can lead to the outer proxy's IP being blocked "
            "during a bruteforce attempt instead of the actual client's IP.",
            value if len(value) <= 200 else value[:200] + "...",
        )
    return networks


TRUSTED_PROXIES = _parse_trusted_proxies(os.environ.get("HP_TRUSTED_PROXY_IPS", ""))

# The entries of the default list after the hard core only log a warning in this release (see
# HP_EXAPP_BIND_MOUNTS_ENFORCE_ALL): AppAPI re-sends the mounts stored at install on every ExApp update, after it
# has removed the old container, so refusing a mount that used to work would leave the ExApp without a container.
_STAGED_BIND_MOUNTS_DENIED = (
    "/dev,/run,/var/run,/root,/var/lib/docker/volumes,/var/lib/containers/storage/volumes,"
    "/etc:rw,/usr:rw,/bin:rw,/sbin:rw,/lib:rw,/lib64:rw,/var/spool:rw"
)
# Host paths ExApp containers may not bind mount unless HP_EXAPP_BIND_MOUNTS_DENIED says otherwise: the hard core
# (kernel interfaces, the container runtimes' sockets and state, host credentials), which no ExApp has a reason to
# mount; then device and runtime trees, root's home, the Docker and Podman volume directories, and read-write access
# to the host's programs and spool directories.
DEFAULT_BIND_MOUNTS_DENIED = (
    "/proc,/sys,/boot,/var/lib/docker,/var/lib/containerd,/var/lib/containers,"
    "/run/docker.sock,/var/run/docker.sock,/run/containerd,/var/run/containerd,"
    "/run/podman,/var/run/podman,/run/crio,/var/run/crio,"
    "/etc/shadow,/etc/gshadow,/etc/sudoers,/etc/sudoers.d,/etc/ssh," + _STAGED_BIND_MOUNTS_DENIED
)
DEFAULT_BIND_MOUNTS_STAGED = frozenset(_STAGED_BIND_MOUNTS_DENIED.split(","))


def _has_control_characters(value: str) -> bool:
    return any(ord(char) < 32 or ord(char) == 127 for char in value)


def _setting_items(value: str) -> list[str]:
    """Split a comma-separated setting; quotes reach us literally from an env-file or a compose list entry."""
    return [item for raw in value.split(",") if (item := raw.strip().strip("'\"").strip())]


def _normalize_mount_path(value: str) -> str:
    """Return the normalized form of an absolute path; raise ValueError for anything else."""
    if not value.startswith("/"):
        raise ValueError("must be an absolute path")
    if ":" in value:
        raise ValueError("must not contain ':'")
    if _has_control_characters(value):
        raise ValueError("must not contain control characters")
    if ".." in value.split("/"):
        raise ValueError("must not contain '..' segments")
    return posixpath.normpath("/" + value.lstrip("/"))


def _parse_mount_rules(env_name: str, value: str, rw_suffix: bool = False) -> list[tuple[str, bool]]:
    """Parse a comma-separated list of host paths into (path, read-write only) rules.

    A malformed entry stops the agent: silently dropping it would change which mounts are permitted.
    """
    rules: list[tuple[str, bool]] = []
    for item in _setting_items(value):
        rw_only = rw_suffix and item.endswith(":rw")
        if rw_only:
            item = item.removesuffix(":rw").strip().strip("'\"").strip()
        words = item.split()
        if len(words) > 1 and all(word.startswith("/") for word in words):
            raise SystemExit(f"ERROR: invalid entry {item!r} in {env_name}: separate the paths with commas.")
        try:
            rules.append((_normalize_mount_path(item), rw_only))
        except ValueError as e:
            raise SystemExit(f"ERROR: invalid entry {item!r} in {env_name}: {e}.") from None
    return rules


def _parse_bool(env_name: str) -> bool:
    """Read a true/false setting that defaults to false; anything else stops the agent instead of meaning false."""
    value = os.environ.get(env_name, "")
    item = value.strip().strip("'\"").strip().lower()
    if item in {"", "0", "false", "no"}:
        return False
    if item in {"1", "true", "yes"}:
        return True
    raise SystemExit(f"ERROR: invalid value {value!r} in {env_name}: expected true or false.")


def _rule_label(path: str, rw_only: bool) -> str:
    return f"{path}:rw" if rw_only else path


BIND_MOUNTS_DISABLED = _parse_bool("HP_EXAPP_BIND_MOUNTS_DISABLED")
BIND_MOUNTS_ENFORCE_ALL = _parse_bool("HP_EXAPP_BIND_MOUNTS_ENFORCE_ALL")
BIND_MOUNTS_ALLOWED = [
    path
    for path, _ in _parse_mount_rules(
        "HP_EXAPP_BIND_MOUNTS_ALLOWED", os.environ.get("HP_EXAPP_BIND_MOUNTS_ALLOWED", "")
    )
]
_denied_setting = os.environ.get("HP_EXAPP_BIND_MOUNTS_DENIED")
BIND_MOUNTS_DENIED = _parse_mount_rules(
    "HP_EXAPP_BIND_MOUNTS_DENIED",
    DEFAULT_BIND_MOUNTS_DENIED if _denied_setting is None else _denied_setting,
    rw_suffix=True,
)
# Rules that only log a warning: the staged entries of the default list. A list the administrator sets is
# always enforced as written.
BIND_MOUNTS_WARN_ONLY = (
    frozenset() if _denied_setting is not None or BIND_MOUNTS_ENFORCE_ALL else DEFAULT_BIND_MOUNTS_STAGED
)
if _denied_setting is not None:
    if BIND_MOUNTS_DENIED:
        LOGGER.warning("HP_EXAPP_BIND_MOUNTS_DENIED replaces the default list of denied host paths.")
    else:
        LOGGER.warning("HP_EXAPP_BIND_MOUNTS_DENIED is empty: no host path is denied to ExApp containers.")
if BIND_MOUNTS_DISABLED:
    LOGGER.info("Bind mounts for ExApp containers are disabled.")
else:
    LOGGER.info(
        "Bind mounts for ExApp containers: allowed below %s, denied %s%s.",
        BIND_MOUNTS_ALLOWED or "any path",
        [_rule_label(path, rw_only) for path, rw_only in BIND_MOUNTS_DENIED] or "nothing",
        f" (only a warning for {sorted(BIND_MOUNTS_WARN_ONLY)})" if BIND_MOUNTS_WARN_ONLY else "",
    )


def _parse_port_ranges(env_name: str, value: str) -> list[tuple[int, int]]:
    """Parse comma-separated ports ("2375") and ranges ("24000-24099"); a malformed entry stops the agent."""
    ranges: list[tuple[int, int]] = []
    for item in _setting_items(value):
        first, dash, last = item.partition("-")
        try:
            first_port = int(first)
            last_port = int(last) if dash else first_port
        except ValueError:
            raise SystemExit(
                f"ERROR: invalid entry {item!r} in {env_name}: expected a port or a range like 24000-24099."
            ) from None
        if not 1 <= first_port <= last_port <= 65535:
            raise SystemExit(f"ERROR: invalid entry {item!r} in {env_name}: ports must be 1-65535, ranges ascending.")
        ranges.append((first_port, last_port))
    return ranges


# An empty value (e.g. a template passing an unset variable) means the default.
DOCKER_ENGINE_PORTS = _parse_port_ranges(
    "HP_DOCKER_ENGINE_PORTS", os.environ.get("HP_DOCKER_ENGINE_PORTS", "")
) or _parse_port_ranges("HP_DOCKER_ENGINE_PORTS", DEFAULT_DOCKER_ENGINE_PORTS)
DOCKER_ENGINE_PORTS_TEXT = ", ".join(
    str(first) if first == last else f"{first}-{last}" for first, last in DOCKER_ENGINE_PORTS
)
LOGGER.info("Docker Engine ports accepted from AppAPI: %s.", DOCKER_ENGINE_PORTS_TEXT)

###############################################################################
# Definitions
###############################################################################


class AccessLevel(IntEnum):
    PUBLIC = 0
    USER = 1
    ADMIN = 2


class ExAppRoute(BaseModel):
    url: str = Field(..., description="REGEX for URL, e.g. r'^/private/.*'")
    verb: str | None = Field(None, description="HTTP methods, e.g. 'GET,POST'. Older AppAPI versions do not send it.")
    access_level: AccessLevel = Field(..., description="ADMIN(2), USER(1), or PUBLIC(0)")
    bruteforce_protection: list[int] = Field(
        [], description="List with HTTP statuses to trigger the bruteforce protection."
    )
    str_bruteforce_protection: str = Field(
        "", description="Private field, that will be automatically initialized from the 'bruteforce_protection' value."
    )

    @model_validator(mode="after")
    def encode_bruteforce_protection_values(self) -> Self:
        self.str_bruteforce_protection = json.dumps(self.bruteforce_protection) if self.bruteforce_protection else ""
        return self

    def methods(self) -> frozenset[str] | None:
        """The methods the route lists, or None (every method) when AppAPI did not send a verb."""
        if self.verb is None:
            return None
        return frozenset(token.upper() for token in re.split(r"[\s,]+", self.verb) if token)

    def accepts(self, method: str) -> bool:
        """HEAD is accepted wherever GET is. Methods are case-sensitive: a backend may treat `get` as not GET."""
        methods = self.methods()
        if methods is None:
            return True
        return method in methods or (method == "HEAD" and "GET" in methods)


class ExApp(BaseModel):
    exapp_token: str = Field(..., repr=False)  # the ExApp's secret; records are written to the log
    exapp_version: str = Field(...)
    host: str = Field(...)
    port: int = Field(...)
    routes: list[ExAppRoute] = Field([])
    resolved_host: str = Field("", description="Contains resolved host field to the IP address.")


class NcUser(BaseModel):
    user_id: str = Field("", description="The Nextcloud user ID if not an anonymous user.")
    access_level: AccessLevel = Field(..., description="ADMIN(2), USER(1), or PUBLIC(0)")


def _sanitize_k8s_name(raw: str) -> str:
    """Convert an arbitrary string into a DNS-1123 compatible name for Kubernetes."""
    name = raw.lower().replace("_", "-")
    name = re.sub(r"[^a-z0-9-]", "-", name)
    name = re.sub(r"-+", "-", name).strip("-")
    if not name:
        name = "exapp"
    if len(name) > K8S_NAME_MAX_LENGTH:
        name = name[:K8S_NAME_MAX_LENGTH].rstrip("-")
    return name


class ExAppName(BaseModel):
    # The three parts end up in container, volume and hostname values and in Docker API URLs,
    # so they are limited to the characters Docker allows in a container name.
    name: str = Field(..., pattern=r"^[a-zA-Z0-9][a-zA-Z0-9_.-]*$", max_length=63, description="ExApp name.")
    instance_id: str = Field("", pattern=r"^[a-zA-Z0-9_.-]*$", max_length=63, description="Nextcloud instance ID.")
    role_suffix: str = Field(
        "", pattern=r"^[a-zA-Z0-9_.-]*$", max_length=63, description="Role suffix, e.g. 'rp', 'idx'."
    )

    @computed_field
    @property
    def exapp_container_name(self) -> str:
        base = f"nc_app_{self.instance_id}_{self.name}" if self.instance_id else f"nc_app_{self.name}"
        if self.role_suffix:
            base = f"{base}_{self.role_suffix}"
        return base

    @computed_field
    @property
    def exapp_container_volume(self) -> str:
        # Role-agnostic multirole K8s Deployments share one PVC and mount it at the same path as APP_PERSISTENT_STORAGE
        base = f"nc_app_{self.instance_id}_{self.name}" if self.instance_id else f"nc_app_{self.name}"
        return f"{base}_data"

    @computed_field
    @property
    def exapp_k8s_name(self) -> str:
        """K8s Deployment name."""
        return _sanitize_k8s_name(self.exapp_container_name)

    @computed_field
    @property
    def exapp_k8s_volume_name(self) -> str:
        """K8s PVC name."""
        base = _sanitize_k8s_name(self.exapp_container_volume)
        if len(base) > K8S_NAME_MAX_LENGTH:
            base = base[:K8S_NAME_MAX_LENGTH].rstrip("-")
        return base


class CreateExAppMounts(BaseModel):
    source: str = Field(...)
    target: str = Field(...)
    # Free text, as `occ app_api:app:register --mount SRC:DST:MODE` stores it; anything but "ro" mounts read-write.
    mode: str = Field("rw")

    @field_validator("source", "target")
    @classmethod
    def normalize_path(cls, value: str) -> str:
        return _normalize_mount_path(value)


class CreateExAppPayload(ExAppName):
    image_id: str = Field(..., description="Docker image ID.")
    network_mode: str = Field(..., description="Desired NetworkMode for the container.")
    environment_variables: list[str] = Field([], description="ExApp environment variables.")
    restart_policy: Literal["", "no", "always", "unless-stopped", "on-failure"] = Field(
        "unless-stopped", description="Desired RestartPolicy for the container."
    )
    compute_device: Literal["cpu", "rocm", "cuda"] = Field(
        "cpu", description="Possible values: 'cpu', 'rocm' or 'cuda'"
    )
    mount_points: list[CreateExAppMounts] = Field([], description="List of mount points for the container.")
    resource_limits: dict[str, Any] = Field({}, description="Resource limits for the container.")
    image_pull_policy: Literal["IfNotPresent", "Never", "Always"] = Field(
        "IfNotPresent", description="Kubernetes imagePullPolicy of the ExApp container (Kubernetes backend only)."
    )

    @model_validator(mode="before")
    @classmethod
    def accept_k8s_friendly_payload(cls, data: Any) -> Any:
        """Map 'image' -> 'image_id' and default network_mode for K8s payloads."""
        if isinstance(data, dict):
            if "image_id" not in data and "image" in data:
                data = {**data, "image_id": data["image"]}  # Allow 'image' instead of 'image_id'
            if "network_mode" not in data:
                data = {**data, "network_mode": "bridge"}  # Default network_mode (used only for Docker)
        return data

    @field_validator("network_mode")
    @classmethod
    def validate_network_mode(cls, value: str) -> str:
        """Accept 'host', 'bridge' or a network name, but not the '<mode>:<argument>' forms like 'container:<id>'."""
        separators = ":" in value or any(char.isspace() for char in value) or _has_control_characters(value)
        if not value or len(value) > 255 or separators:
            raise ValueError("must be 'host', 'bridge' or the name of a network")
        return value


class RemoveExAppPayload(ExAppName):
    remove_data: bool = Field(False, description="Flag indicating whether the Docker ExApp volume should be deleted.")


class InstallCertificatesPayload(ExAppName):
    system_certs_bundle: str | None = Field(None, description="Content of the system CA bundle (concatenated PEMs).")
    install_frp_certs: bool = Field(True, description="Flag to control installation of FRP certificates.")


class ExposeExAppPayload(ExAppName):
    port: int = Field(..., ge=1, le=65535, description="Port on which the ExApp listens inside the Pod/container.")
    expose_type: Literal["nodeport", "clusterip", "loadbalancer", "manual"] = Field("clusterip")
    upstream_host: str | None = Field(None, description="Host override. Required for expose_type=manual.")
    upstream_port: int | None = Field(None, ge=1, le=65535, description="Port override (manual only).")
    service_port: int | None = Field(None, ge=1, le=65535, description="Service port (defaults to payload.port).")
    node_port: int | None = Field(None, ge=30000, le=32767)
    external_traffic_policy: Literal["Cluster", "Local"] | None = Field(None)
    load_balancer_ip: str | None = Field(None)
    service_annotations: dict[str, str] = Field(default_factory=dict)
    service_labels: dict[str, str] = Field(default_factory=dict)
    wait_timeout_seconds: float = Field(60.0, ge=0, le=600)
    wait_interval_seconds: float = Field(1.0, ge=0.1, le=10.0)
    node_address_type: Literal["InternalIP", "ExternalIP"] = Field("InternalIP")
    node_name: str | None = Field(None)
    node_label_selector: str | None = Field(None)

    @model_validator(mode="after")
    def validate_expose_payload(self) -> Self:
        if self.expose_type == "manual" and not self.upstream_host:
            raise ValueError("upstream_host is required when expose_type='manual'")
        return self


class RemoveImagePayload(BaseModel):
    image_ref: str = Field(
        ...,
        min_length=1,
        description="Docker image reference to remove (repo:tag, repo@digest, or image ID).",
    )


###############################################################################
# In-memory caches
###############################################################################

EXAPP_CACHE: dict[str, ExApp] = {}
_EXAPP_INFLIGHT: dict[str, asyncio.Future[ExApp | None]] = {}
_EXAPP_NEGATIVE_CACHE: collections.OrderedDict[str, float] = collections.OrderedDict()
_NEGATIVE_CACHE_TTL = 15.0  # seconds to cache "ExApp not found" results

SESSION_CACHE_LOCK = asyncio.Lock()
SESSION_CACHE: dict[str, tuple[NcUser, float]] = {}  # Stores NcUser and timestamp
SESSION_REQUEST_WINDOW = float(os.environ.get("HP_SESSION_LIFETIME", "3"))  # Keep session information for 3 seconds
if SESSION_REQUEST_WINDOW < 0:
    raise ValueError("`HP_SESSION_LIFETIME` cannot be less than 0")
if SESSION_REQUEST_WINDOW > 10:
    raise ValueError("`HP_SESSION_LIFETIME` cannot be greater than 10")

BLACKLIST_CACHE_LOCK = asyncio.Lock()
BLACKLIST_CACHE: dict[str, list[float]] = {}  # ip_str -> list of timestamps of failures
# 5 minutes in seconds
BLACKLIST_REQUEST_WINDOW = int(os.getenv("HP_BLACKLIST_WINDOW", "300"))
# 10 invalid attempts during BLACKLIST_REQUEST_WINDOW
BLACKLIST_MAX_FAILS_COUNT = int(os.getenv("HP_BLACKLIST_COUNT", "10"))


###############################################################################
# BLACKLIST CACHE functions
###############################################################################


def get_true_client_ip(
    direct_ip: ipaddress.IPv4Address | ipaddress.IPv6Address, headers: dict[str, str]
) -> ipaddress.IPv4Address | ipaddress.IPv6Address:
    """Determine the true client IP by inspecting headers from trusted proxies."""
    if not TRUSTED_PROXIES:
        return direct_ip

    is_trusted = any(direct_ip in network for network in TRUSTED_PROXIES)
    if not is_trusted:
        return direct_ip

    # The request is from a trusted proxy, so we can check the headers.
    # X-Forwarded-For can be a list: client, proxy1, proxy2. We want the first one.
    x_forwarded_for = headers.get("x-forwarded-for")
    if x_forwarded_for:
        true_ip_str = x_forwarded_for.split(",")[0].strip()
        try:
            return ipaddress.ip_address(true_ip_str)
        except ValueError:
            LOGGER.warning("Could not parse IP from X-Forwarded-For header: %s", true_ip_str)

    x_real_ip = headers.get("x-real-ip")
    if x_real_ip:
        try:
            return ipaddress.ip_address(x_real_ip)
        except ValueError:
            LOGGER.warning("Could not parse IP from X-Real-IP header: %s", x_real_ip)
    return direct_ip  # If headers are present but invalid, fall back to the direct IP of the proxy


async def record_ip_failure(ip_address: str | IPv4Address | IPv6Address) -> None:
    """Record a failed request attempt for this IP using BLACKLIST_CACHE."""
    ip_str = str(ip_address)
    now = time.time()
    async with BLACKLIST_CACHE_LOCK:
        attempts = BLACKLIST_CACHE.get(ip_str, [])
        # Purge attempts that are older than the allowed window.
        attempts = [ts for ts in attempts if now - ts < BLACKLIST_REQUEST_WINDOW]
        attempts.append(now)
        BLACKLIST_CACHE[ip_str] = attempts
        LOGGER.warning("Recorded failure for IP %s. Failures in window: %d", ip_str, len(attempts))


def is_shared_key(value: Any) -> bool:
    """Compare a presented key with HP_SHARED_KEY in constant time; an unset or empty key never matches."""
    if not SHARED_KEY or not isinstance(value, str):
        return False
    return hmac.compare_digest(value.encode("utf-8", "surrogatepass"), SHARED_KEY.encode("utf-8", "surrogatepass"))


def is_exapp_frp_token(value: Any) -> bool:
    """Compare a presented FRP token with the one ExApp containers get, in constant time."""
    if not EXAPP_FRP_TOKEN or not isinstance(value, str):
        return False
    return hmac.compare_digest(value.encode("utf-8", "surrogatepass"), EXAPP_FRP_TOKEN.encode())


def with_exapp_frp_token(environment: list[str]) -> list[str]:
    """Replace the shared key that AppAPI gives an ExApp as `HP_SHARED_KEY` with the ExApp FRP token.

    ExApps use it only as their frpc login token. Any other value is left as it is.
    """
    result = []
    for entry in environment:
        name, _, value = entry.partition("=")
        result.append(f"{name}={EXAPP_FRP_TOKEN}" if name == "HP_SHARED_KEY" and is_shared_key(value) else entry)
    return result


async def record_failure_unless_trusted(
    ip_address: str | IPv4Address | IPv6Address, request_headers: dict[str, str]
) -> None:
    """Record a blacklist failure unless the request is authenticated with a valid harp-shared-key.

    Requests from Nextcloud/AppAPI carry `harp-shared-key` matching `SHARED_KEY`; a misconfiguration
    on that trusted path (e.g. reverse-proxy stripping `/exapps/`) would otherwise cause the caller
    to self-DoS once the blacklist window fills up.
    """
    if is_shared_key(request_headers.get("harp-shared-key")):
        return
    await record_ip_failure(ip_address)


async def is_ip_banned(ip_address: str | IPv4Address | IPv6Address) -> bool:
    """Return True if IP has exceeded the maximum allowed failures in the request window."""
    ip_str = str(ip_address)
    now = time.time()
    async with BLACKLIST_CACHE_LOCK:
        attempts = BLACKLIST_CACHE.get(ip_str, [])
        # Purge expired attempts.
        attempts = [ts for ts in attempts if now - ts < BLACKLIST_REQUEST_WINDOW]
        BLACKLIST_CACHE[ip_str] = attempts
        if len(attempts) >= BLACKLIST_MAX_FAILS_COUNT:
            return True
    return False


###############################################################################
# SESSION CACHE functions
###############################################################################


def session_log_id(pass_cookie: str) -> str:
    """A short identifier for log lines about a session; the cookie value itself is never written."""
    return hashlib.sha256(pass_cookie.encode("utf-8", "surrogatepass")).hexdigest()[:12]


async def record_session(pass_cookie: str, nc_user: NcUser) -> None:
    now = time.time()
    async with SESSION_CACHE_LOCK:
        SESSION_CACHE[pass_cookie] = (nc_user, now)
    LOGGER.info("Recorded session %s for user %s", session_log_id(pass_cookie), nc_user.user_id)


async def get_session(pass_cookie: str) -> NcUser | None:
    """Retrieve the session for the given IP address."""
    now = time.time()
    async with SESSION_CACHE_LOCK:
        session_data = SESSION_CACHE.get(pass_cookie)
        if session_data:
            nc_user, timestamp = session_data
            # Check if the session is still valid based on the SESSION_REQUEST_WINDOW
            if now - timestamp <= SESSION_REQUEST_WINDOW:
                return nc_user
            # Session expired, remove it
            del SESSION_CACHE[pass_cookie]
            LOGGER.info("Session %s expired", session_log_id(pass_cookie))
    return None


###############################################################################
# SPOA Handlers
###############################################################################


def _unescaped(match: re.Match) -> str:
    return chr(int(match[0][1:], 16))


def exapp_paths(target_path: str) -> tuple[str, list[str]]:
    """The path to forward to the ExApp, and the ways servers read it, starting with the path its routes are checked on.

    Servers differ on escaped unreserved characters, a `%` that starts no escape, repeated slashes and a raw backslash
    (a slash for WHATWG URL parsers), so the forwarded path has none of them. The checked path is the forwarded one
    decoded once, with `%2F` as a slash, and `//` merged, as for servers that decode before they route.
    """
    forwarded = UNRESERVED_ESCAPE_PATTERN.sub(_unescaped, LONE_PERCENT_PATTERN.sub("%25", target_path))
    forwarded = re.sub("//+", "/", forwarded.replace("\\", "%5C"))
    if len(forwarded) > MAX_FORWARDED_PATH_LENGTH:
        return forwarded, [forwarded]  # refused by the caller
    decoded = unquote(forwarded)
    checked = re.sub("//+", "/", decoded)
    readings = [checked, decoded]  # Starlette keeps the `//` of a decoded `%2F` next to a `/`
    if "%2F" in forwarded or "%2f" in forwarded:
        readings.append(unquote(re.sub("%2[fF]", "%252F", forwarded)))  # Express, aiohttp and Go keep it in its segment
    if ";" in forwarded:
        readings.append(re.sub("//+", "/", unquote(re.sub(";[^/]*", "", forwarded))))  # Tomcat drops `;` parameters
    if "?" in checked or "#" in checked or ";" in checked:
        resent = re.split("[?#]", checked, maxsplit=1)[0]  # a wrapper ExApp re-sending the decoded path cuts it there
        readings += [resent, re.sub("//+", "/", re.sub(";[^/]*", "", resent))]  # Tomcat behind it drops `;` parameters
    readings += [reading[:-1] for reading in readings if len(reading) > 1 and reading.endswith("/")]  # Express, Spring
    return forwarded, list(dict.fromkeys(readings))


def first_route(
    routes: list[ExAppRoute], path: str, flags: int, method: str, exapp_id: str
) -> tuple[ExAppRoute | None, list[ExAppRoute]]:
    """The first route matching both path and method, as in AppAPI's PHP proxy, or None and the routes of the path.

    A request without a method (a custom spoe-agent.conf without `method`) gets the first route of the path.
    """
    path_routes = []
    for route in routes:
        try:
            if not re.match(route.url, path, flags):
                continue
        except re.error as err:
            LOGGER.error("Invalid regex %s in route for exapp %s: %s", route.url, exapp_id, err)
            continue
        if not method or route.accepts(method):
            return route, path_routes
        path_routes.append(route)
    return None, path_routes


def find_route(
    routes: list[ExAppRoute], readings: list[str], method: str, exapp_id: str
) -> tuple[ExAppRoute | None, list[ExAppRoute]]:
    """The route that decides the request: the first one matching the checked path, case-sensitively, and the method.

    A route that another reading of the path (`exapp_paths`, the checked path first) or the path in another letter case
    (Express, AppAPI's PHP proxy) matches can only raise the access level.
    """
    route, path_routes = first_route(routes, readings[0], 0, method, exapp_id)
    if route is None:
        return None, path_routes
    for reading in readings:
        for flags in (0, re.IGNORECASE) if reading != readings[0] else (re.IGNORECASE,):
            other, _ = first_route(routes, reading, flags, method, exapp_id)
            if other is not None and other.access_level > route.access_level:
                route = other
    return route, path_routes


def allowed_methods(routes: list[ExAppRoute]) -> str:
    """The `Allow` header of a 405: the methods of these routes, with HEAD wherever GET is."""
    methods = set().union(*(route.methods() or frozenset() for route in routes))
    if "GET" in methods:
        methods.add("HEAD")
    return ", ".join(sorted(methods))


@SPOA_AGENT.handler("exapps_msg")
async def exapps_msg(
    path: str,
    headers: str,
    client_ip: ipaddress.IPv4Address | ipaddress.IPv6Address,
    pass_cookie: str,
    method: str = "",
) -> AckPayload:
    """An exception would kill the SPOA connection; a zero-action ACK hangs the stream until `timeout processing`."""
    try:
        return await _exapps_msg(path, headers, client_ip, pass_cookie, method)
    except Exception:
        LOGGER.exception("Unhandled error while processing request to path=%s", path)
        return AckPayload().set_txn_var("agent_error", 1)


async def _exapps_msg(
    path: str,
    headers: str,
    client_ip: ipaddress.IPv4Address | ipaddress.IPv6Address,
    pass_cookie: str,
    method: str = "",
) -> AckPayload:
    reply = AckPayload()
    request_headers = parse_headers(headers)
    client_ip_str = str(get_true_client_ip(client_ip, request_headers))
    reply = reply.set_txn_var("true_client_ip", client_ip_str)
    log_headers("Incoming request to ExApp: path=%s, ip=%s, headers=%s", request_headers, path, client_ip_str)

    # Check if the IP is banned based on failed attempts in BLACKLIST_CACHE.
    if await is_ip_banned(client_ip_str):
        LOGGER.warning("IP %s is banned due to excessive failed attempts.", client_ip_str)
        return reply.set_txn_var("bad_request", 1)

    if DOT_SEGMENT_PATTERN.search(path):
        LOGGER.warning("Rejecting path with a dot segment: %s (client_ip=%s)", path, client_ip_str)
        await record_failure_unless_trusted(client_ip_str, request_headers)
        return reply.set_txn_var("bad_request", 1)

    match = APPID_PATTERN.search(path)
    if not match:
        LOGGER.error("Invalid request path, cannot find AppID: %s", path)
        await record_failure_unless_trusted(client_ip_str, request_headers)
        return reply.set_txn_var("not_found", 1)
    exapp_id = match.group(1)
    exapp_id_lower = exapp_id.lower()
    target_path = path.removeprefix(f"/exapps/{exapp_id}")

    # Special handling for AppAPI requests
    if exapp_id == "app_api":
        reply = reply.set_txn_var("target_path", target_path)
        return await handle_app_api_request(target_path, request_headers, client_ip_str, reply)

    exapp_route_bruteforce_protection = None
    authorization_app_api = ""
    exapp_record = None
    if all(
        key in request_headers
        for key in [
            "ex-app-version",
            "ex-app-id",
            "ex-app-host",
            "ex-app-port",
            "authorization-app-api",
            "harp-shared-key",
        ]
    ):
        # This is a direct request from AppAPI to ExApp using AppAPI PHP functions "requestToExAppXXX"
        if not is_shared_key(request_headers["harp-shared-key"]):
            await record_ip_failure(client_ip_str)
            return reply.set_txn_var("bad_request", 1)
        authorization_app_api = request_headers["authorization-app-api"]
        # Use unified fetch path (cache + singleflight + K8s resolution)
        try:
            exapp_record = await _get_or_fetch_exapp(exapp_id_lower)
        except Exception as e:
            # The headers are enough for this request, but requests from users to the ExApp need the lookup.
            LOGGER.warning(
                "Cannot look up ExApp '%s', routing this AppAPI request by its headers: %.300s",
                exapp_id,
                describe_lookup_error(e),
            )
            exapp_record = None
        if not exapp_record:
            # ExApp not in Nextcloud yet (e.g. during /init) — use headers as fallback
            exapp_record = ExApp(
                exapp_token="",
                exapp_version=request_headers["ex-app-version"],
                host=request_headers["ex-app-host"],
                port=int(request_headers["ex-app-port"]),
            )

    if not exapp_record:
        try:
            exapp_record = await _get_or_fetch_exapp(exapp_id_lower)
            if not exapp_record:
                LOGGER.error("No such ExApp enabled: %s", exapp_id)
                await record_failure_unless_trusted(client_ip_str, request_headers)
                return reply.set_txn_var("not_found", 1)
        except ValidationError as e:
            LOGGER.error("Invalid ExApp metadata from Nextcloud: %s", describe_validation_error(e))
            return reply.set_txn_var("not_found", 1)
        except Exception as e:
            LOGGER.exception("Failed to fetch ExApp metadata from Nextcloud", exc_info=e)
            return reply.set_txn_var("not_found", 1)

    route_allowed = False
    if authorization_app_api:
        route_allowed = True  # We skip routes checking for AppAPI signed requests
    else:
        target_path, readings = exapp_paths(target_path)
        # Control characters too: Starlette's `$` also matches before a trailing newline (`/enabled%0A`).
        if (
            len(target_path) > MAX_FORWARDED_PATH_LENGTH
            or CONTROL_CHARACTER_PATTERN.search(readings[0])
            or any(reading.strip("/").lower() in LIFECYCLE_ENDPOINTS for reading in readings)
        ):
            LOGGER.error(
                "Refusing a request to an internal endpoint, with a control character or too long: %.200s", target_path
            )
            await record_ip_failure(client_ip_str)
            return reply.set_txn_var("bad_request", 1)
        route, path_routes = find_route(exapp_record.routes, readings, method, exapp_id)
        if route is None and path_routes:
            # Not counted toward the IP ban: link checkers, monitors and CORS preflights send such requests. So it is
            # decided before the user lookup, or random credentials would get checked by Nextcloud without limit.
            # not_found is also set, so that a custom /haproxy.cfg without the 405 rule answers 404.
            allow = allowed_methods(path_routes)
            LOGGER.info("No route of %s allows %s %s; allowed: %s", exapp_id, method, target_path, allow)
            reply = reply.set_txn_var("allow", allow)
            reply = reply.set_txn_var("method_not_allowed", 1)
            return reply.set_txn_var("not_found", 1)

        nc_user = None
        if pass_cookie or "authorization" in request_headers:
            # We also pass requests with "authorization" to the Nextcloud to support App Passwords and Basic Auth.
            nc_user = await get_session(pass_cookie)
            if not nc_user:
                try:
                    nc_user = await nc_get_user(exapp_id_lower, request_headers)
                    if nc_user and pass_cookie:
                        await record_session(pass_cookie, nc_user)
                except ValidationError as e:
                    LOGGER.error("Invalid user info from Nextcloud: %s", describe_validation_error(e))
                    return reply.set_txn_var("unauthorized", 1)
                except Exception as e:
                    LOGGER.exception("Failed to fetch user info from Nextcloud", exc_info=e)
                    return reply.set_txn_var("unauthorized", 1)

        if route is not None:
            if route.access_level == AccessLevel.PUBLIC or (nc_user and route.access_level <= nc_user.access_level):
                exapp_route_bruteforce_protection = route.str_bruteforce_protection
                route_allowed = True
            else:
                LOGGER.error("Access denied for '%s' to %s", nc_user.user_id if nc_user else "", target_path)
                await record_ip_failure(client_ip_str)
                return reply.set_txn_var("forbidden", 1)

    if not route_allowed:
        LOGGER.error("No defined route for handling %s", target_path)
        await record_ip_failure(client_ip_str)
        return reply.set_txn_var("not_found", 1)

    if not authorization_app_api:
        user_id = nc_user.user_id if nc_user else ""
        authorization_app_api = b64encode(f"{user_id}:{exapp_record.exapp_token}".encode(errors="ignore"))

    if exapp_route_bruteforce_protection:
        reply = reply.set_txn_var("statuses_to_trigger_bp", exapp_route_bruteforce_protection)
        reply = reply.set_txn_var("backend", "ex_apps_backend_w_bruteforce")
    else:
        reply = reply.set_txn_var("backend", "ex_apps_backend")

    if not exapp_record.resolved_host:
        try:
            ip_address(exapp_record.host)
            exapp_record.resolved_host = exapp_record.host
        except ValueError:
            exapp_record.resolved_host = await resolve_ip(exapp_record.host)
        if not exapp_record.resolved_host:
            LOGGER.error("Cannot resolve '%s' to IP address.", exapp_record.host)
            return reply.set_txn_var("not_found", 1)

    LOGGER.info("Rerouting request to %s:%s with path=%s", exapp_record.resolved_host, exapp_record.port, target_path)
    reply = reply.set_txn_var("target_path", target_path)
    reply = reply.set_txn_var("target_ip", exapp_record.resolved_host)
    reply = reply.set_txn_var("target_port", exapp_record.port)
    reply = reply.set_txn_var("exapp_token", authorization_app_api)
    reply = reply.set_txn_var("exapp_version", exapp_record.exapp_version)
    return reply.set_txn_var("exapp_id", exapp_id)


@SPOA_AGENT.handler("exapps_response_status_msg")
async def exapps_response_status_msg(status: int, client_ip: str, statuses_to_trigger_bp: str) -> AckPayload:
    """Top-level SPOA entrypoint, see exapps_msg for why it must never raise."""
    try:
        return await _exapps_response_status_msg(status, client_ip, statuses_to_trigger_bp)
    except Exception:
        LOGGER.exception("Unhandled error while processing a response status message")
        return AckPayload().set_txn_var("bp_error", 1)


async def _exapps_response_status_msg(status: int, client_ip: str, statuses_to_trigger_bp: str) -> AckPayload:
    reply = AckPayload()
    if not statuses_to_trigger_bp:
        return reply.set_txn_var("bp_triggered", 0)
    statuses = json.loads(statuses_to_trigger_bp)
    if status not in statuses:
        return reply.set_txn_var("bp_triggered", 0)
    LOGGER.warning("Bruteforce protection(status=%s) triggered IP=%s.", status, client_ip)
    await record_ip_failure(client_ip)
    return reply.set_txn_var("bp_triggered", 1)


###############################################################################
# Helper functions
###############################################################################


async def handle_app_api_request(
    target_path: str,
    request_headers: dict[str, str],
    str_client_ip: str,
    reply: AckPayload,
) -> AckPayload:
    """Handle the special case where the ExApp ID is 'app_api'."""
    LOGGER.debug("Request from AppAPI received: %s", target_path)
    if not is_shared_key(request_headers.get("harp-shared-key")):
        await record_ip_failure(str_client_ip)
        return reply.set_txn_var("unauthorized", 1)
    docker_engine_port = request_headers.get("docker-engine-port")
    if docker_engine_port and not target_path.startswith("/docker/"):
        engine_port = parse_docker_engine_port(docker_engine_port)
        if engine_port is None:
            return reply.set_txn_var("forbidden", 1)
        reply = reply.set_txn_var("target_port", engine_port)
        return reply.set_txn_var("backend", "docker_engine_backend")
    return reply.set_txn_var("backend", "nextcloud_control_backend")


def redact_headers(headers: dict[str, str]) -> dict[str, str]:
    """Return a copy of the headers fit for the log: credentials and cookies are replaced."""
    return {k: "[REDACTED]" if k.lower() in REDACTED_LOG_HEADERS else v for k, v in headers.items()}


def log_headers(message: str, headers: dict[str, str], *args: Any) -> None:
    """Log request headers at debug level, with credentials and cookies redacted; the headers come last."""
    if LOGGER.isEnabledFor(logging.DEBUG):
        LOGGER.debug(message, *args, redact_headers(headers))


def parse_headers(headers_str: str) -> dict[str, str]:
    """Parse a string containing HTTP headers into a dictionary.

    Each header should be on its own line in the format "Header-Name: value".
    The header names are normalized to lowercase.
    """
    headers = {}
    for line in headers_str.splitlines():
        line = line.strip()
        if not line:
            continue
        if ":" not in line:
            LOGGER.info("Malformed line in header: %s", line)
            continue
        key, value = line.split(":", 1)
        headers[key.strip().lower()] = value.strip()
    return headers


def _get_nc_session() -> aiohttp.ClientSession:
    """Shared pooled session for Nextcloud requests: bounded connector, NC_HTTP_TIMEOUT applied."""
    global _nc_session
    if _nc_session is None or _nc_session.closed:
        # DummyCookieJar: client cookies are forwarded explicitly per request; the shared
        # session must never store Set-Cookie responses, or one user's Nextcloud session
        # cookies would be replayed on other users' requests.
        _nc_session = aiohttp.ClientSession(
            timeout=NC_HTTP_TIMEOUT,
            # limit=100 is the aiohttp default; a tighter cap delays cold-cache re-auth bursts by whole waves.
            connector=aiohttp.TCPConnector(limit=100),
            cookie_jar=aiohttp.DummyCookieJar(),
        )
    return _nc_session


async def _close_nc_session(_app: web.Application = None) -> None:
    global _nc_session
    if _nc_session is not None and not _nc_session.closed:
        await _nc_session.close()
        _nc_session = None


async def nc_get_exapp(app_id: str) -> ExApp | None:
    async with _get_nc_session().get(
        EX_APP_URL, headers={"harp-shared-key": SHARED_KEY}, params={"appId": app_id}
    ) as resp:
        if not resp.ok:
            if resp.status in (401, 404):
                # 404 = ExApp not found; 401 = ExApp not found so key can't be
                # validated against a daemon config (Nextcloud returns 401 for
                # non-existent apps because it has no daemon to check the key).
                return None
            raise Exception("Failed to fetch ExApp metadata from Nextcloud.", await resp.text())
        data = await resp.json()
        return ExApp.model_validate(data)


async def _fetch_exapp_record(exapp_id: str) -> ExApp | None:
    """Fetch ExApp metadata from Nextcloud and optionally resolve K8s upstream.

    Performs network I/O (Nextcloud HTTP + K8s API).  Must NEVER be called
    under a lock.  Returns a fully-populated ExApp ready for caching, or
    None if the ExApp is not found/enabled in Nextcloud.
    """
    exapp_record = await nc_get_exapp(exapp_id)
    if not exapp_record:
        return None
    k8s_upstream = await _k8s_resolve_exapp_upstream(exapp_id)
    if k8s_upstream:
        exapp_record.host, exapp_record.port = k8s_upstream
        exapp_record.resolved_host = ""
        LOGGER.info("Resolved K8s upstream for '%s': %s:%d", exapp_id, *k8s_upstream)
    return exapp_record


def _negative_cache_evict_expired() -> None:
    """Remove expired entries from the front of the ordered negative cache."""
    now = time.monotonic()
    while _EXAPP_NEGATIVE_CACHE:
        _, expiry = next(iter(_EXAPP_NEGATIVE_CACHE.items()))
        if now < expiry:
            break
        _EXAPP_NEGATIVE_CACHE.popitem(last=False)


async def _get_or_fetch_exapp(exapp_id: str) -> ExApp | None:
    """Get ExApp from cache, or fetch with request coalescing.

    Fast path: returns cached record immediately.
    Negative-cache path: if we recently confirmed this ExApp doesn't exist,
    return None without network I/O.
    On cache miss: if another coroutine is already fetching this ExApp,
    awaits its result (singleflight).  Otherwise initiates the fetch and
    lets other coroutines await our result.

    Safety: between the _EXAPP_INFLIGHT.get() check and the dict insert
    there is no ``await``, so no coroutine interleaving can occur in
    single-threaded asyncio — only one coroutine wins the race.
    """
    cached = EXAPP_CACHE.get(exapp_id)
    if cached is not None:
        return cached

    neg_expiry = _EXAPP_NEGATIVE_CACHE.get(exapp_id)
    if neg_expiry is not None:
        if time.monotonic() < neg_expiry:
            return None
        _EXAPP_NEGATIVE_CACHE.pop(exapp_id, None)

    inflight = _EXAPP_INFLIGHT.get(exapp_id)
    if inflight is not None:
        return await inflight

    loop = asyncio.get_running_loop()
    fut: asyncio.Future[ExApp | None] = loop.create_future()
    _EXAPP_INFLIGHT[exapp_id] = fut

    try:
        exapp_record = await _fetch_exapp_record(exapp_id)
        if exapp_record is not None:
            LOGGER.info("Received new ExApp record: %s", exapp_record)
            EXAPP_CACHE[exapp_id] = exapp_record
        else:
            _negative_cache_evict_expired()
            _EXAPP_NEGATIVE_CACHE[exapp_id] = time.monotonic() + _NEGATIVE_CACHE_TTL
        if not fut.done():  # a cancelled waiter cancels the shared future as well
            fut.set_result(exapp_record)
        return exapp_record
    except BaseException as exc:
        if not fut.done():
            fut.set_exception(exc)
            # Waiters get it from the future and our caller from the raise below; marked as retrieved, asyncio
            # does not log it a second time when the future is collected.
            fut.exception()
        raise
    finally:
        _EXAPP_INFLIGHT.pop(exapp_id, None)


async def nc_get_user(app_id: str, all_headers: dict[str, str]) -> NcUser | None:
    ext_headers = {
        k: v
        for k, v in all_headers.items()
        if k.lower() not in EXCLUDE_HEADERS_USER_INFO
        and not k.lower().startswith("content-")
        and not UNSENDABLE_HEADER_CHARS.search(k + v)
    }
    log_headers("Requesting user info for ExApp '%s' with headers %s", ext_headers, app_id)
    async with _get_nc_session().get(
        USER_INFO_URL,
        headers={**ext_headers, "harp-shared-key": SHARED_KEY},
        params={"appId": app_id},
    ) as resp:
        if not resp.ok:
            error_text = await resp.text()
            LOGGER.info("Failed to fetch user info from Nextcloud: %s", error_text)
            if resp.status // 100 == 4:
                return None
            raise Exception("Failed to fetch user info from Nextcloud.", error_text)
        data = await resp.json()
        return NcUser.model_validate(data)


async def resolve_ip(hostname: str) -> str:
    # Blocking getaddrinfo here would freeze the whole event loop; resolve in the executor, time-capped.
    loop = asyncio.get_running_loop()
    try:
        # 10s cap: enough for one full resolver retry cycle, far below the 30s SPOE budget.
        addr_info = await asyncio.wait_for(loop.getaddrinfo(hostname, None), timeout=10.0)
    except (socket.gaierror, TimeoutError):
        return ""
    for family, _, _, _, sockaddr in addr_info:
        if family == socket.AF_INET:  # IPv4
            return sockaddr[0]
    # If no IPv4, return first IPv6
    for family, _, _, _, sockaddr in addr_info:
        if family == socket.AF_INET6:  # IPv6
            return sockaddr[0]
    return ""


###############################################################################
# Misc routes
###############################################################################


async def get_heartbeat(request: web.Request):
    """Liveness probe for healthcheck.sh and the start.sh watchdog; unlike /info it does no network I/O."""
    return web.json_response({"status": "ok"})


async def get_info(request: web.Request):
    """Report HaRP's version and backend status; used by the AppAPI daemon check."""
    k8s_status: dict[str, Any] = {"enabled": K8S_ENABLED}
    if K8S_ENABLED:
        k8s_status["api_server"] = K8S_API_SERVER or ""
        reachable, error = await _k8s_probe()
        k8s_status["reachable"] = reachable
        if error:
            k8s_status["error"] = error

    return web.json_response({
        "version": HARP_VERSION,
        "docker": True,
        "kubernetes": k8s_status,
    })


###############################################################################
# ExApp routes
###############################################################################


async def add_exapp(request: web.Request):
    data = await request.json()
    app_id = request.match_info["app_id"].lower()
    try:
        exapp_record = ExApp.model_validate(data)
    except ValidationError:
        raise web.HTTPBadRequest() from None

    # AppAPI sends the logical host (the appId). For K8s it must be replaced with
    # the real upstream or SPOA would NXDOMAIN. Mirror _fetch_exapp_record's cache-miss path.
    k8s_upstream = await _k8s_resolve_exapp_upstream(app_id)
    if k8s_upstream:
        exapp_record.host, exapp_record.port = k8s_upstream
        exapp_record.resolved_host = ""

    EXAPP_CACHE[app_id] = exapp_record
    _EXAPP_NEGATIVE_CACHE.pop(app_id, None)
    return web.HTTPNoContent()


async def delete_exapp(request: web.Request):
    old = EXAPP_CACHE.pop(request.match_info["app_id"].lower(), None)
    if old is None:
        raise web.HTTPNotFound()
    return web.HTTPNoContent()


###############################################################################
# FRP Plugin Authentication
###############################################################################


def frp_client_host(address: str) -> str:
    """Return the host of an FRP `client_address`: "1.2.3.4:5678" or "[2001:db8::1]:5678"."""
    if address.startswith("["):
        return address[1:].partition("]")[0]
    host, separator, _ = address.rpartition(":")
    return host if separator and ":" not in host else address


async def frp_auth(request: web.Request):
    if request.method != "POST":
        raise web.HTTPBadRequest()
    try:
        json_data = await request.json()
        client_ip = frp_client_host(str(json_data["content"]["client_address"]))
    except Exception:
        raise web.HTTPBadRequest() from None

    if await is_ip_banned(client_ip):
        return web.json_response({"reject": True, "reject_reason": "banned"})

    # frpc sends `metas: null` when no `metadatas` are configured.
    metas = json_data["content"].get("metas")
    auth_token = metas.get("token") if isinstance(metas, dict) else None
    # ExApps deployed by this version log in with their token; HaRP's own frpc, remote Docker engines and ExApps
    # deployed earlier use the shared key.
    if is_shared_key(auth_token) or is_exapp_frp_token(auth_token):
        return web.json_response({"reject": False, "unchange": True})

    await record_ip_failure(client_ip)
    raise web.HTTPBadRequest()


###############################################################################
# Endpoints for AppAPI to work with the Docker API
###############################################################################


def describe_validation_error(error: ValidationError) -> str:
    """Summarize a pydantic error as 'field: message' pairs; the submitted values themselves are left out."""
    return "; ".join(
        f"{'.'.join(str(part) for part in item['loc']) or 'payload'}: {item['msg'].removeprefix('Value error, ')}"
        for item in error.errors()
    )


def describe_lookup_error(error: Exception) -> str:
    """Summarize why an ExApp lookup failed in one line; for a validation error without the values."""
    if isinstance(error, ValidationError):
        return describe_validation_error(error)
    # str() of an aiohttp web exception is only its reason phrase, such as "Service Unavailable".
    reason = error.text if isinstance(error, web.HTTPException) else str(error)
    return f"{type(error).__name__}: {reason}" if reason else type(error).__name__


def parse_docker_engine_port(value: str) -> int | None:
    """Return the port of a Docker Engine the agent may talk to, or None (and log why) when the value is not one."""
    try:
        port = int(value)
    except ValueError:
        port = None
    if port is not None and any(first <= port <= last for first, last in DOCKER_ENGINE_PORTS):
        return port
    # The value comes from the request: shortened and escaped, so control characters do not reach the log.
    LOGGER.error("Invalid 'docker-engine-port' header value: %r (allowed: %s).", value[:32], DOCKER_ENGINE_PORTS_TEXT)
    return None


def get_docker_engine_port(request: web.Request) -> int:
    docker_engine_port_str = request.headers.get("docker-engine-port")
    if not docker_engine_port_str:
        LOGGER.error("Missing 'docker-engine-port' header.")
        raise web.HTTPBadRequest(text="Missing 'docker-engine-port' header.")

    docker_engine_port = parse_docker_engine_port(docker_engine_port_str)
    if docker_engine_port is None:
        raise web.HTTPBadRequest(
            text=f"Invalid 'docker-engine-port' header value: {docker_engine_port_str[:32]!r} "
            f"(allowed: {DOCKER_ENGINE_PORTS_TEXT})"
        )
    return docker_engine_port


def check_mount_modes(mounts: list[CreateExAppMounts], enforce: bool, container_name: str = "") -> str | None:
    """Return why a mount mode is refused; until enforced, a mode other than 'ro' or 'rw' only logs a warning.

    AppAPI stores modes such as "z" or "RO" as typed and re-sends them on every ExApp update; they have always
    mounted read-write, so refusing them at once would leave the ExApp without a container.
    """
    for mount in mounts:
        if mount.mode in ("ro", "rw"):
            continue
        if enforce:
            return (
                f"HP_EXAPP_BIND_MOUNTS_ENFORCE_ALL refuses mode '{mount.mode}' of '{mount.source}': use 'ro' or 'rw'."
            )
        LOGGER.warning(
            "Bind mount of '%s'%s has mode '%s', which mounts it read-write; a later release will refuse it. Use 'ro' "
            "or 'rw'; HP_EXAPP_BIND_MOUNTS_ENFORCE_ALL=true refuses it now.",
            mount.source,
            f" for '{container_name}'" if container_name else "",
            mount.mode,
        )
    return None


def check_docker_resource_limits(resource_limits: dict[str, Any]) -> str | None:
    """Return why the limits cannot be passed to the Docker Engine, or None when they can."""
    for limit in ("memory", "nanoCPUs"):
        value = resource_limits.get(limit)
        if value is not None and (isinstance(value, bool) or not isinstance(value, int) or value < 0):
            return f"Payload validation error: resource_limits.{limit} must be a non-negative integer."
    return None


def _is_same_or_below(path: str, parent: str) -> bool:
    return parent == "/" or path == parent or path.startswith(f"{parent}/")


def _governing_rule(
    path: str, mode: str, allowed: list[str], denied: list[tuple[str, bool]]
) -> tuple[bool, str] | None:
    """Return (is_denied, entry) for the most specific entry covering the path, or None when no entry covers it.

    The entry with the longest path decides, a denied entry wins a tie, and ":rw" entries do not apply to
    read-only mounts. This is what lets an allowed entry below a denied one permit that part of it.
    """
    best: tuple[tuple[int, bool], bool, str] | None = None
    for rule_path, is_denied, rw_only in [(d, True, rw) for d, rw in denied] + [(a, False, False) for a in allowed]:
        if (rw_only and mode == "ro") or not _is_same_or_below(path, rule_path):
            continue
        key = (len(rule_path), is_denied)
        if best is None or key > best[0]:
            best = (key, is_denied, _rule_label(rule_path, rw_only))
    return None if best is None else (best[1], best[2])


def _mount_refusal(
    mount: CreateExAppMounts,
    allowed: list[str],
    denied: list[tuple[str, bool]],
    warn_only: frozenset[str] = frozenset(),
) -> str | None:
    """Return why one bind mount is refused under these rules, or None when it is permitted.

    Denied rules in `warn_only` refuse nothing, but they still count as the most specific entry, so one below an
    enforced rule (`/var/lib/docker/volumes` below `/var/lib/docker`) keeps its part permitted.
    """
    covered = any(_is_same_or_below(mount.source, path) for path in allowed)
    governing = _governing_rule(mount.source, mount.mode, allowed, denied)
    warn_only_decides = governing is not None and governing[0] and governing[1] in warn_only
    # Strict mode (`DENIED=/` with an allowed list): the missing allowed entry is the reason, not the root rule.
    if allowed and not covered and (governing is None or governing[1] in ("/", "/:rw") or warn_only_decides):
        return f"HP_EXAPP_BIND_MOUNTS_ALLOWED does not cover '{mount.source}'."
    if governing is not None and governing[0] and not warn_only_decides:
        return f"HP_EXAPP_BIND_MOUNTS_DENIED rule '{governing[1]}' forbids mounting '{mount.source}' ({mount.mode})."
    for denied_path, rw_only in denied:
        if (rw_only and mount.mode == "ro") or _rule_label(denied_path, rw_only) in warn_only:
            continue
        if _is_same_or_below(denied_path, mount.source):
            return (
                f"HP_EXAPP_BIND_MOUNTS_DENIED rule '{_rule_label(denied_path, rw_only)}' forbids mounting "
                f"'{mount.source}' ({mount.mode}): the mount contains it."
            )
    return None


def check_bind_mounts(
    mounts: list[CreateExAppMounts],
    disabled: bool,
    allowed: list[str],
    denied: list[tuple[str, bool]],
    warn_only: frozenset[str] = frozenset(),
    container_name: str = "",
) -> str | None:
    """Return why a bind mount is refused, or None when all of them are permitted.

    The most specific entry covering the source decides (see `_governing_rule`); with an allowed list, a source
    no entry covers is refused; and a denied path below the source refuses the mount, because mounting a parent
    exposes the path as well. Denied rules listed in `warn_only` refuse nothing (see `_mount_refusal`); a mount
    they would refuse is permitted with a warning. Paths are compared after normalization: links on the host are not
    resolved.
    """
    if disabled and mounts:
        return "HP_EXAPP_BIND_MOUNTS_DISABLED is set, so no bind mount is permitted."
    for mount in mounts:
        refusal = _mount_refusal(mount, allowed, denied, warn_only)
        if refusal:
            return refusal
        if later := _mount_refusal(mount, allowed, denied):
            # Offer only the changes that make the mount pass once every rule is enforced.
            remedies = []
            if mount.mode != "ro" and not _mount_refusal(mount.model_copy(update={"mode": "ro"}), allowed, denied):
                remedies.append("mount it read-only")
            if not _mount_refusal(mount, [*allowed, mount.source], denied):
                # Setting an allowed list for the first time refuses every mount it does not cover.
                remedies.append(
                    "list it in HP_EXAPP_BIND_MOUNTS_ALLOWED"
                    if allowed
                    else "list it, together with every other host path your ExApps mount, in "
                    "HP_EXAPP_BIND_MOUNTS_ALLOWED"
                )
            LOGGER.warning(
                "Bind mount of '%s' (%s)%s is permitted for now, but a later release will refuse it: %s "
                "To keep it, %s; HP_EXAPP_BIND_MOUNTS_ENFORCE_ALL=true refuses it now.",
                mount.source,
                mount.mode,
                f" for '{container_name}'" if container_name else "",
                later,
                " or ".join(remedies) or "set HP_EXAPP_BIND_MOUNTS_DENIED to your own list without that rule",
            )
    return None


async def docker_exapp_exists(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = ExAppName.model_validate(payload_dict)
    except ValidationError as e:
        raise web.HTTPBadRequest(text=f"Payload validation error: {describe_validation_error(e)}") from None

    container_name = payload.exapp_container_name
    docker_api_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/json"
    LOGGER.debug("Checking for container '%s' via Docker API at %s", container_name, docker_api_url)
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15.0)) as session:
        try:
            async with session.get(docker_api_url) as resp:
                if resp.status == 200:
                    LOGGER.info("Container '%s' exists.", container_name)
                    # Surface the image reference the container was created from. AppAPI uses
                    # this for image cleanup so it knows which ref to delete after the container
                    # is gone, without having to track image refs in its own state.
                    image_ref = ""
                    try:
                        data = await resp.json()
                        image_ref = (data.get("Config") or {}).get("Image") or data.get("Image") or ""
                    except (json.JSONDecodeError, aiohttp.ContentTypeError) as e:
                        LOGGER.warning("Container '%s' exists but inspect body was not parseable: %s", container_name, e)
                    return web.json_response({"exists": True, "image_ref": image_ref})
                if resp.status == 404:
                    LOGGER.info("Container '%s' does not exist.", container_name)
                    return web.json_response({"exists": False})
                error_text = await resp.text()
                LOGGER.error(
                    "Error checking container '%s' with Docker API (status %s): %s",
                    container_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(text=f"Error communicating with Docker Engine: Status {resp.status}")
        except aiohttp.ClientConnectorError as e:
            LOGGER.error("Could not connect to Docker Engine at %s:%s: %s", DOCKER_API_HOST, docker_engine_port, e)
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout while trying to communicate with Docker Engine at %s:%s for container '%s'",
                DOCKER_API_HOST,
                docker_engine_port,
                container_name,
            )
            raise web.HTTPGatewayTimeout(
                text=f"Timeout communicating with Docker Engine on port {docker_engine_port}"
            ) from e
        except Exception as e:
            LOGGER.exception("Unexpected error while checking container '%s' existence via Docker API.", container_name)
            raise web.HTTPInternalServerError(
                text="An unexpected error occurred while checking container status."
            ) from e


async def docker_exapp_create(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/create")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = CreateExAppPayload.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/create: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    volume_name = payload.exapp_container_volume
    image_id = payload.image_id

    # Checked before the volume is created, so a refused request leaves nothing behind.
    refusal = check_bind_mounts(
        payload.mount_points,
        BIND_MOUNTS_DISABLED,
        BIND_MOUNTS_ALLOWED,
        BIND_MOUNTS_DENIED,
        BIND_MOUNTS_WARN_ONLY,
        container_name,
    ) or check_mount_modes(payload.mount_points, BIND_MOUNTS_ENFORCE_ALL, container_name)
    refusal = refusal or check_docker_resource_limits(payload.resource_limits)
    if refusal:
        LOGGER.warning("Refusing to create container '%s': %s", container_name, refusal)
        raise web.HTTPBadRequest(text=refusal)

    container_config = {
        "Image": image_id,
        "Hostname": payload.name,
        "HostConfig": {
            "NetworkMode": payload.network_mode,
            "Mounts": [
                {
                    "Type": "volume",
                    "Source": volume_name,
                    "Target": f"/{volume_name}",
                    "ReadOnly": False,
                }
            ],
            "RestartPolicy": {
                "Name": payload.restart_policy,
            },
        },
        "Env": with_exapp_frp_token(payload.environment_variables),
    }
    if payload.network_mode not in ("host", "bridge"):
        container_config["NetworkingConfig"] = {"EndpointsConfig": {payload.network_mode: {"Aliases": [payload.name]}}}

    if payload.compute_device == "cuda":
        container_config["HostConfig"]["DeviceRequests"] = [
            {
                "Driver": "nvidia",
                "Count": -1,
                "Capabilities": [["compute", "utility"]],
            }
        ]
    elif payload.compute_device == "rocm":
        devices = []
        for device in ("/dev/kfd", "/dev/dri"):
            devices.append({"PathOnHost": device, "PathInContainer": device, "CgroupPermissions": "rwm"})
        container_config["HostConfig"]["Devices"] = devices

    if payload.resource_limits:
        if "memory" in payload.resource_limits:
            container_config["HostConfig"]["Memory"] = payload.resource_limits["memory"]
        if "nanoCPUs" in payload.resource_limits:
            container_config["HostConfig"]["NanoCPUs"] = payload.resource_limits["nanoCPUs"]

    for extra_mount in payload.mount_points:
        container_config["HostConfig"]["Mounts"].append(
            {
                "Source": extra_mount.source,
                "Target": extra_mount.target,
                "Type": "bind",
                "Readonly": extra_mount.mode == "ro",
            }
        )

    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=60.0)) as session:
        create_volume_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/volumes/create"
        inspect_volume_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/volumes/{volume_name}"

        LOGGER.debug("Checking/Creating volume '%s' via Docker API", volume_name)
        try:
            async with session.get(inspect_volume_url) as resp_inspect:
                if resp_inspect.status == 200:
                    LOGGER.info("Volume '%s' already exists.", volume_name)
                elif resp_inspect.status == 404:
                    LOGGER.info("Volume '%s' not found, attempting to create.", volume_name)
                    async with session.post(create_volume_url, json={"Name": volume_name}) as resp_create:
                        if resp_create.status == 201:
                            LOGGER.info("Volume '%s' created successfully.", volume_name)
                        else:
                            error_text = await resp_create.text()
                            LOGGER.error(
                                "Failed to create volume '%s' (status %s): %s",
                                volume_name,
                                resp_create.status,
                                error_text,
                            )
                            raise web.HTTPServiceUnavailable(
                                text=f"Failed to create volume '{volume_name}': Status {resp_create.status}"
                            )
                else:
                    error_text = await resp_inspect.text()
                    LOGGER.error(
                        "Error inspecting volume '%s' (status %s): %s", volume_name, resp_inspect.status, error_text
                    )
                    raise web.HTTPServiceUnavailable(
                        text=f"Error inspecting volume '{volume_name}': Status {resp_inspect.status}"
                    )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error("Could not connect to Docker Engine for volume operation: %s", e)
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error("Timeout during volume operation for '%s'", volume_name)
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for volume operation") from e
        except web.HTTPServiceUnavailable:
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error during volume management for '%s'", volume_name)
            raise web.HTTPInternalServerError(text="Unexpected error during volume management.") from e

        create_container_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/create?name={container_name}"
        LOGGER.debug(
            "Attempting to create container '%s' with image '%s' via Docker API at %s",
            container_name,
            image_id,
            create_container_url,
        )
        try:
            async with session.post(create_container_url, json=container_config) as resp:
                if resp.status == 201:
                    container_data = await resp.json()
                    container_id = container_data.get("Id")
                    LOGGER.info("Container '%s' (ID: %s) created successfully.", container_name, container_id)
                    return web.json_response({"id": container_id, "name": container_name}, status=201)
                if resp.status == 409:
                    error_text = await resp.text()
                    LOGGER.warning("Container '%s' already exists (status 409): %s.", container_name, error_text)
                    raise web.HTTPConflict(text=f"Container with name '{container_name}' already exists.")
                error_text = await resp.text()
                LOGGER.error(
                    "Error creating container '%s' with Docker API (status %s): %s",
                    container_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Error creating container '{container_name}': Status {resp.status}"
                )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error("Could not connect to Docker Engine for container creation: %s", e)
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error("Timeout during container creation for '%s'", container_name)
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for container creation") from e
        except (web.HTTPServiceUnavailable, web.HTTPConflict, web.HTTPInternalServerError):
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error during container creation for '%s'", container_name)
            raise web.HTTPInternalServerError(text="An unexpected error occurred during container creation.") from e


async def docker_exapp_start(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/start")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = ExAppName.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/start: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    start_container_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/start"

    LOGGER.info("Attempting to start container '%s' via Docker API at %s", container_name, start_container_url)
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session:
        try:
            async with session.post(start_container_url) as resp:
                if resp.status == 204:
                    LOGGER.info("Container '%s' started successfully.", container_name)
                    return web.HTTPNoContent()
                if resp.status == 304:
                    LOGGER.info("Container '%s' was already started.", container_name)
                    return web.HTTPOk(text="Container already started")
                if resp.status == 404:
                    LOGGER.warning("Container '%s' not found, cannot start.", container_name)
                    raise web.HTTPNotFound(text=f"Container '{container_name}' not found.")
                error_text = await resp.text()
                LOGGER.error(
                    "Error starting container '%s' with Docker API (status %s): %s",
                    container_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Error starting container '{container_name}' via Docker Engine: Status {resp.status}"
                )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error(
                "Could not connect to Docker Engine at %s:%s to start container: %s",
                DOCKER_API_HOST,
                docker_engine_port,
                e,
            )
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout while trying to start container '%s' via Docker Engine at %s:%s",
                DOCKER_API_HOST,
                container_name,
                docker_engine_port,
            )
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for container start") from e
        except (web.HTTPNotFound, web.HTTPServiceUnavailable):
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error while starting container '%s' via Docker API.", container_name)
            raise web.HTTPInternalServerError(text="An unexpected error occurred during container start.") from e


async def docker_exapp_stop(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/stop")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = ExAppName.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/stop: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    stop_container_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/stop"

    LOGGER.info("Attempting to stop container '%s' via Docker API at %s", container_name, stop_container_url)
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session:
        try:
            async with session.post(stop_container_url) as resp:
                if resp.status == 204:
                    LOGGER.info("Container '%s' stopped successfully.", container_name)
                    return web.HTTPNoContent()
                if resp.status == 304:
                    LOGGER.info("Container '%s' was already stopped.", container_name)
                    return web.HTTPOk(text="Container already stopped")
                if resp.status == 404:
                    LOGGER.warning("Container '%s' not found, cannot stop.", container_name)
                    raise web.HTTPNotFound(text=f"Container '{container_name}' not found.")
                error_text = await resp.text()
                LOGGER.error(
                    "Error stopping container '%s' with Docker API (status %s): %s",
                    container_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Error stopping container '{container_name}' via Docker Engine: Status {resp.status}"
                )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error(
                "Could not connect to Docker Engine at %s:%s to stop container: %s",
                DOCKER_API_HOST,
                docker_engine_port,
                e,
            )
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout while trying to stop container '%s' via Docker Engine at %s:%s",
                DOCKER_API_HOST,
                container_name,
                docker_engine_port,
            )
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for container stop") from e
        except (web.HTTPNotFound, web.HTTPServiceUnavailable):
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error while stopping container '%s' via Docker API.", container_name)
            raise web.HTTPInternalServerError(text="An unexpected error occurred during container stop.") from e


async def docker_exapp_wait_for_start(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/wait_for_start")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = ExAppName.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/wait_for_start: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    inspect_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/json"

    max_tries = 180
    sleep_interval = 0.5
    total_wait_time = max_tries * sleep_interval
    client_timeout = aiohttp.ClientTimeout(total=total_wait_time + 15.0)

    LOGGER.info(
        "Waiting for container '%s' to start (max %d tries, interval %.1fs, total wait %.1fs).",
        container_name,
        max_tries,
        sleep_interval,
        total_wait_time,
    )

    last_known_status: str | None = "unknown"
    last_known_health: str | None = None

    async with aiohttp.ClientSession(timeout=client_timeout) as session:
        for attempt in range(max_tries):
            try:
                async with session.get(inspect_url) as resp:
                    if resp.status == 200:
                        container_info = await resp.json()
                        state = container_info.get("State", {})
                        current_status = state.get("Status")
                        current_health = state.get("Health", {}).get("Status")

                        last_known_status = current_status
                        last_known_health = current_health

                        LOGGER.debug(
                            "Container '%s' attempt %d/%d: Status='%s', Health='%s'",
                            container_name,
                            attempt + 1,
                            max_tries,
                            current_status,
                            current_health,
                        )
                        if current_status == "running":
                            if current_health is None or current_health == "healthy":
                                LOGGER.info(
                                    "Container '%s' is running and healthy (or no healthcheck).", container_name
                                )
                                return web.json_response(
                                    {"started": True, "status": current_status, "health": current_health}
                                )
                            if current_health == "unhealthy":
                                LOGGER.warning(
                                    "Container '%s' is running but unhealthy. Reporting as not successfully started.",
                                    container_name,
                                )
                                return web.json_response(
                                    {
                                        "started": True,
                                        "status": current_status,
                                        "health": current_health,
                                        "reason": "unhealthy",
                                    }
                                )
                            if current_health == "starting":
                                LOGGER.info(
                                    "Container '%s' is running, health status is 'starting'. Continuing to wait.",
                                    container_name,
                                )
                        elif current_status in ("created", "restarting"):
                            LOGGER.info("Container '%s' is '%s'. Continuing to wait.", container_name, current_status)
                        elif current_status in ("paused", "exited", "dead"):
                            LOGGER.warning(
                                "Container '%s' is in a non-recoverable state (current state: %s). Stopping wait.",
                                container_name,
                                current_status,
                            )
                            return web.json_response(
                                {
                                    "started": False,
                                    "status": current_status,
                                    "health": current_health,
                                    "reason": f"non-recoverable state: {current_status}",
                                }
                            )
                    elif resp.status == 404:
                        LOGGER.warning(
                            "Container '%s' not found while waiting for start (attempt %d).",
                            container_name,
                            attempt + 1,
                        )
                        return web.json_response(
                            {"started": False, "status": "not_found", "reason": "container not found"}
                        )
                    else:
                        error_text = await resp.text()
                        LOGGER.error(
                            "Error inspecting container '%s' while waiting (status %s, attempt %d): %s",
                            container_name,
                            resp.status,
                            attempt + 1,
                            error_text,
                        )
                        raise web.HTTPServiceUnavailable(
                            text=f"Error inspecting container '{container_name}': Status {resp.status}"
                        )
            except aiohttp.ClientConnectorError as e:
                LOGGER.error(
                    "Could not connect to Docker Engine at %s:%s while waiting for container '%s' (attempt %d): %s",
                    DOCKER_API_HOST,
                    docker_engine_port,
                    container_name,
                    attempt + 1,
                    e,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Could not connect to Docker Engine on port {docker_engine_port}"
                ) from e
            except TimeoutError:
                LOGGER.warning(
                    "Overall timeout reached while waiting for container '%s' to start after %d attempts.",
                    container_name,
                    attempt + 1,
                )
                return web.json_response(
                    {
                        "started": False,
                        "status": last_known_status,
                        "health": last_known_health,
                        "reason": "overall timeout",
                    }
                )
            except web.HTTPServiceUnavailable:
                raise
            except Exception as e:
                LOGGER.exception(
                    "Unexpected error while waiting for container '%s' to start (attempt %d).",
                    container_name,
                    attempt + 1,
                )
                raise web.HTTPInternalServerError(
                    text="An unexpected error occurred while waiting for container start."
                ) from e
            if attempt < max_tries - 1:
                await asyncio.sleep(sleep_interval)
            else:
                LOGGER.info("Max tries reached for container '%s'. Reporting last known state.", container_name)

        LOGGER.warning(
            "Container '%s' did not reach desired 'running' and 'healthy' state within %d attempts.",
            container_name,
            max_tries,
        )
        return web.json_response(
            {
                "started": False,
                "status": last_known_status,
                "health": last_known_health,
                "reason": "timeout after max tries",
            }
        )


async def docker_exapp_remove(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/remove")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = RemoveExAppPayload.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/remove: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    volume_name = payload.exapp_container_volume

    remove_container_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}?force=true"
    remove_volume_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/volumes/{volume_name}"

    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session:
        LOGGER.info("Attempting to remove container '%s' via Docker API at %s", container_name, remove_container_url)
        try:
            async with session.delete(remove_container_url) as resp:
                if resp.status == 204:
                    LOGGER.info("Container '%s' removed successfully.", container_name)
                elif resp.status == 404:
                    LOGGER.info("Container '%s' not found, considering it removed.", container_name)
                elif resp.status == 409:
                    error_text = await resp.text()
                    LOGGER.warning(
                        "Conflict while removing container '%s' (status %s): %s. Assuming it might be gone or handled.",
                        container_name,
                        resp.status,
                        error_text,
                    )
                else:
                    error_text = await resp.text()
                    LOGGER.error(
                        "Error removing container '%s' with Docker API (status %s): %s",
                        container_name,
                        resp.status,
                        error_text,
                    )
                    raise web.HTTPServiceUnavailable(
                        text=f"Error removing container '{container_name}' via Docker Engine: Status {resp.status}"
                    )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error(
                "Could not connect to Docker Engine at %s:%s to remove container: %s",
                DOCKER_API_HOST,
                docker_engine_port,
                e,
            )
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout while trying to remove container '%s' via Docker Engine at %s:%s",
                DOCKER_API_HOST,
                container_name,
                docker_engine_port,
            )
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for container removal") from e
        except web.HTTPServiceUnavailable:
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error while removing container '%s' via Docker API.", container_name)
            raise web.HTTPInternalServerError(text="An unexpected error occurred during container removal.") from e

        if payload.remove_data:
            LOGGER.info("Attempting to remove volume '%s' via Docker API at %s", volume_name, remove_volume_url)
            try:
                async with session.delete(remove_volume_url) as resp:
                    if resp.status == 204:
                        LOGGER.info("Volume '%s' removed successfully.", volume_name)
                    elif resp.status == 404:
                        LOGGER.info("Volume '%s' not found, considering it removed.", volume_name)
                    elif resp.status == 409:
                        error_text = await resp.text()
                        LOGGER.error(
                            "Cannot remove volume '%s' as it is in use (status %s): %s.",
                            volume_name,
                            resp.status,
                            error_text,
                        )
                        raise web.HTTPConflict(text=f"Volume '{volume_name}' is in use and could not be removed.")
                    else:
                        error_text = await resp.text()
                        LOGGER.error(
                            "Error removing volume '%s' with Docker API (status %s): %s",
                            volume_name,
                            resp.status,
                            error_text,
                        )
                        raise web.HTTPServiceUnavailable(
                            text=f"Error removing volume '{volume_name}' via Docker Engine: Status {resp.status}"
                        )
            except aiohttp.ClientConnectorError as e:
                LOGGER.error(
                    "Could not connect to Docker Engine at %s:%s to remove volume: %s",
                    DOCKER_API_HOST,
                    docker_engine_port,
                    e,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Could not connect to Docker Engine on port {docker_engine_port} for volume removal"
                ) from e
            except TimeoutError as e:
                LOGGER.error(
                    "Timeout while trying to remove volume '%s' via Docker Engine at %s:%s",
                    DOCKER_API_HOST,
                    volume_name,
                    docker_engine_port,
                )
                raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for volume removal") from e
            except (web.HTTPServiceUnavailable, web.HTTPConflict):
                raise
            except Exception as e:
                LOGGER.exception("Unexpected error while removing volume '%s' via Docker API.", volume_name)
                raise web.HTTPInternalServerError(text="An unexpected error occurred during volume removal.") from e

    LOGGER.info(
        "ExApp remove operation completed for container '%s' (remove_data=%s).",
        container_name,
        payload.remove_data,
    )
    return web.HTTPNoContent()


async def docker_exapp_image_remove(request: web.Request):
    """Remove a Docker image by reference, returning a structured result.

    Response shape:
        {"deleted": true,  "bytes_freed": <int>}                       # image deleted
        {"deleted": true,  "bytes_freed": 0, "reason": "not_found"}    # image already gone
        {"deleted": false, "reason": "in_use"}                         # Docker refused (409)

    Bytes_freed is best-effort: we inspect the image first to capture its size, but
    if the inspect fails for any reason other than 404 we still try the delete and
    just report bytes_freed=0.
    """
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/image_remove")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = RemoveImagePayload.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/image_remove: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    image_ref = payload.image_ref
    encoded_ref = quote(image_ref, safe="")
    inspect_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/images/{encoded_ref}/json"
    delete_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/images/{encoded_ref}"

    bytes_freed = 0
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=30.0)) as session:
        try:
            async with session.get(inspect_url) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    bytes_freed = int(data.get("Size", 0))
                elif resp.status == 404:
                    LOGGER.info("Image '%s' not found at inspect, treating as already removed.", image_ref)
                    return web.json_response({"deleted": True, "bytes_freed": 0, "reason": "not_found"})
                else:
                    error_text = await resp.text()
                    LOGGER.warning(
                        "Image inspect for '%s' returned status %s: %s. Continuing to delete with bytes_freed=0.",
                        image_ref,
                        resp.status,
                        error_text,
                    )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error(
                "Could not connect to Docker Engine at %s:%s for image inspect: %s",
                DOCKER_API_HOST,
                docker_engine_port,
                e,
            )
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout inspecting image '%s' via Docker Engine at %s:%s",
                image_ref,
                DOCKER_API_HOST,
                docker_engine_port,
            )
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for image inspect") from e
        except Exception as e:
            LOGGER.warning(
                "Unable to inspect image '%s' before removal: %s. Continuing with bytes_freed=0.",
                image_ref,
                e,
            )

        try:
            async with session.delete(delete_url) as resp:
                if resp.status == 200:
                    try:
                        ops = await resp.json()
                    except (json.JSONDecodeError, aiohttp.ContentTypeError):
                        ops = []
                    actually_deleted = isinstance(ops, list) and any(
                        isinstance(op, dict) and "Deleted" in op for op in ops
                    )
                    actual_bytes_freed = bytes_freed if actually_deleted else 0
                    LOGGER.info(
                        "Image '%s' removed (deleted=%s, bytes_freed=%d).",
                        image_ref,
                        actually_deleted,
                        actual_bytes_freed,
                    )
                    return web.json_response({"deleted": True, "bytes_freed": actual_bytes_freed})
                if resp.status == 404:
                    LOGGER.info("Image '%s' already gone at delete time.", image_ref)
                    return web.json_response({"deleted": True, "bytes_freed": 0, "reason": "not_found"})
                if resp.status == 409:
                    error_text = await resp.text()
                    LOGGER.info("Image '%s' is in use, skipping removal: %s", image_ref, error_text)
                    return web.json_response({"deleted": False, "reason": "in_use"})
                error_text = await resp.text()
                LOGGER.error(
                    "Error removing image '%s' via Docker API (status %s): %s",
                    image_ref,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(
                    text=f"Error removing image '{image_ref}' via Docker Engine: Status {resp.status}"
                )
        except aiohttp.ClientConnectorError as e:
            LOGGER.error(
                "Could not connect to Docker Engine at %s:%s to remove image: %s",
                DOCKER_API_HOST,
                docker_engine_port,
                e,
            )
            raise web.HTTPServiceUnavailable(
                text=f"Could not connect to Docker Engine on port {docker_engine_port}"
            ) from e
        except TimeoutError as e:
            LOGGER.error(
                "Timeout removing image '%s' via Docker Engine at %s:%s",
                image_ref,
                DOCKER_API_HOST,
                docker_engine_port,
            )
            raise web.HTTPGatewayTimeout(text="Timeout communicating with Docker Engine for image removal") from e
        except web.HTTPServiceUnavailable:
            raise
        except Exception as e:
            LOGGER.exception("Unexpected error while removing image '%s' via Docker API.", image_ref)
            raise web.HTTPInternalServerError(text="An unexpected error occurred during image removal.") from e


async def docker_exapp_install_certificates(request: web.Request):
    docker_engine_port = get_docker_engine_port(request)
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        LOGGER.warning("Invalid JSON body received for /docker/exapp/install_certificates")
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        payload = InstallCertificatesPayload.model_validate(payload_dict)
    except ValidationError as e:
        reason = describe_validation_error(e)
        LOGGER.warning("Payload validation error for /docker/exapp/install_certificates: %s", reason)
        raise web.HTTPBadRequest(text=f"Payload validation error: {reason}") from None

    container_name = payload.exapp_container_name
    async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=120.0)) as session:
        original_state_running = False
        try:
            inspect_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/json"
            async with session.get(inspect_url) as resp_inspect:
                if resp_inspect.status == 200:
                    inspect_data = await resp_inspect.json()
                    original_state_running = inspect_data.get("State", {}).get("Running", False)
                elif resp_inspect.status == 404:
                    LOGGER.error("Container '%s' not found for certificate installation.", container_name)
                    raise web.HTTPNotFound(text=f"Container '{container_name}' not found.")
                else:
                    error_text = await resp_inspect.text()
                    LOGGER.error(
                        "Failed to inspect container '%s' (status %s): %s",
                        container_name,
                        resp_inspect.status,
                        error_text,
                    )
                    raise web.HTTPServiceUnavailable(text=f"Failed to inspect container: Status {resp_inspect.status}")

            if not original_state_running:
                LOGGER.info(
                    "Container '%s' is not running. Attempting to start for certificate installation.", container_name
                )
                start_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/start"
                async with session.post(start_url) as resp_start:
                    if resp_start.status not in (204, 304):
                        error_text = await resp_start.text()
                        LOGGER.error(
                            "Failed to start container '%s' for cert install (status %s): %s",
                            container_name,
                            resp_start.status,
                            error_text,
                        )
                        raise web.HTTPServiceUnavailable(text=f"Failed to start container: Status {resp_start.status}")
                    LOGGER.info("Container '%s' started/is running for certificate installation.", container_name)

            exit_code, os_info_content = await _execute_command_in_container_simplified(
                session, docker_engine_port, container_name, ["cat", "/etc/os-release"]
            )
            if exit_code != 0:
                LOGGER.error(
                    "Failed to get OS info from container '%s'. Exit code: %s, Raw Output: %s",
                    container_name,
                    exit_code,
                    os_info_content,
                )
                raise web.HTTPInternalServerError(
                    text=f"Failed to get OS info. Exit: {exit_code}. Output: {os_info_content[:200]}"
                )
            LOGGER.info("OS Info for container '%s':\n%s", container_name, os_info_content.strip())

            if payload.system_certs_bundle:
                await _install_system_certificates(
                    session, docker_engine_port, container_name, payload.system_certs_bundle, os_info_content
                )
            else:
                LOGGER.info(
                    "No system_certs_bundle provided for container '%s'. Skipping system cert installation.",
                    container_name,
                )

            if payload.install_frp_certs:
                await _install_frp_certificates(session, docker_engine_port, container_name)
            else:
                LOGGER.info(
                    "install_frp_certs is false. Skipping FRP cert installation for container '%s'.", container_name
                )

            return web.HTTPNoContent()

        except (web.HTTPException, aiohttp.ClientError) as e:
            LOGGER.error("Error during certificate installation for '%s': %s", container_name, e)
            raise
        except Exception as e:
            LOGGER.exception("Unexpected fatal error during certificate installation for '%s'", container_name)
            raise web.HTTPInternalServerError(text=f"Unexpected error during certificate installation: {e}") from e
        finally:
            if not original_state_running:
                LOGGER.info(
                    "Attempting to stop container '%s' as it was started for certificate installation.", container_name
                )
                stop_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/stop"
                try:
                    async with session.post(stop_url) as resp_stop:
                        if resp_stop.status not in (204, 304, 404):
                            error_text = await resp_stop.text()
                            LOGGER.warning(
                                "Failed to stop container '%s' after cert install (status %s): %s",
                                container_name,
                                resp_stop.status,
                                error_text,
                            )
                        else:
                            LOGGER.info(
                                "Container '%s' stopped or was already stopped/gone after cert install.", container_name
                            )
                except Exception as e_stop:
                    LOGGER.error("Error stopping container '%s' in finally block: %s", container_name, e_stop)


async def _install_system_certificates(
    session: aiohttp.ClientSession,
    docker_engine_port: int,
    container_name: str,
    system_certs_bundle: str,
    os_info_content: str,
) -> None:
    target_cert_dir = _get_target_cert_dir(os_info_content)
    if not target_cert_dir:
        LOGGER.warning(
            "OS in container '%s' not supported for sys cert installation, or bundle empty. Skipping.",
            container_name,
        )
        return

    LOGGER.info("Target system cert directory for container '%s': %s", container_name, target_cert_dir)
    exit_code, raw_output = await _execute_command_in_container_simplified(
        session, docker_engine_port, container_name, ["mkdir", "-p", target_cert_dir]
    )
    if exit_code != 0:
        LOGGER.error(
            "Failed to create cert dir '%s' in container '%s'. Exit: %s, Raw Output: %s",
            target_cert_dir,
            container_name,
            exit_code,
            raw_output,
        )
        raise web.HTTPInternalServerError(
            text=f"Failed to create cert directory. Exit: {exit_code}. Output: {raw_output[:200]}"
        )

    certs_to_install = {}
    parsed_certs = _parse_certs_from_bundle(system_certs_bundle)
    for i, cert_content in enumerate(parsed_certs):
        cert_filename = f"custom_ca_cert_{i}.crt"
        certs_to_install[os.path.join(target_cert_dir.lstrip("/"), cert_filename)] = cert_content

    if not certs_to_install:
        LOGGER.info(
            "No individual certificates parsed from system_certs_bundle for container '%s'.",
            container_name,
        )
        return

    tar_bytes = _create_tar_archive_in_memory(certs_to_install)
    await _put_archive_to_container(session, docker_engine_port, container_name, "/", tar_bytes)
    LOGGER.info(
        "Installed %d system CA certificates into '%s' in container '%s'.",
        len(parsed_certs),
        target_cert_dir,
        container_name,
    )

    update_cmd_list = _get_certificate_update_command(os_info_content)
    if update_cmd_list:
        LOGGER.info("Running certificate update command: %s", " ".join(update_cmd_list))
        exit_code, raw_output = await _execute_command_in_container_simplified(
            session, docker_engine_port, container_name, update_cmd_list
        )
        if exit_code != 0:
            LOGGER.error(
                "Certificate update command failed in container '%s'. Exit: %s, Raw Output: %s",
                container_name,
                exit_code,
                raw_output,
            )
        else:
            LOGGER.info("Certificate update command successful. Raw Output: %s", raw_output.strip())
    else:
        LOGGER.warning("No certificate update command found for OS in container '%s'.", container_name)


async def _install_frp_certificates(
    session: aiohttp.ClientSession, docker_engine_port: int, container_name: str
) -> None:
    frp_cert_dir_on_harp = "/certs/frp"
    frp_target_dir_in_container = "/certs/frp"

    frp_files_to_read = {
        "ca.crt": os.path.join(frp_cert_dir_on_harp, "ca.crt"),
        "client.crt": os.path.join(frp_cert_dir_on_harp, "client.crt"),
        "client.key": os.path.join(frp_cert_dir_on_harp, "client.key"),
    }
    frp_certs_content = {}
    all_frp_files_exist = True
    for name, path_on_harp in frp_files_to_read.items():
        if os.path.exists(path_on_harp):
            try:
                with open(path_on_harp, encoding="utf-8") as f:
                    frp_certs_content[name] = f.read()
            except Exception as e_read:
                LOGGER.error("Failed to read FRP cert file '%s' from HaRP agent: %s", path_on_harp, e_read)
                all_frp_files_exist = False
                break
        else:
            LOGGER.warning(
                "FRP certificate file '%s' not found on HaRP agent. Skipping FRP cert installation.",
                path_on_harp,
            )
            all_frp_files_exist = False
            break

    if all_frp_files_exist and frp_certs_content:
        LOGGER.info(
            "Installing FRP certificates from HaRP agent into '%s' in container '%s'.",
            frp_target_dir_in_container,
            container_name,
        )
        exit_code, raw_output = await _execute_command_in_container_simplified(
            session, docker_engine_port, container_name, ["mkdir", "-p", frp_target_dir_in_container]
        )
        if exit_code != 0:
            LOGGER.error(
                "Failed to create FRP cert dir '%s' in container '%s'. Exit: %s, Raw Output: %s",
                frp_target_dir_in_container,
                container_name,
                exit_code,
                raw_output,
            )
            raise web.HTTPInternalServerError(
                text=f"Failed to create FRP cert directory. Exit: {exit_code}. Output: {raw_output[:200]}"
            )

        frp_files_for_tar = {
            os.path.join(frp_target_dir_in_container.lstrip("/"), "ca.crt"): frp_certs_content["ca.crt"],
            os.path.join(frp_target_dir_in_container.lstrip("/"), "client.crt"): frp_certs_content["client.crt"],
            os.path.join(frp_target_dir_in_container.lstrip("/"), "client.key"): frp_certs_content["client.key"],
        }
        tar_bytes_frp = _create_tar_archive_in_memory(frp_files_for_tar)
        await _put_archive_to_container(session, docker_engine_port, container_name, "/", tar_bytes_frp)
        LOGGER.info("FRP certificates installed successfully into '%s'.", frp_target_dir_in_container)
    elif not all_frp_files_exist:
        LOGGER.info(
            "One or more FRP cert files missing, skipping FRP installation for container '%s'",
            container_name,
        )
    else:
        LOGGER.warning(
            "FRP cert content is empty. Skipping FRP installation for '%s'",
            container_name,
        )


def _create_tar_archive_in_memory(files_to_add: dict[str, str]) -> bytes:
    tar_stream = io.BytesIO()
    with tarfile.open(fileobj=tar_stream, mode="w") as tar:
        for path_in_archive, content_str in files_to_add.items():
            content_bytes = content_str.encode("utf-8")
            tarinfo = tarfile.TarInfo(name=path_in_archive)
            tarinfo.size = len(content_bytes)
            tarinfo.mtime = int(time.time())
            tar.addfile(tarinfo, io.BytesIO(content_bytes))
    return tar_stream.getvalue()


async def _put_archive_to_container(
    session: aiohttp.ClientSession, docker_engine_port: int, container_name: str, path: str, data: bytes
):
    upload_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_name}/archive?path={path}"
    headers = {"Content-Type": "application/x-tar"}
    try:
        async with session.put(upload_url, data=data, headers=headers) as resp:
            if resp.status != 200:
                error_text = await resp.text()
                LOGGER.error(
                    "Failed to put archive to container '%s' at path '%s' (status %s): %s",
                    container_name,
                    path,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(text=f"Failed to upload archive: Status {resp.status}")
            LOGGER.info("Successfully put archive to container '%s' at path '%s'", container_name, path)
    except aiohttp.ClientError as e:
        LOGGER.error("Client error putting archive to container '%s': %s", container_name, e)
        raise web.HTTPServiceUnavailable(text=f"Client error uploading archive: {e}") from e
    except Exception as e:
        LOGGER.exception("Unexpected error putting archive to container '%s'", container_name)
        raise web.HTTPInternalServerError(text=f"Unexpected error uploading archive: {e}") from e


def _parse_certs_from_bundle(bundle_content: str) -> list[str]:
    """Parses individual PEM certificates from a bundle string."""
    return re.findall(r"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----", bundle_content, re.DOTALL)


async def _execute_command_in_container_simplified(
    session: aiohttp.ClientSession, docker_engine_port: int, container_id_or_name: str, cmd: list[str]
) -> tuple[int, str]:
    """Executes a command in a running container and returns (exit_code, raw_output_str)."""
    exec_create_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/containers/{container_id_or_name}/exec"
    exec_create_payload = {
        "AttachStdout": True,
        "AttachStderr": True,
        "Cmd": cmd,
    }
    exec_id = None
    raw_output_str = ""
    exit_code = -1  # Default to error

    try:
        async with session.post(exec_create_url, json=exec_create_payload) as resp:
            if resp.status != 201:
                error_text = await resp.text()
                LOGGER.error(
                    "Failed to create exec instance for command '%s' in container '%s' (status %s): %s",
                    " ".join(cmd),
                    container_id_or_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(text=f"Failed to create exec for command: {error_text}")
            exec_id_data = await resp.json()
            exec_id = exec_id_data.get("Id")

        if not exec_id:
            raise web.HTTPInternalServerError(text="Exec ID not found after creation.")

        exec_start_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/exec/{exec_id}/start"
        exec_start_payload = {"Detach": False, "Tty": False}  # Tty=False gives raw stream

        async with session.post(exec_start_url, json=exec_start_payload) as resp:
            if resp.status != 200:
                error_text = await resp.text()
                LOGGER.error(
                    "Failed to start exec instance '%s' for command '%s' in container '%s' (status %s): %s",
                    exec_id,
                    " ".join(cmd),
                    container_id_or_name,
                    resp.status,
                    error_text,
                )
                raise web.HTTPServiceUnavailable(text=f"Failed to start exec: {error_text}")
            raw_output_bytes = await resp.read()
            raw_output_str = raw_output_bytes.decode(errors="ignore")

        exec_inspect_url = f"http://{DOCKER_API_HOST}:{docker_engine_port}/exec/{exec_id}/json"
        async with session.get(exec_inspect_url) as resp_inspect:
            if resp_inspect.status != 200:
                error_text = await resp_inspect.text()
                LOGGER.error(
                    "Failed to inspect exec instance '%s' (status %s): %s. Output was: %s",
                    exec_id,
                    resp_inspect.status,
                    error_text,
                    raw_output_str,
                )
            else:
                exec_info = await resp_inspect.json()
                ret_code = exec_info.get("ExitCode")
                if ret_code is None:  # Should not happen if process exited
                    LOGGER.warning(
                        "Exec inspect for '%s' did not contain ExitCode. Output was: %s", exec_id, raw_output_str
                    )
                else:
                    exit_code = ret_code

            return exit_code, raw_output_str

    except aiohttp.ClientError as e:
        LOGGER.error(
            "Client error during exec command '%s' in container '%s': %s", " ".join(cmd), container_id_or_name, e
        )
        raise web.HTTPServiceUnavailable(text=f"Client error during exec: {e}") from e
    except Exception as e:
        LOGGER.exception(
            "Unexpected error during exec command '%s' in container '%s'", " ".join(cmd), container_id_or_name
        )
        raise web.HTTPInternalServerError(text=f"Unexpected error during exec: {e}") from e


def _get_target_cert_dir(os_info_content: str | None) -> str | None:
    if not os_info_content:
        LOGGER.warning("OS info content is empty, cannot determine target cert directory.")
        return None
    os_info_lower = os_info_content.lower()
    if "alpine" in os_info_lower:
        return "/usr/local/share/ca-certificates"
    if "debian" in os_info_lower or "ubuntu" in os_info_lower:
        return "/usr/local/share/ca-certificates"
    if (
        "centos" in os_info_lower
        or "almalinux" in os_info_lower
        or "rhel" in os_info_lower
        or "fedora" in os_info_lower
    ):
        return "/etc/pki/ca-trust/source/anchors"
    LOGGER.warning(
        "Unsupported OS for SSL certificate installation: %s",
        os_info_content.splitlines()[0] if os_info_content else "Unknown",
    )
    return None


def _get_certificate_update_command(os_info_content: str | None) -> list[str] | None:
    if not os_info_content:
        LOGGER.warning("OS info content is empty, cannot determine certificate update command.")
        return None
    os_info_lower = os_info_content.lower()
    if "alpine" in os_info_lower:
        return ["update-ca-certificates"]
    if "debian" in os_info_lower or "ubuntu" in os_info_lower:
        return ["update-ca-certificates"]
    if (
        "centos" in os_info_lower
        or "almalinux" in os_info_lower
        or "rhel" in os_info_lower
        or "fedora" in os_info_lower
    ):
        return ["update-ca-trust", "extract"]
    return None


###############################################################################
# Kubernetes helpers functions
###############################################################################


async def _parse_json_payload(request: web.Request, model: type[BaseModel]) -> Any:
    """Parse JSON body and validate against a Pydantic model."""
    try:
        payload_dict = await request.json()
    except json.JSONDecodeError:
        raise web.HTTPBadRequest(text="Invalid JSON body") from None
    try:
        return model.model_validate(payload_dict)
    except ValidationError as e:
        raise web.HTTPBadRequest(text=f"Payload validation error: {describe_validation_error(e)}") from None


def _k8s_error_msg(data: dict[str, Any] | None, text: str) -> str:
    """Extract a human-readable message from a K8s API response."""
    if isinstance(data, dict):
        return (data or {}).get("message", text)
    return text


def _get_k8s_token() -> str | None:
    """Get the Kubernetes bearer token.

    If HP_K8S_BEARER_TOKEN env var is set, returns the static value.
    Otherwise re-reads the token file on every call so that kubelet-rotated
    projected service account tokens (default ~1h lifetime) are picked up automatically.
    """
    if K8S_TOKEN_ENV:
        return K8S_TOKEN_ENV.strip()
    if K8S_TOKEN_FILE and os.path.exists(K8S_TOKEN_FILE):
        try:
            with open(K8S_TOKEN_FILE, encoding="utf-8") as f:
                token = f.read().strip()
                if token:
                    return token
        except Exception as e:
            LOGGER.error("Failed to read Kubernetes token file '%s': %s", K8S_TOKEN_FILE, e)
    LOGGER.error(
        "Kubernetes bearer token not found. "
        "Set HP_K8S_BEARER_TOKEN or HP_K8S_BEARER_TOKEN_FILE when HP_K8S_ENABLED=true."
    )
    return None


def _get_k8s_ssl_context() -> ssl.SSLContext | bool:
    """Return SSL context (or False to disable verification) for K8s API."""
    if not K8S_API_SERVER or not K8S_API_SERVER.startswith("https"):
        return False
    if not K8S_VERIFY_SSL:
        return False
    try:
        cafile = K8S_CA_FILE if K8S_CA_FILE and os.path.exists(K8S_CA_FILE) else None
        return ssl.create_default_context(cafile=cafile)
    except Exception as e:
        LOGGER.warning("Failed to create SSL context for Kubernetes API: %s", e)
        return ssl.create_default_context()


def _ensure_k8s_configured() -> None:
    if not K8S_ENABLED:
        LOGGER.error("Kubernetes backend requested but HP_K8S_ENABLED is not true.")
        raise web.HTTPServiceUnavailable(text="Kubernetes backend is disabled in HaRP.")
    if not K8S_API_SERVER:
        LOGGER.error("Kubernetes backend requested but HP_K8S_API_SERVER is not configured.")
        raise web.HTTPServiceUnavailable(text="Kubernetes API server is not configured.")
    if not _get_k8s_token():
        raise web.HTTPServiceUnavailable(text="Kubernetes token is not configured.")


def _get_k8s_session() -> aiohttp.ClientSession:
    """Return a reusable aiohttp session for the Kubernetes API.

    Lazily creates a module-level singleton with connection pooling.
    The session is reused across all K8s API calls, avoiding the overhead
    of a new TCP+TLS handshake per request.
    """
    global _k8s_session
    if _k8s_session is None or _k8s_session.closed:
        ssl_ctx = _get_k8s_ssl_context()
        connector = aiohttp.TCPConnector(ssl=ssl_ctx)
        _k8s_session = aiohttp.ClientSession(
            timeout=K8S_HTTP_TIMEOUT,
            connector=connector,
        )
    return _k8s_session


async def _close_k8s_session(_app: web.Application = None) -> None:
    """Close the K8s session on shutdown to avoid ResourceWarning."""
    global _k8s_session
    if _k8s_session is not None and not _k8s_session.closed:
        await _k8s_session.close()
        _k8s_session = None


async def _k8s_request(
    method: str,
    path: str,
    *,
    query: dict[str, str] | None = None,
    json_body: Any | None = None,
    content_type: str | None = None,
) -> tuple[int, dict[str, Any] | None, str]:
    """Low-level helper for talking to the Kubernetes API."""
    _ensure_k8s_configured()
    token = _get_k8s_token()
    headers: dict[str, str] = {
        "Authorization": f"Bearer {token}",
        "Accept": "application/json",
    }
    if json_body is not None:
        headers["Content-Type"] = content_type or "application/json"

    url = f"{K8S_API_SERVER}{path}"
    session = _get_k8s_session()

    try:
        async with session.request(method.upper(), url, headers=headers, params=query, json=json_body) as resp:
            text = await resp.text()
            data: dict[str, Any] | None = None
            if "application/json" in resp.headers.get("Content-Type", "") and text:
                try:
                    data = json.loads(text)
                except json.JSONDecodeError:
                    LOGGER.warning("Failed to parse JSON from Kubernetes API %s %s: %s", method, url, text[:200])
            return resp.status, data, text
    except aiohttp.ClientError as e:
        LOGGER.error("Error communicating with Kubernetes API (%s %s): %s", method, url, e)
        raise web.HTTPServiceUnavailable(text="Error communicating with Kubernetes API") from e


async def _k8s_probe() -> tuple[bool, str]:
    """Check whether the Kubernetes API server answers; returns ``(reachable, error)``.

    Used by ``/info`` only. Unlike ``_k8s_request`` it fails fast (K8S_PROBE_TIMEOUT) and keeps the reason, so the
    AppAPI daemon check can tell a broken cluster connection from a broken HaRP one.
    """
    try:
        _ensure_k8s_configured()
    except web.HTTPServiceUnavailable as e:
        return False, e.text or "Kubernetes backend is not configured."
    url = f"{K8S_API_SERVER}/api"
    headers = {"Authorization": f"Bearer {_get_k8s_token()}", "Accept": "application/json"}
    try:
        session = _get_k8s_session()  # may raise on a broken HP_K8S_CA_FILE
        async with session.get(url, headers=headers, timeout=K8S_PROBE_TIMEOUT) as resp:
            # Read the body inside the context, like `_k8s_request` does: it carries the Kubernetes `Status.message`
            # that says what actually went wrong, and leaving it unread makes aiohttp drop the pooled connection
            # whenever the body does not arrive with the headers.
            body = (await resp.text())[:200].strip()
            if resp.status == 200:
                return True, ""
            if resp.status == 401:
                error = (
                    "Kubernetes API server answered HTTP 401; check the bearer token "
                    "(HP_K8S_BEARER_TOKEN or HP_K8S_BEARER_TOKEN_FILE)."
                )
            elif resp.status == 403:
                # 401 is authentication, 403 is authorization: a valid token can still be denied by RBAC.
                error = "Kubernetes API server answered HTTP 403; check the bearer token's RBAC permissions."
            else:
                error = f"Kubernetes API server answered HTTP {resp.status}."
            if body:
                error = f"{error} Response: {body}"
    except TimeoutError:
        error = f"Kubernetes API server did not answer within {K8S_PROBE_TIMEOUT.total:g}s (DNS, connect or request)."
    except aiohttp.ClientSSLError as e:
        error = f"TLS error connecting to the Kubernetes API server: {e}"
    except aiohttp.ClientError as e:
        error = f"Cannot connect to the Kubernetes API server: {e}"
    except Exception as e:  # `/info` must always answer
        error = f"Kubernetes API probe failed: {e}"
    LOGGER.warning("Kubernetes API probe (%s) failed: %s", url, error)
    return False, error


def _k8s_parse_env(env_list: list[str]) -> list[dict[str, str]]:
    """Convert ['KEY=VALUE', ...] to Kubernetes env entries."""
    result: list[dict[str, str]] = []
    for raw in env_list:
        if not raw:
            continue
        if "=" in raw:
            name, value = raw.split("=", 1)
            result.append({"name": name, "value": value})
        else:
            result.append({"name": raw, "value": ""})  # No '=', keep name and use empty value
    return result


def _k8s_build_resources(resource_limits: dict[str, Any], compute_device: str = "cpu") -> dict[str, Any]:
    """Convert resource_limits dict to K8s resources spec."""
    if not resource_limits and compute_device == "cpu":
        return {}
    limits: dict[str, str] = {}
    requests: dict[str, str] = {}

    # Memory
    mem_val = resource_limits.get("memory")
    mem_str: str | None = None
    if isinstance(mem_val, int) and mem_val > 0:
        # bytes -> Mi (ceil)
        mem_mi = (mem_val + (1024 * 1024 - 1)) // (1024 * 1024)
        mem_str = f"{mem_mi}Mi"
    elif isinstance(mem_val, str) and mem_val:
        mem_str = mem_val  # Already in K8s units, e.g. "512Mi"

    if mem_str:
        limits["memory"] = mem_str
        requests["memory"] = mem_str  # conservative: same as limit

    # CPU
    cpu_str: str | None = None
    nano_cpus = resource_limits.get("nanoCPUs")
    if isinstance(nano_cpus, int) and nano_cpus > 0:
        milli = (nano_cpus * 1000 + 1_000_000_000 - 1) // 1_000_000_000  # 1e9 nanoCPUs = 1 CPU => millicores
        milli = max(1, milli)
        cpu_str = f"{milli}m"
    else:
        cpu_val = resource_limits.get("cpu")
        if isinstance(cpu_val, str) and cpu_val:
            cpu_str = cpu_val  # Already in K8s units, e.g. "500m"

    if cpu_str:
        limits["cpu"] = cpu_str
        requests["cpu"] = cpu_str

    if compute_device == "cuda":
        limits["nvidia.com/gpu"] = "1"
    elif compute_device == "rocm":
        limits["amd.com/gpu"] = "1"

    res: dict[str, Any] = {}
    if limits:
        res["limits"] = limits
    if requests:
        res["requests"] = requests
    return res


def _k8s_parse_host_aliases() -> list[dict[str, Any]]:
    """Parse HP_K8S_HOST_ALIASES into K8s hostAliases format."""
    if not K8S_HOST_ALIASES_RAW.strip():
        return []

    ip_to_hosts: dict[str, list[str]] = {}
    for entry in K8S_HOST_ALIASES_RAW.split(","):
        entry = entry.strip()
        if not entry or ":" not in entry:
            continue
        # A host name has no ':', so the address is everything after the first one
        hostname, ip_addr = entry.split(":", 1)
        hostname = hostname.strip()
        ip_addr = ip_addr.strip().removeprefix("[").removesuffix("]")
        if hostname and ip_addr:
            ip_to_hosts.setdefault(ip_addr, []).append(hostname)

    return [{"ip": ip, "hostnames": hosts} for ip, hosts in ip_to_hosts.items()]


def _k8s_build_deployment_manifest(payload: CreateExAppPayload, replicas: int) -> dict[str, Any]:
    """Build a Deployment manifest from CreateExAppPayload."""
    deployment_name = payload.exapp_k8s_name
    pvc_name = payload.exapp_k8s_volume_name

    # Base ExApp name (without role suffix) - used for grouping multi-role Deployments.
    base_exapp_name = _sanitize_k8s_name(
        f"nc_app_{payload.instance_id}_{payload.name}" if payload.instance_id else f"nc_app_{payload.name}"
    )

    labels = {
        "app": deployment_name,
        "app.kubernetes.io/name": deployment_name,
        "app.kubernetes.io/part-of": base_exapp_name,
        "app.kubernetes.io/component": "exapp",
    }
    if payload.instance_id:
        labels["app.kubernetes.io/instance"] = payload.instance_id
    if payload.role_suffix:
        labels["app.kubernetes.io/role"] = _sanitize_k8s_name(payload.role_suffix)

    container: dict[str, Any] = {
        "name": "app",
        "image": payload.image_id,
        "imagePullPolicy": payload.image_pull_policy,
        "env": _k8s_parse_env(payload.environment_variables),
    }

    resources = _k8s_build_resources(payload.resource_limits, payload.compute_device)
    if resources:
        container["resources"] = resources

    # Main data volume
    volumes = [
        {
            "name": "data",
            "persistentVolumeClaim": {"claimName": pvc_name},
        }
    ]
    volume_mounts = [
        {
            "name": "data",
            "mountPath": f"/{payload.exapp_container_volume}",
        }
    ]

    if payload.mount_points:
        LOGGER.warning(
            "Kubernetes backend currently ignores additional mount_points for ExApp '%s'.",
            deployment_name,
        )

    container["volumeMounts"] = volume_mounts

    pod_spec: dict[str, Any] = {"containers": [container], "volumes": volumes}
    host_aliases = _k8s_parse_host_aliases()
    if host_aliases:
        pod_spec["hostAliases"] = host_aliases

    return {
        "apiVersion": "apps/v1",
        "kind": "Deployment",
        "metadata": {"name": deployment_name, "labels": labels},
        "spec": {
            "replicas": replicas,
            "selector": {"matchLabels": {"app": deployment_name}},
            "template": {"metadata": {"labels": labels}, "spec": pod_spec},
        },
    }


def _k8s_build_service_manifest(
    payload: ExposeExAppPayload, service_type: Literal["NodePort", "ClusterIP", "LoadBalancer"]
) -> dict[str, Any]:
    service_name = payload.exapp_k8s_name
    base_exapp_name = _sanitize_k8s_name(
        f"nc_app_{payload.instance_id}_{payload.name}" if payload.instance_id else f"nc_app_{payload.name}"
    )
    labels = {
        "app": service_name,
        "app.kubernetes.io/name": service_name,
        "app.kubernetes.io/part-of": base_exapp_name,
        "app.kubernetes.io/component": "exapp",
        "app.kubernetes.io/managed-by": "harp",
        **(payload.service_labels or {}),
    }
    if payload.instance_id:
        labels.setdefault("app.kubernetes.io/instance", payload.instance_id)
    if payload.role_suffix:
        labels["app.kubernetes.io/role"] = _sanitize_k8s_name(payload.role_suffix)

    metadata: dict[str, Any] = {"name": service_name, "labels": labels}
    if payload.service_annotations:
        metadata["annotations"] = payload.service_annotations

    svc_port = payload.service_port or payload.port
    port_entry: dict[str, Any] = {
        "name": "http",
        "port": svc_port,
        "targetPort": payload.port,
    }
    if service_type == "NodePort" and payload.node_port:
        port_entry["nodePort"] = payload.node_port

    spec: dict[str, Any] = {
        "type": service_type,
        "selector": {"app": service_name},
        "ports": [port_entry],
    }

    if payload.external_traffic_policy and service_type in ("NodePort", "LoadBalancer"):
        spec["externalTrafficPolicy"] = payload.external_traffic_policy

    if service_type == "LoadBalancer" and payload.load_balancer_ip:
        spec["loadBalancerIP"] = payload.load_balancer_ip

    return {
        "apiVersion": "v1",
        "kind": "Service",
        "metadata": metadata,
        "spec": spec,
    }


def _extract_manual_upstream(annotations: dict[str, str], deploy_name: str) -> tuple[str, int] | None:
    """Extract manual upstream host/port from Deployment annotations."""
    if annotations.get("appapi.nextcloud.com/expose-type") != "manual":
        return None

    host = annotations.get("appapi.nextcloud.com/upstream-host")
    port_str = annotations.get("appapi.nextcloud.com/upstream-port")
    if not host or not port_str:
        LOGGER.warning(
            "Deployment '%s' has manual expose-type annotation but missing upstream-host/port.",
            deploy_name,
        )
        return None

    try:
        port = int(port_str)
    except ValueError:
        LOGGER.warning("Invalid upstream-port annotation on Deployment '%s': %s", deploy_name, port_str)
        return None
    return (host, port)


async def _k8s_resolve_manual_upstream(deployment_name: str, app_name: str) -> tuple[str, int] | None:
    """Read manual-expose upstream config from Deployment annotations.

    For expose_type=manual, no K8s Service is created.  Instead, the upstream
    host and port are persisted as annotations on the Deployment during the
    expose call.  This function reads those annotations to re-discover the
    upstream after a cache eviction or HaRP restart.

    Returns (host, port) if annotations are present, None otherwise (e.g. a
    Docker-only ExApp that has no K8s Deployment at all).
    """
    # Try the base Deployment name first (single-role ExApps).
    status, data, _ = await _k8s_request(
        "GET",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deployment_name}",
    )
    if status == 200 and isinstance(data, dict):
        annotations = (data.get("metadata") or {}).get("annotations") or {}
        result = _extract_manual_upstream(annotations, deployment_name)
        if result:
            LOGGER.info("Resolved manual upstream for '%s' from Deployment annotations: %s:%d", app_name, *result)
            return result
    elif status != 404:
        LOGGER.warning(
            "K8s API error fetching Deployment '%s' for manual upstream (status %s)",
            deployment_name, status,
        )
        return None

    # Fallback: search for multi-role Deployments by label (the exposed role
    # carries the manual annotations).
    base_k8s_name = _sanitize_k8s_name(f"nc_app_{app_name}")
    label_selector = f"app.kubernetes.io/part-of={base_k8s_name}"
    status, deploy_list, _ = await _k8s_request(
        "GET",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments",
        query={"labelSelector": label_selector},
    )
    if status == 200 and isinstance(deploy_list, dict):
        for item in deploy_list.get("items") or []:
            annotations = (item.get("metadata") or {}).get("annotations") or {}
            d_name = (item.get("metadata") or {}).get("name", "")
            result = _extract_manual_upstream(annotations, d_name)
            if result:
                LOGGER.info("Resolved manual upstream for '%s' from role Deployment '%s': %s:%d", app_name, d_name, *result)
                return result

    return None


async def _k8s_resolve_exapp_upstream(app_name: str) -> tuple[str, int] | None:
    """Look up K8s Service for an ExApp, return (host, port) or None.

    Returns None when K8s is disabled or no Service exists (Docker ExApp).
    Raises web.HTTPServiceUnavailable when the K8s API returned an error
    (e.g. 403 token expired, 500 server error) so the caller knows not to
    cache the result with stale NC-provided host/port.
    Connection-level errors already raise via _k8s_request.
    """
    if not K8S_ENABLED or not K8S_API_SERVER or not _get_k8s_token():
        return None

    # Not validated: the lookup must keep working for ExApps whose id predates the stricter name rules.
    exapp = ExAppName.model_construct(name=app_name)

    svc: dict[str, Any] | None = None
    got_k8s_error = False

    # Try the base Service name first (single-role / legacy ExApps).
    service_name = exapp.exapp_k8s_name
    status, data, _ = await _k8s_request(
        "GET",
        f"/api/v1/namespaces/{K8S_NAMESPACE}/services/{service_name}",
    )
    if status == 200 and isinstance(data, dict):
        svc = data
    elif status != 404:
        got_k8s_error = True

    # Fallback: search for any Service labelled for this ExApp (multi-role).
    if svc is None:
        base_k8s_name = _sanitize_k8s_name(f"nc_app_{app_name}")
        label_selector = f"app.kubernetes.io/part-of={base_k8s_name}"
        status, svc_list, _ = await _k8s_request(
            "GET",
            f"/api/v1/namespaces/{K8S_NAMESPACE}/services",
            query={"labelSelector": label_selector},
        )
        if status == 200 and isinstance(svc_list, dict):
            items = svc_list.get("items") or []
            if items:
                svc = items[0]
                service_name = (svc.get("metadata") or {}).get("name", service_name)
        elif status != 200:
            got_k8s_error = True

    if svc is None:
        if got_k8s_error:
            LOGGER.warning(
                "K8s API returned error during Service lookup for '%s'; "
                "will not cache stale record",
                app_name,
            )
            raise web.HTTPServiceUnavailable(
                text=f"K8s API error during Service lookup for '{app_name}'"
            )

        # No Service found — check if this is a manual-expose ExApp by reading
        # the upstream config stored as annotations on the Deployment.
        return await _k8s_resolve_manual_upstream(exapp.exapp_k8s_name, app_name)

    svc_type = (svc.get("spec") or {}).get("type", "ClusterIP")
    try:
        if svc_type == "NodePort":
            port = _k8s_extract_nodeport(svc)
            host = await _k8s_pick_node_address(preferred_type="InternalIP")
            return (host, port)
        if svc_type == "ClusterIP":
            port = _k8s_extract_service_port(svc)
            # Use the actual ClusterIP address instead of DNS name.  DNS names
            # like "svc.namespace.svc" only resolve inside the cluster, but HaRP
            # may run outside (e.g. Docker on the host).  ClusterIPs are routable
            # via iptables/ipvs from anywhere with kube-proxy access.
            cluster_ip = (svc.get("spec") or {}).get("clusterIP")
            if cluster_ip and cluster_ip != "None":
                host = cluster_ip
            else:
                host = _k8s_service_dns_name(service_name, K8S_NAMESPACE)
            return (host, port)
        if svc_type == "LoadBalancer":
            port = _k8s_extract_service_port(svc)
            host = _k8s_extract_loadbalancer_host(svc)
            if host:
                return (host, port)
    except Exception as e:
        LOGGER.warning("Failed to resolve K8s upstream for '%s': %s", app_name, e)
    return None


def _k8s_service_dns_name(service_name: str, namespace: str) -> str:
    # Cluster domain suffix is typically .svc.cluster.local, but .svc is enough inside most resolvers.
    return f"{service_name}.{namespace}.svc"


async def _k8s_pick_node_address(
    *,
    preferred_type: Literal["InternalIP", "ExternalIP"],
    node_name: str | None = None,
    label_selector: str | None = None,
) -> str:
    query = {"labelSelector": label_selector} if label_selector else None
    status, nodes_data, text = await _k8s_request("GET", "/api/v1/nodes", query=query)
    if status != 200 or not isinstance(nodes_data, dict):
        raise web.HTTPServiceUnavailable(
            text=f"Failed to list K8s nodes: Status {status}, {_k8s_error_msg(nodes_data, text)}"
        )

    items = nodes_data.get("items", [])
    if node_name:
        items = [n for n in items if n.get("metadata", {}).get("name") == node_name]

    if not items:
        raise web.HTTPServiceUnavailable(text="No Kubernetes nodes found (after filtering).")

    def is_ready(node: dict[str, Any]) -> bool:
        for cond in node.get("status", {}).get("conditions", []) or []:
            if cond.get("type") == "Ready" and cond.get("status") == "True":
                return True
        return False

    ready_nodes = [n for n in items if is_ready(n)]
    nodes = ready_nodes or items

    fallback_type = "ExternalIP" if preferred_type == "InternalIP" else "InternalIP"
    address_type_order = [preferred_type, fallback_type, "Hostname"]

    for node in nodes:
        for t in address_type_order:
            for addr in node.get("status", {}).get("addresses", []) or []:
                if addr.get("type") == t and addr.get("address"):
                    return str(addr["address"])

    raise web.HTTPServiceUnavailable(text="Could not determine a node address (no InternalIP/ExternalIP/Hostname).")


def _k8s_extract_nodeport(service: dict[str, Any]) -> int:
    ports = (service.get("spec") or {}).get("ports") or []
    if not ports or "nodePort" not in ports[0]:
        raise web.HTTPServiceUnavailable(text="Service has no nodePort assigned.")
    return int(ports[0]["nodePort"])


def _k8s_extract_service_port(service: dict[str, Any]) -> int:
    ports = (service.get("spec") or {}).get("ports") or []
    if not ports or "port" not in ports[0]:
        raise web.HTTPServiceUnavailable(text="Service has no port defined.")
    return int(ports[0]["port"])


def _k8s_extract_loadbalancer_host(service: dict[str, Any]) -> str | None:
    ingress = ((service.get("status") or {}).get("loadBalancer") or {}).get("ingress") or []
    if not ingress:
        return None
    first = ingress[0] or {}
    return first.get("ip") or first.get("hostname")


async def _k8s_wait_for_loadbalancer_host(service_name: str, timeout_s: float, interval_s: float) -> str:
    deadline = time.time() + max(0.0, timeout_s)
    while True:
        status, svc, text = await _k8s_request(
            "GET",
            f"/api/v1/namespaces/{K8S_NAMESPACE}/services/{service_name}",
        )
        if status != 200 or not isinstance(svc, dict):
            raise web.HTTPServiceUnavailable(
                text=f"Failed to read Service '{service_name}': Status {status}, {_k8s_error_msg(svc, text)}"
            )

        host = _k8s_extract_loadbalancer_host(svc)
        if host:
            return host

        if time.time() >= deadline:
            raise web.HTTPServiceUnavailable(
                text=f"Timed out waiting for LoadBalancer address for Service '{service_name}'"
            )

        await asyncio.sleep(interval_s)


###############################################################################
# Endpoints for AppAPI to work with the Kubernetes API
###############################################################################


async def k8s_exapp_exists(request: web.Request):
    payload = await _parse_json_payload(request, ExAppName)
    deployment_name = payload.exapp_k8s_name

    status, data, text = await _k8s_request(
        "GET",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deployment_name}",
    )
    if status == 200:
        return web.json_response({"exists": True})
    if status == 404:
        return web.json_response({"exists": False})
    LOGGER.error("Error checking deployment '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text))
    raise web.HTTPServiceUnavailable(text=f"Error checking deployment '{deployment_name}': Status {status}")


async def k8s_exapp_create(request: web.Request):
    payload = await _parse_json_payload(request, CreateExAppPayload)
    deployment_name = payload.exapp_k8s_name
    pvc_name = payload.exapp_k8s_volume_name
    base_exapp_name = _sanitize_k8s_name(
        f"nc_app_{payload.instance_id}_{payload.name}" if payload.instance_id else f"nc_app_{payload.name}"
    )

    LOGGER.info("Creating K8s resources for '%s' (ns=%s).", payload.name, K8S_NAMESPACE)

    pvc_manifest: dict[str, Any] = {
        "apiVersion": "v1",
        "kind": "PersistentVolumeClaim",
        "metadata": {
            "name": pvc_name,
            "labels": {
                "app.kubernetes.io/name": base_exapp_name,
                "app.kubernetes.io/part-of": base_exapp_name,
                "app.kubernetes.io/component": "exapp-data",
            },
        },
        "spec": {
            "accessModes": ["ReadWriteOnce"],
            "resources": {"requests": {"storage": K8S_DEFAULT_STORAGE_SIZE}},
        },
    }
    if K8S_STORAGE_CLASS:
        pvc_manifest["spec"]["storageClassName"] = K8S_STORAGE_CLASS

    status, data, text = await _k8s_request(
        "POST",
        f"/api/v1/namespaces/{K8S_NAMESPACE}/persistentvolumeclaims",
        json_body=pvc_manifest,
    )
    if status in (200, 201):
        LOGGER.info("PVC '%s' created.", pvc_name)
    elif status == 409:
        LOGGER.info("PVC '%s' already exists.", pvc_name)
    else:
        LOGGER.error("Failed to create PVC '%s' (status %s): %s", pvc_name, status, _k8s_error_msg(data, text))
        raise web.HTTPServiceUnavailable(text=f"Failed to create PVC '{pvc_name}': Status {status}")

    deployment_manifest = _k8s_build_deployment_manifest(payload, replicas=0)
    status, data, text = await _k8s_request(
        "POST",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments",
        json_body=deployment_manifest,
    )
    if status in (200, 201):
        LOGGER.info("Deployment '%s' created.", deployment_name)
        return web.json_response({"name": deployment_name}, status=201)
    if status == 409:
        raise web.HTTPConflict(text=f"Deployment '{deployment_name}' already exists.")
    LOGGER.error("Error creating deployment '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text))
    raise web.HTTPServiceUnavailable(text=f"Error creating deployment '{deployment_name}': Status {status}")


async def k8s_exapp_start(request: web.Request):
    payload = await _parse_json_payload(request, ExAppName)
    deployment_name = payload.exapp_k8s_name

    status, data, text = await _k8s_request(
        "PATCH",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deployment_name}",
        json_body={"spec": {"replicas": 1}},
        content_type="application/strategic-merge-patch+json",
    )
    if status in (200, 201):
        LOGGER.info("Deployment '%s' scaled to 1.", deployment_name)
        return web.HTTPNoContent()
    if status == 404:
        raise web.HTTPNotFound(text=f"Deployment '{deployment_name}' not found.")
    LOGGER.error("Error starting '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text))
    raise web.HTTPServiceUnavailable(text=f"Error starting deployment '{deployment_name}': Status {status}")


async def k8s_exapp_stop(request: web.Request):
    payload = await _parse_json_payload(request, ExAppName)
    deployment_name = payload.exapp_k8s_name

    status, data, text = await _k8s_request(
        "PATCH",
        f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deployment_name}",
        json_body={"spec": {"replicas": 0}},
        content_type="application/strategic-merge-patch+json",
    )
    if status in (200, 201):
        LOGGER.info("Deployment '%s' scaled to 0.", deployment_name)
        return web.HTTPNoContent()
    if status == 404:
        raise web.HTTPNotFound(text=f"Deployment '{deployment_name}' not found.")
    LOGGER.error("Error stopping '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text))
    raise web.HTTPServiceUnavailable(text=f"Error stopping deployment '{deployment_name}': Status {status}")


async def k8s_exapp_wait_for_start(request: web.Request):
    payload = await _parse_json_payload(request, ExAppName)
    deployment_name = payload.exapp_k8s_name
    label_selector = f"app={deployment_name}"

    # Startup timeout (seconds) applies only *after* the image has been pulled. While the image is being pulled
    # we wait indefinitely (image sizes vary and we cannot predict download speed).
    startup_timeout = 90.0
    sleep_interval = 1.0
    # Safety cap: abort even during image pull after this many seconds to avoid waiting forever if something
    # is truly stuck (e.g. registry unreachable but not yet in ImagePullBackOff).
    image_pull_max_wait = 3600.0

    LOGGER.info(
        "Waiting for Kubernetes pod(s) of deployment '%s' to become Ready "
        "(namespace=%s, startup_timeout=%.0fs, image_pull_max_wait=%.0fs).",
        deployment_name,
        K8S_NAMESPACE,
        startup_timeout,
        image_pull_max_wait,
    )

    last_phase: str | None = None
    last_reason: str | None = None
    last_message: str | None = None
    image_pulled = False
    start_time = time.monotonic()
    post_pull_start: float | None = None

    while True:
        elapsed = time.monotonic() - start_time

        if image_pulled:
            assert post_pull_start is not None
            if time.monotonic() - post_pull_start > startup_timeout:
                LOGGER.warning(
                    "Deployment '%s' did not become Ready within %.0fs after image pull.",
                    deployment_name,
                    startup_timeout,
                )
                break
        elif elapsed > image_pull_max_wait:
            LOGGER.warning(
                "Deployment '%s' image pull did not complete within %.0fs.",
                deployment_name,
                image_pull_max_wait,
            )
            break

        status, data, text = await _k8s_request(
            "GET",
            f"/api/v1/namespaces/{K8S_NAMESPACE}/pods",
            query={"labelSelector": label_selector},
        )
        if status != 200:
            LOGGER.error(
                "Error listing pods for '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text)
            )
            raise web.HTTPServiceUnavailable(
                text=f"Error listing pods for deployment '{deployment_name}': Status {status}"
            )

        items = (data or {}).get("items", []) if isinstance(data, dict) else []
        if not items:
            LOGGER.debug(
                "No pods yet for deployment '%s' (elapsed %.0fs).",
                deployment_name,
                elapsed,
            )
            last_phase = "Pending"
        else:
            # Take the first pod; for single-replica deployments this is enough.
            pod = items[0]
            pod_status = pod.get("status", {})
            phase = pod_status.get("phase", "Unknown")
            last_phase = phase
            conditions = pod_status.get("conditions", [])
            ready = any(c.get("type") == "Ready" and c.get("status") == "True" for c in conditions)
            last_reason = pod_status.get("reason")
            last_message = pod_status.get("message")

            # Inspect container statuses to detect image pull vs app startup.
            container_statuses = pod_status.get("containerStatuses", [])
            waiting_reason = None
            if container_statuses:
                cs = container_statuses[0]
                waiting = cs.get("state", {}).get("waiting", {})
                waiting_reason = waiting.get("reason")

            LOGGER.debug(
                "Pod status for '%s' (elapsed %.0fs): phase=%s, ready=%s, "
                "waiting_reason=%s, reason=%s, message=%s",
                deployment_name,
                elapsed,
                phase,
                ready,
                waiting_reason,
                last_reason,
                last_message,
            )

            if phase == "Running" and ready:
                LOGGER.info("Deployment '%s' pod is Running and Ready.", deployment_name)
                return web.json_response(
                    {
                        "started": True,
                        "status": "running",
                        "health": "ready",
                        "reason": last_reason,
                        "message": last_message,
                    }
                )

            # Succeeded means the pod ran and exited 0 (e.g. init-only apps
            # like app-skeleton-python).  Treat as successful start.
            if phase == "Succeeded":
                LOGGER.info(
                    "Deployment '%s' pod completed successfully (phase=Succeeded).",
                    deployment_name,
                )
                return web.json_response(
                    {
                        "started": True,
                        "status": "succeeded",
                        "health": "ready",
                        "reason": last_reason,
                        "message": last_message,
                    }
                )

            if phase in ("Failed", "Unknown"):
                LOGGER.warning(
                    "Deployment '%s' pod is in phase '%s', treating as not successfully started.",
                    deployment_name,
                    phase,
                )
                return web.json_response(
                    {
                        "started": False,
                        "status": phase,
                        "health": "not_ready",
                        "reason": last_reason,
                        "message": last_message,
                    }
                )

            # Fail fast on image pull errors.
            if waiting_reason in ("ErrImagePull", "ImagePullBackOff", "ErrImageNeverPull", "InvalidImageName"):
                wait_msg = container_statuses[0].get("state", {}).get("waiting", {}).get("message", "")
                LOGGER.error(
                    "Deployment '%s' pod has image pull error: %s - %s",
                    deployment_name,
                    waiting_reason,
                    wait_msg,
                )
                return web.json_response(
                    {
                        "started": False,
                        "status": waiting_reason,
                        "health": "image_error",
                        "reason": waiting_reason,
                        "message": wait_msg,
                    }
                )

            # Detect when image pull is done: once the container has a non-pull waiting reason (e.g. CrashLoopBackOff)
            # or is running/terminated, the image has been pulled.
            if not image_pulled:
                # Still pulling: no container statuses yet, or waiting with ContainerCreating / PodInitializing.
                still_pulling = (
                    not container_statuses
                    or waiting_reason in ("ContainerCreating", "PodInitializing", None)
                )
                if not still_pulling:
                    image_pulled = True
                    post_pull_start = time.monotonic()
                    LOGGER.info(
                        "Deployment '%s' image pull completed after %.0fs. "
                        "Starting startup countdown (%.0fs).",
                        deployment_name,
                        elapsed,
                        startup_timeout,
                    )

        await asyncio.sleep(sleep_interval)

    return web.json_response(
        {
            "started": False,
            "status": last_phase or "unknown",
            "health": "timeout",
            "reason": last_reason,
            "message": last_message,
        }
    )


async def k8s_exapp_remove(request: web.Request):
    payload = await _parse_json_payload(request, RemoveExAppPayload)
    deployment_name = payload.exapp_k8s_name
    pvc_name = payload.exapp_k8s_volume_name

    LOGGER.info(
        "Removing K8s resources for '%s' (ns=%s, remove_data=%s).", deployment_name, K8S_NAMESPACE, payload.remove_data
    )

    status, data, text = await _k8s_request(
        "DELETE", f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deployment_name}",
    )
    if status not in (200, 202, 404):
        LOGGER.error(
            "Error removing deployment '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text)
        )
        raise web.HTTPServiceUnavailable(text=f"Error removing deployment '{deployment_name}': Status {status}")

    if payload.remove_data:
        status, data, text = await _k8s_request(
            "DELETE", f"/api/v1/namespaces/{K8S_NAMESPACE}/persistentvolumeclaims/{pvc_name}",
        )
        if status not in (200, 202, 404):
            LOGGER.error("Error removing PVC '%s' (status %s): %s", pvc_name, status, _k8s_error_msg(data, text))
            raise web.HTTPServiceUnavailable(text=f"Error removing PVC '{pvc_name}': Status {status}")

    # Only delete Services that HaRP created (labelled managed-by=harp).
    # For manual expose, the operator manages the Service — don't delete it.
    svc_path = f"/api/v1/namespaces/{K8S_NAMESPACE}/services/{deployment_name}"
    status, svc_data, text = await _k8s_request("GET", svc_path)
    if status == 200 and isinstance(svc_data, dict):
        svc_labels = (svc_data.get("metadata") or {}).get("labels") or {}
        if svc_labels.get("app.kubernetes.io/managed-by") == "harp":
            status, data, text = await _k8s_request("DELETE", svc_path)
            if status not in (200, 202, 404):
                LOGGER.error("Error removing Service '%s' (status %s): %s", deployment_name, status, _k8s_error_msg(data, text))
                raise web.HTTPServiceUnavailable(text=f"Error removing Service '{deployment_name}': Status {status}")
        else:
            LOGGER.info("Skipping deletion of externally-managed Service '%s'.", deployment_name)
    elif status != 404:
        LOGGER.warning("Error checking Service '%s' (status %s), skipping deletion.", deployment_name, status)
    return web.HTTPNoContent()


async def k8s_exapp_install_certificates(request: web.Request):
    # K8s: certificates are handled via Secrets/volume mounts, this is a no-op.
    await _parse_json_payload(request, InstallCertificatesPayload)
    return web.HTTPNoContent()


async def k8s_exapp_expose(request: web.Request):
    payload = await _parse_json_payload(request, ExposeExAppPayload)
    app_id = payload.name.lower()
    service_name = payload.exapp_k8s_name

    if payload.expose_type == "manual":
        upstream_host = payload.upstream_host
        upstream_port = int(payload.upstream_port or payload.port)

        # Persist the manual upstream config as annotations on the Deployment so
        # that _k8s_resolve_exapp_upstream() can re-discover it after a cache
        # eviction or HaRP restart (no K8s Service exists for manual type).
        # This is best-effort: if K8s is not configured or the PATCH fails, the
        # expose still succeeds (upstream is cached in memory).
        if K8S_ENABLED and K8S_API_SERVER and _get_k8s_token():
            try:
                deploy_name = payload.exapp_k8s_name
                annotation_patch = {
                    "metadata": {
                        "annotations": {
                            "appapi.nextcloud.com/expose-type": "manual",
                            "appapi.nextcloud.com/upstream-host": upstream_host,
                            "appapi.nextcloud.com/upstream-port": str(upstream_port),
                        },
                    },
                }
                status, _, text = await _k8s_request(
                    "PATCH",
                    f"/apis/apps/v1/namespaces/{K8S_NAMESPACE}/deployments/{deploy_name}",
                    json_body=annotation_patch,
                    content_type="application/strategic-merge-patch+json",
                )
                if status not in (200, 201):
                    LOGGER.warning(
                        "Failed to annotate Deployment '%s' with manual upstream (status %s): %s",
                        deploy_name, status, text[:200] if text else "",
                    )
            except Exception as e:
                LOGGER.warning("Best-effort annotation PATCH failed for '%s': %s", payload.name, e)
    else:
        _ensure_k8s_configured()

        type_map: dict[str, Literal["NodePort", "ClusterIP", "LoadBalancer"]] = {
            "nodeport": "NodePort", "clusterip": "ClusterIP", "loadbalancer": "LoadBalancer",
        }
        desired_type = type_map.get(payload.expose_type)
        if not desired_type:
            raise web.HTTPBadRequest(text=f"Unknown expose_type '{payload.expose_type}'")

        service_manifest = _k8s_build_service_manifest(payload, desired_type)
        status, data, text = await _k8s_request(
            "POST", f"/api/v1/namespaces/{K8S_NAMESPACE}/services", json_body=service_manifest,
        )
        if status not in (200, 201, 409):
            LOGGER.error(
                "Failed to create Service '%s' (status %s): %s", service_name, status, _k8s_error_msg(data, text)
            )
            raise web.HTTPServiceUnavailable(text=f"Failed to create Service '{service_name}': Status {status}")

        status, svc, text = await _k8s_request(
            "GET", f"/api/v1/namespaces/{K8S_NAMESPACE}/services/{service_name}",
        )
        if status != 200 or not isinstance(svc, dict):
            LOGGER.error("Failed to read Service '%s' (status %s): %s", service_name, status, _k8s_error_msg(svc, text))
            raise web.HTTPServiceUnavailable(text=f"Failed to read Service '{service_name}': Status {status}")

        if payload.expose_type == "nodeport":
            upstream_port = _k8s_extract_nodeport(svc)
            upstream_host = payload.upstream_host or await _k8s_pick_node_address(
                preferred_type=payload.node_address_type,
                node_name=payload.node_name,
                label_selector=payload.node_label_selector,
            )
        elif payload.expose_type == "clusterip":
            upstream_port = _k8s_extract_service_port(svc)
            # Use the actual ClusterIP address instead of DNS name.  DNS names
            # like "svc.namespace.svc" only resolve inside the cluster, but HaRP
            # may run outside (e.g. Docker on the host).  ClusterIPs are routable
            # via iptables/ipvs from anywhere with kube-proxy access.
            if not payload.upstream_host:
                cluster_ip = (svc.get("spec") or {}).get("clusterIP")
                if cluster_ip and cluster_ip != "None":
                    upstream_host = cluster_ip
                else:
                    upstream_host = _k8s_service_dns_name(service_name, K8S_NAMESPACE)
            else:
                upstream_host = payload.upstream_host
        else:  # loadbalancer
            upstream_port = _k8s_extract_service_port(svc)
            upstream_host = payload.upstream_host or _k8s_extract_loadbalancer_host(svc)
            if not upstream_host:
                upstream_host = await _k8s_wait_for_loadbalancer_host(
                    service_name, timeout_s=payload.wait_timeout_seconds, interval_s=payload.wait_interval_seconds,
                )

    LOGGER.info("Expose '%s' (%s): upstream %s:%d", app_id, payload.expose_type, upstream_host, upstream_port)
    try:
        exapp_meta = await nc_get_exapp(app_id)
        if not exapp_meta:
            LOGGER.error("No ExApp metadata for '%s' in Nextcloud.", app_id)
            raise web.HTTPNotFound(text=f"No ExApp metadata for '{app_id}'")
    except web.HTTPException:
        raise
    except Exception as e:
        LOGGER.exception("Failed to fetch ExApp metadata for '%s'", app_id)
        raise web.HTTPServiceUnavailable(text=f"Failed to fetch metadata for '{app_id}'") from e

    exapp_meta.host = upstream_host
    exapp_meta.port = int(upstream_port)
    exapp_meta.resolved_host = ""

    EXAPP_CACHE[app_id] = exapp_meta
    _EXAPP_NEGATIVE_CACHE.pop(app_id, None)

    return web.json_response(
        {
            "appId": app_id,
            "host": upstream_host,
            "port": int(upstream_port),
            "exposeType": payload.expose_type,
            "serviceName": service_name,
            "namespace": K8S_NAMESPACE,
        }
    )


###############################################################################
# HTTP Server Setup
###############################################################################


# Paths of the agent's HTTP port that answer without the shared key: the health check and the FRP server's login hook.
AGENT_PATHS_WITHOUT_KEY = frozenset({"/heartbeat", "/frp_handler"})


@web.middleware
async def require_shared_key(request: web.Request, handler):
    """Refuse requests to the agent's port that do not carry the shared key.

    HAProxy checks the key before it forwards a request here, but every process in HaRP's network namespace can reach
    the port directly, such as ExApps when HaRP runs with host networking. No failure is recorded: such a request
    comes from the same address as HaRP's own callers, so a ban would lock them out.
    """
    if request.path not in AGENT_PATHS_WITHOUT_KEY and not is_shared_key(request.headers.get("harp-shared-key")):
        LOGGER.warning("Refused a request to %r on the agent's port without the shared key.", request.path[:64])
        raise web.HTTPUnauthorized()
    return await handler(request)


def create_web_app() -> web.Application:
    app = web.Application(middlewares=[require_shared_key])

    app.router.add_get("/heartbeat", get_heartbeat)
    app.router.add_get("/info", get_info)

    # ExApp routes
    app.router.add_post("/exapp_storage/{app_id}", add_exapp)
    app.router.add_delete("/exapp_storage/{app_id}", delete_exapp)

    # FRP auth (FRP Server will call it)
    app.router.add_post("/frp_handler", frp_auth)

    # Docker Engine APIs wrappers
    app.router.add_post("/docker/exapp/exists", docker_exapp_exists)
    app.router.add_post("/docker/exapp/create", docker_exapp_create)
    app.router.add_post("/docker/exapp/start", docker_exapp_start)
    app.router.add_post("/docker/exapp/stop", docker_exapp_stop)
    app.router.add_post("/docker/exapp/wait_for_start", docker_exapp_wait_for_start)
    app.router.add_post("/docker/exapp/remove", docker_exapp_remove)
    app.router.add_post("/docker/exapp/image_remove", docker_exapp_image_remove)
    app.router.add_post("/docker/exapp/install_certificates", docker_exapp_install_certificates)

    # Kubernetes APIs wrappers
    app.router.add_post("/k8s/exapp/exists", k8s_exapp_exists)
    app.router.add_post("/k8s/exapp/create", k8s_exapp_create)
    app.router.add_post("/k8s/exapp/start", k8s_exapp_start)
    app.router.add_post("/k8s/exapp/stop", k8s_exapp_stop)
    app.router.add_post("/k8s/exapp/wait_for_start", k8s_exapp_wait_for_start)
    app.router.add_post("/k8s/exapp/remove", k8s_exapp_remove)
    app.router.add_post("/k8s/exapp/install_certificates", k8s_exapp_install_certificates)
    app.router.add_post("/k8s/exapp/expose", k8s_exapp_expose)
    app.on_shutdown.append(_close_k8s_session)
    app.on_shutdown.append(_close_nc_session)
    return app


async def run_http_server(host="127.0.0.1", port=8200):
    app = create_web_app()
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, host, port)
    LOGGER.info("HTTP server listening at %s:%s", host, port)
    await site.start()
    while True:
        await asyncio.sleep(3600)


###############################################################################
# Main entry point: run both SPOA & HTTP
###############################################################################


async def _loop_lag_monitor():
    """Log when the event loop was blocked: everything in this process shares it, so a stall is a full outage."""
    interval = 1.0
    loop = asyncio.get_running_loop()
    expected = loop.time() + interval
    while True:
        await asyncio.sleep(max(0.0, expected - loop.time()))
        now = loop.time()
        lag = now - expected
        if lag > 1.0:
            LOGGER.warning("Event loop was unresponsive for %.1f seconds", lag)
        expected = now + interval


async def main():
    spoa_task = asyncio.create_task(SPOA_AGENT._run(host=SPOA_HOST, port=SPOA_PORT))  # noqa
    http_task = asyncio.create_task(run_http_server(host="127.0.0.1", port=8200))
    lag_task = asyncio.create_task(_loop_lag_monitor())

    LOGGER.info("Starting both servers: SPOA on %s:%d, HTTP on 127.0.0.1:8200", SPOA_HOST, SPOA_PORT)
    await asyncio.gather(spoa_task, http_task, lag_task)


if __name__ == "__main__":
    asyncio.run(main())