import io import json import unittest from contextlib import redirect_stderr from recovery_service import RecoveryService, authorized_actor from factor_recovery import RecoveryError from test_factor_recovery import Store class RecoveryServiceTests(unittest.TestCase): def setUp(self): self.store=Store() self.store.serials=lambda user:['T1'] if user=='alice' else [] self.claims=dict(sub='operator',principal_type='human',roles=['platform-operator'], assurance=dict(level='aal2',mfa=True,at=1000)) self.service=RecoveryService(self.store,lambda token:self.claims,b'fixture-only',now=lambda:1000) def preview(self): return self.service.operation('signed',dict(action='preview',user='alice',reference='case-1'))['factors'][0] def test_preview_apply_and_replay(self): p=self.preview();self.assertEqual(0,self.store.writes) body=dict(action='apply',confirmation=p['confirmation'],identity_verified=True) self.assertTrue(self.service.operation('signed',body)['changes_applied']) self.assertTrue(self.service.operation('signed',body)['replayed']) self.assertEqual(1,self.store.writes) def test_denies_roles_stale_assurance_and_tampered_ticket(self): ticket=self.preview()['confirmation'] for claims in [dict(self.claims,roles=['tenant-admin']),dict(self.claims,principal_type='service'), dict(self.claims,assurance=dict(level='aal2',mfa=True,at=699)),dict(self.claims,assurance={})]: with self.assertRaises(RecoveryError):authorized_actor(claims,1000) for value in [ticket+'x','garbage']: with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=value,identity_verified=True)) self.assertEqual(0,self.store.writes) def test_ticket_cannot_switch_actor_or_outlive_preview(self): ticket=self.preview()['confirmation'] self.claims['sub']='different' with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=ticket,identity_verified=True)) self.claims['sub']='operator';self.service.now=lambda:2000;self.claims['assurance']['at']=2000 with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=ticket,identity_verified=True)) def test_signed_target_cannot_be_overridden(self): p=self.preview() result=self.service.operation('signed',dict(action='apply',confirmation=p['confirmation'],identity_verified=True,user='bob',actor='attacker')) self.assertEqual('alice',result['user']);self.assertEqual('operator',result['actor']) def test_wsgi_denial_bounds_and_private_errors(self): def invoke(auth='Bearer signed',data=b'{}',length=None): status=[] body=b''.join(self.service(dict(PATH_INFO='/recover',REQUEST_METHOD='POST',HTTP_AUTHORIZATION=auth, CONTENT_LENGTH=str(len(data) if length is None else length),**{'wsgi.input':io.BytesIO(data)}),lambda s,h:status.append(s))) return status[0],json.loads(body) self.assertEqual('403 Forbidden',invoke(auth='')[0]) self.assertEqual('409 Conflict',invoke(length=17000)[0]) def outage(token):raise RuntimeError('private-provider-key') self.service.verify=outage self.assertNotIn('private-provider-key',json.dumps(invoke())) def test_support_reference_recovers_interrupted_confirmation(self): p=self.preview();self.store.fail_complete=True with self.assertRaises(RecoveryError):self.service.operation('signed',dict(action='apply',confirmation=p['confirmation'],identity_verified=True)) status=self.service.operation('signed',dict(action='status',reference='case-1')) self.assertEqual('pending',status['status']);self.assertFalse(status['active']) self.store.fail_complete=False self.service.operation('signed',dict(action='apply',confirmation=status['confirmation'],identity_verified=True)) self.assertEqual('recovered',self.service.operation('signed',dict(action='status',reference='case-1'))['status']) self.assertEqual(1,self.store.writes) def test_directory_subject_preview_apply_and_replay(self): # LLDAP adapter returns entry.DN; OIDC emits that exact user.ID as sub. subject='uid=platform-root,ou=people,dc=netkingdom,dc=local' self.claims['sub']=subject preview=self.preview() self.assertEqual(subject,preview['actor']) self.assertEqual(0,self.store.writes) body=dict(action='apply',confirmation=preview['confirmation'],identity_verified=True) self.assertTrue(self.service.operation('signed',body)['changes_applied']) self.assertTrue(self.service.operation('signed',body)['replayed']) self.assertEqual(subject,self.store.rows['case-1']['actor']) self.assertEqual(1,self.store.writes) def test_directory_subject_ticket_is_bound_to_exact_actor(self): self.claims['sub']='uid=admin,ou=people,dc=example' ticket=self.preview()['confirmation'] self.claims['sub']='uid=other,ou=people,dc=example' with self.assertRaisesRegex(RecoveryError,'preview_expired_or_changed'): self.service.operation('signed',dict(action='apply',confirmation=ticket,identity_verified=True)) self.assertEqual(0,self.store.writes) def test_actor_validation_preserves_identity_and_rejects_controls(self): from factor_recovery import validate_actor subject='cn=Platform Root,ou=people,dc=example' self.assertEqual(subject,validate_actor(subject)) for value in [None,{},'', ' ', 'a'*151, 'uid=admin\nsecret', 'admin\x00', 'admin\x7f', 'ΓΌ']: with self.subTest(value=repr(value)): with self.assertRaisesRegex(RecoveryError,'invalid_actor'): authorized_actor(dict(self.claims,sub=value),1000) self.assertEqual(0,self.store.writes) def test_wsgi_validation_diagnostics_never_contain_values(self): secret='private-value-must-not-appear' cases=[ (dict(action='preview',user=secret+'!',reference='case-1'),None,'user'), (dict(action='preview',user='alice',reference=secret+'!'),None,'reference'), (dict(action=secret),None,'action'), ([secret],None,'body'), (dict(action='preview',user='alice',reference='case-1'),secret,'content_length'), ] for body,length,field in cases: data=json.dumps(body).encode();diagnostic=io.StringIO() with redirect_stderr(diagnostic): result=b''.join(self.service(dict(PATH_INFO='/recover',REQUEST_METHOD='POST', HTTP_AUTHORIZATION='Bearer '+secret,CONTENT_LENGTH=length or str(len(data)), **{'wsgi.input':io.BytesIO(data)}),lambda s,h:None)) record=json.loads(diagnostic.getvalue()) self.assertEqual(dict(event='factor_recovery_validation_failed',code='invalid_request',field=field),record) self.assertNotIn(secret,diagnostic.getvalue()+result.decode()) self.assertEqual(0,self.store.writes) def test_provider_field_failure_is_identifiable_without_serial_value(self): self.store.serials=lambda user:['private-serial!'] data=json.dumps(dict(action='preview',user='alice',reference='case-1')).encode() diagnostic=io.StringIO() with redirect_stderr(diagnostic): result=b''.join(self.service(dict(PATH_INFO='/recover',REQUEST_METHOD='POST', HTTP_AUTHORIZATION='Bearer signed',CONTENT_LENGTH=str(len(data)), **{'wsgi.input':io.BytesIO(data)}),lambda s,h:None)) self.assertEqual('serial',json.loads(diagnostic.getvalue())['field']) self.assertNotIn('private-serial',diagnostic.getvalue()+result.decode()) self.assertEqual(0,self.store.writes)