125 lines
5.4 KiB
Python
125 lines
5.4 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_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_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.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_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"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|