From 849b1c05e9a61227d4fbca9a0b59abb606f7e17e Mon Sep 17 00:00:00 2001 From: Repin Agent Date: Sat, 18 Jul 2026 15:52:16 -0600 Subject: [PATCH 1/2] test(e2e): durable rustisk trunk e2e suite + M2 harness race fix Add tests/e2e-trunk/: a self-contained, CI-runnable end-to-end proof of the inbound-trunk PIN-gate path, turning the one-shot M9 arming harness into a committed, re-runnable suite. - Hermetic: synthetic Chime UAC (chime_caller.py) -> rustisk PinGate -> GRANTED -> Dial(PJSIP/qa-bridge) -> a MOCK registered SIP endpoint (mock_bridge.py: inbound REGISTER + MD5 digest + answer + distinct-tone media) -> two-way RTP, each direction proven by a distinct tone detected on the FAR side (caller 440 Hz, mock 660 Hz). Wrong PIN -> REJECTED -> the bridge is never INVITEd (invite_count==0). No FreeSWITCH/Mumble/docker/ NET_RAW/root -- only the rustisk binary + python3 stdlib (Goertzel, no numpy) + loopback. Runs on fresh 127.0.0.6x / 35xxx-36xxx ports, preflight-asserted free and non-overlapping with the live trunk / m0 / m9 rigs. - Wired into CI as the e2e-trunk job (build rustisk + run.sh on ubuntu-latest). - integration/: the original M9 full-bridge validators (chime_caller.py + qa_probe.py, numpy) + sanitized live config (qa-bridge password redacted). Requires the real Mumble + pymumble bridge voice stack; NOT run in CI -- it is the operator-run live-cutover validator. - Fix the M2 sniffer-startup race in tests/freeswitch-pin-gate: rtp_injector.py gains a --go-file priming gate (keep reading/classifying frames but don't start the match deadline until the caller confirms the flow-triggering ESL round trip returned), and run.sh touches the go-file only after fs_cli returns -- so docker-exec/ESL setup latency no longer eats the 5s window. Adds an m2-standalone case selector. --- .github/workflows/ci.yml | 17 + tests/e2e-trunk/README.md | 109 +++++ tests/e2e-trunk/chime_caller.py | 342 ++++++++++++++ tests/e2e-trunk/gen_prompts.py | 47 ++ tests/e2e-trunk/integration/README.md | 63 +++ tests/e2e-trunk/integration/chime_caller.py | 223 ++++++++++ .../integration/config-live/asterisk.conf | 4 + .../integration/config-live/extensions.conf | 39 ++ .../integration/config-live/manager.conf | 9 + .../integration/config-live/pin_gate.conf | 8 + .../integration/config-live/pjsip.conf | 50 +++ .../integration/config-live/rtp.conf | 3 + tests/e2e-trunk/integration/qa_probe.py | 67 +++ tests/e2e-trunk/mock_bridge.py | 419 ++++++++++++++++++ tests/e2e-trunk/run.sh | 362 +++++++++++++++ tests/freeswitch-pin-gate/rtp_injector.py | 152 +++++-- tests/freeswitch-pin-gate/run.sh | 21 +- 17 files changed, 1886 insertions(+), 49 deletions(-) create mode 100644 tests/e2e-trunk/README.md create mode 100755 tests/e2e-trunk/chime_caller.py create mode 100755 tests/e2e-trunk/gen_prompts.py create mode 100644 tests/e2e-trunk/integration/README.md create mode 100644 tests/e2e-trunk/integration/chime_caller.py create mode 100644 tests/e2e-trunk/integration/config-live/asterisk.conf create mode 100644 tests/e2e-trunk/integration/config-live/extensions.conf create mode 100644 tests/e2e-trunk/integration/config-live/manager.conf create mode 100644 tests/e2e-trunk/integration/config-live/pin_gate.conf create mode 100644 tests/e2e-trunk/integration/config-live/pjsip.conf create mode 100644 tests/e2e-trunk/integration/config-live/rtp.conf create mode 100644 tests/e2e-trunk/integration/qa_probe.py create mode 100755 tests/e2e-trunk/mock_bridge.py create mode 100755 tests/e2e-trunk/run.sh diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 475d01c..05bd676 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -71,3 +71,20 @@ jobs: run: scripts/fetch-pjproject.sh - name: Build shim with pjproject-cffi run: cargo build -p pjsip-shim --features pjproject-cffi + + e2e-trunk: + # Hermetic end-to-end proof of the inbound-trunk PIN-gate path + # (synthetic Chime INVITE -> PinGate -> GRANTED -> Dial(PJSIP/qa-bridge) + # -> a mock registered SIP endpoint -> two-way RTP; and wrong PIN -> + # REJECTED -> no Dial). It is fully self-contained: only the rustisk + # binary + python3 stdlib + loopback (no FreeSWITCH, Mumble, docker, + # NET_RAW or root), so it runs unattended here. See tests/e2e-trunk/README.md. + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: dtolnay/rust-toolchain@1.97.0 + - uses: Swatinem/rust-cache@v2 + - name: Build rustisk binary + run: cargo build --bin rustisk + - name: Run hermetic trunk e2e (PIN gate -> bridge -> two-way RTP + reject) + run: tests/e2e-trunk/run.sh diff --git a/tests/e2e-trunk/README.md b/tests/e2e-trunk/README.md new file mode 100644 index 0000000..6a80739 --- /dev/null +++ b/tests/e2e-trunk/README.md @@ -0,0 +1,109 @@ +# rustisk trunk e2e (`tests/e2e-trunk`) + +Durable, repeatable, **hermetic** end-to-end proof of rustisk's inbound-trunk +PIN-gate path — the same path the live voice trunk runs, but with **no external +services**. It is the committed, CI-runnable successor to the one-shot M9 arming +harness (which drove a real Amazon Chime UAC into the real qa-sip Mumble bridge). + +``` + synthetic Chime UAC ──INVITE(PCMU+RFC2833)──▶ rustisk [pin-gate] + (chime_caller.py) │ PinGate() collects DTMF PIN + │ + wrong PIN ─────────────────────────────▶ REJECTED ─▶ play + Hangup, NO Dial + │ + correct PIN ───────────────────────────▶ GRANTED ─▶ Dial(PJSIP/qa-bridge) + │ + MOCK registered UAS ◀─INVITE─┘ (mock_bridge.py) + answers 200 OK + SDP + ◀──────── two-way RTP ────────▶ + caller 440 Hz ─▶ mock RX (proves A → bridge) + mock 660 Hz ─▶ caller RX (proves bridge → A) +``` + +## What it asserts (receiver-side, like M0/M2) + +**CASE 1 — REJECTED (wrong PIN must not reach the bridge)** +- rustisk log contains `Verbose: PIN_GATE_RESULT=REJECTED`. +- the mock endpoint's `invite_count == 0` — rustisk **never** Dialed the bridge. + +**CASE 2 — GRANTED (correct PIN → Dial → two-way media)** +- rustisk log contains `Verbose: PIN_GATE_RESULT=GRANTED`. +- **A → bridge:** the mock received the caller's relayed tone — + `invite_count == 1`, `rx_voice ≥ 100` frames, and **440 Hz is the dominant + band** in the mock's received audio. +- **bridge → A:** the caller received the bridge's distinct tone — + `voice_rx ≥ 100` frames, and **660 Hz is the dominant band** in the caller's + received audio. + +Distinct tones (caller 440 Hz, mock 660 Hz), each detected on the *far* side, +make each RTP direction independently attributable by frequency — a stronger +proof than an echo (an echo cannot distinguish a relayed frame from a socket +loop). Tone detection is a stdlib Goertzel filter, so **CI needs no numpy**. + +## Isolation (no collision with live / m0 / m9) + +Everything binds fresh loopback IPs + high ports, asserted free by a preflight +(`ss`) before anything starts: + +| role | IP | SIP | RTP | AMI | +|------|-----|-----|-----|-----| +| rustisk (SUT) | `127.0.0.60` | `35060` | `36000-36100` | `35038` | +| synthetic Chime caller | `127.0.0.61` | `35062` | `36200` | — | +| mock qa-bridge | `127.0.0.62` | `35064` | `36300` | — | + +None of these touch the reserved live/arming ports (`:45070` live trunk, +`:25060/:25038/:25062/:21000-21100` M9 arming, `:15060/:15038` M0) — the +preflight fails fast if they ever would. Endpoints are identified by distinct +`/32`s so inbound identification is unambiguous. No `external_media_address` is +set, so rustisk advertises its own loopback bind IP for media (avoids the +external-address blackhole in the SDP path). + +## Run it + +```bash +# uses target/debug/rustisk if present, else `cargo build --bin rustisk` +tests/e2e-trunk/run.sh + +# or point at a prebuilt binary +RUSTISK_BIN=/path/to/rustisk tests/e2e-trunk/run.sh +``` + +Requirements: `python3` (stdlib only), `ss` (iproute2), `stdbuf` (coreutils), +Linux loopback. No docker, no NET_RAW, no root. Exit status **is** the result +(0 = pass). Artifacts (rustisk log, per-leg JSON results, RX WAVs) land in +`target/e2e-trunk/`. + +Knobs: `RUSTISK_BIN`, `MIN_VOICE_FRAMES` (default 100), `MIN_TONE_RATIO` +(default 20), `KEEP_RUNTIME` (keep artifacts). + +## Proven green — 2× consecutive (calibration runs, rustisk @ 461ad74) + +``` +RUN 1 ALL CASES PASSED + REJECTED : PIN_GATE_RESULT=REJECTED, bridge invite_count=0 + GRANTED : mock rx_voice=574 tone440=2004457.92 (660=175.59 350=362.93 880=98.49) + caller voice_rx=695 tone660=1766894.92 (440=216.80 350=154.90 880=155.86) + +RUN 2 ALL CASES PASSED + REJECTED : PIN_GATE_RESULT=REJECTED, bridge invite_count=0 + GRANTED : mock rx_voice=574 tone440=2004457.92 + caller voice_rx=696 tone660=1431051.95 (440=308.06) +``` + +The expected tone dominates the next-strongest band by ~4 orders of magnitude +in both directions on every run (frame counts and dominance are the load-bearing +assertions; the absolute Goertzel magnitude is an unbounded relative metric). + +## CI + +Wired into `.github/workflows/ci.yml` as the `e2e-trunk` job: it builds the +rustisk binary and runs `run.sh` on `ubuntu-latest`. It is hermetic (loopback + +python3 stdlib only), so it runs unattended in CI with no runner privileges. + +## `integration/` — full-stack M9 validator (NOT run in CI) + +`integration/` holds the *original* M9 full-bridge validators +(`chime_caller.py` + `qa_probe.py`, numpy) plus sanitized live config. Those +require the real voice stack (qa-sip Mumble channel + the dedicated pymumble +bridge) and are the end-to-end validator for the live cutover — **not** +hermetic, **not** in CI. See `integration/README.md`. diff --git a/tests/e2e-trunk/chime_caller.py b/tests/e2e-trunk/chime_caller.py new file mode 100755 index 0000000..bee2215 --- /dev/null +++ b/tests/e2e-trunk/chime_caller.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Hermetic synthetic Chime-dialect SIP UAC for the durable e2e-trunk suite. + +Emulates the Chime inbound leg (per REPORT-M0a): PCMU/PT0 + telephone-event/ +PT101, ptime 20 / 50 pps, no 100rel, no session timers. It: + * places a call into rustisk's PIN gate (INVITE, offer PCMU+RFC2833); + * enters a PIN via RFC2833 (in-band RTP) DTMF once answered; + * emits a continuous PCMU tone (default 440 Hz) so that, once GRANTED and + bridged, the tone reaches the far leg; + * captures the return RTP (audio bridged back FROM the far leg) and, over a + late measurement window (after the bridge is up), tone-detects the far + end's distinct tone (default 660 Hz) to prove the bridge->caller direction. + +Receiver-side proof only, exactly like the M0/M2 harnesses. Pure Python +stdlib — the FFT is replaced by a Goertzel filter so CI needs no numpy. +This is the hermetic sibling of the numpy M9 caller in integration/. +""" +import argparse +import json +import math +import os +import random +import socket +import struct +import sys +import threading +import time +import wave + + +def log(*a): + print("[caller]", *a, file=sys.stderr, flush=True) + + +# ---- mulaw + RTP (helpers copied from tests/freeswitch-pin-gate/rtp_injector.py +# lineage; identical math to the proven M9 caller) ---- +def linear_to_mulaw(sample): + bias = 0x84 + clip = 32635 + sign = 0x80 if sample < 0 else 0 + magnitude = min(abs(sample), clip) + bias + exponent = 7 + mask = 0x4000 + while exponent > 0 and not magnitude & mask: + exponent -= 1 + mask >>= 1 + mantissa = (magnitude >> (exponent + 3)) & 0x0F + return (~(sign | (exponent << 4) | mantissa)) & 0xFF + + +_MULAW_DECODE = None + + +def mulaw_to_linear(u): + global _MULAW_DECODE + if _MULAW_DECODE is None: + tab = [] + for i in range(256): + u2 = ~i & 0xFF + sign = u2 & 0x80 + exp = (u2 >> 4) & 0x07 + man = u2 & 0x0F + mag = ((man << 3) + 0x84) << exp + mag -= 0x84 + tab.append(-mag if sign else mag) + _MULAW_DECODE = tab + return _MULAW_DECODE[u & 0xFF] + + +def tone_pcmu(freq, packet_index, amp=10000): + out = bytearray() + for offset in range(160): + absolute = packet_index * 160 + offset + s = round(amp * math.sin(2 * math.pi * freq * absolute / 8000)) + out.append(linear_to_mulaw(s)) + return bytes(out) + + +def rtp_hdr(pt, seq, ts, ssrc, marker=0): + return struct.pack( + "!BBHII", 0x80, ((marker & 1) << 7) | (pt & 0x7F), seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc & 0xFFFFFFFF + ) + + +# ---- Goertzel tone detector (stdlib, replaces numpy FFT) ---- +def goertzel(samples, target_hz, rate=8000): + """Normalized single-bin power at target_hz. Returns a ratio scaled so the + numbers are comparable across runs (mag / total-energy * 1e4).""" + n = len(samples) + if n < 320: + return 0.0 + k = int(0.5 + (n * target_hz) / rate) + w = (2.0 * math.pi / n) * k + coeff = 2.0 * math.cos(w) + s_prev = 0.0 + s_prev2 = 0.0 + energy = 0.0 + for x in samples: + energy += x * x + s = x + coeff * s_prev - s_prev2 + s_prev2 = s_prev + s_prev = s + power = s_prev2 * s_prev2 + s_prev * s_prev - coeff * s_prev * s_prev2 + if energy <= 0: + return 0.0 + return round(math.sqrt(max(power, 0.0)) / math.sqrt(energy) * 1e4, 2) + + +def parse_sdp_pts(body): + """Return (pcmu_pt, tev_pt, remote_ip, remote_port) from an SDP answer.""" + pcmu = 0 + tev = 101 + rip = None + rport = None + for line in body.splitlines(): + line = line.strip() + if line.startswith("m=audio"): + rport = int(line.split()[1]) + elif line.startswith("c=IN IP4"): + rip = line.split()[-1] + elif line.lower().startswith("a=rtpmap:"): + pt, enc = line[len("a=rtpmap:"):].split(None, 1) + if enc.upper().startswith("PCMU/"): + pcmu = int(pt) + if enc.lower().startswith("telephone-event/"): + tev = int(pt) + return pcmu, tev, rip, rport + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--dst-ip", required=True) + ap.add_argument("--dst-port", type=int, required=True) + ap.add_argument("--src-ip", required=True) + ap.add_argument("--sip-port", type=int, default=35062) + ap.add_argument("--rtp-port", type=int, default=36200) + ap.add_argument("--exten", default="9000") + ap.add_argument("--pin", required=True) + ap.add_argument("--tone", type=int, default=440) # A -> far-end tone + ap.add_argument("--detect", type=int, default=660) # far-end -> A tone we expect back + ap.add_argument("--call-secs", type=float, default=14.0) + ap.add_argument("--dtmf-start", type=float, default=1.5) # after 200 OK + ap.add_argument("--measure-start", type=float, default=6.0) # accumulate RX after bridge is up + ap.add_argument("--rx-wav", required=True) + ap.add_argument("--result-file", required=True) + args = ap.parse_args() + + ssrc = random.randint(0, 0x7FFFFFFF) + callid = f"{random.randint(0, 1 << 48):012x}@{args.src_ip}" + fromtag = f"{random.randint(0, 1 << 32):08x}" + branch = f"z9hG4bK{random.randint(0, 1 << 32):08x}" + cseq = 1 + + sipsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sipsock.bind((args.src_ip, args.sip_port)) + sipsock.settimeout(5) + rtpsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rtpsock.bind((args.src_ip, args.rtp_port)) + rtpsock.settimeout(0.25) + + sdp = ( + f"v=0\r\no=- {random.randint(1, 1 << 30)} 1 IN IP4 {args.src_ip}\r\ns=chime\r\n" + f"c=IN IP4 {args.src_ip}\r\nt=0 0\r\nm=audio {args.rtp_port} RTP/AVP 0 101\r\n" + f"a=rtpmap:0 PCMU/8000\r\na=rtpmap:101 telephone-event/8000\r\n" + f"a=fmtp:101 0-15\r\na=ptime:20\r\na=sendrecv\r\n" + ) + ruri = f"sip:{args.exten}@{args.dst_ip}:{args.dst_port}" + invite = ( + f"INVITE {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch={branch};rport\r\n" + f"Max-Forwards: 70\r\n" + f"From: ;tag={fromtag}\r\n" + f"To: <{ruri}>\r\n" + f"Call-ID: {callid}\r\nCSeq: {cseq} INVITE\r\n" + f"Contact: \r\n" + f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}" + ) + sipsock.sendto(invite.encode(), (args.dst_ip, args.dst_port)) + log(f"INVITE -> {ruri}") + + totag = None + ok_body = None + deadline = time.time() + 8 + while time.time() < deadline: + try: + data, _ = sipsock.recvfrom(65535) + except socket.timeout: + sipsock.sendto(invite.encode(), (args.dst_ip, args.dst_port)) + continue + msg = data.decode("latin1") + first = msg.split("\r\n", 1)[0] + log("SIP <=", first) + if " 200 " in first: + for l in msg.split("\r\n"): + if l.lower().startswith("to:") and "tag=" in l.lower(): + totag = l.split("tag=", 1)[1].strip() + ok_body = msg.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in msg else "" + break + # 100/180/183 -> keep waiting + if ok_body is None: + log("no 200 OK; aborting") + _write_result(args, {"error": "no_200_ok", "voice_rx": 0}, exit_code=2) + return 2 + + pcmu_pt, tev_pt, rip, rport = parse_sdp_pts(ok_body) + if not rip or rip == "0.0.0.0": + rip = args.dst_ip + log(f"answered: PCMU_PT={pcmu_pt} TEV_PT={tev_pt} remote_media={rip}:{rport}") + ack = ( + f"ACK {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch=z9hG4bK{random.randint(0, 1 << 32):08x};rport\r\n" + f"Max-Forwards: 70\r\nFrom: ;tag={fromtag}\r\n" + f"To: <{ruri}>;tag={totag}\r\nCall-ID: {callid}\r\nCSeq: {cseq} ACK\r\nContent-Length: 0\r\n\r\n" + ) + sipsock.sendto(ack.encode(), (args.dst_ip, args.dst_port)) + + # ---- media threads ---- + stop = threading.Event() + rx_frames = {"voice": 0, "other": 0, "bytes": 0} + rx_pcm = [] # accumulated ONLY during the measurement window (post-bridge) + measure_open = threading.Event() + + def rx_loop(): + while not stop.is_set(): + try: + pkt, _ = rtpsock.recvfrom(4096) + except socket.timeout: + continue + if len(pkt) < 12: + continue + pt = pkt[1] & 0x7F + rx_frames["bytes"] += len(pkt) + payload = pkt[12:] + if pt == pcmu_pt and len(payload) > 0: + rx_frames["voice"] += 1 + if measure_open.is_set(): + rx_pcm.extend(mulaw_to_linear(b) for b in payload) + else: + rx_frames["other"] += 1 + + threading.Thread(target=rx_loop, daemon=True).start() + + seq = random.randint(0, 0x7FFF) + ts = random.randint(0, 0x7FFFFFFF) + t0 = time.time() + dtmf_sent = False + + def send_voice_packet(idx): + nonlocal seq, ts + pkt = rtp_hdr(pcmu_pt, seq, ts, ssrc) + tone_pcmu(args.tone, idx) + rtpsock.sendto(pkt, (rip, rport)) + seq = (seq + 1) & 0xFFFF + ts = (ts + 160) & 0xFFFFFFFF + + def send_dtmf(digit): + nonlocal seq, ts + evmap = {**{str(d): d for d in range(10)}, "*": 10, "#": 11, "A": 12, "B": 13, "C": 14, "D": 15} + ev = evmap[digit] + ev_ts = ts + dur = 0 + for i in range(8): # 8 event packets (~160 ms), marker on first + dur += 160 + payload = struct.pack("!BBH", ev, (0 << 7) | 10, dur) + pkt = rtp_hdr(tev_pt, seq, ev_ts, ssrc, marker=1 if i == 0 else 0) + payload + rtpsock.sendto(pkt, (rip, rport)) + seq = (seq + 1) & 0xFFFF + time.sleep(0.02) + for _ in range(3): # 3 end packets, E=1 + payload = struct.pack("!BBH", ev, (1 << 7) | 10, dur) + pkt = rtp_hdr(tev_pt, seq, ev_ts, ssrc) + payload + rtpsock.sendto(pkt, (rip, rport)) + seq = (seq + 1) & 0xFFFF + time.sleep(0.02) + ts = (ts + 160 * 11) & 0xFFFFFFFF + + idx = 0 + while time.time() - t0 < args.call_secs: + cyc = time.time() - t0 + if (not dtmf_sent) and cyc >= args.dtmf_start: + log(f"sending PIN via RFC2833 (TEV_PT={tev_pt})") + for d in args.pin: + send_dtmf(d) + time.sleep(0.06) # inter-digit gap + send_dtmf("#") # terminate entry + dtmf_sent = True + continue + if (not measure_open.is_set()) and cyc >= args.measure_start: + measure_open.set() + log(f"measurement window open at t+{cyc:.1f}s") + send_voice_packet(idx) + idx += 1 + time.sleep(0.02) + + stop.set() + time.sleep(0.3) + bye = ( + f"BYE {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch=z9hG4bK{random.randint(0, 1 << 32):08x};rport\r\n" + f"Max-Forwards: 70\r\nFrom: ;tag={fromtag}\r\n" + f"To: <{ruri}>;tag={totag}\r\nCall-ID: {callid}\r\nCSeq: {cseq + 1} BYE\r\nContent-Length: 0\r\n\r\n" + ) + sipsock.sendto(bye.encode(), (args.dst_ip, args.dst_port)) + + # write RX wav (8k mono s16) from the measurement window + with wave.open(args.rx_wav, "wb") as w: + w.setnchannels(1) + w.setsampwidth(2) + w.setframerate(8000) + w.writeframes(struct.pack(f"<{len(rx_pcm)}h", *(max(-32768, min(32767, int(s))) for s in rx_pcm))) + + det = {} + if len(rx_pcm) >= 1600: + mean = sum(rx_pcm) / len(rx_pcm) + centered = [s - mean for s in rx_pcm] + for f0 in (args.detect, args.tone, 350, 880): + det[f0] = goertzel(centered, f0) + + result = { + "voice_rx": rx_frames["voice"], + "other_rx": rx_frames["other"], + "rx_bytes": rx_frames["bytes"], + "rx_samples": len(rx_pcm), + "tone_ratios": det, + "detect_hz": args.detect, + } + _write_result(args, result, exit_code=0) + print( + f"RESULT voice_rx={rx_frames['voice']} other_rx={rx_frames['other']} " + f"rx_bytes={rx_frames['bytes']} rx_samples={len(rx_pcm)} tone_ratios={det}" + ) + return 0 + + +def _write_result(args, result, exit_code): + tmp = args.result_file + ".tmp" + with open(tmp, "w") as f: + json.dump(result, f) + os.replace(tmp, args.result_file) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/e2e-trunk/gen_prompts.py b/tests/e2e-trunk/gen_prompts.py new file mode 100755 index 0000000..abc0f0a --- /dev/null +++ b/tests/e2e-trunk/gen_prompts.py @@ -0,0 +1,47 @@ +#!/usr/bin/env python3 +"""Generate the deterministic 8 kHz mono signed-16 WAV prompt fixtures the +hermetic dialplan references (PinGate prompt + granted/rejected playback). + +Pure stdlib. Tones are chosen so NONE of them collide with the media-proof +tones used by the e2e (caller TX 440 Hz, mock TX 660 Hz are detected on the +FAR side only; the prompts play on the near/A-leg before the bridge exists, +so they can share 440 without contaminating the far-side assertions — but we +still keep granted/rejected distinct for eyeballing captures).""" +import math +import pathlib +import struct +import sys +import wave + +SAMPLE_RATE = 8_000 + + +def write_tone(path: pathlib.Path, frequency: int, duration: float) -> None: + sample_count = round(SAMPLE_RATE * duration) + with wave.open(str(path), "wb") as output: + output.setnchannels(1) + output.setsampwidth(2) + output.setframerate(SAMPLE_RATE) + samples = ( + struct.pack( + " None: + if len(sys.argv) != 2: + raise SystemExit("usage: gen_prompts.py OUTPUT_DIRECTORY") + out = pathlib.Path(sys.argv[1]) + out.mkdir(parents=True, exist_ok=True) + write_tone(out / "pin-prompt.wav", 350, 2.0) + write_tone(out / "granted.wav", 880, 0.8) + write_tone(out / "rejected.wav", 220, 0.8) + print(f"wrote prompt fixtures to {out}") + + +if __name__ == "__main__": + main() diff --git a/tests/e2e-trunk/integration/README.md b/tests/e2e-trunk/integration/README.md new file mode 100644 index 0000000..9f737c4 --- /dev/null +++ b/tests/e2e-trunk/integration/README.md @@ -0,0 +1,63 @@ +# Full-stack M9 trunk validator (`integration/`) — NOT run in CI + +These are the **original, proven** M9 arming-harness validators that drove the +full live media path end-to-end: + +``` +synthetic Chime UAC ─▶ rustisk PIN gate ─▶ Dial(PJSIP/qa-bridge) + (chime_caller.py) │ + the REAL pymumble qa-bridge ◀─┘ + │ (bridges into Mumble) + qa-sip Mumble channel + ▲ + qa_probe.py joins qa-sip as a deterministic + far end: TX 660 Hz, FFT-detect the caller's 440 Hz +``` + +This proved the whole path **3× off-trunk** during M9 arming. Unlike the +hermetic `tests/e2e-trunk/run.sh` (which replaces the bridge with an in-process +mock), this variant exercises the *actual* pymumble bridge and Mumble server, so +it is the validator for the live cutover. + +## Why it is NOT in CI + +It requires infrastructure that does not exist on a CI runner: + +- the **qa-sip Mumble channel** on the voice server (`voice.murphytek.com`), +- the **dedicated pymumble `qa-bridge`** registered to rustisk, +- `numpy` + `pymumble_py3` (the validators use them directly), +- the real `qa-bridge` auth password and the real 6-digit PIN (secrets — never + committed; see below). + +Running it also touches shared live voice infrastructure and must be sequenced +with the live trunk (see `RUNBOOK-M9`). It is a **manual, operator-run** +validator, not an unattended gate. + +## Files + +- `chime_caller.py` — synthetic Chime UAC (numpy). Places the call, enters the + PIN via RFC2833, emits a 440 Hz tone, captures + FFT-analyzes the return RTP. +- `qa_probe.py` — runs inside the `vs-harness` image; joins qa-sip as `m9-probe`, + transmits 660 Hz into the channel and FFT-detects the caller's 440 Hz. +- `config-live/` — the live rustisk config templates (pjsip / extensions / + pin_gate / rtp / manager / asterisk). **Sanitized:** the qa-bridge auth + password is redacted to `` and + `pin_gate.conf` points `secret_file` at a host-local path you stage the real + PIN into at runtime (see its inline comment). **Never commit the real + password or PIN.** + +## Run (operator, with the voice stack up) + +1. Stage the real PIN into the `secret_file` path referenced by + `config-live/pin_gate.conf` (e.g. from the `voice-sip` k8s secret) — chmod 600. +2. Inject the real qa-bridge auth password into `config-live/pjsip.conf` in + place of the placeholder. +3. Start rustisk against `config-live/asterisk.conf`, bring up the pymumble + `qa-bridge` (registers to rustisk) and `qa_probe.py` (in the vs-harness + image, joined to qa-sip). +4. Drive `chime_caller.py` at rustisk. Assert: caller RX shows 660 Hz (probe → + caller) and the probe shows 440 Hz (caller → probe) — two-way through the + real bridge. + +For the unattended, hermetic version of the same assertions, use +`../run.sh`. diff --git a/tests/e2e-trunk/integration/chime_caller.py b/tests/e2e-trunk/integration/chime_caller.py new file mode 100644 index 0000000..aaeb040 --- /dev/null +++ b/tests/e2e-trunk/integration/chime_caller.py @@ -0,0 +1,223 @@ +#!/usr/bin/env python3 +"""Synthetic Chime-dialect SIP UAC for the M9 dry-run. + +Emulates the Chime inbound leg (per REPORT-M0a): PCMU/PT0 + telephone-event/PT101, +ptime 20 / 50pps, no 100rel, no session timers. Places a call into rustisk's +PIN-gate, enters a PIN via RFC2833 (in-band RTP) DTMF, emits a continuous PCMU +tone (so once GRANTED+bridged the tone reaches qa-sip), and captures the return +RTP (the bridged audio coming back FROM qa-sip). Prints receiver-side counters +and writes the decoded RX audio to a WAV for tone analysis. + +Pure stdlib + numpy. No repo imports (helpers copied from rtp_injector.py). +""" +import argparse, math, os, random, socket, struct, sys, threading, time, wave +import numpy as np + +def log(*a): print("[caller]", *a, file=sys.stderr, flush=True) + +# ---- mulaw + RTP (copied verbatim from tests/.../rtp_injector.py) ---- +def linear_to_mulaw(sample): + bias=0x84; clip=32635 + sign=0x80 if sample<0 else 0 + magnitude=min(abs(sample),clip)+bias + exponent=7; mask=0x4000 + while exponent>0 and not magnitude&mask: + exponent-=1; mask>>=1 + mantissa=(magnitude>>(exponent+3))&0x0F + return (~(sign|(exponent<<4)|mantissa))&0xFF + +_MULAW_DECODE=None +def mulaw_to_linear(u): + global _MULAW_DECODE + if _MULAW_DECODE is None: + tab=[] + for i in range(256): + u2=~i&0xFF + sign=u2&0x80; exp=(u2>>4)&0x07; man=u2&0x0F + mag=((man<<3)+0x84)<qa-sip tone + ap.add_argument("--call-secs", type=float, default=12.0) + ap.add_argument("--dtmf-start", type=float, default=1.2) # after 200 OK + ap.add_argument("--rx-wav", required=True) + args=ap.parse_args() + + ssrc=random.randint(0,0x7fffffff) + callid=f"{random.randint(0,1<<48):012x}@{args.src_ip}" + fromtag=f"{random.randint(0,1<<32):08x}" + branch=f"z9hG4bK{random.randint(0,1<<32):08x}" + cseq=1 + + sipsock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM) + sipsock.bind((args.src_ip,args.sip_port)); sipsock.settimeout(5) + rtpsock=socket.socket(socket.AF_INET,socket.SOCK_DGRAM) + rtpsock.bind((args.src_ip,args.rtp_port)); rtpsock.settimeout(0.25) + + sdp=(f"v=0\r\no=- {random.randint(1,1<<30)} 1 IN IP4 {args.src_ip}\r\ns=chime\r\n" + f"c=IN IP4 {args.src_ip}\r\nt=0 0\r\nm=audio {args.rtp_port} RTP/AVP 0 101\r\n" + f"a=rtpmap:0 PCMU/8000\r\na=rtpmap:101 telephone-event/8000\r\n" + f"a=fmtp:101 0-15\r\na=ptime:20\r\na=sendrecv\r\n") + ruri=f"sip:{args.exten}@{args.dst_ip}:{args.dst_port}" + invite=(f"INVITE {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch={branch};rport\r\n" + f"Max-Forwards: 70\r\n" + f"From: ;tag={fromtag}\r\n" + f"To: <{ruri}>\r\n" + f"Call-ID: {callid}\r\nCSeq: {cseq} INVITE\r\n" + f"Contact: \r\n" + f"Content-Type: application/sdp\r\nContent-Length: {len(sdp)}\r\n\r\n{sdp}") + sipsock.sendto(invite.encode(),(args.dst_ip,args.dst_port)) + log(f"INVITE -> {ruri}") + + totag=None; ok_body=None; deadline=time.time()+8 + while time.time() keep waiting + if ok_body is None: + log("no 200 OK; aborting"); return 2 + + pcmu_pt,tev_pt,rip,rport=parse_sdp_pts(ok_body) + if not rip or rip=="0.0.0.0": rip=args.dst_ip + log(f"answered: PCMU_PT={pcmu_pt} TEV_PT={tev_pt} remote_media={rip}:{rport}") + # ACK + ack=(f"ACK {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch=z9hG4bK{random.randint(0,1<<32):08x};rport\r\n" + f"Max-Forwards: 70\r\nFrom: ;tag={fromtag}\r\n" + f"To: <{ruri}>;tag={totag}\r\nCall-ID: {callid}\r\nCSeq: {cseq} ACK\r\nContent-Length: 0\r\n\r\n") + sipsock.sendto(ack.encode(),(args.dst_ip,args.dst_port)) + + # ---- media threads ---- + stop=threading.Event() + rx_frames={"voice":0,"other":0,"bytes":0} + rx_pcm=[] + def rx_loop(): + while not stop.is_set(): + try: pkt,_=rtpsock.recvfrom(4096) + except socket.timeout: continue + if len(pkt)<12: continue + pt=pkt[1]&0x7f; rx_frames["bytes"]+=len(pkt) + payload=pkt[12:] + if pt==pcmu_pt and len(payload)>0: + rx_frames["voice"]+=1 + rx_pcm.extend(mulaw_to_linear(b) for b in payload) + else: + rx_frames["other"]+=1 + threading.Thread(target=rx_loop,daemon=True).start() + + seq=random.randint(0,0x7fff); ts=random.randint(0,0x7fffffff) + t0=time.time(); dtmf_sent=False + def send_voice_packet(idx): + nonlocal seq,ts + pkt=rtp_hdr(pcmu_pt,seq,ts,ssrc)+tone_pcmu(args.tone,idx) + rtpsock.sendto(pkt,(rip,rport)); seq=(seq+1)&0xffff; ts=(ts+160)&0xffffffff + def send_dtmf(digit): + nonlocal seq,ts + evmap={**{str(d):d for d in range(10)},"*":10,"#":11,"A":12,"B":13,"C":14,"D":15} + ev=evmap[digit]; ev_ts=ts; dur=0 + # 8 event packets (~160ms), marker on first + for i in range(8): + dur+=160 + payload=struct.pack("!BBH",ev,(0<<7)|10,dur) # E=0, vol=10 + pkt=rtp_hdr(tev_pt,seq,ev_ts,ssrc,marker=1 if i==0 else 0)+payload + rtpsock.sendto(pkt,(rip,rport)); seq=(seq+1)&0xffff; time.sleep(0.02) + # 3 end packets, E=1 + for _ in range(3): + payload=struct.pack("!BBH",ev,(1<<7)|10,dur) + pkt=rtp_hdr(tev_pt,seq,ev_ts,ssrc)+pkt[12:] if False else rtp_hdr(tev_pt,seq,ev_ts,ssrc)+payload + rtpsock.sendto(pkt,(rip,rport)); seq=(seq+1)&0xffff; time.sleep(0.02) + ts=(ts+160*11)&0xffffffff # advance past the event duration + + idx=0 + while time.time()-t0 < args.call_secs: + cyc=time.time()-t0 + if (not dtmf_sent) and cyc>=args.dtmf_start: + log(f"sending PIN via RFC2833 (TEV_PT={tev_pt})") + for d in args.pin: + send_dtmf(d); time.sleep(0.06) # inter-digit gap + # terminate entry + send_dtmf("#") + dtmf_sent=True + continue + send_voice_packet(idx); idx+=1 + time.sleep(0.02) + + stop.set(); time.sleep(0.3) + # BYE + bye=(f"BYE {ruri} SIP/2.0\r\n" + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch=z9hG4bK{random.randint(0,1<<32):08x};rport\r\n" + f"Max-Forwards: 70\r\nFrom: ;tag={fromtag}\r\n" + f"To: <{ruri}>;tag={totag}\r\nCall-ID: {callid}\r\nCSeq: {cseq+1} BYE\r\nContent-Length: 0\r\n\r\n") + sipsock.sendto(bye.encode(),(args.dst_ip,args.dst_port)) + + # write RX wav (8k mono s16) + pcm=np.array(rx_pcm,dtype=np.int16) if rx_pcm else np.zeros(0,dtype=np.int16) + with wave.open(args.rx_wav,"wb") as w: + w.setnchannels(1); w.setsampwidth(2); w.setframerate(8000); w.writeframes(pcm.tobytes()) + # tone detect on RX + det={} + if len(pcm)>=1600: + x=pcm.astype(np.float64); x-=x.mean() + N=len(x); f=np.fft.rfftfreq(N,1/8000); X=np.abs(np.fft.rfft(x*np.hanning(N))) + def band(f0): + m=(f>=f0-30)&(f<=f0+30); return float(X[m].max()) if m.any() else 0.0 + tot=float(X.sum())+1e-9 + for f0 in (440,660,350,480): + det[f0]=round(band(f0)/tot*1e4,2) + print(f"RESULT voice_rx={rx_frames['voice']} other_rx={rx_frames['other']} " + f"rx_bytes={rx_frames['bytes']} rx_samples={len(pcm)} tone_ratios={det}") + return 0 + +if __name__=="__main__": + sys.exit(main()) diff --git a/tests/e2e-trunk/integration/config-live/asterisk.conf b/tests/e2e-trunk/integration/config-live/asterisk.conf new file mode 100644 index 0000000..856f0cc --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/asterisk.conf @@ -0,0 +1,4 @@ +[directories] +astetcdir = /mnt/data/herodevs-agents/m9-arm/config-live +astrundir = /mnt/data/herodevs-agents/m9-arm/run +astincludedir = /mnt/data/herodevs-agents/m9-arm/run/include diff --git a/tests/e2e-trunk/integration/config-live/extensions.conf b/tests/e2e-trunk/integration/config-live/extensions.conf new file mode 100644 index 0000000..b716e82 --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/extensions.conf @@ -0,0 +1,39 @@ +; ============================================================================ +; rustisk M9 dialplan — extensions.conf +; Chime INVITE (any DID) -> [pin-gate] -> PinGate -> on GRANTED, +; Dial(PJSIP/qa-bridge) two-way-bridges the caller into qa-sip via the +; registered pymumble bridge. On REJECTED/TIMEOUT/ERROR -> play + hangup. +; +; PinGate(prompt, absolute-deadline-secs, inter-digit-timeout-secs) +; - the 6-digit PIN is NOT here; it is loaded from pin_gate.conf secret_file. +; - PINGATESTATUS = GRANTED | REJECTED | TIMEOUT | ERROR +; - TIMEOUT/ERROR hang up fail-closed inside the app. +; Dial(PJSIP/qa-bridge,30) -- BARE NAME, NO @host. Adding @host would bypass +; the registrar and never reach the dynamic bridge. ",30" = ring timeout secs. +; ============================================================================ + +[pin-gate] +; Catch-all: whatever DID Chime dials routes to the PIN gate. rustisk supports +; Asterisk-style pattern extensions (_X. = one+ digits). Explicit 9000 kept for +; the synthetic caller / manual origination. +exten => _X.,1,Answer() + same => n,PinGate(/mnt/data/herodevs-agents/m9-arm/prompts/pin-prompt.wav,20,7) + same => n,GotoIf($["${PINGATESTATUS}" = "GRANTED"]?10:20) + same => 10,Verbose(0,PIN_GATE_RESULT=GRANTED) + same => n,Playback(/mnt/data/herodevs-agents/m9-arm/prompts/granted.wav) + same => n,Dial(PJSIP/qa-bridge,30) + same => n,Hangup() + same => 20,Verbose(0,PIN_GATE_RESULT=REJECTED) + same => n,Playback(/mnt/data/herodevs-agents/m9-arm/prompts/rejected.wav) + same => n,Hangup() + +exten => 9000,1,Answer() + same => n,PinGate(/mnt/data/herodevs-agents/m9-arm/prompts/pin-prompt.wav,20,7) + same => n,GotoIf($["${PINGATESTATUS}" = "GRANTED"]?10:20) + same => 10,Verbose(0,PIN_GATE_RESULT=GRANTED) + same => n,Playback(/mnt/data/herodevs-agents/m9-arm/prompts/granted.wav) + same => n,Dial(PJSIP/qa-bridge,30) + same => n,Hangup() + same => 20,Verbose(0,PIN_GATE_RESULT=REJECTED) + same => n,Playback(/mnt/data/herodevs-agents/m9-arm/prompts/rejected.wav) + same => n,Hangup() diff --git a/tests/e2e-trunk/integration/config-live/manager.conf b/tests/e2e-trunk/integration/config-live/manager.conf new file mode 100644 index 0000000..f266530 --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/manager.conf @@ -0,0 +1,9 @@ +[general] +enabled = yes +bindaddr = 127.0.0.1 +port = 25038 + +[harness] +secret = m9-dryrun-local-only +read = all +write = system diff --git a/tests/e2e-trunk/integration/config-live/pin_gate.conf b/tests/e2e-trunk/integration/config-live/pin_gate.conf new file mode 100644 index 0000000..7db3d0e --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/pin_gate.conf @@ -0,0 +1,8 @@ +[general] +; Point at a mounted file containing the REAL 6 ASCII digits (no trailing junk; +; a trailing newline is tolerated). Reuse the SAME secret the live FS uses: +; the FS trunk reads /etc/voice-sip/inbound-pin (k8s secret voice-sip/inbound-pin). +; Stage that value into a host-local file for rustisk, e.g.: +; microk8s kubectl get secret voice-sip -n voice -o jsonpath='{.data.inbound-pin}' \ +; | base64 -d > /mnt/data/herodevs-agents/m9-arm/secret/pin.live (chmod 600) +secret_file = /mnt/data/herodevs-agents/m9-arm/secret/pin.live diff --git a/tests/e2e-trunk/integration/config-live/pjsip.conf b/tests/e2e-trunk/integration/config-live/pjsip.conf new file mode 100644 index 0000000..29fce30 --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/pjsip.conf @@ -0,0 +1,50 @@ +; ============================================================================ +; rustisk M9 trunk config — pjsip.conf (LIVE — copy over pjsip.conf at cutover) +; Values already substituted from the dry-run rendering. Confirm the two +; items before use. voice.murphytek.com resolved to 66.7.119.163 at arming time +; (2026-07-18) — re-verify at cutover: getent hosts voice.murphytek.com +; ============================================================================ + +[transport-udp] +type = transport +protocol = udp +bind = 192.168.0.109:45070 +external_signaling_address = 66.7.119.163 +external_media_address = 66.7.119.163 +local_net = 192.168.0.0/16 +local_net = 10.0.0.0/8 + +[chime-trunk] +type = endpoint +context = pin-gate +disallow = all +allow = ulaw +direct_media = no +rtp_symmetric = yes +dtmf_mode = rfc4733 + +[chime-trunk-identify] +type = identify +endpoint = chime-trunk +match = 99.77.253.0/24 + +[qa-bridge] +type = endpoint +context = pin-gate +disallow = all +allow = ulaw +direct_media = no +rtp_symmetric = yes +dtmf_mode = rfc4733 +auth = qa-bridge-auth + +[qa-bridge-auth] +type = auth +auth_type = userpass +username = qa-bridge +password = + +[qa-bridge-identify] +type = identify +endpoint = qa-bridge +match = 192.168.0.109/32 diff --git a/tests/e2e-trunk/integration/config-live/rtp.conf b/tests/e2e-trunk/integration/config-live/rtp.conf new file mode 100644 index 0000000..e2f208f --- /dev/null +++ b/tests/e2e-trunk/integration/config-live/rtp.conf @@ -0,0 +1,3 @@ +[general] +rtpstart = 20000 +rtpend = 20100 diff --git a/tests/e2e-trunk/integration/qa_probe.py b/tests/e2e-trunk/integration/qa_probe.py new file mode 100644 index 0000000..c5fb1bb --- /dev/null +++ b/tests/e2e-trunk/integration/qa_probe.py @@ -0,0 +1,67 @@ +#!/usr/bin/env python3 +"""M9 dry-run qa-sip far-end probe (runs inside the vs-harness image). + +Joins the qa-sip Mumble channel as a controlled, DETERMINISTIC far end: + * TRANSMITS a steady 660 Hz tone into qa-sip -> proves qa-sip -> A + (the SIP caller's RTP RX should show a 660 Hz component). + * RECEIVES channel audio and FFT-detects 440 Hz -> proves A -> qa-sip + (the caller emits 440 Hz; it arrives via the ryan-phone bridge). + +Named 'm9-probe' (matches the streaming agent's non-human denylist, so it does +NOT itself register as a human presence — only the ryan-phone bridge does). +""" +import os, sys, time, threading, math +import numpy as np +import pymumble_py3 as pymumble +from pymumble_py3.constants import PYMUMBLE_CLBK_SOUNDRECEIVED + +HOST=os.getenv("MUMBLE_HOST","127.0.0.1"); PORT=int(os.getenv("MUMBLE_PORT","64738")) +PASS=os.getenv("MUMBLE_PASSWORD",""); CH=os.getenv("MUMBLE_CHANNEL","qa-sip") +USER=os.getenv("PROBE_NAME","m9-probe"); SECS=float(os.getenv("PROBE_SECS","16")) +TX_HZ=int(os.getenv("TX_HZ","660")) +def log(*a): print("[probe]",*a,flush=True) + +rx=[] # accumulate received 48k int16 +def on_sound(user, chunk): + a=np.frombuffer(chunk.pcm, dtype="=f0-25)&(f<=f0+25); return float(X[msk].max())/tot*1e4 if msk.any() else 0.0 + res=(f"PROBE_RESULT samples={N} ratio440={band(440):.2f} ratio660={band(660):.2f} " + f"users={sorted(who)}") + print(res) + open("/probe/probe_result.txt","w").write(res+"\n") +else: + res="PROBE_RESULT samples=0 ratio440=0.00 users=[] (NO audio received)" + print(res); open("/probe/probe_result.txt","w").write(res+"\n") +m.stop() diff --git a/tests/e2e-trunk/mock_bridge.py b/tests/e2e-trunk/mock_bridge.py new file mode 100755 index 0000000..989fe1b --- /dev/null +++ b/tests/e2e-trunk/mock_bridge.py @@ -0,0 +1,419 @@ +#!/usr/bin/env python3 +"""Hermetic MOCK registered SIP endpoint that stands in for the pymumble +qa-bridge in the durable e2e-trunk suite. + +It replaces the whole Mumble/voice stack with a tiny, deterministic UAS so the +trunk e2e is self-contained (no external services). It: + + 1. REGISTERs to rustisk as `qa-bridge` with MD5 digest auth (realm "asterisk", + qop=auth) so rustisk's registrar binds a dynamic AOR contact and + `Dial(PJSIP/qa-bridge)` resolves to THIS process (proven registrar + + dynamic-AOR routing path, see rustisk registrar.rs / channel_driver.rs). + 2. Answers the INVITE rustisk sends when a GRANTED PIN triggers the Dial + (100 Trying -> 200 OK with a PCMU + telephone-event SDP answer). + 3. Runs bidirectional media: it TRANSMITS a distinct steady tone (default + 660 Hz) toward the bridge AND RECEIVES the caller's relayed tone, + Goertzel-detecting the caller's tone (default 440 Hz). Distinct tones are + used (rather than a pure echo) precisely so each RTP direction is + independently attributable by frequency — a strictly stronger two-way + proof than an echo, which cannot tell a relayed frame from a socket loop. + +It records an `invite_count`: 0 proves the REJECTED path never Dialed the +bridge; 1 (with RX frames + a detected caller tone) proves the GRANTED path +bridged real two-way media. + +Pure Python stdlib (Goertzel, not numpy) so it runs on a bare CI runner. +""" +import argparse +import hashlib +import json +import math +import os +import random +import signal +import socket +import struct +import sys +import threading +import time + +_TERMINATE = threading.Event() + + +def log(*a): + print("[mock-bridge]", *a, file=sys.stderr, flush=True) + + +# ---- mulaw + RTP + Goertzel (same lineage as chime_caller.py) ---- +def linear_to_mulaw(sample): + bias = 0x84 + clip = 32635 + sign = 0x80 if sample < 0 else 0 + magnitude = min(abs(sample), clip) + bias + exponent = 7 + mask = 0x4000 + while exponent > 0 and not magnitude & mask: + exponent -= 1 + mask >>= 1 + mantissa = (magnitude >> (exponent + 3)) & 0x0F + return (~(sign | (exponent << 4) | mantissa)) & 0xFF + + +_MULAW_DECODE = None + + +def mulaw_to_linear(u): + global _MULAW_DECODE + if _MULAW_DECODE is None: + tab = [] + for i in range(256): + u2 = ~i & 0xFF + sign = u2 & 0x80 + exp = (u2 >> 4) & 0x07 + man = u2 & 0x0F + mag = ((man << 3) + 0x84) << exp + mag -= 0x84 + tab.append(-mag if sign else mag) + _MULAW_DECODE = tab + return _MULAW_DECODE[u & 0xFF] + + +def tone_pcmu(freq, packet_index, amp=10000): + out = bytearray() + for offset in range(160): + absolute = packet_index * 160 + offset + s = round(amp * math.sin(2 * math.pi * freq * absolute / 8000)) + out.append(linear_to_mulaw(s)) + return bytes(out) + + +def rtp_hdr(pt, seq, ts, ssrc, marker=0): + return struct.pack( + "!BBHII", 0x80, ((marker & 1) << 7) | (pt & 0x7F), seq & 0xFFFF, ts & 0xFFFFFFFF, ssrc & 0xFFFFFFFF + ) + + +def goertzel(samples, target_hz, rate=8000): + n = len(samples) + if n < 320: + return 0.0 + k = int(0.5 + (n * target_hz) / rate) + w = (2.0 * math.pi / n) * k + coeff = 2.0 * math.cos(w) + s_prev = 0.0 + s_prev2 = 0.0 + energy = 0.0 + for x in samples: + energy += x * x + s = x + coeff * s_prev - s_prev2 + s_prev2 = s_prev + s_prev = s + power = s_prev2 * s_prev2 + s_prev * s_prev - coeff * s_prev * s_prev2 + if energy <= 0: + return 0.0 + return round(math.sqrt(max(power, 0.0)) / math.sqrt(energy) * 1e4, 2) + + +def md5hex(s): + return hashlib.md5(s.encode()).hexdigest() + + +def header(msg, name): + """First header value (case-insensitive) from a raw SIP message.""" + want = name.lower() + ":" + for line in msg.split("\r\n"): + if line.lower().startswith(want): + return line.split(":", 1)[1].strip() + return None + + +def all_via(msg): + return [line for line in msg.split("\r\n") if line.lower().startswith("via:")] + + +def parse_challenge(msg): + """Extract realm+nonce+qop from a 401 WWW-Authenticate header.""" + wa = header(msg, "WWW-Authenticate") or "" + fields = {} + for part in wa[len("Digest"):].split(",") if wa.lower().startswith("digest") else []: + if "=" in part: + k, v = part.split("=", 1) + fields[k.strip().lower()] = v.strip().strip('"') + return fields.get("realm", "asterisk"), fields.get("nonce", ""), fields.get("qop", "") + + +def parse_sdp(body): + pcmu = 0 + tev = 101 + rip = None + rport = None + for line in body.splitlines(): + line = line.strip() + if line.startswith("m=audio"): + rport = int(line.split()[1]) + elif line.startswith("c=IN IP4"): + rip = line.split()[-1] + elif line.lower().startswith("a=rtpmap:"): + pt, enc = line[len("a=rtpmap:"):].split(None, 1) + if enc.upper().startswith("PCMU/"): + pcmu = int(pt) + if enc.lower().startswith("telephone-event/"): + tev = int(pt) + return pcmu, tev, rip, rport + + +def register(sipsock, args): + """Send REGISTER, answer the 401 digest challenge, expect 200 OK.""" + reg_domain = f"{args.reg_ip}:{args.reg_port}" + ruri = f"sip:{reg_domain}" + callid = f"{random.randint(0, 1 << 48):012x}@{args.src_ip}" + fromtag = f"{random.randint(0, 1 << 32):08x}" + contact = f"" + cseq = 1 + + def build(auth_header=None): + branch = f"z9hG4bK{random.randint(0, 1 << 32):08x}" + lines = [ + f"REGISTER {ruri} SIP/2.0", + f"Via: SIP/2.0/UDP {args.src_ip}:{args.sip_port};branch={branch};rport", + "Max-Forwards: 70", + f"From: ;tag={fromtag}", + f"To: ", + f"Call-ID: {callid}", + f"CSeq: {cseq} REGISTER", + f"Contact: {contact}", + f"Expires: {args.expires}", + ] + if auth_header: + lines.append(auth_header) + lines += ["Content-Length: 0", "", ""] + return "\r\n".join(lines) + + sipsock.sendto(build().encode(), (args.reg_ip, args.reg_port)) + log(f"REGISTER -> {ruri} (as {args.username})") + deadline = time.time() + 8 + while time.time() < deadline: + try: + data, _ = sipsock.recvfrom(65535) + except socket.timeout: + continue + msg = data.decode("latin1") + first = msg.split("\r\n", 1)[0] + if "401" in first or "407" in first: + realm, nonce, qop = parse_challenge(msg) + realm = args.realm or realm + cseq += 1 + ha1 = md5hex(f"{args.username}:{realm}:{args.password}") + ha2 = md5hex(f"REGISTER:{ruri}") + if "auth" in (qop or ""): + cnonce = f"{random.randint(0, 1 << 32):08x}" + nc = "00000001" + resp = md5hex(f"{ha1}:{nonce}:{nc}:{cnonce}:auth:{ha2}") + auth = ( + f'Authorization: Digest username="{args.username}", realm="{realm}", ' + f'nonce="{nonce}", uri="{ruri}", response="{resp}", algorithm=MD5, ' + f'qop=auth, nc={nc}, cnonce="{cnonce}"' + ) + else: + resp = md5hex(f"{ha1}:{nonce}:{ha2}") + auth = ( + f'Authorization: Digest username="{args.username}", realm="{realm}", ' + f'nonce="{nonce}", uri="{ruri}", response="{resp}", algorithm=MD5' + ) + sipsock.sendto(build(auth).encode(), (args.reg_ip, args.reg_port)) + log("REGISTER (authed) -> resending with digest response") + continue + if " 200 " in first: + log("REGISTER: 200 OK (contact bound)") + return True + log("REGISTER <=", first) + return False + + +def respond(sipsock, dst, status, msg, extra_headers=None, body="", to_tag=None): + """Build a response echoing dialog headers from the request `msg`.""" + lines = [f"SIP/2.0 {status}"] + lines += all_via(msg) + frm = header(msg, "From") + to = header(msg, "To") + if to_tag and "tag=" not in (to or "").lower(): + to = f"{to};tag={to_tag}" + lines.append(f"From: {frm}") + lines.append(f"To: {to}") + lines.append(f"Call-ID: {header(msg, 'Call-ID')}") + lines.append(f"CSeq: {header(msg, 'CSeq')}") + if extra_headers: + lines += extra_headers + if body: + lines.append("Content-Type: application/sdp") + lines.append(f"Content-Length: {len(body)}") + lines += ["", body] + else: + lines += ["Content-Length: 0", "", ""] + sipsock.sendto(("\r\n".join(lines)).encode(), dst) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--reg-ip", required=True) # rustisk SIP IP + ap.add_argument("--reg-port", type=int, required=True) # rustisk SIP port + ap.add_argument("--src-ip", required=True) # this mock's IP + ap.add_argument("--sip-port", type=int, default=35064) + ap.add_argument("--rtp-port", type=int, default=36300) + ap.add_argument("--username", default="qa-bridge") + ap.add_argument("--password", required=True) + ap.add_argument("--realm", default="asterisk") + ap.add_argument("--expires", type=int, default=300) + ap.add_argument("--tx-tone", type=int, default=660) # distinct far-end tone + ap.add_argument("--detect", type=int, default=440) # caller tone we expect to receive + ap.add_argument("--run-secs", type=float, default=40.0) + ap.add_argument("--ready-file", required=True) + ap.add_argument("--result-file", required=True) + args = ap.parse_args() + + # SIGTERM (from run.sh, e.g. to end the REJECTED case where no INVITE ever + # arrives) finalizes gracefully: break the loop, write the result file. + signal.signal(signal.SIGTERM, lambda *_: _TERMINATE.set()) + + sipsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sipsock.bind((args.src_ip, args.sip_port)) + sipsock.settimeout(0.25) + rtpsock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + rtpsock.bind((args.src_ip, args.rtp_port)) + rtpsock.settimeout(0.25) + + if not register(sipsock, args): + _write(args, {"registered": False, "invite_count": 0, "rx_voice": 0}, exit_code=3) + return 3 + open(args.ready_file, "w").write("registered\n") + + state = {"invite_count": 0, "rx_voice": 0, "rx_bytes": 0} + seen_callids = set() + stop_media = threading.Event() + got_bye = threading.Event() + rx_pcm = [] + media_started = threading.Event() + remote = {"ip": None, "port": None, "pcmu_pt": 0} + + def rx_loop(): + while not stop_media.is_set(): + try: + pkt, _ = rtpsock.recvfrom(4096) + except socket.timeout: + continue + if len(pkt) < 12: + continue + pt = pkt[1] & 0x7F + state["rx_bytes"] += len(pkt) + payload = pkt[12:] + if pt == remote["pcmu_pt"] and payload: + state["rx_voice"] += 1 + rx_pcm.extend(mulaw_to_linear(b) for b in payload) + + def tx_loop(): + seq = random.randint(0, 0x7FFF) + ts = random.randint(0, 0x7FFFFFFF) + ssrc = random.randint(0, 0x7FFFFFFF) + idx = 0 + media_started.wait() + while not stop_media.is_set(): + if remote["ip"]: + pkt = rtp_hdr(remote["pcmu_pt"], seq, ts, ssrc) + tone_pcmu(args.tx_tone, idx) + try: + rtpsock.sendto(pkt, (remote["ip"], remote["port"])) + except OSError: + pass + seq = (seq + 1) & 0xFFFF + ts = (ts + 160) & 0xFFFFFFFF + idx += 1 + time.sleep(0.02) + + threading.Thread(target=rx_loop, daemon=True).start() + threading.Thread(target=tx_loop, daemon=True).start() + + to_tag = f"{random.randint(0, 1 << 32):08x}" + deadline = time.time() + args.run_secs + while time.time() < deadline and not _TERMINATE.is_set(): + try: + data, addr = sipsock.recvfrom(65535) + except socket.timeout: + if got_bye.is_set(): + break + continue + msg = data.decode("latin1") + first = msg.split("\r\n", 1)[0] + method = first.split(" ", 1)[0].upper() + if method == "INVITE": + callid = header(msg, "Call-ID") + body = msg.split("\r\n\r\n", 1)[1] if "\r\n\r\n" in msg else "" + pcmu_pt, tev_pt, rip, rport = parse_sdp(body) + remote["ip"], remote["port"], remote["pcmu_pt"] = rip, rport, pcmu_pt + if callid not in seen_callids: + seen_callids.add(callid) + state["invite_count"] += 1 + log(f"INVITE #{state['invite_count']} from {addr} remote_media={rip}:{rport} PCMU_PT={pcmu_pt}") + respond(sipsock, addr, "100 Trying", msg) + answer = ( + f"v=0\r\no=- {random.randint(1, 1 << 30)} 1 IN IP4 {args.src_ip}\r\ns=mock-bridge\r\n" + f"c=IN IP4 {args.src_ip}\r\nt=0 0\r\nm=audio {args.rtp_port} RTP/AVP {pcmu_pt} {tev_pt}\r\n" + f"a=rtpmap:{pcmu_pt} PCMU/8000\r\na=rtpmap:{tev_pt} telephone-event/8000\r\n" + f"a=fmtp:{tev_pt} 0-15\r\na=ptime:20\r\na=sendrecv\r\n" + ) + respond( + sipsock, + addr, + "200 OK", + msg, + extra_headers=[f"Contact: "], + body=answer, + to_tag=to_tag, + ) + media_started.set() + elif method == "ACK": + media_started.set() + elif method == "BYE": + respond(sipsock, addr, "200 OK", msg) + log("BYE received -> 200 OK; draining") + got_bye.set() + # brief drain so trailing RTP is counted, then exit the loop + time.sleep(0.4) + break + elif method in ("OPTIONS", "INFO", "UPDATE"): + respond(sipsock, addr, "200 OK", msg) + + stop_media.set() + time.sleep(0.2) + + det = {} + if len(rx_pcm) >= 1600: + mean = sum(rx_pcm) / len(rx_pcm) + centered = [s - mean for s in rx_pcm] + for f0 in (args.detect, args.tx_tone, 350, 880): + det[f0] = goertzel(centered, f0) + + result = { + "registered": True, + "invite_count": state["invite_count"], + "rx_voice": state["rx_voice"], + "rx_bytes": state["rx_bytes"], + "rx_samples": len(rx_pcm), + "tone_ratios": det, + "detect_hz": args.detect, + } + _write(args, result, exit_code=0) + print( + f"RESULT registered=True invite_count={state['invite_count']} rx_voice={state['rx_voice']} " + f"rx_bytes={state['rx_bytes']} rx_samples={len(rx_pcm)} tone_ratios={det}" + ) + return 0 + + +def _write(args, result, exit_code): + tmp = args.result_file + ".tmp" + with open(tmp, "w") as f: + json.dump(result, f) + os.replace(tmp, args.result_file) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/e2e-trunk/run.sh b/tests/e2e-trunk/run.sh new file mode 100755 index 0000000..b85baea --- /dev/null +++ b/tests/e2e-trunk/run.sh @@ -0,0 +1,362 @@ +#!/usr/bin/env bash +# ============================================================================= +# Durable, hermetic e2e for rustisk's TRUNK PATH: +# synthetic Chime INVITE -> rustisk PIN gate -> (correct PIN) GRANTED -> +# Dial(PJSIP/qa-bridge) -> a MOCK registered SIP endpoint answers -> +# two-way RTP (both directions, distinct tones) ; and +# (wrong PIN) REJECTED -> NO Dial (the bridge is never INVITEd). +# +# Self-contained: needs ONLY the rustisk binary + python3 (stdlib) + loopback. +# NO FreeSWITCH, NO Mumble, NO docker, NO NET_RAW, NO root. The pymumble bridge +# is replaced by tests/e2e-trunk/mock_bridge.py (REGISTER + digest + echo tone). +# +# Everything runs on isolated 127.0.0.6x loopback IPs and 35xxx/36xxx ports so +# it CANNOT collide with the live trunk (:45070), the M9 arming rig (:25060/ +# :25038/:25062/:21000-21100) or the M0 instance (:15060/:15038). A preflight +# asserts our ports are free before touching anything. +# +# Exit status is the test result: 0 = all cases passed, non-zero = failure. +# +# Env knobs: +# RUSTISK_BIN path to the rustisk binary (default: build/target/debug/rustisk; +# if absent it is built with `cargo build --bin rustisk`). +# KEEP_RUNTIME set to keep $RUNTIME_DIR artifacts after the run. +# ============================================================================= +set -euo pipefail + +HARNESS_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" +REPO_DIR="$(cd -- "$HARNESS_DIR/../.." && pwd)" +RUNTIME_DIR="$REPO_DIR/target/e2e-trunk" +CONFIG_DIR="$RUNTIME_DIR/config" +RUN_DIR="$RUNTIME_DIR/run" +PROMPT_DIR="$RUNTIME_DIR/prompts" +SECRET_FILE="$RUNTIME_DIR/pin.secret" +RUSTISK_LOG="$RUNTIME_DIR/rustisk.log" +RUSTISK_BIN="${RUSTISK_BIN:-$REPO_DIR/target/debug/rustisk}" + +# --- isolated loopback topology (fresh; no overlap with live/m0/m9 ports) --- +RK_IP=127.0.0.60; RK_SIP=35060; RK_AMI=35038 +RTP_START=36000; RTP_END=36100 +CALLER_IP=127.0.0.61; CALLER_SIP=35062; CALLER_RTP=36200 +MOCK_IP=127.0.0.62; MOCK_SIP=35064; MOCK_RTP=36300 + +# --- TEST-ONLY secrets (never the real PIN / qa-bridge password) --- +TEST_PIN="246813" +WRONG_PIN="999999" +BRIDGE_USER="qa-bridge" +BRIDGE_PASS="e2e-mock-bridge-secret" + +# --- media tones --- +CALLER_TONE=440 # caller -> bridge (proves A->bridge on the mock RX) +BRIDGE_TONE=660 # bridge -> caller (proves bridge->A on the caller RX) + +# --- assertion thresholds (overridable) --- +MIN_VOICE_FRAMES="${MIN_VOICE_FRAMES:-100}" +MIN_TONE_RATIO="${MIN_TONE_RATIO:-20}" + +RUSTISK_PID="" +MOCK_PID="" + +log() { printf '%s\n' "$*"; } +fail() { printf 'FAIL: %s\n' "$*" >&2; exit 1; } + +cleanup() { + [[ -n "$MOCK_PID" ]] && { kill -TERM "$MOCK_PID" 2>/dev/null || true; wait "$MOCK_PID" 2>/dev/null || true; } + if [[ -n "$RUSTISK_PID" ]]; then + kill -TERM "$RUSTISK_PID" 2>/dev/null || true + for _ in {1..20}; do kill -0 "$RUSTISK_PID" 2>/dev/null || break; sleep 0.1; done + kill -KILL "$RUSTISK_PID" 2>/dev/null || true + wait "$RUSTISK_PID" 2>/dev/null || true + fi + if [[ -z "${KEEP_RUNTIME:-}" ]]; then rm -f "$SECRET_FILE" 2>/dev/null || true; fi +} +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' TERM + +require() { command -v "$1" >/dev/null 2>&1 || fail "required command not found: $1"; } + +preflight_ports() { + # Prove none of OUR ports are already bound (guards against colliding with + # a stray instance, and makes a port typo fail fast rather than mysteriously). + local busy="" + local udp tcp + udp="$(ss -Huan 2>/dev/null || true)" + tcp="$(ss -Htan 2>/dev/null || true)" + local p + for p in "$RK_SIP" "$CALLER_SIP" "$MOCK_SIP" "$CALLER_RTP" "$MOCK_RTP" "$RTP_START" "$RTP_END"; do + grep -qE "[.:]$p( |\$)|:$p " <<<"$udp" && busy+="udp/$p " + done + grep -qE "[.:]$RK_AMI( |\$)|:$RK_AMI " <<<"$tcp" && busy+="tcp/$RK_AMI " + # Also assert we are NOT about to touch any known live/m0/m9 port. + local forbidden="45070 25060 25038 25062 21000 21100 15060 15038" + for p in "$RK_SIP" "$RK_AMI" "$CALLER_SIP" "$MOCK_SIP" "$RTP_START" "$RTP_END" "$CALLER_RTP" "$MOCK_RTP"; do + for f in $forbidden; do + [[ "$p" == "$f" ]] && fail "port $p collides with a reserved live/m0/m9 port" + done + done + [[ -z "$busy" ]] || fail "ports already in use: $busy (is a stray rustisk/mock running?)" + log "PREFLIGHT: OK — SIP $RK_SIP / AMI $RK_AMI / RTP $RTP_START-$RTP_END + caller $CALLER_SIP/$CALLER_RTP + mock $MOCK_SIP/$MOCK_RTP all free" +} + +ensure_binary() { + if [[ ! -x "$RUSTISK_BIN" ]]; then + log "rustisk binary not found at $RUSTISK_BIN — building (cargo build --bin rustisk)..." + ( cd "$REPO_DIR" && cargo build --bin rustisk ) + RUSTISK_BIN="$REPO_DIR/target/debug/rustisk" + fi + [[ -x "$RUSTISK_BIN" ]] || fail "no usable rustisk binary at $RUSTISK_BIN" + log "RUSTISK_BIN=$RUSTISK_BIN" +} + +write_configs() { + rm -rf "$RUNTIME_DIR" + mkdir -p "$CONFIG_DIR" "$RUN_DIR" "$PROMPT_DIR" + printf '%s\n' "$TEST_PIN" >"$SECRET_FILE"; chmod 600 "$SECRET_FILE" + python3 "$HARNESS_DIR/gen_prompts.py" "$PROMPT_DIR" >/dev/null + + cat >"$CONFIG_DIR/asterisk.conf" <"$CONFIG_DIR/pin_gate.conf" <"$CONFIG_DIR/rtp.conf" <"$CONFIG_DIR/manager.conf" <"$CONFIG_DIR/pjsip.conf" <"$CONFIG_DIR/extensions.conf" < PIN gate. On GRANTED, Dial(PJSIP/qa-bridge) bridges +; the caller into the registered mock endpoint. On REJECTED, play + hang up +; WITHOUT dialing (the bridge must never be INVITEd). +[pin-gate] +exten => _X.,1,Answer() + same => n,PinGate($PROMPT_DIR/pin-prompt.wav,20,7) + same => n,GotoIf(\$["\${PINGATESTATUS}" = "GRANTED"]?10:20) + same => 10,Verbose(0,PIN_GATE_RESULT=GRANTED) + same => n,Playback($PROMPT_DIR/granted.wav) + same => n,Dial(PJSIP/$BRIDGE_USER,30) + same => n,Hangup() + same => 20,Verbose(0,PIN_GATE_RESULT=REJECTED) + same => n,Playback($PROMPT_DIR/rejected.wav) + same => n,Hangup() + +exten => 9000,1,Answer() + same => n,PinGate($PROMPT_DIR/pin-prompt.wav,20,7) + same => n,GotoIf(\$["\${PINGATESTATUS}" = "GRANTED"]?10:20) + same => 10,Verbose(0,PIN_GATE_RESULT=GRANTED) + same => n,Playback($PROMPT_DIR/granted.wav) + same => n,Dial(PJSIP/$BRIDGE_USER,30) + same => n,Hangup() + same => 20,Verbose(0,PIN_GATE_RESULT=REJECTED) + same => n,Playback($PROMPT_DIR/rejected.wav) + same => n,Hangup() +EOF + log "CONFIG: wrote hermetic pjsip/extensions/pin_gate/rtp/manager to $CONFIG_DIR" +} + +start_rustisk() { + RUST_LOG=info stdbuf -oL -eL "$RUSTISK_BIN" -f -C "$CONFIG_DIR/asterisk.conf" \ + >"$RUSTISK_LOG" 2>&1 & + RUSTISK_PID=$! + for _ in {1..100}; do + kill -0 "$RUSTISK_PID" 2>/dev/null || fail "rustisk exited during boot; log:$(printf '\n'; cat "$RUSTISK_LOG")" + grep -q 'Rustisk is fully booted' "$RUSTISK_LOG" && { log "RUSTISK: booted (pid $RUSTISK_PID) on $RK_IP:$RK_SIP"; return 0; } + sleep 0.1 + done + fail "rustisk did not report fully booted within 10s; log:$(printf '\n'; tail -40 "$RUSTISK_LOG")" +} + +start_mock() { + local ready="$1" result="$2" runsecs="$3" + rm -f "$ready" "$result" + python3 "$HARNESS_DIR/mock_bridge.py" \ + --reg-ip "$RK_IP" --reg-port "$RK_SIP" \ + --src-ip "$MOCK_IP" --sip-port "$MOCK_SIP" --rtp-port "$MOCK_RTP" \ + --username "$BRIDGE_USER" --password "$BRIDGE_PASS" \ + --tx-tone "$BRIDGE_TONE" --detect "$CALLER_TONE" \ + --run-secs "$runsecs" --ready-file "$ready" --result-file "$result" \ + >"$RUNTIME_DIR/$(basename "$result").mock.log" 2>&1 & + MOCK_PID=$! + for _ in {1..100}; do + kill -0 "$MOCK_PID" 2>/dev/null || fail "mock_bridge exited before registering; log:$(printf '\n'; cat "$RUNTIME_DIR/$(basename "$result").mock.log")" + [[ -f "$ready" ]] && { log "MOCK: registered as $BRIDGE_USER (pid $MOCK_PID)"; return 0; } + sleep 0.1 + done + fail "mock_bridge did not register within 10s; log:$(printf '\n'; tail -20 "$RUNTIME_DIR/$(basename "$result").mock.log")" +} + +stop_mock() { + [[ -n "$MOCK_PID" ]] || return 0 + kill -TERM "$MOCK_PID" 2>/dev/null || true + wait "$MOCK_PID" 2>/dev/null || true + MOCK_PID="" +} + +run_caller() { + local pin="$1" rxwav="$2" result="$3" callsecs="$4" + rm -f "$result" + python3 "$HARNESS_DIR/chime_caller.py" \ + --dst-ip "$RK_IP" --dst-port "$RK_SIP" \ + --src-ip "$CALLER_IP" --sip-port "$CALLER_SIP" --rtp-port "$CALLER_RTP" \ + --exten 9000 --pin "$pin" \ + --tone "$CALLER_TONE" --detect "$BRIDGE_TONE" \ + --call-secs "$callsecs" --rx-wav "$rxwav" --result-file "$result" \ + 2>"$RUNTIME_DIR/$(basename "$result").caller.log" || true +} + +# JSON assertion helper: python3 predicate over a result file. Exits non-zero +# (with a message) on failure so `set -e` propagates. +assert_json() { + local file="$1" desc="$2"; shift 2 + python3 - "$file" "$desc" "$@" <<'PY' +import json, sys +path, desc = sys.argv[1], sys.argv[2] +checks = sys.argv[3:] +try: + d = json.load(open(path)) +except Exception as e: + print(f"FAIL: {desc}: cannot read {path}: {e}", file=sys.stderr); sys.exit(1) +def ratio(hz): + return float((d.get("tone_ratios") or {}).get(str(hz), 0) or 0) +env = {"d": d, "ratio": ratio, "tr": d.get("tone_ratios") or {}} +for chk in checks: + try: + ok = eval(chk, {}, env) + except Exception as e: + print(f"FAIL: {desc}: check {chk!r} errored: {e} (result={d})", file=sys.stderr); sys.exit(1) + if not ok: + print(f"FAIL: {desc}: check FAILED: {chk} (result={d})", file=sys.stderr); sys.exit(1) +print(f"OK: {desc}: {d}") +PY +} + +# ============================================================================= +main() { + require python3; require ss; require stdbuf + log "=== rustisk e2e-trunk (hermetic) ===" + preflight_ports + ensure_binary + write_configs + start_rustisk + + # ---------- CASE 1: REJECTED (wrong PIN -> NO Dial) ---------- + log "" + log "--- CASE 1: REJECTED (wrong PIN must NOT reach the bridge) ---" + local rej_ready="$RUNTIME_DIR/mock-reject.ready" rej_res="$RUNTIME_DIR/mock-reject.json" + start_mock "$rej_ready" "$rej_res" 25 + run_caller "$WRONG_PIN" "$RUNTIME_DIR/reject-rx.wav" "$RUNTIME_DIR/caller-reject.json" 9 + sleep 1.0 + stop_mock + grep -q 'Verbose: PIN_GATE_RESULT=REJECTED' "$RUSTISK_LOG" \ + || fail "dialplan did not take the REJECTED branch (no 'Verbose: PIN_GATE_RESULT=REJECTED' in rustisk log)" + assert_json "$rej_res" "REJECTED bridge never INVITEd" \ + "d['registered'] is True" "d['invite_count'] == 0" >/dev/null \ + || fail "REJECTED case: bridge WAS INVITEd (invite_count != 0) — reject leaked to Dial" + log "REJECTED: PASS (PIN_GATE_RESULT=REJECTED, bridge invite_count=0)" + + # ---------- CASE 2: GRANTED (correct PIN -> Dial -> two-way media) ---------- + log "" + log "--- CASE 2: GRANTED (correct PIN -> Dial -> two-way RTP) ---" + local grt_ready="$RUNTIME_DIR/mock-grant.ready" grt_res="$RUNTIME_DIR/mock-grant.json" + local caller_res="$RUNTIME_DIR/caller-grant.json" + start_mock "$grt_ready" "$grt_res" 40 + run_caller "$TEST_PIN" "$RUNTIME_DIR/grant-rx.wav" "$caller_res" 15 + # caller sent BYE; give rustisk time to relay the hangup so the mock finalizes + for _ in {1..50}; do [[ -f "$grt_res" ]] && break; sleep 0.2; done + stop_mock + + grep -q 'Verbose: PIN_GATE_RESULT=GRANTED' "$RUSTISK_LOG" \ + || fail "dialplan did not take the GRANTED branch (no 'Verbose: PIN_GATE_RESULT=GRANTED' in rustisk log)" + + # A -> bridge: the mock received the caller's relayed 440 Hz tone + assert_json "$grt_res" "GRANTED mock RX (A->bridge)" \ + "d['invite_count'] == 1" \ + "d['rx_voice'] >= $MIN_VOICE_FRAMES" \ + "ratio($CALLER_TONE) >= $MIN_TONE_RATIO" \ + "ratio($CALLER_TONE) == max((ratio($CALLER_TONE), ratio($BRIDGE_TONE), ratio(350), ratio(880)))" \ + || fail "GRANTED case: A->bridge media not proven on the mock" + + # bridge -> A: the caller received the bridge's distinct 660 Hz tone + assert_json "$caller_res" "GRANTED caller RX (bridge->A)" \ + "d.get('voice_rx',0) >= $MIN_VOICE_FRAMES" \ + "ratio($BRIDGE_TONE) >= $MIN_TONE_RATIO" \ + "ratio($BRIDGE_TONE) == max((ratio($BRIDGE_TONE), ratio($CALLER_TONE), ratio(350), ratio(880)))" \ + || fail "GRANTED case: bridge->A media not proven on the caller" + + log "GRANTED: PASS (two-way RTP: A->bridge 440 Hz + bridge->A 660 Hz, both directions)" + + log "" + log "================= ALL CASES PASSED =================" + log "Evidence: $(python3 -c "import json;m=json.load(open('$grt_res'));c=json.load(open('$caller_res'));print(f\"mock rx_voice={m['rx_voice']} tone440={m['tone_ratios'].get('440')} tone660={m['tone_ratios'].get('660')} | caller voice_rx={c['voice_rx']} tone660={c['tone_ratios'].get('660')} tone440={c['tone_ratios'].get('440')}\")")" + log "Artifacts in $RUNTIME_DIR" +} + +main "$@" diff --git a/tests/freeswitch-pin-gate/rtp_injector.py b/tests/freeswitch-pin-gate/rtp_injector.py index fedc92d..6a5f1df 100644 --- a/tests/freeswitch-pin-gate/rtp_injector.py +++ b/tests/freeswitch-pin-gate/rtp_injector.py @@ -5,6 +5,7 @@ from collections import Counter import json import math +import os import socket import struct import time @@ -61,8 +62,54 @@ def raw_udp( raw.sendto(header + udp, (destination_ip, destination_port)) +def _read_one(packet_socket: socket.socket, args: argparse.Namespace, totals: Counter, udp_flows: Counter): + """Read and classify a single frame. Returns a match dict, or None if the + frame (if any) didn't match — never blocks longer than the socket's own + recv timeout. Always updates totals/udp_flows so nothing observed during + any phase of the sniff is lost.""" + try: + frame = packet_socket.recv(65535) + except TimeoutError: + return None + totals["frames"] += 1 + if len(frame) < 34 or struct.unpack("!H", frame[12:14])[0] != 0x0800: + return None + totals["ipv4"] += 1 + ip_offset = 14 + ihl = (frame[ip_offset] & 0x0F) * 4 + if len(frame) < ip_offset + ihl + 8 or frame[ip_offset + 9] != socket.IPPROTO_UDP: + return None + totals["udp"] += 1 + source_ip = socket.inet_ntoa(frame[ip_offset + 12 : ip_offset + 16]) + destination_ip = socket.inet_ntoa(frame[ip_offset + 16 : ip_offset + 20]) + udp_offset = ip_offset + ihl + source_port, destination_port = struct.unpack("!HH", frame[udp_offset : udp_offset + 4]) + payload = frame[udp_offset + 8 :] + udp_flows[f"{source_ip}:{source_port}->{destination_ip}:{destination_port}"] += 1 + if ( + source_ip != args.source_ip + or destination_ip != args.destination_ip + or source_port != args.source_port + or destination_port != args.destination_port + or len(payload) < 12 + or payload[0] >> 6 != 2 + ): + return None + payload_type = payload[1] & 0x7F + sequence, timestamp, ssrc = struct.unpack("!HII", payload[2:12]) + return { + "payload_type": payload_type, + "sequence": sequence, + "timestamp": timestamp, + "ssrc": ssrc, + } + + +def _flows_summary(udp_flows: Counter) -> str: + return ", ".join(f"{flow}={count}" for flow, count in udp_flows.most_common(8)) + + def sniff(args: argparse.Namespace) -> None: - deadline = time.monotonic() + args.timeout totals: Counter[str] = Counter() udp_flows: Counter[str] = Counter() with socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0800)) as packet_socket: @@ -70,54 +117,47 @@ def sniff(args: argparse.Namespace) -> None: with open(args.ready_file, "w", encoding="utf-8") as ready: ready.write("ready\n") packet_socket.settimeout(0.2) - while time.monotonic() < deadline: - try: - frame = packet_socket.recv(65535) - except TimeoutError: - continue - totals["frames"] += 1 - if len(frame) < 34 or struct.unpack("!H", frame[12:14])[0] != 0x0800: - continue - totals["ipv4"] += 1 - ip_offset = 14 - ihl = (frame[ip_offset] & 0x0F) * 4 - if len(frame) < ip_offset + ihl + 8 or frame[ip_offset + 9] != socket.IPPROTO_UDP: - continue - totals["udp"] += 1 - source_ip = socket.inet_ntoa(frame[ip_offset + 12 : ip_offset + 16]) - destination_ip = socket.inet_ntoa(frame[ip_offset + 16 : ip_offset + 20]) - udp_offset = ip_offset + ihl - source_port, destination_port = struct.unpack("!HH", frame[udp_offset : udp_offset + 4]) - payload = frame[udp_offset + 8 :] - udp_flows[f"{source_ip}:{source_port}->{destination_ip}:{destination_port}"] += 1 - if ( - source_ip != args.source_ip - or destination_ip != args.destination_ip - or source_port != args.source_port - or destination_port != args.destination_port - or len(payload) < 12 - or payload[0] >> 6 != 2 - ): - continue - payload_type = payload[1] & 0x7F - sequence, timestamp, ssrc = struct.unpack("!HII", payload[2:12]) - print( - json.dumps( - { - "payload_type": payload_type, - "sequence": sequence, - "timestamp": timestamp, - "ssrc": ssrc, - }, - sort_keys=True, - ) + + # Priming phase. The capture loop is already live at this point (the + # ready-file above proves the raw socket is bound), but the caller's + # flow-triggering action (an ESL round trip through a `docker exec` + # into a FreeSWITCH container) has its own, host-load-dependent + # startup latency that has nothing to do with RTP transit time. The + # original implementation started its match deadline the instant the + # socket opened, so that setup latency was silently deducted from the + # assertion window — occasionally consuming all of it before + # FreeSWITCH ever emitted a packet, which read as "no matching RTP + # packet observed" even though RTP was about to flow (or had just + # started) fine. Keep reading and classifying every frame during this + # phase — so an early packet is never missed — but don't start the + # real timeout clock until the caller confirms (via go_file) that it + # has actually issued the flow-triggering action. If a match arrives + # during priming, that's fine too: treat it as an immediate go. + match = None + if args.go_file: + priming_deadline = time.monotonic() + args.priming_timeout + while match is None and not os.path.exists(args.go_file): + if time.monotonic() > priming_deadline: + raise SystemExit( + "go-file was never observed within the priming budget " + f"({args.priming_timeout}s) — the flow-triggering action " + "was never confirmed issued: " + f"frames={totals['frames']} ipv4={totals['ipv4']} " + f"udp={totals['udp']} flows=[{_flows_summary(udp_flows)}]" + ) + match = _read_one(packet_socket, args, totals, udp_flows) + + deadline = time.monotonic() + args.timeout + while match is None and time.monotonic() < deadline: + match = _read_one(packet_socket, args, totals, udp_flows) + + if match is None: + raise SystemExit( + "no matching RTP packet observed: " + f"frames={totals['frames']} ipv4={totals['ipv4']} udp={totals['udp']} " + f"flows=[{_flows_summary(udp_flows)}]" ) - return - flows = ", ".join(f"{flow}={count}" for flow, count in udp_flows.most_common(8)) - raise SystemExit( - "no matching RTP packet observed: " - f"frames={totals['frames']} ipv4={totals['ipv4']} udp={totals['udp']} flows=[{flows}]" - ) + print(json.dumps(match, sort_keys=True)) def inject(args: argparse.Namespace) -> None: @@ -262,6 +302,22 @@ def main() -> None: common_arguments(sniff_parser) sniff_parser.add_argument("--timeout", type=float, default=5.0) sniff_parser.add_argument("--ready-file") + sniff_parser.add_argument( + "--go-file", + help=( + "Path the caller creates once it has confirmed the flow-triggering " + "action was actually issued (e.g. after a synchronous ESL command " + "returns). --timeout is measured from this point, not from socket " + "open, so caller-side setup latency can't eat into the match window. " + "If omitted, --timeout is measured from socket open as before." + ), + ) + sniff_parser.add_argument( + "--priming-timeout", + type=float, + default=20.0, + help="Bounded fallback: how long to wait for --go-file to appear before giving up.", + ) sniff_parser.set_defaults(function=sniff) inject_parser = commands.add_parser("inject") diff --git a/tests/freeswitch-pin-gate/run.sh b/tests/freeswitch-pin-gate/run.sh index 9903ed6..514143f 100755 --- a/tests/freeswitch-pin-gate/run.sh +++ b/tests/freeswitch-pin-gate/run.sh @@ -1083,6 +1083,7 @@ run_m2_two_way_bye_case() { local sniff_pid local metadata_file="$RUNTIME_DIR/m2-a-rtp-metadata.json" local sniffer_ready="$RUNTIME_DIR/m2-sniffer-ready" + local sniffer_go="$RUNTIME_DIR/m2-sniffer-go" local metadata local payload_type local sequence @@ -1117,13 +1118,14 @@ run_m2_two_way_bye_case() { [[ "$negotiated_remote" == "$FS_CONTAINER_IP:$a_source_port" ]] \ || fail "inbound negotiated remote mismatch: expected=$FS_CONTAINER_IP:$a_source_port actual=$negotiated_remote" - rm -f "$sniffer_ready" + rm -f "$sniffer_ready" "$sniffer_go" rtp_sniffer sniff \ --source-ip "$FS_CONTAINER_IP" \ --source-port "$a_source_port" \ --destination-ip "$FS_HOST_IP" \ --destination-port "$a_destination_port" \ --ready-file /runtime/m2-sniffer-ready \ + --go-file /runtime/m2-sniffer-go \ --timeout 5 >"$metadata_file" & sniff_pid=$! for _ in {1..100}; do @@ -1131,7 +1133,17 @@ run_m2_two_way_bye_case() { sleep 0.05 done [[ -f "$sniffer_ready" ]] || fail "RTP sniffer did not become ready" + # The sniffer's 5s match deadline does not start until it observes + # $sniffer_go (see rtp_injector.py's priming phase). Only touch it after + # fs_cli's synchronous ESL round trip confirms FreeSWITCH actually + # accepted uuid_broadcast — this closes the race where docker-exec/ESL + # setup latency was silently consumed out of the match window before any + # real media left FreeSWITCH (previously: deadline started at socket + # open, so a slow `docker exec`+ESL round trip could eat the whole + # window and fail with "no matching RTP packet observed" while real + # frames were still incoming or about to arrive). fs_cli "uuid_broadcast $a_uuid tone_stream://%(1000,0,440) aleg" >/dev/null + touch "$sniffer_go" wait "$sniff_pid" || fail "RTP sniffer did not observe A's negotiated source" sleep 1.2 fs_cli "uuid_broadcast $b_uuid tone_stream://%(1000,0,660) aleg" >/dev/null @@ -1643,6 +1655,13 @@ if [[ "${FREESWITCH_PIN_GATE_CASE:-all}" == "m5-soak" ]]; then exit 0 fi +if [[ "${FREESWITCH_PIN_GATE_CASE:-all}" == "m2-standalone" ]]; then + run_m2_two_way_bye_case + printf '\nPASS: isolated M2 two-way / ingress-hygiene / BYE-silent re-run.\n' + printf 'Proof artifacts: %s\n' "$RUNTIME_DIR" + exit 0 +fi + start_m3_sink_receivers run_case 1 "$WRONG_TEST_PIN" REJECTED run_case 2 "$GRANTED_TEST_PIN" GRANTED From 8018eda982c3599460d1d324217668c68fb0f772 Mon Sep 17 00:00:00 2001 From: Repin Agent Date: Sat, 18 Jul 2026 16:07:35 -0600 Subject: [PATCH 2/2] test(e2e): harden trunk e2e assertions (adversarial review follow-up) - Raise MIN_TONE_RATIO 20 -> 5000. At 20 it sat BELOW the ~150-360 Goertzel noise floor and never discriminated signal from noise; also require the target band to beat every other band by 10x (not merely be the max). Closes the narrow window where a silence/comfort-noise partial relay could pass on a coincidental band "win". - Assert the received RTP source IP == rustisk's media IP on BOTH legs, so the suite proves rustisk RELAYED the audio (defends against a direct_media/ peer-to-peer regression), not merely that media crossed end-to-end. - mock_bridge: wake tx_loop's answer-wait on shutdown so the REJECTED path never parks a thread; extend the post-stop drain to 0.3s (> the 0.25s rx socket timeout) so rx_loop settles before the samples are read. --- tests/e2e-trunk/README.md | 30 ++++++++++++++++++++---------- tests/e2e-trunk/chime_caller.py | 5 ++++- tests/e2e-trunk/mock_bridge.py | 13 ++++++++++--- tests/e2e-trunk/run.sh | 12 +++++++++--- 4 files changed, 43 insertions(+), 17 deletions(-) diff --git a/tests/e2e-trunk/README.md b/tests/e2e-trunk/README.md index 6a80739..7fc84e2 100644 --- a/tests/e2e-trunk/README.md +++ b/tests/e2e-trunk/README.md @@ -29,16 +29,25 @@ harness (which drove a real Amazon Chime UAC into the real qa-sip Mumble bridge) **CASE 2 — GRANTED (correct PIN → Dial → two-way media)** - rustisk log contains `Verbose: PIN_GATE_RESULT=GRANTED`. - **A → bridge:** the mock received the caller's relayed tone — - `invite_count == 1`, `rx_voice ≥ 100` frames, and **440 Hz is the dominant - band** in the mock's received audio. + `invite_count == 1`, `rx_voice ≥ 100` frames, 440 Hz **above the noise floor + AND ≥ 10× every other band**, and the RTP arrived **from rustisk's media IP**. - **bridge → A:** the caller received the bridge's distinct tone — - `voice_rx ≥ 100` frames, and **660 Hz is the dominant band** in the caller's - received audio. - -Distinct tones (caller 440 Hz, mock 660 Hz), each detected on the *far* side, -make each RTP direction independently attributable by frequency — a stronger -proof than an echo (an echo cannot distinguish a relayed frame from a socket -loop). Tone detection is a stdlib Goertzel filter, so **CI needs no numpy**. + `voice_rx ≥ 100` frames, 660 Hz **above the noise floor AND ≥ 10× every other + band**, and the RTP arrived **from rustisk's media IP**. + +Three independent things are asserted per direction, so a passing GRANTED case +cannot be faked: +1. **Distinct tones** (caller 440, mock 660), each detected only on the *far* + side — an echo or socket loop carries the wrong frequency and fails. +2. **Above-noise + 10× dominance** — the target band must clear a floor well + above the Goertzel noise band (~150-360) *and* beat every other band by 10×, + so silence / comfort-noise (which still counts as frames) can't slip through + on a coincidental band win. +3. **Relay-source check** — the received RTP source IP must equal rustisk's + media IP, proving rustisk *relayed* the audio rather than the legs going + peer-to-peer (defends against a `direct_media` regression). + +Tone detection is a stdlib Goertzel filter, so **CI needs no numpy**. ## Isolation (no collision with live / m0 / m9) @@ -74,7 +83,8 @@ Linux loopback. No docker, no NET_RAW, no root. Exit status **is** the result `target/e2e-trunk/`. Knobs: `RUSTISK_BIN`, `MIN_VOICE_FRAMES` (default 100), `MIN_TONE_RATIO` -(default 20), `KEEP_RUNTIME` (keep artifacts). +(default 5000 — an order of magnitude above the noise floor), `KEEP_RUNTIME` +(keep artifacts). ## Proven green — 2× consecutive (calibration runs, rustisk @ 461ad74) diff --git a/tests/e2e-trunk/chime_caller.py b/tests/e2e-trunk/chime_caller.py index bee2215..21eee07 100755 --- a/tests/e2e-trunk/chime_caller.py +++ b/tests/e2e-trunk/chime_caller.py @@ -218,12 +218,13 @@ def main(): stop = threading.Event() rx_frames = {"voice": 0, "other": 0, "bytes": 0} rx_pcm = [] # accumulated ONLY during the measurement window (post-bridge) + rtp_srcs = {} # source-IP histogram of measured voice RTP (proves who relayed) measure_open = threading.Event() def rx_loop(): while not stop.is_set(): try: - pkt, _ = rtpsock.recvfrom(4096) + pkt, addr = rtpsock.recvfrom(4096) except socket.timeout: continue if len(pkt) < 12: @@ -234,6 +235,7 @@ def rx_loop(): if pt == pcmu_pt and len(payload) > 0: rx_frames["voice"] += 1 if measure_open.is_set(): + rtp_srcs[addr[0]] = rtp_srcs.get(addr[0], 0) + 1 rx_pcm.extend(mulaw_to_linear(b) for b in payload) else: rx_frames["other"] += 1 @@ -320,6 +322,7 @@ def send_dtmf(digit): "other_rx": rx_frames["other"], "rx_bytes": rx_frames["bytes"], "rx_samples": len(rx_pcm), + "rtp_src": (max(rtp_srcs, key=rtp_srcs.get) if rtp_srcs else None), "tone_ratios": det, "detect_hz": args.detect, } diff --git a/tests/e2e-trunk/mock_bridge.py b/tests/e2e-trunk/mock_bridge.py index 989fe1b..b432cb3 100755 --- a/tests/e2e-trunk/mock_bridge.py +++ b/tests/e2e-trunk/mock_bridge.py @@ -292,13 +292,14 @@ def main(): stop_media = threading.Event() got_bye = threading.Event() rx_pcm = [] + rtp_srcs = {} # source-IP histogram of received voice RTP (proves who relayed) media_started = threading.Event() remote = {"ip": None, "port": None, "pcmu_pt": 0} def rx_loop(): while not stop_media.is_set(): try: - pkt, _ = rtpsock.recvfrom(4096) + pkt, addr = rtpsock.recvfrom(4096) except socket.timeout: continue if len(pkt) < 12: @@ -308,6 +309,7 @@ def rx_loop(): payload = pkt[12:] if pt == remote["pcmu_pt"] and payload: state["rx_voice"] += 1 + rtp_srcs[addr[0]] = rtp_srcs.get(addr[0], 0) + 1 rx_pcm.extend(mulaw_to_linear(b) for b in payload) def tx_loop(): @@ -315,7 +317,11 @@ def tx_loop(): ts = random.randint(0, 0x7FFFFFFF) ssrc = random.randint(0, 0x7FFFFFFF) idx = 0 - media_started.wait() + # Wait for the call to answer, but wake periodically so a REJECTED-case + # shutdown (no INVITE ever arrives) doesn't leave this thread parked. + while not media_started.wait(timeout=0.2): + if stop_media.is_set(): + return while not stop_media.is_set(): if remote["ip"]: pkt = rtp_hdr(remote["pcmu_pt"], seq, ts, ssrc) + tone_pcmu(args.tx_tone, idx) @@ -382,7 +388,7 @@ def tx_loop(): respond(sipsock, addr, "200 OK", msg) stop_media.set() - time.sleep(0.2) + time.sleep(0.3) # exceed the 0.25s rx socket timeout so rx_loop settles det = {} if len(rx_pcm) >= 1600: @@ -397,6 +403,7 @@ def tx_loop(): "rx_voice": state["rx_voice"], "rx_bytes": state["rx_bytes"], "rx_samples": len(rx_pcm), + "rtp_src": (max(rtp_srcs, key=rtp_srcs.get) if rtp_srcs else None), "tone_ratios": det, "detect_hz": args.detect, } diff --git a/tests/e2e-trunk/run.sh b/tests/e2e-trunk/run.sh index b85baea..e0371f7 100755 --- a/tests/e2e-trunk/run.sh +++ b/tests/e2e-trunk/run.sh @@ -51,8 +51,12 @@ CALLER_TONE=440 # caller -> bridge (proves A->bridge on the mock RX) BRIDGE_TONE=660 # bridge -> caller (proves bridge->A on the caller RX) # --- assertion thresholds (overridable) --- +# MIN_TONE_RATIO is deliberately set an order of magnitude ABOVE the observed +# Goertzel noise-band floor (~150-360) and ~3 orders BELOW a real relayed tone +# (~2e6), so a silence/comfort-noise "relay" can neither clear the floor nor the +# 10x dominance margin the GRANTED assertions also require. MIN_VOICE_FRAMES="${MIN_VOICE_FRAMES:-100}" -MIN_TONE_RATIO="${MIN_TONE_RATIO:-20}" +MIN_TONE_RATIO="${MIN_TONE_RATIO:-5000}" RUSTISK_PID="" MOCK_PID="" @@ -340,15 +344,17 @@ main() { assert_json "$grt_res" "GRANTED mock RX (A->bridge)" \ "d['invite_count'] == 1" \ "d['rx_voice'] >= $MIN_VOICE_FRAMES" \ + "d.get('rtp_src') == '$RK_IP'" \ "ratio($CALLER_TONE) >= $MIN_TONE_RATIO" \ - "ratio($CALLER_TONE) == max((ratio($CALLER_TONE), ratio($BRIDGE_TONE), ratio(350), ratio(880)))" \ + "ratio($CALLER_TONE) >= 10 * max((ratio($BRIDGE_TONE), ratio(350), ratio(880), 1))" \ || fail "GRANTED case: A->bridge media not proven on the mock" # bridge -> A: the caller received the bridge's distinct 660 Hz tone assert_json "$caller_res" "GRANTED caller RX (bridge->A)" \ "d.get('voice_rx',0) >= $MIN_VOICE_FRAMES" \ + "d.get('rtp_src') == '$RK_IP'" \ "ratio($BRIDGE_TONE) >= $MIN_TONE_RATIO" \ - "ratio($BRIDGE_TONE) == max((ratio($BRIDGE_TONE), ratio($CALLER_TONE), ratio(350), ratio(880)))" \ + "ratio($BRIDGE_TONE) >= 10 * max((ratio($CALLER_TONE), ratio(350), ratio(880), 1))" \ || fail "GRANTED case: bridge->A media not proven on the caller" log "GRANTED: PASS (two-way RTP: A->bridge 440 Hz + bridge->A 660 Hz, both directions)"