key-cape/scripts/test_recovery_service.py

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)