"""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 RequestSizeLimitMiddleware class SecurityValidationTests(unittest.TestCase): 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_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_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_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_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()