Blame for deployment/google-cloud/cloud_http_runtime.py

"""Private cloud HTTP composition; no release, executor, or provider authority.



Only the distinct cloud HTTP units own these ports. Do not also schedule the

Windows operator_gateway/frontend_vite registry entries on the cloud host.

The stdlib-only secret worker communicates exclusively through an anonymous pipe.

"""
from __future__ import annotations

import argparse
import base64
import binascii
import hmac
import http.client
import ipaddress
import json
import os
from pathlib import Path
import re
import socket
import subprocess
import sys
import threading
import time
from urllib.parse import urlsplit
from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener

# Isolated-mode secret subprocesses deliberately omit the script directory from
# sys.path. Restore only this resolved, deployed directory so fixed adjacent
# boundary modules remain importable without admitting cwd or user site paths.
_DEPLOYMENT_MODULE_DIR = str(Path(__file__).resolve().parent)
if _DEPLOYMENT_MODULE_DIR not in sys.path:
    sys.path.insert(0, _DEPLOYMENT_MODULE_DIR)

from cloud_credential_boundary import (  # noqa: E402
    CLOUD_CREDENTIAL_ISOLATION_ENV,
    CLOUD_EXCHANGE_CREDENTIALS_PINNED_ENV,
    CLOUD_EXCHANGE_CREDENTIAL_ENV_VARS as CLOUD_EXCHANGE_CREDENTIAL_ENV_VARS,
    CLOUD_EXCHANGE_MODE_ENV_VARS as CLOUD_EXCHANGE_MODE_ENV_VARS,
    CLOUD_ENV_LOADER_CONTROL_VARS as CLOUD_ENV_LOADER_CONTROL_VARS,
    CLOUD_OPERATOR_SAFE_MODE_ENV,
    CLOUD_OPERATOR_SCRUBBED_ENV_VARS as CLOUD_OPERATOR_SCRUBBED_ENV_VARS,
    install_cloud_environment_guard,
    pin_validated_environment,
    scrub_cloud_exchange_authority,
)
from vm_workload_identity import (  # noqa: E402
    BROKER_REQUIRED_ENV,
    TOKEN_FILE_ENV,
    WorkloadIdentityError,
    broker_token,
)


HOLDS = {
    "AUREON_DEPLOYMENT_HOLD": "1",
    "AUREON_DISABLE_REAL_ORDERS": "1",
    "AUREON_DISABLE_EXCHANGE_MUTATIONS": "1",
    "AUREON_ORDER_INTENT_PUBLISH": "0",
    "AUREON_LIVE_TRADING": "0",
    "AUREON_UNIFIED_ORDER_EXECUTOR": "0",
}
CONTROLLER_KEY_VERSION = (
    "projects/aureon-trading/locations/europe-west1/keyRings/aureon-trading-prod/"
    "cryptoKeys/aureon-controller-signing/cryptoKeyVersions/1"
)
PORTS = {"operator": 8080, "ui-gateway": 5173}
HEALTH_PATH = "/__aureon_health"
HEALTH_SCHEMA = "aureon.cloud-http-health.v1"
MAX_REQUEST = 256 * 1024
MAX_RESPONSE = 8 * 1024 * 1024
COGNITION_UPSTREAM_TIMEOUT_SECONDS = 65
AUTH_RESOLUTION_KEY = "HTTP_X_AUREON_AUTH_RESOLUTION"
INTENT_CUSTODY_ENV_KEY = "HTTP_X_AUREON_OPERATOR_INTENT"
INTENT_CUSTODY_HEADER = "X-Aureon-Operator-Intent"
METADATA_BASE = "http://metadata.google.internal/computeMetadata/v1/"
SECRET_DEADLINE = 20
SECRET_CLEANUP = 2
OWNER_PROVIDER_KEYS = (
    "NASA_API_KEY",
    "NOAA_API_KEY",
    "USGS_API_KEY",
    "OLLAMA_API_KEY",
    "AUREON_LLM_BASE_URL",
    "AUREON_LLM_MODEL",
)
PINNED_PROVIDER_ENV_VARS_ENV = "AUREON_PINNED_PROVIDER_ENV_VARS"
OWNER_LLM_BASE_URL = "https://ollama.com/v1"
BROKER_PROJECT_NUMBER_ENV = "AUREON_GCP_PROJECT_NUMBER"
OPERATOR_STATE_PATHS = {
    "AUREON_FEATURE_SWITCHBOARD_DIR": "/var/lib/aureon-operator/feature-switchboard",
    "AUREON_OPERATOR_INTENT_LEDGER": (
        "/var/lib/aureon-operator/operator-intent/operator-intent-replay.sqlite3"
    ),
    "AUREON_RELEASE_LEDGER_PATH": (
        "/var/lib/aureon-control-exchange/release-ledger/release-ledger.jsonl"
    ),
    "AUREON_CLOUD_TRADE_READBACK_PATH": (
        "/var/lib/aureon-control-exchange/trade-readback/latest-terminal-receipt.json"
    ),
    "AUREON_CLOUD_TRADE_BRIDGE_STATUS_PATH": (
        "/var/lib/aureon-control-exchange/trade-readback/"
        "latest-order-intent-bridge.json"
    ),
}


