diff --git a/backend/package/yuxi/channels/adapters/msteams/__init__.py b/backend/package/yuxi/channels/adapters/msteams/__init__.py index a201a6c5..88e6f26e 100644 --- a/backend/package/yuxi/channels/adapters/msteams/__init__.py +++ b/backend/package/yuxi/channels/adapters/msteams/__init__.py @@ -15,8 +15,7 @@ from yuxi.channels.adapters.msteams.cards import ( wrap_as_attachment, ) from yuxi.channels.adapters.msteams.chunking import chunk_text -from yuxi.channels.adapters.msteams.command_gate import CommandGate -from yuxi.channels.adapters.msteams.command_gate import resolve_dual_text_control_command_gate +from yuxi.channels.adapters.msteams.command_gate import CommandGate, resolve_dual_text_control_command_gate from yuxi.channels.adapters.msteams.commands import ( SLASH_COMMANDS, build_command_help_card, @@ -48,11 +47,11 @@ from yuxi.channels.adapters.msteams.feedback import ( ) from yuxi.channels.adapters.msteams.file_upload import ( PendingUploadStore, + _validate_upload_url, build_file_consent_card, build_teams_file_info_card, needs_file_consent, upload_to_sharepoint, - _validate_upload_url, ) from yuxi.channels.adapters.msteams.graph import ( GraphClient, @@ -119,10 +118,10 @@ from yuxi.channels.adapters.msteams.secret_input import ( ) from yuxi.channels.adapters.msteams.security import SecurityPolicy from yuxi.channels.adapters.msteams.send import ( + SILENT_REPLY_TOKEN, + MessageSender, MSTeamsErrorCategory, MSTeamsErrorInfo, - MessageSender, - SILENT_REPLY_TOKEN, classify_msteams_send_error, format_send_error_hint, is_silent_reply_text, @@ -144,12 +143,12 @@ from yuxi.channels.adapters.msteams.sso import ( is_verify_state, ) from yuxi.channels.adapters.msteams.streaming import ( + _STREAM_INFORMATIVE_TEXTS, ReplyStreamController, StreamManager, StreamPhase, StreamState, TeamsHttpStream, - _STREAM_INFORMATIVE_TEXTS, ) from yuxi.channels.adapters.msteams.tenant import TenantValidator from yuxi.channels.adapters.msteams.tool_policy import ToolPolicy, resolve_tool_policy diff --git a/backend/package/yuxi/channels/adapters/msteams/adapter.py b/backend/package/yuxi/channels/adapters/msteams/adapter.py index 16d74b8b..0f1ea259 100644 --- a/backend/package/yuxi/channels/adapters/msteams/adapter.py +++ b/backend/package/yuxi/channels/adapters/msteams/adapter.py @@ -6,6 +6,7 @@ from __future__ import annotations +import asyncio import base64 import logging import os @@ -21,10 +22,11 @@ from jwt import PyJWKClient from yuxi.channels.base import BaseChannelAdapter from yuxi.channels.capabilities import ChannelCapabilities -from yuxi.channels.meta import ChannelMeta from yuxi.channels.exceptions import ( ChannelAuthenticationError, ) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta from yuxi.channels.models import ( ChannelIdentity, ChannelMessage, @@ -46,16 +48,20 @@ from .credentials import DelegatedAuthStore from .debounce import DebounceManager from .feedback import build_feedback_channel_data from .formatter import format_outbound -from .normalizer import normalize_inbound, normalize_conversation_update, normalize_invoke from .invoke_handler import normalize_reaction +from .normalizer import normalize_conversation_update, normalize_inbound, normalize_invoke from .polls import PollStore -from .probe import MSTeamsProbe from .proactive import ( ConversationStore, +) +from .proactive import ( proactive_send as _proactive_send, ) +from .probe import MSTeamsProbe from .security import SecurityPolicy -from .send import MessageSender, send_adaptive_card as _send_adaptive_card, send_media as _send_stream_media +from .send import MessageSender +from .send import send_adaptive_card as _send_adaptive_card +from .send import send_media as _send_stream_media from .sent_message_cache import SentMessageCache from .sso import SSOHandler from .streaming import StreamManager @@ -69,6 +75,8 @@ _DEDUP_WINDOW_SECONDS = 300 _MAX_WEBHOOK_BODY_BYTES = 256 * 1024 _WEBHOOK_HANDLE_TIMEOUT_SECONDS = 30 _WEBHOOK_SIGNATURE_TIMEOUT_SECONDS = 10 +MEDIA_GRAPH_UPLOAD_THRESHOLD = 512 * 1024 +MEDIA_BASE64_MAX_SIZE = 1024 * 1024 BOT_FRAMEWORK_DOMAINS = { "smba.trafficmanager.net", @@ -139,6 +147,7 @@ class MSTeamsAdapter(BaseChannelAdapter): self._stream_mgr = StreamManager() self._dedup_ids: dict[str, float] = {} self._dedup_ttl = self.config.get("dedup_ttl", _DEDUP_WINDOW_SECONDS) + self._dedup_lock = asyncio.Lock() self._http_session: aiohttp.ClientSession | None = None self._jwks_client: PyJWKClient | None = None self._security_policy = SecurityPolicy(self.config) @@ -156,6 +165,7 @@ class MSTeamsAdapter(BaseChannelAdapter): self._conv_store = ConversationStore() self._delegated_auth_store = DelegatedAuthStore() self._sent_cache = SentMessageCache() + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60, channel_id="msteams") async def _get_http_session(self) -> aiohttp.ClientSession: if self._http_session is None or self._http_session.closed: @@ -218,6 +228,7 @@ class MSTeamsAdapter(BaseChannelAdapter): self._streaming_mode = self.config.get("streaming_mode", "block") self._status = ChannelStatus.CONNECTED + self._sent_cache.start_cleanup_task() logger.info(f"MSTeams bot '{self._app_id[:8]}...' connected") async def disconnect(self) -> None: @@ -240,6 +251,7 @@ class MSTeamsAdapter(BaseChannelAdapter): await self._conv_store.flush() await self._poll_store.flush() + self._sent_cache.stop_cleanup_task() self._jwks_client = None self._stream_mgr.clear() @@ -252,45 +264,53 @@ class MSTeamsAdapter(BaseChannelAdapter): if not_connected: return not_connected - conversation_id = self._resolve_conversation_id(response) - chunks = chunk_text(response.content, self.text_chunk_limit) - if not chunks: - return DeliveryResult(success=False, error="Empty content") + async def _do_send() -> DeliveryResult: + conversation_id = self._resolve_conversation_id(response) + chunks = chunk_text(response.content, self.text_chunk_limit) + if not chunks: + raise ChannelAuthenticationError("Empty content") - activity = format_outbound(response, self.text_chunk_limit) - activity["text"] = chunks[0] + activity = format_outbound(response, self.text_chunk_limit) + activity["text"] = chunks[0] - if self._feedback_enabled: - fb_channel_data = build_feedback_channel_data( - feedback_enabled=self._feedback_enabled, - feedback_reflection=self._feedback_reflection, - ) - if fb_channel_data: - existing_cd = activity.get("channelData", {}) or {} - activity["channelData"] = {**existing_cd, **fb_channel_data} + if self._feedback_enabled: + fb_channel_data = build_feedback_channel_data( + feedback_enabled=self._feedback_enabled, + feedback_reflection=self._feedback_reflection, + ) + if fb_channel_data: + existing_cd = activity.get("channelData", {}) or {} + activity["channelData"] = {**existing_cd, **fb_channel_data} + + result = await self._sender.send_activity(conversation_id, activity) + if not result.success or not result.message_id: + raise ChannelAuthenticationError(f"Send failed: {result.error}") + + self._sent_cache.record(result.message_id, conversation_id) + + if len(chunks) <= 1: + return result + + accumulated_text = chunks[0] + for chunk in chunks[1:]: + accumulated_text += chunk + if len(accumulated_text) > self.text_chunk_limit: + accumulated_text = accumulated_text[: self.text_chunk_limit] + edit_activity = { + "type": "message", + "text": accumulated_text, + "textFormat": "markdown", + } + await self._sender.update_activity(conversation_id, result.message_id, edit_activity) - result = await self._sender.send_activity(conversation_id, activity) - if not result.success or not result.message_id: return result - self._sent_cache.record(result.message_id, conversation_id) - - if len(chunks) <= 1: - return result - - accumulated_text = chunks[0] - for chunk in chunks[1:]: - accumulated_text += chunk - if len(accumulated_text) > self.text_chunk_limit: - accumulated_text = accumulated_text[: self.text_chunk_limit] - edit_activity = { - "type": "message", - "text": accumulated_text, - "textFormat": "markdown", - } - await self._sender.update_activity(conversation_id, result.message_id, edit_activity) - - return result + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + except Exception as e: + return DeliveryResult(success=False, error=str(e)) async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: not_connected = self._require_connected() @@ -306,8 +326,15 @@ class MSTeamsAdapter(BaseChannelAdapter): mime_type = content_type_map.get(media_type, "application/octet-stream") if isinstance(data, bytes): + file_size = len(data) token = self._sender.token if self._sender else None - if token and len(data) > 1024 * 1024: + + if file_size > MEDIA_GRAPH_UPLOAD_THRESHOLD: + if not token: + return DeliveryResult( + success=False, + error=f"File too large ({file_size} bytes) and no token available for Graph upload", + ) from .graph import GraphClient client = GraphClient(token) @@ -317,12 +344,19 @@ class MSTeamsAdapter(BaseChannelAdapter): web_url = upload_result.get("webUrl", "") if web_url: return await _send_stream_media(self._sender, chat_id, web_url, mime_type) + return DeliveryResult(success=False, error="Graph upload: no webUrl returned") except Exception as e: logger.error(f"MSTeams media upload via Graph failed: {e}") return DeliveryResult(success=False, error=f"Graph upload failed: {e}") finally: await client.close() + if file_size > MEDIA_BASE64_MAX_SIZE: + return DeliveryResult( + success=False, + error=f"File too large for base64 encoding ({file_size} > {MEDIA_BASE64_MAX_SIZE})", + ) + content_url = f"data:{mime_type};base64,{base64.b64encode(data).decode()}" elif isinstance(data, str): content_url = data @@ -643,7 +677,8 @@ class MSTeamsAdapter(BaseChannelAdapter): if activity_type == "message": msg_id = body.get("id", "") - if self._is_duplicate(msg_id): + if await self._check_and_mark(msg_id): + logger.debug(f"MSTeams: duplicate message {msg_id}, skipping") return None if msg_id and self._sent_cache.was_sent(msg_id): @@ -686,6 +721,21 @@ class MSTeamsAdapter(BaseChannelAdapter): channel_msg = self._classify_message(channel_msg) self._track_message(channel_msg) + + if self.config.get("thread_context_enabled", True): + try: + token = self._sender.token if self._sender else None + if token: + from .graph import GraphClient + from .normalizer import inject_thread_context + + graph_client = GraphClient(token) + session_key = (channel_msg.metadata or {}).get("SessionKey", "") + channel_msg = await inject_thread_context(channel_msg, graph_client, session_key) + await graph_client.close() + except Exception: + pass + return channel_msg if activity_type == "conversationUpdate": @@ -865,20 +915,19 @@ class MSTeamsAdapter(BaseChannelAdapter): self._message_tracker[chat_id] = msg_id while len(self._message_tracker) > _MAX_MESSAGE_TRACKER_SIZE: self._message_tracker.popitem(last=False) - self._mark_seen(msg_id) - def _is_duplicate(self, msg_id: str) -> bool: + async def _check_and_mark(self, msg_id: str) -> bool: if not msg_id: return False - now = time.monotonic() - expired = [mid for mid, ts in self._dedup_ids.items() if now - ts > self._dedup_ttl] - for mid in expired: - self._dedup_ids.pop(mid, None) - return msg_id in self._dedup_ids - - def _mark_seen(self, msg_id: str) -> None: - if msg_id: - self._dedup_ids[msg_id] = time.monotonic() + async with self._dedup_lock: + now = time.monotonic() + expired = [mid for mid, ts in self._dedup_ids.items() if now - ts > self._dedup_ttl] + for mid in expired: + self._dedup_ids.pop(mid, None) + if msg_id in self._dedup_ids: + return True + self._dedup_ids[msg_id] = now + return False def _resolve_conversation_id(self, response: ChannelResponse) -> str: return response.identity.channel_chat_id diff --git a/backend/package/yuxi/channels/adapters/msteams/connection_modes.py b/backend/package/yuxi/channels/adapters/msteams/connection_modes.py new file mode 100644 index 00000000..5a4b37d6 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/msteams/connection_modes.py @@ -0,0 +1,285 @@ +"""Microsoft Teams 连接模式多样化。 + +支持 Webhook (默认)、WebSocket (Bot Framework Streaming Extensions)、 +Polling 三种连接模式，可通过配置项 connection_mode 切换。 +""" + +from __future__ import annotations + +import asyncio +import json +from typing import Any + +import aiohttp + +from yuxi.utils.logging_config import logger + +CONNECTION_MODE_WEBHOOK = "webhook" +CONNECTION_MODE_WEBSOCKET = "websocket" +CONNECTION_MODE_POLLING = "polling" + +VALID_CONNECTION_MODES = {CONNECTION_MODE_WEBHOOK, CONNECTION_MODE_WEBSOCKET, CONNECTION_MODE_POLLING} + +BOT_STREAMING_URL_TEMPLATE = "https://directline.botframework.com/v3/directline/conversations/{conversation_id}/stream" +BOT_ACTIVITIES_URL_TEMPLATE = "https://smba.trafficmanager.net/emea/v3/conversations/{conversation_id}/activities" + +POLLING_DEFAULT_INTERVAL_S = 2.0 +POLLING_MAX_INTERVAL_S = 10.0 +WEBSOCKET_PING_INTERVAL_S = 30.0 +WEBSOCKET_PONG_TIMEOUT_S = 10.0 + + +class WebSocketClient: + """Bot Framework Streaming Extensions WebSocket 客户端。 + + 提供实时双向连接，延迟低于 Webhook 模式。 + """ + + def __init__( + self, + app_id: str, + app_password: str, + stream_url: str = "", + ping_interval: float = WEBSOCKET_PING_INTERVAL_S, + pong_timeout: float = WEBSOCKET_PONG_TIMEOUT_S, + ): + self._app_id = app_id + self._app_password = app_password + self._stream_url = stream_url + self._ping_interval = ping_interval + self._pong_timeout = pong_timeout + self._ws: aiohttp.ClientWebSocketResponse | None = None + self._session: aiohttp.ClientSession | None = None + self._running = False + self._on_message: Any = None + + def set_message_handler(self, handler: Any) -> None: + self._on_message = handler + + async def connect(self, conversation_id: str = "") -> None: + if self._running: + return + + self._session = aiohttp.ClientSession() + stream_url = self._stream_url or BOT_STREAMING_URL_TEMPLATE.format(conversation_id=conversation_id) + + headers = { + "Authorization": f"Bearer {self._app_password}", + "Content-Type": "application/json", + } + + self._ws = await self._session.ws_connect(stream_url, headers=headers, heartbeat=self._ping_interval) + self._running = True + logger.info(f"MSTeams WebSocket connected: {stream_url}") + + asyncio.create_task(self._read_loop()) + + async def _read_loop(self) -> None: + while self._running and self._ws is not None: + try: + msg = await self._ws.receive(timeout=self._pong_timeout) + if msg.type == aiohttp.WSMsgType.TEXT: + try: + data = json.loads(msg.data) + activities = data.get("activities", []) + for activity in activities: + if self._on_message: + await self._on_message(activity) + except json.JSONDecodeError: + logger.warning("MSTeams WebSocket: invalid JSON received") + elif msg.type == aiohttp.WSMsgType.CLOSED: + logger.info("MSTeams WebSocket closed by server") + break + elif msg.type == aiohttp.WSMsgType.ERROR: + logger.error(f"MSTeams WebSocket error: {self._ws.exception()}") + break + except TimeoutError: + continue + except Exception as e: + logger.error(f"MSTeams WebSocket read error: {e}") + await asyncio.sleep(1) + + if self._running: + logger.info("MSTeams WebSocket disconnected, attempting reconnect...") + await asyncio.sleep(2) + if self._running: + asyncio.create_task(self._reconnect()) + + async def _reconnect(self) -> None: + try: + await self.close() + await asyncio.sleep(3) + await self.connect() + except Exception as e: + logger.error(f"MSTeams WebSocket reconnect failed: {e}") + + async def send_activity(self, activity: dict[str, Any]) -> bool: + if not self._ws or self._ws.closed: + return False + try: + data = json.dumps(activity) + await self._ws.send_str(data) + return True + except Exception as e: + logger.error(f"MSTeams WebSocket send error: {e}") + return False + + async def close(self) -> None: + self._running = False + if self._ws and not self._ws.closed: + await self._ws.close() + self._ws = None + if self._session and not self._session.closed: + await self._session.close() + self._session = None + + +class PollingClient: + """Bot Framework Connector API 轮询拉取客户端。 + + 适用于无法使用 Webhook 或 WebSocket 的部署场景。 + """ + + def __init__( + self, + app_id: str, + app_password: str, + service_url: str = "", + poll_interval: float = POLLING_DEFAULT_INTERVAL_S, + max_interval: float = POLLING_MAX_INTERVAL_S, + ): + self._app_id = app_id + self._app_password = app_password + self._service_url = (service_url or "https://smba.trafficmanager.net/emea").rstrip("/") + self._poll_interval = poll_interval + self._max_interval = max_interval + self._session: aiohttp.ClientSession | None = None + self._running = False + self._last_watermark: str = "" + self._on_message: Any = None + + def set_message_handler(self, handler: Any) -> None: + self._on_message = handler + + async def connect(self) -> None: + if self._running: + return + self._session = aiohttp.ClientSession() + self._running = True + logger.info(f"MSTeams Polling started: interval={self._poll_interval}s") + asyncio.create_task(self._poll_loop()) + + async def _poll_loop(self) -> None: + backoff = self._poll_interval + + while self._running: + try: + activities = await self._fetch_activities() + if activities: + backoff = self._poll_interval + for activity in activities: + if self._on_message: + await self._on_message(activity) + else: + backoff = min(backoff * 1.5, self._max_interval) + except Exception as e: + logger.warning(f"MSTeams Polling fetch error: {e}") + backoff = min(backoff * 2, self._max_interval) + + await asyncio.sleep(backoff) + + async def _fetch_activities(self) -> list[dict[str, Any]]: + if not self._session: + return [] + + url = BOT_ACTIVITIES_URL_TEMPLATE.format(conversation_id="all") + headers = { + "Authorization": f"Bearer {self._app_password}", + "Content-Type": "application/json", + } + + async with self._session.get(url, headers=headers) as resp: + if resp.status == 200: + data = await resp.json() + activities = data.get("activities", []) + self._last_watermark = data.get("watermark", self._last_watermark) + return activities + elif resp.status == 429: + logger.warning("MSTeams Polling rate limited") + return [] + else: + body = await resp.text() + logger.warning(f"MSTeams Polling HTTP {resp.status}: {body[:200]}") + return [] + + async def close(self) -> None: + self._running = False + if self._session and not self._session.closed: + await self._session.close() + self._session = None + + +class ConnectionModeManager: + def __init__(self, config: dict[str, Any]): + self._mode = config.get("connection_mode", CONNECTION_MODE_WEBHOOK) + if self._mode not in VALID_CONNECTION_MODES: + logger.warning(f"Invalid connection_mode '{self._mode}', falling back to 'webhook'") + self._mode = CONNECTION_MODE_WEBHOOK + + self._ws_client: WebSocketClient | None = None + self._poll_client: PollingClient | None = None + self._message_router: Any = None + + @property + def mode(self) -> str: + return self._mode + + @property + def is_webhook(self) -> bool: + return self._mode == CONNECTION_MODE_WEBHOOK + + @property + def is_websocket(self) -> bool: + return self._mode == CONNECTION_MODE_WEBSOCKET + + @property + def is_polling(self) -> bool: + return self._mode == CONNECTION_MODE_POLLING + + def set_message_router(self, router: Any) -> None: + self._message_router = router + + async def start( + self, + app_id: str, + app_password: str, + service_url: str = "", + conversation_id: str = "", + ) -> None: + if self._mode == CONNECTION_MODE_WEBSOCKET: + self._ws_client = WebSocketClient(app_id, app_password, stream_url=conversation_id) + if self._message_router: + self._ws_client.set_message_handler(self._message_router) + await self._ws_client.connect(conversation_id) + logger.info("MSTeams connection mode: WebSocket") + elif self._mode == CONNECTION_MODE_POLLING: + self._poll_client = PollingClient(app_id, app_password, service_url) + if self._message_router: + self._poll_client.set_message_handler(self._message_router) + await self._poll_client.connect() + logger.info("MSTeams connection mode: Polling") + else: + logger.info("MSTeams connection mode: Webhook") + + async def stop(self) -> None: + if self._ws_client: + await self._ws_client.close() + self._ws_client = None + if self._poll_client: + await self._poll_client.close() + self._poll_client = None + + async def send_activity(self, activity: dict[str, Any]) -> bool: + if self._mode == CONNECTION_MODE_WEBSOCKET and self._ws_client: + return await self._ws_client.send_activity(activity) + return False diff --git a/backend/package/yuxi/channels/adapters/msteams/credentials.py b/backend/package/yuxi/channels/adapters/msteams/credentials.py index 1c0efaa0..938f6836 100644 --- a/backend/package/yuxi/channels/adapters/msteams/credentials.py +++ b/backend/package/yuxi/channels/adapters/msteams/credentials.py @@ -58,9 +58,9 @@ class FederatedCredential: def _check_prerequisites(self) -> None: try: - from cryptography.hazmat.primitives import serialization # noqa: F401 - from cryptography.hazmat.backends import default_backend # noqa: F401 import jwt as pyjwt # noqa: F401 + from cryptography.hazmat.backends import default_backend # noqa: F401 + from cryptography.hazmat.primitives import serialization # noqa: F401 except ImportError as e: raise FederatedCredentialError( "FederatedCredential requires 'cryptography' and 'PyJWT'. Install with: pip install cryptography pyjwt" @@ -101,9 +101,9 @@ class FederatedCredential: async def _get_certificate_token(self, scope: str) -> str | None: try: - from cryptography.hazmat.primitives import serialization - from cryptography.hazmat.backends import default_backend import jwt as pyjwt + from cryptography.hazmat.backends import default_backend + from cryptography.hazmat.primitives import serialization cert_path = Path(self.certificate_path) if not cert_path.exists(): diff --git a/backend/package/yuxi/channels/adapters/msteams/feedback.py b/backend/package/yuxi/channels/adapters/msteams/feedback.py index 8c6df581..781e5afc 100644 --- a/backend/package/yuxi/channels/adapters/msteams/feedback.py +++ b/backend/package/yuxi/channels/adapters/msteams/feedback.py @@ -208,4 +208,27 @@ def process_feedback( elif is_positive: logger.info(f"MSTeams positive feedback received from {user_name} ({user_id})") + record_feedback_session(result) + return result + + +def record_feedback_session(feedback_result: dict[str, Any], storage_dir: str = "") -> None: + import os + + session_dir = Path( + storage_dir or os.environ.get("MSTEAMS_FEEDBACK_DIR", "") or str(Path.home() / ".yuxi" / "msteams") + ) + session_dir.mkdir(parents=True, exist_ok=True) + session_file = session_dir / "feedback-sessions.jsonl" + + record = { + "timestamp": time.time(), + **feedback_result, + } + + try: + with open(session_file, "a", encoding="utf-8") as f: + f.write(json.dumps(record, ensure_ascii=False) + "\n") + except OSError as e: + logger.warning(f"MSTeams feedback: failed to record session: {e}") diff --git a/backend/package/yuxi/channels/adapters/msteams/file_upload.py b/backend/package/yuxi/channels/adapters/msteams/file_upload.py index e460b689..5a06e178 100644 --- a/backend/package/yuxi/channels/adapters/msteams/file_upload.py +++ b/backend/package/yuxi/channels/adapters/msteams/file_upload.py @@ -5,57 +5,36 @@ FileConsentCard 大文件交互流程 + SharePoint 上传策略 + SSRF 三层防 from __future__ import annotations -import ipaddress -import re import time from pathlib import Path from typing import Any from urllib.parse import urlparse +from yuxi.channels.auth.ssrf_guard import ( + is_hostname_allowed, + is_private_url, +) from yuxi.utils.logging_config import logger FILE_CONSENT_MAX_SIZE = 4 * 1024 * 1024 -_SSRF_ALLOWED_DOMAINS = frozenset( - { - "graph.microsoft.com", - "graph.microsoft.us", - "login.microsoftonline.com", - "login.microsoftonline.us", - "api.botframework.com", - "api.botframework.us", - "smba.trafficmanager.net", - "sharepoint.com", - "sharepoint-df.com", - "onedrive.com", - "office.com", - "office.net", - } -) - -_SSRF_DOMAIN_PARENT_PATTERN = re.compile( - r"^(.+\.)?(" + "|".join(map(re.escape, _SSRF_ALLOWED_DOMAINS)) + r")$", - re.IGNORECASE, -) - -_SSRF_PRIVATE_RANGES = [ - ipaddress.ip_network("10.0.0.0/8"), - ipaddress.ip_network("172.16.0.0/12"), - ipaddress.ip_network("192.168.0.0/16"), - ipaddress.ip_network("127.0.0.0/8"), - ipaddress.ip_network("169.254.0.0/16"), - ipaddress.ip_network("0.0.0.0/8"), - ipaddress.ip_network("fc00::/7"), - ipaddress.ip_network("::1/128"), +_SSRF_ALLOWED_DOMAINS = [ + "graph.microsoft.com", + "graph.microsoft.us", + "login.microsoftonline.com", + "login.microsoftonline.us", + "api.botframework.com", + "api.botframework.us", + "smba.trafficmanager.net", + "*.sharepoint.com", + "*.sharepoint-df.com", + "*.onedrive.com", + "*.office.com", + "*.office.net", ] def _is_ssrf_safe_url(url: str) -> tuple[bool, str]: - """三层 SSRF 防护：协议 → 域名白名单 → DNS 私有地址检测。 - - Returns: - (is_safe, reason) — safe 为 True 表示可安全请求。 - """ parsed = urlparse(url) if parsed.scheme != "https": @@ -65,16 +44,11 @@ def _is_ssrf_safe_url(url: str) -> tuple[bool, str]: if not hostname: return False, "No hostname in URL" - if not _SSRF_DOMAIN_PARENT_PATTERN.match(hostname): + if not is_hostname_allowed(hostname, _SSRF_ALLOWED_DOMAINS): return False, f"Domain '{hostname}' not in SSRF allowlist" - try: - addr = ipaddress.ip_address(hostname) - for net in _SSRF_PRIVATE_RANGES: - if addr in net: - return False, f"IP '{hostname}' is in private range {net}" - except ValueError: - pass + if is_private_url(url): + return False, f"URL '{url}' resolves to private/internal network" return True, "" diff --git a/backend/package/yuxi/channels/adapters/msteams/formatter.py b/backend/package/yuxi/channels/adapters/msteams/formatter.py index e1ab8998..42279edc 100644 --- a/backend/package/yuxi/channels/adapters/msteams/formatter.py +++ b/backend/package/yuxi/channels/adapters/msteams/formatter.py @@ -11,7 +11,7 @@ from typing import Any from yuxi.channels.models import ChannelResponse, MessageType -from .mentions import parse_outbound_mentions, apply_mentions_to_activity +from .mentions import apply_mentions_to_activity, parse_outbound_mentions from .send import is_silent_reply_text, strip_silent_token diff --git a/backend/package/yuxi/channels/adapters/msteams/graph.py b/backend/package/yuxi/channels/adapters/msteams/graph.py index 346bc110..421102e8 100644 --- a/backend/package/yuxi/channels/adapters/msteams/graph.py +++ b/backend/package/yuxi/channels/adapters/msteams/graph.py @@ -14,6 +14,8 @@ import aiohttp from yuxi.utils.logging_config import logger +from .user_agent import get_default_user_agent + GRAPH_BASE_URL = "https://graph.microsoft.com/v1.0" GRAPH_UPLOAD_CHUNK_SIZE = 4 * 1024 * 1024 GRAPH_MAX_RETRIES = 3 @@ -83,6 +85,7 @@ class GraphClient: return { "Authorization": f"Bearer {self._token}", "Content-Type": "application/json", + "User-Agent": get_default_user_agent(), } async def _ensure_session(self) -> aiohttp.ClientSession: diff --git a/backend/package/yuxi/channels/adapters/msteams/group_mgmt.py b/backend/package/yuxi/channels/adapters/msteams/group_mgmt.py index 9b551148..b21748e7 100644 --- a/backend/package/yuxi/channels/adapters/msteams/group_mgmt.py +++ b/backend/package/yuxi/channels/adapters/msteams/group_mgmt.py @@ -5,7 +5,7 @@ addParticipant / removeParticipant / renameGroup 操作。 from __future__ import annotations -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .graph import GraphClient diff --git a/backend/package/yuxi/channels/adapters/msteams/media_tools.py b/backend/package/yuxi/channels/adapters/msteams/media_tools.py new file mode 100644 index 00000000..1e5c0035 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/msteams/media_tools.py @@ -0,0 +1,67 @@ +"""Microsoft Teams 媒体辅助工具。 + +提供 MIME 类型识别、文件名提取、消息 ID 提取等辅助函数。 +""" + +from __future__ import annotations + +import mimetypes +import os +from typing import Any + + +def get_mime_type(filename: str) -> str: + mime_type, _ = mimetypes.guess_type(filename) + return mime_type or "application/octet-stream" + + +def extract_filename(activity: dict[str, Any]) -> str: + attachments = activity.get("attachments", []) or [] + for att in attachments: + name = att.get("name", "").strip() + if name: + return os.path.basename(name) + content = att.get("content", {}) or {} + content_name = content.get("name", "").strip() + if content_name: + return os.path.basename(content_name) + return "" + + +def extract_message_id(activity: dict[str, Any]) -> str: + msg_id = activity.get("id", "") + if msg_id: + return msg_id + + channel_data = activity.get("channelData", {}) or {} + channel_msg_id = channel_data.get("id", "") + if channel_msg_id: + return channel_msg_id + + return "" + + +def extract_attachment_urls(activity: dict[str, Any]) -> list[dict[str, str]]: + urls: list[dict[str, str]] = [] + attachments = activity.get("attachments", []) or [] + for att in attachments: + content_type = att.get("contentType", "") + content_url = att.get("contentUrl", "") + name = att.get("name", "") + if content_url: + urls.append( + { + "url": content_url, + "name": name or os.path.basename(content_url), + "content_type": content_type, + } + ) + return urls + + +def get_edited_timestamp(activity: dict[str, Any]) -> str | None: + channel_data = activity.get("channelData", {}) or {} + edit_time = channel_data.get("editedTimestamp", "") + if edit_time: + return edit_time + return activity.get("editedTimestamp") diff --git a/backend/package/yuxi/channels/adapters/msteams/normalizer.py b/backend/package/yuxi/channels/adapters/msteams/normalizer.py index dfd798cd..f2d6fcc7 100644 --- a/backend/package/yuxi/channels/adapters/msteams/normalizer.py +++ b/backend/package/yuxi/channels/adapters/msteams/normalizer.py @@ -6,7 +6,7 @@ from __future__ import annotations import re -from datetime import datetime, UTC +from datetime import UTC, datetime from typing import Any from yuxi.channels.models import ( @@ -116,6 +116,15 @@ def normalize_inbound(activity: dict[str, Any]) -> ChannelMessage: "ConversationMessageId": thread_root_id or "", } + thread_meta = extract_thread_metadata(activity) + metadata.update(thread_meta) + + quote_info = extract_quote_info(activity) + metadata.update(quote_info) + + edit_info = detect_edited_message(activity) + metadata.update(edit_info) + reply_to = thread_root_id or activity.get("replyToId") return ChannelMessage( @@ -181,12 +190,16 @@ def normalize_invoke(activity: dict[str, Any]) -> ChannelMessage: aad_object_id = from_info.get("aadObjectId", "") or from_info.get("id", "") conversation_id = conversation.get("id", "") + conversation_type = conversation.get("conversationType", "personal") + base_conv_id = extract_base_conversation_id(conversation_id) + + channel_chat_id = resolve_channel_chat_id(conversation_type, base_conv_id, aad_object_id, channel_data) identity = ChannelIdentity( channel_id="msteams", channel_type=ChannelType.MS_TEAMS, channel_user_id=aad_object_id, - channel_chat_id=conversation_id, + channel_chat_id=channel_chat_id, channel_message_id=activity.get("id"), ) @@ -276,6 +289,102 @@ def extract_reply_context(activity: dict[str, Any]) -> dict[str, str]: return result +def extract_quote_info(activity: dict[str, Any]) -> dict[str, str]: + result: dict[str, str] = {} + reply_to_id = activity.get("replyToId", "") + if reply_to_id: + result["quoted_message_id"] = reply_to_id + + channel_data = activity.get("channelData", {}) or {} + quote_target = channel_data.get("quoteMessageId", "") + if quote_target: + result["quote_target_id"] = quote_target + + text = activity.get("text", "") or "" + if "
" in text.lower(): + result["has_blockquote"] = "true" + + return result + + +def detect_edited_message(activity: dict[str, Any]) -> dict[str, str]: + result: dict[str, str] = {} + channel_data = activity.get("channelData", {}) or {} + edit_time = channel_data.get("editedTimestamp", "") + + if not edit_time: + edit_time = activity.get("editedTimestamp", "") + + if edit_time: + result["is_edited"] = "true" + result["edited_timestamp"] = str(edit_time) + + return result + + +def extract_thread_metadata(activity: dict[str, Any]) -> dict[str, str]: + metadata: dict[str, str] = {} + channel_data = activity.get("channelData", {}) or {} + team_id = (channel_data.get("team") or {}).get("id", "") + channel_id = (channel_data.get("channel") or {}).get("id", "") + + conversation = activity.get("conversation", {}) or {} + conversation_id = conversation.get("id", "") + thread_root_id = extract_thread_root_id(conversation_id) + + if team_id: + metadata["team_id"] = team_id + if channel_id: + metadata["channel_id"] = channel_id + if thread_root_id: + metadata["thread_root_id"] = thread_root_id + + reply_to_id = activity.get("replyToId", "") + if reply_to_id: + metadata["reply_to_id"] = reply_to_id + + base_conv_id = extract_base_conversation_id(conversation_id) + metadata["base_conversation_id"] = base_conv_id + + return metadata + + +async def inject_thread_context( + channel_message: ChannelMessage, + graph_client: Any = None, + session_key: str = "", +) -> ChannelMessage: + metadata = channel_message.metadata or {} + team_id = metadata.get("team_id", "") + channel_id = metadata.get("channel_id", "") + thread_root_id = metadata.get("thread_root_id", "") or metadata.get("reply_to_id", "") + + if not graph_client or not team_id or not channel_id or not thread_root_id: + return channel_message + + try: + from .graph import format_thread_context + + parent = await graph_client.fetch_parent_message(team_id, channel_id, thread_root_id) + if not parent: + return channel_message + + replies = await graph_client.fetch_thread_replies(team_id, channel_id, thread_root_id, limit=20) + context_msgs = [parent] + replies + + thread_context = format_thread_context(context_msgs) + + if thread_context and channel_message.content: + channel_message.content = f"{thread_context}\n\n[Current message]\n{channel_message.content}" + metadata["thread_context_injected"] = True + channel_message.metadata = metadata + + except Exception: + pass + + return channel_message + + def extract_html_text(html_content: str) -> str: if not html_content: return "" diff --git a/backend/package/yuxi/channels/adapters/msteams/oauth_flow.py b/backend/package/yuxi/channels/adapters/msteams/oauth_flow.py index 83403094..e67cacd1 100644 --- a/backend/package/yuxi/channels/adapters/msteams/oauth_flow.py +++ b/backend/package/yuxi/channels/adapters/msteams/oauth_flow.py @@ -13,9 +13,9 @@ import os import secrets import sys import webbrowser -from http.server import HTTPServer, BaseHTTPRequestHandler +from http.server import BaseHTTPRequestHandler, HTTPServer from typing import Any -from urllib.parse import urlencode, urlparse, parse_qs +from urllib.parse import parse_qs, urlencode, urlparse import aiohttp diff --git a/backend/package/yuxi/channels/adapters/msteams/pins.py b/backend/package/yuxi/channels/adapters/msteams/pins.py index f7397cdb..61de7379 100644 --- a/backend/package/yuxi/channels/adapters/msteams/pins.py +++ b/backend/package/yuxi/channels/adapters/msteams/pins.py @@ -5,7 +5,7 @@ from __future__ import annotations -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .graph import GraphClient diff --git a/backend/package/yuxi/channels/adapters/msteams/proactive.py b/backend/package/yuxi/channels/adapters/msteams/proactive.py index 40bae92a..b9ae6e46 100644 --- a/backend/package/yuxi/channels/adapters/msteams/proactive.py +++ b/backend/package/yuxi/channels/adapters/msteams/proactive.py @@ -11,7 +11,7 @@ import asyncio import json import time from pathlib import Path -from typing import Any, TYPE_CHECKING +from typing import TYPE_CHECKING, Any from yuxi.channels.models import DeliveryResult from yuxi.utils.logging_config import logger diff --git a/backend/package/yuxi/channels/adapters/msteams/security.py b/backend/package/yuxi/channels/adapters/msteams/security.py index f2ff7563..63ecc428 100644 --- a/backend/package/yuxi/channels/adapters/msteams/security.py +++ b/backend/package/yuxi/channels/adapters/msteams/security.py @@ -7,7 +7,8 @@ DM Policy (open/pairing/allowlist/disabled) 和 Group Policy (open/allowlist/dis from __future__ import annotations import fnmatch -from typing import Any, TYPE_CHECKING +import os +from typing import TYPE_CHECKING, Any from yuxi.utils.logging_config import logger @@ -27,6 +28,7 @@ _VALID_DM_POLICIES = {DM_POLICY_OPEN, DM_POLICY_PAIRING, DM_POLICY_ALLOWLIST, DM _VALID_GROUP_POLICIES = {GROUP_POLICY_OPEN, GROUP_POLICY_ALLOWLIST, GROUP_POLICY_DISABLED} _ALLOW_WILDCARD = "*" +_ALLOW_WILDCARD_ENABLED = os.environ.get("MSTEAMS_ALLOW_WILDCARD", "").lower() in ("1", "true", "yes") AccessGroupResolver = "Callable[[str, list[str]], Awaitable[set[str]]]" @@ -64,6 +66,17 @@ class SecurityPolicy: self.allow_from = self._normalize_allow_entries(config.get("allow_from", [])) self.group_allow_from = self._normalize_allow_entries(config.get("group_allow_from", [])) self.allow_name_matching = config.get("allow_name_matching", False) + + if self.dm_policy == DM_POLICY_ALLOWLIST and _ALLOW_WILDCARD in self.allow_from: + logger.warning( + "MSTeams SecurityPolicy: wildcard '*' detected in DM allowlist. " + "All users will bypass DM policy. Consider using dm_policy='open' for explicit intent." + ) + if self.group_policy == GROUP_POLICY_ALLOWLIST and _ALLOW_WILDCARD in self.group_allow_from: + logger.warning( + "MSTeams SecurityPolicy: wildcard '*' detected in Group allowlist. " + "All groups/users will bypass Group policy." + ) self._teams_config: dict[str, dict[str, Any]] = config.get("teams", {}) self.use_access_groups: bool = config.get("use_access_groups", False) @@ -114,7 +127,10 @@ class SecurityPolicy: if not allowlist: return False if _ALLOW_WILDCARD in allowlist: - return True + if _ALLOW_WILDCARD_ENABLED: + logger.warning("MSTeams SecurityPolicy: wildcard matched — bypassing allowlist check") + return True + return False for entry in allowlist: if self._match_entry(user_id, entry): return True diff --git a/backend/package/yuxi/channels/adapters/msteams/send.py b/backend/package/yuxi/channels/adapters/msteams/send.py index 9489296d..5f577571 100644 --- a/backend/package/yuxi/channels/adapters/msteams/send.py +++ b/backend/package/yuxi/channels/adapters/msteams/send.py @@ -8,7 +8,7 @@ from __future__ import annotations import asyncio import time -from collections.abc import Callable, Awaitable +from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from enum import Enum from typing import Any diff --git a/backend/package/yuxi/channels/adapters/msteams/sent_message_cache.py b/backend/package/yuxi/channels/adapters/msteams/sent_message_cache.py index e53d70ab..9f57a7f4 100644 --- a/backend/package/yuxi/channels/adapters/msteams/sent_message_cache.py +++ b/backend/package/yuxi/channels/adapters/msteams/sent_message_cache.py @@ -6,6 +6,7 @@ from __future__ import annotations +import asyncio import time from collections import OrderedDict from typing import Any @@ -19,6 +20,7 @@ class SentMessageCache: self._max_entries = max_entries self._ttl = ttl self._entries: OrderedDict[str, dict[str, Any]] = OrderedDict() + self._cleanup_task: asyncio.Task | None = None def record(self, message_id: str, chat_id: str) -> None: now = time.monotonic() @@ -48,6 +50,20 @@ class SentMessageCache: def clear(self) -> None: self._entries.clear() + def start_cleanup_task(self, interval: float = 600.0) -> None: + if self._cleanup_task and not self._cleanup_task.done(): + return + self._cleanup_task = asyncio.create_task(self._cleanup_loop(interval)) + + async def _cleanup_loop(self, interval: float) -> None: + while True: + await asyncio.sleep(interval) + self._cleanup(time.monotonic()) + + def stop_cleanup_task(self) -> None: + if self._cleanup_task and not self._cleanup_task.done(): + self._cleanup_task.cancel() + @property def count(self) -> int: return len(self._entries) diff --git a/backend/package/yuxi/channels/adapters/msteams/setup_wizard.py b/backend/package/yuxi/channels/adapters/msteams/setup_wizard.py index 576aa75f..8f6c0b04 100644 --- a/backend/package/yuxi/channels/adapters/msteams/setup_wizard.py +++ b/backend/package/yuxi/channels/adapters/msteams/setup_wizard.py @@ -16,7 +16,6 @@ from typing import Any from yuxi.utils.logging_config import logger - STEP_TITLES = { 1: "Bot 凭据配置", 2: "DM 策略配置", @@ -201,6 +200,26 @@ class MSTeamsSetupWizard: print("\n ℹ 请在 Azure AD 中注册以下重定向 URI：") print(" http://localhost:5353/oauth/msteams/callback") print(" ℹ 授权流程: `make msteams-oauth` 启动本地回调服务器") + + auth_url = "https://login.microsoftonline.com/organizations/oauth2/v2.0/authorize" + if self._confirm("\n是否自动打开浏览器开始 OAuth 授权", True): + tenant_id = self._config.get("tenant_id", "organizations") + client_id = self._config.get("app_id", "") + if client_id: + scope_param = " ".join(scopes.split()) if isinstance(scopes, str) else " ".join(scopes.split()) + oauth_url = ( + f"https://login.microsoftonline.com/{tenant_id}/oauth2/v2.0/authorize" + f"?client_id={client_id}" + f"&response_type=code" + f"&redirect_uri=http://localhost:5353/oauth/msteams/callback" + f"&scope={scope_param.replace(' ', '%20')}" + f"&response_mode=query" + ) + _open_browser(oauth_url) + print(" ℹ 浏览器已打开，完成授权后将回调 localhost:5353") + else: + print(" ⚠ 未配置 App ID，无法生成授权 URL") + print(f" ✓ OAuth: 已启用 (scopes={len(scopes.split())})") else: self._config["delegated_auth"] = {"enabled": False} @@ -348,6 +367,27 @@ def _validate_and_normalize_answers(answers: dict[str, Any]) -> dict[str, Any]: return result +def _open_browser(url: str) -> bool: + """跨平台浏览器自动打开。 + + 支持 Windows (start), macOS (open), Linux (xdg-open)。 + """ + import subprocess + import sys + + try: + if sys.platform == "win32": + subprocess.Popen(["start", url], shell=True) + elif sys.platform == "darwin": + subprocess.Popen(["open", url]) + else: + subprocess.Popen(["xdg-open", url]) + return True + except Exception as e: + logger.warning(f"MSTeams setup wizard: failed to open browser: {e}") + return False + + def run_msteams_setup_wizard( existing_config: dict[str, Any] | None = None, output_file: str | None = None, diff --git a/backend/package/yuxi/channels/adapters/msteams/streaming.py b/backend/package/yuxi/channels/adapters/msteams/streaming.py index a662f2e3..11d9d8ac 100644 --- a/backend/package/yuxi/channels/adapters/msteams/streaming.py +++ b/backend/package/yuxi/channels/adapters/msteams/streaming.py @@ -3,10 +3,10 @@ from __future__ import annotations import asyncio import random import time +from collections.abc import Awaitable, Callable from dataclasses import dataclass from enum import Enum -from typing import Any, TYPE_CHECKING -from collections.abc import Callable, Awaitable +from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from .send import MessageSender @@ -159,13 +159,58 @@ class TeamsHttpStream: def _random_informative_text(self) -> str: return random.choice(self._informative_texts) + async def _fallback_informative(self, chat_id: str) -> DeliveryResult: + text = random.choice(_SIMPLE_INFORMATIVE_TEXTS) + activity = { + "type": "message", + "text": text, + "textFormat": "markdown", + } + result = await self._sender.send_activity(chat_id, activity) + if result.success and result.message_id: + now = time.monotonic() + self._states[chat_id] = StreamState( + phase=StreamPhase.INFORMATIVE, + stream_id=str(int(now * 1000)), + message_id=result.message_id, + chat_id=chat_id, + created_at=now, + last_update_time=now, + has_fallback=True, + ) + return result + + async def _fallback_stream_update(self, state: StreamState, finished: bool) -> DeliveryResult: + now = time.monotonic() + elapsed_since_update = (now - state.last_update_time) * 1000 + + if elapsed_since_update < self._throttle_ms and not finished: + state.total_send_count += 1 + return DeliveryResult(success=True, message_id=state.message_id) + + text = state.accumulated_text[: self._chunk_limit] + activity = { + "type": "message", + "text": text, + "textFormat": "markdown", + } + result = await self._sender.update_activity(state.chat_id, state.message_id, activity) + state.last_update_time = time.monotonic() + state.total_send_count += 1 + + if finished: + state.phase = StreamPhase.FINAL + self._states.pop(state.chat_id, None) + + return result + async def send_informative(self, chat_id: str) -> DeliveryResult: state = self._states.get(chat_id) if state and state.phase == StreamPhase.STREAMING: return DeliveryResult(success=True, message_id=state.message_id) if not self.supports_streaming(chat_id): - return DeliveryResult(success=False, error="Streaming not supported for this chat type, fallback to block") + return await self._fallback_informative(chat_id) text = self._random_informative_text() activity = { @@ -222,6 +267,9 @@ class TeamsHttpStream: state.accumulated_text += chunk state.chunk_count += 1 + if not self.supports_streaming(chat_id): + return await self._fallback_stream_update(state, finished) + if state.phase == StreamPhase.INFORMATIVE: if len(state.accumulated_text) >= self._min_initial_chars: state.phase = StreamPhase.STREAMING diff --git a/backend/package/yuxi/channels/adapters/msteams/thread_history.py b/backend/package/yuxi/channels/adapters/msteams/thread_history.py new file mode 100644 index 00000000..05193134 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/msteams/thread_history.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from yuxi.channels.history_injector import HistoryCache, HistoryFetcher +from yuxi.channels.models import FetchOptions, HistoricalMessage + +if TYPE_CHECKING: + from yuxi.channels.adapters.msteams.graph import GraphClient + + +class TeamsHistoryFetcher(HistoryFetcher): + def __init__(self, graph_client: GraphClient): + self._client = graph_client + self._cache = HistoryCache(max_size=100, ttl_seconds=300) + + async def fetch_thread_history(self, thread_id: str, options: FetchOptions) -> list[HistoricalMessage]: + cached = self._cache.get(thread_id, options) + if cached is not None: + return cached + + team_id, channel_id, message_id = _parse_thread_id(thread_id) + replies = await self._client.fetch_thread_replies( + team_id=team_id, + channel_id=channel_id, + message_id=message_id, + limit=min(options.max_messages, 50), + ) + + messages = [ + HistoricalMessage( + message_id=reply.get("id", ""), + sender_id=(reply.get("from", {}) or {}).get("user", {}).get("id", ""), + sender_name=(reply.get("from", {}) or {}).get("user", {}).get("displayName", "Unknown"), + content=(reply.get("body", {}) or {}).get("content", ""), + timestamp=datetime.fromisoformat(reply.get("createdDateTime", "").replace("Z", "+00:00")), + is_from_bot=(reply.get("from", {}) or {}).get("application") is not None, + ) + for reply in replies + ] + + self._cache.set(thread_id, options, messages) + return messages + + async def fetch_parent_message(self, thread_id: str) -> HistoricalMessage | None: + team_id, channel_id, message_id = _parse_thread_id(thread_id) + parent = await self._client.fetch_parent_message(team_id, channel_id, message_id) + if not parent: + return None + + return HistoricalMessage( + message_id=parent.get("id", ""), + sender_id=(parent.get("from", {}) or {}).get("user", {}).get("id", ""), + sender_name=(parent.get("from", {}) or {}).get("user", {}).get("displayName", "Unknown"), + content=(parent.get("body", {}) or {}).get("content", ""), + timestamp=datetime.fromisoformat(parent.get("createdDateTime", "").replace("Z", "+00:00")), + ) + + +def _parse_thread_id(thread_id: str) -> tuple[str, str, str]: + parts = thread_id.split(":") + if len(parts) >= 3: + return parts[-3], parts[-2], parts[-1] + return "", "", thread_id diff --git a/backend/package/yuxi/channels/adapters/msteams/tool_policy.py b/backend/package/yuxi/channels/adapters/msteams/tool_policy.py index 570e2a3f..e075dae0 100644 --- a/backend/package/yuxi/channels/adapters/msteams/tool_policy.py +++ b/backend/package/yuxi/channels/adapters/msteams/tool_policy.py @@ -1,6 +1,7 @@ """Microsoft Teams 工具策略 (Tools Policy)。 -Team/Channel 级 tools allow/deny + toolsBySender 策略。 +Team/Channel/Sender 三级 tools allow/deny 策略解析。 +支持全局配置 + teams 嵌套覆盖 + toolsBySender 按发送者策略。 """ from __future__ import annotations @@ -9,14 +10,38 @@ from typing import Any class ToolPolicy: + """MSTeams 三级工具策略解析器。 + + 优先级：Sender > Channel > Team > Global deny > Global allow + """ + def __init__(self, config: dict[str, Any] | None = None): config = config or {} self._global_allow: set[str] = set(config.get("tools_allow", [])) self._global_deny: set[str] = set(config.get("tools_deny", [])) - self._tools_by_sender: dict[str, set[str]] = {} - sender_tools = config.get("tools_by_sender", {}) or {} - for sender_id, tools in sender_tools.items(): - self._tools_by_sender[sender_id] = set(tools) + self._tools_by_sender: dict[str, dict[str, set[str]]] = {} + + sender_tools = config.get("tools_by_sender", []) or [] + if isinstance(sender_tools, dict): + for sender_id, tools in sender_tools.items(): + if isinstance(tools, list): + self._tools_by_sender[sender_id] = {"allow": set(tools), "deny": set()} + elif isinstance(tools, dict): + self._tools_by_sender[sender_id] = { + "allow": set(tools.get("allow", [])), + "deny": set(tools.get("deny", [])), + } + elif isinstance(sender_tools, list): + for entry in sender_tools: + if isinstance(entry, dict): + sid = entry.get("sender_id", "") + if sid: + self._tools_by_sender[sid] = { + "allow": set(entry.get("allow", [])), + "deny": set(entry.get("deny", [])), + } + + self._teams_config: dict[str, dict[str, Any]] = config.get("teams", {}) def is_tool_allowed( self, @@ -25,9 +50,24 @@ class ToolPolicy: team_id: str = "", channel_id: str = "", ) -> bool: - sender_tools = self._tools_by_sender.get(sender_id) - if sender_tools is not None: - return tool_name in sender_tools + sender_rules = self._tools_by_sender.get(sender_id) + if sender_rules is not None: + if tool_name in sender_rules["deny"]: + return False + if sender_rules["allow"]: + return tool_name in sender_rules["allow"] + + channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) + if tool_name in channel_deny: + return False + if channel_allow and tool_name not in channel_allow: + return False + + team_allow, team_deny = self._resolve_team_tools(team_id) + if tool_name in team_deny: + return False + if team_allow and tool_name not in team_allow: + return False if tool_name in self._global_deny: return False @@ -37,14 +77,62 @@ class ToolPolicy: return True - def get_allowed_tools(self, sender_id: str = "") -> list[str] | None: - sender_tools = self._tools_by_sender.get(sender_id) - if sender_tools is not None: - return list(sender_tools) + def _resolve_team_tools(self, team_id: str) -> tuple[set[str], set[str]]: + team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {} + team_tools = team_config.get("tools", {}) + allow = set(team_tools.get("allow", [])) + deny = set(team_tools.get("deny", [])) + return allow, deny + + def _resolve_channel_tools(self, team_id: str, channel_id: str) -> tuple[set[str], set[str]]: + if not team_id or not channel_id: + return set(), set() + + team_config = self._teams_config.get(team_id) or self._teams_config.get("*") or {} + channels_config = team_config.get("channels", {}) + ch_config = channels_config.get(channel_id) or channels_config.get("*") or {} + ch_tools = ch_config.get("tools", {}) + allow = set(ch_tools.get("allow", [])) + deny = set(ch_tools.get("deny", [])) + return allow, deny + + def get_allowed_tools( + self, + sender_id: str = "", + team_id: str = "", + channel_id: str = "", + ) -> list[str] | None: + sender_rules = self._tools_by_sender.get(sender_id) + if sender_rules is not None: + if sender_rules["allow"]: + return sorted(sender_rules["allow"] - sender_rules["deny"]) + + channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) + team_allow, team_deny = self._resolve_team_tools(team_id) + + if channel_allow: + return sorted(channel_allow - channel_deny) + if team_allow: + return sorted(team_allow - team_deny) if self._global_allow: - return list(self._global_allow) + return sorted(self._global_allow - self._global_deny) + return None + def get_denied_tools( + self, + sender_id: str = "", + team_id: str = "", + channel_id: str = "", + ) -> set[str]: + sender_rules = self._tools_by_sender.get(sender_id) + sender_deny = sender_rules["deny"] if sender_rules else set() + + channel_allow, channel_deny = self._resolve_channel_tools(team_id, channel_id) + team_allow, team_deny = self._resolve_team_tools(team_id) + + return sender_deny | channel_deny | team_deny | self._global_deny + def resolve_tool_policy( config: dict[str, Any], @@ -53,22 +141,9 @@ def resolve_tool_policy( team_id: str = "", channel_id: str = "", ) -> bool: - teams_config = config.get("teams", {}) or {} - team_config = (teams_config.get(team_id) or teams_config.get("*")) or {} + """快捷函数：三级工具策略解析。 - if team_config: - team_tools = team_config.get("tools", {}) or {} - if tool_name in team_tools.get("deny", []): - return False - if team_tools.get("allow"): - return tool_name in team_tools["allow"] - - global_deny = set(config.get("tools_deny", [])) - if tool_name in global_deny: - return False - - global_allow = set(config.get("tools_allow", [])) - if global_allow: - return tool_name in global_allow - - return True + 优先级：Sender > Channel > Team > Global + """ + policy = ToolPolicy(config) + return policy.is_tool_allowed(tool_name, sender_id, team_id, channel_id)