hectorzhao/lisglosips
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add number library routing and cdr location support

Browse Source
This commit is contained in:
hectorzhao
2026-06-24 11:39:32 +08:00
parent 5fa1bd35e9
commit 7057fd3c42
63 changed files with 6361 additions and 566 deletions
Download Patch File Download Diff File Expand all files Collapse all files
infra/server-a/s28/lisglosips_hotpath.lua
+256
View File
@@ -0,0 +1,256 @@
local active = redis.call('GET', KEYS[1])
if not active or active == '' then
return {
'reject',
'CONFIG_MISSING',
'none',
'none',
'none',
'none',
'no_active_version',
'none',
'none',
'none',
'none',
'UNKNOWN',
'UNKNOWN',
'UNKNOWN',
'UNKNOWN',
'UNKNOWN',
'UNKNOWN',
''
}
end
local source_ip = ARGV[1]
local caller = ARGV[2] or ''
local callee = ARGV[3] or ''
local prefix = 'cfg:v:' .. active
local function field(json, name)
return string.match(json, '"' .. name .. '":"([^"]*)"')
end
local function normalize_callee(value)
local digits = string.gsub(value or '', '[^0-9]', '')
if string.sub(digits, 1, 4) == '0086' then
digits = string.sub(digits, 5)
elseif string.sub(digits, 1, 2) == '86' and string.len(digits) == 13 then
digits = string.sub(digits, 3)
end
return digits
end
local function unknown_number(callee_digits)
return {
city_code = 'UNKNOWN',
city_name = 'UNKNOWN',
province_code = 'UNKNOWN',
province_name = 'UNKNOWN',
carrier = 'UNKNOWN',
number_type = 'UNKNOWN',
normalized = callee_digits or ''
}
end
local function with_geo_city(info)
if info.city_code == 'UNKNOWN' or info.province_code ~= 'UNKNOWN' then
return info
end
local city_json = redis.call('GET', prefix .. ':geo_city:' .. info.city_code)
if city_json then
info.province_code = field(city_json, 'provinceCode') or info.province_code
info.province_name = field(city_json, 'provinceName') or info.province_name
info.city_name = field(city_json, 'cityName') or info.city_name
end
return info
end
local function resolve_carrier(digits, fallback)
if string.len(digits) >= 4 then
local prefix4_json = redis.call('GET', prefix .. ':carrier_prefix:' .. string.sub(digits, 1, 4))
if prefix4_json then
return field(prefix4_json, 'carrier') or fallback
end
end
if string.len(digits) >= 3 then
local prefix3_json = redis.call('GET', prefix .. ':carrier_prefix:' .. string.sub(digits, 1, 3))
if prefix3_json then
return field(prefix3_json, 'carrier') or fallback
end
end
return fallback
end
local function resolve_number(value)
local digits = normalize_callee(value)
local info = unknown_number(digits)
if string.match(digits, '^1%d%d%d%d%d%d%d%d%d%d$') then
info.number_type = 'MOBILE'
local segment_json = redis.call('GET', prefix .. ':phone_segment:' .. string.sub(digits, 1, 7))
if segment_json then
info.city_code = field(segment_json, 'cityCode') or info.city_code
info.city_name = field(segment_json, 'cityName') or info.city_name
info.province_code = field(segment_json, 'provinceCode') or info.province_code
info.province_name = field(segment_json, 'provinceName') or info.province_name
info.carrier = field(segment_json, 'carrier') or info.carrier
end
info.carrier = resolve_carrier(digits, info.carrier)
return with_geo_city(info)
end
if string.sub(digits, 1, 1) == '0' and string.len(digits) >= 3 then
info.number_type = 'LANDLINE'
local area_json = nil
if string.len(digits) >= 4 then
area_json = redis.call('GET', prefix .. ':area_code:' .. string.sub(digits, 1, 4))
end
if not area_json and string.len(digits) >= 3 then
area_json = redis.call('GET', prefix .. ':area_code:' .. string.sub(digits, 1, 3))
end
if area_json then
info.city_code = field(area_json, 'cityCode') or info.city_code
info.city_name = field(area_json, 'cityName') or info.city_name
info.province_code = field(area_json, 'provinceCode') or info.province_code
info.province_name = field(area_json, 'provinceName') or info.province_name
end
return with_geo_city(info)
end
info.carrier = resolve_carrier(digits, info.carrier)
return info
end
local number_info = resolve_number(callee)
local function result(decision, reason, gateway_id, version, line_group_id, policy_id, customer_id, vendor_id, vendor_gateway_id, host, port)
return {
decision,
reason,
gateway_id,
version,
line_group_id,
policy_id,
customer_id,
vendor_id,
vendor_gateway_id,
host,
port,
number_info.city_code,
number_info.city_name,
number_info.province_name,
number_info.carrier,
number_info.number_type,
number_info.province_code,
number_info.normalized
}
end
local gateway_id = redis.call('GET', prefix .. ':auth:ip:' .. source_ip)
if not gateway_id then
return result('reject', 'AUTH_MISSING', 'none', active, 'none', 'none', 'no_auth_ip', 'none', 'none', 'none', 'none')
end
local gateway_json = redis.call('GET', prefix .. ':customer_gateway:' .. gateway_id)
if not gateway_json then
return result('reject', 'GATEWAY_MISSING', gateway_id, active, 'none', 'none', 'gateway_missing', 'none', 'none', 'none', 'none')
end
if not string.find(gateway_json, '"status":"ENABLED"', 1, true) then
return result('reject', 'GATEWAY_DISABLED', gateway_id, active, 'none', 'none', 'gateway_disabled', 'none', 'none', 'none', 'none')
end
local customer_id = string.match(gateway_json, '"customerId":"([^"]+)"')
if not customer_id then
return result('reject', 'CUSTOMER_MISSING', gateway_id, active, 'none', 'none', 'customer_id_missing', 'none', 'none', 'none', 'none')
end
local customer_json = redis.call('GET', prefix .. ':customer:' .. customer_id)
if not customer_json then
return result('reject', 'CUSTOMER_MISSING', gateway_id, active, 'none', 'none', 'customer_missing', 'none', 'none', 'none', 'none')
end
if not string.find(customer_json, '"status":"ENABLED"', 1, true) then
return result('reject', 'CUSTOMER_DISABLED', gateway_id, active, 'none', 'none', 'customer_disabled', 'none', 'none', 'none', 'none')
end
local function region_blocked(vendor_gateway_id)
if number_info.city_code ~= 'UNKNOWN' then
local city_blocked = redis.call('SISMEMBER', prefix .. ':vendor_gateway:' .. vendor_gateway_id .. ':blocked_city_codes', number_info.city_code)
if city_blocked == 1 then
return true
end
end
if number_info.province_code ~= 'UNKNOWN' then
local province_blocked = redis.call('SISMEMBER', prefix .. ':vendor_gateway:' .. vendor_gateway_id .. ':blocked_province_codes', number_info.province_code)
if province_blocked == 1 then
return true
end
end
return false
end
local function first_vendor_route(line_group_id)
local line_group_json = redis.call('GET', prefix .. ':line_group:' .. line_group_id)
if not line_group_json then
return nil, 'LINE_GROUP_MISSING'
end
if not string.find(line_group_json, '"status":"ENABLED"', 1, true) then
return nil, 'LINE_GROUP_DISABLED'
end
local items = redis.call('LRANGE', prefix .. ':line_group:' .. line_group_id .. ':items', 0, -1)
local skipped_region = false
for _, item_json in ipairs(items) do
if string.find(item_json, '"status":"ENABLED"', 1, true) then
local vendor_gateway_id = string.match(item_json, '"vendorGatewayId":"([^"]+)"')
if vendor_gateway_id then
local vendor_gateway_json = redis.call('GET', prefix .. ':vendor_gateway:' .. vendor_gateway_id)
if vendor_gateway_json and string.find(vendor_gateway_json, '"status":"ENABLED"', 1, true) then
if region_blocked(vendor_gateway_id) then
skipped_region = true
else
local vendor_id = string.match(vendor_gateway_json, '"vendorId":"([^"]+)"') or 'none'
local host = string.match(vendor_gateway_json, '"host":"([^"]+)"') or 'none'
local port = string.match(vendor_gateway_json, '"port":([0-9]+)') or '5060'
return {vendor_id, vendor_gateway_id, host, port}, 'OK'
end
end
end
end
end
if skipped_region then
return nil, 'NO_VENDOR_ROUTE_REGION_BLOCKED'
end
return nil, 'NO_VENDOR_ROUTE'
end
local policies = redis.call('LRANGE', prefix .. ':customer_gateway:' .. gateway_id .. ':policies', 0, -1)
for _, policy_json in ipairs(policies) do
if string.find(policy_json, '"status":"ENABLED"', 1, true) then
local caller_mode = string.match(policy_json, '"callerMode":"([^"]+)"') or 'ANY'
local caller_value = string.match(policy_json, '"callerValue":"([^"]*)"') or ''
local callee_mode = string.match(policy_json, '"calleeMode":"([^"]+)"') or 'ANY'
local callee_value = string.match(policy_json, '"calleeValue":"([^"]*)"') or ''
local caller_ok = caller_mode == 'ANY'
or (caller_mode == 'EQUALS' and caller == caller_value)
or (caller_mode == 'PREFIX' and string.sub(caller, 1, string.len(caller_value)) == caller_value)
local callee_ok = callee_mode == 'ANY'
or (callee_mode == 'EQUALS' and callee == callee_value)
or (callee_mode == 'PREFIX' and string.sub(callee, 1, string.len(callee_value)) == callee_value)
if caller_ok and callee_ok then
local policy_id = string.match(policy_json, '"id":"([^"]+)"') or 'none'
local line_group_id = string.match(policy_json, '"lineGroupId":"([^"]+)"') or 'none'
local vendor_route, route_reason = first_vendor_route(line_group_id)
if vendor_route then
return result('allow', 'OK', gateway_id, active, line_group_id, policy_id, customer_id, vendor_route[1], vendor_route[2], vendor_route[3], vendor_route[4])
end
return result('reject', route_reason, gateway_id, active, line_group_id, policy_id, customer_id, 'none', 'none', 'none', 'none')
end
end
end
return result('reject', 'NO_POLICY', gateway_id, active, 'none', 'none', 'no_policy_match', 'none', 'none', 'none', 'none')
infra/server-a/s28/opensips.cfg
+58 -8
View File
@@ -18,11 +18,16 @@ loadmodule "signaling.so"
loadmodule "sl.so"
loadmodule "tm.so"
modparam("tm", "fr_timeout", 5)
modparam("tm", "fr_inv_timeout", 30)
modparam("tm", "fr_inv_timeout", 95)
loadmodule "rr.so"
modparam("rr", "append_fromtag", 1)
loadmodule "dialog.so"
modparam("dialog", "dlg_match_mode", 1)
modparam("dialog", "default_timeout", 7200)
modparam("dialog", "delete_delay", 10)
loadmodule "maxfwd.so"
loadmodule "sipmsgops.so"
loadmodule "textops.so"
@@ -167,10 +172,21 @@ route[S28_INVITE] {
$var(s28_line_group_id) = "none";
$var(s28_policy_id) = "none";
$var(s28_customer_id) = "none";
$var(s28_vendor_id) = "none";
$var(s28_vendor_gateway_id) = "none";
$var(s28_vendor_host) = "none";
$var(s28_vendor_port) = "5060";
$var(s28_callee_city_code) = "UNKNOWN";
$var(s28_callee_city_name) = "UNKNOWN";
$var(s28_callee_province_name) = "UNKNOWN";
$var(s28_callee_operator) = "UNKNOWN";
$var(s28_callee_number_type) = "UNKNOWN";
$var(s28_callee_province_code) = "UNKNOWN";
$var(s28_callee_normalized) = "";
$var(s28_reply_code) = 503;
$var(s28_reply_text) = "Routing Not Ready";
if (!cache_raw_query("redis:s20", "EVALSHA cfdc02cbe5528d37fba617947c09e3380770c918 1 cfg:active_version $si $fU $rU $ci", "$avp(s28_hotpath)")) {
if (!cache_raw_query("redis:s20", "EVALSHA 229c06326b97d3d64b0b02deedec77a764a61045 1 cfg:active_version $si $fU $rU $ci", "$avp(s28_hotpath)")) {
xlog("L_ERR", "S28 Redis hotpath unavailable source=$si callid=$ci\n");
$var(s28_reason) = "REDIS_UNAVAILABLE";
$var(s28_reply_code) = 503;
@@ -187,13 +203,24 @@ route[S28_INVITE] {
$var(s28_line_group_id) = $(avp(s28_hotpath)[4]);
$var(s28_policy_id) = $(avp(s28_hotpath)[5]);
$var(s28_customer_id) = $(avp(s28_hotpath)[6]);
$var(s28_vendor_id) = $(avp(s28_hotpath)[7]);
$var(s28_vendor_gateway_id) = $(avp(s28_hotpath)[8]);
$var(s28_vendor_host) = $(avp(s28_hotpath)[9]);
$var(s28_vendor_port) = $(avp(s28_hotpath)[10]);
$var(s28_callee_city_code) = $(avp(s28_hotpath)[11]);
$var(s28_callee_city_name) = $(avp(s28_hotpath)[12]);
$var(s28_callee_province_name) = $(avp(s28_hotpath)[13]);
$var(s28_callee_operator) = $(avp(s28_hotpath)[14]);
$var(s28_callee_number_type) = $(avp(s28_hotpath)[15]);
$var(s28_callee_province_code) = $(avp(s28_hotpath)[16]);
$var(s28_callee_normalized) = $(avp(s28_hotpath)[17]);
if ($var(s28_decision) != "allow") {
update_stat("s28_hotpath_reject_total", 1);
if ($var(s28_reason) == "CONFIG_MISSING") {
$var(s28_reply_code) = 503;
$var(s28_reply_text) = "Config Missing";
} else if ($var(s28_reason) == "NO_POLICY") {
} else if ($var(s28_reason) == "NO_POLICY" || $var(s28_reason) == "NO_VENDOR_ROUTE" || $var(s28_reason) == "NO_VENDOR_ROUTE_REGION_BLOCKED" || $var(s28_reason) == "LINE_GROUP_MISSING" || $var(s28_reason) == "LINE_GROUP_DISABLED") {
$var(s28_reply_code) = 503;
$var(s28_reply_text) = "No Route Policy";
} else {
@@ -206,7 +233,30 @@ route[S28_INVITE] {
}
update_stat("s28_hotpath_allow_total", 1);
xlog("L_INFO", "S28 hotpath allow source=$si callid=$ci customer=$var(s28_customer_id) gateway=$var(s28_gateway_id) policy=$var(s28_policy_id) line_group=$var(s28_line_group_id) version=$var(s28_config_version)\n");
xlog("L_INFO", "S28 hotpath allow source=$si callid=$ci customer=$var(s28_customer_id) gateway=$var(s28_gateway_id) policy=$var(s28_policy_id) line_group=$var(s28_line_group_id) vendor=$var(s28_vendor_id) vendor_gateway=$var(s28_vendor_gateway_id) dst=$var(s28_vendor_host):$var(s28_vendor_port) version=$var(s28_config_version)\n");
if (!create_dialog()) {
xlog("L_ERR", "S28 create_dialog failed callid=$ci\n");
$var(s28_reason) = "DIALOG_CREATE_FAILED";
$var(s28_reply_code) = 500;
$var(s28_reply_text) = "Dialog Error";
route(S28_CDR_FAILURE);
send_reply($var(s28_reply_code), $var(s28_reply_text));
exit;
}
$dlg_val(s28_customer_id) = $var(s28_customer_id);
$dlg_val(s28_gateway_id) = $var(s28_gateway_id);
$dlg_val(s28_policy_id) = $var(s28_policy_id);
$dlg_val(s28_line_group_id) = $var(s28_line_group_id);
$dlg_val(s28_vendor_id) = $var(s28_vendor_id);
$dlg_val(s28_vendor_gateway_id) = $var(s28_vendor_gateway_id);
$dlg_val(s28_config_version) = $var(s28_config_version);
$dlg_val(s28_callee) = $rU;
$dlg_val(s28_callee_city_code) = $var(s28_callee_city_code);
$dlg_val(s28_callee_city_name) = $var(s28_callee_city_name);
$dlg_val(s28_callee_province_name) = $var(s28_callee_province_name);
$dlg_val(s28_callee_operator) = $var(s28_callee_operator);
$dlg_val(s28_callee_number_type) = $var(s28_callee_number_type);
if (has_body("application/sdp")) {
if (!rtpengine_offer("replace-origin replace-session-connection record-call=on")) {
@@ -221,8 +271,8 @@ route[S28_INVITE] {
}
record_route();
$du = "sip:100.93.185.30:50620";
$ru = "sip:" + $rU + "@100.93.185.30:50620";
$du = "sip:" + $var(s28_vendor_host) + ":" + $var(s28_vendor_port);
$ru = "sip:" + $rU + "@" + $var(s28_vendor_host) + ":" + $var(s28_vendor_port);
t_on_reply("S28_REPLY");
if (!t_relay()) {
@@ -247,7 +297,7 @@ onreply_route[S28_REPLY] {
route[S28_CDR_SUCCESS] {
$var(s28_event_id) = "s28-ok-" + $Ts + "-" + $pp + "-" + $ci;
$var(s28_idempotency_key) = $ci + ":s28-ok:" + $Ts;
if (cache_raw_query("redis:s20", "XADD stream:cdr_payload * schema_version 1 event_id $var(s28_event_id) idempotency_key $var(s28_idempotency_key) call_id $ci node_id a1 opensips_instance opensips-a1 ingress_a_ip 100.90.90.90 rtpengine_node a1 source_ip $si caller $fU callee $rU customer_id cus_s28_t customer_gateway_id cgw_s28_t_ip customer_gateway_policy_id cgp_s28_t_default vendor_id ven_s28_t vendor_gateway_id vgw_s28_t_uas line_group_id llg_s28_t started_at $Ts answered_at $Ts ended_at $Ts duration_sec 6 sip_code 200 hangup_reason NORMAL_CLEARING recording_key none config_version $var(s28_config_version) created_at $Ts", "$avp(s28_cdr_id)")) {
if (cache_raw_query("redis:s20", "XADD stream:cdr_payload * schema_version 1 event_id $var(s28_event_id) idempotency_key $var(s28_idempotency_key) call_id $ci node_id a1 opensips_instance opensips-a1 ingress_a_ip 100.90.90.90 rtpengine_node a1 source_ip $si caller $fU callee $dlg_val(s28_callee) callee_city_code $dlg_val(s28_callee_city_code) callee_city_name $dlg_val(s28_callee_city_name) callee_province_name $dlg_val(s28_callee_province_name) callee_operator $dlg_val(s28_callee_operator) callee_number_type $dlg_val(s28_callee_number_type) customer_id $dlg_val(s28_customer_id) customer_gateway_id $dlg_val(s28_gateway_id) customer_gateway_policy_id $dlg_val(s28_policy_id) vendor_id $dlg_val(s28_vendor_id) vendor_gateway_id $dlg_val(s28_vendor_gateway_id) line_group_id $dlg_val(s28_line_group_id) started_at $Ts answered_at $Ts ended_at $Ts duration_sec 6 sip_code 200 hangup_reason NORMAL_CLEARING recording_key none config_version $dlg_val(s28_config_version) created_at $Ts", "$avp(s28_cdr_id)")) {
update_stat("s28_cdr_xadd_total", 1);
xlog("L_INFO", "S28 success CDR XADD ok redis_id=$avp(s28_cdr_id) event_id=$var(s28_event_id) callid=$ci\n");
} else {
@@ -259,7 +309,7 @@ route[S28_CDR_SUCCESS] {
route[S28_CDR_FAILURE] {
$var(s28_event_id) = "s28-fail-" + $Ts + "-" + $pp + "-" + $ci;
$var(s28_idempotency_key) = $ci + ":s28-fail:" + $Ts;
if (cache_raw_query("redis:s20", "XADD stream:cdr_payload * schema_version 1 event_id $var(s28_event_id) idempotency_key $var(s28_idempotency_key) call_id $ci node_id a1 opensips_instance opensips-a1 ingress_a_ip 100.90.90.90 rtpengine_node a1 source_ip $si caller $fU callee $rU customer_id $var(s28_customer_id) customer_gateway_id $var(s28_gateway_id) customer_gateway_policy_id $var(s28_policy_id) vendor_id none vendor_gateway_id none line_group_id $var(s28_line_group_id) started_at $Ts answered_at none ended_at $Ts duration_sec 0 sip_code $var(s28_reply_code) hangup_reason $var(s28_reason) recording_key none config_version $var(s28_config_version) created_at $Ts", "$avp(s28_cdr_id)")) {
if (cache_raw_query("redis:s20", "XADD stream:cdr_payload * schema_version 1 event_id $var(s28_event_id) idempotency_key $var(s28_idempotency_key) call_id $ci node_id a1 opensips_instance opensips-a1 ingress_a_ip 100.90.90.90 rtpengine_node a1 source_ip $si caller $fU callee $rU callee_city_code $var(s28_callee_city_code) callee_city_name $var(s28_callee_city_name) callee_province_name $var(s28_callee_province_name) callee_operator $var(s28_callee_operator) callee_number_type $var(s28_callee_number_type) customer_id $var(s28_customer_id) customer_gateway_id $var(s28_gateway_id) customer_gateway_policy_id $var(s28_policy_id) vendor_id none vendor_gateway_id none line_group_id $var(s28_line_group_id) started_at $Ts answered_at none ended_at $Ts duration_sec 0 sip_code $var(s28_reply_code) hangup_reason $var(s28_reason) recording_key none config_version $var(s28_config_version) created_at $Ts", "$avp(s28_cdr_id)")) {
update_stat("s28_cdr_xadd_total", 1);
} else {
update_stat("s28_cdr_xadd_error_total", 1);
infra/server-a/s32/lisglosips-call-control
+55
View File
@@ -0,0 +1,55 @@
#!/bin/sh
set -eu
MI_URL=${LISGLOSIPS_MI_URL:-http://127.0.0.1:8888/mi}
ORIGINAL=${SSH_ORIGINAL_COMMAND:-}
if [ "$#" -eq 0 ] && [ -n "$ORIGINAL" ]; then
set -- $ORIGINAL
fi
while [ "$#" -gt 0 ]; do
token=${1:-}
token=${token#\"}
token=${token%\"}
token=${token#\'}
token=${token%\'}
case "$token" in
/usr/local/sbin/lisglosips-call-control|lisglosips-call-control|'')
shift
;;
dlg_list|dlg_end_dlg)
set -- "$token" "${2:-}"
break
;;
*)
shift
;;
esac
done
method=${1:-}
case "$method" in
dlg_list)
payload='{"jsonrpc":"2.0","method":"dlg_list","params":[],"id":1}'
;;
dlg_end_dlg)
dialog_id=${2:-}
dialog_id=${dialog_id#\"}
dialog_id=${dialog_id%\"}
dialog_id=${dialog_id#\'}
dialog_id=${dialog_id%\'}
if ! printf '%s' "$dialog_id" | grep -Eq '^[A-Za-z0-9@._:%+=-]{1,220}$'; then
printf '{"jsonrpc":"2.0","error":{"code":-32602,"message":"invalid dialog id"},"id":1}\n'
exit 0
fi
payload='{"jsonrpc":"2.0","method":"dlg_end_dlg","params":["'"$dialog_id"'"],"id":1}'
;;
*)
printf '{"jsonrpc":"2.0","error":{"code":-32601,"message":"method not allowed"},"id":1}\n'
exit 0
;;
esac
exec curl -fsS -X POST "$MI_URL" -H 'Content-Type: application/json' --data-binary "$payload"
infra/server-b/s05/nginx/lisglosips-security-headers.conf
+1 -2
View File
@@ -3,5 +3,4 @@ add_header X-Content-Type-Options "nosniff" always;
add_header X-Frame-Options "DENY" always;
add_header Referrer-Policy "same-origin" always;
add_header Permissions-Policy "camera=(), microphone=(), geolocation=()" always;
add_header Content-Security-Policy "default-src 'self'; base-uri 'self'; frame-ancestors 'none'; form-action 'self'; object-src 'none'" always;
add_header Content-Security-Policy "default-src 'self'; img-src 'self' data:; base-uri 'self'; frame-ancestors 'none'; form-action 'self'; object-src 'none'" always;
infra/server-t/s40/lisglosips-s40-virtual-calls.py
+388
View File
@@ -0,0 +1,388 @@
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import random
import re
import socket
import string
import struct
import sys
import threading
import time
from dataclasses import dataclass
CRLF = "\r\n"
STOP = threading.Event()
PRINT_LOCK = threading.Lock()
def log(message: str) -> None:
with PRINT_LOCK:
print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {message}", flush=True)
def token(length: int = 8) -> str:
alphabet = string.ascii_lowercase + string.digits
return "".join(random.choice(alphabet) for _ in range(length))
def header(message: str, name: str) -> str:
match = re.search(rf"^{re.escape(name)}\s*:\s*(.+)$", message, re.I | re.M)
return match.group(1).strip() if match else ""
def status(message: str) -> str:
return message.splitlines()[0] if message else "NO RESPONSE"
def call_id(index: int) -> str:
return f"s40-{index:02d}-{int(time.time() * 1000)}-{token()}@lisglosips-t"
def sdp(host: str, port: int) -> str:
return (
"v=0\r\n"
f"o=lisglosips-s40 0 0 IN IP4 {host}\r\n"
"s=lisglosips-s40\r\n"
f"c=IN IP4 {host}\r\n"
"t=0 0\r\n"
f"m=audio {port} RTP/AVP 0 8 101\r\n"
"a=rtpmap:0 PCMU/8000\r\n"
"a=rtpmap:8 PCMA/8000\r\n"
"a=rtpmap:101 telephone-event/8000\r\n"
)
def parse_sdp_media(message: str) -> tuple[str, int] | None:
body = message.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in message else ""
media_host = ""
media_port = 0
for raw_line in body.splitlines():
line = raw_line.strip()
if line.startswith("c=IN IP4 "):
media_host = line.split()[-1]
elif line.startswith("m=audio "):
parts = line.split()
if len(parts) >= 2 and parts[1].isdigit():
media_port = int(parts[1])
if media_host and media_port:
return media_host, media_port
return None
def send_pcmu_rtp(target: tuple[str, int], duration: float, bind_host: str = "0.0.0.0") -> None:
payload = b"\xff" * 160
seq = random.randint(0, 65535)
timestamp = random.randint(0, 2**32 - 1)
ssrc = random.randint(1, 2**32 - 1)
deadline = time.time() + duration
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.bind((bind_host, 0))
while time.time() < deadline and not STOP.is_set():
rtp_header = struct.pack("!BBHII", 0x80, 0x00, seq, timestamp, ssrc)
sock.sendto(rtp_header + payload, target)
seq = (seq + 1) % 65536
timestamp = (timestamp + 160) % 2**32
time.sleep(0.02)
def response(code: int, reason: str, request: str, body: str = "", to_tag: str | None = None) -> str:
to_header = header(request, "To")
if "tag=" not in to_header:
to_header = f"{to_header};tag={to_tag or 's40uas' + token(6)}"
lines = [
f"SIP/2.0 {code} {reason}",
f"Via: {header(request, 'Via')}",
f"From: {header(request, 'From')}",
f"To: {to_header}",
f"Call-ID: {header(request, 'Call-ID')}",
f"CSeq: {header(request, 'CSeq')}",
"Server: lisglosips-s40-uas",
]
if body:
lines.extend(["Content-Type: application/sdp", f"Content-Length: {len(body.encode())}", "", body])
else:
lines.extend(["Content-Length: 0", "", ""])
return CRLF.join(lines)
def extract_callee(request: str) -> str:
first = request.splitlines()[0] if request else ""
match = re.search(r"sip:([^@;>\s]+)", first)
return match.group(1) if match else ""
def scenario_index(callee: str, callee_base: str) -> int:
if callee.startswith(callee_base):
suffix = callee[len(callee_base):]
if suffix.isdigit():
return int(suffix)
digits = re.sub(r"\D", "", callee)
return int(digits[-2:]) if digits else 0
def call_index(call: str) -> int:
match = re.search(r"s40-(\d{2})-", call)
return int(match.group(1)) if match else 0
@dataclass(frozen=True)
class UasConfig:
host: str
port: int
media_host: str
media_port_base: int
answer_count: int
answer_ring: float
no_answer_ring: float
callee_base: str
caller_prefix: str
def extract_user_from_header(value: str) -> str:
match = re.search(r"sip:([^@;>\s]+)", value)
return match.group(1) if match else ""
def handle_invite(sock: socket.socket, request: str, addr: tuple[str, int], config: UasConfig) -> None:
call = header(request, "Call-ID")
callee = extract_callee(request)
caller = extract_user_from_header(header(request, "From"))
index = call_index(call) or scenario_index(callee, config.callee_base) or scenario_index(caller, config.caller_prefix)
should_answer = 1 <= index <= config.answer_count
ring_seconds = config.answer_ring if should_answer else config.no_answer_ring
to_tag = f"s40uas{index:02d}{token(4)}"
sock.sendto(response(100, "Trying", request, to_tag=to_tag).encode(), addr)
sock.sendto(response(180, "Ringing", request, to_tag=to_tag).encode(), addr)
log(f"uas call_id={call} caller={caller} callee={callee} scenario={'answer' if should_answer else 'no-answer'} ringing={ring_seconds}s")
time.sleep(ring_seconds)
if STOP.is_set():
return
if not should_answer:
sock.sendto(response(480, "Temporarily Unavailable", request, to_tag=to_tag).encode(), addr)
log(f"uas call_id={call} final=480 Temporarily Unavailable")
return
media_port = config.media_port_base + max(index, 1)
body = sdp(config.media_host, media_port)
sock.sendto(response(200, "OK", request, body, to_tag=to_tag).encode(), addr)
media = parse_sdp_media(request)
if media:
threading.Thread(target=send_pcmu_rtp, args=(media, 720.0, config.host), daemon=True).start()
log(f"uas call_id={call} final=200 OK media_port={media_port}")
def run_uas(config: UasConfig) -> None:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((config.host, config.port))
sock.settimeout(1.0)
log(f"uas listening {config.host}:{config.port}")
while not STOP.is_set():
try:
data, addr = sock.recvfrom(65535)
except socket.timeout:
continue
request = data.decode(errors="replace")
method = request.split(" ", 1)[0]
if method == "INVITE":
threading.Thread(target=handle_invite, args=(sock, request, addr, config), daemon=True).start()
elif method == "BYE":
sock.sendto(response(200, "OK", request).encode(), addr)
log(f"uas call_id={header(request, 'Call-ID')} bye=200 OK")
@dataclass(frozen=True)
class InviteConfig:
index: int
host: str
port: int
bind_host: str
contact_host: str
caller: str
callee: str
to_domain: str
media_host: str
media_port: int
timeout: float
hold: float
def build_in_dialog(method: str, config: InviteConfig, cid: str, from_tag: str, to_header: str, cseq: int, route: str, local_port: int) -> str:
lines = [
f"{method} sip:{config.callee}@{config.to_domain} SIP/2.0",
f"Via: SIP/2.0/UDP {config.contact_host}:{local_port};branch=z9hG4bK-{token(10)};rport",
"Max-Forwards: 70",
f"From: <sip:{config.caller}@s40.customer.local>;tag={from_tag}",
f"To: {to_header}",
f"Call-ID: {cid}",
f"CSeq: {cseq} {method}",
f"Contact: <sip:{config.caller}@{config.contact_host}:{local_port}>",
]
if route:
lines.append(f"Route: {route}")
lines.extend(["Content-Length: 0", "", ""])
return CRLF.join(lines)
def run_invite(config: InviteConfig) -> None:
cid = call_id(config.index)
from_tag = token()
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
sock.bind((config.bind_host, 0))
local_port = sock.getsockname()[1]
sock.settimeout(config.timeout)
body = sdp(config.media_host, config.media_port)
request = CRLF.join(
[
f"INVITE sip:{config.callee}@{config.to_domain} SIP/2.0",
f"Via: SIP/2.0/UDP {config.contact_host}:{local_port};branch=z9hG4bK-{token(10)};rport",
"Max-Forwards: 70",
f"From: <sip:{config.caller}@s40.customer.local>;tag={from_tag}",
f"To: <sip:{config.callee}@{config.to_domain}>",
f"Call-ID: {cid}",
"CSeq: 1 INVITE",
f"Contact: <sip:{config.caller}@{config.contact_host}:{local_port}>",
"User-Agent: lisglosips-s40",
"Content-Type: application/sdp",
f"Content-Length: {len(body.encode())}",
"",
body,
]
)
log(f"uac index={config.index:02d} call_id={cid} invite callee={config.callee}")
sock.sendto(request.encode(), (config.host, config.port))
final = ""
route = ""
to_header = ""
while not STOP.is_set():
try:
message = sock.recvfrom(65535)[0].decode(errors="replace")
except socket.timeout:
break
line = status(message)
log(f"uac index={config.index:02d} call_id={cid} status={line}")
if message.startswith("SIP/2.0 2"):
final = message
route = header(message, "Record-Route") or f"<sip:{config.host}:{config.port};lr>"
to_header = header(message, "To")
break
if message.startswith(("SIP/2.0 3", "SIP/2.0 4", "SIP/2.0 5", "SIP/2.0 6")):
final = message
break
if not final.startswith("SIP/2.0 2"):
log(f"uac index={config.index:02d} call_id={cid} final={status(final)}")
return
ack = build_in_dialog("ACK", config, cid, from_tag, to_header, 1, route, local_port)
sock.sendto(ack.encode(), (config.host, config.port))
media = parse_sdp_media(final)
rtp_thread = None
if media:
rtp_thread = threading.Thread(target=send_pcmu_rtp, args=(media, config.hold, config.contact_host), daemon=True)
rtp_thread.start()
log(f"uac index={config.index:02d} call_id={cid} connected hold={config.hold}s")
deadline = time.time() + config.hold
while time.time() < deadline and not STOP.is_set():
time.sleep(min(1.0, deadline - time.time()))
if rtp_thread:
rtp_thread.join(timeout=1.0)
bye = build_in_dialog("BYE", config, cid, from_tag, to_header, 2, route, local_port)
sock.sendto(bye.encode(), (config.host, config.port))
try:
bye_response = sock.recvfrom(65535)[0].decode(errors="replace")
except socket.timeout:
bye_response = ""
log(f"uac index={config.index:02d} call_id={cid} bye_status={status(bye_response)}")
def run(args: argparse.Namespace) -> int:
uas_config = UasConfig(
host=args.uas_host,
port=args.uas_port,
media_host=args.media_host,
media_port_base=args.uas_media_port_base,
answer_count=args.answer_count,
answer_ring=args.answer_ring,
no_answer_ring=args.no_answer_ring,
callee_base=args.callee_base,
caller_prefix=args.caller_prefix,
)
uas_thread = threading.Thread(target=run_uas, args=(uas_config,), daemon=True)
uas_thread.start()
time.sleep(1.0)
workers: list[threading.Thread] = []
for index in range(1, args.total + 1):
callee = args.callee or f"{args.callee_base}{index:02d}"
config = InviteConfig(
index=index,
host=args.host,
port=args.port,
bind_host=args.bind_host,
contact_host=args.contact_host,
caller=args.caller or f"{args.caller_prefix}{index:02d}",
callee=callee,
to_domain=args.to_domain,
media_host=args.media_host,
media_port=args.uac_media_port_base + index,
timeout=args.invite_timeout,
hold=args.hold,
)
worker = threading.Thread(target=run_invite, args=(config,), daemon=True)
worker.start()
workers.append(worker)
if index < args.total:
time.sleep(args.interval)
for worker in workers:
worker.join()
STOP.set()
log("scenario finished")
return 0
def main() -> int:
parser = argparse.ArgumentParser(description="LisgloSIPS S40 10-way virtual call scenario")
parser.add_argument("--host", default="100.90.90.90", help="OpenSIPS A host")
parser.add_argument("--port", type=int, default=15060, help="OpenSIPS A SIP port")
parser.add_argument("--bind-host", default="0.0.0.0")
parser.add_argument("--contact-host", default="100.93.185.30")
parser.add_argument("--to-domain", default="100.90.90.90:15060")
parser.add_argument("--media-host", default="100.93.185.30")
parser.add_argument("--uas-host", default="100.93.185.30")
parser.add_argument("--uas-port", type=int, default=50620)
parser.add_argument("--total", type=int, default=10)
parser.add_argument("--interval", type=float, default=15.0)
parser.add_argument("--answer-count", type=int, default=5)
parser.add_argument("--answer-ring", type=float, default=45.0)
parser.add_argument("--no-answer-ring", type=float, default=70.0)
parser.add_argument("--hold", type=float, default=600.0)
parser.add_argument("--invite-timeout", type=float, default=90.0)
parser.add_argument("--callee-base", default="13800140")
parser.add_argument("--callee", default="13800136036", help="Fixed called number. Empty string enables callee-base + index.")
parser.add_argument("--caller-prefix", default="s40-100")
parser.add_argument("--caller", default="s36-1001", help="Fixed caller. Empty string enables caller-prefix + index.")
parser.add_argument("--uac-media-port-base", type=int, default=33100)
parser.add_argument("--uas-media-port-base", type=int, default=34100)
args = parser.parse_args()
if args.total < 1:
parser.error("--total must be >= 1")
if args.answer_count > args.total:
parser.error("--answer-count must be <= --total")
try:
return run(args)
except KeyboardInterrupt:
STOP.set()
log("interrupted")
return 130
if __name__ == "__main__":
sys.exit(main())