143 lines
5.0 KiB
Python
143 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import ssl
|
|
from http.client import HTTPSConnection
|
|
from typing import Any
|
|
from urllib.parse import quote, urlsplit
|
|
|
|
MAX_RESPONSE_BYTES = 64 * 1024
|
|
|
|
|
|
class SipReadonlyError(Exception):
|
|
def __init__(self, code: str, detail: str, *, retryable: bool = False) -> None:
|
|
super().__init__(detail)
|
|
self.code = code
|
|
self.detail = detail
|
|
self.retryable = retryable
|
|
|
|
|
|
class SipReadonlyClient:
|
|
def __init__(
|
|
self,
|
|
base_url: str,
|
|
token: str,
|
|
*,
|
|
ca_file: str | None = None,
|
|
timeout: float = 5.0,
|
|
) -> None:
|
|
if not token or len(token) > 4096 or any(ord(char) < 32 for char in token):
|
|
raise ValueError("SIP read-only token is invalid")
|
|
parsed = urlsplit(base_url.rstrip("/"))
|
|
try:
|
|
port = parsed.port
|
|
except ValueError as exc:
|
|
raise ValueError("SIP read-only URL has an invalid port") 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 ValueError(
|
|
"SIP read-only URL must be HTTPS without credentials or query data"
|
|
)
|
|
if not timeout > 0:
|
|
raise ValueError("SIP_READONLY_TIMEOUT_SECONDS must be positive")
|
|
try:
|
|
context = ssl.create_default_context(cafile=ca_file)
|
|
except (OSError, ssl.SSLError) as exc:
|
|
raise ValueError("SIP read-only CA file could not be loaded") from exc
|
|
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
context.check_hostname = True
|
|
self._host = parsed.hostname
|
|
self._port = port or 443
|
|
self._base_path = parsed.path.rstrip("/")
|
|
self._token = token
|
|
self._context = context
|
|
self._timeout = timeout
|
|
|
|
def get_trunk(self, trunk_id: str) -> dict[str, Any]:
|
|
path = f"{self._base_path}/readonly/v1/sip/trunks/{quote(trunk_id, safe='')}"
|
|
connection: HTTPSConnection | None = None
|
|
try:
|
|
connection = HTTPSConnection(
|
|
self._host,
|
|
self._port,
|
|
context=self._context,
|
|
timeout=self._timeout,
|
|
)
|
|
connection.request(
|
|
"GET",
|
|
path,
|
|
headers={
|
|
"Accept": "application/json",
|
|
"Authorization": f"Bearer {self._token}",
|
|
},
|
|
)
|
|
response = connection.getresponse()
|
|
status = response.status
|
|
raw = response.read(MAX_RESPONSE_BYTES + 1)
|
|
except (OSError, ValueError) as exc:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_UNAVAILABLE",
|
|
"SIP read-only directory is unavailable",
|
|
retryable=True,
|
|
) from exc
|
|
finally:
|
|
if connection is not None:
|
|
connection.close()
|
|
if status == 404:
|
|
raise SipReadonlyError(
|
|
"SIP_TRUNK_NOT_PUBLISHED", "SIP Trunk is not published"
|
|
)
|
|
if status < 200 or status >= 300:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_UNAVAILABLE",
|
|
"SIP read-only directory rejected the request",
|
|
retryable=status >= 500 or status in {408, 429},
|
|
)
|
|
if len(raw) > MAX_RESPONSE_BYTES:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_INVALID", "SIP read-only response is too large"
|
|
)
|
|
try:
|
|
payload = json.loads(raw.decode("utf-8"))
|
|
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_INVALID", "SIP read-only response is invalid JSON"
|
|
) from exc
|
|
if not isinstance(payload, dict) or payload.get("trunk_id") != trunk_id:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_INVALID", "SIP read-only response has the wrong Trunk"
|
|
)
|
|
if payload.get("status") != "published" or not isinstance(
|
|
payload.get("config"), dict
|
|
):
|
|
raise SipReadonlyError(
|
|
"SIP_TRUNK_NOT_PUBLISHED", "SIP Trunk is not published"
|
|
)
|
|
config = payload["config"]
|
|
enabled = config.get("enabled")
|
|
if not isinstance(enabled, bool) or not enabled:
|
|
raise SipReadonlyError("SIP_TRUNK_NOT_PUBLISHED", "SIP Trunk is disabled")
|
|
for field in (
|
|
"codec_profile",
|
|
"caller_ids",
|
|
"dial_prefix",
|
|
"egress_pool_id",
|
|
"max_concurrency",
|
|
"max_cps",
|
|
):
|
|
if field not in config:
|
|
raise SipReadonlyError(
|
|
"SIP_DIRECTORY_INVALID", "SIP Trunk config is incomplete"
|
|
)
|
|
return payload
|
|
|
|
|
|
__all__ = ["SipReadonlyClient", "SipReadonlyError"]
|