tjb-tech
489 lines
15 KiB
Python
489 lines
15 KiB
Python
"""Security regressions for Feishu/Lark channel media handling."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from openharness.channels.bus.queue import MessageBus
|
|
from openharness.channels.bus.events import OutboundMessage
|
|
from openharness.channels.impl.feishu import FeishuChannel, _extract_feishu_mentions, _feishu_mentions_bot
|
|
from openharness.config.schema import FeishuConfig
|
|
from ohmo.group_registry import save_managed_group_record
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_inbound_file_attachment_cannot_escape_media_dir(tmp_path: Path, monkeypatch):
|
|
"""Remote Feishu filenames are metadata and must not be trusted as paths."""
|
|
workspace = tmp_path / "ohmo"
|
|
workspace.mkdir()
|
|
protected_file = workspace / "soul.md"
|
|
protected_file.write_text("ORIGINAL", encoding="utf-8")
|
|
monkeypatch.setenv("OHMO_WORKSPACE", str(workspace))
|
|
|
|
channel = FeishuChannel(
|
|
FeishuConfig(allow_from=["user-open-id"], react_emoji="eyes"), MessageBus()
|
|
)
|
|
|
|
def fake_download(message_id: str, file_key: str, resource_type: str = "file"):
|
|
assert message_id == "message-id"
|
|
assert file_key == "file-key"
|
|
return b"ATTACKER_OVERWRITE", "../../soul.md"
|
|
|
|
async def fake_add_reaction(*args, **kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(channel, "_download_file_sync", fake_download)
|
|
monkeypatch.setattr(channel, "_add_reaction", fake_add_reaction)
|
|
monkeypatch.setattr(
|
|
channel, "_resolve_sender_display_name_sync", lambda sender_id: "Allowed User"
|
|
)
|
|
|
|
forwarded = {}
|
|
|
|
async def fake_handle_message(**kwargs):
|
|
forwarded.update(kwargs)
|
|
|
|
monkeypatch.setattr(channel, "_handle_message", fake_handle_message)
|
|
|
|
event = SimpleNamespace(
|
|
sender=SimpleNamespace(
|
|
sender_type="user",
|
|
sender_id=SimpleNamespace(open_id="user-open-id"),
|
|
),
|
|
message=SimpleNamespace(
|
|
message_id="message-id",
|
|
chat_id="chat-id",
|
|
chat_type="p2p",
|
|
message_type="file",
|
|
content=json.dumps({"file_key": "file-key"}),
|
|
),
|
|
)
|
|
|
|
await channel._on_message(SimpleNamespace(event=event))
|
|
|
|
media_dir = workspace / "attachments" / "feishu"
|
|
saved_paths = [Path(path).resolve() for path in forwarded["media"]]
|
|
assert protected_file.read_text(encoding="utf-8") == "ORIGINAL"
|
|
assert saved_paths == [(media_dir / "soul.md").resolve()]
|
|
assert saved_paths[0].read_bytes() == b"ATTACKER_OVERWRITE"
|
|
assert saved_paths[0].is_relative_to(media_dir.resolve())
|
|
assert "../../" not in forwarded["content"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_send_does_not_reply_in_thread_for_private_messages(monkeypatch):
|
|
channel = FeishuChannel(FeishuConfig(), MessageBus())
|
|
channel._client = object()
|
|
sent: list[str | None] = []
|
|
|
|
def fake_send(*args):
|
|
sent.append(args[-1])
|
|
return True
|
|
|
|
monkeypatch.setattr(channel, "_send_message_sync", fake_send)
|
|
|
|
await channel.send(
|
|
OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="ou_private",
|
|
content="hello",
|
|
metadata={"chat_type": "p2p", "message_id": "om_private"},
|
|
)
|
|
)
|
|
|
|
assert sent == [None]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_send_replies_in_thread_for_group_messages(monkeypatch):
|
|
channel = FeishuChannel(FeishuConfig(), MessageBus())
|
|
channel._client = object()
|
|
sent: list[str | None] = []
|
|
|
|
def fake_send(*args):
|
|
sent.append(args[-1])
|
|
return True
|
|
|
|
monkeypatch.setattr(channel, "_send_message_sync", fake_send)
|
|
|
|
await channel.send(
|
|
OutboundMessage(
|
|
channel="feishu",
|
|
chat_id="oc_group",
|
|
content="hello",
|
|
metadata={"chat_type": "group", "message_id": "om_group"},
|
|
)
|
|
)
|
|
|
|
assert sent == ["om_group"]
|
|
|
|
|
|
def test_feishu_extracts_text_mentions_and_matches_bot_name():
|
|
content = {
|
|
"text": "@_user_1 帮我看看",
|
|
"mentions": [
|
|
{
|
|
"key": "@_user_1",
|
|
"id": {"open_id": "ou_bot"},
|
|
"name": "ohmo",
|
|
}
|
|
],
|
|
}
|
|
|
|
assert _extract_feishu_mentions(content) == [
|
|
{"key": "@_user_1", "name": "ohmo", "open_id": "ou_bot", "user_id": "", "union_id": ""}
|
|
]
|
|
assert _feishu_mentions_bot(content, content["text"], FeishuConfig(bot_names=["ohmo"])) is True
|
|
|
|
|
|
def test_feishu_extracts_sdk_message_mentions():
|
|
mention = SimpleNamespace(
|
|
key="@_user_1",
|
|
id=SimpleNamespace(open_id="ou_bot", user_id="user_bot", union_id=""),
|
|
name="ohmo",
|
|
)
|
|
|
|
assert _extract_feishu_mentions({"text": "@_user_1 帮我看看"}, [mention]) == [
|
|
{
|
|
"key": "@_user_1",
|
|
"name": "ohmo",
|
|
"open_id": "ou_bot",
|
|
"user_id": "user_bot",
|
|
"union_id": "",
|
|
}
|
|
]
|
|
|
|
|
|
def test_feishu_mention_detection_can_use_bot_open_id():
|
|
content = {
|
|
"text": "@_user_1 帮我看看",
|
|
"mentions": [
|
|
{
|
|
"key": "@_user_1",
|
|
"id": {"open_id": "ou_exact_bot"},
|
|
"name": "Different Display Name",
|
|
}
|
|
],
|
|
}
|
|
|
|
assert _feishu_mentions_bot(
|
|
content,
|
|
content["text"],
|
|
FeishuConfig(bot_open_id="ou_exact_bot", bot_names=["ohmo"]),
|
|
) is True
|
|
|
|
|
|
def test_feishu_mention_detection_ignores_other_users():
|
|
content = {
|
|
"text": "@_user_1 帮我看看",
|
|
"mentions": [
|
|
{
|
|
"key": "@_user_1",
|
|
"id": {"open_id": "ou_other"},
|
|
"name": "Alice",
|
|
}
|
|
],
|
|
}
|
|
|
|
assert _feishu_mentions_bot(content, content["text"], FeishuConfig(bot_names=["ohmo"])) is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_group_policy_ignores_unmentioned_unmanaged_group_without_reaction(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
workspace = tmp_path / "ohmo"
|
|
workspace.mkdir()
|
|
monkeypatch.setenv("OHMO_WORKSPACE", str(workspace))
|
|
channel = FeishuChannel(
|
|
FeishuConfig(
|
|
allow_from=["user-open-id"],
|
|
react_emoji="OK",
|
|
group_policy="managed_or_mention",
|
|
bot_names=["ohmo"],
|
|
),
|
|
MessageBus(),
|
|
)
|
|
reactions: list[str] = []
|
|
forwarded: list[dict] = []
|
|
|
|
async def fake_add_reaction(message_id: str, emoji_type: str = "OK") -> None:
|
|
reactions.append(f"{message_id}:{emoji_type}")
|
|
|
|
async def fake_handle_message(**kwargs):
|
|
forwarded.append(kwargs)
|
|
|
|
monkeypatch.setattr(channel, "_add_reaction", fake_add_reaction)
|
|
monkeypatch.setattr(channel, "_handle_message", fake_handle_message)
|
|
monkeypatch.setattr(channel, "_resolve_sender_display_name_sync", lambda sender_id: "Allowed User")
|
|
|
|
event = SimpleNamespace(
|
|
sender=SimpleNamespace(
|
|
sender_type="user",
|
|
sender_id=SimpleNamespace(open_id="user-open-id"),
|
|
),
|
|
message=SimpleNamespace(
|
|
message_id="message-unmanaged",
|
|
chat_id="oc_unmanaged",
|
|
chat_type="group",
|
|
message_type="text",
|
|
content=json.dumps({"text": "这个普通群消息不应该触发"}),
|
|
),
|
|
)
|
|
|
|
await channel._on_message(SimpleNamespace(event=event))
|
|
|
|
assert reactions == []
|
|
assert forwarded == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_group_policy_allows_managed_group_without_mention(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
workspace = tmp_path / "ohmo"
|
|
workspace.mkdir()
|
|
monkeypatch.setenv("OHMO_WORKSPACE", str(workspace))
|
|
save_managed_group_record(
|
|
workspace=workspace,
|
|
channel="feishu",
|
|
chat_id="oc_managed",
|
|
owner_open_id="user-open-id",
|
|
name="Managed Group",
|
|
)
|
|
channel = FeishuChannel(
|
|
FeishuConfig(
|
|
allow_from=["user-open-id"],
|
|
react_emoji="OK",
|
|
group_policy="managed_or_mention",
|
|
bot_names=["ohmo"],
|
|
),
|
|
MessageBus(),
|
|
)
|
|
reactions: list[str] = []
|
|
forwarded: list[dict] = []
|
|
|
|
async def fake_add_reaction(message_id: str, emoji_type: str = "OK") -> None:
|
|
reactions.append(f"{message_id}:{emoji_type}")
|
|
|
|
async def fake_handle_message(**kwargs):
|
|
forwarded.append(kwargs)
|
|
|
|
monkeypatch.setattr(channel, "_add_reaction", fake_add_reaction)
|
|
monkeypatch.setattr(channel, "_handle_message", fake_handle_message)
|
|
monkeypatch.setattr(channel, "_resolve_sender_display_name_sync", lambda sender_id: "Allowed User")
|
|
|
|
event = SimpleNamespace(
|
|
sender=SimpleNamespace(
|
|
sender_type="user",
|
|
sender_id=SimpleNamespace(open_id="user-open-id"),
|
|
),
|
|
message=SimpleNamespace(
|
|
message_id="message-managed",
|
|
chat_id="oc_managed",
|
|
chat_type="group",
|
|
message_type="text",
|
|
content=json.dumps({"text": "managed 群不用 @ 也应该触发"}),
|
|
),
|
|
)
|
|
|
|
await channel._on_message(SimpleNamespace(event=event))
|
|
|
|
assert reactions == ["message-managed:OK"]
|
|
assert len(forwarded) == 1
|
|
assert forwarded[0]["metadata"]["mentions_bot"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_group_policy_allows_mentioned_unmanaged_group(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
workspace = tmp_path / "ohmo"
|
|
workspace.mkdir()
|
|
monkeypatch.setenv("OHMO_WORKSPACE", str(workspace))
|
|
channel = FeishuChannel(
|
|
FeishuConfig(
|
|
allow_from=["user-open-id"],
|
|
react_emoji="OK",
|
|
group_policy="managed_or_mention",
|
|
bot_names=["ohmo"],
|
|
),
|
|
MessageBus(),
|
|
)
|
|
reactions: list[str] = []
|
|
forwarded: list[dict] = []
|
|
|
|
async def fake_add_reaction(message_id: str, emoji_type: str = "OK") -> None:
|
|
reactions.append(f"{message_id}:{emoji_type}")
|
|
|
|
async def fake_handle_message(**kwargs):
|
|
forwarded.append(kwargs)
|
|
|
|
monkeypatch.setattr(channel, "_add_reaction", fake_add_reaction)
|
|
monkeypatch.setattr(channel, "_handle_message", fake_handle_message)
|
|
monkeypatch.setattr(channel, "_resolve_sender_display_name_sync", lambda sender_id: "Allowed User")
|
|
|
|
event = SimpleNamespace(
|
|
sender=SimpleNamespace(
|
|
sender_type="user",
|
|
sender_id=SimpleNamespace(open_id="user-open-id"),
|
|
),
|
|
message=SimpleNamespace(
|
|
message_id="message-mentioned",
|
|
chat_id="oc_unmanaged",
|
|
chat_type="group",
|
|
message_type="text",
|
|
content=json.dumps(
|
|
{
|
|
"text": "@_user_1 帮我看看",
|
|
"mentions": [
|
|
{
|
|
"key": "@_user_1",
|
|
"id": {"open_id": "ou_bot"},
|
|
"name": "ohmo",
|
|
}
|
|
],
|
|
},
|
|
ensure_ascii=False,
|
|
),
|
|
),
|
|
)
|
|
|
|
await channel._on_message(SimpleNamespace(event=event))
|
|
|
|
assert reactions == ["message-mentioned:OK"]
|
|
assert len(forwarded) == 1
|
|
assert forwarded[0]["metadata"]["mentions_bot"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_group_policy_allows_sdk_message_mention(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
workspace = tmp_path / "ohmo"
|
|
workspace.mkdir()
|
|
monkeypatch.setenv("OHMO_WORKSPACE", str(workspace))
|
|
channel = FeishuChannel(
|
|
FeishuConfig(
|
|
allow_from=["user-open-id"],
|
|
react_emoji="OK",
|
|
group_policy="managed_or_mention",
|
|
bot_names=["ohmo"],
|
|
),
|
|
MessageBus(),
|
|
)
|
|
reactions: list[str] = []
|
|
forwarded: list[dict] = []
|
|
|
|
async def fake_add_reaction(message_id: str, emoji_type: str = "OK") -> None:
|
|
reactions.append(f"{message_id}:{emoji_type}")
|
|
|
|
async def fake_handle_message(**kwargs):
|
|
forwarded.append(kwargs)
|
|
|
|
monkeypatch.setattr(channel, "_add_reaction", fake_add_reaction)
|
|
monkeypatch.setattr(channel, "_handle_message", fake_handle_message)
|
|
monkeypatch.setattr(channel, "_resolve_sender_display_name_sync", lambda sender_id: "Allowed User")
|
|
|
|
event = SimpleNamespace(
|
|
sender=SimpleNamespace(
|
|
sender_type="user",
|
|
sender_id=SimpleNamespace(open_id="user-open-id"),
|
|
),
|
|
message=SimpleNamespace(
|
|
message_id="message-sdk-mentioned",
|
|
chat_id="oc_unmanaged",
|
|
chat_type="group",
|
|
message_type="text",
|
|
content=json.dumps({"text": "@_user_1 帮我看看"}, ensure_ascii=False),
|
|
mentions=[
|
|
SimpleNamespace(
|
|
key="@_user_1",
|
|
id=SimpleNamespace(open_id="ou_bot", user_id="", union_id=""),
|
|
name="ohmo",
|
|
)
|
|
],
|
|
),
|
|
)
|
|
|
|
await channel._on_message(SimpleNamespace(event=event))
|
|
|
|
assert reactions == ["message-sdk-mentioned:OK"]
|
|
assert len(forwarded) == 1
|
|
assert forwarded[0]["metadata"]["mentions_bot"] is True
|
|
assert forwarded[0]["metadata"]["mentions"][0]["open_id"] == "ou_bot"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_create_managed_group_builds_expected_request():
|
|
channel = FeishuChannel(FeishuConfig(), MessageBus())
|
|
captured = {}
|
|
|
|
class FakeChat:
|
|
def create(self, request):
|
|
captured["request"] = request
|
|
return SimpleNamespace(
|
|
success=lambda: True,
|
|
data=SimpleNamespace(chat_id="oc_new_group"),
|
|
)
|
|
|
|
channel._client = SimpleNamespace(im=SimpleNamespace(v1=SimpleNamespace(chat=FakeChat())))
|
|
|
|
chat_id = await channel.create_managed_group(user_open_id="ou_user", name="OpenHarness 讨论群")
|
|
|
|
request = captured["request"]
|
|
assert chat_id == "oc_new_group"
|
|
assert request.user_id_type == "open_id"
|
|
assert request.set_bot_manager is True
|
|
assert request.body.name == "OpenHarness 讨论群"
|
|
assert request.body.user_id_list == ["ou_user"]
|
|
assert request.body.chat_mode == "group"
|
|
assert request.body.chat_type == "private"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_create_managed_group_reports_api_failure():
|
|
channel = FeishuChannel(FeishuConfig(), MessageBus())
|
|
|
|
class FakeChat:
|
|
def create(self, request):
|
|
return SimpleNamespace(
|
|
success=lambda: False,
|
|
code=99991663,
|
|
msg="missing scope",
|
|
get_log_id=lambda: "log-1",
|
|
)
|
|
|
|
channel._client = SimpleNamespace(im=SimpleNamespace(v1=SimpleNamespace(chat=FakeChat())))
|
|
|
|
with pytest.raises(RuntimeError, match="99991663.*missing scope.*log-1"):
|
|
await channel.create_managed_group(user_open_id="ou_user", name="OpenHarness 讨论群")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_feishu_rename_group_builds_expected_request():
|
|
channel = FeishuChannel(FeishuConfig(), MessageBus())
|
|
captured = {}
|
|
|
|
class FakeChat:
|
|
def update(self, request):
|
|
captured["request"] = request
|
|
return SimpleNamespace(success=lambda: True)
|
|
|
|
channel._client = SimpleNamespace(im=SimpleNamespace(v1=SimpleNamespace(chat=FakeChat())))
|
|
|
|
await channel.rename_group(chat_id="oc_group", name="New Name")
|
|
|
|
request = captured["request"]
|
|
assert request.user_id_type == "open_id"
|
|
assert request.chat_id == "oc_group"
|
|
assert request.body.name == "New Name"
|