From f7cba67c2c87adbd463a9c4ea663de5d671bcbc3 Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 15 Jun 2026 20:51:20 -0700 Subject: [PATCH] test(attachments): handler-level coverage for /thumbnail + the served-blob gate The /thumbnail endpoint and the _resolve_served_blob ownership/404-leak gate it shares with /content had no handler-level test (only make_thumbnail as a unit + route mounting). Add cases through the real app: image -> 200 image/png with the nosniff + CSP + max-age headers; audio/text -> 415; make_thumbnail None -> 415; cross-workstream id and unowned-ws cross-user -> 404 (no existence leak). --- tests/test_server_attachments_endpoints.py | 62 ++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/tests/test_server_attachments_endpoints.py b/tests/test_server_attachments_endpoints.py index b96ca798..ce674599 100644 --- a/tests/test_server_attachments_endpoints.py +++ b/tests/test_server_attachments_endpoints.py @@ -22,6 +22,9 @@ PNG_1x1 = ( b"\xc0\xc0\xc0\x00\x00\x00\x05\x00\x01\xa5\xf6E@\x00\x00\x00\x00IEND\xaeB`\x82" ) +# Magic-byte-valid minimal WAV (RIFF....WAVE) for audio-kind uploads. +WAV_12 = b"RIFF\x24\x00\x00\x00WAVEfmt " + b"\x00" * 16 + _TEST_JWT_SECRET = "test-jwt-secret-minimum-32-chars!" @@ -364,6 +367,65 @@ class TestGetContent: assert resp.content == b"S" +class TestGetThumbnail: + """The /thumbnail handler + the shared _resolve_served_blob gate it reuses: + 200+png for image/pdf, 415 for non-thumbnailable kinds or a failed render, + and a 404 (no existence leak) for cross-ws / cross-user id access.""" + + def _thumb_url(self, ws_id: str, aid: str) -> str: + return f"/v1/api/workstreams/{ws_id}/attachments/{aid}/thumbnail" + + def test_image_thumbnail_200_png_with_hardening_headers(self, app_client): + client, _ = app_client + aid = _upload(client, "ws-A", "userA", "t.png", PNG_1x1, "image/png") + resp = client.get(self._thumb_url("ws-A", aid), headers=_auth("userA")) + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("image/png") + assert resp.content[:8] == b"\x89PNG\r\n\x1a\n" + assert resp.headers.get("x-content-type-options") == "nosniff" + assert "default-src 'none'" in resp.headers.get("content-security-policy", "") + assert "max-age=300" in resp.headers.get("cache-control", "") + + def test_audio_thumbnail_415(self, app_client): + client, _ = app_client + aid = _upload(client, "ws-A", "userA", "a.wav", WAV_12, "audio/wav") + resp = client.get(self._thumb_url("ws-A", aid), headers=_auth("userA")) + assert resp.status_code == 415 + + def test_text_thumbnail_415(self, app_client): + client, _ = app_client + aid = _upload(client, "ws-A", "userA", "n.md", b"# hi\n", "text/markdown") + resp = client.get(self._thumb_url("ws-A", aid), headers=_auth("userA")) + assert resp.status_code == 415 + + def test_thumbnail_unavailable_returns_415(self, app_client, monkeypatch): + # kind is image (reaches make_thumbnail) but the render yields None. + client, _ = app_client + aid = _upload(client, "ws-A", "userA", "t.png", PNG_1x1, "image/png") + monkeypatch.setattr("turnstone.core.thumbnails.make_thumbnail", lambda *a, **k: None) + resp = client.get(self._thumb_url("ws-A", aid), headers=_auth("userA")) + assert resp.status_code == 415 + + def test_thumbnail_cross_workstream_id_404(self, app_client): + client, _ = app_client + aid = _upload(client, "ws-A", "userA", "t.png", PNG_1x1, "image/png") + # userB owns ws-B; the id belongs to ws-A → 404 (no existence leak). + resp = client.get(self._thumb_url("ws-B", aid), headers=_auth("userB")) + assert resp.status_code == 404 + + def test_thumbnail_unowned_ws_user_isolation_404(self, app_client): + client, _ = app_client + from turnstone.core.memory import register_workstream + + register_workstream("ws-shared-thumb", name="shared") + aid = _upload(client, "ws-shared-thumb", "userA", "s.png", PNG_1x1, "image/png") + # Blank-owner ws is reachable by userB, but userA's blob must not be. + resp = client.get(self._thumb_url("ws-shared-thumb", aid), headers=_auth("userB")) + assert resp.status_code == 404 + resp = client.get(self._thumb_url("ws-shared-thumb", aid), headers=_auth("userA")) + assert resp.status_code == 200 + + class TestDelete: def test_delete_pending(self, app_client): client, _ = app_client