Refuse multi-owner factors and lock ownership during recovery

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

scripts/factor_recovery.py

@@ -76,17 +76,24 @@ class ProviderStore:
   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
+  from privacyidea.models import TokenOwner
+  token=values[0].token
+  if TokenOwner.query.filter_by(token_id=token.id).count()!=1:raise RecoveryError("shared_factor_not_supported")
+  return token
  def snapshot(self,user,serial):
   token=self._token(user,serial)
   if token is None:return None
   data=token.get_vars()
   return {k:data[k] for k in ('id','serial','user_id','resolver','realms','active','rollout_state','tokentype')}
  def disable(self,user,serial,expected):
-  from privacyidea.models import db,Token
+  from privacyidea.models import db,Token,TokenOwner
   token=self._token(user,serial)
   if token is None:raise RecoveryError('factor_not_owned_by_target')
+  if db.engine.dialect.name=="postgresql":
+   from sqlalchemy import text
+   db.session.execute(text("SET LOCAL lock_timeout = '5s'"))
   locked=db.session.query(Token).filter(Token.id==token.id).with_for_update().one()
+  db.session.query(TokenOwner).filter(TokenOwner.token_id==token.id).with_for_update(of=TokenOwner).all()
   db.session.refresh(locked)
   if fingerprint(self.snapshot(user,serial))!=expected:
    db.session.rollback();raise RecoveryError('stale_preview')

scripts/provider-onboarding-contract.py

@@ -126,6 +126,14 @@ def run():
     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'])
+   from privacyidea.models import TokenOwner
+   from privacyidea.lib.token import get_tokens
+   factor_id=get_tokens(serial=serial)[0].token.id
+   other_owner=TokenOwner(token_id=factor_id,user_id='second-fixture-owner',resolver='fixture-users',realmname='fixture')
+   other_owner.save()
+   status,shared=operation(preview_body)
+   check('shared_identity_factor_denied',status==409 and shared['failure']=='shared_factor_not_supported')
+   db.session.delete(other_owner);db.session.commit()
    status,preview=operation(preview_body)
    check('service_preview_owned_factor',status==200 and len(preview['factors'])==1)
    confirmation=preview['factors'][0]['confirmation']