248 lines
10 KiB
Python
Executable File
248 lines
10 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Render an explicit, outbound-only UDP config from the Trunk codec profile."""
|
|
|
|
import argparse
|
|
import ipaddress
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
PUBLIC_IP = "123.56.71.98"
|
|
CODEC_TO_ASTERISK = {"PCMA": "alaw", "PCMU": "ulaw"}
|
|
TRUNK_ID_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]{0,127}")
|
|
|
|
|
|
def scalar(value, name, secret=False):
|
|
if (
|
|
not isinstance(value, str)
|
|
or not value
|
|
or len(value) > 256
|
|
or any(c in value for c in "\r\n;[]\\")
|
|
):
|
|
raise ValueError(f"{name} is missing or contains unsupported INI characters")
|
|
if value.startswith("CHANGE_ME"):
|
|
raise ValueError(f"{name} is still a placeholder")
|
|
if secret and len(value) < 12:
|
|
raise ValueError(f"{name} must have at least 12 characters")
|
|
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 (
|
|
not re.fullmatch(r"[A-Za-z0-9.-]+", host)
|
|
or host == PUBLIC_IP
|
|
or host.endswith((".invalid", ".example"))
|
|
):
|
|
raise ValueError(
|
|
f"{name}.host must be the SIP provider IPv4/DNS address, not our whitelisted IP or a placeholder"
|
|
)
|
|
port = data.get("port", 5060)
|
|
if isinstance(port, bool) or not isinstance(port, int) or not 1 <= port <= 65535:
|
|
raise ValueError(f"{name}.port is invalid")
|
|
mode = data.get("auth_mode")
|
|
if mode not in ("ip", "digest"):
|
|
raise ValueError(f"{name}.auth_mode must be explicitly ip or digest")
|
|
registration = data.get("register", False)
|
|
if not isinstance(registration, bool) or registration and mode != "digest":
|
|
raise ValueError("registration requires explicit digest authentication")
|
|
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(
|
|
r"[A-Za-z0-9_.+\-]+", from_user
|
|
):
|
|
raise ValueError(f"{name}.from_user must be a plain caller identifier")
|
|
text += f"from_user={from_user}\n"
|
|
display_identity = data.get("caller_id")
|
|
if display_identity is not None:
|
|
display_identity = scalar(display_identity, name + ".caller_id")
|
|
if not re.fullmatch(
|
|
r"[A-Za-z0-9_.+\- ]+(?:<[A-Za-z0-9_.+\- ]+>)?", display_identity
|
|
):
|
|
raise ValueError(f"{name}.caller_id contains unsupported characters")
|
|
text += f"callerid={display_identity}\n"
|
|
auth = ""
|
|
if mode == "digest":
|
|
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")
|
|
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={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 += (
|
|
f"\n[{name}-aor]\ntype=aor\ncontact=sip:{host}:{port}\nqualify_frequency=10\n"
|
|
)
|
|
return text + auth
|
|
|
|
|
|
def _trunks(cfg):
|
|
if "trunks" not in cfg:
|
|
primary_data = cfg.get("primary", {})
|
|
if not isinstance(primary_data, dict):
|
|
raise ValueError("primary must be an object")
|
|
rows = [("provider-primary", primary_data)]
|
|
backup_data = cfg.get("backup")
|
|
if backup_data is not None and not isinstance(backup_data, dict):
|
|
raise ValueError("backup must be an object when configured")
|
|
if backup_data:
|
|
rows.append(("provider-backup", backup_data))
|
|
if (primary_data.get("host"), primary_data.get("port", 5060)) == (
|
|
backup_data.get("host"),
|
|
backup_data.get("port", 5060),
|
|
):
|
|
raise ValueError(
|
|
"primary and backup targets must be distinct; shared failure domains still require validation"
|
|
)
|
|
return rows
|
|
if "primary" in cfg or "backup" in cfg:
|
|
raise ValueError("use trunks instead of mixing arbitrary trunks with primary/backup")
|
|
raw = cfg["trunks"]
|
|
if not isinstance(raw, list) or not raw:
|
|
raise ValueError("trunks must be a non-empty list")
|
|
rows = []
|
|
seen = set()
|
|
for index, data in enumerate(raw):
|
|
if not isinstance(data, dict):
|
|
raise TypeError(f"trunks[{index}] must be an object")
|
|
name = data.get("trunk_id")
|
|
if not isinstance(name, str) or not TRUNK_ID_RE.fullmatch(name):
|
|
raise ValueError(
|
|
f"trunks[{index}].trunk_id must be a safe Asterisk endpoint identifier"
|
|
)
|
|
if name in seen:
|
|
raise ValueError(f"duplicate trunk_id: {name}")
|
|
seen.add(name)
|
|
rows.append((name, data))
|
|
return rows
|
|
|
|
|
|
def render(cfg, env):
|
|
if cfg.get("public_ip") != PUBLIC_IP or cfg.get("transport") != "udp":
|
|
raise ValueError(
|
|
"fixed public IP and UDP baseline required; TCP/TLS need a reviewed configuration"
|
|
)
|
|
try:
|
|
network = ipaddress.ip_network(cfg["local_net"], strict=True)
|
|
bind = ipaddress.ip_address(cfg.get("ari_bind", "127.0.0.1"))
|
|
except (KeyError, ValueError) as exc:
|
|
raise ValueError(
|
|
"local_net and ARI bind address must be explicit valid IP configuration"
|
|
) from exc
|
|
if network.version != 4 or not network.is_private or network.prefixlen < 8:
|
|
raise ValueError(
|
|
"local_net must be the actual private IPv4 VPC network, not a default route"
|
|
)
|
|
if bind.version != 4 or bind.is_unspecified or not bind.is_private:
|
|
raise ValueError(
|
|
"ARI must bind to loopback or an approved private IPv4 management address"
|
|
)
|
|
password = scalar(env.get("ARI_PASSWORD"), "ARI_PASSWORD", secret=True)
|
|
if len(password) < 32:
|
|
raise ValueError("ARI_PASSWORD must have at least 32 characters")
|
|
trunks = _trunks(cfg)
|
|
sections = [endpoint(name, data, env) for name, data in trunks]
|
|
primary_name, primary_data = trunks[0]
|
|
transport = f"[global]\ntype=global\nuser_agent=agent-call\n\n[transport-udp]\ntype=transport\nprotocol=udp\nbind=0.0.0.0:5060\nlocal_net={network}\nexternal_signaling_address={PUBLIC_IP}\nexternal_media_address={PUBLIC_IP}\n\n"
|
|
caller = primary_data.get("caller_id") or primary_data.get("from_user")
|
|
extensions = "[deny-inbound]\nexten => s,1,Hangup()\nexten => _.,1,Hangup()\n"
|
|
if caller:
|
|
caller = scalar(caller, f"{primary_name}.caller_id")
|
|
extensions += (
|
|
"\n[outbound]\n"
|
|
"; ARI Cell calls select an approved endpoint with PJSIP/<number>@<trunk_id>.\n"
|
|
"exten => _X.,1,NoOp(agent-call outbound fallback)\n"
|
|
f" same => n,Set(CALLERID(all)={caller})\n"
|
|
f" same => n,Dial(PJSIP/${{EXTEN}}@{primary_name},60)\n"
|
|
" same => n,Hangup()\n"
|
|
)
|
|
return {
|
|
"http.conf": f"[general]\nenabled=yes\nbindaddr={bind}\nbindport=8088\n",
|
|
"ari.conf": f"[general]\nenabled=yes\npretty=no\n\n[outbound]\ntype=user\nread_only=no\npassword={password}\n",
|
|
"pjsip.conf": transport + "\n".join(sections),
|
|
"rtp.conf": "[general]\nrtpstart=10000\nrtpend=10800\nstrictrtp=yes\n",
|
|
"extensions.conf": extensions,
|
|
}
|
|
|
|
|
|
def write_config(files, output):
|
|
output = Path(output)
|
|
if output.exists() or output.is_symlink():
|
|
raise ValueError(
|
|
"output already exists; use a new version directory and review the change before switching"
|
|
)
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = Path(tempfile.mkdtemp(prefix=".asterisk-render-", dir=output.parent))
|
|
try:
|
|
for name, content in files.items():
|
|
path = temporary / name
|
|
path.write_text(content)
|
|
path.chmod(0o600)
|
|
temporary.chmod(0o750)
|
|
temporary.rename(output)
|
|
except Exception:
|
|
shutil.rmtree(temporary)
|
|
raise
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--config", required=True)
|
|
parser.add_argument("--output", default="deploy/asterisk/generated")
|
|
args = parser.parse_args()
|
|
try:
|
|
cfg = json.loads(Path(args.config).read_text())
|
|
files = render(cfg, os.environ)
|
|
write_config(files, args.output)
|
|
print(
|
|
f"Rendered {len(files)} configuration files to {args.output}; services NOT started. Check container UID permissions before deployment."
|
|
)
|
|
except (ValueError, OSError, KeyError) as exc:
|
|
print(f"ERROR: {exc}", file=sys.stderr)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|