Spaces:
Running
Running
Deploy Aureon OS (deploy/selfhost stack: nginx edge + fusion API x2 + console + HNC organ + redis + postgres; paper mode) (part 7)
c626614 verified | """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 | |
| 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 | |
| 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 | |
| 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 | |
| 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 | |
| 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()) | |