fix(auth): fail closed on dev key and version blobs

app/config.py

@@ -52,9 +52,16 @@ class Settings(BaseSettings):
     port: int = 8000
     log_level: str = "info"
 
-    # ── Encryption (auto-generated on first run if empty) ──
+    # ── Encryption ──
+    # NOTE: credential_encryption_key is NOT auto-generated. If empty, the
+    # app seals provider keys with a publicly-known dev key (with a loud
+    # warning), and packages.db.guards refuses to boot once real
+    # credentials are at stake. Generate one: `openssl rand -hex 32`.
     credential_encryption_key: str = ""
     api_key_pepper: str = ""
+    # Explicit opt-out from the startup guard that refuses to run with the
+    # publicly-known dev encryption key when real credentials are at stake.
+    allow_insecure_dev_key: bool = False
 
     # ── Provider keys via env (alternative to UI-stored keys) ──
     # Keep in sync with `_PROVIDERS_FROM_ENV` above. Pydantic-settings reads

app/main.py

@@ -47,6 +47,19 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     async with engine.begin() as conn:
         await conn.run_sync(Base.metadata.create_all)
 
+    # Fail closed before any traffic can be served: refuse to boot when
+    # provider credentials are (or would be) sealed with the publicly-known
+    # dev encryption key. Runs after create_all so a fresh database's empty
+    # provider_keys table counts as "no credentials at risk".
+    from packages.db.guards import assert_credential_encryption_ready
+
+    await assert_credential_encryption_ready(
+        make_session=async_sessionmaker(engine, expire_on_commit=False),
+        database_url=settings.database_url,
+        allow_insecure_dev_key=settings.allow_insecure_dev_key,
+        engine=engine,
+    )
+
     session_mod._session_factory = async_sessionmaker(engine, expire_on_commit=False)
 
     from app.seed import seed_initial_state

packages/auth/encryption.py

@@ -5,21 +5,69 @@
 test fixtures that set the env var directly.
 
 If neither yields a key, derives a deterministic dev key from a fixed seed
