fix(security): serialize SCIM replacement with account merges

docs/merge-unification-flow.md

@@ -31,6 +31,7 @@ only an unverified email, the merge is refused with `422 Unverified email`.
 ```
 merge(survivor S, duplicate D, actor A):
   reject if S == D                         -> 400 SameUser
+  acquire shared user-operation locks for S and D
   load S, D (must exist)                   -> 404 UserNotFound
   reject if S or D disabled                -> 409 InactiveAccount
   decision = decide_match(S, D, explicit)
@@ -55,6 +56,7 @@ merge(survivor S, duplicate D, actor A):
   disable D                                       # D can never authenticate again
   audit "duplicate_tombstoned"
   audit "merge_completed" {moved_*, conflicts}
+  release shared user-operation locks
   return MergeResult{..., audit_id}
 ```
 
@@ -72,6 +74,25 @@ The duplicate is **not deleted**. Its Keycloak user attribute
 (`enabled: false`). This preserves forensic history and lets any stale reference
 resolve to the survivor.
 
+### SCIM/merge serialization invariant
+
+`PUT /scim/v2/Users/{id}` performs a full Keycloak user-representation write and
+can set `active: true`. Its tombstone check and replacement PUT therefore execute
+inside the **same user-operation lock** used by the complete merge transaction.
+A merge cannot create `merged_into_user_id` between those two Admin API calls,
+and SCIM cannot wipe a newly-created tombstone or reactivate the duplicate.
+
+Standalone deployments use a dedicated SQLite sidecar lock database and hold a
+`BEGIN IMMEDIATE` transaction for the complete critical section. This provides a
+crash-safe mutex shared by every worker/process using the same database path;
+process death closes the connection and releases the lock. The current backend
+serializes all user mutations conservatively rather than risking a multi-user
+deadlock. Lock acquisition waits up to 10 seconds, then returns retryable HTTP
+`503` without performing a partial mutation. A clustered Postgres deployment
+must provide the same `UserOperationLocks` contract (for example, ordered
+advisory locks) and wire one shared instance into both the merge service and SCIM
+router.
+
 ## Audit
 
 Every step emits an immutable `account_merge_audit` event sharing one

services/account_unification/app/api.py

@@ -13,6 +13,7 @@
 )
 from .models import FederatedIdentity, MergeRequest, MergeResult, UserAccount
 from .service import UnificationService
+from .user_locks import UserOperationLockTimeout
 
 router = APIRouter()
 
