"""Fast, dependency-light regression tests for P0 request protections.""" from __future__ import annotations import asyncio import os import tempfile import unittest os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-")) from fastapi import HTTPException from core.validation import _validate_character_record, _validate_http_url, _validated_upload_suffix from core.config import _load_settings, _save_settings from server import RequestCorrelationMiddleware, RequestSizeLimitMiddleware class SecurityValidationTests(unittest.TestCase): def test_api_request_correlation_adds_safe_id_header(self): async def endpoint(scope, receive, send): await send({"type": "http.response.start", "status": 204, "headers": []}) await send({"type": "http.response.body", "body": b""}) async def run(request_id): messages = [] scope = {"type": "http", "method": "GET", "path": "/api/diagnostics", "headers": [(b"x-request-id", request_id.encode())]} async def send(message): messages.append(message) await RequestCorrelationMiddleware(endpoint)(scope, lambda: None, send) return dict(messages[0]["headers"]) headers = asyncio.run(run("test-request-123")) self.assertEqual(headers[b"x-request-id"], b"test-request-123") generated = asyncio.run(run("bad\nrequest")) self.assertRegex(generated[b"x-request-id"].decode(), r"^[0-9a-f]{32}$") def test_character_record_rejects_missing_id_and_bad_sheet(self): for record in ( {"name": "Missing id", "sheet": {}}, {"id": "c1", "name": "Bad sheet", "sheet": []}, ): with self.assertRaises(HTTPException): _validate_character_record(record) def test_character_record_accepts_minimal_record(self): record = {"id": "c1", "book": "Book", "name": "Name", "tags": "", "sheet": {}} self.assertIs(_validate_character_record(record), record) def test_character_forms_are_bounded_and_validate_visual_fields(self): record = {"id": "c1", "book": "Book", "name": "Name", "sheet": { "forms": [{"id": "wolf", "name": "Wolf", "physical": "black wolf", "image": "data:image/png;base64,AA=="}], "active_form_id": "wolf", }} self.assertIs(_validate_character_record(record), record) with self.assertRaises(HTTPException): _validate_character_record({**record, "sheet": {"forms": [{}] * 9}}) def test_private_url_is_rejected_when_downloads_are_public_only(self): with self.assertRaises(HTTPException): _validate_http_url("http://127.0.0.1:8188/image", allow_private=False) def test_upload_filename_rejects_paths_and_control_characters(self): for filename in ("../voice.wav", r"folder\\voice.wav", "voice\x00.wav"): with self.assertRaises(HTTPException): _validated_upload_suffix(filename, default=".wav", allowed={".wav"}) self.assertEqual(_validated_upload_suffix("VOICE.WAV", default=".wav", allowed={".wav"}), ".wav") def test_stt_backend_aliases_and_response_shapes_are_normalized(self): from routes.stt import _clean_stt_backend, _transcription_text_from_response class ResponseStub: def __init__(self, payload, text=""): self._payload, self.text = payload, text def json(self): return self._payload self.assertEqual(_clean_stt_backend("faster-whisper"), "faster_whisper") self.assertEqual(_clean_stt_backend("groq"), "groq_whisper") self.assertEqual(_transcription_text_from_response(ResponseStub({"segments": [{"text": " one "}, {"text": "two"}]})), "one two") self.assertEqual(_transcription_text_from_response(ResponseStub(" plain text ")), "plain text") def test_provider_retry_policy_is_bounded_and_closes_discarded_responses(self): from unittest.mock import patch from core.retry import request_with_retries, retry_delay class ResponseStub: def __init__(self, status): self.status_code, self.closed = status, False def close(self): self.closed = True first, second, final = ResponseStub(503), ResponseStub(429), ResponseStub(200) with patch("core.retry.requests.request", side_effect=[first, second, final]) as request, \ patch("core.retry.time.sleep") as sleep: result = request_with_retries("POST", "http://provider", attempts=3) self.assertIs(result, final) self.assertTrue(first.closed) self.assertTrue(second.closed) self.assertEqual(request.call_count, 3) self.assertEqual(sleep.call_count, 2) self.assertLessEqual(retry_delay(20), 8.0) def test_request_size_middleware_rejects_oversized_json(self): sent = [] async def downstream(scope, receive, send): sent.append({"type": "downstream"}) async def send(message): sent.append(message) scope = { "type": "http", "method": "POST", "path": "/api/test", "headers": [(b"content-type", b"application/json"), (b"content-length", b"40000000")], } asyncio.run(RequestSizeLimitMiddleware(downstream)(scope, None, send)) self.assertEqual(sent[0]["status"], 413) self.assertNotIn({"type": "downstream"}, sent) def test_settings_never_returns_provider_credentials(self): from fastapi.testclient import TestClient from server import app settings = _load_settings() settings.update({ "llm_api_key": "provider-secret", "engine_api_keys": {"local": "engine-secret"}, }) _save_settings(settings) with TestClient(app) as client: response = client.get("/api/settings", headers={"origin": "http://testserver"}) self.assertEqual(response.status_code, 200) payload = response.json() self.assertEqual(payload.get("llm_api_key"), "__configured__") self.assertEqual(payload.get("engine_api_keys", {}).get("local"), "__configured__") self.assertIn("no-store", response.headers.get("cache-control", "")) self.assertNotIn("provider-secret", response.text) self.assertNotIn("engine-secret", response.text) def test_empty_masked_secret_does_not_overwrite_saved_value(self): from fastapi.testclient import TestClient from server import app settings = _load_settings() settings["llm_api_key"] = "keep-this-secret" _save_settings(settings) with TestClient(app) as client: response = client.post("/api/settings", json={"llm_api_key": ""}, headers={"origin": "http://testserver"}) self.assertEqual(response.status_code, 200) self.assertEqual(_load_settings().get("llm_api_key"), "keep-this-secret") def test_explicit_external_key_regeneration_is_not_cacheable(self): from fastapi.testclient import TestClient from server import app with TestClient(app) as client: response = client.post("/api/settings/regenerate-api-key", headers={"origin": "http://testserver"}) self.assertEqual(response.status_code, 200) self.assertTrue(response.json().get("external_api_key")) self.assertIn("no-store", response.headers.get("cache-control", "")) def test_transcription_background_job_reports_completion(self): from pathlib import Path from tempfile import TemporaryDirectory from unittest.mock import patch import routes.stt as stt with TemporaryDirectory() as directory: source = Path(directory) / "sample.wav" source.write_bytes(b"RIFF") job_id = "stt-test-job" now = 0.0 stt._STT_JOBS[job_id] = { "stage": "queued", "detail": "Queued", "started": now, "updated": now, "done": False, "cancel_requested": False, "cancelled": False, "error": None, "result": None, } try: with patch.object(stt, "_transcribe_audio", return_value=("hello", "test")): asyncio.run(stt._run_stt_job(job_id, source, {}, "configured")) self.assertTrue(stt._STT_JOBS[job_id]["done"]) self.assertEqual(stt._STT_JOBS[job_id]["result"]["text"], "hello") finally: stt._STT_JOBS.pop(job_id, None) def test_voice_design_background_job_honours_queued_cancellation(self): import routes.tts as tts from unittest.mock import patch job_id = "voice-design-test-job" now = 0.0 tts._VOICE_DESIGN_JOBS[job_id] = { "stage": "queued", "detail": "Queued", "created": now, "updated": now, "done": False, "cancel_requested": True, } try: with patch.object(tts, "_voice_design_execute") as execute: asyncio.run(tts._run_voice_design_job(job_id, {"instruct": "test"}, "", "")) self.assertTrue(tts._VOICE_DESIGN_JOBS[job_id]["done"]) self.assertEqual(tts._VOICE_DESIGN_JOBS[job_id]["stage"], "cancelled") execute.assert_not_called() finally: tts._VOICE_DESIGN_JOBS.pop(job_id, None) def test_visual_pair_review_detects_reused_assets(self): import base64 from unittest.mock import patch import routes.characters as characters from routes.characters import characters_visual_pair_review image = "data:image/png;base64," + base64.b64encode(b"same-image").decode() records = [ {"id": "c1", "name": "One", "image": image, "sheet": {"concept_art_image": image}}, {"id": "c2", "name": "Two", "image": image, "sheet": {"concept_art_image": ""}}, ] with patch.object(characters, "char_get_all", return_value=records): result = asyncio.run(characters_visual_pair_review()) self.assertFalse(result["ok"]) self.assertIn("same_asset_for_profile_and_reference", result["issues"][0]["issues"]) self.assertIn("asset_reused_across_characters", result["issues"][0]["issues"]) def test_visual_pair_review_detects_replaced_current_assets(self): import base64 from unittest.mock import patch import routes.characters as characters from routes.characters import characters_visual_pair_review profile = "data:image/png;base64," + base64.b64encode(b"new-profile").decode() reference = "data:image/png;base64," + base64.b64encode(b"new-reference").decode() records = [{"id": "changed", "name": "Changed", "image": profile, "sheet": { "concept_art_image": reference, "consistent_prompt_version": "v3", "consistent_identity_fingerprint": "", "profile_asset_digest": "old-profile", "reference_asset_digest": "old-reference", }}] with patch.object(characters, "char_get_all", return_value=records): result = asyncio.run(characters_visual_pair_review()) issues = result["issues"][0]["issues"] self.assertIn("profile_asset_changed_since_generation", issues) self.assertIn("reference_asset_changed_since_generation", issues) def test_cast_background_job_honours_queued_cancellation(self): import routes.conversation as conversation from unittest.mock import patch job_id = "cast-test-job" now = 0.0 conversation._CAST_JOBS[job_id] = { "job_id": job_id, "stage": "queued", "detail": "Queued", "started": now, "updated": now, "done": False, "cancel_requested": True, "cancelled": False, } try: with patch.object(conversation, "match_characters_voices") as match: asyncio.run(conversation._run_cast_job(job_id, {}, {})) self.assertTrue(conversation._CAST_JOBS[job_id]["done"]) self.assertEqual(conversation._CAST_JOBS[job_id]["stage"], "cancelled") match.assert_not_called() finally: conversation._CAST_JOBS.pop(job_id, None) def test_live_retry_metadata_is_available_without_exposing_input(self): import routes.tts as tts import routes.conversation as conversation from routes.jobs import _public_job voice_id = "retry-voice-test" cast_id = "retry-cast-test" tts._VOICE_DESIGN_JOBS[voice_id] = {"job_id": voice_id, "kind": "voice_design", "done": True, "error": "temporary", "retry_data": {"instruct": "private prompt"}} conversation._CAST_JOBS[cast_id] = {"job_id": cast_id, "kind": "casting", "done": True, "error": "temporary", "retry_data": {"script": "private source"}} try: self.assertNotIn("retry_data", tts._voice_design_public_job(tts._VOICE_DESIGN_JOBS[voice_id])) public_voice = _public_job({"job_id": voice_id, "kind": "voice_design", "done": True, "error": "temporary", "created": 0}) self.assertTrue(public_voice["retry_available"]) public_cast = _public_job({"job_id": cast_id, "kind": "casting", "done": True, "error": "temporary", "created": 0}) self.assertTrue(public_cast["retry_available"]) finally: tts._VOICE_DESIGN_JOBS.pop(voice_id, None) conversation._CAST_JOBS.pop(cast_id, None) def test_failed_live_jobs_dispatch_safe_retries(self): from unittest.mock import AsyncMock, patch import routes.tts as tts import routes.conversation as conversation voice_id, cast_id = "dispatch-voice-retry", "dispatch-cast-retry" tts._VOICE_DESIGN_JOBS[voice_id] = {"job_id": voice_id, "done": True, "error": "temporary", "retry_data": {"instruct": "voice"}} conversation._CAST_JOBS[cast_id] = {"job_id": cast_id, "done": True, "error": "temporary", "retry_data": {"names": ["A"], "voices": [{"id": "v1"}]}} try: with patch.object(tts, "voice_design", new=AsyncMock(return_value={"job_id": "new-voice"})) as voice_route: self.assertEqual(asyncio.run(tts.voice_design_retry(voice_id)), {"job_id": "new-voice"}) self.assertTrue(voice_route.await_args.args[0]._data["background"]) with patch.object(conversation, "match_characters_voices", new=AsyncMock(return_value={"job_id": "new-cast"})) as cast_route: self.assertEqual(asyncio.run(conversation.match_characters_voices_retry(cast_id)), {"job_id": "new-cast"}) self.assertTrue(cast_route.await_args.args[0]._data["background"]) finally: tts._VOICE_DESIGN_JOBS.pop(voice_id, None) conversation._CAST_JOBS.pop(cast_id, None) def test_background_job_metadata_round_trips_without_large_payloads(self): from core.database import background_job_create, background_job_get, background_job_update job_id = "persistent-job-test" background_job_create(job_id, "test", fingerprint="fingerprint") try: background_job_update(job_id, stage="done", detail="Done", done=True, result={"text": "must not persist", "id": "result-id", "image": "data:image/png;base64,huge"}) job = background_job_get(job_id) self.assertTrue(job["done"]) self.assertEqual(job["result"], {"id": "result-id"}) self.assertEqual(job["fingerprint"], "fingerprint") finally: from core.database import _open with _open() as conn: conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) conn.commit() def test_background_job_artifact_is_atomic_bounded_and_job_scoped(self): from core.database import background_job_artifact_path, background_job_create, background_job_store_artifact, background_job_update job_id = "artifact-job-test" background_job_create(job_id, "test") try: meta = background_job_store_artifact(job_id, "result.wav", b"RIFF-test", "audio/wav") self.assertEqual(meta["name"], "result.wav") path = background_job_artifact_path(job_id, "result.wav") self.assertIsNotNone(path) self.assertEqual(path.read_bytes(), b"RIFF-test") background_job_update(job_id, stage="done", done=True, result={"artifact": meta}) from routes.jobs import jobs_artifact, jobs_get self.assertEqual(asyncio.run(jobs_get(job_id))["result"]["artifact"]["url"], f"/api/jobs/{job_id}/artifacts/result.wav") self.assertEqual(asyncio.run(jobs_artifact(job_id, "result.wav")).path, path) self.assertIsNone(background_job_artifact_path(job_id, "../result.wav")) with self.assertRaises(ValueError): background_job_store_artifact(job_id, "bad.bin", b"x" * (50 * 1024 * 1024 + 1)) finally: from core.database import _open, _BACKGROUND_ARTIFACT_DIR with _open() as conn: conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) conn.commit() import shutil shutil.rmtree(_BACKGROUND_ARTIFACT_DIR / job_id, ignore_errors=True) def test_background_job_persistence_prunes_old_completed_rows(self): from core.database import _open, background_job_create, background_job_update prefix = "prune-job-" try: for index in range(260): job_id = prefix + str(index) background_job_create(job_id, "test") background_job_update(job_id, stage="done", done=True) with _open() as conn: count = conn.execute("SELECT COUNT(*) FROM background_jobs WHERE job_id LIKE ?", (prefix + "%",)).fetchone()[0] self.assertLessEqual(count, 256) finally: with _open() as conn: conn.execute("DELETE FROM background_jobs WHERE job_id LIKE ?", (prefix + "%",)) conn.commit() def test_job_history_endpoint_is_bounded_and_redacts_fingerprints(self): from core.database import background_job_create, background_job_update from routes.jobs import jobs_get, jobs_list job_id = "public-job-test" background_job_create(job_id, "test", fingerprint="private-fingerprint") background_job_update(job_id, stage="done", done=True, result={"id": "safe"}) try: single = asyncio.run(jobs_get(job_id)) listing = asyncio.run(jobs_list(limit=1000)) self.assertNotIn("fingerprint", single) self.assertEqual(single["result"], {"id": "safe"}) self.assertLessEqual(len(listing["jobs"]), 100) finally: from core.database import _open with _open() as conn: conn.execute("DELETE FROM background_jobs WHERE job_id=?", (job_id,)) conn.commit() def test_fish_cache_eviction_is_bounded(self): from pathlib import Path from tempfile import TemporaryDirectory import routes.sources as sources with TemporaryDirectory() as directory: cache = Path(directory) old_count, old_bytes = sources._FISH_CACHE_MAX_FILES, sources._FISH_CACHE_MAX_BYTES try: sources._FISH_CACHE_MAX_FILES = 2 sources._FISH_CACHE_MAX_BYTES = 1024 for index in range(4): (cache / f"{index}.json").write_text("{}") sources._evict_fish_cache(cache) self.assertLessEqual(len(list(cache.glob("*.json"))), 2) finally: sources._FISH_CACHE_MAX_FILES, sources._FISH_CACHE_MAX_BYTES = old_count, old_bytes def test_docker_client_accepts_network_proxy_without_socket_mount(self): from unittest.mock import patch from core.docker_client import _docker_available with patch.dict(os.environ, {"DOCKER_HOST": "tcp://docker-socket-proxy:2375", "DOCKER_SOCKET": "/missing/socket"}): self.assertTrue(_docker_available()) def test_generation_history_is_bounded_and_does_not_require_prompt_text(self): from core.database import generation_history_add, generation_history_list generation_history_add( kind="test", target_id="target-1", provider="test-provider", result={"ok": True, "secret": "must-not-be-added"}, ) rows = generation_history_list("target-1", limit=1) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["kind"], "test") self.assertEqual(rows[0]["prompt"], "") self.assertNotIn("must-not-be-added", rows[0]["result"]) def test_line_audio_cache_eviction_is_bounded(self): from pathlib import Path from tempfile import TemporaryDirectory import routes.tts as tts with TemporaryDirectory() as directory: old_dir, old_count, old_bytes = tts._LINE_AUDIO_DIR, tts._LINE_AUDIO_MAX_FILES, tts._LINE_AUDIO_MAX_BYTES try: tts._LINE_AUDIO_DIR = Path(directory) tts._LINE_AUDIO_MAX_FILES = 2 tts._LINE_AUDIO_MAX_BYTES = 10 book = Path(directory) / "book" book.mkdir() for index in range(4): (book / f"{index:032x}.wav").write_bytes(b"1234") tts._evict_line_audio_cache() files = list(book.glob("*.wav")) self.assertLessEqual(len(files), 2) self.assertLessEqual(sum(path.stat().st_size for path in files), 10) finally: tts._LINE_AUDIO_DIR, tts._LINE_AUDIO_MAX_FILES, tts._LINE_AUDIO_MAX_BYTES = old_dir, old_count, old_bytes if __name__ == "__main__": unittest.main()