400 lines
20 KiB
Python
400 lines
20 KiB
Python
"""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_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()
|