"""Zoom software-provisioning driver (Server-to-Server OAuth). One integration row carries the Zoom Server-to-Server OAuth app credentials (account_id + client_id + client_secret). Per-software bindings carry the desired user type and offboarding action: {"plan_type": "licensed", "action_on_offboard": "disassociate"} {"plan_type": "basic", "action_on_offboard": "delete"} ``plan_type`` values map to Zoom's user ``type`` field: basic → 1 (Basic — free, but see the cap note below) licensed → 2 (Licensed — paid seat) no_meetings → 4 (account with NO Meetings license — the collaboration-only baseline that Zoom's 2024 permissions-based relicensing left most users on) none → 99 (None — keeps a record without a usable seat) **Basic (type 1) is capped per tenant and is NOT the free baseline.** A Zoom plan grants a small fixed number of Basic users (20 on the PingCAP tenant); exceeding it makes both create-as-Basic and Licensed→Basic downgrade fail with code 3412. ``no_meetings`` (4) is the uncapped baseline — 397 of 612 users on the prod tenant sit there, vs exactly 20 on Basic. Prefer it for "give every employee an account" flows; keep Basic for the seat-flow's Licensed→Basic release path. ``action_on_offboard``: disassociate → user is unlinked from the account (default; preserves their personal data, cloud recordings stay accessible to admins during the retention window). delete → user is fully deleted from the account. API reference: https://developers.zoom.us/docs/api/rest/reference/user/methods/ Auth reference: https://developers.zoom.us/docs/internal-apps/s2s-oauth/ """ from __future__ import annotations import asyncio import base64 import logging import re from collections.abc import Awaitable, Callable from dataclasses import dataclass from datetime import UTC, datetime, timedelta from typing import Any from urllib.parse import 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__) _ZOOM_BASE = "https://api.zoom.us/v2" # Hard cap on pagination iterations — see slack_software.py for the # rationale. 200 pages × 300 users/page = 60k user buffer. _MAX_PAGINATOR_PAGES = 200 _ZOOM_TOKEN_URL = "https://zoom.us/oauth/token" _HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0) # Retry policy for genuine HTTP 429 rate-limit responses from Zoom's # API. Each attempt backs off exponentially from _ZOOM_RETRY_BASE_DELAY, # capped at _ZOOM_RETRY_MAX_DELAY. The Retry-After header (if present) # overrides the computed delay. Mass-onboarding/offboarding bursts can # exhaust Zoom's Light-tier RPS budget (~80 RPS); retrying with backoff # avoids permanent failure for the whole batch. Maximum total wait: # attempt 0→1: 2s, 1→2: 4s, 2→3: 8s → up to 14s added latency per # task under full-rate-limit, well within reasonable task timeout. _ZOOM_MAX_RETRIES = 3 _ZOOM_RETRY_BASE_DELAY = 2.0 _ZOOM_RETRY_MAX_DELAY = 30.0 # Zoom user.type enum (their numbering, not ours). # # ``no_meetings`` (4) is undocumented in the public user-type enum # (which still lists only 1/2/3/99) but is what Zoom's June–August 2024 # permissions-based relicensing actually returns for a user who holds a # Zoom Workplace account with no Meetings license. It is the majority # tier on the prod tenant (397/612 users as of 2026-07-29) and, unlike # Basic, is not capped — so it, not Basic, is the "everyone gets an # account" baseline. Kept in the map (rather than special-cased at the # call sites) so ``_PLAN_BY_TYPE`` renders a real label instead of the # bare "4" the vendor-snapshot column used to show. _TYPE_BY_PLAN = { "basic": 1, "licensed": 2, "no_meetings": 4, "none": 99, } _PLAN_BY_TYPE = {v: k for k, v in _TYPE_BY_PLAN.items()} # Zoom user.type ordering for "is this a downgrade?" check. # Higher rank = more privileged seat. Demoting a Licensed (2) user # to Basic (1) silently was a real footgun: an admin who picks the # wrong binding when granting can wipe a paid seat without warning. # We refuse the demotion path; an explicit revoke + re-grant is the # audit-traceable way to change tiers. # # 4 (no Meetings license) and 99 (None) both rank 0: neither can host a # meeting, so moving a Basic/Licensed user onto either is a demotion the # guard must refuse. Ranking 4 explicitly also stops it from falling # through ``.get(..., 0)`` as an *unknown* type — the two cases need to # stay distinguishable if Zoom ever adds another tier. _TYPE_RANK = {1: 1, 2: 2, 4: 0, 99: 0} # ── Structured error tags ──────────────────────────────────────── # # Tagging an error_message with ``[TAG] …`` lets the failure DM # dispatcher (``_dispatch_revoke_failure_dms``) render a clean, # actionable remediation line for the admin without parsing Zoom's # vendor body. The tag is machine-readable AND human-visible — the # admin sees both the tag and the Chinese summary in the Feishu card. # # Bit prod 2026-05-22: P01451 looperth.wong revoke task failed with # managed-domain fallback PATCH user type returned 400 + Zoom code # 3412 (the account had hit its Basic user cap, so the # Licensed→Basic downgrade — the ONLY managed-domain-permitted way # to release the paid seat — was refused). Admin saw a raw Zoom # body and had to look up code 3412 manually. After this PR the # error_message is "[ZOOM_BASIC_CAP_EXCEEDED] Zoom Basic 用户数已达 # 账户上限 (current=N);请在 Zoom Admin 控制台清理闲置 Basic 用户后 #再重试。" and the failure DM appends a remediation hint. ZOOM_TAG_BASIC_CAP_EXCEEDED = "ZOOM_BASIC_CAP_EXCEEDED" ZOOM_TAG_LICENSED_CAP_EXCEEDED = "ZOOM_LICENSED_CAP_EXCEEDED" ZOOM_TAG_USER_IN_OTHER_ACCOUNT = "ZOOM_USER_IN_OTHER_ACCOUNT" # ── Transfer-recipient refusals on DELETE ?action=delete ────────────── # Zoom validates ``transfer_email`` before it deletes anything, and # refuses the WHOLE call when the recipient cannot receive the content. # Both of these mean "the delete did not happen, and the reason is the # RECIPIENT, not the leaver" — a distinction the caller must be able to # read without parsing prose, because the two halves get opposite # treatment (see ``_delete_zoom_account_if_due``): # # ..._NO_WEBINAR: the recipient exists but has no Zoom Webinars # license. Operator policy (2026-08-23) is to delete anyway, # without a transfer, rather than wedge offboarding. # ..._INVALID: anything else about the recipient — chiefly "the data # receiver does not exist", i.e. the resolved successor has no # Zoom account at all. 6 of 416 active people were in exactly # that state on 2026-09-06, so this is one org change away from # live. Deleting without a transfer here would destroy the cloud # recordings the 2026-08-13 decision exists to protect, so the # delete is REFUSED and IT is paged once. ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR = "ZOOM_TRANSFER_RECIPIENT_NO_WEBINAR" ZOOM_TAG_TRANSFER_RECIPIENT_INVALID = "ZOOM_TRANSFER_RECIPIENT_INVALID" # NOTE: an additional tag "ZOOM_MANAGED_DOMAIN_REFUSED" lives in the # DM hint dict (_PROVISIONING_REMEDIATION_HINTS in # app.domains.provisioning.service) as forward-compat for any future # "1107 + PATCH refused for a non-cap reason" code path. No branch # emits it today (1107 always falls through to PATCH=Basic, which # either succeeds or lands in the 3412 / generic failure branches — # both of which already tag correctly), so the Python constant is # omitted to avoid a dead-code smell. # Zoom error codes we explicitly recognise (kept close to the spec # rather than scraping ``message`` text — message strings are # locale-sensitive and the message body has historically changed # between Zoom platform versions, but ``code`` is stable). # # 3412: Basic-user cap reached. Bit us on BOTH paths: # Licensed → Basic downgrade (the managed-domain fallback's # primary failure mode, prod 2026-05-22) AND **create-as-Basic** # (prod 2026-07-24: 16 of 17 identity-sync creates came back # 400/3412 "reached the permitted maximum number of 20 basic # users"). The original comment here assumed create returned # 3413 and only classified that one, so every real create-time # cap refusal fell through untagged — surfaced to the admin as a # raw Zoom body and counted as a retry-worthy ``failed`` instead # of an actionable capacity block. # 3413: documented sibling of 3412 for create-as-Basic. Never observed # in prod, but kept in the set — Zoom has shipped both. # 200 : "User type cannot be modified at this time" — catch-all # when Zoom refuses a PATCH; we distinguish Licensed-cap # vs other-cause by reading ``message``. # 1005: email already exists in ANOTHER Zoom account — admin must # disassociate from the other account first. # 1108: user cannot be associated with this account (same family # as 1005, different Zoom internal state). # 2030: user has been associated with another account. _ZOOM_CODE_BASIC_CAP_ON_DOWNGRADE = 3412 # Every code that means "the Basic pool is full". Membership, not # equality, is the contract — see the 3412-on-create note above. _ZOOM_CODES_BASIC_CAP = {3412, 3413} _ZOOM_CODE_TYPE_PATCH_REFUSED = 200 _ZOOM_CODE_USER_IN_OTHER_ACCOUNT = {1005, 1108, 2030} # Hard cap on how many Zoom users we count for the # "current Basic count" enrichment in 3412 errors. ``list_users`` # already paginates; this is just a sanity ceiling. 60k users >> # any real PingCAP-scale tenant. _ZOOM_BASIC_COUNT_CAP = 60000 # How long one driver instance reuses a Basic headcount. Long enough # that a sweep hitting the cap for many people enumerates once, short # enough that a long-running worker re-reads reality. _BASIC_COUNT_TTL = timedelta(seconds=60) def is_basic_cap_error(result: Any) -> bool: """True when ``result`` failed because the Zoom Basic pool is full. The machine-readable contract for every caller that must distinguish "tenant capacity / wrong tier — a config task for IT" from "transient vendor failure — retry me". Reads the driver-stamped tag rather than the (localised, historically unstable) vendor message. Single home for the predicate: the seat coordinator's ``_result_has_basic_cap_tag`` delegates here, and the identity-sync runner needs the same check but must not import a coordinator. Duck-typed (``Any``) because both ``ProvisioningResult`` and the seat-flow revoke results flow through it. """ raw = getattr(result, "raw_response", None) or {} if isinstance(raw, dict): if raw.get("error_tag") == ZOOM_TAG_BASIC_CAP_EXCEEDED: return True # Defence in depth: a path that bypassed the shared builder still # carries the raw vendor code. if raw.get("zoom_code") in _ZOOM_CODES_BASIC_CAP: return True # Last resort: a future driver revision that drops the tag dict must # not silently disable the capacity-handling paths. msg = getattr(result, "error_message", "") or "" return msg.startswith(f"[{ZOOM_TAG_BASIC_CAP_EXCEEDED}] ") #: The phrase Zoom uses for the ``transfer_email`` party in every #: refusal of this family. Anchoring on the NOUN rather than on a whole #: sentence is deliberate: the one wording we have verified in #: production is "Unable to transfer data because data receiver does not #: have Zoom Webinars licenses." (prod 2026-08-20..08-23), and guessing #: the exact sentence Zoom emits for the other cases would be inventing #: vendor behaviour. Everything that names the data receiver is a #: recipient problem; only the webinar-license variant is sub-classified, #: because only that one has an operator policy attached. _ZOOM_TRANSFER_RECIPIENT_MARKER = "data receiver" def _classify_zoom_transfer_error(zoom_message: str) -> str | None: """Tag a ``DELETE ?action=delete`` failure that is about the ``transfer_email`` RECIPIENT rather than about the user being deleted. Returns ``None`` when the message is not in that family. Only meaningful when the call actually carried a recipient — the caller gates on that, so a leaver-side failure can never be mis-tagged as a recipient refusal. Zoom answers this family with HTTP 400 + ``code: 200`` — the same catch-all code as a refused type PATCH — so the code carries no information here and the message is the only signal. That is a deliberate exception to this module's "classify on code, not on prose" rule, and it is why the tag exists at all: string matching happens ONCE, here, instead of in every caller. See ``_ZOOM_TRANSFER_RECIPIENT_MARKER`` for why the anchor is narrow. """ m = (zoom_message or "").lower() if _ZOOM_TRANSFER_RECIPIENT_MARKER not in m: return None # Truncation-tolerant: prod ledger rows cap the vendor body at 200 # chars, which has historically cut "webinars" mid-word. if "webinar" in m or "webi" in m: return ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR return ZOOM_TAG_TRANSFER_RECIPIENT_INVALID def zoom_transfer_refusal_tag(result: Any) -> str | None: """The transfer-recipient tag on a failed delete, or ``None``. Machine-readable contract for the offboarding sweep, mirroring :func:`is_basic_cap_error`: read the driver-stamped tag rather than re-parsing the (locale-sensitive, historically unstable) vendor message in the caller. Duck-typed for the same reason. The message fallback is defence in depth for a result built by a path that bypassed the tag builder — it must never be the primary signal, and it is gated on evidence that a recipient was actually requested. Without that gate, ``make_room_in_basic_pool``'s no-transfer eviction could be mis-read as a recipient refusal purely because Zoom's message happened to mention the receiver, and the offboarding sweep would then park a delete waiting on a handover person that call never asked for. """ raw = getattr(result, "raw_response", None) or {} if not isinstance(raw, dict): return None tag = raw.get("error_tag") if tag in ( ZOOM_TAG_TRANSFER_RECIPIENT_NO_WEBINAR, ZOOM_TAG_TRANSFER_RECIPIENT_INVALID, ): return str(tag) if not raw.get("transfer_email"): return None message = raw.get("zoom_message") if isinstance(message, str) and message: return _classify_zoom_transfer_error(message) return None def _classify_zoom_user_error( zoom_code: int | None, zoom_message: str, ) -> tuple[str | None, str | None]: """Map a Zoom error ``(code, message)`` onto an admin-actionable ``(tag, remediation_hint)`` pair, or ``(None, None)`` if we don't have a useful classification yet. Centralising this here lets ``grant_seat`` / ``_reconcile_existing_user`` / ``revoke_seat`` share the same admin-facing language. The hint is a one-sentence Chinese remediation pointer (Zoom is the only place the admin can act, regardless of which entry point Helix used). """ if zoom_code in _ZOOM_CODES_BASIC_CAP: return ( ZOOM_TAG_BASIC_CAP_EXCEEDED, "Zoom Basic 用户池已满(该档位有租户上限,不是免费无限档)— " "在 Zoom Admin 清理闲置 Basic 账户,或把基线档位改为 " "no_meetings(无 Meetings 许可,不占 Basic 名额)。", ) if zoom_code in _ZOOM_CODE_USER_IN_OTHER_ACCOUNT: return ( ZOOM_TAG_USER_IN_OTHER_ACCOUNT, "该邮箱已属其他 Zoom 账户 — 让对方账户管理员先 " "disassociate,或在 Zoom 后台提交 transfer 请求。", ) if zoom_code == _ZOOM_CODE_TYPE_PATCH_REFUSED and zoom_message: # Zoom code 200 + a message mentioning "maximum" / "licensed" # is the closest signal for Licensed-cap-hit; otherwise we # leave it untagged (could be a per-user setting conflict # rather than a tenant-wide cap). m = zoom_message.lower() if ("maximum" in m or "limit" in m) and "licensed" in m: return ( ZOOM_TAG_LICENSED_CAP_EXCEEDED, "Zoom Licensed 坐席数已达订阅上限 — 在 Zoom Admin " "回收闲置 Licensed 用户或扩容订阅后重试。", ) return (None, None) def _parse_zoom_error(resp: httpx.Response) -> tuple[int | None, str]: """Pull ``(code, message)`` out of a Zoom error JSON body. Zoom returns ``{"code": int, "message": str}`` on 4xx; we swallow JSON-decode errors so a truncated body / HTML proxy page can't break the caller's failure-path branching. """ try: payload = resp.json() except Exception: # noqa: BLE001 return None, "" if not isinstance(payload, dict): return None, "" code_raw = payload.get("code") code: int | None = code_raw if isinstance(code_raw, int) else None msg_raw = payload.get("message") message = msg_raw if isinstance(msg_raw, str) else "" return code, message def _redact_secrets(blob: str) -> str: """Strip Authorization: Bearer / common API-token prefixes from error-response snippets before they're persisted into ``provisioning_tasks.error_message`` or ``software_connectors.last_error_message``. Vendors occasionally echo request headers in error JSON; we don't want a malformed token to leak across the admin UI when an admin debugs a connector failure. """ 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) # GitHub PATs (ghp_/github_pat_/ghs_/ghu_) blob = re.sub(r"\bghp_[A-Za-z0-9]{20,}\b", "ghp_***", blob) blob = re.sub(r"\bghs_[A-Za-z0-9]{20,}\b", "ghs_***", blob) blob = re.sub(r"\bghu_[A-Za-z0-9]{20,}\b", "ghu_***", blob) blob = re.sub(r"\bgithub_pat_[A-Za-z0-9_]{20,}\b", "github_pat_***", blob) return blob async def _zoom_request( method: str, url: str, *, headers: dict[str, str], json_body: Any = None, params: dict[str, Any] | None = None, refresh_headers: Callable[[], Awaitable[dict[str, str]]] | None = None, ) -> httpx.Response: """Make one Zoom API call with exponential-backoff retry on genuine HTTP 429 rate-limit responses. Zoom's concurrent-disassociate idempotency 429 (message contains "already been made") is NOT retried — that signal means the action already completed and the caller must handle it as a success. On a 401 (Zoom-side S2S token invalidation / early expiry), if a ``refresh_headers`` callable is supplied it is awaited to force a fresh OAuth token, the Authorization header is rebuilt, and the request is retried EXACTLY ONCE before the 401 is surfaced. This stops a single token invalidation from hard-failing every grant/revoke/list until nominal token expiry. Max additional latency per call under full rate-limit: ~14 s (2 + 4 + 8 s backoff across 3 retries). """ did_refresh_401 = False for attempt in range(_ZOOM_MAX_RETRIES + 1): async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, trust_env=False) as client: if method == "GET": resp = await client.get(url, headers=headers, params=params) elif method == "POST": resp = await client.post( url, headers=headers, json=json_body, params=params, ) elif method == "DELETE": resp = await client.delete(url, headers=headers, params=params) elif method == "PATCH": resp = await client.patch(url, headers=headers, json=json_body, params=params) elif method == "PUT": resp = await client.put(url, headers=headers, json=json_body, params=params) else: raise ValueError(f"Unsupported Zoom request method {method!r}") # One-shot token-refresh retry on 401. Does NOT consume a 429 # backoff attempt and never loops more than once. if ( resp.status_code == 401 and refresh_headers is not None and not did_refresh_401 ): did_refresh_401 = True logger.warning( "Zoom API 401 on %s %s — forcing token refresh and " "retrying once", method, url, ) try: headers = await refresh_headers() except Exception as e: # noqa: BLE001 logger.warning( "Zoom token force-refresh failed after 401: %s: %s", type(e).__name__, e, ) return resp continue if resp.status_code != 429: return resp # Pass through the concurrent-disassociate idempotency 429 so # revoke_seat can detect it as an already-completed action. _, zoom_message = _parse_zoom_error(resp) if "already been made" in zoom_message.lower(): return resp if attempt >= _ZOOM_MAX_RETRIES: return resp # Respect the vendor's Retry-After header; fall back to # exponential backoff capped at _ZOOM_RETRY_MAX_DELAY. raw_ra = resp.headers.get("Retry-After") or resp.headers.get("retry-after") delay: float if raw_ra: try: delay = min(float(raw_ra), _ZOOM_RETRY_MAX_DELAY) except (ValueError, TypeError): delay = min( _ZOOM_RETRY_BASE_DELAY * (2.0 ** attempt), _ZOOM_RETRY_MAX_DELAY, ) else: delay = min( _ZOOM_RETRY_BASE_DELAY * (2.0 ** attempt), _ZOOM_RETRY_MAX_DELAY, ) logger.warning( "Zoom API 429 rate-limit on %s %s (attempt %s/%s) " "— backing off %.1fs", method, url, attempt + 1, _ZOOM_MAX_RETRIES, delay, ) await asyncio.sleep(delay) # Unreachable (loop always returns inside), but satisfies mypy. return resp # type: ignore[return-value] def _q_user(user_id_or_email: str) -> str: """URL-encode a Zoom userId path segment. Zoom accepts either a numeric user_id or an email. Emails can contain ``+`` (gmail-style aliases) and ``%`` which httpx leaves untouched in path segments — the resulting URL gets normalised by intermediate proxies, decoding ``+`` into a space and breaking the lookup. A ``/`` in a malformed work_email would also retarget the action to a sibling sub-path of /users. ``quote(safe='@')`` keeps ``@`` readable while encoding everything else. """ return quote(str(user_id_or_email or ""), safe="@") @dataclass(slots=True) class _TokenCache: token: str | None = None expires_at: datetime | None = None # Process-global token cache keyed by integration_id. The driver is # rebuilt on every request via build_driver_by_id(), so an instance- # attribute cache is destroyed with each new instance — meaning every # request fetches a fresh OAuth token from Zoom. Zoom's S2S OAuth # endpoint is rate-limited; in mass-onboarding/offboarding bursts # (50+ concurrent provisioning tasks) this causes 429 on the token # endpoint and every downstream task fails. The module-level cache # persists tokens across instances for the full 1-hour TTL. _TOKEN_CACHE: dict[str, _TokenCache] = {} _TOKEN_LOCK: dict[str, asyncio.Lock] = {} def _clear_token_cache_for_tests(integration_id: str | None = None) -> None: """Evict one or all entries from the process-global token cache. Intended for tests only — clears stale cached tokens between test cases so token-fetch assertions are not confounded by earlier runs. """ if integration_id is None: _TOKEN_CACHE.clear() else: _TOKEN_CACHE.pop(integration_id, None) def _get_or_create_token_lock(integration_id: str) -> asyncio.Lock: if integration_id not in _TOKEN_LOCK: _TOKEN_LOCK[integration_id] = asyncio.Lock() return _TOKEN_LOCK[integration_id] @register_driver class ZoomSoftwareDriver(SoftwareProvisioningDriver): provider = "zoom" display_name = "Zoom" description = ( "Zoom Meetings via Server-to-Server OAuth. 支持创建/删除用户," "按 binding 指定 licensed / basic 档位。每个 Helix 软件绑定一" "个档位;想管理多档位 → 在软件目录建两条(Zoom Licensed / Zoom Basic)。" ) supports_seat = True supports_permission = False supports_multi_instance = True requires_successor_on_revoke = True @classmethod def revoke_requires_successor(cls, binding_config: dict[str, Any] | None) -> bool: """Only ``action_on_offboard = delete`` hard-requires a successor: Zoom's DELETE /users with ``action=delete`` fails for licensed users with hosted content unless a transfer target is provided. ``disassociate`` unlinks without touching content, so no successor is needed. """ action = str((binding_config or {}).get("action_on_offboard") or "disassociate").lower() return action == "delete" config_schema = [ IntegrationFieldInfo( key="account_id", label="Account ID", is_secret=False, required=True, placeholder="1a2B3c4D5eF6gH7i8J9k0L", hint="Zoom Marketplace > Server-to-Server OAuth app > Account ID。", ), IntegrationFieldInfo( key="client_id", label="Client ID", is_secret=False, required=True, placeholder="xxxxxxxxxxxxxxxxxxxxxx", ), IntegrationFieldInfo( key="client_secret", label="Client Secret", is_secret=True, required=True, hint="Server-to-Server OAuth app Client Secret,仅轮换时重贴。", ), ] binding_schema = [ IntegrationFieldInfo( key="plan_type", label="Zoom 档位", is_secret=False, required=True, default="licensed", options=[ {"value": "licensed", "label": "Licensed (付费坐席)"}, {"value": "basic", "label": "Basic (免费,但有租户名额上限)"}, { "value": "no_meetings", "label": "No Meetings license (无会议许可,无名额上限)", }, {"value": "none", "label": "None (占位,不可用)"}, ], hint=( "每个 Helix 软件对应 Zoom 的一个 user type。Basic 有租户" "名额上限(本租户 20),满了会被拒;要给全员发基线账号请用 " "no_meetings。" ), ), IntegrationFieldInfo( key="action_on_offboard", label="离职动作", is_secret=False, required=True, default="disassociate", options=[ {"value": "disassociate", "label": "解除关联 (保留录制与数据)"}, {"value": "delete", "label": "删除 (彻底移除账号)"}, ], hint="disassociate 是常用选择;只有合规要求彻底删除时选 delete。", ), ] # ── Auth ───────────────────────────────────────────────────── def __init__( self, *, integration_id: str, config: dict[str, Any], enabled_capabilities: list[str] | None = None, ) -> None: # Forward ``enabled_capabilities`` so the per-instance set # threaded by ``service._build_driver_from_resolved`` reaches # ``self.enabled_capabilities`` via the base class. Zoom is # not a cloud driver (no per-capability probe today) but the # base call site is unconditional — accepting the kwarg # keeps the call shape consistent across kinds. super().__init__( integration_id=integration_id, config=config, enabled_capabilities=enabled_capabilities, ) # Use the process-global cache/lock; do not store on instance. # Per-instance memo for the Basic headcount enrichment — see # ``_count_basic_users_safe``. Instance-scoped (not process-global) # on purpose: the count is only ever used for admin-facing error # enrichment within one sweep / task, and a process-global cache # would keep serving a stale number to the NEXT run after an admin # freed up Basic seats. self._basic_count_cache: int | None = None self._basic_count_cached_at: datetime | None = None def _account_id(self) -> str: v = self.config.get("account_id") if not v: raise ValueError("Zoom integration missing account_id") return str(v) def _basic_auth(self) -> str: client_id = self.config.get("client_id") client_secret = self.config.get("client_secret") if not client_id or not client_secret: raise ValueError("Zoom integration missing client_id / client_secret") creds = f"{client_id}:{client_secret}".encode() return "Basic " + base64.b64encode(creds).decode("ascii") async def _get_token(self, *, force_refresh: bool = False) -> str: iid = self.integration_id lock = _get_or_create_token_lock(iid) async with lock: now = datetime.now(UTC) cached = _TOKEN_CACHE.get(iid) if ( not force_refresh and cached is not None and cached.token and cached.expires_at and now < cached.expires_at - timedelta(seconds=60) ): return cached.token # type: ignore[return-value] async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, trust_env=False) as client: resp = await client.post( _ZOOM_TOKEN_URL, headers={"Authorization": self._basic_auth()}, params={ "grant_type": "account_credentials", "account_id": self._account_id(), }, ) if resp.status_code != 200: body_preview = _redact_secrets(resp.text[:500]) raise RuntimeError( f"Zoom token endpoint returned {resp.status_code}: {body_preview}", ) payload = resp.json() token = payload.get("access_token") if not token: raise RuntimeError(f"Zoom token response missing access_token: {payload}") expires_in = int(payload.get("expires_in") or 3600) _TOKEN_CACHE[iid] = _TokenCache( token=str(token), expires_at=now + timedelta(seconds=expires_in), ) return _TOKEN_CACHE[iid].token # type: ignore[return-value] async def _auth_headers(self, *, force_refresh: bool = False) -> dict[str, str]: token = await self._get_token(force_refresh=force_refresh) return { "Authorization": f"Bearer {token}", "Accept": "application/json", "Content-Type": "application/json", } # ── Test connection ────────────────────────────────────────── async def test_connection(self) -> IntegrationTestResult: """Probe: list first 1 user on the account. Fails on invalid creds, revoked app, or account_id mismatch. """ try: headers = await self._auth_headers() except Exception as e: # noqa: BLE001 return IntegrationTestResult( ok=False, message=f"Token error: {type(e).__name__}: {e}", ) try: async with httpx.AsyncClient(timeout=_HTTP_TIMEOUT, trust_env=False) as client: resp = await client.get( f"{_ZOOM_BASE}/users", headers=headers, params={"page_size": 1, "status": "active"}, ) except httpx.HTTPError as e: return IntegrationTestResult( ok=False, message=f"{type(e).__name__}: {e}", ) if resp.status_code != 200: body_preview = _redact_secrets(resp.text[:300]) return IntegrationTestResult( ok=False, message=f"Zoom API {resp.status_code}: {body_preview}", detail={"status": resp.status_code}, ) data = resp.json() or {} total = int(data.get("total_records") or 0) return IntegrationTestResult( ok=True, message=f"OK · {total} active users on account", detail={"total_records": total}, ) # ── Provisioning ───────────────────────────────────────────── async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: """Create (or re-activate) a Zoom user with the binding's plan_type. Idempotent: if the email already exists we look it up and, when possible, update its type in-place rather than failing. """ plan_raw = str((ctx.extra or {}).get("plan_type") or "licensed").lower() zoom_type = _TYPE_BY_PLAN.get(plan_raw) if zoom_type is None: return ProvisioningResult( ok=False, error_message=( f"Zoom binding has invalid plan_type {plan_raw!r} " f"(expected one of {list(_TYPE_BY_PLAN)})" ), ) name = (ctx.user_display_name or ctx.user_email.split("@", 1)[0]).strip() first, _, last = name.partition(" ") body = { "action": "create", "user_info": { "email": ctx.user_email, "type": zoom_type, "first_name": first or ctx.user_email, "last_name": last or "-", }, } try: headers = await self._auth_headers() resp = await _zoom_request( "POST", f"{_ZOOM_BASE}/users", headers=headers, json_body=body, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) 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 data.get("email") or ctx.user_email), raw_response=data, ) # User already exists → treat as success if the type already # matches, otherwise try to patch the type. if resp.status_code == 409: allow_demote = bool((ctx.extra or {}).get("allow_demote") is True) return await self._reconcile_existing_user( ctx.user_email, zoom_type, allow_demote=allow_demote, ) return await self.build_create_failure_result( resp=resp, email=ctx.user_email, ) async def _reconcile_existing_user( self, email: str, desired_type: int, *, allow_demote: bool = False, ) -> ProvisioningResult: """Look up an existing Zoom user and patch its type if it doesn't match the desired one. This turns a 409 on create into a successful idempotent grant — but ONLY for upgrades and same-tier no-ops by default. Demotions are refused so a misconfigured binding (e.g. catalog row pointing at the Basic plan when the user already holds a paid Licensed seat) cannot silently downgrade a user without an audit-traceable revoke. ``allow_demote=True`` is plumbed by the admin-driven ``/seat-allocations/{id}/change-plan`` endpoint via ``ctx.extra``. That path writes a ``software_seat.plan_changed`` audit row with the from/to plan ids — combined with this flag-gated PATCH, a deliberate licensed→basic change is traceable end-to-end without losing the protection on the accidental-downgrade path. """ try: headers = await self._auth_headers() get_resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/users/{_q_user(email)}", headers=headers, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"Zoom user lookup failed: {type(e).__name__}: {e}", ) if get_resp.status_code != 200: return ProvisioningResult( ok=False, error_message=( f"Zoom user {email!r} exists (409) but lookup returned " f"{get_resp.status_code}: {_redact_secrets(get_resp.text[:200])}" ), ) existing = get_resp.json() or {} current_type = int(existing.get("type") or 0) if current_type == desired_type: return ProvisioningResult( ok=True, external_ref=str(existing.get("id") or email), raw_response={"already_active": True, "type": current_type}, ) # Refuse silent demote unless caller explicitly opts in via # ``allow_demote``. ``allow_demote`` is set by the admin # change-plan endpoint, which writes a paired # ``software_seat.plan_changed`` audit row with the from/to # plan ids — so the demote is traceable. The default-deny # branch protects the misconfigured-binding case where a # catalog row points at Basic but the user already holds a # paid Licensed seat: silent PATCH would tear down a paid # seat without leaving a seat-allocation history row. current_rank = _TYPE_RANK.get(current_type, 0) desired_rank = _TYPE_RANK.get(desired_type, 0) if desired_rank < current_rank and not allow_demote: return ProvisioningResult( ok=False, error_message=( f"Zoom user {email!r} currently holds a " f"{_PLAN_BY_TYPE.get(current_type, current_type)!r} " f"seat (type={current_type}); refusing to silently " f"downgrade to " f"{_PLAN_BY_TYPE.get(desired_type, desired_type)!r} " f"(type={desired_type}). Use the admin " "/seat-allocations/{id}/change-plan endpoint for " "an audit-traceable tier change, or revoke the " "existing seat first." ), raw_response={ "refused_demote": True, "current_type": current_type, "desired_type": desired_type, }, ) # Patch the type — PATCH /users/{id} (upgrade path only). try: patch_resp = await _zoom_request( "PATCH", f"{_ZOOM_BASE}/users/{_q_user(existing.get('id') or email)}", headers=headers, json_body={"type": desired_type}, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"Zoom user patch failed: {type(e).__name__}: {e}", ) if patch_resp.status_code in (200, 204): return ProvisioningResult( ok=True, external_ref=str(existing.get("id") or email), raw_response={ "upgraded_from_type": current_type, "upgraded_to_type": desired_type, }, ) body_preview = _redact_secrets(patch_resp.text[:200]) zoom_code, zoom_message = _parse_zoom_error(patch_resp) tag, hint = _classify_zoom_user_error(zoom_code, zoom_message) prefix = f"[{tag}] " if tag else "" suffix = f" {hint}" if hint else "" return ProvisioningResult( ok=False, error_message=( f"{prefix}Zoom user {email!r} exists but type patch " f"returned {patch_resp.status_code} (code={zoom_code}): " f"{body_preview}{suffix}" ), raw_response={ "status": patch_resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": tag, "patch_path": "upgrade", }, ) async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: """Disassociate or delete the user per binding.action_on_offboard. When ``action=delete`` and a ``successor_email`` is supplied in the context, Zoom's DELETE /users accepts ``transfer_email`` + ``transfer_meeting`` / ``transfer_webinar`` / ``transfer_recording`` query params so hosted content moves to the successor before the user is gone. Without a successor on a licensed user, Zoom returns 400 for users owning content — we guard up-front so the error message is clear instead of a raw vendor 400. """ action = str( (ctx.extra or {}).get("action_on_offboard") or "disassociate", ).lower() if action not in ("disassociate", "delete"): return ProvisioningResult( ok=False, error_message=( f"Zoom binding invalid action_on_offboard {action!r} " "(expected disassociate|delete)" ), ) params: dict[str, Any] = {"action": action} if action == "delete": if not ctx.successor_email: return ProvisioningResult( ok=False, error_message=( "Zoom action_on_offboard=delete requires a successor " "to receive meetings/webinars/recordings. Retry the " "revoke with a successor selected." ), ) params["transfer_email"] = ctx.successor_email params["transfer_meeting"] = "true" params["transfer_webinar"] = "true" params["transfer_recording"] = "true" # Zoom accepts the user's email as userId for DELETE. Path-quote # to defuse "+" / "%" / "/" in the local-part (gmail aliases or # malformed HRIS imports). try: headers = await self._auth_headers() resp = await _zoom_request( "DELETE", f"{_ZOOM_BASE}/users/{_q_user(ctx.user_email)}", headers=headers, params=params, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) if resp.status_code in (200, 204): raw: dict[str, Any] = {"action": action} if action == "delete": raw["transferred_to"] = ctx.successor_email return ProvisioningResult( ok=True, external_ref=f"zoom:{ctx.user_email}:{action}", raw_response=raw, ) if resp.status_code == 404: # Already not on the account — treat as success for # offboarding idempotency. return ProvisioningResult( ok=True, external_ref=f"zoom:{ctx.user_email}:already_absent", raw_response={"already_absent": True}, ) # Parse Zoom's structured error so we can branch on code 1107 # (managed-domain users cannot be disassociated) and on the 429 # concurrent-disassociate guard. zoom_code, zoom_message = _parse_zoom_error(resp) # Concurrent-disassociate idempotency: a leaver who holds N add-on # seats on the same Zoom account (e.g. Workplace + Large Meeting + # Webinar) creates N provisioning tasks, each issuing # ``DELETE /users/{email}?action=disassociate`` against the SAME # Zoom user. Zoom serialises these at the account level and # rejects the 2nd/3rd with HTTP 429 + "A request to disassociate # this user has already been made." The first call already moved # the user out of the account; subsequent calls are no-ops. Treat # this as success for offboarding idempotency — same semantics as # the 404 branch above. Caught 2026-05-08 on P01121: 1 of 3 # Zoom revokes succeeded, the other 2 dead-lettered the leaver # for the day, leaving the offboarding stuck on a non-error. if ( resp.status_code == 429 and "already been made" in zoom_message.lower() ): return ProvisioningResult( ok=True, external_ref=f"zoom:{ctx.user_email}:already_in_progress", raw_response={ "already_in_progress": True, "status": 429, "zoom_message": zoom_message, "note": ( "Zoom serialised disassociate at the account " "level — first call already removed the user; " "this call is a no-op." ), }, ) # Managed-domain fallback: an SSO-bound email domain (e.g. a # tenant where the email TLD is owned by Zoom for SAML/SSO) # PERMANENTLY refuses ``DELETE /users/{id}?action=disassociate`` # with HTTP 400 / code 1107: "You can not disassociate a user # with managed domain." The closest equivalent that achieves the # same billing outcome (paid license released, every add-on # cleared) is to PATCH ``user.type = 1`` (Basic). The user # remains in the tenant — required because SSO would re-create # them on next login anyway — but no longer holds a paid seat. # Bug class caught 2026-05-08 (kevin.lu@pingcap.com — 3 Zoom # seats showed revoked in Helix while Zoom kept all 3 licensed # because every disassociate returned 1107). Without this # fallback, every managed-domain Zoom tenant is a guaranteed # "ghost revoke" tenant. if action == "disassociate" and resp.status_code == 400 and zoom_code == 1107: return await self._disassociate_via_type_downgrade(ctx) # Other vendor codes that admins act on differently — tag the # error_message so the failure DM can render a clean # remediation line. ``raw_response`` carries the original # Zoom code for log / audit / future analytics. body_preview = _redact_secrets(resp.text[:300]) tag, hint = _classify_zoom_user_error(zoom_code, zoom_message) prefix = f"[{tag}] " if tag else "" suffix = f" {hint}" if hint else "" return ProvisioningResult( ok=False, error_message=( f"{prefix}Zoom delete user {resp.status_code} " f"(code={zoom_code}): {body_preview}{suffix}" ), raw_response={ "status": resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": tag, }, ) async def _disassociate_via_type_downgrade( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """Managed-domain fallback for ``revoke_seat`` (Zoom error 1107). Issues ``PATCH /users/{id}`` setting ``type=1`` (Basic). For managed-domain tenants this is the only Zoom-permitted way to release a paid license without removing the user from the account (which SSO requires). The PATCH also clears any type-2-only add-ons (Large Meeting, Webinar) since Basic users cannot hold them. Returns ``ok=True`` on: - 200/204: type successfully downgraded. - 404: user not found at provider — already absent (idempotent). Returns ``ok=False`` for any other status, with the original 400/ 1107 + the fallback's response code in ``raw_response`` so the operator can correlate both legs in a single ``provisioning_ tasks.error_message`` row. """ # OPEN QUESTION (2026-07-30, deliberately NOT changed here): this # releases the paid license by moving the user into the Basic pool # — the one tenant-capped tier (20 slots) — which is both why the # 3412 deadlock below exists AND a plausible reason the pool is # full at all (20/20 on prod). Downgrading to ``no_meetings`` (4) # instead would be uncapped and free the license just the same. # Not done in this PR because it is a seat-flow behaviour change # whose PATCH-to-type-4 acceptance is unverified against Zoom, and # getting it wrong means paid licenses stop being released. Verify # on one real user before switching. try: headers = await self._auth_headers() patch_resp = await _zoom_request( "PATCH", f"{_ZOOM_BASE}/users/{_q_user(ctx.user_email)}", headers=headers, json_body={"type": _TYPE_BY_PLAN["basic"]}, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=( f"Zoom managed-domain fallback PATCH failed: " f"{type(e).__name__}: {e}" ), ) if patch_resp.status_code in (200, 204): logger.info( "Zoom revoke_seat for %s: managed-domain fallback " "downgraded type to Basic (1107 → PATCH type=1)", ctx.user_email, ) return ProvisioningResult( ok=True, external_ref=f"zoom:{ctx.user_email}:downgraded_to_basic", raw_response={ "fallback": "managed_domain_type_downgrade", "from_action": "disassociate", "to_type": _TYPE_BY_PLAN["basic"], "note": ( "Disassociate refused (1107 managed-domain). " "Downgraded type Licensed→Basic — paid license " "released, add-ons cleared, account retained " "for SSO." ), }, ) if patch_resp.status_code == 404: return ProvisioningResult( ok=True, external_ref=f"zoom:{ctx.user_email}:already_absent", raw_response={ "already_absent": True, "via": "managed_domain_fallback", }, ) body_preview = _redact_secrets(patch_resp.text[:300]) zoom_code, zoom_message = _parse_zoom_error(patch_resp) # 3412 is the only path that lands an admin in a "both legs # refused" deadlock: disassociate was already refused by 1107 # (managed domain), and now PATCH→Basic is refused because the # Basic-user pool is full. The admin's ONLY remediation is on # the Zoom side (free up Basic seats) — surface that explicitly # plus the current Basic count so they know how much headroom # they need to make. if ( patch_resp.status_code == 400 and zoom_code == _ZOOM_CODE_BASIC_CAP_ON_DOWNGRADE ): basic_count = await self._count_basic_users_safe() count_clause = ( f"(current Basic users={basic_count})" if basic_count is not None else "(current Basic count unavailable)" ) logger.warning( "Zoom revoke_seat for %s: managed-domain fallback hit " "Basic-cap (3412) %s — admin must clean up dormant " "Basic users before retry.", ctx.user_email, count_clause, ) return ProvisioningResult( ok=False, error_message=( f"[{ZOOM_TAG_BASIC_CAP_EXCEEDED}] Zoom Basic 用户数" f"已达账户上限 {count_clause};managed-domain 租户" "释放付费 license 的唯一路径是 Licensed→Basic 降级," "Basic 池满则该路径被拒。请到 Zoom Admin 控制台清理" "闲置 Basic 用户后重试本回收任务。" ), raw_response={ "fallback": "managed_domain_type_downgrade", "status": patch_resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": ZOOM_TAG_BASIC_CAP_EXCEEDED, "current_basic_users": basic_count, }, ) # Any other PATCH failure: keep the original opaque-but-traceable # behaviour, but include the parsed code so logs / future # branches can act on it. return ProvisioningResult( ok=False, error_message=( f"Zoom managed-domain fallback PATCH user type returned " f"{patch_resp.status_code} (code={zoom_code}): {body_preview}" ), raw_response={ "fallback": "managed_domain_type_downgrade", "status": patch_resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, }, ) async def delete_user_unconditional( self, email: str, *, transfer_email: str | None = None, ) -> ProvisioningResult: """Permanently DELETE a Zoom user with ``action=delete``, no successor *guard* — optionally handing their hosted content to ``transfer_email`` on the way out. Used by two callers with different needs: * ``zoom_seat_coordinator.make_room_in_basic_pool`` evicts an idle Basic user when the Basic-user pool is full — the only Zoom-permitted path that releases a Basic slot on a managed-domain tenant without violating the 1107 disassociate-refused constraint (``action=delete`` is allowed for managed-domain users; ``action=disassociate`` is not). The eviction target is by construction a long-idle Basic user (sorted by ``last_login_time`` asc, never-logged-in first), so it passes no ``transfer_email``. * The offboarding terminal-delete sweep passes the resolved successor so meetings / webinars / cloud recordings move before the account disappears. **Zoom deletes are PERMANENT — there is no undelete window like Google's ~20 days** — so omitting the transfer here destroys recordings irrecoverably. That is exactly what happened to two prod leavers before 2026-08-13; the sweep now refuses to delete without a successor. Do not drop this parameter. Distinct from :meth:`revoke_seat`, which early-returns with "successor required" for ``action=delete`` and is driven by the binding's ``action_on_offboard``. This method is the account-level path where the caller has already decided the disposition and resolved (or deliberately declined) a recipient. Returns ``ok=True`` on: * 200/204 — user deleted on Zoom side. * 404 — already absent (idempotent). On failure ``raw_response["zoom_code"]`` and ``raw_response["error_tag"]`` are populated so the caller can decide whether to advance or abort. A refused transfer (bad / non-existent recipient) surfaces as a normal failure — the caller must NOT retry without the transfer, or it silently destroys the content the transfer existed to save. """ params: dict[str, Any] = {"action": "delete"} recipient = (transfer_email or "").strip() if recipient: params["transfer_email"] = recipient params["transfer_meeting"] = "true" params["transfer_webinar"] = "true" params["transfer_recording"] = "true" try: headers = await self._auth_headers() resp = await _zoom_request( "DELETE", f"{_ZOOM_BASE}/users/{_q_user(email)}", headers=headers, params=params, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=( f"Zoom delete_user_unconditional {type(e).__name__}: {e}" ), ) if resp.status_code in (200, 204): raw: dict[str, Any] = {"action": "delete"} if recipient: # Persisted by the caller as the handover receipt — the # ledger line an auditor reads to answer "who got their # recordings?" without a Zoom console round-trip. raw["transferred_to"] = recipient return ProvisioningResult( ok=True, external_ref=f"zoom:{email}:deleted", raw_response=raw, ) if resp.status_code == 404: return ProvisioningResult( ok=True, external_ref=f"zoom:{email}:already_absent", raw_response={"already_absent": True}, ) body_preview = _redact_secrets(resp.text[:300]) zoom_code, zoom_message = _parse_zoom_error(resp) tag, _hint = _classify_zoom_user_error(zoom_code, zoom_message) if recipient: # A recipient refusal is about the OTHER party and outranks # the generic classification: Zoom validates transfer_email # before it deletes anything, so the leaver's account is # untouched and the caller's next move depends entirely on # WHICH recipient problem this is. Only reachable when we # actually asked for a transfer, so a leaver-side failure can # never be mis-tagged as a recipient refusal. tag = _classify_zoom_transfer_error(zoom_message) or tag return ProvisioningResult( ok=False, error_message=( f"Zoom delete user {resp.status_code} (code={zoom_code}): " f"{body_preview}" + ( f" [transfer_email={recipient}]" if recipient else "" ) ), raw_response={ "transfer_email": recipient or None, "status": resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": tag, }, ) async def _count_basic_users_safe(self) -> int | None: """Count Zoom users with ``type=1`` (Basic) across active + pending status, capped at ``_ZOOM_BASIC_COUNT_CAP``. Wrapped in a broad exception swallow so a transient list-users failure doesn't mask the underlying 3412 error — the caller falls back to "unavailable" in the error_message. Used purely for admin-facing enrichment; no business logic branches on the count. DoS / rate-limit considerations: each call paginates active + pending /users, worst-case 2 × 200 pages = 400 Zoom API calls on a 60k-user tenant. It used to run only on the failure path of a single revoke task (serialised per-(user_id, software_id) by the provisioning dispatcher, so back-to-back 3412s on one task could not fan out). The identity-sync create path can hit the cap for MANY people in one sweep, which is exactly the "daily wave of 3412 failures" the previous version of this docstring said would need a cache — so the count is now memoised per driver instance for ``_BASIC_COUNT_TTL``. Driver instances are built per sweep / per task, so the memo also bounds a single run to one enumeration. (The identity-sync runner additionally stops calling create at all once the pool is known full.) """ now = datetime.now(UTC) if ( self._basic_count_cached_at is not None and now - self._basic_count_cached_at < _BASIC_COUNT_TTL ): return self._basic_count_cache try: total = 0 for status in ("active", "pending"): users = await self.list_users(status=status) for u in users: if int(u.get("type") or 0) == _TYPE_BY_PLAN["basic"]: total += 1 if total >= _ZOOM_BASIC_COUNT_CAP: break if total >= _ZOOM_BASIC_COUNT_CAP: break self._basic_count_cache = total self._basic_count_cached_at = now return total except Exception as e: # noqa: BLE001 logger.warning( "Zoom _count_basic_users_safe failed (enrichment only): %s", e, ) # Deliberately NOT memoised: a transient list failure should # not pin "unavailable" for the rest of the run. return None async def build_create_failure_result( self, *, resp: httpx.Response, email: str, ) -> ProvisioningResult: """Shared failure envelope for both create-a-user call sites. ``grant_seat`` (seat flow) and :meth:`create_user` (identity-sync baseline) both POST /users and must classify the response the same way — they drifted before: only one of them was ever taught that a Basic-cap refusal needs the live Basic headcount attached. Keeping one builder means a new Zoom error code is understood by both surfaces the moment it is added to ``_classify_zoom_user_error``. ``raw_response['error_tag']`` is the machine-readable contract callers branch on (see :func:`is_basic_cap_error`); everything else in there is for the admin-facing ledger. """ body_preview = _redact_secrets(resp.text[:300]) zoom_code, zoom_message = _parse_zoom_error(resp) tag, hint = _classify_zoom_user_error(zoom_code, zoom_message) prefix = f"[{tag}] " if tag else "" suffix = f" {hint}" if hint else "" # The headcount is what tells an admin whether this is "free up one # dormant Basic user" or "this tier is structurally wrong for the # tenant" — without it the caller only learns "full". basic_count: int | None = None if zoom_code in _ZOOM_CODES_BASIC_CAP: basic_count = await self._count_basic_users_safe() logger.warning( "Zoom create user %s refused — Basic pool full " "(code=%s, current Basic users=%s). Basic is tenant-capped; " "'no_meetings' (type 4) is the uncapped baseline tier.", email, zoom_code, basic_count, ) return ProvisioningResult( ok=False, error_message=( f"{prefix}Zoom create user {resp.status_code} " f"(code={zoom_code}): {body_preview}{suffix}" ), raw_response={ "status": resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": tag, "current_basic_users": basic_count, }, ) # ── Directory / subscription helpers ───────────────────────── # # Used by the ``zoom.seat_sync`` scheduled job to discover who # currently holds a licensed seat on the Zoom side, so Helix can # (a) back-fill :class:`SoftwareSeatAllocation` rows for # pre-existing Zoom users, and (b) pick a least-recently-logged-in # victim when a new seat request must evict someone to grant the # new user. They are NOT part of the ``SoftwareProvisioningDriver`` # mixin contract — just Zoom-specific extras, analogous to # ``GoogleWorkspaceDriver.list_all_groups``. async def list_users( self, *, status: str = "active", page_size: int = 300, ) -> list[dict[str, Any]]: """Return every Zoom user at the given status. ``status`` is Zoom's native filter (``active`` / ``inactive`` / ``pending``). Paginates via ``next_page_token`` until Zoom returns empty; default page size 300 is Zoom's max. """ out: list[dict[str, Any]] = [] next_token: str | None = None seen_tokens: set[str] = set() headers = await self._auth_headers() for _page in range(_MAX_PAGINATOR_PAGES): params: dict[str, Any] = { "page_size": page_size, "status": status, } if next_token: params["next_page_token"] = next_token resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/users", headers=headers, params=params, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) if resp.status_code != 200: raise RuntimeError( f"Zoom /users list returned {resp.status_code}: " f"{_redact_secrets(resp.text[:300])}" ) payload = resp.json() or {} out.extend(payload.get("users") or []) next_token = payload.get("next_page_token") or None if not next_token: break if next_token in seen_tokens: raise RuntimeError( "Zoom /users paginator returned a repeating " f"next_page_token (len={len(next_token)}); " "aborting.", ) seen_tokens.add(next_token) else: raise RuntimeError( "Zoom /users paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages.", ) return out async def get_user(self, user_id_or_email: str) -> dict[str, Any] | None: """Fetch a single user record. Returns ``None`` on 404. Used by the eviction flow to pull ``last_login_time`` for a candidate victim when the bulk ``list_users`` response doesn't include it (Zoom truncates some fields in bulk listings depending on account plan). """ headers = await self._auth_headers() resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/users/{_q_user(user_id_or_email)}", headers=headers, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) if resp.status_code == 404: return None if resp.status_code != 200: raise RuntimeError( f"Zoom /users/{{id}} returned {resp.status_code}: " f"{_redact_secrets(resp.text[:300])}" ) return resp.json() or {} # ── Identity-lifecycle primitives ──────────────────────────── # # These back the ``zoom`` identity-sync runner (create the baseline # account on hire, close it on leave) — a DIFFERENT concern from the # seat/license flow above (``grant_seat`` / ``revoke_seat``). They are # deliberately kept as thin, policy-free primitives: the runner owns # the "create-if-missing, never downgrade a paid user, reactivate a # rejoiner" orchestration so this driver stays a pure Zoom API client. async def create_user( self, *, email: str, first_name: str, last_name: str, plan_type: str = "basic", ) -> ProvisioningResult: """Create a Zoom user of ``plan_type`` (default Basic — free). Idempotent-friendly wrapper for the identity-sync runner. Unlike :meth:`grant_seat`, a 409 (email already on the account) is surfaced as ``ok=True`` + ``raw_response['already_exists']=True`` rather than routed through :meth:`_reconcile_existing_user` — the latter refuses a Licensed→Basic demotion and would misfire when an existing PAID user is re-synced as the Basic baseline. The runner inspects ``already_exists`` to decide whether to reactivate / no-op. """ # Unknown tier name is a configuration error, not a reason to # silently create the account on the WRONG tier. It used to fall # back to Basic — the one tenant-capped tier — so a typo in # ``runner_options.zoom_plan_type`` would have quietly recreated # the exact 3412 outage this default was changed to avoid. plan_key = str(plan_type).lower().strip() if plan_key not in _TYPE_BY_PLAN: return ProvisioningResult( ok=False, error_message=( f"Zoom create user aborted: unknown plan_type " f"{plan_type!r} (expected one of {sorted(_TYPE_BY_PLAN)})" ), raw_response={"config_error": "unknown_plan_type"}, ) zoom_type = _TYPE_BY_PLAN[plan_key] body = { "action": "create", "user_info": { "email": email, "type": zoom_type, "first_name": (first_name or email).strip() or email, "last_name": (last_name or "-").strip() or "-", }, } try: headers = await self._auth_headers() resp = await _zoom_request( "POST", f"{_ZOOM_BASE}/users", headers=headers, json_body=body, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) 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 data.get("email") or email), raw_response=data, ) if resp.status_code == 409: return ProvisioningResult( ok=True, external_ref=email, raw_response={"already_exists": True}, ) return await self.build_create_failure_result(resp=resp, email=email) async def set_user_status( self, *, user_id_or_email: str, activate: bool, ) -> ProvisioningResult: """Activate / deactivate a Zoom user via ``PUT /users/{id}/status``. ``user_id_or_email`` may be the numeric Zoom user id (preferred — stable across an email change) or the email; Zoom accepts either as the ``userId`` path segment. This is the identity-lifecycle "open / close the account" primitive — distinct from :meth:`revoke_seat` (disassociate / delete). Deactivation keeps the user in the tenant (so SSO does not silently re-create them on next login), retains their meetings / recordings, and frees any paid seat they held. A 404 is treated as idempotent success (user already gone from the account). Works on managed-domain tenants where ``disassociate`` is refused (1107). """ action = "activate" if activate else "deactivate" try: headers = await self._auth_headers() resp = await _zoom_request( "PUT", f"{_ZOOM_BASE}/users/{_q_user(user_id_or_email)}/status", headers=headers, json_body={"action": action}, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) except httpx.HTTPError as e: return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) if resp.status_code in (200, 204): return ProvisioningResult( ok=True, external_ref=f"zoom:{user_id_or_email}:{action}", raw_response={"action": action}, ) if resp.status_code == 404: return ProvisioningResult( ok=True, external_ref=f"zoom:{user_id_or_email}:already_absent", raw_response={"already_absent": True, "action": action}, ) body_preview = _redact_secrets(resp.text[:300]) zoom_code, zoom_message = _parse_zoom_error(resp) tag, _hint = _classify_zoom_user_error(zoom_code, zoom_message) return ProvisioningResult( ok=False, error_message=( f"Zoom {action} user {resp.status_code} " f"(code={zoom_code}): {body_preview}" ), raw_response={ "status": resp.status_code, "body": body_preview, "zoom_code": zoom_code, "zoom_message": zoom_message, "error_tag": tag, }, ) async def get_user_settings( self, user_id_or_email: str, ) -> dict[str, Any] | None: """Fetch ``/users/{id}/settings`` — source of per-user add-on flags (``feature.large_meeting``, ``feature.webinar`` + their capacity ints). Requires ``user:read:settings:admin`` on the S2S app. Returns ``None`` on 400/401/403/404 so the seat-sync handler can skip add-on population gracefully if the admin hasn't granted the scope yet. """ headers = await self._auth_headers() resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/users/{_q_user(user_id_or_email)}/settings", headers=headers, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) if resp.status_code in (400, 401, 403, 404): return None if resp.status_code != 200: raise RuntimeError( f"Zoom /users/{{id}}/settings returned {resp.status_code}: " f"{_redact_secrets(resp.text[:300])}" ) return resp.json() or None async def get_account_plans_usage(self) -> dict[str, Any] | None: """Fetch ``/accounts/me/plans/usage`` — per-SKU purchased vs used host counts. Drives ``max_seats`` auto-population for the base plan *and* the Large Meeting / Webinar add-on plans. Shape (observed): { "plan_base": {"type": "...", "hosts": 200, "usage": 190}, "plan_large_meeting": [ {"type": "large1000_yearly", "hosts": N, "usage": M} ], "plan_zoom_sessions": [...], ... } Returns ``None`` on missing scope so the sync job can proceed without add-on max_seats (seat population still works). """ headers = await self._auth_headers() resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/accounts/me/plans/usage", headers=headers, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) if resp.status_code in (400, 401, 403, 404): logger.warning( "Zoom /accounts/me/plans/usage returned %s; add-on " "max_seats will not auto-populate. Grant " "billing:read:plan_usage:admin on the S2S app.", resp.status_code, ) return None if resp.status_code != 200: raise RuntimeError( f"Zoom /accounts/me/plans/usage returned {resp.status_code}: " f"{_redact_secrets(resp.text[:300])}" ) return resp.json() or None async def get_account_plan(self) -> dict[str, Any] | None: """Return the account's subscription plan envelope or ``None``. Drives auto-population of ``SoftwareLicensePlan.max_seats``. Zoom's response shape is ``{"plan_base": {"type": "...", "hosts": N, ...}, "plan_recording": {...}, ...}``; callers typically read ``plan_base.hosts``. Endpoint requires the ``billing:read:plan_information:admin`` granular scope on the S2S app (the old ``account:read:admin`` no longer covers this endpoint after Zoom's granular-scope migration). Returns ``None`` on 400/401/403/404 — Zoom returns HTTP 400 with ``code:4711`` ("Invalid access token, does not contain scopes:[...]") for missing granular scopes on this endpoint, so 400 has to be in the swallow list alongside the classic auth status codes; otherwise a missing scope tanks the whole seat-sync run instead of just leaving ``max_seats`` untouched. """ headers = await self._auth_headers() resp = await _zoom_request( "GET", f"{_ZOOM_BASE}/accounts/me/plans", headers=headers, refresh_headers=lambda: self._auth_headers(force_refresh=True), ) if resp.status_code in (400, 401, 403, 404): logger.warning( "Zoom /accounts/me/plans returned %s; falling back " "to null max_seats. Add the " "billing:read:plan_information:admin scope to the S2S " "app to auto-populate paid seat count.", resp.status_code, ) return None if resp.status_code != 200: raise RuntimeError( f"Zoom /accounts/me/plans returned {resp.status_code}: " f"{_redact_secrets(resp.text[:300])}" ) return resp.json() or None @staticmethod def extract_licensed_seats(plan_payload: dict[str, Any] | None) -> int | None: """Pull the paid-host count out of a ``get_account_plan`` reply. Standalone so the sync handler can call it without re-implementing Zoom's payload key. ``plan_base.hosts`` is the canonical field (total count of licensed user slots the account is paying for); we fall back to ``plan_base.service_effective`` where ``type == 'Licensed'`` in case Zoom's schema evolves. Returns ``None`` if nothing sensible can be parsed so the caller leaves ``max_seats`` untouched instead of zeroing it. """ if not isinstance(plan_payload, dict): return None base = plan_payload.get("plan_base") if isinstance(base, dict): hosts = base.get("hosts") if isinstance(hosts, int) and hosts > 0: return hosts # Some accounts expose hosts via ``plan_base.type`` + # ``plan_base.service_effective``; try that before giving up. if isinstance(base, dict): effective = base.get("service_effective") if isinstance(effective, int) and effective > 0: return effective return None