fix(server): k8s rollout readiness
ci / lint-prettier (push) Failing after 10s
ci / lint-ruff (push) Failing after 4s
ci / lint-yaml (push) Successful in 5s
ci / lint-dockerfiles (push) Successful in 5s
ci / validate (push) Successful in 5s
ci / lint-audit (push) Failing after 49s
ci / publish (push) Has been skipped

- TRUSTED_HOSTS env: configurable trusted hostnames for proxy-domain access (default stays strict: localhost/loopback/private IP); k8s manifest sets tg.workstation.internal (L-8 follow-up)
- Media allowlist +10: mkv/mk3d/heic/tgs/flv/3gp/ogv/asf/wmv/djvu (live disk has .tgs x44, .mkv x2)
- Cache buster: app.js?v=4 -> ?v=5 so browsers pick up the new bundle
- +6 tests (64 passing); REVIEW.md updated with live-cluster rollout notes
This commit is contained in:
2026-09-07 13:36:16 +02:00
parent bc2e93353a
commit 2a75537fb9
5 changed files with 126 additions and 2 deletions
+97
View File
@@ -977,6 +977,103 @@ class TestTrustedHost:
assert h3._check_same_origin() is True
class TestTrustedHostsEnv:
"""TRUSTED_HOSTS env: proxy-domain fix with strict default.
_TRUSTED_HOSTS_ENV is read at import, so tests patch the module
attribute directly (monkeypatch) instead of mutating os.environ.
"""
def _ws(self):
import webui_server as ws_module
return ws_module
def test_default_env_domain_untrusted(self, monkeypatch):
ws = self._ws()
monkeypatch.setattr(ws, "_TRUSTED_HOSTS_ENV", frozenset())
assert ws._is_trusted_host("tg.workstation.internal") is False
# regression guard: loopback/private literals still trusted
for host in ("localhost", "127.0.0.1", "10.0.0.5", "192.168.1.50"):
assert ws._is_trusted_host(host) is True, f"{host!r} should be trusted"
def test_configured_domain_trusted(self, monkeypatch):
ws = self._ws()
monkeypatch.setattr(ws, "_TRUSTED_HOSTS_ENV", {"tg.workstation.internal"})
assert ws._is_trusted_host("tg.workstation.internal") is True
assert ws._is_trusted_host("TG.WORKSTATION.INTERNAL") is True
assert ws._is_trusted_host("tg.workstation.internal.") is True
assert ws._is_trusted_host("unknown.example") is False
def test_check_same_origin_with_trusted_domain(self, monkeypatch):
ws = self._ws()
monkeypatch.setattr(ws, "_TRUSTED_HOSTS_ENV", {"tg.workstation.internal"})
h = _make_ws_handler(
headers={
"Host": "tg.workstation.internal",
"Origin": "https://tg.workstation.internal",
}
)
assert h._check_same_origin() is True
class TestExtendedMediaServing:
"""Extended media allowlist: .tgs/.mkv/... served, .exe/.ts still 403."""
def _make_full_media_handler(self, relative, data_dir):
import webui_server as ws_module
self._ws_orig_data = ws_module.DATA_DIR
ws_module.DATA_DIR = data_dir
handler = object.__new__(ws_module.TelegramScraperRequestHandler)
handler.headers = {}
handler.rfile = io.BytesIO()
handler.wfile = io.BytesIO()
handler.path = "/media/" + relative
handler.command = "GET"
handler.client_address = ("127.0.0.1", 4321)
handler.server = MagicMock()
handler.send_error_json = MagicMock()
handler.send_response = MagicMock()
handler.send_header = MagicMock()
handler.end_headers = MagicMock()
return handler
def _serve_and_status(self, relative):
import webui_server as ws_module
h = self._make_full_media_handler(relative, TEST_DATA)
try:
ws_module.DATA_DIR = TEST_DATA
h.serve_media(relative)
finally:
ws_module.DATA_DIR = self._ws_orig_data
if h.send_error_json.called:
return int(h.send_error_json.call_args[0][0])
return int(h.send_response.call_args[0][0])
def test_serves_tgs_and_mkv(self):
media_dir = TEST_DATA / "ext-media"
media_dir.mkdir(parents=True, exist_ok=True)
(media_dir / "sticker.tgs").write_bytes(b"\x1f\x8b\x08\x00")
(media_dir / "clip.mkv").write_bytes(b"\x1a\x45\xdf\xa3")
assert self._serve_and_status("ext-media/sticker.tgs") == 200
assert self._serve_and_status("ext-media/clip.mkv") == 200
def test_guess_media_kind_new_suffixes_fall_through_to_file(self):
import webui_server as ws_module
for name in (
"a.tgs", "a.mkv", "a.mk3d", "a.heic", "a.flv",
"a.3gp", "a.ogv", "a.asf", "a.wmv", "a.djvu",
):
assert ws_module.guess_media_kind(name, None) == "file", name
def test_exe_and_ts_still_forbidden(self):
media_dir = TEST_DATA / "ext-media"
media_dir.mkdir(parents=True, exist_ok=True)
(media_dir / "evil.exe").write_bytes(b"MZ")
(media_dir / "stream.ts").write_bytes(b"\x47" * 188)
assert self._serve_and_status("ext-media/evil.exe") == 403
assert self._serve_and_status("ext-media/stream.ts") == 403
class TestMediaServingLockdown:
"""M-1: /media/ must never serve state.json / *.db / *.session."""