"""Slack software-provisioning driver. One integration row stores up to three credentials, each optional, so the driver degrades gracefully to whatever the workspace plan allows: * ``bot_token`` (``xoxb-...``) — required. Used for ``auth.test`` probe and for ``list_users`` (the scheduled sync job needs it). Every Slack workspace plan supports these read scopes. * ``user_token`` (``xoxp-...``) — optional. If present AND the workspace is on the Business+ plan or Enterprise Grid, the driver can invite / deactivate users via SCIM (plan-gated server-side). * ``scim_base_url`` — optional override for Enterprise Grid customers with a custom SCIM endpoint. Defaults to Slack's public SCIM v1 base. Because PingCAP's current workspace is on the Pro plan (no SCIM — Slack returns ``plus_teams_only`` for every SCIM call), provisioning runs in a **manual_notify** fallback mode: a human IT admin receives a Feishu DM with "please invite X@pingcap.com" and marks the task done in the Helix provisioning admin page. This is selected via the software_connectors binding config ``execution_mode = "manual_notify"`` (see migration 056). On the Plus/Enterprise plans the admin flips the binding to ``execution_mode = "auto_api"`` and no code changes are needed — the ``grant_seat`` / ``revoke_seat`` methods below will start being invoked and SCIM will fire. SCIM reference https://api.slack.com/admins/scim (v1; Slack has no v2) Web API reference https://api.slack.com/methods/users.list https://api.slack.com/methods/auth.test """ from __future__ import annotations import logging import re from typing import Any from urllib.parse import quote as _url_quote import httpx from app.integrations.driver_base import ( ProvisioningContext, ProvisioningResult, SoftwareProvisioningDriver, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _DEFAULT_SCIM_BASE = "https://api.slack.com/scim/v1" _SLACK_WEB_BASE = "https://slack.com/api" _HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0) _SCIM_CONTENT = "application/json" # Slack SCIM v1 uses plain JSON # Hard cap on pagination iterations. A buggy vendor response that # returns the same cursor twice or an unbounded next_cursor chain # would otherwise wedge the worker forever. 200 pages × 200/page is # 40k entries — well above any realistic 500-employee tenant. _MAX_PAGINATOR_PAGES = 200 # ── SCIM retry / error tags (Round-1 audit hardening) ──────────── # # Slack SCIM endpoints (Enterprise Grid / Business+) inherit the # tier-3 rate budget (~50 RPM) plus per-second burst caps. A single # leaver sweep that fires N SCIM revokes can pile up 429s during # peak hours, and 5xx happens during Slack's data-center cutovers. # Without retry the offboarding tasks dead-letter immediately and # admin sees N "transient" failures across the next hour. Mirrors # the pattern already used by ``list_full_members`` (line ~340). _SCIM_MAX_429_RETRIES = 5 _SCIM_MAX_5XX_RETRIES = 3 _SCIM_DEFAULT_BACKOFF = 1.0 _SCIM_MAX_BACKOFF = 30.0 # Error tags. Pairs with new entries in # ``app.domains.provisioning.service._PROVISIONING_REMEDIATION_HINTS``. SLACK_TAG_RATE_LIMITED = "SLACK_RATE_LIMITED" SLACK_TAG_SCIM_UNAVAILABLE = "SLACK_SCIM_UNAVAILABLE" async def _scim_request_with_retry( method: str, url: str, *, headers: dict[str, str], json_body: dict[str, Any] | None = None, params: dict[str, Any] | None = None, ) -> tuple[httpx.Response | None, str | None]: """Issue a SCIM HTTP request with bounded 429 + 5xx retry. Returns ``(response, None)`` on a non-retried response (caller decides ok-ness), or ``(None, error_message)`` when the budget is exhausted / a transport exception is unrecoverable. The ``error_message`` is pre-tagged with ``[SLACK_RATE_LIMITED]`` when relevant so callers can pass it through ``ProvisioningResult. error_message`` unmodified and the failure-DM dispatcher will append the matching remediation hint. Mirrors the 429 loop already in ``list_full_members`` (line ~340) — same Slack ``Retry-After`` header handling, same 30s upper bound on backoff so a wedged Slack tier doesn't hang the worker beyond the dispatcher's tick budget. """ import asyncio as _asyncio # noqa: PLC0415 — local to defer import attempt_429 = 0 attempt_5xx = 0 while True: try: from app.integrations._url_safety import safe_async_http_transport transport = await safe_async_http_transport( url, allowed_suffixes=("slack.com",), ) async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, transport=transport, ) as client: resp = await client.request( method, url, headers=headers, json=json_body, params=params, ) except httpx.HTTPError as e: return None, f"Slack SCIM {method} failed: {type(e).__name__}: {e}" if resp.status_code == 429 and attempt_429 < _SCIM_MAX_429_RETRIES: # Increment BEFORE the sleep/log so on-budget-exhaust the # log line shows the correct total attempt count (Round-3 # logic scan MEDIUM #1 — off-by-one in audit trail). attempt_429 += 1 retry_after = resp.headers.get("Retry-After") or "1" try: delay = float(retry_after) except ValueError: delay = _SCIM_DEFAULT_BACKOFF delay = min(max(delay, _SCIM_DEFAULT_BACKOFF), _SCIM_MAX_BACKOFF) logger.warning( "Slack SCIM %s %s 429 (attempt %d/%d); sleeping %.1fs", method, url, attempt_429, _SCIM_MAX_429_RETRIES, delay, ) await _asyncio.sleep(delay) continue if ( 500 <= resp.status_code < 600 and attempt_5xx < _SCIM_MAX_5XX_RETRIES ): # Same pre-sleep increment for symmetric log accuracy. attempt_5xx += 1 # Slack data-center cutover / overload — exponential backoff # capped at _SCIM_MAX_BACKOFF. Note the back-off uses # ``attempt_5xx - 1`` so attempt 1 sleeps 1s (2**0), attempt # 2 sleeps 2s, attempt 3 sleeps 4s — preserves the original # exponential schedule despite the pre-increment. delay = min( _SCIM_DEFAULT_BACKOFF * (2 ** (attempt_5xx - 1)), _SCIM_MAX_BACKOFF, ) logger.warning( "Slack SCIM %s %s %d (attempt %d/%d); sleeping %.1fs", method, url, resp.status_code, attempt_5xx, _SCIM_MAX_5XX_RETRIES, delay, ) await _asyncio.sleep(delay) continue # 429 / 5xx budget exhausted → tag the error so failure-DM # surfaces a "Slack throttle, retry later" hint instead of a # raw vendor body. if resp.status_code == 429 or 500 <= resp.status_code < 600: return resp, ( f"[{SLACK_TAG_RATE_LIMITED}] Slack SCIM {method} " f"returned {resp.status_code} after retries; " "Slack throttled / unavailable. Retry the revoke " "later — no admin action required for the throttle " "itself, but check Slack's status page if " "persistent." ) return resp, None def _redact_secrets(blob: str) -> str: """Strip Slack tokens and Authorization headers from error snippets. Slack SCIM error responses occasionally echo back request metadata including token values. This prevents them from leaking into provisioning_tasks.error_message or admin UI error surfaces. """ if not blob: return blob # Authorization: Bearer xxx → "Authorization: Bearer ***" blob = re.sub( r"(?i)(authorization\s*:?\s*(?:bearer|basic)\s+)[A-Za-z0-9._\-+/=]+", r"\1***", blob, ) # Slack tokens (xoxb-/xoxp-/xoxa-/xapp-) blob = re.sub(r"\bxox[abps]-[A-Za-z0-9-]{8,}\b", "xox*-***", blob) return blob class SlackSCIMUnavailable(RuntimeError): """Slack SCIM is not accessible on this workspace plan. Raised by driver helpers that must fail fast rather than returning a generic HTTP error, so the caller can steer the task into the manual_notify fallback queue instead of retrying a provisioning call that will never succeed until the workspace upgrades. """ @register_driver class SlackSoftwareDriver(SoftwareProvisioningDriver): provider = "slack" display_name = "Slack" description = ( "Slack 工作区成员管理。Bot Token (xoxb-) 用于同步成员列表;" "User Token (xoxp- 带 admin scope) + Slack Business+/Enterprise " "Grid 档位才能通过 SCIM 自动 invite/deactivate。Pro 档下坐席申请" "走 manual_notify 通知 IT 管理员手工操作(绑定里选 execution_mode=manual_notify)。" ) supports_seat = True supports_permission = False supports_multi_instance = True config_schema = [ IntegrationFieldInfo( key="bot_token", label="Bot Token (xoxb-)", is_secret=True, required=True, placeholder="xoxb-...", hint=( "Slack App OAuth & Permissions > Bot User OAuth Token。" "需要 scopes: users:read, users:read.email, chat:write。" "所有 Slack 档位都支持。用于读成员列表 + 发 DM。" ), ), IntegrationFieldInfo( key="user_token", label="User Token (xoxp-)", is_secret=True, required=False, placeholder="xoxp-... (可选,Business+ / Enterprise Grid 才生效)", hint=( "可选。Workspace owner/admin 安装 app 时授权的 User OAuth " "Token,且 User Token Scopes 里勾选了 admin。Pro 档填了也" "没用(Slack 会返回 plus_teams_only)。升级到 Business+ 后" "把绑定 execution_mode 改为 auto_api 即可自动 invite。" ), ), IntegrationFieldInfo( key="scim_base_url", label="SCIM Base URL", is_secret=False, required=False, default=_DEFAULT_SCIM_BASE, placeholder=_DEFAULT_SCIM_BASE, hint="通常留空。Enterprise Grid 自定义 SCIM endpoint 才需要改。", ), ] binding_schema = [ IntegrationFieldInfo( key="execution_mode", label="执行模式", is_secret=False, required=True, default="manual_notify", options=[ { "value": "manual_notify", "label": "手动通知 (Pro 档位必选)", }, { "value": "auto_api", "label": "SCIM 自动 (Business+ / Enterprise Grid 才可用)", }, ], hint=( "Pro 档没有 SCIM API,必须选 manual_notify: 坐席申请通过" "后,Helix 通过飞书 DM 通知 IT 管理员手工邀请,完成后在" "「/admin/provisioning」点「完成」。升到 Business+ 后可改为 " "auto_api 全自动。" ), ), IntegrationFieldInfo( key="action_on_offboard", label="离职动作", is_secret=False, required=True, default="deactivate", options=[ {"value": "deactivate", "label": "停用 (active=false,保留数据)"}, {"value": "delete", "label": "删除 (彻底移除用户)"}, ], hint=( "deactivate 是标准选择;delete 仅用于合规彻底移除。" "仅在 execution_mode=auto_api 时生效。" ), ), ] # ── Config accessors ────────────────────────────────────────── def _bot_token(self) -> str: v = self.config.get("bot_token") if not v: raise ValueError("Slack integration missing bot_token") return str(v) def _user_token(self) -> str | None: v = self.config.get("user_token") return str(v) if v else None async def _scim_base(self) -> str: url = str( self.config.get("scim_base_url") or _DEFAULT_SCIM_BASE, ).rstrip("/") return url async def _scim_transport(self): from app.integrations._url_safety import safe_async_http_transport return await safe_async_http_transport( await self._scim_base(), allowed_suffixes=("slack.com",), ) def _bot_headers(self) -> dict[str, str]: return { "Authorization": f"Bearer {self._bot_token()}", "Accept": "application/json", } def _scim_headers(self) -> dict[str, str]: token = self._user_token() if not token: raise SlackSCIMUnavailable( "Slack integration has no user_token (xoxp-); " "SCIM calls require it.", ) return { "Authorization": f"Bearer {token}", "Accept": _SCIM_CONTENT, "Content-Type": _SCIM_CONTENT, } @staticmethod def _is_plus_teams_only(status_code: int, body: str) -> bool: """Slack returns 403 with ``plus_teams_only`` when the SCIM endpoint is plan-gated. Detect so callers can steer into the manual fallback instead of surfacing a cryptic 403. """ if status_code != 403: return False return "plus_teams_only" in (body or "") # ── test_connection ────────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: """Probe the bot token, and opportunistically SCIM if a user token is configured. Reports what's usable so the admin can tell at a glance whether the binding needs manual_notify or auto_api. """ try: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, ) as client: auth_resp = await client.post( f"{_SLACK_WEB_BASE}/auth.test", headers=self._bot_headers(), ) except httpx.HTTPError as e: return IntegrationTestResult( ok=False, message=f"auth.test failed: {type(e).__name__}: {e}", ) except Exception as e: # noqa: BLE001 — config error, etc return IntegrationTestResult( ok=False, message=f"auth.test failed: {type(e).__name__}: {e}", ) if auth_resp.status_code != 200: return IntegrationTestResult( ok=False, message=f"Slack auth.test HTTP {auth_resp.status_code}: {_redact_secrets(auth_resp.text[:200])}", ) auth_data = auth_resp.json() or {} if not auth_data.get("ok"): return IntegrationTestResult( ok=False, message=f"Slack auth.test error: {auth_data.get('error')!r}", detail=auth_data, ) team = auth_data.get("team") or "" team_url = auth_data.get("url") or "" scim_status: str if self._user_token() is None: scim_status = "no user_token → manual_notify only" else: scim_status = await self._probe_scim() return IntegrationTestResult( ok=True, message=f"OK · team={team} · SCIM: {scim_status}", detail={ "team": team, "team_url": team_url, "scim_status": scim_status, "is_enterprise_install": bool( auth_data.get("is_enterprise_install"), ), }, ) async def _probe_scim(self) -> str: try: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, transport=await self._scim_transport(), ) as client: r = await client.get( f"{await self._scim_base()}/Users", headers=self._scim_headers(), params={"count": 1}, ) except SlackSCIMUnavailable as e: return f"unavailable ({e})" except httpx.HTTPError as e: return f"probe error: {type(e).__name__}: {e}" if r.status_code == 200: return "available (Business+ / Enterprise Grid)" if self._is_plus_teams_only(r.status_code, r.text): return "plus_teams_only (Pro plan — upgrade to Business+ for SCIM)" return f"HTTP {r.status_code}: {_redact_secrets(r.text[:120])}" # ── list_users (bot-token-based, always available) ──────────── async def list_full_members(self) -> list[dict[str, Any]]: """Return every active full workspace member. Filters out: * deleted users (``deleted=True``) * bots (``is_bot=True`` or the reserved ``USLACKBOT`` id) * single-channel / multi-channel guests (``is_restricted`` / ``is_ultra_restricted`` true). Guests are typically free/low-cost and tracking them as seats adds noise. Uses ``users.list`` with cursor pagination. Requires ``users:read`` and, for the email field (which is what Helix matches against ``identity_persons.work_email``), ``users:read.email``. Both are granted on the bot token. """ out: list[dict[str, Any]] = [] cursor: str | None = None seen_cursors: set[str] = set() # Max-page cap (P1 from CR2): a buggy Slack response that # returns the same cursor twice or an unbounded next_cursor # chain would otherwise wedge the worker forever. 200 pages × # 200 members/page = 40k headroom — well above any realistic # tenant size for a 500-employee company. # Plus: Cap the per-call retry budget so a sustained 429 storm # raises rather than spinning forever — a 429-flood on # users.list must NOT silently degrade to "no members" or the # seat-sync sweep mass-revokes every Slack seat. max_429_retries = 5 for _page in range(_MAX_PAGINATOR_PAGES): params: dict[str, Any] = {"limit": 200} if cursor: params["cursor"] = cursor attempt_429 = 0 while True: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, ) as client: resp = await client.get( f"{_SLACK_WEB_BASE}/users.list", headers=self._bot_headers(), params=params, ) if resp.status_code == 429 and attempt_429 < max_429_retries: # Honor Slack's ``Retry-After`` header (seconds). # Without this, the previous bare ``raise`` aborted # the whole sync — the seat allocation reconciliation # then dropped to 0 members and downstream # "revoke seats not in tenant" mass-revoked every # Slack seat. retry_after = resp.headers.get("Retry-After") or "1" try: delay = float(retry_after) except ValueError: delay = 1.0 delay = min(max(delay, 1.0), 30.0) import asyncio as _asyncio # noqa: PLC0415 await _asyncio.sleep(delay) attempt_429 += 1 continue break if resp.status_code != 200: # Loud failure — the caller MUST treat this as # "transient, do NOT interpret as authoritative empty". # Seat-sync handler maps this to a sync error rather # than a successful 0-member snapshot. raise RuntimeError( f"Slack users.list HTTP {resp.status_code} " "(transient — caller must abort, NOT interpret as " f"authoritative empty): {_redact_secrets(resp.text[:200])}", ) data = resp.json() or {} if not data.get("ok"): raise RuntimeError( f"Slack users.list error: {data.get('error')!r}", ) for m in data.get("members") or []: if not isinstance(m, dict): continue if m.get("deleted"): continue if m.get("is_bot"): continue if m.get("id") == "USLACKBOT": continue if m.get("is_restricted") or m.get("is_ultra_restricted"): continue out.append(m) cursor = ( ((data.get("response_metadata") or {}).get("next_cursor")) or None ) if not cursor: break if cursor in seen_cursors: raise RuntimeError( "Slack users.list returned a repeating cursor " f"(len={len(cursor)}); aborting to avoid infinite " "loop.", ) seen_cursors.add(cursor) else: raise RuntimeError( f"Slack users.list paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages; likely a buggy response.", ) return out # ── grant_seat / revoke_seat (SCIM) ─────────────────────────── # # These are only invoked when the binding's ``execution_mode`` is # ``auto_api`` — the coordinator short-circuits to manual_notify # otherwise. If SCIM is gated (Pro plan), we return a structured # error mentioning plus_teams_only so admin sees the real reason. async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: """Invite a user via SCIM v1 POST /Users. On 409 (user already exists), look them up and PATCH active=true so retries after a prior deactivation are idempotent. """ try: headers = self._scim_headers() except SlackSCIMUnavailable as e: return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( f"Slack SCIM not usable: {e}. Flip binding " "execution_mode to manual_notify." ), ) # Normalise the email before sending — Slack stores userName # case-insensitively, but keeping our own payloads lowercase # makes external_ref values stable and matches the email we # pass to ``_scim_find_user`` when the 409 fallback fires. email_norm = (ctx.user_email or "").lower() display = (ctx.user_display_name or email_norm.split("@", 1)[0]).strip() first, _, last = display.partition(" ") payload = { "schemas": ["urn:scim:schemas:core:1.0"], "userName": email_norm, "name": { "givenName": first or email_norm, "familyName": last or "-", }, "displayName": display, "emails": [{"value": email_norm, "primary": True}], "active": True, } try: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, transport=await self._scim_transport(), ) as client: resp = await client.post( f"{await self._scim_base()}/Users", headers=headers, json=payload, ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) if resp.status_code in (200, 201): data = resp.json() or {} return ProvisioningResult( ok=True, external_ref=str(data.get("id") or ctx.user_email), raw_response=data, ) if resp.status_code == 409: return await self._reactivate_existing(ctx, headers=headers) if self._is_plus_teams_only(resp.status_code, resp.text): return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( "Slack SCIM is plan-gated (plus_teams_only). Either " "upgrade this Slack workspace to Business+/Enterprise " "Grid, or flip the Slack connector binding " "execution_mode to manual_notify." ), raw_response={"status": 403, "slack_error": "plus_teams_only"}, ) return ProvisioningResult( ok=False, error_message=f"Slack SCIM create {resp.status_code}: {_redact_secrets(resp.text[:300])}", raw_response={"status": resp.status_code, "body": _redact_secrets(resp.text[:300])}, ) async def _reactivate_existing( self, ctx: ProvisioningContext, *, headers: dict[str, str], ) -> ProvisioningResult: find = await self._scim_find_user(ctx.user_email, headers=headers) if not find.ok: return find.to_result() user = find.user if user is None: return ProvisioningResult( ok=False, error_message=( f"Slack SCIM create returned 409 but lookup found 0 " f"users for {ctx.user_email!r}" ), ) user_id = user.get("id") if not user_id: return ProvisioningResult( ok=False, error_message=( f"Slack SCIM lookup for {ctx.user_email!r} returned a " "user record without an id field — cannot reactivate" ), raw_response=user, ) if bool(user.get("active")): return ProvisioningResult( ok=True, external_ref=str(user_id), raw_response={"already_active": True}, ) # SCIM v1 accepts a direct PUT on active; PATCH syntax isn't # universally supported by Slack's SCIM v1 implementation, so # we use PUT with a MINIMAL body. Sending ``{**user, "active": # True}`` round-trips server-controlled fields (``meta``, # ``groups``, computed timestamps) — SCIM v1 treats PUT as a # full replacement, so any read-only field Slack returned could # be rejected (occasional 400s on edge fields) or — worse — # accepted as a write. Build the smallest schema-compliant body # we can. # Round-49: Slack SCIM v1 legacy records sometimes omit # ``userName``. Sending ``"userName": null`` would either be # rejected (400, lost reactivation) or silently clear the # workspace identity (decoupling Slack user from email). # Fall back to the email's canonical lowercase — Slack # convention sets ``userName`` to the email anyway. If that # too is missing, fail loud rather than ship null. scim_username = user.get("userName") if not isinstance(scim_username, str) or not scim_username.strip(): fallback = (ctx.user_email or "").strip().lower() if not fallback: return ProvisioningResult( ok=False, error_message=( f"Slack SCIM lookup for {ctx.user_email!r} returned " "a user record without ``userName`` and no fallback " "email is available — refusing to PUT a null-valued " "userName which would clear Slack's workspace identity" ), raw_response=user, ) scim_username = fallback minimal_body = { "schemas": ["urn:scim:schemas:core:1.0"], "id": user_id, "userName": scim_username, "active": True, } try: async with httpx.AsyncClient( timeout=_HTTP_TIMEOUT, trust_env=False, transport=await self._scim_transport(), ) as client: put = await client.put( f"{await self._scim_base()}/Users/{_url_quote(str(user_id), safe='')}", headers=headers, json=minimal_body, ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"Slack SCIM reactivate failed: {type(e).__name__}: {e}", ) if put.status_code in (200, 204): # Pre-launch audit Round-29 fix: validate Slack actually # honoured the active=True PUT. SCIM v1 implementations can # silently ignore the minimal body and return 200 with the # user still inactive — we'd then mark the seat reactivated # in Helix while Slack still treats the user as suspended. # 204 (No Content) has no body to inspect; trust the status # code in that case since Slack never returns 204 today (it # always returns 200 + JSON body), but defend in depth by # treating an explicit ``active=False`` in the response body # as a failure. try: body_json = put.json() if put.status_code == 200 else {} except Exception: # noqa: BLE001 — malformed JSON: trust status body_json = {} active_resp = body_json.get("active") if isinstance(body_json, dict) else None if active_resp is False: return ProvisioningResult( ok=False, error_message=( "Slack SCIM PUT returned 200 but body still " "reports active=false — reactivation rejected " "silently. Check Slack workspace SCIM settings." ), raw_response={"reactivated": False, "body": body_json}, ) return ProvisioningResult( ok=True, external_ref=str(user_id), raw_response={"reactivated": True}, ) return ProvisioningResult( ok=False, error_message=f"Slack SCIM reactivate {put.status_code}: {_redact_secrets(put.text[:200])}", ) async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: """Deactivate (PUT active=false) or delete per binding config.""" try: headers = self._scim_headers() except SlackSCIMUnavailable as e: return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( f"Slack SCIM not usable: {e}. Flip binding " "execution_mode to manual_notify." ), ) action = str( (ctx.extra or {}).get("action_on_offboard") or "deactivate", ).lower() if action not in ("deactivate", "delete"): return ProvisioningResult( ok=False, error_message=( f"Slack binding invalid action_on_offboard {action!r} " "(expected deactivate|delete)" ), ) find = await self._scim_find_user(ctx.user_email, headers=headers) if not find.ok: if find.plus_teams_only: return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( "Slack SCIM is plan-gated (plus_teams_only); " "flip the Slack connector binding execution_mode " "to manual_notify." ), raw_response={ "status": 403, "slack_error": "plus_teams_only", }, ) return find.to_result() if find.user is None: # Already absent — idempotent success for offboarding. return ProvisioningResult( ok=True, external_ref=f"slack:{(ctx.user_email or '').lower()}:already_absent", raw_response={"already_absent": True}, ) user_id = find.user.get("id") if not user_id: return ProvisioningResult( ok=False, error_message=( f"Slack SCIM lookup for {ctx.user_email!r} returned a " "user record without an id field — cannot revoke" ), raw_response=find.user, ) if action == "deactivate": # Same minimization as ``_reactivate_existing`` — never # round-trip the full resource on a PUT. # Fall back to ctx.user_email if the SCIM record's userName # is missing (non-SCIM-spec, but defensive): a null userName in # the PUT body can cause some SCIM servers to silently reject # the active=false field while returning a 200. user_name = find.user.get("userName") or ctx.user_email minimal_body = { "schemas": ["urn:scim:schemas:core:1.0"], "id": user_id, "userName": user_name, "active": False, } put, err = await _scim_request_with_retry( "PUT", f"{await self._scim_base()}/Users/{_url_quote(str(user_id), safe='')}", headers=headers, json_body=minimal_body, ) if put is None: return ProvisioningResult( ok=False, error_message=err or "Slack SCIM deactivate failed", ) if put.status_code in (200, 204): # Validate that the response confirms active=False. # A 200 with active still True means the vendor silently # ignored the deactivation (no content on 204 — skip check). if put.status_code == 200: resp_body = put.json() if put.content else {} if resp_body.get("active") is True: return ProvisioningResult( ok=False, error_message=( "[SLACK_DEACTIVATE_SILENT_FAIL] Slack returned " "200 but active=true in the response body — " "deactivation was not applied. Retry or " "deactivate manually in Slack admin." ), raw_response={ "error_tag": "SLACK_DEACTIVATE_SILENT_FAIL", "response_active": resp_body.get("active"), }, ) logger.info( "slack scim deactivate: user_id=%s email=%s", user_id, ctx.user_email, ) return ProvisioningResult( ok=True, external_ref=str(user_id), raw_response={"deactivated": True}, ) # TOCTOU between _scim_find_user and PUT — admin removed the # user out-of-band between lookup and update. Same idempotency # contract as the `delete` branch below: desired end-state # (user not active in workspace) is already met. Without this, # offboarding blockers stay open for users whose Slack account # was hard-deleted between the find and the PUT. # # external_ref uses the synthetic email-keyed form to match # the upstream `find.user is None` already-absent path # (line ~588) — by the time SCIM PUT 404s, the user_id we # captured from the find may already be reaped on Slack's # side, so an audit replay keying on it would 404 again. if put.status_code == 404: return ProvisioningResult( ok=True, external_ref=f"slack:{(ctx.user_email or '').lower()}:already_absent", raw_response={"already_absent": True, "user_id": user_id}, ) # If retry helper tagged the error (429/5xx budget # exhausted), surface its message unchanged so the failure- # DM dispatcher reads the [SLACK_RATE_LIMITED] tag. if err is not None: return ProvisioningResult( ok=False, error_message=err, raw_response={ "error_tag": SLACK_TAG_RATE_LIMITED, "status": put.status_code, }, ) return ProvisioningResult( ok=False, error_message=f"Slack SCIM deactivate {put.status_code}: {_redact_secrets(put.text[:200])}", ) # action == "delete" delete, err = await _scim_request_with_retry( "DELETE", f"{await self._scim_base()}/Users/{_url_quote(str(user_id), safe='')}", headers=headers, ) if delete is None: return ProvisioningResult( ok=False, error_message=err or "Slack SCIM delete failed", ) if delete.status_code in (200, 204): logger.info( "slack scim delete: user_id=%s email=%s", user_id, ctx.user_email, ) return ProvisioningResult( ok=True, external_ref=str(user_id), raw_response={"deleted": True}, ) if delete.status_code == 404: return ProvisioningResult( ok=True, external_ref=str(user_id), raw_response={"already_absent": True}, ) if err is not None: return ProvisioningResult( ok=False, error_message=err, raw_response={ "error_tag": SLACK_TAG_RATE_LIMITED, "status": delete.status_code, }, ) return ProvisioningResult( ok=False, error_message=f"Slack SCIM delete {delete.status_code}: {_redact_secrets(delete.text[:200])}", ) @staticmethod def _escape_scim_filter_value(v: str) -> str: """Escape backslash + double-quote in a SCIM filter string value. Prevents injection when the caller-supplied email contains a quote (RFC 3986 allows it in the local-part). Without this a crafted email like ``foo"or 1 eq 1"@bar.com`` would break out of the ``userName eq "…"`` boundary. """ return v.replace("\\", "\\\\").replace('"', '\\"') async def _scim_find_user( self, email: str, *, headers: dict[str, str], ) -> _ScimFindResult: email_norm = (email or "").lower() filter_value = self._escape_scim_filter_value(email_norm) # Round-1 audit CRITICAL: Slack SCIM lookups must retry 429/5xx # — otherwise a throttle blip in the find step instantly # dead-letters the offboarding revoke. Pattern mirrors # list_full_members (already retries) and the new PUT / DELETE # paths below. resp, err = await _scim_request_with_retry( "GET", f"{await self._scim_base()}/Users", headers=headers, params={"filter": f'userName eq "{filter_value}"'}, ) if resp is None: return _ScimFindResult(ok=False, error=err or "Slack SCIM lookup failed") r = resp if self._is_plus_teams_only(r.status_code, r.text): return _ScimFindResult( ok=False, error="plus_teams_only", plus_teams_only=True, ) if r.status_code != 200: return _ScimFindResult( ok=False, error=( err or ( f"Slack SCIM lookup for {email!r} returned " f"{r.status_code}: {_redact_secrets(r.text[:200])}" ) ), ) resources = (r.json() or {}).get("Resources") or [] if len(resources) > 1: logger.warning( "Slack SCIM lookup for %r returned %d users; expected 0 or 1", email, len(resources), ) # Return the failure through the result contract instead of # raising — callers (revoke_seat, _reactivate_existing) # consume _ScimFindResult without a try/except, so a raise # here becomes an unhandled traceback that dead-letters the # whole task. Surface it as a normal handled failure. return _ScimFindResult( ok=False, error=( "ambiguous SCIM results: multiple Slack accounts for " "this email" ), ) return _ScimFindResult(ok=True, user=resources[0] if resources else None) class _ScimFindResult: """Local result holder for ``_scim_find_user`` — avoids tuple unpacking and keeps the plus_teams_only signal out of generic error strings. """ __slots__ = ("ok", "user", "error", "plus_teams_only") def __init__( self, *, ok: bool, user: dict[str, Any] | None = None, error: str | None = None, plus_teams_only: bool = False, ) -> None: self.ok = ok self.user = user self.error = error self.plus_teams_only = plus_teams_only def to_result(self) -> ProvisioningResult: return ProvisioningResult( ok=False, error_message=self.error or "Slack SCIM lookup failed", )