fix(secret): authenticate bottled-secret encryption (#468)
prd-number-check / require-numbered-prds (pull_request) Failing after 14s
tracker-policy-pr / check-pr (pull_request) Successful in 12s
test / integration-docker (pull_request) Successful in 46s
test / unit (pull_request) Successful in 56s
lint / lint (push) Successful in 1m6s
test / coverage (pull_request) Successful in 24s
prd-number-check / require-numbered-prds (pull_request) Failing after 14s
tracker-policy-pr / check-pr (pull_request) Successful in 12s
test / integration-docker (pull_request) Successful in 46s
test / unit (pull_request) Successful in 56s
lint / lint (push) Successful in 1m6s
test / coverage (pull_request) Successful in 24s
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>
This commit is contained in:
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user