1aacf4aa20
Protect the research lifecycle with X-API-Key validation backed by persistent user and key records. Store only salted scrypt hashes, support expiry and revocation, and expose a local admin CLI for create/list/revoke workflows. Initialize only the authentication schema at startup, prevent SQL echo from exposing sensitive bound values, and keep health probes public. Add coverage for valid, missing, invalid, expired, and revoked keys. Document deployment and key administration, update the local CLI to send NSCT_API_KEY, and record the reset handoff state.
77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
"""Integration tests for persistent X-API-Key authentication."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timedelta
|
|
from typing import Generator
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from nsct.api.main import create_app
|
|
from nsct.security.api_keys import IssuedAPIKey, create_api_key, revoke_api_key
|
|
from nsct.storage.engine import get_session_factory
|
|
|
|
|
|
@pytest.fixture
|
|
def authenticated_client(monkeypatch: pytest.MonkeyPatch, tmp_path) -> Generator[tuple[TestClient, IssuedAPIKey], None, None]:
|
|
monkeypatch.setenv("NSCT_DB_URL", f"sqlite+aiosqlite:///{tmp_path / 'auth.db'}")
|
|
app = create_app()
|
|
with TestClient(app) as client:
|
|
async def issue(expires_at: datetime | None = None) -> IssuedAPIKey:
|
|
async with get_session_factory()() as session:
|
|
key = await create_api_key(
|
|
session, username="test-user", name="test key", expires_at=expires_at
|
|
)
|
|
await session.commit()
|
|
return key
|
|
|
|
key = client.portal.call(issue)
|
|
client.issue_key = issue # type: ignore[attr-defined]
|
|
yield client, key
|
|
|
|
|
|
def test_missing_key_is_rejected(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, _ = authenticated_client
|
|
response = client.get("/v1/research")
|
|
assert response.status_code == 401
|
|
assert response.headers["www-authenticate"] == "APIKey"
|
|
|
|
|
|
def test_invalid_key_is_rejected(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, key = authenticated_client
|
|
response = client.get("/v1/research", headers={"X-API-Key": key.key[:-1] + "x"})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_valid_key_is_accepted(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, key = authenticated_client
|
|
response = client.get("/v1/research", headers={"X-API-Key": key.key})
|
|
assert response.status_code == 200
|
|
|
|
|
|
def test_expired_key_is_rejected(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, _ = authenticated_client
|
|
expired = client.portal.call(client.issue_key, datetime.utcnow() - timedelta(seconds=1)) # type: ignore[attr-defined]
|
|
response = client.get("/v1/research", headers={"X-API-Key": expired.key})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_revoked_key_is_rejected(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, key = authenticated_client
|
|
|
|
async def revoke() -> bool:
|
|
async with get_session_factory()() as session:
|
|
result = await revoke_api_key(session, key.key_id)
|
|
await session.commit()
|
|
return result
|
|
|
|
assert client.portal.call(revoke)
|
|
response = client.get("/v1/research", headers={"X-API-Key": key.key})
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_health_remains_public(authenticated_client: tuple[TestClient, IssuedAPIKey]) -> None:
|
|
client, _ = authenticated_client
|
|
assert client.get("/health").status_code == 200
|