From 2fd34a91e211d21b8a6d210d49b248bd2527daa8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 29 Jul 2026 13:13:29 +0000 Subject: [PATCH 01/21] fix(healthcheck): allow-list URL scheme and clear Semgrep dynamic-urllib finding The central SAST Semgrep gate fails on the base branch because p/default's python.lang.security.audit.dynamic-urllib-use-detected flags app/healthcheck.py: urlopen() receives a non-literal url, which urllib would happily open as a file:// path. This Medium finding blocks every open keyverse PR, since each PR scans a tree that still contains this file. Harden the probe by rejecting any URL whose scheme is not http/https before opening it, so a stray value can never coerce urlopen into a file:// read or another protocol handler. The residual audit finding on the (still non-literal) urlopen call is suppressed narrowly with an inline `# nosemgrep: dynamic-urllib-use-detected`, justified by the scheme allow-list and the fact that the container self-probe URL is not attacker-controlled. Add a regression test for the rejected-scheme path. Verified locally: semgrep marks the finding suppressed (gate passes), ruff is clean, and the healthcheck tests pass. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01HdCssGnNMhKHNu3TXFstWH --- services/account_unification/app/healthcheck.py | 11 +++++++++++ .../account_unification/tests/test_healthcheck.py | 11 +++++++++++ 2 files changed, 22 insertions(+) diff --git a/services/account_unification/app/healthcheck.py b/services/account_unification/app/healthcheck.py index 4284510..713e31b 100644 --- a/services/account_unification/app/healthcheck.py +++ b/services/account_unification/app/healthcheck.py @@ -7,14 +7,25 @@ import json import sys +import urllib.parse import urllib.request DEFAULT_URL = "http://127.0.0.1:8099/healthz" +_ALLOWED_SCHEMES = frozenset({"http", "https"}) 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: + # Reject non-HTTP(S) schemes so a stray value can never coerce urlopen into + # reading a local ``file://`` path or reaching another protocol handler. + print(f"healthcheck failed: unsupported URL scheme {scheme!r}", file=sys.stderr) + return 1 try: + # Internal container self-probe; the scheme is allow-listed to http/https + # above, so this urlopen cannot be redirected to a file:// path or other handler. + # nosemgrep: dynamic-urllib-use-detected with urllib.request.urlopen(url, timeout=5) as response: # noqa: S310 body = json.loads(response.read().decode("utf-8")) except Exception as exc: # pragma: no cover - network failure path diff --git a/services/account_unification/tests/test_healthcheck.py b/services/account_unification/tests/test_healthcheck.py index e538b16..af99c34 100644 --- a/services/account_unification/tests/test_healthcheck.py +++ b/services/account_unification/tests/test_healthcheck.py @@ -48,3 +48,14 @@ def fake_urlopen(url: str, *, timeout: int) -> _Response: assert healthcheck.main("http://service/healthz") == 1 assert "healthcheck failed: connection refused" in capsys.readouterr().err + + +def test_healthcheck_rejects_non_http_scheme(monkeypatch, capsys): + """A non-HTTP(S) URL is rejected before urllib ever opens it.""" + def fail_urlopen(*args: object, **kwargs: object) -> _Response: + raise AssertionError("urlopen must not run for a rejected scheme") + + monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fail_urlopen) + + assert healthcheck.main("file:///etc/passwd") == 1 + assert "unsupported URL scheme 'file'" in capsys.readouterr().err From 5e9c4eb125fb35adb27125ab179e2ea290feeeba Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 30 Jul 2026 11:09:26 +0000 Subject: [PATCH 02/21] fix(scim): refuse SCIM PUT that would resurrect a tombstoned merged duplicate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A merged-away duplicate is tombstoned (disabled + a `merged_into_user_id` pointer to the survivor) so it can never authenticate again, per docs/merge-unification-flow.md and the CLAUDE.md merge invariant. But the SCIM shim's `PUT /scim/v2/Users/{id}` (`replace_user`) only did a 404 existence check and then translated the resource (whose `active` defaults to true) into a replace — with no tombstone guard. SCIM PUT is the *only* reactivation vector: `create` guards uniqueness and `patch`/`delete` only ever disable. So a routine upstream HR/IGA full-sync PUT that still lists the decommissioned person silently re-enabled the tombstoned account (restoring its untouched passkey/WebAuthn login), and against a live Keycloak the PUT would also overwrite the user representation, wiping the `merged_into_user_id` pointer that resolves stale references to the survivor. SCIM endpoints carry no app-level authz (trust terminates at the WAF edge), so the trigger is unprivileged. Fix: in `replace_user`, refuse with SCIM 409 when the target carries the tombstone attribute, keeping a merged duplicate immutable via SCIM. Adds `get_user_attribute` to the `AdminApi` protocol, the HTTP client, and the mock so the guard reads the pointer uniformly. Regression test asserts the PUT is refused and the duplicate stays disabled with its survivor pointer intact (verified red→green: without the guard the test fails as the account is re-enabled). Full suite 54 passed; interrogate 100%. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HdCssGnNMhKHNu3TXFstWH --- .../app/keycloak_client.py | 16 ++++++++++++++ services/account_unification/app/scim.py | 13 +++++++++++ .../tests/mock_keycloak.py | 4 ++++ .../account_unification/tests/test_scim.py | 22 +++++++++++++++++++ 4 files changed, 55 insertions(+) diff --git a/services/account_unification/app/keycloak_client.py b/services/account_unification/app/keycloak_client.py index 7d6ab05..4ab2d57 100644 --- a/services/account_unification/app/keycloak_client.py +++ b/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.""" diff --git a/services/account_unification/app/scim.py b/services/account_unification/app/scim.py index 2cb7609..f76f6ea 100644 --- a/services/account_unification/app/scim.py +++ b/services/account_unification/app/scim.py @@ -20,6 +20,7 @@ from .keycloak_client import AdminApi from .models import UserAccount +from .service import TOMBSTONE_ATTRIBUTE_KEY SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" SCIM_LIST_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:ListResponse" @@ -192,6 +193,18 @@ def replace_user( 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. SCIM PUT is the only + # reactivation vector -- create guards uniqueness, and patch/delete only + # ever disable -- and a live Keycloak PUT would also overwrite the whole + # user representation, wiping the survivor pointer. Refuse it outright so a + # routine upstream full-sync cannot resurrect a decommissioned identity. + 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) return _scim_response(_to_scim_resource(provisioner.get_user(user_id))) diff --git a/services/account_unification/tests/mock_keycloak.py b/services/account_unification/tests/mock_keycloak.py index f844391..26f9c38 100644 --- a/services/account_unification/tests/mock_keycloak.py +++ b/services/account_unification/tests/mock_keycloak.py @@ -148,3 +148,7 @@ def set_user_attribute(self, user_id: str, key: str, value: str) -> None: self.users[user_id] = self.users[user_id].model_copy( update={"external_id": value} ) + + def get_user_attribute(self, user_id: str, key: str) -> str | None: + self.calls.append(f"get_user_attribute:{user_id}:{key}") + return self.attributes.get((user_id, key)) diff --git a/services/account_unification/tests/test_scim.py b/services/account_unification/tests/test_scim.py index bdcef85..cfd57d4 100644 --- a/services/account_unification/tests/test_scim.py +++ b/services/account_unification/tests/test_scim.py @@ -82,6 +82,28 @@ def test_scim_replace_updates_user(client, api): assert api.get_user(created["id"]).email == "jane.doe@corp.com" +def test_scim_replace_refuses_to_resurrect_a_tombstoned_duplicate(client, api): + """A merged-away (tombstoned) duplicate must not be re-enabled via SCIM PUT. + + After a merge the duplicate is disabled and carries a merged_into_user_id + pointer. A routine upstream full-sync PUT (``active`` defaults to true) must + be refused with 409, leaving the duplicate disabled with its survivor pointer + intact -- never silently reactivated. + """ + created = client.post("/scim/v2/Users", json=_scim_user()).json() + dup_id = created["id"] + # Simulate the post-merge tombstone state (service._tombstone does exactly this). + api.set_user_attribute(dup_id, "merged_into_user_id", "survivor-id") + api.deactivate_user(dup_id) + + response = client.put(f"/scim/v2/Users/{dup_id}", json=_scim_user()) + + assert response.status_code == 409 + assert dup_id in api.deactivated + assert api.get_user(dup_id).state == "disabled" + assert api.get_user_attribute(dup_id, "merged_into_user_id") == "survivor-id" + + def test_scim_patch_deactivates_user(client, api): created = client.post("/scim/v2/Users", json=_scim_user()).json() response = client.patch( From e95e87b442d35d383f6547408af1934b33eca5ef Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 01:39:10 +0000 Subject: [PATCH 03/21] fix(merge): refuse explicit-link merge when the only tie is an unverified email MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `UnificationService.merge_accounts` nested the unverified-email refusal inside `if not decision.matched:`. Because `decide_match(..., explicit_link=True)` returns `matched=True, reason=EXPLICIT_LINK`, an operator "explicit link" merge skipped the guard entirely and merged + tombstoned two accounts whose only shared tie was an UNVERIFIED email — the account-takeover vector the hard rule exists to block (an attacker registers a duplicate holding the victim's unverified email, then one explicit_link=True merge folds it into the victim). This violates three contract sources: - `app/models.py` MergeRequest.explicit_link docstring: "Even so, the service refuses if the only tie is an UNVERIFIED email." - `docs/merge-unification-flow.md`: "reject if only tie is unverified email -> 422 UnverifiedEmailMerge" is an unconditional step after decide_match. - `CLAUDE.md`: "Never link or merge accounts on an unverified email." Fix: hoist the unverified-email guard out of the not-matched branch and run it for every decision reason except a genuine tie (EXACT_IDP_SUBJECT / VERIFIED_ EMAIL), so explicit-link and no-match are both covered. Legitimate merges are preserved: an explicit link with different/absent emails still merges, and verified-email / exact-(idp,subject) matches are exempt. TDD: added test_refuse_explicit_merge_on_shared_unverified_email — confirmed it fails on the pre-fix code ("DID NOT RAISE UnverifiedEmailMergeError", duplicate gets tombstoned) and passes after. No existing test changes. Verified (CI parity, py3.12, services/account_unification): ruff clean, interrogate 100% (>=80 gate), pytest all pass. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HdCssGnNMhKHNu3TXFstWH --- services/account_unification/app/service.py | 21 +++++++++++++------ .../account_unification/tests/test_merge.py | 10 +++++++++ 2 files changed, 25 insertions(+), 6 deletions(-) diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py index 3354c6f..6c77371 100644 --- a/services/account_unification/app/service.py +++ b/services/account_unification/app/service.py @@ -27,6 +27,7 @@ from .matching import decide_match, have_matching_verified_email from .models import ( FederatedIdentity, + MatchReason, MergeConflict, MergeRequest, MergeResult, @@ -85,19 +86,27 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult: decision = decide_match( survivor, duplicate, explicit_link=request.explicit_link ) - # Guard: refuse when the accounts only coincide on an unverified email. - # (decide_match already refuses to *call* that a verified match; here we - # produce the precise error for the operator + audit trail.) - if not decision.matched: + # Guard: refuse whenever the ONLY shared tie is an unverified email -- + # even when an operator asserts an explicit link. This is the + # account-takeover vector the hard rule blocks (an attacker registers a + # duplicate holding the victim's unverified email). A verified-email or + # exact (idp, subject) match is a genuine tie and is exempt, so the guard + # runs for every other decision reason (explicit link and no-match alike) + # rather than only when decide_match found no rule. + if decision.reason not in ( + MatchReason.EXACT_IDP_SUBJECT, + MatchReason.VERIFIED_EMAIL, + ): same_email = ( - (survivor.email or "").strip().lower() + bool((survivor.email or "").strip()) + and (survivor.email or "").strip().lower() == (duplicate.email or "").strip().lower() - and bool(survivor.email) ) if same_email and not have_matching_verified_email(survivor, duplicate): raise UnverifiedEmailMergeError( "refusing merge: accounts share only an UNVERIFIED email" ) + if not decision.matched: raise NoMatchError(decision.detail or "no matching rule satisfied") audit_id = self._audit.new_correlation_id() diff --git a/services/account_unification/tests/test_merge.py b/services/account_unification/tests/test_merge.py index 1acfd12..b4b559c 100644 --- a/services/account_unification/tests/test_merge.py +++ b/services/account_unification/tests/test_merge.py @@ -168,6 +168,16 @@ def test_explicit_link_allows_merge_without_shared_signal(service, api): assert result.duplicate_tombstoned +def test_refuse_explicit_merge_on_shared_unverified_email(service, api): + """An explicit link must not launder a shared UNVERIFIED email into a merge.""" + api.create_test_user("survivor", email="jane@corp.com", is_email_verified=False) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) + with pytest.raises(UnverifiedEmailMergeError): + service.merge_accounts(_merge(explicit=True)) + # nothing mutated: duplicate not tombstoned. + assert "dup" not in api.deactivated + + def test_refuse_self_merge(service, api): api.create_test_user("same", email="a@x.com", is_email_verified=True) with pytest.raises(SameUserError): From 4e2b9d245d77650dfe22b768763f2b73334571b2 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 01:41:22 +0000 Subject: [PATCH 04/21] Revert "fix(merge): refuse explicit-link merge when the only tie is an unverified email" This reverts commit e95e87b442d35d383f6547408af1934b33eca5ef. --- services/account_unification/app/service.py | 21 ++++++------------- .../account_unification/tests/test_merge.py | 10 --------- 2 files changed, 6 insertions(+), 25 deletions(-) diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py index 6c77371..3354c6f 100644 --- a/services/account_unification/app/service.py +++ b/services/account_unification/app/service.py @@ -27,7 +27,6 @@ from .matching import decide_match, have_matching_verified_email from .models import ( FederatedIdentity, - MatchReason, MergeConflict, MergeRequest, MergeResult, @@ -86,27 +85,19 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult: decision = decide_match( survivor, duplicate, explicit_link=request.explicit_link ) - # Guard: refuse whenever the ONLY shared tie is an unverified email -- - # even when an operator asserts an explicit link. This is the - # account-takeover vector the hard rule blocks (an attacker registers a - # duplicate holding the victim's unverified email). A verified-email or - # exact (idp, subject) match is a genuine tie and is exempt, so the guard - # runs for every other decision reason (explicit link and no-match alike) - # rather than only when decide_match found no rule. - if decision.reason not in ( - MatchReason.EXACT_IDP_SUBJECT, - MatchReason.VERIFIED_EMAIL, - ): + # Guard: refuse when the accounts only coincide on an unverified email. + # (decide_match already refuses to *call* that a verified match; here we + # produce the precise error for the operator + audit trail.) + if not decision.matched: same_email = ( - bool((survivor.email or "").strip()) - and (survivor.email or "").strip().lower() + (survivor.email or "").strip().lower() == (duplicate.email or "").strip().lower() + and bool(survivor.email) ) if same_email and not have_matching_verified_email(survivor, duplicate): raise UnverifiedEmailMergeError( "refusing merge: accounts share only an UNVERIFIED email" ) - if not decision.matched: raise NoMatchError(decision.detail or "no matching rule satisfied") audit_id = self._audit.new_correlation_id() diff --git a/services/account_unification/tests/test_merge.py b/services/account_unification/tests/test_merge.py index b4b559c..1acfd12 100644 --- a/services/account_unification/tests/test_merge.py +++ b/services/account_unification/tests/test_merge.py @@ -168,16 +168,6 @@ def test_explicit_link_allows_merge_without_shared_signal(service, api): assert result.duplicate_tombstoned -def test_refuse_explicit_merge_on_shared_unverified_email(service, api): - """An explicit link must not launder a shared UNVERIFIED email into a merge.""" - api.create_test_user("survivor", email="jane@corp.com", is_email_verified=False) - api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) - with pytest.raises(UnverifiedEmailMergeError): - service.merge_accounts(_merge(explicit=True)) - # nothing mutated: duplicate not tombstoned. - assert "dup" not in api.deactivated - - def test_refuse_self_merge(service, api): api.create_test_user("same", email="a@x.com", is_email_verified=True) with pytest.raises(SameUserError): From ee085a32da741ed6f74722951a777d60ffdcf56a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 31 Jul 2026 01:46:44 +0000 Subject: [PATCH 05/21] fix(healthcheck): constrain redirect targets to HTTP(S), drop ftp/file handlers The healthcheck validated only the *initial* URL scheme, then used the default `urllib` opener -- which follows redirects and carries an `FTPHandler`. A `http:// -> ftp://` (or `file://`) redirect from the probed endpoint would have been followed by another protocol handler (the code comment even wrongly claimed it could not be). CodeRabbit flagged it (CWE-918 SSRF, Major). Fix: route the probe through a purpose-built opener that (1) carries only HTTP/HTTPS handlers -- no ftp/file/data handler exists to open such a target -- and (2) uses `_HttpOnlyRedirectHandler`, which drops any redirect whose `Location` scheme is not in the http/https allow-list. Both are belt-and- suspenders; either alone fails the ftp redirect closed. `main` now opens via the patchable `_open_health_url` seam. Added `test_healthcheck_opener_drops_non_http_redirect_target` (ftp target dropped, same-scheme redirect kept, no ftp/file/data handler on the opener); the three existing tests re-point to the new seam. Verified (CI parity, services/account_unification): pytest all pass, ruff clean, interrogate 99.4% (>=80 gate). Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01HdCssGnNMhKHNu3TXFstWH --- .../account_unification/app/healthcheck.py | 40 +++++++++++++-- .../tests/test_healthcheck.py | 51 +++++++++++++++---- 2 files changed, 78 insertions(+), 13 deletions(-) diff --git a/services/account_unification/app/healthcheck.py b/services/account_unification/app/healthcheck.py index 713e31b..93902fe 100644 --- a/services/account_unification/app/healthcheck.py +++ b/services/account_unification/app/healthcheck.py @@ -14,6 +14,39 @@ _ALLOWED_SCHEMES = frozenset({"http", "https"}) +class _HttpOnlyRedirectHandler(urllib.request.HTTPRedirectHandler): + """Redirect handler that drops any redirect whose target scheme is not HTTP(S). + + The initial-URL scheme check does not cover a ``Location`` header, so an + ``http:// -> ftp://`` (or ``file://``) redirect would otherwise be followed + by whichever protocol handler the opener carries. + """ + + def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001, D102 + 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) -- no ftp/file/data handlers. + + Even if a redirect target slipped past :class:`_HttpOnlyRedirectHandler`, the + opener has no handler able to open it, so ``ftp://``/``file://`` fail closed. + """ + 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 scheme-restricted, ftp/file-less 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() @@ -23,10 +56,11 @@ def main(url: str = DEFAULT_URL) -> int: print(f"healthcheck failed: unsupported URL scheme {scheme!r}", file=sys.stderr) return 1 try: - # Internal container self-probe; the scheme is allow-listed to http/https - # above, so this urlopen cannot be redirected to a file:// path or other handler. + # Internal container self-probe. Both the initial scheme (above) and any + # redirect target are constrained to http/https, and the opener carries no + # ftp/file handler, so this cannot reach another protocol handler. # nosemgrep: dynamic-urllib-use-detected - with urllib.request.urlopen(url, timeout=5) as response: # noqa: S310 + 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) diff --git a/services/account_unification/tests/test_healthcheck.py b/services/account_unification/tests/test_healthcheck.py index af99c34..60e4ff1 100644 --- a/services/account_unification/tests/test_healthcheck.py +++ b/services/account_unification/tests/test_healthcheck.py @@ -1,6 +1,8 @@ """Container healthcheck command behavior.""" from __future__ import annotations +import urllib.request + from app import healthcheck @@ -19,32 +21,31 @@ def read(self) -> bytes: def test_healthcheck_returns_zero_for_ok_status(monkeypatch, capsys): - def fake_urlopen(url: str, *, timeout: int) -> _Response: + def fake_open(url: str) -> _Response: assert url == "http://service/healthz" - assert timeout == 5 return _Response(b'{"status":"ok"}') - monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen) + monkeypatch.setattr(healthcheck, "_open_health_url", fake_open) assert healthcheck.main("http://service/healthz") == 0 assert capsys.readouterr().out == "ok\n" def test_healthcheck_returns_one_for_non_ok_status(monkeypatch, capsys): - def fake_urlopen(url: str, *, timeout: int) -> _Response: + def fake_open(url: str) -> _Response: return _Response(b'{"status":"starting"}') - monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen) + monkeypatch.setattr(healthcheck, "_open_health_url", fake_open) assert healthcheck.main("http://service/healthz") == 1 assert "not ready" in capsys.readouterr().err def test_healthcheck_returns_one_for_request_error(monkeypatch, capsys): - def fake_urlopen(url: str, *, timeout: int) -> _Response: + def fake_open(url: str) -> _Response: raise OSError("connection refused") - monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fake_urlopen) + monkeypatch.setattr(healthcheck, "_open_health_url", fake_open) assert healthcheck.main("http://service/healthz") == 1 assert "healthcheck failed: connection refused" in capsys.readouterr().err @@ -52,10 +53,40 @@ def fake_urlopen(url: str, *, timeout: int) -> _Response: def test_healthcheck_rejects_non_http_scheme(monkeypatch, capsys): """A non-HTTP(S) URL is rejected before urllib ever opens it.""" - def fail_urlopen(*args: object, **kwargs: object) -> _Response: - raise AssertionError("urlopen must not run for a rejected scheme") + def fail_open(*args: object, **kwargs: object) -> _Response: + raise AssertionError("the opener must not run for a rejected scheme") - monkeypatch.setattr(healthcheck.urllib.request, "urlopen", fail_urlopen) + monkeypatch.setattr(healthcheck, "_open_health_url", fail_open) assert healthcheck.main("file:///etc/passwd") == 1 assert "unsupported URL scheme 'file'" in capsys.readouterr().err + + +def test_healthcheck_opener_drops_non_http_redirect_target(): + """A ``http:// -> ftp://`` redirect is dropped, and no ftp/file handler exists.""" + handler = healthcheck._HttpOnlyRedirectHandler() + dropped = handler.redirect_request( + urllib.request.Request("http://127.0.0.1:8099/healthz"), + None, + 302, + "Found", + {}, + "ftp://127.0.0.1/secret", + ) + assert dropped is None + # A same-scheme redirect is still honoured (returns a Request, not None). + kept = handler.redirect_request( + urllib.request.Request("http://127.0.0.1:8099/healthz"), + None, + 302, + "Found", + {}, + "http://127.0.0.1:8099/ready", + ) + assert kept is not None + # The opener carries no protocol handler that could open ftp/file targets. + opener = healthcheck._build_http_only_opener() + assert not any( + type(h).__name__ in {"FTPHandler", "FileHandler", "DataHandler"} + for h in opener.handlers + ) From 73428599aebe3326efca01d52c19261becef5e9a Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:19:29 +0900 Subject: [PATCH 06/21] test(security): reproduce SCIM merge tombstone race --- .../account_unification/tests/test_scim.py | 121 +++++++++++++++++- 1 file changed, 120 insertions(+), 1 deletion(-) diff --git a/services/account_unification/tests/test_scim.py b/services/account_unification/tests/test_scim.py index cfd57d4..9b4d30d 100644 --- a/services/account_unification/tests/test_scim.py +++ b/services/account_unification/tests/test_scim.py @@ -1,16 +1,83 @@ """Inbound SCIM 2.0 provisioning shim -> Keycloak Admin API.""" from __future__ import annotations +import threading +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager + import pytest from fastapi.testclient import TestClient from app.main import create_app +from app.models import MergeRequest +from app.service import TOMBSTONE_ATTRIBUTE_KEY, UnificationService + +from .mock_keycloak import MockKeycloakAdminApi + + +class _TestUserOperationLocks: + """Small keyed lock manager used to prove cross-path serialization.""" + + def __init__(self) -> None: + self._guard = threading.Lock() + self._locks: dict[str, threading.RLock] = {} + + @contextmanager + def hold(self, *user_ids: str): + """Hold all requested user locks in stable order.""" + ordered_ids = sorted(set(user_ids)) + with self._guard: + locks = [self._locks.setdefault(user_id, threading.RLock()) for user_id in ordered_ids] + for lock in locks: + lock.acquire() + try: + yield + finally: + for lock in reversed(locks): + lock.release() + + +class _BlockingReplaceApi(MockKeycloakAdminApi): + """Pause SCIM replacement after its tombstone check to expose the race.""" + + def __init__(self) -> None: + super().__init__() + self.replace_started = threading.Event() + self.allow_replace = threading.Event() + self.tombstone_started = threading.Event() + + def replace_user(self, user_id, user) -> None: + """Wait until the test permits the full Keycloak representation PUT.""" + self.replace_started.set() + if not self.allow_replace.wait(timeout=5): + raise AssertionError("test did not release the blocked SCIM replacement") + super().replace_user(user_id, user) + # Keycloak's full user-representation PUT can remove attributes omitted + # from the payload and re-enable the account via SCIM's active=true. + self.attributes = { + attribute: value + for attribute, value in self.attributes.items() + if attribute[0] != user_id + } + self.deactivated.discard(user_id) + + def set_user_attribute(self, user_id: str, key: str, value: str) -> None: + """Signal when merge begins writing the duplicate tombstone.""" + if user_id == "dup" and key == TOMBSTONE_ATTRIBUTE_KEY: + self.tombstone_started.set() + super().set_user_attribute(user_id, key, value) + + +@pytest.fixture +def user_operation_locks(): + return _TestUserOperationLocks() @pytest.fixture -def client(api): +def client(api, user_operation_locks): app = create_app(wire=False) app.state.keycloak_api = api + app.state.user_operation_locks = user_operation_locks with TestClient(app) as test_client: yield test_client @@ -104,6 +171,58 @@ def test_scim_replace_refuses_to_resurrect_a_tombstoned_duplicate(client, api): assert api.get_user_attribute(dup_id, "merged_into_user_id") == "survivor-id" +def test_scim_replace_is_serialized_with_concurrent_merge(config, audit): + """A concurrent merge cannot slip between SCIM's tombstone check and PUT.""" + api = _BlockingReplaceApi() + locks = _TestUserOperationLocks() + service = UnificationService( + api, + audit, + config, + user_operation_locks=locks, + ) + app = create_app(wire=False) + app.state.keycloak_api = api + app.state.user_operation_locks = locks + app.state.unification_service = service + api.create_test_user( + "survivor", email="jane@corp.com", is_email_verified=True + ) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=True) + merge_invoked = threading.Event() + + def run_merge(): + merge_invoked.set() + return service.merge_accounts( + MergeRequest( + survivor_user_id="survivor", + duplicate_user_id="dup", + actor="admin@cwl", + ) + ) + + with TestClient(app) as test_client, ThreadPoolExecutor(max_workers=2) as executor: + scim_future = executor.submit( + test_client.put, + "/scim/v2/Users/dup", + json=_scim_user(username="dup"), + ) + assert api.replace_started.wait(timeout=2) + merge_future = executor.submit(run_merge) + assert merge_invoked.wait(timeout=2) + + merge_was_serialized = not api.tombstone_started.wait(timeout=0.25) + api.allow_replace.set() + response = scim_future.result(timeout=5) + merge_result = merge_future.result(timeout=5) + + assert merge_was_serialized + assert response.status_code == 200 + assert merge_result.duplicate_tombstoned is True + assert api.get_user("dup").state == "disabled" + assert api.get_user_attribute("dup", TOMBSTONE_ATTRIBUTE_KEY) == "survivor" + + def test_scim_patch_deactivates_user(client, api): created = client.post("/scim/v2/Users", json=_scim_user()).json() response = client.patch( From 06abe3aae952dc47152f73392072f3db874fccae Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:24:07 +0900 Subject: [PATCH 07/21] fix(security): add shared user-operation lock abstraction --- .../account_unification/app/user_locks.py | 141 ++++++++++++++++++ 1 file changed, 141 insertions(+) create mode 100644 services/account_unification/app/user_locks.py diff --git a/services/account_unification/app/user_locks.py b/services/account_unification/app/user_locks.py new file mode 100644 index 0000000..c9a3d11 --- /dev/null +++ b/services/account_unification/app/user_locks.py @@ -0,0 +1,141 @@ +"""Cross-path serialization for mutations of Keycloak user records. + +SCIM replacement and account merge both write complete or partial Keycloak user +representations. They must share one lock boundary so a merge cannot tombstone a +duplicate between SCIM's tombstone check and its replacement PUT. + +The standalone runtime uses :class:`SqliteUserOperationLocks`, backed by a +dedicated sidecar SQLite database. ``BEGIN IMMEDIATE`` provides a crash-safe, +cross-process mutex for every service worker sharing that database file. The +current implementation intentionally serializes all user mutations rather than +risking a multi-key deadlock; the public interface remains user-ID keyed so a +future Postgres advisory-lock implementation can safely increase concurrency. +""" +from __future__ import annotations + +import sqlite3 +import threading +from contextlib import contextmanager +from typing import ContextManager, Iterator, Protocol + + +class UserOperationLockTimeout(RuntimeError): + """Raised when a shared user-operation lock cannot be acquired in time.""" + + +class UserOperationLocks(Protocol): + """Serialize mutations that involve one or more Keycloak user IDs.""" + + def hold(self, *user_ids: str) -> ContextManager[None]: + """Return a context manager holding the requested user-operation locks.""" + ... + + +def _normalise_user_ids(user_ids: tuple[str, ...]) -> tuple[str, ...]: + """Return unique, non-empty user IDs in deterministic acquisition order.""" + ordered = tuple(sorted(set(user_ids))) + if not ordered or any(not user_id for user_id in ordered): + raise ValueError("at least one non-empty user ID is required") + return ordered + + +class InMemoryUserOperationLocks: + """Process-local keyed lock manager for tests and explicit single-worker use.""" + + def __init__(self) -> None: + """Create an empty keyed re-entrant lock registry.""" + self._registry_guard = threading.Lock() + self._locks: dict[str, threading.RLock] = {} + + @contextmanager + def hold(self, *user_ids: str) -> Iterator[None]: + """Hold all requested user locks in stable order to avoid deadlocks.""" + ordered_ids = _normalise_user_ids(user_ids) + with self._registry_guard: + locks = [ + self._locks.setdefault(user_id, threading.RLock()) + for user_id in ordered_ids + ] + for lock in locks: + lock.acquire() + try: + yield + finally: + for lock in reversed(locks): + lock.release() + + +class SqliteUserOperationLocks: + """Cross-process mutex backed by a dedicated SQLite sidecar database. + + SQLite permits only one writer holding a ``BEGIN IMMEDIATE`` transaction. + Every manager instance pointed at the same database file therefore shares a + crash-safe mutex: process termination closes the connection and releases the + lock automatically. This is deliberately coarser than the user-ID-keyed + protocol, but it fully serializes the SCIM and merge critical sections for + the supported SQLite deployment without introducing a lease-expiry race. + """ + + _SCHEMA = """ + CREATE TABLE IF NOT EXISTS user_operation_lock_state ( + lock_name TEXT PRIMARY KEY, + requested_user_ids TEXT NOT NULL + ); + """ + + def __init__(self, database_path: str, *, timeout_seconds: float = 10.0) -> None: + """Create a manager using ``database_path`` and an acquisition timeout.""" + if not database_path: + raise ValueError("database_path is required") + if timeout_seconds <= 0: + raise ValueError("timeout_seconds must be positive") + self._database_path = database_path + self._timeout_seconds = timeout_seconds + self._initialize() + + def _connect(self) -> sqlite3.Connection: + """Open one autocommit connection configured with the lock timeout.""" + connection = sqlite3.connect( + self._database_path, + timeout=self._timeout_seconds, + isolation_level=None, + ) + connection.execute( + f"PRAGMA busy_timeout = {int(self._timeout_seconds * 1000)}" + ) + return connection + + def _initialize(self) -> None: + """Create the sidecar schema before requests begin competing for it.""" + connection = self._connect() + try: + connection.execute(self._SCHEMA) + finally: + connection.close() + + @contextmanager + def hold(self, *user_ids: str) -> Iterator[None]: + """Hold the shared SQLite mutex for the complete user mutation.""" + ordered_ids = _normalise_user_ids(user_ids) + connection = self._connect() + try: + try: + connection.execute("BEGIN IMMEDIATE") + except sqlite3.OperationalError as exc: + if "locked" in str(exc).lower(): + raise UserOperationLockTimeout( + "timed out waiting for another user mutation to finish" + ) from exc + raise + connection.execute( + "INSERT INTO user_operation_lock_state " + "(lock_name, requested_user_ids) VALUES ('global', ?) " + "ON CONFLICT(lock_name) DO UPDATE SET " + "requested_user_ids = excluded.requested_user_ids", + (",".join(ordered_ids),), + ) + yield + finally: + if connection.in_transaction: + connection.rollback() + connection.close() From 9375b4e70dca54eac9c2efc91a65a9ad05fd47a0 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:24:56 +0900 Subject: [PATCH 08/21] fix(security): serialize merge mutations with SCIM --- services/account_unification/app/service.py | 17 ++++++++++++++++- 1 file changed, 16 insertions(+), 1 deletion(-) diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py index 3354c6f..165d7e5 100644 --- a/services/account_unification/app/service.py +++ b/services/account_unification/app/service.py @@ -32,6 +32,7 @@ MergeResult, UserAccount, ) +from .user_locks import UserOperationLocks # Keycloak user attribute stamped on a tombstoned duplicate (two-word snake_case). TOMBSTONE_ATTRIBUTE_KEY = "merged_into_user_id" @@ -46,11 +47,13 @@ def __init__( api: AdminApi, audit: AuditLogger, config: ServiceConfig, + user_operation_locks: UserOperationLocks, ) -> None: - """Create a service around admin API, audit, and config dependencies.""" + """Create a service around admin, audit, config, and lock dependencies.""" self._api = api self._audit = audit self._config = config + self._user_operation_locks = user_operation_locks # -- (a) inspect identities ------------------------------------------- def get_account(self, user_id: str) -> UserAccount: @@ -70,6 +73,18 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult: if request.survivor_user_id == request.duplicate_user_id: raise SameUserError("survivor and duplicate are the same account") + # The complete merge, including the final tombstone write, shares the + # same duplicate-user lock as SCIM replacement. This closes the TOCTOU + # window where SCIM could pass its tombstone check, then overwrite a + # concurrently-created tombstone with an active user representation. + with self._user_operation_locks.hold( + request.survivor_user_id, + request.duplicate_user_id, + ): + return self._merge_accounts_locked(request) + + def _merge_accounts_locked(self, request: MergeRequest) -> MergeResult: + """Perform a merge while both participating user IDs are serialized.""" survivor = self._load_user(request.survivor_user_id) duplicate = self._load_user(request.duplicate_user_id) survivor.federated_identities = self._api.list_federated_identities( From ce436a2dbddde8e8fb583b365355370b12e11ae4 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:25:43 +0900 Subject: [PATCH 09/21] fix(security): make SCIM tombstone check and PUT atomic --- services/account_unification/app/scim.py | 54 ++++++++++++++++-------- 1 file changed, 37 insertions(+), 17 deletions(-) diff --git a/services/account_unification/app/scim.py b/services/account_unification/app/scim.py index f76f6ea..4c0b9fb 100644 --- a/services/account_unification/app/scim.py +++ b/services/account_unification/app/scim.py @@ -21,6 +21,10 @@ 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" @@ -39,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( @@ -187,27 +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 - # A merged-away duplicate is tombstoned (disabled + a merged_into_user_id - # pointer) so it can never authenticate again. SCIM PUT is the only - # reactivation vector -- create guards uniqueness, and patch/delete only - # ever disable -- and a live Keycloak PUT would also overwrite the whole - # user representation, wiping the survivor pointer. Refuse it outright so a - # routine upstream full-sync cannot resurrect a decommissioned identity. - if provisioner.get_user_attribute(user_id, TOMBSTONE_ATTRIBUTE_KEY): + 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( - 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) - return _scim_response(_to_scim_resource(provisioner.get_user(user_id))) + 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}") From ac3b05620e5fcabacf1f6da7cd06513515d3f2e6 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:26:03 +0900 Subject: [PATCH 10/21] fix(security): wire one shared mutation lock manager --- services/account_unification/app/main.py | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/services/account_unification/app/main.py b/services/account_unification/app/main.py index ddff44a..355bc09 100644 --- a/services/account_unification/app/main.py +++ b/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 From 5220bab330e11dff8dd4d73b2944ac3933ba9582 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:26:19 +0900 Subject: [PATCH 11/21] test: inject shared user-operation locks into service fixtures --- services/account_unification/tests/conftest.py | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/services/account_unification/tests/conftest.py b/services/account_unification/tests/conftest.py index 3deba76..64d7ecf 100644 --- a/services/account_unification/tests/conftest.py +++ b/services/account_unification/tests/conftest.py @@ -11,6 +11,7 @@ from app.audit import AuditLogger, InMemoryAuditSink # noqa: E402 from app.config import ServiceConfig # noqa: E402 from app.service import UnificationService # noqa: E402 +from app.user_locks import InMemoryUserOperationLocks # noqa: E402 from .mock_keycloak import MockKeycloakAdminApi # noqa: E402 @@ -42,8 +43,16 @@ def config() -> ServiceConfig: ) +@pytest.fixture +def user_operation_locks() -> InMemoryUserOperationLocks: + return InMemoryUserOperationLocks() + + @pytest.fixture def service( - api: MockKeycloakAdminApi, audit: AuditLogger, config: ServiceConfig + api: MockKeycloakAdminApi, + audit: AuditLogger, + config: ServiceConfig, + user_operation_locks: InMemoryUserOperationLocks, ) -> UnificationService: - return UnificationService(api, audit, config) + return UnificationService(api, audit, config, user_operation_locks) From 3c4d09889994c087f2b53879b729ab158984a810 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:26:40 +0900 Subject: [PATCH 12/21] fix(api): surface user-operation lock contention as retryable --- services/account_unification/app/api.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/services/account_unification/app/api.py b/services/account_unification/app/api.py index 3e8119b..fe042a7 100644 --- a/services/account_unification/app/api.py +++ b/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"]) From 68ad2a1a42d9ee836f17ee9ea013be13411e9807 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:27:04 +0900 Subject: [PATCH 13/21] test(security): verify shared lock serialization and timeout --- .../tests/test_user_locks.py | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) create mode 100644 services/account_unification/tests/test_user_locks.py diff --git a/services/account_unification/tests/test_user_locks.py b/services/account_unification/tests/test_user_locks.py new file mode 100644 index 0000000..97f0271 --- /dev/null +++ b/services/account_unification/tests/test_user_locks.py @@ -0,0 +1,71 @@ +"""Shared user-operation serialization for merge and SCIM writes.""" +from __future__ import annotations + +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from app.user_locks import ( + InMemoryUserOperationLocks, + SqliteUserOperationLocks, + UserOperationLockTimeout, +) + + +def _assert_overlapping_operation_waits(first_manager, second_manager) -> None: + first_entered = threading.Event() + release_first = threading.Event() + second_entered = threading.Event() + + def hold_first() -> None: + with first_manager.hold("survivor", "dup"): + first_entered.set() + assert release_first.wait(timeout=5) + + def hold_second() -> None: + assert first_entered.wait(timeout=5) + with second_manager.hold("dup"): + second_entered.set() + + with ThreadPoolExecutor(max_workers=2) as executor: + first_future = executor.submit(hold_first) + second_future = executor.submit(hold_second) + assert first_entered.wait(timeout=2) + was_serialized = not second_entered.wait(timeout=0.25) + release_first.set() + first_future.result(timeout=5) + second_future.result(timeout=5) + + assert was_serialized + assert second_entered.is_set() + + +def test_in_memory_locks_serialize_overlapping_user_ids(): + manager = InMemoryUserOperationLocks() + _assert_overlapping_operation_waits(manager, manager) + + +def test_sqlite_locks_serialize_distinct_manager_instances(tmp_path): + database_path = str(tmp_path / "user-operation-locks.sqlite3") + first_manager = SqliteUserOperationLocks(database_path) + second_manager = SqliteUserOperationLocks(database_path) + _assert_overlapping_operation_waits(first_manager, second_manager) + + +def test_sqlite_lock_timeout_is_explicit_and_retryable(tmp_path): + database_path = str(tmp_path / "user-operation-locks.sqlite3") + first_manager = SqliteUserOperationLocks(database_path) + impatient_manager = SqliteUserOperationLocks(database_path, timeout_seconds=0.05) + + with first_manager.hold("dup"): + with pytest.raises(UserOperationLockTimeout): + with impatient_manager.hold("dup"): + pytest.fail("contending operation unexpectedly acquired the lock") + + +def test_lock_manager_rejects_empty_user_ids(): + manager = InMemoryUserOperationLocks() + with pytest.raises(ValueError): + with manager.hold(""): + pytest.fail("empty user ID unexpectedly acquired a lock") From cbd283ca8322394674b2132742b5f7d78e024a6f Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:27:32 +0900 Subject: [PATCH 14/21] docs: define SCIM and merge serialization invariant --- docs/merge-unification-flow.md | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/docs/merge-unification-flow.md b/docs/merge-unification-flow.md index a776a27..35a88f9 100644 --- a/docs/merge-unification-flow.md +++ b/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 From 1051cebf520afb321134e2e2db313201a5027056 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:49:02 +0900 Subject: [PATCH 15/21] fix(security): preserve the unverified-email hard rule under serialization --- services/account_unification/app/service.py | 85 ++++++++++++++------- 1 file changed, 58 insertions(+), 27 deletions(-) diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py index 165d7e5..16c42d8 100644 --- a/services/account_unification/app/service.py +++ b/services/account_unification/app/service.py @@ -27,6 +27,7 @@ from .matching import decide_match, have_matching_verified_email from .models import ( FederatedIdentity, + MatchReason, MergeConflict, MergeRequest, MergeResult, @@ -100,19 +101,28 @@ def _merge_accounts_locked(self, request: MergeRequest) -> MergeResult: decision = decide_match( survivor, duplicate, explicit_link=request.explicit_link ) - # Guard: refuse when the accounts only coincide on an unverified email. - # (decide_match already refuses to *call* that a verified match; here we - # produce the precise error for the operator + audit trail.) - if not decision.matched: - same_email = ( - (survivor.email or "").strip().lower() - == (duplicate.email or "").strip().lower() - and bool(survivor.email) + # Hard rule (enforced here per docs/merge-unification-flow.md and the + # MergeRequest.explicit_link contract): never merge when the only tie is + # an UNVERIFIED email. An unverified address is attacker-registerable, + # so even an operator's explicit_link assertion must not promote a shared + # unverified email into a merge — only a strong tie (exact idp subject or + # a mutually verified email) justifies it. This guard therefore runs + # regardless of decision.matched, catching the explicit_link path too. + shares_unverified_email = ( + bool((survivor.email or "").strip()) + and (survivor.email or "").strip().lower() + == (duplicate.email or "").strip().lower() + and not have_matching_verified_email(survivor, duplicate) + ) + strong_tie = decision.reason in ( + MatchReason.EXACT_IDP_SUBJECT, + MatchReason.VERIFIED_EMAIL, + ) + if shares_unverified_email and not strong_tie: + raise UnverifiedEmailMergeError( + "refusing merge: accounts share only an UNVERIFIED email" ) - if same_email and not have_matching_verified_email(survivor, duplicate): - raise UnverifiedEmailMergeError( - "refusing merge: accounts share only an UNVERIFIED email" - ) + if not decision.matched: raise NoMatchError(decision.detail or "no matching rule satisfied") audit_id = self._audit.new_correlation_id() @@ -196,8 +206,10 @@ def _move_federated_identities( duplicate.user_id, link.identity_provider ) self._audit.emit( - audit_id=audit_id, event_type="federated_identity_conflict", - actor=actor, survivor_user_id=survivor.user_id, + audit_id=audit_id, + event_type="federated_identity_conflict", + actor=actor, + survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, payload={"identifier": identifier, "resolution": "survivor_wins"}, ) @@ -208,8 +220,11 @@ def _move_federated_identities( ) moved.append(identifier) self._audit.emit( - audit_id=audit_id, event_type="federated_identity_moved", actor=actor, - survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, + audit_id=audit_id, + event_type="federated_identity_moved", + actor=actor, + survivor_user_id=survivor.user_id, + duplicate_user_id=duplicate.user_id, payload={"identifier": identifier}, ) return moved @@ -234,7 +249,9 @@ def _move_role_mappings( # survivor-wins: survivor already has it; drop the duplicate's. self._api.remove_role_mapping(duplicate.user_id, role) self._audit.emit( - audit_id=audit_id, event_type="role_mapping_conflict", actor=actor, + audit_id=audit_id, + event_type="role_mapping_conflict", + actor=actor, survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, payload={"identifier": identifier, "resolution": "survivor_wins"}, @@ -244,10 +261,16 @@ def _move_role_mappings( self._api.remove_role_mapping(duplicate.user_id, role) moved.append(identifier) self._audit.emit( - audit_id=audit_id, event_type="role_mapping_moved", actor=actor, - survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, - payload={"identifier": identifier, "role_name": role.role_name, - "client_id": role.client_id}, + audit_id=audit_id, + event_type="role_mapping_moved", + actor=actor, + survivor_user_id=survivor.user_id, + duplicate_user_id=duplicate.user_id, + payload={ + "identifier": identifier, + "role_name": role.role_name, + "client_id": role.client_id, + }, ) return moved @@ -267,8 +290,10 @@ def _move_group_memberships( ) self._api.remove_group_membership(duplicate.user_id, group) self._audit.emit( - audit_id=audit_id, event_type="group_membership_conflict", - actor=actor, survivor_user_id=survivor.user_id, + audit_id=audit_id, + event_type="group_membership_conflict", + actor=actor, + survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, payload={"identifier": identifier, "resolution": "survivor_wins"}, ) @@ -277,8 +302,11 @@ def _move_group_memberships( self._api.remove_group_membership(duplicate.user_id, group) moved.append(identifier) self._audit.emit( - audit_id=audit_id, event_type="group_membership_moved", actor=actor, - survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, + audit_id=audit_id, + event_type="group_membership_moved", + actor=actor, + survivor_user_id=survivor.user_id, + duplicate_user_id=duplicate.user_id, payload={"identifier": identifier}, ) return moved @@ -292,8 +320,11 @@ def _tombstone(self, duplicate, survivor, audit_id, actor) -> None: ) self._api.deactivate_user(duplicate.user_id) self._audit.emit( - audit_id=audit_id, event_type="duplicate_tombstoned", actor=actor, - survivor_user_id=survivor.user_id, duplicate_user_id=duplicate.user_id, + audit_id=audit_id, + event_type="duplicate_tombstoned", + actor=actor, + survivor_user_id=survivor.user_id, + duplicate_user_id=duplicate.user_id, payload={"attribute_key": TOMBSTONE_ATTRIBUTE_KEY}, ) From 5dbba0a3624975bc10bff60fcbf2853bd4f84728 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:49:36 +0900 Subject: [PATCH 16/21] fix(security): harden healthcheck redirects and document the handler --- .../account_unification/app/healthcheck.py | 27 ++++++------------- 1 file changed, 8 insertions(+), 19 deletions(-) diff --git a/services/account_unification/app/healthcheck.py b/services/account_unification/app/healthcheck.py index 93902fe..22a1148 100644 --- a/services/account_unification/app/healthcheck.py +++ b/services/account_unification/app/healthcheck.py @@ -15,25 +15,17 @@ class _HttpOnlyRedirectHandler(urllib.request.HTTPRedirectHandler): - """Redirect handler that drops any redirect whose target scheme is not HTTP(S). + """Drop redirects whose target scheme is not HTTP(S).""" - The initial-URL scheme check does not cover a ``Location`` header, so an - ``http:// -> ftp://`` (or ``file://``) redirect would otherwise be followed - by whichever protocol handler the opener carries. - """ - - def redirect_request(self, req, fp, code, msg, headers, newurl): # noqa: ANN001, D102 + 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) -- no ftp/file/data handlers. - - Even if a redirect target slipped past :class:`_HttpOnlyRedirectHandler`, the - opener has no handler able to open it, so ``ftp://``/``file://`` fail closed. - """ + """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()) @@ -43,7 +35,7 @@ def _build_http_only_opener() -> urllib.request.OpenerDirector: def _open_health_url(url: str): # noqa: ANN202 - """Open an HTTP(S) health URL through the scheme-restricted, ftp/file-less opener.""" + """Open an HTTP(S) health URL through the restricted opener.""" return _build_http_only_opener().open(url, timeout=5) @@ -51,15 +43,12 @@ 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: - # Reject non-HTTP(S) schemes so a stray value can never coerce urlopen into - # reading a local ``file://`` path or reaching another protocol handler. print(f"healthcheck failed: unsupported URL scheme {scheme!r}", file=sys.stderr) return 1 try: - # Internal container self-probe. Both the initial scheme (above) and any - # redirect target are constrained to http/https, and the opener carries no - # ftp/file handler, so this cannot reach another protocol handler. - # nosemgrep: dynamic-urllib-use-detected + # 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 From 624e0e5d254baa95829d0ed6c6309f1ebbbd3144 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:52:07 +0900 Subject: [PATCH 17/21] test(security): retain explicit-link unverified-email regressions --- .../account_unification/tests/test_merge.py | 129 +++++++++++++----- 1 file changed, 93 insertions(+), 36 deletions(-) diff --git a/services/account_unification/tests/test_merge.py b/services/account_unification/tests/test_merge.py index 1acfd12..de00138 100644 --- a/services/account_unification/tests/test_merge.py +++ b/services/account_unification/tests/test_merge.py @@ -35,9 +35,13 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api): email="jane@corp.com", is_email_verified=True, federated_identities=[ - FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp") + FederatedIdentity( + identity_provider="employer-adfs", external_user_id="jane@corp" + ) + ], + role_mappings=[ + RoleMapping(role_id="r-s", role_name="viewer", client_id="naruon") ], - role_mappings=[RoleMapping(role_id="r-s", role_name="viewer", client_id="naruon")], group_memberships=[GroupMembership(group_id="g-org", group_path="/org")], ) api.create_test_user( @@ -45,9 +49,13 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api): email="jane@corp.com", is_email_verified=True, federated_identities=[ - FederatedIdentity(identity_provider="google", external_user_id="jane@gmail") + FederatedIdentity( + identity_provider="google", external_user_id="jane@gmail" + ) + ], + role_mappings=[ + RoleMapping(role_id="r-d", role_name="editor", client_id="clearfolio") ], - role_mappings=[RoleMapping(role_id="r-d", role_name="editor", client_id="clearfolio")], group_memberships=[GroupMembership(group_id="g-proj", group_path="/pg-erd")], ) @@ -55,93 +63,127 @@ def test_merge_moves_links_roles_groups_and_tombstones(service, api): assert result.match_reason is MatchReason.VERIFIED_EMAIL assert result.duplicate_tombstoned is True - # survivor gained the duplicate's external identity... - survivor_idps = {f.identity_provider for f in api.list_federated_identities("survivor")} + survivor_idps = { + identity.identity_provider + for identity in api.list_federated_identities("survivor") + } assert survivor_idps == {"employer-adfs", "google"} - # ...its client role... - survivor_roles = {r.role_name for r in api.list_role_mappings("survivor")} + survivor_roles = {role.role_name for role in api.list_role_mappings("survivor")} assert survivor_roles == {"viewer", "editor"} - # ...and its group. - survivor_groups = {g.group_id for g in api.list_group_memberships("survivor")} + survivor_groups = { + group.group_id for group in api.list_group_memberships("survivor") + } assert survivor_groups == {"g-org", "g-proj"} - # duplicate is emptied + tombstoned + disabled. assert api.list_federated_identities("dup") == [] assert "dup" in api.deactivated assert api.attributes[("dup", TOMBSTONE_ATTRIBUTE_KEY)] == "survivor" def test_merge_by_exact_idp_subject(service, api): - shared = FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp") + shared = FederatedIdentity( + identity_provider="employer-adfs", external_user_id="jane@corp" + ) api.create_test_user("survivor", email="a@x.com", federated_identities=[shared]) api.create_test_user("dup", email="b@y.com", federated_identities=[shared]) result = service.merge_accounts(_merge()) assert result.match_reason is MatchReason.EXACT_IDP_SUBJECT - # shared link stays on survivor exactly once (survivor-wins conflict). - assert [f.external_user_id for f in api.list_federated_identities("survivor")] == ["jane@corp"] - assert any(c.kind == "federated_identity" for c in result.conflicts) + assert [ + identity.external_user_id + for identity in api.list_federated_identities("survivor") + ] == ["jane@corp"] + assert any(conflict.kind == "federated_identity" for conflict in result.conflicts) def test_federated_identity_provider_conflict_is_survivor_wins(service, api): - # Same provider alias, different external subject: Keycloak allows only one - # link per provider, so survivor-wins keeps the survivor's. api.create_test_user( - "survivor", email="j@x.com", is_email_verified=True, + "survivor", + email="j@x.com", + is_email_verified=True, federated_identities=[ - FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane@corp") + FederatedIdentity( + identity_provider="employer-adfs", external_user_id="jane@corp" + ) ], ) api.create_test_user( - "dup", email="j@x.com", is_email_verified=True, + "dup", + email="j@x.com", + is_email_verified=True, federated_identities=[ - FederatedIdentity(identity_provider="employer-adfs", external_user_id="jane2@corp") + FederatedIdentity( + identity_provider="employer-adfs", external_user_id="jane2@corp" + ) ], ) result = service.merge_accounts(_merge()) survivor_links = api.list_federated_identities("survivor") - assert [f.external_user_id for f in survivor_links] == ["jane@corp"] - assert any(c.kind == "federated_identity" for c in result.conflicts) + assert [identity.external_user_id for identity in survivor_links] == ["jane@corp"] + assert any(conflict.kind == "federated_identity" for conflict in result.conflicts) def test_role_conflict_is_survivor_wins(service, api): api.create_test_user( - "survivor", email="j@x.com", is_email_verified=True, - role_mappings=[RoleMapping(role_id="r-s", role_name="admin", client_id="naruon")], + "survivor", + email="j@x.com", + is_email_verified=True, + role_mappings=[ + RoleMapping(role_id="r-s", role_name="admin", client_id="naruon") + ], ) api.create_test_user( - "dup", email="j@x.com", is_email_verified=True, - role_mappings=[RoleMapping(role_id="r-d", role_name="admin", client_id="naruon")], + "dup", + email="j@x.com", + is_email_verified=True, + role_mappings=[ + RoleMapping(role_id="r-d", role_name="admin", client_id="naruon") + ], ) result = service.merge_accounts(_merge()) survivor_roles = api.list_role_mappings("survivor") - # only the survivor's admin role on naruon survives. assert len(survivor_roles) == 1 assert survivor_roles[0].role_id == "r-s" - assert any(c.kind == "role_mapping" and c.resolution == "survivor_wins" for c in result.conflicts) + assert any( + conflict.kind == "role_mapping" + and conflict.resolution == "survivor_wins" + for conflict in result.conflicts + ) def test_realm_role_moves(service, api): api.create_test_user("survivor", email="j@x.com", is_email_verified=True) api.create_test_user( - "dup", email="j@x.com", is_email_verified=True, - role_mappings=[RoleMapping(role_id="r-realm", role_name="ecosystem-user", client_id=None)], + "dup", + email="j@x.com", + is_email_verified=True, + role_mappings=[ + RoleMapping( + role_id="r-realm", role_name="ecosystem-user", client_id=None + ) + ], ) result = service.merge_accounts(_merge()) assert "realm:ecosystem-user" in result.moved_role_mappings - assert any(r.client_id is None for r in api.list_role_mappings("survivor")) + assert any( + role.client_id is None for role in api.list_role_mappings("survivor") + ) def test_group_conflict_is_survivor_wins(service, api): api.create_test_user( - "survivor", email="j@x.com", is_email_verified=True, + "survivor", + email="j@x.com", + is_email_verified=True, group_memberships=[GroupMembership(group_id="g1", group_path="/owners")], ) api.create_test_user( - "dup", email="j@x.com", is_email_verified=True, + "dup", + email="j@x.com", + is_email_verified=True, group_memberships=[GroupMembership(group_id="g1", group_path="/owners")], ) result = service.merge_accounts(_merge()) assert len(api.list_group_memberships("survivor")) == 1 - assert any(c.kind == "group_membership" for c in result.conflicts) + assert any(conflict.kind == "group_membership" for conflict in result.conflicts) def test_refuse_merge_on_unverified_email(service, api): @@ -149,7 +191,6 @@ def test_refuse_merge_on_unverified_email(service, api): api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) with pytest.raises(UnverifiedEmailMergeError): service.merge_accounts(_merge()) - # nothing mutated: duplicate not tombstoned. assert "dup" not in api.deactivated @@ -168,6 +209,22 @@ def test_explicit_link_allows_merge_without_shared_signal(service, api): assert result.duplicate_tombstoned +def test_explicit_link_cannot_override_shared_unverified_email(service, api): + api.create_test_user("survivor", email="jane@corp.com", is_email_verified=False) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) + with pytest.raises(UnverifiedEmailMergeError): + service.merge_accounts(_merge(explicit=True)) + assert "dup" not in api.deactivated + + +def test_explicit_link_cannot_override_case_variant_unverified_email(service, api): + api.create_test_user("survivor", email="Jane@Corp.com", is_email_verified=True) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) + with pytest.raises(UnverifiedEmailMergeError): + service.merge_accounts(_merge(explicit=True)) + assert "dup" not in api.deactivated + + def test_refuse_self_merge(service, api): api.create_test_user("same", email="a@x.com", is_email_verified=True) with pytest.raises(SameUserError): From 4e2cf589a72440103540338c5132b368972f3da1 Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 10:59:06 +0900 Subject: [PATCH 18/21] fix(account-unification): enforce unverified-email hard rule (#34) Reject explicit-link merges when the only shared signal is an unverified email, including case-normalized variants, while preserving exact-subject and mutually verified-email matches. --- services/account_unification/app/service.py | 34 ++++++++++++------- .../account_unification/tests/test_merge.py | 24 +++++++++++++ 2 files changed, 46 insertions(+), 12 deletions(-) diff --git a/services/account_unification/app/service.py b/services/account_unification/app/service.py index 3354c6f..0ecacb1 100644 --- a/services/account_unification/app/service.py +++ b/services/account_unification/app/service.py @@ -27,6 +27,7 @@ from .matching import decide_match, have_matching_verified_email from .models import ( FederatedIdentity, + MatchReason, MergeConflict, MergeRequest, MergeResult, @@ -85,19 +86,28 @@ def merge_accounts(self, request: MergeRequest) -> MergeResult: decision = decide_match( survivor, duplicate, explicit_link=request.explicit_link ) - # Guard: refuse when the accounts only coincide on an unverified email. - # (decide_match already refuses to *call* that a verified match; here we - # produce the precise error for the operator + audit trail.) - if not decision.matched: - same_email = ( - (survivor.email or "").strip().lower() - == (duplicate.email or "").strip().lower() - and bool(survivor.email) + # Hard rule (enforced here per docs/merge-unification-flow.md and the + # MergeRequest.explicit_link contract): never merge when the only tie is + # an UNVERIFIED email. An unverified address is attacker-registerable, + # so even an operator's explicit_link assertion must not promote a shared + # unverified email into a merge — only a strong tie (exact idp subject or + # a mutually verified email) justifies it. This guard therefore runs + # regardless of decision.matched, catching the explicit_link path too. + shares_unverified_email = ( + bool((survivor.email or "").strip()) + and (survivor.email or "").strip().lower() + == (duplicate.email or "").strip().lower() + and not have_matching_verified_email(survivor, duplicate) + ) + strong_tie = decision.reason in ( + MatchReason.EXACT_IDP_SUBJECT, + MatchReason.VERIFIED_EMAIL, + ) + if shares_unverified_email and not strong_tie: + raise UnverifiedEmailMergeError( + "refusing merge: accounts share only an UNVERIFIED email" ) - if same_email and not have_matching_verified_email(survivor, duplicate): - raise UnverifiedEmailMergeError( - "refusing merge: accounts share only an UNVERIFIED email" - ) + if not decision.matched: raise NoMatchError(decision.detail or "no matching rule satisfied") audit_id = self._audit.new_correlation_id() diff --git a/services/account_unification/tests/test_merge.py b/services/account_unification/tests/test_merge.py index 1acfd12..ebcc1d2 100644 --- a/services/account_unification/tests/test_merge.py +++ b/services/account_unification/tests/test_merge.py @@ -168,6 +168,30 @@ def test_explicit_link_allows_merge_without_shared_signal(service, api): assert result.duplicate_tombstoned +def test_explicit_link_cannot_override_shared_unverified_email(service, api): + # Hard rule (CLAUDE.md, docs/merge-unification-flow.md, and the + # MergeRequest.explicit_link contract): "Even so, the service refuses if the + # only tie is an UNVERIFIED email." An unverified address is + # attacker-registerable, so flipping explicit_link=True must NOT promote a + # shared unverified email into a merge. + api.create_test_user("survivor", email="jane@corp.com", is_email_verified=False) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) + with pytest.raises(UnverifiedEmailMergeError): + service.merge_accounts(_merge(explicit=True)) + # nothing mutated: duplicate not tombstoned. + assert "dup" not in api.deactivated + + +def test_explicit_link_cannot_override_case_variant_unverified_email(service, api): + # Same rule, exercised through case-insensitive email normalization: one + # side verified is not enough — both must be verified for an email tie. + api.create_test_user("survivor", email="Jane@Corp.com", is_email_verified=True) + api.create_test_user("dup", email="jane@corp.com", is_email_verified=False) + with pytest.raises(UnverifiedEmailMergeError): + service.merge_accounts(_merge(explicit=True)) + assert "dup" not in api.deactivated + + def test_refuse_self_merge(service, api): api.create_test_user("same", email="a@x.com", is_email_verified=True) with pytest.raises(SameUserError): From c87d7dc9854f776a0c67962438f3ed67e5ac501d Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 12:47:05 +0900 Subject: [PATCH 19/21] fix(security): remove formatted SQLite PRAGMA --- services/account_unification/app/user_locks.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/services/account_unification/app/user_locks.py b/services/account_unification/app/user_locks.py index c9a3d11..0e940aa 100644 --- a/services/account_unification/app/user_locks.py +++ b/services/account_unification/app/user_locks.py @@ -95,15 +95,11 @@ def __init__(self, database_path: str, *, timeout_seconds: float = 10.0) -> None def _connect(self) -> sqlite3.Connection: """Open one autocommit connection configured with the lock timeout.""" - connection = sqlite3.connect( + return sqlite3.connect( self._database_path, timeout=self._timeout_seconds, isolation_level=None, ) - connection.execute( - f"PRAGMA busy_timeout = {int(self._timeout_seconds * 1000)}" - ) - return connection def _initialize(self) -> None: """Create the sidecar schema before requests begin competing for it.""" From 78673bcc6819113c2efb4c149635fea6a05d030c Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 12:47:25 +0900 Subject: [PATCH 20/21] test: wire user-operation locks into API fixtures --- services/account_unification/tests/test_api.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/services/account_unification/tests/test_api.py b/services/account_unification/tests/test_api.py index 51c5693..d9ce523 100644 --- a/services/account_unification/tests/test_api.py +++ b/services/account_unification/tests/test_api.py @@ -9,11 +9,16 @@ @pytest.fixture -def client(api, audit, config): +def client(api, audit, config, user_operation_locks): from app.service import UnificationService app = create_app(wire=False) - 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 with TestClient(app) as test_client: From 62083f10aff72f649b3edd03e7fb6ba9bb7f7c5b Mon Sep 17 00:00:00 2001 From: Seongho Bae Date: Mon, 3 Aug 2026 12:47:47 +0900 Subject: [PATCH 21/21] test: provide locks to standalone audit service --- services/account_unification/tests/test_audit.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/services/account_unification/tests/test_audit.py b/services/account_unification/tests/test_audit.py index a12de6d..6777200 100644 --- a/services/account_unification/tests/test_audit.py +++ b/services/account_unification/tests/test_audit.py @@ -7,6 +7,7 @@ from app.config import ServiceConfig from app.models import FederatedIdentity, MergeRequest, RoleMapping from app.service import UnificationService +from app.user_locks import InMemoryUserOperationLocks from .mock_keycloak import MockKeycloakAdminApi @@ -70,7 +71,12 @@ def test_sqlite_audit_sink_persists(tmp_path): keycloak_client_id="svc", keycloak_client_secret="secret", ) - service = UnificationService(api, audit, config) + service = UnificationService( + api, + audit, + config, + InMemoryUserOperationLocks(), + ) result = service.merge_accounts( MergeRequest( survivor_user_id="survivor",