Spaces:
Sleeping
Sleeping
File size: 6,293 Bytes
d03af21 e44fdef d03af21 e44fdef d03af21 e44fdef d03af21 e44fdef d03af21 e44fdef d03af21 | 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 180 181 | """Tests for the optional API-key middleware.
This is the only security control in the codebase and had no tests at all:
mutation testing showed that replacing `secrets.compare_digest` with `==`
(a timing-oracle regression) or disabling the guard's condition outright
both left the full suite green. These pin it.
`SERPENT_API_KEY` is read into a module global that the middleware resolves
at call time, so monkeypatching the module attribute is enough - no import
juggling or module reloading needed.
"""
import httpx
import pytest
from httpx import ASGITransport
import app as app_module
import services as services_module
@pytest.fixture
async def client():
transport = ASGITransport(app=app_module.app)
async with httpx.AsyncClient(transport=transport, base_url="http://test") as c:
yield c
@pytest.fixture
def api_key(monkeypatch):
"""Configure a known key for the duration of one test."""
monkeypatch.setattr(app_module, "SERPENT_API_KEY", "s3cret-key")
return "s3cret-key"
@pytest.fixture(autouse=True)
def _unset_by_default(monkeypatch):
"""Every test starts from the shipped default (no key configured), so a
stray SERPENT_API_KEY in the developer's environment can't change the
outcome of the tests below.
"""
monkeypatch.setattr(app_module, "SERPENT_API_KEY", None)
# ------------------------------- disabled by default -------------------------------
async def test_no_key_configured_leaves_the_api_open(client, monkeypatch):
"""The middleware is a no-op unless SERPENT_API_KEY is set - this keeps
the default deployment exactly as open as it was before it existed.
"""
monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
assert resp.status_code == 200
async def _stub_arxiv(client_arg, q, n):
return [{"title": q, "href": "h", "body": "b", "id": "1"}]
# ------------------------------- gating when enabled -------------------------------
async def test_missing_key_is_rejected(client, api_key):
resp = await client.post("/serp/search_arxiv", json={"queries": ["a"]})
assert resp.status_code == 401
assert "API key" in resp.json()["detail"]
async def test_wrong_key_is_rejected(client, api_key):
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
assert resp.status_code == 401
async def test_correct_key_via_x_api_key_header_is_accepted(client, api_key, monkeypatch):
monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": api_key})
assert resp.status_code == 200
async def test_correct_key_via_bearer_header_is_accepted(client, api_key, monkeypatch):
monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]},
headers={"Authorization": f"Bearer {api_key}"})
assert resp.status_code == 200
async def test_bearer_scheme_match_is_case_insensitive(client, api_key, monkeypatch):
monkeypatch.setattr(services_module, "query_arxiv", _stub_arxiv)
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]},
headers={"Authorization": f"bearer {api_key}"})
assert resp.status_code == 200
# --------------------------------- public paths ---------------------------------
@pytest.mark.parametrize("path", ["/", "/openapi.json"])
async def test_docs_paths_stay_public(client, api_key, path):
"""The docs UI and the schema it fetches must stay reachable, otherwise
a key-protected deployment serves an unusable landing page.
"""
resp = await client.get(path)
assert resp.status_code == 200
async def test_a_key_that_is_a_prefix_of_the_real_one_is_rejected(client, api_key):
"""Guards the length check inside the comparison."""
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "s3cret"})
assert resp.status_code == 401
# ------------------------------ constant-time comparison ------------------------------
async def test_key_comparison_is_constant_time(client, api_key, monkeypatch):
"""`==` and `secrets.compare_digest` are functionally identical - they
differ only in timing, so no black-box assertion can tell them apart.
(Confirmed by mutation: swapping in `==` leaves every other test in this
file green.) A timing oracle on the API key is worth pinning anyway, so
assert structurally that the comparison goes through compare_digest.
"""
import secrets as secrets_module
calls = []
real = secrets_module.compare_digest
def spy(a, b):
calls.append((a, b))
return real(a, b)
monkeypatch.setattr(secrets_module, "compare_digest", spy)
await client.post(
"/serp/search_arxiv", json={"queries": ["a"]}, headers={"X-API-Key": "wrong"})
assert calls, "the API key must be compared with secrets.compare_digest"
async def test_non_ascii_key_is_rejected_not_crashed(client, api_key):
"""`secrets.compare_digest` raises TypeError ("comparing strings with
non-ASCII characters is not supported") on str arguments outside ASCII,
and Starlette decodes headers as latin-1 - so an accented byte in the
header reached it intact and turned the security control into an
unhandled 500 for any anonymous caller. It must fail closed with a 401
like any other wrong key.
The header is passed as raw latin-1 bytes because that is what arrives
off the wire; handing httpx a str with non-ASCII in it fails in the
client before the request is ever sent, which would test nothing.
"""
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]},
headers={"X-API-Key": "café".encode("latin-1")})
assert resp.status_code == 401
async def test_non_ascii_key_is_rejected_via_bearer_header_too(client, api_key):
resp = await client.post(
"/serp/search_arxiv", json={"queries": ["a"]},
headers={"Authorization": "Bearer café".encode("latin-1")})
assert resp.status_code == 401
|