File size: 6,468 Bytes
d0eee72
 
 
 
 
 
08565c0
 
d0eee72
 
 
 
 
 
 
 
 
 
 
 
6b41207
d0eee72
6b41207
 
 
 
 
d0eee72
6b41207
 
d0eee72
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
08565c0
 
 
 
 
 
 
d0eee72
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
"""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": "<script>alert(1)</script>"}])
    page = client.get(f"/t/{created['id']}")
    assert "<script>alert(1)</script>" not in page.text
    assert "&lt;script&gt;" 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