tests/test_auth.py

"""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