252 lines
9.8 KiB
Python
252 lines
9.8 KiB
Python
"""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"<!doctype html><html>account-specific bootstrap</html>"
|
|
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)
|