@@ -75,6 +76,11 @@ def merge_accounts(
         raise HTTPException(status_code=409, detail=str(exc)) from exc
     except InactiveAccountError as exc:
         raise HTTPException(status_code=409, detail=str(exc)) from exc
+    except UserOperationLockTimeout as exc:
+        raise HTTPException(
+            status_code=503,
+            detail="one of the requested accounts is being modified; retry",
+        ) from exc
 
 
 @router.get("/merges/{audit_id}/audit", tags=["merge"])

services/account_unification/app/healthcheck.py

@@ -7,16 +7,49 @@
 
 import json
 import sys
+import urllib.parse
 import urllib.request
 
 DEFAULT_URL = "http://127.0.0.1:8099/healthz"
+_ALLOWED_SCHEMES = frozenset({"http", "https"})
+
+
+class _HttpOnlyRedirectHandler(urllib.request.HTTPRedirectHandler):
+    """Drop redirects whose target scheme is not HTTP(S)."""
+
+    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
+        """Return an HTTP(S) redirect request, or reject another scheme."""
+        if urllib.parse.urlsplit(newurl).scheme.lower() not in _ALLOWED_SCHEMES:
+            return None
+        return super().redirect_request(req, fp, code, msg, headers, newurl)
+
+
+def _build_http_only_opener() -> urllib.request.OpenerDirector:
+    """Build an opener that can speak only HTTP(S), without file/FTP handlers."""
+    opener = urllib.request.OpenerDirector()
+    opener.add_handler(urllib.request.HTTPHandler())
+    opener.add_handler(urllib.request.HTTPSHandler())
+    opener.add_handler(_HttpOnlyRedirectHandler())
+    opener.add_handler(urllib.request.HTTPErrorProcessor())
+    return opener
+
+
+def _open_health_url(url: str):  # noqa: ANN202
+    """Open an HTTP(S) health URL through the restricted opener."""
+    return _build_http_only_opener().open(url, timeout=5)
 
 
 def main(url: str = DEFAULT_URL) -> int:
     """Check the configured health endpoint and return a shell status code."""
+    scheme = urllib.parse.urlsplit(url).scheme.lower()
+    if scheme not in _ALLOWED_SCHEMES:
+        print(f"healthcheck failed: unsupported URL scheme {scheme!r}", file=sys.stderr)
+        return 1
     try:
-        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected -- container healthcheck against a hardcoded loopback default (127.0.0.1); any override is a deployment-controlled target, not user input.
-        with urllib.request.urlopen(url, timeout=5) as response:  # noqa: S310
+        # The initial and redirected schemes are constrained to HTTP(S), and the
+        # opener carries no handler for file, FTP, or data URLs.
+        # nosemgrep: python.lang.security.audit.dynamic-urllib-use-detected.dynamic-urllib-use-detected
+        with _open_health_url(url) as response:  # noqa: S310
             body = json.loads(response.read().decode("utf-8"))
     except Exception as exc:  # pragma: no cover - network failure path
         print(f"healthcheck failed: {exc}", file=sys.stderr)

services/account_unification/app/keycloak_client.py

@@ -108,6 +108,10 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
         """Set one single-valued user attribute."""
         ...
 
+    def get_user_attribute(self, user_id: str, key: str) -> str | None:
+        """Return one single-valued user attribute, or ``None`` if unset."""
+        ...
+
 
 class HttpAdminApi:
     """httpx-backed :class:`AdminApi` for a live Keycloak instance.
@@ -339,6 +343,18 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None:
             f"/admin/realms/{self._realm}/users/{user_id}", {"attributes": attributes}
         )
 
+    def get_user_attribute(self, user_id: str, key: str) -> str | None:
+        """Return one single-valued Keycloak user attribute, or ``None``.
+
+        Keycloak stores attributes as string lists; this returns the first
+        element (or a bare string, defensively), or ``None`` when unset.
+        """
+        data = self._get(f"/admin/realms/{self._realm}/users/{user_id}")
+        value = (data.get("attributes") or {}).get(key)
+        if isinstance(value, list):
+            return value[0] if value else None
+        return value if isinstance(value, str) else None
+
     # -- transport ---------------------------------------------------------
     def _get(self, path: str, params: dict | None = None) -> dict | list:
         """Issue an authenticated GET and parse JSON."""

services/account_unification/app/main.py

@@ -20,6 +20,7 @@
 from .keycloak_client import HttpAdminApi
 from .scim import scim_router
 from .service import UnificationService
+from .user_locks import SqliteUserOperationLocks
 
 
 def build_service(app: FastAPI) -> None:
@@ -39,10 +40,22 @@ def build_service(app: FastAPI) -> None:
     # a Postgres-backed sink writing account_merge_audit.
     audit_path = descriptor.sqlite_path or "account_unification.db"
     audit = AuditLogger(SqliteAuditSink(audit_path))
+    # Use a dedicated sidecar database so the serialization transaction never
+    # blocks config reads/writes or audit persistence. Every service worker that
+    # shares this path also shares the same crash-safe SQLite mutex.
+    user_operation_locks = SqliteUserOperationLocks(
+        f"{audit_path}.user-operation-locks.sqlite3"
+    )
 
-    app.state.unification_service = UnificationService(api, audit, config)
+    app.state.unification_service = UnificationService(
+        api,
+        audit,
+        config,
+        user_operation_locks,
+    )
     app.state.audit_logger = audit
     app.state.keycloak_api = api
+    app.state.user_operation_locks = user_operation_locks
     app.state.ready = True
 
 

services/account_unification/app/scim.py

@@ -20,6 +20,11 @@
 
 from .keycloak_client import AdminApi
 from .models import UserAccount
+from .service import TOMBSTONE_ATTRIBUTE_KEY
+from .user_locks import (
+    UserOperationLocks,
+    UserOperationLockTimeout,
+)
 
 SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User"
 SCIM_LIST_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:ListResponse"
@@ -38,6 +43,14 @@ def get_provisioner(request: Request) -> AdminApi:
     return api
 
 
+def get_user_operation_locks(request: Request) -> UserOperationLocks:
+    """Return the shared lock manager used by both SCIM and merge paths."""
+    locks = getattr(request.app.state, "user_operation_locks", None)
+    if locks is None:  # pragma: no cover - only when misconfigured / test wiring
+        raise HTTPException(status_code=503, detail="user operation locks not wired")
+    return locks
+
+
 def _scim_error(status: int, detail: str) -> HTTPException:
     """Build a SCIM-shaped HTTP error."""
     return HTTPException(
@@ -186,15 +199,35 @@ def replace_user(
     user_id: str,
     resource: dict[str, Any],
     provisioner: AdminApi = Depends(get_provisioner),
+    user_operation_locks: UserOperationLocks = Depends(get_user_operation_locks),
 ) -> Response:
     """Replace a provisioned user from a SCIM PUT request."""
     try:
-        provisioner.get_user(user_id)
-    except KeyError as exc:
-        raise _scim_error(404, f"user '{user_id}' not found") from exc
-    account = _to_user_account(resource, user_id=user_id)
-    provisioner.replace_user(user_id, account)
-    return _scim_response(_to_scim_resource(provisioner.get_user(user_id)))
+        with user_operation_locks.hold(user_id):
+            try:
+                provisioner.get_user(user_id)
+            except KeyError as exc:
+                raise _scim_error(404, f"user '{user_id}' not found") from exc
+            # A merged-away duplicate is tombstoned (disabled + a
+            # merged_into_user_id pointer) so it can never authenticate again.
+            # Keep the check and the full replacement PUT under the same lock
+            # used by merge_accounts; otherwise merge can create the tombstone
+            # between these two Admin API calls and SCIM can wipe it again.
+            if provisioner.get_user_attribute(user_id, TOMBSTONE_ATTRIBUTE_KEY):
+                raise _scim_error(
+                    409,
+                    f"user '{user_id}' has been merged into another account "
+                    "and cannot be modified",
+                )
+            account = _to_user_account(resource, user_id=user_id)
+            provisioner.replace_user(user_id, account)
+            replaced = provisioner.get_user(user_id)
+    except UserOperationLockTimeout as exc:
+        raise _scim_error(
+            503,
+            f"user '{user_id}' is being modified; retry the request",
+        ) from exc
+    return _scim_response(_to_scim_resource(replaced))
 
 
 @scim_router.patch("/Users/{user_id}")