Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions gateway/platforms/feishu.py
Original file line number Diff line number Diff line change
Expand Up @@ -1514,8 +1514,10 @@ def _load_settings(extra: Dict[str, Any]) -> FeishuAdapterSettings:
connection_mode=str(
extra.get("connection_mode") or os.getenv("FEISHU_CONNECTION_MODE", "websocket")
).strip().lower(),
encrypt_key=os.getenv("FEISHU_ENCRYPT_KEY", "").strip(),
verification_token=os.getenv("FEISHU_VERIFICATION_TOKEN", "").strip(),
encrypt_key=str(extra.get("encrypt_key") or os.getenv("FEISHU_ENCRYPT_KEY", "")).strip(),
verification_token=str(
extra.get("verification_token") or os.getenv("FEISHU_VERIFICATION_TOKEN", "")
).strip(),
group_policy=os.getenv("FEISHU_GROUP_POLICY", "allowlist").strip().lower(),
allowed_group_users=frozenset(
item.strip()
Expand Down Expand Up @@ -1642,6 +1644,11 @@ async def connect(self) -> bool:
self._connection_mode,
)
return False
if self._connection_mode == "webhook" and not (self._verification_token or self._encrypt_key):
logger.error(
"[Feishu] Webhook mode requires FEISHU_VERIFICATION_TOKEN or FEISHU_ENCRYPT_KEY."
)
return False

try:
self._app_lock_identity = self._app_id
Expand Down
33 changes: 33 additions & 0 deletions tests/gateway/test_feishu.py
Original file line number Diff line number Diff line change
Expand Up @@ -3191,6 +3191,39 @@ def test_webhook_request_rejects_bad_signature(self):
response = asyncio.run(adapter._handle_webhook_request(request))
self.assertEqual(response.status, 401)

@patch.dict(os.environ, {}, clear=True)
def test_webhook_connect_requires_inbound_auth_secret(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter

adapter = FeishuAdapter(
PlatformConfig(
enabled=True,
extra={"app_id": "cli_app", "app_secret": "secret_app", "connection_mode": "webhook"},
)
)
self.assertFalse(asyncio.run(adapter.connect()))

@patch.dict(os.environ, {}, clear=True)
def test_webhook_loads_auth_secrets_from_platform_extra(self):
from gateway.config import PlatformConfig
from gateway.platforms.feishu import FeishuAdapter

adapter = FeishuAdapter(
PlatformConfig(
enabled=True,
extra={
"app_id": "cli_app",
"app_secret": "secret_app",
"connection_mode": "webhook",
"verification_token": "token_from_extra",
"encrypt_key": "encrypt_from_extra",
},
)
)
self.assertEqual(adapter._verification_token, "token_from_extra")
self.assertEqual(adapter._encrypt_key, "encrypt_from_extra")

@patch.dict(os.environ, {}, clear=True)
def test_webhook_url_verification_challenge_passes_without_signature(self):
"""Challenge requests must succeed even when no encrypt_key is set."""
Expand Down
Loading