131 lines
4.6 KiB
Python
131 lines
4.6 KiB
Python
"""Security and transport tests for the tenant-local Telegram media handoff."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
from pathlib import Path
|
|
import threading
|
|
from urllib.error import HTTPError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
import pytest
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[2] / "services/hermes/scripts/telegram_media_server.py"
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def media_server_module():
|
|
"""Load the deployed script without requiring it to be a Python package."""
|
|
spec = importlib.util.spec_from_file_location("telegram_media_server", SCRIPT)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_normalize_media_path_allows_clean_paths_and_rejects_traversal(media_server_module, tmp_path):
|
|
root = tmp_path / "images"
|
|
root.mkdir()
|
|
selected, components = media_server_module.normalize_media_path(
|
|
f"{root}/./renders//portrait.png", (root,)
|
|
)
|
|
assert selected == root
|
|
assert components == ("renders", "portrait.png")
|
|
|
|
for unsafe in (
|
|
f"{root}/../secret.png",
|
|
f"{root}-other/portrait.png",
|
|
"relative.png",
|
|
f"{root}\\..\\secret.png",
|
|
):
|
|
with pytest.raises(media_server_module.MediaPathError):
|
|
media_server_module.normalize_media_path(unsafe, (root,))
|
|
|
|
|
|
def test_open_media_file_rejects_symlink_escape_and_missing_file(media_server_module, tmp_path):
|
|
root = tmp_path / "images"
|
|
root.mkdir()
|
|
outside = tmp_path / "private.png"
|
|
outside.write_bytes(b"private")
|
|
(root / "escape.png").symlink_to(outside)
|
|
|
|
with pytest.raises(OSError):
|
|
media_server_module.open_media_file(str(root / "escape.png"), (root,))
|
|
with pytest.raises(FileNotFoundError):
|
|
media_server_module.open_media_file(str(root / "missing.png"), (root,))
|
|
|
|
|
|
def test_open_media_file_rejects_symlink_in_allowed_root(media_server_module, tmp_path):
|
|
logical_parent = tmp_path / "tenant"
|
|
logical_parent.mkdir()
|
|
outside = tmp_path / "outside"
|
|
(outside / "images").mkdir(parents=True)
|
|
(outside / "images" / "private.png").write_bytes(b"private")
|
|
(logical_parent / "cache").symlink_to(outside)
|
|
logical_root = logical_parent / "cache" / "images"
|
|
|
|
with pytest.raises(OSError):
|
|
media_server_module.open_media_file(str(logical_root / "private.png"), (logical_root,))
|
|
|
|
|
|
def test_open_media_file_returns_only_regular_file_beneath_root(media_server_module, tmp_path):
|
|
root = tmp_path / "images"
|
|
nested = root / "generated"
|
|
nested.mkdir(parents=True)
|
|
expected = b"generated-image"
|
|
target = nested / "portrait.png"
|
|
target.write_bytes(expected)
|
|
|
|
descriptor, metadata, filename = media_server_module.open_media_file(str(target), (root,))
|
|
try:
|
|
assert os.read(descriptor, len(expected)) == expected
|
|
finally:
|
|
os.close(descriptor)
|
|
assert metadata.st_size == len(expected)
|
|
assert filename == "portrait.png"
|
|
|
|
|
|
def test_http_handoff_requires_relay_key_and_exact_tenant_slot(media_server_module, tmp_path):
|
|
root = tmp_path / "images"
|
|
root.mkdir()
|
|
target = root / "portrait.png"
|
|
target.write_bytes(b"generated-image")
|
|
server = media_server_module.TenantMediaServer(
|
|
("127.0.0.1", 0), "relay-secret", "3", (root,)
|
|
)
|
|
thread = threading.Thread(target=server.serve_forever, daemon=True)
|
|
thread.start()
|
|
url = f"http://127.0.0.1:{server.server_port}/media?{urlencode({'path': str(target)})}"
|
|
try:
|
|
for headers in (
|
|
{"Authorization": "Bearer wrong", "X-Hermes-Tenant-Slot": "3"},
|
|
{"Authorization": "Bearer relay-secret", "X-Hermes-Tenant-Slot": "2"},
|
|
):
|
|
with pytest.raises(HTTPError) as error:
|
|
urlopen(Request(url, headers=headers), timeout=2)
|
|
assert error.value.code == 401
|
|
|
|
request = Request(
|
|
url,
|
|
headers={
|
|
"Authorization": "Bearer relay-secret",
|
|
"X-Hermes-Tenant-Slot": "3",
|
|
},
|
|
)
|
|
with urlopen(request, timeout=2) as response:
|
|
assert response.read() == b"generated-image"
|
|
assert response.headers.get_content_type() == "image/png"
|
|
finally:
|
|
server.shutdown()
|
|
server.server_close()
|
|
thread.join(timeout=2)
|
|
|
|
|
|
def test_tenant_slot_requires_statefulset_ordinal(media_server_module):
|
|
assert media_server_module.tenant_slot("hermes-chat-tenant-7") == "7"
|
|
with pytest.raises(RuntimeError):
|
|
media_server_module.tenant_slot("hermes-chat-tenant")
|