Implement audited lost-factor recovery and track remaining P04 acceptance

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

scripts/test_factor_recovery.py

@@ -0,0 +1,47 @@
+import copy,unittest
+from factor_recovery import recover,fingerprint,RecoveryError
+class Store:
+ realm='coulomb'
+ def __init__(self):
+  self.state={'id':1,'serial':'T1','user_id':'u1','resolver':'directory','realms':['coulomb'],'active':True,'rollout_state':'enrolled','tokentype':'totp'};self.rows={};self.writes=0;self.fail_audit=False;self.fail_complete=False;self.race=False
+ def snapshot(self,user,serial):return copy.deepcopy(self.state) if user=='alice' and serial=='T1' else None
+ def receipt(self,ref):return self.rows.get(ref)
+ def record(self,request,version,complete):
+  if self.fail_audit or self.fail_complete and complete:raise RecoveryError('audit_unavailable')
+  self.rows[request['reference']]=dict(request,version=version,complete=complete)
+ def disable(self,user,serial,expected):
+  if self.race:self.state['id']=2
+  if fingerprint(self.state)!=expected:raise RecoveryError('stale_preview')
+  self.state['active']=False;self.writes+=1
+class RecoveryTests(unittest.TestCase):
+ def setUp(self):self.store=Store();self.request=dict(user='alice',serial='T1',realm='coulomb',actor='operator-entity',reference='support-1')
+ def approval(self):return dict(self.request,apply=True,identity_verified=True,expected_version=recover(self.store,self.request)['version'])
+ def test_preview_never_mutates_and_names_global_scope(self):
+  p=recover(self.store,self.request);self.assertFalse(p['changes_applied']);self.assertEqual(p['scope'],'shared_identity_across_applications');self.assertEqual(self.store.writes,0)
+ def test_verified_recovery_and_replay_disable_exactly_once(self):
+  r=self.approval();self.assertTrue(recover(self.store,r)['changes_applied']);self.assertTrue(recover(self.store,r)['replayed']);self.assertEqual(self.store.writes,1)
+ def test_verification_and_ownership_denied_before_mutation(self):
+  for r in [dict(self.approval(),identity_verified=False),dict(self.request,user='bob'),dict(self.request,realm='other')]:
+   with self.assertRaises(RecoveryError):recover(self.store,r)
+  self.assertEqual(self.store.writes,0)
+ def test_stale_confirmation_and_concurrent_replacement_rejected(self):
+  r=self.approval();self.store.state['id']=2
+  with self.assertRaises(RecoveryError):recover(self.store,r)
+  self.store.state['id']=1;self.store.race=True
+  with self.assertRaises(RecoveryError):recover(self.store,r)
+  self.assertEqual(self.store.writes,0)
+ def test_audit_failure_prevents_mutation(self):
+  r=self.approval();self.store.fail_audit=True
+  with self.assertRaises(RecoveryError):recover(self.store,r)
+  self.assertEqual(self.store.writes,0)
+ def test_completion_audit_retry_does_not_repeat_mutation(self):
+  r=self.approval();self.store.fail_complete=True
+  with self.assertRaises(RecoveryError):recover(self.store,r)
+  self.store.fail_complete=False;self.assertTrue(recover(self.store,r)['replayed']);self.assertEqual(self.store.writes,1)
+ def test_conflicting_reference_rejected(self):
+  r=self.approval();recover(self.store,r)
+  with self.assertRaises(RecoveryError):recover(self.store,dict(r,actor='different-operator'))
+
+ def test_replaced_inactive_factor_cannot_replay_old_recovery(self):
+  r=self.approval();recover(self.store,r);self.store.state['id']=2
+  with self.assertRaises(RecoveryError):recover(self.store,r)