Add contained issuer-only configuration check and revision-guarded pin

Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc

sso-mfa/k8s/keycape/test_issuer_pin.py

@@ -0,0 +1,115 @@
+"""Issuer-only mutation, concurrency and output-containment regression checks."""
+import argparse
+import base64
+import copy
+import importlib.util
+import io
+import json
+import subprocess
+import unittest
+from contextlib import redirect_stdout, redirect_stderr
+from pathlib import Path
+from unittest.mock import patch
+
+spec = importlib.util.spec_from_file_location('owner_config', Path(__file__).with_name('openbao-client-config.py'))
+m = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(m)
+UID = '2e94519d-1550-41c7-9701-2efe47fe1fd3'
+SENTINEL = 'DO-NOT-EMIT-fixture-private-value'
+RAW = '# retained comment\nauthelia:\n  baseURL: https://auth.coulomb.social\n  clientSecret: '+SENTINEL+'\nclients:\n  - clientId: existing\n    secretRef: env:EXISTING\n'
+
+
+def secret(raw=RAW):
+    return {'kind': 'Secret', 'metadata': {'name': 'keycape-config', 'namespace': 'sso', 'uid': UID, 'resourceVersion': '12'},
+            'data': {'config.yaml': base64.b64encode(raw.encode()).decode(), 'key.pem': base64.b64encode(SENTINEL.encode()).decode()}}
+
+
+class IssuerPinTest(unittest.TestCase):
+    def test_add_preserves_every_other_byte(self):
+        encoded, changed, prior = m.issuer_replacement(secret())
+        updated = base64.b64decode(encoded).decode()
+        self.assertTrue(changed)
+        self.assertEqual(prior, 'absent')
+        self.assertEqual(updated.replace('  issuer: "https://auth.coulomb.social"\n', ''), RAW)
+
+    def test_replace_retains_comments_and_other_fields(self):
+        raw = RAW.replace('authelia:\n', 'authelia:\n  issuer: "http://old.example" # keep\n')
+        encoded, changed, prior = m.issuer_replacement(secret(raw))
+        self.assertTrue(changed)
+        self.assertEqual(prior, 'different')
+        self.assertEqual(base64.b64decode(encoded).decode(), raw.replace('"http://old.example"', '"https://auth.coulomb.social"'))
+
+    def test_matching_is_byte_identical_noop(self):
+        raw = RAW.replace('authelia:\n', 'authelia:\n  issuer: https://auth.coulomb.social\n')
+        source = secret(raw)
+        self.assertEqual(m.issuer_replacement(source), (source['data']['config.yaml'], False, 'matching'))
+
+    def test_ambiguous_documents_and_aliases_are_refused(self):
+        for raw in [RAW+'authelia: {}\n', RAW.replace('  baseURL:', '  issuer: a\n  issuer: b\n  baseURL:'),
+                    'authelia: {issuer: old}\n', RAW+'---\nauthelia: {}\n', RAW+'alias: &shared value\ncopy: *shared\n']:
+            with self.subTest(raw=raw), self.assertRaises(Exception):
+                m.issuer_replacement(secret(raw))
+
+    def test_check_never_patches_or_reports_old_value(self):
+        source = secret(RAW.replace('authelia:\n', 'authelia:\n  issuer: '+SENTINEL+'\n'))
+        with patch.object(m, 'contained_kube', return_value=source) as kube:
+            result = m.issuer_live('issuer-check-live')
+        self.assertEqual(kube.call_count, 1)
+        self.assertFalse(result['issuer_matches'])
+        self.assertNotIn(SENTINEL, json.dumps(result))
+
+    def test_pin_checks_observed_revision_before_mutation(self):
+        with patch.object(m, 'contained_kube', return_value=secret()) as kube:
+            with self.assertRaisesRegex(m.IssuerPinError, 'observed_revision_changed'):
+                m.issuer_live('issuer-pin-live', UID, '11')
+        self.assertEqual(kube.call_count, 1)
+
+    def test_pin_uses_cas_and_proves_other_secret_data_unchanged(self):
+        state = secret()
+        calls = []
+        def kube(args, payload=None):
+            calls.append((args, payload))
+            if args[0] == 'patch':
+                self.assertEqual(payload[:2], [{'op': 'test', 'path': '/metadata/uid', 'value': UID},
+                                              {'op': 'test', 'path': '/metadata/resourceVersion', 'value': '12'}])
+                self.assertEqual(payload[2]['path'], '/data/config.yaml')
+                state['data']['config.yaml'] = payload[2]['value']
+                state['metadata']['resourceVersion'] = '13'
+            return copy.deepcopy(state)
+        with patch.object(m, 'contained_kube', side_effect=kube):
+            result = m.issuer_live('issuer-pin-live', UID, '12')
+        self.assertEqual(len(calls), 3)
+        self.assertTrue(result['issuer_matches'])
+        self.assertTrue(result['other_secret_data_unchanged'])
+        self.assertNotIn(SENTINEL, json.dumps(result))
+
+    def test_readback_race_is_not_retried_or_overwritten(self):
+        after = secret()
+        after['data']['key.pem'] = 'changed-by-another-actor'
+        with patch.object(m, 'contained_kube', side_effect=[secret(), {}, after]) as kube:
+            with self.assertRaisesRegex(m.IssuerPinError, 'readback_mismatch'):
+                m.issuer_live('issuer-pin-live', UID, '12')
+        self.assertEqual(kube.call_count, 3)
+
+    def test_subprocess_error_output_and_parser_errors_stay_contained(self):
+        args = argparse.Namespace(mode='issuer-check-live', expected_uid=None, expected_resource_version=None)
+        failures = [ValueError(SENTINEL), subprocess.TimeoutExpired(['kubectl'], 30, output=SENTINEL)]
+        for failure in failures:
+            output = io.StringIO()
+            with patch.object(m, 'contained_kube', side_effect=failure), redirect_stdout(output), redirect_stderr(output):
+                self.assertEqual(m.issuer_main(args), 1)
+            self.assertNotIn(SENTINEL, output.getvalue())
+            self.assertEqual(json.loads(output.getvalue())['reason'], 'contained_operation_failed')
+
+    def test_patch_payload_only_goes_to_captured_child_stdin(self):
+        with patch.object(m.subprocess, 'run', return_value=subprocess.CompletedProcess([], 1, SENTINEL, SENTINEL)) as run:
+            with self.assertRaisesRegex(m.IssuerPinError, '^kubernetes_operation_failed$'):
+                m.contained_kube(['patch', 'secret', 'keycape-config'], {'data': SENTINEL})
+        args, kwargs = run.call_args
+        self.assertNotIn(SENTINEL, json.dumps(args))
+        self.assertIn(SENTINEL, kwargs['input'])
+        self.assertTrue(kwargs['capture_output'])
+
+
+if __name__ == '__main__':
+    unittest.main()