Files
hold-slayer/tests/lab/sounds/generate.py
Robert Helewka c00cf02676
All checks were successful
CVE Scan & Docker Build / security-scan (push) Successful in 45s
CVE Scan & Docker Build / build-and-push (push) Successful in 2m12s
test(lab): add Asterisk lab — a fake PSTN for media validation
An Asterisk instance that answers calls, plays an IVR, holds with music and
connects a "human", so the gateway has something real to dial that is not the
PSTN: no charges, no strangers, no E911 exposure.

Asterisk rather than Kamailio because the unproven risks are media risks.
Kamailio is a proxy — it routes signalling and answers nothing, so it would
forward the INVITE and find nobody home. Asterisk is a B2BUA: it answers,
plays prompts and collects DTMF, which is the hold-slayer scenario itself.
Kamailio remains the better model for trunk registration/digest auth later.

No application changes are needed to use it. SIP_TRUNK_HOST is just an
address, so the production code path runs unmodified — there is no test-only
branch anywhere in the gateway. It also means safety is structural: while the
trunk points at the lab there is no route to the PSTN at all, an absence of
route rather than a policy that could be misconfigured.

Nine scenarios (1001-1008 plus an echo test) cover the baseline call, the
IVR/DTMF path, hold-then-human, long hold, busy, no-answer, remote hangup and
silence.

The image ships no sound files, so sounds/generate.py synthesises three
fixtures from fixed seeds — byte-identical on every run, which is what makes
a classifier regression distinguishable from noise. Verified against
AudioClassifier: music→MUSIC 0.85, speech→LIVE_HUMAN 0.75,
silence→SILENCE 1.00. The speech formants deliberately avoid the DTMF bands;
the first version landed on a valid pair and classified as a keypress.

Anonymous inbound calls are refused, and endpoint matching is by source
address — Asterisk's default matches the From-header domain, which Hold
Slayer populates from its SIP bind address (0.0.0.0 on a wildcard bind).

Generated audio and the rendered per-host configs are gitignored: the former
is reproducible from a fixed seed, the latter carry a host-specific IP and
the lab password.

Known limit, documented in the README: MediaPipeline.create_tap is a stub, so
the classifier receives no audio on a live call. RTP flows and Asterisk plays
audio, but the tap is never fed — the fixture results above were measured by
feeding the classifier directly. This blocks scenarios 1002/1003/1004.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-07-29 06:06:14 -04:00

114 lines
4.7 KiB
Python

#!/usr/bin/env python3
"""Generate the lab's audio fixtures.
The Asterisk container ships with no sound files, and the design calls for
*deterministic* audio: real hold music varies per call, so a classifier
regression on the PSTN is indistinguishable from noise. These are synthesised
from a fixed seed, so every run classifies identical input.
Output is 8 kHz 16-bit mono signed-linear (.sln), which Asterisk plays without
transcoding — the format is implied by the extension, so `Playback(lab-music)`
finds `lab-music.sln`.
python generate.py [outdir]
"""
import struct
import sys
from pathlib import Path
import numpy as np
RATE = 8000 # Asterisk's native rate for ulaw/alaw telephony
def _write_sln(path: Path, samples: np.ndarray) -> None:
"""Write float samples in [-1, 1] as 16-bit signed little-endian PCM."""
clipped = np.clip(samples, -1.0, 1.0)
pcm = (clipped * 32767).astype("<i2")
path.write_bytes(pcm.tobytes())
print(f" {path.name}: {len(pcm) / RATE:.1f}s ({path.stat().st_size} bytes)")
def make_music(seconds: float = 30.0) -> np.ndarray:
"""Sustained multi-harmonic tones — what the classifier must call MUSIC.
A chord progression with stable pitch and strong harmonic structure. The
steady spectrum across a long window is what distinguishes music from
speech; this deliberately has no pauses.
"""
t = np.linspace(0, seconds, int(RATE * seconds), endpoint=False)
# A-minor-ish progression, one chord per 2s bar.
chords = [(220.0, 261.6, 329.6), (196.0, 246.9, 293.7),
(174.6, 220.0, 261.6), (196.0, 246.9, 329.6)]
out = np.zeros_like(t)
bar = 2.0
for i, chord in enumerate(chords * int(np.ceil(seconds / (bar * len(chords))))):
start, end = i * bar, (i + 1) * bar
if start >= seconds:
break
mask = (t >= start) & (t < end)
for j, freq in enumerate(chord):
# Fundamental plus two harmonics, decaying — a plucked-string feel.
for h, amp in ((1, 0.30), (2, 0.12), (3, 0.05)):
out[mask] += amp / (j + 1) * np.sin(2 * np.pi * freq * h * t[mask])
# Gentle per-bar envelope so bars are distinguishable but never silent.
env = 0.8 + 0.2 * np.sin(2 * np.pi * (t[mask] - start) / bar)
out[mask] *= env
return out * 0.45
def make_speech(seconds: float = 8.0, seed: int = 1337) -> np.ndarray:
"""Formant-like bursts with pauses — what the classifier must call SPEECH.
Not real speech, but it carries the features the classifier keys on: a
fundamental in the human range, shifting formants, and syllable-rate
amplitude modulation with genuine silence between utterances.
"""
rng = np.random.default_rng(seed)
t = np.linspace(0, seconds, int(RATE * seconds), endpoint=False)
out = np.zeros_like(t)
pos = 0.3 # leading pause
while pos < seconds - 0.4:
syl = rng.uniform(0.12, 0.28) # syllable length
mask = (t >= pos) & (t < pos + syl)
if mask.any():
local = t[mask] - pos
f0 = rng.uniform(95, 165) # fundamental — adult speaking range
# Two formants, swept slightly across the syllable. The ranges
# deliberately avoid the DTMF bands (rows 697-941, columns
# 1209-1633): a formant pair landing on both trips the Goertzel
# detector and the whole utterance is classified as a keypress.
f1 = rng.uniform(300, 620) + rng.uniform(-40, 40) * local / syl
f2 = rng.uniform(1750, 2600) + rng.uniform(-120, 120) * local / syl
sig = (0.50 * np.sin(2 * np.pi * f0 * local)
+ 0.30 * np.sin(2 * np.pi * f1 * local)
+ 0.18 * np.sin(2 * np.pi * f2 * local))
# Raised-cosine envelope: no clicks at syllable edges.
sig *= np.sin(np.pi * local / syl) ** 0.6
out[mask] += sig
# Inter-syllable gap; occasionally a longer between-word pause.
pos += syl + (rng.uniform(0.25, 0.5) if rng.random() < 0.25
else rng.uniform(0.04, 0.12))
return out * 0.55
def make_silence(seconds: float = 5.0) -> np.ndarray:
"""Near-silence with a trace of noise — real lines are never digitally flat."""
rng = np.random.default_rng(4242)
return rng.normal(0, 0.0006, int(RATE * seconds))
def main() -> None:
outdir = Path(sys.argv[1] if len(sys.argv) > 1 else Path(__file__).parent)
outdir.mkdir(parents=True, exist_ok=True)
print(f"Generating lab audio into {outdir}/")
_write_sln(outdir / "lab-music.sln", make_music())
_write_sln(outdir / "lab-speech.sln", make_speech())
_write_sln(outdir / "lab-silence.sln", make_silence())
print("Done. Deterministic: same bytes on every run.")
if __name__ == "__main__":
main()