Connect P04 audited recovery to fresh-MFA platform browser flow

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c

tests/test_factor_recovery_journey.py

@@ -0,0 +1,52 @@
+import re
+import time
+from html import unescape
+from dataclasses import replace
+from test_journey_roles import JourneyFixture
+from test_web import invoke
+
+class FakeRecovery:
+    def __init__(self):self.calls=[];self.applied=False;self.outage=False
+    def call(self,token,body):
+        self.calls.append((token,body))
+        if self.outage:raise RuntimeError('private-secret')
+        if body['action']=='preview':
+            return dict(success=True,status='preview',factors=[dict(user='alice',serial='T1',reference='case-1',confirmation='signed-preview')])
+        if not body['identity_verified']:return dict(success=False,failure='identity_verification_required')
+        replay=self.applied;self.applied=True
+        return dict(success=True,status='recovered',user='alice',serial='T1',reference='case-1',replayed=replay)
+
+class FactorRecoveryJourney(JourneyFixture):
+    def setUp(self):
+        super().setUp();self.provider=FakeRecovery();self.app.factor_recovery=self.provider
+        s=self.oidc.sessions['operator']
+        self.oidc.sessions['operator']=replace(s,id_token='server-only-token',claims=dict(s.claims,assurance=dict(level='aal2',mfa=True,at=time.time())))
+    def test_role_csrf_and_freshness_denied_before_provider(self):
+        for who in ['member','admin']:
+            response,_=self.post('/platform/factor-recovery',who=who,action='preview')
+            self.assertEqual('403 Forbidden',response['status'])
+        response,_=self.post('/platform/factor-recovery',who='operator',csrf_token='bad',action='preview')
+        self.assertEqual('403 Forbidden',response['status'])
+        self.oidc.sessions['operator'].claims['assurance']['at']=time.time()-301
+        response,body=self.post('/platform/factor-recovery',who='operator',action='preview')
+        self.assertIn(b'Verify my sign-in again',body);self.assertEqual([],self.provider.calls)
+    def test_preview_cancel_apply_retry_and_secret_not_rendered(self):
+        response,body=self.post('/platform/factor-recovery',who='operator',action='preview',user='alice',reference='case-1')
+        self.assertIn(b'all applications',body);self.assertIn(b'Cancel',body);self.assertFalse(self.provider.applied)
+        self.assertNotIn(b'server-only-token',body)
+        _,body=self.post('/platform/factor-recovery',who='operator',action='apply',confirmation='signed-preview')
+        self.assertIn(b'Confirm that you verified',body);self.assertFalse(self.provider.applied)
+        self.provider.outage=True
+        _,body=self.post('/platform/factor-recovery',who='operator',action='apply',confirmation='signed-preview',identity_verified='yes')
+        self.assertIn(b'Retry this recovery',body);self.assertNotIn(b'private-secret',body)
+        self.provider.outage=False
+        for i in range(2):
+            _,body=self.post('/platform/factor-recovery',who='operator',action='apply',confirmation='signed-preview',identity_verified='yes')
+            self.assertIn(b'Authenticator recovery recorded',body)
+            self.assertIn(b'Enroll a replacement',body)
+        self.assertIn(b'without repeating it',body)
+    def test_stepup_requests_fresh_mfa_and_binds_return(self):
+        response,_=invoke(self.app,'/login',query='recovery=1')
+        location=dict(response['headers'])['Location']
+        self.assertIn('prompt=login',location);self.assertIn('max_age=0',location);self.assertIn('acr_values=aal2',location)
+        self.assertTrue(next(iter(self.oidc.pending.values())).recovery)