class CloudHTTPError(RuntimeError):
    """Only fixed, non-sensitive error codes may leave this module."""


def require_holds(environment=None):
    environment = os.environ if environment is None else environment
    if any(environment.get(name) != value for name, value in HOLDS.items()):
        raise CloudHTTPError("cloud_deployment_holds_required")


def holds_ready():
    return all(os.environ.get(name) == value for name, value in HOLDS.items())


def validate_secret_reference(project, secret, version):
    if not isinstance(project, str) or not re.fullmatch(r"(?:[a-z][a-z0-9-]{4,61}[a-z0-9]|[0-9]{6,20})", project):
        raise CloudHTTPError("cloud_project_invalid")
    if not isinstance(secret, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,255}", secret):
        raise CloudHTTPError("cloud_secret_reference_invalid")
    if not isinstance(version, str) or not re.fullmatch(r"[1-9][0-9]{0,18}", version):
        raise CloudHTTPError("cloud_secret_version_must_be_pinned")


def validate_bearer(raw):
    if not isinstance(raw, bytes) or not re.fullmatch(rb"[A-Za-z0-9_-]{48,128}", raw):
        raise CloudHTTPError("cloud_secret_payload_invalid")
    return raw.decode("ascii")


def validate_supabase_tenant_config(raw):
    """Validate the complete public Supabase tenant-auth configuration.



    The publishable key is intentionally client-visible, but it still travels

    through Secret Manager so one pinned, checksummed deployment input owns the

    operator's tenant-verification configuration.  Extra fields are rejected so

    this payload can never grow into a provider/service credential carrier.

    """
    if not isinstance(raw, bytes) or not 1 <= len(raw) <= 8192:
        raise CloudHTTPError("cloud_tenant_secret_payload_invalid")
    try:
        payload = json.loads(raw.decode("utf-8"))
    except (UnicodeError, ValueError, TypeError) as exc:
        raise CloudHTTPError("cloud_tenant_secret_payload_invalid") from exc
    if not isinstance(payload, dict) or set(payload) != {"project_url", "publishable_key"}:
        raise CloudHTTPError("cloud_tenant_secret_payload_invalid")
    project_url = payload.get("project_url")
    publishable_key = payload.get("publishable_key")
    if not isinstance(project_url, str) or not isinstance(publishable_key, str):
        raise CloudHTTPError("cloud_tenant_secret_payload_invalid")
    parsed = urlsplit(project_url)
    host = parsed.hostname or ""
    if (parsed.scheme != "https" or parsed.port is not None or parsed.username is not None
            or parsed.password is not None or parsed.path not in {"", "/"} or parsed.query
            or parsed.fragment or not re.fullmatch(r"[a-z0-9]{20}\.supabase\.co", host)):
        raise CloudHTTPError("cloud_tenant_project_url_invalid")
    if (not 20 <= len(publishable_key) <= 4096
            or re.search(r"[^\x21-\x7e]", publishable_key)
            or not (publishable_key.startswith("sb_publishable_") or publishable_key.count(".") == 2)):
        raise CloudHTTPError("cloud_tenant_publishable_key_invalid")
    if publishable_key.count(".") == 2:
        try:
            segment = publishable_key.split(".")[1]
            segment += "=" * (-len(segment) % 4)
            claims = json.loads(base64.urlsafe_b64decode(segment.encode("ascii")))
        except (ValueError, UnicodeError, binascii.Error, TypeError) as exc:
            raise CloudHTTPError("cloud_tenant_publishable_key_invalid") from exc
        if (not isinstance(claims, dict) or claims.get("role") != "anon"
                or claims.get("ref") not in {None, host.split(".", 1)[0]}):
            raise CloudHTTPError("cloud_tenant_publishable_key_invalid")
    return {"project_url": project_url.rstrip("/"), "publishable_key": publishable_key}


def validate_owner_provider_config(raw):
    """Validate the complete owner-only provider configuration bundle.



    This object is deliberately closed: adding another provider credential

    requires a code/release review rather than silently widening which secrets

    reach the operator and selected read-only organs.

    """
    if not isinstance(raw, bytes) or not 1 <= len(raw) <= 32768:
        raise CloudHTTPError("cloud_owner_provider_payload_invalid")

    def unique_object(pairs):
        result = {}
        for key, value in pairs:
            if key in result:
                raise ValueError("duplicate key")
            result[key] = value
        return result

    try:
        payload = json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object)
    except (UnicodeError, ValueError, TypeError) as exc:
        raise CloudHTTPError("cloud_owner_provider_payload_invalid") from exc
    if not isinstance(payload, dict) or set(payload) != set(OWNER_PROVIDER_KEYS):
        raise CloudHTTPError("cloud_owner_provider_payload_invalid")

    result = {}
    for key in OWNER_PROVIDER_KEYS[:4]:
        value = payload.get(key)
        if (not isinstance(value, str) or not 8 <= len(value) <= 4096
                or re.fullmatch(r"[\x21-\x7e]+", value) is None
                or value.casefold() in {"demo_key", "changeme", "placeholder", "replace_me"}):
            raise CloudHTTPError("cloud_owner_provider_key_invalid")
        result[key] = value

    base_url = payload.get("AUREON_LLM_BASE_URL")
    if base_url != OWNER_LLM_BASE_URL:
        raise CloudHTTPError("cloud_owner_llm_base_url_invalid")
    model = payload.get("AUREON_LLM_MODEL")
    if (not isinstance(model, str) or not 1 <= len(model) <= 128
            or re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.:/-]*", model) is None):
        raise CloudHTTPError("cloud_owner_llm_model_invalid")
    result["AUREON_LLM_BASE_URL"] = base_url
    result["AUREON_LLM_MODEL"] = model
    return result


