fix(api-keys)!: use salted, peppered scrypt digests

CHANGELOG.md

@@ -4,6 +4,14 @@ This repository starts with a fresh history. Releases are listed here once tagge
 
 ## [Unreleased]
 
+- API keys (breaking): keys are stored as salted, peppered scrypt hashes
+  (`API_KEY_HASH_PEPPER`) and found by a non-secret indexed lookup id
+  (migration 054). A bounded 60 s cache of recent verifications keeps repeat
+  requests cheap. The plain SHA-256 lookup is removed, so all existing keys
+  stop working at deploy and must be reissued. Agent-comms owner digests use
+  peppered scrypt too. Production-like environments refuse to boot without
+  the pepper; local and test use a public dev pepper. See
+  docs/api-key-hashing.md.
 - Upstream MCP (Preview): optional multi-tool discovery and exact-name allowlist with
   stable public IDs, per-tool permit scope, and unchanged per-call credits.
   Health reports `upstream_tool_mode: multi_tool_preview` while it is on.

gateway/.env.example

@@ -158,6 +158,14 @@ JEV_RISK_GUARD_TIERS=medium,high
 TYPESAFE_API_KEY=
 TYPESAFE_BASE_URL=https://api.typesafe.ai
 
+# Server pepper for HMAC-SHA256 API-key digests. REQUIRED in production-like
+# environments: startup refuses to boot and keys are rejected without it.
+# Generate a long random value (e.g. 32+ random bytes) in your secret store;
+# never commit it. Changing it invalidates every key. Local/test fall back to
+# a public non-secret dev pepper when unset.
+# See docs/api-key-hashing.md.
+API_KEY_HASH_PEPPER=
+
 # --- IoT Protocol Bridge ---
 MQTT_BROKER_URL=mqtt://localhost:1883
 MQTT_DEFAULT_QOS=1

gateway/app/core/config.py

@@ -252,6 +252,10 @@ def demo_allowed_tools_list(self) -> list[str]:
     JEV_RISK_GUARD_TIMEOUT_SECONDS: float = 1.5
     JEV_RISK_GUARD_TIERS: str = "medium,high"
     TYPESAFE_API_KEY: SecretStr = SecretStr("")
+    # Server pepper for HMAC-SHA256 API-key digests. Required in
+    # production-like environments (startup refuses to boot without it).
+    # See docs/api-key-hashing.md. Never log it.
+    API_KEY_HASH_PEPPER: SecretStr = SecretStr("")
     TYPESAFE_BASE_URL: str = "https://api.typesafe.ai"
     # Default repeat detection window in seconds. Duplicates outside this window
     # are not detected.

gateway/app/db/models.py

@@ -273,7 +273,10 @@ class APIKeyModel(SQLModel, table=True):
     key_id: str = Field(primary_key=True, max_length=50)
     wallet_id: str = Field(max_length=50, foreign_key="wallets.wallet_id", index=True)
 
-    key_hash: str = Field(max_length=64, index=True)
+    # Salted, peppered scrypt hash (see docs/api-key-hashing.md).
+    key_hash: str = Field(max_length=255, index=True)
+    # Non-secret lookup id (first 16 chars of the key), indexed for lookup.
+    key_lookup: str | None = Field(default=None, max_length=16, index=True)
     key_prefix: str = Field(max_length=8)
 
     status: str = Field(default="active", max_length=20)

gateway/app/main.py

@@ -25,6 +25,7 @@
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import HTMLResponse, JSONResponse
 
+from app.services.api_key_service import validate_api_key_hashing_config
 from .core.auth import AuthContext, debug_open_auth_armed, get_auth_context
 from .core.api_key_usage import APIKeyUsageMiddleware
 from .core.build_metadata import get_build_commit_sha, get_build_provenance
@@ -176,6 +177,7 @@
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     validate_trust_mode_guardrails(settings)
+    validate_api_key_hashing_config()
     warn_if_trust_mode_permissive(settings)
     # Operator-facing posture record. The unauthenticated /health/dependencies
     # payload no longer publishes per-service simulation modes when proof

gateway/app/routers/comms.py

