diff --git a/backend/secuscan/auth.py b/backend/secuscan/auth.py index b4ef744cb..911dcc70c 100644 --- a/backend/secuscan/auth.py +++ b/backend/secuscan/auth.py @@ -1,11 +1,23 @@ """ API key authentication for SecuScan backend. -A random key is generated at startup and written to /.api_key. +A random key is generated at startup and written to /.api_key as +JSON (``{"key": ..., "created_at": }``), which lets the key +carry an age/expiry (issue #1619) instead of being valid indefinitely with +no revocation path short of deleting the file and restarting the process. +Older deployments with a plaintext key file are migrated in place on next +boot -- the existing key is kept, wrapped with a fresh created_at. + Clients must supply it via: - Authorization: Bearer - X-Api-Key: +Rotation and expiry (admin-key-gated, issue #1619): + - POST /api/v1/admin/api-key/rotate — generate a new key, invalidate the old one + - GET /api/v1/admin/api-key/status — report key age and (if configured) expiry + - SECUSCAN_API_KEY_TTL_SECONDS (default 0 = disabled) rejects requests once the + key is older than the configured TTL, forcing rotation. + Session management (signed cookie, no server-side state): - POST /api/v1/auth/session — validate API key and set HttpOnly session cookie - GET /api/v1/auth/session/check — verify active session cookie @@ -28,6 +40,8 @@ _api_key_header = APIKeyHeader(name="X-Api-Key", auto_error=False) _api_key: str | None = None +_api_key_created_at: float | None = None +_api_key_file: Path | None = None SESSION_TTL_SECONDS = 3600 # 1 hour COOKIE_NAME = "secuscan_session" @@ -108,6 +122,16 @@ async def create_session(request: Request, response: Response): if not candidate or not secrets.compare_digest(candidate, _api_key): raise HTTPException(status_code=401, detail="Invalid API key") + from .config import settings + + ttl = settings.api_key_ttl_seconds + if ttl > 0 and _api_key_created_at is not None and time.time() - _api_key_created_at > ttl: + raise HTTPException( + status_code=401, + detail="API key has expired. An administrator must rotate it via " + "POST /api/v1/admin/api-key/rotate.", + ) + token = _make_signed_token() response.set_cookie( key=COOKIE_NAME, @@ -141,27 +165,105 @@ def is_authenticated_by_session(request: Request) -> bool: return bool(token and _verify_signed_token(token)) +def _resolve_key_file(data_dir: str) -> Path: + # Allow operators to redirect the key file via env var (e.g. Docker secrets). + custom_path = os.environ.get("SECUSCAN_API_KEY_FILE", "").strip() + return Path(custom_path) if custom_path else Path(data_dir) / ".api_key" + + +def _write_key_file(key_file: Path, key: str, created_at: float) -> None: + key_file.parent.mkdir(parents=True, exist_ok=True) + payload = json.dumps({"key": key, "created_at": created_at}) + # Write to a temp file in the same directory and atomically rename, so a + # crash mid-write can never leave a truncated/corrupt key file behind. + tmp_path = key_file.with_suffix(f"{key_file.suffix}.tmp") + tmp_path.write_text(payload) + tmp_path.chmod(0o600) + tmp_path.replace(key_file) + + def init_api_key(data_dir: str) -> str: """ Load the persisted API key, or generate and persist a new one. Called once during application startup; the returned key is also stored in the module-level ``_api_key`` variable so the FastAPI dependency can reach it. + + The key file is JSON (``{"key": ..., "created_at": ...}``) so the key's age + can be checked against ``SECUSCAN_API_KEY_TTL_SECONDS`` and reported via + ``GET /api/v1/admin/api-key/status``. A pre-existing plaintext key file + (from before issue #1619) is migrated in place: the key is kept, wrapped + with a fresh ``created_at`` timestamp. """ - global _api_key - # Allow operators to redirect the key file via env var (e.g. Docker secrets). - custom_path = os.environ.get("SECUSCAN_API_KEY_FILE", "").strip() - key_file = Path(custom_path) if custom_path else Path(data_dir) / ".api_key" + global _api_key, _api_key_created_at, _api_key_file + key_file = _resolve_key_file(data_dir) + _api_key_file = key_file + if key_file.exists(): - _api_key = key_file.read_text().strip() + raw = key_file.read_text().strip() + try: + data = json.loads(raw) + _api_key = str(data["key"]) + _api_key_created_at = float(data["created_at"]) + except (json.JSONDecodeError, KeyError, TypeError, ValueError): + # Legacy plaintext key file -- keep the existing key, but it has + # no known creation time, so treat it as freshly issued and + # migrate the file to the new JSON format. + _api_key = raw + _api_key_created_at = time.time() + _write_key_file(key_file, _api_key, _api_key_created_at) else: _api_key = secrets.token_hex(32) - key_file.parent.mkdir(parents=True, exist_ok=True) - key_file.write_text(_api_key) - key_file.chmod(0o600) + _api_key_created_at = time.time() + _write_key_file(key_file, _api_key, _api_key_created_at) + return _api_key +def rotate_api_key() -> dict: + """Generate a new API key, persist it, and invalidate the old one. + + The old key stops working the instant this returns, since ``_api_key`` + is the single value ``require_api_key`` compares against -- there is no + grace period. Returns the new key so the caller (an admin-authenticated + request) can retrieve it; there is no other way to read it back out, + since the key file is not exposed over the API. + """ + global _api_key, _api_key_created_at + if _api_key_file is None: + raise RuntimeError("API key not initialised -- call init_api_key() first") + + new_key = secrets.token_hex(32) + created_at = time.time() + _write_key_file(_api_key_file, new_key, created_at) + _api_key = new_key + _api_key_created_at = created_at + return {"key": new_key, "created_at": created_at} + + +def get_api_key_status() -> dict: + """Report the current key's age and, if SECUSCAN_API_KEY_TTL_SECONDS is + set, when it expires -- without exposing the key itself.""" + from .config import settings + + ttl = settings.api_key_ttl_seconds + age_seconds = None + expires_at = None + expired = False + if _api_key_created_at is not None: + age_seconds = time.time() - _api_key_created_at + if ttl > 0: + expires_at = _api_key_created_at + ttl + expired = age_seconds > ttl + return { + "created_at": _api_key_created_at, + "age_seconds": age_seconds, + "ttl_seconds": ttl or None, + "expires_at": expires_at, + "expired": expired, + } + + async def require_api_key( request: Request = None, bearer: HTTPAuthorizationCredentials | None = Depends(_bearer_scheme), @@ -204,6 +306,17 @@ async def require_api_key( headers={"WWW-Authenticate": "Bearer"}, ) + from .config import settings + + ttl = settings.api_key_ttl_seconds + if ttl > 0 and _api_key_created_at is not None and time.time() - _api_key_created_at > ttl: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="API key has expired. An administrator must rotate it via " + "POST /api/v1/admin/api-key/rotate.", + headers={"WWW-Authenticate": "Bearer"}, + ) + return candidate diff --git a/backend/secuscan/config.py b/backend/secuscan/config.py index cd66b1b1c..07cf858a2 100644 --- a/backend/secuscan/config.py +++ b/backend/secuscan/config.py @@ -66,6 +66,13 @@ class Settings(BaseSettings): denied_capabilities: List[str] = [] admin_api_key: Optional[str] = None + # API key expiry (issue #1619): the client API key stored at + # /.api_key is otherwise valid indefinitely with no way to + # revoke it short of deleting the file and restarting the process. 0 + # (default) disables expiry so existing deployments keep working + # unchanged until an operator opts in. + api_key_ttl_seconds: int = 0 + # Network Policy Configuration network_allowlist: List[str] = [] # IPs/networks to allow (CIDR); empty = deny all egress network_denylist: List[str] = [ # IPs/networks to deny (CIDR) diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index f4b3289c9..2334eb728 100644 --- a/backend/secuscan/routes.py +++ b/backend/secuscan/routes.py @@ -117,7 +117,7 @@ def _json_payload(value: Any, fallback: str) -> str: from .reporting import reporting from .vault import VaultCrypto from .workflows import scheduler, _finalize_workflow_run -from .auth import require_api_key, get_current_owner +from .auth import require_api_key, get_current_owner, rotate_api_key, get_api_key_status from .execution_context import is_offensive_validation, normalize_execution_context from .finding_intelligence import build_asset_summary, build_finding_groups from .knowledgebase import KnowledgeBase @@ -2660,6 +2660,42 @@ def verify_admin_access( ) return candidate +# ── Client API key rotation & expiry (issue #1619) ────────────────────────── +# +# The client API key (backend/data/.api_key) previously had no TTL, rotation +# endpoint, or revocation mechanism -- once issued it was valid forever, so a +# key leaked via a co-located container, path traversal, or leaked backup +# stayed usable indefinitely. These endpoints are gated by the *admin* API +# key (a separate, statically-configured secret) rather than the client key +# itself, since an attacker holding a leaked client key must not be able to +# use it to mint itself a fresh one. + +@router.post( + "/admin/api-key/rotate", + dependencies=[Depends(verify_admin_access), Depends(admin_limiter)], +) +async def rotate_client_api_key(): + """Generate a new client API key and invalidate the old one immediately. + + Returns the new key -- this is the only way to retrieve it, since it is + never exposed anywhere else. Existing sessions established via the old + key continue until their own TTL expires (POST /auth/session/logout to + revoke immediately); the old key itself can no longer authenticate a new + request or a new session as soon as this returns. + """ + return rotate_api_key() + + +@router.get( + "/admin/api-key/status", + dependencies=[Depends(verify_admin_access), Depends(admin_limiter)], +) +async def get_client_api_key_status(): + """Report the client API key's age and, if SECUSCAN_API_KEY_TTL_SECONDS + is configured, its expiry -- without exposing the key itself.""" + return get_api_key_status() + + @router.get( "/admin/diagnostics/notifications", response_model=NotificationDiagnosticsResponse, diff --git a/docs/api-authentication.md b/docs/api-authentication.md index 8837f9530..6177c8d83 100644 --- a/docs/api-authentication.md +++ b/docs/api-authentication.md @@ -12,15 +12,47 @@ writes it to `/.api_key` (mode `0600`), and prints it to the console: ✓ API key authentication ready (key file: backend/data/.api_key) ``` -On every subsequent start the same key is loaded from the file. +On every subsequent start the same key is loaded from the file. The key file is +JSON (`{"key": ..., "created_at": }`) so the key's age can be +tracked; a pre-existing plaintext key file from before this format is migrated +in place on next startup, keeping the same key. -To rotate the key, delete the file and restart the backend: +### Rotation and expiry + +Rotating the key no longer requires deleting the file and restarting the +process. `POST /api/v1/admin/api-key/rotate` (gated by the separate admin API +key, `SECUSCAN_ADMIN_API_KEY`) generates a new key and invalidates the old one +immediately: + +```bash +curl -X POST -H "X-API-Key: $ADMIN_API_KEY" \ + http://localhost:8000/api/v1/admin/api-key/rotate +# {"key": "", "created_at": 1783...} +``` + +The response is the only place the new key is returned — it is not exposed +anywhere else, so save it immediately. `GET /api/v1/admin/api-key/status` +reports the current key's age (and expiry, if configured) without exposing the +key itself: ```bash -rm backend/data/.api_key -python -m secuscan # a new key is generated on startup +curl -H "X-API-Key: $ADMIN_API_KEY" \ + http://localhost:8000/api/v1/admin/api-key/status +# {"created_at": 1783..., "age_seconds": 42, "ttl_seconds": null, "expires_at": null, "expired": false} ``` +Set `SECUSCAN_API_KEY_TTL_SECONDS` (default `0`, disabled) to have the backend +automatically reject the client key once it is older than the configured TTL, +forcing an operator to rotate it: + +```bash +export SECUSCAN_API_KEY_TTL_SECONDS=2592000 # 30 days +``` + +The admin key itself is a separate, statically-configured secret (not subject +to this rotation flow) specifically so that a leaked *client* key can never be +used to mint itself a replacement. + ## Frontend / UI The web UI does **not** fetch the key from the backend. You must configure it diff --git a/testing/backend/unit/test_api_auth.py b/testing/backend/unit/test_api_auth.py index 53a8b587b..a5705d944 100644 --- a/testing/backend/unit/test_api_auth.py +++ b/testing/backend/unit/test_api_auth.py @@ -49,7 +49,10 @@ def test_secuscan_api_key_file_env_var(self, tmp_path, monkeypatch): monkeypatch.setenv("SECUSCAN_API_KEY_FILE", str(custom_path)) key = auth_module.init_api_key(str(tmp_path)) assert custom_path.exists() - assert custom_path.read_text().strip() == key + # Key file is JSON ({"key": ..., "created_at": ...}, issue #1619) so + # the key's age can be tracked, not a bare plaintext key. + import json as _json + assert _json.loads(custom_path.read_text())["key"] == key def test_secuscan_api_key_file_loads_existing(self, tmp_path, monkeypatch): custom_path = tmp_path / "my_key" diff --git a/testing/backend/unit/test_api_key_rotation.py b/testing/backend/unit/test_api_key_rotation.py new file mode 100644 index 000000000..87a5bdbbd --- /dev/null +++ b/testing/backend/unit/test_api_key_rotation.py @@ -0,0 +1,171 @@ +""" +Unit tests for API key rotation and expiry (issue #1619). + +The client API key file previously had no created_at, TTL, or rotation +path -- once issued it was valid forever with no way to revoke it short of +deleting the file and restarting the process. These tests cover: + +- The key file now stores created_at alongside the key (JSON format). +- A pre-existing plaintext key file is migrated in place without losing + the key itself. +- POST /admin/api-key/rotate invalidates the old key immediately and + returns a usable new one. +- GET /admin/api-key/status reports age/expiry without exposing the key. +- SECUSCAN_API_KEY_TTL_SECONDS rejects requests once the key is stale. +- Both admin endpoints are gated by the *admin* key, not the client key. +""" + +import asyncio +import json +import time + +import pytest +from fastapi.testclient import TestClient + +from backend.secuscan import auth as auth_module +from backend.secuscan.main import app +from backend.secuscan.config import settings +from backend.secuscan.database import init_db +from backend.secuscan.plugins import init_plugins + + +ADMIN_KEY = "valid-admin-key-for-rotation-tests" + + +@pytest.fixture() +def client_with_key(setup_test_environment, monkeypatch): + """TestClient with a valid client API key pre-seeded and a valid + (>= 16 char) admin key, since verify_admin_access rejects the + conftest default ("test-admin-key", 14 chars) as too weak.""" + monkeypatch.setattr(settings, "admin_api_key", ADMIN_KEY) + asyncio.run(init_db(settings.database_path)) + asyncio.run(init_plugins(settings.plugins_dir)) + api_key = auth_module.init_api_key(settings.data_dir) + with TestClient(app) as c: + yield c, api_key + + +class TestKeyFileFormat: + def test_key_file_stores_created_at(self, tmp_path): + auth_module.init_api_key(str(tmp_path)) + data = json.loads((tmp_path / ".api_key").read_text()) + assert "key" in data + assert "created_at" in data + assert isinstance(data["created_at"], (int, float)) + + def test_legacy_plaintext_key_file_is_migrated(self, tmp_path): + key_file = tmp_path / ".api_key" + key_file.write_text("legacy-plaintext-key-123") + + loaded = auth_module.init_api_key(str(tmp_path)) + + assert loaded == "legacy-plaintext-key-123" + data = json.loads(key_file.read_text()) + assert data["key"] == "legacy-plaintext-key-123" + assert isinstance(data["created_at"], (int, float)) + + def test_reloading_json_key_file_preserves_created_at(self, tmp_path): + auth_module.init_api_key(str(tmp_path)) + first = json.loads((tmp_path / ".api_key").read_text()) + + auth_module.init_api_key(str(tmp_path)) + second = json.loads((tmp_path / ".api_key").read_text()) + + assert first["key"] == second["key"] + assert first["created_at"] == second["created_at"] + + +class TestRotateEndpoint: + def test_rotate_requires_admin_key(self, client_with_key): + client, _ = client_with_key + resp = client.post("/api/v1/admin/api-key/rotate", headers={}) + assert resp.status_code == 401 + + def test_client_key_cannot_rotate_itself(self, client_with_key): + client, api_key = client_with_key + resp = client.post("/api/v1/admin/api-key/rotate", headers={"X-API-Key": api_key}) + assert resp.status_code == 401 + + def test_rotate_returns_new_key(self, client_with_key): + client, old_key = client_with_key + resp = client.post("/api/v1/admin/api-key/rotate", headers={"X-API-Key": ADMIN_KEY}) + assert resp.status_code == 200 + body = resp.json() + assert "key" in body + assert body["key"] != old_key + assert len(body["key"]) == 64 # 32 bytes -> 64 hex chars + + def test_old_key_rejected_immediately_after_rotation(self, client_with_key): + client, old_key = client_with_key + client.post("/api/v1/admin/api-key/rotate", headers={"X-API-Key": ADMIN_KEY}) + + resp = client.get("/api/v1/plugins", headers={"X-Api-Key": old_key}) + assert resp.status_code == 401 + + def test_new_key_works_after_rotation(self, client_with_key): + client, _ = client_with_key + rotate_resp = client.post("/api/v1/admin/api-key/rotate", headers={"X-API-Key": ADMIN_KEY}) + new_key = rotate_resp.json()["key"] + + resp = client.get("/api/v1/plugins", headers={"X-Api-Key": new_key}) + assert resp.status_code == 200 + + def test_rotation_persists_to_disk(self, client_with_key): + client, _ = client_with_key + rotate_resp = client.post("/api/v1/admin/api-key/rotate", headers={"X-API-Key": ADMIN_KEY}) + new_key = rotate_resp.json()["key"] + + on_disk = json.loads((__import__("pathlib").Path(settings.data_dir) / ".api_key").read_text()) + assert on_disk["key"] == new_key + + +class TestStatusEndpoint: + def test_status_requires_admin_key(self, client_with_key): + client, _ = client_with_key + resp = client.get("/api/v1/admin/api-key/status", headers={}) + assert resp.status_code == 401 + + def test_status_reports_age_without_exposing_key(self, client_with_key): + client, api_key = client_with_key + resp = client.get("/api/v1/admin/api-key/status", headers={"X-API-Key": ADMIN_KEY}) + assert resp.status_code == 200 + body = resp.json() + assert "created_at" in body + assert "age_seconds" in body + assert body["age_seconds"] >= 0 + assert api_key not in json.dumps(body) + + def test_status_ttl_disabled_by_default(self, client_with_key): + client, _ = client_with_key + resp = client.get("/api/v1/admin/api-key/status", headers={"X-API-Key": ADMIN_KEY}) + body = resp.json() + assert body["ttl_seconds"] is None + assert body["expired"] is False + + +class TestTTLExpiry: + def test_expired_key_rejected(self, client_with_key, monkeypatch): + client, api_key = client_with_key + monkeypatch.setattr(settings, "api_key_ttl_seconds", 1) + # Backdate the in-memory created_at so the key reads as already expired. + monkeypatch.setattr(auth_module, "_api_key_created_at", time.time() - 10) + + resp = client.get("/api/v1/plugins", headers={"X-Api-Key": api_key}) + assert resp.status_code == 401 + assert "expired" in resp.json()["detail"].lower() + + def test_non_expired_key_within_ttl_is_accepted(self, client_with_key, monkeypatch): + client, api_key = client_with_key + monkeypatch.setattr(settings, "api_key_ttl_seconds", 3600) + monkeypatch.setattr(auth_module, "_api_key_created_at", time.time()) + + resp = client.get("/api/v1/plugins", headers={"X-Api-Key": api_key}) + assert resp.status_code == 200 + + def test_session_creation_rejects_expired_key(self, client_with_key, monkeypatch): + client, api_key = client_with_key + monkeypatch.setattr(settings, "api_key_ttl_seconds", 1) + monkeypatch.setattr(auth_module, "_api_key_created_at", time.time() - 10) + + resp = client.post("/api/v1/auth/session", headers={"X-Api-Key": api_key}) + assert resp.status_code == 401 diff --git a/testing/backend/unit/test_auth.py b/testing/backend/unit/test_auth.py index 9698fb726..9bd21074c 100644 --- a/testing/backend/unit/test_auth.py +++ b/testing/backend/unit/test_auth.py @@ -10,6 +10,7 @@ - init_api_key raises OSError on unwritable directory """ +import json import os import stat import tempfile @@ -30,6 +31,11 @@ def fresh_key(data_dir: str) -> str: return auth_module.init_api_key(data_dir) +def _read_key(key_file: Path) -> str: + """Key files are JSON ({"key": ..., "created_at": ...}, issue #1619).""" + return json.loads(key_file.read_text())["key"] + + class TestInitApiKey: def test_generates_new_key_when_no_file_exists(self, tmp_path: Path): """A new key is generated and written to /.api_key.""" @@ -39,10 +45,24 @@ def test_generates_new_key_when_no_file_exists(self, tmp_path: Path): assert key is not None assert len(key) == 64 # secrets.token_hex(32) -> 64 hex chars assert key_file.exists() - assert key_file.read_text().strip() == key + assert _read_key(key_file) == key def test_loads_existing_key_without_regenerating(self, tmp_path: Path): - """An existing key file is not overwritten.""" + """An existing JSON key file's key is loaded unchanged.""" + key_file = tmp_path / ".api_key" + existing = "deadbeef" * 8 # 64-char hex + key_file.write_text(json.dumps({"key": existing, "created_at": 1700000000.0})) + + loaded = fresh_key(str(tmp_path)) + + assert loaded == existing + # created_at (and therefore the key) must not have been regenerated. + data = json.loads(key_file.read_text()) + assert data["key"] == existing + assert data["created_at"] == 1700000000.0 + + def test_legacy_plaintext_key_is_migrated_not_replaced(self, tmp_path: Path): + """A pre-#1619 plaintext key file keeps its key, wrapped in JSON.""" key_file = tmp_path / ".api_key" existing = "deadbeef" * 8 # 64-char hex key_file.write_text(existing + "\n") @@ -50,8 +70,7 @@ def test_loads_existing_key_without_regenerating(self, tmp_path: Path): loaded = fresh_key(str(tmp_path)) assert loaded == existing - # File must not have been rewritten - assert key_file.read_text().strip() == existing + assert _read_key(key_file) == existing def test_respects_env_var_custom_path(self, tmp_path: Path, monkeypatch): """SECUSCAN_API_KEY_FILE redirects the key file location.""" @@ -61,7 +80,7 @@ def test_respects_env_var_custom_path(self, tmp_path: Path, monkeypatch): key = fresh_key(str(tmp_path)) assert custom_file.exists() - assert custom_file.read_text().strip() == key + assert _read_key(custom_file) == key # No .api_key in data_dir either assert not (tmp_path / ".api_key").exists() @@ -76,7 +95,7 @@ def test_file_mode_is_0600(self, tmp_path: Path): def test_returns_the_loaded_or_generated_key(self, tmp_path: Path): """init_api_key returns the same value that is written to the file.""" key = fresh_key(str(tmp_path)) - assert key == (tmp_path / ".api_key").read_text().strip() + assert key == _read_key(tmp_path / ".api_key") def test_creates_parent_directories(self, tmp_path: Path): """init_api_key creates parent directories if they do not exist."""