import base64 import copy import importlib.util import json from pathlib import Path import tempfile import unittest from unittest.mock import patch import yaml PATH = Path(__file__).with_name('approval-clients-rollout.py') spec = importlib.util.spec_from_file_location('rollout', PATH) m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) def fixture(raw=None): raw = raw or 'authelia:\n issuer: https://auth.coulomb.social\n clientSecret: synthetic-fixture\nclients:\n - clientId: existing-human\n allowedScopes: [openid]\n# trailing owner comment\nother: preserved\n' return {'kind': 'Secret', 'metadata': {'name': 'keycape-config', 'namespace': 'sso', 'uid': '2e94519d-1550-41c7-9701-2efe47fe1fd3', 'resourceVersion': '100'}, 'data': {'config.yaml': base64.b64encode(raw.encode()).decode(), 'key.pem': 'c3ludGhldGlj'}} def deployment(): source = next(yaml.safe_load_all((PATH.parent / 'deployment.yaml').read_text())) source['metadata'].update(uid='99ddd83c-cb3f-4847-bcf8-35f1aa87627f', resourceVersion='101') return source class RolloutTests(unittest.TestCase): def test_append_preserves_original_bytes_and_signing_key(self): secret = fixture(); before = copy.deepcopy(secret) actual = base64.b64decode(m.replacement(secret, m.registrations())).decode() original = base64.b64decode(secret['data']['config.yaml']).decode() insertion_start = original.index('other:') self.assertTrue(actual.startswith(original[:insertion_start])) self.assertTrue(actual.endswith(original[insertion_start:])) self.assertEqual(secret, before) self.assertEqual(len(yaml.safe_load(actual)['clients']), 3) def test_indentless_and_final_client_sequence(self): for text in ['authelia:\n issuer: https://auth.coulomb.social\nclients:\n- clientId: human\nother: keep\n', 'authelia:\n issuer: https://auth.coulomb.social\nclients:\n - clientId: human']: parsed = yaml.safe_load(base64.b64decode(m.replacement(fixture(text), m.registrations()))) self.assertEqual(len(parsed['clients']), 3) def test_reject_duplicate_existing_registration(self): text = 'authelia:\n issuer: https://auth.coulomb.social\nclients:\n - clientId: secrets-engine-approval\n' with self.assertRaisesRegex(m.LaneError, 'existing_client_registration'): m.replacement(fixture(text), m.registrations()) def test_reject_ambiguous_yaml_and_wrong_issuer(self): for raw in ['authelia: {}\nauthelia: {}\nclients: []', 'authelia:\n issuer: https://wrong.invalid\nclients:\n - clientId: human\n']: with self.assertRaises((m.LaneError, m.pin.IssuerPinError)): m.replacement(fixture(raw), m.registrations()) def test_candidate_preserves_existing_settings(self): old = deployment(); result = m.candidate_spec(old) before = old['spec']['template']['spec']['containers'][0] after = result['template']['spec']['containers'][0] self.assertEqual(result['strategy'], {'type': 'Recreate'}) self.assertEqual(after['env'][:len(before['env'])], before['env']) for field in ['livenessProbe', 'resources', 'volumeMounts', 'startupProbe']: self.assertEqual(after[field], before[field]) self.assertEqual(after['image'], m.IMAGE) def test_cas_patch_contains_both_preconditions_and_stdin_only(self): with patch.object(m, 'command') as command: command.return_value.stdout = b'{}' m.patch_object(['kubectl'], 'secret', fixture(), '/data/config.yaml', 'synthetic-payload') args, kwargs = command.call_args self.assertNotIn('synthetic-payload', args[0]) self.assertEqual([p['path'] for p in kwargs['payload'][:2]], ['/metadata/uid', '/metadata/resourceVersion']) def test_failed_acceptance_restores_compatible_pair(self): state = {'secret': fixture(), 'deployment': deployment()} original = copy.deepcopy(state) def fake_patch(kube, kind, obj, path, value, dry=False): changed = copy.deepcopy(obj) if path == '/spec': changed['spec'] = value else: changed['data']['config.yaml'] = value if not dry: changed['metadata']['resourceVersion'] = str(int(changed['metadata']['resourceVersion']) + 1) state[kind] = changed return copy.deepcopy(changed) with tempfile.TemporaryDirectory() as directory, patch.object(m, 'assert_cluster'), patch.object(m, 'verify_artifact'), \ patch.object(m, 'get', side_effect=lambda kube, kind, name: copy.deepcopy(state[kind])), \ patch.object(m, 'patch_object', side_effect=fake_patch), patch.object(m, 'ready', return_value={}), \ patch.object(m, 'acceptance', side_effect=m.LaneError('synthetic_acceptance_failure')): receipt = {} with self.assertRaisesRegex(m.LaneError, 'synthetic_acceptance_failure'): m.rollout([], receipt, Path(directory) / 'recovery.json') self.assertTrue(receipt['compatible_pair_restored']) self.assertEqual(state['secret']['data'], original['secret']['data']) self.assertEqual(state['deployment']['spec'], original['deployment']['spec']) self.assertEqual((Path(directory) / 'recovery.json').stat().st_mode & 0o777, 0o600) def test_ready_matches_manifest_imageid_not_runtime_config_id(self): dep = deployment(); dep['metadata']['generation'] = 30 dep['status'] = {'observedGeneration': 30, 'updatedReplicas': 1, 'readyReplicas': 1, 'availableReplicas': 1, 'replicas': 1} pod = {'metadata': {'uid': 'fixture-pod'}, 'status': {'containerStatuses': [ {'name': 'keycape', 'ready': True, 'image': 'sha256:runtime-config-id', 'imageID': m.IMAGE}]}} from types import SimpleNamespace with patch.object(m, 'get', return_value=dep), patch.object(m, 'command', return_value=SimpleNamespace(stdout=json.dumps({'items': [pod]}).encode())): self.assertTrue(m.ready([], m.IMAGE, timeout=1)['single_ready_replica']) def test_iat_skew_matches_native_contract_without_extending_expiry(self): from cryptography.hazmat.primitives.asymmetric import rsa key = rsa.generate_private_key(public_exponent=65537, key_size=2048) now = int(m.time.time()) claims = {'iss': m.ISSUER, 'aud': 'approval-engine', 'sub': 'synthetic-service', 'iat': now + 2, 'exp': now + 902} encoded = m.jwt.encode(claims, key, algorithm='RS256') self.assertEqual(m.verified_claims(encoded, key.public_key())['iat'], now + 2) future = m.jwt.encode(dict(claims, iat=now+60), key, algorithm='RS256') with self.assertRaisesRegex(m.LaneError, 'issued_at_binding_failed'): m.verified_claims(future, key.public_key()) expired = m.jwt.encode(dict(claims, iat=now-900, exp=now-1), key, algorithm='RS256') with self.assertRaises(m.jwt.ExpiredSignatureError): m.verified_claims(expired, key.public_key()) def test_unrelated_refusal_never_passes(self): for status, body in [(500, {}), (400, {'error': 'invalid_profile_usage', 'feature': 'client_id'}), (401, {'error': 'invalid_profile_usage', 'feature': 'scope'})]: with self.assertRaises(m.LaneError): m.denied((status, body), 400, 'scope') if __name__ == '__main__': unittest.main()