431 lines
18 KiB
Python
431 lines
18 KiB
Python
"""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)
|