diff --git a/gateway/platforms/feishu.py b/gateway/platforms/feishu.py index 16f5467b2..7fce74def 100644 --- a/gateway/platforms/feishu.py +++ b/gateway/platforms/feishu.py @@ -34,6 +34,9 @@ from datetime import datetime from pathlib import Path from types import SimpleNamespace from typing import Any, Dict, List, Optional +from urllib.error import HTTPError, URLError +from urllib.parse import urlencode +from urllib.request import Request, urlopen # aiohttp/websockets are independent optional deps — import outside lark_oapi # so they remain available for tests and webhook mode even if lark_oapi is missing. @@ -169,6 +172,19 @@ _FEISHU_CARD_ACTION_DEDUP_TTL_SECONDS = 15 * 60 # card action token dedup win _FEISHU_BOT_MSG_TRACK_SIZE = 512 # LRU size for tracking sent message IDs _FEISHU_REPLY_FALLBACK_CODES = frozenset({230011, 231003}) # reply target withdrawn/missing → create fallback _FEISHU_ACK_EMOJI = "OK" + +# QR onboarding constants +_ONBOARD_ACCOUNTS_URLS = { + "feishu": "https://accounts.feishu.cn", + "lark": "https://accounts.larksuite.com", +} +_ONBOARD_OPEN_URLS = { + "feishu": "https://open.feishu.cn", + "lark": "https://open.larksuite.com", +} +_REGISTRATION_PATH = "/oauth/v1/app/registration" +_ONBOARD_REQUEST_TIMEOUT_S = 10 + # --------------------------------------------------------------------------- # Fallback display strings # --------------------------------------------------------------------------- @@ -3621,3 +3637,328 @@ class FeishuAdapter(BasePlatformAdapter): return _FEISHU_FILE_UPLOAD_TYPE, "file" return _FEISHU_FILE_UPLOAD_TYPE, "file" + + +# ============================================================================= +# QR scan-to-create onboarding +# +# Device-code flow: user scans a QR code with Feishu/Lark mobile app and the +# platform creates a fully configured bot application automatically. +# Called by `hermes gateway setup` via _setup_feishu() in hermes_cli/gateway.py. +# ============================================================================= + + +def _accounts_base_url(domain: str) -> str: + return _ONBOARD_ACCOUNTS_URLS.get(domain, _ONBOARD_ACCOUNTS_URLS["feishu"]) + + +def _onboard_open_base_url(domain: str) -> str: + return _ONBOARD_OPEN_URLS.get(domain, _ONBOARD_OPEN_URLS["feishu"]) + + +def _post_registration(base_url: str, body: Dict[str, str]) -> dict: + """POST form-encoded data to the registration endpoint, return parsed JSON. + + The registration endpoint returns JSON even on 4xx (e.g. poll returns + authorization_pending as a 400). We always parse the body regardless of + HTTP status. + """ + url = f"{base_url}{_REGISTRATION_PATH}" + data = urlencode(body).encode("utf-8") + req = Request(url, data=data, headers={"Content-Type": "application/x-www-form-urlencoded"}) + try: + with urlopen(req, timeout=_ONBOARD_REQUEST_TIMEOUT_S) as resp: + return json.loads(resp.read().decode("utf-8")) + except HTTPError as exc: + body_bytes = exc.read() + if body_bytes: + try: + return json.loads(body_bytes.decode("utf-8")) + except (ValueError, json.JSONDecodeError): + raise exc from None + raise + + +def _init_registration(domain: str = "feishu") -> None: + """Verify the environment supports client_secret auth. + + Raises RuntimeError if not supported. + """ + base_url = _accounts_base_url(domain) + res = _post_registration(base_url, {"action": "init"}) + methods = res.get("supported_auth_methods") or [] + if "client_secret" not in methods: + raise RuntimeError( + f"Feishu / Lark registration environment does not support client_secret auth. " + f"Supported: {methods}" + ) + + +def _begin_registration(domain: str = "feishu") -> dict: + """Start the device-code flow. Returns device_code, qr_url, user_code, interval, expire_in.""" + base_url = _accounts_base_url(domain) + res = _post_registration(base_url, { + "action": "begin", + "archetype": "PersonalAgent", + "auth_method": "client_secret", + "request_user_info": "open_id", + }) + device_code = res.get("device_code") + if not device_code: + raise RuntimeError("Feishu / Lark registration did not return a device_code") + qr_url = res.get("verification_uri_complete", "") + if "?" in qr_url: + qr_url += "&from=hermes&tp=hermes" + else: + qr_url += "?from=hermes&tp=hermes" + return { + "device_code": device_code, + "qr_url": qr_url, + "user_code": res.get("user_code", ""), + "interval": res.get("interval") or 5, + "expire_in": res.get("expire_in") or 600, + } + + +def _poll_registration( + *, + device_code: str, + interval: int, + expire_in: int, + domain: str = "feishu", +) -> Optional[dict]: + """Poll until the user scans the QR code, or timeout/denial. + + Returns dict with app_id, app_secret, domain, open_id on success. + Returns None on failure. + """ + deadline = time.time() + expire_in + current_domain = domain + domain_switched = False + poll_count = 0 + + while time.time() < deadline: + base_url = _accounts_base_url(current_domain) + try: + res = _post_registration(base_url, { + "action": "poll", + "device_code": device_code, + "tp": "ob_app", + }) + except (URLError, OSError, json.JSONDecodeError): + time.sleep(interval) + continue + + poll_count += 1 + if poll_count == 1: + print(" Fetching configuration results...", end="", flush=True) + elif poll_count % 6 == 0: + print(".", end="", flush=True) + + # Domain auto-detection + user_info = res.get("user_info") or {} + tenant_brand = user_info.get("tenant_brand") + if tenant_brand == "lark" and not domain_switched: + current_domain = "lark" + domain_switched = True + # Fall through — server may return credentials in this same response. + + # Success + if res.get("client_id") and res.get("client_secret"): + if poll_count > 0: + print() # newline after "Fetching configuration results..." dots + return { + "app_id": res["client_id"], + "app_secret": res["client_secret"], + "domain": current_domain, + "open_id": user_info.get("open_id"), + } + + # Terminal errors + error = res.get("error", "") + if error in ("access_denied", "expired_token"): + if poll_count > 0: + print() + logger.warning("[Feishu onboard] Registration %s", error) + return None + + # authorization_pending or unknown — keep polling + time.sleep(interval) + + if poll_count > 0: + print() + logger.warning("[Feishu onboard] Poll timed out after %ds", expire_in) + return None + + +try: + import qrcode as _qrcode_mod +except (ImportError, TypeError): + _qrcode_mod = None # type: ignore[assignment] + + +def _render_qr(url: str) -> bool: + """Try to render a QR code in the terminal. Returns True if successful.""" + if _qrcode_mod is None: + return False + try: + qr = _qrcode_mod.QRCode() + qr.add_data(url) + qr.make(fit=True) + qr.print_ascii(invert=True) + return True + except Exception: + return False + + +def probe_bot(app_id: str, app_secret: str, domain: str) -> Optional[dict]: + """Verify bot connectivity via /open-apis/bot/v3/info. + + Uses lark_oapi SDK when available, falls back to raw HTTP otherwise. + Returns {"bot_name": ..., "bot_open_id": ...} on success, None on failure. + """ + if FEISHU_AVAILABLE: + return _probe_bot_sdk(app_id, app_secret, domain) + return _probe_bot_http(app_id, app_secret, domain) + + +def _build_onboard_client(app_id: str, app_secret: str, domain: str) -> Any: + """Build a lark Client for the given credentials and domain.""" + sdk_domain = LARK_DOMAIN if domain == "lark" else FEISHU_DOMAIN + return ( + lark.Client.builder() + .app_id(app_id) + .app_secret(app_secret) + .domain(sdk_domain) + .log_level(lark.LogLevel.WARNING) + .build() + ) + + +def _parse_bot_response(data: dict) -> Optional[dict]: + """Extract bot_name and bot_open_id from a /bot/v3/info response.""" + if data.get("code") != 0: + return None + bot = data.get("bot") or data.get("data", {}).get("bot") or {} + return { + "bot_name": bot.get("bot_name"), + "bot_open_id": bot.get("open_id"), + } + + +def _probe_bot_sdk(app_id: str, app_secret: str, domain: str) -> Optional[dict]: + """Probe bot info using lark_oapi SDK.""" + try: + client = _build_onboard_client(app_id, app_secret, domain) + resp = client.request( + method="GET", + url="/open-apis/bot/v3/info", + body=None, + raw_response=True, + ) + return _parse_bot_response(json.loads(resp.content)) + except Exception as exc: + logger.debug("[Feishu onboard] SDK probe failed: %s", exc) + return None + + +def _probe_bot_http(app_id: str, app_secret: str, domain: str) -> Optional[dict]: + """Fallback probe using raw HTTP (when lark_oapi is not installed).""" + base_url = _onboard_open_base_url(domain) + try: + token_data = json.dumps({"app_id": app_id, "app_secret": app_secret}).encode("utf-8") + token_req = Request( + f"{base_url}/open-apis/auth/v3/tenant_access_token/internal", + data=token_data, + headers={"Content-Type": "application/json"}, + ) + with urlopen(token_req, timeout=_ONBOARD_REQUEST_TIMEOUT_S) as resp: + token_res = json.loads(resp.read().decode("utf-8")) + + access_token = token_res.get("tenant_access_token") + if not access_token: + return None + + bot_req = Request( + f"{base_url}/open-apis/bot/v3/info", + headers={ + "Authorization": f"Bearer {access_token}", + "Content-Type": "application/json", + }, + ) + with urlopen(bot_req, timeout=_ONBOARD_REQUEST_TIMEOUT_S) as resp: + bot_res = json.loads(resp.read().decode("utf-8")) + + return _parse_bot_response(bot_res) + except (URLError, OSError, KeyError, json.JSONDecodeError) as exc: + logger.debug("[Feishu onboard] HTTP probe failed: %s", exc) + return None + + +def qr_register( + *, + initial_domain: str = "feishu", + timeout_seconds: int = 600, +) -> Optional[dict]: + """Run the Feishu / Lark scan-to-create QR registration flow. + + Returns on success:: + + { + "app_id": str, + "app_secret": str, + "domain": "feishu" | "lark", + "open_id": str | None, + "bot_name": str | None, + "bot_open_id": str | None, + } + + Returns None on expected failures (network, auth denied, timeout). + Unexpected errors (bugs, protocol regressions) propagate to the caller. + """ + try: + return _qr_register_inner(initial_domain=initial_domain, timeout_seconds=timeout_seconds) + except (RuntimeError, URLError, OSError, json.JSONDecodeError) as exc: + logger.warning("[Feishu onboard] Registration failed: %s", exc) + return None + + +def _qr_register_inner( + *, + initial_domain: str, + timeout_seconds: int, +) -> Optional[dict]: + """Run init → begin → poll → probe. Raises on network/protocol errors.""" + print(" Connecting to Feishu / Lark...", end="", flush=True) + _init_registration(initial_domain) + begin = _begin_registration(initial_domain) + print(" done.") + + print() + qr_url = begin["qr_url"] + if _render_qr(qr_url): + print(f"\n Scan the QR code above, or open this URL directly:\n {qr_url}") + else: + print(f" Open this URL in Feishu / Lark on your phone:\n\n {qr_url}\n") + print(" Tip: pip install qrcode to display a scannable QR code here next time") + print() + + result = _poll_registration( + device_code=begin["device_code"], + interval=begin["interval"], + expire_in=min(begin["expire_in"], timeout_seconds), + domain=initial_domain, + ) + if not result: + return None + + # Probe bot — best-effort, don't fail the registration + bot_info = probe_bot(result["app_id"], result["app_secret"], result["domain"]) + if bot_info: + result["bot_name"] = bot_info.get("bot_name") + result["bot_open_id"] = bot_info.get("bot_open_id") + else: + result["bot_name"] = None + result["bot_open_id"] = None + + return result diff --git a/hermes_cli/gateway.py b/hermes_cli/gateway.py index 908d8992a..a0a4d6735 100644 --- a/hermes_cli/gateway.py +++ b/hermes_cli/gateway.py @@ -2290,6 +2290,183 @@ def _setup_weixin(): print_info(f" User ID: {user_id}") +def _setup_feishu(): + """Interactive setup for Feishu / Lark — scan-to-create or manual credentials.""" + print() + print(color(" ─── 🪽 Feishu / Lark Setup ───", Colors.CYAN)) + + existing_app_id = get_env_value("FEISHU_APP_ID") + existing_secret = get_env_value("FEISHU_APP_SECRET") + if existing_app_id and existing_secret: + print() + print_success("Feishu / Lark is already configured.") + if not prompt_yes_no(" Reconfigure Feishu / Lark?", False): + return + + # ── Choose setup method ── + print() + method_choices = [ + "Scan QR code to create a new bot automatically (recommended)", + "Enter existing App ID and App Secret manually", + ] + method_idx = prompt_choice(" How would you like to set up Feishu / Lark?", method_choices, 0) + + credentials = None + used_qr = False + + if method_idx == 0: + # ── QR scan-to-create ── + try: + from gateway.platforms.feishu import qr_register + except Exception as exc: + print_error(f" Feishu / Lark onboard import failed: {exc}") + qr_register = None + + if qr_register is not None: + try: + credentials = qr_register() + except KeyboardInterrupt: + print() + print_warning(" Feishu / Lark setup cancelled.") + return + except Exception as exc: + print_warning(f" QR registration failed: {exc}") + if credentials: + used_qr = True + if not credentials: + print_info(" QR setup did not complete. Continuing with manual input.") + + # ── Manual credential input ── + if not credentials: + print() + print_info(" Go to https://open.feishu.cn/ (or https://open.larksuite.com/ for Lark)") + print_info(" Create an app, enable the Bot capability, and copy the credentials.") + print() + app_id = prompt(" App ID", password=False) + if not app_id: + print_warning(" Skipped — Feishu / Lark won't work without an App ID.") + return + app_secret = prompt(" App Secret", password=True) + if not app_secret: + print_warning(" Skipped — Feishu / Lark won't work without an App Secret.") + return + + domain_choices = ["feishu (China)", "lark (International)"] + domain_idx = prompt_choice(" Domain", domain_choices, 0) + domain = "lark" if domain_idx == 1 else "feishu" + + # Try to probe the bot with manual credentials + bot_name = None + try: + from gateway.platforms.feishu import probe_bot + bot_info = probe_bot(app_id, app_secret, domain) + if bot_info: + bot_name = bot_info.get("bot_name") + print_success(f" Credentials verified — bot: {bot_name or 'unnamed'}") + else: + print_warning(" Could not verify bot connection. Credentials saved anyway.") + except Exception as exc: + print_warning(f" Credential verification skipped: {exc}") + + credentials = { + "app_id": app_id, + "app_secret": app_secret, + "domain": domain, + "open_id": None, + "bot_name": bot_name, + } + + # ── Save core credentials ── + app_id = credentials["app_id"] + app_secret = credentials["app_secret"] + domain = credentials.get("domain", "feishu") + open_id = credentials.get("open_id") + bot_name = credentials.get("bot_name") + + save_env_value("FEISHU_APP_ID", app_id) + save_env_value("FEISHU_APP_SECRET", app_secret) + save_env_value("FEISHU_DOMAIN", domain) + # Bot identity is resolved at runtime via _hydrate_bot_identity(). + + # ── Connection mode ── + if used_qr: + connection_mode = "websocket" + else: + print() + mode_choices = [ + "WebSocket (recommended — no public URL needed)", + "Webhook (requires a reachable HTTP endpoint)", + ] + mode_idx = prompt_choice(" Connection mode", mode_choices, 0) + connection_mode = "webhook" if mode_idx == 1 else "websocket" + if connection_mode == "webhook": + print_info(" Webhook defaults: 127.0.0.1:8765/feishu/webhook") + print_info(" Override with FEISHU_WEBHOOK_HOST / FEISHU_WEBHOOK_PORT / FEISHU_WEBHOOK_PATH") + print_info(" For signature verification, set FEISHU_ENCRYPT_KEY and FEISHU_VERIFICATION_TOKEN") + save_env_value("FEISHU_CONNECTION_MODE", connection_mode) + + if bot_name: + print() + print_success(f" Bot created: {bot_name}") + + # ── DM security policy ── + print() + access_choices = [ + "Use DM pairing approval (recommended)", + "Allow all direct messages", + "Only allow listed user IDs", + "Disable direct messages", + ] + access_idx = prompt_choice(" How should direct messages be authorized?", access_choices, 0) + if access_idx == 0: + save_env_value("FEISHU_ALLOW_ALL_USERS", "false") + save_env_value("FEISHU_ALLOWED_USERS", "") + print_success(" DM pairing enabled.") + print_info(" Unknown users can request access; approve with `hermes pairing approve`.") + elif access_idx == 1: + save_env_value("FEISHU_ALLOW_ALL_USERS", "true") + save_env_value("FEISHU_ALLOWED_USERS", "") + print_warning(" Open DM access enabled for Feishu / Lark.") + elif access_idx == 2: + save_env_value("FEISHU_ALLOW_ALL_USERS", "false") + default_allow = open_id or "" + allowlist = prompt(" Allowed user IDs (comma-separated)", default_allow, password=False).replace(" ", "") + save_env_value("FEISHU_ALLOWED_USERS", allowlist) + print_success(" Allowlist saved.") + else: + save_env_value("FEISHU_ALLOW_ALL_USERS", "false") + save_env_value("FEISHU_ALLOWED_USERS", "") + print_warning(" Direct messages disabled.") + + # ── Group policy ── + print() + group_choices = [ + "Respond only when @mentioned in groups (recommended)", + "Disable group chats", + ] + group_idx = prompt_choice(" How should group chats be handled?", group_choices, 0) + if group_idx == 0: + save_env_value("FEISHU_GROUP_POLICY", "open") + print_info(" Group chats enabled (bot must be @mentioned).") + else: + save_env_value("FEISHU_GROUP_POLICY", "disabled") + print_info(" Group chats disabled.") + + # ── Home channel ── + print() + home_channel = prompt(" Home chat ID (optional, for cron/notifications)", password=False) + if home_channel: + save_env_value("FEISHU_HOME_CHANNEL", home_channel) + print_success(f" Home channel set to {home_channel}") + + print() + print_success("🪽 Feishu / Lark configured!") + print_info(f" App ID: {app_id}") + print_info(f" Domain: {domain}") + if bot_name: + print_info(f" Bot: {bot_name}") + + def _setup_signal(): """Interactive setup for Signal messenger.""" import shutil @@ -2467,6 +2644,8 @@ def gateway_setup(): _setup_signal() elif platform["key"] == "weixin": _setup_weixin() + elif platform["key"] == "feishu": + _setup_feishu() else: _setup_standard_platform(platform) diff --git a/tests/gateway/test_feishu_onboard.py b/tests/gateway/test_feishu_onboard.py new file mode 100644 index 000000000..cb998fa5a --- /dev/null +++ b/tests/gateway/test_feishu_onboard.py @@ -0,0 +1,436 @@ +"""Tests for gateway.platforms.feishu — Feishu scan-to-create registration.""" + +import json +from unittest.mock import patch, MagicMock +import pytest + + +def _mock_urlopen(response_data, status=200): + """Create a mock for urllib.request.urlopen that returns JSON response_data.""" + mock_response = MagicMock() + mock_response.read.return_value = json.dumps(response_data).encode("utf-8") + mock_response.status = status + mock_response.__enter__ = lambda s: s + mock_response.__exit__ = MagicMock(return_value=False) + return mock_response + + +class TestPostRegistration: + """Tests for the low-level HTTP helper.""" + + @patch("gateway.platforms.feishu.urlopen") + def test_post_registration_returns_parsed_json(self, mock_urlopen_fn): + from gateway.platforms.feishu import _post_registration + + mock_urlopen_fn.return_value = _mock_urlopen({"nonce": "abc", "supported_auth_methods": ["client_secret"]}) + result = _post_registration("https://accounts.feishu.cn", {"action": "init"}) + assert result["nonce"] == "abc" + assert "client_secret" in result["supported_auth_methods"] + + @patch("gateway.platforms.feishu.urlopen") + def test_post_registration_sends_form_encoded_body(self, mock_urlopen_fn): + from gateway.platforms.feishu import _post_registration + + mock_urlopen_fn.return_value = _mock_urlopen({}) + _post_registration("https://accounts.feishu.cn", {"action": "init", "key": "val"}) + call_args = mock_urlopen_fn.call_args + request = call_args[0][0] + body = request.data.decode("utf-8") + assert "action=init" in body + assert "key=val" in body + assert request.get_header("Content-type") == "application/x-www-form-urlencoded" + + +class TestInitRegistration: + """Tests for the init step.""" + + @patch("gateway.platforms.feishu.urlopen") + def test_init_succeeds_when_client_secret_supported(self, mock_urlopen_fn): + from gateway.platforms.feishu import _init_registration + + mock_urlopen_fn.return_value = _mock_urlopen({ + "nonce": "abc", + "supported_auth_methods": ["client_secret"], + }) + _init_registration("feishu") + + @patch("gateway.platforms.feishu.urlopen") + def test_init_raises_when_client_secret_not_supported(self, mock_urlopen_fn): + from gateway.platforms.feishu import _init_registration + + mock_urlopen_fn.return_value = _mock_urlopen({ + "nonce": "abc", + "supported_auth_methods": ["other_method"], + }) + with pytest.raises(RuntimeError, match="client_secret"): + _init_registration("feishu") + + @patch("gateway.platforms.feishu.urlopen") + def test_init_uses_lark_url_for_lark_domain(self, mock_urlopen_fn): + from gateway.platforms.feishu import _init_registration + + mock_urlopen_fn.return_value = _mock_urlopen({ + "nonce": "abc", + "supported_auth_methods": ["client_secret"], + }) + _init_registration("lark") + call_args = mock_urlopen_fn.call_args + request = call_args[0][0] + assert "larksuite.com" in request.full_url + + +class TestBeginRegistration: + """Tests for the begin step.""" + + @patch("gateway.platforms.feishu.urlopen") + def test_begin_returns_device_code_and_qr_url(self, mock_urlopen_fn): + from gateway.platforms.feishu import _begin_registration + + mock_urlopen_fn.return_value = _mock_urlopen({ + "device_code": "dc_123", + "verification_uri_complete": "https://accounts.feishu.cn/qr/abc", + "user_code": "ABCD-1234", + "interval": 5, + "expire_in": 600, + }) + result = _begin_registration("feishu") + assert result["device_code"] == "dc_123" + assert "qr_url" in result + assert "accounts.feishu.cn" in result["qr_url"] + assert result["user_code"] == "ABCD-1234" + assert result["interval"] == 5 + assert result["expire_in"] == 600 + + @patch("gateway.platforms.feishu.urlopen") + def test_begin_sends_correct_archetype(self, mock_urlopen_fn): + from gateway.platforms.feishu import _begin_registration + + mock_urlopen_fn.return_value = _mock_urlopen({ + "device_code": "dc_123", + "verification_uri_complete": "https://example.com/qr", + "user_code": "X", + "interval": 5, + "expire_in": 600, + }) + _begin_registration("feishu") + request = mock_urlopen_fn.call_args[0][0] + body = request.data.decode("utf-8") + assert "archetype=PersonalAgent" in body + assert "auth_method=client_secret" in body + + +class TestPollRegistration: + """Tests for the poll step.""" + + @patch("gateway.platforms.feishu.time") + @patch("gateway.platforms.feishu.urlopen") + def test_poll_returns_credentials_on_success(self, mock_urlopen_fn, mock_time): + from gateway.platforms.feishu import _poll_registration + + mock_time.time.side_effect = [0, 1] + mock_time.sleep = MagicMock() + + mock_urlopen_fn.return_value = _mock_urlopen({ + "client_id": "cli_app123", + "client_secret": "secret456", + "user_info": {"open_id": "ou_owner", "tenant_brand": "feishu"}, + }) + result = _poll_registration( + device_code="dc_123", interval=1, expire_in=60, domain="feishu" + ) + assert result is not None + assert result["app_id"] == "cli_app123" + assert result["app_secret"] == "secret456" + assert result["domain"] == "feishu" + assert result["open_id"] == "ou_owner" + + @patch("gateway.platforms.feishu.time") + @patch("gateway.platforms.feishu.urlopen") + def test_poll_switches_domain_on_lark_tenant_brand(self, mock_urlopen_fn, mock_time): + from gateway.platforms.feishu import _poll_registration + + mock_time.time.side_effect = [0, 1, 2] + mock_time.sleep = MagicMock() + + pending_resp = _mock_urlopen({ + "error": "authorization_pending", + "user_info": {"tenant_brand": "lark"}, + }) + success_resp = _mock_urlopen({ + "client_id": "cli_lark", + "client_secret": "secret_lark", + "user_info": {"open_id": "ou_lark", "tenant_brand": "lark"}, + }) + mock_urlopen_fn.side_effect = [pending_resp, success_resp] + + result = _poll_registration( + device_code="dc_123", interval=0, expire_in=60, domain="feishu" + ) + assert result is not None + assert result["domain"] == "lark" + + @patch("gateway.platforms.feishu.time") + @patch("gateway.platforms.feishu.urlopen") + def test_poll_success_with_lark_brand_in_same_response(self, mock_urlopen_fn, mock_time): + """Credentials and lark tenant_brand in one response must not be discarded.""" + from gateway.platforms.feishu import _poll_registration + + mock_time.time.side_effect = [0, 1] + mock_time.sleep = MagicMock() + + mock_urlopen_fn.return_value = _mock_urlopen({ + "client_id": "cli_lark_direct", + "client_secret": "secret_lark_direct", + "user_info": {"open_id": "ou_lark_direct", "tenant_brand": "lark"}, + }) + result = _poll_registration( + device_code="dc_123", interval=1, expire_in=60, domain="feishu" + ) + assert result is not None + assert result["app_id"] == "cli_lark_direct" + assert result["domain"] == "lark" + assert result["open_id"] == "ou_lark_direct" + + @patch("gateway.platforms.feishu.time") + @patch("gateway.platforms.feishu.urlopen") + def test_poll_returns_none_on_access_denied(self, mock_urlopen_fn, mock_time): + from gateway.platforms.feishu import _poll_registration + + mock_time.time.side_effect = [0, 1] + mock_time.sleep = MagicMock() + + mock_urlopen_fn.return_value = _mock_urlopen({ + "error": "access_denied", + }) + result = _poll_registration( + device_code="dc_123", interval=1, expire_in=60, domain="feishu" + ) + assert result is None + + @patch("gateway.platforms.feishu.time") + @patch("gateway.platforms.feishu.urlopen") + def test_poll_returns_none_on_timeout(self, mock_urlopen_fn, mock_time): + from gateway.platforms.feishu import _poll_registration + + mock_time.time.side_effect = [0, 999] + mock_time.sleep = MagicMock() + + mock_urlopen_fn.return_value = _mock_urlopen({ + "error": "authorization_pending", + }) + result = _poll_registration( + device_code="dc_123", interval=1, expire_in=1, domain="feishu" + ) + assert result is None + + +class TestRenderQr: + """Tests for QR code terminal rendering.""" + + @patch("gateway.platforms.feishu._qrcode_mod", create=True) + def test_render_qr_returns_true_on_success(self, mock_qrcode_mod): + from gateway.platforms.feishu import _render_qr + + mock_qr = MagicMock() + mock_qrcode_mod.QRCode.return_value = mock_qr + assert _render_qr("https://example.com/qr") is True + mock_qr.add_data.assert_called_once_with("https://example.com/qr") + mock_qr.make.assert_called_once_with(fit=True) + mock_qr.print_ascii.assert_called_once() + + def test_render_qr_returns_false_when_qrcode_missing(self): + from gateway.platforms.feishu import _render_qr + + with patch("gateway.platforms.feishu._qrcode_mod", None): + assert _render_qr("https://example.com/qr") is False + + +class TestProbeBot: + """Tests for bot connectivity verification.""" + + def test_probe_returns_bot_info_on_success(self): + from gateway.platforms.feishu import probe_bot + + with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk: + mock_sdk.return_value = {"bot_name": "TestBot", "bot_open_id": "ou_bot123"} + result = probe_bot("cli_app", "secret", "feishu") + + assert result is not None + assert result["bot_name"] == "TestBot" + assert result["bot_open_id"] == "ou_bot123" + + def test_probe_returns_none_on_failure(self): + from gateway.platforms.feishu import probe_bot + + with patch("gateway.platforms.feishu._probe_bot_sdk") as mock_sdk: + mock_sdk.return_value = None + result = probe_bot("bad_id", "bad_secret", "feishu") + + assert result is None + + @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False) + @patch("gateway.platforms.feishu.urlopen") + def test_http_fallback_when_sdk_unavailable(self, mock_urlopen_fn): + """Without lark_oapi, probe falls back to raw HTTP.""" + from gateway.platforms.feishu import probe_bot + + token_resp = _mock_urlopen({"code": 0, "tenant_access_token": "t-123"}) + bot_resp = _mock_urlopen({"code": 0, "bot": {"bot_name": "HttpBot", "open_id": "ou_http"}}) + mock_urlopen_fn.side_effect = [token_resp, bot_resp] + + result = probe_bot("cli_app", "secret", "feishu") + assert result is not None + assert result["bot_name"] == "HttpBot" + + @patch("gateway.platforms.feishu.FEISHU_AVAILABLE", False) + @patch("gateway.platforms.feishu.urlopen") + def test_http_fallback_returns_none_on_network_error(self, mock_urlopen_fn): + from gateway.platforms.feishu import probe_bot + from urllib.error import URLError + + mock_urlopen_fn.side_effect = URLError("connection refused") + result = probe_bot("cli_app", "secret", "feishu") + assert result is None + + +class TestQrRegister: + """Tests for the public qr_register entry point.""" + + @patch("gateway.platforms.feishu.probe_bot") + @patch("gateway.platforms.feishu._render_qr") + @patch("gateway.platforms.feishu._poll_registration") + @patch("gateway.platforms.feishu._begin_registration") + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_success_flow( + self, mock_init, mock_begin, mock_poll, mock_render, mock_probe + ): + from gateway.platforms.feishu import qr_register + + mock_begin.return_value = { + "device_code": "dc_123", + "qr_url": "https://example.com/qr", + "user_code": "ABCD", + "interval": 1, + "expire_in": 60, + } + mock_poll.return_value = { + "app_id": "cli_app", + "app_secret": "secret", + "domain": "feishu", + "open_id": "ou_owner", + } + mock_probe.return_value = {"bot_name": "MyBot", "bot_open_id": "ou_bot"} + + result = qr_register() + assert result is not None + assert result["app_id"] == "cli_app" + assert result["app_secret"] == "secret" + assert result["bot_name"] == "MyBot" + mock_init.assert_called_once() + mock_render.assert_called_once() + + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_returns_none_on_init_failure(self, mock_init): + from gateway.platforms.feishu import qr_register + + mock_init.side_effect = RuntimeError("not supported") + result = qr_register() + assert result is None + + @patch("gateway.platforms.feishu._render_qr") + @patch("gateway.platforms.feishu._poll_registration") + @patch("gateway.platforms.feishu._begin_registration") + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_returns_none_on_poll_failure( + self, mock_init, mock_begin, mock_poll, mock_render + ): + from gateway.platforms.feishu import qr_register + + mock_begin.return_value = { + "device_code": "dc_123", + "qr_url": "https://example.com/qr", + "user_code": "ABCD", + "interval": 1, + "expire_in": 60, + } + mock_poll.return_value = None + + result = qr_register() + assert result is None + + # -- Contract: expected errors → None, unexpected errors → propagate -- + + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_returns_none_on_network_error(self, mock_init): + """URLError (network down) is an expected failure → None.""" + from gateway.platforms.feishu import qr_register + from urllib.error import URLError + + mock_init.side_effect = URLError("DNS resolution failed") + result = qr_register() + assert result is None + + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_returns_none_on_json_error(self, mock_init): + """Malformed server response is an expected failure → None.""" + from gateway.platforms.feishu import qr_register + + mock_init.side_effect = json.JSONDecodeError("bad json", "", 0) + result = qr_register() + assert result is None + + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_propagates_unexpected_errors(self, mock_init): + """Bugs (e.g. AttributeError) must not be swallowed — they propagate.""" + from gateway.platforms.feishu import qr_register + + mock_init.side_effect = AttributeError("some internal bug") + with pytest.raises(AttributeError, match="some internal bug"): + qr_register() + + # -- Negative paths: partial/malformed server responses -- + + @patch("gateway.platforms.feishu._render_qr") + @patch("gateway.platforms.feishu._begin_registration") + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_returns_none_when_begin_missing_device_code( + self, mock_init, mock_begin, mock_render + ): + """Server returns begin response without device_code → RuntimeError → None.""" + from gateway.platforms.feishu import qr_register + + mock_begin.side_effect = RuntimeError("Feishu registration did not return a device_code") + result = qr_register() + assert result is None + + @patch("gateway.platforms.feishu.probe_bot") + @patch("gateway.platforms.feishu._render_qr") + @patch("gateway.platforms.feishu._poll_registration") + @patch("gateway.platforms.feishu._begin_registration") + @patch("gateway.platforms.feishu._init_registration") + def test_qr_register_succeeds_even_when_probe_fails( + self, mock_init, mock_begin, mock_poll, mock_render, mock_probe + ): + """Registration succeeds but probe fails → result with bot_name=None.""" + from gateway.platforms.feishu import qr_register + + mock_begin.return_value = { + "device_code": "dc_123", + "qr_url": "https://example.com/qr", + "user_code": "ABCD", + "interval": 1, + "expire_in": 60, + } + mock_poll.return_value = { + "app_id": "cli_app", + "app_secret": "secret", + "domain": "feishu", + "open_id": "ou_owner", + } + mock_probe.return_value = None # probe failed + + result = qr_register() + assert result is not None + assert result["app_id"] == "cli_app" + assert result["bot_name"] is None + assert result["bot_open_id"] is None diff --git a/tests/gateway/test_setup_feishu.py b/tests/gateway/test_setup_feishu.py new file mode 100644 index 000000000..0b977cde9 --- /dev/null +++ b/tests/gateway/test_setup_feishu.py @@ -0,0 +1,284 @@ +"""Tests for _setup_feishu() in hermes_cli/gateway.py. + +Verifies that the interactive setup writes env vars that correctly drive the +Feishu adapter: credentials, connection mode, DM policy, and group policy. +""" + +import os +from unittest.mock import patch + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _run_setup_feishu( + *, + qr_result=None, + prompt_yes_no_responses=None, + prompt_choice_responses=None, + prompt_responses=None, + existing_env=None, +): + """Run _setup_feishu() with mocked I/O and return the env vars that were saved. + + Returns a dict of {env_var_name: value} for all save_env_value calls. + """ + existing_env = existing_env or {} + prompt_yes_no_responses = list(prompt_yes_no_responses or [True]) + # QR path: method(0), dm(0), group(0) — 3 choices (no connection mode) + # Manual path: method(1), domain(0), connection(0), dm(0), group(0) — 5 choices + prompt_choice_responses = list(prompt_choice_responses or [0, 0, 0]) + prompt_responses = list(prompt_responses or [""]) + + saved_env = {} + + def mock_save(name, value): + saved_env[name] = value + + def mock_get(name): + return existing_env.get(name, "") + + with patch("hermes_cli.gateway.save_env_value", side_effect=mock_save), \ + patch("hermes_cli.gateway.get_env_value", side_effect=mock_get), \ + patch("hermes_cli.gateway.prompt_yes_no", side_effect=prompt_yes_no_responses), \ + patch("hermes_cli.gateway.prompt_choice", side_effect=prompt_choice_responses), \ + patch("hermes_cli.gateway.prompt", side_effect=prompt_responses), \ + patch("hermes_cli.gateway.print_info"), \ + patch("hermes_cli.gateway.print_success"), \ + patch("hermes_cli.gateway.print_warning"), \ + patch("hermes_cli.gateway.print_error"), \ + patch("hermes_cli.gateway.color", side_effect=lambda t, c: t), \ + patch("gateway.platforms.feishu.qr_register", return_value=qr_result): + + from hermes_cli.gateway import _setup_feishu + _setup_feishu() + + return saved_env + + +# --------------------------------------------------------------------------- +# QR scan-to-create path +# --------------------------------------------------------------------------- + +class TestSetupFeishuQrPath: + """Tests for the QR scan-to-create happy path.""" + + def test_qr_success_saves_core_credentials(self): + env = _run_setup_feishu( + qr_result={ + "app_id": "cli_test", + "app_secret": "secret_test", + "domain": "feishu", + "open_id": "ou_owner", + "bot_name": "TestBot", + "bot_open_id": "ou_bot", + }, + prompt_yes_no_responses=[True], # Start QR + prompt_choice_responses=[0, 0, 0], # method=QR, dm=pairing, group=open + prompt_responses=[""], # home channel: skip + ) + assert env["FEISHU_APP_ID"] == "cli_test" + assert env["FEISHU_APP_SECRET"] == "secret_test" + assert env["FEISHU_DOMAIN"] == "feishu" + + def test_qr_success_does_not_persist_bot_identity(self): + """Bot identity is discovered at runtime by _hydrate_bot_identity — not persisted + in env, so it stays fresh if the user renames the bot later.""" + env = _run_setup_feishu( + qr_result={ + "app_id": "cli_test", + "app_secret": "secret_test", + "domain": "feishu", + "open_id": "ou_owner", + "bot_name": "TestBot", + "bot_open_id": "ou_bot", + }, + prompt_yes_no_responses=[True], + prompt_choice_responses=[0, 0, 0], + prompt_responses=[""], + ) + assert "FEISHU_BOT_OPEN_ID" not in env + assert "FEISHU_BOT_NAME" not in env + + +# --------------------------------------------------------------------------- +# Connection mode +# --------------------------------------------------------------------------- + +class TestSetupFeishuConnectionMode: + """Connection mode: QR always websocket, manual path lets user choose.""" + + def test_qr_path_defaults_to_websocket(self): + env = _run_setup_feishu( + qr_result={ + "app_id": "cli_test", "app_secret": "s", "domain": "feishu", + "open_id": None, "bot_name": None, "bot_open_id": None, + }, + prompt_choice_responses=[0, 0, 0], # method=QR, dm=pairing, group=open + prompt_responses=[""], + ) + assert env["FEISHU_CONNECTION_MODE"] == "websocket" + + @patch("gateway.platforms.feishu.probe_bot", return_value=None) + def test_manual_path_websocket(self, _mock_probe): + env = _run_setup_feishu( + qr_result=None, + prompt_choice_responses=[1, 0, 0, 0, 0], # method=manual, domain=feishu, connection=ws, dm=pairing, group=open + prompt_responses=["cli_manual", "secret_manual", ""], # app_id, app_secret, home_channel + ) + assert env["FEISHU_CONNECTION_MODE"] == "websocket" + + @patch("gateway.platforms.feishu.probe_bot", return_value=None) + def test_manual_path_webhook(self, _mock_probe): + env = _run_setup_feishu( + qr_result=None, + prompt_choice_responses=[1, 0, 1, 0, 0], # method=manual, domain=feishu, connection=webhook, dm=pairing, group=open + prompt_responses=["cli_manual", "secret_manual", ""], # app_id, app_secret, home_channel + ) + assert env["FEISHU_CONNECTION_MODE"] == "webhook" + + +# --------------------------------------------------------------------------- +# DM security policy +# --------------------------------------------------------------------------- + +class TestSetupFeishuDmPolicy: + """DM policy must use platform-scoped FEISHU_ALLOW_ALL_USERS, not the global flag.""" + + def _run_with_dm_choice(self, dm_choice_idx, prompt_responses=None): + return _run_setup_feishu( + qr_result={ + "app_id": "cli_test", "app_secret": "s", "domain": "feishu", + "open_id": "ou_owner", "bot_name": None, "bot_open_id": None, + }, + prompt_yes_no_responses=[True], + prompt_choice_responses=[0, dm_choice_idx, 0], # method=QR, dm=, group=open + prompt_responses=prompt_responses or [""], + ) + + def test_pairing_sets_feishu_allow_all_false(self): + env = self._run_with_dm_choice(0) + assert env["FEISHU_ALLOW_ALL_USERS"] == "false" + assert env["FEISHU_ALLOWED_USERS"] == "" + assert "GATEWAY_ALLOW_ALL_USERS" not in env + + def test_allow_all_sets_feishu_allow_all_true(self): + env = self._run_with_dm_choice(1) + assert env["FEISHU_ALLOW_ALL_USERS"] == "true" + assert env["FEISHU_ALLOWED_USERS"] == "" + assert "GATEWAY_ALLOW_ALL_USERS" not in env + + def test_allowlist_sets_feishu_allow_all_false_with_list(self): + env = self._run_with_dm_choice(2, prompt_responses=["ou_user1,ou_user2", ""]) + assert env["FEISHU_ALLOW_ALL_USERS"] == "false" + assert env["FEISHU_ALLOWED_USERS"] == "ou_user1,ou_user2" + assert "GATEWAY_ALLOW_ALL_USERS" not in env + + def test_allowlist_prepopulates_with_scan_owner_open_id(self): + """When open_id is available from QR scan, it should be the default allowlist value.""" + # We return the owner's open_id from prompt (+ empty home channel). + env = self._run_with_dm_choice(2, prompt_responses=["ou_owner", ""]) + assert env["FEISHU_ALLOWED_USERS"] == "ou_owner" + + def test_disabled_sets_feishu_allow_all_false(self): + env = self._run_with_dm_choice(3) + assert env["FEISHU_ALLOW_ALL_USERS"] == "false" + assert env["FEISHU_ALLOWED_USERS"] == "" + assert "GATEWAY_ALLOW_ALL_USERS" not in env + + +# --------------------------------------------------------------------------- +# Group policy +# --------------------------------------------------------------------------- + +class TestSetupFeishuGroupPolicy: + + def test_open_with_mention(self): + env = _run_setup_feishu( + qr_result={ + "app_id": "cli_test", "app_secret": "s", "domain": "feishu", + "open_id": None, "bot_name": None, "bot_open_id": None, + }, + prompt_yes_no_responses=[True], + prompt_choice_responses=[0, 0, 0], # method=QR, dm=pairing, group=open + prompt_responses=[""], + ) + assert env["FEISHU_GROUP_POLICY"] == "open" + + def test_disabled(self): + env = _run_setup_feishu( + qr_result={ + "app_id": "cli_test", "app_secret": "s", "domain": "feishu", + "open_id": None, "bot_name": None, "bot_open_id": None, + }, + prompt_yes_no_responses=[True], + prompt_choice_responses=[0, 0, 1], # method=QR, dm=pairing, group=disabled + prompt_responses=[""], + ) + assert env["FEISHU_GROUP_POLICY"] == "disabled" + + +# --------------------------------------------------------------------------- +# Adapter integration: env vars → FeishuAdapterSettings +# --------------------------------------------------------------------------- + +class TestSetupFeishuAdapterIntegration: + """Verify that env vars written by _setup_feishu() produce a valid adapter config. + + This bridges the gap between 'setup wrote the right env vars' and + 'the adapter will actually initialize correctly from those vars'. + """ + + def _make_env_from_setup(self, dm_idx=0, group_idx=0): + """Run _setup_feishu via QR path and return the env vars it would write.""" + return _run_setup_feishu( + qr_result={ + "app_id": "cli_test_app", + "app_secret": "test_secret_value", + "domain": "feishu", + "open_id": "ou_owner", + "bot_name": "IntegrationBot", + "bot_open_id": "ou_bot_integration", + }, + prompt_yes_no_responses=[True], + prompt_choice_responses=[0, dm_idx, group_idx], # method=QR, dm, group + prompt_responses=[""], + ) + + @patch.dict(os.environ, {}, clear=True) + def test_qr_env_produces_valid_adapter_settings(self): + """QR setup → adapter initializes with websocket mode.""" + env = self._make_env_from_setup() + + with patch.dict(os.environ, env, clear=True): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + adapter = FeishuAdapter(PlatformConfig()) + assert adapter._app_id == "cli_test_app" + assert adapter._app_secret == "test_secret_value" + assert adapter._domain_name == "feishu" + assert adapter._connection_mode == "websocket" + + @patch.dict(os.environ, {}, clear=True) + def test_open_dm_env_sets_correct_adapter_state(self): + """Setup with 'allow all DMs' → adapter sees allow-all flag.""" + env = self._make_env_from_setup(dm_idx=1) + + with patch.dict(os.environ, env, clear=True): + from gateway.platforms.feishu import FeishuAdapter + from gateway.config import PlatformConfig + # Verify adapter initializes without error and env var is correct. + FeishuAdapter(PlatformConfig()) + assert os.getenv("FEISHU_ALLOW_ALL_USERS") == "true" + + @patch.dict(os.environ, {}, clear=True) + def test_group_open_env_sets_adapter_group_policy(self): + """Setup with 'open groups' → adapter group_policy is 'open'.""" + env = self._make_env_from_setup(group_idx=0) + + with patch.dict(os.environ, env, clear=True): + from gateway.config import PlatformConfig + from gateway.platforms.feishu import FeishuAdapter + adapter = FeishuAdapter(PlatformConfig()) + assert adapter._group_policy == "open" diff --git a/website/docs/user-guide/messaging/feishu.md b/website/docs/user-guide/messaging/feishu.md index ac4bad239..4d9783d40 100644 --- a/website/docs/user-guide/messaging/feishu.md +++ b/website/docs/user-guide/messaging/feishu.md @@ -31,12 +31,25 @@ Set it to `false` only if you explicitly want one shared conversation per chat. ## Step 1: Create a Feishu / Lark App +### Recommended: Scan-to-Create (one command) + +```bash +hermes gateway setup +``` + +Select **Feishu / Lark** and scan the QR code with your Feishu or Lark mobile app. Hermes will automatically create a bot application with the correct permissions and save the credentials. + +### Alternative: Manual Setup + +If scan-to-create is not available, the wizard falls back to manual input: + 1. Open the Feishu or Lark developer console: - Feishu: [https://open.feishu.cn/](https://open.feishu.cn/) - Lark: [https://open.larksuite.com/](https://open.larksuite.com/) 2. Create a new app. 3. In **Credentials & Basic Info**, copy the **App ID** and **App Secret**. 4. Enable the **Bot** capability for the app. +5. Run `hermes gateway setup`, select **Feishu / Lark**, and enter the credentials when prompted. :::warning Keep the App Secret private. Anyone with it can impersonate your app.