"""
Cloudonix implementation of the TelephonyProvider interface.
"""
import asyncio
import json
import re
from typing import TYPE_CHECKING, Any, Dict, List, Optional
from urllib.parse import quote
import aiohttp
from fastapi import HTTPException
from loguru import logger
from api.db import db_client
from api.enums import TelephonyCallStatus, WorkflowRunMode
from api.services.telephony import ws_auth
from api.services.telephony.base import (
CallInitiationResult,
NormalizedInboundData,
ProviderPhoneNumberLookupError,
ProviderSyncResult,
SIPConnectivityDetails,
SIPRegionDetails,
SIPTransportDetails,
TelephonyProvider,
)
from api.services.workflow.initial_context import merge_external_initial_context
from api.utils.common import get_backend_endpoints
from api.utils.telephony_address import normalize_telephony_address
from .config import normalize_cloudonix_domain
from .regions import CLOUDONIX_REGIONS, CloudonixRegion
if TYPE_CHECKING:
from fastapi import WebSocket
CLOUDONIX_API_BASE_URL = "https://api.cloudonix.io"
# Cloudonix sends the connected/start handshake immediately after the media
# stream opens. The agent-stream route holds an org concurrency slot while we
# wait, so an idle socket must not be able to hold it indefinitely.
AGENT_STREAM_HANDSHAKE_TIMEOUT_S = 10
def _key_tail(key: str | None) -> str:
"""Last 8 characters of an API key, for logs that compare two keys."""
if not key:
return "MISSING"
return key[-8:] if len(key) > 8 else "SHORT_KEY"
def _inbound_transports(
hostname: str, region: CloudonixRegion
) -> list[SIPTransportDetails]:
"""The three SIP transports Cloudonix accepts on a region's edge."""
return [
SIPTransportDetails(
transport="UDP",
hostname=hostname,
port=region.sip_port,
uri=f"{hostname}:{region.sip_port}",
),
SIPTransportDetails(
transport="TCP",
hostname=hostname,
port=region.sip_port,
uri=f"{hostname}:{region.sip_port};transport=tcp;",
),
SIPTransportDetails(
transport="TLS",
hostname=hostname,
port=region.tls_port,
uri=f"{hostname}:{region.tls_port};transport=tls;",
),
]
class CloudonixProvider(TelephonyProvider):
"""
Cloudonix implementation of TelephonyProvider.
Uses Bearer token authentication and is TwiML-compatible for WebSocket audio.
"""
PROVIDER_NAME = WorkflowRunMode.CLOUDONIX.value
WEBHOOK_ENDPOINT = "twiml" # Cloudonix is TwiML-compatible
def __init__(self, config: Dict[str, Any]):
"""
Initialize CloudonixProvider with configuration.
Args:
config: Dictionary containing:
- bearer_token: Cloudonix API Bearer Token
- domain_id: Configured Cloudonix domain name
- domain_uuid: UUID returned by Cloudonix's domainGet API
- application_name: Cloudonix Voice Application name whose
url is updated by ``configure_inbound``
- trunks: Dograh-managed outbound voice trunks on this domain
- trunk_id_by_number: which trunk each active number dials out
on, so a call is pinned to the carrier that authorised its
caller ID
- from_numbers: List of phone numbers to use (optional, fetched from API if not provided)
"""
self.bearer_token = config.get("bearer_token")
self.domain_id = self._normalize_domain(config.get("domain_id"))
self.domain_uuid = config.get("domain_uuid")
self.application_name = config.get("application_name")
self.trunks = config.get("trunks") or []
self.trunk_id_by_number = config.get("trunk_id_by_number") or {}
self.from_numbers = config.get("from_numbers", [])
self.default_from_number = config.get("default_from_number")
# Handle both single number (string) and multiple numbers (list)
if isinstance(self.from_numbers, str):
self.from_numbers = [self.from_numbers]
self.base_url = CLOUDONIX_API_BASE_URL
def _pin_trunk(self, data: Dict[str, Any], from_number: Optional[str]) -> None:
"""Route the call over the trunk that authorised its caller ID.
Cloudonix otherwise tries every active trunk in creation order, which
can present a number one carrier owns to a different carrier. When no
trunk resolves we leave the selector off rather than guess — and we
drop any the caller passed through kwargs, since the stored assignment
is what decides this.
"""
trunk = self.select_trunk(from_number)
name = trunk.get("name") if trunk else None
if isinstance(name, str) and name.strip():
data["trunk"] = name.strip()
else:
data.pop("trunk", None)
@staticmethod
def _normalize_domain(domain: Optional[str]) -> Optional[str]:
"""Normalize short names without rewriting custom Cloudonix FQDNs."""
return normalize_cloudonix_domain(domain)
def _get_auth_headers(self) -> Dict[str, str]:
"""Generate authorization headers for Cloudonix API."""
return {
"Authorization": f"Bearer {self.bearer_token}",
"Content-Type": "application/json",
}
def get_sip_connectivity_details(self) -> SIPConnectivityDetails | None:
"""Return Cloudonix's regional SIP connection details."""
if not isinstance(self.domain_uuid, str):
return None
domain_uuid = self.domain_uuid.strip()
if not domain_uuid:
return None
return SIPConnectivityDetails(
provider_display_name="Cloudonix",
regions=[
SIPRegionDetails(
region=region.name,
inbound_transports=_inbound_transports(
region.hostname(domain_uuid), region
),
outbound_origin_ip=region.edge_ip,
)
for region in CLOUDONIX_REGIONS
],
)
async def initiate_call(
self,
to_number: str,
webhook_url: str,
workflow_run_id: Optional[int] = None,
from_number: Optional[str] = None,
**kwargs: Any,
) -> CallInitiationResult:
"""
Initiate an outbound call via Cloudonix.
Note: webhook_url parameter is ignored for Cloudonix. Unlike Twilio/Vonage,
Cloudonix embeds CXML directly in the API call rather than using webhook callbacks.
"""
if not self.validate_config():
raise ValueError("Cloudonix provider not properly configured")
endpoint = f"{self.base_url}/calls/{self.domain_id}/application"
# A caller-id is REQUIRED by Cloudonix
from_number = self.select_from_number(from_number)
if from_number is None:
raise ValueError(
"No phone numbers configured for Cloudonix provider. "
"At least one phone number is required as 'caller-id' for outbound calls. "
"Please configure phone numbers in the telephony settings."
)
logger.info(
f"Selected phone number {from_number} for outbound call to {to_number}"
)
workflow_id, organization_id = (
kwargs["workflow_id"],
kwargs["organization_id"],
)
# Prepare call data using Cloudonix callObject schema
# Note: 'caller-id' is REQUIRED by Cloudonix API
backend_endpoint, wss_backend_endpoint = await get_backend_endpoints()
ws_url = ws_auth.build_media_ws_url(
wss_backend_endpoint, workflow_id, organization_id, workflow_run_id
)
data: Dict[str, Any] = {
"destination": to_number,
"cxml": f"""
""",
"caller-id": from_number, # Required field
}
# TODO: Cloudonix status callbacks are spammy, so commenting it out. Can send it to
# some persistent logging system instead of transcational database.
# Add status callback if workflow_run_id provided
# if workflow_run_id:
# callback_url = f"{backend_endpoint}/api/v1/telephony/cloudonix/status-callback/{workflow_run_id}"
# data["callback"] = callback_url
# Merge any additional kwargs
data.update(kwargs)
self._pin_trunk(data, from_number)
# Make the API request
headers = self._get_auth_headers()
# Log request details (mask sensitive token)
masked_headers = {
k: v if k != "Authorization" else f"Bearer {self.bearer_token[:8]}..."
for k, v in headers.items()
}
logger.info(
f"[Cloudonix] Initiating outbound call:\n"
f" Endpoint: {endpoint}\n"
f" To: {to_number}\n"
f" From: {from_number}\n"
f" Workflow Run ID: {workflow_run_id}"
)
# Redacted: the embedded CXML's stream URL carries a bearer capability token.
logger.debug(
f"[Cloudonix] Request details:\n"
f" Headers: {masked_headers}\n"
f" Payload: {ws_auth.redact_token(json.dumps(data, indent=2))}"
)
async with aiohttp.ClientSession() as session:
async with session.post(endpoint, json=data, headers=headers) as response:
response_text = await response.text()
response_status = response.status
if response_status != 200:
raise HTTPException(
status_code=response_status,
detail=f"Failed to initiate call via Cloudonix (HTTP {response_status}): {response_text}",
)
response_data = await response.json()
# Extract session token (call ID) and other metadata
session_token = response_data.get("token")
domain_id = response_data.get("domainId")
subscriber_id = response_data.get("subscriberId")
if not session_token:
raise Exception("No session token returned from Cloudonix")
logger.info(
f"[Cloudonix] Call initiated successfully:\n"
f" Session Token: {session_token}\n"
f" Domain ID: {domain_id}\n"
f" Subscriber ID: {subscriber_id}\n"
f" To: {to_number}\n"
f" From: {from_number}\n"
f" Workflow Run ID: {workflow_run_id}"
)
return CallInitiationResult(
call_id=session_token,
status="initiated",
caller_number=from_number,
provider_metadata={
"call_id": session_token,
"domain_id": domain_id,
"subscriber_id": subscriber_id,
},
raw_response=response_data,
)
async def get_call_status(self, call_id: str) -> Dict[str, Any]:
"""
Get the current status of a Cloudonix call (session).
Args:
call_id: The session token returned from call initiation
"""
if not self.validate_config():
raise ValueError("Cloudonix provider not properly configured")
endpoint = (
f"{self.base_url}/customers/self/domains/"
f"{self.domain_id}/sessions/{call_id}"
)
headers = self._get_auth_headers()
async with aiohttp.ClientSession() as session:
async with session.get(endpoint, headers=headers) as response:
if response.status != 200:
error_data = await response.text()
logger.error(f"Failed to get call status: {error_data}")
raise Exception(f"Failed to get call status: {error_data}")
return await response.json()
async def get_available_phone_numbers(self) -> List[str]:
"""
Get list of available Cloudonix phone numbers (DNIDs).
"""
# If phone numbers are already configured, return them
if self.from_numbers:
return self.from_numbers
# Otherwise, fetch from API
if not self.validate_config():
raise ValueError("Cloudonix provider not properly configured")
endpoint = f"{self.base_url}/customers/self/domains/{self.domain_id}/dnids"
headers = self._get_auth_headers()
try:
async with aiohttp.ClientSession() as session:
async with session.get(endpoint, headers=headers) as response:
if response.status != 200:
logger.warning(
f"Failed to fetch DNIDs from Cloudonix: {response.status}"
)
return []
dnids = await response.json()
# Extract phone numbers from DNID objects
# Use "source" field which contains the original phone number
phone_numbers = [
dnid.get("source") or dnid.get("dnid")
for dnid in dnids
if dnid.get("source") or dnid.get("dnid")
]
# Cache the fetched numbers
self.from_numbers = phone_numbers
return phone_numbers
except Exception as e:
logger.error(f"Exception fetching Cloudonix DNIDs: {e}")
return []
def validate_config(self) -> bool:
"""
Validate Cloudonix configuration.
"""
return bool(self.bearer_token and self.domain_id)
async def verify_webhook_signature(
self, url: str, params: Dict[str, Any], signature: str
) -> bool:
"""
Dummy implementation - Cloudonix doesn't use webhook signature verification.
Cloudonix embeds CXML directly in the API call during initiate_call(),
so webhook endpoints are never called and signature verification is not needed.
This method only exists to satisfy the abstract base class requirement.
Always returns True since no actual webhook verification is performed.
"""
logger.warning(
"verify_webhook_signature called for Cloudonix - this should not happen. "
"Cloudonix embeds CXML directly in API calls and doesn't use webhook callbacks."
)
return True
async def get_call_cost(self, call_id: str) -> Dict[str, Any]:
"""
Get cost information for a completed Cloudonix call.
Note: Cloudonix does not currently support call cost retrieval via API.
This method returns zero cost.
Args:
call_id: The Cloudonix session token
Returns:
Dict containing cost information (all zeros for now)
"""
logger.info(
f"Cloudonix does not support call cost retrieval - returning zero cost for call {call_id}"
)
return {
"cost_usd": 0.0,
"duration": 0,
"status": "unknown",
"error": "Cloudonix does not support cost retrieval",
}
def parse_status_callback(self, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Parse Cloudonix status callback data into generic format.
Note: The exact format of Cloudonix status callbacks needs to be confirmed.
This implementation assumes a similar structure to Twilio.
"""
# Map Cloudonix status values to common format
# These mappings may need adjustment based on actual Cloudonix callback format
status_map = {
"initiated": TelephonyCallStatus.INITIATED,
"ringing": TelephonyCallStatus.RINGING,
"answered": TelephonyCallStatus.ANSWERED,
"completed": TelephonyCallStatus.COMPLETED,
"failed": TelephonyCallStatus.FAILED,
"busy": TelephonyCallStatus.BUSY,
"no-answer": TelephonyCallStatus.NO_ANSWER,
"canceled": TelephonyCallStatus.CANCELED,
"error": TelephonyCallStatus.ERROR,
}
call_status = data.get("status", "")
mapped_status = status_map.get(call_status.lower(), call_status)
return {
"call_id": data.get("token")
or data.get("session_id")
or data.get("CallSid", ""),
"status": mapped_status,
"from_number": data.get("caller_id") or data.get("From"),
"to_number": data.get("destination") or data.get("To"),
"direction": data.get("direction"),
"duration": data.get("duration") or data.get("CallDuration"),
"extra": data, # Include all original data
}
@staticmethod
def parse_cdr_status_callback(data: Dict[str, Any]) -> Dict[str, Any]:
"""Parse Cloudonix CDR data into generic status callback format."""
disposition_map = {
"ANSWER": TelephonyCallStatus.COMPLETED,
"BUSY": TelephonyCallStatus.BUSY,
"CANCEL": TelephonyCallStatus.CANCELED,
"FAILED": TelephonyCallStatus.FAILED,
"CONGESTION": TelephonyCallStatus.FAILED,
"NOANSWER": TelephonyCallStatus.NO_ANSWER,
}
disposition = data.get("disposition") or ""
session = data.get("session")
billsec = data.get("billsec")
return {
"call_id": session.get("token") if isinstance(session, dict) else "",
"status": disposition_map.get(disposition.upper(), disposition.lower()),
"from_number": data.get("from"),
"to_number": data.get("to"),
"duration": str(
billsec if billsec is not None else (data.get("duration") or 0)
),
"extra": data,
}
async def get_webhook_response(
self, workflow_id: int, organization_id: int, workflow_run_id: int
) -> str:
"""
Dummy implementation - Cloudonix doesn't use webhook responses.
Cloudonix embeds CXML directly in the API call during initiate_call(),
so this webhook endpoint is never actually called. This method only
exists to satisfy the abstract base class requirement.
"""
logger.warning(
"get_webhook_response called for Cloudonix - this should not happen. "
"Cloudonix embeds CXML directly in API calls."
)
return """
Error: This endpoint should not be called for Cloudonix
"""
async def handle_websocket(
self,
websocket: "WebSocket",
workflow_id: int,
organization_id: int,
workflow_run_id: int,
) -> None:
"""
Handle Cloudonix-specific WebSocket connection.
Cloudonix WebSocket is compatible with Twilio, so we use the same handler.
Cloudonix sends:
1. "connected" event first
2. "start" event with streamSid and callSid
3. Then audio messages
"""
from api.services.pipecat.run_pipeline import run_pipeline_telephony
try:
# Wait for "connected" event
first_msg = await websocket.receive_text()
msg = json.loads(first_msg)
logger.debug(f"Received first message: {msg}")
if msg.get("event") != "connected":
logger.error(f"Expected 'connected' event, got: {msg.get('event')}")
await websocket.close(code=4400, reason="Expected connected event")
return
# Wait for "start" event with stream details
start_msg = await websocket.receive_text()
logger.debug(f"Received start message: {start_msg}")
start_msg = json.loads(start_msg)
if start_msg.get("event") != "start":
logger.error("Expected 'start' event second")
await websocket.close(code=4400, reason="Expected start event")
return
start = start_msg.get("start")
if not isinstance(start, dict):
logger.error("Cloudonix start message missing start object")
await websocket.close(code=4400, reason="Missing start metadata")
return
try:
stream_sid = start["streamSid"]
call_sid = start["callSid"]
except KeyError:
logger.error("Missing streamSid or callSid in start message")
await websocket.close(code=4400, reason="Missing stream identifiers")
return
logger.debug(
f"Cloudonix WebSocket connected for workflow_run {workflow_run_id} "
f"stream_sid: {stream_sid} call_sid: {call_sid}"
)
workflow_run = await db_client.get_workflow_run_by_id(workflow_run_id)
call_id = (
workflow_run.gathered_context.get("call_id")
if workflow_run and workflow_run.gathered_context
else None
)
if not call_id:
logger.error(
f"call_id not found in gathered_context for workflow run {workflow_run_id}"
)
await websocket.close(code=4400, reason="Missing call_id")
return
await run_pipeline_telephony(
websocket,
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
organization_id=organization_id,
call_id=call_id,
transport_kwargs={"call_id": call_id, "stream_sid": stream_sid},
)
except Exception as e:
logger.error(f"Error in Cloudonix WebSocket handler: {e}")
raise
async def handle_external_websocket(
self,
websocket: "WebSocket",
*,
organization_id: int,
workflow_id: int,
workflow_run_id: int,
params: Dict[str, str],
) -> None:
"""Agent-stream entry point.
The Cloudonix domain is read from the ``start.accountSid`` field
in the start message. The bearer token comes from the stored
Cloudonix telephony configuration matched by ``domain_id`` within
the workflow's organization — never from the URL or stream payload.
The websocket handshake (connected / start) is identical to the
standard inbound flow.
Before starting the pipeline we (a) require an existing Cloudonix
telephony configuration for the stream's ``domain_id`` and (b)
validate the call session with Cloudonix using the bearer token
from that configuration. Either failure closes the socket with
4400.
"""
from api.services.pipecat.run_pipeline import run_pipeline_telephony
try:
try:
first_msg = await asyncio.wait_for(
websocket.receive_text(),
timeout=AGENT_STREAM_HANDSHAKE_TIMEOUT_S,
)
msg = json.loads(first_msg)
if msg.get("event") != "connected":
logger.error(f"Expected 'connected' event, got: {msg.get('event')}")
await websocket.close(code=4400, reason="Expected connected event")
return
start_msg = json.loads(
await asyncio.wait_for(
websocket.receive_text(),
timeout=AGENT_STREAM_HANDSHAKE_TIMEOUT_S,
)
)
except asyncio.TimeoutError:
logger.warning(
f"Cloudonix agent-stream handshake timed out for workflow_run "
f"{workflow_run_id}"
)
await websocket.close(code=4408, reason="Handshake timeout")
return
if start_msg.get("event") != "start":
logger.error("Expected 'start' event second")
await websocket.close(code=4400, reason="Expected start event")
return
start = start_msg.get("start")
if not isinstance(start, dict):
logger.error(
"Cloudonix agent-stream start message missing start object"
)
await websocket.close(code=4400, reason="Missing start metadata")
return
try:
stream_sid = start["streamSid"]
call_sid = start["callSid"]
call_session = start["session"]
domain_id = self._normalize_domain(start["accountSid"])
except KeyError:
logger.error(
"Missing streamSid, callSid, session, or accountSid in start message"
)
await websocket.close(code=4400, reason="Missing stream identifiers")
return
if not domain_id:
logger.error("Cloudonix agent-stream start message missing accountSid")
await websocket.close(
code=4400, reason="Missing Cloudonix domain in start message"
)
return
config = await self._find_config_by_domain(organization_id, domain_id)
if not config:
logger.error(
f"Cloudonix agent-stream: no telephony configuration found "
f"for domain_id={domain_id}"
)
await websocket.close(
code=4400, reason=f"Unknown Cloudonix domain: {domain_id}"
)
return
bearer_token = (config.credentials or {}).get("bearer_token")
if not bearer_token:
logger.error(
f"Cloudonix agent-stream: telephony configuration {config.id} "
f"is missing bearer_token in credentials"
)
await websocket.close(
code=4400, reason="Cloudonix configuration missing bearer_token"
)
return
if not await self._validate_session(domain_id, call_session, bearer_token):
await websocket.close(
code=4400, reason="Cloudonix session validation failed"
)
return
start_context = start.get("context")
custom_parameters = start.get("customParameters")
builtin_context = {
"caller_number": start.get("from"),
"called_number": start.get("to"),
"direction": (
"outbound" if start_context == "outbound-api" else "inbound"
),
"cloudonix_context": start_context,
}
await db_client.update_workflow_run(
run_id=workflow_run_id,
initial_context={
# Flatten customParameters, but never let them overwrite a
# built-in key even when the built-in's value is None.
key: value
for key, value in {
**{
k: v
for k, v in merge_external_initial_context(
{},
custom_parameters
if isinstance(custom_parameters, dict)
else {},
).items()
if k not in builtin_context
},
**builtin_context,
}.items()
if value is not None
},
gathered_context={
"call_id": call_session,
"cloudonix_call_sid": call_sid,
"cloudonix_stream_sid": stream_sid,
},
logs={
"inbound_webhook": {
"domain": domain_id,
"session": call_session,
"callSid": call_sid,
"streamSid": stream_sid,
"from": start.get("from"),
"to": start.get("to"),
"context": start_context,
"tracks": start.get("tracks"),
"mediaFormat": start.get("mediaFormat"),
},
},
)
logger.info(
f"Cloudonix agent-stream connected for workflow_run "
f"{workflow_run_id} stream_sid={stream_sid} call_sid={call_sid} "
f"session={call_session} "
f"telephony_configuration_id={config.id}"
)
await run_pipeline_telephony(
websocket,
provider_name=self.PROVIDER_NAME,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
organization_id=organization_id,
call_id=call_session,
transport_kwargs={
"call_id": call_session,
"stream_sid": stream_sid,
"bearer_token": bearer_token,
"domain_id": domain_id,
},
)
except Exception as e:
logger.error(f"Error in Cloudonix agent-stream handler: {e}")
raise
async def _validate_session(
self, domain_id: str, call_session: str, bearer_token: str
) -> bool:
"""Confirm the session is live with Cloudonix.
Hits ``GET /customers/self/domains/{domain_id}/sessions/{call_session}``
with the supplied bearer token. A 200 response means both the
token is valid and the session exists.
"""
endpoint = f"{self.base_url}/customers/self/domains/{domain_id}/sessions/{call_session}"
headers = {
"Authorization": f"Bearer {bearer_token}",
"Content-Type": "application/json",
}
try:
async with aiohttp.ClientSession() as http:
async with http.get(endpoint, headers=headers) as response:
if response.status == 200:
return True
body = await response.text()
logger.warning(
f"Cloudonix session validation failed: "
f"HTTP {response.status} domain_id={domain_id} "
f"call_id={call_session} body={body}"
)
return False
except Exception as e:
logger.error(
f"Cloudonix session validation error for domain_id={domain_id} "
f"call_id={call_session}: {e}"
)
return False
async def _find_config_by_domain(self, organization_id: int, domain_id: str):
"""Find a Cloudonix config by its normalized ``domain_id`` within
``organization_id`` — scoped lookup so credentials from a different
org can never be used."""
normalized = self._normalize_domain(domain_id)
if not normalized:
return None
candidates = await db_client.list_telephony_configurations_by_provider(
organization_id, self.PROVIDER_NAME
)
for cand in candidates:
if (cand.credentials or {}).get("domain_id") == normalized:
return cand
return None
# ======== INBOUND CALL METHODS ========
@classmethod
def can_handle_webhook(
cls, webhook_data: Dict[str, Any], headers: Dict[str, str]
) -> bool:
"""
Determine if this provider can handle the incoming webhook.
"""
# 1: Check User-Agent header
user_agent = headers.get("user-agent", "").lower()
if "cloudonix" in user_agent:
return True
# 2: Check for Cloudonix-specific headers
cloudonix_headers = [
"x-cx-apikey",
"x-cx-domain",
"x-cx-session",
"x-cx-source",
]
if any(header in headers for header in cloudonix_headers):
return True
# 3: Check data structure for Cloudonix-specific fields
if (
"SessionData" in webhook_data
and "Domain" in webhook_data
and webhook_data.get("Domain", "").endswith(".cloudonix.net")
):
return True
# Check if AccountSid is a Cloudonix domain
account_sid = webhook_data.get("AccountSid", "")
if account_sid.endswith(".cloudonix.net"):
return True
return False
@staticmethod
def parse_inbound_webhook(webhook_data: Dict[str, Any]) -> NormalizedInboundData:
"""
Parse Cloudonix-specific inbound webhook data into normalized format.
Cloudonix webhook structure includes:
- CallSid: Call id
- From: Caller number
- To: Called number
- AccountSid: Domain (e.g., "abc.cloudonix.net")
- SessionData: Contains additional call info including underlying provider details
"""
session_data = webhook_data.get("SessionData", {})
token = session_data.get("token", "") if isinstance(session_data, dict) else ""
call_id = webhook_data.get("Session") or webhook_data.get("CallSid") or token
raw_domain = webhook_data.get("Domain")
raw_account_sid = webhook_data.get("AccountSid", "")
account_id = CloudonixProvider._normalize_domain(raw_domain or raw_account_sid)
# The domain is what picks the telephony configuration — and therefore
# the organization — for this call, so record both fields it can come
# from before they are collapsed into one normalized value.
logger.info(
f"Cloudonix inbound webhook domain fields — Domain={raw_domain!r} "
f"AccountSid={raw_account_sid!r} normalized={account_id!r} "
f"To={webhook_data.get('To')!r} From={webhook_data.get('From')!r}"
)
# Extract underlying provider information from SessionData if available
session_data = webhook_data.get("SessionData", {})
underlying_provider = None
if isinstance(session_data, dict):
profile = session_data.get("profile", {})
trunk_headers = profile.get("trunk-sip-headers", {})
if "Twilio-AccountSid" in trunk_headers:
underlying_provider = "twilio"
direction = webhook_data.get("Direction", "inbound").lower()
if direction in {"inbound", "subscriber"}:
direction = "inbound"
return NormalizedInboundData(
provider=CloudonixProvider.PROVIDER_NAME,
call_id=call_id,
from_number=webhook_data.get("From", ""),
to_number=webhook_data.get("To", ""),
direction=direction,
call_status=webhook_data.get("CallStatus", "in-progress"),
account_id=account_id,
from_country=webhook_data.get("FromCountry"),
to_country=webhook_data.get("ToCountry"),
raw_data={
**webhook_data,
"underlying_provider": underlying_provider,
},
)
@staticmethod
def validate_account_id(config_data: dict, webhook_account_id: str) -> bool:
"""
Validate that the account_id from webhook matches the Cloudonix configuration.
For Cloudonix:
- webhook_account_id is the Domain field (e.g., "test1.cloudonix.net")
- config domain_id stores the same domain string
"""
if not webhook_account_id:
return False
# Get stored domain from config (stored under 'domain_id' key)
stored_domain = config_data.get("domain_id")
if not stored_domain:
return False
return CloudonixProvider._normalize_domain(
webhook_account_id
) == CloudonixProvider._normalize_domain(stored_domain)
async def verify_inbound_signature(
self,
url: str,
webhook_data: Dict[str, Any],
headers: Dict[str, str],
body: str = "",
) -> bool:
"""
Verify the API key of an inbound Cloudonix webhook for security.
Cloudonix uses ``x-cx-apikey`` header validation instead of signature
verification. The API key from the webhook should match the
bearer_token in our configuration.
"""
api_key = headers.get("x-cx-apikey", "")
if not api_key:
logger.warning("No x-cx-apikey provided in Cloudonix webhook")
return False
# The bearer_token in config is the same as x-cx-apikey header value
if not self.bearer_token:
logger.warning("No bearer_token configured for Cloudonix provider")
return False
# Compare the API keys
is_valid = api_key == self.bearer_token
if is_valid:
logger.info(
f"Cloudonix x-cx-apikey validation successful for "
f"domain={self.domain_id}"
)
else:
# A mismatch here usually means the config this request resolved to
# is not the one Cloudonix called: the key belongs to whichever
# domain took the call, the expected key to whatever the inbound
# route matched. Print both tails so the two can be told apart.
logger.warning(
f"Cloudonix x-cx-apikey validation failed. Received key ending "
f"with ...{_key_tail(api_key)}, expected key ending with "
f"...{_key_tail(self.bearer_token)} "
f"(matched config domain={self.domain_id}, "
f"application={self.application_name})"
)
return True # TODO: update this post clarification from cloudonix
async def configure_inbound(
self, address: str, webhook_url: Optional[str]
) -> ProviderSyncResult:
"""Update the ``url`` on the Cloudonix Voice Application.
PATCH is partial, so we send only ``url`` and ``method=POST`` (our
``/inbound/run`` is POST-only); ``type``, ``active``, and ``profile``
are preserved as configured in the cockpit. The URL is shared across
every DNID on the application — clearing is a no-op to avoid
silently breaking inbound for sibling numbers.
"""
if webhook_url is None:
logger.info(
f"Cloudonix configure_inbound clear for {address}: skipping "
f"application update (url is shared across all DNIDs on Voice "
f"Application {self.application_name})"
)
return ProviderSyncResult(ok=True)
if not self.validate_config():
return ProviderSyncResult(
ok=False, message="Cloudonix provider not properly configured"
)
if not self.application_name:
return ProviderSyncResult(
ok=False,
message=(
"Cloudonix application_name is not configured. Set it in "
"the telephony configuration so inbound webhooks can be "
"synced to the right Voice Application."
),
)
app_endpoint = (
f"{self.base_url}/customers/self/domains/{self.domain_id}/"
f"applications/{self.application_name}"
)
data = {
"url": webhook_url,
"method": "POST",
}
try:
async with aiohttp.ClientSession() as session:
async with session.patch(
app_endpoint, json=data, headers=self._get_auth_headers()
) as response:
if response.status != 200:
body = await response.text()
logger.error(
f"Cloudonix Voice Application update failed for "
f"{self.application_name} on domain "
f"{self.domain_id}: {response.status} {body}"
)
return ProviderSyncResult(
ok=False,
message=f"Cloudonix API {response.status}: {body}",
)
except Exception as e:
logger.error(
f"Exception updating Cloudonix Voice Application "
f"{self.application_name}: {e}"
)
return ProviderSyncResult(ok=False, message=f"Cloudonix update failed: {e}")
logger.info(
f"Cloudonix url set on Voice Application {self.application_name} "
f"(domain={self.domain_id}, triggered by address {address})"
)
return ProviderSyncResult(ok=True)
@staticmethod
def _dnid_matches_address(data: Any, expected: str) -> bool:
if not isinstance(data, dict):
return False
source = data.get("source")
if source is not None:
try:
source = normalize_telephony_address(str(source)).canonical
except ValueError:
source = str(source).strip()
if source == expected:
return True
dnid = data.get("dnid")
if not isinstance(dnid, str):
return False
escaped_address = re.escape(expected)
return dnid.strip() in {
expected,
f"^{escaped_address}$",
f"^({escaped_address})$",
}
async def provision_phone_number(self, address: str) -> ProviderSyncResult | None:
"""Ensure ``address`` exists as a DNID on the configured application."""
if not (self.bearer_token and self.domain_id):
return ProviderSyncResult(
ok=False,
message="Cloudonix bearer token and domain are required to create a DNID",
)
if not self.application_name:
return ProviderSyncResult(
ok=False,
message=(
"Cloudonix application_name is required to associate the "
"new DNID with a Voice Application"
),
)
expected = normalize_telephony_address(address).canonical
encoded_domain = quote(self.domain_id, safe="")
encoded_application = quote(self.application_name, safe="")
domain_endpoint = f"{self.base_url}/domains/{encoded_domain}"
dnids_endpoint = f"{domain_endpoint}/dnids"
application_endpoint = f"{domain_endpoint}/applications/{encoded_application}"
headers = self._get_auth_headers()
try:
async with aiohttp.ClientSession() as session:
async with session.get(dnids_endpoint, headers=headers) as response:
if response.status == 200:
dnids = await response.json()
if not isinstance(dnids, list):
return ProviderSyncResult(
ok=False,
message=(
"Cloudonix dnidList response was not an array"
),
)
if any(
self._dnid_matches_address(dnid, expected) for dnid in dnids
):
return ProviderSyncResult(ok=True)
elif response.status != 404:
body = await response.text()
return ProviderSyncResult(
ok=False,
message=(
f"Cloudonix DNID list failed with HTTP "
f"{response.status}: {body}"
),
)
async with session.get(
application_endpoint, headers=headers
) as response:
if response.status != 200:
body = await response.text()
return ProviderSyncResult(
ok=False,
message=(
"Cloudonix Voice Application lookup failed with "
f"HTTP {response.status}: {body}"
),
)
application = await response.json()
application_uuid = (
application.get("uuid") if isinstance(application, dict) else None
)
if (
not isinstance(application_uuid, str)
or not application_uuid.strip()
):
return ProviderSyncResult(
ok=False,
message=(
"Cloudonix applicationGet response did not include "
"an application UUID"
),
)
encoded_application_uuid = quote(application_uuid.strip(), safe="")
application_dnids_endpoint = (
f"{domain_endpoint}/applications/{encoded_application_uuid}/dnids"
)
payload = {
"source": expected,
"prefix": False,
"expression": False,
"asteriskCompatible": False,
}
redacted_headers = {
key: "Bearer [REDACTED]"
if key.lower() == "authorization"
else value
for key, value in headers.items()
}
logger.info(
f"[Cloudonix] dnidCreate request:\n"
f" Method: POST\n"
f" Endpoint: {application_dnids_endpoint}\n"
f" Headers: {json.dumps(redacted_headers, sort_keys=True)}\n"
f" Payload: {json.dumps(payload, sort_keys=True)}"
)
async with session.post(
application_dnids_endpoint, json=payload, headers=headers
) as response:
if response.status not in (200, 201, 204):
body = await response.text()
return ProviderSyncResult(
ok=False,
message=(
f"Cloudonix dnidCreate failed with HTTP "
f"{response.status}: {body}"
),
)
except Exception as e:
logger.error(
f"Exception provisioning Cloudonix DNID {expected} on domain "
f"{self.domain_id}: {e}"
)
return ProviderSyncResult(
ok=False, message=f"Cloudonix DNID provisioning failed: {e}"
)
logger.info(
f"Cloudonix DNID {expected} created on domain {self.domain_id} "
f"for Voice Application {self.application_name}"
)
return ProviderSyncResult(ok=True)
async def validate_phone_number(self, address: str) -> ProviderSyncResult:
"""Verify that the address exists as a DNID in this Cloudonix domain."""
if not (self.bearer_token and self.domain_id):
raise ProviderPhoneNumberLookupError(
"Cloudonix bearer token and domain are required to validate "
"phone-number ownership"
)
normalized = normalize_telephony_address(address)
expected = normalized.canonical
encoded_address = quote(expected, safe="")
endpoint = (
f"{self.base_url}/customers/self/domains/{self.domain_id}/dnids/"
f"{encoded_address}"
)
try:
async with aiohttp.ClientSession() as session:
async with session.get(
endpoint, headers=self._get_auth_headers()
) as response:
if response.status == 404:
return ProviderSyncResult(
ok=False,
message=(
f"Address {expected} is not configured as a DNID "
f"in Cloudonix domain {self.domain_id}. Add it in "
"the Cloudonix Cockpit first."
),
)
if response.status != 200:
body = await response.text()
raise ProviderPhoneNumberLookupError(
f"Cloudonix API {response.status}: {body}",
status_code=response.status,
)
data = await response.json()
except ProviderPhoneNumberLookupError:
raise
except Exception as e:
raise ProviderPhoneNumberLookupError(
f"Cloudonix DNID lookup failed: {e}"
) from e
if self._dnid_matches_address(data, expected):
return ProviderSyncResult(ok=True)
return ProviderSyncResult(
ok=False,
message=(
f"Address {expected} is not configured as a DNID in Cloudonix "
f"domain {self.domain_id}. Add it in the Cloudonix Cockpit first."
),
)
async def start_inbound_stream(
self,
*,
websocket_url: str,
workflow_run_id: int,
normalized_data,
backend_endpoint: str,
):
"""
Generate the appropriate CXML response for an inbound Cloudonix webhook.
Returns CXML to connect to WebSocket, same format as outbound calls.
"""
from fastapi import Response
# Generate CXML response (same format as outbound calls)
cxml_content = f"""
"""
# Redacted: the stream URL carries a bearer capability token, and this
# log line is the one place it would otherwise reach a log sink.
logger.info("Cloudonix inbound CXML response content:")
logger.info(ws_auth.redact_token(cxml_content))
response = Response(content=cxml_content, media_type="application/xml")
logger.info(f"Cloudonix inbound response object: {response}")
logger.info(f"Response headers: {response.headers}")
logger.info(f"Response media type: {response.media_type}")
return response
@staticmethod
def generate_validation_error_response(error_type) -> tuple:
"""
Generate Cloudonix-specific error response for validation failures.
Since Cloudonix is TwiML-compatible, we use the same XML format.
"""
from fastapi import Response
from api.errors.telephony_errors import TELEPHONY_ERROR_MESSAGES, TelephonyError
message = TELEPHONY_ERROR_MESSAGES.get(
error_type, TELEPHONY_ERROR_MESSAGES[TelephonyError.GENERAL_AUTH_FAILED]
)
twiml_content = f"""
{message}
"""
return Response(content=twiml_content, media_type="application/xml")
@staticmethod
def generate_error_response(error_type: str, message: str) -> tuple:
"""
Generate a Cloudonix-specific error response.
Since Cloudonix is TwiML-compatible, we use TwiML format.
"""
from fastapi import Response
# Map error types to appropriate TwiML responses
if error_type == "auth_failed":
twiml = """
Authentication failed. This call cannot be processed.
"""
elif error_type == "not_configured":
twiml = """
Service not configured. Please contact support.
"""
elif error_type == "invalid_number":
twiml = """
Invalid phone number. This call cannot be processed.
"""
else:
# Generic error
twiml = f"""
An error occurred: {message}
"""
return Response(content=twiml, media_type="application/xml"), "application/xml"
# ======== CALL TRANSFER METHODS ========
@staticmethod
def _conference_join_cxml(conference_name: str, callback_url: str) -> str:
"""CXML the destination leg runs once it answers: join the conference."""
return (
''
""
"You have answered a transfer call. Connecting you now."
""
f'{conference_name}'
""
""
)
async def transfer_call(
self,
destination: str,
transfer_id: str,
conference_name: str,
timeout: int = 30,
**kwargs: Any,
) -> Dict[str, Any]:
"""Dial the transfer destination into a conference via Cloudonix.
Places an outbound call whose inline CXML joins ``conference_name`` when
the destination answers, and sets the call object's ``callback`` to the
Cloudonix transfer-result route so the destination's session-status
transitions (``connected`` / terminal) drive transfer completion. The
original caller leg is later forked into the same conference by
``CloudonixConferenceStrategy``.
Supports both PSTN numbers and SIP URIs as ``destination``.
Returns a dict with the destination leg's ``call_sid`` (session token).
"""
if not self.validate_config():
raise ValueError("Cloudonix provider not properly configured")
from_number = self.select_from_number()
if from_number is None:
raise ValueError(
"No phone numbers configured for Cloudonix provider; a caller-id "
"is required to place the transfer call."
)
backend_endpoint, _ = await get_backend_endpoints()
callback_url = f"{backend_endpoint}/api/v1/telephony/cloudonix/transfer-result/{transfer_id}"
endpoint = f"{self.base_url}/calls/{self.domain_id}/application"
data: Dict[str, Any] = {
"destination": destination,
"caller-id": from_number,
"cxml": self._conference_join_cxml(conference_name, callback_url),
"callback": callback_url,
"timeout": timeout,
}
data.update(kwargs)
self._pin_trunk(data, from_number)
headers = self._get_auth_headers()
masked_destination = f"***{destination[-4:]}" if len(destination) > 4 else "***"
logger.info(
f"[Cloudonix Transfer] Dialing {masked_destination} into conference "
f"{conference_name} (transfer_id={transfer_id})"
)
async with aiohttp.ClientSession() as session:
async with session.post(endpoint, json=data, headers=headers) as response:
response_text = await response.text()
if response.status != 200:
logger.error(
f"[Cloudonix Transfer] Dial failed: HTTP {response.status}, "
f"body: {response_text}"
)
raise Exception(
f"Cloudonix transfer dial failed (HTTP {response.status}): "
f"{response_text}"
)
response_data = await response.json()
session_token = response_data.get("token")
if not session_token:
raise Exception(
"No session token returned from Cloudonix transfer dial"
)
logger.info(
f"[Cloudonix Transfer] Destination leg initiated "
f"(token={session_token}, transfer_id={transfer_id})"
)
return {
"call_sid": session_token,
"status": response_data.get("status", "initiated"),
"provider": self.PROVIDER_NAME,
"from_number": from_number,
"to_number": destination,
"raw_response": response_data,
}
def supports_transfers(self) -> bool:
"""Cloudonix supports conference-based call transfers."""
return True