lct-hack/scripts/smoke_sip.py

#!/usr/bin/env python3
"""Local SIP/RTP acceptance check for the bundled Asterisk service.

The check registers extension 6001, calls the Echo application at 7000 and
measures round-trip time for PCMU RTP packets.  It intentionally uses only the
Python standard library so the offline delivery does not need a SIP test tool.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import random
import re
import socket
import statistics
import struct
import subprocess
import time
import uuid
from pathlib import Path


CRLF = "\r\n"


def md5(value: str) -> str:
    return hashlib.md5(value.encode("utf-8")).hexdigest()  # noqa: S324 - SIP Digest requires MD5


def parse_message(data: bytes) -> tuple[str, dict[str, str], str]:
    text = data.decode("utf-8", "replace")
    head, _, body = text.partition(CRLF + CRLF)
    lines = head.split(CRLF)
    headers: dict[str, str] = {}
    for line in lines[1:]:
        if ":" in line:
            key, value = line.split(":", 1)
            headers[key.strip().lower()] = value.strip()
    return lines[0], headers, body


def digest_header(challenge: str, user: str, password: str, method: str, uri: str) -> str:
    parsed = {key: quoted or plain for key, quoted, plain in re.findall(
        r'(\w+)=(?:"([^"]*)"|([^,\s]+))', challenge
    )}
    realm = parsed["realm"]
    nonce = parsed["nonce"]
    response_parts = [md5(f"{user}:{realm}:{password}"), nonce]
    attrs = [
        f'username="{user}"', f'realm="{realm}"', f'nonce="{nonce}"',
        f'uri="{uri}"',
    ]
    qop = parsed.get("qop", "").split(",")[0]
    if qop:
        nc = "00000001"
        cnonce = uuid.uuid4().hex[:16]
        response_parts.extend([nc, cnonce, qop])
        attrs.extend([f"qop={qop}", f"nc={nc}", f'cnonce="{cnonce}"'])
    response_parts.append(md5(f"{method}:{uri}"))
    attrs.extend([f'response="{md5(":".join(response_parts))}"', "algorithm=MD5"])
    return "Digest " + ", ".join(attrs)


def request(
    method: str,
    uri: str,
    host: str,
    local_port: int,
    user: str,
    call_id: str,
    cseq: int,
    branch: str,
    *,
    authorization: str | None = None,
    body: str = "",
    to_user: str | None = None,
    from_tag: str | None = None,
    to_header: str | None = None,
) -> bytes:
    target = to_user or user
    from_tag = from_tag or uuid.uuid4().hex[:10]
    headers = [
        f"{method} {uri} SIP/2.0",
        f"Via: SIP/2.0/UDP 127.0.0.1:{local_port};branch={branch};rport",
        "Max-Forwards: 70",
        f'From: <sip:{user}@{host}>;tag={from_tag}',
        f"To: {to_header or f'<sip:{target}@{host}>'}",
        f"Call-ID: {call_id}",
        f"CSeq: {cseq} {method}",
        f"Contact: <sip:{user}@127.0.0.1:{local_port};transport=udp>",
        "User-Agent: LCT-offline-smoke/1.0",
    ]
    if authorization:
        headers.append(f"Authorization: {authorization}")
    if body:
        headers.append("Content-Type: application/sdp")
    headers.extend([f"Content-Length: {len(body.encode())}", "", body])
    return CRLF.join(headers).encode()


def receive_final(sock: socket.socket, timeout: float = 3.0) -> tuple[str, dict[str, str], str]:
    deadline = time.monotonic() + timeout
    last = None
    while time.monotonic() < deadline:
        sock.settimeout(max(0.05, deadline - time.monotonic()))
        parsed = parse_message(sock.recvfrom(65535)[0])
        last = parsed
        status = parsed[0]
        if status.startswith("SIP/2.0") and not status.startswith("SIP/2.0 1"):
            return parsed
    raise TimeoutError(f"No final SIP response; last={last and last[0]}")


def compose_password(user: str) -> str:
    result = subprocess.run(
        [
            "docker", "compose", "exec", "-T", "sip", "cat", "/var/lib/lct-sip/credentials.env",
        ],
        check=True,
        capture_output=True,
        text=True,
    )
    prefix = f"SIP_{user}_PASSWORD="
    for line in result.stdout.splitlines():
        if line.startswith(prefix):
            return line[len(prefix):]
    raise RuntimeError(f"No password for SIP extension {user}")


def send_ack(
    sock: socket.socket, host: str, port: int, user: str, call_id: str, cseq: int,
    target: str, from_tag: str, to_header: str | None = None,
) -> None:
    payload = request(
        "ACK", f"sip:{target}@{host}", host, sock.getsockname()[1], user, call_id, cseq,
        f"z9hG4bK{uuid.uuid4().hex}", to_user=target, from_tag=from_tag,
        to_header=to_header,
    )
    sock.sendto(payload, (host, port))


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--host", default="127.0.0.1")
    parser.add_argument("--port", type=int, default=5060)
    parser.add_argument("--rtp-host-offset", type=int, default=0,
                        help="Host-to-container RTP port offset (1000 for 11000:10000 mappings)")
    parser.add_argument("--user", default="6001")
    parser.add_argument("--password", default=os.getenv("SIP_PASSWORD"))
    parser.add_argument("--packets", type=int, default=30)
    parser.add_argument("--output", type=Path)
    args = parser.parse_args()
    password = args.password or compose_password(args.user)

    sip = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sip.bind(("127.0.0.1", 0))
    sip_port = sip.getsockname()[1]
    registrar = f"sip:{args.host}"

    register_id = f"{uuid.uuid4().hex}@lct-smoke"
    packet = request("REGISTER", registrar, args.host, sip_port, args.user, register_id, 1,
                     f"z9hG4bK{uuid.uuid4().hex}")
    sip.sendto(packet, (args.host, args.port))
    status, headers, _ = receive_final(sip)
    if "401" not in status:
        raise RuntimeError(f"Expected REGISTER challenge, got {status}")
    challenge = headers.get("www-authenticate")
    if not challenge:
        raise RuntimeError("REGISTER challenge has no WWW-Authenticate header")
    auth = digest_header(challenge, args.user, password, "REGISTER", registrar)
    packet = request("REGISTER", registrar, args.host, sip_port, args.user, register_id, 2,
                     f"z9hG4bK{uuid.uuid4().hex}", authorization=auth)
    sip.sendto(packet, (args.host, args.port))
    status, _, _ = receive_final(sip)
    if "200" not in status:
        raise RuntimeError(f"REGISTER failed: {status}")

    rtp = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    rtp.bind(("0.0.0.0", 0))
    rtp.settimeout(1.0)
    rtp_port = rtp.getsockname()[1]
    sdp = CRLF.join([
        "v=0", f"o=lct 1 1 IN IP4 host.docker.internal", "s=LCT RTP smoke",
        "c=IN IP4 host.docker.internal", "t=0 0", f"m=audio {rtp_port} RTP/AVP 0",
        "a=rtpmap:0 PCMU/8000", "a=sendrecv", "",
    ])
    call_id = f"{uuid.uuid4().hex}@lct-smoke"
    from_tag = uuid.uuid4().hex[:10]
    invite_uri = f"sip:7000@{args.host}"
    invite_cseq = 1
    packet = request("INVITE", invite_uri, args.host, sip_port, args.user, call_id, invite_cseq,
                     f"z9hG4bK{uuid.uuid4().hex}", body=sdp, to_user="7000",
                     from_tag=from_tag)
    sip.sendto(packet, (args.host, args.port))
    status, headers, _ = receive_final(sip)
    if "401" in status or "407" in status:
        send_ack(sip, args.host, args.port, args.user, call_id, invite_cseq, "7000",
                 from_tag, headers.get("to"))
        challenge = headers.get("www-authenticate") or headers.get("proxy-authenticate")
        if not challenge:
            raise RuntimeError("INVITE challenge has no authentication header")
        invite_cseq += 1
        auth = digest_header(challenge, args.user, password, "INVITE", invite_uri)
        packet = request("INVITE", invite_uri, args.host, sip_port, args.user, call_id, invite_cseq,
                         f"z9hG4bK{uuid.uuid4().hex}", authorization=auth, body=sdp,
                         to_user="7000", from_tag=from_tag)
        sip.sendto(packet, (args.host, args.port))
        status, answer_headers, answer_sdp = receive_final(sip)
    else:
        answer_headers = headers
        answer_sdp = ""
    if "200" not in status:
        raise RuntimeError(f"INVITE failed: {status}")
    send_ack(sip, args.host, args.port, args.user, call_id, invite_cseq, "7000",
             from_tag, answer_headers.get("to"))

    match = re.search(r"^m=audio\s+(\d+)", answer_sdp, re.MULTILINE)
    if not match:
        raise RuntimeError("SIP 200 response contains no RTP audio port")
    asterisk_rtp_port = int(match.group(1))
    target = (args.host, asterisk_rtp_port + args.rtp_host_offset)
    ssrc = random.getrandbits(32)
    sequence = random.getrandbits(16)
    timestamp = random.getrandbits(32)
    samples: list[float] = []

    # Prime symmetric RTP, then measure matched echoed sequence numbers.
    for index in range(args.packets + 3):
        seq = (sequence + index) & 0xFFFF
        packet = struct.pack("!BBHII", 0x80, 0x00, seq, (timestamp + index * 160) & 0xFFFFFFFF, ssrc)
        marker = struct.pack("!I", index)
        packet += marker + b"\xff" * 156
        started = time.perf_counter()
        rtp.sendto(packet, target)
        deadline = started + 0.5
        while time.perf_counter() < deadline:
            try:
                response, _ = rtp.recvfrom(2048)
            except socket.timeout:
                break
            if len(response) >= 16 and response[12:16] == marker:
                if index >= 3:
                    samples.append((time.perf_counter() - started) * 1000)
                break
            rtp.settimeout(max(0.001, deadline - time.perf_counter()))
        rtp.settimeout(0.5)
        time.sleep(0.02)

    if not samples:
        raise RuntimeError(f"No echoed RTP received from {target}")
    ordered = sorted(samples)
    p95 = ordered[min(len(ordered) - 1, int(len(ordered) * 0.95))]
    result = {
        "checked_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
        "sip_registration": "ok",
        "sip_echo_call": "ok",
        "rtp_packets_received": len(samples),
        "rtp_rtt_ms_median": round(statistics.median(samples), 3),
        "rtp_rtt_ms_p95": round(p95, 3),
        "requirement_ms": 150,
        "passed": p95 <= 150,
    }

    bye = request(
        "BYE", invite_uri, args.host, sip_port, args.user, call_id, invite_cseq + 1,
        f"z9hG4bK{uuid.uuid4().hex}", to_user="7000", from_tag=from_tag,
        to_header=answer_headers.get("to"),
    )
    sip.sendto(bye, (args.host, args.port))
    bye_status, _, _ = receive_final(sip)
    result["sip_hangup"] = "ok" if "200" in bye_status else bye_status
    result["passed"] = result["passed"] and result["sip_hangup"] == "ok"
    rendered = json.dumps(result, ensure_ascii=False, indent=2)
    print(rendered)
    if args.output:
        args.output.parent.mkdir(parents=True, exist_ok=True)
        args.output.write_text(rendered + "\n", encoding="utf-8")
    return 0 if result["passed"] else 1


if __name__ == "__main__":
    raise SystemExit(main())