fix: recover keycloak user on create error

ariadne/manager/provisioning.py

@@ -470,8 +470,22 @@ class ProvisioningManager:
         email = self._require_verified_email(ctx)
         self._ensure_email_unused(email, ctx.username)
         payload = self._new_user_payload(ctx.username, email, ctx.mailu_email)
-        created_id = keycloak_admin.create_user(payload)
+        try:
-        return keycloak_admin.get_user(created_id)
+            created_id = keycloak_admin.create_user(payload)
+            return keycloak_admin.get_user(created_id)
+        except Exception as exc:
+            detail = safe_error_detail(exc, "create user failed")
+            logger.warning(
+                "keycloak create user failed, checking for existing user",
+                extra={"event": "keycloak_user_fallback", "username": ctx.username, "detail": detail},
+            )
+            user = keycloak_admin.find_user(ctx.username)
+            if user:
+                return user
+            user = keycloak_admin.find_user_by_email(email)
+            if user:
+                return user
+            raise
 
     def _fetch_full_user(self, user_id: str, fallback: dict[str, Any]) -> dict[str, Any]:
         try:

tests/test_provisioning.py

@@ -228,6 +228,75 @@ def test_provisioning_creates_user_and_password(monkeypatch) -> None:
     assert admin.reset_calls
 
 
+def test_provisioning_create_user_fallback(monkeypatch) -> None:
+    dummy_settings = types.SimpleNamespace(
+        mailu_domain="bstein.dev",
+        mailu_sync_url="http://mailu",
+        mailu_mailbox_wait_timeout_sec=1.0,
+        nextcloud_namespace="nextcloud",
+        nextcloud_mail_sync_cronjob="nextcloud-mail-sync",
+        provision_retry_cooldown_sec=0.0,
+        default_user_groups=["dev"],
+        allowed_flag_groups=["demo"],
+        welcome_email_enabled=False,
+        portal_public_base_url="https://bstein.dev",
+    )
+    monkeypatch.setattr(prov, "settings", dummy_settings)
+    _patch_mailu_ready(monkeypatch, dummy_settings)
+
+    class Admin(DummyAdmin):
+        def __init__(self):
+            super().__init__()
+            self.find_calls = 0
+            self.reset_calls = []
+
+        def find_user(self, username):
+            self.find_calls += 1
+            if self.find_calls >= 2:
+                return {"id": "user-123", "username": username, "attributes": {}, "requiredActions": []}
+            return None
+
+        def get_user(self, user_id):
+            return {"id": user_id, "username": "alice", "attributes": {}, "requiredActions": []}
+
+        def create_user(self, payload):
+            raise RuntimeError("boom")
+
+        def reset_password(self, user_id, password, temporary=False):
+            self.reset_calls.append((user_id, temporary))
+
+    admin = Admin()
+    monkeypatch.setattr(prov, "keycloak_admin", admin)
+    monkeypatch.setattr(prov.mailu, "sync", lambda reason, force=False: None)
+    monkeypatch.setattr(prov.mailu, "wait_for_mailbox", lambda email, timeout: True)
+    monkeypatch.setattr(prov.nextcloud, "sync_mail", lambda username, wait=True: {"status": "ok"})
+    monkeypatch.setattr(prov.wger, "sync_user", lambda username, email, password, wait=True: {"status": "ok"})
+    monkeypatch.setattr(prov.firefly, "sync_user", lambda email, password, wait=True: {"status": "ok"})
+    monkeypatch.setattr(prov.vaultwarden, "invite_user", lambda email: VaultwardenInvite(True, "invited"))
+    monkeypatch.setattr(prov.ProvisioningManager, "_all_tasks_ok", lambda *args, **kwargs: True)
+    monkeypatch.setattr(prov.ProvisioningManager, "_send_welcome_email", lambda *args, **kwargs: None)
+
+    row = {
+        "username": "alice",
+        "contact_email": "alice@example.com",
+        "email_verified_at": datetime.now(timezone.utc),
+        "status": "approved",
+        "initial_password": None,
+        "initial_password_revealed_at": None,
+        "provision_attempted_at": None,
+        "approval_flags": ["demo"],
+    }
+
+    db = DummyDB(row)
+    storage = DummyStorage()
+    manager = prov.ProvisioningManager(db, storage)
+    outcome = manager.provision_access_request("REQ125")
+
+    assert outcome.status == "awaiting_onboarding"
+    assert admin.find_calls >= 2
+    assert admin.reset_calls
+
+
 def test_extract_attr_variants() -> None:
     assert prov._extract_attr("bad", "key") == ""
     assert prov._extract_attr({"key": ["value"]}, "key") == "value"