"""Generic SCIM 2.0 client runner. SCIM (System for Cross-domain Identity Management, RFC 7643/7644) is the lingua franca for HR/IdP-to-app user provisioning. Most modern SaaS apps that take outbound identity sync — Okta, Workday inbound, Navan, Rippling, Slack Enterprise — speak SCIM 2.0 over HTTPS with a bearer token. This runner targets the standard endpoints: * ``GET {base}/Users?filter=userName eq ""`` — find an existing user by its normal login name; a uniqueness-conflict reconciliation also tries the standard ``emails.value`` attribute for legacy accounts * ``POST {base}/Users`` — create * ``PUT {base}/Users/{id}`` — full replace (preferred — predictable) * ``PATCH {base}/Users/{id}`` — partial update (used for deactivate) * ``DELETE {base}/Users/{id}`` — hard delete Conventions: * Bearer token from the integration's ``api_token`` field. Some vendors expect a different scheme (Basic auth for SAP, OAuth2 client credentials for Azure AD); those go through the ``auth_scheme`` runner option (``bearer`` default | ``basic``). * The user is normally matched by ``userName`` = work_email. During a create/update uniqueness conflict only, a miss also checks standard ``emails.value`` so an existing legacy account with a different login name can be reconciled rather than re-created. ``externalId`` is not used as an identity key. * When the runner's prior_external_id is set, we PUT directly to ``/Users/{id}`` instead of looking up first. This is the steady- state path — the lookup only runs on first sync per person. * ``404 on PUT`` means the downstream system deleted the user behind our back. Falls back to POST to recreate. Configuration (under target.runner_options): { "base_url": "https://api.navan.com/scim/v2", "auth_scheme": "bearer", # or "basic" "extra_headers": {"X-Tenant": "..."}, "verify_ssl": true, "timeout_seconds": 30, "schemas_user_create": [...], # optional override "rate_limit_per_minute": 60, # client-side guard, defaults to 60 } The token (or basic-auth password) is read from the integration's ``api_token`` field; the username for basic-auth is read from ``api_username``. Both are encrypted at rest by the integration row. """ from __future__ import annotations import asyncio import base64 import logging from typing import Any import httpx from app.domains.identity_sync.runners.base import PersonOutcome, SyncRunner logger = logging.getLogger(__name__) # Standard SCIM 2.0 schemas (RFC 7643). SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" SCIM_USER_ENTERPRISE_SCHEMA = ( "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User" ) SCIM_LIST_RESPONSE = "urn:ietf:params:scim:api:messages:2.0:ListResponse" SCIM_PATCH_OP = "urn:ietf:params:scim:api:messages:2.0:PatchOp" SCIM_GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group" # Navan's vendor extension (user-reference field ``travelerApprover`` # lives here). Harmlessly absent from non-Navan payloads. NAVAN_EXT_SCHEMA = "urn:ietf:params:scim:schemas:extension:navan:2.0:User" def _scim_filter_quote(value: str) -> str: """Escape a value for a SCIM ``attr eq "value"`` filter literal. RFC 7644 §3.4.2.2 string literals use double quotes; embedded quotes and backslashes are backslash-escaped. Dept names are the only caller input and they're HRIS-controlled, but escaping keeps a stray quote from truncating the filter (and from acting as a filter-injection vector on lax servers). """ return value.replace("\\", "\\\\").replace('"', '\\"') # Vendor phrasings that mean "the user you targeted isn't here". # Used to make deactivate idempotent: a clean 404 already covers # "no such resource", but some vendors (Navan) instead answer a # deactivate PATCH with 400 + one of these in the SCIM ``detail`` # when the resolved id no longer maps to a live account. _USER_ABSENT_MARKERS = ( "does not exist", "not found", "no such user", "unknown user", ) def _scim_says_user_absent(resp_body: Any, raw: str) -> bool: """True when a SCIM error response indicates the target user is already gone. Checks the structured ``detail`` first, then the raw text as a fallback. Case-insensitive.""" haystack = "" if isinstance(resp_body, dict): haystack = str(resp_body.get("detail") or "") if not haystack: haystack = raw or "" lowered = haystack.lower() return any(marker in lowered for marker in _USER_ABSENT_MARKERS) # Vendor refuses to manage a user because their email domain isn't in # the tenant's configured allow-list. This is PERMANENT and per-user: # no payload Helix can send will satisfy it. Navan (Universal SCIM) # rejects EVERY write to such a user — create, update, and even a no-op # ``active=false`` deactivate — with HTTP 400 ``invalidValue`` and a # detail like (captured from prod 2026-06-08, request to deactivate an # already-inactive user):: # # "896581f8-... :: -1: User email selva.ganesh@pingcap-c.com does not # match any allowed company domains: [pingcap.com]" # # Note: PR #528 guessed this message was "does not exist" from a # truncated (``raw[:200]``) prod log and added it to # ``_USER_ABSENT_MARKERS`` — the guess never matched the real wording, # so the run kept going ``partial`` (retry_count climbed 9 → 22). The # marker below is the verified live string, not a guess. _DOMAIN_NOT_ALLOWED_MARKERS = ( "does not match any allowed company domain", # singular + plural "not in the allowed company domain", ) def _scim_says_domain_not_allowed(resp_body: Any, raw: str) -> bool: """True when a SCIM 400 means the vendor refuses to manage this user because their email domain isn't in the tenant allow-list. Permanent, non-retryable. Checks ``detail`` first, then raw text.""" haystack = "" if isinstance(resp_body, dict): haystack = str(resp_body.get("detail") or "") if not haystack: haystack = raw or "" lowered = haystack.lower() return any(marker in lowered for marker in _DOMAIN_NOT_ALLOWED_MARKERS) # Vendor refuses the user's specific email even though the domain IS # allowed — Navan returns HTTP 409 with a detail like (captured from # prod 2026-07-07, creating a China-entity manager auto-included as an # approver of foreign reports):: # # "a3f4779c-... :: Email 'xulijia@pingcap.com' can't be used for this # company" # # Root cause (confirmed 2026-07-16, xulijia + zhangyuanjia): the email # is already registered on the Navan side in a NON-User shape — # typically a finance-invited **guest** — which claims the address but # is invisible to SCIM (GET /Users?filter finds nothing; there is no # guest resource on the SCIM surface at all, so Helix can neither see # nor convert it). A cross-org claim produces the same symptom. Either # way no payload satisfies the create, and it is NOT a # userName-uniqueness conflict within THIS company (so the POST-409 # ``_reconcile_by_email`` path would GET-filter, find nothing, and fail # every sweep — retry_count climbing forever). Treat it like # ``_scim_says_domain_not_allowed``: a non-failing skip. NOT permanent # though: the skip outcome re-attempts on the ≤7-day drift-reverify # clock (service.py skip handling bumps last_verified_at), so once an # admin removes/converts the guest in the Navan console, the account # creates itself on the next re-verify pass with zero manual re-push. _EMAIL_NOT_USABLE_MARKERS = ( "can't be used for this company", "cannot be used for this company", ) def _scim_says_email_not_usable(resp_body: Any, raw: str) -> bool: """True when the vendor permanently refuses this user's specific email (claimed by another tenant / blocked), regardless of HTTP status. Checks ``detail`` first, then raw text.""" haystack = "" if isinstance(resp_body, dict): haystack = str(resp_body.get("detail") or "") if not haystack: haystack = raw or "" lowered = haystack.lower() return any(marker in lowered for marker in _EMAIL_NOT_USABLE_MARKERS) def _is_uniqueness_conflict(status: int, raw: str) -> bool: """True when a user create/update was refused because the ``userName`` is already taken by ANOTHER resource — AWS IdC answers 409 (detail "Duplicate UserName"), some servers 400 + a 'uniqueness' scimType. Signals that our stored ``external_id`` is stale/orphan and the email's true owner has a different id, so we should relink by email.""" if status == 409: return True return status == 400 and "uniqueness" in (raw or "").lower() def _scim_says_already_member(status: int, raw: str) -> bool: """True when a group-member ``add`` was refused because the member is already present — AWS IdC answers 409, or 400 + 'uniqueness'. The goal state (member in group) already holds, so it's a success, not a failure. Used to keep membership reconcile idempotent under IdC's eventual consistency (a GET snapshot can miss a just-landed member, leaving a stale entry in the add set).""" if status == 409: return True return status == 400 and "uniqueness" in (raw or "").lower() class ScimError(RuntimeError): """Wraps an HTTP failure from a SCIM endpoint.""" def __init__( self, message: str, *, status_code: int | None = None, body_excerpt: str | None = None, ): super().__init__(message) self.status_code = status_code self.body_excerpt = body_excerpt class ScimRunner(SyncRunner): """Generic SCIM 2.0 / RFC 7644 outbound client.""" def __init__( self, *, base_url: str, auth_scheme: str = "bearer", api_token: str, api_username: str | None = None, extra_headers: dict[str, str] | None = None, verify_ssl: bool = True, ca_bundle_path: str | None = None, timeout_seconds: int = 30, rate_limit_per_minute: int = 60, schemas_user_create: list[str] | None = None, put_merge: bool = False, label: str = "SCIM", transport: httpx.AsyncBaseTransport | None = None, ): if not base_url: raise ValueError("scim runner requires base_url") if not api_token: raise ValueError("scim runner requires api_token") self._base_url = base_url.rstrip("/") self._auth_scheme = auth_scheme.lower() if self._auth_scheme not in ("bearer", "basic"): raise ValueError( f"unsupported auth_scheme {auth_scheme!r}; " "use 'bearer' or 'basic'" ) if self._auth_scheme == "basic" and not api_username: raise ValueError("basic auth requires api_username") self._api_token = api_token self._api_username = api_username self._extra_headers = dict(extra_headers or {}) self._timeout = timeout_seconds # ``follow_redirects=False`` is the httpx default but we set # it explicitly so a future global default change doesn't # silently turn the SCIM client into an open redirect proxy. # ``trust_env=False`` so an ambient ``HTTPS_PROXY`` envvar # cannot redirect bearer-token-bearing traffic through an # unintended proxy on a misconfigured worker host. from app.domains.identity_sync.runners.tls import build_identity_sync_tls_verify tls_verify = build_identity_sync_tls_verify( base_url=self._base_url, verify_ssl=verify_ssl, ca_bundle_path=ca_bundle_path, ) self._client = httpx.AsyncClient( timeout=timeout_seconds, verify=tls_verify, follow_redirects=False, trust_env=False, transport=transport, ) # Tiny client-side rate-limiter — enough to avoid the obvious # 429 storm when an admin clicks "Sync Now" on a 500-employee # target. Per-second token bucket, derived from the per-minute # limit. Vendors typically allow 60-120/min for SCIM; default # of 60 is conservative. self._min_interval = 60.0 / max(1, rate_limit_per_minute) self._rl_lock = asyncio.Lock() self._last_call_at: float = 0.0 self._schemas_user_create = schemas_user_create or [ SCIM_USER_SCHEMA, SCIM_USER_ENTERPRISE_SCHEMA, ] # GET-merge before a full-replace PUT — OPT-IN (default off). ON for # a vendor that CLEARS omitted attributes on PUT (AWS IdC, RFC 7644 # §3.5.1) so a transiently-empty mapped field doesn't erase the # vendor value; OFF avoids the extra GET-per-PUT read on vendors that # don't need it. (PR2's required-field preflight + value_map-miss # signal already catch the most common blanking subclass upstream; # #1965 protects the manager/travelerApprover reference slots.) self._put_merge = bool(put_merge) self._label = label # email → SCIM resource id (or None = lookup found nothing) for # user-reference slot resolution. Per-runner (= per sweep), so a # 200-person sweep whose reports share ~40 managers costs ~40 # lookups, not 200. self._user_ref_cache: dict[str, str | None] = {} # ── Wire helpers ───────────────────────────────────────────────── def _auth_header(self) -> str: if self._auth_scheme == "bearer": return f"Bearer {self._api_token}" creds = f"{self._api_username}:{self._api_token}".encode() return f"Basic {base64.b64encode(creds).decode('ascii')}" def _headers(self) -> dict[str, str]: # Build extra_headers FIRST, then layer auth/accept/ # content-type ON TOP so admin-supplied entries cannot # override the bearer token (or wrap a stolen token onto a # different host's request). The validator at # ``service._validate_runner_options`` ALSO rejects these # forbidden keys at save time — defence in depth. headers: dict[str, str] = {} headers.update(self._extra_headers) headers["Authorization"] = self._auth_header() headers["Accept"] = "application/scim+json, application/json" headers["Content-Type"] = "application/scim+json" return headers async def _rate_limit_wait(self) -> None: async with self._rl_lock: now = asyncio.get_event_loop().time() wait = self._min_interval - (now - self._last_call_at) if wait > 0: await asyncio.sleep(wait) self._last_call_at = asyncio.get_event_loop().time() async def _request( self, method: str, path: str, *, json_body: dict[str, Any] | None = None, params: dict[str, str] | None = None, retry_on_429: int = 2, ) -> tuple[int, dict[str, Any] | None, str]: """Issue a SCIM request with bounded retries on 429/5xx. Returns (status_code, body_json_or_None, raw_text). Raises ``ScimError`` on transport failures or non-recoverable 4xx (after retry budget is exhausted). """ url = f"{self._base_url}/{path.lstrip('/')}" attempt = 0 last_exc: Exception | None = None while attempt <= retry_on_429: await self._rate_limit_wait() try: resp = await self._client.request( method, url, headers=self._headers(), json=json_body, params=params, ) except httpx.HTTPError as e: last_exc = e attempt += 1 if attempt > retry_on_429: raise ScimError( f"transport error: {type(e).__name__}: {e!s:.200}" ) from e await asyncio.sleep(2 ** attempt) continue text = resp.text body: dict[str, Any] | None = None if text: try: body = resp.json() except ValueError: body = None # Retry on 429 (rate limit) and 5xx (transient server). if resp.status_code in (429, 500, 502, 503, 504): attempt += 1 if attempt > retry_on_429: raise ScimError( f"{method} {path} → {resp.status_code} after " f"{retry_on_429 + 1} attempts: {text[:300]}", status_code=resp.status_code, body_excerpt=text[:500], ) # Honor Retry-After when present; otherwise exponential. ra = resp.headers.get("Retry-After") if ra and ra.isdigit(): await asyncio.sleep(min(int(ra), 30)) else: await asyncio.sleep(2 ** attempt) continue return resp.status_code, body, text # Loop only exits via return or raise; this is unreachable but # keeps the type checker happy. raise ScimError( f"unreachable: exhausted retries with last_exc={last_exc!r}" ) def _domain_reject_skip( self, status: int, resp_body: Any, raw: str, person_email: str, ) -> PersonOutcome | None: """If the vendor permanently refuses this user because their email domain isn't in the tenant allow-list, return a non-failing ``skipped`` outcome (visible in state, not retried every sweep). Otherwise ``None`` — the caller handles the failure normally.""" if status == 400 and _scim_says_domain_not_allowed(resp_body, raw): logger.info( "scim upsert: %s rejected by %s as outside the tenant's " "allowed email domains; skipping (permanent, no retry).", person_email, self._label, ) return PersonOutcome( status="skipped", error=( f"vendor refused {person_email}: email domain not in " "the tenant allow-list (permanent — remove from the " "target filter or add the domain on the vendor side)" ), ) # Email already taken on the vendor side in a non-User shape # (typically a finance-invited GUEST, invisible to SCIM) — # Navan answers 409 here (NOT a userName-uniqueness conflict), so # this must be checked BEFORE the 409 reconcile path or the sweep # fails on it every night. if _scim_says_email_not_usable(resp_body, raw): logger.info( "scim upsert: %s rejected by %s — email can't be used for " "this company (likely a vendor-side guest invite holding " "the address); skipping, auto-retries on the drift-" "reverify clock.", person_email, self._label, ) return PersonOutcome( status="skipped", error=( f"vendor refused {person_email}: this email can't be " "used for this company — the address is already taken " "on the vendor side, typically a guest invite " "(invisible to SCIM). Remove/convert the guest in the " "vendor admin console; this row then auto-retries " "within the ≤7-day re-verify window (or push the " "person manually for immediate effect)" ), ) return None # ── User-reference resolution (email → SCIM resource id) ──────── # RFC 7643 §4.3 defines ``enterprise.manager.value`` as the *id* of # the manager's User resource — not an email. The mapping engine # renders emails (that's all HRIS knows), and Navan tolerates that # on POST by resolving the email itself. On PUT it does NOT: an # email-valued reference is silently dropped and the vendor keeps # its old value. Verified live 2026-07-16 against Navan prod: # * POST create with manager.value=email → manager linked (id). # * PUT update with manager.value=email → 200, but the record # keeps its PREVIOUS manager (or none) — 33/180 in-scope users # had drifted managers, every one a post-create change. # Since every steady-state change flows through PUT, manager # changes NEVER landed. Fix: resolve email → id ourselves before # any write. An unresolvable email (manager not provisioned on the # vendor, e.g. blocked by Navan's cross-tenant email lock) is sent # unchanged — on POST the vendor may still resolve it, on PUT it's # a proven no-op, and never omitted so a full-replace PUT can't # clear a vendor-side value we couldn't map. Convergence: a manager # who becomes resolvable later lands on the next real push (the # ≤7-day drift re-verify or a forced re-push), because the payload # hash is computed on the email-form payload upstream. async def _resolve_user_ref(self, email: str) -> str | None: key = email.strip().lower() if key in self._user_ref_cache: return self._user_ref_cache[key] try: rid = await self._find_user_id_by_email(email.strip()) except ScimError as e: # Lookup endpoint unhealthy — keep the email fallback and # cache the miss so one broken filter endpoint costs one # round-trip per manager per sweep, not per person. logger.warning( "scim user-ref lookup failed for %s on %s: %s", email, self._label, e, ) rid = None self._user_ref_cache[key] = rid return rid async def _resolve_body_user_refs(self, body: dict[str, Any]) -> None: """Swap email-valued user references in ``body`` for SCIM ids. Slots covered: ``enterprise.manager.value`` and the Navan extension's ``travelerApprover`` (string or ``{value}`` object). Values without ``@`` are already ids — untouched. Copy-on-write on the nested extension dicts: ``body`` was built by shallow-merging the caller's rendered payload, whose hash / ledger record must keep the email form (stable across id churn), so the caller's dicts are never mutated.""" ent = body.get(SCIM_USER_ENTERPRISE_SCHEMA) if isinstance(ent, dict): mgr = ent.get("manager") if ( isinstance(mgr, dict) and isinstance(mgr.get("value"), str) and "@" in mgr["value"] ): rid = await self._resolve_user_ref(mgr["value"]) if rid: body[SCIM_USER_ENTERPRISE_SCHEMA] = { **ent, "manager": {**mgr, "value": rid}, } nav = body.get(NAVAN_EXT_SCHEMA) if isinstance(nav, dict): ta = nav.get("travelerApprover") if isinstance(ta, str) and "@" in ta: rid = await self._resolve_user_ref(ta) if rid: body[NAVAN_EXT_SCHEMA] = {**nav, "travelerApprover": rid} elif ( isinstance(ta, dict) and isinstance(ta.get("value"), str) and "@" in ta["value"] ): rid = await self._resolve_user_ref(ta["value"]) if rid: body[NAVAN_EXT_SCHEMA] = { **nav, "travelerApprover": {**ta, "value": rid}, } # SCIM server-managed / read-only attributes that must NOT be echoed # back on a PUT (they're computed by the server; re-sending them can # 400 or, worse, be interpreted as an attempt to change immutable state). _PUT_MERGE_STRIP_KEYS = frozenset({ "meta", "groups", "id", "externalId", }) async def _get_user_resource(self, user_id: str) -> dict[str, Any] | None: """GET the full current SCIM resource by id (for PUT merge). Returns the resource dict, or None on any non-200 / error — the caller then falls back to a plain full-replace PUT.""" try: status, body, _raw = await self._request("GET", f"Users/{user_id}") except Exception: # noqa: BLE001 — merge is best-effort return None if status == 200 and isinstance(body, dict): return body return None def _merge_put_body( self, *, current: dict[str, Any], mapped: dict[str, Any], ) -> dict[str, Any]: """Deep-merge the freshly-rendered ``mapped`` body over the vendor's ``current`` resource so a full-replace PUT preserves attributes the mapping doesn't manage (RFC 7644 §3.5.1: a PUT is the FULL resource, and AWS IdC / Navan CLEAR any attribute omitted from it). Mapped keys win (one level of dict deep-merge for the SCIM extension objects); server-managed keys are stripped from ``current`` first. Without this a field that transiently rendered empty this sweep (HRIS blank / a STRICT value_map miss) would be ERASED vendor-side — the general case of the #1965 manager/travelerApprover protection.""" merged: dict[str, Any] = { k: v for k, v in current.items() if k not in self._PUT_MERGE_STRIP_KEYS } for k, v in mapped.items(): if ( isinstance(v, dict) and isinstance(merged.get(k), dict) ): merged[k] = {**merged[k], **v} # extension objects: mapped wins else: merged[k] = v return merged # ── SyncRunner implementation ──────────────────────────────────── async def upsert( self, *, person_email: str, payload: dict[str, Any], prior_external_id: str | None, ) -> PersonOutcome: # Build a SCIM-shaped payload — userName is always the email # (our matching key); schemas list is whatever the runner was # configured with; the rest of the rendered fields land on the # body verbatim. The mapping engine is responsible for placing # each field on its correct nested path (name.familyName, etc.). body = { "schemas": list(self._schemas_user_create), "userName": person_email, "active": True, **payload, } # Resolve user-reference emails → ids BEFORE any write; the # same body feeds the PUT, POST, and 409-reconcile paths below. await self._resolve_body_user_refs(body) # Steady-state path: PUT directly when we already have the id. # Avoids the lookup-by-email round-trip on every push and is # what every modern SCIM server expects for "I know who you # are, here's the new state." if prior_external_id: # AWS IAM Identity Center (and RFC 7644 §3.5.1, where the # body is the full resource representation) require the PUT # body to carry the resource ``id`` matching the URI path. # Omitting it trips IdC's 400 "User ID attribute provided # within request body doesn't match the one in URI". Other # SCIM servers (Navan) accept a body ``id`` that matches the # URI, so injecting it unconditionally is spec-safe. # # GET-merge (防清场, OPT-IN via runner_options.put_merge): a # full-replace PUT CLEARS any attribute the body omits, so a # transiently-blank HRIS field / STRICT value_map miss would ERASE # the vendor's value on servers like AWS IdC. When enabled, GET # the current resource and merge the mapped fields over it (mapped # wins; unmapped preserved) so a PUT only ever CHANGES what the # mapping actually rendered. Best-effort: a failed GET falls back # to the plain full-replace PUT. Default OFF — avoids the extra # GET-per-PUT read on vendors that don't clear omitted attrs. put_body = dict(body) if self._put_merge: current = await self._get_user_resource(prior_external_id) if current is not None: put_body = self._merge_put_body(current=current, mapped=body) put_body["id"] = prior_external_id status, resp_body, raw = await self._request( "PUT", f"Users/{prior_external_id}", json_body=put_body, ) if status in (200, 204): return PersonOutcome( status="success", external_id=(resp_body or {}).get("id") or prior_external_id, ) if status == 404: logger.warning( "scim PUT 404 for %s (id=%s); recreating", person_email, prior_external_id, ) # Fall through to the POST-first flow below. elif _is_uniqueness_conflict(status, raw): # A permanent per-email rejection also arrives as 409 on # some vendors ("can't be used for this company") — skip # rather than relink (the filter would find nothing). skip = self._domain_reject_skip( status, resp_body, raw, person_email, ) if skip is not None: return skip # Our stored id is stale/orphan: the userName (email) is # owned by a DIFFERENT resource on the IdP (e.g. AWS IdC # auto-provisioned its own user via federation). PUTting our # id would rename it onto a taken userName → 409. Relink to # the real owner by email and PUT there — same reconcile the # POST-409 path runs. Self-heals the stored external_id. logger.info( "scim PUT %s (id=%s) → %d duplicate userName; " "relinking by email", person_email, prior_external_id, status, ) return await self._reconcile_by_email( person_email, body, context=f"PUT /Users/{prior_external_id} → {status}", ) else: skip = self._domain_reject_skip( status, resp_body, raw, person_email, ) if skip is not None: return skip return PersonOutcome( status="failed", external_id=prior_external_id, error=f"PUT /Users/{prior_external_id} → {status}: {raw[:200]}", ) # First sync OR prior id 404'd. POST-first flow: # # We deliberately do NOT call ``GET /Users?filter=userName eq ...`` # before the POST. Why: # * Saves one round-trip per first-time sync (~ N API calls # per Nth fresh employee on a new target). # * SCIM filter handling is the most under-tested corner of # vendor implementations — Navan in particular returned # HTTP 500 on the standard filter syntax during early # verification (2026-05-06). POST + 409-fallback bypasses # that whole code path on the vendor side. # * Per RFC 7644 §3.3 every compliant server MUST emit 409 # on a duplicate userName, so the fallback is well-defined. status, resp_body, raw = await self._request( "POST", "Users", json_body=body, ) if status in (200, 201): new_id = (resp_body or {}).get("id") if not new_id: # Spec violation — RFC 7644 §3.3 requires the response # body of a successful POST to include the resource's # server-assigned ``id``. Without it, every subsequent # sweep does another POST → vendor 409 → GET-by-email # lookup → PUT, costing 3 round-trips per person per # sweep. Surface as a one-line warning so admins can # spot a non-compliant vendor without grep-debugging. logger.warning( "scim POST %s returned %d without an 'id' field; " "next sync will re-POST and 409. Vendor is not " "RFC 7644 §3.3 compliant.", person_email, status, ) return PersonOutcome(status="success", external_id=new_id) if status == 409: # A 409 is USUALLY a duplicate userName → reconcile by email # (GET filter → PUT to the real id). But some vendors overload # 409 for a permanent per-email rejection ("can't be used for # this company"); that must skip, not reconcile (the filter # would find nothing and fail every sweep). Check first. skip = self._domain_reject_skip(status, resp_body, raw, person_email) if skip is not None: return skip return await self._reconcile_by_email( person_email, body, context="POST /Users → 409", ) skip = self._domain_reject_skip(status, resp_body, raw, person_email) if skip is not None: return skip return PersonOutcome( status="failed", error=f"POST /Users → {status}: {raw[:300]}", ) async def _reconcile_by_email( self, person_email: str, body: dict[str, Any], *, context: str, ) -> PersonOutcome: """Resolve the user's REAL id by ``userName`` filter and PUT to it. Invoked when a create or update collided on a duplicate userName — either a first-time POST against an already-existing user, or a steady-state PUT whose stored ``external_id`` is stale/orphan. The returned ``external_id`` is the reconciled id so the caller heals ``identity_sync_state``. ``context`` names the originating call for error messages. """ try: existing_id = await self._find_user_id_by_email( person_email, include_primary_email=True, ) except ScimError as e: # Vendor's GET endpoint is broken (Navan circa 2026-05-06 # returned 500). Surface the conflict + the lookup failure # together so admins can see exactly why we couldn't reconcile. return PersonOutcome( status="failed", error=( f"{context} (already exists) but GET /Users lookup " f"failed: {e!s:.200}. Vendor's SCIM filter endpoint is " "unhealthy; manually pass externalId from the vendor " "console into identity_sync_state.external_id so " "subsequent PUTs work." ), ) if not existing_id: return PersonOutcome( status="failed", error=( f"{context} but GET filter found no existing user; " f"cannot reconcile {person_email!r}." ), ) # 防清场 (put_merge): a full-replace PUT clears any attribute the # body omits, so the relink PUT must GET-merge exactly like the # steady-state PUT path (upsert). Without this, the most common # self-heal path — AWS IdC auto-reprovisioned the user under a new # id via federation → our stored id is stale → 409 → relink here — # would erase every vendor-side attribute the Helix mapping doesn't # render (manager / travelerApprover / …), the exact clobber # ``put_merge`` exists to prevent (#1965). Best-effort: a failed GET # falls back to the plain full-replace PUT. put_body = {**body, "id": existing_id} if self._put_merge: current = await self._get_user_resource(existing_id) if current is not None: merged = self._merge_put_body(current=current, mapped=body) merged["id"] = existing_id put_body = merged # AWS IdC requires the PUT body to echo the resource id from the URI. status2, resp2, raw2 = await self._request( "PUT", f"Users/{existing_id}", json_body=put_body, ) if status2 in (200, 204): return PersonOutcome( status="success", external_id=(resp2 or {}).get("id") or existing_id, ) skip = self._domain_reject_skip(status2, resp2, raw2, person_email) if skip is not None: return skip return PersonOutcome( status="failed", external_id=existing_id, error=( f"{context}; PUT /Users/{existing_id} → {status2}: " f"{raw2[:200]}" ), ) async def deactivate( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: target_id = prior_external_id or await self._find_user_id_by_email(person_email) if target_id is None: # No stored id AND the email resolves to nothing on the vendor # — the account is genuinely absent; goal already met. return PersonOutcome(status="deactivated", external_id=None) outcome = await self._patch_deactivate(target_id, person_email) if outcome is not None: return outcome # The stored id 404'd / is absent. ZERO-TRUST re-verify (2026-07 # offboarding review): a 404 on the STORED id does NOT prove the # leaver has no live account — the vendor may hold the SAME EMAIL # under a DIFFERENT id (AWS IdC federation auto-reprovision after an # admin delete/recreate; a Navan-side recreate). Claiming # "deactivated" off the stale id would leave that live account open # while the dashboard shows the leaver revoked. So when a STORED id # 404s, look the email up again; if it now resolves to a DIFFERENT # live id, deactivate THAT. Only when the email lookup is also empty # is the account truly gone. if prior_external_id: alt_id = await self._safe_find_user_id_by_email(person_email) if alt_id and alt_id != prior_external_id: logger.warning( "scim deactivate: stored id %s for %s 404'd but the " "email resolves to a DIFFERENT live id %s — deactivating " "that (stale-id zero-trust re-verify)", prior_external_id, person_email, alt_id, ) alt_outcome = await self._patch_deactivate(alt_id, person_email) if alt_outcome is not None: return alt_outcome # alt id also 404'd → genuinely gone now. return PersonOutcome(status="deactivated", external_id=None) async def _safe_find_user_id_by_email(self, email: str) -> str | None: """``_find_user_id_by_email`` that swallows lookup errors → None. Used only on the offboarding re-verify path: if the confirming GET itself fails (SCIM collection 404 / transport blip), we must fall back to the idempotent "already gone" outcome rather than turn a stale-id 404 into a hard failure that re-fires every sweep. A genuine live account under a different id is still caught on any sweep where the lookup succeeds.""" try: return await self._find_user_id_by_email(email) except Exception: # noqa: BLE001 — re-verify is best-effort logger.info( "scim offboard re-verify lookup for %s failed — treating as " "no alternate live id (idempotent gone)", email, ) return None async def _patch_deactivate( self, target_id: str, person_email: str, ) -> PersonOutcome | None: """PATCH active=false on ``target_id``. Returns a terminal PersonOutcome, or ``None`` when the id 404'd / is absent (so the caller can re-verify by email before claiming success).""" body = { "schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "replace", "path": "active", "value": False}, ], } status, resp_body, raw = await self._request( "PATCH", f"Users/{target_id}", json_body=body, ) if status in (200, 204): return PersonOutcome(status="deactivated", external_id=target_id) # A clean 404, or a Navan 400 + "user does not exist" detail, means # THIS id no longer resolves — signal the caller to re-verify. if status == 404 or ( status == 400 and _scim_says_user_absent(resp_body, raw) ): return None # Vendor refuses to manage this user because their email domain # isn't in the tenant allow-list (Navan: "does not match any # allowed company domains"). We cannot PATCH active=false — but # such a user is outside the vendor's managed population (and in # practice already inactive there), so the offboarding goal is # met and there is nothing actionable. Retrying every sweep just # re-400s forever (this is the recurring Navan-prod failure). # Treat as a terminal, non-error deactivation. if status == 400 and _scim_says_domain_not_allowed(resp_body, raw): logger.info( "scim deactivate: %s rejected by %s as outside the " "tenant's allowed email domains; treating as " "already-unmanaged (terminal, no retry).", person_email, self._label, ) return PersonOutcome(status="deactivated", external_id=None) return PersonOutcome( status="failed", external_id=target_id, error=f"PATCH /Users/{target_id} (deactivate) → {status}: {raw[:200]}", ) async def delete( self, *, person_email: str, prior_external_id: str | None, ) -> PersonOutcome: target_id = prior_external_id or await self._find_user_id_by_email(person_email) if target_id is None: return PersonOutcome(status="deleted", external_id=None) status, _, raw = await self._request( "DELETE", f"Users/{target_id}", ) if status in (200, 202, 204): return PersonOutcome(status="deleted", external_id=None) if status == 404: # Zero-trust re-verify (parity with deactivate): a 404 on a # STORED id doesn't prove the leaver is gone — the same email # may live under a different id. Only claim deleted when the # email also resolves to nothing; otherwise delete the live id. if prior_external_id: alt_id = await self._safe_find_user_id_by_email(person_email) if alt_id and alt_id != prior_external_id: logger.warning( "scim delete: stored id %s for %s 404'd but the " "email resolves to a DIFFERENT live id %s — deleting " "that (stale-id zero-trust re-verify)", prior_external_id, person_email, alt_id, ) st2, _, raw2 = await self._request( "DELETE", f"Users/{alt_id}", ) if st2 in (200, 202, 204, 404): return PersonOutcome(status="deleted", external_id=None) return PersonOutcome( status="failed", external_id=alt_id, error=f"DELETE /Users/{alt_id} → {st2}: {raw2[:200]}", ) return PersonOutcome(status="deleted", external_id=None) return PersonOutcome( status="failed", external_id=target_id, error=f"DELETE /Users/{target_id} → {status}: {raw[:200]}", ) async def _find_user_id_by_email( self, email: str, *, include_primary_email: bool = False, ) -> str | None: # userName is the configured SCIM identifier. Only the upsert 409 # reconciliation asks the additional standard email attribute: changing # offboarding lookup semantics would be a separate safety decision. escaped_email = _scim_filter_quote(email) attributes = ("userName", "emails.value") if include_primary_email else ("userName",) for attribute in attributes: status, body, raw = await self._request( "GET", "Users", params={"filter": f'{attribute} eq "{escaped_email}"'}, ) if status != 200 or body is None: # 404 on the collection itself means SCIM isn't even mounted — # surface it early instead of treating it as an absent user. raise ScimError( f"GET /Users?filter → {status}: {raw[:200]}", status_code=status, ) # Some vendors return the resource directly when the filter narrows # to one; others wrap in ListResponse. Handle both. if body.get("schemas") == [SCIM_LIST_RESPONSE] or "Resources" in body: resources = body.get("Resources") or [] if resources: return resources[0].get("id") continue return body.get("id") return None # ── Read-only roster listing (vendor snapshot / 对端数据) ──────── async def _list_resources( self, collection: str, *, max_pages: int = 200, ) -> list[dict[str, Any]]: """Page a SCIM collection via startIndex/count (RFC 7644 §3.4.2.4). STRICT: any non-200 page raises — a partial roster silently read as "complete" would make vendor-only detection lie. The identical-page guard stops servers that ignore startIndex. Bound: 200 pages × 100 = 20k resources, far above any tenant. ``totalResults`` is NOT trusted on a full page. AWS IAM Identity Center answers ListUsers with ``totalResults == itemsPerPage == len(Resources)`` and ``startIndex: 1`` no matter what we ask for, and re-sends the same first 100 users for every ``startIndex`` — it supports no pagination at all (verified against both prod IdC tenants 2026-08-14: no ``nextToken``, no ``Link`` header, filters limited to ``eq``). Trusting that total broke on page 1, so the ``aws-ai`` / ``aws-sso`` snapshots had silently held exactly 100 of 464 users since day one — a leaver-assurance vendor signal that covered 22% of the roster while reporting itself complete. A SHORT page is the only unambiguous end-of-collection proof; on a full page we always fetch the next one, and a full page of already-seen ids proves the server ignores paging → raise. """ out: list[dict[str, Any]] = [] seen_ids: set[str] = set() start_index = 1 count = 100 for _ in range(max_pages): status, body, raw = await self._request( "GET", collection, params={"startIndex": str(start_index), "count": str(count)}, ) if status != 200 or not isinstance(body, dict): raise ScimError( f"GET /{collection} page start={start_index} → " f"{status}: {raw[:200]}", status_code=status, ) resources = body.get("Resources") or [] fresh = 0 for r in resources: if not isinstance(r, dict): continue rid = str(r.get("id") or "") if not rid or rid in seen_ids: continue seen_ids.add(rid) out.append(r) fresh += 1 total = body.get("totalResults") page_len = len(resources) start_index += page_len # Clean completion: the server told us the total, we've # collected all of it, AND the page came back short. The short # page is the load-bearing half — a FULL page that claims # ``total == len(out)`` is exactly what a server that ignores # startIndex looks like on page 1 (AWS IdC), and it is # indistinguishable from a collection whose size happens to be # a multiple of ``count``. Both cases are resolved by asking # for one more page: a real end answers empty, a paging-blind # server re-sends what we already have and trips the guard # below. One extra request per roster pull. if isinstance(total, int) and len(out) >= total and page_len < count: break # #1786: no progress this page — either an empty page, OR the # server ignored ``startIndex`` and re-sent a page of # already-seen ids (so ``fresh == 0``). Previously this broke # unconditionally, silently treating a TRUNCATED roster as a # complete snapshot — which makes ``compute_drift`` report every # unlisted person as vendor-missing (the exact STRICT-contract # violation the docstring warns about). # # An EMPTY page is the clean end of the collection. A non-empty # page carrying nothing new is the paging-blind server: it # re-served a page we already hold, so everything past what we # have is unreachable and the roster we could return would be a # truncation wearing a "complete" label. Raise on that # regardless of the declared total — under this server the # total is the page size, so ``len(out) < total`` is false # precisely when we are being lied to. if not resources or fresh == 0: if resources: raise ScimError( f"GET /{collection}: page at start=" f"{start_index - page_len} returned {page_len} " f"resources but none new (declared total={total}) — " "the server ignores startIndex paging, so the roster " f"cannot be enumerated past {len(out)}. Refusing a " "truncated roster.", ) if isinstance(total, int) and len(out) < total: raise ScimError( f"GET /{collection}: no new resources at " f"start={start_index} but only {len(out)}/{total} " "collected — refusing a truncated roster (server " "likely ignores startIndex paging).", ) break else: raise ScimError( f"GET /{collection}: exceeded {max_pages} pages — " "refusing a possibly-truncated roster.", ) return out async def list_users(self) -> list[dict[str, Any]]: """Full vendor user roster (read-only).""" return await self._list_resources("Users") async def list_groups(self) -> list[dict[str, Any]]: """Full vendor group roster (read-only). NOTE: AWS IdC omits ``members`` on GET /Groups — group rows describe existence and displayName, not membership.""" return await self._list_resources("Groups") # ── Group push (dept tree → SCIM Groups) ──────────────────────── # Used by the orchestrator's group phase (``push_groups`` runner # option) to mirror what Authing's downstream SCIM sync did for # AWS IAM Identity Center: one Group per in-scope HRIS dept with # DIRECT members. AWS IdC quirks honored here: # * GET /Groups does NOT return members — replace-all PATCH is # the only reliable membership write. # * externalId is round-tripped but not always filterable — # match externalId-first with displayName fallback. # * PUT /Groups is not supported — PATCH only. async def find_group_id( self, *, display_name: str, external_code: str | None = None, ) -> str | None: """Resolve an existing group id, externalId-first. Raises :class:`ScimError` only when BOTH lookup paths fail at the transport/server level — a filter-unsupported externalId lookup (400) quietly falls through to the displayName path. """ if external_code: params = {"filter": f'externalId eq "{_scim_filter_quote(external_code)}"'} status, body, _raw = await self._request("GET", "Groups", params=params) if status == 200 and isinstance(body, dict): resources = body.get("Resources") or [] if resources and resources[0].get("id"): return resources[0]["id"] # fall through to displayName on empty — older rows # may have been created without externalId. params = {"filter": f'displayName eq "{_scim_filter_quote(display_name)}"'} status, body, raw = await self._request("GET", "Groups", params=params) if status != 200 or not isinstance(body, dict): raise ScimError( f"GET /Groups?filter=displayName → {status}: {raw[:200]}", status_code=status, ) resources = body.get("Resources") or [] if not resources: return None return resources[0].get("id") async def create_group( self, *, display_name: str, external_code: str | None = None, ) -> tuple[str | None, str | None]: """POST a new group. Returns (group_id, error). A 409 duplicate falls back to lookup (raced another writer).""" body: dict[str, Any] = { "schemas": [SCIM_GROUP_SCHEMA], "displayName": display_name, } if external_code: body["externalId"] = external_code status, resp, raw = await self._request("POST", "Groups", json_body=body) if status in (200, 201): gid = (resp or {}).get("id") if gid: return gid, None return None, ( f"POST /Groups returned {status} without an id " f"(non-compliant server): {raw[:200]}" ) if status == 409: try: gid = await self.find_group_id( display_name=display_name, external_code=external_code, ) except ScimError as e: return None, f"POST 409 then lookup failed: {e!s:.200}" if gid: return gid, None return None, "POST /Groups 409 but lookup found nothing" return None, f"POST /Groups → {status}: {raw[:300]}" # AWS IdC caps membership changes per PATCH request at 100; bigger # ops 400. We chunk add/remove ops at this cap. _PATCH_MEMBER_CAP = 100 async def replace_group_members( self, *, group_id: str, member_ids: list[str], display_name: str | None = None, ) -> str | None: """Reconcile membership to exactly ``member_ids`` via add/remove deltas, and keep the vendor displayName in step with HRIS. Returns an error string, a soft sentinel, or None. AWS IAM Identity Center rejects ``op:replace`` on ``members`` ("Replacing group memberships is not supported") — only ``add`` and ``remove`` are honored. The list endpoint omits members, but the single-resource ``GET /Groups/{id}`` DOES return them, so we read the current set and diff: add newcomers, remove leavers. Diffing (vs. a blind add-only pass) is what makes membership correct on dept moves — the person leaves the old group. Sentinels / soft errors: * ``"group_absent"`` — group vanished server-side (404); caller clears its stored id and recreates next sweep. * a ``"name_patch_failed: ..."`` prefix — membership landed but the displayName op was rejected; surfaced WITHOUT failing membership (access > label). """ # Current membership — single-resource GET returns members even # though the collection GET /Groups does not. status, gbody, raw = await self._request("GET", f"Groups/{group_id}") if status == 404: return "group_absent" if status != 200 or gbody is None: return f"GET /Groups/{group_id} → {status}: {raw[:300]}" current = { str(m.get("value")) for m in (gbody.get("members") or []) if m.get("value") } desired = {mid for mid in member_ids if mid} to_add = sorted(desired - current) to_remove = sorted(current - desired) name_warning: str | None = None # displayName drift: a standalone attribute replace (IdC accepts # replacing scalar attrs — only the members ARRAY replace is # unsupported). Skip when already in step to save a round-trip. if display_name and str(gbody.get("displayName") or "") != display_name: st, _r, dr = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "replace", "path": "displayName", "value": display_name}, ]}, ) if st == 404: return "group_absent" if st not in (200, 204): name_warning = ( f"name_patch_failed: displayName replace rejected " f"({dr[:150]}); membership written without it." ) def _chunks(ids: list[str]) -> list[list[str]]: return [ ids[i:i + self._PATCH_MEMBER_CAP] for i in range(0, len(ids), self._PATCH_MEMBER_CAP) ] # Per-member failures accumulate here. CRITICAL: the GET above # already proved the group exists, so a 404 on a member op is a # *member-reference* problem (a bad/absent user id), NOT a missing # group — it must NOT return "group_absent" (which would clear the # group's stored id and re-create-loop forever, never landing the # other valid members). We land every good member and surface the # bad ones as a non-fatal error so the dept retries next sweep # without losing the group id. member_errors: list[str] = [] async def _add_one(mid: str) -> None: st, _r, ar = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "add", "path": "members", "value": [{"value": mid}]}, ]}, ) if st in (200, 204) or _scim_says_already_member(st, ar): return member_errors.append(f"add {mid} → {st}: {ar[:120]}") # Batch add first (one round-trip for the common all-new case). # IdC ops are atomic, so a batch fails wholesale if ANY member is # already present (409/uniqueness) or unresolvable (400/404). On a # 4xx batch failure, retry per-member so the good members land and # only the genuinely-bad ids are isolated; a 5xx (already retried # by _request) is transient → fail fast for a clean next-sweep. for chunk in _chunks(to_add): st, _r, ar = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "add", "path": "members", "value": [{"value": mid} for mid in chunk]}, ]}, ) if st in (200, 204): continue if 500 <= st <= 599: return f"PATCH /Groups/{group_id} add-members → {st}: {ar[:300]}" for mid in chunk: await _add_one(mid) # Removes: drop each leaver. Primary = IdC's documented filter-path # form; fallback = the value-list form (some servers honor only # that). A 404 on a remove means the member is ALREADY absent — # that IS the goal (success), not a missing group. PATCH ops are # atomic so a rejected primary can't double-remove. async def _remove_one(mid: str) -> None: st, _r, rr = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "remove", "path": f'members[value eq "{_scim_filter_quote(mid)}"]'}, ]}, ) if st in (200, 204, 404): return st, _r, rr = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "remove", "path": "members", "value": [{"value": mid}]}, ]}, ) if st in (200, 204, 404): return member_errors.append(f"remove {mid} → {st}: {rr[:120]}") # Batch remove first (one round-trip per chunk), mirroring the # batched adds above — a department reorg / target-filter change can # drop hundreds of members in one sweep, and one PATCH-per-member # with per-call retry/backoff would stall the whole target for # minutes. IdC ops are atomic, so on a 4xx wholesale failure (one # bad/absent id fails the batch) retry per-member so the good ones # still land; a 5xx (already retried by ``_request``) is transient → # fail fast for a clean next sweep. for chunk in _chunks(to_remove): if len(chunk) == 1: # Single member: skip the batch round-trip and use the # per-member filter→value-list form directly (no redundant # filter retry). Batching only pays off for multi-member # drops, which is exactly the bulk case this optimizes. await _remove_one(chunk[0]) continue st, _r, rr = await self._request( "PATCH", f"Groups/{group_id}", json_body={"schemas": [SCIM_PATCH_OP], "Operations": [ {"op": "remove", "path": f'members[value eq "{_scim_filter_quote(mid)}"]'} for mid in chunk ]}, ) if st in (200, 204): continue if 500 <= st <= 599: return f"PATCH /Groups/{group_id} remove-members → {st}: {rr[:300]}" for mid in chunk: await _remove_one(mid) if member_errors: head = "; ".join(member_errors[:5]) extra = ( f" (+{len(member_errors) - 5} more)" if len(member_errors) > 5 else "" ) return f"PATCH /Groups/{group_id} member ops failed: {head}{extra}" return name_warning async def add_group_member( self, *, group_id: str, member_id: str, ) -> str | None: """Idempotent single-member add (hire-day event path). AWS IdC answers a duplicate add with 409 uniqueness — treated as success (the goal state holds). """ body = { "schemas": [SCIM_PATCH_OP], "Operations": [{ "op": "add", "path": "members", "value": [{"value": member_id}], }], } status, resp_body, raw = await self._request( "PATCH", f"Groups/{group_id}", json_body=body, ) if status in (200, 204): return None if _scim_says_already_member(status, raw): return None # already a member — goal achieved return f"PATCH /Groups/{group_id} add → {status}: {raw[:300]}" def describe(self) -> str: return f"{self._label} {self._base_url}" async def aclose(self) -> None: await self._client.aclose()