From e556af85714325c00f7bf30ba8d2a7e3c47daed3 Mon Sep 17 00:00:00 2001 From: tegwick Date: Thu, 29 Oct 2026 22:18:30 +0100 Subject: [PATCH] NK-WP-0054-T05 phase 4: realm_bootstrap --switch-bind moves the resolver to svc-privacyidea Only from the legacy uid=admin bind, only BINDDN/BINDPW, nothing else drifted. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 308701@bnt-lap001 Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56 --- sso-mfa/k8s/privacyidea/realm_bootstrap.py | 35 +++++++++++++++++----- tests/test_realm_bootstrap.py | 26 ++++++++++++++++ 2 files changed, 53 insertions(+), 8 deletions(-) diff --git a/sso-mfa/k8s/privacyidea/realm_bootstrap.py b/sso-mfa/k8s/privacyidea/realm_bootstrap.py index 28ff238..109f516 100644 --- a/sso-mfa/k8s/privacyidea/realm_bootstrap.py +++ b/sso-mfa/k8s/privacyidea/realm_bootstrap.py @@ -17,6 +17,8 @@ REALM = 'coulomb' CORE = {'LDAPURI': 'ldap://lldap.sso.svc.cluster.local:3890', 'LDAPBASE': 'dc=netkingdom,dc=local', 'BINDDN': 'uid=svc-privacyidea,ou=people,dc=netkingdom,dc=local'} TUNING = {'TIMEOUT': '5', 'CACHE_TIMEOUT': '120', 'SIZELIMIT': '500'} +# Before NK-WP-0054-T05 the resolver bound as the human admin; only --switch-bind moves it. +LEGACY_BINDDN = 'uid=admin,ou=people,dc=netkingdom,dc=local' class Refused(Exception): @@ -64,7 +66,7 @@ class Client: return {'resolver': resolvers.get(RESOLVER), 'realms': realms} -def plan(snapshot): +def plan(snapshot, switch_bind=False): resolver, realms = snapshot['resolver'], snapshot['realms'] updates = {} if resolver is None: @@ -77,11 +79,17 @@ def plan(snapshot): if not isinstance(resolver, dict) or resolver.get('type') != 'ldapresolver' or not isinstance(resolver.get('data'), dict): raise Refused() data = resolver['data'] - if any(data.get(key) != value for key, value in CORE.items()): raise Refused() + if any(data.get(key) != value for key, value in CORE.items() if key != 'BINDDN'): raise Refused() + if data.get('BINDDN') != CORE['BINDDN']: + if not (switch_bind and data.get('BINDDN') == LEGACY_BINDDN): raise Refused() + switching = True + else: + switching = False for key, default in TUNING.items(): value = data.get(key) if value in (None, ''): updates[key] = default elif type(value) not in (str, int) or not str(value).isascii() or not str(value).isdigit(): raise Refused() + if updates and switching: raise Refused() # one change at a time: tune first, then switch if updates: updates['type'] = 'ldapresolver' realm = realms.get(REALM) if realm is not None: @@ -89,25 +97,35 @@ def plan(snapshot): if (not isinstance(bindings, list) or not any(isinstance(v, dict) and v.get('name') == RESOLVER for v in bindings)): raise Refused() return {'resolver_update': updates, 'create_resolver': resolver is None, + 'switch_bind': resolver is not None and switching, 'create_realm': realm is None, 'set_initial_default': not realms} -def repair(client, bind_input, *, apply=False): +def repair(client, bind_input, *, apply=False, switch_bind=False): initial = client.snapshot() - changes = plan(initial) + changes = plan(initial, switch_bind) actions = [] + if changes['switch_bind']: actions.append('switch-bind') if changes['resolver_update']: actions.append('create-resolver' if changes['create_resolver'] else 'fill-missing-tuning') if changes['create_realm']: actions.append('create-realm') if changes['set_initial_default']: actions.append('set-first-realm-default') if not apply: return {'result': 'planned', 'actions': actions, 'configuration_writes': 0, 'policies_changed': False, 'mfa_verified': False} - bind = password(bind_input()) if changes['create_resolver'] else None + bind = password(bind_input()) if changes['create_resolver'] or changes['switch_bind'] else None # Recheck the full observed target before each operation. API has no CAS; # an exclusive owner window remains required, and ambiguous writes are not retried. expected = initial for action in actions: if client.snapshot() != expected: raise Refused() - if action in ('create-resolver', 'fill-missing-tuning'): + if action == 'switch-bind': + value = client.call('/resolver/' + RESOLVER, {'type': 'ldapresolver', 'BINDDN': CORE['BINDDN'], 'BINDPW': bind}) + if type(value) is not int or value < 1: raise Refused() + after = client.snapshot() + old, data = expected['resolver']['data'], after['resolver'].get('data', {}) + if data.get('BINDDN') != CORE['BINDDN']: raise Refused() + if any(data.get(k) != v for k, v in old.items() if k not in ('BINDDN', 'BINDPW')): raise Refused() + if after['realms'] != expected['realms']: raise Refused() + elif action in ('create-resolver', 'fill-missing-tuning'): payload = dict(changes['resolver_update']) if action == 'create-resolver': payload['BINDPW'] = bind value = client.call('/resolver/' + RESOLVER, payload) @@ -158,20 +176,21 @@ def main(): parser.add_argument('--attended', action='store_true') parser.add_argument('--apply', action='store_true') parser.add_argument('--verify-user') + parser.add_argument('--switch-bind', action='store_true') args = parser.parse_args() base = endpoint(args.pi_url or args.url) if args.verify_user and not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9._-]{0,127}', args.verify_user): raise Refused() if args.attended: if args.root is not None or not sys.stdin.isatty(): raise Refused() admin = password(getpass.getpass('privacyIDEA pi-admin password: ')) - bind_input = lambda: getpass.getpass('LLDAP bind password for missing resolver: ') + bind_input = lambda: getpass.getpass('LLDAP bind password for the resolver (svc-privacyidea): ') else: if args.root is None or args.verify_user: raise Refused() admin, bind = private_inputs(args.root) bind_input = lambda: bind phase = 'configuration' client = Client(base, admin) - report = repair(client, bind_input, apply=args.apply) + report = repair(client, bind_input, apply=args.apply, switch_bind=args.switch_bind) if args.verify_user and args.apply: phase = 'functional-verification' from verify_mfa import verify diff --git a/tests/test_realm_bootstrap.py b/tests/test_realm_bootstrap.py index 37876bc..bba3f91 100644 --- a/tests/test_realm_bootstrap.py +++ b/tests/test_realm_bootstrap.py @@ -205,3 +205,29 @@ def test_default_readback_refuses_unexpected_other_realm(): client=Concurrent({'resolver':None,'realms':{}}) with pytest.raises(r.Refused):r.repair(client,lambda:'fixture-bind',apply=True) assert len(client.calls)==3 + + +def legacy(): + state=snapshot();state['resolver']['data']['BINDDN']=r.LEGACY_BINDDN;return state + + +def test_legacy_human_bind_is_refused_without_explicit_switch(): + with pytest.raises(r.Refused):r.repair(Fake(legacy()),lambda:'unused',apply=True) + + +def test_switch_bind_changes_only_the_bind_and_needs_the_service_password(): + client=Fake(legacy());before=copy.deepcopy(client.state) + preview=r.repair(Fake(legacy()),lambda:(_ for _ in ()).throw(AssertionError('no prompt in preview')),switch_bind=True) + assert preview['actions']==['switch-bind'] and preview['configuration_writes']==0 + report=r.repair(client,lambda:'svc-privacyidea-synthetic',apply=True,switch_bind=True) + assert report['actions']==['switch-bind'] and report['configuration_writes']==1 + assert client.calls==[('/resolver/'+r.RESOLVER,{'type':'ldapresolver','BINDDN':r.CORE['BINDDN'],'BINDPW':'svc-privacyidea-synthetic'})] + before['resolver']['data']['BINDDN']=r.CORE['BINDDN'];assert client.state==before + assert r.repair(client,lambda:'unused',apply=True,switch_bind=True)['configuration_writes']==0 + + +def test_switch_bind_refuses_other_drift_or_unknown_bind(): + state=legacy();del state['resolver']['data']['TIMEOUT'] + with pytest.raises(r.Refused):r.repair(Fake(state),lambda:'x',apply=True,switch_bind=True) + state=snapshot();state['resolver']['data']['BINDDN']='uid=someone,ou=people,dc=netkingdom,dc=local' + with pytest.raises(r.Refused):r.repair(Fake(state),lambda:'x',apply=True,switch_bind=True)