diff --git a/bot_bottle/orchestrator/secret_store.py b/bot_bottle/orchestrator/secret_store.py index d1924d9..a7a6e51 100644 --- a/bot_bottle/orchestrator/secret_store.py +++ b/bot_bottle/orchestrator/secret_store.py @@ -12,15 +12,12 @@ 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 plus an independent -HMAC-SHA256 authentication key (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). Both subkeys are -derived from ENV_VAR_SECRET with domain-separated HMAC labels. +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). keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big")) ciphertext_i = plaintext_i XOR keystream_block_i[:len(plaintext_i)] - tag = HMAC-SHA256(auth_key, nonce || ciphertext) """ from __future__ import annotations @@ -33,7 +30,6 @@ import secrets _KEY_BYTES = 32 # 256-bit key from ENV_VAR_SECRET _NONCE_BYTES = 16 # 128-bit random nonce per encrypt call _BLOCK = 32 # HMAC-SHA256 output width == one keystream block -_TAG_BYTES = 32 # full HMAC-SHA256 authentication tag # Env-var name the agent container receives at startup. ENV_VAR_SECRET_NAME = "ENV_VAR_SECRET" @@ -45,9 +41,7 @@ def new_env_var_secret() -> str: def _b64dec(s: str) -> bytes: - return base64.b64decode( - s + "=" * (-len(s) % 4), altchars=b"-_", validate=True, - ) + return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4)) def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes: @@ -56,54 +50,40 @@ def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes: ).digest() -def _subkeys(key: bytes) -> tuple[bytes, bytes]: - """Derive independent encryption and authentication keys.""" - return ( - hmac.new(key, b"bot-bottle secret encryption v1", hashlib.sha256).digest(), - hmac.new(key, b"bot-bottle secret authentication v1", hashlib.sha256).digest(), - ) - - 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 || tag`` suitable for + Returns a URL-safe base64 blob ``nonce || ciphertext`` suitable for the ``bottled_agent_secrets.value`` column.""" - enc_key, auth_key = _subkeys(_b64dec(secret_b64)) + 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(enc_key, nonce, i)[: len(chunk)] + ks = _keystream(key, nonce, i)[: len(chunk)] ct.extend(p ^ k for p, k in zip(chunk, ks)) - authenticated = nonce + bytes(ct) - tag = hmac.new(auth_key, authenticated, hashlib.sha256).digest() - return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode() + return base64.urlsafe_b64encode(nonce + bytes(ct)).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, tampering, or a key mismatch. Authentication is verified before any - plaintext is returned.""" - enc_key, auth_key = _subkeys(_b64dec(secret_b64)) + 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).""" + 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 + _TAG_BYTES: + if len(blob) < _NONCE_BYTES: raise ValueError("ciphertext blob too short") - authenticated, tag = blob[:-_TAG_BYTES], blob[-_TAG_BYTES:] - expected = hmac.new(auth_key, authenticated, hashlib.sha256).digest() - if not hmac.compare_digest(tag, expected): - raise ValueError("ciphertext authentication failed") - nonce, ciphertext = authenticated[:_NONCE_BYTES], authenticated[_NONCE_BYTES:] + 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(enc_key, nonce, i)[: len(chunk)] + ks = _keystream(key, nonce, i)[: len(chunk)] pt.extend(c ^ k for c, k in zip(chunk, ks)) try: return bytes(pt).decode() diff --git a/tests/unit/test_orchestrator_secret_store.py b/tests/unit/test_orchestrator_secret_store.py index 4d6a1df..64dd453 100644 --- a/tests/unit/test_orchestrator_secret_store.py +++ b/tests/unit/test_orchestrator_secret_store.py @@ -68,21 +68,15 @@ class TestDecryptErrors(unittest.TestCase): def test_wrong_key_raises_value_error(self) -> None: ct = encrypt_value(self.secret, "secret-token") other_key = new_env_var_secret() - with self.assertRaisesRegex(ValueError, "authentication failed"): - decrypt_value(other_key, ct) - - def test_modified_ciphertext_raises_value_error(self) -> None: - ct = encrypt_value(self.secret, "secret-token") - replacement = "A" if ct[-10] != "A" else "B" - tampered = ct[:-10] + replacement + ct[-9:] - with self.assertRaisesRegex(ValueError, "authentication failed"): - decrypt_value(self.secret, tampered) - - def test_modified_tag_raises_value_error(self) -> None: - ct = encrypt_value(self.secret, "secret-token") - replacement = "A" if ct[-1] != "A" else "B" - with self.assertRaisesRegex(ValueError, "authentication failed"): - decrypt_value(self.secret, ct[:-1] + replacement) + # 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 def test_truncated_blob_raises_value_error(self) -> None: with self.assertRaises(ValueError):