Skip to content

Secrets API

features.secrets

Envelope-encrypt JSON secrets (api_keys, bearer tokens, saga state).

Master key resolution order: 1. OS keychain (keyring library) — recommended for production 2. .env file in project root (MCP_MASTER_KEY=...) 3. crypto.master_key_hex in config.yaml 4. MCP_MASTER_KEY environment variable (argon2id KDF) 5. Fail loud if none available

encrypt_json

encrypt_json(data)

Encrypt JSON data. Returns nonce(24) || ciphertext.

Source code in features/secrets.py
def encrypt_json(data: dict | list) -> bytes:
    """Encrypt JSON data. Returns nonce(24) || ciphertext."""
    box = SecretBox(_get_master_key())
    nonce = nacl_random(SecretBox.NONCE_SIZE)
    plaintext = json.dumps(data, ensure_ascii=False, sort_keys=True).encode()
    return nonce + box.encrypt(plaintext, nonce).ciphertext

decrypt_json

decrypt_json(blob)

Decrypt blob back to JSON.

Source code in features/secrets.py
def decrypt_json(blob: bytes) -> Any:
    """Decrypt blob back to JSON."""
    if len(blob) < _NONCE_SIZE + _MAC_SIZE:
        raise ValueError("blob too short for valid SecretBox message")
    nonce, ct = blob[:_NONCE_SIZE], blob[_NONCE_SIZE:]
    box = SecretBox(_get_master_key())
    return json.loads(box.decrypt(ct, nonce).decode("utf-8"))

is_encrypted_blob

is_encrypted_blob(path)

Check if file is encrypted (not plain JSON).

Heuristic: encrypted blobs start with random 24 bytes (nonce), JSON starts with { or [.

Source code in features/secrets.py
def is_encrypted_blob(path: Path) -> bool:
    """Check if file is encrypted (not plain JSON).

    Heuristic: encrypted blobs start with random 24 bytes (nonce),
    JSON starts with { or [.
    """
    if not path.exists():
        return False
    with open(path, "rb") as f:
        head = f.read(1)
    return head not in (b"{", b"[", b" ", b"\n")

_load_master_key

_load_master_key()

Load or derive master key from keyring, .env, config, or environment.

If no key is found, auto-generates one and saves to .env for dev convenience.

Source code in features/secrets.py
def _load_master_key() -> bytes:
    """Load or derive master key from keyring, .env, config, or environment.

    If no key is found, auto-generates one and saves to .env for dev convenience.
    """
    if not _HAS_NACL:
        raise ImportError("pynacl is required for encryption. Install with: pip install pynacl")

    # Try OS keychain first (recommended for production)
    try:
        import keyring

        stored = keyring.get_password(_KEYRING_SERVICE, _KEYRING_USERNAME)
        if stored:
            return bytes.fromhex(stored)
    except Exception:
        pass

    # Try .env file
    _load_dotenv()

    # Try config
    try:
        from config import config

        cfg_key = config.get("crypto", "master_key_hex", default="")
        if cfg_key:
            return bytes.fromhex(cfg_key)
    except Exception:
        pass

    # Try environment variable with argon2id KDF
    env_seed = os.environ.get(_ENV_VAR)
    if env_seed:
        return argon2id.kdf(
            size=_MASTER_KEY_LEN,
            password=env_seed.encode("utf-8"),
            salt=_KDF_SALT,
            opslimit=argon2id.OPSLIMIT_MODERATE,
            memlimit=argon2id.MEMLIMIT_MODERATE,
        )

    # Auto-generate key for dev convenience
    import secrets as _secrets

    auto_key = _secrets.token_hex(32)
    logger.warning("No master key found. Auto-generating key and saving to .env. For production, use keyring or set MCP_MASTER_KEY explicitly.")
    _save_dotenv(_ENV_VAR, auto_key)
    return argon2id.kdf(
        size=_MASTER_KEY_LEN,
        password=auto_key.encode("utf-8"),
        salt=_KDF_SALT,
        opslimit=argon2id.OPSLIMIT_MODERATE,
        memlimit=argon2id.MEMLIMIT_MODERATE,
    )

_get_master_key

_get_master_key()

Get cached master key.

Source code in features/secrets.py
def _get_master_key() -> bytes:
    """Get cached master key."""
    key = _master_cache.get("k")
    if key is None:
        key = _load_master_key()
        _master_cache["k"] = key
    return key