admin / Synapse-Cortex

"""Credential vault: a Fernet symmetric key generated once from Admin ->
Vault and wrapped (encrypted) with the deployment's SECRET_KEY before being
stored on the Tenant row, so a database leak alone never exposes it - an
attacker would also need SECRET_KEY. There is deliberately no rotate or
recreate path anywhere in this module or the API surface above it: rotating
would make every already-vaulted credential (and the stored Claude API key)
permanently undecryptable, so create_vault() raises if a vault already
exists rather than overwriting it.

encrypt_secret/decrypt_secret both take the tenant row directly (not a bare
key) so they can raise a clear, specific VaultError when the vault hasn't
been created yet, rather than a confusing Fernet decode error. Decryption
happens exclusively inside the remediation execution path, in-memory,
immediately before dispatching a command - no API response ever returns a
decrypted secret, the encrypted blob, or the wrapped vault key itself.
"""
import base64
import hashlib
import os
from datetime import datetime, timezone

from cryptography.fernet import Fernet, InvalidToken


class VaultError(RuntimeError):
    pass


class VaultAlreadyExistsError(VaultError):
    pass


def _wrapping_fernet() -> Fernet:
    """Derives a stable Fernet key from SECRET_KEY to wrap the real vault
    key at rest. Using SECRET_KEY (already a required, deployment-wide
    secret used for session signing) means protecting the vault key needs
    no new environment variable of its own."""
    secret_key = os.getenv("SECRET_KEY", "change-me-in-production")
    digest = hashlib.sha256(secret_key.encode()).digest()
    return Fernet(base64.urlsafe_b64encode(digest))


def create_vault(tenant) -> str:
    """Generates a fresh Fernet key, wraps it with the SECRET_KEY-derived
    wrapping key, and stores it on the tenant row. Returns the *raw*
    (unwrapped) key so the caller can show it to the admin exactly once -
    it is never recoverable from the API again after this call returns.
    Raises VaultAlreadyExistsError if this tenant already has a vault;
    there is no update path, by design. Caller is responsible for
    committing the session."""
    if tenant.vault_key_wrapped:
        raise VaultAlreadyExistsError("A vault already exists for this tenant and cannot be recreated.")

    raw_key = Fernet.generate_key()
    tenant.vault_key_wrapped = _wrapping_fernet().encrypt(raw_key).decode()
    tenant.vault_created_at = datetime.now(timezone.utc)
    return raw_key.decode()


def _tenant_fernet(tenant) -> Fernet:
    if not tenant.vault_key_wrapped:
        raise VaultError(
            "No vault has been created for this tenant yet. Create one from Admin → Vault before "
            "vaulting credentials or saving a Claude API key."
        )
    try:
        raw_key = _wrapping_fernet().decrypt(tenant.vault_key_wrapped.encode())
    except InvalidToken as exc:
        raise VaultError(
            "The stored vault key could not be unwrapped - SECRET_KEY may have changed since the "
            "vault was created."
        ) from exc
    return Fernet(raw_key)


def encrypt_secret(tenant, plaintext: str) -> str:
    return _tenant_fernet(tenant).encrypt(plaintext.encode()).decode()


def decrypt_secret(tenant, ciphertext: str) -> str:
    try:
        return _tenant_fernet(tenant).decrypt(ciphertext.encode()).decode()
    except InvalidToken as exc:
        raise VaultError("Stored value could not be decrypted.") from exc