"""HTTP control/query adapter for the contract-defined non-execution endpoints.""" from __future__ import annotations import hashlib import json import os from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any from urllib.parse import unquote, urlsplit from .core import AgentCallService, ServiceError, ValidationError, _integer, _loads READ_SCOPE = "outbound.read" CONTROL_SCOPE = "outbound.control" UPLOAD_SCOPE = "recording.upload" COMPLETE_SCOPE = "recording.complete" REPLAY_SCOPE = "outbound.replay" AI_CONFIG_READ_SCOPE = "ai.config.read" AI_CONFIG_PUBLISH_SCOPE = "ai.config.publish" class AgentCallHandler(BaseHTTPRequestHandler): service: AgentCallService server_version = "agent-call/0.1" def log_message(self, format: str, *args: Any) -> None: # Request IDs are enough for local diagnostics; never log auth headers or bodies. return def _request_id(self) -> str: value = self.headers.get("X-Request-ID", "") if not value or len(value) > 128 or any(ch.isspace() for ch in value): raise ServiceError("REQUEST_ID_REQUIRED", "X-Request-ID is required", 400) return value def _tenant_id(self) -> str: value = self.headers.get("X-Tenant-ID", "") if not value or len(value) > 128 or any(ch.isspace() for ch in value): raise ServiceError("TENANT_ID_REQUIRED", "X-Tenant-ID is required", 400) return value def _authorize( self, scope: str, tenant_id: str | None = None, auth_domain: str | None = None, ) -> dict[str, Any]: token_header = self.headers.get("Authorization", "") if not token_header.startswith("Bearer "): raise ServiceError("UNAUTHORIZED", "bearer authentication is required", 401) token = token_header[7:] token_map = _loads(os.environ.get("HTTP_TOKENS"), {}) if not isinstance(token_map, dict): raise ServiceError("UNAUTHORIZED", "token configuration is invalid", 401) identity = token_map.get(token) if not isinstance(identity, dict): raise ServiceError("UNAUTHORIZED", "bearer token is not authorized", 401) scopes = identity.get("scopes", []) tenant_ids = identity.get("tenant_ids", []) if not isinstance(scopes, list) or not all( isinstance(item, str) for item in scopes ): raise ServiceError("UNAUTHORIZED", "token scopes are invalid", 401) if not isinstance(tenant_ids, list) or not all( isinstance(item, str) for item in tenant_ids ): raise ServiceError("UNAUTHORIZED", "token tenant scope is invalid", 401) if scope not in scopes and "*" not in scopes: raise ServiceError("FORBIDDEN", "required scope is missing", 403) if auth_domain is not None: issuer = os.environ.get("AI_CONFIG_ISSUER") audience = os.environ.get("AI_CONFIG_AUDIENCE") if not issuer or not audience: raise ServiceError( "UNAUTHORIZED", "AI configuration authentication domain is not configured", 401, ) if identity.get("auth_domain") != auth_domain: raise ServiceError( "UNAUTHORIZED", "token authentication domain is invalid", 401 ) if identity.get("issuer") != issuer or identity.get("audience") != audience: raise ServiceError( "UNAUTHORIZED", "token issuer or audience is invalid", 401 ) if ( tenant_id is not None and "*" not in tenant_ids and tenant_id not in tenant_ids ): raise ServiceError("FORBIDDEN", "tenant scope is missing", 403) return identity def _body(self) -> dict[str, Any]: raw_length = self.headers.get("Content-Length") if not raw_length: raise ValidationError("BODY_REQUIRED", "JSON request body is required") length = _integer(raw_length) if length < 1 or length > self.service.profile.get("limits", {}).get( "max_http_bytes", 65536 ): raise ServiceError( "BODY_TOO_LARGE", "request body exceeds the HTTP limit", 413 ) try: raw = self.rfile.read(length) if len(raw) != length: raise ValidationError("BODY_TRUNCATED", "request body was truncated") body = _loads(raw.decode("utf-8")) except UnicodeDecodeError as exc: raise ValidationError( "INVALID_JSON", "request body must be UTF-8 JSON" ) from exc if not isinstance(body, dict): raise ValidationError("INVALID_JSON", "request body must be a JSON object") return body def _header(self, name: str) -> str: value = self.headers.get(name, "") if not value or len(value) > 128 or any(ch.isspace() for ch in value): raise ServiceError("HEADER_REQUIRED", f"{name} is required", 400) return value def _actor_id(self) -> str: token_header = self.headers.get("Authorization", "") token = token_header[7:] if token_header.startswith("Bearer ") else "" identity = _loads(os.environ.get("HTTP_TOKENS"), {}) principal = identity.get(token) if isinstance(identity, dict) else None if isinstance(principal, dict): value = principal.get("client_id") or principal.get("subject") if isinstance(value, str) and value: return value return "token:" + hashlib.sha256(token.encode("utf-8")).hexdigest()[:16] def _send( self, status: int, body: Any, content_type: str = "application/json" ) -> None: encoded = json.dumps( body, ensure_ascii=False, sort_keys=True, separators=(",", ":") ).encode("utf-8") self.send_response(status) self.send_header("Content-Type", f"{content_type}; charset=utf-8") self.send_header("Content-Length", str(len(encoded))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(encoded) def _send_text(self, status: int, body: str, content_type: str) -> None: encoded = body.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(encoded))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(encoded) def _error(self, exc: ServiceError, request_id: str | None = None) -> None: body = { "type": "about:blank", "title": HTTPStatus(exc.status).phrase if exc.status in HTTPStatus._value2member_map_ else "Request failed", "status": exc.status, "code": exc.code, "detail": exc.detail, "request_id": request_id or self.headers.get("X-Request-ID", "unknown"), "retryable": exc.retryable, } try: self._send(exc.status, body, "application/problem+json") except (BrokenPipeError, ConnectionResetError): return def _dispatch(self, method: str) -> None: request_id: str | None = None try: path = urlsplit(self.path).path parts = [unquote(part) for part in path.split("/") if part] if method == "GET" and path == "/healthz/live": self._send(200, {"status": "live", "mode": self.service.mode}) return if method == "GET" and path == "/healthz/ready": if not self.service.ready(): raise ServiceError( "NOT_READY", "dependency health check failed", 503, True ) self._send(200, {"status": "ready", "mode": self.service.mode}) return if method == "GET" and path == "/metrics": self._send_text( 200, self.service.metrics(), "text/plain; version=0.0.4" ) return if not ( method == "PUT" and len(parts) == 3 and parts[:2] == ["_mock", "oss"] ): request_id = self._request_id() if ( method == "POST" and len(parts) == 4 and parts == ["internal", "v1", "ai", "agent-versions"] ): tenant_id = self._tenant_id() self._authorize(AI_CONFIG_PUBLISH_SCOPE, tenant_id, "ai-config") body = self._body() if ( set(body) != {"agent_version_id", "config"} or not isinstance(body["agent_version_id"], str) or not isinstance(body["config"], dict) ): raise ValidationError( "SCHEMA_INVALID", "agent version publish requires agent_version_id and config", ) result = self.service.publish_agent_version( tenant_id, body["agent_version_id"], body["config"], self._actor_id(), ) self._send(201 if result["status"] == "published" else 200, result) return if ( method == "GET" and len(parts) == 5 and parts[:4] == ["internal", "v1", "ai", "agent-versions"] ): tenant_id = self._tenant_id() self._authorize(AI_CONFIG_READ_SCOPE, tenant_id, "ai-config") self._send(200, self.service.get_agent_version(tenant_id, parts[4])) return if ( method == "GET" and len(parts) == 5 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "commands" ): tenant_id = self._tenant_id() self._authorize(READ_SCOPE, tenant_id) self._send(200, self.service.get_command(tenant_id, parts[4])) return if ( method == "GET" and len(parts) == 5 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "calls" ): tenant_id = self._tenant_id() self._authorize(READ_SCOPE, tenant_id) self._send(200, self.service.get_call(tenant_id, parts[4])) return if ( method == "POST" and len(parts) == 6 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "tasks" and parts[5] == "controls" ): tenant_id = self._tenant_id() self._authorize(CONTROL_SCOPE, tenant_id) body = self._body() command_id = self._header("Idempotency-Key") result = self.service.control_task( tenant_id, parts[4], body, command_id, allow_hangup=self._has_scope("outbound.hangup", tenant_id), ) self._send(202, result) return if ( method == "POST" and len(parts) == 6 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "calls" and parts[5] == "replays" ): tenant_id = self._tenant_id() self._authorize(REPLAY_SCOPE, tenant_id) result = self.service.create_replay( tenant_id, "call", parts[4], self._body(), self._header("Idempotency-Key"), ) self._send(202, result) return if ( method == "POST" and len(parts) == 6 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "commands" and parts[5] == "replays" ): tenant_id = self._tenant_id() self._authorize(REPLAY_SCOPE, tenant_id) result = self.service.create_replay( tenant_id, "command", parts[4], self._body(), self._header("Idempotency-Key"), ) self._send(202, result) return if ( method == "POST" and len(parts) == 4 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "recording-uploads" ): tenant_id = self._tenant_id() self._authorize(UPLOAD_SCOPE, tenant_id) idempotency_key = self._header("Idempotency-Key") result_status, result = self.service.create_upload( tenant_id, self._body(), idempotency_key, self._actor_id() ) self._send(result_status, result) return if ( method == "POST" and len(parts) == 6 and parts[:3] == ["internal", "v1", "outbound"] and parts[3] == "recording-uploads" and parts[5] == "complete" ): tenant_id = self._tenant_id() self._authorize(COMPLETE_SCOPE, tenant_id) idempotency_key = self._header("Idempotency-Key") result = self.service.complete_upload( tenant_id, parts[4], self._body(), idempotency_key, self._actor_id() ) self._send(200, result) return if method == "PUT" and len(parts) == 3 and parts[:2] == ["_mock", "oss"]: if self.service.mode == "real": raise ServiceError( "MOCK_DISABLED", "mock object storage is disabled in real mode", 404, ) token = self._header("X-Mock-Upload-Token") raw_length = self.headers.get("Content-Length") if not raw_length: raise ValidationError("BODY_REQUIRED", "upload body is required") length = _integer(raw_length) if length < 0 or length > self.service.profile.get("limits", {}).get( "max_recording_bytes", 16777216 ): raise ServiceError( "BODY_TOO_LARGE", "recording exceeds the configured size limit", 413, ) try: data = self.rfile.read(length) except (BrokenPipeError, ConnectionResetError) as exc: raise ServiceError( "UPLOAD_INTERRUPTED", "upload connection interrupted", 400 ) from exc if len(data) != length: raise ServiceError( "UPLOAD_INTERRUPTED", "upload body was truncated", 400 ) upload = self.service.store.one( "SELECT tenant_id FROM uploads WHERE upload_id=?", (parts[2],) ) if upload is None: raise ServiceError("NOT_FOUND", "upload session not found", 404) self.service.put_upload(parts[2], token, data) self._send(200, {"status": "uploaded", "upload_id": parts[2]}) return raise ServiceError("NOT_FOUND", "route not found", 404) except ServiceError as exc: self._error(exc, request_id) except (BrokenPipeError, ConnectionResetError): return except (OSError, RuntimeError, TypeError, ValueError, KeyError): self._error( ServiceError("INTERNAL_ERROR", "request could not be completed", 500), request_id, ) def _has_scope(self, scope: str, tenant_id: str) -> bool: token_header = self.headers.get("Authorization", "") if not token_header.startswith("Bearer "): return False token_map = _loads(os.environ.get("HTTP_TOKENS"), {}) identity = ( token_map.get(token_header[7:]) if isinstance(token_map, dict) else None ) if not isinstance(identity, dict): return False tenant_ids = identity.get("tenant_ids", []) scopes = identity.get("scopes", []) if not isinstance(tenant_ids, list) or not all( isinstance(item, str) for item in tenant_ids ): return False if not isinstance(scopes, list) or not all( isinstance(item, str) for item in scopes ): return False return ("*" in tenant_ids or tenant_id in tenant_ids) and ( scope in scopes or "*" in scopes ) def do_GET(self) -> None: self._dispatch("GET") def do_POST(self) -> None: self._dispatch("POST") def do_PUT(self) -> None: self._dispatch("PUT") def make_server( service: AgentCallService, host: str = "127.0.0.1", port: int = 8080 ) -> ThreadingHTTPServer: class BoundHandler(AgentCallHandler): pass BoundHandler.service = service return ThreadingHTTPServer((host, port), BoundHandler)