"""Feishu (Lark) software-provisioning driver. Four operation modes, selected by per-software binding: 1. ``user_suspend`` (default, recommended since 2026-05-30) — freeze / unfreeze (or DELETE on opt-in) an EXISTING Feishu account. Used for leaver lockout when Feishu access itself is the resource to revoke. Default revoke = PATCH ``is_frozen=true`` (reversible, preserves account / chats / file ownership); ``hard_delete_on_offboard`` binding flag opts into DELETE /contact/v3/users for compliance- mandated removal. Grant = PATCH unfreeze (best-effort — accounts are typically created by HRIS sync, so grant only runs on rehire). 2. ``group_member`` — manage user membership in a Feishu user group. Used for SSO-governed SaaS where group membership is the access gate. Binding requires ``group_chat_id``. Account itself is not touched (grant/revoke only adds/removes group membership). 3. ``tenant_users`` — read-only seat sync. ``feishu.seat_sync`` mirrors every active Feishu tenant account to ``software_seat_allocations``. ``grant_seat`` / ``revoke_seat`` HARD-ERROR by design (loud failure vs silent no-op) because the account itself is owned by another channel (HRIS sync — 飞书人事 / SuccessFactors → Feishu — or a third-party IDP). Pick this mode only when Helix must NOT touch Feishu accounts. Was the historical driver default (≤ 2026-05-29) but flipped to ``user_suspend`` after a prod incident where admins accepted the default expecting real revocation. 4. ``user_create`` (added 2026-05-12, round-19) — actively CREATE a Feishu account on hire via POST /contact/v3/users. Default OFF in the dropdown — HRIS sync remains the source of truth for account existence; this mode is for the narrow case where Helix must front-run HRIS (e.g. day-0 access before HRIS catches up the next morning, or a Helix-managed contractor tenant without HRIS coverage). Revoke shares the user_suspend freeze/delete paths. Binding REQUIRES ``default_department_id`` because Feishu's create endpoint refuses requests with no department. **Flag OFF means default OFF**: an admin must explicitly select ``user_create`` in the operation dropdown AND configure ``default_department_id`` before any user creation fires. Existing bindings on the other 3 modes are unaffected. Identical pattern to google_workspace.py's ``operation`` field — the driver dispatches ``grant_seat`` / ``revoke_seat`` based on the binding's ``operation`` value. Mode semantics: group_member: grant → add user to group (POST .../group/{id}/member/add) revoke → remove user from group (POST .../group/{id}/member/remove) list → :meth:`list_group_members_with_emails` walks the ``simplelist`` endpoint and resolves each member's email — used by the ``feishu.seat_sync`` scheduled job to back-fill ``software_seat_allocations`` from the canonical Feishu user-group membership. user_suspend: grant → best-effort PATCH to mark user active. In practice users are created via HRIS sync, not via Helix; so "grant" typically runs when an employee returns from leave. If Feishu has no active record we no-op ok. revoke → freeze (PATCH is_frozen=true) by default; DELETE /contact/v3/users when ``hard_delete_on_offboard`` is set. Either way removes the user from Feishu access. user_create (round-19): grant → POST /contact/v3/users with display_name + email + department_ids=[binding.default_department_id]. If Feishu reports the user already exists (e.g. HRIS raced ahead), collapse to idempotent ok. revoke → reuses user_suspend's freeze (default) / delete (binding opt-in) — the create-on-hire vs revoke-on-leave asymmetry is intentional; both directions converge on the same Feishu account lifecycle once it exists. API doc: https://open.feishu.cn/document/server-docs/contact-v3/user/ """ from __future__ import annotations import logging import re from typing import Any from urllib.parse import quote from app.approval.feishu_client import FeishuApprovalClient from app.core.exceptions import AppException 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__) _OP_GROUP_MEMBER = "group_member" _OP_USER_SUSPEND = "user_suspend" _OP_TENANT_USERS = "tenant_users" _OP_USER_CREATE = "user_create" # Hard cap on pagination iterations across every paginated Feishu # call below. A buggy Feishu response that returns ``has_more=true`` # with the same ``page_token`` (or perpetually-different ones) # would otherwise wedge the worker forever. 500 pages × 100/page # is 50k entries — well above any realistic 500-employee tenant. _MAX_PAGINATOR_PAGES = 500 # Feishu API codes that map to "user / member already in target state" # (200 + numeric body code). Sourced from open.feishu.cn error tables; # code numbers are stable across the contact-v3 API. # 99991663 — user_id not found (account already gone / never existed) # 230003 — user_id not found (alt path) # 230002 — user is already frozen / resigned # 99991401 — already a member of the group (group/member/add idempotent) # 99991402 — not a member (group/member/remove idempotent) _FEISHU_ALREADY_GONE_CODES = {99991663, 230003} # AppException ``code`` values that explicitly signal "user not found # in Feishu" — flagged either by the Feishu client's resolve-by-email # helper (FEISHU_USER_NOT_FOUND) or the generic NotFoundError wrapper # raised at the integrations layer. Both convey the same semantic: # "the upstream confirmed the user is not present" — collapsing to # already-absent / no-op is correct. Other AppException codes (token # rotated, scope downgraded, transport blip) MUST NOT be treated as # "user not found" — those should bubble as ok=False so the worker # retries instead of silently claiming success. _FEISHU_USER_NOT_FOUND_CODES: frozenset[str] = frozenset({ "FEISHU_USER_NOT_FOUND", "NOT_FOUND", }) _FEISHU_ALREADY_FROZEN_CODES = {230002} _FEISHU_ALREADY_IN_GROUP_CODES = {99991401} _FEISHU_NOT_IN_GROUP_CODES = {99991402} # Feishu codes for "group / chat / department not found" — used by # group_member remove to fall through to idempotent ok=True when the # target group itself is gone. Without this a renamed / deleted group # (or a stale ``group_chat_id`` in the binding) would loop the revoke # in dead-letter forever. Source: open.feishu.cn/document/server-docs/ # group-v1 + contact-v3 error tables. # 99991407 — group / chat not found # 1006 — chat_id invalid (some endpoints use this) # 230004 — group_id not found (alt path) _FEISHU_GROUP_NOT_FOUND_CODES = {99991407, 1006, 230004} # ── Error tags ─────────────────────────────────────────────────── # # Tagging an error_message with ``[TAG] …`` lets the failure DM # dispatcher (``_dispatch_revoke_failure_dms``) render a clean, # actionable remediation hint for the admin. Mirrors the Zoom # ``ZOOM_TAG_*`` pattern. The provisioning service's # ``_PROVISIONING_REMEDIATION_HINTS`` dict carries the per-tag hint # text. Hard-delete fallback to freeze emits this tag in the # raw_response so audit can distinguish freeze-because-asked from # freeze-because-delete-refused. FEISHU_TAG_HARD_DELETE_REFUSED = "FEISHU_HARD_DELETE_REFUSED" # Feishu API codes that POST /contact/v3/users returns when the # requested user already exists in the tenant. Listed under the # create-user error table at # https://open.feishu.cn/document/server-docs/contact-v3/user/create. # 43002 — mobile already exists # 43003 — email already exists # 43005 — employee_no already exists # We treat all three as "user already provisioned" and collapse to # idempotent ok — HRIS sync may have created the account between the # scan and the dispatch, or a retry after a partial failure may hit a # half-committed user. Other create errors (43006 invalid email, # 99991664 department not found, …) bubble as ok=False. _FEISHU_USER_ALREADY_EXISTS_CODES = {43002, 43003, 43005} # Department-not-found code for POST /contact/v3/users — surfaces # clearly to the admin so they fix the binding's # ``default_department_id`` rather than letting the worker retry # a structurally invalid request forever. _FEISHU_DEPARTMENT_NOT_FOUND_CODES = {99991664} @register_driver class FeishuSoftwareDriver(SoftwareProvisioningDriver): kind = "software_provisioning" provider = "feishu_group" display_name = "飞书软件供给" description = ( "通过飞书控制 SaaS / 飞书自身的访问。三种模式:" "(1) 全员坐席同步(默认,推荐):read-only,把飞书租户内" "全部 active 账号镜像到 Helix software_seat_allocations," "等同 Google Workspace 的 base seat 台账;不参与 grant/revoke。" "(2) 用户组成员:加入/移除某个飞书用户组(用户组 ≠ 部门)。" "(3) 账号禁用:离职二次保障,删除/恢复整个飞书账号。" "通讯录权限范围必须包含全部成员;权限要求:" "contact:user.employee_id:readonly + contact:user.base:readonly + " "contact:user.email:readonly + contact:department.base:readonly。" "用户组模式额外需要 contact:group.member:write。" "账号禁用模式额外需要 contact:user.base:write。" ) supports_seat = True supports_permission = False supports_multi_instance = True config_schema = [ IntegrationFieldInfo( key="app_id", label="App ID", is_secret=False, required=True, placeholder="cli_xxxxxxxxxxxxxxxx", ), IntegrationFieldInfo( key="app_secret", label="App Secret", is_secret=True, required=True, ), ] binding_schema = [ IntegrationFieldInfo( key="operation", label="操作模式", is_secret=False, required=True, default=_OP_USER_SUSPEND, options=[ { "value": _OP_USER_SUSPEND, "label": "账号禁用 (推荐 — 离职冻结飞书账号,入职解冻;保留账号、聊天、文件)", }, { "value": _OP_GROUP_MEMBER, "label": "用户组成员 (加入/移除指定飞书用户组,适合按权限组隔离)", }, { "value": _OP_TENANT_USERS, "label": "全员坐席同步 (read-only,仅用于计费台账镜像 — 不会真实冻结/解冻账号)", }, { "value": _OP_USER_CREATE, "label": "账号创建 (高级,默认关 — 入职时主动新建飞书账号;HRIS 通常是源)", }, ], hint=( "默认推荐「账号禁用」 — Helix 在离职时通过 PATCH 真实冻结飞书账号," "入职时解冻;账号、聊天记录、文件归属全部保留,合规允许冷静期内复职。" "「用户组成员」用于只想管控特定权限组的场景。" "「全员坐席同步」是 read-only 镜像模式 — Helix 不会调用任何写接口," "仅用于把飞书全员台账映射成坐席行做计费 / 盘点;离职时不会真实回收账号," "仅在另有渠道(飞书人事 / 第三方 IDP)做账号生命周期时启用。" "「账号创建」仅在 Helix 必须抢在 HRIS 之前发账号(承包商租户 / day-0 临时访问)" "时启用,需配合 default_department_id 设置。" ), ), IntegrationFieldInfo( key="group_chat_id", label="飞书用户组 ID", is_secret=False, required=True, visible_when={"operation": _OP_GROUP_MEMBER}, placeholder="如 c2ce53cg(注意是飞书用户组 ID,不是聊天群 oc_…)", hint=( "在飞书后台 通讯录管理 → 用户组 列表里查到的组 ID。" "授权 = 加入该用户组,回收 = 从该用户组移出。" ), ), IntegrationFieldInfo( key="hard_delete_on_offboard", label="离职时硬删除账号", is_secret=False, required=False, default=False, visible_when={"operation": _OP_USER_SUSPEND}, hint=( "默认关闭 — 离职 revoke 走 PATCH 冻结(is_frozen=true)," "保留账号、聊天记录、文件归属,合规允许冷静期内复职。" "勾选后 revoke 会改为 DELETE /contact/v3/users 永久删除" "整个飞书账号(不可恢复),仅在合规要求强制清除时启用。" ), ), IntegrationFieldInfo( key="default_department_id", label="新建账号默认部门 (open_department_id)", is_secret=False, required=True, visible_when={"operation": _OP_USER_CREATE}, placeholder="od-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx", hint=( "Feishu POST /contact/v3/users 接口要求 department_ids 不能为空。" "此 ID 是飞书管理后台 通讯录 → 部门 详情页 URL 中的" "open_department_id(以 od- 开头)。新建账号会落到该部门," "后续 HRIS 同步会自动重定位到正确部门。仅在 operation=user_create " "模式下可见且必填 — 没有部门 ID 飞书 API 会拒绝创建。" ), ), IntegrationFieldInfo( key="user_create_revoke_to_delete", label="user_create 模式离职时硬删除", is_secret=False, required=False, default=False, visible_when={"operation": _OP_USER_CREATE}, hint=( "默认关闭 — user_create 模式离职 revoke 也走 PATCH 冻结" "(默认与 user_suspend 模式一致,保留账号 + 数据归属)。" "勾选后 revoke 会改为 DELETE 永久删除。" ), ), ] def _client(self) -> FeishuApprovalClient: return FeishuApprovalClient( app_id=str(self.config.get("app_id") or ""), app_secret=str(self.config.get("app_secret") or ""), ) async def test_connection(self) -> IntegrationTestResult: try: await self._client().get_tenant_access_token() except Exception as e: # noqa: BLE001 return IntegrationTestResult( ok=False, message=f"{type(e).__name__}: {e}", ) return IntegrationTestResult(ok=True, message="OK · tenant token") # ── Dispatch on binding.operation ─ _VALID_OPERATIONS = frozenset( {_OP_USER_SUSPEND, _OP_GROUP_MEMBER, _OP_TENANT_USERS, _OP_USER_CREATE}, ) @classmethod def _resolve_operation( cls, extra: dict[str, Any] | None, ) -> tuple[str | None, str | None]: """Return ``(operation, error_message)`` for a dispatch call. Production-grade strictness: never silently fall back to a hard-coded default when the binding row is malformed. The schema marks ``operation`` ``required=True`` so a row created through the admin UI always carries it; this guard catches the path where a binding is written through MCP / direct DB / a legacy migration that left the JSON column empty — those paths bypass the form validator and would otherwise pick up whatever the runtime fallback happened to be at deploy time (a moving target now that the recommended default has changed once). Fail loud + name every valid value so the operator can fix it. Type-tolerant on input (binding_config is JSON) but lossless on signal — ``None``, missing key, and empty string all map to the same "missing" error; non-string types (bool / int / list) get a distinct "wrong type" error so the operator knows it's a casting bug rather than an omission. (Round-55 follow-up.) """ raw = (extra or {}).get("operation") if raw is None or raw == "": return None, ( "Feishu binding is missing required 'operation' field. " f"Set binding_config.operation to one of: " f"{_OP_USER_SUSPEND}|{_OP_GROUP_MEMBER}|" f"{_OP_TENANT_USERS}|{_OP_USER_CREATE}." ) if not isinstance(raw, str): return None, ( f"Feishu binding 'operation' must be a string, got " f"{type(raw).__name__}: {raw!r}. Expected one of: " f"{_OP_USER_SUSPEND}|{_OP_GROUP_MEMBER}|" f"{_OP_TENANT_USERS}|{_OP_USER_CREATE}." ) op = raw.strip() if not op: return None, ( "Feishu binding 'operation' is whitespace-only. " f"Set binding_config.operation to one of: " f"{_OP_USER_SUSPEND}|{_OP_GROUP_MEMBER}|" f"{_OP_TENANT_USERS}|{_OP_USER_CREATE}." ) if op not in cls._VALID_OPERATIONS: return None, ( f"Feishu binding has unknown 'operation': {op!r} " f"(expected {_OP_USER_SUSPEND}|{_OP_GROUP_MEMBER}|" f"{_OP_TENANT_USERS}|{_OP_USER_CREATE})" ) return op, None async def grant_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: op, err = self._resolve_operation(ctx.extra) if err is not None: return ProvisioningResult(ok=False, error_message=err) if op == _OP_GROUP_MEMBER: return await self._modify_member(ctx, op="add") if op == _OP_USER_SUSPEND: return await self._user_reactivate(ctx) if op == _OP_USER_CREATE: return await self._user_create(ctx) if op == _OP_TENANT_USERS: # Tenant-users mode mirrors who has a Feishu account; the # account itself is provisioned by HRIS sync (飞书人事 / # SuccessFactors → Feishu), not by Helix's grant flow. # Returning a hard error (rather than ok=True no-op) is # deliberate: it surfaces a misconfiguration if anyone # routes a grant_seat task at this binding. # ``capability_unsupported`` (2026-06-10, execution-plane # PR3): a BY-DESIGN refusal, not an error — this binding can # never perform the operation, so retrying is pointless. # The flag routes the dispatcher to its auto-downgrade-to- # manual_notify branch (same as Slack's plan-gate) so a # human gets pinged with this text as the instruction, # instead of the task rotting in FAILED while silently # pinning its offboarding blocker (prod: 11 such rows). return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( "Feishu tenant_users binding is read-only. The Feishu " "account itself is the identity — its lifecycle " "(create on hire / freeze on leave) is owned by the " "``feishu-contact`` identity_sync target, not by the " "seat-layer driver. If this software's catalog row is " "the Feishu account, set its offboarding_policy to " "``identity_sync_managed`` so the offboarding scan " "stops creating tasks against it. If you meant to " "add a user to a specific group, switch this binding " "to operation=group_member; if you meant to re-enable " "a frozen account, use operation=user_suspend." ), ) # Defensive default — ``_resolve_operation`` already rejected # unknown ops, so this is unreachable in practice. return ProvisioningResult( ok=False, error_message="Feishu binding: unsupported grant_seat operation.", ) async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult: op, err = self._resolve_operation(ctx.extra) if err is not None: return ProvisioningResult(ok=False, error_message=err) if op == _OP_GROUP_MEMBER: return await self._modify_member(ctx, op="remove") if op == _OP_USER_SUSPEND: # Default revoke = PATCH freeze (preserves account, chat # history, file ownership). Only DELETE the account when # the binding explicitly opts in via # ``hard_delete_on_offboard``. This honours the ~500-emp # cooling-off / rehire policy: a 30-day freeze window # lets ops un-freeze without rebuilding identity from # scratch. hard = bool( (ctx.extra or {}).get("hard_delete_on_offboard") is True, ) if hard: return await self._user_delete_with_freeze_fallback(ctx) return await self._user_freeze(ctx) if op == _OP_USER_CREATE: # user_create's revoke reuses the user_suspend lifecycle — # the asymmetry is intentional: Helix creates the account # via POST on hire, then freezes (default) or deletes (opt # in) on leave through the same Feishu endpoints # ``_user_suspend`` already uses. Reading the binding's # ``user_create_revoke_to_delete`` flag keeps the create-side # and suspend-side hard-delete opt-ins separate, so an # admin who flips on aggressive removal for legacy # user_suspend bindings doesn't accidentally cascade that # policy to user_create bindings — each binding decides for # itself. hard = bool( (ctx.extra or {}).get("user_create_revoke_to_delete") is True, ) if hard: return await self._user_delete_with_freeze_fallback(ctx) return await self._user_freeze(ctx) if op == _OP_TENANT_USERS: # See the grant-side twin above — by-design refusal, route # to the dispatcher's manual_notify downgrade instead of a # terminal FAILED. return ProvisioningResult( ok=False, capability_unsupported=True, error_message=( "Feishu tenant_users binding is read-only. If this " "software's catalog row IS the Feishu account, set " "its offboarding_policy to ``identity_sync_managed`` " "— freezes are run by the ``feishu-contact`` " "identity_sync target's deactivate hook on offboard, " "not by a seat-layer revoke task (running both " "duplicates the call and forks the audit trail). " "To revoke a specific Feishu user group seat, " "switch this binding to operation=group_member. To " "drive freeze from the seat layer (legacy path), " "switch to operation=user_suspend — but prefer the " "identity_sync route." ), ) # Defensive default — ``_resolve_operation`` already rejected # unknown ops, so this is unreachable in practice. return ProvisioningResult( ok=False, error_message="Feishu binding: unsupported revoke_seat operation.", ) @classmethod def revoke_requires_successor( cls, binding_config: dict[str, Any] | None, ) -> bool: """Hard delete (DELETE /contact/v3/users) cascades file ownership / chat ownership the same way Google does on ``user_suspend`` — without a successor, those artifacts are orphaned. Freeze (the default) does NOT need a successor: the account stays addressable so file owners can still be resolved during retention. Applies to both ``user_suspend`` (revoke = freeze/delete via ``hard_delete_on_offboard``) and ``user_create`` (revoke = same path via ``user_create_revoke_to_delete``). Reading the per-mode opt-in keeps the two binding policies independent. """ cfg = binding_config or {} # Missing / unknown operation falls through to ``return False`` # (no successor demanded). The actual ``revoke_seat`` dispatch # will then surface the missing-field error via # ``_resolve_operation`` — keeping the two surfaces consistent # (don't ask the operator for a successor when the binding # itself is malformed and the revoke is going to fail anyway). op = str(cfg.get("operation") or "").strip() if op == _OP_USER_SUSPEND: return bool(cfg.get("hard_delete_on_offboard") is True) if op == _OP_USER_CREATE: return bool(cfg.get("user_create_revoke_to_delete") is True) return False # ── group_member ─ @staticmethod def _feishu_error_code(exc: AppException) -> int | None: """Pull the numeric Feishu error code from an AppException raised by FeishuApprovalClient. The client formats messages as ``"Feishu API error : "`` (see feishu_client.py); we re-parse the code here so dispatch logic doesn't depend on localised message substrings (Chinese-tenant Feishu returns translated messages — substring "not found" never matches "用户不存在"). Anchored on the ``Feishu API error :`` prefix specifically so a numeric run inside ```` (e.g. an upstream message body that happens to contain digits like an internal request id) cannot be misread as the API code. (Round 4 security P2.) """ msg = exc.message or "" # re.search (not re.match) so wrapped error messages like # "delete_user: Feishu API error 1234: ..." still match. The # anchored ^re.match silently failed when a caller prepended # context before re-raising, causing the freeze fallback to # not trigger and leaving offboarded users' accounts active. m = re.search(r"Feishu API error (\d{4,8}):", msg) if not m: return None try: return int(m.group(1)) except (TypeError, ValueError): return None async def _modify_member( self, ctx: ProvisioningContext, *, op: str, ) -> ProvisioningResult: group_chat_id = str((ctx.extra or {}).get("group_chat_id") or "") if not group_chat_id: return ProvisioningResult( ok=False, error_message="binding missing 'group_chat_id'", ) try: client = self._client() feishu_user_id = await client.resolve_user_id_by_email(ctx.user_email) path = ( f"/contact/v3/group/{quote(group_chat_id, safe='')}/member/" + ("add" if op == "add" else "remove") ) await client._request( # noqa: SLF001 — driver lives next to the client "POST", f"{path}?member_type=user&member_id_type=user_id", json_body={"member_id": feishu_user_id}, ) except AppException as e: # Idempotent short-circuit: "already in group" on add / # "not in group" on remove are not failures. Without this, # a re-grant after a partial sync would loop in dead-letter. code = self._feishu_error_code(e) if op == "add" and code in _FEISHU_ALREADY_IN_GROUP_CODES: return ProvisioningResult( ok=True, external_ref=f"{group_chat_id}:{ctx.user_email}", raw_response={"already_member": True, "code": code}, ) if op == "remove" and code in _FEISHU_NOT_IN_GROUP_CODES: return ProvisioningResult( ok=True, external_ref=f"{group_chat_id}:{ctx.user_email}", raw_response={"already_absent": True, "code": code}, ) # Idempotency expansion (Round-1 audit HIGH #2): when the # USER themselves was hard-deleted from Feishu between the # offboarding scan and this revoke task firing, # ``resolve_user_id_by_email`` raises FEISHU_USER_NOT_FOUND # (caught here). Remove-from-group on a non-existent user # is by definition already-absent — desired end-state # (user not in group) is met. Otherwise the revoke task # dead-letters and the offboarding blocker stays open # because of a phantom Feishu account that's already gone. if op == "remove" and ( e.code in _FEISHU_USER_NOT_FOUND_CODES or code in _FEISHU_ALREADY_GONE_CODES ): return ProvisioningResult( ok=True, external_ref=f"{group_chat_id}:{ctx.user_email}", raw_response={ "already_absent": True, "reason": "user_not_found_in_feishu", "code": code, }, ) # Idempotency expansion (Round-1 audit MEDIUM #4): if the # group itself was renamed / deleted out-of-band, the # ``group_chat_id`` in the binding becomes a 404. For # ``remove`` the desired end-state (user not in group) is # trivially met when the group doesn't exist; for ``add`` # we still fail because the admin needs to fix the binding. if op == "remove" and code in _FEISHU_GROUP_NOT_FOUND_CODES: return ProvisioningResult( ok=True, external_ref=f"{group_chat_id}:{ctx.user_email}", raw_response={ "already_absent": True, "reason": "feishu_group_not_found", "code": code, }, ) return ProvisioningResult(ok=False, error_message=e.message) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=f"{group_chat_id}:{ctx.user_email}", ) # ── user_suspend ─ async def _user_delete( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """DELETE /contact/v3/users/{user_id} — full Feishu account removal. Used only when the binding explicitly opts into ``hard_delete_on_offboard``. Treats "user not found" as success for idempotency, but ONLY when the upstream raises the explicit ``FEISHU_USER_NOT_FOUND`` AppException code or the Feishu numeric code is in the known-gone set. A regex parse miss / other transport error is NOT treated as already-absent, because that would silently swallow token-revoked / scope- downgraded / network-blip failures into successful revokes — flipping the seat to revoked locally while the vendor account stays active. (Pre-launch round 2 security P0.) """ try: client = self._client() feishu_user_id = await client.resolve_user_id_by_email(ctx.user_email) except AppException as e: code = self._feishu_error_code(e) if e.code in _FEISHU_USER_NOT_FOUND_CODES or code in _FEISHU_ALREADY_GONE_CODES: logger.info( "feishu user_delete: cannot resolve %s — treating as already absent (%s)", ctx.user_email, e.message, ) return ProvisioningResult( ok=True, external_ref=f"feishu:{ctx.user_email}:already_absent", raw_response={"already_absent": True, "code": code}, ) return ProvisioningResult( ok=False, error_message=f"resolve_user: {e.message}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"resolve_user: {type(e).__name__}: {e}", ) try: await client._request( # noqa: SLF001 "DELETE", f"/contact/v3/users/{quote(feishu_user_id, safe='')}" "?user_id_type=user_id", ) except AppException as e: code = self._feishu_error_code(e) # Collapse to already-absent when EITHER the numeric code # is in the known-gone set OR the AppException code is an # explicit "user not found" signal (FEISHU_USER_NOT_FOUND # from the Feishu client OR generic NOT_FOUND wrapper). # Other AppException codes (token rotated, scope downgrade, # transport blip) bubble as ok=False. if ( code in _FEISHU_ALREADY_GONE_CODES or e.code in _FEISHU_USER_NOT_FOUND_CODES ): return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:already_absent", raw_response={"already_absent": True, "code": code}, ) # Round-1 audit MEDIUM #5 — refined: tag the failure with # FEISHU_HARD_DELETE_REFUSED ONLY when it's a recoverable # BUSINESS refusal (4xx with a numeric Feishu code present, # OR 4xx without a code that isn't a transport blip). The # wrapping ``_user_delete_with_freeze_fallback`` reads the # tag to decide whether to fall back to freeze. We must NOT # tag transients (5xx, network reset, token revoked) — # otherwise the fallback would silently switch from DELETE # to freeze on a transient that would have succeeded on # retry, masking real upstream bugs and leaving the seat in # a confusing "fallback freeze" state. # # ``AppException.status_code`` is the canonical signal — 4xx # = business, 5xx = transient. The Feishu numeric ``code`` # being present strengthens the "business refusal" case # because Feishu only emits its code numbers for # application-level rejections (not for 5xx infra issues). status_code = getattr(e, "status_code", None) is_business_refusal = ( isinstance(status_code, int) and 400 <= status_code < 500 ) if is_business_refusal: return ProvisioningResult( ok=False, error_message=( f"[{FEISHU_TAG_HARD_DELETE_REFUSED}] {e.message}" ), raw_response={ "feishu_code": code, "error_tag": FEISHU_TAG_HARD_DELETE_REFUSED, "vendor_message": e.message, "status_code": status_code, }, ) # Transient — surface as-is so the worker retries. return ProvisioningResult(ok=False, error_message=e.message) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:deleted", raw_response={"deleted": True}, ) async def _user_delete_with_freeze_fallback( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """Hard-delete with reversible-freeze fallback (Round-1 CRITICAL). When the binding opts into ``hard_delete_on_offboard=true`` and Feishu refuses the DELETE for a non-idempotent business reason (file ownership not transferred, parent department in cascade freeze, content-handoff required by tenant policy), the prior implementation surfaced the failure and the offboarding task dead-lettered after 10 retries — manual admin intervention required to clear the blocker. The seat-revoke semantic is "remove the user from the active roster"; freeze (PATCH ``is_frozen=true``) achieves exactly that without touching content, and is the reversible path admin would have chosen anyway for cooling-off / rehire. So we treat a tagged hard-delete refusal as a recoverable failure and fall back to freeze, stamping ``hard_delete_fallback=True`` on the success ``raw_response`` so audit can distinguish: * clean freeze (binding default, never tried delete) * fallback freeze (binding asked for delete; Feishu refused; we freeze instead — flagged so compliance review can re-attempt the hard delete manually once the blocking condition is cleared) """ delete_result = await self._user_delete(ctx) if delete_result.ok: return delete_result raw = delete_result.raw_response or {} is_recoverable_refusal = ( isinstance(raw, dict) and raw.get("error_tag") == FEISHU_TAG_HARD_DELETE_REFUSED ) or ( (delete_result.error_message or "").startswith( f"[{FEISHU_TAG_HARD_DELETE_REFUSED}] " ) ) if not is_recoverable_refusal: # Transport / token / scope error — surface as-is so the # worker retries. Falling back to freeze on a transient # would silently flip the seat to revoked while Feishu # still had a hard-delete attempt pending in the queue. return delete_result logger.warning( "feishu hard-delete refused for %s — falling back to " "freeze. vendor=%s", ctx.user_email, delete_result.error_message, ) freeze_result = await self._user_freeze(ctx) if freeze_result.ok: freeze_raw = dict(freeze_result.raw_response or {}) freeze_raw.update({ "hard_delete_fallback": True, "original_delete_error": delete_result.error_message, }) return ProvisioningResult( ok=True, external_ref=freeze_result.external_ref, raw_response=freeze_raw, ) # Even the freeze failed — surface the freeze error (more # informative than the original hard-delete refusal at this # point because freeze should always work). Preserve the # original delete error in raw_response so audit can # reconstruct the full sequence (Round-3 logic scan LOW #3). freeze_raw = dict(freeze_result.raw_response or {}) freeze_raw["skipped_hard_delete_error"] = delete_result.error_message return ProvisioningResult( ok=freeze_result.ok, external_ref=freeze_result.external_ref, error_message=freeze_result.error_message, raw_response=freeze_raw, ) async def _user_freeze( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """PATCH /contact/v3/users/{user_id} with ``status.is_frozen=true``. Default revoke path for ``operation=user_suspend`` bindings. Preserves the account, file ownership, chat history — Feishu admin can later un-freeze for rehire / cooling-off without rebuilding identity from scratch. The hard delete path (``hard_delete_on_offboard=true``) remains available for compliance-mandated removal. """ try: client = self._client() feishu_user_id = await client.resolve_user_id_by_email(ctx.user_email) except AppException as e: code = self._feishu_error_code(e) # Only collapse to already-absent on EXPLICIT not-found # signal — same hardening as ``_user_delete``. A regex # parse miss must NOT be misread as "user gone"; # otherwise a transient token / scope failure would # silently flip the seat to revoked locally while the # Feishu account stayed active. (Round 2 security P0.) if e.code in _FEISHU_USER_NOT_FOUND_CODES or code in _FEISHU_ALREADY_GONE_CODES: return ProvisioningResult( ok=True, external_ref=f"feishu:{ctx.user_email}:already_absent", raw_response={"already_absent": True, "code": code}, ) return ProvisioningResult( ok=False, error_message=f"resolve_user: {e.message}", ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"resolve_user: {type(e).__name__}: {e}", ) try: await client._request( # noqa: SLF001 "PATCH", f"/contact/v3/users/{quote(feishu_user_id, safe='')}" "?user_id_type=user_id", json_body={ "status": { "is_frozen": True, }, }, ) except AppException as e: code = self._feishu_error_code(e) if code in _FEISHU_ALREADY_FROZEN_CODES: return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:already_frozen", raw_response={"already_frozen": True, "code": code}, ) if ( code in _FEISHU_ALREADY_GONE_CODES or e.code in _FEISHU_USER_NOT_FOUND_CODES ): return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:already_absent", raw_response={"already_absent": True, "code": code}, ) return ProvisioningResult(ok=False, error_message=e.message) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:frozen", raw_response={"frozen": True}, ) async def _user_reactivate( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """Best-effort reactivation for user_suspend mode. Users are normally created via HRIS sync, not through Helix's provisioning queue. This path runs when an employee "returns" from leave — if Feishu still has the user record, we ensure they're marked active; if Feishu has no record at all, we no-op rather than create one (HRIS is authoritative for account creation). """ try: client = self._client() feishu_user_id = await client.resolve_user_id_by_email(ctx.user_email) except AppException as e: code = self._feishu_error_code(e) # Same hardening as _user_freeze / _user_delete: only # collapse to no-op when the upstream is EXPLICITLY # signalling user-not-found. Other AppExceptions (token # rotation, scope downgrade, rate limit) are surfaced as # ok=False so the worker retries instead of silently # claiming the reactivation succeeded. (Round 2 security P0.) if ( e.code not in _FEISHU_USER_NOT_FOUND_CODES and code not in _FEISHU_ALREADY_GONE_CODES ): return ProvisioningResult( ok=False, error_message=f"resolve_user: {e.message}", ) logger.info( "feishu user_reactivate: %s not present in Feishu — " "no-op (HRIS will recreate on rehire). msg=%s", ctx.user_email, e.message, ) return ProvisioningResult( ok=True, external_ref=f"feishu:{ctx.user_email}:no_feishu_record", raw_response={"no_feishu_record": True}, ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"resolve_user: {type(e).__name__}: {e}", ) # PATCH /contact/v3/users/{user_id} with status clearing any # resigned/frozen flag. Exact body shape is tolerant: Feishu # ignores unknown fields, so sending both ``is_frozen=false`` # and ``is_resigned=false`` covers older and newer dialects. try: await client._request( # noqa: SLF001 "PATCH", f"/contact/v3/users/{quote(feishu_user_id, safe='')}" "?user_id_type=user_id", json_body={ "status": { "is_frozen": False, "is_resigned": False, }, }, ) except AppException as e: # Round 4 logic-scan P2: previously this branch swallowed # EVERY AppException as a soft success — a token-revoked / # scope-downgraded / rate-limit error was reported as # "reactivated" while Feishu still had the user frozen. # Now: only collapse to soft ok when Feishu explicitly # returns "not-supported" / "feature-not-available" codes # AND the user already exists (which we proved earlier in # this function via resolve_user_id_by_email). Other # AppExceptions bubble as ok=False so the worker retries. code = self._feishu_error_code(e) # 99991674 = feature not enabled on tenant plan; 230050 = # endpoint deprecated for legacy plans. Add new codes here # only when verified to be "PATCH not supported" semantics. patch_not_supported_codes = {99991674, 230050} if code in patch_not_supported_codes: logger.warning( "feishu user_reactivate PATCH not supported on this " "tenant plan for %s: %s — treating as soft success " "(HRIS is authoritative)", ctx.user_email, e.message, ) return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:soft_ok", raw_response={ "soft_ok": True, "reason": e.message, "code": code, }, ) return ProvisioningResult( ok=False, error_message=e.message, ) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:reactivated", raw_response={"reactivated": True}, ) # ── user_create (round-19) ───────────────────────────────── async def _user_create( self, ctx: ProvisioningContext, ) -> ProvisioningResult: """POST /contact/v3/users — actively create a new Feishu account. Default-OFF entry point: only fires when a binding explicitly sets ``operation=user_create``. The flow: 1. Resolve target department from ``ctx.extra['default_department_id']`` (set on the binding, required by the binding_schema). If absent, fail clean — a Feishu create with an empty ``department_ids`` array is rejected at the API boundary anyway. 2. Build minimal create body — name + email + department. ``employee_no`` is intentionally omitted; HRIS sync will backfill it on its next pass. Mobile + gender + country + city are all omitted for the same reason — keeping the initial create payload small lowers the surface area for Feishu validation errors and makes the rehire / move-dept reconciliation purely HRIS-driven. 3. Idempotency: if Feishu returns "already exists" (mobile / email / employee_no collision — codes 43002 / 43003 / 43005), collapse to ok with ``already_exists=True``. This handles the HRIS-races-Helix case where the Feishu account got created between Helix's hire-day scan and the dispatcher firing the grant — re-running the grant must NOT fail. 4. Department-not-found (99991664) bubbles as ok=False with a clear message pointing the admin to the binding config — silent retry would loop forever on a misconfigured binding. 5. Other AppExceptions (token rotated, scope downgrade, transport blip) bubble as ok=False so the worker retries. """ cfg = ctx.extra or {} dept_id = str(cfg.get("default_department_id") or "").strip() if not dept_id: return ProvisioningResult( ok=False, error_message=( "Feishu user_create binding is missing required " "'default_department_id'. Set it on the integration " "binding (an open_department_id from Feishu admin) " "before this mode can fire." ), ) if not ctx.user_email: return ProvisioningResult( ok=False, error_message=( "Feishu user_create requires user_email — caller " "passed an empty email so we cannot register a Feishu " "account against an unknown identity." ), ) body: dict[str, Any] = { "name": ctx.user_display_name or ctx.user_email, "email": ctx.user_email, "department_ids": [dept_id], } try: client = self._client() resp = await client._request( # noqa: SLF001 — driver lives next to client "POST", "/contact/v3/users?user_id_type=user_id" "&department_id_type=open_department_id", json_body=body, ) except AppException as e: code = self._feishu_error_code(e) if code in _FEISHU_USER_ALREADY_EXISTS_CODES: # HRIS raced Helix or a retry hit a half-committed user. # Best-effort: resolve the user_id by email so the # external_ref points at the actual Feishu record; # if the resolve itself fails we still return ok with # an email-based external_ref so the seat allocation # ledger has SOMETHING to anchor on. try: feishu_user_id = await self._client().resolve_user_id_by_email( ctx.user_email, ) except Exception: # noqa: BLE001 feishu_user_id = ctx.user_email logger.info( "feishu user_create: %s already exists (code=%s) — " "treating as idempotent ok", ctx.user_email, code, ) return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:already_exists", raw_response={ "already_exists": True, "code": code, "message": e.message, }, ) if code in _FEISHU_DEPARTMENT_NOT_FOUND_CODES: return ProvisioningResult( ok=False, error_message=( f"Feishu department {dept_id!r} not found " f"(Feishu code={code}). Fix the binding's " f"default_department_id (use an open_department_id " f"that starts with 'od-' from Feishu admin → " f"通讯录 → 部门), then retry." ), ) return ProvisioningResult(ok=False, error_message=e.message) except Exception as e: # noqa: BLE001 return ProvisioningResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) # Success path — Feishu returns the new user record under # ``user.user_id``. Stamp that as the external_ref so future # revokes / freezes can resolve directly without re-querying. feishu_user_id = "" if isinstance(resp, dict): user_node = resp.get("user") or {} if isinstance(user_node, dict): feishu_user_id = str(user_node.get("user_id") or "").strip() if not feishu_user_id: # Defensive: shape changed upstream. Fall back to email # so the row still has a non-empty external_ref. feishu_user_id = ctx.user_email return ProvisioningResult( ok=True, external_ref=f"feishu:{feishu_user_id}:created", raw_response={"created": True, "user_id": feishu_user_id}, ) # ── seat_sync helpers ───────────────────────────────────── async def list_all_tenant_users_with_emails( self, ) -> dict[str, Any]: """Return every active user the app can see in the Feishu tenant. Used by the ``feishu.seat_sync`` scheduled job when a binding is in :data:`_OP_TENANT_USERS` mode — i.e., "every active Feishu account is one seat on this plan", same shape as Google Workspace's base seat sync. Implementation strategy: 1. ``GET /contact/v3/departments?parent_department_id=0& fetch_child=true&page_size=50`` paginated → enumerates every department the app's contact-scope can reach (page size is capped at 50 by Feishu's validator on this endpoint). 2. For each department — **root "0" included** — ``GET /contact/v3/users/find_by_department ?department_id=&department_id_type=open_department_id& fetch_child=false&page_size=50&user_id_type=user_id`` paginated → returns the *direct* members of that department. ``fetch_child=false`` (not true) is deliberate: Feishu allows a user to live in multiple departments, so descending the tree with fetch_child=true causes the same user to land in every ancestor's response. Walking direct members only + de-duping by ``user_id`` is both faster and easier to reason about. Root "0" must be walked explicitly because the departments listing in step 1 only returns root's *descendants* — it never returns root itself, yet Feishu parks users with no department assignment directly on root. Production 2026-07-02: admin console showed 451 members while this walk mirrored 450 — the missing one was a root-parked account. Same contract as identity-sync's ``fetch_feishu_directory`` ("root included — Feishu parks dept-less users there"). 3. Merge: dedup by ``user_id``; drop resigned + frozen accounts (those don't occupy a paid Feishu seat) but COUNT what was dropped (see below); return one record per unique active user. Return shape:: { "users": [ { "user_id": "", # always non-empty "email": "user@company.com", # may be "" if hidden "name": "", # best-effort }, ... ], "roster_total": int, # distinct accounts the walk SAW "skipped_frozen": int, # dropped: status.is_frozen "skipped_resigned": int, # dropped: status.is_resigned "departments_attempted": int, # total deps the walk tried "departments_failed": int, # deps that raised AppException "failure_examples": list[str], # up to 5, ": " } **Why the skip counters exist (2026-07-29).** ``len(users)`` alone cannot be reconciled against the Feishu admin console: 成员与部门 → 成员 counts 总人数 as *active + frozen* (已离职 lives on its own tab), so the console number is ``len(users) + skipped_frozen`` while this method returns only the seat-occupying subset. A user asking "后台 446, 同步 445, 哪儿漏了?" previously had no way to tell a deliberate frozen-account skip from a silently-missed department — the two look identical from the run result, and an out-of-scope department does NOT raise (``/contact/v3/departments`` simply omits it, so ``departments_failed`` stays 0). Emitting the counters makes the arithmetic self-serve:: console 总人数 == roster_total - skipped_resigned console 已激活+未激活 == len(users) (a.k.a. members_listed) ``roster_total`` counts DISTINCT ``user_id``s, so a user living in several departments still counts once. Precedence when an account carries both flags: ``is_resigned`` wins (a resigned account is usually frozen too, and the console files it under 已离职), so the two counters never double-count. Per-department isolation: when a single department rejects ``find_by_department`` (most commonly Feishu code ``2200 call CheckAppTenantScopes failed`` — that department was moved outside the app's 通讯录范围 on the Feishu admin side, or a new restricted subtree was created), this method records the failure and continues with the remaining departments instead of aborting the entire walk. Surfaced via ``departments_failed`` + ``failure_examples`` so the seat-sync handler can attach the count to the per-plan summary and the admin sees "X/Y departments unreachable — fix app scope" as an actionable signal in the run result. Only raised on: * Top-level ``/contact/v3/departments`` failure (no roster can be enumerated at all — auth/token/global outage). * Every single department failing (cumulative failure == attempted > 0 — again, a global condition like a rotated tenant token, not per-dept scope drift). * Paginator divergence (``has_more=true`` past the hard cap) on either endpoint — protects the worker from a wedged loop. Members with no email aren't dropped here — the caller (the seat-sync handler) buckets them into ``members_without_email`` so the admin can see how many account holders couldn't be matched to a Helix person via email. Hard limits on the Feishu side that this method respects: ``/contact/v3/departments`` and ``/contact/v3/users/find_by_ department`` both cap ``page_size`` at 50. Each rejects 51+ with HTTP 400 ``99992402 field validation failed``. Confirmed against the live PingCAP tenant 2026-05-02. Page-size caveat (separately confirmed): ``/contact/v3/group/.../member/simplelist`` caps at 100 but ``/contact/v3/users/find_by_department`` caps at 50 — every Feishu endpoint has its own validator. Don't extrapolate. """ client = self._client() # ── Step 1: enumerate every accessible department ─ # # Root "0" is seeded up front: the /departments listing below # only enumerates root's descendants and never root itself, # but Feishu parks dept-less users (e.g. accounts created # before their department assignment lands) directly on root. # Without this seed those users are invisible to the walk and # the mirrored seat count runs permanently one-or-more short # of the Feishu admin console (observed 450 vs 451, # 2026-07-02). all_depts: list[str] = ["0"] page_token: str | None = None for _page in range(_MAX_PAGINATOR_PAGES): qs = ( "parent_department_id=0&fetch_child=true&page_size=50" "&department_id_type=open_department_id" ) if page_token: qs += f"&page_token={quote(page_token, safe='')}" try: page = await client._request( # noqa: SLF001 — same pkg "GET", f"/contact/v3/departments?{qs}", ) except AppException as e: raise RuntimeError( f"Feishu departments list failed: {e.message}", ) from e for d in page.get("items") or []: if not isinstance(d, dict): continue did = ( d.get("open_department_id") or d.get("department_id") or "" ) did = str(did).strip() if did: all_depts.append(did) if not page.get("has_more"): break page_token = page.get("page_token") or None if not page_token: break else: raise RuntimeError( f"Feishu departments paginator did not terminate after " f"{_MAX_PAGINATOR_PAGES} pages.", ) # ── Step 2: walk users per department (fetch_child=false) ─ # # We iterate departments serially (not asyncio.gather) because # Feishu's tenant_access_token rate limits pile up on parallel # bursts and the daily sync is not latency-sensitive — ~150 # departments × ~80ms each ≈ 12 s, well under the worker's # per-job budget. # # Per-department isolation: production incident 2026-05-28 # showed a single department returning Feishu code ``2200 # CheckAppTenantScopes failed`` (an admin moved a subtree # outside the app's 通讯录范围) aborts the entire walk and # zeroes the sync — yesterday's run mirrored 477 members, this # run produced 0. We now collect per-dept failures into a # diagnostics bucket and continue. The total-failure raise # below still catches genuine global outages (rotated token → # every dept fails). seen: dict[str, dict[str, Any]] = {} # Deliberately-dropped accounts, deduped by user_id (a frozen # user living in 3 departments is ONE skip, not three). Sets # rather than counters for exactly that reason. skipped_resigned: set[str] = set() skipped_frozen: set[str] = set() departments_failed = 0 failure_examples: list[str] = [] _failure_example_cap = 5 for did in all_depts: page_token = None dept_aborted = False for _page in range(_MAX_PAGINATOR_PAGES): qs = ( f"department_id={quote(did, safe='')}" "&department_id_type=open_department_id" "&fetch_child=false&page_size=50" "&user_id_type=user_id" ) if page_token: qs += f"&page_token={quote(page_token, safe='')}" try: page = await client._request( # noqa: SLF001 "GET", f"/contact/v3/users/find_by_department?{qs}", ) except AppException as e: departments_failed += 1 if len(failure_examples) < _failure_example_cap: failure_examples.append( f"{did}: {e.message}", ) logger.warning( "feishu.seat_sync: find_by_department(%r) " "failed; skipping department: %s", did, e.message, ) dept_aborted = True break for u in page.get("items") or []: if not isinstance(u, dict): continue uid = str(u.get("user_id") or "").strip() if not uid: continue if ( uid in seen or uid in skipped_resigned or uid in skipped_frozen ): # User belongs to multiple departments — already # counted (as a member OR as a skip); subsequent # appearances add no info. continue status = u.get("status") or {} if not isinstance(status, dict): status = {} # Resigned / frozen accounts don't occupy a paid # Feishu seat. Skip them deliberately — but record # WHICH so the handler can explain the delta against # the admin console instead of leaving the operator # to guess (see the docstring's arithmetic). # is_resigned wins when both flags are set. if status.get("is_resigned"): skipped_resigned.add(uid) continue if status.get("is_frozen"): skipped_frozen.add(uid) continue seen[uid] = { "user_id": uid, "email": str(u.get("email") or "").strip(), "name": str(u.get("name") or "").strip(), } if not page.get("has_more"): break page_token = page.get("page_token") or None if not page_token: break else: # Paginator divergence within a single department is a # protocol-level wedge (we'd loop forever), not a per- # dept scope drift — bubble it up so the worker doesn't # silently stop after _MAX_PAGINATOR_PAGES of garbage. raise RuntimeError( f"Feishu find_by_department({did!r}) paginator did " f"not terminate after {_MAX_PAGINATOR_PAGES} pages.", ) if dept_aborted: continue # If every department failed, it's a global condition (token # rotated, app suspended, Feishu open-API outage) — not the # per-dept scope drift this method is built to absorb. Raise # so raise_if_total_failure escalates instead of silently # mirroring an empty roster and triggering revoke against # every Helix seat (safe_to_drift_revoke would refuse, but # we'd rather surface the outage explicitly than rely on the # safety helper to dead-cat the run). if departments_failed == len(all_depts) and departments_failed > 0: raise RuntimeError( f"Feishu find_by_department failed for every department " f"({departments_failed}/{len(all_depts)}); likely a " f"global condition (token rotated / app scope revoked / " f"Feishu outage). First failure: " f"{failure_examples[0] if failure_examples else 'n/a'}" ) # Surface a truncation sentinel when more departments failed # than fit in the example cap, so admins reading per_plan see # "+95 more" rather than just the first 5 — without that the # count and the examples can disagree confusingly (e.g. # failed=100 but list has only 5 entries). if departments_failed > _failure_example_cap: failure_examples.append( f"... and {departments_failed - _failure_example_cap} more" ) return { "users": list(seen.values()), "roster_total": ( len(seen) + len(skipped_resigned) + len(skipped_frozen) ), "skipped_frozen": len(skipped_frozen), "skipped_resigned": len(skipped_resigned), "departments_attempted": len(all_depts), "departments_failed": departments_failed, "failure_examples": failure_examples, } async def list_group_members_with_emails( self, group_chat_id: str, ) -> dict[str, Any]: """Return active group members with their resolved Feishu email. Two-step join because Feishu's group-member endpoint only returns ids (no email/name): 1. ``GET /contact/v3/group/{group_id}/member/simplelist`` walks every page (``page_token`` cursor, ``page_size=200``) and collects every entry with ``member_type='user'``. Departments and bots are skipped — they don't map to a Helix person. 2. ``GET /contact/v3/users/batch?user_ids=…`` resolves each user_id to a contact record and pulls the ``email`` field. Feishu caps the batch at 50 ids per call, so we chunk. Return shape:: { "users": [ { "user_id": "", # always set "email": "user@company.com", # may be "" if hidden "name": "", # best-effort }, ... ], "batches_attempted": int, # total /users/batch chunks tried "batches_failed": int, # chunks that raised AppException "failure_examples": list[str], # up to 5, capped } Per-batch isolation (symmetric to ``list_all_tenant_users_with_emails``): when a single ``/users/batch`` chunk fails (rate limit, transient 500, a single deleted user_id in the chunk poisoning the response), the failure is recorded and the walk proceeds with the remaining chunks. Members in the failed chunk are still emitted with ``email=""`` so the handler buckets them into ``members_without_email`` rather than dropping them silently — the email gap is identical to "Feishu hid the email" from the handler's perspective. Only raise if EVERY batch fails (global outage / rotated token / app suspended), so a single bad chunk doesn't zero the whole group sync. Members whose contact record exposes no email (privacy off, external contact, frozen account) are returned with ``email=""`` so the caller can count and surface them in ``per_plan`` output without dropping them silently. Raises ``RuntimeError`` on: * Top-level simplelist failure (no roster can be enumerated). * Every users/batch chunk failing (global condition). * Paginator divergence on simplelist (wedged loop protection). """ client = self._client() # ── Step 1: walk simplelist with pagination ─ # Feishu caps simplelist at page_size=100 (other contact endpoints # accept 200 but this one has its own validator that rejects # >100 with HTTP 400 ``99992402 field validation failed``). # Confirmed against the live tenant 2026-05-02. members_raw: list[dict[str, Any]] = [] page_token: str | None = None for _page in range(_MAX_PAGINATOR_PAGES): qs = "member_id_type=user_id&page_size=100" if page_token: # ``page_token`` values are opaque Feishu cursors; # quote defensively in case Feishu ever ships a # cursor with reserved chars. qs += f"&page_token={quote(page_token, safe='')}" try: page = await client._request( # noqa: SLF001 "GET", f"/contact/v3/group/{group_chat_id}/member/simplelist?{qs}", ) except AppException as e: raise RuntimeError( f"Feishu group member/simplelist failed for " f"{group_chat_id!r}: {e.message}", ) from e for m in page.get("memberlist") or []: if not isinstance(m, dict): continue if (m.get("member_type") or "").lower() != "user": continue mid = str(m.get("member_id") or "").strip() if mid: members_raw.append({"user_id": mid}) if not page.get("has_more"): break page_token = page.get("page_token") or None if not page_token: break else: raise RuntimeError( f"Feishu group simplelist paginator did not terminate " f"after {_MAX_PAGINATOR_PAGES} pages for " f"{group_chat_id!r}.", ) if not members_raw: return { "users": [], "batches_attempted": 0, "batches_failed": 0, "failure_examples": [], } # ── Step 2: batch-resolve user_id → email ─ # # Per-batch isolation: a single chunk failure (rate limit, # transient 500, a deleted user_id poisoning the response) # used to abort the entire group walk via `raise RuntimeError`. # Symmetric to the tenant_users fix (incident 2026-05-28) — we # now record the failure, emit ALL members from the failed # chunk with email="" (so the handler can still count them in # ``members_without_email`` rather than silently dropping the # chunk's 50 user_ids), and continue. Only raises if EVERY # chunk fails (= global condition). out: list[dict[str, Any]] = [] chunk_size = 50 # Feishu's documented hard cap on /users/batch batches_attempted = 0 batches_failed = 0 failure_examples: list[str] = [] _failure_example_cap = 5 for i in range(0, len(members_raw), chunk_size): chunk = members_raw[i : i + chunk_size] qs_parts = ["user_id_type=user_id"] for m in chunk: qs_parts.append(f"user_ids={quote(m['user_id'], safe='')}") batches_attempted += 1 try: batch = await client._request( # noqa: SLF001 "GET", "/contact/v3/users/batch?" + "&".join(qs_parts), ) except AppException as e: batches_failed += 1 if len(failure_examples) < _failure_example_cap: failure_examples.append( f"chunk[{i}:{i + len(chunk)}] " f"({len(chunk)} ids): {e.message}", ) logger.warning( "feishu.seat_sync: users/batch failed for group %r " "chunk[%d:%d]; skipping batch: %s", group_chat_id, i, i + len(chunk), e.message, ) # Emit chunk members with email="" so the handler's # ``members_without_email`` counter accounts for them # — silently dropping 50 ids would shrink the # `members_present_user_ids` set passed to the drift # cleanup, risking false-positive revokes on the next # tick. Empty-email rows fall through both import # (no email lookup) AND drift cleanup (the seat's # `user_id` isn't in the empty-email bucket → unchanged). for m in chunk: out.append({ "user_id": m["user_id"], "email": "", "name": "", }) continue by_id: dict[str, dict[str, Any]] = {} for item in batch.get("items") or []: if not isinstance(item, dict): continue uid = str(item.get("user_id") or "").strip() if not uid: continue by_id[uid] = item for m in chunk: rec = by_id.get(m["user_id"]) or {} out.append({ "user_id": m["user_id"], "email": str(rec.get("email") or "").strip(), "name": str(rec.get("name") or "").strip(), }) # Same total-failure guard as tenant_users mode: every batch # failing means a global condition (rotated token / app # suspended / Feishu open-API outage), not per-chunk transient # noise. Raise so raise_if_total_failure escalates. if batches_failed == batches_attempted and batches_attempted > 0: raise RuntimeError( f"Feishu users/batch failed for every chunk " f"({batches_failed}/{batches_attempted}) of group " f"{group_chat_id!r}; likely a global condition (token " f"rotated / app scope revoked / Feishu outage). First " f"failure: " f"{failure_examples[0] if failure_examples else 'n/a'}" ) # Same truncation sentinel as the tenant walk — don't let the # count and the examples-list disagree silently. if batches_failed > _failure_example_cap: failure_examples.append( f"... and {batches_failed - _failure_example_cap} more" ) return { "users": out, "batches_attempted": batches_attempted, "batches_failed": batches_failed, "failure_examples": failure_examples, }