fix real cell RTP bridge readiness

2026-09-17 14:29:37 +08:00
parent b6d0af1a56
commit 668c12e523
10 changed files with 781 additions and 51 deletions
@@ -11,9 +11,11 @@ from __future__ import annotations
 import base64
 import contextlib
 import http.client
+import ipaddress
 import json
 import os
 import queue
+import re
 import socket
 import sqlite3
 import ssl
@@ -56,6 +58,7 @@ class CellCallConfig:
    ari_app: str = "agent-call"
    trunk_id: str = "provider-primary"
    caller_id: str = "BD93205882"
+    dial_prefix: str = "7089"
    rtp_bind_host: str = "127.0.0.1"
    rtp_bind_port: int = 0
    rtp_format: str = "alaw"
@@ -80,6 +83,13 @@ class CellCallConfig:
            raise CellCallError(
                "CELL_CONFIG_INVALID", "Cell ARI/trunk identity is incomplete"
            )
+        if len(self.dial_prefix) > 32 or (
+            self.dial_prefix and not re.fullmatch(r"[A-Za-z0-9]+", self.dial_prefix)
+        ):
+            raise CellCallError(
+                "DIAL_PREFIX_INVALID",
+                "Cell dial prefix must be an ASCII alphanumeric token",
+            )
        if self.rtp_format != "alaw":
            raise CellCallError(
                "RTP_FORMAT_UNSUPPORTED", "real Cell currently requires PCMA/alaw"
@@ -106,6 +116,7 @@ class CellCallConfig:
            ari_app=os.environ.get("CELL_ARI_APP", "agent-call"),
            trunk_id=os.environ.get("CELL_SIP_TRUNK_ID", "provider-primary"),
            caller_id=os.environ.get("CELL_CALLER_ID", "BD93205882"),
+            dial_prefix=os.environ.get("CELL_DIAL_PREFIX", "7089"),
            rtp_bind_host=os.environ.get("CELL_RTP_BIND_HOST", "127.0.0.1"),
            rtp_bind_port=rtp_bind_port,
            recording_dir=os.environ.get("CELL_RECORDING_DIR", "/data/recordings"),
@@ -235,6 +246,7 @@ class RTPMedia:
            self.socket.close()
            raise
        self.peer: tuple[str, int] | None = None
+        self.recv_ssrc: int | None = None
        self.sequence = 0
        self.timestamp = 0
        self.ssrc = uuid.uuid4().int & 0xFFFFFFFF
@@ -258,7 +270,7 @@ class RTPMedia:
        self.socket.settimeout(max(0.0, timeout_s))
        try:
            packet, peer = self.socket.recvfrom(2048)
-        except (TimeoutError, ConnectionRefusedError):
+        except (BlockingIOError, TimeoutError, ConnectionRefusedError):
            return None
        except OSError as exc:
            if self.socket.fileno() < 0:
@@ -266,15 +278,26 @@ class RTPMedia:
            raise CellCallError(
                "RTP_RECEIVE_FAILED", "RTP receive failed", True
            ) from exc
-        payload = self._payload(packet)
-        if payload is None:
+        parsed = self._packet_info(packet)
+        if parsed is None:
+            return None
+        payload, payload_type, ssrc = parsed
+        if payload_type != 8:
            return None
        try:
-            self.peer = (str(peer[0]), int(peer[1]))
+            candidate = (str(peer[0]), int(peer[1]))
        except (IndexError, TypeError, ValueError) as exc:
            raise CellCallError(
                "RTP_PEER_INVALID", "RTP peer address is invalid"
            ) from exc
+        if self.peer is not None and candidate != self.peer:
+            return None
+        if self.recv_ssrc is not None and ssrc != self.recv_ssrc:
+            return None
+        if self.peer is None:
+            self.peer = candidate
+        if self.recv_ssrc is None:
+            self.recv_ssrc = ssrc
        return payload

    def send_pcm16(self, raw: bytes, input_rate_hz: int = 24000) -> int:
@@ -282,7 +305,7 @@ class RTPMedia:
            return 0
        if self.peer is None:
            raise CellCallError(
-                "RTP_PEER_MISSING", "no inbound RTP peer has been observed"
+                "RTP_PEER_MISSING", "no Asterisk externalMedia RTP peer is configured"
            )
        payload = pcm16_to_alaw(raw, input_rate_hz, 8000)
        sent = 0
@@ -312,9 +335,11 @@ class RTPMedia:
        return sent

    @staticmethod
-    def _payload(packet: bytes) -> bytes | None:
+    def _packet_info(packet: bytes) -> tuple[bytes, int, int] | None:
        if len(packet) < 12 or packet[0] >> 6 != 2:
            return None
+        payload_type = packet[1] & 0x7F
+        ssrc = struct.unpack("!I", packet[8:12])[0]
        cc = packet[0] & 0x0F
        offset = 12 + cc * 4
        if len(packet) < offset:
@@ -332,7 +357,13 @@ class RTPMedia:
            if not padding or padding > end - offset:
                return None
            end -= padding
-        return packet[offset:end] or None
+        payload = packet[offset:end]
+        return (payload, payload_type, ssrc) if payload else None
+
+    @classmethod
+    def _payload(cls, packet: bytes) -> bytes | None:
+        parsed = cls._packet_info(packet)
+        return parsed[0] if parsed is not None else None


 def alaw_to_pcm16(payload: bytes) -> bytes:
@@ -462,16 +493,17 @@ class RealCellCall:
                },
            )
            self.external_channel_id = self._channel_id(external)
