389 lines
14 KiB
Python
389 lines
14 KiB
Python
#!/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())
|