Implement scoped P06 authentication policy and guarded optional onboarding

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6

sso-mfa/k8s/keycape/authentication-policy-rollout.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Migrate only the two reviewed baseline registrations; never export Secret values.
+
+Runtime policy overrides are issuer-owned in its policy PVC. This migration is
+for initial activation; subsequent changes use the authenticated policy journey.
+"""
+import argparse
+import base64
+import copy
+import importlib.util
+import json
+from pathlib import Path
+import subprocess
+import yaml
+
+ROOT = Path(__file__).resolve().parent
+spec = importlib.util.spec_from_file_location('config_pin', ROOT/'openbao-client-config.py')
+pin = importlib.util.module_from_spec(spec); spec.loader.exec_module(pin)
+CLIENTS = {'vergabe-demo-company', 'user-engine-portal'}
+CLUSTER_UID = 'a553c742-0115-43d4-99a4-a5ca56fe0786'
+
+class Refused(Exception): pass
+
+def require(condition, reason):
+    if not condition: raise Refused(reason)
+
+def replacement(secret):
+    try: raw,config,_ = pin.issuer_document(secret)
+    except Exception: raise Refused("issuer_configuration_unreadable") from None
+    clients = config.get('clients',[])
+    ids = [c.get('clientId') for c in clients]
+    require(len(ids)==len(set(ids)) and CLIENTS.issubset(ids), 'reviewed_clients_required')
+    document = yaml.compose(raw)
+    nodes = next(v for k,v in document.value if k.value=='clients')
+    edits=[]; expected=copy.deepcopy(config)
+    for index,client in enumerate(clients):
+        if client['clientId'] not in CLIENTS: continue
+        require(client.get('clientType')=='public' and client.get('grantTypes')==['authorization_code'], 'client_shape_changed')
+        require('mfaRequired' not in client, 'explicit_policy_requires_review')
+        if client.get('mfaOptional') is True: continue
+        require('mfaOptional' not in client, 'explicit_policy_requires_review')
+        node=nodes.value[index]
+        require(not node.flow_style, 'block_client_required')
+        column=node.value[0][0].start_mark.column
+        position=node.end_mark.index
+        line_start=raw.rfind('\n',0,position)+1
+        if not raw[line_start:position].strip(): position=line_start
+        addition=('' if position==0 or raw[position-1]=='\n' else '\n')+' '*column+'mfaOptional: true\n'
+        edits.append((position,addition));expected['clients'][index]['mfaOptional']=True
+    updated=raw
+    for position,addition in sorted(edits,reverse=True):updated=updated[:position]+addition+updated[position:]
+    require(yaml.load(updated,Loader=pin.UniqueLoader)==expected, 'unrelated_configuration_changed')
+    return base64.b64encode(updated.encode()).decode(),bool(edits)
+
+def kube(args,payload=None):
+    result=subprocess.run(['kubectl','--request-timeout=20s',*args],input=json.dumps(payload) if payload is not None else None,capture_output=True,text=True,timeout=30)
+    require(result.returncode==0,'kubernetes_operation_failed')
+    return json.loads(result.stdout)
+
+def run(args):
+    require(kube(['get','namespace','kube-system','-o','json'])['metadata']['uid']==CLUSTER_UID,'cluster_changed')
+    secret=kube(['-n','sso','get','secret','keycape-config','-o','json'])
+    require(not secret['metadata'].get('ownerReferences'),'controller_owned_configuration')
+    value,changed=replacement(secret);metadata=pin.safe_metadata(secret)
+    receipt={'clients':sorted(CLIENTS),'before':metadata,'change_needed':changed,'other_clients_unchanged':True,'secret_values_emitted':False,'mode':args.mode}
+    if args.mode=='inspect':return receipt
+    require(metadata=={'uid':args.expected_uid,'resource_version':args.expected_resource_version},'observed_revision_changed')
+    if changed:
+        patch=[{'op':'test','path':'/metadata/uid','value':metadata['uid']},{'op':'test','path':'/metadata/resourceVersion','value':metadata['resource_version']},{'op':'replace','path':'/data/config.yaml','value':value}]
+        command=['-n','sso','patch','secret','keycape-config','--type=json','--patch-file=/dev/stdin','-o','json']
+        if args.mode=='dry-run':command.append('--dry-run=server')
+        response=kube(command,patch)
+        require(response['data']==dict(secret['data'],**{'config.yaml':value}),'patch_readback_mismatch')
+    if args.mode=='apply':
+        response=kube(['-n','sso','get','secret','keycape-config','-o','json'])
+        require(response['metadata']['uid']==metadata['uid'] and response['data']==dict(secret['data'],**{'config.yaml':value}),'readback_changed')
+        receipt['after']=pin.safe_metadata(response)
+    return receipt
+
+if __name__=='__main__':
+    parser=argparse.ArgumentParser();parser.add_argument('mode',choices=['inspect','dry-run','apply']);parser.add_argument('--expected-uid');parser.add_argument('--expected-resource-version')
+    try: print(json.dumps(run(parser.parse_args())))
+    except Refused as error:print(json.dumps({'success':False,'failure':str(error)}));raise SystemExit(1)

sso-mfa/k8s/keycape/policy-pvc.yaml

@@ -0,0 +1,11 @@
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: keycape-authentication-policy
+  namespace: sso
+spec:
+  accessModes: [ReadWriteOnce]
+  storageClassName: local-path
+  resources:
+    requests:
+      storage: 64Mi

sso-mfa/k8s/keycape/test_authentication_policy_rollout.py

@@ -0,0 +1,35 @@
+import base64,importlib.util,json,unittest
+from pathlib import Path
+import yaml
+spec=importlib.util.spec_from_file_location('rollout',Path(__file__).with_name('authentication-policy-rollout.py'))
+module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+class PolicyMigration(unittest.TestCase):
+    def secret(self):
+        raw='''# exact bytes outside reviewed fields must survive
+issuer: https://fixture.test
+authelia:
+  issuer: https://primary.test
+clients:
+  - clientId: user-engine-portal
+    clientType: public
+    grantTypes: [authorization_code]
+  - clientId: untouched
+    mfaRequired: true
+    clientSecret: fixture-secret
+  - clientId: vergabe-demo-company
+    clientType: public
+    grantTypes: [authorization_code]
+privacyidea:
+  requireForAll: true
+'''
+        return {'data':{'config.yaml':base64.b64encode(raw.encode()).decode(),'key.pem':'fixture-key'}}
+    def test_exact_migration_preserves_siblings_and_retries(self):
+        secret=self.secret();value,changed=module.replacement(secret);self.assertTrue(changed)
+        raw=base64.b64decode(value).decode();self.assertIn('# exact bytes',raw);self.assertIn('    clientSecret: fixture-secret\n',raw)
+        config=yaml.safe_load(raw);self.assertTrue(config['privacyidea']['requireForAll'])
+        self.assertTrue(config['clients'][0]['mfaOptional']);self.assertTrue(config['clients'][2]['mfaOptional'])
+        secret['data']['config.yaml']=value;again,changed=module.replacement(secret);self.assertFalse(changed);self.assertEqual(value,again)
+    def test_explicit_or_duplicate_registration_refused(self):
+        for edit in [lambda s:s.replace('    clientType: public','    mfaRequired: false\n    clientType: public',1),lambda s:s.replace('clientId: untouched','clientId: user-engine-portal')]:
+            secret=self.secret();secret['data']['config.yaml']=base64.b64encode(edit(base64.b64decode(secret['data']['config.yaml']).decode()).encode()).decode()
+            with self.assertRaises(module.Refused):module.replacement(secret)

sso-mfa/k8s/privacyidea/activate-onboarding-policy.py

@@ -0,0 +1,25 @@
+"""Provider-local activation after the guarded deployment; never changes tokens."""
+import contextlib,io,json,logging
+result={'success':False}
+with contextlib.redirect_stdout(io.StringIO()),contextlib.redirect_stderr(io.StringIO()):
+    try:
+        logging.disable(logging.CRITICAL)
+        from privacyidea.app import create_app
+        from privacyidea.lib.policy import PolicyClass,set_policy
+        app=create_app(config_name='production',silent=True)
+        with app.app_context():
+            if app.config.get('PI_INIT_CHECK_HOOK')!='keycape_onboarding_guard.check':raise ValueError('hook_not_active')
+            from keycape_onboarding_guard import check
+            rows=[r for r in PolicyClass().policies if r['scope']=='user' and r['active']]
+            expected=next(r for r in rows if r['name']=='totp-self-enrollment')
+            if expected['realm']!=['coulomb'] or expected.get('conditions') or expected['action'] not in [{'enrollTOTP':True,'delete':True,'disable':True},{'enrollTOTP':True}]:raise ValueError('self_service_policy_changed')
+            if any(r['name'] not in {'totp-self-enrollment','keycape-pending-enrollment-cancel'} and (not r['realm'] or 'coulomb' in r['realm']) for r in rows):raise ValueError('additional_user_policy_requires_review')
+            set_policy(name='totp-self-enrollment',scope='user',action='enrollTOTP',realm='coulomb')
+            set_policy(name='keycape-pending-enrollment-cancel',scope='user',action='delete',realm='coulomb',conditions=[('token','rollout_state','equals','verify',True)])
+            fresh=PolicyClass().policies
+            active=next(r for r in fresh if r['name']=='totp-self-enrollment')
+            pending=next(r for r in fresh if r['name']=='keycape-pending-enrollment-cancel')
+            if active['action']!={'enrollTOTP':True} or not pending['conditions']:raise ValueError('readback_failed')
+            result={'success':True,'hook':'keycape_onboarding_guard.check','self_service_enrollment':True,'pending_cancel_only':True,'active_factor_changes_require_recovery':True,'existing_tokens_changed':False}
+    except Exception as error:result={'success':False,'failure_type':type(error).__name__}
+print(json.dumps(result));raise SystemExit(0 if result['success'] else 1)

sso-mfa/k8s/privacyidea/deploy-factor-recovery.py

@@ -68,6 +68,7 @@ def main():
         'resources':{'requests':{'cpu':'10m','memory':'128Mi'},'limits':{'cpu':'500m','memory':'384Mi'}},
         'readinessProbe':{'httpGet':{'path':'/healthz','port':8091},'initialDelaySeconds':10,'periodSeconds':10},
         'livenessProbe':{'httpGet':{'path':'/healthz','port':8091},'initialDelaySeconds':60,'periodSeconds':20}}
+    sidecar['env'] += [entry for entry in main.get('env',[]) if entry['name']=='PYTHONPATH']
     containers=[c for c in spec['containers'] if c['name']!='factor-recovery']+[sidecar]
     volumes=[v for v in spec['volumes'] if v['name']!='factor-recovery-code']+[{'name':'factor-recovery-code','configMap':{'name':name}}]
     patch=[{'op':'test','path':'/metadata/resourceVersion','value':provider['metadata']['resourceVersion']},

sso-mfa/k8s/privacyidea/deploy-onboarding-guard.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Deploy the reviewed provider enrollment hook while preserving runtime secrets."""
+import argparse
+import copy
+import hashlib
+import json
+from pathlib import Path
+import subprocess
+
+CLUSTER='a553c742-0115-43d4-99a4-a5ca56fe0786'
+PROVIDER='58c7f96d-61cb-4dd4-bca2-54661c0ac375'
+CONFIG='423d857b-6b10-40ad-aba8-2cda2d3645ef'
+SUFFIX='\n# P06: fail startup if the enrollment guard is unavailable.\nfrom keycape_onboarding_guard import check as _p06_onboarding_check\nPI_INIT_CHECK_HOOK = "keycape_onboarding_guard.check"\n'
+
+def kube(args,payload=None):
+    result=subprocess.run(['kubectl','--request-timeout=20s',*args],input=json.dumps(payload) if payload is not None else None,capture_output=True,text=True,timeout=30)
+    if result.returncode:raise RuntimeError('kubernetes_operation_failed')
+    return json.loads(result.stdout)
+
+def plan(source,config,deployment):
+    code=(source/'scripts/keycape_onboarding_guard.py').read_text()
+    digest=hashlib.sha256(code.encode()).hexdigest();name='keycape-onboarding-'+digest[:16]
+    cm={'apiVersion':'v1','kind':'ConfigMap','metadata':{'name':name,'namespace':'mfa'},'immutable':True,'data':{'keycape_onboarding_guard.py':code}}
+    original=config['data']['pi.cfg']
+    if 'PI_INIT_CHECK_HOOK' in original and not original.endswith(SUFFIX):raise RuntimeError('existing_hook_requires_review')
+    updated=original if original.endswith(SUFFIX) else original+SUFFIX
+    spec=copy.deepcopy(deployment['spec']['template']['spec'])
+    for container in spec['containers']:
+        if container['name'] not in {'privacyidea','factor-recovery'}:continue
+        env=container.setdefault('env',[])
+        path=next((e for e in env if e['name']=='PYTHONPATH'),None)
+        if path and path.get('value') not in {'/opt/keycape-onboarding'}:raise RuntimeError('existing_python_path_requires_review')
+        if not path:env.append({'name':'PYTHONPATH','value':'/opt/keycape-onboarding'})
+        mounts=container.setdefault('volumeMounts',[])
+        existing=next((m for m in mounts if m['name']=='onboarding-guard'),None)
+        desired={'name':'onboarding-guard','mountPath':'/opt/keycape-onboarding','readOnly':True}
+        if existing and existing!=desired:raise RuntimeError('existing_guard_mount_differs')
+        if not existing:mounts.append(desired)
+    spec['volumes']=[v for v in spec['volumes'] if v['name']!='onboarding-guard']+[{'name':'onboarding-guard','configMap':{'name':name}}]
+    return cm,updated,spec
+
+def main():
+    parser=argparse.ArgumentParser();parser.add_argument('--source',type=Path,required=True);parser.add_argument('--apply',action='store_true');args=parser.parse_args()
+    if kube(['get','namespace','kube-system','-o','json'])['metadata']['uid']!=CLUSTER:raise RuntimeError('cluster_changed')
+    config=kube(['-n','mfa','get','configmap','privacyidea-cfg','-o','json']);deployment=kube(['-n','mfa','get','deployment','privacyidea','-o','json'])
+    if config['metadata']['uid']!=CONFIG or deployment['metadata']['uid']!=PROVIDER:raise RuntimeError('target_identity_changed')
+    cm,updated,spec=plan(args.source,config,deployment)
+    print(json.dumps({'apply':args.apply,'guard_config':cm['metadata']['name'],'target':'mfa/privacyidea','config_uid':CONFIG,'provider_uid':PROVIDER,'secret_values_emitted':False}),flush=True)
+    if not args.apply:return
+    committed=subprocess.run(['git','show','HEAD:scripts/keycape_onboarding_guard.py'],cwd=args.source,capture_output=True,text=True)
+    if committed.returncode or committed.stdout!=cm['data']['keycape_onboarding_guard.py']:raise RuntimeError('guard_source_must_be_committed')
+    kube(['apply','-f','-','-o','json'],cm)
+    if updated!=config['data']['pi.cfg']:
+        kube(['-n','mfa','patch','configmap','privacyidea-cfg','--type=json','--patch-file=/dev/stdin','-o','json'],[
+            {'op':'test','path':'/metadata/uid','value':CONFIG},{'op':'test','path':'/metadata/resourceVersion','value':config['metadata']['resourceVersion']},
+            {'op':'replace','path':'/data/pi.cfg','value':updated}])
+    kube(['-n','mfa','patch','deployment','privacyidea','--type=json','--patch-file=/dev/stdin','-o','json'],[
+        {'op':'test','path':'/metadata/uid','value':PROVIDER},{'op':'test','path':'/metadata/resourceVersion','value':deployment['metadata']['resourceVersion']},
+        {'op':'replace','path':'/spec/template/spec','value':spec}])
+    after=kube(['-n','mfa','get','configmap','privacyidea-cfg','-o','json'])
+    if after['data']!=dict(config['data'],**{'pi.cfg':updated}):raise RuntimeError('configuration_readback_changed')
+    print(json.dumps({'guard_deployed':True,'wait_for_rollout_and_policy_activation':True}))
+
+if __name__=='__main__':main()

