Spaces:
Sleeping
Test the API-key middleware, and stop it 500ing on a non-ASCII key
Browse filesThe optional API-key guard was the only security control in the codebase
and had no tests at all. Mutation testing showed what that cost: replacing
`secrets.compare_digest` with `==` (a timing-oracle regression) or
disabling the guard's condition outright both left the whole suite green.
Add tests/test_auth.py covering the behaviour that was already there -
open when unset, 401 on a missing/wrong/prefix key, both header forms
accepted, bearer scheme matched case-insensitively, and the docs paths
staying public - all confirmed green against the unchanged middleware
first. Three of the four mutants above now fail the suite.
The fourth cannot be caught behaviourally: `==` and `compare_digest`
return the same answer for every input and differ only in timing. Since a
timing oracle on the API key is worth pinning anyway, that one test asserts
structurally that the comparison goes through compare_digest.
Then fix a real defect those tests exposed. `compare_digest` raises
TypeError on str arguments containing non-ASCII characters, and Starlette
decodes headers as latin-1, so `X-API-Key: café` reached it intact and
turned the security control into an unhandled 500 that any anonymous
caller could trigger. Compare as bytes instead; it now fails closed with a
401 like any other wrong key.
Note for anyone extending these tests: the header has to be passed as raw
latin-1 bytes, because handing httpx a str with non-ASCII in it fails
inside the client before a request is ever sent.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MSNSYnceqvdVz7Csis4K9e
- app.py +6 -1
- tests/test_auth.py +179 -0
|
@@ -99,7 +99,12 @@ async def api_key_guard(request: Request, call_next):
|
|
| 99 |
"""No-op unless SERPENT_API_KEY is set; then gates every path but the docs."""
|
| 100 |
if SERPENT_API_KEY and request.url.path not in _PUBLIC_PATHS:
|
| 101 |
supplied = _extract_supplied_key(request)
|
| 102 |
-
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 103 |
return JSONResponse(
|
| 104 |
{"detail": "Missing or invalid API key. Supply it via the X-API-Key or Authorization: Bearer header."},
|
| 105 |
status_code=401,
|
|
|
|
| 99 |
"""No-op unless SERPENT_API_KEY is set; then gates every path but the docs."""
|
| 100 |
if SERPENT_API_KEY and request.url.path not in _PUBLIC_PATHS:
|
| 101 |
supplied = _extract_supplied_key(request)
|
| 102 |
+
# Compare as bytes: compare_digest raises TypeError on str arguments
|
| 103 |
+
# containing non-ASCII characters, and Starlette decodes headers as
|
| 104 |
+
# latin-1, so an accented byte in the header would otherwise turn
|
| 105 |
+
# this check into an unhandled 500 for any anonymous caller.
|
| 106 |
+
if not supplied or not secrets.compare_digest(
|
| 107 |
+
supplied.encode("utf-8"), SERPENT_API_KEY.encode("utf-8")):
|
| 108 |
return JSONResponse(
|
| 109 |
{"detail": "Missing or invalid API key. Supply it via the X-API-Key or Authorization: Bearer header."},
|
| 110 |
status_code=401,
|
|
@@ -0,0 +1,179 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
+
"""Tests for the optional API-key middleware.
|
| 2 |
+
|
| 3 |
+
This is the only security control in the codebase and had no tests at all:
|
| 4 |
+
mutation testing showed that replacing `secrets.compare_digest` with `==`
|
| 5 |
+
(a timing-oracle regression) or disabling the guard's condition outright
|
| 6 |
+
both left the full suite green. These pin it.
|
| 7 |
+
|
| 8 |
+
`SERPENT_API_KEY` is read into a module global that the middleware resolves
|
| 9 |
+
at call time, so monkeypatching the module attribute is enough - no import
|
| 10 |
+
juggling or module reloading needed.
|
| 11 |
+
"""
|
| 12 |
+
|
| 13 |
+
import httpx
|
| 14 |
+
import pytest
|
| 15 |
+
from httpx import ASGITransport
|
| 16 |
+
|
| 17 |
+
import app as app_module
|
| 18 |
+
|
| 19 |
+
|
| 20 |
+
@pytest.fixture
|
| 21 |
+
async def client():
|
| 22 |
+
transport = ASGITransport(app=app_module.app)
|
| 23 |
+
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
|
| 24 |
+
yield c
|
| 25 |
+
|
| 26 |
+
|
| 27 |
+
@pytest.fixture
|
| 28 |
+
def api_key(monkeypatch):
|
| 29 |
+
"""Configure a known key for the duration of one test."""
|
| 30 |
+
monkeypatch.setattr(app_module, "SERPENT_API_KEY", "s3cret-key")
|
| 31 |
+
return "s3cret-key"
|
| 32 |
+
|
| 33 |
+
|
| 34 |
+
@pytest.fixture(autouse=True)
|
| 35 |
+
def _unset_by_default(monkeypatch):
|
| 36 |
+
"""Every test starts from the shipped default (no key configured), so a
|
| 37 |
+
stray SERPENT_API_KEY in the developer's environment can't change the
|
| 38 |
+
outcome of the tests below.
|
| 39 |
+
"""
|
| 40 |
+
monkeypatch.setattr(app_module, "SERPENT_API_KEY", None)
|
| 41 |
+
|
| 42 |
+
|
| 43 |
+
# ------------------------------- disabled by default -------------------------------
|
| 44 |
+
|
| 45 |
+
|
| 46 |
+
async def test_no_key_configured_leaves_the_api_open(client, monkeypatch):
|
| 47 |
+
"""The middleware is a no-op unless SERPENT_API_KEY is set - this keeps
|
| 48 |
+
the default deployment exactly as open as it was before it existed.
|
| 49 |
+
"""
|
| 50 |
+
monkeypatch.setattr(app_module, "query_arxiv", _stub_arxiv)
|
| 51 |
+
|
| 52 |
+
resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
|
| 53 |
+
|
| 54 |
+
assert resp.status_code == 200
|
| 55 |
+
|
| 56 |
+
|
| 57 |
+
async def _stub_arxiv(client_arg, q, n):
|
| 58 |
+
return [{"title": q, "href": "h", "body": "b", "id": "1"}]
|
| 59 |
+
|
| 60 |
+
|
| 61 |
+
# ------------------------------- gating when enabled -------------------------------
|
| 62 |
+
|
| 63 |
+
|
| 64 |
+
async def test_missing_key_is_rejected(client, api_key):
|
| 65 |
+
resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
|
| 66 |
+
|
| 67 |
+
assert resp.status_code == 401
|
| 68 |
+
assert "API key" in resp.json()["detail"]
|
| 69 |
+
|
| 70 |
+
|
| 71 |
+
async def test_wrong_key_is_rejected(client, api_key):
|
| 72 |
+
resp = await client.post(
|
| 73 |
+
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
|
| 74 |
+
|
| 75 |
+
assert resp.status_code == 401
|
| 76 |
+
|
| 77 |
+
|
| 78 |
+
async def test_correct_key_via_x_api_key_header_is_accepted(client, api_key, monkeypatch):
|
| 79 |
+
monkeypatch.setattr(app_module, "query_arxiv", _stub_arxiv)
|
| 80 |
+
|
| 81 |
+
resp = await client.post(
|
| 82 |
+
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": api_key})
|
| 83 |
+
|
| 84 |
+
assert resp.status_code == 200
|
| 85 |
+
|
| 86 |
+
|
| 87 |
+
async def test_correct_key_via_bearer_header_is_accepted(client, api_key, monkeypatch):
|
| 88 |
+
monkeypatch.setattr(app_module, "query_arxiv", _stub_arxiv)
|
| 89 |
+
|
| 90 |
+
resp = await client.post(
|
| 91 |
+
"/serp/search_arxiv", json={"queries": ["a"]},
|
| 92 |
+
headers={"Authorization": f"Bearer {api_key}"})
|
| 93 |
+
|
| 94 |
+
assert resp.status_code == 200
|
| 95 |
+
|
| 96 |
+
|
| 97 |
+
async def test_bearer_scheme_match_is_case_insensitive(client, api_key, monkeypatch):
|
| 98 |
+
monkeypatch.setattr(app_module, "query_arxiv", _stub_arxiv)
|
| 99 |
+
|
| 100 |
+
resp = await client.post(
|
| 101 |
+
"/serp/search_arxiv", json={"queries": ["a"]},
|
| 102 |
+
headers={"Authorization": f"bearer {api_key}"})
|
| 103 |
+
|
| 104 |
+
assert resp.status_code == 200
|
| 105 |
+
|
| 106 |
+
|
| 107 |
+
# --------------------------------- public paths ---------------------------------
|
| 108 |
+
|
| 109 |
+
|
| 110 |
+
@pytest.mark.parametrize("path", ["/", "/openapi.json"])
|
| 111 |
+
async def test_docs_paths_stay_public(client, api_key, path):
|
| 112 |
+
"""The docs UI and the schema it fetches must stay reachable, otherwise
|
| 113 |
+
a key-protected deployment serves an unusable landing page.
|
| 114 |
+
"""
|
| 115 |
+
resp = await client.get(path)
|
| 116 |
+
|
| 117 |
+
assert resp.status_code == 200
|
| 118 |
+
|
| 119 |
+
|
| 120 |
+
async def test_a_key_that_is_a_prefix_of_the_real_one_is_rejected(client, api_key):
|
| 121 |
+
"""Guards the length check inside the comparison."""
|
| 122 |
+
resp = await client.post(
|
| 123 |
+
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "s3cret"})
|
| 124 |
+
|
| 125 |
+
assert resp.status_code == 401
|
| 126 |
+
|
| 127 |
+
|
| 128 |
+
# ------------------------------ constant-time comparison ------------------------------
|
| 129 |
+
|
| 130 |
+
|
| 131 |
+
async def test_key_comparison_is_constant_time(client, api_key, monkeypatch):
|
| 132 |
+
"""`==` and `secrets.compare_digest` are functionally identical - they
|
| 133 |
+
differ only in timing, so no black-box assertion can tell them apart.
|
| 134 |
+
(Confirmed by mutation: swapping in `==` leaves every other test in this
|
| 135 |
+
file green.) A timing oracle on the API key is worth pinning anyway, so
|
| 136 |
+
assert structurally that the comparison goes through compare_digest.
|
| 137 |
+
"""
|
| 138 |
+
import secrets as secrets_module
|
| 139 |
+
|
| 140 |
+
calls = []
|
| 141 |
+
real = secrets_module.compare_digest
|
| 142 |
+
|
| 143 |
+
def spy(a, b):
|
| 144 |
+
calls.append((a, b))
|
| 145 |
+
return real(a, b)
|
| 146 |
+
|
| 147 |
+
monkeypatch.setattr(secrets_module, "compare_digest", spy)
|
| 148 |
+
|
| 149 |
+
await client.post(
|
| 150 |
+
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
|
| 151 |
+
|
| 152 |
+
assert calls, "the API key must be compared with secrets.compare_digest"
|
| 153 |
+
|
| 154 |
+
|
| 155 |
+
async def test_non_ascii_key_is_rejected_not_crashed(client, api_key):
|
| 156 |
+
"""`secrets.compare_digest` raises TypeError ("comparing strings with
|
| 157 |
+
non-ASCII characters is not supported") on str arguments outside ASCII,
|
| 158 |
+
and Starlette decodes headers as latin-1 - so an accented byte in the
|
| 159 |
+
header reached it intact and turned the security control into an
|
| 160 |
+
unhandled 500 for any anonymous caller. It must fail closed with a 401
|
| 161 |
+
like any other wrong key.
|
| 162 |
+
|
| 163 |
+
The header is passed as raw latin-1 bytes because that is what arrives
|
| 164 |
+
off the wire; handing httpx a str with non-ASCII in it fails in the
|
| 165 |
+
client before the request is ever sent, which would test nothing.
|
| 166 |
+
"""
|
| 167 |
+
resp = await client.post(
|
| 168 |
+
"/serp/search_arxiv", json={"queries": ["a"]},
|
| 169 |
+
headers={"X-API-Key": "café".encode("latin-1")})
|
| 170 |
+
|
| 171 |
+
assert resp.status_code == 401
|
| 172 |
+
|
| 173 |
+
|
| 174 |
+
async def test_non_ascii_key_is_rejected_via_bearer_header_too(client, api_key):
|
| 175 |
+
resp = await client.post(
|
| 176 |
+
"/serp/search_arxiv", json={"queries": ["a"]},
|
| 177 |
+
headers={"Authorization": "Bearer café".encode("latin-1")})
|
| 178 |
+
|
| 179 |
+
assert resp.status_code == 401
|