didericis
44e2b5a897
test / integration-docker (pull_request) Successful in 13s
tracker-policy-pr / check-pr (pull_request) Successful in 14s
test / unit (pull_request) Successful in 42s
lint / lint (push) Failing after 54s
test / integration-firecracker (pull_request) Successful in 3m17s
test / coverage (pull_request) Successful in 17s
test / publish-infra (pull_request) Has been skipped
Give each service its own store package + manager, and cut the supervise module
along the control/data-plane boundary so nothing in the shared layer reaches up
into the orchestrator.
Stores, by owner:
- bot_bottle/store/ keeps only the shared base (DbStore, migrations) and the
concrete stores that aren't service-owned (audit_store, config_store).
- bot_bottle/orchestrator/store/ now houses the orchestrator-owned stores —
queue_store (supervise queue), secret_store, config_store — plus a new
orchestrator store_manager that migrates them (composing audit/config
downward from the base). The old shared store_manager is gone.
Supervise plane, by tier:
- bot_bottle/supervisor/ (NEUTRAL, importable by every tier including the
gateway): types.py (the Proposal/Response/AuditEntry wire types + the tool/
status/poll constants + the shared daemon constants moved out of
supervise.py) and plan.py (SupervisePlan, a pure DTO).
- bot_bottle/orchestrator/supervisor/ (orchestrator-only): queue.py (the
queue/audit I/O wrappers + render_diff + sha256_hex) and supervise.py (the
Supervise lifecycle that stages the DB via the store manager). Its __init__
re-exports the neutral vocabulary so orchestrator-side callers import from
one place.
The gateway now imports only bot_bottle.supervisor.types (never
bot_bottle.supervise), so the data plane holds no code dependency on the
orchestrator — it reaches the queue over the control-plane RPC. This removes
the circular import that moving queue_store under orchestrator introduced
(supervise -> orchestrator -> service -> supervise).
supervise_types.py -> supervisor/types.py; supervise.py deleted (split). Full
unit suite green (2251).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
97 lines
3.3 KiB
Python
97 lines
3.3 KiB
Python
"""Unit tests for per-bottle egress secret encryption (PRD prd-new-secret-provider)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import unittest
|
|
|
|
from bot_bottle.orchestrator.store.secret_store import (
|
|
ENV_VAR_SECRET_NAME,
|
|
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_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
|
|
|
|
def test_truncated_blob_raises_value_error(self) -> None:
|
|
with self.assertRaises(ValueError):
|
|
decrypt_value(self.secret, "dG9vc2hvcnQ") # "tooshort" — under 16 nonce bytes
|
|
|
|
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()
|