diff --git a/bot_bottle/orchestrator/store/secret_store.py b/bot_bottle/orchestrator/store/secret_store.py index 7daf0767..a6271c24 100644 --- a/bot_bottle/orchestrator/store/secret_store.py +++ b/bot_bottle/orchestrator/store/secret_store.py @@ -12,12 +12,22 @@ reattachment path reads ENV_VAR_SECRET from the running agent container via ``POST /bottles//reprovision_gateway``; the orchestrator decrypts the stored rows and re-populates ``_tokens``. -Encryption scheme: HMAC-SHA256 used as a PRF in CTR mode (stdlib-only, -no external deps). Each value is encrypted independently. The output blob is -``nonce (16 bytes) || ciphertext`` encoded as URL-safe base64 (no padding). +Encryption scheme: HMAC-SHA256 used as a PRF in CTR mode, **authenticated** +encrypt-then-MAC (stdlib-only, no external deps). Each value is encrypted +independently. The output blob is ``nonce (16 bytes) || ciphertext || tag +(32 bytes)`` encoded as URL-safe base64 (no padding). keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big")) ciphertext_i = plaintext_i XOR keystream_block_i[:len(plaintext_i)] + mac_key = HMAC-SHA256(key, "bottled-secret-mac-v1") + tag = HMAC-SHA256(mac_key, nonce || ciphertext) + +The tag is what makes a **wrong key deterministically detectable**: without it, +CTR decryption with the wrong key yields garbage that only fails when it isn't +valid UTF-8 (so ``reprovision`` would sometimes "succeed" with a wrong +ENV_VAR_SECRET and inject garbage egress tokens). The MAC key is derived from +the ENV_VAR_SECRET by a domain-separated HMAC so the same key never both +generates the keystream and signs the tag with the same message shape. """ from __future__ import annotations @@ -29,6 +39,7 @@ import secrets _KEY_BYTES = 32 # 256-bit key from ENV_VAR_SECRET _NONCE_BYTES = 16 # 128-bit random nonce per encrypt call +_TAG_BYTES = 32 # HMAC-SHA256 authentication tag _BLOCK = 32 # HMAC-SHA256 output width == one keystream block # Env-var name the agent container receives at startup. @@ -50,45 +61,58 @@ def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes: ).digest() +def _tag(key: bytes, nonce: bytes, ciphertext: bytes) -> bytes: + """The authentication tag over ``nonce || ciphertext``, keyed by a MAC + subkey domain-separated from the keystream key.""" + mac_key = hmac.new(key, b"bottled-secret-mac-v1", hashlib.sha256).digest() + return hmac.new(mac_key, nonce + ciphertext, hashlib.sha256).digest() + + +def _ctr(key: bytes, nonce: bytes, data: bytes) -> bytes: + """CTR keystream XOR — its own inverse, so it both encrypts and decrypts.""" + out = bytearray() + for i in range(0, len(data), _BLOCK): + chunk = data[i : i + _BLOCK] + ks = _keystream(key, nonce, i)[: len(chunk)] + out.extend(b ^ k for b, k in zip(chunk, ks)) + return bytes(out) + + def encrypt_value(secret_b64: str, plaintext: str) -> str: """Encrypt a single string value with *secret_b64* (the ENV_VAR_SECRET). - Returns a URL-safe base64 blob ``nonce || ciphertext`` suitable for + Returns a URL-safe base64 blob ``nonce || ciphertext || tag`` suitable for the ``bottled_agent_secrets.value`` column.""" key = _b64dec(secret_b64) - pt = plaintext.encode() nonce = secrets.token_bytes(_NONCE_BYTES) - ct = bytearray() - for i in range(0, len(pt), _BLOCK): - chunk = pt[i : i + _BLOCK] - ks = _keystream(key, nonce, i)[: len(chunk)] - ct.extend(p ^ k for p, k in zip(chunk, ks)) - return base64.urlsafe_b64encode(nonce + bytes(ct)).rstrip(b"=").decode() + ct = _ctr(key, nonce, plaintext.encode()) + tag = _tag(key, nonce, ct) + return base64.urlsafe_b64encode(nonce + ct + tag).rstrip(b"=").decode() def decrypt_value(secret_b64: str, blob_b64: str) -> str: """Decrypt a blob produced by :func:`encrypt_value`. Returns the original plaintext string. Raises ``ValueError`` for malformed - input or a key mismatch (wrong key produces garbage, not an error, unless - the plaintext is non-UTF-8 — treat all such failures as wrong key).""" + input, a **wrong key**, or a tampered ciphertext — all caught by the + authentication tag before any plaintext is returned, so a wrong + ENV_VAR_SECRET is rejected deterministically (never a garbage token).""" key = _b64dec(secret_b64) try: blob = _b64dec(blob_b64) except Exception as exc: raise ValueError(f"invalid ciphertext blob: {exc}") from exc - if len(blob) < _NONCE_BYTES: + if len(blob) < _NONCE_BYTES + _TAG_BYTES: raise ValueError("ciphertext blob too short") - nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:] - pt = bytearray() - for i in range(0, len(ciphertext), _BLOCK): - chunk = ciphertext[i : i + _BLOCK] - ks = _keystream(key, nonce, i)[: len(chunk)] - pt.extend(c ^ k for c, k in zip(chunk, ks)) + nonce = blob[:_NONCE_BYTES] + tag = blob[-_TAG_BYTES:] + ciphertext = blob[_NONCE_BYTES:-_TAG_BYTES] + if not hmac.compare_digest(tag, _tag(key, nonce, ciphertext)): + raise ValueError("ciphertext failed authentication (wrong key or tampered)") try: - return bytes(pt).decode() - except UnicodeDecodeError as exc: - raise ValueError(f"decryption produced non-UTF-8 output (wrong key?): {exc}") from exc + return _ctr(key, nonce, ciphertext).decode() + except UnicodeDecodeError as exc: # pragma: no cover - authenticated, so unreachable + raise ValueError(f"decryption produced non-UTF-8 output: {exc}") from exc __all__ = ["ENV_VAR_SECRET_NAME", "new_env_var_secret", "encrypt_value", "decrypt_value"] diff --git a/tests/unit/test_orchestrator_secret_store.py b/tests/unit/test_orchestrator_secret_store.py index 7301b126..22acd079 100644 --- a/tests/unit/test_orchestrator_secret_store.py +++ b/tests/unit/test_orchestrator_secret_store.py @@ -2,10 +2,12 @@ from __future__ import annotations +import base64 import unittest from bot_bottle.orchestrator.store.secret_store import ( ENV_VAR_SECRET_NAME, + _NONCE_BYTES, decrypt_value, encrypt_value, new_env_var_secret, @@ -65,22 +67,27 @@ class TestDecryptErrors(unittest.TestCase): def setUp(self) -> None: self.secret = new_env_var_secret() - def test_wrong_key_raises_value_error(self) -> None: + def test_wrong_key_always_raises_value_error(self) -> None: + # Deterministic: the authentication tag rejects a wrong key every time, + # so reprovision can never inject a garbage token. Repeat across many + # random keys (the old unauthenticated scheme let ~5% through when the + # garbage happened to decode as valid UTF-8). + for _ in range(200): + ct = encrypt_value(self.secret, "secret-token") + with self.assertRaises(ValueError): + decrypt_value(new_env_var_secret(), ct) + + def test_tampered_ciphertext_raises_value_error(self) -> None: ct = encrypt_value(self.secret, "secret-token") - other_key = new_env_var_secret() - # Wrong key produces garbage bytes; decrypt_value raises ValueError - # when the result is non-UTF-8 (which is very likely for 12-char data). - # We allow it to succeed only if garbage happens to be valid UTF-8, but - # the plaintext must not match. - try: - result = decrypt_value(other_key, ct) - self.assertNotEqual("secret-token", result) - except ValueError: - pass + raw = bytearray(base64.urlsafe_b64decode(ct + "=" * (-len(ct) % 4))) + raw[_NONCE_BYTES] ^= 0x01 # flip a bit in the ciphertext body → tag mismatch + tampered = base64.urlsafe_b64encode(bytes(raw)).rstrip(b"=").decode() + with self.assertRaises(ValueError): + decrypt_value(self.secret, tampered) def test_truncated_blob_raises_value_error(self) -> None: with self.assertRaises(ValueError): - decrypt_value(self.secret, "dG9vc2hvcnQ") # "tooshort" — under 16 nonce bytes + decrypt_value(self.secret, "dG9vc2hvcnQ") # "tooshort" — under nonce+tag def test_invalid_base64_raises_value_error(self) -> None: with self.assertRaises(ValueError):