Reconcile recovery by support reference and bound provider lookup

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

scripts/factor_recovery.py

@@ -56,6 +56,7 @@ class ProviderStore:
    if db.engine.dialect.name != 'postgresql':
     yield;return
    with db.engine.connect() as connection:
+    connection.execute(text("SET lock_timeout = '5s'"))
     connection.execute(text('SELECT pg_advisory_lock(4912040036)'))
     try:yield
     finally:connection.execute(text('SELECT pg_advisory_unlock(4912040036)'))
@@ -63,13 +64,17 @@ class ProviderStore:
  def serials(self,user):
   from privacyidea.lib.token import get_tokens
   from privacyidea.lib.user import User
-  values=get_tokens(user=User(user,self.realm))
+  identity=User(user,self.realm)
+  if not identity.resolver or identity.uid in (None,''):raise RecoveryError('account_not_found')
+  values=get_tokens(user=identity)
   if len(values)>20:raise RecoveryError('too_many_factors')
   return [token.token.serial for token in values]
  def _token(self,user,serial):
   from privacyidea.lib.token import get_tokens
   from privacyidea.lib.user import User
-  values=get_tokens(user=User(user,self.realm),serial=serial)
+  identity=User(user,self.realm)
+  if not identity.resolver or identity.uid in (None,''):return None
+  values=get_tokens(user=identity,serial=serial)
   if len(values)!=1:return None
   return values[0].token
  def snapshot(self,user,serial):

scripts/provider-onboarding-contract.py

@@ -124,6 +124,8 @@ def run():
    for name,token in [('tenant_admin',signed(roles=['tenant-admin'])),('wrong_audience',signed(aud='other')),('stale_mfa',signed(assurance=dict(level='aal2',mfa=True,at=now-301))),('unsigned','invalid')]:
     status,_=operation(preview_body,token)
     check('service_denies_'+name,status==403 and store.snapshot('alice',serial)['active'])
+   status,_=operation(dict(preview_body,user='missing-fixture-user'))
+   check('service_unknown_user_not_realm_wide',status==409 and store.snapshot('alice',serial)['active'])
    status,preview=operation(preview_body)
    check('service_preview_owned_factor',status==200 and len(preview['factors'])==1)
    confirmation=preview['factors'][0]['confirmation']
@@ -134,6 +136,8 @@ def run():
    check('service_audited_recovery',status==200 and receipt['changes_applied'] and store.receipt('fixture-browser-recovery')['complete'])
    status,receipt=operation(apply_body)
    check('service_safe_retry',status==200 and receipt['replayed'])
+   status,receipt=operation(dict(action='status',reference='fixture-browser-recovery'))
+   check('service_support_reference_readback',status==200 and receipt['status']=='recovered')
   phase='replacement_possession'
   detail=enroll();serial=detail['serial']
   seed=urllib.parse.parse_qs(urllib.parse.urlsplit(detail['googleurl']['value']).query)['secret'][0]

scripts/recovery_service.py

@@ -4,7 +4,7 @@ import hashlib
 import hmac
 import json
 import time
-from factor_recovery import ProviderStore, RecoveryError, recover
+from factor_recovery import ProviderStore, RecoveryError, recover, fingerprint
 
 
 def authorized_actor(claims, now=None):
@@ -64,6 +64,27 @@ class RecoveryService:
                     preview['confirmation'] = self.ticket(dict(request, expected_version=preview['version']))
                     factors.append(preview)
             return dict(success=True, status='preview', user=user, reference=reference, factors=factors)
+        if body.get('action') == 'status':
+            import re
+            reference = body.get('reference', '')
+            if not isinstance(reference,str) or not re.fullmatch(r'[A-Za-z0-9_.@:/-]{1,150}',reference):
+                raise RecoveryError('invalid_request')
+            row = self.store.receipt(reference)
+            if row is None:
+                return dict(success=True,status='not_found',reference=reference)
+            current = self.store.snapshot(row['user'],row['serial'])
+            if current is None or fingerprint(dict(current,active=True)) != row['version']:
+                raise RecoveryError('factor_changed_after_recovery')
+            result = dict(success=True,status='pending',reference=reference,user=row['user'],
+                          serial=row['serial'],active=current['active'],actor=row['actor'])
+            if row['complete']:
+                if current['active']:
+                    raise RecoveryError('factor_changed_after_recovery')
+                result.update(status='recovered',replayed=True,changes_applied=False)
+            elif row['actor'] == actor:
+                result['confirmation'] = self.ticket(dict(user=row['user'],serial=row['serial'],
+                    realm=row['realm'],actor=actor,reference=reference,expected_version=row['version']))
+            return result
         if body.get('action') == 'apply':
             if body.get('identity_verified') is not True:
                 raise RecoveryError('identity_verification_required')

scripts/test_recovery_service.py

@@ -49,3 +49,13 @@ class RecoveryServiceTests(unittest.TestCase):
         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)