652 lines
24 KiB
Python
652 lines
24 KiB
Python
"""Fail-closed tests for the G7 authenticated public voice soak runner."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import importlib.util
|
|
import json
|
|
import math
|
|
import struct
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
RUNNER_PATH = Path(__file__).with_name("soak-public-voice-websocket.py")
|
|
|
|
|
|
def load_runner():
|
|
spec = importlib.util.spec_from_file_location("g7_external_voice_soak", RUNNER_PATH)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError("G7 public voice soak runner could not be loaded")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
class FakeWebSocket:
|
|
def __init__(self, frames: list[str | bytes]) -> None:
|
|
self.frames = list(frames)
|
|
self.sent: list[str | bytes] = []
|
|
|
|
async def send(self, payload: str | bytes) -> None:
|
|
self.sent.append(payload)
|
|
|
|
async def recv(self) -> str | bytes:
|
|
await asyncio.sleep(0)
|
|
return self.frames.pop(0)
|
|
|
|
async def close(self) -> None:
|
|
return None
|
|
|
|
|
|
class G7ExternalVoiceSoakTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls) -> None:
|
|
cls.runner = load_runner()
|
|
|
|
def test_default_duration_preserves_the_common_window_margin(self) -> None:
|
|
args = self.runner.parser().parse_args(
|
|
[
|
|
"--expected-stt-provider",
|
|
"deepgram",
|
|
"--expected-stt-model",
|
|
"nova-3",
|
|
"--expected-tts-provider",
|
|
"commercial-tts",
|
|
"--expected-tts-model",
|
|
"persona-v1",
|
|
"--microphone-device",
|
|
"Physical Microphone",
|
|
]
|
|
)
|
|
self.assertEqual(3_120.0, args.duration_seconds)
|
|
self.assertFalse(args.preflight_only)
|
|
self.assertFalse(args.confirm_physical_capture)
|
|
|
|
def test_full_soak_rejects_a_duration_without_overlap_margin(self) -> None:
|
|
cli = self.runner.parser()
|
|
args = cli.parse_args(
|
|
[
|
|
"--expected-stt-provider",
|
|
"local_whisper",
|
|
"--expected-stt-model",
|
|
"small",
|
|
"--expected-tts-provider",
|
|
"melotts",
|
|
"--expected-tts-model",
|
|
"melotts-korean",
|
|
"--microphone-device",
|
|
"Physical Microphone",
|
|
"--duration-seconds",
|
|
"3000",
|
|
]
|
|
)
|
|
with (
|
|
patch.object(cli, "error", side_effect=ValueError) as error,
|
|
self.assertRaises(ValueError),
|
|
):
|
|
self.runner.validate_args(args, cli)
|
|
error.assert_called_once_with(
|
|
"--duration-seconds must be >= 3120 for a production physical soak"
|
|
)
|
|
|
|
def test_public_url_injects_session_without_changing_target_metadata(self) -> None:
|
|
raw_session_id = "private-session-id"
|
|
url = self.runner.with_session_id(
|
|
"wss://api-vignette.chanpaca.net/voice/ws?source=a",
|
|
raw_session_id,
|
|
)
|
|
self.assertIn("session_id=private-session-id", url)
|
|
self.assertEqual(
|
|
("api-vignette.chanpaca.net", "/voice/ws"),
|
|
self.runner.target_metadata(url),
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence(
|
|
target_host="api-vignette.chanpaca.net",
|
|
target_path="/voice/ws",
|
|
cookie_present=True,
|
|
session_id_present=True,
|
|
session_id_sha256=self.runner.sha256_text(raw_session_id),
|
|
)
|
|
serialized = self.runner.evidence_json(evidence)
|
|
self.assertNotIn(raw_session_id, serialized)
|
|
|
|
def test_pcm_metrics_detects_signal_without_retaining_audio(self) -> None:
|
|
samples = [int(12_000 * math.sin(index / 5)) for index in range(16_000)]
|
|
audio = struct.pack(f"<{len(samples)}h", *samples)
|
|
metrics = self.runner.pcm_metrics(audio)
|
|
self.assertEqual(32_000, metrics.bytes_captured)
|
|
self.assertEqual(1.0, metrics.duration_seconds)
|
|
self.assertGreater(metrics.rms_dbfs, -20)
|
|
self.assertGreater(metrics.peak_dbfs, metrics.rms_dbfs)
|
|
|
|
def test_full_soak_blockers_are_explicit_and_stable(self) -> None:
|
|
blockers = self.runner.external_blockers(
|
|
cookie_present=False,
|
|
session_id_present=False,
|
|
device_enumerated=True,
|
|
capture_confirmed=False,
|
|
microphone_rms_dbfs=None,
|
|
minimum_rms_dbfs=-65.0,
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
"authenticated_session_cookie_missing",
|
|
"owned_voice_session_id_missing",
|
|
"physical_microphone_capture_not_confirmed",
|
|
],
|
|
blockers,
|
|
)
|
|
|
|
def test_full_soak_never_enumerates_or_captures_without_confirmation(self) -> None:
|
|
args = self.runner.parser().parse_args(
|
|
[
|
|
"--expected-stt-provider",
|
|
"deepgram",
|
|
"--expected-stt-model",
|
|
"nova-3",
|
|
"--expected-tts-provider",
|
|
"commercial-tts",
|
|
"--expected-tts-model",
|
|
"persona-v1",
|
|
"--microphone-device",
|
|
"Physical Microphone",
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
with (
|
|
patch.object(
|
|
self.runner,
|
|
"dshow_audio_devices",
|
|
side_effect=AssertionError("confirmation must precede enumeration"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"capture_microphone_pcm",
|
|
side_effect=AssertionError("confirmation must precede capture"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"open_authenticated",
|
|
new=AsyncMock(),
|
|
) as open_authenticated,
|
|
):
|
|
with self.assertRaises(self.runner.GateBlocked) as caught:
|
|
asyncio.run(
|
|
self.runner.run_soak(
|
|
args,
|
|
evidence,
|
|
cookie_value="opaque-cookie",
|
|
session_id="owned-session",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
["physical_microphone_capture_not_confirmed"],
|
|
caught.exception.blockers,
|
|
)
|
|
open_authenticated.assert_not_awaited()
|
|
|
|
def test_ready_metadata_is_allowlisted_recorded_and_exactly_validated(self) -> None:
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
ready = {
|
|
"type": "ready",
|
|
"session_id": "must-not-be-recorded",
|
|
"state": "idle",
|
|
"voice": "must-not-be-recorded",
|
|
"stt_provider": "deepgram",
|
|
"stt_model": "nova-3",
|
|
"tts_provider": "commercial-tts",
|
|
"tts_model": "persona-v1",
|
|
}
|
|
|
|
self.runner.validate_ready_provider_metadata(
|
|
ready,
|
|
evidence,
|
|
expected_stt_provider="deepgram",
|
|
expected_stt_model="nova-3",
|
|
expected_tts_provider="commercial-tts",
|
|
expected_tts_model="persona-v1",
|
|
)
|
|
|
|
self.assertTrue(evidence.ready_provider_metadata_validated)
|
|
self.assertEqual("deepgram", evidence.ready_stt_provider)
|
|
self.assertEqual("nova-3", evidence.ready_stt_model)
|
|
self.assertEqual("commercial-tts", evidence.ready_tts_provider)
|
|
self.assertEqual("persona-v1", evidence.ready_tts_model)
|
|
serialized = self.runner.evidence_json(evidence)
|
|
self.assertNotIn("must-not-be-recorded", serialized)
|
|
|
|
def test_ready_metadata_missing_or_mismatch_fails_without_echoing_values(
|
|
self,
|
|
) -> None:
|
|
complete = {
|
|
"stt_provider": "unexpected-private-provider",
|
|
"stt_model": "nova-3",
|
|
"tts_provider": "commercial-tts",
|
|
"tts_model": "persona-v1",
|
|
}
|
|
with self.assertRaises(self.runner.GateFailure) as mismatch:
|
|
self.runner.validate_ready_provider_metadata(
|
|
complete,
|
|
self.runner.ExternalSoakEvidence(),
|
|
expected_stt_provider="deepgram",
|
|
expected_stt_model="nova-3",
|
|
expected_tts_provider="commercial-tts",
|
|
expected_tts_model="persona-v1",
|
|
)
|
|
self.assertEqual(
|
|
"authenticated_public_wss_stt_provider_mismatch",
|
|
str(mismatch.exception),
|
|
)
|
|
self.assertNotIn("unexpected-private-provider", str(mismatch.exception))
|
|
|
|
missing = dict(complete)
|
|
missing.pop("tts_model")
|
|
with self.assertRaises(self.runner.GateFailure) as absent:
|
|
self.runner.validate_ready_provider_metadata(
|
|
missing,
|
|
self.runner.ExternalSoakEvidence(),
|
|
expected_stt_provider="unexpected-private-provider",
|
|
expected_stt_model="nova-3",
|
|
expected_tts_provider="commercial-tts",
|
|
expected_tts_model="persona-v1",
|
|
)
|
|
self.assertEqual(
|
|
"authenticated_public_wss_ready_metadata_missing:tts_model",
|
|
str(absent.exception),
|
|
)
|
|
|
|
def test_authenticated_open_validates_ready_provider_metadata(self) -> None:
|
|
secret_session_id = "private-owned-session"
|
|
websocket = FakeWebSocket(
|
|
[
|
|
json.dumps(
|
|
{
|
|
"type": "ready",
|
|
"session_id": secret_session_id,
|
|
"state": "idle",
|
|
"voice": "private-voice-preset",
|
|
"stt_provider": "deepgram",
|
|
"stt_model": "nova-3",
|
|
"tts_provider": "commercial-tts",
|
|
"tts_model": "persona-v1",
|
|
}
|
|
)
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
with patch.object(
|
|
self.runner,
|
|
"connect",
|
|
new=AsyncMock(return_value=websocket),
|
|
):
|
|
result = asyncio.run(
|
|
self.runner.open_authenticated(
|
|
wss_url="wss://api-vignette.chanpaca.net/voice/ws",
|
|
origin="https://vignette.chanpaca.net",
|
|
session_id=secret_session_id,
|
|
cookie_name="__Host-vignette_sid",
|
|
cookie_value="opaque-cookie",
|
|
timeout_seconds=1,
|
|
evidence=evidence,
|
|
expected_stt_provider="deepgram",
|
|
expected_stt_model="nova-3",
|
|
expected_tts_provider="commercial-tts",
|
|
expected_tts_model="persona-v1",
|
|
)
|
|
)
|
|
|
|
self.assertIs(websocket, result)
|
|
self.assertTrue(evidence.authenticated_public_wss_ready)
|
|
self.assertTrue(evidence.ready_provider_metadata_validated)
|
|
serialized = self.runner.evidence_json(evidence)
|
|
self.assertNotIn(secret_session_id, serialized)
|
|
self.assertNotIn("private-voice-preset", serialized)
|
|
|
|
def test_preflight_never_enumerates_or_opens_microphone(self) -> None:
|
|
args = self.runner.parser().parse_args(
|
|
[
|
|
"--preflight-only",
|
|
"--expected-stt-provider",
|
|
"deepgram",
|
|
"--expected-stt-model",
|
|
"nova-3",
|
|
"--expected-tts-provider",
|
|
"commercial-tts",
|
|
"--expected-tts-model",
|
|
"persona-v1",
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
websocket = FakeWebSocket([])
|
|
|
|
with (
|
|
patch.object(
|
|
self.runner,
|
|
"unauthenticated_route_probe",
|
|
new=AsyncMock(return_value=None),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"dshow_audio_devices",
|
|
side_effect=AssertionError("preflight must not enumerate microphones"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"capture_microphone_pcm",
|
|
side_effect=AssertionError("microphone must remain unopened"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"open_authenticated",
|
|
new=AsyncMock(return_value=websocket),
|
|
) as open_authenticated,
|
|
patch.object(
|
|
self.runner,
|
|
"application_ping",
|
|
new=AsyncMock(return_value=None),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"expect_close_code",
|
|
new=AsyncMock(return_value=1000),
|
|
),
|
|
):
|
|
asyncio.run(
|
|
self.runner.run_preflight(
|
|
args,
|
|
evidence,
|
|
cookie_value="opaque-cookie",
|
|
session_id="owned-session",
|
|
)
|
|
)
|
|
|
|
open_authenticated.assert_awaited_once()
|
|
call = open_authenticated.await_args.kwargs
|
|
self.assertEqual("deepgram", call["expected_stt_provider"])
|
|
self.assertEqual("nova-3", call["expected_stt_model"])
|
|
self.assertFalse(evidence.microphone_device_enumerated)
|
|
self.assertFalse(evidence.physical_capture_confirmed)
|
|
self.assertFalse(evidence.physical_microphone_used)
|
|
|
|
def test_preflight_missing_auth_never_checks_microphone(self) -> None:
|
|
args = self.runner.parser().parse_args(
|
|
[
|
|
"--preflight-only",
|
|
"--expected-stt-provider",
|
|
"deepgram",
|
|
"--expected-stt-model",
|
|
"nova-3",
|
|
"--expected-tts-provider",
|
|
"commercial-tts",
|
|
"--expected-tts-model",
|
|
"persona-v1",
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
|
|
with (
|
|
patch.object(
|
|
self.runner,
|
|
"unauthenticated_route_probe",
|
|
new=AsyncMock(return_value=None),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"dshow_audio_devices",
|
|
side_effect=AssertionError("preflight must not enumerate microphones"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"capture_microphone_pcm",
|
|
side_effect=AssertionError("microphone must remain unopened"),
|
|
),
|
|
patch.object(
|
|
self.runner,
|
|
"open_authenticated",
|
|
new=AsyncMock(),
|
|
) as open_authenticated,
|
|
):
|
|
with self.assertRaises(self.runner.GateBlocked) as caught:
|
|
asyncio.run(
|
|
self.runner.run_preflight(
|
|
args,
|
|
evidence,
|
|
cookie_value="",
|
|
session_id="",
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
[
|
|
"authenticated_session_cookie_missing",
|
|
"owned_voice_session_id_missing",
|
|
],
|
|
caught.exception.blockers,
|
|
)
|
|
open_authenticated.assert_not_awaited()
|
|
|
|
def test_turn_counts_contract_frames_but_never_records_text(self) -> None:
|
|
interim_transcript_secret = "private interim learner transcript"
|
|
final_transcript_secret = "private final learner transcript"
|
|
reply_secret = "private client reply"
|
|
frames: list[str | bytes] = [
|
|
json.dumps({"type": "state", "state": "listening"}),
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": interim_transcript_secret,
|
|
"final": False,
|
|
"speech_final": False,
|
|
}
|
|
),
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": final_transcript_secret,
|
|
"final": True,
|
|
"speech_final": True,
|
|
}
|
|
),
|
|
json.dumps({"type": "reply", "text": reply_secret}),
|
|
b"tts-bytes",
|
|
json.dumps({"type": "tts_end", "chunks": 1}),
|
|
json.dumps({"type": "state", "state": "idle"}),
|
|
]
|
|
websocket = FakeWebSocket(frames)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
asyncio.run(
|
|
self.runner.run_audio_turn(
|
|
websocket,
|
|
b"\x00\x00" * 32,
|
|
evidence,
|
|
timeout_seconds=1,
|
|
)
|
|
)
|
|
self.assertEqual(1, evidence.turns_succeeded)
|
|
self.assertEqual("vignette.g7-public-voice-soak.v4", evidence.schema_version)
|
|
self.assertEqual(2, evidence.transcript_frames)
|
|
self.assertEqual(1, evidence.interim_transcript_frames)
|
|
self.assertEqual(1, evidence.speech_final_transcript_frames)
|
|
self.assertEqual(1, evidence.reply_frames)
|
|
self.assertEqual(1, evidence.tts_binary_frames)
|
|
self.assertEqual(1, len(evidence.turn_transcript_metrics))
|
|
turn_metrics = evidence.turn_transcript_metrics[0]
|
|
self.assertEqual(1, turn_metrics.turn_number)
|
|
self.assertEqual(1, turn_metrics.interim_transcript_frames)
|
|
self.assertEqual(1, turn_metrics.speech_final_transcript_frames)
|
|
self.assertIsNotNone(turn_metrics.first_interim_latency_ms)
|
|
self.assertIsNotNone(turn_metrics.speech_final_latency_ms)
|
|
self.assertEqual(
|
|
turn_metrics.first_interim_latency_ms,
|
|
evidence.first_interim_latency_ms_min,
|
|
)
|
|
self.assertEqual(
|
|
turn_metrics.first_interim_latency_ms,
|
|
evidence.first_interim_latency_ms_max,
|
|
)
|
|
self.assertEqual(
|
|
turn_metrics.speech_final_latency_ms,
|
|
evidence.speech_final_latency_ms_min,
|
|
)
|
|
self.assertEqual(
|
|
turn_metrics.speech_final_latency_ms,
|
|
evidence.speech_final_latency_ms_max,
|
|
)
|
|
serialized = self.runner.evidence_json(evidence)
|
|
self.assertNotIn(interim_transcript_secret, serialized)
|
|
self.assertNotIn(final_transcript_secret, serialized)
|
|
self.assertNotIn(reply_secret, serialized)
|
|
|
|
def test_turn_fails_closed_when_interim_transcript_is_missing(self) -> None:
|
|
websocket = FakeWebSocket(
|
|
[
|
|
json.dumps({"type": "state", "state": "listening"}),
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": "private final transcript",
|
|
"final": True,
|
|
"speech_final": True,
|
|
}
|
|
),
|
|
json.dumps({"type": "reply", "text": "private reply"}),
|
|
b"tts-bytes",
|
|
json.dumps({"type": "tts_end"}),
|
|
json.dumps({"type": "state", "state": "idle"}),
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
|
|
with self.assertRaises(self.runner.GateFailure) as caught:
|
|
asyncio.run(
|
|
self.runner.run_audio_turn(
|
|
websocket,
|
|
b"\x00\x00" * 32,
|
|
evidence,
|
|
timeout_seconds=1,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
"public_wss_interim_transcript_missing",
|
|
str(caught.exception),
|
|
)
|
|
self.assertEqual(0, evidence.interim_transcript_frames)
|
|
self.assertEqual(1, evidence.speech_final_transcript_frames)
|
|
self.assertEqual(0, evidence.turns_succeeded)
|
|
|
|
def test_turn_fails_closed_when_speech_final_transcript_is_missing(self) -> None:
|
|
websocket = FakeWebSocket(
|
|
[
|
|
json.dumps({"type": "state", "state": "listening"}),
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": "private interim transcript",
|
|
"final": False,
|
|
"speech_final": False,
|
|
}
|
|
),
|
|
json.dumps({"type": "reply", "text": "private reply"}),
|
|
b"tts-bytes",
|
|
json.dumps({"type": "tts_end"}),
|
|
json.dumps({"type": "state", "state": "idle"}),
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
|
|
with self.assertRaises(self.runner.GateFailure) as caught:
|
|
asyncio.run(
|
|
self.runner.run_audio_turn(
|
|
websocket,
|
|
b"\x00\x00" * 32,
|
|
evidence,
|
|
timeout_seconds=1,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
"public_wss_speech_final_transcript_missing",
|
|
str(caught.exception),
|
|
)
|
|
self.assertEqual(1, evidence.interim_transcript_frames)
|
|
self.assertEqual(0, evidence.speech_final_transcript_frames)
|
|
self.assertEqual(0, evidence.turns_succeeded)
|
|
|
|
def test_turn_fails_closed_when_speech_final_transcript_is_duplicated(
|
|
self,
|
|
) -> None:
|
|
final_frames = [
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": f"private final transcript {index}",
|
|
"final": True,
|
|
"speech_final": True,
|
|
}
|
|
)
|
|
for index in range(2)
|
|
]
|
|
websocket = FakeWebSocket(
|
|
[
|
|
json.dumps({"type": "state", "state": "listening"}),
|
|
json.dumps(
|
|
{
|
|
"type": "transcript",
|
|
"text": "private interim transcript",
|
|
"final": False,
|
|
"speech_final": False,
|
|
}
|
|
),
|
|
*final_frames,
|
|
json.dumps({"type": "reply", "text": "private reply"}),
|
|
b"tts-bytes",
|
|
json.dumps({"type": "tts_end"}),
|
|
json.dumps({"type": "state", "state": "idle"}),
|
|
]
|
|
)
|
|
evidence = self.runner.ExternalSoakEvidence()
|
|
|
|
with self.assertRaises(self.runner.GateFailure) as caught:
|
|
asyncio.run(
|
|
self.runner.run_audio_turn(
|
|
websocket,
|
|
b"\x00\x00" * 32,
|
|
evidence,
|
|
timeout_seconds=1,
|
|
)
|
|
)
|
|
|
|
self.assertEqual(
|
|
"public_wss_speech_final_transcript_duplicate",
|
|
str(caught.exception),
|
|
)
|
|
self.assertEqual(1, evidence.interim_transcript_frames)
|
|
self.assertEqual(2, evidence.speech_final_transcript_frames)
|
|
self.assertIsNone(evidence.turn_transcript_metrics[0].speech_final_latency_ms)
|
|
self.assertEqual(0, evidence.turns_succeeded)
|
|
|
|
def test_error_frame_discards_server_detail(self) -> None:
|
|
secret_detail = "private provider message"
|
|
websocket = FakeWebSocket(
|
|
[json.dumps({"type": "error", "code": "consent", "detail": secret_detail})]
|
|
)
|
|
with self.assertRaises(self.runner.GateFailure) as caught:
|
|
asyncio.run(
|
|
self.runner.receive_until(
|
|
websocket,
|
|
lambda _raw, _payload: False,
|
|
timeout_seconds=1,
|
|
)
|
|
)
|
|
self.assertEqual("public_wss_error_frame:consent", str(caught.exception))
|
|
self.assertNotIn(secret_detail, str(caught.exception))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|