feat(feishu): add scan-to-create onboarding for Feishu / Lark

Add a QR-based onboarding flow to `hermes gateway setup` for Feishu / Lark.
Users scan a QR code with their phone and the platform creates a fully
configured bot application automatically — matching the existing WeChat
QR login experience.

Setup flow:
- Choose between QR scan-to-create (new app) or manual credential input (existing app)
- Connection mode selection (WebSocket / Webhook)
- DM security policy (pairing / open / allowlist / disabled)
- Group chat policy (open with @mention / disabled)

Implementation:
- Onboard functions (init/begin/poll/QR/probe) in gateway/platforms/feishu.py
- _setup_feishu() in hermes_cli/gateway.py with manual fallback
- probe_bot uses lark_oapi SDK when available, raw HTTP fallback otherwise
- qr_register() catches expected errors (network/protocol), propagates bugs
- Poll handles HTTP 4xx JSON responses and feishu/lark domain auto-detection

Tests:
- 25 tests for onboard module (registration, QR, probe, contract, negative paths)
- 16 tests for setup flow (credentials, connection mode, DM policy, group policy,
  adapter integration verifying env vars produce valid FeishuAdapterSettings)

Change-Id: I720591ee84755f32dda95fbac4b26dc82cbcf823
This commit is contained in:
Shuo 2026-04-11 16:57:16 +08:00 committed by Teknium
parent a9ebb331bc
commit d7785f4d5b
5 changed files with 1253 additions and 0 deletions

View file

@ -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

View file

@ -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)

View file

@ -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

View file

@ -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=<choice>, 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"

View file

@ -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.