-so local development doesn't require any setup. **In production, set a real
-64-char hex string** — the dev fallback is publicly known via the source
-code, so anyone with read access to the SQLite file could decrypt provider
-keys with it.
+so local development doesn't require any setup. The dev fallback is
+publicly known via the source code, so anyone with read access to the
+SQLite file could decrypt provider keys with it:
+
+- Every use logs a prominent WARNING (`insecure_dev_encryption_key`).
+- `packages.db.guards.assert_credential_encryption_ready` fail-closes at
+  startup when the fallback would protect real credentials (existing
+  provider rows, or any non-SQLite database) unless
+  ORCA_ALLOW_INSECURE_DEV_KEY=1 is set explicitly.
+
+Ciphertext format:
+
+- v1 (current): ``b"\\x01" + nonce(12) + ciphertext+tag``
+- legacy:       ``nonce(12) + ciphertext+tag`` (no version byte)
+
+Decrypt auto-detects. A legacy blob whose first nonce byte happens to be
+``0x01`` (~0.4% of legacy blobs) is attempted as v1 first and falls back
+to the legacy parse when authentication fails, so upgrades never brick
+stored credentials.
 """
 
 from __future__ import annotations
 
 import hashlib
+import logging
 import os
 
+from cryptography.exceptions import InvalidTag
 from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
+logger = logging.getLogger("orca.encryption")
+
+VERSION_BYTE = b"\x01"
+_NONCE_LEN = 12
+_TAG_LEN = 16
+
+_dev_fallback_warned = False
+
+
+def _warn_dev_fallback_once() -> None:
+    global _dev_fallback_warned
+    if _dev_fallback_warned:
+        return
+    _dev_fallback_warned = True
+    logger.warning(
+        "insecure_dev_encryption_key: CREDENTIAL_ENCRYPTION_KEY is not set; "
+        "provider credentials are being sealed with a PUBLICLY-KNOWN dev "
+        "key. Anyone with read access to the database can decrypt them. "
+        "Generate one with `openssl rand -hex 32` (or set "
+        "ORCA_ALLOW_INSECURE_DEV_KEY=1 to silence this check)."
+    )
+
 
 def _get_encryption_key() -> bytes:
+    key, _source = _resolve_key_material()
+    return key
+
+
+def _resolve_key_material() -> tuple[bytes, str]:
+    """Return (key_bytes, source) where source names how the key was obtained.
+
+    Sources: "config" (Settings/.env), "env" (os.environ), "dev-fallback".
+    """
     # Prefer Settings (which loads .env) over raw os.environ, because
     # pydantic-settings does NOT propagate .env values into os.environ.
     # Without this lookup, a user who follows the README and writes
@@ -33,27 +81,61 @@ def _get_encryption_key() -> bytes:
         # Settings may not be importable in some isolated test contexts;
         # fall through to env-only behavior.
         pass
-    if not key_hex:
-        key_hex = os.environ.get("CREDENTIAL_ENCRYPTION_KEY", "")
     if key_hex:
         try:
             raw = bytes.fromhex(key_hex)
             if len(raw) >= 32:
-                return raw[:32]
+                return raw[:32], "config"
         except ValueError:
             pass
-        return hashlib.sha256(key_hex.encode()).digest()
+        return hashlib.sha256(key_hex.encode()).digest(), "config"
+    key_hex = os.environ.get("CREDENTIAL_ENCRYPTION_KEY", "")
+    if key_hex:
+        try:
+            raw = bytes.fromhex(key_hex)
+            if len(raw) >= 32:
+                return raw[:32], "env"
+        except ValueError:
+            pass
+        return hashlib.sha256(key_hex.encode()).digest(), "env"
     # Dev fallback so test fixtures and `docker compose up` Just Work.
-    return hashlib.sha256(b"orcarouter-lite-dev-key").digest()
+    _warn_dev_fallback_once()
+    return hashlib.sha256(b"orcarouter-lite-dev-key").digest(), "dev-fallback"
+
+
+def is_using_insecure_dev_key() -> bool:
+    try:
+        return _resolve_key_material()[1] == "dev-fallback"
+    except Exception:
+        return False
 
 
 def encrypt_credential(plaintext: str) -> bytes:
     aes = AESGCM(_get_encryption_key())
-    nonce = os.urandom(12)
-    return nonce + aes.encrypt(nonce, plaintext.encode("utf-8"), None)
+    nonce = os.urandom(_NONCE_LEN)
+    return VERSION_BYTE + nonce + aes.encrypt(nonce, plaintext.encode("utf-8"), None)
 
 
 def decrypt_credential(blob: bytes) -> str:
-    aes = AESGCM(_get_encryption_key())
-    nonce, ciphertext = blob[:12], blob[12:]
+    key = _get_encryption_key()
+    aes = AESGCM(key)
+
+    if blob[:1] == VERSION_BYTE and len(blob) >= 1 + _NONCE_LEN + _TAG_LEN:
+        try:
+            return aes.decrypt(
+                blob[1:1 + _NONCE_LEN], blob[1 + _NONCE_LEN:], None
+            ).decode("utf-8")
+        except InvalidTag:
+            # Could be a LEGACY blob whose first nonce byte happens to be
+            # 0x01 (~0.4%). Fall through and try the unversioned layout
+            # before giving up.
+            pass
+
+    # Legacy unversioned blob: nonce(12) || ciphertext+tag.
+    # Validate length explicitly so truncated/malformed input raises
+    # InvalidTag (the contract callers test for) rather than a bare
+    # ValueError from cryptography's parameter checks.
+    if len(blob) < _NONCE_LEN + _TAG_LEN:
+        raise InvalidTag("ciphertext too short")
+    nonce, ciphertext = blob[:_NONCE_LEN], blob[_NONCE_LEN:]
     return aes.decrypt(nonce, ciphertext, None).decode("utf-8")

packages/db/guards.py

@@ -0,0 +1,137 @@
+"""Startup safety guards that need DB access.
+
+`assert_credential_encryption_ready` fail-closes boot when provider
+credentials would be (or are) protected by the publicly-known dev
+encryption key. Kept separate from `packages.auth.encryption` so the
+crypto module stays free of SQLAlchemy imports.
+"""
+
+from __future__ import annotations
+
+from sqlalchemy import func, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from packages.auth.encryption import is_using_insecure_dev_key
+
+_ALLOW_FLAG_ENV = "ORCA_ALLOW_INSECURE_DEV_KEY"
+
+
+def _allow_flag_enabled(settings_value: bool, os_environ) -> bool:
+    if settings_value:
+        return True
+    return str(os_environ.get(_ALLOW_FLAG_ENV, "")).lower() in ("1", "true", "yes")
+
+
+async def _count_provider_keys(session: AsyncSession) -> int:
+    from packages.db.models.provider_key import ProviderKey
+
+    return int(
+        (await session.execute(select(func.count()).select_from(ProviderKey))).scalar_one()
+    )
+
+
+async def assert_credential_encryption_ready(
+    *,
+    make_session,
+    database_url: str,
+    allow_insecure_dev_key: bool = False,
+    os_environ=None,
+    engine=None,
+) -> None:
+    """Refuse to start when the dev encryption key would guard real secrets.
+
+    - SQLite + zero stored provider keys  -> allowed (fresh dev install),
+      `encryption.py` warns loudly at first use.
+    - Anything else without a configured key -> RuntimeError with remediation.
+    """
+    import os as _os
+
+    environ = os_environ if os_environ is not None else _os.environ
+    if not is_using_insecure_dev_key():
+        return
+    if _allow_flag_enabled(allow_insecure_dev_key, environ):
+        return
+
+    is_sqlite = database_url.startswith("sqlite")
+
+    # Prefer an explicit engine for table-existence checks; fall back to
+    # extracting it from the session factory so we don't rely on fragile
+    # string-matching of exception messages.
+    if engine is None:
+        try:
+            engine = getattr(make_session, "kw", {}).get("bind")  # async_sessionmaker
+        except Exception:
+            engine = None
+        if engine is None:
+            engine = getattr(make_session, "bind", None)
+
+    if engine is not None:
+        # Use async-safe inspector: for AsyncEngine we must run through
+        # run_sync inside an async connection, otherwise MissingGreenlet.
+        has_table = False
+        try:
+            if hasattr(engine, "connect") and hasattr(engine, "sync_engine"):
+                # AsyncEngine path — use run_sync
+                async with engine.connect() as conn:
+                    def _check(sync_conn):
+                        from sqlalchemy import inspect as _inspect
+
+                        return _inspect(sync_conn).has_table("provider_keys")
+
+                    has_table = await conn.run_sync(_check)
+            else:
+                from sqlalchemy import inspect
+
+                sync_engine = engine.sync_engine if hasattr(engine, "sync_engine") else engine
+                has_table = inspect(sync_engine).has_table("provider_keys")
+        except RuntimeError:
+            raise
+        except Exception:
+            # Inspector itself failed — fail closed, don't silently allow boot.
+            raise
+        if not has_table:
+            key_rows = 0
+            if is_sqlite and key_rows == 0:
+                return
+            raise RuntimeError(
+                "CREDENTIAL_ENCRYPTION_KEY is not set, so provider API keys would be "
+                "sealed with a publicly-known development key. "
+                + "A non-SQLite database requires an explicit encryption key. "
+                + "Generate one with `openssl rand -hex 32`, set it as "
+                "CREDENTIAL_ENCRYPTION_KEY, and re-save your provider keys. "
+                "(If you knowingly want to keep using the insecure dev key, set "
+                "ORCA_ALLOW_INSECURE_DEV_KEY=1.)"
+            )
+        # Table exists — count rows; any failure here is not "missing table"
+        # and must fail closed.
+        async with make_session() as session:
+            key_rows = await _count_provider_keys(session)
+    else:
+        # No engine available (test helper without bind) — fall back to
+        # counting and narrowly treat only a missing-table error as zero.
+        try:
+            async with make_session() as session:
+                key_rows = await _count_provider_keys(session)
+        except Exception as exc:
+            msg = str(exc).lower()
+            if "no such table" in msg or "no such relation" in msg:
+                key_rows = 0
+            else:
+                raise
+
+    if is_sqlite and key_rows == 0:
+        return
+
+    raise RuntimeError(
+        "CREDENTIAL_ENCRYPTION_KEY is not set, so provider API keys would be "
+        "sealed with a publicly-known development key. "
+        + (
+            f"{key_rows} provider key(s) already exist in this database."
+            if key_rows
+            else "A non-SQLite database requires an explicit encryption key."
+        )
+        + " Generate one with `openssl rand -hex 32`, set it as "
+        "CREDENTIAL_ENCRYPTION_KEY, and re-save your provider keys. "
+        "(If you knowingly want to keep using the insecure dev key, set "
+        "ORCA_ALLOW_INSECURE_DEV_KEY=1.)"
+    )