"""Outbound identity-sync endpoint drivers (Navan, generic SCIM 2.0). These drivers exist to register a credential shape in the integrations table (base_url + api_token + optional auth_scheme + extra_headers) so the IntegrationsPage admin UI renders the right form. They delegate all wire-protocol work to ``app.domains.identity_sync.runners.scim`` (or .google_schema for Google), which the Identity Sync Center service constructs when a target fires. Why split driver-as-catalog from runner-as-execution: * The integrations table is the single place tenants store encrypted credentials for external systems. Reusing it for identity-sync targets keeps secret handling in one audited place. * The runner is per-target (one Navan instance can host multiple syncs with different mappings); the driver is per-integration (one credential set). Splitting them lets a single integration back many sync targets without copying tokens. * ``test_connection`` issues a real but minimal call (``GET /Users?count=1&startIndex=1``) — verifies bearer token + endpoint reachability without enumerating users or mutating anything. Returns the SCIM error detail when auth fails so admins see the exact reason (401 / 403 / 404 / unreachable host) instead of a vague "click Sync to find out." To add a new SCIM-compatible system: subclass :class:`IdentitySyncEndpointDriver`, set kind/provider, decorate with ``@register_driver``. Then add a preset in ``app.domains.identity_sync.presets`` so admins can scaffold a target with sensible default mappings. """ from __future__ import annotations import re from app.integrations.driver_base import IntegrationDriver from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) _COMMON_FIELDS: list[IntegrationFieldInfo] = [ IntegrationFieldInfo( key="base_url", label="Base URL", hint=( "The SCIM endpoint base URL, e.g. " "'https://app.navan.com/universal/scim/v2' (Navan), " "'https://acme.okta.com/api/v1/scim'. Identity Sync Center " "appends '/Users' / '/Users/{id}' to this. Must include " "the version path segment." ), placeholder="https://app.navan.com/universal/scim/v2", required=True, ), IntegrationFieldInfo( key="api_token", label="API Token", hint=( "Bearer token issued by the downstream system. For Navan, " "generate under Admin → Integrations → Generate API key. " "Encrypted at rest." ), is_secret=True, required=True, ), IntegrationFieldInfo( key="auth_scheme", label="Auth Scheme", hint="Most modern SCIM systems use bearer.", options=[ {"value": "bearer", "label": "Bearer (Authorization: Bearer …)"}, {"value": "basic", "label": "Basic (Authorization: Basic …)"}, ], default="bearer", ), IntegrationFieldInfo( key="api_username", label="API Username (basic only)", hint="Only used when auth_scheme=basic.", visible_when={"auth_scheme": "basic"}, ), IntegrationFieldInfo( key="ca_bundle_path", label="Approved CA Bundle Path", hint=( "Optional private-PKI CA file mounted under " "/etc/helix/ca-certificates (.pem/.crt). Certificate and " "hostname verification remain mandatory." ), placeholder="/etc/helix/ca-certificates/vendor.pem", ), ] def _redact_connection_error(blob: str, *known_secrets: str) -> str: """Remove configured credentials from admin-visible probe failures.""" for secret in known_secrets: if secret: blob = blob.replace(secret, "***") return re.sub( r"(?i)(authorization\s*:?\s*(?:bearer|basic)\s+)[A-Za-z0-9._\-+/=]+", r"\1***", blob, ) class IdentitySyncEndpointDriver(IntegrationDriver): """Base for outbound SCIM-flavored identity-sync targets. Subclasses set :attr:`provider` + :attr:`display_name` and (optionally) add provider-specific fields by extending ``config_schema``. """ kind = "identity_sync_endpoint" config_schema = list(_COMMON_FIELDS) async def test_connection(self) -> IntegrationTestResult: """Verify bearer token + endpoint reachability via a minimal SCIM ``GET /Users?count=1&startIndex=1`` call. The call enumerates AT MOST one user (the count=1 cap is what the Universal SCIM spec defines as the smallest well-formed pagination request). It does not mutate state and does not enumerate user data — the response body's Resources array is read only for shape-validation, never logged or persisted. Failures surface the actual SCIM error body so the admin sees exactly why auth failed (401 / 403 / 404 / DNS) rather than a vague "Sync to find out." """ import asyncio import base64 import httpx base_url = (self.config.get("base_url") or "").rstrip("/") api_token = self.config.get("api_token") if not base_url: return IntegrationTestResult( ok=False, message="base_url is required", ) if not api_token: return IntegrationTestResult( ok=False, message="api_token is required", ) # Defense-in-depth SSRF guard — the canonical guard runs at # save time in the identity_sync service validator, but this # driver may also be constructed off rows persisted before # that validator existed (or via a future direct DB edit). # Without re-checking here, a stale ``base_url`` like # ``http://169.254.169.254/...`` (cloud-metadata IMDS) reaches # the request layer with the bearer ``Authorization`` header # already attached. Belt-and-suspenders mirrors the # SuccessFactors driver. (Round-17 audit.) from app.domains.identity_sync.runners.tls import ( build_identity_sync_tls_verify, ) try: tls_verify = build_identity_sync_tls_verify( base_url=base_url, verify_ssl=True, ca_bundle_path=self.config.get("ca_bundle_path"), ) except ValueError as exc: return IntegrationTestResult(ok=False, message=f"TLS config invalid: {exc}") from app.integrations._url_safety import ( UnsafeURLError, safe_async_http_transport, ) try: transport = await safe_async_http_transport( base_url, allowed_suffixes=None, verify=tls_verify, ) except UnsafeURLError as exc: return IntegrationTestResult( ok=False, message=f"base_url rejected by SSRF guard: {exc.message}", ) auth_scheme = (self.config.get("auth_scheme") or "bearer").lower() if auth_scheme == "basic": api_username = self.config.get("api_username") or "" if not api_username: return IntegrationTestResult( ok=False, message="api_username is required for basic auth", ) creds = f"{api_username}:{api_token}".encode() auth_header = f"Basic {base64.b64encode(creds).decode('ascii')}" elif auth_scheme == "bearer": auth_header = f"Bearer {api_token}" else: return IntegrationTestResult( ok=False, message=f"unsupported auth_scheme {auth_scheme!r}", ) headers = { "Authorization": auth_header, "Accept": "application/scim+json, application/json", } extra = self.config.get("extra_headers") or {} if isinstance(extra, dict): for k, v in extra.items(): if isinstance(k, str) and isinstance(v, str): headers[k] = v try: # follow_redirects=False + trust_env=False — same SSRF / # ambient-proxy posture as the runner. We don't re-check # the base_url against private-IP ranges here because # this driver is only constructed AFTER the integrations # service decrypts the row; the canonical SSRF guard # lives in the identity_sync service validator at save # time. test_connection is the post-save smoke check. async with httpx.AsyncClient( timeout=10.0, verify=tls_verify, follow_redirects=False, trust_env=False, transport=transport, ) as client: resp = await client.get( f"{base_url}/Users", params={"count": 1, "startIndex": 1}, headers=headers, ) except httpx.HTTPError as e: return IntegrationTestResult( ok=False, message=_redact_connection_error( f"transport error: {type(e).__name__}: {e!s:.200}", str(api_token), auth_header, ), ) except asyncio.CancelledError: raise if resp.status_code == 200: try: body = resp.json() except ValueError: return IntegrationTestResult( ok=False, message="endpoint returned 200 but body was not JSON", ) if not isinstance(body, dict): return IntegrationTestResult( ok=False, message="endpoint returned 200 but body was not a SCIM ListResponse object", ) return IntegrationTestResult( ok=True, message=( f"OK · totalResults={body.get('totalResults', '?')} " f"· itemsPerPage={body.get('itemsPerPage', '?')}" ), ) body_excerpt = _redact_connection_error( resp.text[:300] if resp.text else "(empty body)", str(api_token), auth_header, ) return IntegrationTestResult( ok=False, message=f"HTTP {resp.status_code}: {body_excerpt}", ) @register_driver class NavanDriver(IdentitySyncEndpointDriver): provider = "navan" display_name = "Navan (Travel & Expense)" description = ( "Navan Universal SCIM 2.0 endpoint " "(app.navan.com/universal/scim/v2). Create an " "API token in Navan Admin → Integrations and paste it below. " "Then attach this integration to a Navan target in the " "Identity Sync Center to push employee directory data." ) @register_driver class ScimGenericDriver(IdentitySyncEndpointDriver): provider = "scim_generic" display_name = "Generic SCIM 2.0 Endpoint" description = ( "Any SCIM 2.0-compliant outbound endpoint (Okta, Workday " "Outbound, Rippling, custom HRMS connectors). Configure " "base_url + api_token; the Identity Sync Center maps " "IdentityPerson fields to SCIM userName / name / " "enterprise extension via per-target mapping rules." )