"""Exercise the attended controller against synthetic Kubernetes/Bao boundaries.""" import copy import io import json from pathlib import Path import subprocess import sys import tempfile import unittest from unittest.mock import patch, Mock sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts')) import email_connect_smtp_foundation as foundation class Native: def __init__(self, proposal): self.bundle = proposal['foundation'] self.created = [] self.objects = {} self.bad_cluster = False self.bad_binding = False self.target_exists = False self.fail_kind = None self.not_ready = False self.drift = False self.changed_application = False self.bad_ownership = False def get(self, kind, name, namespace=foundation.NS): if kind == 'namespace': key = 'cluster_uid' if name == 'kube-system' else 'namespace_uid' return {'metadata': {'uid': 'wrong' if self.bad_cluster else self.bundle[key]}} if kind == 'deployment': return {'metadata': {'uid': self.bundle['deployment_uid'], 'generation': 74}, 'spec': {'unchanged': not (self.created and self.changed_application)}} if name in ('email-connect-config', 'email-connect-runtime'): return {'metadata': {'uid': name}, 'spec': {}, 'data': {}} return copy.deepcopy(self.objects.get((kind, name))) def target_metadata(self, name): if self.target_exists: return {'uid': 'foreign'} if 'ExternalSecret' in self.created: return {'ownerReferences': [{'uid': 'wrong' if self.bad_ownership else 'uid-ExternalSecret', 'controller': True}]} return None def create(self, wanted): kind = wanted['kind'] self.created.append(kind) if self.fail_kind == kind: raise RuntimeError('synthetic provider secret text') obj = copy.deepcopy(wanted) obj['metadata']['uid'] = 'uid-'+kind obj['status'] = {'conditions': [{'type': 'Ready', 'status': 'False' if self.not_ready else 'True'}], 'refreshTime': '2026-10-31T12:00:00Z'} if self.drift and kind == 'SecretStore': obj['spec']['provider']['vault']['path'] = 'wrong' self.objects[(kind, wanted['metadata']['name'])] = obj return copy.deepcopy(obj) def jwt(self, *args): return 'fixture-jwt' def token_review(self, jwt): return {'authenticated': not self.bad_binding, 'audiences': ['openbao'], 'user': {'username': 'system:serviceaccount:email-connect:'+foundation.SA, 'uid': 'uid-ServiceAccount'}} def wait(self, resource): pass def completed(value, code=0): return subprocess.CompletedProcess([], code, json.dumps(value).encode(), b'403 permission denied' if code else b'') class FoundationTests(unittest.TestCase): def setUp(self): self.ccr, self.proposal, self.digest = foundation.prepare() self.native = Native(self.proposal) self.validator = Mock() self.records = [] self.owner_calls = [] def owner(self, args, **kwargs): if args[0] == 'status': return completed({'cluster_id': foundation.admission.OPENBAO_CLUSTER_ID, 'initialized': True, 'sealed': False}) return completed({'data': {'policies': ['platform-admin']}}) def run_controller(self): approved = copy.deepcopy(self.ccr) approved['status'] = 'approved' approved['openbao']['auth']['bound_claims_confirmed'] = True def admit(row, checkpoint): self.owner_calls.append('admit') row.update(status='reader_admitted_verified'); checkpoint(row) receipt = {'proposal_digest': self.digest, 'status': 'started'} with patch.object(foundation, 'bao', side_effect=self.owner), \ patch.object(foundation.admission, 'module', return_value=self.validator), \ patch.object(foundation.admission, 'prepare', return_value=(approved, {}, self.proposal['admission_digest'])), \ patch.object(foundation.admission, 'admit', side_effect=admit), \ patch.object(foundation, 'verify_login', return_value={'native_consumer_auth_verified': True}): foundation.run(self.proposal, receipt, lambda r: self.records.append(copy.deepcopy(r)), self.native) return receipt def test_offline_proposal_and_live_default_refusal(self): self.assertEqual(self.ccr['status'], 'proposed') self.assertEqual(len(self.proposal['foundation']['resources']), 3) with tempfile.TemporaryDirectory() as tmp: path = Path(tmp)/'receipt' with self.assertRaises(foundation.admission.Refused): foundation.execute(self.ccr, self.proposal, self.digest, self.digest, path) self.assertFalse(path.exists()) def test_complete_controller_requires_binding_before_admission_and_ready_after(self): result = self.run_controller() self.assertEqual(self.native.created, ['ServiceAccount', 'SecretStore', 'ExternalSecret']) self.validator.confirm_binding.assert_called_once() self.assertEqual(result['status'], 'foundation_admitted_projection_ready') self.assertFalse(result['application_sender_changed']) phases = [r['phase'] for r in self.records] self.assertLess(phases.index('binding_verified'), phases.index('owner_admission')) self.assertLess(phases.index('native_login_verification'), phases.index('creating_secretstore')) def test_wrong_cluster_and_existing_targets_precede_mutation(self): for mode in ('cluster', 'target', 'existing'): with self.subTest(mode=mode): self.native = Native(self.proposal) self.native.bad_cluster = mode == 'cluster' self.native.target_exists = mode == 'target' if mode == 'existing': self.native.objects[('SecretStore', 'openbao-email-connect-platform-smtp')] = {'foreign': True} with self.assertRaises(foundation.admission.Refused): self.run_controller() self.assertFalse(self.native.created) self.assertFalse(self.owner_calls) def test_bad_tokenreview_never_confirms_or_admits(self): self.native.bad_binding = True with self.assertRaises(foundation.admission.Refused): self.run_controller() self.assertEqual(self.native.created, ['ServiceAccount']) self.validator.confirm_binding.assert_not_called() self.assertFalse(self.owner_calls) def test_projection_drift_not_ready_foreign_owner_and_application_changes_refuse(self): for field in ('drift', 'not_ready', 'bad_ownership', 'changed_application'): with self.subTest(field=field): self.native = Native(self.proposal); setattr(self.native, field, True) with self.assertRaises(foundation.admission.Refused): self.run_controller() self.assertFalse(any(r.get('status') == 'foundation_admitted_projection_ready' for r in self.records)) def test_creation_failure_keeps_checkpoint_and_does_not_rollback_or_retry(self): self.native.fail_kind = 'SecretStore' with self.assertRaises(RuntimeError): self.run_controller() self.assertEqual(self.native.created, ['ServiceAccount', 'SecretStore']) self.assertEqual(self.records[-1]['phase'], 'creating_secretstore') self.assertEqual(self.records[-1]['mutation_outcome'], 'unconfirmed') def test_journal_failure_silent_private_and_exclusive(self): self.ccr['status'] = 'approved' with tempfile.TemporaryDirectory() as tmp: path = Path(tmp)/'receipt' def fail(proposal, receipt, checkpoint): receipt.update(phase='creating_service_account', mutation_outcome='unconfirmed') checkpoint(receipt) raise RuntimeError('synthetic private provider text') self.assertEqual(foundation.execute(self.ccr, self.proposal, self.digest, self.digest, path, fail), 1) raw = path.read_bytes() rows = [json.loads(line) for line in raw.splitlines()] self.assertEqual(rows[-1]['status'], 'failed_requires_inspection') self.assertEqual(rows[-1]['phase'], 'creating_service_account') self.assertNotIn(b'private provider', raw) self.assertEqual(path.stat().st_mode & 0o777, 0o600) with self.assertRaises(FileExistsError): foundation.execute(self.ccr, self.proposal, self.digest, self.digest, path, fail) self.assertEqual(path.read_bytes(), raw) def test_interruption_cannot_be_success(self): self.ccr['status'] = 'approved' with tempfile.TemporaryDirectory() as tmp: path = Path(tmp)/'receipt' def interrupt(*args): raise SystemExit(9) with self.assertRaises(SystemExit): foundation.execute(self.ccr, self.proposal, self.digest, self.digest, path, interrupt) last = json.loads(path.read_text().splitlines()[-1]) self.assertEqual(last['status'], 'started') self.assertIn('interrupted_at', last) def test_native_login_scope_denials_and_all_tokens_revoked(self): calls = [] def bao(args, **kw): calls.append((args, kw)) if args[:2] == ['write', '-format=json']: # The fake JWT identifies the requested service account/audience. if kw['payload']['jwt'] != foundation.SA+':openbao': return completed({}, 1) return completed({'auth': {'client_token': 'fixture', 'token_policies': [foundation.SA], 'lease_duration': 900}}) if args[0] == 'read': return completed({'data': {'data': dict(foundation.admission.lane_module().FIELDS, SMTP_PASSWORD='fixture'), 'metadata': {'version': 2}}}) if args[0] == 'write': return completed({}) return completed({}, 1) def caps(token, paths): return dict(zip(paths, [['read'], ['deny'], ['deny'], ['deny'], ['read'], ['update']])) native = Mock() native.jwt.side_effect = lambda sa=foundation.SA, audience='openbao': sa+':'+audience with patch.object(foundation, 'bao', side_effect=bao), patch.object(foundation, 'capabilities', side_effect=caps): result = foundation.verify_login(native) self.assertTrue(result['other_service_account_denied']) self.assertTrue(result['workload_token_revocation_verified']) self.assertEqual(sum(a == ['write', '-f', 'auth/token/revoke-self'] for a, _ in calls), 1) def test_wrong_digest_and_cli_refusal_are_before_receipt_or_owner_calls(self): self.ccr['status'] = 'approved' with tempfile.TemporaryDirectory() as tmp: path = Path(tmp)/'receipt' with self.assertRaises(foundation.admission.Refused): foundation.execute(self.ccr, self.proposal, self.digest, 'wrong', path) result = subprocess.run([sys.executable, str(Path(foundation.__file__)), '--execute', '--expect-digest', 'wrong', '--receipt', str(path)], capture_output=True, timeout=10) self.assertNotEqual(result.returncode, 0) self.assertEqual((result.stdout, result.stderr), (b'', b'')) self.assertFalse(path.exists()) def test_metadata_query_negotiates_partial_object_and_cleans_proxy(self): for status, payload, refuses in [ (404, {}, False), (200, {'kind': 'PartialObjectMetadata', 'apiVersion': 'meta.k8s.io/v1', 'metadata': {'uid': 'fixture'}}, False), (200, {'kind': 'Secret', 'apiVersion': 'v1', 'data': {'fixture': 'must-not-be-accepted'}}, True), (403, {}, True), ]: with self.subTest(status=status, refuses=refuses): response = io.BytesIO(json.dumps(payload).encode()); response.status = status process = Mock(); process.poll.return_value = None conn = Mock(); conn.getresponse.return_value = response with patch.object(foundation.subprocess, 'Popen', return_value=process), \ patch.object(foundation.socket, 'socket'), \ patch.object(foundation.http.client, 'HTTPConnection', return_value=conn), \ patch.object(foundation.time, 'sleep'): if refuses: with self.assertRaises(foundation.admission.Refused): foundation.Native().target_metadata('email-connect-platform-smtp') else: result = foundation.Native().target_metadata('email-connect-platform-smtp') self.assertEqual(result, None if status == 404 else {'uid': 'fixture'}) self.assertEqual(conn.request.call_args.kwargs['headers']['Accept'], 'application/json;as=PartialObjectMetadata;g=meta.k8s.io;v=v1') process.terminate.assert_called_once(); process.wait.assert_called_once() conn.close.assert_called_once() def test_native_scope_failure_still_revokes_workload_token(self): calls = [] def bao(args, **kwargs): calls.append(args) if args[:2] == ['write', '-format=json']: return completed({'auth': {'client_token': 'fixture', 'token_policies': ['too-broad'], 'lease_duration': 900}}) if args[:2] == ['write', '-f']: return completed({}) return completed({}, 1) with patch.object(foundation, 'bao', side_effect=bao): with self.assertRaises(foundation.admission.Refused): foundation.verify_login(Mock()) self.assertIn(['write', '-f', 'auth/token/revoke-self'], calls) if __name__ == '__main__': unittest.main()