226 lines
8.3 KiB
Python
Executable File
226 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import random
|
|
import re
|
|
import socket
|
|
import struct
|
|
import string
|
|
import sys
|
|
import threading
|
|
import time
|
|
|
|
CRLF = "\r\n"
|
|
|
|
|
|
def token(n: int = 8) -> str:
|
|
return "".join(random.choice(string.ascii_lowercase + string.digits) for _ in range(n))
|
|
|
|
|
|
def header(msg: str, name: str) -> str:
|
|
m = re.search(rf"^{re.escape(name)}\s*:\s*(.+)$", msg, re.I | re.M)
|
|
return m.group(1).strip() if m else ""
|
|
|
|
|
|
def call_id() -> str:
|
|
return f"s28-{int(time.time() * 1000)}-{token()}@lisglosips-t"
|
|
|
|
|
|
def status(msg: str) -> str:
|
|
return msg.splitlines()[0] if msg else "NO RESPONSE"
|
|
|
|
|
|
def sdp(host: str, port: int) -> str:
|
|
return (
|
|
"v=0\r\n"
|
|
f"o=lisglosips-s28 0 0 IN IP4 {host}\r\n"
|
|
"s=lisglosips-s28\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(msg: str) -> tuple[str, int] | None:
|
|
body = msg.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in msg else msg.split("\n\n", 1)[1] if "\n\n" in msg else ""
|
|
media_host = ""
|
|
media_port = 0
|
|
for raw in body.splitlines():
|
|
line = raw.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:
|
|
header = struct.pack("!BBHII", 0x80, 0x00, seq, timestamp, ssrc)
|
|
sock.sendto(header + payload, target)
|
|
seq = (seq + 1) % 65536
|
|
timestamp = (timestamp + 160) % 2**32
|
|
time.sleep(0.02)
|
|
|
|
|
|
def response(code: int, reason: str, req: str, body: str = "") -> str:
|
|
to_h = header(req, "To")
|
|
if "tag=" not in to_h:
|
|
to_h = f"{to_h};tag=s28uas{token(6)}"
|
|
lines = [
|
|
f"SIP/2.0 {code} {reason}",
|
|
f"Via: {header(req, 'Via')}",
|
|
f"From: {header(req, 'From')}",
|
|
f"To: {to_h}",
|
|
f"Call-ID: {header(req, 'Call-ID')}",
|
|
f"CSeq: {header(req, 'CSeq')}",
|
|
"Server: lisglosips-s28-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 uas(args: argparse.Namespace) -> int:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind((args.host, args.port))
|
|
while True:
|
|
data, addr = sock.recvfrom(65535)
|
|
req = data.decode(errors="replace")
|
|
method = req.split(" ", 1)[0]
|
|
if method == "INVITE":
|
|
sock.sendto(response(100, "Trying", req).encode(), addr)
|
|
body = sdp(args.media_host, args.media_port)
|
|
sock.sendto(response(200, "OK", req, body).encode(), addr)
|
|
media = parse_sdp_media(req)
|
|
if media:
|
|
threading.Thread(target=send_pcmu_rtp, args=(media, args.rtp_duration, args.host), daemon=True).start()
|
|
elif method == "BYE":
|
|
sock.sendto(response(200, "OK", req).encode(), addr)
|
|
|
|
|
|
def invite(args: argparse.Namespace) -> int:
|
|
cid = call_id()
|
|
from_tag = token()
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind((args.bind_host, args.local_port))
|
|
args.local_port = sock.getsockname()[1]
|
|
sock.settimeout(args.timeout)
|
|
body = sdp(args.media_host, args.media_port)
|
|
req = CRLF.join(
|
|
[
|
|
f"INVITE sip:{args.callee}@{args.to_domain} SIP/2.0",
|
|
f"Via: SIP/2.0/UDP {args.contact_host}:{args.local_port};branch=z9hG4bK-{token(10)};rport",
|
|
"Max-Forwards: 70",
|
|
f"From: <sip:{args.caller}@s28.customer.local>;tag={from_tag}",
|
|
f"To: <sip:{args.callee}@{args.to_domain}>",
|
|
f"Call-ID: {cid}",
|
|
"CSeq: 1 INVITE",
|
|
f"Contact: <sip:{args.caller}@{args.contact_host}:{args.local_port}>",
|
|
"User-Agent: lisglosips-s28",
|
|
"Content-Type: application/sdp",
|
|
f"Content-Length: {len(body.encode())}",
|
|
"",
|
|
body,
|
|
]
|
|
)
|
|
sock.sendto(req.encode(), (args.host, args.port))
|
|
final = ""
|
|
route = ""
|
|
to_h = ""
|
|
while True:
|
|
try:
|
|
msg = sock.recvfrom(65535)[0].decode(errors="replace")
|
|
except socket.timeout:
|
|
break
|
|
print(f"call_id={cid} status={status(msg)}")
|
|
if msg.startswith("SIP/2.0 2"):
|
|
final = msg
|
|
route = header(msg, "Record-Route") or f"<sip:{args.host}:{args.port};lr>"
|
|
to_h = header(msg, "To")
|
|
break
|
|
if msg.startswith("SIP/2.0 3") or msg.startswith("SIP/2.0 4") or msg.startswith("SIP/2.0 5") or msg.startswith("SIP/2.0 6"):
|
|
final = msg
|
|
break
|
|
if not final.startswith("SIP/2.0 2"):
|
|
print(f"call_id={cid} final={status(final)}")
|
|
return 1
|
|
ack = build_in_dialog("ACK", args, cid, from_tag, to_h, 1, route)
|
|
sock.sendto(ack.encode(), (args.host, args.port))
|
|
media = parse_sdp_media(final)
|
|
if media:
|
|
send_pcmu_rtp(media, args.hold, args.contact_host)
|
|
time.sleep(args.hold)
|
|
bye = build_in_dialog("BYE", args, cid, from_tag, to_h, 2, route)
|
|
sock.sendto(bye.encode(), (args.host, args.port))
|
|
try:
|
|
bye_resp = sock.recvfrom(65535)[0].decode(errors="replace")
|
|
except socket.timeout:
|
|
bye_resp = ""
|
|
print(f"call_id={cid} bye_status={status(bye_resp)}")
|
|
return 0 if bye_resp.startswith("SIP/2.0 200") else 1
|
|
|
|
|
|
def build_in_dialog(method: str, args: argparse.Namespace, cid: str, from_tag: str, to_h: str, cseq: int, route: str) -> str:
|
|
lines = [
|
|
f"{method} sip:{args.callee}@{args.to_domain} SIP/2.0",
|
|
f"Via: SIP/2.0/UDP {args.contact_host}:{args.local_port};branch=z9hG4bK-{token(10)};rport",
|
|
"Max-Forwards: 70",
|
|
f"From: <sip:{args.caller}@s28.customer.local>;tag={from_tag}",
|
|
f"To: {to_h}",
|
|
f"Call-ID: {cid}",
|
|
f"CSeq: {cseq} {method}",
|
|
f"Contact: <sip:{args.caller}@{args.contact_host}:{args.local_port}>",
|
|
]
|
|
if route:
|
|
lines.append(f"Route: {route}")
|
|
lines.extend(["Content-Length: 0", "", ""])
|
|
return CRLF.join(lines)
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser()
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
u = sub.add_parser("uas")
|
|
u.add_argument("--host", default="100.93.185.30")
|
|
u.add_argument("--port", type=int, default=50620)
|
|
u.add_argument("--media-host", default="100.93.185.30")
|
|
u.add_argument("--media-port", type=int, default=32000)
|
|
u.add_argument("--rtp-duration", type=float, default=4.0)
|
|
i = sub.add_parser("invite")
|
|
i.add_argument("--host", default="100.90.90.90")
|
|
i.add_argument("--port", type=int, default=15060)
|
|
i.add_argument("--bind-host", default="0.0.0.0")
|
|
i.add_argument("--contact-host", default="100.93.185.30")
|
|
i.add_argument("--local-port", type=int, default=0)
|
|
i.add_argument("--caller", default="s28-ip-1001")
|
|
i.add_argument("--callee", default="13800138000")
|
|
i.add_argument("--to-domain", default="100.90.90.90:15060")
|
|
i.add_argument("--media-host", default="100.93.185.30")
|
|
i.add_argument("--media-port", type=int, default=31002)
|
|
i.add_argument("--timeout", type=float, default=5.0)
|
|
i.add_argument("--hold", type=float, default=3.0)
|
|
args = p.parse_args()
|
|
return uas(args) if args.cmd == "uas" else invite(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|