refactor(#2695): centralize WebUI static asset resolution

This commit is contained in:
Rod Boev committed 2026-07-06 06:45:36 -04:00
1 parent f80e641a9a
commit 624e03c60a
7 files changed
+187 -56

No files matched your search

+11 -1
View File
@@ -7598,7 +7598,17 @@ def get_available_models_for_session_visit() -> dict:
# ── Static file path ───────────────────────────────────────────────────────── # ── Static file path ─────────────────────────────────────────────────────────
_INDEX_HTML_PATH = REPO_ROOT / "static" / "index.html"
def get_static_root() -> Path:
return REPO_ROOT / "static"
def get_index_html_path() -> Path:
return get_static_root() / "index.html"
_INDEX_HTML_PATH = get_index_html_path()
# ── Thread synchronisation ─────────────────────────────────────────────────── # ── Thread synchronisation ───────────────────────────────────────────────────
LOCK = threading.Lock() LOCK = threading.Lock()
+9 -8
View File
@@ -2779,7 +2779,6 @@ from api.config import (
STREAM_LAST_EVENT_ID, STREAM_LAST_EVENT_ID,
SERVER_START_TIME, SERVER_START_TIME,
_resolve_cli_toolsets, _resolve_cli_toolsets,
_INDEX_HTML_PATH,
get_available_models, get_available_models,
get_available_models_for_session_visit, get_available_models_for_session_visit,
_provider_is_known_or_configured, _provider_is_known_or_configured,
@@ -2818,6 +2817,7 @@ from api.config import (
_cfg_lock, _cfg_lock,
PENDING_BG_TASK_COMPLETIONS, PENDING_BG_TASK_COMPLETIONS,
) )
from api import config as api_config
from api.helpers import ( from api.helpers import (
require, require,
bad, bad,
@@ -11075,7 +11075,7 @@ def _serve_manifest(handler) -> bool:
routes so Firefox Android can fetch the manifest when installing from routes so Firefox Android can fetch the manifest when installing from
a /session/<id> page. See #2226. a /session/<id> page. See #2226.
""" """
static_root = Path(__file__).parent.parent / "static" static_root = api_config.get_static_root()
manifest_path = (static_root / "manifest.json").resolve() manifest_path = (static_root / "manifest.json").resolve()
if manifest_path.exists(): if manifest_path.exists():
data = manifest_path.read_bytes() data = manifest_path.read_bytes()
@@ -11136,8 +11136,9 @@ def _render_index_shell_base() -> str:
""" """
from api.updates import WEBUI_VERSION from api.updates import WEBUI_VERSION
st = _INDEX_HTML_PATH.stat() index_path = api_config.get_index_html_path()
sig = (st.st_size, st.st_mtime_ns) st = index_path.stat()
sig = (index_path, st.st_size, st.st_mtime_ns)
with _INDEX_SHELL_CACHE_LOCK: with _INDEX_SHELL_CACHE_LOCK:
cached = _INDEX_SHELL_CACHE.get("base") cached = _INDEX_SHELL_CACHE.get("base")
if cached and cached[0] == sig: if cached and cached[0] == sig:
@@ -11146,7 +11147,7 @@ def _render_index_shell_base() -> str:
version_token = quote(WEBUI_VERSION, safe="") version_token = quote(WEBUI_VERSION, safe="")
base = ( base = (
_INDEX_HTML_PATH.read_text(encoding="utf-8") index_path.read_text(encoding="utf-8")
.replace("__WEBUI_VERSION__", version_token) .replace("__WEBUI_VERSION__", version_token)
.replace("__MAX_UPLOAD_BYTES__", str(MAX_UPLOAD_BYTES)) .replace("__MAX_UPLOAD_BYTES__", str(MAX_UPLOAD_BYTES))
) )
@@ -11320,7 +11321,7 @@ def handle_get(handler, parsed) -> bool:
return _serve_manifest(handler) return _serve_manifest(handler)
if parsed.path == "/sw.js": if parsed.path == "/sw.js":
static_root = Path(__file__).parent.parent / "static" static_root = api_config.get_static_root()
sw_path = (static_root / "sw.js").resolve() sw_path = (static_root / "sw.js").resolve()
if sw_path.exists(): if sw_path.exists():
# Inject the current git-derived version as the cache name so the # Inject the current git-derived version as the cache name so the
@@ -11343,7 +11344,7 @@ def handle_get(handler, parsed) -> bool:
return j(handler, {"error": "not found"}, status=404) return j(handler, {"error": "not found"}, status=404)
if parsed.path == "/favicon.ico": if parsed.path == "/favicon.ico":
static_root = Path(__file__).parent.parent / "static" static_root = api_config.get_static_root()
ico_path = (static_root / "favicon.ico").resolve() ico_path = (static_root / "favicon.ico").resolve()
if ico_path.exists() and ico_path.is_file(): if ico_path.exists() and ico_path.is_file():
data = ico_path.read_bytes() data = ico_path.read_bytes()
@@ -15438,7 +15439,7 @@ _STATIC_CACHE_LOCK = threading.Lock()
def _serve_static(handler, parsed): def _serve_static(handler, parsed):
static_root = (Path(__file__).parent.parent / "static").resolve() static_root = api_config.get_static_root().resolve()
# Strip the leading '/static/' prefix, then resolve and sandbox # Strip the leading '/static/' prefix, then resolve and sandbox
rel = parsed.path[len("/static/") :] rel = parsed.path[len("/static/") :]
static_file = (static_root / rel).resolve() static_file = (static_root / rel).resolve()
+7 -1
View File
@@ -6,6 +6,7 @@ their normal JSON error behavior; this only pins the shell route contract.
""" """
from urllib.parse import urlparse from urllib.parse import urlparse
from types import SimpleNamespace
class _FakeHandler: class _FakeHandler:
@@ -36,14 +37,19 @@ class _FakeHandler:
class _BrokenIndexPath: class _BrokenIndexPath:
def stat(self):
return SimpleNamespace(st_size=1, st_mtime_ns=1)
def read_text(self, *args, **kwargs): def read_text(self, *args, **kwargs):
raise RuntimeError("simulated index.html read failure") raise RuntimeError("simulated index.html read failure")
def test_home_route_internal_error_returns_html_503_not_json(monkeypatch): def test_home_route_internal_error_returns_html_503_not_json(monkeypatch):
from api import config as api_config
from api import routes from api import routes
monkeypatch.setattr(routes, "_INDEX_HTML_PATH", _BrokenIndexPath()) monkeypatch.setattr(api_config, "get_index_html_path", lambda: _BrokenIndexPath())
monkeypatch.setattr(routes, "_INDEX_SHELL_CACHE", {})
handler = _FakeHandler() handler = _FakeHandler()
assert routes.handle_get(handler, urlparse("http://example.com/")) is True assert routes.handle_get(handler, urlparse("http://example.com/")) is True
+25 -3
View File
@@ -17,14 +17,16 @@ import os
import time import time
from urllib.parse import quote from urllib.parse import quote
import api.config as api_config
import api.routes as routes import api.routes as routes
from api.updates import WEBUI_VERSION from api.updates import WEBUI_VERSION
def _old_inline_render(csrf_token: str) -> str: def _old_inline_render(csrf_token: str) -> str:
"""Reproduce the pre-cache inline render exactly, for equivalence checks.""" """Reproduce the pre-cache inline render exactly, for equivalence checks."""
index_path = api_config.get_index_html_path()
return ( return (
routes._INDEX_HTML_PATH.read_text(encoding="utf-8") index_path.read_text(encoding="utf-8")
.replace("__WEBUI_VERSION__", quote(WEBUI_VERSION, safe="")) .replace("__WEBUI_VERSION__", quote(WEBUI_VERSION, safe=""))
.replace("__MAX_UPLOAD_BYTES__", str(routes.MAX_UPLOAD_BYTES)) .replace("__MAX_UPLOAD_BYTES__", str(routes.MAX_UPLOAD_BYTES))
.replace("__CSRF_TOKEN_JSON__", json.dumps(csrf_token)) .replace("__CSRF_TOKEN_JSON__", json.dumps(csrf_token))
@@ -69,10 +71,10 @@ def test_second_call_returns_cached_object():
def test_cache_invalidates_on_mtime_change(tmp_path, monkeypatch): def test_cache_invalidates_on_mtime_change(tmp_path, monkeypatch):
# Point the module at a temp copy so we can mutate its mtime safely. # Point the module at a temp copy so we can mutate its mtime safely.
src = routes._INDEX_HTML_PATH.read_text(encoding="utf-8") src = api_config.get_index_html_path().read_text(encoding="utf-8")
fake = tmp_path / "index.html" fake = tmp_path / "index.html"
fake.write_text(src, encoding="utf-8") fake.write_text(src, encoding="utf-8")
monkeypatch.setattr(routes, "_INDEX_HTML_PATH", fake) monkeypatch.setattr(api_config, "get_index_html_path", lambda: fake)
# Reset the shared cache so this test is order-independent. # Reset the shared cache so this test is order-independent.
monkeypatch.setattr(routes, "_INDEX_SHELL_CACHE", {}) monkeypatch.setattr(routes, "_INDEX_SHELL_CACHE", {})
@@ -85,3 +87,23 @@ def test_cache_invalidates_on_mtime_change(tmp_path, monkeypatch):
sig_after = routes._INDEX_SHELL_CACHE["base"][0] sig_after = routes._INDEX_SHELL_CACHE["base"][0]
assert sig_after != sig_before assert sig_after != sig_before
def test_cache_invalidates_on_index_path_change(tmp_path, monkeypatch):
first = tmp_path / "first.html"
second = tmp_path / "second.html"
first.write_text("alpha __WEBUI_VERSION__ __MAX_UPLOAD_BYTES__ __CSRF_TOKEN_JSON__", encoding="utf-8")
second.write_text("bravo __WEBUI_VERSION__ __MAX_UPLOAD_BYTES__ __CSRF_TOKEN_JSON__", encoding="utf-8")
stamp = time.time() + 5
os.utime(first, (stamp, stamp))
os.utime(second, (stamp, stamp))
selected = {"path": first}
monkeypatch.setattr(api_config, "get_index_html_path", lambda: selected["path"])
monkeypatch.setattr(routes, "_INDEX_SHELL_CACHE", {})
assert "alpha" in routes._render_index_shell_base()
selected["path"] = second
rendered = routes._render_index_shell_base()
assert "bravo" in rendered
assert "alpha" not in rendered
+2 -1
View File
@@ -6,6 +6,7 @@ import time
from types import SimpleNamespace from types import SimpleNamespace
import api.auth as auth import api.auth as auth
import api.config as api_config
import api.routes as routes import api.routes as routes
@@ -152,7 +153,7 @@ def test_login_route_remains_csrf_exempt(monkeypatch):
def test_index_shell_includes_csrf_fetch_and_sendbeacon_injection(): def test_index_shell_includes_csrf_fetch_and_sendbeacon_injection():
src = (routes._INDEX_HTML_PATH).read_text(encoding="utf-8") src = api_config.get_index_html_path().read_text(encoding="utf-8")
assert "csrfToken:__CSRF_TOKEN_JSON__" in src assert "csrfToken:__CSRF_TOKEN_JSON__" in src
assert "X-Hermes-CSRF-Token" in src assert "X-Hermes-CSRF-Token" in src
@@ -22,6 +22,8 @@ import gzip
from types import SimpleNamespace from types import SimpleNamespace
from urllib.parse import urlparse from urllib.parse import urlparse
import api.config as api_config
class _FakeHandler: class _FakeHandler:
"""Minimal request handler stand-in matching tests/test_session_static_assets.py.""" """Minimal request handler stand-in matching tests/test_session_static_assets.py."""
@@ -66,24 +68,6 @@ def _serve(routes, path, query="", request_headers=None):
return h return h
def _patch_static_root(monkeypatch, static_root):
"""Force _serve_static to read from a temp directory and clear its cache."""
from api import routes
monkeypatch.setattr(
routes, "_serve_static",
lambda handler, parsed, _root=static_root, _orig=routes._serve_static: _orig(handler, parsed),
)
# Tests redirect by writing files to the real static dir's parent layout
# via a fixture; instead we monkeypatch the module-level Path computation.
# _serve_static derives static_root from `Path(__file__).parent.parent / "static"`,
# so we monkeypatch __file__ via a closure that re-resolves with our temp tree.
# Simpler: patch the cache and call the real function with a parsed path that
# resolves under the real static dir. We use the fixture below instead.
# ── Fixture: build a tiny isolated static tree and rebind paths ───────────
import pytest import pytest
@@ -100,30 +84,7 @@ def isolated_static(tmp_path, monkeypatch):
# Patch the cache so cross-test state cannot leak. # Patch the cache so cross-test state cannot leak.
monkeypatch.setattr(routes, "_STATIC_CACHE", {}, raising=True) monkeypatch.setattr(routes, "_STATIC_CACHE", {}, raising=True)
monkeypatch.setattr(api_config, "get_static_root", lambda: static_root)
# _serve_static derives static_root from Path(__file__).parent.parent.
# Rebind by monkeypatching Path resolution: we wrap the function so the
# caller-visible signature is unchanged.
original = routes._serve_static
def wrapped(handler, parsed):
# Trick: temporarily monkeypatch Path so the function sees our temp tree.
import api.routes as ar
orig_file = ar.__file__
# Place a sentinel api/routes.py "next to" tmp_path so the relative
# walk lands in our static_root.
fake_api_dir = tmp_path / "api"
fake_api_dir.mkdir(exist_ok=True)
fake_routes = fake_api_dir / "routes.py"
if not fake_routes.exists():
fake_routes.write_text("# stub for path resolution\n")
monkeypatch.setattr(ar, "__file__", str(fake_routes))
try:
return original(handler, parsed)
finally:
monkeypatch.setattr(ar, "__file__", orig_file)
monkeypatch.setattr(routes, "_serve_static", wrapped)
yield static_root yield static_root
+130
View File
@@ -0,0 +1,130 @@
from __future__ import annotations
import json
from pathlib import Path
from types import SimpleNamespace
from urllib.parse import quote
import api.config as api_config
import api.routes as routes
from api.updates import WEBUI_VERSION
ROOT = Path(__file__).resolve().parent.parent
class _FakeHandler:
def __init__(self, request_headers=None):
self.status = None
self.sent_headers = []
self.body = bytearray()
self.headers = dict(request_headers or {})
self.wfile = self
def send_response(self, status):
self.status = status
def send_header(self, name, value):
self.sent_headers.append((name, value))
def end_headers(self):
pass
def write(self, data):
self.body.extend(data)
def header(self, name):
for key, value in self.sent_headers:
if key.lower() == name.lower():
return value
return None
def _get(path, request_headers=None):
handler = _FakeHandler(request_headers)
routes.handle_get(handler, SimpleNamespace(path=path, query=""))
return handler
def test_config_owner_returns_checkout_static_root():
static_root = ROOT / "static"
assert api_config.get_static_root() == static_root
assert api_config.get_index_html_path() == static_root / "index.html"
def test_manifest_routes_follow_selected_static_root(tmp_path, monkeypatch):
static_root = tmp_path / "static"
static_root.mkdir()
manifest_path = static_root / "manifest.json"
payload = json.dumps({"name": "temp", "display": "standalone"}).encode("utf-8")
manifest_path.write_bytes(payload)
monkeypatch.setattr(api_config, "get_static_root", lambda: static_root)
handler = _get("/manifest.json")
assert handler.status == 200
assert handler.header("Content-Type") == "application/manifest+json; charset=utf-8"
assert handler.header("Cache-Control") == "no-store"
assert bytes(handler.body) == payload
session_handler = _get("/session/manifest.webmanifest")
assert session_handler.status == 200
assert bytes(session_handler.body) == payload
def test_service_worker_and_favicon_follow_selected_static_root(tmp_path, monkeypatch):
static_root = tmp_path / "static"
static_root.mkdir()
sw_path = static_root / "sw.js"
sw_path.write_text("const version = '__WEBUI_VERSION__';\n", encoding="utf-8")
favicon_path = static_root / "favicon.ico"
favicon_path.write_bytes(b"favicon-bytes")
monkeypatch.setattr(api_config, "get_static_root", lambda: static_root)
sw_handler = _get("/sw.js")
expected = sw_path.read_text(encoding="utf-8").replace(
"__WEBUI_VERSION__", quote(WEBUI_VERSION, safe="")
).encode("utf-8")
assert sw_handler.status == 200
assert sw_handler.header("Service-Worker-Allowed") == "/"
assert sw_handler.header("Cache-Control") == "no-store"
assert bytes(sw_handler.body) == expected
favicon_handler = _get("/favicon.ico")
assert favicon_handler.status == 200
assert favicon_handler.header("Content-Type") == "image/x-icon"
assert bytes(favicon_handler.body) == b"favicon-bytes"
favicon_path.unlink()
missing_favicon_handler = _get("/favicon.ico")
assert missing_favicon_handler.status == 204
def test_index_shell_and_static_route_use_selected_root(tmp_path, monkeypatch):
static_root = tmp_path / "static"
static_root.mkdir()
index_path = static_root / "index.html"
index_path.write_bytes(
b"<html>__WEBUI_VERSION__ __MAX_UPLOAD_BYTES__ __CSRF_TOKEN_JSON__ temp</html>"
)
ui_path = static_root / "ui.js"
ui_path.write_bytes(b"console.log('temp static');\n")
monkeypatch.setattr(api_config, "get_static_root", lambda: static_root)
monkeypatch.setattr(api_config, "get_index_html_path", lambda: index_path)
monkeypatch.setattr(routes, "_INDEX_SHELL_CACHE", {})
monkeypatch.setattr(routes, "_STATIC_CACHE", {})
shell = routes._render_index_shell_base()
assert "temp" in shell
assert "__WEBUI_VERSION__" not in shell
assert "__MAX_UPLOAD_BYTES__" not in shell
assert "__CSRF_TOKEN_JSON__" in shell
temp_static = _get("/static/ui.js")
assert temp_static.status == 200
assert bytes(temp_static.body) == b"console.log('temp static');\n"
assert bytes(temp_static.body) != (ROOT / "static" / "ui.js").read_bytes()
traversal = _get("/static/../api/routes.py")
assert traversal.status == 404