Implement persistent API-key authentication

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.
This commit is contained in:
faligam
2026-09-06 17:23:05 +02:00
parent 64eb4e8465
commit 1aacf4aa20
14 changed files with 1065 additions and 60 deletions

View File

@@ -0,0 +1,76 @@
"""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