209 lines
9.3 KiB
Python
209 lines
9.3 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI, HTTPException, Request, WebSocket
|
|
from fastapi.responses import Response, StreamingResponse
|
|
from fastapi.testclient import TestClient
|
|
from starlette.middleware.gzip import GZipMiddleware
|
|
|
|
from hockey_data import program_log
|
|
from hockey_data.auth_bridge import HockeyUser
|
|
from hockey_data.auth_bridge import HockeyAuthDependencies, LocalBrowserUserAdapter
|
|
from hockey_data.auth_router import create_hockey_auth_router
|
|
from hockey_data.program_log import ProgramLogStore, reset_log_context, set_log_context
|
|
from hockey_data.web_observability import (WebAuditMiddleware, apply_web_cache_policy,
|
|
create_browser_event_router, install_runtime_audit)
|
|
|
|
|
|
@pytest.fixture
|
|
def journal(tmp_path, monkeypatch):
|
|
store = ProgramLogStore(tmp_path / "journal")
|
|
store.start()
|
|
assert store._ready.wait(3)
|
|
monkeypatch.setattr(program_log, "_DEFAULT_STORE", store)
|
|
yield store
|
|
assert store.close()
|
|
|
|
|
|
def test_audit_preserves_body_and_actor_and_masks_sensitive_details(journal):
|
|
app = FastAPI()
|
|
app.add_middleware(WebAuditMiddleware)
|
|
received = []
|
|
|
|
@app.post("/api/hockey/games/12/control/timers")
|
|
async def change(request: Request):
|
|
request.state.wfl_user = HockeyUser("42", "operator", "Operator")
|
|
payload = await request.json()
|
|
received.append(payload)
|
|
return {"ok": True}
|
|
|
|
payload = {"time": "12:05", "device_id": "GFX-OPERATOR", "password": "VERY_PRIVATE", "selected": {"device_secret": "NEVER_LOG"}}
|
|
with TestClient(app) as client:
|
|
response = client.post("/api/hockey/games/12/control/timers", json=payload)
|
|
assert response.status_code == 200 and received == [payload]
|
|
assert response.headers["x-request-id"] and "app;dur=" in response.headers["server-timing"]
|
|
assert journal.flush()
|
|
event = journal.query(preset="all")["items"][0]
|
|
assert event["actor_id"] == "42" and event["game_id"] == "12"
|
|
assert event["device_id"] == "GFX-OPERATOR"
|
|
assert event["details"]["action"]["time"] == "12:05"
|
|
assert "VERY_PRIVATE" not in json.dumps(event) and "NEVER_LOG" not in json.dumps(event)
|
|
|
|
|
|
def test_streaming_response_is_not_buffered_and_secret_queries_are_not_saved(journal):
|
|
app = FastAPI()
|
|
app.add_middleware(WebAuditMiddleware)
|
|
|
|
@app.get("/api/export")
|
|
def export():
|
|
return StreamingResponse(iter([b"A" * 4096] * 50), media_type="application/octet-stream")
|
|
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/export?token=PRIVATE_QUERY&date=2026-10-21")
|
|
assert len(response.content) == 204800
|
|
assert journal.flush()
|
|
event = journal.query(preset="all")["items"][0]
|
|
assert event["details"]["response_bytes"] == 204800
|
|
assert event["details"]["query_keys"] == ["token", "date"]
|
|
assert "PRIVATE_QUERY" not in json.dumps(event)
|
|
|
|
|
|
def test_websocket_normal_close_is_not_an_error(journal):
|
|
app = FastAPI()
|
|
app.add_middleware(WebAuditMiddleware)
|
|
|
|
@app.websocket("/ws/test")
|
|
async def socket(ws: WebSocket):
|
|
await ws.accept()
|
|
await ws.send_json({"ok": True})
|
|
await ws.close()
|
|
|
|
with TestClient(app) as client:
|
|
with client.websocket_connect("/ws/test") as ws:
|
|
assert ws.receive_json() == {"ok": True}
|
|
assert journal.flush()
|
|
rows = journal.query(preset="all")["items"]
|
|
assert {row["event"] for row in rows} == {"websocket.connected", "websocket.closed"}
|
|
assert all(row["level"] == "INFO" for row in rows)
|
|
|
|
|
|
def test_login_form_body_is_never_saved(journal):
|
|
app = FastAPI()
|
|
app.add_middleware(WebAuditMiddleware)
|
|
|
|
@app.post("/login")
|
|
async def login(request: Request):
|
|
assert b"FORM_PASSWORD" in await request.body()
|
|
return Response(status_code=401)
|
|
|
|
with TestClient(app) as client:
|
|
client.post("/login", content="username=me&password=FORM_PASSWORD", headers={"content-type": "application/x-www-form-urlencoded"})
|
|
assert journal.flush()
|
|
event = journal.query(preset="all")["items"][0]
|
|
assert "action" not in event["details"]
|
|
assert "FORM_PASSWORD" not in json.dumps(event)
|
|
|
|
|
|
def test_compressed_static_can_be_cached_while_live_data_stays_fresh(journal):
|
|
app = FastAPI()
|
|
app.add_middleware(GZipMiddleware, minimum_size=1024, compresslevel=5)
|
|
app.add_middleware(WebAuditMiddleware)
|
|
|
|
@app.middleware("http")
|
|
async def cache(request, call_next):
|
|
return apply_web_cache_policy(request, await call_next(request))
|
|
|
|
@app.get("/{rest:path}")
|
|
def resource(rest: str):
|
|
return Response("const value = 1;\n" * 10000, media_type="text/javascript")
|
|
|
|
with TestClient(app) as client:
|
|
static = client.get("/ui-builder-assets/app.js?v=contenthash")
|
|
live = client.get("/api/hockey/games/12/control?v=untrusted")
|
|
plain = client.get("/hockey-assets/module.js")
|
|
assert static.headers["content-encoding"] == "gzip"
|
|
assert int(static.headers["content-length"]) < len(static.content) / 10
|
|
assert "immutable" in static.headers["cache-control"]
|
|
assert "no-store" in live.headers["cache-control"]
|
|
assert "no-store" in plain.headers["cache-control"]
|
|
|
|
|
|
def test_browser_metrics_are_authenticated_bounded_and_searchable(journal):
|
|
app = FastAPI()
|
|
|
|
async def auth(request: Request):
|
|
if not request.headers.get("x-test-login"):
|
|
raise HTTPException(401)
|
|
return HockeyUser("42", "operator", "Operator")
|
|
|
|
app.include_router(create_browser_event_router(auth))
|
|
payload = {"event": "ready", "metrics": {"runtime_ready_ms": 1234.5, "bad": -1}, "build": "190"}
|
|
with TestClient(app) as client:
|
|
assert client.post("/api/hockey/client-events", json=payload).status_code == 401
|
|
assert client.post("/api/hockey/client-events", json=payload, headers={"x-test-login": "yes"}).status_code == 200
|
|
assert client.post("/api/hockey/client-events", json={**payload, "event": "something_else"}, headers={"x-test-login": "yes"}).status_code == 422
|
|
assert journal.flush()
|
|
event = journal.query(preset="all", category="browser")["items"][0]
|
|
assert event["actor_id"] == "42"
|
|
assert event["details"]["metrics"] == {"runtime_ready_ms": 1234.5}
|
|
|
|
|
|
def test_agent_audit_preserves_initiator_and_records_failed_ack(journal):
|
|
async def send(device, payload): return True
|
|
async def ack(device, message): return None
|
|
async def register(ws, hello): return {"device_id": "gfx", "owner": "operator", "current_match_id": "12"}
|
|
hub = SimpleNamespace(send=send, receive_command_ack=ack, register=register)
|
|
service = SimpleNamespace()
|
|
install_runtime_audit(service, hub)
|
|
|
|
async def scenario():
|
|
await hub.register(None, {})
|
|
token = set_log_context(actor_id="admin-id", actor_login="admin")
|
|
try:
|
|
assert await hub.send("gfx", {"type": "vmix.command", "command": {"Function": "SetText", "Value": "3"}})
|
|
finally:
|
|
reset_log_context(token)
|
|
await hub.receive_command_ack("gfx", {"type": "command.ack", "ok": False, "reason": "HTTP 500"})
|
|
await hub.receive_command_ack("gfx", {"type": "command.batch.ack", "ok": "false", "reason": "malformed"})
|
|
await hub.receive_command_ack("gfx", {"type": "command.batch.ack", "ok": True, "reason": "confirmed"})
|
|
asyncio.run(scenario())
|
|
assert journal.flush()
|
|
rows = journal.query(preset="all", category="vmix")["items"]
|
|
sent = next(row for row in rows if row["event"] == "vmix.command.sent")
|
|
assert sent["actor_login"] == "admin"
|
|
assert any(row["level"] == "WARNING" and row["details"]["reason"] == "HTTP 500" for row in rows)
|
|
assert next(row for row in rows if row["details"].get("reason") == "malformed")["level"] == "WARNING"
|
|
assert next(row for row in rows if row["details"].get("reason") == "confirmed")["level"] == "INFO"
|
|
|
|
|
|
def test_phone_login_returns_to_journal_and_rejects_external_redirects(journal):
|
|
app = FastAPI()
|
|
app.include_router(create_hockey_auth_router(HockeyAuthDependencies(LocalBrowserUserAdapter())))
|
|
with TestClient(app) as client:
|
|
page = client.get("/login?next=/logs")
|
|
assert 'name="next" value="/logs"' in page.text
|
|
response = client.post("/login", data={"username": "admin", "password": "test", "next": "/logs"}, follow_redirects=False)
|
|
assert response.status_code == 303 and response.headers["location"] == "/logs"
|
|
client.post("/logout", follow_redirects=False)
|
|
response = client.post("/login", data={"username": "admin", "password": "test", "next": "https://invalid.example/"}, follow_redirects=False)
|
|
assert response.headers["location"] == "/"
|
|
|
|
|
|
def test_logout_still_revokes_cookie_when_auth_database_unavailable(journal):
|
|
class Unavailable:
|
|
revoked = []
|
|
async def current_user(self, request): raise HTTPException(503, "database unavailable")
|
|
def logout(self, token): self.revoked.append(token)
|
|
adapter = Unavailable()
|
|
app = FastAPI()
|
|
app.include_router(create_hockey_auth_router(HockeyAuthDependencies(adapter)))
|
|
with TestClient(app) as client:
|
|
client.cookies.set("hockey_auth_token", "old-session")
|
|
response = client.post("/logout", follow_redirects=False)
|
|
assert response.status_code == 303 and adapter.revoked == ["old-session"]
|
|
assert "Max-Age=0" in response.headers["set-cookie"]
|