322 lines
10 KiB
Python
322 lines
10 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from collections.abc import Callable, Mapping, Sequence
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
from starlette.types import ASGIApp, Message, Receive, Scope, Send
|
|
|
|
from game_data_capture.config import ServerConfig
|
|
from game_data_capture.web.app import (
|
|
CaptureCorpusPort,
|
|
CaptureRecord,
|
|
CaptureSessionStatus,
|
|
ControlOutcome,
|
|
MediaDescriptor,
|
|
MutationOutcome,
|
|
WebConfig,
|
|
create_app,
|
|
)
|
|
from game_data_capture.web.security import (
|
|
LocalAuthority,
|
|
LocalRequestLimitsMiddleware,
|
|
RequestLimits,
|
|
new_bootstrap_token,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class Control:
|
|
start_calls: int = 0
|
|
|
|
def status(self) -> CaptureSessionStatus:
|
|
return CaptureSessionStatus(False, 0, 1)
|
|
|
|
def start_session(self) -> ControlOutcome:
|
|
self.start_calls += 1
|
|
return "started"
|
|
|
|
def stop_session(self) -> ControlOutcome:
|
|
return "already_stopped"
|
|
|
|
|
|
class Corpus(CaptureCorpusPort):
|
|
def list_captures(self, *, limit: int) -> Sequence[CaptureRecord]:
|
|
return ()
|
|
|
|
def get_capture(self, capture_id: str) -> CaptureRecord | None:
|
|
return None
|
|
|
|
def open_media(self, capture_id: str) -> MediaDescriptor | None:
|
|
return None
|
|
|
|
def label(self, capture_id: str, classification: Mapping[str, object]) -> MutationOutcome:
|
|
return "not_found"
|
|
|
|
def reject_preserving_evidence(self, capture_id: str) -> MutationOutcome:
|
|
return "not_found"
|
|
|
|
def delete_with_retention(self, capture_id: str) -> MutationOutcome:
|
|
return "not_found"
|
|
|
|
|
|
class Clock:
|
|
def __init__(self) -> None:
|
|
self.value = 0.0
|
|
|
|
def __call__(self) -> float:
|
|
return self.value
|
|
|
|
|
|
def _app(
|
|
control: Control, *, now: Callable[[], float] | None = None, max_body: int = 16 * 1024
|
|
) -> tuple[FastAPI, str]:
|
|
token = new_bootstrap_token()
|
|
config = WebConfig(ServerConfig(), request_limits=RequestLimits(max_body_bytes=max_body))
|
|
if now is None:
|
|
app = create_app(
|
|
config=config,
|
|
capture_control=control,
|
|
corpus=Corpus(),
|
|
bootstrap_token=token,
|
|
)
|
|
else:
|
|
app = create_app(
|
|
config=config,
|
|
capture_control=control,
|
|
corpus=Corpus(),
|
|
bootstrap_token=token,
|
|
now=now,
|
|
)
|
|
return app, token
|
|
|
|
|
|
def _bootstrap(client: TestClient, token: str) -> str:
|
|
result = client.post(
|
|
"/api/bootstrap", headers={"Origin": "http://127.0.0.1:8765"}, json={"token": token}
|
|
)
|
|
assert result.status_code == 200
|
|
return str(result.json()["csrf_token"])
|
|
|
|
|
|
def test_host_origin_and_csrf_fail_before_start_side_effect() -> None:
|
|
control = Control()
|
|
app, token = _app(control)
|
|
client = TestClient(app, base_url="http://127.0.0.1:8765")
|
|
csrf = _bootstrap(client, token)
|
|
|
|
bad_host = client.post(
|
|
"/api/session/start",
|
|
headers={"Host": "evil.test:8765", "Origin": "http://127.0.0.1:8765", "X-CSRF-Token": csrf},
|
|
json={"confirmation": "START"},
|
|
)
|
|
missing_origin = client.post(
|
|
"/api/session/start", headers={"X-CSRF-Token": csrf}, json={"confirmation": "START"}
|
|
)
|
|
cross_origin = client.post(
|
|
"/api/session/start",
|
|
headers={"Origin": "http://evil.test", "X-CSRF-Token": csrf},
|
|
json={"confirmation": "START"},
|
|
)
|
|
bad_csrf = client.post(
|
|
"/api/session/start",
|
|
headers={"Origin": "http://127.0.0.1:8765", "X-CSRF-Token": "x" * 43},
|
|
json={"confirmation": "START"},
|
|
)
|
|
|
|
assert bad_host.status_code == 400
|
|
assert missing_origin.status_code == 403
|
|
assert cross_origin.status_code == 403
|
|
assert bad_csrf.status_code == 403
|
|
assert control.start_calls == 0
|
|
|
|
|
|
def test_session_expiry_and_process_restart_invalidate_cookie() -> None:
|
|
control = Control()
|
|
clock = Clock()
|
|
app, token = _app(control, now=clock)
|
|
client = TestClient(app, base_url="http://127.0.0.1:8765")
|
|
_bootstrap(client, token)
|
|
assert client.get("/api/session").status_code == 200
|
|
clock.value = 901
|
|
assert client.get("/api/session").status_code == 401
|
|
|
|
new_app, _ = _app(control)
|
|
restarted = TestClient(new_app, base_url="http://127.0.0.1:8765")
|
|
restarted.cookies.update(client.cookies)
|
|
assert restarted.get("/api/session").status_code == 401
|
|
|
|
|
|
def test_duplicate_json_and_form_payloads_fail_before_start_side_effect() -> None:
|
|
control = Control()
|
|
app, token = _app(control)
|
|
client = TestClient(app, base_url="http://127.0.0.1:8765")
|
|
csrf = _bootstrap(client, token)
|
|
headers = {"Origin": "http://127.0.0.1:8765", "X-CSRF-Token": csrf}
|
|
|
|
duplicate_json = client.post(
|
|
"/api/session/start",
|
|
headers={**headers, "Content-Type": "application/json"},
|
|
content=b'{"confirmation":"START","confirmation":"START"}',
|
|
)
|
|
form = client.post(
|
|
"/api/session/start",
|
|
headers={**headers, "Content-Type": "application/x-www-form-urlencoded"},
|
|
content=b"confirmation=START&confirmation=START",
|
|
)
|
|
|
|
assert duplicate_json.status_code == 403
|
|
assert form.status_code == 403
|
|
assert control.start_calls == 0
|
|
|
|
|
|
def test_ipv6_host_requires_bracketed_exact_authority() -> None:
|
|
control = Control()
|
|
token = new_bootstrap_token()
|
|
app = create_app(
|
|
config=WebConfig(ServerConfig(host="::1", port=8765)),
|
|
capture_control=control,
|
|
corpus=Corpus(),
|
|
bootstrap_token=token,
|
|
)
|
|
client = TestClient(app, base_url="http://127.0.0.1:8765")
|
|
|
|
accepted = client.get("/health", headers={"Host": "[::1]:8765"})
|
|
rejected = client.get("/health", headers={"Host": "::1:8765"})
|
|
|
|
assert accepted.status_code == 200
|
|
assert rejected.status_code == 400
|
|
|
|
|
|
def test_all_success_and_error_responses_have_local_only_security_headers() -> None:
|
|
control = Control()
|
|
app, _ = _app(control)
|
|
client = TestClient(app, base_url="http://127.0.0.1:8765")
|
|
responses = (
|
|
client.get("/health"),
|
|
client.get("/health", headers={"Host": "evil.test:8765"}),
|
|
client.get("/api/session"),
|
|
)
|
|
for response in responses:
|
|
csp = response.headers["content-security-policy"]
|
|
assert "default-src 'self'" in csp
|
|
assert "script-src 'self'" in csp
|
|
assert "style-src 'self'" in csp
|
|
assert "font-src 'self'" in csp
|
|
assert "img-src 'self'" in csp
|
|
assert "media-src 'self'" in csp
|
|
assert "connect-src 'self'" in csp
|
|
assert "object-src 'none'" in csp
|
|
assert "base-uri 'none'" in csp
|
|
assert "frame-ancestors 'none'" in csp
|
|
assert response.headers["referrer-policy"] == "no-referrer"
|
|
assert response.headers["x-content-type-options"] == "nosniff"
|
|
assert response.headers["x-frame-options"] == "DENY"
|
|
|
|
|
|
async def _call_raw(app: ASGIApp, scope: Scope, messages: list[Message]) -> list[Message]:
|
|
sent: list[Message] = []
|
|
|
|
async def receive() -> Message:
|
|
return messages.pop(0)
|
|
|
|
async def send(message: Message) -> None:
|
|
sent.append(message)
|
|
|
|
await app(scope, receive, send)
|
|
return sent
|
|
|
|
|
|
def _scope(path: str, headers: list[tuple[bytes, bytes]], *, method: str = "POST") -> Scope:
|
|
return {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0"},
|
|
"http_version": "1.1",
|
|
"method": method,
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode(),
|
|
"query_string": b"",
|
|
"headers": headers,
|
|
"client": ("127.0.0.1", 50000),
|
|
"server": ("127.0.0.1", 8765),
|
|
}
|
|
|
|
|
|
def test_declared_streamed_duplicate_and_disconnect_requests_fail_before_start() -> None:
|
|
control = Control()
|
|
app, _ = _app(control, max_body=32)
|
|
base_headers = [(b"host", b"127.0.0.1:8765"), (b"content-type", b"application/json")]
|
|
|
|
declared = asyncio.run(
|
|
_call_raw(
|
|
app,
|
|
_scope("/api/session/start", [*base_headers, (b"content-length", b"33")]),
|
|
[{"type": "http.request", "body": b"", "more_body": False}],
|
|
)
|
|
)
|
|
duplicated = asyncio.run(
|
|
_call_raw(
|
|
app,
|
|
_scope("/api/session/start", [*base_headers, (b"host", b"127.0.0.1:8765")]),
|
|
[{"type": "http.request", "body": b"{}", "more_body": False}],
|
|
)
|
|
)
|
|
streamed = asyncio.run(
|
|
_call_raw(
|
|
app,
|
|
_scope(
|
|
"/api/bootstrap",
|
|
[*base_headers, (b"origin", b"http://127.0.0.1:8765")],
|
|
),
|
|
[
|
|
{"type": "http.request", "body": b'{"token":"', "more_body": True},
|
|
{"type": "http.request", "body": b"x" * 24, "more_body": False},
|
|
],
|
|
)
|
|
)
|
|
disconnected = asyncio.run(
|
|
_call_raw(
|
|
app,
|
|
_scope(
|
|
"/api/bootstrap",
|
|
[*base_headers, (b"origin", b"http://127.0.0.1:8765")],
|
|
),
|
|
[{"type": "http.disconnect"}],
|
|
)
|
|
)
|
|
|
|
assert declared[0]["status"] == 413
|
|
assert duplicated[0]["status"] == 400
|
|
assert streamed[0]["status"] == 413
|
|
assert disconnected[0]["status"] == 400
|
|
assert control.start_calls == 0
|
|
|
|
|
|
def test_post_completion_disconnect_is_normal_for_streaming_response_lifecycle() -> None:
|
|
async def streaming_lifecycle(scope: Scope, receive: Receive, send: Send) -> None:
|
|
await receive()
|
|
assert (await receive())["type"] == "http.disconnect"
|
|
await send({"type": "http.response.start", "status": 200, "headers": []})
|
|
await send({"type": "http.response.body", "body": b"ok"})
|
|
|
|
guarded = LocalRequestLimitsMiddleware(
|
|
streaming_lifecycle,
|
|
authority=LocalAuthority("127.0.0.1", 8765),
|
|
limits=RequestLimits(),
|
|
)
|
|
sent = asyncio.run(
|
|
_call_raw(
|
|
guarded,
|
|
_scope("/stream", [(b"host", b"127.0.0.1:8765")], method="GET"),
|
|
[
|
|
{"type": "http.request", "body": b"", "more_body": False},
|
|
{"type": "http.disconnect"},
|
|
],
|
|
)
|
|
)
|
|
|
|
assert sent[0]["status"] == 200
|