From 11faa8a789b4f546605df641af308f7756d7a3b9 Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Sat, 11 Jul 2026 08:51:33 +0800 Subject: [PATCH] refactor(connectors): decompose Slack and Feishu adapters --- .../platforms/feishu_delivery_handlers.py | 421 ++++++ .../platforms/feishu_ingress_handlers.py | 462 +++++++ .../platforms/feishu_installation_handlers.py | 198 +++ connector/platforms/feishu_webhook.py | 1124 ++--------------- .../platforms/slack_delivery_handlers.py | 345 +++++ connector/platforms/slack_ingress_handlers.py | 497 ++++++++ .../platforms/slack_installation_handlers.py | 171 +++ connector/platforms/slack_webhook.py | 1008 +-------------- scripts/verify_platform_adapter_contract.py | 158 +++ tests/exception_boundary_policy.json | 45 +- tests/platform_adapter_contract_r223.json | 170 +++ tests/static_analysis_policy.json | 80 +- tests/test_r219_exception_boundary_phase2.py | 5 +- ...est_r223_platform_adapter_decomposition.py | 97 ++ tests/test_s36s37r79_egress_hardening.py | 2 +- 15 files changed, 2711 insertions(+), 2072 deletions(-) create mode 100644 connector/platforms/feishu_delivery_handlers.py create mode 100644 connector/platforms/feishu_ingress_handlers.py create mode 100644 connector/platforms/feishu_installation_handlers.py create mode 100644 connector/platforms/slack_delivery_handlers.py create mode 100644 connector/platforms/slack_ingress_handlers.py create mode 100644 connector/platforms/slack_installation_handlers.py create mode 100644 scripts/verify_platform_adapter_contract.py create mode 100644 tests/platform_adapter_contract_r223.json create mode 100644 tests/test_r223_platform_adapter_decomposition.py diff --git a/connector/platforms/feishu_delivery_handlers.py b/connector/platforms/feishu_delivery_handlers.py new file mode 100644 index 0000000..5048711 --- /dev/null +++ b/connector/platforms/feishu_delivery_handlers.py @@ -0,0 +1,421 @@ +"""Owned Feishu card, response, and media-delivery mixin.""" + +# ruff: noqa: UP006, UP035, UP045 -- preserve frozen facade annotations. + +from __future__ import annotations + +import json +import secrets +from dataclasses import dataclass +from typing import Any, Dict, Optional + +from services.safe_io import STANDARD_OUTBOUND_POLICY, SafeIOHTTPError + +from ..reply_visibility import decide_reply_visibility +from .feishu_installation_manager import FeishuBinding + +# mypy: disable-error-code="attr-defined,no-any-return" + + +@dataclass +class FeishuDeliveryTarget: + channel_id: str + reply_to_message_id: str = "" + workspace_id: str = "" + account_id: str = "" + + +class FeishuDeliveryMixin: + def _build_card_button_value( + self, + button: Dict[str, Any], + *, + target: FeishuDeliveryTarget, + binding: FeishuBinding, + signing_secret: str, + ) -> Dict[str, Any]: + contract = self._callback_contract_for_binding( + binding=binding, + signing_secret=signing_secret, + ) + command_text = str(button.get("value", "") or "").strip() + callback_payload = { + "label": str(button.get("label", "") or "").strip(), + "command": command_text, + "approval_id": str(button.get("approval_id", "") or "").strip(), + "workspace_id": target.workspace_id or binding.workspace_id, + "account_id": target.account_id or binding.account_id, + "channel_id": target.channel_id, + "message_id": target.reply_to_message_id, + } + envelope = contract.build_envelope( + request_id=secrets.token_hex(12), + workspace_id=callback_payload["workspace_id"], + action_type=self._adapter_infer_callback_action_type(command_text, button), + payload=callback_payload, + ) + return { + "callback_envelope": dict(envelope.__dict__), + "payload": callback_payload, + } + + def _build_interactive_card( + self, + target: FeishuDeliveryTarget, + text: str, + buttons: list[dict], + *, + binding: FeishuBinding, + secrets: Dict[str, str], + ) -> Dict[str, Any]: + signing_secret = str( + secrets.get("app_secret", "") or binding.app_secret or "" + ).strip() + if not signing_secret: + raise RuntimeError("feishu_callback_signing_secret_missing") + actions = [] + for button in buttons[:6]: + command_text = str(button.get("value", "") or "").strip() + if not command_text: + continue + actions.append( + { + "tag": "button", + "type": str(button.get("style", "") or "default"), + "text": { + "tag": "plain_text", + "content": str(button.get("label", "") or "OpenClaw"), + }, + "value": self._build_card_button_value( + button, + target=target, + binding=binding, + signing_secret=signing_secret, + ), + } + ) + return { + "config": {"wide_screen_mode": True}, + "header": { + "template": "blue", + "title": {"tag": "plain_text", "content": "OpenClaw"}, + }, + "elements": [ + {"tag": "markdown", "content": text or "OpenClaw"}, + {"tag": "action", "actions": actions}, + ], + } + + async def _send_interactive_reply( + self, + target: FeishuDeliveryTarget, + text: str, + buttons: list[dict], + ) -> None: + resolution, binding, secrets = self._resolve_delivery_binding( + workspace_id=target.workspace_id, + account_id=target.account_id, + ) + if binding is None or not resolution.ok: + self._adapter_logger().warning( + "Feishu interactive reply dropped: no workspace binding available (%s / %s)", + target.workspace_id or "no-workspace", + target.account_id or "no-account", + ) + return + token = await self._get_tenant_access_token( + binding=binding, + workspace_id=target.workspace_id, + account_id=target.account_id, + ) + api_base = self._adapter_resolve_domain_base(binding.domain) + card = self._build_interactive_card( + target, + text, + buttons, + binding=binding, + secrets=secrets, + ) + payload = { + "content": json.dumps(card, ensure_ascii=False), + "msg_type": "interactive", + } + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json; charset=utf-8", + } + if target.reply_to_message_id: + url = ( + f"{api_base}/open-apis/im/v1/messages/" + f"{target.reply_to_message_id}/reply" + ) + else: + url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" + payload["receive_id"] = target.channel_id + try: + data = self._adapter_safe_request_json( + method="POST", + url=url, + json_body=payload, + headers=headers, + content_type="application/json; charset=utf-8", + timeout_sec=15, + allow_hosts=self._adapter_allowed_api_hosts(binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "interactive_reply"}, + ) + self._adapter_logger().warning( + "Feishu interactive reply failed: status=%s", exc.status_code + ) + return + if data.get("code", 0) != 0: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=str(data.get("msg", "unknown") or "unknown"), + status_code=200, + details={"phase": "interactive_reply"}, + ) + self._adapter_logger().warning( + "Feishu interactive reply failed: %s", data.get("msg", "unknown") + ) + + async def _send_reply( + self, + target: FeishuDeliveryTarget, + text: str, + *, + delivery_context: Optional[Dict[str, Any]] = None, + ) -> None: + ctx = dict(delivery_context or {}) + if target.workspace_id: + ctx.setdefault("workspace_id", target.workspace_id) + if target.account_id: + ctx.setdefault("account_id", target.account_id) + if target.reply_to_message_id: + ctx.setdefault("thread_id", target.reply_to_message_id) + decision = decide_reply_visibility( + delivery_context=ctx, + platform="feishu", + channel_kind=str(ctx.get("chat_type", "") or ""), + in_thread=bool(target.reply_to_message_id), + text=text, + ) + if decision.suppressed: + self._adapter_logger().info( + "Suppressed Feishu reply channel=%s reason=%s", + target.channel_id, + decision.reason, + ) + return + resolution, binding, _ = self._resolve_delivery_binding( + workspace_id=target.workspace_id, + account_id=target.account_id, + ) + if binding is None or not resolution.ok: + self._adapter_logger().warning( + "Feishu reply dropped: no workspace binding available (%s / %s)", + target.workspace_id or "no-workspace", + target.account_id or "no-account", + ) + return + token = await self._get_tenant_access_token( + binding=binding, + workspace_id=target.workspace_id, + account_id=target.account_id, + ) + api_base = self._adapter_resolve_domain_base(binding.domain) + payload = { + "content": json.dumps({"text": text}, ensure_ascii=False), + "msg_type": "text", + } + headers = { + "Authorization": f"Bearer {token}", + "Content-Type": "application/json; charset=utf-8", + } + if target.reply_to_message_id: + url = ( + f"{api_base}/open-apis/im/v1/messages/" + f"{target.reply_to_message_id}/reply" + ) + else: + url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" + payload["receive_id"] = target.channel_id + try: + data = self._adapter_safe_request_json( + method="POST", + url=url, + json_body=payload, + headers=headers, + content_type="application/json; charset=utf-8", + timeout_sec=15, + allow_hosts=self._adapter_allowed_api_hosts(binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "reply"}, + ) + self._adapter_logger().warning( + "Feishu reply failed: status=%s", exc.status_code + ) + return + if data.get("code", 0) != 0: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=str(data.get("msg", "unknown") or "unknown"), + status_code=200, + details={"phase": "reply"}, + ) + self._adapter_logger().warning( + "Feishu reply failed: %s", + data.get("msg", "unknown"), + ) + + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[Dict[str, Any]] = None, + ): + ctx = dict(delivery_context or {}) + await self._send_reply( + self._adapter_delivery_target( + channel_id=channel_id, + reply_to_message_id=str(ctx.get("thread_id", "") or "").strip(), + workspace_id=str(ctx.get("workspace_id", "") or "").strip(), + account_id=str(ctx.get("account_id", "") or "").strip(), + ), + text, + delivery_context=ctx, + ) + + async def send_image( + self, + channel_id: str, + image_data: bytes, + filename: str = "image.png", + caption: Optional[str] = None, + delivery_context: Optional[Dict[str, Any]] = None, + ): + ctx = dict(delivery_context or {}) + resolution, binding, _ = self._resolve_delivery_binding( + workspace_id=str(ctx.get("workspace_id", "") or "").strip(), + account_id=str(ctx.get("account_id", "") or "").strip(), + ) + if binding is None or not resolution.ok: + self._adapter_logger().warning( + "Feishu image dropped: no workspace binding available (%s / %s)", + str(ctx.get("workspace_id", "") or "").strip() or "no-workspace", + str(ctx.get("account_id", "") or "").strip() or "no-account", + ) + return + token = await self._get_tenant_access_token( + binding=binding, + workspace_id=str(ctx.get("workspace_id", "") or "").strip(), + account_id=str(ctx.get("account_id", "") or "").strip(), + ) + api_base = self._adapter_resolve_domain_base(binding.domain) + upload_headers = { + "Accept": "application/json", + "Authorization": f"Bearer {token}", + } + upload_body, upload_content_type = self._adapter_build_multipart_form( + fields={"image_type": "message"}, + file_field="image", + filename=filename, + file_bytes=image_data, + file_content_type="image/png", + ) + try: + upload_payload = self._adapter_safe_request_json( + method="POST", + url=f"{api_base}/open-apis/im/v1/images", + raw_body=upload_body, + headers=upload_headers, + content_type=upload_content_type, + timeout_sec=30, + allow_hosts=self._adapter_allowed_api_hosts(binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "image_upload"}, + ) + self._adapter_logger().warning( + "Feishu image upload failed: status=%s", exc.status_code + ) + return + image_key = str( + (upload_payload.get("data") or {}).get("image_key", "") or "" + ).strip() + if upload_payload.get("code", 0) != 0 or not image_key: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=str(upload_payload.get("msg", "unknown") or "unknown"), + status_code=200, + details={"phase": "image_upload"}, + ) + self._adapter_logger().warning( + "Feishu image upload failed: %s", + upload_payload.get("msg", "unknown"), + ) + return + message_payload = { + "content": json.dumps({"image_key": image_key}, ensure_ascii=False), + "msg_type": "image", + } + thread_id = str(ctx.get("thread_id", "") or "").strip() + if thread_id: + send_url = f"{api_base}/open-apis/im/v1/messages/{thread_id}/reply" + else: + send_url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" + message_payload["receive_id"] = channel_id + try: + self._adapter_safe_request_json( + method="POST", + url=send_url, + json_body=message_payload, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {token}", + }, + content_type="application/json; charset=utf-8", + timeout_sec=30, + allow_hosts=self._adapter_allowed_api_hosts(binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "image_send"}, + ) + self._adapter_logger().warning( + "Feishu image send failed: status=%s", exc.status_code + ) + if caption: + await self.send_message( + channel_id, + caption, + delivery_context=ctx, + ) diff --git a/connector/platforms/feishu_ingress_handlers.py b/connector/platforms/feishu_ingress_handlers.py new file mode 100644 index 0000000..ccf7c0e --- /dev/null +++ b/connector/platforms/feishu_ingress_handlers.py @@ -0,0 +1,462 @@ +"""Owned Feishu webhook ingress and callback transaction mixin.""" + +# ruff: noqa: UP006, UP035, UP045 -- preserve frozen facade annotations. + +from __future__ import annotations + +import json +import secrets +import time +from typing import Any, Dict, Optional, Tuple + +from services.connector_callback_contract import ( + CallbackActorContext, + CallbackDecisionCode, + ConnectorCallbackContract, +) + +from ..contract import CommandRequest +from .feishu_installation_manager import FeishuBinding + +# mypy: disable-error-code="attr-defined,index,no-any-return" + + +class FeishuIngressMixin: + async def handle_event(self, request): + _, web = self._adapter_import_aiohttp_web() + try: + body = await request.read() + except Exception: + return self._adapter_make_response(web, status=400, text="Bad request") + if len(body) > self._adapter_max_body_bytes(): + return self._adapter_make_response( + web, status=413, text="Payload too large" + ) + try: + payload = json.loads(body or b"{}") + except json.JSONDecodeError: + return self._adapter_make_response(web, status=400, text="Bad JSON") + if self._is_challenge(payload): + if not self._verify_request_token(payload): + return self._adapter_make_response( + web, status=401, text="Invalid verification token" + ) + return self._adapter_make_json_response( + web, {"challenge": str(payload.get("challenge", "") or "")} + ) + if not self._verify_request_token(payload): + return self._adapter_make_response( + web, status=401, text="Invalid verification token" + ) + try: + await self.process_event_payload(payload) + except ValueError as exc: + safe_code = self._adapter_safe_external_error_code("event_rejected", exc) + self._adapter_logger().warning("Feishu event rejected: %s", safe_code) + return self._adapter_make_response( + web, + status=400, + text=safe_code, + ) + return self._adapter_make_response(web, status=200, text="OK") + + async def handle_callback(self, request): + _, web = self._adapter_import_aiohttp_web() + try: + body = await request.read() + except Exception: + return self._adapter_make_response(web, status=400, text="Bad request") + if len(body) > self._adapter_max_body_bytes(): + return self._adapter_make_response( + web, status=413, text="Payload too large" + ) + try: + payload = json.loads(body or b"{}") + except json.JSONDecodeError: + return self._adapter_make_response(web, status=400, text="Bad JSON") + try: + response = await self.process_callback_payload(payload) + except ValueError as exc: + safe_code = self._adapter_safe_external_error_code("callback_rejected", exc) + self._adapter_logger().warning("Feishu callback rejected: %s", safe_code) + return self._adapter_make_json_response( + web, + { + "ok": False, + "error": safe_code, + }, + status=403, + ) + return self._adapter_make_json_response(web, response) + + def _is_challenge(self, payload: Dict[str, Any]) -> bool: + return bool( + payload.get("challenge") + and str(payload.get("type", "") or "").strip().lower() == "url_verification" + ) + + def _verify_request_token(self, payload: Dict[str, Any]) -> bool: + try: + self._resolve_inbound_binding(payload) + return True + except ValueError: + return False + + def _extract_callback_action(self, payload: Dict[str, Any]) -> Tuple[ + Dict[str, Any], + Dict[str, Any], + Dict[str, Any], + Dict[str, Any], + str, + str, + ]: + header = payload.get("header") or {} + event = payload.get("event") or {} + action = payload.get("action") or event.get("action") or {} + if not action and isinstance(event.get("actions"), list): + first_action = event.get("actions")[0] if event.get("actions") else {} + if isinstance(first_action, dict): + action = first_action + if not isinstance(action, dict): + raise ValueError("invalid_callback_action") + raw_value = action.get("value") or {} + if isinstance(raw_value, str): + raw_value = self._adapter_json_loads_safe(raw_value) + if not isinstance(raw_value, dict): + raise ValueError("invalid_callback_value") + envelope = raw_value.get("callback_envelope") or {} + callback_payload = raw_value.get("payload") or {} + if not isinstance(envelope, dict) or not isinstance(callback_payload, dict): + raise ValueError("invalid_callback_envelope") + workspace_id = str( + header.get("tenant_key") + or event.get("tenant_key") + or callback_payload.get("workspace_id") + or "" + ).strip() + account_id = str(callback_payload.get("account_id", "") or "").strip() + return header, event, envelope, callback_payload, workspace_id, account_id + + def _callback_contract_for_binding( + self, + *, + binding: FeishuBinding, + signing_secret: str, + ) -> ConnectorCallbackContract: + cache_key = self._cache_key_for_binding(binding) + if ( + self._callback_contracts.get(cache_key) is not None + and self._callback_contract_secrets.get(cache_key) == signing_secret + ): + return self._callback_contracts[cache_key] + contract = ConnectorCallbackContract( + signing_secret=signing_secret, + installation_registry=self._installation_manager.registry, + action_policy_map=self._adapter_callback_policy_map(), + ) + self._callback_contracts[cache_key] = contract + self._callback_contract_secrets[cache_key] = signing_secret + return contract + + def _actor_context_for_callback( + self, + *, + actor_id: str, + actor_open_id: str, + channel_id: str, + message_id: str, + workspace_id: str, + account_id: str, + command_text: str, + ) -> Tuple[CallbackActorContext, CommandRequest]: + request = CommandRequest( + platform="feishu", + sender_id=actor_id or actor_open_id, + channel_id=channel_id or actor_id or actor_open_id, + username=actor_id or actor_open_id, + message_id=message_id or f"cb-{secrets.token_hex(4)}", + text=command_text, + timestamp=time.time(), + workspace_id=workspace_id, + thread_id=message_id, + metadata={ + "account_id": account_id, + "sender_open_id": actor_open_id, + "interactive_callback": True, + }, + ) + actor = CallbackActorContext( + is_admin=self.router._is_admin(request.sender_id), + is_trusted=self.router._is_trusted(request), + user_id=request.sender_id, + tenant_id=workspace_id or request.workspace_id or "", + ) + return actor, request + + def _build_callback_response( + self, + *, + ok: bool, + text: str, + response_type: str = "info", + card: Optional[Dict[str, Any]] = None, + duplicate: bool = False, + decision_code: str = "", + ) -> Dict[str, Any]: + response = { + "ok": ok, + "duplicate": duplicate, + "decision_code": decision_code, + "toast": { + "type": response_type, + "content": text[:500] if text else "", + }, + } + if card is not None: + response["card"] = card + return response + + def _build_request( + self, + payload: Dict[str, Any], + *, + binding: FeishuBinding, + bot_open_id: str, + ) -> Optional[CommandRequest]: + header = payload.get("header") or {} + if ( + str(header.get("event_type", "") or "").strip() + not in self._adapter_supported_event_types() + ): + return None + event = payload.get("event") or {} + message = event.get("message") or {} + sender = event.get("sender") or {} + sender_id = sender.get("sender_id") or {} + mentions = self._adapter_normalize_mentions(message) + sender_user_id = str(sender_id.get("user_id", "") or "").strip() + sender_open_id = str(sender_id.get("open_id", "") or "").strip() + chat_id = str(message.get("chat_id", "") or "").strip() + chat_type = str(message.get("chat_type", "") or "").strip().lower() + message_id = str(message.get("message_id", "") or "").strip() + workspace_id = ( + str(header.get("tenant_key", "") or "").strip() or binding.workspace_id + ) + if not sender_user_id and not sender_open_id: + return None + if not chat_id or not message_id: + return None + if sender_open_id and bot_open_id and sender_open_id == bot_open_id: + return None + raw_text = self._adapter_parse_message_text(message) + if not raw_text: + return None + mentioned_bot = False + if bot_open_id: + for mention in mentions: + open_id = str(((mention.get("id") or {}).get("open_id")) or "").strip() + if open_id and open_id == bot_open_id: + mentioned_bot = True + break + text = self._adapter_strip_bot_mention(raw_text, mentions, bot_open_id) + if ( + chat_type == "group" + and self.config.feishu_require_mention + and not mentioned_bot + ): + return None + effective_sender = sender_user_id or sender_open_id + return CommandRequest( + platform="feishu", + sender_id=effective_sender, + channel_id=chat_id, + username=effective_sender, + message_id=message_id, + text=text, + timestamp=time.time(), + workspace_id=workspace_id, + thread_id=( + str(message.get("root_id", "") or "").strip() + or (message_id if self.config.feishu_reply_in_thread else "") + ), + metadata={ + "account_id": binding.account_id, + "chat_type": chat_type, + "mentioned_bot": mentioned_bot, + "message_type": str(message.get("message_type", "") or "").strip(), + "sender_open_id": sender_open_id, + }, + ) + + async def process_event_payload( + self, + payload: Dict[str, Any], + *, + binding: Optional[FeishuBinding] = None, + ) -> None: + header = payload.get("header") or {} + event_id = str(header.get("event_id", "") or "").strip() + if not event_id: + raise ValueError("Missing event_id") + if not self._replay_guard.check_and_record(event_id): + return + effective_binding = binding or self._resolve_inbound_binding(payload) + bot_open_id = self._cached_bot_open_id(effective_binding) + message = (payload.get("event") or {}).get("message") or {} + chat_type = str(message.get("chat_type", "") or "").strip().lower() + if not bot_open_id and chat_type == "group": + bot_open_id = await self._fetch_bot_open_id( + binding=effective_binding, allow_degrade=True + ) + request = self._build_request( + payload, + binding=effective_binding, + bot_open_id=bot_open_id, + ) + if request is None: + return + if self._user_allowlist.entries: + user_result = self._user_allowlist.evaluate(str(request.sender_id)) + if user_result.decision == "deny": + return + if self._chat_allowlist.entries: + chat_result = self._chat_allowlist.evaluate(str(request.channel_id)) + if chat_result.decision == "deny": + return + response = await self.router.handle(request) + resp_text = str(getattr(response, "text", "") or "").strip() + buttons = getattr(response, "buttons", []) or [] + target = self._adapter_delivery_target( + channel_id=request.channel_id, + reply_to_message_id=request.thread_id, + workspace_id=request.workspace_id, + account_id=str(request.metadata.get("account_id", "") or ""), + ) + if buttons: + await self._send_interactive_reply(target, resp_text, buttons) + elif resp_text: + await self._send_reply( + target, + resp_text, + delivery_context={ + "workspace_id": request.workspace_id, + "thread_id": request.thread_id, + "account_id": str(request.metadata.get("account_id", "") or ""), + "chat_type": str(request.metadata.get("chat_type", "") or ""), + "mentioned_bot": bool(request.metadata.get("mentioned_bot")), + }, + ) + + async def process_callback_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: + _, _, envelope_dict, callback_payload, workspace_id, account_id = ( + self._extract_callback_action(payload) + ) + resolution, binding, secrets = self._resolve_delivery_binding( + workspace_id=workspace_id, + account_id=account_id, + ) + if binding is None or not resolution.ok: + raise ValueError(resolution.reject_reason or "missing_binding") + signing_secret = str( + secrets.get("app_secret", "") or binding.app_secret or "" + ).strip() + if not signing_secret: + raise ValueError("missing_callback_signing_secret") + contract = self._callback_contract_for_binding( + binding=binding, + signing_secret=signing_secret, + ) + event = payload.get("event") or {} + operator = payload.get("operator") or event.get("operator") or {} + operator_id = operator.get("operator_id") or operator.get("sender_id") or {} + actor_id = str( + operator.get("user_id") + or operator_id.get("user_id") + or callback_payload.get("actor_user_id") + or "" + ).strip() + actor_open_id = str( + operator.get("open_id") + or operator_id.get("open_id") + or callback_payload.get("actor_open_id") + or "" + ).strip() + command_text = str(callback_payload.get("command", "") or "").strip() + actor, request = self._actor_context_for_callback( + actor_id=actor_id, + actor_open_id=actor_open_id, + channel_id=str( + payload.get("open_chat_id") + or event.get("open_chat_id") + or callback_payload.get("channel_id") + or "" + ).strip(), + message_id=str( + payload.get("open_message_id") + or event.get("open_message_id") + or callback_payload.get("message_id") + or "" + ).strip(), + workspace_id=workspace_id or binding.workspace_id, + account_id=binding.account_id, + command_text=command_text, + ) + decision = contract.evaluate( + platform="feishu", + envelope_dict=envelope_dict, + payload=callback_payload, + actor=actor, + ) + if decision.decision_code == CallbackDecisionCode.REJECT_REPLAY.value: + return self._build_callback_response( + ok=True, + text="Action already processed.", + response_type="info", + duplicate=True, + decision_code=decision.decision_code, + ) + if not decision.ok and not decision.requires_approval: + raise ValueError(decision.message or decision.decision_code) + request.text = ( + self._adapter_force_approval_command(request.text) + if decision.requires_approval + else request.text + ) + request_id = str(envelope_dict.get("request_id", "") or "") + contract.acknowledge_request(request_id) + try: + response = await self.router.handle(request) + except Exception: + # IMPORTANT: failures before route completion remain retryable. + # After router.handle returns, the action may already have side effects, + # so completion failures must not release the claim for rerouting. + contract.release_request_retryable( + request_id, reason="feishu_callback_failed_before_commit" + ) + raise + contract.complete_request(request_id) + response_text = str(getattr(response, "text", "") or "").strip() or ( + "Action processed." + ) + response_buttons = getattr(response, "buttons", []) or [] + card = None + if response_buttons: + card = self._build_interactive_card( + self._adapter_delivery_target( + channel_id=request.channel_id, + reply_to_message_id=request.thread_id, + workspace_id=request.workspace_id, + account_id=binding.account_id, + ), + response_text, + response_buttons, + binding=binding, + secrets=secrets, + ) + return self._build_callback_response( + ok=True, + text=response_text, + response_type="success", + card=card, + decision_code=decision.decision_code, + ) diff --git a/connector/platforms/feishu_installation_handlers.py b/connector/platforms/feishu_installation_handlers.py new file mode 100644 index 0000000..57281a0 --- /dev/null +++ b/connector/platforms/feishu_installation_handlers.py @@ -0,0 +1,198 @@ +"""Owned Feishu installation, tenant-token, and bot-identity mixin.""" + +# ruff: noqa: UP006, UP035, UP045 -- preserve frozen facade annotations. + +from __future__ import annotations + +import time +from typing import Any, Dict, Optional, Tuple + +from services.connector_installation_registry import InstallationResolution +from services.safe_io import STANDARD_OUTBOUND_POLICY, SafeIOHTTPError + +from .feishu_installation_manager import FeishuBinding + +# mypy: disable-error-code="attr-defined,no-any-return" + + +class FeishuInstallationMixin: + def _resolve_inbound_binding(self, payload: Dict[str, Any]) -> FeishuBinding: + header = payload.get("header") or {} + verification_token = ( + str(payload.get("token", "") or "").strip() + or str(header.get("token", "") or "").strip() + or str(((payload.get("event") or {}).get("token")) or "").strip() + ) + workspace_id = str(header.get("tenant_key", "") or "").strip() + return self._installation_manager.resolve_inbound_binding( + verification_token=verification_token, + workspace_id=workspace_id, + account_id=self._bound_account_id, + ) + + def _cache_key_for_binding(self, binding: FeishuBinding) -> str: + return binding.installation_id or binding.account_id + + def _cached_bot_open_id(self, binding: FeishuBinding) -> str: + return ( + self._bot_open_ids.get(self._cache_key_for_binding(binding), "") + or self._bot_open_id + ) + + def _resolve_delivery_binding( + self, *, workspace_id: str = "", account_id: str = "" + ) -> Tuple[InstallationResolution, Optional[FeishuBinding], Dict[str, str]]: + return self._installation_manager.resolve_binding( + workspace_id=workspace_id, + account_id=account_id or self._bound_account_id, + ) + + async def _get_tenant_access_token( + self, + *, + binding: Optional[FeishuBinding] = None, + workspace_id: str = "", + account_id: str = "", + ) -> str: + resolution, effective_binding, secrets = self._resolve_delivery_binding( + workspace_id=workspace_id, + account_id=account_id or (binding.account_id if binding else ""), + ) + if effective_binding is None or not resolution.ok: + raise RuntimeError( + f"feishu_binding_resolution_failed:{resolution.reject_reason or 'missing_binding'}" + ) + cache_key = self._cache_key_for_binding(effective_binding) + if self._tenant_access_tokens.get( + cache_key + ) and self._tenant_access_token_expires_at.get(cache_key, 0.0) > ( + time.time() + 30 + ): + return self._tenant_access_tokens[cache_key] + app_secret = str( + secrets.get("app_secret", "") or effective_binding.app_secret + ).strip() + payload = { + "app_id": effective_binding.app_id, + "app_secret": app_secret, + } + url = f"{self._adapter_resolve_domain_base(effective_binding.domain)}/open-apis/auth/v3/tenant_access_token/internal" + try: + data = self._adapter_safe_request_json( + method="POST", + url=url, + json_body=payload, + headers={"Accept": "application/json"}, + content_type="application/json; charset=utf-8", + timeout_sec=15, + allow_hosts=self._adapter_allowed_api_hosts(effective_binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "tenant_access_token"}, + ) + raise RuntimeError( + f"feishu_token_fetch_failed:{exc.status_code}:{exc.reason}" + ) from exc + if data.get("code", 0) != 0: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=str(data.get("msg", "unknown") or "unknown"), + status_code=200, + details={"phase": "tenant_access_token"}, + ) + raise RuntimeError( + f"feishu_token_fetch_failed:200:{data.get('msg', 'unknown')}" + ) + token = str(data.get("tenant_access_token", "") or "").strip() + if not token: + raise RuntimeError("feishu_token_fetch_failed:missing_token") + expire = int( + data.get("expire", self._adapter_token_ttl_sec()) + or self._adapter_token_ttl_sec() + ) + self._tenant_access_tokens[cache_key] = token + self._tenant_access_token_expires_at[cache_key] = time.time() + max(60, expire) + if resolution.installation is not None: + self._installation_manager.mark_resolution_success( + resolution.installation.installation_id, + effective_binding.workspace_id, + ) + return token + + async def _fetch_bot_open_id( + self, + *, + binding: Optional[FeishuBinding] = None, + workspace_id: str = "", + account_id: str = "", + allow_degrade: bool = False, + ) -> str: + resolution, effective_binding, _ = self._resolve_delivery_binding( + workspace_id=workspace_id, + account_id=account_id or (binding.account_id if binding else ""), + ) + if effective_binding is None or not resolution.ok: + return "" + cache_key = self._cache_key_for_binding(effective_binding) + if self._bot_open_ids.get(cache_key): + return self._bot_open_ids[cache_key] + token = await self._get_tenant_access_token(binding=effective_binding) + url = f"{self._adapter_resolve_domain_base(effective_binding.domain)}/open-apis/bot/v3/info" + try: + data = self._adapter_safe_request_json( + method="GET", + url=url, + headers={ + "Accept": "application/json", + "Authorization": f"Bearer {token}", + }, + timeout_sec=15, + allow_hosts=self._adapter_allowed_api_hosts(effective_binding.domain), + policy=STANDARD_OUTBOUND_POLICY, + ) + except SafeIOHTTPError as exc: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=exc.reason, + status_code=exc.status_code, + details={"phase": "bot_info"}, + ) + if allow_degrade: + return "" + return "" + if data.get("code", 0) != 0: + if resolution.installation is not None: + self._installation_manager.mark_api_error( + resolution.installation.installation_id, + error_code=str(data.get("msg", "unknown") or "unknown"), + status_code=200, + details={"phase": "bot_info"}, + ) + return "" + bot_open_id = str( + (((data.get("data") or {}).get("bot") or {}).get("open_id")) or "" + ).strip() + if bot_open_id: + self._bot_open_ids[cache_key] = bot_open_id + self._bot_open_id = bot_open_id + return bot_open_id + + async def prime_bot_identity(self) -> None: + try: + await self._fetch_bot_open_id( + account_id=self._bound_account_id + or str(self.config.feishu_account_id or "").strip() + or str(self.config.feishu_default_account_id or "").strip(), + workspace_id=str(self.config.feishu_workspace_id or "").strip(), + allow_degrade=True, + ) + except Exception as exc: + self._adapter_logger().debug("Feishu bot identity fetch failed: %s", exc) diff --git a/connector/platforms/feishu_webhook.py b/connector/platforms/feishu_webhook.py index ca1607a..a593e54 100644 --- a/connector/platforms/feishu_webhook.py +++ b/connector/platforms/feishu_webhook.py @@ -14,45 +14,29 @@ Notes: from __future__ import annotations -import asyncio import json import logging import secrets -import time -from dataclasses import dataclass from typing import Any, Dict, Optional, Tuple from urllib.parse import urlparse from ..config import ConnectorConfig -from ..contract import CommandRequest -from ..reply_visibility import decide_reply_visibility from ..router import CommandRouter from ..security_profile import AllowlistPolicy, ReplayGuard -from .feishu_installation_manager import FeishuBinding, FeishuInstallationManager +from .feishu_delivery_handlers import FeishuDeliveryMixin, FeishuDeliveryTarget +from .feishu_ingress_handlers import FeishuIngressMixin +from .feishu_installation_handlers import FeishuInstallationMixin +from .feishu_installation_manager import FeishuInstallationManager try: - from services.safe_io import ( - STANDARD_OUTBOUND_POLICY, - SafeIOHTTPError, - safe_request_json, - ) + from services.safe_io import safe_request_json except ImportError: # pragma: no cover - from services.safe_io import ( # type: ignore - STANDARD_OUTBOUND_POLICY, - SafeIOHTTPError, - safe_request_json, - ) + from services.safe_io import safe_request_json # type: ignore try: - from services.connector_callback_contract import ( - CallbackActorContext, - CallbackDecisionCode, - ConnectorCallbackContract, - ) + from services.connector_callback_contract import ConnectorCallbackContract except ImportError: # pragma: no cover from services.connector_callback_contract import ( # type: ignore - CallbackActorContext, - CallbackDecisionCode, ConnectorCallbackContract, ) @@ -264,15 +248,11 @@ def _force_approval_command(command_text: str) -> str: return normalized -@dataclass -class FeishuDeliveryTarget: - channel_id: str - reply_to_message_id: str = "" - workspace_id: str = "" - account_id: str = "" - - -class FeishuWebhookServer: +class FeishuWebhookServer( + FeishuInstallationMixin, + FeishuIngressMixin, + FeishuDeliveryMixin, +): REPLAY_WINDOW_SEC = 300 NONCE_CACHE_SIZE = 5000 @@ -310,6 +290,87 @@ class FeishuWebhookServer: self._callback_contracts: Dict[str, ConnectorCallbackContract] = {} self._callback_contract_secrets: Dict[str, str] = {} + # IMPORTANT: keep facade patch seams live across extracted protocol owners. + @staticmethod + def _adapter_import_aiohttp_web(): + return _import_aiohttp_web() + + @staticmethod + def _adapter_make_response(*args, **kwargs): + return _make_response(*args, **kwargs) + + @staticmethod + def _adapter_make_json_response(*args, **kwargs): + return _make_json_response(*args, **kwargs) + + @staticmethod + def _adapter_safe_external_error_code(*args, **kwargs): + return _safe_external_error_code(*args, **kwargs) + + @staticmethod + def _adapter_resolve_domain_base(*args, **kwargs): + return _resolve_domain_base(*args, **kwargs) + + @staticmethod + def _adapter_allowed_api_hosts(*args, **kwargs): + return _allowed_api_hosts(*args, **kwargs) + + @staticmethod + def _adapter_build_multipart_form(*args, **kwargs): + return _build_multipart_form(*args, **kwargs) + + @staticmethod + def _adapter_json_loads_safe(*args, **kwargs): + return _json_loads_safe(*args, **kwargs) + + @staticmethod + def _adapter_normalize_mentions(*args, **kwargs): + return _normalize_mentions(*args, **kwargs) + + @staticmethod + def _adapter_strip_bot_mention(*args, **kwargs): + return _strip_bot_mention(*args, **kwargs) + + @staticmethod + def _adapter_parse_message_text(*args, **kwargs): + return parse_feishu_message_text(*args, **kwargs) + + @staticmethod + def _adapter_infer_callback_action_type(*args, **kwargs): + return _infer_callback_action_type(*args, **kwargs) + + @staticmethod + def _adapter_force_approval_command(*args, **kwargs): + return _force_approval_command(*args, **kwargs) + + @staticmethod + def _adapter_safe_request_json(*args, **kwargs): + return safe_request_json(*args, **kwargs) + + @staticmethod + def _adapter_max_body_bytes(): + return FEISHU_WEBHOOK_MAX_BODY_BYTES + + @staticmethod + def _adapter_token_ttl_sec(): + return FEISHU_TOKEN_TTL_SEC + + @staticmethod + def _adapter_callback_policy_map(): + return _FEISHU_CALLBACK_POLICY_MAP + + @staticmethod + def _adapter_supported_event_types(): + return _SUPPORTED_EVENT_TYPES + + @staticmethod + def _adapter_delivery_target(*args, **kwargs): + return FeishuDeliveryTarget(*args, **kwargs) + + @staticmethod + def _adapter_logger(): + return logger + async def start(self): aiohttp, web = _import_aiohttp_web() if aiohttp is None or web is None: @@ -361,1002 +422,3 @@ class FeishuWebhookServer: await self.site.stop() if self.runner: await self.runner.cleanup() - - async def handle_event(self, request): - _, web = _import_aiohttp_web() - try: - body = await request.read() - except Exception: - return _make_response(web, status=400, text="Bad request") - if len(body) > FEISHU_WEBHOOK_MAX_BODY_BYTES: - return _make_response(web, status=413, text="Payload too large") - try: - payload = json.loads(body or b"{}") - except json.JSONDecodeError: - return _make_response(web, status=400, text="Bad JSON") - if self._is_challenge(payload): - if not self._verify_request_token(payload): - return _make_response( - web, status=401, text="Invalid verification token" - ) - return _make_json_response( - web, {"challenge": str(payload.get("challenge", "") or "")} - ) - if not self._verify_request_token(payload): - return _make_response(web, status=401, text="Invalid verification token") - try: - await self.process_event_payload(payload) - except ValueError as exc: - safe_code = _safe_external_error_code("event_rejected", exc) - logger.warning("Feishu event rejected: %s", safe_code) - return _make_response( - web, - status=400, - text=safe_code, - ) - return _make_response(web, status=200, text="OK") - - async def handle_callback(self, request): - _, web = _import_aiohttp_web() - try: - body = await request.read() - except Exception: - return _make_response(web, status=400, text="Bad request") - if len(body) > FEISHU_WEBHOOK_MAX_BODY_BYTES: - return _make_response(web, status=413, text="Payload too large") - try: - payload = json.loads(body or b"{}") - except json.JSONDecodeError: - return _make_response(web, status=400, text="Bad JSON") - try: - response = await self.process_callback_payload(payload) - except ValueError as exc: - safe_code = _safe_external_error_code("callback_rejected", exc) - logger.warning("Feishu callback rejected: %s", safe_code) - return _make_json_response( - web, - { - "ok": False, - "error": safe_code, - }, - status=403, - ) - return _make_json_response(web, response) - - def _is_challenge(self, payload: Dict[str, Any]) -> bool: - return bool( - payload.get("challenge") - and str(payload.get("type", "") or "").strip().lower() == "url_verification" - ) - - def _verify_request_token(self, payload: Dict[str, Any]) -> bool: - try: - self._resolve_inbound_binding(payload) - return True - except ValueError: - return False - - def _extract_callback_action(self, payload: Dict[str, Any]) -> Tuple[ - Dict[str, Any], - Dict[str, Any], - Dict[str, Any], - Dict[str, Any], - str, - str, - ]: - header = payload.get("header") or {} - event = payload.get("event") or {} - action = payload.get("action") or event.get("action") or {} - if not action and isinstance(event.get("actions"), list): - first_action = event.get("actions")[0] if event.get("actions") else {} - if isinstance(first_action, dict): - action = first_action - if not isinstance(action, dict): - raise ValueError("invalid_callback_action") - raw_value = action.get("value") or {} - if isinstance(raw_value, str): - raw_value = _json_loads_safe(raw_value) - if not isinstance(raw_value, dict): - raise ValueError("invalid_callback_value") - envelope = raw_value.get("callback_envelope") or {} - callback_payload = raw_value.get("payload") or {} - if not isinstance(envelope, dict) or not isinstance(callback_payload, dict): - raise ValueError("invalid_callback_envelope") - workspace_id = str( - header.get("tenant_key") - or event.get("tenant_key") - or callback_payload.get("workspace_id") - or "" - ).strip() - account_id = str(callback_payload.get("account_id", "") or "").strip() - return header, event, envelope, callback_payload, workspace_id, account_id - - def _callback_contract_for_binding( - self, - *, - binding: FeishuBinding, - signing_secret: str, - ) -> ConnectorCallbackContract: - cache_key = self._cache_key_for_binding(binding) - if ( - self._callback_contracts.get(cache_key) is not None - and self._callback_contract_secrets.get(cache_key) == signing_secret - ): - return self._callback_contracts[cache_key] - contract = ConnectorCallbackContract( - signing_secret=signing_secret, - installation_registry=self._installation_manager.registry, - action_policy_map=_FEISHU_CALLBACK_POLICY_MAP, - ) - self._callback_contracts[cache_key] = contract - self._callback_contract_secrets[cache_key] = signing_secret - return contract - - def _actor_context_for_callback( - self, - *, - actor_id: str, - actor_open_id: str, - channel_id: str, - message_id: str, - workspace_id: str, - account_id: str, - command_text: str, - ) -> Tuple[CallbackActorContext, CommandRequest]: - request = CommandRequest( - platform="feishu", - sender_id=actor_id or actor_open_id, - channel_id=channel_id or actor_id or actor_open_id, - username=actor_id or actor_open_id, - message_id=message_id or f"cb-{secrets.token_hex(4)}", - text=command_text, - timestamp=time.time(), - workspace_id=workspace_id, - thread_id=message_id, - metadata={ - "account_id": account_id, - "sender_open_id": actor_open_id, - "interactive_callback": True, - }, - ) - actor = CallbackActorContext( - is_admin=self.router._is_admin(request.sender_id), - is_trusted=self.router._is_trusted(request), - user_id=request.sender_id, - tenant_id=workspace_id or request.workspace_id or "", - ) - return actor, request - - def _build_callback_response( - self, - *, - ok: bool, - text: str, - response_type: str = "info", - card: Optional[Dict[str, Any]] = None, - duplicate: bool = False, - decision_code: str = "", - ) -> Dict[str, Any]: - response = { - "ok": ok, - "duplicate": duplicate, - "decision_code": decision_code, - "toast": { - "type": response_type, - "content": text[:500] if text else "", - }, - } - if card is not None: - response["card"] = card - return response - - def _resolve_inbound_binding(self, payload: Dict[str, Any]) -> FeishuBinding: - header = payload.get("header") or {} - verification_token = ( - str(payload.get("token", "") or "").strip() - or str(header.get("token", "") or "").strip() - or str(((payload.get("event") or {}).get("token")) or "").strip() - ) - workspace_id = str(header.get("tenant_key", "") or "").strip() - return self._installation_manager.resolve_inbound_binding( - verification_token=verification_token, - workspace_id=workspace_id, - account_id=self._bound_account_id, - ) - - def _cache_key_for_binding(self, binding: FeishuBinding) -> str: - return binding.installation_id or binding.account_id - - def _cached_bot_open_id(self, binding: FeishuBinding) -> str: - return ( - self._bot_open_ids.get(self._cache_key_for_binding(binding), "") - or self._bot_open_id - ) - - def _build_request( - self, - payload: Dict[str, Any], - *, - binding: FeishuBinding, - bot_open_id: str, - ) -> Optional[CommandRequest]: - header = payload.get("header") or {} - if ( - str(header.get("event_type", "") or "").strip() - not in _SUPPORTED_EVENT_TYPES - ): - return None - event = payload.get("event") or {} - message = event.get("message") or {} - sender = event.get("sender") or {} - sender_id = sender.get("sender_id") or {} - mentions = _normalize_mentions(message) - sender_user_id = str(sender_id.get("user_id", "") or "").strip() - sender_open_id = str(sender_id.get("open_id", "") or "").strip() - chat_id = str(message.get("chat_id", "") or "").strip() - chat_type = str(message.get("chat_type", "") or "").strip().lower() - message_id = str(message.get("message_id", "") or "").strip() - workspace_id = ( - str(header.get("tenant_key", "") or "").strip() or binding.workspace_id - ) - if not sender_user_id and not sender_open_id: - return None - if not chat_id or not message_id: - return None - if sender_open_id and bot_open_id and sender_open_id == bot_open_id: - return None - raw_text = parse_feishu_message_text(message) - if not raw_text: - return None - mentioned_bot = False - if bot_open_id: - for mention in mentions: - open_id = str(((mention.get("id") or {}).get("open_id")) or "").strip() - if open_id and open_id == bot_open_id: - mentioned_bot = True - break - text = _strip_bot_mention(raw_text, mentions, bot_open_id) - if ( - chat_type == "group" - and self.config.feishu_require_mention - and not mentioned_bot - ): - return None - effective_sender = sender_user_id or sender_open_id - return CommandRequest( - platform="feishu", - sender_id=effective_sender, - channel_id=chat_id, - username=effective_sender, - message_id=message_id, - text=text, - timestamp=time.time(), - workspace_id=workspace_id, - thread_id=( - str(message.get("root_id", "") or "").strip() - or (message_id if self.config.feishu_reply_in_thread else "") - ), - metadata={ - "account_id": binding.account_id, - "chat_type": chat_type, - "mentioned_bot": mentioned_bot, - "message_type": str(message.get("message_type", "") or "").strip(), - "sender_open_id": sender_open_id, - }, - ) - - async def process_event_payload( - self, - payload: Dict[str, Any], - *, - binding: Optional[FeishuBinding] = None, - ) -> None: - header = payload.get("header") or {} - event_id = str(header.get("event_id", "") or "").strip() - if not event_id: - raise ValueError("Missing event_id") - if not self._replay_guard.check_and_record(event_id): - return - effective_binding = binding or self._resolve_inbound_binding(payload) - bot_open_id = self._cached_bot_open_id(effective_binding) - message = (payload.get("event") or {}).get("message") or {} - chat_type = str(message.get("chat_type", "") or "").strip().lower() - if not bot_open_id and chat_type == "group": - bot_open_id = await self._fetch_bot_open_id( - binding=effective_binding, allow_degrade=True - ) - request = self._build_request( - payload, - binding=effective_binding, - bot_open_id=bot_open_id, - ) - if request is None: - return - if self._user_allowlist.entries: - user_result = self._user_allowlist.evaluate(str(request.sender_id)) - if user_result.decision == "deny": - return - if self._chat_allowlist.entries: - chat_result = self._chat_allowlist.evaluate(str(request.channel_id)) - if chat_result.decision == "deny": - return - response = await self.router.handle(request) - resp_text = str(getattr(response, "text", "") or "").strip() - buttons = getattr(response, "buttons", []) or [] - target = FeishuDeliveryTarget( - channel_id=request.channel_id, - reply_to_message_id=request.thread_id, - workspace_id=request.workspace_id, - account_id=str(request.metadata.get("account_id", "") or ""), - ) - if buttons: - await self._send_interactive_reply(target, resp_text, buttons) - elif resp_text: - await self._send_reply( - target, - resp_text, - delivery_context={ - "workspace_id": request.workspace_id, - "thread_id": request.thread_id, - "account_id": str(request.metadata.get("account_id", "") or ""), - "chat_type": str(request.metadata.get("chat_type", "") or ""), - "mentioned_bot": bool(request.metadata.get("mentioned_bot")), - }, - ) - - def _resolve_delivery_binding( - self, *, workspace_id: str = "", account_id: str = "" - ) -> Tuple[InstallationResolution, Optional[FeishuBinding], Dict[str, str]]: - return self._installation_manager.resolve_binding( - workspace_id=workspace_id, - account_id=account_id or self._bound_account_id, - ) - - async def process_callback_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: - _, _, envelope_dict, callback_payload, workspace_id, account_id = ( - self._extract_callback_action(payload) - ) - resolution, binding, secrets = self._resolve_delivery_binding( - workspace_id=workspace_id, - account_id=account_id, - ) - if binding is None or not resolution.ok: - raise ValueError(resolution.reject_reason or "missing_binding") - signing_secret = str( - secrets.get("app_secret", "") or binding.app_secret or "" - ).strip() - if not signing_secret: - raise ValueError("missing_callback_signing_secret") - contract = self._callback_contract_for_binding( - binding=binding, - signing_secret=signing_secret, - ) - event = payload.get("event") or {} - operator = payload.get("operator") or event.get("operator") or {} - operator_id = operator.get("operator_id") or operator.get("sender_id") or {} - actor_id = str( - operator.get("user_id") - or operator_id.get("user_id") - or callback_payload.get("actor_user_id") - or "" - ).strip() - actor_open_id = str( - operator.get("open_id") - or operator_id.get("open_id") - or callback_payload.get("actor_open_id") - or "" - ).strip() - command_text = str(callback_payload.get("command", "") or "").strip() - actor, request = self._actor_context_for_callback( - actor_id=actor_id, - actor_open_id=actor_open_id, - channel_id=str( - payload.get("open_chat_id") - or event.get("open_chat_id") - or callback_payload.get("channel_id") - or "" - ).strip(), - message_id=str( - payload.get("open_message_id") - or event.get("open_message_id") - or callback_payload.get("message_id") - or "" - ).strip(), - workspace_id=workspace_id or binding.workspace_id, - account_id=binding.account_id, - command_text=command_text, - ) - decision = contract.evaluate( - platform="feishu", - envelope_dict=envelope_dict, - payload=callback_payload, - actor=actor, - ) - if decision.decision_code == CallbackDecisionCode.REJECT_REPLAY.value: - return self._build_callback_response( - ok=True, - text="Action already processed.", - response_type="info", - duplicate=True, - decision_code=decision.decision_code, - ) - if not decision.ok and not decision.requires_approval: - raise ValueError(decision.message or decision.decision_code) - request.text = ( - _force_approval_command(request.text) - if decision.requires_approval - else request.text - ) - request_id = str(envelope_dict.get("request_id", "") or "") - contract.acknowledge_request(request_id) - try: - response = await self.router.handle(request) - except Exception: - # IMPORTANT: failures before route completion remain retryable. - # After router.handle returns, the action may already have side effects, - # so completion failures must not release the claim for rerouting. - contract.release_request_retryable( - request_id, reason="feishu_callback_failed_before_commit" - ) - raise - contract.complete_request(request_id) - response_text = str(getattr(response, "text", "") or "").strip() or ( - "Action processed." - ) - response_buttons = getattr(response, "buttons", []) or [] - card = None - if response_buttons: - card = self._build_interactive_card( - FeishuDeliveryTarget( - channel_id=request.channel_id, - reply_to_message_id=request.thread_id, - workspace_id=request.workspace_id, - account_id=binding.account_id, - ), - response_text, - response_buttons, - binding=binding, - secrets=secrets, - ) - return self._build_callback_response( - ok=True, - text=response_text, - response_type="success", - card=card, - decision_code=decision.decision_code, - ) - - async def _get_tenant_access_token( - self, - *, - binding: Optional[FeishuBinding] = None, - workspace_id: str = "", - account_id: str = "", - ) -> str: - resolution, effective_binding, secrets = self._resolve_delivery_binding( - workspace_id=workspace_id, - account_id=account_id or (binding.account_id if binding else ""), - ) - if effective_binding is None or not resolution.ok: - raise RuntimeError( - f"feishu_binding_resolution_failed:{resolution.reject_reason or 'missing_binding'}" - ) - cache_key = self._cache_key_for_binding(effective_binding) - if self._tenant_access_tokens.get( - cache_key - ) and self._tenant_access_token_expires_at.get(cache_key, 0.0) > ( - time.time() + 30 - ): - return self._tenant_access_tokens[cache_key] - app_secret = str( - secrets.get("app_secret", "") or effective_binding.app_secret - ).strip() - payload = { - "app_id": effective_binding.app_id, - "app_secret": app_secret, - } - url = f"{_resolve_domain_base(effective_binding.domain)}/open-apis/auth/v3/tenant_access_token/internal" - try: - data = safe_request_json( - method="POST", - url=url, - json_body=payload, - headers={"Accept": "application/json"}, - content_type="application/json; charset=utf-8", - timeout_sec=15, - allow_hosts=_allowed_api_hosts(effective_binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "tenant_access_token"}, - ) - raise RuntimeError( - f"feishu_token_fetch_failed:{exc.status_code}:{exc.reason}" - ) from exc - if data.get("code", 0) != 0: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=str(data.get("msg", "unknown") or "unknown"), - status_code=200, - details={"phase": "tenant_access_token"}, - ) - raise RuntimeError( - f"feishu_token_fetch_failed:200:{data.get('msg', 'unknown')}" - ) - token = str(data.get("tenant_access_token", "") or "").strip() - if not token: - raise RuntimeError("feishu_token_fetch_failed:missing_token") - expire = int(data.get("expire", FEISHU_TOKEN_TTL_SEC) or FEISHU_TOKEN_TTL_SEC) - self._tenant_access_tokens[cache_key] = token - self._tenant_access_token_expires_at[cache_key] = time.time() + max(60, expire) - if resolution.installation is not None: - self._installation_manager.mark_resolution_success( - resolution.installation.installation_id, - effective_binding.workspace_id, - ) - return token - - async def _fetch_bot_open_id( - self, - *, - binding: Optional[FeishuBinding] = None, - workspace_id: str = "", - account_id: str = "", - allow_degrade: bool = False, - ) -> str: - resolution, effective_binding, _ = self._resolve_delivery_binding( - workspace_id=workspace_id, - account_id=account_id or (binding.account_id if binding else ""), - ) - if effective_binding is None or not resolution.ok: - return "" - cache_key = self._cache_key_for_binding(effective_binding) - if self._bot_open_ids.get(cache_key): - return self._bot_open_ids[cache_key] - token = await self._get_tenant_access_token(binding=effective_binding) - url = f"{_resolve_domain_base(effective_binding.domain)}/open-apis/bot/v3/info" - try: - data = safe_request_json( - method="GET", - url=url, - headers={ - "Accept": "application/json", - "Authorization": f"Bearer {token}", - }, - timeout_sec=15, - allow_hosts=_allowed_api_hosts(effective_binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "bot_info"}, - ) - if allow_degrade: - return "" - return "" - if data.get("code", 0) != 0: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=str(data.get("msg", "unknown") or "unknown"), - status_code=200, - details={"phase": "bot_info"}, - ) - return "" - bot_open_id = str( - (((data.get("data") or {}).get("bot") or {}).get("open_id")) or "" - ).strip() - if bot_open_id: - self._bot_open_ids[cache_key] = bot_open_id - self._bot_open_id = bot_open_id - return bot_open_id - - def _build_card_button_value( - self, - button: Dict[str, Any], - *, - target: FeishuDeliveryTarget, - binding: FeishuBinding, - signing_secret: str, - ) -> Dict[str, Any]: - contract = self._callback_contract_for_binding( - binding=binding, - signing_secret=signing_secret, - ) - command_text = str(button.get("value", "") or "").strip() - callback_payload = { - "label": str(button.get("label", "") or "").strip(), - "command": command_text, - "approval_id": str(button.get("approval_id", "") or "").strip(), - "workspace_id": target.workspace_id or binding.workspace_id, - "account_id": target.account_id or binding.account_id, - "channel_id": target.channel_id, - "message_id": target.reply_to_message_id, - } - envelope = contract.build_envelope( - request_id=secrets.token_hex(12), - workspace_id=callback_payload["workspace_id"], - action_type=_infer_callback_action_type(command_text, button), - payload=callback_payload, - ) - return { - "callback_envelope": dict(envelope.__dict__), - "payload": callback_payload, - } - - def _build_interactive_card( - self, - target: FeishuDeliveryTarget, - text: str, - buttons: list[dict], - *, - binding: FeishuBinding, - secrets: Dict[str, str], - ) -> Dict[str, Any]: - signing_secret = str( - secrets.get("app_secret", "") or binding.app_secret or "" - ).strip() - if not signing_secret: - raise RuntimeError("feishu_callback_signing_secret_missing") - actions = [] - for button in buttons[:6]: - command_text = str(button.get("value", "") or "").strip() - if not command_text: - continue - actions.append( - { - "tag": "button", - "type": str(button.get("style", "") or "default"), - "text": { - "tag": "plain_text", - "content": str(button.get("label", "") or "OpenClaw"), - }, - "value": self._build_card_button_value( - button, - target=target, - binding=binding, - signing_secret=signing_secret, - ), - } - ) - return { - "config": {"wide_screen_mode": True}, - "header": { - "template": "blue", - "title": {"tag": "plain_text", "content": "OpenClaw"}, - }, - "elements": [ - {"tag": "markdown", "content": text or "OpenClaw"}, - {"tag": "action", "actions": actions}, - ], - } - - async def _send_interactive_reply( - self, - target: FeishuDeliveryTarget, - text: str, - buttons: list[dict], - ) -> None: - resolution, binding, secrets = self._resolve_delivery_binding( - workspace_id=target.workspace_id, - account_id=target.account_id, - ) - if binding is None or not resolution.ok: - logger.warning( - "Feishu interactive reply dropped: no workspace binding available (%s / %s)", - target.workspace_id or "no-workspace", - target.account_id or "no-account", - ) - return - token = await self._get_tenant_access_token( - binding=binding, - workspace_id=target.workspace_id, - account_id=target.account_id, - ) - api_base = _resolve_domain_base(binding.domain) - card = self._build_interactive_card( - target, - text, - buttons, - binding=binding, - secrets=secrets, - ) - payload = { - "content": json.dumps(card, ensure_ascii=False), - "msg_type": "interactive", - } - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json; charset=utf-8", - } - if target.reply_to_message_id: - url = ( - f"{api_base}/open-apis/im/v1/messages/" - f"{target.reply_to_message_id}/reply" - ) - else: - url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" - payload["receive_id"] = target.channel_id - try: - data = safe_request_json( - method="POST", - url=url, - json_body=payload, - headers=headers, - content_type="application/json; charset=utf-8", - timeout_sec=15, - allow_hosts=_allowed_api_hosts(binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "interactive_reply"}, - ) - logger.warning( - "Feishu interactive reply failed: status=%s", exc.status_code - ) - return - if data.get("code", 0) != 0: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=str(data.get("msg", "unknown") or "unknown"), - status_code=200, - details={"phase": "interactive_reply"}, - ) - logger.warning( - "Feishu interactive reply failed: %s", data.get("msg", "unknown") - ) - - async def _send_reply( - self, - target: FeishuDeliveryTarget, - text: str, - *, - delivery_context: Optional[Dict[str, Any]] = None, - ) -> None: - ctx = dict(delivery_context or {}) - if target.workspace_id: - ctx.setdefault("workspace_id", target.workspace_id) - if target.account_id: - ctx.setdefault("account_id", target.account_id) - if target.reply_to_message_id: - ctx.setdefault("thread_id", target.reply_to_message_id) - decision = decide_reply_visibility( - delivery_context=ctx, - platform="feishu", - channel_kind=str(ctx.get("chat_type", "") or ""), - in_thread=bool(target.reply_to_message_id), - text=text, - ) - if decision.suppressed: - logger.info( - "Suppressed Feishu reply channel=%s reason=%s", - target.channel_id, - decision.reason, - ) - return - resolution, binding, _ = self._resolve_delivery_binding( - workspace_id=target.workspace_id, - account_id=target.account_id, - ) - if binding is None or not resolution.ok: - logger.warning( - "Feishu reply dropped: no workspace binding available (%s / %s)", - target.workspace_id or "no-workspace", - target.account_id or "no-account", - ) - return - token = await self._get_tenant_access_token( - binding=binding, - workspace_id=target.workspace_id, - account_id=target.account_id, - ) - api_base = _resolve_domain_base(binding.domain) - payload = { - "content": json.dumps({"text": text}, ensure_ascii=False), - "msg_type": "text", - } - headers = { - "Authorization": f"Bearer {token}", - "Content-Type": "application/json; charset=utf-8", - } - if target.reply_to_message_id: - url = ( - f"{api_base}/open-apis/im/v1/messages/" - f"{target.reply_to_message_id}/reply" - ) - else: - url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" - payload["receive_id"] = target.channel_id - try: - data = safe_request_json( - method="POST", - url=url, - json_body=payload, - headers=headers, - content_type="application/json; charset=utf-8", - timeout_sec=15, - allow_hosts=_allowed_api_hosts(binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "reply"}, - ) - logger.warning("Feishu reply failed: status=%s", exc.status_code) - return - if data.get("code", 0) != 0: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=str(data.get("msg", "unknown") or "unknown"), - status_code=200, - details={"phase": "reply"}, - ) - logger.warning( - "Feishu reply failed: %s", - data.get("msg", "unknown"), - ) - - async def send_message( - self, - channel_id: str, - text: str, - delivery_context: Optional[Dict[str, Any]] = None, - ): - ctx = dict(delivery_context or {}) - await self._send_reply( - FeishuDeliveryTarget( - channel_id=channel_id, - reply_to_message_id=str(ctx.get("thread_id", "") or "").strip(), - workspace_id=str(ctx.get("workspace_id", "") or "").strip(), - account_id=str(ctx.get("account_id", "") or "").strip(), - ), - text, - delivery_context=ctx, - ) - - async def send_image( - self, - channel_id: str, - image_data: bytes, - filename: str = "image.png", - caption: Optional[str] = None, - delivery_context: Optional[Dict[str, Any]] = None, - ): - ctx = dict(delivery_context or {}) - resolution, binding, _ = self._resolve_delivery_binding( - workspace_id=str(ctx.get("workspace_id", "") or "").strip(), - account_id=str(ctx.get("account_id", "") or "").strip(), - ) - if binding is None or not resolution.ok: - logger.warning( - "Feishu image dropped: no workspace binding available (%s / %s)", - str(ctx.get("workspace_id", "") or "").strip() or "no-workspace", - str(ctx.get("account_id", "") or "").strip() or "no-account", - ) - return - token = await self._get_tenant_access_token( - binding=binding, - workspace_id=str(ctx.get("workspace_id", "") or "").strip(), - account_id=str(ctx.get("account_id", "") or "").strip(), - ) - api_base = _resolve_domain_base(binding.domain) - upload_headers = { - "Accept": "application/json", - "Authorization": f"Bearer {token}", - } - upload_body, upload_content_type = _build_multipart_form( - fields={"image_type": "message"}, - file_field="image", - filename=filename, - file_bytes=image_data, - file_content_type="image/png", - ) - try: - upload_payload = safe_request_json( - method="POST", - url=f"{api_base}/open-apis/im/v1/images", - raw_body=upload_body, - headers=upload_headers, - content_type=upload_content_type, - timeout_sec=30, - allow_hosts=_allowed_api_hosts(binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "image_upload"}, - ) - logger.warning("Feishu image upload failed: status=%s", exc.status_code) - return - image_key = str( - (upload_payload.get("data") or {}).get("image_key", "") or "" - ).strip() - if upload_payload.get("code", 0) != 0 or not image_key: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=str(upload_payload.get("msg", "unknown") or "unknown"), - status_code=200, - details={"phase": "image_upload"}, - ) - logger.warning( - "Feishu image upload failed: %s", - upload_payload.get("msg", "unknown"), - ) - return - message_payload = { - "content": json.dumps({"image_key": image_key}, ensure_ascii=False), - "msg_type": "image", - } - thread_id = str(ctx.get("thread_id", "") or "").strip() - if thread_id: - send_url = f"{api_base}/open-apis/im/v1/messages/{thread_id}/reply" - else: - send_url = f"{api_base}/open-apis/im/v1/messages?receive_id_type=chat_id" - message_payload["receive_id"] = channel_id - try: - safe_request_json( - method="POST", - url=send_url, - json_body=message_payload, - headers={ - "Accept": "application/json", - "Authorization": f"Bearer {token}", - }, - content_type="application/json; charset=utf-8", - timeout_sec=30, - allow_hosts=_allowed_api_hosts(binding.domain), - policy=STANDARD_OUTBOUND_POLICY, - ) - except SafeIOHTTPError as exc: - if resolution.installation is not None: - self._installation_manager.mark_api_error( - resolution.installation.installation_id, - error_code=exc.reason, - status_code=exc.status_code, - details={"phase": "image_send"}, - ) - logger.warning("Feishu image send failed: status=%s", exc.status_code) - if caption: - await self.send_message( - channel_id, - caption, - delivery_context=ctx, - ) - - async def prime_bot_identity(self) -> None: - try: - await self._fetch_bot_open_id( - account_id=self._bound_account_id - or str(self.config.feishu_account_id or "").strip() - or str(self.config.feishu_default_account_id or "").strip(), - workspace_id=str(self.config.feishu_workspace_id or "").strip(), - allow_degrade=True, - ) - except Exception as exc: - logger.debug("Feishu bot identity fetch failed: %s", exc) diff --git a/connector/platforms/slack_delivery_handlers.py b/connector/platforms/slack_delivery_handlers.py new file mode 100644 index 0000000..76f703b --- /dev/null +++ b/connector/platforms/slack_delivery_handlers.py @@ -0,0 +1,345 @@ +"""Owned Slack response and media-delivery mixin.""" + +# ruff: noqa: SIM117, UP006, UP035, UP045 -- preserve frozen behavior/signatures. + +from typing import Any, Dict, Optional + +from ..reply_visibility import decide_reply_visibility + +# mypy: disable-error-code="attr-defined,no-any-return" + + +class SlackDeliveryMixin: + async def _send_interactive_reply( + self, + *, + channel_id: str, + text: str, + buttons: list[dict], + thread_ts: str = "", + delivery_context: Optional[Dict[str, Any]] = None, + ) -> None: + """Send a Slack Block Kit message with bounded button actions.""" + try: + import aiohttp as _aiohttp + except ImportError: + self._adapter_logger().warning( + "aiohttp not available; cannot send Slack interactive reply" + ) + return + + ctx = dict(delivery_context or {}) + if not thread_ts: + thread_ts = str(ctx.get("thread_id", "") or "").strip() + installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( + str(ctx.get("workspace_id", "") or "").strip() + ) + if not bot_token: + self._adapter_logger().warning( + "Slack interactive reply dropped: no workspace token available (workspace=%s)", + workspace_id or "legacy", + ) + return + + elements: list[dict] = [] + for idx, button in enumerate(buttons[:5]): + value = str(button.get("value", "") or "").strip() + if not value: + continue + label = str(button.get("label", "") or "OpenClaw").strip()[:75] + action_id = str( + button.get("action_type") + or button.get("action_id") + or f"openclaw.{idx}" + ).strip()[:255] + element: Dict[str, Any] = { + "type": "button", + "text": {"type": "plain_text", "text": label or "OpenClaw"}, + "value": value[:2000], + "action_id": action_id or f"openclaw.{idx}", + } + style = self._adapter_style_to_slack(str(button.get("style", "") or "")) + if style: + element["style"] = style + elements.append(element) + if not elements: + if text: + await self._send_reply( + channel_id=channel_id, + text=text, + thread_ts=thread_ts, + delivery_context=ctx, + ) + return + + payload: Dict[str, Any] = { + "channel": channel_id, + "text": text or "OpenClaw", + "blocks": [ + { + "type": "section", + "text": { + "type": "mrkdwn", + "text": (text or "OpenClaw")[:3000], + }, + }, + {"type": "actions", "elements": elements}, + ], + } + if thread_ts: + payload["thread_ts"] = thread_ts + + headers = { + "Authorization": f"Bearer {bot_token}", + "Content-Type": "application/json; charset=utf-8", + } + try: + async with _aiohttp.ClientSession() as session: + async with session.post( + "https://slack.com/api/chat.postMessage", + json=payload, + headers=headers, + timeout=_aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status != 200: + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=f"http_{resp.status}", + status_code=resp.status, + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + "interactive": True, + }, + ) + return + data = await resp.json() + if not data.get("ok"): + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=str(data.get("error", "unknown")), + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + "interactive": True, + }, + ) + elif installation_id: + self._installation_manager.mark_installation_health( + installation_id, + health_code="ok", + reason="chat_post_message_interactive_ok", + details={"workspace_id": workspace_id}, + ) + except Exception as e: + self._adapter_logger().warning("Slack interactive reply failed: %s", e) + + async def _send_reply( + self, + channel_id: str, + text: str, + thread_ts: str = "", + delivery_context: Optional[Dict[str, Any]] = None, + ) -> None: + """Send a message via Slack Web API (chat.postMessage).""" + ctx = dict(delivery_context or {}) + if not thread_ts: + thread_ts = str(ctx.get("thread_id", "") or "").strip() + decision = decide_reply_visibility( + delivery_context=ctx, + platform="slack", + channel_kind=self._adapter_channel_kind(channel_id), + in_thread=bool(thread_ts), + text=text, + ) + if decision.suppressed: + self._adapter_logger().info( + "Suppressed Slack reply channel=%s reason=%s", + channel_id, + decision.reason, + ) + return + try: + import aiohttp as _aiohttp + except ImportError: + self._adapter_logger().warning( + "aiohttp not available; cannot send Slack reply" + ) + return + + installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( + str(ctx.get("workspace_id", "") or "").strip() + ) + if not bot_token: + self._adapter_logger().warning( + "Slack reply dropped: no workspace token available (workspace=%s)", + workspace_id or "legacy", + ) + return + + url = "https://slack.com/api/chat.postMessage" + headers = { + "Authorization": f"Bearer {bot_token}", + "Content-Type": "application/json; charset=utf-8", + } + payload: Dict[str, Any] = { + "channel": channel_id, + "text": text, + } + if thread_ts: + payload["thread_ts"] = thread_ts + + try: + async with _aiohttp.ClientSession() as session: + async with session.post( + url, + json=payload, + headers=headers, + timeout=_aiohttp.ClientTimeout(total=10), + ) as resp: + if resp.status != 200: + body = await resp.text() + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=f"http_{resp.status}", + status_code=resp.status, + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + }, + ) + self._adapter_logger().warning( + f"Slack API error: status={resp.status} body={body[:200]}" + ) + else: + data = await resp.json() + if not data.get("ok"): + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=str(data.get("error", "unknown")), + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + }, + ) + self._adapter_logger().warning( + f"Slack API error: {data.get('error', 'unknown')}" + ) + elif installation_id: + self._installation_manager.mark_installation_health( + installation_id, + health_code="ok", + reason="chat_post_message_ok", + details={"workspace_id": workspace_id}, + ) + except Exception as e: + self._adapter_logger().warning(f"Slack reply failed: {e}") + + # ------------------------------------------------------------------ + # Platform contract: send_message / send_image + # ------------------------------------------------------------------ + + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[Dict[str, Any]] = None, + ): + """Platform contract: send text message.""" + await self._send_reply( + channel_id=channel_id, + text=text, + delivery_context=delivery_context, + ) + + async def send_image( + self, + channel_id: str, + image_data: bytes, + filename: str = "image.png", + caption: Optional[str] = None, + delivery_context: Optional[Dict[str, Any]] = None, + ): + """Platform contract: send image (Slack files.upload).""" + try: + import aiohttp as _aiohttp + except ImportError: + self._adapter_logger().warning( + "aiohttp not available; cannot upload Slack image" + ) + return + + ctx = dict(delivery_context or {}) + thread_ts = str(ctx.get("thread_id", "") or "").strip() + installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( + str(ctx.get("workspace_id", "") or "").strip() + ) + if not bot_token: + self._adapter_logger().warning( + "Slack image dropped: no workspace token available (workspace=%s)", + workspace_id or "legacy", + ) + return + + url = "https://slack.com/api/files.upload" + headers = { + "Authorization": f"Bearer {bot_token}", + } + data = _aiohttp.FormData() + data.add_field("file", image_data, filename=filename, content_type="image/png") + data.add_field("channels", channel_id) + if caption: + data.add_field("initial_comment", caption) + if thread_ts: + data.add_field("thread_ts", thread_ts) + + try: + async with _aiohttp.ClientSession() as session: + async with session.post( + url, + data=data, + headers=headers, + timeout=_aiohttp.ClientTimeout(total=30), + ) as resp: + if resp.status != 200: + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=f"http_{resp.status}", + status_code=resp.status, + details={ + "workspace_id": workspace_id, + "path": "files.upload", + }, + ) + self._adapter_logger().warning( + f"Slack file upload error: status={resp.status}" + ) + else: + resp_data = await resp.json() + if not resp_data.get("ok"): + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=str(resp_data.get("error", "unknown")), + details={ + "workspace_id": workspace_id, + "path": "files.upload", + }, + ) + self._adapter_logger().warning( + f"Slack file upload error: {resp_data.get('error')}" + ) + elif installation_id: + self._installation_manager.mark_installation_health( + installation_id, + health_code="ok", + reason="files_upload_ok", + details={"workspace_id": workspace_id}, + ) + except Exception as e: + self._adapter_logger().warning(f"Slack image upload failed: {e}") diff --git a/connector/platforms/slack_ingress_handlers.py b/connector/platforms/slack_ingress_handlers.py new file mode 100644 index 0000000..63c910f --- /dev/null +++ b/connector/platforms/slack_ingress_handlers.py @@ -0,0 +1,497 @@ +"""Owned Slack signed-ingress and interaction transaction mixin.""" + +import json +import time +from typing import Any, Dict, Optional +from urllib.parse import parse_qs + +from ..contract import CommandRequest + +# ruff: noqa: SIM102, UP006, UP035, UP045 -- preserve frozen behavior/signatures. +# mypy: disable-error-code="attr-defined,no-any-return" + + +class SlackIngressMixin: + async def handle_event(self, request): + """POST handler for Slack Events API.""" + _, web = self._adapter_import_aiohttp_web() + + try: + body_bytes = await request.read() + except Exception: + return self._adapter_make_response(web, status=400, text="Bad request") + + # -- Step 1: Signature verification (fail-closed) -- + timestamp = "" + signature = "" + if hasattr(request, "headers"): + timestamp = request.headers.get("X-Slack-Request-Timestamp", "") + signature = request.headers.get("X-Slack-Signature", "") + + if not self._adapter_verify_slack_signature( + signing_secret=self.config.slack_signing_secret or "", + timestamp=timestamp, + body=body_bytes, + signature=signature, + ): + self._adapter_logger().warning( + "Slack signature verification failed (rejected)" + ) + return self._adapter_make_response( + web, status=401, text="Invalid signature" + ) + + # -- Step 2: Parse payload -- + try: + payload = json.loads(body_bytes) + except json.JSONDecodeError: + return self._adapter_make_response(web, status=400, text="Bad JSON") + + # -- Step 3: url_verification challenge (Webhook only) -- + if payload.get("type") == "url_verification": + challenge = payload.get("challenge", "") + return self._adapter_make_json_response(web, {"challenge": challenge}) + + # -- Step 4: Process event -- + try: + await self.process_event_payload(payload) + except ValueError: + return self._adapter_make_response(web, status=400, text="Bad Request") + return self._adapter_make_response(web, status=200, text="OK") + + async def handle_interaction(self, request): + """POST handler for Slack Block Kit interactivity callbacks.""" + _, web = self._adapter_import_aiohttp_web() + + try: + body_bytes = await request.read() + except Exception: + return self._adapter_make_response(web, status=400, text="Bad request") + + timestamp = "" + signature = "" + if hasattr(request, "headers"): + timestamp = request.headers.get("X-Slack-Request-Timestamp", "") + signature = request.headers.get("X-Slack-Signature", "") + + if not self._adapter_verify_slack_signature( + signing_secret=self.config.slack_signing_secret or "", + timestamp=timestamp, + body=body_bytes, + signature=signature, + ): + self._adapter_logger().warning( + "Slack interaction signature verification failed (rejected)" + ) + return self._adapter_make_response( + web, status=401, text="Invalid signature" + ) + + parsed = parse_qs(body_bytes.decode("utf-8"), keep_blank_values=True) + raw_payload = (parsed.get("payload") or [""])[0] + if not raw_payload: + return self._adapter_make_response(web, status=400, text="Missing payload") + + try: + payload = json.loads(raw_payload) + except json.JSONDecodeError: + return self._adapter_make_response(web, status=400, text="Bad payload") + if not isinstance(payload, dict): + return self._adapter_make_response(web, status=400, text="Bad payload") + + try: + routed = await self.process_interaction_payload(payload) + except ValueError: + return self._adapter_make_response(web, status=400, text="Bad Request") + except Exception as exc: + safe_text = self._adapter_safe_external_error_text( + "Slack interaction failed", exc + ) + self._adapter_logger().warning("Slack interaction failed: %s", safe_text) + return self._adapter_make_response(web, status=500, text=safe_text) + + # Slack requires a fast acknowledgement for interactivity requests. + # Keep the external response bounded; detailed action results are routed + # through the existing reply/deferred-response surfaces. + return self._adapter_make_json_response( + web, {"ok": True, "routed": bool(routed)} + ) + + async def process_event_payload(self, payload: Dict[str, Any]) -> None: + """ + Shared event processing path for both webhook and socket mode transports. + """ + if payload.get("type") != "event_callback": + return + + event = payload.get("event", {}) + event_id = payload.get("event_id", "") + event_type = event.get("type", "") + workspace_id = self._installation_manager.extract_workspace_id(payload) + + if event_type in ("app_uninstalled", "tokens_revoked", "app_rate_limited"): + if workspace_id: + self._handle_lifecycle_event(workspace_id, event_type) + return + + # -- Step 5: Replay / dedupe guard -- + if not event_id: + self._adapter_logger().warning("Slack event missing event_id (rejected)") + raise ValueError("Missing event_id") + + if not self._replay_guard.check_and_record(event_id): + self._adapter_logger().debug( + f"Slack duplicate event_id={event_id} (accepted, no-op)" + ) + return + + # -- Step 6: Bot-loop prevention -- + # Resolve bot user ID from authorizations or cache. + bot_user_id = self._get_bot_user_id(payload, workspace_id) + + sender_id = event.get("user", "") + if sender_id and bot_user_id and sender_id == bot_user_id: + return + + if event.get("bot_id"): + return + + subtype = event.get("subtype", "") + if subtype and subtype not in ("", "file_share"): + return + + # -- Step 7: Event normalization -- + text = event.get("text", "").strip() + channel_id = event.get("channel", "") + thread_ts = event.get("thread_ts", "") + message_ts = event.get("ts", "") + + if event_type not in ("message", "app_mention"): + return + + if not text or not sender_id: + return + + # S67: Require mention in group channels. + is_dm = channel_id.startswith("D") + mentioned_bot = event_type == "app_mention" or ( + bool(bot_user_id) and f"<@{bot_user_id}>" in text + ) + if not is_dm and self.config.slack_require_mention: + if event_type != "app_mention": + if bot_user_id and f"<@{bot_user_id}>" not in text: + return + + if bot_user_id: + text = text.replace(f"<@{bot_user_id}>", "").strip() + + # -- Step 8: Allowlist checks (S67) -- + if self._user_allowlist.entries: + user_result = self._user_allowlist.evaluate(sender_id) + if user_result.decision == "deny": + self._adapter_logger().warning( + f"Slack user {sender_id} denied by allowlist" + ) + return + + if self._channel_allowlist.entries and channel_id: + chan_result = self._channel_allowlist.evaluate(channel_id) + if chan_result.decision == "deny": + self._adapter_logger().warning( + f"Slack channel {channel_id} denied by allowlist" + ) + return + + # -- Step 9: Build CommandRequest and route -- + req = CommandRequest( + platform="slack", + sender_id=sender_id, + channel_id=channel_id, + username=sender_id, + message_id=event_id, + text=text, + timestamp=float(message_ts) if message_ts else time.time(), + workspace_id=workspace_id, + thread_id=thread_ts + or (message_ts if self.config.slack_reply_in_thread else ""), + ) + + try: + resp = await self.router.handle(req) + resp_text = getattr(resp, "text", "") + if not isinstance(resp_text, str): + resp_text = str(resp_text) if resp_text is not None else "" + + buttons = getattr(resp, "buttons", []) or [] + if resp_text or buttons: + if buttons: + await self._send_interactive_reply( + channel_id=channel_id, + text=resp_text or "OpenClaw", + buttons=buttons, + thread_ts=req.thread_id, + delivery_context={ + "workspace_id": workspace_id, + "thread_id": req.thread_id, + "channel_kind": self._adapter_channel_kind(channel_id), + "mentioned": mentioned_bot, + }, + ) + else: + await self._send_reply( + channel_id=channel_id, + text=resp_text, + thread_ts=req.thread_id, + delivery_context={ + "workspace_id": workspace_id, + "thread_id": req.thread_id, + "channel_kind": self._adapter_channel_kind(channel_id), + "mentioned": mentioned_bot, + }, + ) + except Exception as e: + self._adapter_logger().error( + "Slack event handling failed (error_type=%s)", type(e).__name__ + ) + + async def process_interaction_payload(self, payload: Dict[str, Any]) -> bool: + interaction_type = str(payload.get("type", "") or "").strip() + if interaction_type not in self._adapter_interaction_types(): + return False + + request = self._build_interaction_request(payload) + if request is None: + return False + + replay_key = self._interaction_replay_key(payload, request) + if self._interaction_lifecycle is None: # pragma: no cover + if not self._replay_guard.check_and_record(replay_key): + self._adapter_logger().debug( + "Slack duplicate interaction %s (accepted, no-op)", replay_key + ) + return False + claim = None + else: + claim = self._interaction_lifecycle.claim( + replay_key, + metadata={ + "platform": "slack", + "workspace_id": request.workspace_id, + "interaction_type": str(payload.get("type", "") or ""), + }, + ) + if claim is not None and not claim.accepted: + self._adapter_logger().debug( + "Slack duplicate interaction %s state=%s code=%s (accepted, no-op)", + replay_key, + claim.record.state, + claim.code, + ) + return False + + # IMPORTANT: interactive run-like payloads must be routed through the same + # approval semantics as text commands. Untrusted users get approval forced + # before CommandRouter sees the request, avoiding a parallel bypass path. + if request.text.startswith("/run") and not ( + self.router._is_admin(request) or self.router._is_trusted(request) + ): + request.text = self._adapter_force_approval_command(request.text) + + try: + response = await self.router.handle(request) + except Exception: + # IMPORTANT: only failures before router completion are retryable. + # Once router.handle returns, duplicate user actions must not reroute. + if self._interaction_lifecycle is not None: + self._interaction_lifecycle.release_retryable( + replay_key, reason="slack_interaction_failed_before_commit" + ) + raise + + if self._interaction_lifecycle is not None: + self._interaction_lifecycle.commit_success(replay_key, reason="routed") + response_text = str(getattr(response, "text", "") or "").strip() + response_buttons = getattr(response, "buttons", []) or [] + if response_text or response_buttons: + if response_buttons: + await self._send_interactive_reply( + channel_id=request.channel_id, + text=response_text or "Action processed.", + buttons=response_buttons, + thread_ts=request.thread_id, + delivery_context={ + "workspace_id": request.workspace_id, + "thread_id": request.thread_id, + }, + ) + elif response_text: + await self._send_reply( + channel_id=request.channel_id, + text=response_text, + thread_ts=request.thread_id, + delivery_context={ + "workspace_id": request.workspace_id, + "thread_id": request.thread_id, + }, + ) + return True + + def _build_interaction_request( + self, payload: Dict[str, Any] + ) -> Optional[CommandRequest]: + interaction_type = str(payload.get("type", "") or "").strip() + command_text = self._extract_interaction_command(payload) + if not command_text: + return None + + team = payload.get("team") or {} + user = payload.get("user") or {} + container = payload.get("container") or {} + channel = payload.get("channel") or {} + view = payload.get("view") or {} + message = payload.get("message") or {} + action = self._first_action(payload) + + workspace_id = self._adapter_first_non_empty( + team.get("id"), + payload.get("team_id"), + ( + payload.get("enterprise", {}).get("id") + if isinstance(payload.get("enterprise"), dict) + else "" + ), + ) + sender_id = self._adapter_first_non_empty( + user.get("id"), payload.get("user_id") + ) + channel_id = self._adapter_first_non_empty( + channel.get("id"), + container.get("channel_id"), + payload.get("channel_id"), + ) + message_id = self._adapter_first_non_empty( + view.get("id"), + action.get("action_ts"), + container.get("message_ts"), + payload.get("trigger_id"), + f"slack-interaction-{int(time.time())}", + ) + thread_id = self._adapter_first_non_empty( + container.get("thread_ts"), + message.get("thread_ts") if isinstance(message, dict) else "", + container.get("message_ts"), + ) + if not thread_id and self.config.slack_reply_in_thread: + thread_id = self._adapter_first_non_empty( + container.get("message_ts"), message.get("ts") + ) + + return CommandRequest( + platform="slack", + sender_id=sender_id, + channel_id=channel_id or sender_id, + username=self._adapter_first_non_empty( + user.get("username"), user.get("name"), sender_id + ), + message_id=message_id, + text=command_text, + timestamp=time.time(), + workspace_id=workspace_id, + thread_id=thread_id, + metadata={ + "interactive_callback": True, + "interaction_type": interaction_type, + "action_id": self._adapter_first_non_empty( + action.get("action_id"), view.get("callback_id") + ), + "response_url": str(payload.get("response_url", "") or ""), + }, + ) + + def _extract_interaction_command(self, payload: Dict[str, Any]) -> str: + interaction_type = str(payload.get("type", "") or "").strip() + if interaction_type == "block_actions": + action = self._first_action(payload) + selected = action.get("selected_option") or {} + value = self._adapter_first_non_empty( + action.get("value"), + selected.get("value") if isinstance(selected, dict) else "", + action.get("action_id"), + ) + parsed = self._adapter_json_loads_safe(value) + return self._adapter_first_non_empty( + parsed.get("command"), parsed.get("value"), value + ) + if interaction_type == "view_submission": + view = payload.get("view") or {} + private_meta = self._adapter_first_non_empty(view.get("private_metadata")) + parsed = self._adapter_json_loads_safe(private_meta) + if parsed: + return self._adapter_first_non_empty( + parsed.get("command"), parsed.get("value") + ) + if private_meta: + return private_meta + state = (view.get("state") or {}).get("values") or {} + return self._extract_command_from_view_state(state) + if interaction_type == "workflow_step_execute": + workflow_step = payload.get("workflow_step") or {} + inputs = workflow_step.get("inputs") or {} + command = inputs.get("command") or {} + if isinstance(command, dict): + return self._adapter_first_non_empty(command.get("value")) + return self._adapter_first_non_empty(workflow_step.get("callback_id")) + return "" + + def _extract_command_from_view_state(self, state: Dict[str, Any]) -> str: + if not isinstance(state, dict): + return "" + for block_value in state.values(): + if not isinstance(block_value, dict): + continue + for action_value in block_value.values(): + if not isinstance(action_value, dict): + continue + candidate = self._adapter_first_non_empty( + action_value.get("value"), + ( + (action_value.get("selected_option") or {}).get("value") + if isinstance(action_value.get("selected_option"), dict) + else "" + ), + ) + parsed = self._adapter_json_loads_safe(candidate) + command = self._adapter_first_non_empty( + parsed.get("command"), parsed.get("value"), candidate + ) + if command: + return command + return "" + + def _first_action(self, payload: Dict[str, Any]) -> Dict[str, Any]: + actions = payload.get("actions") or [] + if isinstance(actions, list) and actions and isinstance(actions[0], dict): + return actions[0] + return {} + + def _interaction_replay_key( + self, payload: Dict[str, Any], request: CommandRequest + ) -> str: + action = self._first_action(payload) + key_parts = [ + "interaction", + str(payload.get("type", "") or ""), + request.workspace_id, + request.sender_id, + request.channel_id, + request.message_id, + str(payload.get("trigger_id", "") or ""), + str(action.get("action_id", "") or ""), + str(action.get("action_ts", "") or ""), + request.text, + ] + return ":".join(key_parts) + + # ------------------------------------------------------------------ + # Slack Web API reply + # ------------------------------------------------------------------ diff --git a/connector/platforms/slack_installation_handlers.py b/connector/platforms/slack_installation_handlers.py new file mode 100644 index 0000000..6cf1d5f --- /dev/null +++ b/connector/platforms/slack_installation_handlers.py @@ -0,0 +1,171 @@ +"""Owned Slack installation, OAuth, and workspace-identity mixin.""" + +# ruff: noqa: UP006, UP035, UP045 -- preserve frozen facade annotations. + +from typing import Any, Dict, Optional, Tuple + +# mypy: disable-error-code="attr-defined,has-type,no-any-return" + + +class SlackInstallationMixin: + async def handle_oauth_install(self, request): + _, web = self._adapter_import_aiohttp_web() + if not self._installation_manager.can_handle_oauth(): + return self._adapter_make_response( + web, status=503, text="Slack OAuth not configured" + ) + state = self._installation_manager.issue_install_state() + return self._adapter_make_redirect_response( + web, self._installation_manager.build_install_url(state) + ) + + async def handle_oauth_callback(self, request): + _, web = self._adapter_import_aiohttp_web() + if not self._installation_manager.can_handle_oauth(): + return self._adapter_make_response( + web, status=503, text="Slack OAuth not configured" + ) + query = getattr(request, "query", {}) or {} + if query.get("error"): + return self._adapter_make_response( + web, + status=400, + text=f"Slack OAuth rejected: {query.get('error')}", + ) + state = str(query.get("state", "") or "").strip() + code = str(query.get("code", "") or "").strip() + if not state or not code: + return self._adapter_make_response( + web, status=400, text="Missing OAuth callback fields" + ) + if not self._installation_manager.consume_install_state(state): + return self._adapter_make_response( + web, status=400, text="Invalid or replayed OAuth state" + ) + try: + payload = await self._installation_manager.exchange_code(code) + installation = self._installation_manager.upsert_from_oauth_payload(payload) + return self._adapter_make_response( + web, + status=200, + text=( + "Slack installation complete for " + f"{installation.workspace_id} ({installation.installation_id})." + ), + ) + except Exception as exc: + safe_text = self._adapter_safe_external_error_text( + "Slack OAuth processing failed", exc + ) + self._adapter_logger().warning("Slack OAuth callback failed: %s", safe_text) + return self._adapter_make_response( + web, + status=502, + text=safe_text, + ) + + def _get_bot_user_id(self, payload: Dict[str, Any], workspace_id: str) -> str: + candidate = "" + if workspace_id and workspace_id in self._bot_user_ids: + return self._bot_user_ids[workspace_id] + if self._bot_user_id: + return self._bot_user_id + authorizations = payload.get("authorizations", []) + if authorizations and isinstance(authorizations, list): + candidate = str((authorizations[0] or {}).get("user_id", "") or "").strip() + if candidate: + self._bot_user_id = candidate + if workspace_id: + self._bot_user_ids[workspace_id] = candidate + return candidate + if workspace_id: + workspace_resolution, _ = ( + self._installation_manager.resolve_workspace_tokens(workspace_id) + ) + candidate = self._installation_manager.bot_user_id_for_installation( + workspace_resolution.installation if workspace_resolution.ok else None + ) + if candidate: + self._bot_user_ids[workspace_id] = candidate + if self._bot_user_id is None: + self._bot_user_id = candidate + return candidate + + def _resolve_workspace_credentials( + self, workspace_id: str + ) -> Tuple[Optional[str], Optional[str], Optional[str]]: + workspace_id = str(workspace_id or "").strip() + if workspace_id: + resolution, tokens = self._installation_manager.resolve_workspace_tokens( + workspace_id + ) + if resolution.ok and resolution.installation is not None: + bot_token = tokens.get("bot_token") + if bot_token: + self._installation_manager.mark_resolution_success( + resolution.installation.installation_id, workspace_id + ) + return ( + resolution.installation.installation_id, + bot_token, + workspace_id, + ) + self._adapter_logger().warning( + "Slack workspace %s resolved without bot token secret", workspace_id + ) + return ( + resolution.installation.installation_id, + None, + workspace_id, + ) + if ( + not self._installation_manager.oauth_enabled + and self.config.slack_bot_token + ): + return (None, self.config.slack_bot_token, workspace_id) + self._adapter_logger().warning( + "Slack workspace resolution failed for %s: %s (%s)", + workspace_id, + resolution.reject_reason, + resolution.health_code, + ) + return (None, None, workspace_id) + if self.config.slack_bot_token: + return (None, self.config.slack_bot_token, "") + return (None, None, workspace_id) + + def _handle_lifecycle_event(self, workspace_id: str, event_type: str) -> None: + installation_id = self._installation_manager.installation_id_for_workspace( + workspace_id + ) + try: + if event_type == "app_uninstalled": + self._installation_manager.mark_installation_health( + installation_id, + health_code="revoked", + reason="slack_app_uninstalled", + details={"workspace_id": workspace_id}, + ) + self._installation_manager.uninstall_installation( + installation_id, reason="slack_app_uninstalled" + ) + elif event_type == "tokens_revoked": + self._installation_manager.mark_installation_health( + installation_id, + health_code="invalid_token", + reason="slack_tokens_revoked", + details={"workspace_id": workspace_id}, + ) + elif event_type == "app_rate_limited": + self._installation_manager.mark_installation_health( + installation_id, + health_code="degraded", + reason="slack_app_rate_limited", + details={"workspace_id": workspace_id}, + ) + except ValueError: + self._adapter_logger().warning( + "Slack lifecycle event for unbound workspace %s (%s)", + workspace_id, + event_type, + ) diff --git a/connector/platforms/slack_webhook.py b/connector/platforms/slack_webhook.py index 1b233f2..a25348b 100644 --- a/connector/platforms/slack_webhook.py +++ b/connector/platforms/slack_webhook.py @@ -33,14 +33,14 @@ import hmac import json import logging import time -from typing import Any, Dict, Optional, Tuple -from urllib.parse import parse_qs +from typing import Any, Dict, Optional from ..config import ConnectorConfig -from ..contract import CommandRequest, CommandResponse -from ..reply_visibility import decide_reply_visibility from ..router import CommandRouter from ..security_profile import AllowlistPolicy, ReplayGuard +from .slack_delivery_handlers import SlackDeliveryMixin +from .slack_ingress_handlers import SlackIngressMixin +from .slack_installation_handlers import SlackInstallationMixin from .slack_installation_manager import SlackInstallationManager try: @@ -203,7 +203,11 @@ def verify_slack_signature( # -- Slack adapter ---------------------------------------------------------- -class SlackWebhookServer: +class SlackWebhookServer( + SlackInstallationMixin, + SlackIngressMixin, + SlackDeliveryMixin, +): """ F56 -- Slack Events API adapter. @@ -246,9 +250,63 @@ class SlackWebhookServer: self._installation_manager = SlackInstallationManager(config) # Bot user ID (resolved on first event or set from config) - self._bot_user_id: Optional[str] = None + self._bot_user_id: Optional[str] = None # type: ignore[assignment] self._bot_user_ids: Dict[str, str] = {} + # IMPORTANT: resolve facade globals at call time; integration suites and + # minimal-host shims patch these security/protocol seams directly. + @staticmethod + def _adapter_import_aiohttp_web(): + return _import_aiohttp_web() + + @staticmethod + def _adapter_make_response(*args, **kwargs): + return _make_response(*args, **kwargs) + + @staticmethod + def _adapter_make_json_response(*args, **kwargs): + return _make_json_response(*args, **kwargs) + + @staticmethod + def _adapter_make_redirect_response(*args, **kwargs): + return _make_redirect_response(*args, **kwargs) + + @staticmethod + def _adapter_safe_external_error_text(*args, **kwargs): + return _safe_external_error_text(*args, **kwargs) + + @staticmethod + def _adapter_verify_slack_signature(*args, **kwargs): + return verify_slack_signature(*args, **kwargs) + + @staticmethod + def _adapter_json_loads_safe(*args, **kwargs): + return _json_loads_safe(*args, **kwargs) + + @staticmethod + def _adapter_first_non_empty(*args, **kwargs): + return _first_non_empty(*args, **kwargs) + + @staticmethod + def _adapter_force_approval_command(*args, **kwargs): + return _force_approval_command(*args, **kwargs) + + @staticmethod + def _adapter_style_to_slack(*args, **kwargs): + return _style_to_slack(*args, **kwargs) + + @staticmethod + def _adapter_channel_kind(*args, **kwargs): + return _slack_channel_kind(*args, **kwargs) + + @staticmethod + def _adapter_interaction_types(): + return _SLACK_INTERACTION_TYPES + + @staticmethod + def _adapter_logger(): + return logger + # ------------------------------------------------------------------ # Lifecycle # ------------------------------------------------------------------ @@ -310,941 +368,3 @@ class SlackWebhookServer: # ------------------------------------------------------------------ # Event handler # ------------------------------------------------------------------ - - async def handle_oauth_install(self, request): - _, web = _import_aiohttp_web() - if not self._installation_manager.can_handle_oauth(): - return _make_response(web, status=503, text="Slack OAuth not configured") - state = self._installation_manager.issue_install_state() - return _make_redirect_response( - web, self._installation_manager.build_install_url(state) - ) - - async def handle_oauth_callback(self, request): - _, web = _import_aiohttp_web() - if not self._installation_manager.can_handle_oauth(): - return _make_response(web, status=503, text="Slack OAuth not configured") - query = getattr(request, "query", {}) or {} - if query.get("error"): - return _make_response( - web, - status=400, - text=f"Slack OAuth rejected: {query.get('error')}", - ) - state = str(query.get("state", "") or "").strip() - code = str(query.get("code", "") or "").strip() - if not state or not code: - return _make_response(web, status=400, text="Missing OAuth callback fields") - if not self._installation_manager.consume_install_state(state): - return _make_response( - web, status=400, text="Invalid or replayed OAuth state" - ) - try: - payload = await self._installation_manager.exchange_code(code) - installation = self._installation_manager.upsert_from_oauth_payload(payload) - return _make_response( - web, - status=200, - text=( - "Slack installation complete for " - f"{installation.workspace_id} ({installation.installation_id})." - ), - ) - except Exception as exc: - safe_text = _safe_external_error_text("Slack OAuth processing failed", exc) - logger.warning("Slack OAuth callback failed: %s", safe_text) - return _make_response( - web, - status=502, - text=safe_text, - ) - - def _get_bot_user_id(self, payload: Dict[str, Any], workspace_id: str) -> str: - candidate = "" - if workspace_id and workspace_id in self._bot_user_ids: - return self._bot_user_ids[workspace_id] - if self._bot_user_id: - return self._bot_user_id - authorizations = payload.get("authorizations", []) - if authorizations and isinstance(authorizations, list): - candidate = str((authorizations[0] or {}).get("user_id", "") or "").strip() - if candidate: - self._bot_user_id = candidate - if workspace_id: - self._bot_user_ids[workspace_id] = candidate - return candidate - if workspace_id: - workspace_resolution, _ = ( - self._installation_manager.resolve_workspace_tokens(workspace_id) - ) - candidate = self._installation_manager.bot_user_id_for_installation( - workspace_resolution.installation if workspace_resolution.ok else None - ) - if candidate: - self._bot_user_ids[workspace_id] = candidate - if self._bot_user_id is None: - self._bot_user_id = candidate - return candidate - - def _resolve_workspace_credentials( - self, workspace_id: str - ) -> Tuple[Optional[str], Optional[str], Optional[str]]: - workspace_id = str(workspace_id or "").strip() - if workspace_id: - resolution, tokens = self._installation_manager.resolve_workspace_tokens( - workspace_id - ) - if resolution.ok and resolution.installation is not None: - bot_token = tokens.get("bot_token") - if bot_token: - self._installation_manager.mark_resolution_success( - resolution.installation.installation_id, workspace_id - ) - return ( - resolution.installation.installation_id, - bot_token, - workspace_id, - ) - logger.warning( - "Slack workspace %s resolved without bot token secret", workspace_id - ) - return ( - resolution.installation.installation_id, - None, - workspace_id, - ) - if ( - not self._installation_manager.oauth_enabled - and self.config.slack_bot_token - ): - return (None, self.config.slack_bot_token, workspace_id) - logger.warning( - "Slack workspace resolution failed for %s: %s (%s)", - workspace_id, - resolution.reject_reason, - resolution.health_code, - ) - return (None, None, workspace_id) - if self.config.slack_bot_token: - return (None, self.config.slack_bot_token, "") - return (None, None, workspace_id) - - def _handle_lifecycle_event(self, workspace_id: str, event_type: str) -> None: - installation_id = self._installation_manager.installation_id_for_workspace( - workspace_id - ) - try: - if event_type == "app_uninstalled": - self._installation_manager.mark_installation_health( - installation_id, - health_code="revoked", - reason="slack_app_uninstalled", - details={"workspace_id": workspace_id}, - ) - self._installation_manager.uninstall_installation( - installation_id, reason="slack_app_uninstalled" - ) - elif event_type == "tokens_revoked": - self._installation_manager.mark_installation_health( - installation_id, - health_code="invalid_token", - reason="slack_tokens_revoked", - details={"workspace_id": workspace_id}, - ) - elif event_type == "app_rate_limited": - self._installation_manager.mark_installation_health( - installation_id, - health_code="degraded", - reason="slack_app_rate_limited", - details={"workspace_id": workspace_id}, - ) - except ValueError: - logger.warning( - "Slack lifecycle event for unbound workspace %s (%s)", - workspace_id, - event_type, - ) - - async def handle_event(self, request): - """POST handler for Slack Events API.""" - _, web = _import_aiohttp_web() - - try: - body_bytes = await request.read() - except Exception: - return _make_response(web, status=400, text="Bad request") - - # -- Step 1: Signature verification (fail-closed) -- - timestamp = "" - signature = "" - if hasattr(request, "headers"): - timestamp = request.headers.get("X-Slack-Request-Timestamp", "") - signature = request.headers.get("X-Slack-Signature", "") - - if not verify_slack_signature( - signing_secret=self.config.slack_signing_secret or "", - timestamp=timestamp, - body=body_bytes, - signature=signature, - ): - logger.warning("Slack signature verification failed (rejected)") - return _make_response(web, status=401, text="Invalid signature") - - # -- Step 2: Parse payload -- - try: - payload = json.loads(body_bytes) - except json.JSONDecodeError: - return _make_response(web, status=400, text="Bad JSON") - - # -- Step 3: url_verification challenge (Webhook only) -- - if payload.get("type") == "url_verification": - challenge = payload.get("challenge", "") - return _make_json_response(web, {"challenge": challenge}) - - # -- Step 4: Process event -- - try: - await self.process_event_payload(payload) - except ValueError: - return _make_response(web, status=400, text="Bad Request") - return _make_response(web, status=200, text="OK") - - async def handle_interaction(self, request): - """POST handler for Slack Block Kit interactivity callbacks.""" - _, web = _import_aiohttp_web() - - try: - body_bytes = await request.read() - except Exception: - return _make_response(web, status=400, text="Bad request") - - timestamp = "" - signature = "" - if hasattr(request, "headers"): - timestamp = request.headers.get("X-Slack-Request-Timestamp", "") - signature = request.headers.get("X-Slack-Signature", "") - - if not verify_slack_signature( - signing_secret=self.config.slack_signing_secret or "", - timestamp=timestamp, - body=body_bytes, - signature=signature, - ): - logger.warning("Slack interaction signature verification failed (rejected)") - return _make_response(web, status=401, text="Invalid signature") - - parsed = parse_qs(body_bytes.decode("utf-8"), keep_blank_values=True) - raw_payload = (parsed.get("payload") or [""])[0] - if not raw_payload: - return _make_response(web, status=400, text="Missing payload") - - try: - payload = json.loads(raw_payload) - except json.JSONDecodeError: - return _make_response(web, status=400, text="Bad payload") - if not isinstance(payload, dict): - return _make_response(web, status=400, text="Bad payload") - - try: - routed = await self.process_interaction_payload(payload) - except ValueError: - return _make_response(web, status=400, text="Bad Request") - except Exception as exc: - safe_text = _safe_external_error_text("Slack interaction failed", exc) - logger.warning("Slack interaction failed: %s", safe_text) - return _make_response(web, status=500, text=safe_text) - - # Slack requires a fast acknowledgement for interactivity requests. - # Keep the external response bounded; detailed action results are routed - # through the existing reply/deferred-response surfaces. - return _make_json_response(web, {"ok": True, "routed": bool(routed)}) - - async def process_event_payload(self, payload: Dict[str, Any]) -> None: - """ - Shared event processing path for both webhook and socket mode transports. - """ - if payload.get("type") != "event_callback": - return - - event = payload.get("event", {}) - event_id = payload.get("event_id", "") - event_type = event.get("type", "") - workspace_id = self._installation_manager.extract_workspace_id(payload) - - if event_type in ("app_uninstalled", "tokens_revoked", "app_rate_limited"): - if workspace_id: - self._handle_lifecycle_event(workspace_id, event_type) - return - - # -- Step 5: Replay / dedupe guard -- - if not event_id: - logger.warning("Slack event missing event_id (rejected)") - raise ValueError("Missing event_id") - - if not self._replay_guard.check_and_record(event_id): - logger.debug(f"Slack duplicate event_id={event_id} (accepted, no-op)") - return - - # -- Step 6: Bot-loop prevention -- - # Resolve bot user ID from authorizations or cache. - bot_user_id = self._get_bot_user_id(payload, workspace_id) - - sender_id = event.get("user", "") - if sender_id and bot_user_id and sender_id == bot_user_id: - return - - if event.get("bot_id"): - return - - subtype = event.get("subtype", "") - if subtype and subtype not in ("", "file_share"): - return - - # -- Step 7: Event normalization -- - text = event.get("text", "").strip() - channel_id = event.get("channel", "") - thread_ts = event.get("thread_ts", "") - message_ts = event.get("ts", "") - - if event_type not in ("message", "app_mention"): - return - - if not text or not sender_id: - return - - # S67: Require mention in group channels. - is_dm = channel_id.startswith("D") - mentioned_bot = event_type == "app_mention" or ( - bool(bot_user_id) and f"<@{bot_user_id}>" in text - ) - if not is_dm and self.config.slack_require_mention: - if event_type != "app_mention": - if bot_user_id and f"<@{bot_user_id}>" not in text: - return - - if bot_user_id: - text = text.replace(f"<@{bot_user_id}>", "").strip() - - # -- Step 8: Allowlist checks (S67) -- - if self._user_allowlist.entries: - user_result = self._user_allowlist.evaluate(sender_id) - if user_result.decision == "deny": - logger.warning(f"Slack user {sender_id} denied by allowlist") - return - - if self._channel_allowlist.entries and channel_id: - chan_result = self._channel_allowlist.evaluate(channel_id) - if chan_result.decision == "deny": - logger.warning(f"Slack channel {channel_id} denied by allowlist") - return - - # -- Step 9: Build CommandRequest and route -- - req = CommandRequest( - platform="slack", - sender_id=sender_id, - channel_id=channel_id, - username=sender_id, - message_id=event_id, - text=text, - timestamp=float(message_ts) if message_ts else time.time(), - workspace_id=workspace_id, - thread_id=thread_ts - or (message_ts if self.config.slack_reply_in_thread else ""), - ) - - try: - resp = await self.router.handle(req) - resp_text = getattr(resp, "text", "") - if not isinstance(resp_text, str): - resp_text = str(resp_text) if resp_text is not None else "" - - buttons = getattr(resp, "buttons", []) or [] - if resp_text or buttons: - if buttons: - await self._send_interactive_reply( - channel_id=channel_id, - text=resp_text or "OpenClaw", - buttons=buttons, - thread_ts=req.thread_id, - delivery_context={ - "workspace_id": workspace_id, - "thread_id": req.thread_id, - "channel_kind": _slack_channel_kind(channel_id), - "mentioned": mentioned_bot, - }, - ) - else: - await self._send_reply( - channel_id=channel_id, - text=resp_text, - thread_ts=req.thread_id, - delivery_context={ - "workspace_id": workspace_id, - "thread_id": req.thread_id, - "channel_kind": _slack_channel_kind(channel_id), - "mentioned": mentioned_bot, - }, - ) - except Exception as e: - logger.error( - "Slack event handling failed (error_type=%s)", type(e).__name__ - ) - - async def process_interaction_payload(self, payload: Dict[str, Any]) -> bool: - interaction_type = str(payload.get("type", "") or "").strip() - if interaction_type not in _SLACK_INTERACTION_TYPES: - return False - - request = self._build_interaction_request(payload) - if request is None: - return False - - replay_key = self._interaction_replay_key(payload, request) - if self._interaction_lifecycle is None: # pragma: no cover - if not self._replay_guard.check_and_record(replay_key): - logger.debug( - "Slack duplicate interaction %s (accepted, no-op)", replay_key - ) - return False - claim = None - else: - claim = self._interaction_lifecycle.claim( - replay_key, - metadata={ - "platform": "slack", - "workspace_id": request.workspace_id, - "interaction_type": str(payload.get("type", "") or ""), - }, - ) - if claim is not None and not claim.accepted: - logger.debug( - "Slack duplicate interaction %s state=%s code=%s (accepted, no-op)", - replay_key, - claim.record.state, - claim.code, - ) - return False - - # IMPORTANT: interactive run-like payloads must be routed through the same - # approval semantics as text commands. Untrusted users get approval forced - # before CommandRouter sees the request, avoiding a parallel bypass path. - if request.text.startswith("/run") and not ( - self.router._is_admin(request) or self.router._is_trusted(request) - ): - request.text = _force_approval_command(request.text) - - try: - response = await self.router.handle(request) - except Exception: - # IMPORTANT: only failures before router completion are retryable. - # Once router.handle returns, duplicate user actions must not reroute. - if self._interaction_lifecycle is not None: - self._interaction_lifecycle.release_retryable( - replay_key, reason="slack_interaction_failed_before_commit" - ) - raise - - if self._interaction_lifecycle is not None: - self._interaction_lifecycle.commit_success(replay_key, reason="routed") - response_text = str(getattr(response, "text", "") or "").strip() - response_buttons = getattr(response, "buttons", []) or [] - if response_text or response_buttons: - if response_buttons: - await self._send_interactive_reply( - channel_id=request.channel_id, - text=response_text or "Action processed.", - buttons=response_buttons, - thread_ts=request.thread_id, - delivery_context={ - "workspace_id": request.workspace_id, - "thread_id": request.thread_id, - }, - ) - elif response_text: - await self._send_reply( - channel_id=request.channel_id, - text=response_text, - thread_ts=request.thread_id, - delivery_context={ - "workspace_id": request.workspace_id, - "thread_id": request.thread_id, - }, - ) - return True - - def _build_interaction_request( - self, payload: Dict[str, Any] - ) -> Optional[CommandRequest]: - interaction_type = str(payload.get("type", "") or "").strip() - command_text = self._extract_interaction_command(payload) - if not command_text: - return None - - team = payload.get("team") or {} - user = payload.get("user") or {} - container = payload.get("container") or {} - channel = payload.get("channel") or {} - view = payload.get("view") or {} - message = payload.get("message") or {} - action = self._first_action(payload) - - workspace_id = _first_non_empty( - team.get("id"), - payload.get("team_id"), - ( - payload.get("enterprise", {}).get("id") - if isinstance(payload.get("enterprise"), dict) - else "" - ), - ) - sender_id = _first_non_empty(user.get("id"), payload.get("user_id")) - channel_id = _first_non_empty( - channel.get("id"), - container.get("channel_id"), - payload.get("channel_id"), - ) - message_id = _first_non_empty( - view.get("id"), - action.get("action_ts"), - container.get("message_ts"), - payload.get("trigger_id"), - f"slack-interaction-{int(time.time())}", - ) - thread_id = _first_non_empty( - container.get("thread_ts"), - message.get("thread_ts") if isinstance(message, dict) else "", - container.get("message_ts"), - ) - if not thread_id and self.config.slack_reply_in_thread: - thread_id = _first_non_empty(container.get("message_ts"), message.get("ts")) - - return CommandRequest( - platform="slack", - sender_id=sender_id, - channel_id=channel_id or sender_id, - username=_first_non_empty( - user.get("username"), user.get("name"), sender_id - ), - message_id=message_id, - text=command_text, - timestamp=time.time(), - workspace_id=workspace_id, - thread_id=thread_id, - metadata={ - "interactive_callback": True, - "interaction_type": interaction_type, - "action_id": _first_non_empty( - action.get("action_id"), view.get("callback_id") - ), - "response_url": str(payload.get("response_url", "") or ""), - }, - ) - - def _extract_interaction_command(self, payload: Dict[str, Any]) -> str: - interaction_type = str(payload.get("type", "") or "").strip() - if interaction_type == "block_actions": - action = self._first_action(payload) - selected = action.get("selected_option") or {} - value = _first_non_empty( - action.get("value"), - selected.get("value") if isinstance(selected, dict) else "", - action.get("action_id"), - ) - parsed = _json_loads_safe(value) - return _first_non_empty(parsed.get("command"), parsed.get("value"), value) - if interaction_type == "view_submission": - view = payload.get("view") or {} - private_meta = _first_non_empty(view.get("private_metadata")) - parsed = _json_loads_safe(private_meta) - if parsed: - return _first_non_empty(parsed.get("command"), parsed.get("value")) - if private_meta: - return private_meta - state = (view.get("state") or {}).get("values") or {} - return self._extract_command_from_view_state(state) - if interaction_type == "workflow_step_execute": - workflow_step = payload.get("workflow_step") or {} - inputs = workflow_step.get("inputs") or {} - command = inputs.get("command") or {} - if isinstance(command, dict): - return _first_non_empty(command.get("value")) - return _first_non_empty(workflow_step.get("callback_id")) - return "" - - def _extract_command_from_view_state(self, state: Dict[str, Any]) -> str: - if not isinstance(state, dict): - return "" - for block_value in state.values(): - if not isinstance(block_value, dict): - continue - for action_value in block_value.values(): - if not isinstance(action_value, dict): - continue - candidate = _first_non_empty( - action_value.get("value"), - ( - (action_value.get("selected_option") or {}).get("value") - if isinstance(action_value.get("selected_option"), dict) - else "" - ), - ) - parsed = _json_loads_safe(candidate) - command = _first_non_empty( - parsed.get("command"), parsed.get("value"), candidate - ) - if command: - return command - return "" - - def _first_action(self, payload: Dict[str, Any]) -> Dict[str, Any]: - actions = payload.get("actions") or [] - if isinstance(actions, list) and actions and isinstance(actions[0], dict): - return actions[0] - return {} - - def _interaction_replay_key( - self, payload: Dict[str, Any], request: CommandRequest - ) -> str: - action = self._first_action(payload) - key_parts = [ - "interaction", - str(payload.get("type", "") or ""), - request.workspace_id, - request.sender_id, - request.channel_id, - request.message_id, - str(payload.get("trigger_id", "") or ""), - str(action.get("action_id", "") or ""), - str(action.get("action_ts", "") or ""), - request.text, - ] - return ":".join(key_parts) - - # ------------------------------------------------------------------ - # Slack Web API reply - # ------------------------------------------------------------------ - - async def _send_interactive_reply( - self, - *, - channel_id: str, - text: str, - buttons: list[dict], - thread_ts: str = "", - delivery_context: Optional[Dict[str, Any]] = None, - ) -> None: - """Send a Slack Block Kit message with bounded button actions.""" - try: - import aiohttp as _aiohttp - except ImportError: - logger.warning("aiohttp not available; cannot send Slack interactive reply") - return - - ctx = dict(delivery_context or {}) - if not thread_ts: - thread_ts = str(ctx.get("thread_id", "") or "").strip() - installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( - str(ctx.get("workspace_id", "") or "").strip() - ) - if not bot_token: - logger.warning( - "Slack interactive reply dropped: no workspace token available (workspace=%s)", - workspace_id or "legacy", - ) - return - - elements: list[dict] = [] - for idx, button in enumerate(buttons[:5]): - value = str(button.get("value", "") or "").strip() - if not value: - continue - label = str(button.get("label", "") or "OpenClaw").strip()[:75] - action_id = str( - button.get("action_type") - or button.get("action_id") - or f"openclaw.{idx}" - ).strip()[:255] - element: Dict[str, Any] = { - "type": "button", - "text": {"type": "plain_text", "text": label or "OpenClaw"}, - "value": value[:2000], - "action_id": action_id or f"openclaw.{idx}", - } - style = _style_to_slack(str(button.get("style", "") or "")) - if style: - element["style"] = style - elements.append(element) - if not elements: - if text: - await self._send_reply( - channel_id=channel_id, - text=text, - thread_ts=thread_ts, - delivery_context=ctx, - ) - return - - payload: Dict[str, Any] = { - "channel": channel_id, - "text": text or "OpenClaw", - "blocks": [ - { - "type": "section", - "text": { - "type": "mrkdwn", - "text": (text or "OpenClaw")[:3000], - }, - }, - {"type": "actions", "elements": elements}, - ], - } - if thread_ts: - payload["thread_ts"] = thread_ts - - headers = { - "Authorization": f"Bearer {bot_token}", - "Content-Type": "application/json; charset=utf-8", - } - try: - async with _aiohttp.ClientSession() as session: - async with session.post( - "https://slack.com/api/chat.postMessage", - json=payload, - headers=headers, - timeout=_aiohttp.ClientTimeout(total=10), - ) as resp: - if resp.status != 200: - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=f"http_{resp.status}", - status_code=resp.status, - details={ - "workspace_id": workspace_id, - "path": "chat.postMessage", - "interactive": True, - }, - ) - return - data = await resp.json() - if not data.get("ok"): - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=str(data.get("error", "unknown")), - details={ - "workspace_id": workspace_id, - "path": "chat.postMessage", - "interactive": True, - }, - ) - elif installation_id: - self._installation_manager.mark_installation_health( - installation_id, - health_code="ok", - reason="chat_post_message_interactive_ok", - details={"workspace_id": workspace_id}, - ) - except Exception as e: - logger.warning("Slack interactive reply failed: %s", e) - - async def _send_reply( - self, - channel_id: str, - text: str, - thread_ts: str = "", - delivery_context: Optional[Dict[str, Any]] = None, - ) -> None: - """Send a message via Slack Web API (chat.postMessage).""" - ctx = dict(delivery_context or {}) - if not thread_ts: - thread_ts = str(ctx.get("thread_id", "") or "").strip() - decision = decide_reply_visibility( - delivery_context=ctx, - platform="slack", - channel_kind=_slack_channel_kind(channel_id), - in_thread=bool(thread_ts), - text=text, - ) - if decision.suppressed: - logger.info( - "Suppressed Slack reply channel=%s reason=%s", - channel_id, - decision.reason, - ) - return - try: - import aiohttp as _aiohttp - except ImportError: - logger.warning("aiohttp not available; cannot send Slack reply") - return - - installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( - str(ctx.get("workspace_id", "") or "").strip() - ) - if not bot_token: - logger.warning( - "Slack reply dropped: no workspace token available (workspace=%s)", - workspace_id or "legacy", - ) - return - - url = "https://slack.com/api/chat.postMessage" - headers = { - "Authorization": f"Bearer {bot_token}", - "Content-Type": "application/json; charset=utf-8", - } - payload: Dict[str, Any] = { - "channel": channel_id, - "text": text, - } - if thread_ts: - payload["thread_ts"] = thread_ts - - try: - async with _aiohttp.ClientSession() as session: - async with session.post( - url, - json=payload, - headers=headers, - timeout=_aiohttp.ClientTimeout(total=10), - ) as resp: - if resp.status != 200: - body = await resp.text() - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=f"http_{resp.status}", - status_code=resp.status, - details={ - "workspace_id": workspace_id, - "path": "chat.postMessage", - }, - ) - logger.warning( - f"Slack API error: status={resp.status} body={body[:200]}" - ) - else: - data = await resp.json() - if not data.get("ok"): - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=str(data.get("error", "unknown")), - details={ - "workspace_id": workspace_id, - "path": "chat.postMessage", - }, - ) - logger.warning( - f"Slack API error: {data.get('error', 'unknown')}" - ) - elif installation_id: - self._installation_manager.mark_installation_health( - installation_id, - health_code="ok", - reason="chat_post_message_ok", - details={"workspace_id": workspace_id}, - ) - except Exception as e: - logger.warning(f"Slack reply failed: {e}") - - # ------------------------------------------------------------------ - # Platform contract: send_message / send_image - # ------------------------------------------------------------------ - - async def send_message( - self, - channel_id: str, - text: str, - delivery_context: Optional[Dict[str, Any]] = None, - ): - """Platform contract: send text message.""" - await self._send_reply( - channel_id=channel_id, - text=text, - delivery_context=delivery_context, - ) - - async def send_image( - self, - channel_id: str, - image_data: bytes, - filename: str = "image.png", - caption: Optional[str] = None, - delivery_context: Optional[Dict[str, Any]] = None, - ): - """Platform contract: send image (Slack files.upload).""" - try: - import aiohttp as _aiohttp - except ImportError: - logger.warning("aiohttp not available; cannot upload Slack image") - return - - ctx = dict(delivery_context or {}) - thread_ts = str(ctx.get("thread_id", "") or "").strip() - installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( - str(ctx.get("workspace_id", "") or "").strip() - ) - if not bot_token: - logger.warning( - "Slack image dropped: no workspace token available (workspace=%s)", - workspace_id or "legacy", - ) - return - - url = "https://slack.com/api/files.upload" - headers = { - "Authorization": f"Bearer {bot_token}", - } - data = _aiohttp.FormData() - data.add_field("file", image_data, filename=filename, content_type="image/png") - data.add_field("channels", channel_id) - if caption: - data.add_field("initial_comment", caption) - if thread_ts: - data.add_field("thread_ts", thread_ts) - - try: - async with _aiohttp.ClientSession() as session: - async with session.post( - url, - data=data, - headers=headers, - timeout=_aiohttp.ClientTimeout(total=30), - ) as resp: - if resp.status != 200: - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=f"http_{resp.status}", - status_code=resp.status, - details={ - "workspace_id": workspace_id, - "path": "files.upload", - }, - ) - logger.warning(f"Slack file upload error: status={resp.status}") - else: - resp_data = await resp.json() - if not resp_data.get("ok"): - if installation_id: - self._installation_manager.mark_api_error( - installation_id, - error_code=str(resp_data.get("error", "unknown")), - details={ - "workspace_id": workspace_id, - "path": "files.upload", - }, - ) - logger.warning( - f"Slack file upload error: {resp_data.get('error')}" - ) - elif installation_id: - self._installation_manager.mark_installation_health( - installation_id, - health_code="ok", - reason="files_upload_ok", - details={"workspace_id": workspace_id}, - ) - except Exception as e: - logger.warning(f"Slack image upload failed: {e}") diff --git a/scripts/verify_platform_adapter_contract.py b/scripts/verify_platform_adapter_contract.py new file mode 100644 index 0000000..73a1bc0 --- /dev/null +++ b/scripts/verify_platform_adapter_contract.py @@ -0,0 +1,158 @@ +"""Verify the frozen R223 Slack and Feishu adapter contracts.""" + +from __future__ import annotations + +import argparse +import ast +import hashlib +import inspect +import json +import sys +import textwrap +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) +CONTRACT_PATH = ROOT / "tests" / "platform_adapter_contract_r223.json" + + +def _canonical_json(value: Any) -> str: + return json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) + "\n" + + +def _method_signatures(owner: type) -> dict[str, str]: + names = { + name + for cls in owner.__mro__ + if cls is not object + for name, value in cls.__dict__.items() + if callable(value) + and (not name.startswith("__") or name == "__init__") + and not name.startswith("_adapter_") + } + return { + name: str(inspect.signature(getattr(owner, name))) for name in sorted(names) + } + + +def _instance_ownership(owner: type) -> list[str]: + tree = ast.parse(textwrap.dedent(inspect.getsource(owner.__init__))) # type: ignore[misc] + names = set() + for node in ast.walk(tree): + target = None + if isinstance(node, ast.Assign) and len(node.targets) == 1: + target = node.targets[0] + elif isinstance(node, ast.AnnAssign): + target = node.target + if ( + isinstance(target, ast.Attribute) + and isinstance(target.value, ast.Name) + and target.value.id == "self" + ): + names.add(target.attr) + return sorted(names) + + +def build_contract() -> dict[str, Any]: + from connector.platforms.feishu_webhook import ( + FEISHU_DOMAIN_BASES, + FEISHU_TOKEN_TTL_SEC, + FEISHU_WEBHOOK_MAX_BODY_BYTES, + FeishuDeliveryTarget, + FeishuWebhookServer, + ) + from connector.platforms.slack_webhook import ( + SLACK_SIGNING_VERSION, + SLACK_TIMESTAMP_MAX_DRIFT_SEC, + SlackWebhookServer, + ) + + router_contract = ROOT / "tests" / "connector_router_contract_r222.json" + return { + "schema_version": 1, + "slack": { + "class_constants": { + "REPLAY_WINDOW_SEC": SlackWebhookServer.REPLAY_WINDOW_SEC, + "NONCE_CACHE_SIZE": SlackWebhookServer.NONCE_CACHE_SIZE, + "SLACK_SIGNING_VERSION": SLACK_SIGNING_VERSION, + "SLACK_TIMESTAMP_MAX_DRIFT_SEC": SLACK_TIMESTAMP_MAX_DRIFT_SEC, + }, + "method_signatures": _method_signatures(SlackWebhookServer), + "instance_ownership": _instance_ownership(SlackWebhookServer), + "routes": [ + ["POST", "slack_webhook_path", "handle_event"], + ["POST", "slack_interactions_path", "handle_interaction"], + ["GET", "slack_oauth_install_path", "handle_oauth_install"], + ["GET", "slack_oauth_callback_path", "handle_oauth_callback"], + ], + "patch_seams": [ + "_import_aiohttp_web", + "_make_response", + "_make_json_response", + "_make_redirect_response", + "verify_slack_signature", + "logger", + ], + }, + "feishu": { + "class_constants": { + "REPLAY_WINDOW_SEC": FeishuWebhookServer.REPLAY_WINDOW_SEC, + "NONCE_CACHE_SIZE": FeishuWebhookServer.NONCE_CACHE_SIZE, + "FEISHU_WEBHOOK_MAX_BODY_BYTES": FEISHU_WEBHOOK_MAX_BODY_BYTES, + "FEISHU_TOKEN_TTL_SEC": FEISHU_TOKEN_TTL_SEC, + "FEISHU_DOMAIN_BASES": FEISHU_DOMAIN_BASES, + }, + "method_signatures": _method_signatures(FeishuWebhookServer), + "delivery_target_signature": str(inspect.signature(FeishuDeliveryTarget)), + "instance_ownership": _instance_ownership(FeishuWebhookServer), + "routes": [ + ["POST", "feishu_webhook_path", "handle_event"], + ["POST", "feishu_callback_path", "handle_callback"], + ], + "patch_seams": [ + "_import_aiohttp_web", + "_make_response", + "_make_json_response", + "safe_request_json", + "logger", + ], + }, + "response_matrix_owners": [ + "tests.test_r124_slack_ingress_contract", + "tests.test_r125_slack_real_backend_lane", + "tests.test_f57_slack_transport_parity", + "tests.test_f58_slack_oauth_installations", + "tests.test_f59_slack_interactions", + "tests.test_f67_feishu_transport_parity", + "tests.test_f68_feishu_installations", + "tests.test_f69_feishu_callbacks", + "tests.test_f74_reply_visibility_policy", + "tests.security.test_s80_connector_ingress", + ], + "router_contract_digest": hashlib.sha256( + router_contract.read_bytes() + ).hexdigest(), + } + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--write-baseline", action="store_true") + args = parser.parse_args() + actual = build_contract() + if args.write_baseline: + CONTRACT_PATH.write_text(_canonical_json(actual), encoding="utf-8") + print(f"PLATFORM-ADAPTER-CONTRACT-WRITTEN: {CONTRACT_PATH}") + return 0 + expected = json.loads(CONTRACT_PATH.read_text(encoding="utf-8")) + if actual != expected: + print("PLATFORM-ADAPTER-CONTRACT-FAIL") + return 1 + print("PLATFORM-ADAPTER-CONTRACT-PASS") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/exception_boundary_policy.json b/tests/exception_boundary_policy.json index 0e57c62..16d1464 100644 --- a/tests/exception_boundary_policy.json +++ b/tests/exception_boundary_policy.json @@ -184,26 +184,33 @@ } ] }, - "connector/platforms/slack_webhook.py": { + "connector/platforms/slack_installation_handlers.py": { "coverage": "selected_scopes", "selected_scopes": [ - "SlackWebhookServer.handle_oauth_callback", - "SlackWebhookServer.handle_event", - "SlackWebhookServer.handle_interaction", - "SlackWebhookServer.process_event_payload", - "SlackWebhookServer.process_interaction_payload" + "SlackInstallationMixin.handle_oauth_callback" ], "broad_catches": [ { - "scope": "SlackWebhookServer.handle_oauth_callback", + "scope": "SlackInstallationMixin.handle_oauth_callback", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "OAuth callback failures translate to a constant external response.", "regression_owner": "tests/test_f58_slack_oauth_installations.py", "review_after": "2027-01-11" - }, + } + ] + }, + "connector/platforms/slack_ingress_handlers.py": { + "coverage": "selected_scopes", + "selected_scopes": [ + "SlackIngressMixin.handle_event", + "SlackIngressMixin.handle_interaction", + "SlackIngressMixin.process_event_payload", + "SlackIngressMixin.process_interaction_payload" + ], + "broad_catches": [ { - "scope": "SlackWebhookServer.handle_event", + "scope": "SlackIngressMixin.handle_event", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Request-read failures translate to the existing fixed 400 response.", @@ -211,7 +218,7 @@ "review_after": "2027-01-11" }, { - "scope": "SlackWebhookServer.handle_interaction", + "scope": "SlackIngressMixin.handle_interaction", "expected_count": 2, "classification": "allowed_boundary_guard", "reason": "Request-read and routed interaction failures retain fixed acknowledgement responses.", @@ -219,7 +226,7 @@ "review_after": "2027-01-11" }, { - "scope": "SlackWebhookServer.process_event_payload", + "scope": "SlackIngressMixin.process_event_payload", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Slack event dispatch is acknowledged while failures log only a safe type classification.", @@ -227,7 +234,7 @@ "review_after": "2027-01-11" }, { - "scope": "SlackWebhookServer.process_interaction_payload", + "scope": "SlackIngressMixin.process_interaction_payload", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Pre-commit router failures release replay claims and re-raise for fixed outer translation.", @@ -236,16 +243,16 @@ } ] }, - "connector/platforms/feishu_webhook.py": { + "connector/platforms/feishu_ingress_handlers.py": { "coverage": "selected_scopes", "selected_scopes": [ - "FeishuWebhookServer.handle_event", - "FeishuWebhookServer.handle_callback", - "FeishuWebhookServer.process_callback_payload" + "FeishuIngressMixin.handle_event", + "FeishuIngressMixin.handle_callback", + "FeishuIngressMixin.process_callback_payload" ], "broad_catches": [ { - "scope": "FeishuWebhookServer.handle_event", + "scope": "FeishuIngressMixin.handle_event", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Request-read failures translate to the existing fixed 400 response.", @@ -253,7 +260,7 @@ "review_after": "2027-01-11" }, { - "scope": "FeishuWebhookServer.handle_callback", + "scope": "FeishuIngressMixin.handle_callback", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Callback request-read failures translate to the existing fixed 400 response.", @@ -261,7 +268,7 @@ "review_after": "2027-01-11" }, { - "scope": "FeishuWebhookServer.process_callback_payload", + "scope": "FeishuIngressMixin.process_callback_payload", "expected_count": 1, "classification": "allowed_boundary_guard", "reason": "Pre-completion router failures release the callback request for retry and re-raise.", diff --git a/tests/platform_adapter_contract_r223.json b/tests/platform_adapter_contract_r223.json new file mode 100644 index 0000000..3afd4a6 --- /dev/null +++ b/tests/platform_adapter_contract_r223.json @@ -0,0 +1,170 @@ +{ + "feishu": { + "class_constants": { + "FEISHU_DOMAIN_BASES": { + "feishu": "https://open.feishu.cn", + "lark": "https://open.larksuite.com" + }, + "FEISHU_TOKEN_TTL_SEC": 3600, + "FEISHU_WEBHOOK_MAX_BODY_BYTES": 262144, + "NONCE_CACHE_SIZE": 5000, + "REPLAY_WINDOW_SEC": 300 + }, + "delivery_target_signature": "(channel_id: 'str', reply_to_message_id: 'str' = '', workspace_id: 'str' = '', account_id: 'str' = '') -> None", + "instance_ownership": [ + "_bot_open_id", + "_bot_open_ids", + "_bound_account_id", + "_callback_contract_secrets", + "_callback_contracts", + "_chat_allowlist", + "_installation_manager", + "_replay_guard", + "_tenant_access_token_expires_at", + "_tenant_access_tokens", + "_user_allowlist", + "app", + "config", + "router", + "runner", + "site" + ], + "method_signatures": { + "__init__": "(self, config: 'ConnectorConfig', router: 'CommandRouter', *, installation_manager: 'Optional[FeishuInstallationManager]' = None, bound_account_id: 'str' = '')", + "_actor_context_for_callback": "(self, *, actor_id: 'str', actor_open_id: 'str', channel_id: 'str', message_id: 'str', workspace_id: 'str', account_id: 'str', command_text: 'str') -> 'Tuple[CallbackActorContext, CommandRequest]'", + "_build_callback_response": "(self, *, ok: 'bool', text: 'str', response_type: 'str' = 'info', card: 'Optional[Dict[str, Any]]' = None, duplicate: 'bool' = False, decision_code: 'str' = '') -> 'Dict[str, Any]'", + "_build_card_button_value": "(self, button: 'Dict[str, Any]', *, target: 'FeishuDeliveryTarget', binding: 'FeishuBinding', signing_secret: 'str') -> 'Dict[str, Any]'", + "_build_interactive_card": "(self, target: 'FeishuDeliveryTarget', text: 'str', buttons: 'list[dict]', *, binding: 'FeishuBinding', secrets: 'Dict[str, str]') -> 'Dict[str, Any]'", + "_build_request": "(self, payload: 'Dict[str, Any]', *, binding: 'FeishuBinding', bot_open_id: 'str') -> 'Optional[CommandRequest]'", + "_cache_key_for_binding": "(self, binding: 'FeishuBinding') -> 'str'", + "_cached_bot_open_id": "(self, binding: 'FeishuBinding') -> 'str'", + "_callback_contract_for_binding": "(self, *, binding: 'FeishuBinding', signing_secret: 'str') -> 'ConnectorCallbackContract'", + "_extract_callback_action": "(self, payload: 'Dict[str, Any]') -> 'Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any], Dict[str, Any], str, str]'", + "_fetch_bot_open_id": "(self, *, binding: 'Optional[FeishuBinding]' = None, workspace_id: 'str' = '', account_id: 'str' = '', allow_degrade: 'bool' = False) -> 'str'", + "_get_tenant_access_token": "(self, *, binding: 'Optional[FeishuBinding]' = None, workspace_id: 'str' = '', account_id: 'str' = '') -> 'str'", + "_is_challenge": "(self, payload: 'Dict[str, Any]') -> 'bool'", + "_resolve_delivery_binding": "(self, *, workspace_id: 'str' = '', account_id: 'str' = '') -> 'Tuple[InstallationResolution, Optional[FeishuBinding], Dict[str, str]]'", + "_resolve_inbound_binding": "(self, payload: 'Dict[str, Any]') -> 'FeishuBinding'", + "_send_interactive_reply": "(self, target: 'FeishuDeliveryTarget', text: 'str', buttons: 'list[dict]') -> 'None'", + "_send_reply": "(self, target: 'FeishuDeliveryTarget', text: 'str', *, delivery_context: 'Optional[Dict[str, Any]]' = None) -> 'None'", + "_verify_request_token": "(self, payload: 'Dict[str, Any]') -> 'bool'", + "handle_callback": "(self, request)", + "handle_event": "(self, request)", + "prime_bot_identity": "(self) -> 'None'", + "process_callback_payload": "(self, payload: 'Dict[str, Any]') -> 'Dict[str, Any]'", + "process_event_payload": "(self, payload: 'Dict[str, Any]', *, binding: 'Optional[FeishuBinding]' = None) -> 'None'", + "send_image": "(self, channel_id: 'str', image_data: 'bytes', filename: 'str' = 'image.png', caption: 'Optional[str]' = None, delivery_context: 'Optional[Dict[str, Any]]' = None)", + "send_message": "(self, channel_id: 'str', text: 'str', delivery_context: 'Optional[Dict[str, Any]]' = None)", + "start": "(self)", + "stop": "(self)" + }, + "patch_seams": [ + "_import_aiohttp_web", + "_make_response", + "_make_json_response", + "safe_request_json", + "logger" + ], + "routes": [ + [ + "POST", + "feishu_webhook_path", + "handle_event" + ], + [ + "POST", + "feishu_callback_path", + "handle_callback" + ] + ] + }, + "response_matrix_owners": [ + "tests.test_r124_slack_ingress_contract", + "tests.test_r125_slack_real_backend_lane", + "tests.test_f57_slack_transport_parity", + "tests.test_f58_slack_oauth_installations", + "tests.test_f59_slack_interactions", + "tests.test_f67_feishu_transport_parity", + "tests.test_f68_feishu_installations", + "tests.test_f69_feishu_callbacks", + "tests.test_f74_reply_visibility_policy", + "tests.security.test_s80_connector_ingress" + ], + "router_contract_digest": "78360537c9129f1671d1c62b663b1ea8a53d5b29fe067e2d73ab7c6bad5824aa", + "schema_version": 1, + "slack": { + "class_constants": { + "NONCE_CACHE_SIZE": 5000, + "REPLAY_WINDOW_SEC": 300, + "SLACK_SIGNING_VERSION": "v0", + "SLACK_TIMESTAMP_MAX_DRIFT_SEC": 300 + }, + "instance_ownership": [ + "_bot_user_id", + "_bot_user_ids", + "_channel_allowlist", + "_installation_manager", + "_interaction_lifecycle", + "_replay_guard", + "_user_allowlist", + "app", + "config", + "router", + "runner", + "site" + ], + "method_signatures": { + "__init__": "(self, config: connector.config.ConnectorConfig, router: connector.router.CommandRouter)", + "_build_interaction_request": "(self, payload: Dict[str, Any]) -> Optional[connector.contract.CommandRequest]", + "_extract_command_from_view_state": "(self, state: Dict[str, Any]) -> str", + "_extract_interaction_command": "(self, payload: Dict[str, Any]) -> str", + "_first_action": "(self, payload: Dict[str, Any]) -> Dict[str, Any]", + "_get_bot_user_id": "(self, payload: Dict[str, Any], workspace_id: str) -> str", + "_handle_lifecycle_event": "(self, workspace_id: str, event_type: str) -> None", + "_interaction_replay_key": "(self, payload: Dict[str, Any], request: connector.contract.CommandRequest) -> str", + "_resolve_workspace_credentials": "(self, workspace_id: str) -> Tuple[Optional[str], Optional[str], Optional[str]]", + "_send_interactive_reply": "(self, *, channel_id: str, text: str, buttons: list[dict], thread_ts: str = '', delivery_context: Optional[Dict[str, Any]] = None) -> None", + "_send_reply": "(self, channel_id: str, text: str, thread_ts: str = '', delivery_context: Optional[Dict[str, Any]] = None) -> None", + "handle_event": "(self, request)", + "handle_interaction": "(self, request)", + "handle_oauth_callback": "(self, request)", + "handle_oauth_install": "(self, request)", + "process_event_payload": "(self, payload: Dict[str, Any]) -> None", + "process_interaction_payload": "(self, payload: Dict[str, Any]) -> bool", + "send_image": "(self, channel_id: str, image_data: bytes, filename: str = 'image.png', caption: Optional[str] = None, delivery_context: Optional[Dict[str, Any]] = None)", + "send_message": "(self, channel_id: str, text: str, delivery_context: Optional[Dict[str, Any]] = None)", + "start": "(self)", + "stop": "(self)" + }, + "patch_seams": [ + "_import_aiohttp_web", + "_make_response", + "_make_json_response", + "_make_redirect_response", + "verify_slack_signature", + "logger" + ], + "routes": [ + [ + "POST", + "slack_webhook_path", + "handle_event" + ], + [ + "POST", + "slack_interactions_path", + "handle_interaction" + ], + [ + "GET", + "slack_oauth_install_path", + "handle_oauth_install" + ], + [ + "GET", + "slack_oauth_callback_path", + "handle_oauth_callback" + ] + ] + } +} diff --git a/tests/static_analysis_policy.json b/tests/static_analysis_policy.json index 6f6ff9f..9caf74c 100644 --- a/tests/static_analysis_policy.json +++ b/tests/static_analysis_policy.json @@ -658,20 +658,6 @@ "message": "\"None\" has no attribute \"start\"", "count": 1 }, - { - "tool": "mypy", - "path": "connector/platforms/feishu_webhook.py", - "code": "index", - "message": "Value of type \"Any | None\" is not indexable", - "count": 1 - }, - { - "tool": "mypy", - "path": "connector/platforms/feishu_webhook.py", - "code": "name-defined", - "message": "Name \"InstallationResolution\" is not defined", - "count": 1 - }, { "tool": "mypy", "path": "connector/platforms/feishu_webhook.py", @@ -756,13 +742,6 @@ "message": "Returning Any from function declared to return \"str | None\"", "count": 1 }, - { - "tool": "mypy", - "path": "connector/platforms/slack_webhook.py", - "code": "arg-type", - "message": "Argument 1 to \"_is_admin\" of \"RouterDispatchMixin\" has incompatible type \"CommandRequest\"; expected \"str\"", - "count": 1 - }, { "tool": "mypy", "path": "connector/platforms/slack_webhook.py", @@ -3178,33 +3157,19 @@ "message": "Use `X | None` for type annotations", "count": 2 }, - { - "tool": "ruff", - "path": "connector/platforms/feishu_webhook.py", - "code": "F401", - "message": "`asyncio` imported but unused", - "count": 1 - }, - { - "tool": "ruff", - "path": "connector/platforms/feishu_webhook.py", - "code": "F821", - "message": "Undefined name `InstallationResolution`", - "count": 1 - }, { "tool": "ruff", "path": "connector/platforms/feishu_webhook.py", "code": "UP006", "message": "Use `dict` instead of `Dict` for type annotation", - "count": 34 + "count": 12 }, { "tool": "ruff", "path": "connector/platforms/feishu_webhook.py", "code": "UP006", "message": "Use `tuple` instead of `Tuple` for type annotation", - "count": 4 + "count": 1 }, { "tool": "ruff", @@ -3232,7 +3197,7 @@ "path": "connector/platforms/feishu_webhook.py", "code": "UP045", "message": "Use `X | None` for type annotations", - "count": 12 + "count": 2 }, { "tool": "ruff", @@ -3339,40 +3304,12 @@ "message": "Use `X | None` for type annotations", "count": 2 }, - { - "tool": "ruff", - "path": "connector/platforms/slack_webhook.py", - "code": "F401", - "message": "`..contract.CommandResponse` imported but unused", - "count": 1 - }, - { - "tool": "ruff", - "path": "connector/platforms/slack_webhook.py", - "code": "SIM102", - "message": "Use a single `if` statement instead of nested `if` statements", - "count": 2 - }, - { - "tool": "ruff", - "path": "connector/platforms/slack_webhook.py", - "code": "SIM117", - "message": "Use a single `with` statement with multiple contexts instead of nested `with` statements", - "count": 3 - }, { "tool": "ruff", "path": "connector/platforms/slack_webhook.py", "code": "UP006", "message": "Use `dict` instead of `Dict` for type annotation", - "count": 18 - }, - { - "tool": "ruff", - "path": "connector/platforms/slack_webhook.py", - "code": "UP006", - "message": "Use `tuple` instead of `Tuple` for type annotation", - "count": 1 + "count": 2 }, { "tool": "ruff", @@ -3381,19 +3318,12 @@ "message": "`typing.Dict` is deprecated, use `dict` instead", "count": 1 }, - { - "tool": "ruff", - "path": "connector/platforms/slack_webhook.py", - "code": "UP035", - "message": "`typing.Tuple` is deprecated, use `tuple` instead", - "count": 1 - }, { "tool": "ruff", "path": "connector/platforms/slack_webhook.py", "code": "UP045", "message": "Use `X | None` for type annotations", - "count": 11 + "count": 2 }, { "tool": "ruff", diff --git a/tests/test_r219_exception_boundary_phase2.py b/tests/test_r219_exception_boundary_phase2.py index 3997cf9..2a673c0 100644 --- a/tests/test_r219_exception_boundary_phase2.py +++ b/tests/test_r219_exception_boundary_phase2.py @@ -42,8 +42,9 @@ class TestPolicyV2(unittest.TestCase): "services/route_bootstrap.py", "api/config_projection_handlers.py", "api/config_llm_handlers.py", - "connector/platforms/slack_webhook.py", - "connector/platforms/feishu_webhook.py", + "connector/platforms/slack_installation_handlers.py", + "connector/platforms/slack_ingress_handlers.py", + "connector/platforms/feishu_ingress_handlers.py", }, ) for module in policy["selected_modules"].values(): diff --git a/tests/test_r223_platform_adapter_decomposition.py b/tests/test_r223_platform_adapter_decomposition.py new file mode 100644 index 0000000..f688010 --- /dev/null +++ b/tests/test_r223_platform_adapter_decomposition.py @@ -0,0 +1,97 @@ +"""Contract-first tests for R223 Slack/Feishu adapter decomposition.""" + +from __future__ import annotations + +import importlib.util +import inspect +import json +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] + + +def _verifier(): + path = ROOT / "scripts" / "verify_platform_adapter_contract.py" + spec = importlib.util.spec_from_file_location("r223_adapter_contract", path) + if spec is None or spec.loader is None: + raise RuntimeError("unable to load R223 verifier") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +class TestR223PlatformAdapterDecomposition(unittest.TestCase): + def test_frozen_platform_contract_matches_byte_for_byte(self): + verifier = _verifier() + fixture = (ROOT / "tests" / "platform_adapter_contract_r223.json").read_text( + encoding="utf-8" + ) + self.assertEqual(verifier._canonical_json(verifier.build_contract()), fixture) + + def test_platform_owner_modules_are_substantive_and_one_way(self): + from connector.platforms import ( + feishu_delivery_handlers, + feishu_ingress_handlers, + feishu_installation_handlers, + slack_delivery_handlers, + slack_ingress_handlers, + slack_installation_handlers, + ) + + modules = ( + slack_installation_handlers, + slack_ingress_handlers, + slack_delivery_handlers, + feishu_installation_handlers, + feishu_ingress_handlers, + feishu_delivery_handlers, + ) + for module in modules: + source = inspect.getsource(module) + self.assertGreater(len(source.splitlines()), 40) + self.assertNotIn("from . import slack_webhook", source) + self.assertNotIn("from . import feishu_webhook", source) + self.assertNotIn("import connector.platforms.slack_webhook", source) + self.assertNotIn("import connector.platforms.feishu_webhook", source) + + def test_platforms_do_not_share_protocol_implementation_owner(self): + expected = json.loads( + (ROOT / "tests" / "platform_adapter_contract_r223.json").read_text( + encoding="utf-8" + ) + ) + self.assertNotEqual(expected["slack"]["routes"], expected["feishu"]["routes"]) + + def test_facade_patch_seams_remain_present(self): + from connector.platforms import feishu_webhook, slack_webhook + + expected = json.loads( + (ROOT / "tests" / "platform_adapter_contract_r223.json").read_text( + encoding="utf-8" + ) + ) + for seam in expected["slack"]["patch_seams"]: + self.assertTrue( + callable(getattr(slack_webhook, seam, None)) or seam == "logger" + ) + for seam in expected["feishu"]["patch_seams"]: + self.assertTrue( + callable(getattr(feishu_webhook, seam, None)) or seam == "logger" + ) + + def test_r222_router_contract_digest_is_unchanged(self): + verifier = _verifier() + expected = json.loads( + (ROOT / "tests" / "platform_adapter_contract_r223.json").read_text( + encoding="utf-8" + ) + ) + self.assertEqual( + verifier.build_contract()["router_contract_digest"], + expected["router_contract_digest"], + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_s36s37r79_egress_hardening.py b/tests/test_s36s37r79_egress_hardening.py index 5351ece..abff11f 100644 --- a/tests/test_s36s37r79_egress_hardening.py +++ b/tests/test_s36s37r79_egress_hardening.py @@ -289,7 +289,7 @@ class TestR79EgressCompliance(unittest.TestCase): # IMPORTANT: Keep connector platform adapters in parity here. # Missing a newly-added adapter causes false-positive R79 failures # in full-gate runs even when egress behavior is intentional. - "connector/platforms/slack_webhook.py", + "connector/platforms/slack_delivery_handlers.py", "connector/platforms/slack_socket_mode.py", # Providers "services/providers/anthropic.py",