diff --git a/repeater/web/http_server.py b/repeater/web/http_server.py index 4a52cc0..0064215 100644 --- a/repeater/web/http_server.py +++ b/repeater/web/http_server.py @@ -353,6 +353,54 @@ class StatsApp: current = self._resolve_html_dir() return previous != current + _IMMUTABLE_CACHE = "public, max-age=31536000, immutable" + _REVALIDATE_CACHE = "no-cache" + # Vite names hashed files ``name-XXXXXXXX.ext``; a hash carries an uppercase letter or a + # digit, which a plain two-word name such as ``red-btn-down.svg`` does not. + _HASHED_NAME = re.compile(r"-([A-Za-z0-9_-]{8})\.[A-Za-z0-9]+$") + + @classmethod + def _is_hashed(cls, name: str) -> bool: + match = cls._HASHED_NAME.search(name) + return match is not None and re.search(r"[A-Z0-9]", match.group(1)) is not None + + @classmethod + def _send_static(cls, target: Path, *, immutable: bool) -> bytes: + """Send a file, its precompressed sibling when accepted, with cache headers and 304.""" + guessed_type, _ = mimetypes.guess_type(str(target)) + chosen, encoding = target, None + accepted = { + e.value.lower() + for e in cherrypy.request.headers.elements("Accept-Encoding") + if e.qvalue > 0 + } + for token, suffix in (("br", ".br"), ("gzip", ".gz")): + candidate = target.with_name(target.name + suffix) + if ( + token in accepted + and candidate.is_file() + and not candidate.is_symlink() + and candidate.stat().st_mtime >= target.stat().st_mtime + ): + chosen, encoding = candidate, token + break + + stat = chosen.stat() + etag = f'W/"{stat.st_size:x}-{int(stat.st_mtime):x}{"-" + encoding if encoding else ""}"' + headers = cherrypy.response.headers + headers["Content-Type"] = guessed_type or "application/octet-stream" + headers["Cache-Control"] = cls._IMMUTABLE_CACHE if immutable else cls._REVALIDATE_CACHE + headers["ETag"] = etag + headers["Vary"] = "Accept-Encoding" + if encoding: + headers["Content-Encoding"] = encoding + + if_none_match = str(cherrypy.request.headers.get("If-None-Match", "") or "") + if etag in [tag.strip() for tag in if_none_match.split(",")]: + cherrypy.response.status = 304 + return b"" + return chosen.read_bytes() + def _serve_static_file(self, root_dir: str, relative_parts: tuple[str, ...]): if not relative_parts: raise cherrypy.NotFound() @@ -360,9 +408,7 @@ class StatsApp: target = (root.joinpath(*relative_parts)).resolve() if not target.is_relative_to(root) or not target.is_file(): raise cherrypy.NotFound() - guessed_type, _ = mimetypes.guess_type(str(target)) - cherrypy.response.headers["Content-Type"] = guessed_type or "application/octet-stream" - return target.read_bytes() + return self._send_static(target, immutable=self._is_hashed(target.name)) def _serve_plugin_ui(self, plugin_id: str, relative_parts: tuple[str, ...]): """Serve static assets for an enabled application UI plugin.""" @@ -411,17 +457,12 @@ class StatsApp: raise cherrypy.HTTPError(400, "invalid path") if target.is_file(): - guessed_type, _ = mimetypes.guess_type(str(target)) - cherrypy.response.headers["Content-Type"] = ( - guessed_type or "application/octet-stream" - ) - return target.read_bytes() + return self._send_static(target, immutable=self._is_hashed(target.name)) # SPA fallback to entry HTML for client-side routes entry_target = safe_join(doc_root, entry_name) if entry_target.is_file(): - cherrypy.response.headers["Content-Type"] = "text/html; charset=utf-8" - return entry_target.read_bytes() + return self._send_static(entry_target, immutable=False) raise cherrypy.NotFound() except cherrypy.HTTPError: raise @@ -441,6 +482,7 @@ class StatsApp: self._resolve_html_dir() index_path = os.path.join(self.html_dir, "index.html") try: + cherrypy.response.headers["Cache-Control"] = self._REVALIDATE_CACHE with open(index_path, "r", encoding="utf-8") as f: return f.read() except FileNotFoundError: diff --git a/tests/test_web_static_caching.py b/tests/test_web_static_caching.py new file mode 100644 index 0000000..42ae772 --- /dev/null +++ b/tests/test_web_static_caching.py @@ -0,0 +1,123 @@ +import gzip +import os +import time +from pathlib import Path +from types import SimpleNamespace + +import cherrypy +import pytest +from cherrypy.lib.httputil import HeaderMap + +from repeater.web.http_server import StatsApp +from tests.test_plugin_static_security import plugin_ui # noqa: F401 + +JS = "index-DKwPA4Vw.js" +IMMUTABLE = "public, max-age=31536000, immutable" + + +@pytest.fixture +def http(monkeypatch): + request = SimpleNamespace(headers=HeaderMap()) + response = SimpleNamespace(headers={}, status=200) + monkeypatch.setattr(cherrypy, "request", request) + monkeypatch.setattr(cherrypy, "response", response) + return request, response + + +def _bundle(root: Path) -> Path: + assets = root / "assets" + assets.mkdir(parents=True) + (assets / JS).write_bytes(b"raw") + (assets / f"{JS}.br").write_bytes(b"BR") + (assets / f"{JS}.gz").write_bytes(gzip.compress(b"raw")) + (assets / "knob.svg").write_bytes(b"") + return assets + + +def _serve(assets, request, response, name=JS, **headers): + request.headers = HeaderMap(headers) + response.headers.clear() + response.status = 200 + return object.__new__(StatsApp)._serve_static_file(str(assets), (name,)) + + +def test_precompressed_siblings_by_accept_encoding(tmp_path, http): + request, response = http + assets = _bundle(tmp_path) + + assert _serve(assets, request, response, **{"Accept-Encoding": "gzip, deflate, br"}) == b"BR" + assert response.headers["Content-Encoding"] == "br" + assert response.headers["Content-Type"] == "text/javascript" + assert response.headers["Vary"] == "Accept-Encoding" + assert response.headers["Cache-Control"] == IMMUTABLE + assert response.headers["ETag"].endswith('-br"') + + body = _serve(assets, request, response, **{"Accept-Encoding": "gzip"}) + assert gzip.decompress(body) == b"raw" + assert response.headers["Content-Encoding"] == "gzip" + + assert _serve(assets, request, response, **{"Accept-Encoding": "br;q=0, gzip;q=0"}) == b"raw" + assert "Content-Encoding" not in response.headers + + +def test_only_hashed_names_are_immutable(tmp_path, http): + request, response = http + assets = _bundle(tmp_path) + _serve(assets, request, response, "knob.svg") + assert response.headers["Cache-Control"] == "no-cache" + assert StatsApp._is_hashed("login-bunny-CdOv7mzf.webm") + assert StatsApp._is_hashed("ChartCard-DvNDqE-i.js") + assert not StatsApp._is_hashed("red-btn-down.svg") + assert not StatsApp._is_hashed("index-DKwPA4Vw.js.map") + + +def test_bundled_frontend_matches_the_hash_rule(): + assets = Path(__file__).parents[1] / "repeater" / "web" / "html" / "assets" + if not assets.is_dir(): + pytest.skip("no bundled frontend") + built = [p.name for p in assets.iterdir() if p.suffix in (".js", ".css")] + assert built and all(StatsApp._is_hashed(name) for name in built) + + +def test_matching_etag_answers_304(tmp_path, http): + request, response = http + assets = _bundle(tmp_path) + _serve(assets, request, response, **{"Accept-Encoding": "br"}) + etag = response.headers["ETag"] + assert ( + _serve(assets, request, response, **{"Accept-Encoding": "br", "If-None-Match": etag}) == b"" + ) + assert response.status == 304 + assert ( + _serve(assets, request, response, **{"Accept-Encoding": "gzip", "If-None-Match": etag}) + != b"" + ) + assert response.status == 200 + + +def test_symlinked_or_stale_siblings_are_skipped(tmp_path, http): + request, response = http + assets = _bundle(tmp_path) + (assets / f"{JS}.br").unlink() + (tmp_path / "outside").write_bytes(b"OUTSIDE") + (assets / f"{JS}.br").symlink_to(tmp_path / "outside") + assert _serve(assets, request, response, **{"Accept-Encoding": "br, gzip"}) != b"OUTSIDE" + assert response.headers["Content-Encoding"] == "gzip" + old = time.time() - 3600 + os.utime(assets / f"{JS}.gz", (old, old)) + assert _serve(assets, request, response, **{"Accept-Encoding": "gzip"}) == b"raw" + assert "Content-Encoding" not in response.headers + + +def test_plugin_ui_takes_the_same_path(plugin_ui, http): # noqa: F811 + request, response = http + app, release, _manifest = plugin_ui + _bundle(release / "ui") + (release / "ui" / "index.html").write_text("ok") + request.headers["Accept-Encoding"] = "br" + assert app._serve_plugin_ui("audit.ui", ("assets", JS)) == b"BR" + assert response.headers["Cache-Control"] == IMMUTABLE + response.headers.clear() + assert b"ok" in app._serve_plugin_ui("audit.ui", ("messages", "companion")) + assert response.headers["Cache-Control"] == "no-cache" + assert response.headers["Content-Type"] == "text/html"