337 lines
12 KiB
Python
Executable File
337 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""LisgloSIPS S21 SIP scenario runner.
|
|
|
|
Self-contained UDP SIP helper for Server T customer simulation.
|
|
It intentionally accepts HA1 values instead of plaintext SIP passwords.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import random
|
|
import re
|
|
import socket
|
|
import string
|
|
import sys
|
|
import threading
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
|
|
CRLF = "\r\n"
|
|
|
|
|
|
def md5_hex(value: str) -> str:
|
|
return hashlib.md5(value.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def rand_token(length: int = 10) -> str:
|
|
alphabet = string.ascii_lowercase + string.digits
|
|
return "".join(random.choice(alphabet) for _ in range(length))
|
|
|
|
|
|
def call_id(prefix: str = "s21") -> str:
|
|
return f"{prefix}-{int(time.time() * 1000)}-{rand_token(8)}@lisglosips-t"
|
|
|
|
|
|
def parse_authenticate(header: str) -> dict[str, str]:
|
|
header = header.strip()
|
|
if header.lower().startswith("digest "):
|
|
header = header[7:]
|
|
values: dict[str, str] = {}
|
|
for part in re.split(r",\s*", header):
|
|
if "=" not in part:
|
|
continue
|
|
key, raw = part.split("=", 1)
|
|
values[key.strip().lower()] = raw.strip().strip('"')
|
|
return values
|
|
|
|
|
|
def header_value(message: str, name: str) -> str | None:
|
|
pattern = re.compile(rf"^{re.escape(name)}\s*:\s*(.+)$", re.IGNORECASE | re.MULTILINE)
|
|
match = pattern.search(message)
|
|
return match.group(1).strip() if match else None
|
|
|
|
|
|
def status_line(message: str) -> str:
|
|
return message.splitlines()[0] if message else "NO RESPONSE"
|
|
|
|
|
|
@dataclass
|
|
class SipTarget:
|
|
host: str
|
|
port: int
|
|
|
|
|
|
def udp_roundtrip(payload: str, target: SipTarget, timeout: float = 3.0, bind_host: str = "0.0.0.0") -> str:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind((bind_host, 0))
|
|
sock.settimeout(timeout)
|
|
sock.sendto(payload.encode("utf-8"), (target.host, target.port))
|
|
try:
|
|
data, _ = sock.recvfrom(65535)
|
|
except socket.timeout:
|
|
return ""
|
|
return data.decode("utf-8", errors="replace")
|
|
|
|
|
|
def build_register(
|
|
username: str,
|
|
domain: str,
|
|
contact_host: str,
|
|
local_port: int,
|
|
cseq: int,
|
|
cid: str,
|
|
branch: str,
|
|
authorization: str | None = None,
|
|
) -> str:
|
|
lines = [
|
|
f"REGISTER sip:{domain} SIP/2.0",
|
|
f"Via: SIP/2.0/UDP {contact_host}:{local_port};branch=z9hG4bK-{branch};rport",
|
|
"Max-Forwards: 70",
|
|
f"From: <sip:{username}@{domain}>;tag={rand_token(8)}",
|
|
f"To: <sip:{username}@{domain}>",
|
|
f"Call-ID: {cid}",
|
|
f"CSeq: {cseq} REGISTER",
|
|
f"Contact: <sip:{username}@{contact_host}:{local_port}>;expires=120",
|
|
"Expires: 120",
|
|
"User-Agent: lisglosips-s21",
|
|
]
|
|
if authorization:
|
|
lines.append(f"Authorization: {authorization}")
|
|
lines.extend(["Content-Length: 0", "", ""])
|
|
return CRLF.join(lines)
|
|
|
|
|
|
def digest_authorization(username: str, realm: str, uri: str, method: str, ha1: str, challenge: dict[str, str]) -> str:
|
|
nonce = challenge["nonce"]
|
|
qop = challenge.get("qop", "").split(",")[0].strip()
|
|
algorithm = challenge.get("algorithm", "MD5")
|
|
ha2 = md5_hex(f"{method}:{uri}")
|
|
if qop:
|
|
nc = "00000001"
|
|
cnonce = rand_token(16)
|
|
response = md5_hex(f"{ha1}:{nonce}:{nc}:{cnonce}:{qop}:{ha2}")
|
|
return (
|
|
f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{uri}", '
|
|
f'response="{response}", algorithm={algorithm}, qop={qop}, nc={nc}, cnonce="{cnonce}"'
|
|
)
|
|
response = md5_hex(f"{ha1}:{nonce}:{ha2}")
|
|
return (
|
|
f'Digest username="{username}", realm="{realm}", nonce="{nonce}", uri="{uri}", '
|
|
f'response="{response}", algorithm={algorithm}'
|
|
)
|
|
|
|
|
|
def cmd_register(args: argparse.Namespace) -> int:
|
|
cid = call_id("s21-register")
|
|
target = SipTarget(args.host, args.port)
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind((args.bind_host, 0))
|
|
local_host, local_port = sock.getsockname()
|
|
if args.contact_host:
|
|
local_host = args.contact_host
|
|
sock.settimeout(args.timeout)
|
|
first = build_register(args.username, args.domain, local_host, local_port, 1, cid, rand_token())
|
|
sock.sendto(first.encode("utf-8"), (target.host, target.port))
|
|
try:
|
|
data, _ = sock.recvfrom(65535)
|
|
except socket.timeout:
|
|
print(f"call_id={cid} status=NO_RESPONSE")
|
|
return 2
|
|
response = data.decode("utf-8", errors="replace")
|
|
print(f"call_id={cid} first_status={status_line(response)}")
|
|
www = header_value(response, "WWW-Authenticate") or header_value(response, "Proxy-Authenticate")
|
|
if not www:
|
|
return 0 if response.startswith("SIP/2.0 200") else 1
|
|
challenge = parse_authenticate(www)
|
|
realm = challenge.get("realm", args.domain)
|
|
uri = f"sip:{args.domain}"
|
|
auth = digest_authorization(args.username, realm, uri, "REGISTER", args.ha1, challenge)
|
|
second = build_register(args.username, args.domain, local_host, local_port, 2, cid, rand_token(), auth)
|
|
sock.sendto(second.encode("utf-8"), (target.host, target.port))
|
|
try:
|
|
data, _ = sock.recvfrom(65535)
|
|
except socket.timeout:
|
|
print(f"call_id={cid} final_status=NO_RESPONSE")
|
|
return 2
|
|
final = data.decode("utf-8", errors="replace")
|
|
print(f"call_id={cid} final_status={status_line(final)}")
|
|
return 0 if final.startswith("SIP/2.0 200") else 1
|
|
|
|
|
|
def build_invite(args: argparse.Namespace, cid: str, cseq: int = 1) -> str:
|
|
branch = rand_token(10)
|
|
from_tag = rand_token(8)
|
|
from_uri = f"sip:{args.caller}@{args.from_domain}"
|
|
to_uri = f"sip:{args.callee}@{args.to_domain}"
|
|
request_uri = f"sip:{args.callee}@{args.to_domain}"
|
|
sdp = (
|
|
"v=0\r\n"
|
|
f"o=lisglosips-s21 0 0 IN IP4 {args.media_host}\r\n"
|
|
"s=lisglosips-s21\r\n"
|
|
f"c=IN IP4 {args.media_host}\r\n"
|
|
"t=0 0\r\n"
|
|
f"m=audio {args.media_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"
|
|
)
|
|
lines = [
|
|
f"INVITE {request_uri} SIP/2.0",
|
|
f"Via: SIP/2.0/UDP {args.contact_host}:{args.local_port};branch=z9hG4bK-{branch};rport",
|
|
"Max-Forwards: 70",
|
|
f"From: <{from_uri}>;tag={from_tag}",
|
|
f"To: <{to_uri}>",
|
|
f"Call-ID: {cid}",
|
|
f"CSeq: {cseq} INVITE",
|
|
f"Contact: <sip:{args.caller}@{args.contact_host}:{args.local_port}>",
|
|
"User-Agent: lisglosips-s21",
|
|
"Content-Type: application/sdp",
|
|
f"Content-Length: {len(sdp.encode('utf-8'))}",
|
|
"",
|
|
sdp,
|
|
]
|
|
return CRLF.join(lines)
|
|
|
|
|
|
def cmd_invite(args: argparse.Namespace) -> int:
|
|
cid = call_id("s21-invite")
|
|
target = SipTarget(args.host, args.port)
|
|
payload = build_invite(args, cid)
|
|
response = udp_roundtrip(payload, target, args.timeout, args.bind_host)
|
|
print(f"call_id={cid} status={status_line(response)}")
|
|
return 0 if response else 2
|
|
|
|
|
|
def uas_once(host: str, port: int, mode: str, ready: threading.Event) -> None:
|
|
code_map = {
|
|
"success": ("200", "OK"),
|
|
"busy": ("486", "Busy Here"),
|
|
"reject": ("603", "Decline"),
|
|
}
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
|
|
sock.bind((host, port))
|
|
sock.settimeout(5)
|
|
ready.set()
|
|
try:
|
|
data, addr = sock.recvfrom(65535)
|
|
except socket.timeout:
|
|
return
|
|
if mode == "timeout":
|
|
time.sleep(3)
|
|
return
|
|
request = data.decode("utf-8", errors="replace")
|
|
via = header_value(request, "Via") or ""
|
|
from_h = header_value(request, "From") or ""
|
|
to_h = header_value(request, "To") or ""
|
|
cid = header_value(request, "Call-ID") or call_id("s21-uas")
|
|
cseq = header_value(request, "CSeq") or "1 INVITE"
|
|
code, reason = code_map[mode]
|
|
if "tag=" not in to_h:
|
|
to_h = f"{to_h};tag={rand_token(8)}"
|
|
response = CRLF.join(
|
|
[
|
|
f"SIP/2.0 {code} {reason}",
|
|
f"Via: {via}",
|
|
f"From: {from_h}",
|
|
f"To: {to_h}",
|
|
f"Call-ID: {cid}",
|
|
f"CSeq: {cseq}",
|
|
"Server: lisglosips-s21-uas",
|
|
"Content-Length: 0",
|
|
"",
|
|
"",
|
|
]
|
|
)
|
|
sock.sendto(response.encode("utf-8"), addr)
|
|
|
|
|
|
def cmd_local_scenario(args: argparse.Namespace) -> int:
|
|
ready = threading.Event()
|
|
thread = threading.Thread(target=uas_once, args=(args.listen_host, args.listen_port, args.mode, ready), daemon=True)
|
|
thread.start()
|
|
ready.wait(2)
|
|
args.host = args.listen_host
|
|
args.port = args.listen_port
|
|
args.to_domain = f"{args.listen_host}:{args.listen_port}"
|
|
rc = cmd_invite(args)
|
|
thread.join(4)
|
|
if args.mode == "timeout":
|
|
return 0 if rc == 2 else 1
|
|
return rc
|
|
|
|
|
|
def cmd_over_cps(args: argparse.Namespace) -> int:
|
|
target = SipTarget(args.host, args.port)
|
|
statuses: list[str] = []
|
|
for index in range(args.count):
|
|
cid = call_id(f"s21-cps-{index + 1}")
|
|
args.callee = f"{args.callee_prefix}{index + 1:03d}"
|
|
payload = build_invite(args, cid)
|
|
response = udp_roundtrip(payload, target, args.timeout, args.bind_host)
|
|
line = status_line(response)
|
|
statuses.append(line)
|
|
print(f"call_id={cid} status={line}")
|
|
if args.interval > 0:
|
|
time.sleep(args.interval)
|
|
print(f"sent={args.count} unique_statuses={sorted(set(statuses))}")
|
|
return 0
|
|
|
|
|
|
def add_common_invite_args(parser: argparse.ArgumentParser) -> None:
|
|
parser.add_argument("--host", default="100.90.90.90")
|
|
parser.add_argument("--port", type=int, default=15060)
|
|
parser.add_argument("--bind-host", default="0.0.0.0")
|
|
parser.add_argument("--contact-host", default="100.93.185.30")
|
|
parser.add_argument("--local-port", type=int, default=50620)
|
|
parser.add_argument("--caller", default="s21-ip-1001")
|
|
parser.add_argument("--callee", default="13800138000")
|
|
parser.add_argument("--from-domain", default="s21-ip.lisglosips.test")
|
|
parser.add_argument("--to-domain", default="100.90.90.90:15060")
|
|
parser.add_argument("--media-host", default="100.93.185.30")
|
|
parser.add_argument("--media-port", type=int, default=31000)
|
|
parser.add_argument("--timeout", type=float, default=3.0)
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description="LisgloSIPS S21 SIP scenario runner")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
|
|
reg = sub.add_parser("register")
|
|
reg.add_argument("--host", default="100.93.185.30")
|
|
reg.add_argument("--port", type=int, default=5060)
|
|
reg.add_argument("--bind-host", default="0.0.0.0")
|
|
reg.add_argument("--contact-host", default="100.93.185.30")
|
|
reg.add_argument("--username", required=True)
|
|
reg.add_argument("--domain", required=True)
|
|
reg.add_argument("--ha1", required=True)
|
|
reg.add_argument("--timeout", type=float, default=3.0)
|
|
reg.set_defaults(func=cmd_register)
|
|
|
|
inv = sub.add_parser("invite")
|
|
add_common_invite_args(inv)
|
|
inv.set_defaults(func=cmd_invite)
|
|
|
|
local = sub.add_parser("local-scenario")
|
|
add_common_invite_args(local)
|
|
local.add_argument("--mode", choices=["success", "busy", "reject", "timeout"], required=True)
|
|
local.add_argument("--listen-host", default="127.0.0.1")
|
|
local.add_argument("--listen-port", type=int, default=15070)
|
|
local.set_defaults(func=cmd_local_scenario)
|
|
|
|
cps = sub.add_parser("over-cps")
|
|
add_common_invite_args(cps)
|
|
cps.add_argument("--count", type=int, default=20)
|
|
cps.add_argument("--interval", type=float, default=0.02)
|
|
cps.add_argument("--callee-prefix", default="1390000")
|
|
cps.set_defaults(func=cmd_over_cps)
|
|
|
|
args = parser.parse_args()
|
|
return args.func(args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|