"""Feishu (Lark) chat_approval driver. Hosts multiple approval workflows per integration. Each workflow is a row in ``integration_approval_flows`` (flow_key → external_code). Adding a new workflow = admin adds a row; zero code change. Webhook security supports v2 X-Lark-Signature (preferred) and legacy token verification (fallback when encrypt_key isn't configured). """ from __future__ import annotations import hashlib import hmac import json import logging import time from typing import Any from uuid import UUID from redis.exceptions import RedisError from app.approval.feishu_client import FeishuApprovalClient from app.integrations.driver_base import ( ApprovalSubmission, ApprovalSubmissionResult, ChatApprovalDriver, WebhookVerificationResult, ) from app.integrations.registry import register_driver from app.integrations.schemas import ( IntegrationFieldInfo, IntegrationTestResult, ) logger = logging.getLogger(__name__) _CONFIG_SCHEMA = [ IntegrationFieldInfo( key="app_id", label="App ID", is_secret=False, required=True, placeholder="cli_xxxxxxxxxxxxxxxx", hint="飞书开放平台 > 凭证与基础信息 > App ID。", ), IntegrationFieldInfo( key="app_secret", label="App Secret", is_secret=True, required=True, hint="App Secret,仅在轮换时重新粘贴。", ), IntegrationFieldInfo( key="event_verification_token", label="Event Verification Token", is_secret=True, required=False, hint="飞书事件订阅 Verification Token(旧版回调校验)。", ), IntegrationFieldInfo( key="event_encrypt_key", label="Event Encrypt Key", is_secret=True, required=False, hint="v2 X-Lark-Signature 验签用 Encrypt Key(建议配置)。", ), IntegrationFieldInfo( key="alert_webhook_url", label="运维告警 Webhook URL", is_secret=False, required=False, hint="飞书自定义机器人 Webhook,留空仅写日志不推送。", ), ] @register_driver class FeishuChatApprovalDriver(ChatApprovalDriver): provider = "feishu" display_name = "飞书 (Lark) 审批" description = ( "飞书开放平台 Approval v4。用于资产/软件坐席/API Key 申请等企业审批流。" "每个流程在连接器的「审批流程」子面板里数据驱动新增。" ) supports_multi_instance = True config_schema = _CONFIG_SCHEMA def _client(self) -> FeishuApprovalClient: return FeishuApprovalClient( app_id=str(self.config.get("app_id") or ""), app_secret=str(self.config.get("app_secret") or ""), ) # ─── Health ─ 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}", detail={"type": type(e).__name__}, ) return IntegrationTestResult( ok=True, message="OK · tenant_access_token 获取成功", ) # ─── Approval submission ─ async def submit_approval( self, submission: ApprovalSubmission, ) -> ApprovalSubmissionResult: # ``submission.fields`` is the canonical dict. The router layer # already resolved flow_key → external_code via the service, so # we expect ``submission.fields["__external_code__"]`` + # ``submission.fields["__form__"]`` (a pre-built Feishu form # JSON string). Keeping the driver dumb — all widget-id mapping # lives in the approval service using ``payload_schema``. external_code = submission.fields.get("__external_code__") form_json = submission.fields.get("__form__") user_id = submission.open_user_id provider_request_uuid = (submission.provider_request_uuid or "").strip() try: UUID(provider_request_uuid) except (TypeError, ValueError, AttributeError): return ApprovalSubmissionResult( ok=False, error_message=( "Internal error: submission missing a valid durable " "provider_request_uuid" ), ) if not external_code or not form_json or not user_id: return ApprovalSubmissionResult( ok=False, error_message=( "Internal error: submission missing __external_code__, " "__form__, or open_user_id" ), ) try: data = await self._client().create_instance( approval_code=str(external_code), user_id=str(user_id), form=str(form_json), instance_uuid=provider_request_uuid, ) except Exception as e: # noqa: BLE001 return ApprovalSubmissionResult( ok=False, error_message=f"{type(e).__name__}: {e}", ) return ApprovalSubmissionResult( ok=True, instance_code=data.get("instance_code"), raw_response=data, ) async def cancel_approval( self, instance_code: str, *, user_id: str | None = None, approval_code: str | None = None, ) -> bool: # Caller-error vs Feishu-rejection are now distinguishable. # Previously ``return False`` for missing args silently # short-circuited the cancel-on-timeout flow (employee withdraws), # leaving the Feishu instance pending forever. if not (user_id and approval_code): raise ValueError( "cancel_approval requires user_id and approval_code; " f"caller passed user_id={user_id!r} " f"approval_code={approval_code!r}", ) try: await self._client().cancel_instance( instance_code=instance_code, user_id=user_id, approval_code=approval_code, ) return True except Exception as e: # noqa: BLE001 logger.warning("feishu: cancel instance %s failed: %s", instance_code, e) return False async def resolve_user_id_by_email(self, email: str) -> str | None: try: return await self._client().resolve_user_id_by_email(email) except Exception: # noqa: BLE001 return None async def get_approval_definition(self, external_code: str) -> dict[str, Any]: return await self._client().get_approval_definition(external_code) async def get_instance_detail(self, instance_code: str) -> dict[str, Any]: return await self._client().get_instance_detail(instance_code) # ─── Webhook verification ─ async def verify_webhook( self, *, body: bytes, headers: dict[str, str], ) -> WebhookVerificationResult: try: payload = json.loads(body) if body else {} except ValueError as e: return WebhookVerificationResult( ok=False, error_message=f"Invalid JSON body: {e}", ) # URL verification challenge — no signature required, Feishu sends this # during endpoint registration. if payload.get("type") == "url_verification": return WebhookVerificationResult( ok=True, challenge=payload.get("challenge"), ) # Feishu webhooks come in two flavors: # * v2 events (user-scope): signed with X-Lark-Signature header # * v1 events (``approval_instance`` v1.0, the one we subscribe # for instance status changes): no signature header, token # embedded in the JSON body. # Accept either path (see PR #216 for the production incident # that motivated this fallback). encrypt_key = str(self.config.get("event_encrypt_key") or "") signature = headers.get("x-lark-signature") or headers.get("X-Lark-Signature") if signature and encrypt_key: # v2 header-signed path with replay protection. ts = headers.get("x-lark-request-timestamp") or headers.get( "X-Lark-Request-Timestamp", "", ) nonce = headers.get("x-lark-request-nonce") or headers.get( "X-Lark-Request-Nonce", "", ) try: ts_int = int(ts) except (ValueError, TypeError): return WebhookVerificationResult( ok=False, error_message="Invalid X-Lark-Request-Timestamp", ) if abs(time.time() - ts_int) > 300: return WebhookVerificationResult( ok=False, error_message="Request timestamp too old (replay)", ) content = (ts + nonce + encrypt_key).encode() + body expected = hashlib.sha256(content).hexdigest() if not hmac.compare_digest(expected, signature): return WebhookVerificationResult( ok=False, error_message="X-Lark-Signature mismatch", ) # Nonce-replay cache (Round-44). The HMAC + timestamp window # ensures a captured request can't be replayed outside ±300s, # but it CAN be replayed within that window. SOC2 group-add # callbacks are conditionally idempotent on local state; a # rapid add/remove/add race could be amplified by a captured # replay. SETNX with a TTL longer than the window keeps the # gate honest. Fail-open on Redis unavailability so a Redis # blip doesn't refuse legitimate live callbacks. try: from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: key = f"feishu:webhook_replay:{nonce}:{ts}" set_result = await r.set(key, "1", nx=True, ex=600) if not set_result: return WebhookVerificationResult( ok=False, error_message=( "Replayed webhook (nonce+timestamp " "already processed within ±10min window)" ), ) except Exception: # noqa: BLE001 — fail-open on Redis fault logger.warning( "feishu webhook replay cache unavailable — " "accepting without dedup", ) return WebhookVerificationResult(ok=True, event=payload) # v1-style body-token path (also the fallback for v2 when the # sender happens not to sign). Legacy setups without # event_encrypt_key also land here. expected_token = str(self.config.get("event_verification_token") or "") if expected_token: token = payload.get("token") or payload.get("header", {}).get("token") if not token or not hmac.compare_digest( expected_token.encode(), str(token).encode(), ): return WebhookVerificationResult( ok=False, error_message="Invalid Feishu webhook token", ) # Pre-launch hardening (Round-29 audit): MANDATORY # X-Lark-Request-Timestamp on the v1 body-token path. The # static ``event_verification_token`` alone is replayable # forever — an attacker who captured a single approval # callback could replay it indefinitely if we accepted # callbacks without a timestamp header. ±300s mirrors v2. # ``approval/router.py`` enforces the same window at the # HTTP layer; the driver-level enforcement here guarantees # any future caller (MCP, batch reprocessor, integration # test) inherits the same replay protection. ts_v1 = headers.get("x-lark-request-timestamp") or headers.get( "X-Lark-Request-Timestamp", "", ) if not ts_v1: return WebhookVerificationResult( ok=False, error_message=( "Missing X-Lark-Request-Timestamp header — " "v1 token-only path requires timestamp for " "replay protection" ), ) try: ts_int = int(ts_v1) except (ValueError, TypeError): return WebhookVerificationResult( ok=False, error_message="Invalid X-Lark-Request-Timestamp", ) if abs(time.time() - ts_int) > 300: return WebhookVerificationResult( ok=False, error_message=( "Request timestamp too old (replay protection)" ), ) # Round-49: SETNX nonce-replay cache mirroring v2 (v1 has # no nonce header so we hash the body itself). Without # this, a captured callback inside the ±300s window could # be replayed and amplify any conditionally-idempotent # state change (SOC2 group-add, MCP-routed handler). try: # #1653: was ``hashlib.sha256(content)`` — ``content`` is # only bound in the v2 branch above, so on the v1 path this # raised NameError EVERY time, was swallowed by the broad # ``except Exception`` below, and fell through to ok=True → # the entire v1 SETNX replay cache was dead code + fail-open. # The v1 raw body is the ``body`` param. body_hash = hashlib.sha256(body).hexdigest()[:32] from app.core.redis_client import get_redis # noqa: PLC0415 r = get_redis() if r is not None: key = f"feishu:webhook_replay_v1:{body_hash}:{ts_int}" set_result = await r.set(key, "1", nx=True, ex=600) if not set_result: return WebhookVerificationResult( ok=False, error_message=( "Replayed webhook (body-hash + " "timestamp already processed within " "±10min window)" ), ) except RedisError: # Fail-open ONLY on a genuine Redis fault (outage / timeout): # blocking every approval callback while Redis is down is a # worse outage than the bounded replay risk inside the ±300s # window. Narrowed from ``except Exception`` so a coding bug # (the NameError above) can never again masquerade as a # transient Redis fault and silently disable dedup. logger.warning( "feishu v1 webhook replay cache unavailable — " "accepting without dedup", ) return WebhookVerificationResult(ok=True, event=payload) # Neither v2 (encrypt_key + signature) NOR v1 (token) configured. # Refuse the callback — accepting would expose any # ChatApprovalDriver-routed handler as an unauthenticated # state-changing endpoint that can replay approvals. Production # today routes Feishu callbacks through ``approval/router.py`` # (which fails closed in ``verify_callback_token``), but the # driver's contract is documented as the canonical entry — any # future router / MCP path that uses it would inherit a wide- # open gate. Fail closed by default. logger.error( "feishu integration %s has no event_encrypt_key/X-Lark-" "Signature NOR event_verification_token configured — " "rejecting webhook to prevent open callback acceptance", self.integration_id, ) return WebhookVerificationResult( ok=False, error_message=( "Feishu webhook secret not configured: set " "event_encrypt_key (preferred) or event_verification_" "token on the integration before enabling callbacks." ), )