def crc32c(data):
    """Castagnoli CRC, verified with the standard 123456789 test vector."""
    value = 0xFFFFFFFF
    for byte in data:
        value ^= byte
        for _ in range(8):
            value = (value >> 1) ^ (0x82F63B78 if value & 1 else 0)
    return value ^ 0xFFFFFFFF


class NoRedirects(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise CloudHTTPError("cloud_secret_redirect_denied")


def _secret_http_read(url, headers, *, metadata=False):
    # Only fetch_secret_worker constructs URLs. No environment proxies, alternate
    # credentials, redirect traversal, or unbounded response body is accepted.
    opener = build_opener(ProxyHandler({}), NoRedirects())
    with opener.open(Request(url, headers=headers), timeout=5) as response:
        if response.status != 200:
            raise CloudHTTPError("cloud_secret_access_failed")
        if metadata and response.headers.get("Metadata-Flavor") != "Google":
            raise CloudHTTPError("cloud_metadata_identity_invalid")
        data = response.read(16385)
        if len(data) > 16384:
            raise CloudHTTPError("cloud_secret_response_too_large")
        return data


def decode_secret_payload(raw, *, resource):
    try:
        result = json.loads(raw)
        if not isinstance(result, dict) or result.get("name") != resource:
            raise CloudHTTPError("cloud_secret_resource_mismatch")
        payload = result["payload"]
        if not isinstance(payload, dict) or not isinstance(payload.get("data"), str):
            raise CloudHTTPError("cloud_secret_payload_invalid")
        data = base64.b64decode(payload["data"], validate=True)
        checksum = payload.get("dataCrc32c")
        if not isinstance(checksum, str) or not re.fullmatch(r"[0-9]{1,10}", checksum):
            raise CloudHTTPError("cloud_secret_checksum_missing")
        if int(checksum) != crc32c(data):
            raise CloudHTTPError("cloud_secret_checksum_mismatch")
        return data
    except (ValueError, KeyError, TypeError, binascii.Error) as exc:
        raise CloudHTTPError("cloud_secret_response_invalid") from exc


def decode_secret_response(raw, *, resource):
    return validate_bearer(decode_secret_payload(raw, resource=resource))


def fetch_secret_worker(project, secret, version, reader=_secret_http_read, *, secret_kind="bearer"):
    """Runs only in the isolated, deadline-bounded child, never logs payloads."""
    validate_secret_reference(project, secret, version)
    header = {"Metadata-Flavor": "Google"}
    try:
        token = broker_token()
    except WorkloadIdentityError as exc:
        raise CloudHTTPError(str(exc)) from exc
    if token is None:
        project_id = reader(METADATA_BASE + "project/project-id", header, metadata=True).decode("ascii")
        project_number = reader(METADATA_BASE + "project/numeric-project-id", header, metadata=True).decode("ascii")
        if project not in {project_id, project_number} or not re.fullmatch(r"[0-9]{6,20}", project_number):
            raise CloudHTTPError("cloud_metadata_project_mismatch")
        identity = json.loads(reader(METADATA_BASE + "instance/service-accounts/default/token", header, metadata=True))
        token = identity.get("access_token")
        expires = identity.get("expires_in")
        if (not isinstance(token, str) or not 20 <= len(token) <= 8192
                or re.search(r"[^\x21-\x7e]", token)
                or identity.get("token_type", "").lower() != "bearer"
                or type(expires) is not int or not 30 <= expires <= 86400):
            raise CloudHTTPError("cloud_metadata_token_invalid")
    else:
        project_id = project
        project_number = str(os.environ.get(BROKER_PROJECT_NUMBER_ENV) or "")
        if project_id != "aureon-trading" or project_number != "4435775023":
            raise CloudHTTPError("cloud_broker_project_mismatch")
    resource = f"projects/{project_number}/secrets/{secret}/versions/{version}"
    raw = reader("https://secretmanager.googleapis.com/v1/" + resource + ":access",
                 {"Authorization": "Bearer " + token, "Accept": "application/json"})
    payload = decode_secret_payload(raw, resource=resource)
    if secret_kind == "bearer":
        return validate_bearer(payload)
    if secret_kind == "supabase-tenant":
        return json.dumps(validate_supabase_tenant_config(payload), sort_keys=True, separators=(",", ":"))
    if secret_kind == "owner-provider":
        return json.dumps(validate_owner_provider_config(payload), sort_keys=True, separators=(",", ":"))
    raise CloudHTTPError("cloud_secret_kind_invalid")


def worker_environment(environment=None):
    environment = os.environ if environment is None else environment
    # No provider/API credentials, proxy settings, PYTHONPATH, or SDK key files.
    allowed = (
        "SYSTEMROOT", "WINDIR", "TEMP", "TMP", "LANG", "LC_ALL",
        TOKEN_FILE_ENV, BROKER_REQUIRED_ENV, BROKER_PROJECT_NUMBER_ENV,
    )
    return {key: environment[key] for key in allowed if key in environment}


def load_cloud_secret(project, secret, version, *, popen=subprocess.Popen):
    validate_secret_reference(project, secret, version)
    command = [sys.executable, "-I", "-S", "-B", str(Path(__file__).resolve()),
               "--secret-worker", "--project", project, "--secret", secret, "--secret-version", version]
    process = popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE, env=worker_environment(), close_fds=True)
    try:
        output, errors = process.communicate(timeout=SECRET_DEADLINE)
    except subprocess.TimeoutExpired:
        process.kill()
        try:
            process.communicate(timeout=SECRET_CLEANUP)
        except subprocess.TimeoutExpired:
            raise CloudHTTPError("cloud_secret_worker_cleanup_failed") from None
        raise CloudHTTPError("cloud_secret_deadline_exceeded") from None
    if process.returncode != 0 or errors:
        # Neither stdout nor stderr (nor the process exception) is logged.
        raise CloudHTTPError("cloud_secret_access_failed")
    return validate_bearer(output)