@@ -9,6 +9,8 @@
 This is the nervous system of swarm intelligence.
 """
 
+import asyncio
+
 from fastapi import APIRouter, Depends, HTTPException, Query, status
 from pydantic import BaseModel, Field
 from datetime import datetime
@@ -20,6 +22,7 @@
     MessageType,
     MessagePriority,
     caller_owns_agent,
+    owner_candidates,
     owner_identity,
 )
 
@@ -174,7 +177,7 @@ async def require_agent_owner(
                 "message": f"Agent '{agent_id}' is not registered.",
             },
         )
-    if caller_owns_agent(agent, auth):
+    if await asyncio.to_thread(caller_owns_agent, agent, auth):
         return
     if not agent.owner_key and auth.is_bootstrap_admin:
         return
@@ -218,7 +221,7 @@ async def register_agent(
         name=request.name,
         capabilities=request.capabilities,
         webhook_url=request.webhook_url,
-        owner_key=owner_identity(auth),
+        owner_key=await asyncio.to_thread(owner_identity, auth),
     )
     return AgentRegistrationResponse(
         agent_id=agent.agent_id,
@@ -254,7 +257,8 @@ async def list_agents(
     if not auth.is_bootstrap_admin:
         # Tenants never enumerate each other's agents. Capability handoff
         # routing reads the registry server-side and is unaffected.
-        agents = [a for a in agents if caller_owns_agent(a, auth)]
+        candidates = await asyncio.to_thread(owner_candidates, auth)
+        agents = [a for a in agents if caller_owns_agent(a, auth, candidates)]
 
     return {
         "agents": [

gateway/app/services/agent_comms.py

@@ -27,6 +27,8 @@
 from ..core.runtime_mode import is_simulation, require_simulation
 from .agent_comms_store import CommsMessageStore, compute_payload_hash
 
+from .api_key_service import APIKeyHashingUnavailable, _api_key_hash_pepper
+
 logger = logging.getLogger(__name__)
 
 
@@ -121,11 +123,34 @@ class RegisteredAgent:
 # durable registry holds no secrets.
 
 _OWNER_WALLET_PREFIX = "wallet:"
-_DIGEST_PREFIX = "sha256:"
+_DIGEST_PREFIX = "scrypt:"
+# Records written before the scrypt switch. They are treated as already
+# scrubbed (never re-hashed) but no longer match any caller.
+_RETIRED_DIGEST_PREFIXES = ("sha256:",)
+_OWNER_DIGEST_DOMAIN = b"amw-agent-comms-owner-v1:"
 
 
 def _digest(secret: str) -> str:
-    return _DIGEST_PREFIX + hashlib.sha256(secret.encode()).hexdigest()
+    """Deterministic peppered scrypt digest of a credential.
+
+    Ownership is an equality match, so this digest uses a fixed salt (domain
+    tag plus the API-key pepper) instead of a per-record random one; the
+    pepper keeps a registry dump from being attacked offline. Production-like
+    environments without the pepper fail closed.
+    """
+    pepper = _api_key_hash_pepper()
+    if pepper is None:
+        raise APIKeyHashingUnavailable("API_KEY_HASH_PEPPER is not set")
+    derived = hashlib.scrypt(
+        secret.encode("utf-8"),
+        salt=_OWNER_DIGEST_DOMAIN + pepper,
+        n=2**14,
+        r=8,
+        p=1,
+        maxmem=64 * 1024 * 1024,
+        dklen=32,
+    )
+    return _DIGEST_PREFIX + derived.hex()
 
 
 def owner_identity(auth: AuthContext) -> str:
@@ -135,7 +160,21 @@ def owner_identity(auth: AuthContext) -> str:
     return _digest(auth.raw_key)
 
 
-def caller_owns_agent(agent: RegisteredAgent, auth: AuthContext) -> bool:
+def owner_candidates(auth: AuthContext) -> list[str]:
+    """Identities ``auth`` may own records under; compute once per request."""
+    candidates = []
+    if auth.wallet_id:
+        candidates.append(_OWNER_WALLET_PREFIX + auth.wallet_id)
+    if not auth.wallet_id or auth.source != "jwt":
+        candidates.append(_digest(auth.raw_key))
+    return candidates
+
+
+def caller_owns_agent(
+    agent: RegisteredAgent,
+    auth: AuthContext,
+    candidates: list[str] | None = None,
+) -> bool:
     """True only when ``auth`` is the tenant that registered ``agent``.
 
     An owner-less record matches nobody. A record registered before ownership
@@ -146,9 +185,8 @@ def caller_owns_agent(agent: RegisteredAgent, auth: AuthContext) -> bool:
     """
     if not agent.owner_key:
         return False
-    candidates = [owner_identity(auth)]
-    if auth.source != "jwt":
-        candidates.append(_digest(auth.raw_key))
+    if candidates is None:
+        candidates = owner_candidates(auth)
     stored = agent.owner_key.encode()
     return any(hmac.compare_digest(stored, c.encode()) for c in candidates)
 
@@ -194,11 +232,13 @@ def _scrub_plaintext_credentials(agent: RegisteredAgent) -> bool:
         """
         changed = False
         if agent.owner_key and not agent.owner_key.startswith(
-            (_OWNER_WALLET_PREFIX, _DIGEST_PREFIX)
+            (_OWNER_WALLET_PREFIX, _DIGEST_PREFIX, *_RETIRED_DIGEST_PREFIXES)
         ):
             agent.owner_key = _digest(agent.owner_key)
             changed = True
-        if agent.api_key and not agent.api_key.startswith(_DIGEST_PREFIX):
+        if agent.api_key and not agent.api_key.startswith(
+            (_DIGEST_PREFIX, *_RETIRED_DIGEST_PREFIXES)
+        ):
             agent.api_key = _digest(agent.api_key)
             changed = True
         return changed
@@ -223,7 +263,12 @@ async def _hydrate_if_needed(self):
                             "Skipping corrupt comms agent record: %s", agent_id
                         )
                         continue
-                    scrubbed = self._scrub_plaintext_credentials(agent) or scrubbed
+                    scrubbed = (
+                        await asyncio.to_thread(
+                            self._scrub_plaintext_credentials, agent
+                        )
+                        or scrubbed
+                    )
                     loaded[agent_id] = agent
                 self._agents = loaded
 
@@ -270,7 +315,7 @@ async def register(
             api_key=agent_key,
             owner_key=owner_key,
         )
-        self._scrub_plaintext_credentials(agent)
+        await asyncio.to_thread(self._scrub_plaintext_credentials, agent)
         async with self._lock:
             self._agents[agent.agent_id] = agent
             await self._persist_locked()