diff --git a/docs/keycape-upstream-issuer-pin.md b/docs/keycape-upstream-issuer-pin.md new file mode 100644 index 0000000..318f1e4 --- /dev/null +++ b/docs/keycape-upstream-issuer-pin.md @@ -0,0 +1,52 @@ +# Pin KeyCape's verified upstream issuer + +Exercise status: read-only check exercised 2026-09-09 by codex; guarded write unexercised. + +Owner implementation: `sso-mfa/k8s/keycape/openbao-client-config.py`. +Dependency: KEY-WP-0013-T02 / HFACT-WP-0001-T03. The actual signed upstream +issuer was verified as `https://auth.coulomb.social` at 2026-09-08T21:44:44Z; +KeyCape retains that signed-token receipt. This procedure ensures its exact +`authelia.issuer` configuration pin before the separately reviewed client rollout. + +The existing `patch` mode also reconciles client registrations and LLDAP defaults. +Use the dedicated issuer modes for this operation. They read the live Secret in +captured process memory and emit only metadata and fixed result fields. They do +not print the configuration, private key, prior field value or Kubernetes errors. +No Secret backup or patch is written to disk. The patch travels only through +child stdin, never command arguments. The current Kubernetes owner context is +used; no credential is fetched through a generic or unrelated Warden route. + +```bash +python3 -B sso-mfa/k8s/keycape/openbao-client-config.py issuer-check-live +python3 -B sso-mfa/k8s/keycape/openbao-client-config.py issuer-pin-live \ + --expected-uid \ + --expected-resource-version +``` + +The target is fixed to `sso/keycape-config`; the issuer is fixed to the verified +HTTPS value. The write requires both the observed UID and resourceVersion and +tests them atomically in a JSON patch. A matching pin is a no-op. A concurrent +change refuses the write instead of replaying stale configuration. Duplicate +mapping keys, aliases, anchors and unsupported YAML shapes fail closed. + +The mutation inserts or replaces only the issuer scalar. Every other configuration +byte is preserved, and readback compares every Secret data entry, including the +existing private key, without publishing hashes or values. API or parser failures +return fixed reason codes. A readback mismatch stops without another write; +the owner must reconcile the current revision before retrying. There is no +automatic rollback that could overwrite a concurrent credential rotation. + +This step changes no Deployment, process, client registration, credential or +custody policy. The existing process is not restarted. Its separate compatible +image/configuration cutover must still prove readiness and existing human login, +and must retain its owner-controlled rollback pair. Do not call this stored +configuration check proof that the current process has reloaded the field. + +The CCR-2026-0017/0018 named reviews and attended custody window remain separate +gates. This field pin does not approve either request. + +Validation: `python3 -B -m unittest discover -s sso-mfa/k8s/keycape -p 'test_*.py' -v` +passes 13 tests, covering byte preservation, duplicate/alias rejection, stale +revision refusal, atomic preconditions, readback conflict and secret-free failure +output. The first live check on 2026-09-09 found the pin absent at resourceVersion +`51346058`, UID `2e94519d-1550-41c7-9701-2efe47fe1fd3`. diff --git a/sso-mfa/k8s/keycape/openbao-client-config.py b/sso-mfa/k8s/keycape/openbao-client-config.py index 6bcef65..80899bb 100644 --- a/sso-mfa/k8s/keycape/openbao-client-config.py +++ b/sso-mfa/k8s/keycape/openbao-client-config.py @@ -11,7 +11,11 @@ from __future__ import annotations import argparse import base64 import json +import copy +import re +import subprocess import sys +from datetime import datetime, timezone from typing import Any try: @@ -52,6 +56,135 @@ LLDAP_REQUIRED = { "groupOU": "ou=groups", } +# Verified against an actual signed upstream token on 2026-09-08 (KEY-WP-0013). +VERIFIED_UPSTREAM_ISSUER = "https://auth.coulomb.social" + + +class IssuerPinError(Exception): + """Only fixed, non-secret reason codes may leave the contained operation.""" + + +class UniqueLoader(yaml.SafeLoader): + pass + + +def unique_mapping(loader: Any, node: Any) -> dict[str, Any]: + result = {} + for key_node, value_node in node.value: + key = loader.construct_object(key_node, deep=True) + if not isinstance(key, str) or key in result: + raise IssuerPinError("ambiguous_yaml_mapping") + result[key] = loader.construct_object(value_node, deep=True) + return result + + +UniqueLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, unique_mapping) + + +def issuer_document(secret: dict[str, Any]) -> tuple[str, dict[str, Any], Any]: + """Parse privately; never include a parser exception or document in output.""" + raw = base64.b64decode(secret['data']['config.yaml'], validate=True).decode('utf-8') + if any(isinstance(t, (yaml.tokens.AnchorToken, yaml.tokens.AliasToken)) for t in yaml.scan(raw)): + raise IssuerPinError("yaml_alias_or_anchor") + config = yaml.load(raw, Loader=UniqueLoader) + if not isinstance(config, dict) or not isinstance(config.get('authelia'), dict): + raise IssuerPinError("unexpected_configuration_shape") + node = yaml.compose(raw) + authelia = next(v for k, v in node.value if k.value == 'authelia') + if authelia.flow_style or not authelia.value: + raise IssuerPinError("unsupported_authelia_mapping") + return raw, config, authelia + + +def issuer_replacement(secret: dict[str, Any]) -> tuple[str, bool, str]: + raw, config, authelia = issuer_document(secret) + old = config['authelia'].get('issuer') + prior = 'absent' if 'issuer' not in config['authelia'] else 'matching' if old == VERIFIED_UPSTREAM_ISSUER else 'different' + if old == VERIFIED_UPSTREAM_ISSUER: + return secret['data']['config.yaml'], False, prior + field = next((v for k, v in authelia.value if k.value == 'issuer'), None) + if field is not None: + if not isinstance(field, yaml.nodes.ScalarNode): + raise IssuerPinError("unsupported_issuer_value") + updated = raw[:field.start_mark.index] + json.dumps(VERIFIED_UPSTREAM_ISSUER) + raw[field.end_mark.index:] + else: + index = authelia.start_mark.index - authelia.start_mark.column + indent = ' ' * authelia.start_mark.column + updated = raw[:index] + indent + 'issuer: ' + json.dumps(VERIFIED_UPSTREAM_ISSUER) + '\n' + raw[index:] + expected = copy.deepcopy(config) + expected['authelia']['issuer'] = VERIFIED_UPSTREAM_ISSUER + if yaml.load(updated, Loader=UniqueLoader) != expected: + raise IssuerPinError("unexpected_configuration_change") + return base64.b64encode(updated.encode()).decode('ascii'), True, prior + + +def contained_kube(args: list[str], payload: Any = None) -> Any: + # Secret JSON and patch bodies stay in process memory and child stdin. + # In particular, never use -p , a shell pipeline, or inherited stderr. + result = subprocess.run( + ['kubectl', '--request-timeout=20s', '-n', 'sso', *args], + input=None if payload is None else json.dumps(payload), + capture_output=True, text=True, timeout=30, check=False, + ) + if result.returncode: + raise IssuerPinError("kubernetes_operation_failed") + return json.loads(result.stdout) + + +def safe_metadata(obj: dict[str, Any]) -> dict[str, str]: + metadata = obj['metadata'] + uid, revision = metadata['uid'], metadata['resourceVersion'] + if not re.fullmatch(r'[0-9a-f-]{36}', uid) or not re.fullmatch(r'[0-9]+', revision): + raise IssuerPinError("unexpected_metadata_shape") + return {'uid': uid, 'resource_version': revision} + + +def issuer_live(mode: str, expected_uid: str | None = None, expected_revision: str | None = None) -> dict[str, Any]: + if mode == 'issuer-pin-live' and (not expected_uid or not expected_revision): + raise IssuerPinError("pin_requires_observed_uid_and_revision") + before = contained_kube(['get', 'secret', 'keycape-config', '-o', 'json']) + if before.get('kind') != 'Secret' or before['metadata'].get('name') != 'keycape-config' or before['metadata'].get('namespace') != 'sso': + raise IssuerPinError("unexpected_target") + metadata = safe_metadata(before) + if mode == 'issuer-pin-live' and metadata != {'uid': expected_uid, 'resource_version': expected_revision}: + raise IssuerPinError("observed_revision_changed") + replacement, changed, prior = issuer_replacement(before) + receipt = {'schema': 'netkingdom.keycape-issuer-pin.v1', 'mode': mode, + 'observed_at': datetime.now(timezone.utc).isoformat(), + 'target': 'sso/keycape-config', 'verified_issuer': VERIFIED_UPSTREAM_ISSUER, + 'before': metadata, 'prior_pin_state': prior, + 'issuer_matches': not changed, 'config_changed': False, + 'secret_values_emitted': False, 'deployment_changed': False} + if mode == 'issuer-check-live': + return receipt + if changed: + patch = [ + {'op': 'test', 'path': '/metadata/uid', 'value': expected_uid}, + {'op': 'test', 'path': '/metadata/resourceVersion', 'value': expected_revision}, + {'op': 'replace', 'path': '/data/config.yaml', 'value': replacement}, + ] + contained_kube(['patch', 'secret', 'keycape-config', '--type=json', '--patch-file=/dev/stdin', '-o', 'json'], patch) + after = contained_kube(['get', 'secret', 'keycape-config', '-o', 'json']) + expected_data = dict(before['data'], **{'config.yaml': replacement}) + if after['metadata']['uid'] != metadata['uid'] or after['data'] != expected_data: + raise IssuerPinError("readback_mismatch_stop_without_replaying_stale_config") + receipt.update({'after': safe_metadata(after), 'issuer_matches': True, + 'config_changed': changed, 'other_secret_data_unchanged': True, + 'unrelated_config_bytes_preserved': True}) + return receipt + + +def issuer_main(args: Any) -> int: + try: + receipt = issuer_live(args.mode, args.expected_uid, args.expected_resource_version) + except IssuerPinError as exc: + receipt = {'schema': 'netkingdom.keycape-issuer-pin.v1', 'status': 'failed', 'reason': str(exc)} + except Exception: + # Kubernetes/YAML exceptions may contain entire response/config fragments. + receipt = {'schema': 'netkingdom.keycape-issuer-pin.v1', 'status': 'failed', 'reason': 'contained_operation_failed'} + print(json.dumps(receipt, sort_keys=True)) + return 1 if receipt.get('status') == 'failed' else 0 + def load_config() -> dict[str, Any]: secret = json.load(sys.stdin) @@ -142,8 +275,12 @@ def verify(config: dict[str, Any]) -> None: def main() -> None: parser = argparse.ArgumentParser() - parser.add_argument("mode", choices=("patch", "verify")) + parser.add_argument("mode", choices=("patch", "verify", "issuer-check-live", "issuer-pin-live")) + parser.add_argument("--expected-uid") + parser.add_argument("--expected-resource-version") args = parser.parse_args() + if args.mode.startswith('issuer-'): + raise SystemExit(issuer_main(args)) config = load_config() if args.mode == "patch": render_patch(config) diff --git a/sso-mfa/k8s/keycape/test_issuer_pin.py b/sso-mfa/k8s/keycape/test_issuer_pin.py new file mode 100644 index 0000000..b84b652 --- /dev/null +++ b/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()