def load_cloud_tenant_config(project, secret, version, *, popen=subprocess.Popen):
    validate_secret_reference(project, secret, version)
    command = [sys.executable, "-I", "-S", "-B", str(Path(__file__).resolve()),
               "--secret-worker", "--secret-kind", "supabase-tenant", "--project", project,
               "--secret", secret, "--secret-version", version]
    process = popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE, env=worker_environment(), close_fds=True)
    try:
        output, errors = process.communicate(timeout=SECRET_DEADLINE)
    except subprocess.TimeoutExpired:
        process.kill()
        try:
            process.communicate(timeout=SECRET_CLEANUP)
        except subprocess.TimeoutExpired:
            raise CloudHTTPError("cloud_secret_worker_cleanup_failed") from None
        raise CloudHTTPError("cloud_secret_deadline_exceeded") from None
    if process.returncode != 0 or errors:
        raise CloudHTTPError("cloud_secret_access_failed")
    return validate_supabase_tenant_config(output)


def load_cloud_owner_provider_config(project, secret, version, *, popen=subprocess.Popen):
    validate_secret_reference(project, secret, version)
    command = [sys.executable, "-I", "-S", "-B", str(Path(__file__).resolve()),
               "--secret-worker", "--secret-kind", "owner-provider", "--project", project,
               "--secret", secret, "--secret-version", version]
    process = popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
                    stderr=subprocess.PIPE, env=worker_environment(), close_fds=True)
    try:
        output, errors = process.communicate(timeout=SECRET_DEADLINE)
    except subprocess.TimeoutExpired:
        process.kill()
        try:
            process.communicate(timeout=SECRET_CLEANUP)
        except subprocess.TimeoutExpired:
            raise CloudHTTPError("cloud_secret_worker_cleanup_failed") from None
        raise CloudHTTPError("cloud_secret_deadline_exceeded") from None
    if process.returncode != 0 or errors:
        raise CloudHTTPError("cloud_secret_access_failed")
    return validate_owner_provider_config(output)


def _loopback(value, *, allow_name=False):
    if allow_name and value == "localhost":
        return True
    try:
        address = ipaddress.ip_address(value)
        if isinstance(address, ipaddress.IPv6Address) and address.ipv4_mapped:
            address = address.ipv4_mapped
        return address.is_loopback
    except ValueError:
        return False


def _origin(value, *, host=False):
    if not isinstance(value, str) or not value or len(value) > (255 if host else 512):
        return None
    if re.search(r"[\s\\,@?#]", value) or (host and "/" in value):
        return None
    try:
        parsed = urlsplit("http://" + value if host else value)
        if (parsed.scheme != "http" or not _loopback(parsed.hostname, allow_name=True)
                or parsed.username or parsed.password or parsed.path not in {"", "/"}
                or parsed.query or parsed.fragment):
            return None
        port = parsed.port or 80
        return parsed.scheme, parsed.hostname, port
    except (ValueError, TypeError):
        return None


def inspect_ingress(environ):
    if not _loopback(environ.get("REMOTE_ADDR", "")):
        return "loopback_client_required"
    expected = _origin(environ.get("HTTP_HOST"), host=True)
    if expected is None:
        return "loopback_host_required"
    if "HTTP_SEC_FETCH_SITE" in environ and environ["HTTP_SEC_FETCH_SITE"] not in {"same-origin", "none"}:
        return "cross_site_request_denied"
    if "HTTP_ORIGIN" in environ and _origin(environ["HTTP_ORIGIN"]) != expected:
        return "same_origin_required"
    for name, limit in (
        ("HTTP_AUTHORIZATION", 8192),
        (AUTH_RESOLUTION_KEY, 128),
        (INTENT_CUSTODY_ENV_KEY, 8192),
    ):
        if name in environ and (not isinstance(environ[name], str) or len(environ[name]) > limit
                                or re.search(r"[\r\n\x00]", environ[name])):
            return "bounded_auth_header_required"
    return None


