diff --git a/connector/__main__.py b/connector/__main__.py index 1fecd0c..49e5a0d 100644 --- a/connector/__main__.py +++ b/connector/__main__.py @@ -262,7 +262,7 @@ async def main(): logger.warning("Feishu adapter disabled due to invalid mode config.") else: platforms["feishu"] = feishu_server or feishu_long_clients[0] - if config.feishu_mode == "webhook" and feishu_server is not None: + if feishu_server is not None: await feishu_server.start() for feishu_long_client in feishu_long_clients: await feishu_long_client.start() diff --git a/connector/config.py b/connector/config.py index df66798..6c50246 100644 --- a/connector/config.py +++ b/connector/config.py @@ -135,6 +135,7 @@ class ConnectorConfig: feishu_bind_host: str = "127.0.0.1" feishu_bind_port: int = 8094 feishu_webhook_path: str = "/feishu/events" + feishu_callback_path: str = "/feishu/callback" feishu_domain: str = "feishu" # feishu | lark feishu_mode: str = "websocket" # websocket | webhook feishu_require_mention: bool = True @@ -367,6 +368,9 @@ def load_config() -> ConnectorConfig: cfg.feishu_webhook_path = os.environ.get( "OPENCLAW_CONNECTOR_FEISHU_PATH", "/feishu/events" ) + cfg.feishu_callback_path = os.environ.get( + "OPENCLAW_CONNECTOR_FEISHU_CALLBACK_PATH", "/feishu/callback" + ) cfg.feishu_domain = ( os.environ.get("OPENCLAW_CONNECTOR_FEISHU_DOMAIN", "feishu").strip() or "feishu" ) diff --git a/connector/platforms/feishu_installation_manager.py b/connector/platforms/feishu_installation_manager.py index bf28f06..131c7e6 100644 --- a/connector/platforms/feishu_installation_manager.py +++ b/connector/platforms/feishu_installation_manager.py @@ -158,6 +158,10 @@ class FeishuInstallationManager: def binding_count(self) -> int: return len(self._bindings) + @property + def registry(self): + return self._registry + def bindings(self) -> List[FeishuBinding]: return list(self._bindings.values()) diff --git a/connector/platforms/feishu_webhook.py b/connector/platforms/feishu_webhook.py index 91dad11..7568b9a 100644 --- a/connector/platforms/feishu_webhook.py +++ b/connector/platforms/feishu_webhook.py @@ -42,6 +42,19 @@ except ImportError: # pragma: no cover safe_request_json, ) +try: + from services.connector_callback_contract import ( + CallbackActorContext, + CallbackDecisionCode, + ConnectorCallbackContract, + ) +except ImportError: # pragma: no cover + from services.connector_callback_contract import ( # type: ignore + CallbackActorContext, + CallbackDecisionCode, + ConnectorCallbackContract, + ) + logger = logging.getLogger(__name__) FEISHU_WEBHOOK_MAX_BODY_BYTES = 256 * 1024 @@ -58,6 +71,12 @@ _PLACEHOLDER_TYPES = { "media": "", "sticker": "", } +_FEISHU_CALLBACK_POLICY_MAP = { + "approval.approve": "admin", + "approval.reject": "admin", + "command.status": "public", + "command.run": "run", +} def _import_aiohttp_web(): @@ -213,6 +232,31 @@ def parse_feishu_message_text(message: Dict[str, Any]) -> str: return str(parsed.get("text", "") or "").strip() +def _infer_callback_action_type(command_text: str, button: Dict[str, Any]) -> str: + explicit = str(button.get("action_type", "") or "").strip() + if explicit: + return explicit + normalized = str(command_text or "").strip().lower() + if normalized.startswith("/approve"): + return "approval.approve" + if normalized.startswith("/reject"): + return "approval.reject" + if normalized.startswith("/run"): + return "command.run" + if normalized.startswith("/status"): + return "command.status" + return "command.unknown" + + +def _force_approval_command(command_text: str) -> str: + normalized = str(command_text or "").strip() + if not normalized: + return normalized + if normalized.startswith("/run") and "--approval" not in normalized: + return f"{normalized} --approval" + return normalized + + @dataclass class FeishuDeliveryTarget: channel_id: str @@ -256,6 +300,8 @@ class FeishuWebhookServer: self._tenant_access_token_expires_at: Dict[str, float] = {} self._bot_open_ids: Dict[str, str] = {} self._bot_open_id: str = "" + self._callback_contracts: Dict[str, ConnectorCallbackContract] = {} + self._callback_contract_secrets: Dict[str, str] = {} async def start(self): aiohttp, web = _import_aiohttp_web() @@ -268,13 +314,15 @@ class FeishuWebhookServer: "(OPENCLAW_CONNECTOR_FEISHU_APP_ID / APP_SECRET missing)" ) return - if not any( + has_event_ingress = any( binding.verification_token for binding in self._installation_manager.bindings() - ): + ) + has_callback_ingress = bool(str(self.config.feishu_callback_path or "").strip()) + if not has_event_ingress and not has_callback_ingress: logger.info( "Feishu webhook adapter disabled " - "(OPENCLAW_CONNECTOR_FEISHU_VERIFICATION_TOKEN missing)" + "(verification token and callback path missing)" ) return logger.info( @@ -285,7 +333,13 @@ class FeishuWebhookServer: self.config.feishu_domain, ) self.app = web.Application(client_max_size=FEISHU_WEBHOOK_MAX_BODY_BYTES) - self.app.router.add_post(self.config.feishu_webhook_path, self.handle_event) + if has_event_ingress: + self.app.router.add_post(self.config.feishu_webhook_path, self.handle_event) + if has_callback_ingress: + self.app.router.add_post( + self.config.feishu_callback_path, + self.handle_callback, + ) self.runner = web.AppRunner(self.app) await self.runner.setup() self.site = web.TCPSite( @@ -330,6 +384,29 @@ class FeishuWebhookServer: return _make_response(web, status=400, text=str(exc)) 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: + logger.warning("Feishu callback rejected: %s", exc) + return _make_json_response( + web, + {"ok": False, "error": str(exc)}, + status=403, + ) + return _make_json_response(web, response) + def _is_challenge(self, payload: Dict[str, Any]) -> bool: return bool( payload.get("challenge") @@ -343,6 +420,120 @@ class FeishuWebhookServer: 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 = ( @@ -474,16 +665,17 @@ class FeishuWebhookServer: return response = await self.router.handle(request) resp_text = str(getattr(response, "text", "") or "").strip() - if resp_text: - await self._send_reply( - 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 ""), - ), - resp_text, - ) + 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) def _resolve_delivery_binding( self, *, workspace_id: str = "", account_id: str = "" @@ -493,6 +685,110 @@ class FeishuWebhookServer: 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 + ) + contract.acknowledge_request(envelope_dict.get("request_id", "")) + response = await self.router.handle(request) + contract.complete_request(envelope_dict.get("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, *, @@ -628,6 +924,167 @@ class FeishuWebhookServer: 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) -> None: resolution, binding, _ = self._resolve_delivery_binding( workspace_id=target.workspace_id, diff --git a/connector/router.py b/connector/router.py index 237a369..e5ac9a7 100644 --- a/connector/router.py +++ b/connector/router.py @@ -524,6 +524,7 @@ class CommandRouter: pending_count = res.get("pending_count") lines = [] + buttons = [] for i in items: # IMPORTANT (stability): the backend approval schema uses: # `approval_id`, `template_id`, `status`, `requested_by`, `source`. @@ -539,11 +540,35 @@ class CommandRouter: lines.append( f"- {approval_id} [{status}] template={template_id} by={requested_by} source={source}" ) + for i in items[:3]: + approval_id = i.get("approval_id") or i.get("id") or "unknown" + short_id = str(approval_id)[:8] + buttons.append( + { + "label": f"Approve {short_id}", + "value": f"/approve {approval_id}", + "action_type": "approval.approve", + "approval_id": approval_id, + "style": "primary", + } + ) + buttons.append( + { + "label": f"Reject {short_id}", + "value": f"/reject {approval_id}", + "action_type": "approval.reject", + "approval_id": approval_id, + "style": "danger", + } + ) header = "Pending Approvals" if isinstance(pending_count, int): header += f" ({pending_count})" - return CommandResponse(text=header + ":\n" + "\n".join(lines)) + return CommandResponse( + text=header + ":\n" + "\n".join(lines), + buttons=buttons, + ) async def _handle_approve( self, req: CommandRequest, args: List[str] diff --git a/tests/chat_connector/test_config.py b/tests/chat_connector/test_config.py index c863b23..b0c0cd1 100644 --- a/tests/chat_connector/test_config.py +++ b/tests/chat_connector/test_config.py @@ -53,6 +53,7 @@ class TestConnectorConfig(unittest.TestCase): "OPENCLAW_CONNECTOR_FEISHU_WORKSPACE_ID": "tenant-alpha", "OPENCLAW_CONNECTOR_FEISHU_WORKSPACE_NAME": "Alpha Workspace", "OPENCLAW_CONNECTOR_FEISHU_BINDINGS_JSON": '[{"account_id":"acct-extra","workspace_id":"tenant-beta","app_id":"cli_extra","app_secret":"sec_extra"}]', + "OPENCLAW_CONNECTOR_FEISHU_CALLBACK_PATH": "/feishu/cards", "OPENCLAW_CONNECTOR_FEISHU_ALLOWED_USERS": "u1,u2", "OPENCLAW_CONNECTOR_FEISHU_ALLOWED_CHATS": "oc_a,oc_b", "OPENCLAW_CONNECTOR_FEISHU_DOMAIN": "lark", @@ -71,6 +72,7 @@ class TestConnectorConfig(unittest.TestCase): self.assertEqual(cfg.feishu_workspace_id, "tenant-alpha") self.assertEqual(cfg.feishu_workspace_name, "Alpha Workspace") self.assertIn("acct-extra", cfg.feishu_bindings_json) + self.assertEqual(cfg.feishu_callback_path, "/feishu/cards") self.assertEqual(cfg.feishu_allowed_users, ["u1", "u2"]) self.assertEqual(cfg.feishu_allowed_chats, ["oc_a", "oc_b"]) self.assertEqual(cfg.feishu_domain, "lark") diff --git a/tests/test_f69_feishu_callbacks.py b/tests/test_f69_feishu_callbacks.py new file mode 100644 index 0000000..97ba7ae --- /dev/null +++ b/tests/test_f69_feishu_callbacks.py @@ -0,0 +1,235 @@ +import json +import os +import sys +import tempfile +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from connector.config import ConnectorConfig +from connector.contract import CommandResponse +from connector.platforms.feishu_installation_manager import FeishuInstallationManager +from connector.platforms.feishu_webhook import FeishuDeliveryTarget, FeishuWebhookServer +from services.connector_installation_registry import ConnectorInstallationRegistry +from services.secret_store import SecretStore + + +def _event_payload(): + return { + "schema": "2.0", + "header": { + "event_id": "fe-evt-f69-1", + "event_type": "im.message.receive_v1", + "tenant_key": "tenant-1", + "token": "verify-token", + }, + "event": { + "sender": { + "sender_id": { + "user_id": "u_sender", + "open_id": "ou_sender", + } + }, + "message": { + "message_id": "om_1", + "chat_id": "oc_dm_1", + "chat_type": "p2p", + "message_type": "text", + "content": json.dumps({"text": "/approvals"}), + "mentions": [], + "root_id": "om_root_1", + }, + }, + } + + +class TestF69FeishuCallbacks(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.tmpdir = tempfile.TemporaryDirectory() + self.state_dir = self.tmpdir.name + self.secret_store = SecretStore(state_dir=self.state_dir) + self.registry = ConnectorInstallationRegistry( + state_dir=self.state_dir, + secret_store=self.secret_store, + ) + self.config = ConnectorConfig() + self.config.feishu_app_id = "cli_test" + self.config.feishu_app_secret = "sec_test" + self.config.feishu_account_id = "acct-default" + self.config.feishu_default_account_id = "acct-default" + self.config.feishu_verification_token = "verify-token" + self.config.feishu_workspace_id = "tenant-1" + self.config.feishu_mode = "webhook" + self.config.feishu_callback_path = "/feishu/callback" + self.router = MagicMock() + self.router.handle = AsyncMock(return_value=CommandResponse(text="OK")) + self.router._is_admin = MagicMock(return_value=False) + self.router._is_trusted = MagicMock(return_value=False) + self.manager = FeishuInstallationManager( + self.config, + registry=self.registry, + secret_store=self.secret_store, + state_dir=self.state_dir, + ) + self.server = FeishuWebhookServer( + self.config, + self.router, + installation_manager=self.manager, + ) + self.binding = self.manager.get_binding("acct-default") + + async def asyncTearDown(self): + self.tmpdir.cleanup() + + def _callback_body( + self, + *, + button, + actor_user_id="admin-user", + actor_open_id="ou_admin", + ): + value = self.server._build_card_button_value( + button, + target=FeishuDeliveryTarget( + channel_id="oc_dm_1", + reply_to_message_id="om_card_1", + workspace_id="tenant-1", + account_id="acct-default", + ), + binding=self.binding, + signing_secret=self.binding.app_secret, + ) + return { + "header": {"tenant_key": "tenant-1"}, + "event": { + "open_chat_id": "oc_dm_1", + "open_message_id": "om_card_1", + "operator": { + "operator_id": { + "user_id": actor_user_id, + "open_id": actor_open_id, + } + }, + "action": {"value": value}, + }, + } + + async def test_approvals_response_uses_interactive_card(self): + self.router.handle = AsyncMock( + return_value=CommandResponse( + text="Pending Approvals (1):\n- apr_1 [pending]", + buttons=[ + { + "label": "Approve apr_1", + "value": "/approve apr_1", + "action_type": "approval.approve", + "approval_id": "apr_1", + "style": "primary", + } + ], + ) + ) + with ( + patch.object( + self.server, "_get_tenant_access_token", AsyncMock(return_value="tok_1") + ), + patch( + "connector.platforms.feishu_webhook.safe_request_json", + return_value={"code": 0, "data": {}}, + ) as mock_safe, + ): + await self.server.process_event_payload(_event_payload()) + + kwargs = mock_safe.call_args.kwargs + self.assertEqual(kwargs["json_body"]["msg_type"], "interactive") + card = json.loads(kwargs["json_body"]["content"]) + action = card["elements"][1]["actions"][0] + self.assertEqual(action["text"]["content"], "Approve apr_1") + self.assertEqual( + action["value"]["callback_envelope"]["action_type"], "approval.approve" + ) + self.assertEqual(action["value"]["payload"]["command"], "/approve apr_1") + + async def test_admin_callback_routes_and_duplicate_is_deduped(self): + self.router._is_admin = MagicMock(return_value=True) + self.router._is_trusted = MagicMock(return_value=True) + self.router.handle = AsyncMock( + return_value=CommandResponse(text="[Approved] apr_1") + ) + body = self._callback_body( + button={ + "label": "Approve apr_1", + "value": "/approve apr_1", + "action_type": "approval.approve", + "approval_id": "apr_1", + "style": "primary", + } + ) + + first = await self.server.process_callback_payload(body) + second = await self.server.process_callback_payload(body) + + self.assertTrue(first["ok"]) + self.assertEqual(first["decision_code"], "cb_accept_admin") + req = self.router.handle.call_args.args[0] + self.assertEqual(req.text, "/approve apr_1") + self.assertTrue(second["ok"]) + self.assertTrue(second["duplicate"]) + self.assertEqual(self.router.handle.await_count, 1) + + async def test_run_callback_degrades_to_approval_for_untrusted_actor(self): + self.router._is_admin = MagicMock(return_value=False) + self.router._is_trusted = MagicMock(return_value=False) + self.router.handle = AsyncMock( + return_value=CommandResponse(text="[Approval Requested]") + ) + body = self._callback_body( + button={ + "label": "Run template", + "value": "/run template_x prompt=city", + "action_type": "command.run", + }, + actor_user_id="u_untrusted", + actor_open_id="ou_untrusted", + ) + + response = await self.server.process_callback_payload(body) + + self.assertTrue(response["ok"]) + self.assertEqual(response["decision_code"], "cb_require_approval") + req = self.router.handle.call_args.args[0] + self.assertIn("--approval", req.text) + + async def test_stale_callback_is_rejected(self): + body = self._callback_body( + button={ + "label": "Approve apr_1", + "value": "/approve apr_1", + "action_type": "approval.approve", + "approval_id": "apr_1", + } + ) + envelope = body["event"]["action"]["value"]["callback_envelope"] + envelope["timestamp"] = envelope["timestamp"] - 1000 + + with self.assertRaisesRegex(ValueError, "timestamp_out_of_window"): + await self.server.process_callback_payload(body) + + async def test_invalid_signature_is_rejected(self): + body = self._callback_body( + button={ + "label": "Approve apr_1", + "value": "/approve apr_1", + "action_type": "approval.approve", + "approval_id": "apr_1", + } + ) + body["event"]["action"]["value"]["callback_envelope"]["signature"] = "bad" + + with self.assertRaisesRegex(ValueError, "signature_mismatch"): + await self.server.process_callback_payload(body) + + +if __name__ == "__main__": + unittest.main()