#!/usr/bin/env python3 """SCIM 2.0 -> NexusCRM public REST bridge (PingCAP tenant integration). Helix (PingCAP's HR/IdP hub) speaks generic SCIM 2.0. NexusCRM main does not expose a SCIM endpoint (the in-core engine was reverted; the platform-blessed L1 AppService is blocked on serviceification serving). This bridge closes the gap WITHOUT touching NexusCRM core: it terminates Helix's SCIM requests and drives NexusCRM through its existing, production public data API (/services/data sobjects + SOQL), exactly like an admin API client. Tenant-boundary note: this is PingCAP-specific integration glue (Helix is a PingCAP-internal system), which is why it lives in this L2 repo and not upstream. When the upstream L1 SCIM AppService lands (see NexusCRM docs/design/scim-l1-rebuild.md), Helix re-points at it and this bridge retires. Contract (matches Helix's generic SCIM runner exactly): GET {base}/Users?filter=userName eq "" find by email POST {base}/Users create GET {base}/Users/{id} read PUT {base}/Users/{id} full replace PATCH {base}/Users/{id} partial (deactivate) DELETE {base}/Users/{id} mapped to deactivate: NexusCRM hard-blocks User DELETE (deactivate-only lifecycle), and PATCH IsActive=false transactionally revokes the user's sessions, so deprovision is always a deactivation here. Auth is stateless and secretless: the inbound bearer must be "." of a NexusCRM confidential Connected App with a RunAs user (client_credentials). The bridge exchanges it for a NexusCRM token (cached) and forwards calls under that identity, so NexusCRM's own CRUD/FLS/session machinery — not this process — is the security authority. Provisioned accounts are SSO-only: the bridge NEVER sends a Password field, so the records engine stamps an unguessable placeholder hash. Stdlib only (Python 3.9), matching this repo's tooling conventions. """ from __future__ import annotations import hashlib import json import os import re import threading import time import urllib.error import urllib.parse import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Optional, Tuple SCIM_USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" SCIM_ENTERPRISE_SCHEMA = "urn:ietf:params:scim:schemas:extension:enterprise:2.0:User" SCIM_LIST_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:ListResponse" SCIM_ERROR_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:Error" SCIM_PATCHOP_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:PatchOp" SCIM_ZERO_ACCESS_PROFILE = "SCIM Zero Access" class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): # 3xx -> treated as error return None _NO_REDIRECT_OPENER = urllib.request.build_opener(_NoRedirectHandler) USER_FILTER_RE = re.compile(r'^\s*(userName)\s+eq\s+"([^"]*)"\s*$', re.IGNORECASE) USERS_PATH_RE = re.compile(r"^/scim/v2/Users(?:/([^/]+))?/?$") class BridgeConfig: """Environment-driven configuration; no secrets are ever stored here.""" def __init__(self, env: Optional[dict] = None) -> None: e = os.environ if env is None else env self.nexus_base_url = e.get("BRIDGE_NEXUS_BASE_URL", "http://127.0.0.1:3001").rstrip("/") # The inbound bearer carries the connected-app secret, so plaintext # HTTP to a non-loopback NexusCRM is forbidden: it would expose a # credential that can create/deactivate users. Loopback is exempt # (local dev / a TLS-terminating proxy on the same host). parsed = urllib.parse.urlparse(self.nexus_base_url) host = (parsed.hostname or "").lower() insecure_ok = e.get("BRIDGE_ALLOW_INSECURE_HTTP", "") == "1" if (parsed.scheme.lower() != "https" and host not in ("127.0.0.1", "localhost", "::1") and not insecure_ok): raise ValueError( "BRIDGE_NEXUS_BASE_URL must be https:// for non-loopback hosts " "(set BRIDGE_ALLOW_INSECURE_HTTP=1 only behind a trusted TLS proxy)") self.api_version = e.get("BRIDGE_NEXUS_API_VERSION", "v63.0") self.listen_host = e.get("BRIDGE_LISTEN_HOST", "127.0.0.1") # The listener itself is plain HTTP; exposing it beyond loopback is # only safe behind a TLS-terminating proxy the operator explicitly # acknowledges (the inbound bearer carries the connected-app secret). if (self.listen_host not in ("127.0.0.1", "localhost", "::1") and e.get("BRIDGE_TLS_TERMINATED", "") != "1"): raise ValueError( "BRIDGE_LISTEN_HOST beyond loopback requires BRIDGE_TLS_TERMINATED=1 " "(deploy the bridge behind a TLS-terminating proxy)") self.listen_port = int(e.get("BRIDGE_LISTEN_PORT", "8790")) self.default_profile_name = e.get( "BRIDGE_DEFAULT_PROFILE_NAME", SCIM_ZERO_ACCESS_PROFILE, ).strip() if self.default_profile_name != SCIM_ZERO_ACCESS_PROFILE: raise ValueError( "BRIDGE_DEFAULT_PROFILE_NAME must be SCIM Zero Access; " "SCIM onboarding cannot grant business access" ) self.timezone = e.get("BRIDGE_TIMEZONE_SID_KEY", "Asia/Shanghai") self.locale = e.get("BRIDGE_LOCALE_SID_KEY", "en_US") self.email_encoding = e.get("BRIDGE_EMAIL_ENCODING_KEY", "UTF-8") self.language = e.get("BRIDGE_LANGUAGE_LOCALE_KEY", "en_US") # Deactivation safety valves: a bounded budget per rolling window plus # a never-deactivate exemption list (bootstrap admin is always exempt). self.deactivation_max = int(e.get("BRIDGE_DEACTIVATION_MAX", "25")) self.deactivation_window_s = int(e.get("BRIDGE_DEACTIVATION_WINDOW_SECONDS", "3600")) if self.deactivation_max <= 0 or self.deactivation_window_s <= 0: raise ValueError( "BRIDGE_DEACTIVATION_MAX and BRIDGE_DEACTIVATION_WINDOW_SECONDS " "must be positive: the deactivation circuit breaker cannot be disabled") exempt = [s.strip().lower() for s in e.get("BRIDGE_EXEMPT_EMAILS", "").split(",") if s.strip()] if not exempt: raise ValueError( "BRIDGE_EXEMPT_EMAILS is required: list the bootstrap/break-glass " "admin account(s) that must never be deactivated via sync") # The actual bootstrap admin varies by deployment (VERIFY_USERNAME), so # it must be named explicitly and must appear in the exemption list; a # hard-coded default cannot be trusted to protect a renamed admin. # BRIDGE_BOOTSTRAP_ADMIN and VERIFY_USERNAME may both name a canonical # admin. Do NOT let one silently override the other: EVERY configured # admin identity must be present in the exemption list, so a typo in one # cannot leave the real deployment admin deactivatable. admin_identities = { (e.get(key) or "").strip().lower() for key in ("BRIDGE_BOOTSTRAP_ADMIN", "VERIFY_USERNAME") } - {""} if not admin_identities: raise ValueError( "BRIDGE_BOOTSTRAP_ADMIN (or VERIFY_USERNAME) is required: name the " "deployment's bootstrap admin so it is provably exemption-protected") missing = sorted(admin_identities - set(exempt)) if missing: raise ValueError( f"bootstrap admin identit(y/ies) {missing} must be included in " "BRIDGE_EXEMPT_EMAILS (they must never be deactivated via sync)") self.exempt_emails = set(exempt) self.http_timeout_s = int(e.get("BRIDGE_HTTP_TIMEOUT_SECONDS", "30")) # Breaker state must survive restarts, or a crash-loop/redeploy would # hand mass deactivation a fresh budget every time. Single-writer file; # run exactly one bridge instance per state file. self.state_file = e.get("BRIDGE_STATE_FILE", "scim-bridge-state.json") class BridgeError(Exception): def __init__(self, status: int, detail: str, scim_type: Optional[str] = None) -> None: super().__init__(detail) self.status = status self.detail = detail self.scim_type = scim_type def upstream_error_summary(payload: Any) -> str: """Reduce an upstream error payload to error codes only. Raw API payloads can embed customer data (field values, names, emails), so neither logs nor SCIM error bodies may carry them verbatim. """ codes = [] items = payload if isinstance(payload, list) else [payload] for item in items: if isinstance(item, dict) and item.get("errorCode"): codes.append(str(item["errorCode"])) return ",".join(codes) if codes else "unavailable" def soql_quote(value: str) -> str: return value.replace("\\", "\\\\").replace("'", "\\'") class NexusClient: """Minimal NexusCRM public-API client bound to one inbound credential. Token cache is keyed by a one-way credential fingerprint (client_id AND secret), so a wrong secret can never ride a previously cached token; entries hold only the NexusCRM-issued bearer (never the client secret) and are refreshed on expiry or one 401. """ _token_cache: dict = {} _cache_lock = threading.Lock() TOKEN_TTL_MARGIN_S = 300 TOKEN_ASSUMED_TTL_S = 23 * 3600 # engine sessions last 24h; renew early def __init__(self, config: BridgeConfig, client_id: str, client_secret: str) -> None: self._config = config self._client_id = client_id self._client_secret = client_secret # Cache key must bind BOTH halves of the credential: keying by # client_id alone would let a request with a wrong secret ride a # token cached by an earlier valid request. The fingerprint is a # one-way hash, so the secret itself is never stored. self._cache_key = hashlib.sha256( f"{client_id}\x00{client_secret}".encode("utf-8")).hexdigest() def _data_url(self, suffix: str) -> str: return f"{self._config.nexus_base_url}/services/data/{self._config.api_version}{suffix}" def _http(self, method: str, url: str, body: Optional[dict], headers: dict) -> Tuple[int, Any]: data = None if body is None else json.dumps(body).encode("utf-8") request = urllib.request.Request(url, data=data, method=method) for name, value in headers.items(): request.add_header(name, value) if data is not None: request.add_header("Content-Type", "application/json") try: with _NO_REDIRECT_OPENER.open(request, timeout=self._config.http_timeout_s) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else None) except urllib.error.HTTPError as err: raw = err.read() try: parsed = json.loads(raw) if raw else None except ValueError: parsed = raw.decode("utf-8", "replace") return err.code, parsed except (urllib.error.URLError, OSError) as err: raise BridgeError(502, f"NexusCRM unreachable: {err}") from err def _mint_token(self) -> str: form = urllib.parse.urlencode({ "grant_type": "client_credentials", "client_id": self._client_id, "client_secret": self._client_secret, }).encode("ascii") request = urllib.request.Request( f"{self._config.nexus_base_url}/services/oauth2/token", data=form, method="POST") request.add_header("Content-Type", "application/x-www-form-urlencoded") try: with _NO_REDIRECT_OPENER.open(request, timeout=self._config.http_timeout_s) as resp: payload = json.loads(resp.read()) except urllib.error.HTTPError as err: raise BridgeError(401, f"NexusCRM rejected the connected-app credential ({err.code})") except (urllib.error.URLError, OSError) as err: raise BridgeError(502, f"NexusCRM token endpoint unreachable: {err}") from err token = payload.get("access_token") if isinstance(payload, dict) else None if not token: raise BridgeError(502, "NexusCRM token response carried no access_token") try: ttl = float(payload.get("expires_in")) except (TypeError, ValueError): ttl = float(self.TOKEN_ASSUMED_TTL_S) if ttl <= 0: ttl = float(self.TOKEN_ASSUMED_TTL_S) return token, ttl def _token(self, force_refresh: bool = False) -> str: now = time.monotonic() with self._cache_lock: entry = self._token_cache.get(self._cache_key) if not force_refresh and entry and entry[1] > now: return entry[0] token, ttl = self._mint_token() # Honor the server-provided lifetime, refreshing early by a margin so a # token never expires mid-sync; short lifetimes keep a small floor. usable = max(30.0, ttl - self.TOKEN_TTL_MARGIN_S) with self._cache_lock: self._token_cache[self._cache_key] = (token, now + usable) return token def call(self, method: str, suffix: str, body: Optional[dict] = None) -> Tuple[int, Any]: headers = {"Authorization": f"Bearer {self._token()}"} status, payload = self._http(method, self._data_url(suffix), body, headers) if status == 401 and method == "GET": # Token expired/revoked -> refresh and retry exactly once, for # reads only. Mutations (POST/PATCH/PUT/DELETE) are never replayed: # if NexusCRM committed before the 401 surfaced, a replay could # duplicate the user or re-fire record automation/events. The # failure is surfaced and the IdP's own sync retry converges it. headers = {"Authorization": f"Bearer {self._token(force_refresh=True)}"} status, payload = self._http(method, self._data_url(suffix), body, headers) elif status == 401: self._token(force_refresh=True) # heal the cache for the next call return status, payload def query(self, soql: str) -> list: status, payload = self.call("GET", "/query?q=" + urllib.parse.quote(soql)) if status != 200 or not isinstance(payload, dict): raise BridgeError(502, f"SOQL query failed ({status}): codes={upstream_error_summary(payload)}") return payload.get("records") or [] class ScimBridgeService: def __init__(self, config: BridgeConfig) -> None: self._config = config self._deactivation_lock = threading.Lock() self._deactivations = self._load_deactivations() def _load_deactivations(self) -> list: try: with open(self._config.state_file, "r", encoding="utf-8") as handle: raw = json.load(handle) items = raw.get("deactivations") if isinstance(raw, dict) else None if not isinstance(items, list): raise ValueError("state file is not a {deactivations: [...]} object") return [float(t) for t in items] except FileNotFoundError: return [] except (OSError, ValueError, TypeError) as err: # Fail closed: a corrupt state file must not silently reset the # deactivation budget. The operator repairs or removes it. raise ValueError( f"breaker state file {self._config.state_file} is unreadable " f"({err}); refusing to start with a reset deactivation budget") from err def _persist_deactivations(self) -> None: tmp_path = self._config.state_file + ".tmp" with open(tmp_path, "w", encoding="utf-8") as handle: json.dump({"deactivations": self._deactivations}, handle) handle.flush() os.fsync(handle.fileno()) os.replace(tmp_path, self._config.state_file) # -- mapping --------------------------------------------------------- def _profile_id(self, client: NexusClient) -> str: name = self._config.default_profile_name records = client.query( "SELECT Id, IsActive, IsSystem, UserType FROM Profile " f"WHERE Name = '{soql_quote(name)}' LIMIT 2" ) if len(records) != 1: raise BridgeError( 503, "zero-access onboarding profile is missing or duplicate; " "user creation refused", ) profile = records[0] if ( not profile.get("Id") or profile.get("IsActive") is not True or profile.get("IsSystem") is not False or profile.get("UserType") != "Standard" ): raise BridgeError( 503, "zero-access onboarding profile attributes drifted; " "user creation refused", ) profile_id = str(profile["Id"]) quoted_profile_id = soql_quote(profile_id) permission_sets = client.query( "SELECT Id FROM PermissionSet " f"WHERE ProfileId = '{quoted_profile_id}' " "AND IsOwnedByProfile = true AND IsActive = true LIMIT 2" ) if len(permission_sets) != 1 or not permission_sets[0].get("Id"): raise BridgeError( 503, "zero-access onboarding profile permission owner drifted; " "user creation refused", ) permission_set_id = soql_quote(str(permission_sets[0]["Id"])) grant_queries = ( ( "profile system permissions", "SELECT Id FROM _System_ProfilePermission " f"WHERE ProfileId = '{quoted_profile_id}' AND IsEnabled = true LIMIT 1", ), ( "permission-set system permissions", "SELECT Id FROM _System_PermissionSetPermission " f"WHERE PermissionSetId = '{permission_set_id}' " "AND IsEnabled = true LIMIT 1", ), ( "object permissions", "SELECT Id FROM ObjectPermissions " f"WHERE ParentId = '{permission_set_id}' AND IsDeleted = false " "AND (PermissionsRead = true OR PermissionsCreate = true " "OR PermissionsEdit = true OR PermissionsDelete = true " "OR PermissionsViewAllRecords = true " "OR PermissionsModifyAllRecords = true) LIMIT 1", ), ( "field permissions", "SELECT Id FROM FieldPermissions " f"WHERE ParentId = '{permission_set_id}' AND IsDeleted = false " "AND (PermissionsRead = true OR PermissionsEdit = true) LIMIT 1", ), ( "permission-set custom permissions", "SELECT Id FROM PermissionSetCustomPermission " f"WHERE PermissionSetId = '{permission_set_id}' LIMIT 1", ), ( "profile custom permissions", "SELECT Id FROM ProfileCustomPermission " f"WHERE ProfileId = '{quoted_profile_id}' LIMIT 1", ), ( "record-type visibility", "SELECT Id FROM PermissionSetRecordTypeVisibility " f"WHERE PermissionSetId = '{permission_set_id}' " "AND IsVisible = true LIMIT 1", ), ) for label, query in grant_queries: if client.query(query): raise BridgeError( 503, f"zero-access onboarding profile has {label}; " "user creation refused", ) return profile_id def _resolve_manager_id(self, client: NexusClient, manager_value: str) -> Optional[str]: value = (manager_value or "").strip() if not value: return None if "@" not in value: # already a NexusCRM User id return value quoted = soql_quote(value) records = client.query( "SELECT Id FROM User WHERE Email = " f"'{quoted}' OR Username = '{quoted}' LIMIT 1") if not records: raise BridgeError( 400, f"manager not found in NexusCRM: {value}", "invalidValue") return records[0]["Id"] REPLACE_MANAGED_FIELDS = ("Title", "Phone", "MobilePhone", "Department", "Division", "Company", "ManagerId", "FirstName") def _require_valid_user_payload(self, payload: dict) -> None: schemas = payload.get("schemas") if not isinstance(schemas, list) or SCIM_USER_SCHEMA not in schemas: raise BridgeError( 400, f"payload is not a SCIM User (missing {SCIM_USER_SCHEMA} in schemas)", "invalidValue") user_name = str(payload.get("userName") or "").strip() primary_email = "" for item in payload.get("emails") or []: if isinstance(item, dict) and item.get("value"): primary_email = str(item["value"]).strip() break if not user_name and not primary_email: raise BridgeError( 400, "SCIM User requires userName (or a primary email)", "invalidValue") def _fields_from_scim(self, client: NexusClient, payload: dict, replace: bool = False) -> Tuple[dict, Optional[bool]]: """Map a SCIM User payload to NexusCRM User fields. Returns (fields, active_or_None). Absent SCIM attributes are simply not written (Helix PUTs are complete for the attributes it manages, so this never clobbers unmanaged fields). Password is never mapped: omitting it makes the records engine stamp an unguessable hash -> SSO-only account. """ fields: dict = {} user_name = str(payload.get("userName") or "").strip() if user_name: fields["Username"] = user_name emails = payload.get("emails") or [] email = "" for item in emails: if isinstance(item, dict) and item.get("value"): email = str(item["value"]).strip() if item.get("primary"): break fields_email = email or user_name if fields_email: fields["Email"] = fields_email name = payload.get("name") or {} if isinstance(name, dict): if name.get("givenName"): fields["FirstName"] = str(name["givenName"]).strip() if name.get("familyName"): fields["LastName"] = str(name["familyName"]).strip() if payload.get("displayName") and "LastName" not in fields: fields["LastName"] = str(payload["displayName"]).strip() if payload.get("title"): fields["Title"] = str(payload["title"]).strip() for phone in payload.get("phoneNumbers") or []: if not (isinstance(phone, dict) and phone.get("value")): continue if (phone.get("type") or "").lower() == "mobile": fields["MobilePhone"] = str(phone["value"]).strip() else: fields.setdefault("Phone", str(phone["value"]).strip()) enterprise = payload.get(SCIM_ENTERPRISE_SCHEMA) or {} if isinstance(enterprise, dict): if enterprise.get("department"): fields["Department"] = str(enterprise["department"]).strip() if enterprise.get("division"): fields["Division"] = str(enterprise["division"]).strip() if enterprise.get("organization"): fields["Company"] = str(enterprise["organization"]).strip() manager = enterprise.get("manager") if isinstance(manager, dict) and "value" in manager: value = str(manager.get("value") or "").strip() if not value: fields["ManagerId"] = None # explicit empty -> clear manager else: # Unresolved non-empty manager raises (400) instead of # silently clearing an existing manager on PUT/PATCH. fields["ManagerId"] = self._resolve_manager_id(client, value) if replace: # PUT is a full replacement of the managed attribute set: an # attribute absent from the payload clears the mapped field, so a # removed title/phone/department/manager never lingers. for field_name in self.REPLACE_MANAGED_FIELDS: fields.setdefault(field_name, None) if "LastName" not in fields: # LastName is schema-required and cannot be cleared; a full # replacement without a name re-derives it the same way create # does, so it can never linger stale. source = fields.get("Username") or fields.get("Email") or "" if source: fields["LastName"] = source.split("@", 1)[0] active: Optional[bool] = None if "active" in payload: raw_active = payload.get("active") if isinstance(raw_active, bool): active = raw_active elif isinstance(raw_active, str) and raw_active.strip().lower() in ("true", "false"): active = raw_active.strip().lower() == "true" else: raise BridgeError( 400, f"active must be a boolean, got: {raw_active!r}", "invalidValue") return fields, active def _create_defaults(self, client: NexusClient, fields: dict) -> dict: username = fields.get("Username") or fields.get("Email") or "" if not username: raise BridgeError(400, "userName (or a primary email) is required", "invalidValue") defaults = { "Alias": re.sub(r"[^A-Za-z0-9]", "", username.split("@", 1)[0])[:8] or "scimuser", "ProfileId": self._profile_id(client), "TimeZoneSidKey": self._config.timezone, "LocaleSidKey": self._config.locale, "EmailEncodingKey": self._config.email_encoding, "LanguageLocaleKey": self._config.language, "IsActive": True, } if "LastName" not in fields: # required by the User schema defaults["LastName"] = username.split("@", 1)[0] merged = dict(defaults) merged.update(fields) return merged # -- deactivation safety valves -------------------------------------- def _guard_deactivation(self, record: dict) -> None: identities = {(record.get("Username") or "").strip().lower(), (record.get("Email") or "").strip().lower()} - {""} exempt = identities & self._config.exempt_emails if exempt: raise BridgeError( 403, f"user {sorted(exempt)[0]} is on the never-deactivate exemption list", "mutability") now = time.time() # wall clock: the budget must survive restarts with self._deactivation_lock: cutoff = now - self._config.deactivation_window_s self._deactivations = [t for t in self._deactivations if t > cutoff] if len(self._deactivations) >= self._config.deactivation_max: raise BridgeError( 429, "deactivation circuit breaker tripped: " f"more than {self._config.deactivation_max} deactivations in " f"{self._config.deactivation_window_s}s; investigate before retrying", ) self._deactivations.append(now) self._persist_deactivations() return now def _release_deactivation(self, timestamp: float) -> None: # A failed upstream PATCH must not burn budget, or NexusCRM flakiness # would self-inflict a 429 lockout of the deprovision path. with self._deactivation_lock: try: self._deactivations.remove(timestamp) except ValueError: return self._persist_deactivations() # -- NexusCRM record helpers ------------------------------------------ def _get_user(self, client: NexusClient, user_id: str) -> dict: records = client.query( "SELECT Id, Username, Email, FirstName, LastName, IsActive, Title, Phone, " "MobilePhone, Department, Division, Company, ManagerId " f"FROM User WHERE Id = '{soql_quote(user_id)}' LIMIT 1") if not records: raise BridgeError(404, f"User {user_id} not found") return records[0] def _find_by_username(self, client: NexusClient, username: str) -> Optional[dict]: records = client.query( "SELECT Id, Username, Email, FirstName, LastName, IsActive, Title, Phone, " "MobilePhone, Department, Division, Company, ManagerId " f"FROM User WHERE Username = '{soql_quote(username)}' LIMIT 1") return records[0] if records else None def _apply_update(self, client: NexusClient, user_id: str, fields: dict, active: Optional[bool]) -> None: record = self._get_user(client, user_id) identities = {(record.get("Username") or "").strip().lower(), (record.get("Email") or "").strip().lower()} - {""} if identities & self._config.exempt_emails: for key in ("Username", "Email"): new_value = (fields.get(key) or "").strip().lower() current = (record.get(key) or "").strip().lower() if key in fields and new_value and new_value != current: raise BridgeError( 403, f"{key} of an exemption-listed user cannot be changed via sync " "(rename would bypass the never-deactivate exemption)", "mutability") write = {k: v for k, v in fields.items() if k != "Username" or v} reserved_budget = None if active is not None: if active is False and record.get("IsActive"): reserved_budget = self._guard_deactivation(record) elif active is True and not record.get("IsActive"): print(f"[scim-bridge] reactivating user id={record.get('Id')}") write["IsActive"] = active if not write: return status, payload = client.call("PATCH", f"/sobjects/User/{user_id}", write) if status not in (200, 204): if reserved_budget is not None and 400 <= status < 500: # A 4xx is a definite rejection (nothing committed), so the # budget is returned. 5xx and transport errors are AMBIGUOUS — # the deactivation may have committed before the error — so the # reservation is kept: a safety valve must fail closed, never # allow committed deactivations to slip past the budget. self._release_deactivation(reserved_budget) raise BridgeError(502, f"NexusCRM update failed ({status}): codes={upstream_error_summary(payload)}") def _scim_user(self, record: dict) -> dict: scim_payload = { "schemas": [SCIM_USER_SCHEMA], "id": record["Id"], "userName": record.get("Username") or "", "active": bool(record.get("IsActive")), "name": {"givenName": record.get("FirstName") or "", "familyName": record.get("LastName") or ""}, "emails": [{"value": record.get("Email") or "", "primary": True}], "meta": {"resourceType": "User", "location": f"/scim/v2/Users/{record['Id']}"}, } if record.get("Title"): scim_payload["title"] = record["Title"] phones = [] if record.get("Phone"): phones.append({"type": "work", "value": record["Phone"]}) if record.get("MobilePhone"): phones.append({"type": "mobile", "value": record["MobilePhone"]}) if phones: scim_payload["phoneNumbers"] = phones enterprise = {} for scim_attr, field_name in (("department", "Department"), ("division", "Division"), ("organization", "Company")): if record.get(field_name): enterprise[scim_attr] = record[field_name] if record.get("ManagerId"): enterprise["manager"] = {"value": record["ManagerId"]} if enterprise: scim_payload["schemas"] = [SCIM_USER_SCHEMA, SCIM_ENTERPRISE_SCHEMA] scim_payload[SCIM_ENTERPRISE_SCHEMA] = enterprise return scim_payload # -- SCIM operations --------------------------------------------------- def list_users(self, client: NexusClient, raw_filter: Optional[str]) -> dict: resources = [] if raw_filter: match = USER_FILTER_RE.match(raw_filter) if not match: raise BridgeError(400, f"unsupported filter: {raw_filter}", "invalidFilter") # Only userName matching is supported: externalId is not persisted, # so accepting it would silently alias it to Username and invite # duplicate provisioning. Helix's runner matches by userName. record = self._find_by_username(client, match.group(2)) if record: resources = [self._scim_user(record)] return {"schemas": [SCIM_LIST_SCHEMA], "totalResults": len(resources), "startIndex": 1, "itemsPerPage": len(resources), "Resources": resources} def create_user(self, client: NexusClient, payload: dict) -> Tuple[int, dict]: self._require_valid_user_payload(payload) fields, active = self._fields_from_scim(client, payload) username = fields.get("Username") or fields.get("Email") or "" existing = self._find_by_username(client, username) if username else None if existing: # idempotent re-provision converges instead of erroring # active absent -> preserve the record's current IsActive: a stray # re-create must never silently resurrect a deprovisioned account. self._apply_update(client, existing["Id"], fields, active) return 200, self._scim_user(self._get_user(client, existing["Id"])) body = self._create_defaults(client, fields) body.setdefault("Username", body.get("Email")) if active is False: body["IsActive"] = False body.pop("ManagerId", None) if body.get("ManagerId") is None else None status, payload_out = client.call("POST", "/sobjects/User", body) if status not in (200, 201) or not isinstance(payload_out, dict) or not payload_out.get("id"): raise BridgeError( 409 if status == 400 else 502, f"NexusCRM create failed ({status}): codes={upstream_error_summary(payload_out)}", "uniqueness" if status == 400 else None) return 201, self._scim_user(self._get_user(client, payload_out["id"])) def replace_user(self, client: NexusClient, user_id: str, payload: dict) -> dict: self._require_valid_user_payload(payload) fields, active = self._fields_from_scim(client, payload, replace=True) self._apply_update(client, user_id, fields, active) return self._scim_user(self._get_user(client, user_id)) PATCH_PATH_FIELDS = { "username": "Username", "title": "Title", "displayname": "LastName", "name.givenname": "FirstName", "name.familyname": "LastName", "department": "Department", "division": "Division", "organization": "Company", 'phonenumbers[type eq "mobile"].value': "MobilePhone", 'phonenumbers[type eq "work"].value': "Phone", } def _patch_path_field(self, path: str) -> Optional[str]: normalized = path.strip().lower() prefix = SCIM_ENTERPRISE_SCHEMA.lower() + ":" if normalized.startswith(prefix): normalized = normalized[len(prefix):] return self.PATCH_PATH_FIELDS.get(normalized) def patch_user(self, client: NexusClient, user_id: str, payload: dict) -> dict: fields: dict = {} active: Optional[bool] = None for op in payload.get("Operations") or []: op_name = (op.get("op") or "").lower() if isinstance(op, dict) else "" if op_name not in ("replace", "add", "remove"): raise BridgeError(400, f"unsupported PATCH op: {op!r}", "invalidValue") path = (op.get("path") or "").strip() value = op.get("value") normalized = path.lower() enterprise_prefix = SCIM_ENTERPRISE_SCHEMA.lower() + ":" if normalized.startswith(enterprise_prefix): normalized = normalized[len(enterprise_prefix):] if op_name == "remove": if normalized in ("manager", "manager.value"): fields["ManagerId"] = None continue field_name = self._patch_path_field(path) if field_name is None or field_name == "Username": raise BridgeError( 400, f"unsupported PATCH remove path: {path}", "invalidPath") fields[field_name] = None continue if normalized == "active": if isinstance(value, bool): active = value elif isinstance(value, str) and value.strip().lower() in ("true", "false"): active = value.strip().lower() == "true" else: raise BridgeError( 400, f"active must be a boolean, got: {value!r}", "invalidValue") elif not path and isinstance(value, dict): sub_fields, sub_active = self._fields_from_scim(client, value) fields.update(sub_fields) if sub_active is not None: active = sub_active elif normalized in ("manager", "manager.value"): raw = value.get("value") if isinstance(value, dict) else value raw = str(raw or "").strip() fields["ManagerId"] = ( None if not raw else self._resolve_manager_id(client, raw)) elif normalized == "name" and isinstance(value, dict): if value.get("givenName") is not None: fields["FirstName"] = str(value["givenName"]).strip() if value.get("familyName") is not None: fields["LastName"] = str(value["familyName"]).strip() else: field_name = self._patch_path_field(path) if field_name is None: raise BridgeError(400, f"unsupported PATCH path: {path}", "invalidPath") fields[field_name] = "" if value is None else str(value).strip() self._apply_update(client, user_id, fields, active) return self._scim_user(self._get_user(client, user_id)) def delete_user(self, client: NexusClient, user_id: str) -> None: # NexusCRM's engine forbids hard-deleting Users (deactivate-only # lifecycle; deactivation revokes sessions transactionally), so SCIM # DELETE deprovisions by deactivation. self._apply_update(client, user_id, {}, active=False) class _Handler(BaseHTTPRequestHandler): service: ScimBridgeService = None # type: ignore[assignment] config: BridgeConfig = None # type: ignore[assignment] server_version = "nexuscrm-scim-bridge/1.0" def log_message(self, fmt: str, *args: Any) -> None: # no PII in stdout print(f"[scim-bridge] {self.command} {self.path.split('?')[0]} {args[-1] if args else ''}") def _send(self, status: int, payload: Optional[dict]) -> None: raw = b"" if payload is None else json.dumps(payload).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/scim+json") self.send_header("Content-Length", str(len(raw))) self.end_headers() if raw: self.wfile.write(raw) def _error(self, err: BridgeError) -> None: detail = err.detail if err.status >= 500: # Never leak upstream payloads/exception text across the public # SCIM boundary; full detail stays in the server log. print(f"[scim-bridge] error {err.status}: {err.detail}") detail = "upstream request failed; see bridge logs" body = {"schemas": [SCIM_ERROR_SCHEMA], "status": str(err.status), "detail": detail} if err.scim_type: body["scimType"] = err.scim_type self._send(err.status, body) def _client(self) -> NexusClient: header = self.headers.get("Authorization") or "" token = header[7:].strip() if header.lower().startswith("bearer ") else "" for sep in (".", ":"): if sep in token: client_id, secret = token.split(sep, 1) if client_id and secret: return NexusClient(self.config, client_id, secret) raise BridgeError(401, 'expected Authorization: Bearer "."') MAX_BODY_BYTES = 1024 * 1024 def _body(self) -> dict: if self.headers.get("Transfer-Encoding"): raise BridgeError(411, "chunked bodies are not accepted; send Content-Length") length = int(self.headers.get("Content-Length") or 0) if length > self.MAX_BODY_BYTES: raise BridgeError(413, f"body exceeds {self.MAX_BODY_BYTES} bytes") if length <= 0: raise BridgeError(400, "a JSON body is required", "invalidSyntax") try: payload = json.loads(self.rfile.read(length)) except ValueError as err: raise BridgeError(400, f"invalid JSON body: {err}", "invalidSyntax") if not isinstance(payload, dict): raise BridgeError(400, "JSON object body required", "invalidSyntax") return payload def _route(self) -> None: parsed = urllib.parse.urlparse(self.path) if parsed.path.rstrip("/") == "/scim/v2/ServiceProviderConfig" and self.command == "GET": self._send(200, { "schemas": ["urn:ietf:params:scim:schemas:core:2.0:ServiceProviderConfig"], "patch": {"supported": True}, "filter": {"supported": True, "maxResults": 1}, "bulk": {"supported": False}, "changePassword": {"supported": False}, "sort": {"supported": False}, "etag": {"supported": False}, "authenticationSchemes": [{"type": "oauthbearertoken", "name": "NexusCRM connected-app credential"}]}) return match = USERS_PATH_RE.match(parsed.path) if not match: raise BridgeError(404, f"no such resource: {parsed.path}") user_id = match.group(1) client = self._client() if self.command == "GET" and user_id is None: raw_filter = (urllib.parse.parse_qs(parsed.query).get("filter") or [None])[0] self._send(200, self.service.list_users(client, raw_filter)) elif self.command == "GET": self._send(200, self.service._scim_user(self.service._get_user(client, user_id))) elif self.command == "POST" and user_id is None: status, body = self.service.create_user(client, self._body()) self._send(status, body) elif self.command == "PUT" and user_id: self._send(200, self.service.replace_user(client, user_id, self._body())) elif self.command == "PATCH" and user_id: self._send(200, self.service.patch_user(client, user_id, self._body())) elif self.command == "DELETE" and user_id: self.service.delete_user(client, user_id) self._send(204, None) else: raise BridgeError(405, f"{self.command} not supported on {parsed.path}") def _handle(self) -> None: try: self._route() except BridgeError as err: self._error(err) except Exception as err: # never leak a traceback to the IdP self._error(BridgeError(500, f"bridge internal error: {err}")) do_GET = do_POST = do_PUT = do_PATCH = do_DELETE = _handle def serve(config: Optional[BridgeConfig] = None) -> ThreadingHTTPServer: config = config or BridgeConfig() handler = type("BoundHandler", (_Handler,), { "service": ScimBridgeService(config), "config": config}) server = ThreadingHTTPServer((config.listen_host, config.listen_port), handler) return server def main() -> int: config = BridgeConfig() server = serve(config) print(f"[scim-bridge] listening on {config.listen_host}:{config.listen_port} " f"-> {config.nexus_base_url} ({config.api_version})") try: server.serve_forever() except KeyboardInterrupt: pass finally: server.server_close() return 0 if __name__ == "__main__": raise SystemExit(main())