rogee/agent-call
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: deliver versioned SIP config to Cells

Browse Source
This commit is contained in:
rogee
2026-09-14 14:38:15 +08:00
parent 3a47914a6c
commit c8716608a2
26 changed files with 4665 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
deploy/render_asterisk.py
+37 -5
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
"""Render an explicit, outbound-only UDP/PCMA baseline. Does not start services."""
"""Render an explicit, outbound-only UDP config from the Trunk codec profile."""
import argparse
import ipaddress
@@ -12,6 +12,7 @@ import tempfile
from pathlib import Path
PUBLIC_IP = "123.56.71.98"
CODEC_TO_ASTERISK = {"PCMA": "alaw", "PCMU": "ulaw"}
def scalar(value, name, secret=False):
@@ -29,6 +30,32 @@ def scalar(value, name, secret=False):
return value
def codec_allow(name, data):
profile = data.get("codec_profile")
if not isinstance(profile, dict):
raise ValueError( # noqa: TRY004 - render() exposes one config-error type
f"{name}.codec_profile must be explicit"
)
allowed = profile.get("allowed")
if not isinstance(allowed, list) or not allowed:
raise ValueError(f"{name}.codec_profile.allowed must be a non-empty list")
normalized = []
for codec in allowed:
if not isinstance(codec, str) or codec.upper() not in CODEC_TO_ASTERISK:
raise ValueError(f"{name}.codec_profile contains an unsupported codec")
codec = codec.upper()
if codec in normalized:
raise ValueError(f"{name}.codec_profile contains a duplicate codec")
normalized.append(codec)
preferred = profile.get("preferred", normalized[0])
if not isinstance(preferred, str) or preferred.upper() not in normalized:
raise ValueError(f"{name}.codec_profile.preferred must be allowed")
ordered = [preferred.upper()] + [
codec for codec in normalized if codec != preferred.upper()
]
return ",".join(CODEC_TO_ASTERISK[codec] for codec in ordered)
def endpoint(name, data, env):
host = scalar(data.get("host"), name + ".host")
if (
@@ -48,7 +75,8 @@ def endpoint(name, data, env):
registration = data.get("register", False)
if not isinstance(registration, bool) or registration and mode != "digest":
raise ValueError("registration requires explicit digest authentication")
text = f"[{name}]\ntype=endpoint\ntransport=transport-udp\ncontext=deny-inbound\ndisallow=all\nallow=alaw\ndirect_media=no\nrtp_symmetric=yes\nforce_rport=yes\nrewrite_contact=yes\naors={name}-aor\n"
allow = codec_allow(name, data)
text = f"[{name}]\ntype=endpoint\ntransport=transport-udp\ncontext=deny-inbound\ndisallow=all\nallow={allow}\ndirect_media=no\nrtp_symmetric=yes\nforce_rport=yes\nrewrite_contact=yes\naors={name}-aor\n"
from_user = data.get("from_user")
if from_user is not None:
if not isinstance(from_user, str) or not re.fullmatch(
@@ -69,12 +97,16 @@ def endpoint(name, data, env):
user = scalar(data.get("username"), name + ".username")
if not re.fullmatch(r"[A-Za-z0-9_.+-]+", user):
raise ValueError("SIP username must be a plain user identifier")
password_key = "SIP_" + name.removeprefix("provider-").upper() + "_PASSWORD"
password = scalar(env.get(password_key), password_key, secret=True)
credential_env_name = (
"SIP_" + name.removeprefix("provider-").upper() + "_PASSWORD"
)
credential_value = scalar(
env.get(credential_env_name), credential_env_name, secret=True
)
text += f"outbound_auth={name}-auth\n"
if from_user is None:
text += f"from_user={user}\n"
auth = f"\n[{name}-auth]\ntype=auth\nauth_type=userpass\nusername={user}\npassword={password}\n"
auth = f"\n[{name}-auth]\ntype=auth\nauth_type=userpass\nusername={user}\npassword={credential_value}\n"
if registration:
auth += f"\n[{name}-registration]\ntype=registration\ntransport=transport-udp\noutbound_auth={name}-auth\nserver_uri=sip:{host}:{port}\nclient_uri=sip:{user}@{host}:{port}\nretry_interval=60\n"
text += (