didericis-claude
b5753b5921
The per-bottle egress-secret encryption (secret_store.py) was unauthenticated CTR/XOR: decrypting with the WRONG ENV_VAR_SECRET produced garbage that decrypt_value only rejected when it wasn't valid UTF-8. For short token values that garbage is coincidentally valid UTF-8 ~5% of the time, so `reprovision_from_secret` would occasionally "succeed" with a wrong key and inject a garbage egress credential — and test_reprovision_rejects_missing_rows_and_wrong_key failed ~5% of runs (flaky CI, surfaced by this stack's unit job). Switch to authenticated encrypt-then-MAC: append an HMAC-SHA256 tag over `nonce || ciphertext`, keyed by a domain-separated MAC subkey derived from the ENV_VAR_SECRET. decrypt_value verifies the tag (constant-time) before returning any plaintext, so a wrong key or tampered ciphertext is rejected deterministically. Blob format is now `nonce || ciphertext || tag` (the stored rows are transient — re-written every launch — so no migration is needed). Pre-existing bug on main, unrelated to the transport work, but it blocks this stack's CI. Tests: wrong key rejected 200/200; tampered ciphertext rejected; round-trips unchanged. Deterministic now (was ~5% flaky). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
104 lines
3.8 KiB
Python
104 lines
3.8 KiB
Python
"""Unit tests for per-bottle egress secret encryption (PRD 0080)."""
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
class TestNewEnvVarSecret(unittest.TestCase):
|
|
def test_returns_non_empty_string(self) -> None:
|
|
s = new_env_var_secret()
|
|
self.assertIsInstance(s, str)
|
|
self.assertTrue(len(s) > 0)
|
|
|
|
def test_secrets_are_unique(self) -> None:
|
|
keys = {new_env_var_secret() for _ in range(50)}
|
|
self.assertEqual(50, len(keys))
|
|
|
|
def test_no_padding_characters(self) -> None:
|
|
# URL-safe base64, padding stripped — should round-trip cleanly
|
|
for _ in range(20):
|
|
self.assertNotIn("=", new_env_var_secret())
|
|
|
|
|
|
class TestEncryptDecryptRoundtrip(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.secret = new_env_var_secret()
|
|
|
|
def _rt(self, plaintext: str) -> str:
|
|
return decrypt_value(self.secret, encrypt_value(self.secret, plaintext))
|
|
|
|
def test_roundtrip_short_value(self) -> None:
|
|
self.assertEqual("sk-abc123", self._rt("sk-abc123"))
|
|
|
|
def test_roundtrip_empty_string(self) -> None:
|
|
self.assertEqual("", self._rt(""))
|
|
|
|
def test_roundtrip_long_value_crosses_block_boundary(self) -> None:
|
|
# 32 bytes is exactly one HMAC-SHA256 block; 65 bytes crosses two.
|
|
plaintext = "x" * 65
|
|
self.assertEqual(plaintext, self._rt(plaintext))
|
|
|
|
def test_roundtrip_unicode(self) -> None:
|
|
self.assertEqual("héllo wörld", self._rt("héllo wörld"))
|
|
|
|
def test_encrypt_produces_different_ciphertexts_each_call(self) -> None:
|
|
ct1 = encrypt_value(self.secret, "same")
|
|
ct2 = encrypt_value(self.secret, "same")
|
|
self.assertNotEqual(ct1, ct2) # fresh nonce each call
|
|
|
|
def test_ciphertext_is_url_safe_base64(self) -> None:
|
|
ct = encrypt_value(self.secret, "hello")
|
|
# no '+', '/', '=' — URL-safe and padding-stripped
|
|
for ch in ("+", "/", "="):
|
|
self.assertNotIn(ch, ct)
|
|
|
|
|
|
class TestDecryptErrors(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.secret = new_env_var_secret()
|
|
|
|
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")
|
|
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 nonce+tag
|
|
|
|
def test_invalid_base64_raises_value_error(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
decrypt_value(self.secret, "!!not-base64!!")
|
|
|
|
|
|
class TestConstant(unittest.TestCase):
|
|
def test_env_var_secret_name(self) -> None:
|
|
self.assertEqual("ENV_VAR_SECRET", ENV_VAR_SECRET_NAME)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|