def same_origin_evidence(environ):
    if inspect_ingress(environ) is not None:
        return False
    if "HTTP_ORIGIN" in environ:
        return _origin(environ["HTTP_ORIGIN"]) == _origin(environ.get("HTTP_HOST"), host=True)
    return environ.get("HTTP_SEC_FETCH_SITE") == "same-origin"


def proxy_route(path):
    if path == "/api/orca-observations":
        # The operator exposes a passive, strictly sanitized projection of the
        # current terminal-state receipts.  The legacy Orca process on 8888 is
        # intentionally not launched because it also owns mutation-capable
        # command-center code.
        return 8080, "/api/orca-observations", True
    if path in {"/api/terminal-state", "/api/flight-test"}:
        return 8791, path, True
    if path == "/api" or path.startswith("/api/"):
        return 8080, path, False
    raise CloudHTTPError("cloud_api_route_invalid")


def proxy_headers(environ, bearer, port, *, forward_intent=True):
    result = {}
    for original, forwarded in (("HTTP_ACCEPT", "Accept"), ("CONTENT_TYPE", "Content-Type")):
        if original in environ:
            value = environ[original]
            if len(value) > 2048 or re.search(r"[\r\n\x00]", value):
                raise CloudHTTPError("cloud_proxy_header_invalid")
            result[forwarded] = value
    if port == 8080:
        if "HTTP_AUTHORIZATION" in environ:
            # Including empty/invalid: caller identity can never fall back to admin.
            result["Authorization"] = environ["HTTP_AUTHORIZATION"]
        elif AUTH_RESOLUTION_KEY not in environ and same_origin_evidence(environ):
            result["Authorization"] = "Bearer " + bearer
        if forward_intent and INTENT_CUSTODY_ENV_KEY in environ:
            if not same_origin_evidence(environ):
                raise CloudHTTPError("cloud_intent_custody_origin_required")
            result[INTENT_CUSTODY_HEADER] = environ[INTENT_CUSTODY_ENV_KEY]
    return result


def loopback_request(port, method, target, headers, body=b"", *, timeout=30, max_bytes=MAX_RESPONSE):
    if port not in {8080, 8791} or not target.startswith("/") or target.startswith("//"):
        raise CloudHTTPError("cloud_upstream_route_invalid")
    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=timeout)
    deadline = time.monotonic() + timeout
    timer = None
    try:
        connection.connect()
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise CloudHTTPError("cloud_upstream_unavailable")
        wire_socket = connection.sock

        def expire():
            # Socket shutdown also interrupts buffered headers/chunk framing;
            # a read timeout alone would reset on every byte of a slow response.
            try:
                wire_socket.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass

        timer = threading.Timer(remaining, expire)
        timer.daemon = True
        timer.start()
        connection.request(method, target, body=body, headers=headers)
        response = connection.getresponse()
        if 300 <= response.status < 400:
            raise CloudHTTPError("cloud_upstream_redirect_denied")
        if response.getheader("Content-Encoding", "identity") != "identity":
            raise CloudHTTPError("cloud_upstream_encoding_unsupported")
        content = response.read(max_bytes + 1)
        if time.monotonic() >= deadline:
            raise CloudHTTPError("cloud_upstream_unavailable")
        if len(content) > max_bytes:
            raise CloudHTTPError("cloud_upstream_response_too_large")
        # Whitelisted response metadata only; no cookies, redirects, CORS or auth.
        response_headers = {"Content-Type": response.getheader("Content-Type", "application/octet-stream")}
        retry = response.getheader("Retry-After")
        if retry and re.fullmatch(r"[0-9]{1,6}(?:\.[0-9]{1,3})?", retry):
            response_headers["Retry-After"] = retry
        return response.status, response_headers, content
    except (OSError, http.client.HTTPException, ValueError):
        raise CloudHTTPError("cloud_upstream_unavailable") from None
    finally:
        if timer is not None:
            timer.cancel()
        connection.close()


def api_upstream_timeout(path: str) -> int:
    clean = path.split("?", 1)[0]
    if clean in {
        "/api/cognition/reason", "/api/cognition/stream",
        "/api/operator/respond", "/api/operator/stream",
    }:
        return COGNITION_UPSTREAM_TIMEOUT_SECONDS
    return 30


def _health(role, revision, ready, *, tenant_auth_ready=False):
    return {"schema": HEALTH_SCHEMA, "role": role, "revision": revision,
            "ready": bool(ready), "deployment_hold": holds_ready(), "release_authority": False,
            "tenant_auth_ready": bool(tenant_auth_ready)}


def _json_wsgi(start_response, status, data):
    body = json.dumps(data, separators=(",", ":")).encode("utf-8")
    start_response(status, [("Content-Type", "application/json"), ("Content-Length", str(len(body))),
                            ("Cache-Control", "no-store"), ("X-Content-Type-Options", "nosniff")])
    return [body]


