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. Returns nonce(24) || ciphertext.
Source code in features/secrets.py
| def encrypt_json(data: dict[str, Any] | list[Any]) -> bytes:
"""Encrypt JSON data. Returns nonce(24) || ciphertext."""
return _encrypt_json(data, _get_master_key())
|
decrypt_json
Decrypt blob back to JSON.
Source code in features/secrets.py
| def decrypt_json(blob: bytes) -> Any:
"""Decrypt blob back to JSON."""
return _decrypt_json(blob, _get_master_key())
|
is_encrypted_blob
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
# Path is verified to be within app data dir by caller, safe.
with path.open("rb") as f:
head = f.read(1)
return bool(_is_encrypted_blob(head))
|
_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)
with contextlib.suppress(Exception):
import keyring
stored = keyring.get_password(_KEYRING_SERVICE, _KEYRING_USERNAME)
if stored:
return bytes.fromhex(stored)
# Try .env file
_load_dotenv()
# Try config
with contextlib.suppress(Exception):
from config import config
cfg_key = config.get("crypto", "master_key_hex", default="")
if cfg_key:
return bytes.fromhex(cfg_key)
# Try environment variable with argon2id KDF
env_seed = os.environ.get(_ENV_VAR)
if env_seed:
res_kdf = argon2id.kdf(
size=_MASTER_KEY_LEN,
password=env_seed.encode("utf-8"),
salt=_KDF_SALT,
opslimit=argon2id.OPSLIMIT_MODERATE,
memlimit=argon2id.MEMLIMIT_MODERATE,
)
return bytes(res_kdf)
# 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)
res_auto = argon2id.kdf(
size=_MASTER_KEY_LEN,
password=auto_key.encode("utf-8"),
salt=_KDF_SALT,
opslimit=argon2id.OPSLIMIT_MODERATE,
memlimit=argon2id.MEMLIMIT_MODERATE,
)
return bytes(res_auto)
|
_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
|