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 | 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 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
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 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 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
|