def operator_envelope(app, revision, *, tenant_auth_ready=False):
    """Health is the only exact-path bypass. All API identity/HNC stays in app."""
    def wrapped(environ, start_response):
        reason = inspect_ingress(environ)
        if reason:
            return _json_wsgi(start_response, "403 Forbidden", {"error": reason})
        if environ.get("PATH_INFO") == HEALTH_PATH and environ.get("REQUEST_METHOD") == "GET":
            ready = holds_ready() and tenant_auth_ready
            return _json_wsgi(start_response, "200 OK" if ready else "503 Service Unavailable",
                              _health("operator", revision, ready, tenant_auth_ready=tenant_auth_ready))
        if not holds_ready():
            return _json_wsgi(start_response, "503 Service Unavailable", {"error": "cloud_deployment_holds_required"})
        return app(environ, start_response)
    return wrapped


def validate_dist(dist_root):
    root = Path(dist_root).resolve(strict=True)
    index = root / "index.html"
    if not index.is_file() or index.is_symlink() or not 0 < index.stat().st_size <= 256 * 1024:
        raise CloudHTTPError("cloud_built_index_required")
    text = index.read_text(encoding="utf-8")
    if any(marker in text for marker in ("@vite/client", "/src/main.", "jsx-dev-runtime")):
        raise CloudHTTPError("cloud_production_build_required")
    assets = re.findall(r'(?:src|href)=["\'](/assets/[^"\']+)["\']', text)
    if not assets or not any(asset.endswith(".js") for asset in assets):
        raise CloudHTTPError("cloud_built_assets_required")
    for asset in assets:
        candidate = (root / asset.lstrip("/")).resolve(strict=True)
        if not candidate.is_relative_to(root) or not candidate.is_file():
            raise CloudHTTPError("cloud_built_asset_invalid")
    return root, tuple((root / asset.lstrip("/")).resolve() for asset in assets)