+            self._try_set_external_media_peer()
            self._add_when_ready(self.external_channel_id)
            target = self.ari.request(
                "POST",
                "channels",
                {
-                    "endpoint": f"PJSIP/{'7089' + callee}@{self.config.trunk_id}",
+                    "endpoint": f"PJSIP/{self.config.dial_prefix + callee}@{self.config.trunk_id}",
                    "app": self.config.ari_app,
                    "appArgs": self.call_id,
                    "callerId": self.config.caller_id,
-                    "timeout": 30000,
+                    "timeout": 30,
                    "channelId": f"target_{self.call_id}",
                },
            )
@@ -615,30 +647,64 @@ class RealCellCall:

    def _wait_answer(self, result: CellCallResult) -> None:
        deadline = time.monotonic() + self.config.event_timeout_s
+        answered = False
+        expected_channels = {
+            channel_id
+            for channel_id in (self.external_channel_id, self.target_channel_id)
+            if channel_id
+        }
        while time.monotonic() < deadline:
            try:
                event = self.events.get(timeout=min(0.5, deadline - time.monotonic()))
            except queue.Empty:
-                continue
-            event_type = event.get("type")
-            channel = event.get("channel") or {}
-            channel_id = channel.get("id")
-            if event_type == "StasisStart" and isinstance(channel_id, str):
-                self._add_when_ready(channel_id)
-            if (
-                event_type == "ChannelStateChange"
-                and channel_id == self.target_channel_id
-                and channel.get("state") == "Up"
-            ):
-                return
-            if (
-                event_type in {"ChannelDestroyed", "StasisEnd", "ChannelHangupRequest"}
-                and channel_id == self.target_channel_id
-            ):
-                self._call_ended = True
-                raise CellCallError(
-                    "CALL_NOT_ANSWERED", "target channel ended before answer"
-                )
+                event = None
+            if event is not None:
+                event_type = event.get("type")
+                channel = event.get("channel") or {}
+                channel_id = channel.get("id")
+                if (
+                    event_type == "StasisStart"
+                    and isinstance(channel_id, str)
+                    and channel_id in expected_channels
+                ):
+                    if channel_id == self.external_channel_id:
+                        self._try_set_external_media_peer()
+                    self._add_when_ready(channel_id)
+                if (
+                    event_type == "ChannelStateChange"
+                    and channel_id == self.target_channel_id
+                    and channel.get("state") == "Up"
+                ):
+                    answered = True
+                if (
+                    event_type
+                    in {"ChannelDestroyed", "StasisEnd", "ChannelHangupRequest"}
+                    and channel_id == self.target_channel_id
+                ):
+                    self._call_ended = True
+                    raise CellCallError(
+                        "CALL_NOT_ANSWERED", "target channel ended before answer"
+                    )
+            if answered:
+                self._try_set_external_media_peer()
+                try:
+                    bridge_channels = self._bridge_channels()
+                except CellCallError as exc:
+                    if exc.code not in {"ARI_HTTP_404", "ARI_HTTP_409"}:
+                        raise
+                else:
+                    if (
+                        expected_channels.issubset(bridge_channels)
+                        and self.media is not None
+                        and self.media.peer is not None
+                    ):
+                        return
+        if answered:
+            raise CellCallError(
+                "ARI_BRIDGE_NOT_READY",
+                "answered channels did not join the expected bridge",
+                True,
+            )
        raise CellCallError(
            "CALL_ANSWER_TIMEOUT", "ARI did not report an answered channel", True
        )
@@ -724,16 +790,89 @@ class RealCellCall:
            self._out_bytes += bytes_sent
        return sent

