lisglosips/infra/server-t/s21/scripts/lisglosips-s21-sip.py

#!/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())