def create_ui_gateway(dist_root, revision, bearer, transport=loopback_request):
    from flask import Flask, Response, jsonify, request, send_file

    require_holds()
    validate_bearer(bearer.encode("ascii"))
    root, assets = validate_dist(dist_root)
    app = Flask("aureon_cloud_ui", static_folder=None)
    app.config["MAX_CONTENT_LENGTH"] = MAX_REQUEST

    @app.before_request
    def ingress():
        reason = inspect_ingress(request.environ)
        if reason:
            return jsonify(error=reason), 403
        if INTENT_CUSTODY_ENV_KEY in request.environ and not same_origin_evidence(request.environ):
            return jsonify(error="cloud_intent_custody_origin_required"), 403
        if not holds_ready() and request.path != HEALTH_PATH:
            return jsonify(error="cloud_deployment_holds_required"), 503
        return None

    @app.after_request
    def private_response(response):
        response.headers["Cache-Control"] = "no-store"
        response.headers["X-Content-Type-Options"] = "nosniff"
        response.headers["Referrer-Policy"] = "no-referrer"
        response.headers["X-Frame-Options"] = "DENY"
        return response

    @app.get(HEALTH_PATH)
    def health():
        ready = False
        tenant_auth_ready = False
        if holds_ready() and (root / "index.html").is_file() and all(path.is_file() for path in assets):
            try:
                status, _, body = transport(8080, "GET", HEALTH_PATH, {}, timeout=3, max_bytes=4096)
                upstream = json.loads(body)
                tenant_auth_ready = upstream.get("tenant_auth_ready") is True if isinstance(upstream, dict) else False
                ready = (status == 200 and isinstance(upstream, dict)
                         and upstream.get("schema") == HEALTH_SCHEMA and upstream.get("role") == "operator"
                         and upstream.get("revision") == revision and upstream.get("ready") is True
                         and upstream.get("deployment_hold") is True and upstream.get("release_authority") is False
                         and tenant_auth_ready)
            except (CloudHTTPError, ValueError, TypeError):
                pass
        return jsonify(_health("ui-gateway", revision, ready,
                               tenant_auth_ready=tenant_auth_ready)), 200 if ready else 503

    @app.route("/api", defaults={"rest": ""}, methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"])
    @app.route("/api/<path:rest>", methods=["GET", "HEAD", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"])
    def api(rest):
        del rest
        try:
            port, target, passive = proxy_route(request.path)
            if passive and request.method not in {"GET", "HEAD"}:
                return jsonify(error="cloud_observation_route_read_only"), 405
            if request.query_string:
                target += "?" + request.query_string.decode("ascii")
            headers = proxy_headers(request.environ, bearer, port, forward_intent=not passive)
            status, response_headers, body = transport(
                port, request.method, target, headers, request.get_data(),
                timeout=api_upstream_timeout(request.path),
            )
            return Response(body, status=status, headers=response_headers)
        except CloudHTTPError:
            # Uniform explicit unavailable response; no fallback/synthetic data.
            return jsonify(error="cloud_upstream_unavailable", available=False), 502
        except UnicodeError:
            return jsonify(error="cloud_request_encoding_invalid"), 400

    @app.get("/", defaults={"path": ""})
    @app.get("/<path:path>")
    def static_ui(path):
        parts = path.split("/")
        if "\\" in path or any(part.startswith(".") for part in parts if part):
            return jsonify(error="cloud_static_path_invalid"), 404
        candidate = (root / path).resolve()
        if not candidate.is_relative_to(root):
            return jsonify(error="cloud_static_path_invalid"), 404
        if path and candidate.is_file():
            return send_file(candidate, conditional=True)
        if path.startswith("assets/") or (path and Path(path).suffix):
            return jsonify(error="cloud_static_asset_missing"), 404
        return send_file(root / "index.html", conditional=True)

    return app


def _operator_boot_settings(root, revision, bearer, tenant_config, provider_config):
    validate_bearer(bearer.encode("ascii"))
    tenant_config = validate_supabase_tenant_config(
        json.dumps(tenant_config, sort_keys=True, separators=(",", ":")).encode("utf-8")
    )
    provider_config = validate_owner_provider_config(
        json.dumps(provider_config, sort_keys=True, separators=(",", ":")).encode("utf-8")
    )
    configured_state = {}
    for key, expected in OPERATOR_STATE_PATHS.items():
        value = str(os.environ.get(key) or expected)
        if value != expected:
            raise CloudHTTPError("cloud_operator_state_path_invalid")
        configured_state[key] = value
    owner_subject = str(os.environ.get("AUREON_SUPABASE_OWNER_SUBJECT") or "").strip().lower()
    owner_email_sha256 = str(
        os.environ.get("AUREON_SUPABASE_OWNER_EMAIL_SHA256") or ""
    ).strip().lower()
    if (
        re.fullmatch(
            r"[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
            owner_subject,
        )
        is None
        or re.fullmatch(r"[0-9a-f]{64}", owner_email_sha256) is None
    ):
        raise CloudHTTPError("cloud_owner_identity_binding_required")
    return {
        "AUREON_OPERATOR_API_KEY": bearer, "AUREON_OPERATOR_ENV": "production",
        "AUREON_OPERATOR_DEV": "0", "AUREON_OPERATOR_HOST": "127.0.0.1", "AUREON_OPERATOR_PORT": "8080",
        "AUREON_OPERATOR_RATE_RPS": "40", "AUREON_OPERATOR_RATE_BURST": "160",
        "AUREON_OPERATOR_HTTP_PROCESSES": "1", "AUREON_OPERATOR_REPLICAS": "1",
        "AUREON_OPERATOR_MAX_BODY": str(MAX_REQUEST), "AUREON_OPERATOR_TRUSTED_PROXY_CIDRS": "",
        "SERO_WORKSPACE": str(root.parent),
        **configured_state,
        "AUREON_SUPABASE_URL": tenant_config["project_url"],
        "AUREON_SUPABASE_PUBLISHABLE_KEY": tenant_config["publishable_key"],
        "AUREON_SUPABASE_OWNER_SUBJECT": owner_subject,
        "AUREON_SUPABASE_OWNER_EMAIL_SHA256": owner_email_sha256,
        "AUREON_OPERATOR_INTENT_CERT_FILE": "/etc/aureon-public/bridge-cert.pem",
        "AUREON_OPERATOR_REVISION": revision,
        "AUREON_CONTROLLER_KEY_VERSION": CONTROLLER_KEY_VERSION,
        PINNED_PROVIDER_ENV_VARS_ENV: ",".join(OWNER_PROVIDER_KEYS),
        # Exchange values stay outside the operator process in the isolated
        # executor's pinned Secret Manager bundle.  This non-secret marker
        # reserves their owner write authority even when a readback is stale.
        CLOUD_EXCHANGE_CREDENTIALS_PINNED_ENV: "1",
        CLOUD_CREDENTIAL_ISOLATION_ENV: "1",
        "PYTHON_DOTENV_DISABLED": "1",
        **CLOUD_OPERATOR_SAFE_MODE_ENV,
        **provider_config,
    }


def isolate_operator_environment(environment) -> None:
    """Remove ambient exchange authority before importing operator code.



    The mapping is mutated in place so the same boundary is usable with

    ``os.environ`` and with isolated test mappings.  Safe mode values are set

    explicitly after removal; the deployment holds remain a separate mandatory

    layer and are verified again after the application is built.

    """
    scrub_cloud_exchange_authority(environment)


def build_operator(source_root, revision, bearer, tenant_config, provider_config):
    require_holds()
    root = Path(source_root).resolve(strict=True)
    if not (root / "aureon/operator/operator_server.py").is_file():
        raise CloudHTTPError("cloud_operator_source_missing")
    settings = _operator_boot_settings(root, revision, bearer, tenant_config, provider_config)
    isolate_operator_environment(os.environ)
    # The systemd/PYTHONPATH site hook installs this before any script code in
    # production.  Install again here so direct invocations fail closed too.
    install_cloud_environment_guard()
    os.environ.update(settings)
    # Provider values were retrieved and validated in-process, after the early
    # guard started. Pin them now so a later load_dotenv(override=True) cannot
    # replace the approved bundle while exchange aliases remain impossible.
    pin_validated_environment(provider_config)
    sys.path[:0] = [str(root), str(root.parent)]
    os.chdir(root)
    from aureon.operator.operator_server import build_boot_app
    app = build_boot_app()
    require_holds()
    if any(not hmac.compare_digest(os.environ.get(key, ""), value) for key, value in settings.items()):
        raise CloudHTTPError("cloud_operator_boot_profile_changed")
    if getattr(app, "debug", False) or getattr(app, "testing", False):
        raise CloudHTTPError("cloud_operator_production_app_required")
    return operator_envelope(app, revision, tenant_auth_ready=True)


def reserve_port(role):
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        # Linux may retain a recently closed listener while public requests
        # drain.  Reuse that address after the prior owner exits, but never set
        # SO_REUSEPORT: a concurrent second owner must still fail closed.
        if os.name == "posix":
            listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        listener.bind(("127.0.0.1", PORTS[role]))
        listener.listen(128)
    except OSError:
        listener.close()
        raise CloudHTTPError("cloud_http_port_already_owned") from None
    return listener


def run(role=None, argv=None):
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--role", choices=tuple(PORTS), default=role)
    parser.add_argument("--source-root")
    parser.add_argument("--dist-root")
    parser.add_argument("--revision")
    parser.add_argument("--project", required=True)
    parser.add_argument("--secret", required=True)
    parser.add_argument("--secret-version", required=True)
    parser.add_argument("--tenant-secret")
    parser.add_argument("--tenant-secret-version")
    parser.add_argument("--provider-secret")
    parser.add_argument("--provider-secret-version")
    parser.add_argument("--secret-worker", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--secret-kind", choices=("bearer", "supabase-tenant", "owner-provider"), default="bearer",
                        help=argparse.SUPPRESS)
    args = parser.parse_args(argv)
    try:
        validate_secret_reference(args.project, args.secret, args.secret_version)
        if args.secret_worker:
            # Refuse interactive use; the credential goes only to the parent pipe.
            if sys.stdout.isatty():
                raise CloudHTTPError("cloud_secret_worker_pipe_required")
            token = fetch_secret_worker(args.project, args.secret, args.secret_version,
                                        secret_kind=args.secret_kind)
            sys.stdout.buffer.write(token.encode("ascii"))
            sys.stdout.buffer.flush()
            return 0
        require_holds()
        if args.role not in PORTS or (role is not None and args.role != role):
            raise CloudHTTPError("cloud_http_role_invalid")
        tenant_reference = bool(args.tenant_secret or args.tenant_secret_version)
        provider_reference = bool(args.provider_secret or args.provider_secret_version)
        if args.role == "operator" and not (args.tenant_secret and args.tenant_secret_version):
            raise CloudHTTPError("cloud_tenant_auth_reference_required")
        if args.role != "operator" and tenant_reference:
            raise CloudHTTPError("cloud_tenant_auth_reference_operator_only")
        if args.role == "operator":
            validate_secret_reference(args.project, args.tenant_secret, args.tenant_secret_version)
            if not (args.provider_secret and args.provider_secret_version):
                raise CloudHTTPError("cloud_owner_provider_reference_required")
            validate_secret_reference(args.project, args.provider_secret, args.provider_secret_version)
        elif provider_reference:
            raise CloudHTTPError("cloud_owner_provider_reference_operator_only")
        if not re.fullmatch(r"(?:[0-9a-f]{40}|[0-9a-f]{64})", args.revision or ""):
            raise CloudHTTPError("cloud_revision_required")
        if not args.source_root or not Path(args.source_root).is_dir():
            raise CloudHTTPError("cloud_source_root_required")
        if args.role == "ui-gateway":
            if not args.dist_root:
                raise CloudHTTPError("cloud_dist_root_required")
            validate_dist(args.dist_root)
        # Resolve production server before any cognition/services are imported.
        from waitress import create_server
        with reserve_port(args.role) as listener:
            bearer = load_cloud_secret(args.project, args.secret, args.secret_version)
            tenant_config = (load_cloud_tenant_config(args.project, args.tenant_secret,
                                                       args.tenant_secret_version)
                             if args.role == "operator" else None)
            provider_config = (load_cloud_owner_provider_config(
                args.project, args.provider_secret, args.provider_secret_version
            ) if args.role == "operator" else None)
            app = (build_operator(args.source_root, args.revision, bearer, tenant_config,
                                  provider_config) if args.role == "operator"
                   else create_ui_gateway(args.dist_root, args.revision, bearer))
            server = create_server(app, sockets=[listener], threads=8, max_request_body_size=MAX_REQUEST,
                                   max_request_header_size=32768, expose_tracebacks=False, clear_untrusted_proxy_headers=True)
            try:
                server.run()
            finally:
                server.close()
                server.task_dispatcher.shutdown(cancel_pending=True, timeout=5)
        return 0
    except CloudHTTPError as error:
        print(str(error), file=sys.stderr)
        return 1
    except Exception:
        # Do not serialize credential, upstream, configuration, or import errors.
        print("cloud_http_startup_failed", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(run())