sso-mfa/k8s/privacyidea/test_onboarding_guard_deploy.py

@@ -0,0 +1,39 @@
+import copy
+import importlib.util
+from pathlib import Path
+import tempfile
+import unittest
+
+spec = importlib.util.spec_from_file_location("guard_deploy", Path(__file__).with_name("deploy-onboarding-guard.py"))
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+class GuardDeploymentTest(unittest.TestCase):
+    def test_preserves_custody_and_recovery_and_replays(self):
+        config = {"data": {"pi.cfg": "EXISTING=True\n", "other": "unchanged"}}
+        deployment = {"spec": {"template": {"spec": {
+            "containers": [{"name": name, "image": "pinned", "envFrom": [{"secretRef": {"name": "existing"}}], "volumeMounts": [{"name": "data", "mountPath": "/etc/privacyidea"}]} for name in ["privacyidea", "factor-recovery", "unrelated"]],
+            "volumes": [{"name": "data", "persistentVolumeClaim": {"claimName": "existing"}}],
+        }}}}
+        original = copy.deepcopy(deployment)
+        with tempfile.TemporaryDirectory() as directory:
+            source = Path(directory); (source / "scripts").mkdir()
+            (source / "scripts/keycape_onboarding_guard.py").write_text("def check(request, action): return True\n")
+            cm, updated, planned = module.plan(source, config, deployment)
+            self.assertEqual(deployment, original)
+            self.assertEqual(updated, config["data"]["pi.cfg"] + module.SUFFIX)
+            self.assertTrue(cm["immutable"])
+            for before, after in zip(original["spec"]["template"]["spec"]["containers"], planned["containers"]):
+                if before["name"] == "unrelated":
+                    self.assertEqual(before, after)
+                else:
+                    self.assertEqual(before["envFrom"], after["envFrom"])
+                    self.assertEqual(before["image"], after["image"])
+                    self.assertEqual(after["env"], [{"name": "PYTHONPATH", "value": "/opt/keycape-onboarding"}])
+                    self.assertIn(before["volumeMounts"][0], after["volumeMounts"])
+            replay = module.plan(source, {"data": {"pi.cfg": updated}}, {"spec": {"template": {"spec": planned}}})
+            self.assertEqual(replay, (cm, updated, planned))
+            with self.assertRaisesRegex(RuntimeError, "existing_hook_requires_review"):
+                module.plan(source, {"data": {"pi.cfg": "PI_INIT_CHECK_HOOK='another.check'"}}, deployment)
+
+if __name__ == "__main__": unittest.main()