Coverage for node / src / stigmem_node / auth.py: 92%
191 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-06-18 05:34 +0000
1"""API-key authentication for the Stigmem reference node.
3Auth model (spec §3.5):
4 - Callers present `Authorization: Bearer <raw-key>` on every request.
5 - Raw keys are never stored; only an Argon2id hash is persisted.
6 - Legacy SHA-256 hex digests are accepted during the v0.9.x migration
7 window and are opportunistically rehashed to Argon2id on successful use.
8 - Each key maps to an entity_uri and a JSON-array of permissions:
9 ["read"], ["read","write"], or ["read","write","federate"].
10 - `STIGMEM_AUTH_REQUIRED` defaults to **True** — every request must
11 present a valid Bearer token. Set `STIGMEM_AUTH_REQUIRED=false` to
12 opt into anonymous-mode for single-operator development; this is
13 NOT appropriate for any deployment that accepts requests from
14 agents you don't fully control. See LIMITATIONS.md §"LLM agents
15 holding admin-scope API keys" for context.
17Bootstrapping the first key (single-operator install):
18 $ KEY=$(openssl rand -hex 32)
19 $ stigmem auth bootstrap-key --key "$KEY"
20 # then use $KEY as `Authorization: Bearer $KEY` for subsequent requests.
22 The system NEVER generates the key — the caller provides the value and
23 retains full custody. We hash and store it. This is by design:
24 removing the system as the credential-generation surface eliminates
25 the "reveal channel" risk entirely. The command refuses to run when
26 api_keys is non-empty (bootstrap is one-shot); after bootstrap,
27 additional keys go through `POST /v1/auth/keys`.
28"""
30from __future__ import annotations
32import hashlib
33import hmac
34import json
35import logging
36import re
37import uuid
38from datetime import UTC, datetime, timedelta
39from typing import Annotated, Any
41from argon2 import PasswordHasher
42from argon2.exceptions import InvalidHash, VerificationError, VerifyMismatchError
43from argon2.low_level import Type
44from fastapi import Depends, HTTPException, status
45from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
47from .db import db
48from .multi_tenant_gate import warn_if_tenant_not_isolatable
49from .plugins import Deny, TenantContext, get_registry
50from .settings import settings as settings
51from .tenant import DEFAULT_TENANT_ID, TenantIdError, validate_tenant_id
53_ARGON2_HASHER = PasswordHasher(
54 time_cost=2,
55 memory_cost=19_456,
56 parallelism=1,
57 hash_len=32,
58 salt_len=16,
59 type=Type.ID,
60)
61_LEGACY_SHA256_HEX = re.compile(r"^[0-9a-f]{64}$")
63logger = logging.getLogger("stigmem.auth")
66def _hash_key(raw: str) -> str:
67 """Return the persisted hash format for newly issued API keys."""
68 return _ARGON2_HASHER.hash(raw)
71def _legacy_sha256(raw: str) -> str:
72 return hashlib.sha256(raw.encode()).hexdigest()
75def _is_legacy_sha256_hash(stored_hash: str) -> bool:
76 return bool(_LEGACY_SHA256_HEX.fullmatch(stored_hash))
79def _legacy_sha256_allowed() -> bool:
80 deadline = settings.legacy_sha256_accept_until
81 if deadline is None:
82 return True
83 if deadline.tzinfo is None: 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true
84 deadline = deadline.replace(tzinfo=UTC)
85 return datetime.now(UTC) <= deadline.astimezone(UTC)
88def _raise_legacy_sha256_disabled() -> None:
89 raise HTTPException(
90 status_code=status.HTTP_401_UNAUTHORIZED,
91 detail="Legacy API key hashes are no longer accepted; rotate the key.",
92 )
95def _verify_key_hash(raw_key: str, stored_hash: str) -> bool:
96 """Verify *raw_key* against either current Argon2id or legacy SHA-256."""
97 if stored_hash.startswith("$argon2id$"):
98 try:
99 return _ARGON2_HASHER.verify(stored_hash, raw_key)
100 except VerifyMismatchError:
101 return False
102 except VerificationError:
103 return False
104 except InvalidHash:
105 return False
106 if _is_legacy_sha256_hash(stored_hash): 106 ↛ 123line 106 didn't jump to line 123 because the condition on line 106 was always true
107 if not _legacy_sha256_allowed():
108 _raise_legacy_sha256_disabled()
109 matched = hmac.compare_digest(stored_hash, _legacy_sha256(raw_key))
110 if matched:
111 # F-SAUTH2: a legacy unsalted SHA-256 key hash was accepted. Emit a
112 # per-acceptance deprecation warning so operators see exactly which
113 # auth events still rely on the weaker legacy format and can force
114 # rotation before setting a cutoff (legacy_sha256_accept_until).
115 logger.warning(
116 "DEPRECATION: accepted a legacy unsalted SHA-256 API-key hash. "
117 "Unsalted SHA-256 is weaker than the Argon2id default; this key "
118 "is opportunistically rehashed on success — rotate any keys that "
119 "do not rehash, and set STIGMEM_LEGACY_SHA256_ACCEPT_UNTIL to bound "
120 "the migration window."
121 )
122 return matched
123 return False
126def _row_expired(row: Any, now: str) -> bool:
127 expires_at = row["expires_at"]
128 return bool(expires_at and expires_at < now)
131def _parse_api_key_expiry(expires_at: str) -> datetime:
132 normalized = expires_at.replace("Z", "+00:00")
133 parsed = datetime.fromisoformat(normalized)
134 if parsed.tzinfo is None: 134 ↛ 135line 134 didn't jump to line 135 because the condition on line 134 was never true
135 parsed = parsed.replace(tzinfo=UTC)
136 return parsed.astimezone(UTC)
139def _normalize_api_key_expiry(
140 expires_at: str | None,
141 *,
142 created_at: datetime,
143) -> str | None:
144 """Apply static API-key max-age policy and return the persisted expiry."""
145 max_age_days = settings.api_key_max_age_days
146 if max_age_days <= 0: 146 ↛ 147line 146 didn't jump to line 147 because the condition on line 146 was never true
147 return expires_at
149 max_expires_at = created_at + timedelta(days=max_age_days)
150 if expires_at is None:
151 return max_expires_at.isoformat()
153 requested_expires_at = _parse_api_key_expiry(expires_at)
154 if requested_expires_at > max_expires_at:
155 raise ValueError(
156 "expires_at exceeds configured static API key max age "
157 f"({max_age_days} days)"
158 )
159 return requested_expires_at.isoformat()
162def _rehash_legacy_key(conn: Any, row: Any, raw_key: str) -> None:
163 """Rewrite one verified legacy SHA-256 key row to Argon2id and audit it."""
164 new_hash = _hash_key(raw_key)
165 conn.execute(
166 "UPDATE api_keys SET key_hash = ? WHERE id = ? AND key_hash = ?",
167 (new_hash, row["id"], row["key_hash"]),
168 )
170 from .observability.audit_event import emit
172 emit(
173 "api_key_rehashed",
174 entity_uri=row["entity_uri"],
175 tenant_id=row["tenant_id"] or "default",
176 oidc_sub=row["oidc_sub"],
177 source="system:auth",
178 detail={
179 "key_id": row["id"],
180 "from": "sha256",
181 "to": "argon2id",
182 },
183 conn=conn,
184 )
187def _select_key_rows(conn: Any) -> list[Any]:
188 return list(
189 conn.execute(
190 "SELECT id, key_hash, entity_uri, permissions, expires_at, oidc_sub, tenant_id"
191 " FROM api_keys"
192 ).fetchall()
193 )
196def _find_key_row(
197 raw_key: str,
198 *,
199 include_expired: bool = False,
200 rehash_legacy: bool = False,
201) -> Any | None:
202 now = datetime.now(UTC).isoformat()
203 with db() as conn:
204 legacy_row = conn.execute(
205 "SELECT id, key_hash, entity_uri, permissions, expires_at, oidc_sub, tenant_id"
206 " FROM api_keys WHERE key_hash = ?",
207 (_legacy_sha256(raw_key),),
208 ).fetchone()
209 if legacy_row is not None and (include_expired or not _row_expired(legacy_row, now)):
210 if not _legacy_sha256_allowed():
211 _raise_legacy_sha256_disabled()
212 if rehash_legacy:
213 _rehash_legacy_key(conn, legacy_row, raw_key)
214 legacy_row = conn.execute(
215 "SELECT id, key_hash, entity_uri, permissions, expires_at, oidc_sub, tenant_id"
216 " FROM api_keys WHERE id = ?",
217 (legacy_row["id"],),
218 ).fetchone()
219 return legacy_row
221 for row in _select_key_rows(conn):
222 if not include_expired and _row_expired(row, now):
223 continue
224 if _verify_key_hash(raw_key, row["key_hash"]):
225 if rehash_legacy and _is_legacy_sha256_hash(row["key_hash"]): 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true
226 _rehash_legacy_key(conn, row, raw_key)
227 row = conn.execute(
228 "SELECT id, key_hash, entity_uri, permissions, expires_at, oidc_sub,"
229 " tenant_id FROM api_keys WHERE id = ?",
230 (row["id"],),
231 ).fetchone()
232 return row
233 return None
236def find_api_key_id_by_raw_key(raw_key: str) -> str | None:
237 """Return the stored key id for *raw_key*, including expired rows."""
238 row = _find_key_row(raw_key, include_expired=True, rehash_legacy=False)
239 return row["id"] if row is not None else None
242def lookup_principal(raw_key: str) -> tuple[str, str, str | None] | None:
243 """Return (entity_uri, tenant_id, oidc_sub) for a non-expired raw key."""
244 row = _find_key_row(raw_key, include_expired=False, rehash_legacy=False)
245 if row is None:
246 return None
247 return (row["entity_uri"], row["tenant_id"] or "default", row["oidc_sub"])
250BEARER = HTTPBearer(auto_error=False)
253class Identity:
254 """Resolved caller identity (spec §3.5)."""
256 def __init__(
257 self,
258 entity_uri: str,
259 permissions: list[str],
260 oidc_sub: str | None = None,
261 tenant_id: str = "default",
262 ) -> None:
263 self.entity_uri = entity_uri
264 self.permissions = set(permissions)
265 self.oidc_sub = oidc_sub
266 self.tenant_id = tenant_id
268 def can_read(self) -> bool:
269 return self._has_capability("read")
271 def can_write(self) -> bool:
272 return self._has_capability("write")
274 def can_write_instruction(self) -> bool:
275 return self._has_capability("instruction:write")
277 def can_federate(self) -> bool:
278 return self._has_capability("federate")
280 def can_admin_federation(self) -> bool:
281 return self._has_capability("admin:federation")
283 def can_audit(self) -> bool:
284 """True when the principal holds the audit.read capability (spec §22.3)."""
285 return self._has_capability("audit.read")
287 def is_admin(self) -> bool:
288 """True when the principal holds the admin capability (spec §24.3.2)."""
289 return self._has_capability("admin")
291 def _has_capability(self, capability: str) -> bool:
292 if capability not in self.permissions:
293 return False
294 decision = get_registry().fire_voting(
295 "capability_check",
296 identity=self,
297 capability=capability,
298 tenant=TenantContext(
299 tenant_id=self.tenant_id,
300 metadata={"tenant_context_source": "hook"},
301 ),
302 )
303 return not isinstance(decision, Deny)
306_ANON = Identity("anon:trusted", ["read", "write", "federate"], tenant_id="default")
309def resolve_identity(
310 creds: Annotated[HTTPAuthorizationCredentials | None, Depends(BEARER)],
311) -> Identity:
312 """Dependency: resolve caller identity from Bearer token.
314 If auth is disabled, returns a fully-trusted anonymous identity.
315 If auth is enabled, the token must match a non-expired api_keys row.
316 """
317 if not settings.auth_required:
318 if creds is not None:
319 # Still try to resolve a real identity even in non-required mode
320 identity = _lookup(creds.credentials)
321 if identity:
322 return _apply_identity_hooks(identity, raw_credentials=creds.credentials)
323 return _apply_identity_hooks(_ANON, raw_credentials=None)
325 if creds is None:
326 raise HTTPException(
327 status_code=status.HTTP_401_UNAUTHORIZED,
328 detail="Authorization header required",
329 headers={"WWW-Authenticate": "Bearer"},
330 )
331 identity = _lookup(creds.credentials)
332 if identity is None:
333 raise HTTPException(
334 status_code=status.HTTP_401_UNAUTHORIZED,
335 detail="Invalid or expired API key",
336 headers={"WWW-Authenticate": "Bearer"},
337 )
338 return _apply_identity_hooks(identity, raw_credentials=creds.credentials)
341def resolve_identity_optional(
342 creds: Annotated[HTTPAuthorizationCredentials | None, Depends(BEARER)],
343) -> Identity | None:
344 """Dependency: resolve identity from Bearer token without the auth_required gate.
346 Returns None when no credential is supplied or the credential is invalid/expired.
347 Used by routes that handle auth themselves (e.g. /metrics with metrics_require_auth).
348 """
349 if creds is None:
350 return None
351 identity = _lookup(creds.credentials)
352 if identity is None: 352 ↛ 353line 352 didn't jump to line 353 because the condition on line 352 was never true
353 return None
354 return _apply_identity_hooks(identity, raw_credentials=creds.credentials)
357def _apply_identity_hooks(identity: Identity, raw_credentials: str | None) -> Identity:
358 registry = get_registry()
359 resolved = registry.fire_filter_chain(
360 "identity_resolve",
361 identity,
362 raw_credentials=raw_credentials,
363 )
364 tenant = registry.fire_filter_chain(
365 "tenant_resolve",
366 TenantContext(
367 tenant_id=DEFAULT_TENANT_ID,
368 metadata={
369 "source_tenant_id": resolved.tenant_id,
370 "tenant_context_source": "hook",
371 },
372 ),
373 identity=resolved,
374 )
375 try:
376 resolved_tenant_id = validate_tenant_id(tenant.tenant_id)
377 except TenantIdError:
378 resolved_tenant_id = DEFAULT_TENANT_ID
379 if resolved_tenant_id != resolved.tenant_id:
380 return Identity(
381 entity_uri=resolved.entity_uri,
382 permissions=sorted(resolved.permissions),
383 oidc_sub=resolved.oidc_sub,
384 tenant_id=resolved_tenant_id,
385 )
386 return resolved
389def _lookup(raw_key: str) -> Identity | None:
390 row = _find_key_row(raw_key, include_expired=False, rehash_legacy=True)
391 if row is None:
392 return None
393 perms: list[str] = json.loads(row["permissions"])
394 return Identity(
395 entity_uri=row["entity_uri"],
396 permissions=perms,
397 oidc_sub=row["oidc_sub"],
398 tenant_id=row["tenant_id"] or "default",
399 )
402def register_api_key(
403 raw_key: str,
404 entity_uri: str,
405 permissions: list[str] | None = None,
406 description: str | None = None,
407 expires_at: str | None = None,
408 oidc_sub: str | None = None,
409 tenant_id: str = "default",
410) -> str:
411 """Persist the hash of a caller-provided raw API key. Returns the key_id.
413 The caller is responsible for generating `raw_key` (e.g., from
414 `secrets.token_hex(32)` or `openssl rand -hex 32`) and for storing it
415 securely. This function never touches the key after hashing.
416 """
417 if permissions is None: 417 ↛ 418line 417 didn't jump to line 418 because the condition on line 417 was never true
418 permissions = ["read", "write"]
419 if find_api_key_id_by_raw_key(raw_key) is not None:
420 raise ValueError("raw API key already exists")
421 normalized_tenant_id = validate_tenant_id(tenant_id)
422 # F-ID-1: a non-default tenant collapses to the default tenant without the
423 # multi-tenant plugin — warn loudly rather than silently mislead the operator.
424 warn_if_tenant_not_isolatable(normalized_tenant_id)
425 key_id = str(uuid.uuid4())
426 created_at = datetime.now(UTC).replace(microsecond=0)
427 normalized_expires_at = _normalize_api_key_expiry(
428 expires_at,
429 created_at=created_at,
430 )
431 with db() as conn:
432 conn.execute(
433 """INSERT INTO api_keys
434 (id, key_hash, entity_uri, permissions, description,
435 created_at, expires_at, oidc_sub, tenant_id)
436 VALUES (?,?,?,?,?,?,?,?,?)""",
437 (
438 key_id,
439 _hash_key(raw_key),
440 entity_uri,
441 json.dumps(permissions),
442 description,
443 created_at.isoformat(),
444 normalized_expires_at,
445 oidc_sub,
446 normalized_tenant_id,
447 ),
448 )
449 return key_id
452def create_api_key(
453 entity_uri: str,
454 permissions: list[str] | None = None,
455 description: str | None = None,
456 expires_at: str | None = None,
457 oidc_sub: str | None = None,
458 tenant_id: str = "default",
459) -> str:
460 """Mint a new raw API key, persist its hash, and return the raw key.
462 Internal/test-only convenience over `register_api_key`. Adopter-facing
463 flows should require the caller to provide the key material (see
464 `stigmem auth bootstrap-key`) so the system is never the credential
465 generation surface.
466 """
467 raw = str(uuid.uuid4()).replace("-", "") # 32-char hex
468 register_api_key(
469 raw_key=raw,
470 entity_uri=entity_uri,
471 permissions=permissions,
472 description=description,
473 expires_at=expires_at,
474 oidc_sub=oidc_sub,
475 tenant_id=tenant_id,
476 )
477 return raw