6801272f51
Maintainer review changes on the Edge TTS PR: - edge-tts was added as a HARD base requirement, but the /api/tts handler is designed optional (ImportError returns 503). Moved it out of requirements.txt with an OPTIONAL comment + install hint; updated the 503 message from the nonexistent 'see docs' to the actual install command. Keeps minimal base deps. - PR shipped NO tests for a new auth+rate-limited+allowlisted network endpoint. Added tests/test_issue2931_edge_tts_endpoint.py covering method (405), missing text (400), over-length (400), voice allowlist (400), per-client rate limit (429) — all in-process, no real synthesis/network.
97 lines
3.1 KiB
Python
97 lines
3.1 KiB
Python
"""Validation + security-path coverage for the Edge TTS endpoint (#2931).
|
|
|
|
These exercise the guard rails of _handle_tts (method, input cap, voice
|
|
allowlist, rate limiting) in-process via a fake handler — no network and no
|
|
real edge-tts synthesis required, since every rejection happens before the
|
|
edge_tts import / Communicate call.
|
|
"""
|
|
import io
|
|
import json
|
|
|
|
import api.routes as routes
|
|
|
|
|
|
class _FakeHandler:
|
|
def __init__(self, body: bytes, command: str = "POST", headers=None, client="1.2.3.4"):
|
|
self.command = command
|
|
self.rfile = io.BytesIO(body)
|
|
self.wfile = io.BytesIO()
|
|
self.headers = headers or {}
|
|
self.headers.setdefault("Content-Length", str(len(body)))
|
|
self.client_address = (client, 12345)
|
|
self.status = None
|
|
self.sent_headers = {}
|
|
|
|
def send_response(self, status):
|
|
self.status = status
|
|
|
|
def send_header(self, key, value):
|
|
self.sent_headers[key] = value
|
|
|
|
def end_headers(self):
|
|
pass
|
|
|
|
def payload(self):
|
|
try:
|
|
return json.loads(self.wfile.getvalue().decode("utf-8"))
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _post(body_dict, **kw):
|
|
body = json.dumps(body_dict).encode()
|
|
return _FakeHandler(body, **kw)
|
|
|
|
|
|
def _reset_limiter():
|
|
# Drop any limiter state carried between tests so rate-limit assertions are
|
|
# deterministic regardless of run order.
|
|
if hasattr(routes._handle_tts, "_tts_limiter"):
|
|
del routes._handle_tts._tts_limiter
|
|
|
|
|
|
def test_tts_requires_post():
|
|
_reset_limiter()
|
|
h = _post({"text": "hello"}, command="GET")
|
|
routes._handle_tts(h, None)
|
|
assert h.status == 405
|
|
|
|
|
|
def test_tts_requires_text():
|
|
_reset_limiter()
|
|
h = _post({"text": " "})
|
|
routes._handle_tts(h, None)
|
|
assert h.status == 400
|
|
assert "text is required" in (h.payload() or {}).get("error", "")
|
|
|
|
|
|
def test_tts_rejects_overlong_text():
|
|
_reset_limiter()
|
|
h = _post({"text": "x" * 5001})
|
|
routes._handle_tts(h, None)
|
|
assert h.status == 400
|
|
assert "too long" in (h.payload() or {}).get("error", "")
|
|
|
|
|
|
def test_tts_rejects_unknown_voice():
|
|
_reset_limiter()
|
|
h = _post({"text": "hello", "voice": "evil-voice-injection"}, client="5.6.7.8")
|
|
routes._handle_tts(h, None)
|
|
assert h.status == 400
|
|
assert "invalid voice" in (h.payload() or {}).get("error", "")
|
|
|
|
|
|
def test_tts_rate_limits_second_immediate_request():
|
|
_reset_limiter()
|
|
# The limiter runs (and records the client) BEFORE the voice allowlist and
|
|
# before any edge-tts synthesis. Use an invalid voice so the first request
|
|
# still registers with the limiter but returns at the allowlist (400) without
|
|
# making a real network call; the second immediate request from the SAME
|
|
# client is then throttled (429).
|
|
h1 = _post({"text": "hello", "voice": "not-a-real-voice"}, client="9.9.9.9")
|
|
routes._handle_tts(h1, None)
|
|
assert h1.status == 400 # rejected at allowlist, limiter recorded the client
|
|
h2 = _post({"text": "hello", "voice": "not-a-real-voice"}, client="9.9.9.9")
|
|
routes._handle_tts(h2, None)
|
|
assert h2.status == 429
|
|
_reset_limiter()
|