"""Checks for the properties that make link-only sharing actually hold."""
import json
import os
import tempfile
os.environ["TRACE_DIR"] = tempfile.mkdtemp()
os.environ["UPLOAD_TOKEN"] = "s3cret" # exercised below; unset in deployment
import pytest # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
import app as app_module # noqa: E402
import render # noqa: E402
client = TestClient(app_module.app, raise_server_exceptions=False)
AUTH = {"x-upload-token": "s3cret"}
@pytest.fixture(autouse=True)
def fresh_state():
"""Every test looks like a new client; otherwise the shared bucket from the
TestClient's single IP starts returning 429 partway through the suite.
UPLOAD_TOKEN is pinned here rather than inherited from import order — the
app module is shared with test_auth.py, which turns the token off.
"""
app_module._hits.clear()
app_module.UPLOAD_TOKEN = "s3cret"
client.cookies.clear()
yield
app_module._hits.clear()
def create(trace, title="t"):
response = client.post("/api/traces", json={"title": title, "trace": trace}, headers=AUTH)
assert response.status_code == 201, response.text
return response.json()
def test_ids_are_128_bit_and_unique():
ids = {app_module.new_id() for _ in range(2000)}
assert len(ids) == 2000
assert all(app_module.ID_RE.match(i) for i in ids)
assert app_module.ID_BYTES * 8 == 128
# and the id the API hands back has the same shape
assert app_module.ID_RE.match(create("hello")["id"])
def test_roundtrip_and_raw():
messages = [
{"role": "user", "content": "what's 2+2"},
{"role": "assistant", "content": [{"type": "text", "text": "4"}]},
]
created = create({"messages": messages}, title="math")
page = client.get(f"/t/{created['id']}")
assert page.status_code == 200
assert "what's 2+2" in page.text or "what's 2+2" in page.text
assert "math" in page.text
raw = client.get(f"/t/{created['id']}/raw")
assert raw.json()["messages"] == messages
def test_no_listing_or_index_leak():
created = create("secret trace")
home = client.get("/")
assert created["id"] not in home.text
# No route enumerates traces.
for path in ("/traces", "/api/traces/list", "/t/", "/t"):
assert client.get(path).status_code in (404, 405)
def test_unknown_and_malformed_ids_are_indistinguishable():
unknown = client.get("/t/" + "A" * 22) # well-formed, absent
malformed = client.get("/t/short") # wrong shape
assert unknown.status_code == malformed.status_code == 404
assert unknown.json() == malformed.json()
def test_security_headers_present():
created = create("x")
headers = client.get(f"/t/{created['id']}").headers
assert headers["referrer-policy"] == "no-referrer"
assert "noindex" in headers["x-robots-tag"]
assert "default-src 'none'" in headers["content-security-policy"]
assert headers["cache-control"] == "private, no-store"
def test_robots_disallows_everything():
assert "Disallow: /" in client.get("/robots.txt").text
def test_uploads_are_open_when_no_token_configured(monkeypatch):
"""The deployed configuration: anyone may upload."""
monkeypatch.setattr(app_module, "UPLOAD_TOKEN", "")
assert client.post("/api/traces", json={"trace": "anyone"}).status_code == 201
assert client.post("/", data={"content": "anyone"}).status_code == 201
def test_upload_requires_token():
assert client.post("/api/traces", json={"trace": "x"}).status_code == 401
assert client.post("/api/traces", json={"trace": "x"},
headers={"x-upload-token": "wrong"}).status_code == 401
assert client.post("/api/traces", json={"trace": "x"},
headers={"authorization": "Bearer s3cret"}).status_code == 201
def test_oversized_trace_rejected():
original = app_module.MAX_BYTES
app_module.MAX_BYTES = 500
try:
response = client.post("/api/traces", json={"trace": "x" * 2000}, headers=AUTH)
assert response.status_code == 413
finally:
app_module.MAX_BYTES = original
def test_html_is_escaped():
created = create([{"role": "user", "content": ""}])
page = client.get(f"/t/{created['id']}")
assert "" not in page.text
assert "<script>" in page.text
def test_read_rate_limit_blocks_scanning():
app_module._hits.clear()
limit, _ = app_module.LIMITS["read"]
codes = {client.get("/t/" + "B" * 22).status_code for _ in range(limit + 10)}
assert 429 in codes
app_module._hits.clear()
def test_render_handles_dialects():
anthropic = render.normalize({"messages": [
{"role": "assistant", "content": [
{"type": "thinking", "thinking": "hmm"},
{"type": "tool_use", "name": "bash", "input": {"cmd": "ls"}},
]},
{"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "1", "content": "a.txt", "is_error": False}
]},
]})
assert anthropic["kind"] == "messages"
kinds = [b["type"] for m in anthropic["messages"] for b in m["blocks"]]
assert kinds == ["thinking", "tool_use", "tool_result"]
# a user turn carrying only tool results is displayed as the tool speaking
assert anthropic["messages"][1]["role"] == "tool"
openai = render.normalize([{
"role": "assistant",
"content": None,
"tool_calls": [{"function": {"name": "search", "arguments": '{"q": "x"}'}}],
}])
assert openai["messages"][0]["blocks"][0]["name"] == "search"
langchain = render.normalize({"messages": [{"type": "human", "content": "hi"}]})
assert langchain["messages"][0]["role"] == "user"
assert render.normalize("just text")["kind"] == "text"
assert render.normalize({"nested": {"a": 1}})["kind"] == "json"
def test_remote_images_are_not_rendered():
view = render.normalize([{
"role": "user",
"content": [{"type": "image_url", "image_url": {"url": "https://evil.test/p.png"}}],
}])
block = view["messages"][0]["blocks"][0]
assert block["type"] == "note"
assert "evil.test" not in json.dumps(view)
def test_form_post_creates_trace():
app_module._hits.clear()
response = client.post("/", data={"content": "hello", "title": "via form", "token": "s3cret"})
assert response.status_code == 201
assert "/t/" in response.text