tegwick
d15f4dde0b
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a092fe-13b1-7f12-ac74-7d258af4d79c
61 lines
4.1 KiB
Python
61 lines
4.1 KiB
Python
import io
|
|
import json
|
|
import unittest
|
|
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)
|