1872 lines
69 KiB
Python
1872 lines
69 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import ssl
|
|
import threading
|
|
import uuid
|
|
from collections.abc import Iterator, Mapping
|
|
from contextlib import contextmanager
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from functools import partial
|
|
from http.client import HTTPSConnection
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from urllib.parse import quote, unquote, urlsplit
|
|
|
|
CODEC_TO_ASTERISK = {"PCMA": "alaw", "PCMU": "ulaw"}
|
|
ADMIN_AUDIENCE = "agent-call.sip-admin"
|
|
READ_AUDIENCE = "agent-call.sip-read"
|
|
MAX_BODY_BYTES = 64 * 1024
|
|
_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
|
|
|
|
|
|
class SipManagementError(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 ConfigurationError(ValueError):
|
|
pass
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Principal:
|
|
subject: str
|
|
issuer: str
|
|
audience: str
|
|
scopes: frozenset[str]
|
|
trunk_ids: frozenset[str]
|
|
token_class: str
|
|
|
|
|
|
def _now() -> str:
|
|
return (
|
|
datetime.now(timezone.utc)
|
|
.replace(microsecond=0)
|
|
.isoformat()
|
|
.replace("+00:00", "Z")
|
|
)
|
|
|
|
|
|
def _json(value: Any) -> str:
|
|
try:
|
|
return json.dumps(
|
|
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
|
|
)
|
|
except (TypeError, ValueError) as exc:
|
|
raise ConfigurationError("value is not JSON serializable") from exc
|
|
|
|
|
|
def _copy(value: Any) -> Any:
|
|
try:
|
|
return json.loads(json.dumps(value, ensure_ascii=False))
|
|
except (TypeError, ValueError, json.JSONDecodeError) as exc:
|
|
raise ConfigurationError("value is not JSON serializable") from exc
|
|
|
|
|
|
def _decode_json(value: str, detail: str) -> Any:
|
|
try:
|
|
return json.loads(value)
|
|
except json.JSONDecodeError as exc:
|
|
raise SipManagementError(500, "CORRUPT_STATE", detail) from exc
|
|
|
|
|
|
def _json_object(raw: str | None, detail: str) -> dict[str, Any]:
|
|
try:
|
|
value = json.loads(raw or "{}")
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigurationError(detail) from exc
|
|
if not isinstance(value, dict):
|
|
raise ConfigurationError(detail)
|
|
return value
|
|
|
|
|
|
def _id(value: Any, field: str) -> str:
|
|
if not isinstance(value, str) or not _ID_RE.fullmatch(value):
|
|
raise SipManagementError(400, "INVALID_" + field.upper(), f"{field} is invalid")
|
|
return value
|
|
|
|
|
|
def _text(value: Any, field: str, *, max_length: int = 256) -> str:
|
|
if not isinstance(value, str) or not value or len(value) > max_length:
|
|
raise SipManagementError(400, "INVALID_" + field.upper(), f"{field} is invalid")
|
|
if value != value.strip() or any(ord(char) < 32 for char in value):
|
|
raise SipManagementError(
|
|
400, "INVALID_" + field.upper(), f"{field} contains control characters"
|
|
)
|
|
return value
|
|
|
|
|
|
def _integer(
|
|
value: Any, field: str, *, minimum: int = 0, maximum: int = 2**31 - 1
|
|
) -> int:
|
|
if (
|
|
isinstance(value, bool)
|
|
or not isinstance(value, int)
|
|
or not minimum <= value <= maximum
|
|
):
|
|
raise SipManagementError(400, "INVALID_" + field.upper(), f"{field} is invalid")
|
|
return value
|
|
|
|
|
|
def _keys(value: Mapping[str, Any], allowed: set[str], field: str) -> None:
|
|
unknown = sorted(set(value) - allowed)
|
|
if unknown:
|
|
raise SipManagementError(
|
|
400,
|
|
"UNKNOWN_CONFIG_FIELD",
|
|
f"{field} contains unsupported fields",
|
|
{"fields": unknown},
|
|
)
|
|
|
|
|
|
def _codec_list(value: Any, field: str) -> list[str]:
|
|
if not isinstance(value, list) or not value or len(value) > len(CODEC_TO_ASTERISK):
|
|
raise SipManagementError(
|
|
400, "INVALID_CODEC_PROFILE", f"{field} must be a non-empty list"
|
|
)
|
|
codecs: list[str] = []
|
|
for item in value:
|
|
if not isinstance(item, str) or item.upper() not in CODEC_TO_ASTERISK:
|
|
raise SipManagementError(
|
|
400, "UNSUPPORTED_CODEC", f"unsupported codec in {field}"
|
|
)
|
|
codec = item.upper()
|
|
if codec in codecs:
|
|
raise SipManagementError(
|
|
400, "DUPLICATE_CODEC", f"duplicate codec in {field}"
|
|
)
|
|
codecs.append(codec)
|
|
return codecs
|
|
|
|
|
|
def _validate_trunk(trunk_id: str, raw: Any) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
if not isinstance(raw, dict):
|
|
raise SipManagementError(
|
|
400, "INVALID_TRUNK_CONFIG", "trunk config must be an object"
|
|
)
|
|
_keys(
|
|
raw,
|
|
{
|
|
"display_name",
|
|
"enabled",
|
|
"sip",
|
|
"codec_profile",
|
|
"caller_ids",
|
|
"dial_prefix",
|
|
"egress_pool_id",
|
|
"max_concurrency",
|
|
"max_cps",
|
|
},
|
|
"trunk",
|
|
)
|
|
display_name = _text(raw.get("display_name"), "display_name")
|
|
enabled = raw.get("enabled", True)
|
|
if not isinstance(enabled, bool):
|
|
raise SipManagementError(400, "INVALID_ENABLED", "enabled must be boolean")
|
|
|
|
sip = raw.get("sip")
|
|
if not isinstance(sip, dict):
|
|
raise SipManagementError(400, "INVALID_SIP_CONFIG", "sip must be an object")
|
|
_keys(
|
|
sip,
|
|
{"host", "port", "transport", "auth_mode", "credential_ref", "register"},
|
|
"sip",
|
|
)
|
|
host = _text(sip.get("host"), "sip.host", max_length=253)
|
|
if "/" in host or "@" in host:
|
|
raise SipManagementError(
|
|
400, "INVALID_SIP_HOST", "sip.host must be a host, not a URL or credential"
|
|
)
|
|
port = _integer(sip.get("port"), "sip.port", minimum=1, maximum=65535)
|
|
transport = sip.get("transport", "udp")
|
|
if transport not in {"udp", "tcp", "tls"}:
|
|
raise SipManagementError(
|
|
400, "INVALID_SIP_TRANSPORT", "sip.transport is unsupported"
|
|
)
|
|
auth_mode = sip.get("auth_mode", "ip")
|
|
if auth_mode not in {"ip", "digest"}:
|
|
raise SipManagementError(
|
|
400, "INVALID_AUTH_MODE", "sip.auth_mode is unsupported"
|
|
)
|
|
credential_ref = sip.get("credential_ref")
|
|
if credential_ref is not None:
|
|
credential_ref = _text(credential_ref, "sip.credential_ref", max_length=512)
|
|
if not credential_ref.startswith(("secret://", "vault://")):
|
|
raise SipManagementError(
|
|
400,
|
|
"INVALID_CREDENTIAL_REF",
|
|
"sip.credential_ref must reference a secret store",
|
|
)
|
|
if auth_mode == "digest" and not credential_ref:
|
|
raise SipManagementError(
|
|
400, "MISSING_CREDENTIAL_REF", "digest auth requires sip.credential_ref"
|
|
)
|
|
register = sip.get("register", False)
|
|
if not isinstance(register, bool):
|
|
raise SipManagementError(
|
|
400, "INVALID_REGISTER", "sip.register must be boolean"
|
|
)
|
|
|
|
codec_profile = raw.get("codec_profile")
|
|
if not isinstance(codec_profile, dict):
|
|
raise SipManagementError(
|
|
400, "INVALID_CODEC_PROFILE", "codec_profile must be an object"
|
|
)
|
|
_keys(codec_profile, {"allowed", "preferred"}, "codec_profile")
|
|
allowed_codecs = _codec_list(codec_profile.get("allowed"), "codec_profile.allowed")
|
|
preferred = codec_profile.get("preferred", allowed_codecs[0])
|
|
if not isinstance(preferred, str) or preferred.upper() not in allowed_codecs:
|
|
raise SipManagementError(
|
|
400, "INVALID_CODEC_PREFERENCE", "codec_profile.preferred must be allowed"
|
|
)
|
|
preferred = preferred.upper()
|
|
|
|
caller_ids = raw.get("caller_ids")
|
|
if not isinstance(caller_ids, list) or not caller_ids or len(caller_ids) > 100:
|
|
raise SipManagementError(
|
|
400, "INVALID_CALLER_IDS", "caller_ids must be a non-empty list"
|
|
)
|
|
normalized_callers = [
|
|
_text(item, "caller_ids", max_length=128) for item in caller_ids
|
|
]
|
|
if len(set(normalized_callers)) != len(normalized_callers):
|
|
raise SipManagementError(
|
|
400, "DUPLICATE_CALLER_ID", "caller_ids contains duplicates"
|
|
)
|
|
|
|
dial_prefix = raw.get("dial_prefix", "")
|
|
if (
|
|
not isinstance(dial_prefix, str)
|
|
or len(dial_prefix) > 32
|
|
or any(char.isspace() for char in dial_prefix)
|
|
):
|
|
raise SipManagementError(400, "INVALID_DIAL_PREFIX", "dial_prefix is invalid")
|
|
egress_pool_id = _id(raw.get("egress_pool_id"), "egress_pool_id")
|
|
|
|
return {
|
|
"display_name": display_name,
|
|
"enabled": enabled,
|
|
"sip": {
|
|
"host": host,
|
|
"port": port,
|
|
"transport": transport,
|
|
"auth_mode": auth_mode,
|
|
"credential_ref": credential_ref,
|
|
"register": register,
|
|
},
|
|
"codec_profile": {"allowed": allowed_codecs, "preferred": preferred},
|
|
"caller_ids": normalized_callers,
|
|
"dial_prefix": dial_prefix,
|
|
"egress_pool_id": egress_pool_id,
|
|
"max_concurrency": _integer(
|
|
raw.get("max_concurrency"), "max_concurrency", minimum=1
|
|
),
|
|
"max_cps": _integer(raw.get("max_cps"), "max_cps", minimum=1),
|
|
}
|
|
|
|
|
|
def _validate_cell(cell_id: str, raw: Any) -> dict[str, Any]:
|
|
cell_id = _id(cell_id, "cell_id")
|
|
if not isinstance(raw, dict):
|
|
raise SipManagementError(
|
|
400, "INVALID_CELL_CONFIG", "cell config must be an object"
|
|
)
|
|
_keys(
|
|
raw,
|
|
{
|
|
"egress_pool_id",
|
|
"codec_capabilities",
|
|
"status",
|
|
"max_concurrency",
|
|
"management_url",
|
|
},
|
|
"cell",
|
|
)
|
|
status = raw.get("status", "healthy")
|
|
if status not in {"healthy", "draining", "disabled"}:
|
|
raise SipManagementError(
|
|
400, "INVALID_CELL_STATUS", "cell.status is unsupported"
|
|
)
|
|
capabilities = _codec_list(raw.get("codec_capabilities"), "codec_capabilities")
|
|
management_url = raw.get("management_url")
|
|
if management_url is not None:
|
|
if not isinstance(management_url, str) or not management_url:
|
|
raise SipManagementError(
|
|
400, "INVALID_MANAGEMENT_URL", "cell.management_url must be a URL"
|
|
)
|
|
parsed = urlsplit(management_url)
|
|
try:
|
|
port = parsed.port
|
|
except ValueError as exc:
|
|
raise SipManagementError(
|
|
400, "INVALID_MANAGEMENT_URL", "cell.management_url port is invalid"
|
|
) from exc
|
|
if (
|
|
parsed.scheme != "https"
|
|
or not parsed.hostname
|
|
or parsed.username
|
|
or parsed.password
|
|
or parsed.query
|
|
or parsed.fragment
|
|
or (port is not None and not 1 <= port <= 65535)
|
|
):
|
|
raise SipManagementError(
|
|
400,
|
|
"INVALID_MANAGEMENT_URL",
|
|
"cell.management_url must be an HTTPS URL without credentials or query data",
|
|
)
|
|
management_url = management_url.rstrip("/")
|
|
return {
|
|
"egress_pool_id": _id(raw.get("egress_pool_id"), "egress_pool_id"),
|
|
"codec_capabilities": capabilities,
|
|
"status": status,
|
|
"max_concurrency": _integer(
|
|
raw.get("max_concurrency"), "max_concurrency", minimum=1
|
|
),
|
|
"management_url": management_url,
|
|
}
|
|
|
|
|
|
def asterisk_codec_allow(codec_profile: Mapping[str, Any]) -> list[str]:
|
|
"""Map the public codec names to Asterisk names without exposing Asterisk syntax upstream."""
|
|
return [CODEC_TO_ASTERISK[codec] for codec in codec_profile["allowed"]]
|
|
|
|
|
|
def _public_config(
|
|
config: Mapping[str, Any], *, include_asterisk: bool = False
|
|
) -> dict[str, Any]:
|
|
result = _copy(config)
|
|
result.pop("trunk_id", None)
|
|
sip = result["sip"]
|
|
credential_ref = sip.pop("credential_ref", None)
|
|
result["credential_configured"] = bool(credential_ref)
|
|
if include_asterisk:
|
|
result["asterisk_allow"] = asterisk_codec_allow(result["codec_profile"])
|
|
return result
|
|
|
|
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS trunks (
|
|
trunk_id TEXT PRIMARY KEY,
|
|
latest_revision INTEGER NOT NULL,
|
|
active_revision INTEGER NOT NULL,
|
|
active_status TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS trunk_versions (
|
|
trunk_id TEXT NOT NULL REFERENCES trunks(trunk_id),
|
|
revision INTEGER NOT NULL,
|
|
config_json TEXT NOT NULL,
|
|
state TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
created_by TEXT NOT NULL,
|
|
PRIMARY KEY (trunk_id, revision)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS cells (
|
|
cell_id TEXT PRIMARY KEY,
|
|
revision INTEGER NOT NULL,
|
|
config_json TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
updated_by TEXT NOT NULL
|
|
);
|
|
CREATE TABLE IF NOT EXISTS publications (
|
|
trunk_id TEXT NOT NULL,
|
|
revision INTEGER NOT NULL,
|
|
cell_id TEXT NOT NULL REFERENCES cells(cell_id),
|
|
status TEXT NOT NULL,
|
|
error_code TEXT,
|
|
updated_at TEXT NOT NULL,
|
|
PRIMARY KEY (trunk_id, revision, cell_id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS audit_log (
|
|
audit_id TEXT PRIMARY KEY,
|
|
resource_type TEXT NOT NULL,
|
|
resource_id TEXT NOT NULL,
|
|
action TEXT NOT NULL,
|
|
revision INTEGER NOT NULL,
|
|
actor TEXT NOT NULL,
|
|
request_id TEXT,
|
|
details_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
|
|
|
|
class SipManagementStore:
|
|
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 foreign_keys = ON")
|
|
self._db.execute("PRAGMA busy_timeout = 5000")
|
|
if self.path != ":memory:":
|
|
self._db.execute("PRAGMA journal_mode = WAL")
|
|
self._db.executescript(SCHEMA)
|
|
self._db.commit()
|
|
|
|
@contextmanager
|
|
def write(self) -> Iterator[sqlite3.Connection]:
|
|
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) -> Iterator[sqlite3.Connection]:
|
|
with self._lock:
|
|
yield self._db
|
|
|
|
def close(self) -> None:
|
|
with self._lock:
|
|
self._db.close()
|
|
|
|
|
|
class CellPublishError(Exception):
|
|
def __init__(self, code: str, detail: str) -> None:
|
|
super().__init__(detail)
|
|
self.code = code
|
|
self.detail = detail
|
|
|
|
|
|
class MtlsCellPublisher:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
ca_file: str,
|
|
cert_file: str,
|
|
key_file: str,
|
|
timeout: float = 15.0,
|
|
) -> None:
|
|
if not timeout > 0:
|
|
raise ConfigurationError(
|
|
"SIP_CELL_PUBLISH_TIMEOUT_SECONDS must be positive"
|
|
)
|
|
try:
|
|
context = ssl.create_default_context(
|
|
ssl.Purpose.SERVER_AUTH, cafile=ca_file
|
|
)
|
|
context.load_cert_chain(certfile=cert_file, keyfile=key_file)
|
|
except (OSError, ssl.SSLError) as exc:
|
|
raise ConfigurationError(
|
|
"Cell publisher TLS files could not be loaded"
|
|
) from exc
|
|
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
context.check_hostname = True
|
|
self._context = context
|
|
self._timeout = timeout
|
|
|
|
def publish(
|
|
self,
|
|
cell: Mapping[str, Any],
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: Mapping[str, Any],
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
management_url = cell.get("management_url")
|
|
if not management_url:
|
|
raise CellPublishError(
|
|
"CELL_ENDPOINT_MISSING", "Cell Agent endpoint is not configured"
|
|
)
|
|
parsed = urlsplit(management_url)
|
|
try:
|
|
port = parsed.port or 443
|
|
except ValueError as exc:
|
|
raise CellPublishError(
|
|
"CELL_ENDPOINT_INVALID", "Cell Agent endpoint is invalid"
|
|
) from exc
|
|
if parsed.scheme != "https" or not parsed.hostname:
|
|
raise CellPublishError(
|
|
"CELL_ENDPOINT_INVALID", "Cell Agent endpoint must use HTTPS"
|
|
)
|
|
path = (
|
|
f"{parsed.path.rstrip('/')}/v1/sip/trunks/{quote(trunk_id, safe='')}/apply"
|
|
)
|
|
if not path.startswith("/"):
|
|
path = "/" + path
|
|
digest = hashlib.sha256(_json(config).encode("utf-8")).hexdigest()
|
|
body = _json(
|
|
{
|
|
"mode": "real",
|
|
"cell_id": cell["cell_id"],
|
|
"trunk_id": trunk_id,
|
|
"revision": revision,
|
|
"config": config,
|
|
"config_sha256": digest,
|
|
}
|
|
).encode("utf-8")
|
|
connection: HTTPSConnection | None = None
|
|
try:
|
|
connection = HTTPSConnection(
|
|
parsed.hostname,
|
|
port,
|
|
context=self._context,
|
|
timeout=self._timeout,
|
|
)
|
|
connection.request(
|
|
"POST",
|
|
path,
|
|
body=body,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json",
|
|
"Content-Length": str(len(body)),
|
|
"X-Request-ID": request_id,
|
|
},
|
|
)
|
|
response = connection.getresponse()
|
|
raw = response.read(MAX_BODY_BYTES + 1)
|
|
except (OSError, ValueError) as exc:
|
|
raise CellPublishError(
|
|
"CELL_UNREACHABLE", "Cell Agent request failed"
|
|
) from exc
|
|
finally:
|
|
if connection is not None:
|
|
connection.close()
|
|
if response.status < 200 or response.status >= 300:
|
|
code = "CELL_REJECTED"
|
|
try:
|
|
error = json.loads(raw.decode("utf-8")).get("error", {})
|
|
if isinstance(error, dict) and isinstance(error.get("code"), str):
|
|
code = error["code"]
|
|
except (UnicodeDecodeError, json.JSONDecodeError, AttributeError):
|
|
pass
|
|
raise CellPublishError(code, "Cell Agent rejected the publication")
|
|
try:
|
|
acknowledgement = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise CellPublishError(
|
|
"CELL_INVALID_ACK", "Cell Agent returned invalid JSON"
|
|
) from exc
|
|
if not isinstance(acknowledgement, dict):
|
|
raise CellPublishError(
|
|
"CELL_INVALID_ACK", "Cell Agent acknowledgement is invalid"
|
|
)
|
|
if (
|
|
acknowledgement.get("status") != "applied"
|
|
or acknowledgement.get("mode") != "real"
|
|
or acknowledgement.get("cell_id") != cell["cell_id"]
|
|
or acknowledgement.get("trunk_id") != trunk_id
|
|
or acknowledgement.get("revision") != revision
|
|
or acknowledgement.get("config_sha256") != digest
|
|
):
|
|
raise CellPublishError(
|
|
"CELL_INVALID_ACK",
|
|
"Cell Agent acknowledgement does not match publication",
|
|
)
|
|
return acknowledgement
|
|
|
|
|
|
class SipManagementService:
|
|
def __init__(
|
|
self,
|
|
db_path: str | Path,
|
|
*,
|
|
mode: str = "mock",
|
|
publisher: MtlsCellPublisher | Any | None = None,
|
|
) -> None:
|
|
if mode not in {"mock", "real"}:
|
|
raise ConfigurationError("SIP_MANAGEMENT_MODE must be mock or real")
|
|
if mode == "real" and publisher is None:
|
|
raise ConfigurationError("real mode requires an mTLS Cell publisher")
|
|
self.mode = mode
|
|
self.publisher = publisher
|
|
self.store = SipManagementStore(db_path)
|
|
|
|
def close(self) -> None:
|
|
self.store.close()
|
|
|
|
def _audit(
|
|
self,
|
|
db: sqlite3.Connection,
|
|
resource_type: str,
|
|
resource_id: str,
|
|
action: str,
|
|
revision: int,
|
|
actor: str,
|
|
request_id: str | None,
|
|
details: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
db.execute(
|
|
"INSERT INTO audit_log(audit_id,resource_type,resource_id,action,revision,actor,request_id,details_json,created_at) VALUES(?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
str(uuid.uuid4()),
|
|
resource_type,
|
|
resource_id,
|
|
action,
|
|
revision,
|
|
actor,
|
|
request_id,
|
|
_json(details or {}),
|
|
_now(),
|
|
),
|
|
)
|
|
|
|
@staticmethod
|
|
def _row_config(row: sqlite3.Row | None) -> dict[str, Any] | None:
|
|
return (
|
|
_decode_json(row["config_json"], "stored Trunk config is invalid")
|
|
if row
|
|
else None
|
|
)
|
|
|
|
def _compatible_cell_records(
|
|
self,
|
|
db: sqlite3.Connection,
|
|
config: Mapping[str, Any],
|
|
*,
|
|
healthy_only: bool = True,
|
|
) -> list[dict[str, Any]]:
|
|
cells = db.execute(
|
|
"SELECT cell_id, config_json FROM cells ORDER BY cell_id"
|
|
).fetchall()
|
|
allowed = set(config["codec_profile"]["allowed"])
|
|
result: list[dict[str, Any]] = []
|
|
for cell in cells:
|
|
cell_config = _decode_json(
|
|
cell["config_json"], "stored Cell config is invalid"
|
|
)
|
|
if healthy_only and cell_config["status"] != "healthy":
|
|
continue
|
|
if cell_config["egress_pool_id"] != config["egress_pool_id"]:
|
|
continue
|
|
if allowed.intersection(cell_config["codec_capabilities"]):
|
|
result.append({"cell_id": cell["cell_id"], **cell_config})
|
|
return result
|
|
|
|
def _compatible_cells(
|
|
self, db: sqlite3.Connection, config: Mapping[str, Any]
|
|
) -> list[str]:
|
|
return [cell["cell_id"] for cell in self._compatible_cell_records(db, config)]
|
|
|
|
def _trunk_summary(
|
|
self, db: sqlite3.Connection, trunk_id: str, *, admin: bool
|
|
) -> dict[str, Any]:
|
|
trunk = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id = ?", (trunk_id,)
|
|
).fetchone()
|
|
if not trunk:
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
latest = db.execute(
|
|
"SELECT * FROM trunk_versions WHERE trunk_id = ? AND revision = ?",
|
|
(trunk_id, trunk["latest_revision"]),
|
|
).fetchone()
|
|
active = None
|
|
if trunk["active_revision"]:
|
|
active = db.execute(
|
|
"SELECT * FROM trunk_versions WHERE trunk_id = ? AND revision = ?",
|
|
(trunk_id, trunk["active_revision"]),
|
|
).fetchone()
|
|
latest_config = self._row_config(latest)
|
|
active_config = self._row_config(active)
|
|
selected_config = active_config or latest_config
|
|
compatible = (
|
|
self._compatible_cells(db, selected_config) if selected_config else []
|
|
)
|
|
active_view = (
|
|
_public_config(active_config, include_asterisk=admin)
|
|
if active_config
|
|
else None
|
|
)
|
|
if not admin:
|
|
return {
|
|
"mode": self.mode,
|
|
"trunk_id": trunk_id,
|
|
"revision": trunk["active_revision"],
|
|
"status": trunk["active_status"],
|
|
"updated_at": trunk["updated_at"],
|
|
"config": active_view,
|
|
}
|
|
result: dict[str, Any] = {
|
|
"mode": self.mode,
|
|
"trunk_id": trunk_id,
|
|
"latest_revision": trunk["latest_revision"],
|
|
"active_revision": trunk["active_revision"],
|
|
"status": trunk["active_status"],
|
|
"updated_at": trunk["updated_at"],
|
|
"compatible_cell_ids": compatible,
|
|
"latest": _public_config(latest_config, include_asterisk=True)
|
|
if latest_config
|
|
else None,
|
|
"active": active_view,
|
|
}
|
|
result["versions"] = [
|
|
{
|
|
"revision": row["revision"],
|
|
"state": row["state"],
|
|
"created_at": row["created_at"],
|
|
"created_by": row["created_by"],
|
|
}
|
|
for row in db.execute(
|
|
"SELECT revision,state,created_at,created_by FROM trunk_versions WHERE trunk_id = ? ORDER BY revision",
|
|
(trunk_id,),
|
|
).fetchall()
|
|
]
|
|
return result
|
|
|
|
def upsert_cell(
|
|
self,
|
|
cell_id: str,
|
|
payload: Any,
|
|
*,
|
|
expected_revision: int,
|
|
actor: str,
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
config = _validate_cell(cell_id, payload)
|
|
if self.mode == "real" and not config["management_url"]:
|
|
raise SipManagementError(
|
|
422,
|
|
"CELL_ENDPOINT_MISSING",
|
|
"real mode requires cell.management_url",
|
|
)
|
|
_integer(expected_revision, "expected_revision")
|
|
with self.store.write() as db:
|
|
row = db.execute(
|
|
"SELECT revision FROM cells WHERE cell_id = ?", (cell_id,)
|
|
).fetchone()
|
|
current = row["revision"] if row else 0
|
|
if expected_revision != current:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"cell revision changed",
|
|
{
|
|
"expected_revision": expected_revision,
|
|
"current_revision": current,
|
|
},
|
|
)
|
|
revision = current + 1
|
|
now = _now()
|
|
db.execute(
|
|
"INSERT INTO cells(cell_id,revision,config_json,updated_at,updated_by) VALUES(?,?,?,?,?) "
|
|
"ON CONFLICT(cell_id) DO UPDATE SET revision=excluded.revision,config_json=excluded.config_json,updated_at=excluded.updated_at,updated_by=excluded.updated_by",
|
|
(cell_id, revision, _json(config), now, actor),
|
|
)
|
|
self._audit(db, "cell", cell_id, "upsert", revision, actor, request_id)
|
|
return self.get_cell(cell_id)
|
|
|
|
def get_cell(self, cell_id: str) -> dict[str, Any]:
|
|
cell_id = _id(cell_id, "cell_id")
|
|
with self.store.read() as db:
|
|
row = db.execute(
|
|
"SELECT * FROM cells WHERE cell_id = ?", (cell_id,)
|
|
).fetchone()
|
|
if not row:
|
|
raise SipManagementError(404, "CELL_NOT_FOUND", "cell does not exist")
|
|
config = _decode_json(row["config_json"], "stored Cell config is invalid")
|
|
return {
|
|
"mode": self.mode,
|
|
"cell_id": cell_id,
|
|
"revision": row["revision"],
|
|
"config": config,
|
|
"updated_at": row["updated_at"],
|
|
"updated_by": row["updated_by"],
|
|
}
|
|
|
|
def list_cells(self) -> list[dict[str, Any]]:
|
|
with self.store.read() as db:
|
|
rows = db.execute("SELECT * FROM cells ORDER BY cell_id").fetchall()
|
|
return [
|
|
{
|
|
"mode": self.mode,
|
|
"cell_id": row["cell_id"],
|
|
"revision": row["revision"],
|
|
"config": _decode_json(
|
|
row["config_json"], "stored Cell config is invalid"
|
|
),
|
|
"updated_at": row["updated_at"],
|
|
"updated_by": row["updated_by"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
def upsert_trunk(
|
|
self,
|
|
trunk_id: str,
|
|
payload: Any,
|
|
*,
|
|
expected_revision: int,
|
|
actor: str,
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
config = _validate_trunk(trunk_id, payload)
|
|
_integer(expected_revision, "expected_revision")
|
|
with self.store.write() as db:
|
|
row = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id = ?", (trunk_id,)
|
|
).fetchone()
|
|
current = row["latest_revision"] if row else 0
|
|
if expected_revision != current:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"trunk revision changed",
|
|
{
|
|
"expected_revision": expected_revision,
|
|
"current_revision": current,
|
|
},
|
|
)
|
|
latest = (
|
|
db.execute(
|
|
"SELECT state FROM trunk_versions WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, current),
|
|
).fetchone()
|
|
if row
|
|
else None
|
|
)
|
|
if latest and latest["state"] == "publishing":
|
|
raise SipManagementError(
|
|
409,
|
|
"PUBLISH_IN_PROGRESS",
|
|
"trunk publication is still in progress",
|
|
)
|
|
revision = current + 1
|
|
now = _now()
|
|
if row:
|
|
db.execute(
|
|
"UPDATE trunks SET latest_revision=?,updated_at=? WHERE trunk_id=?",
|
|
(revision, now, trunk_id),
|
|
)
|
|
else:
|
|
db.execute(
|
|
"INSERT INTO trunks(trunk_id,latest_revision,active_revision,active_status,created_at,updated_at) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, 0, "draft", now, now),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO trunk_versions(trunk_id,revision,config_json,state,created_at,created_by) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, _json(config), "draft", now, actor),
|
|
)
|
|
self._audit(db, "trunk", trunk_id, "upsert", revision, actor, request_id)
|
|
return self.get_trunk(trunk_id)
|
|
|
|
def get_trunk(self, trunk_id: str) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
with self.store.read() as db:
|
|
return self._trunk_summary(db, trunk_id, admin=True)
|
|
|
|
def list_trunks(self) -> list[dict[str, Any]]:
|
|
with self.store.read() as db:
|
|
rows = db.execute(
|
|
"SELECT trunk_id FROM trunks ORDER BY trunk_id"
|
|
).fetchall()
|
|
return [
|
|
self._trunk_summary(db, row["trunk_id"], admin=True) for row in rows
|
|
]
|
|
|
|
def list_readonly_trunks(
|
|
self, allowed_trunk_ids: frozenset[str]
|
|
) -> list[dict[str, Any]]:
|
|
with self.store.read() as db:
|
|
rows = db.execute(
|
|
"SELECT trunk_id FROM trunks WHERE active_status = 'published' ORDER BY trunk_id"
|
|
).fetchall()
|
|
result: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
if (
|
|
"*" not in allowed_trunk_ids
|
|
and row["trunk_id"] not in allowed_trunk_ids
|
|
):
|
|
continue
|
|
result.append(self._trunk_summary(db, row["trunk_id"], admin=False))
|
|
return result
|
|
|
|
def get_readonly_trunk(
|
|
self, trunk_id: str, allowed_trunk_ids: frozenset[str]
|
|
) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
if "*" not in allowed_trunk_ids and trunk_id not in allowed_trunk_ids:
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
with self.store.read() as db:
|
|
summary = self._trunk_summary(db, trunk_id, admin=False)
|
|
if summary["status"] != "published":
|
|
raise SipManagementError(
|
|
404, "TRUNK_NOT_FOUND", "trunk is not published"
|
|
)
|
|
return summary
|
|
|
|
def _staged_publication_cells(
|
|
self,
|
|
db: sqlite3.Connection,
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: Mapping[str, Any],
|
|
) -> list[dict[str, Any]]:
|
|
rows = db.execute(
|
|
"SELECT p.cell_id,c.config_json FROM publications p LEFT JOIN cells c ON c.cell_id=p.cell_id WHERE p.trunk_id=? AND p.revision=? ORDER BY p.cell_id",
|
|
(trunk_id, revision),
|
|
).fetchall()
|
|
if not rows:
|
|
raise SipManagementError(
|
|
409, "PUBLICATION_NOT_FOUND", "publication has no Cell intents"
|
|
)
|
|
result: list[dict[str, Any]] = []
|
|
for row in rows:
|
|
if row["config_json"] is None:
|
|
result.append(
|
|
{
|
|
"cell_id": row["cell_id"],
|
|
"egress_pool_id": config["egress_pool_id"],
|
|
"codec_capabilities": [],
|
|
"status": "unavailable",
|
|
"max_concurrency": 1,
|
|
"management_url": None,
|
|
}
|
|
)
|
|
continue
|
|
cell_config = _decode_json(
|
|
row["config_json"], "stored Cell config is invalid"
|
|
)
|
|
result.append({"cell_id": row["cell_id"], **cell_config})
|
|
return result
|
|
|
|
def _deliver_publication(
|
|
self,
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: Mapping[str, Any],
|
|
cells: list[dict[str, Any]],
|
|
request_id: str,
|
|
) -> list[dict[str, str]]:
|
|
if self.mode == "mock":
|
|
return []
|
|
publisher: Any = self.publisher
|
|
if publisher is None:
|
|
raise ConfigurationError("real mode requires an mTLS Cell publisher")
|
|
failures: list[dict[str, str]] = []
|
|
for cell in cells:
|
|
with self.store.read() as db:
|
|
publication = db.execute(
|
|
"SELECT status FROM publications WHERE trunk_id=? AND revision=? AND cell_id=?",
|
|
(trunk_id, revision, cell["cell_id"]),
|
|
).fetchone()
|
|
if publication and publication["status"] == "applied":
|
|
continue
|
|
try:
|
|
acknowledgement = publisher.publish(
|
|
cell, trunk_id, revision, config, request_id
|
|
)
|
|
if (
|
|
not isinstance(acknowledgement, dict)
|
|
or acknowledgement.get("status") != "applied"
|
|
):
|
|
raise CellPublishError(
|
|
"CELL_INVALID_ACK", "Cell Agent did not acknowledge applied"
|
|
)
|
|
status = "applied"
|
|
error_code = None
|
|
except CellPublishError as exc:
|
|
status = "failed"
|
|
error_code = exc.code
|
|
failures.append({"cell_id": cell["cell_id"], "error_code": exc.code})
|
|
except (OSError, TypeError, ValueError):
|
|
status = "failed"
|
|
error_code = "CELL_PUBLISH_FAILED"
|
|
failures.append({"cell_id": cell["cell_id"], "error_code": error_code})
|
|
with self.store.write() as db:
|
|
db.execute(
|
|
"UPDATE publications SET status=?,error_code=?,updated_at=? WHERE trunk_id=? AND revision=? AND cell_id=?",
|
|
(
|
|
status,
|
|
error_code,
|
|
_now(),
|
|
trunk_id,
|
|
revision,
|
|
cell["cell_id"],
|
|
),
|
|
)
|
|
return failures
|
|
|
|
def _complete_publication(
|
|
self,
|
|
trunk_id: str,
|
|
revision: int,
|
|
config: Mapping[str, Any],
|
|
cells: list[dict[str, Any]],
|
|
failures: list[dict[str, str]],
|
|
*,
|
|
action: str,
|
|
actor: str,
|
|
request_id: str,
|
|
details: Mapping[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
details = {"cell_ids": [cell["cell_id"] for cell in cells], **(details or {})}
|
|
with self.store.write() as db:
|
|
trunk = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone()
|
|
if not trunk or trunk["latest_revision"] != revision:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"publication revision changed before completion",
|
|
)
|
|
if failures:
|
|
db.execute(
|
|
"UPDATE trunk_versions SET state='draft' WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, revision),
|
|
)
|
|
self._audit(
|
|
db,
|
|
"trunk",
|
|
trunk_id,
|
|
f"{action}_failed",
|
|
revision,
|
|
actor,
|
|
request_id,
|
|
{
|
|
**details,
|
|
"delivery_status": "failed",
|
|
"failures": failures,
|
|
},
|
|
)
|
|
else:
|
|
if trunk["active_revision"]:
|
|
db.execute(
|
|
"UPDATE trunk_versions SET state='superseded' WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, trunk["active_revision"]),
|
|
)
|
|
db.execute(
|
|
"UPDATE trunk_versions SET state='published' WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, revision),
|
|
)
|
|
status = "published" if config["enabled"] else "disabled"
|
|
db.execute(
|
|
"UPDATE trunks SET active_revision=?,active_status=?,updated_at=? WHERE trunk_id=?",
|
|
(revision, status, _now(), trunk_id),
|
|
)
|
|
if self.mode == "real":
|
|
db.execute(
|
|
"UPDATE publications SET status='applied',error_code=NULL,updated_at=? WHERE trunk_id=? AND revision=?",
|
|
(_now(), trunk_id, revision),
|
|
)
|
|
self._audit(
|
|
db,
|
|
"trunk",
|
|
trunk_id,
|
|
action,
|
|
revision,
|
|
actor,
|
|
request_id,
|
|
{
|
|
**details,
|
|
"delivery_status": "applied"
|
|
if self.mode == "real"
|
|
else "pending",
|
|
},
|
|
)
|
|
return self._trunk_summary(db, trunk_id, admin=True)
|
|
|
|
def _raise_delivery_failure(
|
|
self, action: str, failures: list[dict[str, str]]
|
|
) -> None:
|
|
if failures:
|
|
raise SipManagementError(
|
|
502,
|
|
"CELL_PUBLISH_FAILED",
|
|
f"{action} was not applied by every Cell",
|
|
{"failures": failures},
|
|
)
|
|
|
|
def publish_trunk(
|
|
self,
|
|
trunk_id: str,
|
|
*,
|
|
expected_revision: int,
|
|
actor: str,
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
_integer(expected_revision, "expected_revision")
|
|
with self.store.write() as db:
|
|
trunk = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone()
|
|
if not trunk:
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
if trunk["latest_revision"] != expected_revision:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"trunk revision changed",
|
|
{
|
|
"expected_revision": expected_revision,
|
|
"current_revision": trunk["latest_revision"],
|
|
},
|
|
)
|
|
version = db.execute(
|
|
"SELECT * FROM trunk_versions WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, expected_revision),
|
|
).fetchone()
|
|
if not version:
|
|
raise SipManagementError(
|
|
409, "VERSION_NOT_FOUND", "trunk version does not exist"
|
|
)
|
|
resume = version["state"] == "publishing"
|
|
if (
|
|
trunk["active_revision"] == expected_revision
|
|
and trunk["active_status"] == "published"
|
|
):
|
|
return self._trunk_summary(db, trunk_id, admin=True)
|
|
config = _decode_json(
|
|
version["config_json"], "stored Trunk config is invalid"
|
|
)
|
|
if not config["enabled"]:
|
|
raise SipManagementError(
|
|
409, "TRUNK_DISABLED", "disabled config cannot be published"
|
|
)
|
|
if resume:
|
|
if self.mode != "real":
|
|
raise SipManagementError(
|
|
409,
|
|
"PUBLISH_IN_PROGRESS",
|
|
"mock publication cannot be resumed as a Cell delivery",
|
|
)
|
|
cells = self._staged_publication_cells(
|
|
db, trunk_id, expected_revision, config
|
|
)
|
|
else:
|
|
cells = self._compatible_cell_records(db, config)
|
|
if not cells:
|
|
raise SipManagementError(
|
|
409,
|
|
"NO_CAPABLE_CELL",
|
|
"no healthy Cell supports the Trunk egress pool and codec policy",
|
|
)
|
|
now = _now()
|
|
db.execute(
|
|
"UPDATE trunk_versions SET state='publishing' WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, expected_revision),
|
|
)
|
|
db.execute(
|
|
"DELETE FROM publications WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, expected_revision),
|
|
)
|
|
for cell in cells:
|
|
db.execute(
|
|
"INSERT INTO publications(trunk_id,revision,cell_id,status,error_code,updated_at) VALUES(?,?,?,?,?,?)",
|
|
(
|
|
trunk_id,
|
|
expected_revision,
|
|
cell["cell_id"],
|
|
"pending",
|
|
None,
|
|
now,
|
|
),
|
|
)
|
|
failures = self._deliver_publication(
|
|
trunk_id, expected_revision, config, cells, request_id
|
|
)
|
|
result = self._complete_publication(
|
|
trunk_id,
|
|
expected_revision,
|
|
config,
|
|
cells,
|
|
failures,
|
|
action="publish",
|
|
actor=actor,
|
|
request_id=request_id,
|
|
)
|
|
self._raise_delivery_failure("publish", failures)
|
|
return result
|
|
|
|
def disable_trunk(
|
|
self,
|
|
trunk_id: str,
|
|
*,
|
|
expected_revision: int,
|
|
actor: str,
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
_integer(expected_revision, "expected_revision")
|
|
with self.store.write() as db:
|
|
trunk = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone()
|
|
if not trunk:
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
if trunk["latest_revision"] != expected_revision:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"trunk revision changed",
|
|
{
|
|
"expected_revision": expected_revision,
|
|
"current_revision": trunk["latest_revision"],
|
|
},
|
|
)
|
|
latest = db.execute(
|
|
"SELECT * FROM trunk_versions WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, expected_revision),
|
|
).fetchone()
|
|
if not latest:
|
|
raise SipManagementError(
|
|
409, "VERSION_NOT_FOUND", "trunk version does not exist"
|
|
)
|
|
resume = latest["state"] == "publishing"
|
|
if resume:
|
|
config = _decode_json(
|
|
latest["config_json"], "stored Trunk config is invalid"
|
|
)
|
|
if config["enabled"]:
|
|
raise SipManagementError(
|
|
409,
|
|
"PUBLISH_IN_PROGRESS",
|
|
"another publication is still in progress",
|
|
)
|
|
cells = self._staged_publication_cells(
|
|
db, trunk_id, expected_revision, config
|
|
)
|
|
revision = expected_revision
|
|
else:
|
|
if trunk["active_status"] == "disabled":
|
|
return self._trunk_summary(db, trunk_id, admin=True)
|
|
config = _decode_json(
|
|
latest["config_json"], "stored Trunk config is invalid"
|
|
)
|
|
config["enabled"] = False
|
|
config = _validate_trunk(trunk_id, config)
|
|
cells = self._compatible_cell_records(db, config, healthy_only=False)
|
|
if self.mode == "real" and not cells:
|
|
raise SipManagementError(
|
|
409,
|
|
"NO_CAPABLE_CELL",
|
|
"no Cell can receive the disable publication",
|
|
)
|
|
revision = expected_revision + 1
|
|
now = _now()
|
|
db.execute(
|
|
"UPDATE trunks SET latest_revision=?,updated_at=? WHERE trunk_id=?",
|
|
(revision, now, trunk_id),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO trunk_versions(trunk_id,revision,config_json,state,created_at,created_by) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, _json(config), "publishing", now, actor),
|
|
)
|
|
for cell in cells:
|
|
db.execute(
|
|
"INSERT INTO publications(trunk_id,revision,cell_id,status,error_code,updated_at) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, cell["cell_id"], "pending", None, now),
|
|
)
|
|
failures = self._deliver_publication(
|
|
trunk_id, revision, config, cells, request_id
|
|
)
|
|
result = self._complete_publication(
|
|
trunk_id,
|
|
revision,
|
|
config,
|
|
cells,
|
|
failures,
|
|
action="disable",
|
|
actor=actor,
|
|
request_id=request_id,
|
|
)
|
|
self._raise_delivery_failure("disable", failures)
|
|
return result
|
|
|
|
def rollback_trunk(
|
|
self,
|
|
trunk_id: str,
|
|
target_revision: int,
|
|
*,
|
|
expected_revision: int,
|
|
actor: str,
|
|
request_id: str,
|
|
) -> dict[str, Any]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
_integer(target_revision, "target_revision", minimum=1)
|
|
_integer(expected_revision, "expected_revision")
|
|
with self.store.write() as db:
|
|
trunk = db.execute(
|
|
"SELECT * FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone()
|
|
if not trunk:
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
if trunk["latest_revision"] != expected_revision:
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"trunk revision changed",
|
|
{
|
|
"expected_revision": expected_revision,
|
|
"current_revision": trunk["latest_revision"],
|
|
},
|
|
)
|
|
latest = db.execute(
|
|
"SELECT state,config_json FROM trunk_versions WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, expected_revision),
|
|
).fetchone()
|
|
resume = bool(latest and latest["state"] == "publishing")
|
|
source = db.execute(
|
|
"SELECT config_json FROM trunk_versions WHERE trunk_id=? AND revision=?",
|
|
(trunk_id, target_revision),
|
|
).fetchone()
|
|
if not source:
|
|
raise SipManagementError(
|
|
404, "VERSION_NOT_FOUND", "rollback target does not exist"
|
|
)
|
|
config = _decode_json(
|
|
source["config_json"], "stored Trunk config is invalid"
|
|
)
|
|
if not config["enabled"]:
|
|
raise SipManagementError(
|
|
409, "TRUNK_DISABLED", "disabled config cannot be rolled back to"
|
|
)
|
|
if resume:
|
|
staged = _decode_json(
|
|
latest["config_json"], "stored Trunk config is invalid"
|
|
)
|
|
if _json(staged) != _json(config):
|
|
raise SipManagementError(
|
|
409,
|
|
"REVISION_CONFLICT",
|
|
"latest publication is for a different rollback target",
|
|
)
|
|
cells = self._staged_publication_cells(
|
|
db, trunk_id, expected_revision, config
|
|
)
|
|
revision = expected_revision
|
|
else:
|
|
cells = self._compatible_cell_records(db, config)
|
|
if not cells:
|
|
raise SipManagementError(
|
|
409, "NO_CAPABLE_CELL", "rollback target has no capable Cell"
|
|
)
|
|
revision = expected_revision + 1
|
|
now = _now()
|
|
db.execute(
|
|
"UPDATE trunks SET latest_revision=?,updated_at=? WHERE trunk_id=?",
|
|
(revision, now, trunk_id),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO trunk_versions(trunk_id,revision,config_json,state,created_at,created_by) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, _json(config), "publishing", now, actor),
|
|
)
|
|
for cell in cells:
|
|
db.execute(
|
|
"INSERT INTO publications(trunk_id,revision,cell_id,status,error_code,updated_at) VALUES(?,?,?,?,?,?)",
|
|
(trunk_id, revision, cell["cell_id"], "pending", None, now),
|
|
)
|
|
failures = self._deliver_publication(
|
|
trunk_id, revision, config, cells, request_id
|
|
)
|
|
result = self._complete_publication(
|
|
trunk_id,
|
|
revision,
|
|
config,
|
|
cells,
|
|
failures,
|
|
action="rollback",
|
|
actor=actor,
|
|
request_id=request_id,
|
|
details={"target_revision": target_revision},
|
|
)
|
|
self._raise_delivery_failure("rollback", failures)
|
|
return result
|
|
|
|
def list_publications(self, trunk_id: str) -> list[dict[str, Any]]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
with self.store.read() as db:
|
|
if not db.execute(
|
|
"SELECT 1 FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone():
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
return [
|
|
dict(row)
|
|
for row in db.execute(
|
|
"SELECT trunk_id,revision,cell_id,status,error_code,updated_at FROM publications WHERE trunk_id=? ORDER BY revision,cell_id",
|
|
(trunk_id,),
|
|
).fetchall()
|
|
]
|
|
|
|
def list_audit(self, trunk_id: str) -> list[dict[str, Any]]:
|
|
trunk_id = _id(trunk_id, "trunk_id")
|
|
with self.store.read() as db:
|
|
if not db.execute(
|
|
"SELECT 1 FROM trunks WHERE trunk_id=?", (trunk_id,)
|
|
).fetchone():
|
|
raise SipManagementError(404, "TRUNK_NOT_FOUND", "trunk does not exist")
|
|
return [
|
|
dict(row)
|
|
for row in db.execute(
|
|
"SELECT audit_id,resource_type,resource_id,action,revision,actor,request_id,details_json,created_at FROM audit_log WHERE resource_type='trunk' AND resource_id=? ORDER BY rowid",
|
|
(trunk_id,),
|
|
).fetchall()
|
|
]
|
|
|
|
|
|
def _principal_map(raw: Mapping[str, Any], access_class: str) -> dict[str, Principal]:
|
|
result: dict[str, Principal] = {}
|
|
expected_audience = ADMIN_AUDIENCE if access_class == "admin" else READ_AUDIENCE
|
|
for token, value in raw.items():
|
|
if not isinstance(token, str) or not token or not isinstance(value, dict):
|
|
raise ConfigurationError(f"invalid {access_class} token map")
|
|
subject = value.get("subject")
|
|
issuer = value.get("issuer")
|
|
audience = value.get("audience")
|
|
scopes = value.get("scopes")
|
|
trunk_ids = value.get("trunk_ids", [])
|
|
if (
|
|
not isinstance(subject, str)
|
|
or not subject
|
|
or not isinstance(issuer, str)
|
|
or not issuer
|
|
):
|
|
raise ConfigurationError(f"invalid {access_class} token identity")
|
|
if not isinstance(audience, str) or audience != expected_audience:
|
|
raise ConfigurationError(f"{access_class} token has wrong audience")
|
|
if not isinstance(scopes, list) or not all(
|
|
isinstance(item, str) for item in scopes
|
|
):
|
|
raise ConfigurationError(f"invalid {access_class} token scopes")
|
|
if trunk_ids != "*" and (
|
|
not isinstance(trunk_ids, list)
|
|
or not all(isinstance(item, str) for item in trunk_ids)
|
|
):
|
|
raise ConfigurationError(f"invalid {access_class} token trunk_ids")
|
|
result[token] = Principal(
|
|
subject=subject,
|
|
issuer=issuer,
|
|
audience=audience,
|
|
scopes=frozenset(scopes),
|
|
trunk_ids=frozenset(trunk_ids if trunk_ids != "*" else ["*"]),
|
|
token_class=access_class,
|
|
)
|
|
return result
|
|
|
|
|
|
def load_token_map(raw: str | None, access_class: str) -> dict[str, Principal]:
|
|
try:
|
|
value = json.loads(raw or "{}")
|
|
except json.JSONDecodeError as exc:
|
|
raise ConfigurationError(f"invalid {access_class} token JSON") from exc
|
|
if not isinstance(value, dict):
|
|
raise ConfigurationError(f"{access_class} token JSON must be an object")
|
|
return _principal_map(value, access_class)
|
|
|
|
|
|
def validate_token_separation(
|
|
admin_tokens: Mapping[str, Principal],
|
|
read_tokens: Mapping[str, Principal],
|
|
ordinary_tokens: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
ordinary_tokens = ordinary_tokens or {}
|
|
if set(admin_tokens) & set(read_tokens):
|
|
raise ConfigurationError("admin and read-only SIP tokens must be different")
|
|
if (set(admin_tokens) | set(read_tokens)) & set(ordinary_tokens):
|
|
raise ConfigurationError(
|
|
"SIP management tokens must differ from ordinary scheduling tokens"
|
|
)
|
|
|
|
|
|
class SipManagementHandler(BaseHTTPRequestHandler):
|
|
server_version = "agent-call-sip-management/1"
|
|
|
|
def __init__(
|
|
self,
|
|
request: Any,
|
|
client_address: Any,
|
|
server: Any,
|
|
*,
|
|
service: SipManagementService,
|
|
admin_tokens: Mapping[str, Principal],
|
|
read_tokens: Mapping[str, Principal],
|
|
) -> None:
|
|
self.service = service
|
|
self.admin_tokens = admin_tokens
|
|
self.read_tokens = read_tokens
|
|
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 _error(self, exc: SipManagementError) -> None:
|
|
self._send(exc.status, exc.response())
|
|
|
|
def _path(self) -> list[str]:
|
|
path = urlsplit(self.path).path
|
|
parts = [unquote(part) for part in path.split("/") if part]
|
|
if any(part in {".", ".."} for part in parts):
|
|
raise SipManagementError(400, "INVALID_PATH", "path is invalid")
|
|
return parts
|
|
|
|
def _token(self, tokens: Mapping[str, Principal], scope: str) -> Principal:
|
|
header = self.headers.get("Authorization", "")
|
|
scheme, _, token = header.partition(" ")
|
|
principal = (
|
|
tokens.get(token)
|
|
if scheme == "Bearer" and token and not token.isspace()
|
|
else None
|
|
)
|
|
if principal is None:
|
|
raise SipManagementError(401, "UNAUTHORIZED", "valid bearer token required")
|
|
if scope not in principal.scopes and "*" not in principal.scopes:
|
|
raise SipManagementError(403, "FORBIDDEN", "token lacks required scope")
|
|
return principal
|
|
|
|
def _request_id(self) -> str:
|
|
value = self.headers.get("X-Request-ID") or self.headers.get("X-Request-Id")
|
|
if (
|
|
not value
|
|
or len(value) > 128
|
|
or value != value.strip()
|
|
or any(ord(char) < 32 for char in value)
|
|
):
|
|
raise SipManagementError(
|
|
400, "MISSING_REQUEST_ID", "X-Request-ID is required"
|
|
)
|
|
return value
|
|
|
|
def _expected_revision(self) -> int:
|
|
raw = self.headers.get("If-Match")
|
|
if raw is None:
|
|
raise SipManagementError(
|
|
428, "PRECONDITION_REQUIRED", "If-Match revision is required"
|
|
)
|
|
raw = raw.strip().strip('"')
|
|
try:
|
|
value = int(raw)
|
|
except ValueError as exc:
|
|
raise SipManagementError(
|
|
400, "INVALID_REVISION", "If-Match must be an integer revision"
|
|
) from exc
|
|
if value < 0:
|
|
raise SipManagementError(
|
|
400, "INVALID_REVISION", "revision must be non-negative"
|
|
)
|
|
return value
|
|
|
|
def _body(self) -> Any:
|
|
raw_length = self.headers.get("Content-Length")
|
|
try:
|
|
length = int(raw_length or "-1")
|
|
except ValueError as exc:
|
|
raise SipManagementError(
|
|
400, "INVALID_CONTENT_LENGTH", "Content-Length is invalid"
|
|
) from exc
|
|
if length < 0 or length > MAX_BODY_BYTES:
|
|
raise SipManagementError(
|
|
413, "PAYLOAD_TOO_LARGE", "request body is too large"
|
|
)
|
|
try:
|
|
return json.loads(self.rfile.read(length).decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise SipManagementError(
|
|
400, "INVALID_JSON", "request body must be JSON"
|
|
) from exc
|
|
|
|
def do_GET(self) -> None:
|
|
try:
|
|
parts = self._path()
|
|
if parts == ["healthz", "live"]:
|
|
self._send(200, {"status": "ok", "mode": self.service.mode})
|
|
return
|
|
if parts[:2] == ["admin", "v1"]:
|
|
if parts == ["admin", "v1", "cells"]:
|
|
self._token(self.admin_tokens, "sip.cell.read")
|
|
self._send(
|
|
200,
|
|
{"mode": self.service.mode, "cells": self.service.list_cells()},
|
|
)
|
|
return
|
|
self._token(self.admin_tokens, "sip.trunk.read")
|
|
if parts == ["admin", "v1", "trunks"]:
|
|
self._send(
|
|
200,
|
|
{
|
|
"mode": self.service.mode,
|
|
"trunks": self.service.list_trunks(),
|
|
},
|
|
)
|
|
elif len(parts) == 4 and parts[2] == "trunks":
|
|
self._send(200, self.service.get_trunk(parts[3]))
|
|
elif (
|
|
len(parts) == 5
|
|
and parts[2] == "trunks"
|
|
and parts[4] == "publications"
|
|
):
|
|
self._send(
|
|
200,
|
|
{
|
|
"mode": self.service.mode,
|
|
"publications": self.service.list_publications(parts[3]),
|
|
},
|
|
)
|
|
elif len(parts) == 5 and parts[2] == "trunks" and parts[4] == "audit":
|
|
self._send(
|
|
200,
|
|
{
|
|
"mode": self.service.mode,
|
|
"audit": self.service.list_audit(parts[3]),
|
|
},
|
|
)
|
|
else:
|
|
raise SipManagementError(
|
|
404, "NOT_FOUND", "endpoint does not exist"
|
|
)
|
|
return
|
|
if parts[:4] == ["readonly", "v1", "sip", "trunks"]:
|
|
principal = self._token(self.read_tokens, "sip.trunk.read")
|
|
if len(parts) == 4:
|
|
self._send(
|
|
200,
|
|
{
|
|
"mode": self.service.mode,
|
|
"trunks": self.service.list_readonly_trunks(
|
|
principal.trunk_ids
|
|
),
|
|
},
|
|
)
|
|
elif len(parts) == 5:
|
|
self._send(
|
|
200,
|
|
self.service.get_readonly_trunk(parts[4], principal.trunk_ids),
|
|
)
|
|
else:
|
|
raise SipManagementError(
|
|
404, "NOT_FOUND", "endpoint does not exist"
|
|
)
|
|
return
|
|
raise SipManagementError(404, "NOT_FOUND", "endpoint does not exist")
|
|
except SipManagementError as exc:
|
|
self._error(exc)
|
|
|
|
def do_PUT(self) -> None:
|
|
try:
|
|
parts = self._path()
|
|
if parts[:3] == ["admin", "v1", "cells"] and len(parts) == 4:
|
|
principal = self._token(self.admin_tokens, "sip.cell.write")
|
|
request_id = self._request_id()
|
|
try:
|
|
self.service.get_cell(parts[3])
|
|
status = 200
|
|
except SipManagementError as exc:
|
|
if exc.code != "CELL_NOT_FOUND":
|
|
raise
|
|
status = 201
|
|
self._send(
|
|
status,
|
|
self.service.upsert_cell(
|
|
parts[3],
|
|
self._body(),
|
|
expected_revision=self._expected_revision(),
|
|
actor=principal.subject,
|
|
request_id=request_id,
|
|
),
|
|
)
|
|
return
|
|
if parts[:3] == ["admin", "v1", "trunks"] and len(parts) == 4:
|
|
principal = self._token(self.admin_tokens, "sip.trunk.write")
|
|
request_id = self._request_id()
|
|
try:
|
|
self.service.get_trunk(parts[3])
|
|
status = 200
|
|
except SipManagementError as exc:
|
|
if exc.code != "TRUNK_NOT_FOUND":
|
|
raise
|
|
status = 201
|
|
self._send(
|
|
status,
|
|
self.service.upsert_trunk(
|
|
parts[3],
|
|
self._body(),
|
|
expected_revision=self._expected_revision(),
|
|
actor=principal.subject,
|
|
request_id=request_id,
|
|
),
|
|
)
|
|
return
|
|
raise SipManagementError(404, "NOT_FOUND", "endpoint does not exist")
|
|
except SipManagementError as exc:
|
|
self._error(exc)
|
|
|
|
def do_POST(self) -> None:
|
|
try:
|
|
parts = self._path()
|
|
if len(parts) == 5 and parts[:3] == ["admin", "v1", "trunks"]:
|
|
action = parts[4]
|
|
principal = self._token(
|
|
self.admin_tokens,
|
|
{
|
|
"publish": "sip.trunk.publish",
|
|
"disable": "sip.trunk.write",
|
|
"rollback": "sip.trunk.rollback",
|
|
}.get(action, "sip.trunk.write"),
|
|
)
|
|
request_id = self._request_id()
|
|
expected = self._expected_revision()
|
|
if action == "publish":
|
|
result = self.service.publish_trunk(
|
|
parts[3],
|
|
expected_revision=expected,
|
|
actor=principal.subject,
|
|
request_id=request_id,
|
|
)
|
|
elif action == "disable":
|
|
result = self.service.disable_trunk(
|
|
parts[3],
|
|
expected_revision=expected,
|
|
actor=principal.subject,
|
|
request_id=request_id,
|
|
)
|
|
elif action == "rollback":
|
|
body = self._body()
|
|
if not isinstance(body, dict) or set(body) != {"target_revision"}:
|
|
raise SipManagementError(
|
|
400, "INVALID_ROLLBACK", "rollback requires target_revision"
|
|
)
|
|
result = self.service.rollback_trunk(
|
|
parts[3],
|
|
body["target_revision"],
|
|
expected_revision=expected,
|
|
actor=principal.subject,
|
|
request_id=request_id,
|
|
)
|
|
else:
|
|
raise SipManagementError(
|
|
404, "NOT_FOUND", "endpoint does not exist"
|
|
)
|
|
self._send(200, result)
|
|
return
|
|
if parts[:4] == ["readonly", "v1", "sip", "trunks"]:
|
|
raise SipManagementError(
|
|
405, "READ_ONLY", "SaaS Trunk API is read-only"
|
|
)
|
|
raise SipManagementError(404, "NOT_FOUND", "endpoint does not exist")
|
|
except SipManagementError as exc:
|
|
self._error(exc)
|
|
|
|
|
|
class _Server(ThreadingHTTPServer):
|
|
daemon_threads = True
|
|
allow_reuse_address = True
|
|
|
|
|
|
def make_server(
|
|
service: SipManagementService,
|
|
host: str = "127.0.0.1",
|
|
port: int = 8090,
|
|
*,
|
|
admin_tokens: Mapping[str, Principal],
|
|
read_tokens: Mapping[str, Principal],
|
|
) -> ThreadingHTTPServer:
|
|
if not admin_tokens or not read_tokens:
|
|
raise ConfigurationError("admin and read-only SIP token maps are required")
|
|
validate_token_separation(admin_tokens, read_tokens)
|
|
handler = partial(
|
|
SipManagementHandler,
|
|
service=service,
|
|
admin_tokens=admin_tokens,
|
|
read_tokens=read_tokens,
|
|
)
|
|
return _Server((host, port), handler)
|
|
|
|
|
|
def build_from_env() -> tuple[SipManagementService, ThreadingHTTPServer]:
|
|
admin_tokens = load_token_map(os.environ.get("SIP_ADMIN_TOKENS"), "admin")
|
|
read_tokens = load_token_map(os.environ.get("SIP_READ_TOKENS"), "read")
|
|
ordinary_tokens = _json_object(
|
|
os.environ.get("HTTP_TOKENS"), "HTTP_TOKENS must be an object"
|
|
)
|
|
if not admin_tokens or not read_tokens:
|
|
raise ConfigurationError("SIP_ADMIN_TOKENS and SIP_READ_TOKENS are required")
|
|
validate_token_separation(admin_tokens, read_tokens, ordinary_tokens)
|
|
mode = os.environ.get("SIP_MANAGEMENT_MODE", "mock")
|
|
publisher: MtlsCellPublisher | None = None
|
|
if mode == "real":
|
|
ca_file = os.environ.get("SIP_CELL_TLS_CA_FILE")
|
|
cert_file = os.environ.get("SIP_CELL_TLS_CERT_FILE")
|
|
key_file = os.environ.get("SIP_CELL_TLS_KEY_FILE")
|
|
if not ca_file or not cert_file or not key_file:
|
|
raise ConfigurationError(
|
|
"real mode requires SIP_CELL_TLS_CA_FILE, SIP_CELL_TLS_CERT_FILE and SIP_CELL_TLS_KEY_FILE"
|
|
)
|
|
try:
|
|
timeout = float(os.environ.get("SIP_CELL_PUBLISH_TIMEOUT_SECONDS", "15"))
|
|
except ValueError as exc:
|
|
raise ConfigurationError(
|
|
"SIP_CELL_PUBLISH_TIMEOUT_SECONDS must be a number"
|
|
) from exc
|
|
publisher = MtlsCellPublisher(
|
|
ca_file=ca_file,
|
|
cert_file=cert_file,
|
|
key_file=key_file,
|
|
timeout=timeout,
|
|
)
|
|
service = SipManagementService(
|
|
os.environ.get("SIP_MANAGEMENT_DB", "data/sip-management.sqlite3"),
|
|
mode=mode,
|
|
publisher=publisher,
|
|
)
|
|
try:
|
|
port = int(os.environ.get("SIP_MANAGEMENT_PORT", "8090"))
|
|
except ValueError as exc:
|
|
service.close()
|
|
raise ConfigurationError("SIP_MANAGEMENT_PORT must be an integer") from exc
|
|
server = make_server(
|
|
service,
|
|
os.environ.get("SIP_MANAGEMENT_HOST", "127.0.0.1"),
|
|
port,
|
|
admin_tokens=admin_tokens,
|
|
read_tokens=read_tokens,
|
|
)
|
|
return service, server
|
|
|
|
|
|
__all__ = [
|
|
"ADMIN_AUDIENCE",
|
|
"CODEC_TO_ASTERISK",
|
|
"READ_AUDIENCE",
|
|
"CellPublishError",
|
|
"ConfigurationError",
|
|
"MtlsCellPublisher",
|
|
"Principal",
|
|
"SipManagementError",
|
|
"SipManagementService",
|
|
"SipManagementStore",
|
|
"asterisk_codec_allow",
|
|
"build_from_env",
|
|
"load_token_map",
|
|
"make_server",
|
|
"validate_token_separation",
|
|
]
|