tegwick
e556af8571
Only from the legacy uid=admin bind, only BINDDN/BINDPW, nothing else drifted. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 308701@bnt-lap001 Assistant-Session: 5548e447-444c-4dcc-bed1-05c16eebfe56
209 lines
11 KiB
Python
209 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Repair missing realm/resolver configuration while preserving established policy."""
|
|
import argparse
|
|
import getpass
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
import sys
|
|
|
|
from mfa_state import QuietParser, endpoint
|
|
from pi_api import request
|
|
|
|
RESOLVER = 'lldap-coulomb'
|
|
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):
|
|
pass
|
|
|
|
|
|
def password(value):
|
|
if not isinstance(value, str) or not value or len(value.encode()) > 4096 or any(ord(c) < 32 for c in value):
|
|
raise Refused()
|
|
return value
|
|
|
|
|
|
def private_inputs(root):
|
|
path = Path(__file__).resolve().parents[2] / 'bootstrap/apply-service-secrets.py'
|
|
spec = importlib.util.spec_from_file_location('realm_custody', path)
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
|
with module.custody(root, ['privacyidea', 'lldap']) as dirs:
|
|
admin = password(module.read_env(dirs['privacyidea']).get('PI_ADMIN_PASSWORD'))
|
|
bind = password(module.read_env(dirs['lldap']).get('LLDAP_SVC_PRIVACYIDEA_PASS'))
|
|
return admin, bind
|
|
|
|
|
|
class Client:
|
|
def __init__(self, url, admin_password):
|
|
self.url = endpoint(url)
|
|
self.token = None
|
|
auth = self.call('/auth', {'username': 'pi-admin', 'password': password(admin_password)})
|
|
if not isinstance(auth, dict) or auth.get('role') != 'admin' or not isinstance(auth.get('token'), str) or not auth['token']:
|
|
raise Refused()
|
|
self.token = auth['token']
|
|
|
|
def call(self, path, payload=None):
|
|
status, body = request(self.url + path, payload, self.token)
|
|
result = body.get('result') if isinstance(body, dict) else None
|
|
if status != 200 or not isinstance(result, dict) or result.get('status') is not True:
|
|
raise Refused()
|
|
return result.get('value')
|
|
|
|
def snapshot(self):
|
|
resolvers = self.call('/resolver/' + RESOLVER)
|
|
realms = self.call('/realm/')
|
|
if not isinstance(resolvers, dict) or set(resolvers) - {RESOLVER} or not isinstance(realms, dict):
|
|
raise Refused()
|
|
if any(not isinstance(v, dict) for v in realms.values()): raise Refused()
|
|
return {'resolver': resolvers.get(RESOLVER), 'realms': realms}
|
|
|
|
|
|
def plan(snapshot, switch_bind=False):
|
|
resolver, realms = snapshot['resolver'], snapshot['realms']
|
|
updates = {}
|
|
if resolver is None:
|
|
updates = dict(CORE, **TUNING, type='ldapresolver', LOGINNAMEATTRIBUTE='uid',
|
|
LDAPSEARCHFILTER='(objectClass=inetOrgPerson)', LDAPFILTER='(&(objectClass=inetOrgPerson)(uid=%s))',
|
|
USERINFO=json.dumps({'username':'uid','phone':'telephoneNumber','mobile':'mobile','email':'mail',
|
|
'surname':'sn','givenname':'givenName'}),
|
|
UIDTYPE='uid', NOREFERRALS=True, NOSCHEMAS=True)
|
|
else:
|
|
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() 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:
|
|
bindings = realm.get('resolver')
|
|
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, switch_bind=False):
|
|
initial = client.snapshot()
|
|
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'] 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 == '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)
|
|
if type(value) is not int or value < 1: raise Refused()
|
|
after = client.snapshot()
|
|
current = after['resolver']
|
|
if not isinstance(current, dict) or current.get('type') != 'ldapresolver': raise Refused()
|
|
data = current.get('data', {})
|
|
for key, desired in payload.items():
|
|
if key in ('type', 'BINDPW'): continue
|
|
actual = data.get(key)
|
|
if isinstance(desired, bool):
|
|
if str(actual).lower() != str(desired).lower(): raise Refused()
|
|
elif str(actual) != str(desired): raise Refused()
|
|
if action == 'fill-missing-tuning':
|
|
old = expected['resolver']['data']
|
|
if any(data.get(k) != v for k, v in old.items() if k not in payload): raise Refused()
|
|
if after['realms'] != expected['realms']: raise Refused()
|
|
elif action == 'create-realm':
|
|
value = client.call('/realm/' + REALM, {'resolvers': [RESOLVER], 'priority.'+RESOLVER: 1})
|
|
if not isinstance(value, dict) or value.get('failed') != [] or value.get('added') != [RESOLVER]: raise Refused()
|
|
after = client.snapshot()
|
|
realm = after['realms'].get(REALM)
|
|
if not isinstance(realm, dict) or [r.get('name') for r in realm.get('resolver', [])] != [RESOLVER]: raise Refused()
|
|
if {k:v for k,v in after['realms'].items() if k != REALM} != expected['realms']: raise Refused()
|
|
if after['resolver'] != expected['resolver']: raise Refused()
|
|
else:
|
|
value = client.call('/defaultrealm/' + REALM, {})
|
|
if type(value) is not int or value < 1: raise Refused()
|
|
after = client.snapshot()
|
|
anticipated = json.loads(json.dumps(expected))
|
|
anticipated['realms'][REALM]['default'] = True
|
|
if after != anticipated: raise Refused()
|
|
expected = after
|
|
final = client.snapshot()
|
|
if final != expected or plan(final)['resolver_update'] or plan(final)['create_realm']: raise Refused()
|
|
return {'result': 'configured', 'actions': actions, 'configuration_writes': len(actions),
|
|
'policies_changed': False, 'mfa_verified': False, 'onboarding_policy': 'owner-managed'}
|
|
|
|
|
|
def main():
|
|
phase = 'input'
|
|
try:
|
|
parser = QuietParser(description=__doc__)
|
|
parser.add_argument('root', nargs='?')
|
|
parser.add_argument('url', nargs='?', default=os.environ.get('PI_URL', 'https://pink.coulomb.social'))
|
|
parser.add_argument('--pi-url')
|
|
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 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, switch_bind=args.switch_bind)
|
|
if args.verify_user and args.apply:
|
|
phase = 'functional-verification'
|
|
from verify_mfa import verify
|
|
proof = verify(base, args.verify_user, REALM, RESOLVER, admin,
|
|
lambda:getpass.getpass('Fresh MFA code (include token PIN if required): '))
|
|
if proof['result'] != 'PASS': raise Refused()
|
|
report['mfa_verified'] = True
|
|
print(json.dumps(report, sort_keys=True))
|
|
return 0
|
|
except (Exception, KeyboardInterrupt):
|
|
print(json.dumps({'result':'unverified','phase':phase,'partial_changes_possible':phase!='input'}), file=sys.stderr)
|
|
return 1
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|