-    def _add_when_ready(self, channel_id: str | None) -> None:
-        if not channel_id or channel_id in self._known_channels:
-            return
-        self._known_channels.add(channel_id)
-        with contextlib.suppress(CellCallError):
-            self.ari.request(
-                "POST",
-                f"bridges/{self.bridge_id}/addChannel",
-                {"channel": channel_id},
+    def _try_set_external_media_peer(self) -> bool:
+        if self.external_channel_id is None:
+            return False
+        try:
+            self._set_external_media_peer(self.external_channel_id)
+        except CellCallError as exc:
+            if exc.code in {
+                "ARI_HTTP_400",
+                "ARI_HTTP_404",
+                "ARI_HTTP_409",
+                "RTP_EXTERNAL_PEER_UNAVAILABLE",
+            }:
+                return False
+            raise
+        return True
+
+    def _set_external_media_peer(self, channel_id: str) -> None:
+        if self.media is None:
+            raise CellCallError("RTP_UNAVAILABLE", "RTP media is not initialized")
+        values: dict[str, str] = {}
+        for variable in ("UNICASTRTP_LOCAL_ADDRESS", "UNICASTRTP_LOCAL_PORT"):
+            response = self.ari.request(
+                "GET",
+                f"channels/{channel_id}/variable",
+                {"variable": variable},
            )
+            value = response.get("value") if isinstance(response, dict) else None
+            if not isinstance(value, str) or not value:
+                raise CellCallError(
+                    "RTP_EXTERNAL_PEER_UNAVAILABLE",
+                    f"ARI did not provide {variable}",
+                    True,
+                )
+            values[variable] = value
+        try:
+            address = ipaddress.ip_address(values["UNICASTRTP_LOCAL_ADDRESS"])
+            port = int(values["UNICASTRTP_LOCAL_PORT"])
+        except (ValueError, TypeError) as exc:
+            raise CellCallError(
+                "RTP_EXTERNAL_PEER_INVALID", "ARI external RTP peer is invalid"
+            ) from exc
+        if address.is_unspecified or address.is_multicast or not 1 <= port <= 65535:
+            raise CellCallError(
+                "RTP_EXTERNAL_PEER_INVALID", "ARI external RTP peer is invalid"
+            )
+        self.media.peer = (str(address), port)
+
+    def _bridge_channels(self) -> set[str]:
+        response = self.ari.request("GET", f"bridges/{self.bridge_id}")
+        channels = response.get("channels") if isinstance(response, dict) else None
+        if not isinstance(channels, list) or not all(
+            isinstance(channel_id, str) for channel_id in channels
+        ):
+            raise CellCallError(
+                "ARI_BRIDGE_INVALID", "ARI did not return bridge channels"
+            )
+        return set(channels)
+
+    def _add_when_ready(self, channel_id: str | None) -> bool:
+        if not channel_id or channel_id in self._known_channels:
+            return channel_id in self._known_channels
+        for attempt in range(3):
+            try:
+                self.ari.request(
+                    "POST",
+                    f"bridges/{self.bridge_id}/addChannel",
+                    {"channel": channel_id},
+                )
+            except CellCallError as exc:
+                if exc.code not in {
+                    "ARI_HTTP_400",
+                    "ARI_HTTP_404",
+                    "ARI_HTTP_409",
+                    "ARI_HTTP_422",
+                }:
+                    raise
+                if attempt == 2:
+                    return False
+                time.sleep(0.05 * (attempt + 1))
+            else:
+                self._known_channels.add(channel_id)
+                return True
+        return False

    @staticmethod
    def _channel_id(value: dict[str, Any] | bytes) -> str:
@@ -745,12 +884,25 @@ class RealCellCall:
        return channel_id

    def _finish_recording(self) -> str | None:
-        with contextlib.suppress(CellCallError):
-            self.ari.request("DELETE", f"recordings/live/{self.recording_name}")
        try:
-            raw = self.ari.request(
-                "GET", f"recordings/stored/{self.recording_name}/file"
-            )
+            try:
+                self.ari.request("POST", f"recordings/live/{self.recording_name}/stop")
+            except CellCallError as exc:
+                if exc.code != "ARI_HTTP_404":
+                    return None
+            deadline = time.monotonic() + min(self.config.event_timeout_s, 5.0)
+            while True:
+                try:
+                    raw = self.ari.request(
+                        "GET", f"recordings/stored/{self.recording_name}/file"
+                    )
+                    break
+                except CellCallError as exc:
+                    if exc.code != "ARI_HTTP_404":
+                        return None
+                    if time.monotonic() >= deadline:
+                        return None
+                    time.sleep(0.1)
        except CellCallError:
            return None
        if not isinstance(raw, bytes) or not raw: