from __future__ import annotations import hashlib import json import os import sqlite3 import ssl import subprocess import tempfile import threading import uuid from collections.abc import Mapping from contextlib import contextmanager, suppress from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import unquote, urlsplit from .sip_management import ( SipManagementError, _id, _json, _validate_trunk, asterisk_codec_allow, ) CELL_BODY_BYTES = 256 * 1024 class CellAgentError(Exception): def __init__( self, status: int, code: str, detail: str, extra: Mapping[str, Any] | None = None, ) -> None: super().__init__(detail) self.status = status self.code = code self.detail = detail self.extra = dict(extra or {}) def response(self) -> dict[str, Any]: return {"error": {"code": self.code, "message": self.detail, **self.extra}} class CellAgentStore: def __init__(self, db_path: str | Path) -> None: self.path = str(db_path) if self.path != ":memory:": Path(self.path).parent.mkdir(parents=True, exist_ok=True) self._lock = threading.RLock() self._db = sqlite3.connect(self.path, timeout=5, check_same_thread=False) self._db.row_factory = sqlite3.Row self._db.execute( "PRAGMA journal_mode = WAL" ) if self.path != ":memory:" else None self._db.execute("PRAGMA busy_timeout = 5000") self._db.executescript( """ CREATE TABLE IF NOT EXISTS trunk_state ( trunk_id TEXT PRIMARY KEY, desired_revision INTEGER NOT NULL, desired_hash TEXT NOT NULL, desired_config_json TEXT NOT NULL, applied_revision INTEGER NOT NULL, applied_hash TEXT, applied_config_json TEXT, status TEXT NOT NULL, last_error TEXT, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS audit_log ( audit_id TEXT PRIMARY KEY, trunk_id TEXT NOT NULL, revision INTEGER NOT NULL, action TEXT NOT NULL, details_json TEXT NOT NULL, created_at TEXT NOT NULL ); """ ) self._db.commit() @contextmanager def write(self): with self._lock: self._db.execute("BEGIN IMMEDIATE") try: yield self._db except Exception: self._db.rollback() raise else: self._db.commit() @contextmanager def read(self): with self._lock: yield self._db def close(self) -> None: with self._lock: self._db.close() def _now() -> str: from datetime import datetime, timezone return ( datetime.now(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) def _decode(value: str, detail: str) -> Any: try: return json.loads(value) except json.JSONDecodeError as exc: raise CellAgentError(500, "CORRUPT_STATE", detail) from exc def _hash_config(config: Mapping[str, Any]) -> str: return hashlib.sha256(_json(config).encode("utf-8")).hexdigest() def _request_id(value: str | None) -> str: if not value or len(value) > 128 or value != value.strip(): raise CellAgentError(400, "MISSING_REQUEST_ID", "X-Request-ID is required") if any(ord(char) < 32 for char in value): raise CellAgentError(400, "INVALID_REQUEST_ID", "X-Request-ID is invalid") return value def _parse_config(body: Any, trunk_id: str) -> tuple[dict[str, Any], str]: if not isinstance(body, dict): raise CellAgentError( 400, "INVALID_PUBLICATION", "publication must be an object" ) required = {"mode", "cell_id", "trunk_id", "revision", "config", "config_sha256"} if set(body) != required: raise CellAgentError( 400, "INVALID_PUBLICATION", "publication fields are invalid" ) if body["mode"] != "real": raise CellAgentError( 400, "INVALID_MODE", "Cell Agent accepts real publications only" ) if body["trunk_id"] != trunk_id: raise CellAgentError(400, "TRUNK_MISMATCH", "path and body trunk_id differ") if ( not isinstance(body["revision"], int) or isinstance(body["revision"], bool) or body["revision"] < 1 ): raise CellAgentError( 400, "INVALID_REVISION", "revision must be a positive integer" ) digest = body["config_sha256"] if ( not isinstance(digest, str) or len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest) ): raise CellAgentError( 400, "INVALID_CONFIG_HASH", "config_sha256 must be lowercase SHA-256" ) try: config = _validate_trunk(trunk_id, body["config"]) except SipManagementError as exc: raise CellAgentError(exc.status, exc.code, exc.detail, exc.extra) from exc actual = _hash_config(config) if actual != digest: raise CellAgentError( 400, "CONFIG_HASH_MISMATCH", "config_sha256 does not match config" ) return config, digest def _asterisk_name(trunk_id: str) -> str: return "agent_trunk_" + trunk_id.encode("utf-8").hex() def render_pjsip_trunk(config: Mapping[str, Any], trunk_id: str | None = None) -> str: if config["sip"]["auth_mode"] == "digest": raise CellAgentError( 422, "DIGEST_SECRET_RESOLUTION_REQUIRED", "Cell Agent requires a local secret resolver for digest trunks", ) name = _asterisk_name( trunk_id or config.get("trunk_id") or config.get("display_name", "trunk") ) sip = config["sip"] host = sip["host"] contact_host = f"[{host}]" if ":" in host and not host.startswith("[") else host scheme = "sips" if sip["transport"] == "tls" else "sip" allow = ",".join(asterisk_codec_allow(config["codec_profile"])) caller = config["caller_ids"][0] aor = f"{name}-aor" return ( f"[{name}]\n" "type=endpoint\n" f"transport=transport-{sip['transport']}\n" "context=deny-inbound\n" "disallow=all\n" f"allow={allow}\n" "direct_media=no\n" "rtp_symmetric=yes\n" "force_rport=yes\n" "rewrite_contact=yes\n" f"from_user={caller}\n" f"aors={aor}\n\n" f"[{aor}]\n" "type=aor\n" f"contact={scheme}:{contact_host}:{sip['port']}\n" "qualify_frequency=30\n" ) class AsteriskConfigApplier: def __init__( self, config_dir: str | Path, *, asterisk_cli: str = "/usr/sbin/asterisk", reload_timeout: float = 30.0, ) -> None: self.config_dir = Path(config_dir) self.asterisk_cli = asterisk_cli self.reload_timeout = reload_timeout def _target(self, trunk_id: str) -> Path: return self.config_dir / f"{_asterisk_name(trunk_id)}.conf" def _reload(self) -> None: try: result = subprocess.run( [self.asterisk_cli, "-rx", "pjsip reload"], stdin=subprocess.DEVNULL, capture_output=True, check=False, timeout=self.reload_timeout, text=True, ) except (OSError, subprocess.SubprocessError) as exc: raise CellAgentError( 502, "ASTERISK_RELOAD_FAILED", "Asterisk reload command failed" ) from exc if result.returncode != 0: raise CellAgentError( 502, "ASTERISK_RELOAD_FAILED", "Asterisk rejected the PJSIP reload" ) def apply(self, trunk_id: str, config: Mapping[str, Any]) -> None: self.config_dir.mkdir(parents=True, exist_ok=True) target = self._target(trunk_id) previous = target.read_bytes() if target.exists() else None try: if config["enabled"]: content = render_pjsip_trunk(config, trunk_id).encode("utf-8") fd, temporary = tempfile.mkstemp( prefix=target.name + ".", dir=self.config_dir ) try: with os.fdopen(fd, "wb") as stream: stream.write(content) stream.flush() os.fsync(stream.fileno()) os.chmod(temporary, 0o640) os.replace(temporary, target) finally: if os.path.exists(temporary): os.unlink(temporary) elif target.exists(): target.unlink() self._reload() except CellAgentError: self._restore(target, previous) raise def _restore(self, target: Path, previous: bytes | None) -> None: try: if previous is None: if target.exists(): target.unlink() else: fd, temporary = tempfile.mkstemp( prefix=target.name + ".restore.", dir=self.config_dir ) try: with os.fdopen(fd, "wb") as stream: stream.write(previous) stream.flush() os.fsync(stream.fileno()) os.chmod(temporary, 0o640) os.replace(temporary, target) finally: if os.path.exists(temporary): os.unlink(temporary) with suppress(CellAgentError): self._reload() except OSError: pass class CellAgentService: def __init__( self, db_path: str | Path, cell_id: str, applier: Any, *, mode: str = "real", ) -> None: if mode != "real": raise ValueError("Cell Agent only supports real mode") self.cell_id = _id(cell_id, "cell_id") self.mode = mode self.applier = applier self.store = CellAgentStore(db_path) def close(self) -> None: self.store.close() def apply(self, body: Any, request_id: str) -> dict[str, Any]: request_id = _request_id(request_id) if not isinstance(body, dict) or body.get("cell_id") != self.cell_id: raise CellAgentError( 403, "CELL_MISMATCH", "publication is for another Cell" ) trunk_id = _id(body.get("trunk_id"), "trunk_id") config, digest = _parse_config(body, trunk_id) revision = body["revision"] with self.store.write() as db: row = db.execute( "SELECT * FROM trunk_state WHERE trunk_id = ?", (trunk_id,) ).fetchone() applied_revision = row["applied_revision"] if row else 0 applied_hash = row["applied_hash"] if row else None if revision == applied_revision and digest == applied_hash: return { "mode": self.mode, "cell_id": self.cell_id, "trunk_id": trunk_id, "revision": revision, "config_sha256": digest, "status": "applied", "idempotent": True, } if revision <= applied_revision: raise CellAgentError( 409, "REVISION_CONFLICT", "publication revision is older than the applied revision", {"applied_revision": applied_revision}, ) if revision != applied_revision + 1: raise CellAgentError( 409, "REVISION_GAP", "Cell requires the next revision in sequence", {"applied_revision": applied_revision}, ) now = _now() db.execute( "INSERT INTO trunk_state(trunk_id,desired_revision,desired_hash,desired_config_json,applied_revision,applied_hash,applied_config_json,status,last_error,updated_at) VALUES(?,?,?,?,?,?,?,?,?,?) " "ON CONFLICT(trunk_id) DO UPDATE SET desired_revision=excluded.desired_revision,desired_hash=excluded.desired_hash,desired_config_json=excluded.desired_config_json,status=excluded.status,last_error=NULL,updated_at=excluded.updated_at", ( trunk_id, revision, digest, _json(config), applied_revision, applied_hash, row["applied_config_json"] if row else None, "applying", None, now, ), ) db.execute( "INSERT INTO audit_log(audit_id,trunk_id,revision,action,details_json,created_at) VALUES(?,?,?,?,?,?)", ( str(uuid.uuid4()), trunk_id, revision, "apply_started", _json({"request_id": request_id}), now, ), ) try: self.applier.apply(trunk_id, config) except CellAgentError as exc: self._mark_failed(trunk_id, revision, exc.code) raise except Exception as exc: self._mark_failed(trunk_id, revision, "APPLIER_ERROR") raise CellAgentError( 502, "ASTERISK_APPLY_FAILED", "Cell configuration apply failed" ) from exc with self.store.write() as db: now = _now() db.execute( "UPDATE trunk_state SET applied_revision=?,applied_hash=?,applied_config_json=?,status='applied',last_error=NULL,updated_at=? WHERE trunk_id=?", (revision, digest, _json(config), now, trunk_id), ) db.execute( "INSERT INTO audit_log(audit_id,trunk_id,revision,action,details_json,created_at) VALUES(?,?,?,?,?,?)", ( str(uuid.uuid4()), trunk_id, revision, "applied", _json({"request_id": request_id}), now, ), ) return { "mode": self.mode, "cell_id": self.cell_id, "trunk_id": trunk_id, "revision": revision, "config_sha256": digest, "status": "applied", "idempotent": False, } def _mark_failed(self, trunk_id: str, revision: int, error_code: str) -> None: with self.store.write() as db: db.execute( "UPDATE trunk_state SET status='failed',last_error=?,updated_at=? WHERE trunk_id=? AND desired_revision=?", (error_code, _now(), trunk_id, revision), ) db.execute( "INSERT INTO audit_log(audit_id,trunk_id,revision,action,details_json,created_at) VALUES(?,?,?,?,?,?)", ( str(uuid.uuid4()), trunk_id, revision, "apply_failed", _json({"error_code": error_code}), _now(), ), ) def state(self, trunk_id: str) -> dict[str, Any]: trunk_id = _id(trunk_id, "trunk_id") with self.store.read() as db: row = db.execute( "SELECT * FROM trunk_state WHERE trunk_id=?", (trunk_id,) ).fetchone() if not row: raise CellAgentError( 404, "TRUNK_NOT_FOUND", "trunk has not been applied" ) return { "mode": self.mode, "cell_id": self.cell_id, "trunk_id": trunk_id, "desired_revision": row["desired_revision"], "applied_revision": row["applied_revision"], "status": row["status"], "last_error": row["last_error"], "updated_at": row["updated_at"], } class CellAgentHandler(BaseHTTPRequestHandler): server_version = "agent-call-cell-agent/1" def __init__( self, request: Any, client_address: Any, server: Any, *, service: CellAgentService, ) -> None: self.service = service super().__init__(request, client_address, server) def log_message(self, format: str, *args: Any) -> None: return def _send(self, status: int, payload: Mapping[str, Any]) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _body(self) -> Any: raw_length = self.headers.get("Content-Length") try: length = int(raw_length or "-1") except ValueError as exc: raise CellAgentError( 400, "INVALID_CONTENT_LENGTH", "Content-Length is invalid" ) from exc if length < 0 or length > CELL_BODY_BYTES: raise CellAgentError( 413, "PAYLOAD_TOO_LARGE", "publication body is too large" ) try: return json.loads(self.rfile.read(length).decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise CellAgentError( 400, "INVALID_JSON", "publication body must be JSON" ) from exc def _path(self) -> list[str]: parts = [unquote(part) for part in urlsplit(self.path).path.split("/") if part] if any(part in {".", ".."} for part in parts): raise CellAgentError(400, "INVALID_PATH", "path is invalid") return parts def do_GET(self) -> None: try: parts = self._path() if parts == ["healthz", "live"]: self._send( 200, { "status": "ok", "mode": self.service.mode, "cell_id": self.service.cell_id, }, ) return if ( len(parts) == 5 and parts[:3] == ["v1", "sip", "trunks"] and parts[4] == "state" ): self._send(200, self.service.state(parts[3])) return raise CellAgentError(404, "NOT_FOUND", "endpoint does not exist") except CellAgentError as exc: self._send(exc.status, exc.response()) def do_POST(self) -> None: try: parts = self._path() if ( len(parts) != 5 or parts[:3] != ["v1", "sip", "trunks"] or parts[4] != "apply" ): raise CellAgentError(404, "NOT_FOUND", "endpoint does not exist") body = self._body() if not isinstance(body, dict) or body.get("trunk_id") != parts[3]: raise CellAgentError( 400, "TRUNK_MISMATCH", "path and body trunk_id differ" ) result = self.service.apply( body, _request_id(self.headers.get("X-Request-ID")) ) self._send(200, result) except CellAgentError as exc: self._send(exc.status, exc.response()) class CellAgentHTTPServer(ThreadingHTTPServer): allow_reuse_address = True daemon_threads = True def __init__( self, address: tuple[str, int], handler: Any, *, ssl_context: ssl.SSLContext | None = None, ) -> None: self.ssl_context = ssl_context super().__init__(address, handler) def get_request(self): connection, address = super().get_request() if self.ssl_context is None: return connection, address try: return self.ssl_context.wrap_socket(connection, server_side=True), address except Exception: connection.close() raise def make_server( service: CellAgentService, host: str = "127.0.0.1", port: int = 9443, *, ssl_context: ssl.SSLContext | None = None, ) -> CellAgentHTTPServer: return CellAgentHTTPServer( (host, port), lambda request, client_address, server: CellAgentHandler( request, client_address, server, service=service ), ssl_context=ssl_context, ) def build_server_tls_context( ca_file: str, cert_file: str, key_file: str ) -> ssl.SSLContext: context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) context.minimum_version = ssl.TLSVersion.TLSv1_2 context.verify_mode = ssl.CERT_REQUIRED try: context.load_cert_chain(certfile=cert_file, keyfile=key_file) context.load_verify_locations(cafile=ca_file) except (OSError, ssl.SSLError) as exc: raise ValueError("Cell Agent TLS files could not be loaded") from exc return context def build_from_env() -> tuple[CellAgentService, CellAgentHTTPServer]: cell_id = os.environ.get("CELL_ID") if not cell_id: raise ValueError("CELL_ID is required") config_dir = os.environ.get("CELL_ASTERISK_CONFIG_DIR", "/etc/asterisk/pjsip.d") db_path = os.environ.get("CELL_AGENT_DB", "/var/lib/agent-call/cell-agent.sqlite3") applier = AsteriskConfigApplier( config_dir, asterisk_cli=os.environ.get("CELL_ASTERISK_CLI", "/usr/sbin/asterisk"), ) service = CellAgentService(db_path, cell_id, applier) ca_file = os.environ.get("CELL_TLS_CA_FILE") cert_file = os.environ.get("CELL_TLS_CERT_FILE") key_file = os.environ.get("CELL_TLS_KEY_FILE") if not ca_file or not cert_file or not key_file: service.close() raise ValueError( "CELL_TLS_CA_FILE, CELL_TLS_CERT_FILE and CELL_TLS_KEY_FILE are required" ) try: port = int(os.environ.get("CELL_AGENT_PORT", "9443")) except ValueError as exc: service.close() raise ValueError("CELL_AGENT_PORT must be an integer") from exc server = make_server( service, os.environ.get("CELL_AGENT_HOST", "127.0.0.1"), port, ssl_context=build_server_tls_context(ca_file, cert_file, key_file), ) return service, server __all__ = [ "AsteriskConfigApplier", "CellAgentError", "CellAgentHTTPServer", "CellAgentService", "CellAgentStore", "build_from_env", "build_server_tls_context", "make_server", "render_pjsip_trunk", ]