From 3b6f68d26d8d0fab6855c63a42d4659cf4819fcd Mon Sep 17 00:00:00 2001 From: didericis Date: Tue, 21 Jul 2026 22:23:18 -0400 Subject: [PATCH] test(secrets): cover per-bottle egress secret encryption Add unit coverage for the encrypted at-rest egress secrets: the secret_store round-trip (new_env_var_secret/encrypt_value/decrypt_value) and the registry/service wiring that injects ENV_VAR_SECRET. Recovered from the bot-bottle-claude-agent-1 VM after the CI runner's disk filled and forced its rootfs read-only; committed work was already on origin, these were the agent's uncommitted changes. Co-Authored-By: Claude Opus 4.8 --- tests/unit/test_orchestrator_registry.py | 52 +++++++++++ tests/unit/test_orchestrator_secret_store.py | 96 ++++++++++++++++++++ tests/unit/test_orchestrator_service.py | 1 + 3 files changed, 149 insertions(+) create mode 100644 tests/unit/test_orchestrator_secret_store.py diff --git a/tests/unit/test_orchestrator_registry.py b/tests/unit/test_orchestrator_registry.py index 3328a71..90d0ff4 100644 --- a/tests/unit/test_orchestrator_registry.py +++ b/tests/unit/test_orchestrator_registry.py @@ -173,6 +173,58 @@ if __name__ == "__main__": unittest.main() +class TestAgentSecrets(unittest.TestCase): + """store/get/delete for the bottled_agent_secrets table.""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db = Path(self._tmp.name) / "registry.db" + self.store = RegistryStore(self.db) + self.store.migrate() + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_store_and_get_roundtrip(self) -> None: + self.store.store_agent_secrets("bottle-1", {"EGRESS_TOKEN_1": "enc-val-a"}) + got = self.store.get_agent_secrets("bottle-1") + self.assertEqual({"EGRESS_TOKEN_1": "enc-val-a"}, got) + + def test_get_returns_empty_when_none_stored(self) -> None: + self.assertEqual({}, self.store.get_agent_secrets("no-such-bottle")) + + def test_store_replaces_existing_rows(self) -> None: + self.store.store_agent_secrets("bottle-1", {"K": "old"}) + self.store.store_agent_secrets("bottle-1", {"K": "new", "K2": "v2"}) + got = self.store.get_agent_secrets("bottle-1") + self.assertEqual({"K": "new", "K2": "v2"}, got) + + def test_delete_removes_secrets(self) -> None: + self.store.store_agent_secrets("bottle-1", {"K": "v"}) + self.store.delete_agent_secrets("bottle-1") + self.assertEqual({}, self.store.get_agent_secrets("bottle-1")) + + def test_delete_is_idempotent_on_missing(self) -> None: + self.store.delete_agent_secrets("no-such-bottle") # must not raise + + def test_secrets_are_isolated_by_bottle_id(self) -> None: + self.store.store_agent_secrets("bottle-1", {"K": "for-1"}) + self.store.store_agent_secrets("bottle-2", {"K": "for-2"}) + self.assertEqual({"K": "for-1"}, self.store.get_agent_secrets("bottle-1")) + self.assertEqual({"K": "for-2"}, self.store.get_agent_secrets("bottle-2")) + + def test_secrets_isolated_by_type(self) -> None: + self.store.store_agent_secrets("bottle-1", {"K": "injected"}, secret_type="injected_env_var") + self.store.store_agent_secrets("bottle-1", {"K": "other"}, secret_type="other_type") + self.assertEqual({"K": "injected"}, self.store.get_agent_secrets("bottle-1")) + self.assertEqual({"K": "other"}, self.store.get_agent_secrets("bottle-1", secret_type="other_type")) + + def test_secrets_persist_across_reopen(self) -> None: + self.store.store_agent_secrets("bottle-1", {"K": "v"}) + reopened = RegistryStore(self.db) + self.assertEqual({"K": "v"}, reopened.get_agent_secrets("bottle-1")) + + class TestReapAbsent(unittest.TestCase): """`reap_absent` — the self-heal for rows whose bottle is gone. diff --git a/tests/unit/test_orchestrator_secret_store.py b/tests/unit/test_orchestrator_secret_store.py new file mode 100644 index 0000000..64dd453 --- /dev/null +++ b/tests/unit/test_orchestrator_secret_store.py @@ -0,0 +1,96 @@ +"""Unit tests for per-bottle egress secret encryption (PRD prd-new-secret-provider).""" + +from __future__ import annotations + +import unittest + +from bot_bottle.orchestrator.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() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index efee477..67c4eca 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -13,6 +13,7 @@ from unittest.mock import patch from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker from bot_bottle.orchestrator.registry import RegistryStore +from bot_bottle.orchestrator.secret_store import new_env_var_secret from bot_bottle.orchestrator.service import Orchestrator from bot_bottle.orchestrator.gateway import Gateway from bot_bottle.store_manager import StoreManager