338 lines
11 KiB
Python
338 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable, Iterator, Mapping, Sequence
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
from fastapi.testclient import TestClient
|
|
from starlette.routing import Mount
|
|
from starlette.staticfiles import StaticFiles
|
|
from starlette.types import Scope
|
|
|
|
from game_data_capture.config import ServerConfig
|
|
from game_data_capture.web.app import (
|
|
CaptureRecord,
|
|
CaptureSessionStatus,
|
|
ControlOutcome,
|
|
MediaDescriptor,
|
|
MutationOutcome,
|
|
WebConfig,
|
|
create_app,
|
|
)
|
|
from game_data_capture.web.security import new_bootstrap_token
|
|
|
|
_CAPTURE_ID = "capture-00000000000000000000000000000001"
|
|
_LABEL: dict[str, object] = {
|
|
"family": "weapon",
|
|
"view": "detail",
|
|
"gear_slot": None,
|
|
"record_types": ["weapon", "weapon_talent"],
|
|
"confidence": 1.0,
|
|
"abstention_reason": None,
|
|
}
|
|
|
|
|
|
@dataclass
|
|
class FakeControl:
|
|
armed: bool = False
|
|
starts: int = 0
|
|
stops: int = 0
|
|
|
|
def status(self) -> CaptureSessionStatus:
|
|
return CaptureSessionStatus(self.armed, 0, 64)
|
|
|
|
def start_session(self) -> ControlOutcome:
|
|
if self.armed:
|
|
return "already_armed"
|
|
self.starts += 1
|
|
self.armed = True
|
|
return "started"
|
|
|
|
def stop_session(self) -> ControlOutcome:
|
|
if not self.armed:
|
|
return "already_stopped"
|
|
self.stops += 1
|
|
self.armed = False
|
|
return "stopped"
|
|
|
|
|
|
@dataclass
|
|
class FakeCorpus:
|
|
records: dict[str, CaptureRecord] = field(
|
|
default_factory=lambda: {
|
|
_CAPTURE_ID: CaptureRecord(_CAPTURE_ID, "stored", "2026-07-31T00:00:00Z", None)
|
|
}
|
|
)
|
|
calls: list[str] = field(default_factory=list)
|
|
media: MediaDescriptor | None = None
|
|
|
|
def list_captures(self, *, limit: int) -> Sequence[CaptureRecord]:
|
|
self.calls.append("list")
|
|
return tuple(self.records.values())[:limit]
|
|
|
|
def get_capture(self, capture_id: str) -> CaptureRecord | None:
|
|
self.calls.append("get")
|
|
return self.records.get(capture_id)
|
|
|
|
def open_media(self, capture_id: str) -> MediaDescriptor | None:
|
|
self.calls.append("media")
|
|
if capture_id not in self.records:
|
|
return None
|
|
if self.media is not None:
|
|
return self.media
|
|
return MediaDescriptor("image/png", 4, [b"data"])
|
|
|
|
def label(self, capture_id: str, classification: Mapping[str, object]) -> MutationOutcome:
|
|
self.calls.append("label")
|
|
if capture_id not in self.records:
|
|
return "not_found"
|
|
old = self.records[capture_id]
|
|
self.records[capture_id] = CaptureRecord(
|
|
old.capture_id, old.state, old.received_at, dict(classification)
|
|
)
|
|
return "updated"
|
|
|
|
def reject_preserving_evidence(self, capture_id: str) -> MutationOutcome:
|
|
self.calls.append("reject")
|
|
return "updated" if capture_id in self.records else "not_found"
|
|
|
|
def delete_with_retention(self, capture_id: str) -> MutationOutcome:
|
|
self.calls.append("delete")
|
|
return "retained" if capture_id in self.records else "not_found"
|
|
|
|
|
|
def _client(control: FakeControl, corpus: FakeCorpus) -> tuple[TestClient, str]:
|
|
token = new_bootstrap_token()
|
|
app = create_app(
|
|
config=WebConfig(ServerConfig()),
|
|
capture_control=control,
|
|
corpus=corpus,
|
|
bootstrap_token=token,
|
|
)
|
|
return TestClient(app, base_url="http://127.0.0.1:8765"), token
|
|
|
|
|
|
def _bootstrap(client: TestClient, token: str) -> str:
|
|
response = client.post(
|
|
"/api/bootstrap", headers={"Origin": "http://127.0.0.1:8765"}, json={"token": token}
|
|
)
|
|
assert response.status_code == 200
|
|
return str(response.json()["csrf_token"])
|
|
|
|
|
|
def _mutation_headers(csrf: str) -> dict[str, str]:
|
|
return {"Origin": "http://127.0.0.1:8765", "X-CSRF-Token": csrf}
|
|
|
|
|
|
def test_factory_and_authenticated_gets_never_arm_capture() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
|
|
assert control.starts == 0
|
|
assert client.get("/health").json() == {"status": "ok"}
|
|
csrf = _bootstrap(client, token)
|
|
assert client.get("/api/session").status_code == 200
|
|
assert client.get("/api/captures").status_code == 200
|
|
assert client.get(f"/api/captures/{_CAPTURE_ID}").status_code == 200
|
|
assert control.starts == 0
|
|
assert csrf
|
|
|
|
|
|
def test_static_contract_is_package_local_and_does_not_accept_a_filesystem_root() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
app = create_app(
|
|
config=WebConfig(ServerConfig()),
|
|
capture_control=control,
|
|
corpus=corpus,
|
|
bootstrap_token=new_bootstrap_token(),
|
|
)
|
|
static_mount = next(
|
|
route for route in app.routes if isinstance(route, Mount) and route.path == "/static"
|
|
)
|
|
|
|
assert isinstance(static_mount.app, StaticFiles)
|
|
assert static_mount.app.follow_symlink is False
|
|
assert static_mount.app.directory is not None
|
|
assert Path(static_mount.app.directory).name == "static"
|
|
assert not Path(
|
|
static_mount.app.get_path(
|
|
cast(Scope, {"path": "/static/../../etc/passwd", "root_path": ""})
|
|
)
|
|
).is_absolute()
|
|
|
|
|
|
def test_bootstrap_cookie_is_host_only_http_only_and_never_echoes_bootstrap_token() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
|
|
response = client.post(
|
|
"/api/bootstrap", headers={"Origin": "http://127.0.0.1:8765"}, json={"token": token}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert token not in response.text
|
|
cookie = response.headers["set-cookie"].lower()
|
|
assert "httponly" in cookie
|
|
assert "samesite=strict" in cookie
|
|
assert "domain=" not in cookie
|
|
assert token not in cookie
|
|
assert response.json().keys() == {"csrf_token"}
|
|
|
|
|
|
def test_start_is_explicit_csrf_protected_idempotent_and_rotates_session() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
csrf = _bootstrap(client, token)
|
|
old_session = str(client.cookies["capture_session"])
|
|
|
|
response = client.post(
|
|
"/api/session/start", headers=_mutation_headers(csrf), json={"confirmation": "START"}
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["result"] == "started"
|
|
assert control.starts == 1
|
|
next_csrf = str(response.json()["csrf_token"])
|
|
stale = client.post(
|
|
"/api/session/start", headers=_mutation_headers(csrf), json={"confirmation": "START"}
|
|
)
|
|
assert stale.status_code == 403
|
|
assert control.starts == 1
|
|
expired_session = client.post(
|
|
"/api/session/start",
|
|
headers={**_mutation_headers(next_csrf), "Cookie": f"capture_session={old_session}"},
|
|
json={"confirmation": "START"},
|
|
)
|
|
assert expired_session.status_code == 401
|
|
assert control.starts == 1
|
|
duplicate = client.post(
|
|
"/api/session/start", headers=_mutation_headers(next_csrf), json={"confirmation": "START"}
|
|
)
|
|
assert duplicate.status_code == 200
|
|
assert duplicate.json()["result"] == "already_armed"
|
|
assert control.starts == 1
|
|
|
|
|
|
def test_label_reject_and_retention_deletion_use_closed_confirmed_contracts() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
csrf = _bootstrap(client, token)
|
|
|
|
labeled = client.post(
|
|
f"/api/captures/{_CAPTURE_ID}/label", headers=_mutation_headers(csrf), json=_LABEL
|
|
)
|
|
assert labeled.status_code == 200
|
|
csrf = str(labeled.json()["csrf_token"])
|
|
detail = client.get(f"/api/captures/{_CAPTURE_ID}")
|
|
assert detail.json()["capture"]["classification"] == _LABEL
|
|
rejected = client.post(
|
|
f"/api/captures/{_CAPTURE_ID}/reject",
|
|
headers=_mutation_headers(csrf),
|
|
json={"confirmation": "REJECT"},
|
|
)
|
|
assert rejected.status_code == 200
|
|
csrf = str(rejected.json()["csrf_token"])
|
|
denied = client.post(
|
|
f"/api/captures/{_CAPTURE_ID}/delete",
|
|
headers=_mutation_headers(csrf),
|
|
json={"confirmation": "delete"},
|
|
)
|
|
assert denied.status_code == 403
|
|
assert "delete" not in corpus.calls
|
|
retained = client.post(
|
|
f"/api/captures/{_CAPTURE_ID}/delete",
|
|
headers=_mutation_headers(csrf),
|
|
json={"confirmation": "DELETE"},
|
|
)
|
|
assert retained.status_code == 409
|
|
assert retained.json() == {"error": "retained"}
|
|
assert corpus.calls.count("label") == 1
|
|
assert corpus.calls.count("reject") == 1
|
|
assert corpus.calls.count("delete") == 1
|
|
|
|
|
|
def test_opaque_media_route_has_private_nosniff_response_and_no_path_parameter() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
_bootstrap(client, token)
|
|
|
|
response = client.get(f"/api/captures/{_CAPTURE_ID}/media")
|
|
|
|
assert response.status_code == 200
|
|
assert response.content == b"data"
|
|
assert response.headers["content-type"] == "image/png"
|
|
assert response.headers["cache-control"] == "private, no-store"
|
|
assert response.headers["x-content-type-options"] == "nosniff"
|
|
assert client.get("/api/captures/../../etc/passwd/media").status_code == 404
|
|
assert corpus.calls.count("media") == 1
|
|
|
|
|
|
class RaisingChunks:
|
|
def __iter__(self) -> Iterator[bytes]:
|
|
raise RuntimeError("raw-secret /private/capture/path")
|
|
|
|
|
|
def test_malformed_media_streams_fail_before_success_headers_or_content() -> None:
|
|
descriptors = (
|
|
MediaDescriptor("image/png", 4, [b"abc"]),
|
|
MediaDescriptor("image/png", 3, [b"abcd"]),
|
|
MediaDescriptor("image/png", 3, [b""]),
|
|
MediaDescriptor("image/png", 3, cast(Iterable[bytes], ["bad"])),
|
|
MediaDescriptor("image/png", 3, RaisingChunks()),
|
|
)
|
|
for descriptor in descriptors:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus(media=descriptor)
|
|
client, token = _client(control, corpus)
|
|
_bootstrap(client, token)
|
|
|
|
response = client.get(f"/api/captures/{_CAPTURE_ID}/media")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"error": "service_unavailable"}
|
|
assert "raw-secret" not in response.text
|
|
assert "/private/capture/path" not in response.text
|
|
|
|
|
|
class FailingCorpus(FakeCorpus):
|
|
def list_captures(self, *, limit: int) -> Sequence[CaptureRecord]:
|
|
raise RuntimeError("raw-secret /private/capture/path")
|
|
|
|
|
|
def test_port_exceptions_are_sanitized_without_an_uncaught_client_error() -> None:
|
|
control = FakeControl()
|
|
corpus = FailingCorpus()
|
|
client, token = _client(control, corpus)
|
|
_bootstrap(client, token)
|
|
|
|
response = client.get("/api/captures")
|
|
|
|
assert response.status_code == 503
|
|
assert response.json() == {"error": "service_unavailable"}
|
|
assert "raw-secret" not in response.text
|
|
assert "/private/capture/path" not in response.text
|
|
|
|
|
|
def test_taxonomy_payload_rejects_unknown_and_ambiguous_values_before_corpus_mutation() -> None:
|
|
control = FakeControl()
|
|
corpus = FakeCorpus()
|
|
client, token = _client(control, corpus)
|
|
csrf = _bootstrap(client, token)
|
|
unsupported = dict(_LABEL)
|
|
unsupported["view"] = "not-a-view"
|
|
unsupported["record_types"] = ["gear"]
|
|
|
|
response = client.post(
|
|
f"/api/captures/{_CAPTURE_ID}/label", headers=_mutation_headers(csrf), json=unsupported
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert "label" not in corpus.calls
|