Authorize self identity links by stored ownership
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

This commit is contained in:
tegwick 2026-08-14 16:16:59 +02:00
parent 10ee042c41
commit 5e37e9c012
2 changed files with 11 additions and 1 deletions

View file

@ -1948,6 +1948,7 @@ class UserEngineService:
return existing
correlation_id = correlation_id or new_id("corr")
actor_identity = self.store.find_identity(*actor.identity_key)
decision = self._authorize(
actor,
action="identity.link",
@ -1956,6 +1957,10 @@ class UserEngineService:
tenant=actor.tenant,
correlation_id=correlation_id,
target_user_id=user_id,
context={
"self": actor_identity is not None
and actor_identity.user_id == user_id,
},
)
identity = ExternalIdentity(
identity_id=new_id("idn"),

View file

@ -42,7 +42,7 @@ class IsolatedMvpTests(unittest.TestCase):
self.assertEqual(len(service.outbox_events()), 1)
def test_account_lifecycle_and_identity_linking(self):
service, _, _ = _service()
service, _, authorization = _service()
session = service.me(human_actor_claims(), correlation_id="corr-me")
disabled = service.set_account_status(
@ -67,6 +67,11 @@ class IsolatedMvpTests(unittest.TestCase):
self.assertEqual(disabled.status, AccountStatus.DISABLED)
self.assertEqual(linked.user_id, session.user.user_id)
self.assertEqual(linked_session.user.user_id, session.user.user_id)
link_request = next(
request for request in authorization.requests
if request.action == "identity.link"
)
self.assertTrue(link_request.context["self"])
self.assertIn(
"identity.linked",
[event.event_type for event in service.outbox_events()],