"""Asset transfer, cache isolation and bounded-memory proxy regressions.""" from __future__ import annotations import asyncio import gzip from contextlib import asynccontextmanager from types import SimpleNamespace import httpx import pytest import pytest_asyncio from fastapi import FastAPI, HTTPException from app.gateway.weknora_embed import WEKNORA_EMBED_COOKIE, sign_weknora_embed_payload from app.gateway.weknora_static import StaticAssetCache, prepare_static_asset @pytest_asyncio.fixture async def static_api(monkeypatch): from app.gateway.routers import llmwiki state = SimpleNamespace( body=('export const asset="/assets/lazy.js";\n' * 8000).encode(), content_type="application/javascript", headers={}, status=200, requests=[], runtime=SimpleNamespace(web_base_url="http://weknora-web.test", api_base_url="http://weknora-api.test"), prefix="/deerflow", ) async def upstream(request): state.requests.append(request) await asyncio.sleep(0) return httpx.Response(state.status, content=gzip.compress(state.body), headers={"content-type": state.content_type, "content-encoding": "gzip", **state.headers}) def no_store(_): pytest.fail("Static resources must not query knowledge-base ACLs") async def no_login(*_, **__): pytest.fail("Static resources must not log in to the upstream API") app = FastAPI() app.include_router(llmwiki.router) app.include_router(llmwiki.proxy_router, prefix="/deerflow") monkeypatch.setattr(llmwiki, "_store", no_store) monkeypatch.setattr(llmwiki, "_get_weknora_admin_session", no_login) monkeypatch.setattr(llmwiki, "_runtime_for_iframe_proxy", lambda _: state.runtime) monkeypatch.setattr(llmwiki, "_request_public_prefix", lambda _: state.prefix) async with httpx.AsyncClient(transport=httpx.MockTransport(upstream)) as remote: @asynccontextmanager async def pool(_): yield remote monkeypatch.setattr(llmwiki, "_proxy_http_client", pool) async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://gateway.test") as api: api.cookies.set( WEKNORA_EMBED_COOKIE, sign_weknora_embed_payload( { "mapping_id": "local", "weknora_id": "remote", "user_id": "owner", "is_admin": False, "can_write": False, "exp": 4_102_444_800, } ), ) state.api = api yield state @pytest.mark.asyncio async def test_static_cold_concurrent_and_warm_reads_share_one_fetch_and_compressed_bytes(static_api): state = static_api path = "/deerflow/assets/main-Cg_NDvUs.js" results = await asyncio.gather(*(state.api.get(path, headers={"Accept-Encoding": "gzip"}) for _ in range(6))) assert all(response.status_code == 200 for response in results) assert len(state.requests) == 1 response = results[0] assert response.headers["content-encoding"] == "gzip" assert response.headers["vary"] == "Accept-Encoding" assert "31536000, immutable" in response.headers["cache-control"] assert response.content == state.body.replace(b'"/assets/', b'"/deerflow/assets/') assert int(response.headers["content-length"]) < len(response.content) / 10 assert (await state.api.get(path)).content == response.content assert len(state.requests) == 1 assert state.requests[0].headers["accept-encoding"] == "gzip, deflate" assert "authorization" not in state.requests[0].headers assert "cookie" not in state.requests[0].headers @pytest.mark.asyncio async def test_variant_etags_head_and_conditional_requests_are_consistent(static_api): api = static_api.api path = "/deerflow/assets/main-Cg_NDvUs.js" compressed = await api.get(path, headers={"Accept-Encoding": "gzip"}) identity = await api.get(path, headers={"Accept-Encoding": "gzip;q=0, identity"}) assert "content-encoding" not in identity.headers assert identity.content == compressed.content assert identity.headers["etag"] != compressed.headers["etag"] cached = await api.get(path, headers={"Accept-Encoding": "gzip", "If-None-Match": compressed.headers["etag"]}) assert cached.status_code == 304 and cached.content == b"" other_variant = await api.get(path, headers={"Accept-Encoding": "identity", "If-None-Match": compressed.headers["etag"]}) assert other_variant.status_code == 200 head = await api.head(path, headers={"Accept-Encoding": "gzip"}) assert head.content == b"" assert head.headers["etag"] == compressed.headers["etag"] assert head.headers["content-length"] == compressed.headers["content-length"] assert len(static_api.requests) == 1 @pytest.mark.asyncio async def test_cache_keys_separate_upstream_prefix_and_query(static_api): state = static_api path = "/deerflow/assets/main-Cg_NDvUs.js" await state.api.get(path) await state.api.get(path + "?variant=2") state.runtime.web_base_url = "http://another-weknora.test" await state.api.get(path) state.prefix = "/another-prefix" response = await state.api.get(path) assert b'"/another-prefix/assets/' in response.content assert len(state.requests) == 4 @pytest.mark.asyncio @pytest.mark.parametrize("content_type", ["text/html", "application/octet-stream"]) async def test_spa_html_fallback_is_never_cached_or_exposed_as_an_asset(static_api, content_type): state = static_api state.body = b"account-specific bootstrap" state.content_type = content_type path = "/deerflow/assets/missing.js" assert (await state.api.get(path)).status_code == 404 assert (await state.api.get(path)).status_code == 404 assert len(state.requests) == 2 @pytest.mark.asyncio async def test_static_cache_still_requires_a_valid_embed_cookie(static_api): state = static_api path = "/deerflow/assets/main-Cg_NDvUs.js" await state.api.get(path) state.api.cookies.clear() response = await state.api.get(path) assert response.status_code == 401 assert len(state.requests) == 1 @pytest.mark.asyncio async def test_html_and_api_keep_full_permission_checks(static_api, monkeypatch): from app.gateway.routers import llmwiki checked = [] async def deny(request): checked.append(request.url.path) raise HTTPException(status_code=403, detail="Permission revoked") monkeypatch.setattr(llmwiki, "_read_embed_context", deny) for path in ("/deerflow/platform/knowledge-bases/remote", "/deerflow/api/v1/knowledge-bases/remote"): assert (await static_api.api.get(path)).status_code == 403 assert len(checked) == 2 assert not static_api.requests @pytest.mark.asyncio @pytest.mark.parametrize( "headers", [ {"set-cookie": "session=private"}, {"vary": "Authorization"}, {"cache-control": "private"}, {"cache-control": "no-store"}, {"cache-control": "no-cache"}, {"cache-control": "max-age=0"}, {"cache-control": 'public, max-age = "0"'}, ], ) async def test_upstream_identity_responses_are_not_cached(static_api, headers): state = static_api state.headers = headers path = "/deerflow/assets/main-Cg_NDvUs.js" first = await state.api.get(path) await state.api.get(path) assert "no-store" in first.headers["cache-control"] assert "set-cookie" not in first.headers assert len(state.requests) == 2 @pytest.mark.asyncio async def test_config_js_is_revalidated_instead_of_immutable(static_api): state = static_api await state.api.get("/deerflow/config.js") state.body = b"window.config='changed';" changed = await state.api.get("/deerflow/config.js") assert changed.content == state.body assert changed.headers["cache-control"] == "private, no-cache" assert len(state.requests) == 2 @pytest.mark.asyncio async def test_browser_hard_refresh_bypasses_the_server_asset_cache(static_api): state = static_api path = "/deerflow/assets/styles.css" state.content_type = "text/css" first = await state.api.get(path) state.body = b"body { color: red; }" refreshed = await state.api.get(path, headers={"Cache-Control": "no-cache", "If-None-Match": first.headers["etag"]}) assert refreshed.status_code == 200 and refreshed.content == state.body assert len(state.requests) == 2 assert (await state.api.get(path)).content == state.body assert len(state.requests) == 2 state.headers = {"Cache-Control": "no-cache"} await state.api.get(path, headers={"Cache-Control": "no-cache"}) state.body = b"body { color: blue; }" assert (await state.api.get(path)).content == state.body assert len(state.requests) == 4 @pytest.mark.asyncio async def test_static_cache_enforces_byte_and_single_asset_bounds(): cache = StaticAssetCache(max_bytes=10, max_asset_bytes=8, max_entries=2) calls = [] async def get(key, content): async def load(): calls.append(key) return prepare_static_asset(content, content_type="text/plain", cache_control="private", ttl=60) return await cache.get((key,), load) await get("one", b"123456") await get("two", b"123456") await get("two", b"123456") await get("one", b"123456") assert calls == ["one", "two", "one"] await get("huge", b"123456789") await get("huge", b"123456789") assert calls[-2:] == ["huge", "huge"] assert cache._bytes <= 10 def test_static_fast_path_rejects_traversal_and_arbitrary_data_paths(): from app.gateway.routers.llmwiki import _is_weknora_static_asset_path for path in ("/api/v1/data.json", "/platform/data.json", "/assets/../api/data.json", "/assets/..\\api/data.json", "/assets/private.html"): assert not _is_weknora_static_asset_path(path)