e77f3989b4
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
262 lines
14 KiB
Python
262 lines
14 KiB
Python
"""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()
|