diff --git a/backend/package/yuxi/channels/adapters/googlechat/__init__.py b/backend/package/yuxi/channels/adapters/googlechat/__init__.py new file mode 100644 index 00000000..57648b88 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/__init__.py @@ -0,0 +1,57 @@ +from .adapter import GoogleChatAdapter +from .approval_auth import ( + build_exec_approval_context, + is_fully_approved, + normalize_approver_id, + record_approval_decision, + resolve_approver_ids, +) +from .auth import GoogleChatCertCache, get_cert_cache, verify_project_number_token +from .directory import list_groups, list_peers, normalize_id +from .doctor import run_diagnostics +from .pairing import check_pairing_approval, send_pairing_challenge +from .proxy import build_google_auth_request, build_proxies_dict, resolve_proxy_config, resolve_tls_config +from .secret_contract import get_secret_contracts +from .setup import SetupWizard, get_setup_guide +from .ssrf import ( + apply_ssrf_guard, + check_auth_response_size, + check_auth_response_text, + sanitize_google_auth_init, + sanitize_request_kwargs, +) +from .target import is_googlechat_space_target, is_googlechat_user_target, normalize_googlechat_target, resolve_targets + +__all__ = [ + "GoogleChatAdapter", + "send_pairing_challenge", + "check_pairing_approval", + "sanitize_google_auth_init", + "sanitize_request_kwargs", + "apply_ssrf_guard", + "check_auth_response_size", + "check_auth_response_text", + "normalize_approver_id", + "resolve_approver_ids", + "build_exec_approval_context", + "record_approval_decision", + "is_fully_approved", + "list_peers", + "list_groups", + "normalize_id", + "GoogleChatCertCache", + "get_cert_cache", + "verify_project_number_token", + "run_diagnostics", + "build_google_auth_request", + "build_proxies_dict", + "resolve_proxy_config", + "resolve_tls_config", + "get_secret_contracts", + "SetupWizard", + "get_setup_guide", + "is_googlechat_space_target", + "is_googlechat_user_target", + "normalize_googlechat_target", + "resolve_targets", +] diff --git a/backend/package/yuxi/channels/adapters/googlechat/adapter.py b/backend/package/yuxi/channels/adapters/googlechat/adapter.py new file mode 100644 index 00000000..a20e1b1a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/adapter.py @@ -0,0 +1,1373 @@ +from __future__ import annotations + +import asyncio +import json +import os +import time +from collections import defaultdict +from collections.abc import AsyncIterator +from datetime import UTC +from typing import Any, ClassVar + +from yuxi.channels.base import BaseChannelAdapter +from yuxi.channels.capabilities import ChannelCapabilities +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelException, + ChannelNotConnectedError, +) +from yuxi.channels.infra.circuit_breaker import CircuitBreaker, CircuitBreakerOpenError +from yuxi.channels.meta import ChannelMeta +from yuxi.channels.models import ( + ChannelIdentity, + ChannelMessage, + ChannelResponse, + ChannelStatus, + ChannelType, + DeliveryResult, + HealthStatus, +) +from yuxi.channels.registry import register_builtin_adapter +from yuxi.utils.logging_config import logger + +from .formatter import format_outbound +from .media import download_media as _download_media +from .normalizer import is_bot_message, normalize_inbound +from .policy import GoogleChatPolicy +from .send import ( + delete_message as _delete_message, +) +from .send import ( + send_media, + send_message, + send_reaction, + update_message, + upload_file_message, + upload_image_message, +) +from .streaming import StreamManager +from . import auth +from . import proxy + +_CONNECT_TIMEOUT_S = 30.0 +_PREAUTH_BODY_MAX_BYTES = 16 * 1024 +_PREAUTH_TIMEOUT_S = 3.0 + +_GOOGLE_CHAT_ISSUER_RE = r"^https://chat\.google\.com$" +_ADDON_ISSUER_RE = r"^(https://workspace\.google\.com/)?accounts\.google\.com$" +_ADDON_SA_ISSUER_PATTERN = r"^service-\d+@gcp-sa-gsuiteaddons\.iam\.gserviceaccount\.com$" + +_AUDIENCE_TYPE_CANONICAL: dict[str, str] = { + "app-url": "app-url", + "app_url": "app-url", + "app": "app-url", + "project-number": "project-number", + "project_number": "project-number", + "project": "project-number", +} + + +def _verify_jwt(token: str, audience: str) -> bool: + try: + from google.auth.transport import requests as ga_requests + from google.oauth2 import id_token + except ImportError: + logger.info("google-auth not installed, skipping JWT verification") + return True + + try: + id_token.verify_oauth2_token(token, ga_requests.Request(), audience=audience) + return True + except Exception as e: + logger.warning(f"JWT verification failed: {e}") + return False + + +class ChatRateLimiter: + def __init__(self, ops_per_second: float = 0.9): + self._interval = 1.0 / ops_per_second + self._last_call: dict[str, float] = defaultdict(float) + + async def acquire(self, space_id: str) -> None: + now = time.monotonic() + wait = self._last_call[space_id] + self._interval - now + if wait > 0: + await asyncio.sleep(wait) + self._last_call[space_id] = time.monotonic() + + +@register_builtin_adapter +class GoogleChatAdapter(BaseChannelAdapter): + channel_id: ClassVar[str] = "googlechat" + channel_type: ClassVar[ChannelType] = ChannelType.GOOGLE_CHAT + webhook_path: ClassVar[str | None] = "/api/webhook/googlechat" + + text_chunk_limit: ClassVar[int] = 4000 + supports_markdown: ClassVar[bool] = True + supports_streaming: ClassVar[bool] = True + streaming_modes: ClassVar[list[str]] = ["off", "partial", "block"] + max_media_size_mb: ClassVar[int] = 20 + min_send_interval_ms: ClassVar[int] = 1000 + + capabilities = ChannelCapabilities( + chat_types=["direct", "group", "thread"], + reply=True, + threads=True, + media=True, + reactions=True, + edit=True, + unsend=True, + supports_markdown=True, + supports_streaming=True, + streaming_modes=["off", "partial", "block"], + text_chunk_limit=4000, + max_media_size_mb=20, + ) + meta = ChannelMeta( + id="googlechat", + label="Google Chat", + aliases=["gchat", "google-chat"], + markdown_capable=True, + selection_label="Google Chat (Google Workspace)", + system_image="https://www.gstatic.com/companion/icon_assets/chat_2x.png", + docs_path="/docs/channels/googlechat", + docs_label="Google Chat Setup Guide", + selection_docs_prefix="googlechat", + blurb="AI bot for Google Chat spaces and direct messages via Google Workspace", + ) + + def __init__(self, config: dict[str, Any] | None = None): + super().__init__(config) + self._status = ChannelStatus.DISCONNECTED + self._chat_service = None + self._credentials = None + self._service_account_email: str = "" + self._pubsub_subscription: str = "" + self._connected_at: float | None = None + self._stream_mgr = StreamManager( + update_interval_ms=config.get("streamUpdateIntervalMs", 800) if config else 800, + coalesce_min_chars=config.get("blockStreamingCoalesceMinChars", 1500) if config else 1500, + coalesce_idle_ms=config.get("blockStreamingCoalesceIdleMs", 1000) if config else 1000, + ) + self._rate_limiter = ChatRateLimiter() + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0) + self._webhook_semaphore = asyncio.Semaphore(64) + self._auth_cache: dict[str, Any] = {} + self._auth_cache_max = 32 + self._credential_source: str | None = None + self._audience_type = self._normalize_audience_type(config) + self._audience = config.get("audience", "") if config else "" + self._bot_user = config.get("bot_user", config.get("botUser", "")) if config else "" + self._app_principal = config.get("app_principal", config.get("appPrincipal", "")) if config else "" + self._allow_bots = bool(config.get("allow_bots", config.get("allowBots", False))) if config else False + self._reply_to_mode = config.get("reply_to_mode", config.get("replyToMode", "off")) if config else "off" + self._actions = config.get("actions", {}) if config else {} + self._proxy_config = proxy.resolve_proxy_config(config) if config else proxy.resolve_proxy_config() + self._tls_config = proxy.resolve_tls_config(config) if config else {} + self._default_account = ( + config.get("default_account", config.get("defaultAccount", "default")) if config else "default" + ) + self._per_account_reply_to_mode: dict[str, str] = self._parse_per_account_reply_to_mode(config) + self._policy = GoogleChatPolicy.from_config(config) + self._accounts_config: dict[str, dict[str, Any]] = self._parse_accounts_config(config) + self._message_queues: dict[str, asyncio.Queue] = {} + self._message_queue_tasks: dict[str, asyncio.Task] = {} + + async def connect(self) -> None: + self._status = ChannelStatus.CONNECTING + + try: + await asyncio.wait_for( + self._connect_impl(), + timeout=_CONNECT_TIMEOUT_S, + ) + except TimeoutError: + self._status = ChannelStatus.ERROR + raise ChannelException( + f"Google Chat connection timed out after {_CONNECT_TIMEOUT_S}s", + retryable=True, + ) + + self._status = ChannelStatus.CONNECTED + self._connected_at = time.time() + logger.info(f"Google Chat connected. SA: {self._service_account_email}") + + async def _connect_impl(self) -> None: + await self._init_service_account() + + try: + self._chat_service.spaces().list(pageSize=1).execute() + except Exception as e: + raise ChannelException( + f"Google Chat API probe failed: {e}", + retryable=False, + ) + + await self._init_pubsub_subscription() + + @staticmethod + def _normalize_audience_type(config: dict[str, Any] | None) -> str: + if not config: + return "app-url" + raw = str(config.get("audience_type", config.get("audienceType", ""))).strip().lower() + if not raw: + return "app-url" + canonical = _AUDIENCE_TYPE_CANONICAL.get(raw) + if canonical: + return canonical + logger.warning(f"Unknown audienceType '{raw}', falling back to 'app-url'") + return "app-url" + + async def disconnect(self) -> None: + self._status = ChannelStatus.DISCONNECTED + self._stream_mgr.clear() + self._circuit_breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0) + self._chat_service = None + self._credentials = None + for task in self._message_queue_tasks.values(): + task.cancel() + self._message_queue_tasks.clear() + self._message_queues.clear() + + async def probe(self) -> dict[str, Any]: + if not self._chat_service: + return {"ok": False, "status": "not_connected", "error": "chat_service not initialized"} + try: + result = self._chat_service.spaces().list(pageSize=1).execute() + return { + "ok": True, + "status": "connected", + "spaces_count_hint": len(result.get("spaces", [])), + "service_account_email": self._service_account_email, + "credential_source": self._credential_source, + "audience_type": self._audience_type, + } + except Exception as e: + return {"ok": False, "status": "error", "error": str(e)} + + async def send(self, response: ChannelResponse) -> DeliveryResult: + chat_id = response.identity.channel_chat_id + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + + if chat_id.startswith("users/"): + resolved = await self.find_direct_message_space(chat_id) + if resolved: + response.identity.channel_chat_id = resolved + chat_id = resolved + else: + return DeliveryResult(success=False, error=f"Cannot resolve user target to DM space: {chat_id}") + + body = format_outbound(response) + + async def _do_send(): + return await send_message(self._chat_service, chat_id, body) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def find_direct_message_space(self, user_name: str) -> str | None: + from .send import find_direct_message_space as _find_dm + + try: + return await _find_dm(self._chat_service, user_name) + except Exception as e: + logger.warning(f"find_direct_message_space failed for {user_name}: {e}") + return None + + async def resolve_outbound_space(self, target: str) -> str | None: + from .send import resolve_outbound_space as _resolve + + try: + return await _resolve(self._chat_service, target) + except Exception as e: + logger.warning(f"resolve_outbound_space failed for {target}: {e}") + return None + + async def send_media(self, chat_id: str, media_type: str, data: Any) -> DeliveryResult: + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + + async def _do_send(): + if isinstance(data, str): + return await send_media(self._chat_service, chat_id, data) + if isinstance(data, bytes): + if media_type in ("image", "IMAGE"): + return await upload_image_message(self._chat_service, chat_id, data) + return await upload_file_message( + self._chat_service, + chat_id, + data, + filename="upload", + mime_type=media_type, + ) + return DeliveryResult(success=False, error=f"Unsupported media data type: {type(data)}") + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def edit_message(self, chat_id: str, msg_id: str, content: str) -> DeliveryResult: + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + body: dict[str, Any] = {"text": content} + + async def _do_send(): + return await update_message(self._chat_service, msg_id, body) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def delete_message(self, chat_id: str, msg_id: str) -> DeliveryResult: + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + + async def _do_send(): + return await _delete_message(self._chat_service, msg_id) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def send_reaction(self, chat_id: str, msg_id: str, emoji: str) -> DeliveryResult: + reactions_enabled = self._actions.get("reactions", True) + if not reactions_enabled: + return DeliveryResult(success=False, error="Agent reactions disabled by actions.reactions config") + + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + + async def _do_send(): + return await send_reaction(self._chat_service, msg_id, emoji) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def list_reactions(self, msg_id: str) -> list[dict[str, Any]]: + from .send import list_reactions as _list_reactions + + try: + return await _list_reactions(self._chat_service, msg_id) + except Exception as e: + logger.warning(f"list_reactions failed: {e}") + return [] + + async def delete_reaction(self, reaction_id: str) -> DeliveryResult: + from .send import delete_reaction as _delete_reaction + + try: + return await _delete_reaction(self._chat_service, reaction_id) + except Exception as e: + return DeliveryResult(success=False, error=str(e)) + + async def send_typing_indicator(self, chat_id: str) -> DeliveryResult: + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + body = format_outbound( + ChannelResponse( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content="", + ) + ) + + async def _do_send(): + return await send_message(self._chat_service, chat_id, body) + + try: + return await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + async def send_stream_chunk( + self, + chat_id: str, + msg_id: str, + chunk: str, + finished: bool, + ) -> DeliveryResult: + if not msg_id: + space_name = chat_id.split("/threads/")[0] + await self._rate_limiter.acquire(space_name) + body = format_outbound( + ChannelResponse( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content=chunk, + ) + ) + + async def _do_send(): + return await send_message(self._chat_service, chat_id, body) + + try: + result = await self._circuit_breaker.call(_do_send) + except CircuitBreakerOpenError: + return DeliveryResult(success=False, error="Circuit breaker open") + + if result.success and result.message_id: + self._stream_mgr.register_message(chat_id, result.message_id, chunk) + return result + + self._stream_mgr.append_text(chat_id, chunk) + + if not self._stream_mgr.should_update(chat_id) and not finished: + return DeliveryResult(success=True, message_id=msg_id) + + self._stream_mgr.mark_update(chat_id) + + result = await self._stream_mgr.send_update(self._chat_service, chat_id, finished=finished) + if result: + return result + + return DeliveryResult(success=False, error="No pending stream message") + + async def receive(self) -> AsyncIterator[ChannelMessage]: + return + yield # type: ignore[misc] + + def normalize_inbound(self, raw: dict) -> ChannelMessage: + return normalize_inbound(self.channel_id, self.channel_type, raw, self._bot_user) + + def format_outbound(self, response: ChannelResponse) -> dict: + return format_outbound(response) + + async def health_check(self) -> HealthStatus: + from datetime import datetime + + if self._status != ChannelStatus.CONNECTED: + return HealthStatus(status="unhealthy", last_error="not connected") + + start = time.monotonic() + try: + self._chat_service.spaces().list(pageSize=1).execute() + latency_ms = (time.monotonic() - start) * 1000 + + warnings = self.collect_security_warnings() + status_issues = self.collect_status_issues() + + return HealthStatus( + status="healthy", + latency_ms=latency_ms, + last_connected_at=datetime.fromtimestamp(self._connected_at, tz=UTC) if self._connected_at else None, + metadata={ + "service_account": self._service_account_email, + "project_id": self.config.get("project_id"), + "pubsub_subscription": self._pubsub_subscription, + "pending_stream_msgs": self._stream_mgr.pending_count, + "circuit_breaker_state": self._circuit_breaker.state, + "audience_type": self._audience_type, + "audience": self._audience, + "credential_source": self._credential_source, + "webhook_path": self._resolve_webhook_path(self.config.get("accountId", "default")), + "warnings": warnings, + "status_issues": status_issues, + }, + ) + except Exception as e: + return HealthStatus(status="unhealthy", last_error=str(e)) + + def collect_security_warnings(self) -> list[str]: + warnings: list[str] = [] + + if self._policy.dm_policy == "open": + warnings.append("dmPolicy is 'open': any user can DM the bot") + if self._policy.group_policy == "open": + warnings.append("groupPolicy is 'open': bot responds in all groups") + + allow_from = self._policy.allow_from + for entry in allow_from: + if "@" in entry and not entry.startswith("users/") and not entry.startswith("user:"): + warnings.append(f"allowFrom contains email '{entry}': consider migrating to 'users/' format") + + group_allow_from = self._policy.group_allow_from + for entry in group_allow_from: + if "@" in entry and not entry.startswith("spaces/"): + warnings.append( + f"groupAllowFrom contains email '{entry}': group allowlist should use 'spaces/' format" + ) + + if not self._audience: + warnings.append("audience is not configured: webhook JWT audience validation may be incomplete") + if not self._audience_type: + warnings.append("audienceType is not configured: using default 'app-url'") + + if self._policy.dm_policy == "allowlist" and not allow_from: + warnings.append("dmPolicy is 'allowlist' but allowFrom is empty: no user can DM the bot") + + if self._policy.group_policy == "allowlist" and not group_allow_from: + warnings.append("groupPolicy is 'allowlist' but groupAllowFrom is empty: bot won't respond in any group") + + return warnings + + def collect_status_issues(self) -> list[dict[str, Any]]: + issues: list[dict[str, Any]] = [] + + if not self._audience: + issues.append( + { + "severity": "warning", + "field": "audience", + "message": "audience is not configured: webhook JWT audience validation may be incomplete", + } + ) + if not self._audience_type: + issues.append( + { + "severity": "warning", + "field": "audienceType", + "message": "audienceType is not configured: using default 'app-url'", + } + ) + if not self._service_account_email: + issues.append( + { + "severity": "error", + "field": "serviceAccount", + "message": "No service account email resolved: credentials may not be loaded", + } + ) + + if self._policy.dm_policy == "open": + issues.append( + { + "severity": "info", + "field": "dmPolicy", + "message": "dmPolicy is 'open': any user can DM the bot", + } + ) + if self._policy.group_policy == "open": + issues.append( + { + "severity": "info", + "field": "groupPolicy", + "message": "groupPolicy is 'open': bot responds in all groups", + } + ) + + allow_from = self._policy.allow_from + has_email_entries = any( + "@" in entry and not entry.startswith("users/") and not entry.startswith("user:") for entry in allow_from + ) + if has_email_entries: + issues.append( + { + "severity": "warning", + "field": "allowFrom", + "message": "allowFrom contains email addresses: consider migrating to 'users/' format", + } + ) + + if not self.config.get("project_id"): + issues.append( + { + "severity": "info", + "field": "projectId", + "message": "GCP_PROJECT_ID not configured: Pub/Sub event subscription will not be initialized", + } + ) + + if self._status.value != "connected": + issues.append( + { + "severity": "error", + "field": "connection", + "message": f"Adapter status is '{self._status.value}': channel not operational", + } + ) + + return issues + + async def verify_webhook_signature(self, headers: dict, body: bytes) -> bool: + content_type = headers.get("Content-Type", headers.get("content-type", "")) + if "application/json" not in content_type: + logger.warning("Google Chat webhook: invalid Content-Type, expected application/json") + return False + + if len(body) > _PREAUTH_BODY_MAX_BYTES: + logger.warning(f"Google Chat webhook: body too large ({len(body)} bytes > {_PREAUTH_BODY_MAX_BYTES})") + return False + + body_json: dict = {} + try: + body_json = json.loads(body) + except (json.JSONDecodeError, TypeError): + pass + + common_event = body_json.get("commonEventObject", {}) + if common_event: + if not self._validate_addon_common_event(common_event): + return False + auth_header = headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + token = auth_header.removeprefix("Bearer ").strip() + if token: + return self._verify_addon_token(token, body_json) + logger.debug("Google Chat webhook: accepted Add-on commonEventObject format (no token)") + return True + + auth_header = headers.get("Authorization", "") + webhook_urls = self._get_webhook_urls(body_json) + + if auth_header.startswith("Bearer "): + token = auth_header.removeprefix("Bearer ").strip() + if not token: + return False + + if self._audience_type == "project-number" and self._audience: + return self._verify_project_number_token(token) + + for url in webhook_urls: + if _verify_jwt(token, url): + return True + return False + + system_token = body_json.get("systemIdToken", "") + if system_token: + if self._audience_type == "project-number" and self._audience: + return self._verify_project_number_token(system_token) + + for url in webhook_urls: + if _verify_jwt(system_token, url): + return True + return False + + logger.warning("Google Chat webhook: no Authorization header or systemIdToken found") + return False + + def verify_webhook_with_status(self, headers: dict, body: bytes) -> tuple[bool, int, str]: + if not body: + return False, 400, "Empty request body" + + content_type = headers.get("Content-Type", headers.get("content-type", "")) + if "application/json" not in content_type: + return False, 400, "Invalid Content-Type, expected application/json" + + if len(body) > _PREAUTH_BODY_MAX_BYTES: + return False, 400, f"Body too large ({len(body)} bytes > {_PREAUTH_BODY_MAX_BYTES})" + + try: + body_json = json.loads(body) + except (json.JSONDecodeError, TypeError): + return False, 400, "Invalid JSON body" + + common_event = body_json.get("commonEventObject", {}) + if common_event: + if not self._validate_addon_common_event(common_event): + return False, 401, "Add-on event validation failed" + auth_header = headers.get("Authorization", "") + if auth_header.startswith("Bearer "): + token = auth_header.removeprefix("Bearer ").strip() + if token: + if self._verify_addon_token(token, body_json): + return True, 200, "OK" + return False, 401, "Add-on token verification failed" + return True, 200, "OK" + + auth_header = headers.get("Authorization", "") + webhook_urls = self._get_webhook_urls(body_json) + + if auth_header.startswith("Bearer "): + token = auth_header.removeprefix("Bearer ").strip() + if not token: + return False, 401, "Empty Bearer token" + if self._audience_type == "project-number" and self._audience: + if self._verify_project_number_token(token): + return True, 200, "OK" + return False, 401, "Project-number token verification failed" + for url in webhook_urls: + if _verify_jwt(token, url): + return True, 200, "OK" + return False, 401, "JWT verification failed" + + system_token = body_json.get("systemIdToken", "") + if system_token: + if self._audience_type == "project-number" and self._audience: + if self._verify_project_number_token(system_token): + return True, 200, "OK" + return False, 401, "Project-number systemIdToken verification failed" + for url in webhook_urls: + if _verify_jwt(system_token, url): + return True, 200, "OK" + return False, 401, "systemIdToken verification failed" + + return False, 401, "No valid authentication found" + + def _verify_project_number_token(self, token: str) -> bool: + import asyncio + + try: + cert_cache = auth.get_cert_cache() + except Exception as e: + logger.warning(f"Failed to get cert cache: {e}") + return False + + loop = None + try: + loop = asyncio.get_running_loop() + except RuntimeError: + loop = self._get_or_create_event_loop() + + if loop is not None: + certs = loop.run_until_complete(cert_cache.get_certs()) if not loop.is_running() else {} + else: + import asyncio as _asyncio + + certs = _asyncio.run(cert_cache.get_certs()) + + return auth.verify_project_number_token(token, self._audience, certs) + + @staticmethod + def _get_or_create_event_loop(): + import asyncio as _asyncio + + try: + return _asyncio.get_event_loop() + except RuntimeError: + loop = _asyncio.new_event_loop() + _asyncio.set_event_loop(loop) + return loop + + def _get_webhook_urls(self, body_json: dict) -> list[str]: + base_url = self.config.get("base_url", "") + default_path = self.config.get("webhook_path", "/api/webhook/googlechat") + urls = [self.config.get("webhook_url", f"{base_url}{default_path}")] + + account_id = self._extract_account_id_from_event(body_json) + if account_id: + account_path = self._resolve_webhook_path(account_id) + account_url = self.config.get("webhook_url", f"{base_url}{account_path}") + if account_url not in urls: + urls.append(account_url) + + return urls + + def _extract_account_id_from_event(self, body_json: dict) -> str | None: + space = body_json.get("event", {}).get("space", {}) or body_json.get("space", {}) + space_name = space.get("name", "") + if not space_name: + return None + + for account_id, cfg in self._accounts_config.items(): + project_id = cfg.get("project_id", "") + if project_id and project_id in space_name: + return account_id + + return None + + def _validate_addon_common_event(self, common_event: dict) -> bool: + if not common_event.get("type"): + return False + if self._app_principal: + user = common_event.get("user", {}) + if user.get("name", "") != self._app_principal: + logger.warning( + f"Add-on event appPrincipal mismatch: expected {self._app_principal}, got {user.get('name', '')}" + ) + return False + return True + + def _verify_addon_token(self, token: str, body_json: dict) -> bool: + import re + + try: + from google.auth.transport import requests as ga_requests + from google.oauth2 import id_token + except ImportError: + logger.info("google-auth not installed, accepting Add-on token without verification") + return True + + try: + payload = id_token.verify_oauth2_token(token, ga_requests.Request(), audience=None) + except Exception as e: + logger.warning(f"Add-on token verification failed: {e}") + return False + + issuer = payload.get("iss", "") + is_standard = bool(re.match(_ADDON_ISSUER_RE, issuer)) + is_sa_issuer = bool(re.match(_ADDON_SA_ISSUER_PATTERN, issuer)) + + if not is_standard and not is_sa_issuer: + logger.warning(f"Add-on token issuer mismatch: {issuer}") + return False + + event_user = body_json.get("commonEventObject", {}).get("user", {}) + event_email = event_user.get("email", "") + token_email = payload.get("email", "") + if event_email and token_email and event_email.lower() != token_email.lower(): + logger.warning(f"Add-on token email mismatch: event={event_email}, token={token_email}") + return False + + return True + + async def get_user_info(self, channel_user_id: str) -> dict[str, Any]: + try: + result = self._chat_service.users().get(userId=channel_user_id).execute() + return { + "name": result.get("name", ""), + "display_name": result.get("displayName", ""), + "email": result.get("email", ""), + "avatar_url": result.get("avatarUrl", ""), + } + except Exception: + return {} + + async def download_media(self, file_id: str) -> bytes: + if not self._chat_service: + raise ChannelNotConnectedError() + return await _download_media(self._chat_service, file_id) + + async def _refresh_token_if_needed(self) -> bool: + if not self._credentials or not self._credentials.valid: + return False + try: + ga_request = proxy.build_google_auth_request(self._proxy_config) + if not ga_request: + return False + if self._credentials.expired and self._credentials.refresh_token: + self._credentials.refresh(ga_request) + logger.info("Google Chat credentials refreshed") + return True + except Exception as e: + logger.warning(f"Token refresh failed: {e}") + return False + + async def _handle_pubsub_event(self, event_data: dict) -> None: + async with self._webhook_semaphore: + common_event = event_data.get("commonEventObject") + if common_event: + event_data = self._convert_addon_event(common_event) + + message = event_data.get("event", {}).get("message", {}) + if not self._allow_bots and is_bot_message(message): + logger.debug("Google Chat webhook: filtered bot message") + return + + msg = self.normalize_inbound(event_data) + if not self._policy.check_inbound(msg): + logger.debug( + "Google Chat webhook: rejected by policy: " + f"user={msg.identity.channel_user_id}, " + f"space={msg.metadata.get('space_name')}" + ) + return + + space_name = msg.metadata.get("space_name", "") + queue = self._message_queues.setdefault(space_name, asyncio.Queue()) + await queue.put(msg) + + if space_name not in self._message_queue_tasks or self._message_queue_tasks[space_name].done(): + task = asyncio.ensure_future(self._process_message_queue(space_name)) + self._message_queue_tasks[space_name] = task + + async def _process_message_queue(self, space_name: str) -> None: + queue = self._message_queues.get(space_name) + if queue is None: + return + try: + while not queue.empty(): + msg = await queue.get() + try: + if msg.chat_type == "direct" and self._policy.dm_policy == "pairing": + user_id = msg.identity.channel_user_id + if self._chat_service: + from .pairing import check_pairing_approval, send_pairing_challenge + + if not await check_pairing_approval( + self._chat_service, space_name, user_id, msg.content or "" + ): + await send_pairing_challenge(self._chat_service, space_name, user_id) + queue.task_done() + continue + + if msg.message_type == "command" and msg.metadata.get("slash_command"): + handled = await self._handle_slash_command(msg, space_name) + if handled: + queue.task_done() + continue + + per_group_prompt = self._policy.get_per_group_system_prompt(space_name) + if per_group_prompt: + msg.metadata["per_group_system_prompt"] = per_group_prompt + + await self._handle_message(msg) + except Exception as e: + logger.error(f"Error processing message in {space_name}: {e}") + finally: + queue.task_done() + except asyncio.CancelledError: + pass + except Exception as e: + logger.error(f"Message queue processing error for {space_name}: {e}") + + async def _handle_slash_command(self, msg: ChannelMessage, space_name: str) -> bool: + command = msg.metadata.get("slash_command", "") + chat_id = msg.identity.channel_chat_id + + if command == "/help": + from .slash_commands import get_command_help + + help_text = get_command_help() + response = ChannelResponse( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content=help_text, + ) + await self.send(response) + return True + + if command == "/status": + health = await self.health_check() + status_text = ( + f"*Google Chat Bot 状态*\n" + f"• 状态: {health.status}\n" + f"• 服务帐户: {self._service_account_email}\n" + f"• 延迟: {health.latency_ms:.0f}ms\n" + f"• 待处理流消息: {self._stream_mgr.pending_count}\n" + f"• 熔断器: {self._circuit_breaker.state}" + ) + response = ChannelResponse( + identity=ChannelIdentity( + channel_id=self.channel_id, + channel_type=self.channel_type, + channel_user_id="", + channel_chat_id=chat_id, + ), + content=status_text, + ) + await self.send(response) + return True + + return False + + def _convert_addon_event(self, common_event: dict) -> dict: + event_type = common_event.get("type", "MESSAGE") + return { + "event": { + "type": event_type, + "space": common_event.get("space", {}), + "message": common_event.get("message", {}), + "user": common_event.get("user", {}), + } + } + + async def _init_service_account(self) -> None: + from googleapiclient.discovery import build + + creds = await self._load_credentials() + if not creds: + raise ChannelAuthenticationError("No valid Google Chat service account credentials found") + + self._credentials = creds + self._service_account_email = getattr(creds, "service_account_email", "") + self._chat_service = build("chat", "v1", credentials=creds, cache_discovery=False) + + async def _load_credentials(self): + + SCOPES = ["https://www.googleapis.com/auth/chat.bot"] + + account_id = self.config.get("accountId", "default") + account_config = self._resolve_account_config(account_id) + + cache_key_prefix = self._build_auth_cache_key_prefix(account_config) + cached = self._auth_cache.get(cache_key_prefix) + if cached: + self._credential_source = "cache" + logger.debug(f"Auth cache hit for key: {cache_key_prefix}") + return cached + + self._credential_source = None + + creds = self._try_inline_credentials(SCOPES, account_config) + if creds: + self._cache_credential(cache_key_prefix, creds) + return creds + + creds = self._try_file_credentials(SCOPES, account_config) + if creds: + self._cache_credential(cache_key_prefix, creds) + return creds + + creds = self._try_env_json_credentials(SCOPES) + if creds: + self._cache_credential(cache_key_prefix, creds) + return creds + + creds = self._try_env_file_credentials(SCOPES) + if creds: + self._cache_credential(cache_key_prefix, creds) + return creds + + creds = self._try_secret_ref_credentials(SCOPES, account_config) + if creds: + self._cache_credential(cache_key_prefix, creds) + return creds + + return None + + def _cache_credential(self, cache_key: str, creds) -> None: + if cache_key in self._auth_cache: + cached = self._auth_cache[cache_key] + cached_email = getattr(cached, "service_account_email", "") + new_email = getattr(creds, "service_account_email", "") + if cached_email and new_email and cached_email != new_email: + logger.info(f"Auth cache: credential changed (email: {cached_email} -> {new_email}), rebuilding") + self._auth_cache.pop(cache_key) + if len(self._auth_cache) >= self._auth_cache_max: + first_key = next(iter(self._auth_cache)) + self._auth_cache.pop(first_key) + self._auth_cache[cache_key] = creds + + @staticmethod + def _build_auth_cache_key_prefix(account_config: dict[str, Any]) -> str: + raw = account_config.get("service_account") + if raw: + import hashlib + + if isinstance(raw, dict): + raw = json.dumps(raw, sort_keys=True) + inline_hash = hashlib.sha256(str(raw).encode()).hexdigest()[:12] + return f"inline:{inline_hash}" + file_path = account_config.get("service_account_file", "") + if file_path: + return f"file:{file_path}" + env_json = os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT", "") + if env_json: + import hashlib + + env_hash = hashlib.sha256(env_json.encode()).hexdigest()[:12] + return f"env_json:{env_hash}" + env_file = os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE", "") or os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_FILE", "") + if env_file: + return f"env_file:{env_file}" + secret_ref = account_config.get("serviceAccountRef", account_config.get("service_account_ref", "")) + if secret_ref: + return f"secret_ref:{secret_ref}" + return "auth:default" + + def _parse_accounts_config(self, config: dict[str, Any] | None) -> dict[str, dict[str, Any]]: + if not config: + return {} + accounts = config.get("accounts", {}) + if not isinstance(accounts, dict): + return {} + result: dict[str, dict[str, Any]] = {} + for key, value in accounts.items(): + if isinstance(value, dict) and value.get("enabled", True): + result[key] = value + return result + + def _resolve_account_config(self, account_id: str = "default") -> dict[str, Any]: + if account_id == "default" and self._default_account != "default": + account_id = self._default_account + merged = dict(self.config) + account_overrides = self._accounts_config.get(account_id, {}) + if account_overrides: + merged.update(account_overrides) + return merged + + @staticmethod + def _parse_per_account_reply_to_mode(config: dict[str, Any] | None) -> dict[str, str]: + if not config: + return {} + accounts = config.get("accounts", {}) + if not isinstance(accounts, dict): + return {} + result: dict[str, str] = {} + for account_id, cfg in accounts.items(): + if isinstance(cfg, dict): + mode = cfg.get("replyToMode", cfg.get("reply_to_mode", "")) + if mode: + result[account_id] = mode + return result + + def get_reply_to_mode(self, account_id: str = "default") -> str: + per_account = self._per_account_reply_to_mode.get(account_id, "") + if per_account: + return per_account + return self._reply_to_mode + + def list_accounts(self) -> list[dict[str, Any]]: + accounts = [] + if not self._accounts_config: + return [{"account_id": "default", "email": self._service_account_email}] + + for account_id, cfg in self._accounts_config.items(): + accounts.append( + { + "account_id": account_id, + "email": cfg.get("service_account_email", ""), + "project_id": cfg.get("project_id", ""), + "enabled": cfg.get("enabled", True), + } + ) + return accounts + + def _resolve_webhook_path(self, account_id: str | None = None) -> str: + if account_id and account_id in self._accounts_config: + override = self._accounts_config[account_id].get("webhook_path", "") + if override: + return override + return f"/api/webhook/googlechat/{account_id}" + return "/api/webhook/googlechat" + + def _try_inline_credentials(self, scopes: list[str], config: dict[str, Any] | None = None): + cfg = config if config is not None else self.config + raw = cfg.get("service_account") + if not raw: + return None + + sa_json = self._validate_service_account_json(raw) + if not sa_json: + return None + + from google.oauth2 import service_account + + creds = service_account.Credentials.from_service_account_info(sa_json, scopes=scopes) + self._credential_source = "inline" + return creds + + def _try_file_credentials(self, scopes: list[str], config: dict[str, Any] | None = None): + cfg = config if config is not None else self.config + file_path = cfg.get("service_account_file", "") + if not file_path: + return None + + expanded = os.path.expanduser(file_path) + if not self._validate_credential_file(expanded): + return None + + sa_json = self._load_and_validate_sa_file(expanded) + if not sa_json: + return None + + from google.oauth2 import service_account + + creds = service_account.Credentials.from_service_account_info(sa_json, scopes=scopes) + self._credential_source = "file" + return creds + + def _try_env_json_credentials(self, scopes: list[str]): + env_val = os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT", "") + if not env_val: + return None + + from google.oauth2 import service_account + + try: + sa_json = json.loads(env_val) + except (json.JSONDecodeError, TypeError): + logger.warning("GOOGLE_CHAT_SERVICE_ACCOUNT is not valid JSON") + return None + + sa_json = self._validate_service_account_json(sa_json) + if not sa_json: + return None + + creds = service_account.Credentials.from_service_account_info(sa_json, scopes=scopes) + self._credential_source = "env" + return creds + + def _try_env_file_credentials(self, scopes: list[str]): + file_path = os.getenv("GOOGLE_SERVICE_ACCOUNT_FILE", "") or os.getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_FILE", "") + if not file_path: + return None + + expanded = os.path.expanduser(file_path) + if not self._validate_credential_file(expanded): + return None + + sa_json = self._load_and_validate_sa_file(expanded) + if not sa_json: + return None + + from google.oauth2 import service_account + + creds = service_account.Credentials.from_service_account_info(sa_json, scopes=scopes) + self._credential_source = "env" + return creds + + def _try_secret_ref_credentials(self, scopes: list[str], config: dict[str, Any] | None = None): + cfg = config if config is not None else self.config + secret_ref = cfg.get("serviceAccountRef", cfg.get("service_account_ref", "")) + if not secret_ref: + return None + + try: + from yuxi.channels.secret import resolve_secret_ref + + resolved = resolve_secret_ref(secret_ref) + if not resolved: + return None + sa_json = resolved + if isinstance(sa_json, str): + try: + sa_json = json.loads(sa_json) + except (json.JSONDecodeError, TypeError): + logger.warning(f"serviceAccountRef resolved to invalid JSON: {secret_ref}") + return None + sa_json = self._validate_service_account_json(sa_json) + if not sa_json: + return None + from google.oauth2 import service_account + + creds = service_account.Credentials.from_service_account_info(sa_json, scopes=scopes) + self._credential_source = "secret_ref" + return creds + except ImportError: + logger.warning("Secret ref resolution not available") + return None + except Exception as e: + logger.warning(f"Failed to resolve serviceAccountRef '{secret_ref}': {e}") + return None + + def _validate_credential_file(self, path: str) -> bool: + if not os.path.isfile(path): + logger.warning(f"Credential path is not a regular file: {path}") + return False + + try: + size = os.path.getsize(path) + except OSError: + logger.warning(f"Cannot stat credential file: {path}") + return False + + if size > 64 * 1024: + logger.warning(f"Credential file too large ({size} bytes > 64KB): {path}") + return False + + if size == 0: + logger.warning(f"Credential file is empty: {path}") + return False + + return True + + def _load_and_validate_sa_file(self, path: str) -> dict | None: + try: + with open(path) as f: + sa_json = json.load(f) + except (json.JSONDecodeError, OSError) as e: + logger.warning(f"Failed to read credential file {path}: {e}") + return None + + return self._validate_service_account_json(sa_json) + + def _validate_service_account_json(self, sa_json: dict | str) -> dict | None: + if isinstance(sa_json, str): + try: + sa_json = json.loads(sa_json) + except (json.JSONDecodeError, TypeError): + return None + + if not isinstance(sa_json, dict): + return None + + if sa_json.get("type") != "service_account": + logger.warning("Credential type is not 'service_account'") + return None + + if not sa_json.get("private_key"): + logger.warning("Credential missing 'private_key'") + return None + + if not sa_json.get("client_email"): + logger.warning("Credential missing 'client_email'") + return None + + if not sa_json.get("token_uri"): + logger.warning("Credential missing 'token_uri'") + return None + + auth_uri = sa_json.get("auth_uri", "") + if auth_uri and "google.com" not in auth_uri: + logger.warning(f"Credential auth_uri points to non-Google domain: {auth_uri}") + if not auth_uri: + logger.debug("Credential missing 'auth_uri' (may be valid for some setups)") + + client_x509 = sa_json.get("client_x509_cert_url", "") + if not client_x509: + logger.debug("Credential missing 'client_x509_cert_url' (may be valid for some setups)") + + universe_domain = sa_json.get("universe_domain", "googleapis.com") + if universe_domain != "googleapis.com": + logger.warning(f"Credential universe_domain is '{universe_domain}', expected 'googleapis.com'") + + client_email = sa_json["client_email"] + if client_email: + self._service_account_email = client_email + + return sa_json + + async def _init_pubsub_subscription(self) -> None: + project_id = self.config.get("project_id") + if not project_id: + logger.info("GCP_PROJECT_ID not configured, skipping Pub/Sub init") + return + + if not self._credentials: + logger.warning("No credentials available, skipping Pub/Sub init") + return + + try: + from google.cloud import pubsub_v1 + except ImportError: + logger.warning("google-cloud-pubsub not installed, skipping Pub/Sub") + return + + topic = self.config.get("pubsub_topic", "forcepilot-googlechat-events") + sub = self.config.get("pubsub_subscription", "forcepilot-googlechat-sub") + + try: + subscriber = pubsub_v1.SubscriberClient(credentials=self._credentials) + sub_path = subscriber.subscription_path(project_id, sub) + + try: + subscriber.get_subscription(subscription=sub_path) + logger.info(f"Pub/Sub subscription already exists: {sub_path}") + except Exception: + base_url = self.config.get("base_url", "") + webhook_url = self.config.get( + "webhook_url", + f"{base_url}/api/webhook/googlechat", + ) + push_config = pubsub_v1.PushConfig(push_endpoint=webhook_url) + topic_path = subscriber.topic_path(project_id, topic) + + try: + subscriber.create_subscription( + name=sub_path, + topic=topic_path, + push_config=push_config, + ack_deadline_seconds=60, + ) + logger.info(f"Created Pub/Sub subscription: {sub_path} -> {webhook_url}") + except Exception as create_err: + if hasattr(create_err, "code") and getattr(create_err, "code") == 409: + logger.info(f"Pub/Sub subscription already exists (race): {sub_path}") + else: + logger.error(f"Failed to create Pub/Sub subscription: {create_err}") + return + + self._pubsub_subscription = sub_path + except Exception as e: + logger.error(f"Pub/Sub initialization failed for project={project_id}: {e}") diff --git a/backend/package/yuxi/channels/adapters/googlechat/approval_auth.py b/backend/package/yuxi/channels/adapters/googlechat/approval_auth.py new file mode 100644 index 00000000..3a735755 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/approval_auth.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + + +def normalize_approver_id(raw: str) -> str: + if not raw: + return "" + if raw.startswith("users/") or raw.startswith("user:"): + return raw + if "@" in raw: + return f"users/{raw}" + return f"users/{raw}" + + +def resolve_approver_ids( + dm_allow_from: list[str], + default_to: str = "", +) -> list[str]: + approvers: list[str] = [] + for entry in dm_allow_from: + if entry == "*": + continue + normalized = normalize_approver_id(entry) + if normalized: + approvers.append(normalized) + if not approvers and default_to: + approvers.append(normalize_approver_id(default_to)) + return approvers + + +def build_exec_approval_context( + action_id: str, + approvers: list[str], + metadata: dict[str, Any] | None = None, +) -> dict[str, Any]: + return { + "action_id": action_id, + "approvers": approvers, + "approved_by": [], + "rejected_by": [], + "metadata": metadata or {}, + } + + +def record_approval_decision( + context: dict[str, Any], + user_id: str, + approved: bool, +) -> bool: + normalized = normalize_approver_id(user_id) + if normalized not in context.get("approvers", []): + logger.warning(f"User {normalized} is not an authorized approver") + return False + if approved: + if normalized in context.get("approved_by", []): + return True + context.setdefault("approved_by", []).append(normalized) + else: + if normalized in context.get("rejected_by", []): + return True + context.setdefault("rejected_by", []).append(normalized) + return True + + +def is_fully_approved(context: dict[str, Any]) -> bool: + approvers = context.get("approvers", []) + approved = context.get("approved_by", []) + return bool(approvers) and set(approvers) == set(approved) diff --git a/backend/package/yuxi/channels/adapters/googlechat/audit.py b/backend/package/yuxi/channels/adapters/googlechat/audit.py new file mode 100644 index 00000000..eb4af26e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/audit.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.utils.logging_config import logger + + +async def check_space_membership( + chat_service: Resource, + space_name: str, + user_name: str, +) -> bool: + try: + membership = chat_service.spaces().members().get(name=f"{space_name}/members/{user_name}").execute() + return membership is not None + except Exception as e: + _log_audit_error("check_space_membership", space_name, e) + return False + + +async def list_space_members( + chat_service: Resource, + space_name: str, +) -> list[dict]: + try: + result = chat_service.spaces().members().list(parent=space_name).execute() + return result.get("memberships", []) + except Exception as e: + _log_audit_error("list_space_members", space_name, e) + return [] + + +async def get_member_role( + chat_service: Resource, + space_name: str, + user_name: str, +) -> str | None: + try: + membership = chat_service.spaces().members().get(name=f"{space_name}/members/{user_name}").execute() + return membership.get("role") + except Exception as e: + _log_audit_error("get_member_role", space_name, e) + return None + + +def _log_audit_error(operation: str, space_name: str, error: Exception) -> None: + try: + from googleapiclient.errors import HttpError + + if isinstance(error, HttpError): + logger.warning( + f"Audit operation '{operation}' failed for {space_name}: HTTP {error.resp.status} {error.resp.reason}" + ) + return + except ImportError: + pass + logger.warning(f"Audit operation '{operation}' failed for {space_name}: {error}") diff --git a/backend/package/yuxi/channels/adapters/googlechat/auth.py b/backend/package/yuxi/channels/adapters/googlechat/auth.py new file mode 100644 index 00000000..f7f87e65 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/auth.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import json +import time + +import httpx + +from yuxi.utils.logging_config import logger + +_CERT_URL = "https://www.googleapis.com/service_accounts/v1/metadata/x509/chat@system.gserviceaccount.com" +_CERT_CACHE_TTL_S = 600 +_CERT_FETCH_TIMEOUT_S = 10.0 + + +class GoogleChatCertCache: + def __init__(self, ttl_s: int = _CERT_CACHE_TTL_S): + self._ttl = ttl_s + self._cache: dict[str, str] | None = None + self._cache_at: float = 0.0 + + @property + def is_valid(self) -> bool: + return self._cache is not None and (time.monotonic() - self._cache_at) < self._ttl + + async def get_certs(self) -> dict[str, str]: + if self.is_valid and self._cache is not None: + return dict(self._cache) + + try: + async with httpx.AsyncClient(timeout=_CERT_FETCH_TIMEOUT_S) as client: + resp = await client.get(_CERT_URL) + resp.raise_for_status() + certs: dict[str, str] = resp.json() + self._cache = certs + self._cache_at = time.monotonic() + logger.info(f"Google Chat cert cache refreshed: {len(certs)} certificates") + return dict(certs) + except Exception as e: + logger.warning(f"Failed to fetch Google Chat certs: {e}") + if self._cache is not None: + logger.info("Using stale cert cache") + return dict(self._cache) + return {} + + def clear(self) -> None: + self._cache = None + self._cache_at = 0.0 + + +def verify_project_number_token(token: str, project_number: str, certs: dict[str, str]) -> bool: + try: + from google.auth import jwt + except ImportError: + logger.warning("google-auth not installed, skipping project-number token verification") + return True + + if not certs: + logger.warning("No certificates available for project-number verification") + return False + + certs_json = json.dumps(certs) + try: + payload = jwt.decode(token, certs=certs_json, audience=project_number) + issuer = payload.get("iss", "") + expected_issuer = "chat@system.gserviceaccount.com" + if issuer != expected_issuer: + logger.warning(f"project-number token issuer mismatch: expected '{expected_issuer}', got '{issuer}'") + return False + return True + except Exception as e: + logger.warning(f"project-number token verification failed: {e}") + return False + + +_global_cert_cache: GoogleChatCertCache | None = None + + +def get_cert_cache() -> GoogleChatCertCache: + global _global_cert_cache + if _global_cert_cache is None: + _global_cert_cache = GoogleChatCertCache() + return _global_cert_cache diff --git a/backend/package/yuxi/channels/adapters/googlechat/cards.py b/backend/package/yuxi/channels/adapters/googlechat/cards.py new file mode 100644 index 00000000..2a1a8691 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/cards.py @@ -0,0 +1,218 @@ +from __future__ import annotations + + +def build_approval_card(title: str, action_id: str, confirm_text: str = "确认", reject_text: str = "拒绝") -> dict: + return { + "cards_v2": [ + { + "card_id": action_id, + "card": { + "header": {"title": title, "imageType": "SQUARE"}, + "sections": [ + { + "widgets": [ + { + "buttonList": { + "buttons": [ + _make_button(confirm_text, "approve", green=True), + _make_button(reject_text, "reject", green=False), + ] + } + } + ] + } + ], + }, + } + ] + } + + +def build_poll_card(question: str, options: list[str], action_id: str) -> dict: + buttons = [_make_button(opt, f"poll_vote_{i}_{action_id}") for i, opt in enumerate(options)] + + sections: list[dict] = [] + while buttons: + chunk = buttons[:2] + buttons = buttons[2:] + sections.append({"widgets": [{"buttonList": {"buttons": chunk}}]}) + + return { + "cards_v2": [ + { + "card_id": action_id, + "card": { + "header": {"title": question, "imageType": "SQUARE"}, + "sections": sections, + }, + } + ] + } + + +def build_info_card(title: str, fields: dict[str, str]) -> dict: + widgets = [] + for key, value in fields.items(): + widgets.append( + { + "decoratedText": { + "topLabel": key, + "text": value, + } + } + ) + + return { + "cards_v2": [ + { + "card": { + "header": {"title": title}, + "sections": [{"widgets": widgets}], + } + } + ] + } + + +def build_form_card(title: str, fields: list[dict], submit_action: str) -> dict: + widgets = [] + for field in fields: + field_input_type = field.get("input_type", "SINGLE_LINE") + + if field_input_type in ("SINGLE_SELECT", "MULTI_SELECT"): + widgets.append( + { + "selectionInput": { + "label": field.get("label", ""), + "name": field.get("name", ""), + "type": field_input_type, + "items": _build_selection_items(field.get("options", [])), + } + } + ) + elif field_input_type == "DIVIDER": + widgets.append({"divider": {}}) + else: + widgets.append( + { + "textInput": { + "label": field.get("label", ""), + "name": field.get("name", ""), + "type": field_input_type, + } + } + ) + + widgets.append({"buttonList": {"buttons": [_make_button("提交", submit_action, green=True)]}}) + + return { + "cards_v2": [ + { + "card": { + "header": {"title": title}, + "sections": [{"widgets": widgets}], + } + } + ] + } + + +def build_exec_approval_card( + title: str, + action_id: str, + detail_fields: dict[str, str] | None = None, +) -> dict: + widgets = [] + + if detail_fields: + for key, value in detail_fields.items(): + widgets.append( + { + "decoratedText": { + "topLabel": key, + "text": value, + } + } + ) + + widgets.append( + { + "buttonList": { + "buttons": [ + _make_button("确认执行", f"exec_approve_{action_id}", green=True), + _make_button("拒绝执行", f"exec_reject_{action_id}", green=False), + ] + } + } + ) + + return { + "cards_v2": [ + { + "card": { + "header": {"title": title, "imageType": "SQUARE"}, + "sections": [{"widgets": widgets}], + } + } + ] + } + + +def build_selection_card( + title: str, + selection_name: str, + label: str, + options: list[dict[str, str]], + selection_type: str = "SINGLE_SELECT", + submit_action: str | None = None, +) -> dict: + widgets = [ + { + "selectionInput": { + "label": label, + "name": selection_name, + "type": selection_type, + "items": _build_selection_items(options), + } + } + ] + + if submit_action: + widgets.append({"buttonList": {"buttons": [_make_button("提交", submit_action, green=True)]}}) + + return { + "cards_v2": [ + { + "card": { + "header": {"title": title}, + "sections": [{"widgets": widgets}], + } + } + ] + } + + +def _make_divider() -> dict: + return {"divider": {}} + + +def _build_selection_items(options: list[dict[str, str]]) -> list[dict]: + items = [] + for opt in options: + items.append( + { + "text": opt.get("text", opt.get("label", "")), + "value": opt.get("value", ""), + "selected": opt.get("selected", False), + } + ) + return items + + +def _make_button(text: str, action: str, green: bool = False) -> dict: + color = {"red": 0, "green": 0.6, "blue": 0} if green else {"red": 0.6, "green": 0, "blue": 0} + return { + "text": text, + "color": color, + "onClick": {"action": {"function": action}}, + } diff --git a/backend/package/yuxi/channels/adapters/googlechat/directory.py b/backend/package/yuxi/channels/adapters/googlechat/directory.py new file mode 100644 index 00000000..590bcaec --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/directory.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from typing import Any + + +def list_peers(allow_from: list[str]) -> list[dict[str, Any]]: + peers: list[dict[str, Any]] = [] + for entry in allow_from: + if entry == "*": + continue + normalized = normalize_id(entry) + peers.append({"id": normalized, "raw": entry, "type": "user"}) + return peers + + +def list_groups(groups_config: dict[str, Any]) -> list[dict[str, Any]]: + groups: list[dict[str, Any]] = [] + for space_name, cfg in groups_config.items(): + if not isinstance(cfg, dict): + continue + groups.append( + { + "id": space_name, + "type": "space", + "name": cfg.get("name", space_name), + "enabled": cfg.get("enabled", True), + "requires_mention": cfg.get("require_mention", cfg.get("requireMention", False)), + } + ) + return groups + + +def normalize_id(raw: str) -> str: + if not raw: + return "" + raw = raw.strip() + if raw.startswith("spaces/"): + return raw + if raw.startswith("users/"): + return raw + if raw.startswith("user:"): + return raw.replace("user:", "users/", 1) + if "@" in raw: + return f"users/{raw}" + return f"users/{raw}" diff --git a/backend/package/yuxi/channels/adapters/googlechat/doctor.py b/backend/package/yuxi/channels/adapters/googlechat/doctor.py new file mode 100644 index 00000000..4b0c6a6a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/doctor.py @@ -0,0 +1,164 @@ +from __future__ import annotations + +import re +from typing import Any + + +def run_diagnostics(config: dict[str, Any] | None) -> list[dict[str, Any]]: + if not config: + return [] + + issues: list[dict[str, Any]] = [] + issues.extend(_check_mutable_allowlist(config)) + issues.extend(_check_deprecated_formats(config)) + issues.extend(_check_legacy_stream_mode(config)) + issues.extend(_check_credential_readiness(config)) + return issues + + +def _check_mutable_allowlist(config: dict[str, Any]) -> list[dict[str, Any]]: + issues: list[dict[str, Any]] = [] + + allow_from = _normalize_allowlist(config.get("allowFrom", config.get("allow_from", []))) + for entry in allow_from: + if "@" in entry and not _is_users_prefix(entry): + issues.append( + { + "severity": "warning", + "category": "mutable_allowlist", + "field": "allowFrom", + "value": entry, + "message": ( + f"allowFrom contains email '{entry}' without 'users/' prefix. " + "Email-based allowlists are mutable; consider migrating to 'users/' format" + ), + } + ) + + group_allow_from = _normalize_allowlist(config.get("groupAllowFrom", config.get("group_allow_from", []))) + for entry in group_allow_from: + if "@" in entry and not entry.startswith("spaces/"): + issues.append( + { + "severity": "warning", + "category": "mutable_allowlist", + "field": "groupAllowFrom", + "value": entry, + "message": ( + f"groupAllowFrom contains email '{entry}'. " + "Group allowlists should use 'spaces/' format as emails are mutable" + ), + } + ) + + return issues + + +def _check_deprecated_formats(config: dict[str, Any]) -> list[dict[str, Any]]: + issues: list[dict[str, Any]] = [] + + allow_from = _normalize_allowlist(config.get("allowFrom", config.get("allow_from", []))) + for entry in allow_from: + if _is_users_prefix_with_email_legacy(entry): + issues.append( + { + "severity": "warning", + "category": "deprecated_format", + "field": "allowFrom", + "value": entry, + "message": ( + f"allowFrom entry '{entry}' uses deprecated format. " + "Remove the trailing '/' portion: use 'users/' instead" + ), + } + ) + + if config.get("streamMode"): + issues.append( + { + "severity": "info", + "category": "deprecated_config", + "field": "streamMode", + "value": config["streamMode"], + "message": ( + "streamMode config is deprecated; " + "streaming is configured via 'supports_streaming' capability" + ), + } + ) + + if config.get("enableThreadReply"): + issues.append( + { + "severity": "info", + "category": "deprecated_config", + "field": "enableThreadReply", + "value": config["enableThreadReply"], + "message": "enableThreadReply is deprecated; use replyToMode instead", + } + ) + + return issues + + +def _check_legacy_stream_mode(config: dict[str, Any]) -> list[dict[str, Any]]: + issues: list[dict[str, Any]] = [] + if config.get("streamMode"): + issues.append( + { + "severity": "info", + "category": "legacy_config", + "field": "streamMode", + "value": config["streamMode"], + "message": "Remove 'streamMode' config key; streaming is now controlled by capability flags", + } + ) + return issues + + +def _check_credential_readiness(config: dict[str, Any]) -> list[dict[str, Any]]: + issues: list[dict[str, Any]] = [] + + has_inline = bool(config.get("service_account")) + has_file = bool(config.get("service_account_file")) + has_secret_ref = bool(config.get("serviceAccountRef", config.get("service_account_ref"))) + has_env = bool( + __import__("os").getenv("GOOGLE_CHAT_SERVICE_ACCOUNT") + or __import__("os").getenv("GOOGLE_SERVICE_ACCOUNT_FILE") + or __import__("os").getenv("GOOGLE_CHAT_SERVICE_ACCOUNT_FILE") + ) + + if not (has_inline or has_file or has_secret_ref or has_env): + issues.append( + { + "severity": "error", + "category": "credential_readiness", + "field": "serviceAccount", + "message": ( + "No Google Chat service account configured. " + "Provide one via service_account, service_account_file, " + "serviceAccountRef, or environment variables" + ), + } + ) + + return issues + + +def _normalize_allowlist(raw: Any) -> list[str]: + if isinstance(raw, str): + return [x.strip() for x in raw.split(",") if x.strip()] + if isinstance(raw, (list, tuple)): + return [str(x).strip() for x in raw if x] + return [] + + +def _is_users_prefix(entry: str) -> bool: + return entry.startswith("users/") or entry.startswith("user:") + + +def _is_users_prefix_with_email_legacy(entry: str) -> bool: + if not entry.startswith("users/"): + return False + inner = entry.removeprefix("users/") + return bool(re.match(r"^[\w.+-]+@[\w-]+\.[\w.-]+/", inner)) diff --git a/backend/package/yuxi/channels/adapters/googlechat/formatter.py b/backend/package/yuxi/channels/adapters/googlechat/formatter.py new file mode 100644 index 00000000..eef24b8b --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/formatter.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import re +from typing import Any + +from yuxi.channels.models import ChannelResponse + +from . import cards + +_MAX_TEXT_CHARS = 4000 + + +def format_outbound(response: ChannelResponse) -> dict[str, Any]: + body: dict[str, Any] = {} + content = sanitize_text(response.content or "") + if not content: + content = " " + body["text"] = _truncate_text(content) + + body = _apply_card(body, response) + + if response.attachments: + for att in response.attachments: + if att.url and att.type in ("image", "IMAGE", "video", "VIDEO"): + body.setdefault("cards_v2", []) + body["cards_v2"].append({"card": {"sections": [{"widgets": [{"image": {"imageUrl": att.url}}]}]}}) + + if response.metadata and response.metadata.get("thread_key"): + body["messageReplyOption"] = "REPLY_MESSAGE_FALLBACK_TO_NEW_THREAD" + body["thread"] = {"threadKey": response.metadata["thread_key"]} + + if response.metadata and response.metadata.get("silent"): + body["disableNotification"] = True + + return body + + +def _apply_card(body: dict[str, Any], response: ChannelResponse) -> dict[str, Any]: + if response.metadata and response.metadata.get("card"): + body["cards_v2"] = response.metadata["card"].get("cards_v2", []) + return body + + card_type = (response.metadata or {}).get("card_type", "") + if not card_type: + return body + + card_params = (response.metadata or {}).get("card_params", {}) + card_body = _build_card(card_type, response.content, card_params) + if card_body: + body["cards_v2"] = card_body.get("cards_v2", []) + + return body + + +def _build_card(card_type: str, content: str, params: dict[str, Any]) -> dict | None: + action_id = params.get("action_id", "card_action") + + if card_type == "approval": + return cards.build_approval_card( + title=params.get("title", content or "审批"), + action_id=action_id, + confirm_text=params.get("confirm_text", "确认"), + reject_text=params.get("reject_text", "拒绝"), + ) + + if card_type == "poll": + options = params.get("options", []) + if not options: + return None + return cards.build_poll_card( + question=params.get("title", content or "投票"), + options=options, + action_id=action_id, + ) + + if card_type == "info": + fields = params.get("fields", {}) + if not fields: + return None + return cards.build_info_card( + title=params.get("title", content or "信息"), + fields=fields, + ) + + if card_type == "form": + fields = params.get("fields", []) + if not fields: + return None + return cards.build_form_card( + title=params.get("title", content or "表单"), + fields=fields, + submit_action=params.get("submit_action", action_id), + ) + + if card_type == "selection": + options = params.get("options", []) + if not options: + return None + return cards.build_selection_card( + title=params.get("title", content or "选择"), + selection_name=params.get("selection_name", "selection"), + label=params.get("label", "请选择"), + options=options, + selection_type=params.get("selection_type", "SINGLE_SELECT"), + submit_action=params.get("submit_action"), + ) + + if card_type == "exec_approval": + return cards.build_exec_approval_card( + title=params.get("title", content or "执行审批"), + action_id=action_id, + detail_fields=params.get("detail_fields"), + ) + + return None + + +def resolve_reply_to_mode(config: dict | None = None, default: str = "off") -> str: + if not config: + return default + mode = str(config.get("reply_to_mode", config.get("replyToMode", default))).strip().lower() + if mode in ("first", "all", "off"): + return mode + return default + + +def sanitize_text(text: str) -> str: + text = text.replace("\x00", "").rstrip() + text = re.sub(r"[\x00-\x08\x0b\x0c\x0e-\x1f]", "", text) + return text + + +def chunk_text_for_outbound(text: str, chunk_limit: int = _MAX_TEXT_CHARS) -> list[str]: + if not text: + return [] + if len(text) <= chunk_limit: + return [text] + + chunks: list[str] = [] + remaining = text + + while remaining: + if len(remaining) <= chunk_limit: + chunks.append(remaining) + break + + split_at = _find_split_point(remaining, chunk_limit) + chunk = remaining[:split_at].rstrip() + chunks.append(chunk) + remaining = remaining[split_at:].lstrip() + + return chunks + + +def _find_split_point(text: str, limit: int) -> int: + window = text[:limit] + + code_block_match = re.search(r"```[\s\S]*?```", window) + if code_block_match and code_block_match.end() <= limit: + after_end = code_block_match.end() + if after_end < limit: + return after_end + + open_fence = re.search(r"```\w*\n", window) + if open_fence: + fence_start = open_fence.start() + closing = text.find("```", fence_start + 3) + if fence_start > 0 and (closing == -1 or closing >= limit): + return fence_start + + for pat in [r"\n\n", r"\n", r"\. ", r"。", r"\.\n"]: + matches = list(re.finditer(pat, window)) + if matches: + last = matches[-1] + boundary = last.end() + if boundary > limit * 0.6: + return boundary + + for pat_word in [r"\s", r"[,，;；:：!！?？]"]: + m = re.search(pat_word + r"[^\s]*$", window) + if m: + return m.start() + + return limit + + +def _truncate_text(text: str) -> str: + if len(text) <= _MAX_TEXT_CHARS: + return text + + truncated = text[: _MAX_TEXT_CHARS - 3] + + markdown_boundary = _find_markdown_break(truncated) + if markdown_boundary > _MAX_TEXT_CHARS * 0.8: + truncated = truncated[:markdown_boundary].rstrip() + + return truncated + "..." + + +def _find_markdown_break(text: str) -> int: + best = 0 + for pat in [r"\n\n", r"\n", r"\. ", r"。"]: + matches = list(re.finditer(pat, text)) + if matches: + best = matches[-1].end() + break + return best if best > 0 else len(text) diff --git a/backend/package/yuxi/channels/adapters/googlechat/media.py b/backend/package/yuxi/channels/adapters/googlechat/media.py new file mode 100644 index 00000000..ac3bbbeb --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/media.py @@ -0,0 +1,153 @@ +from __future__ import annotations + +import io +import json +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.channels.exceptions import DeliveryFailedError +from yuxi.channels.models import DeliveryResult +from yuxi.utils.logging_config import logger + +_MAX_DOWNLOAD_SIZE_BYTES = 100 * 1024 * 1024 + + +async def upload_image( + chat_service: Resource, + chat_id: str, + image_data: bytes, + filename: str = "image.png", + caption: str = "", +) -> DeliveryResult: + from googleapiclient.http import MediaIoBaseUpload + + try: + media = MediaIoBaseUpload( + io.BytesIO(image_data), + mimetype="image/png", + resumable=True, + ) + body = {"text": caption or " "} + result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute() + return DeliveryResult( + success=True, + message_id=result.get("name", ""), + ) + except Exception as e: + logger.error(f"upload_image failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def upload_file( + chat_service: Resource, + chat_id: str, + file_data: bytes, + filename: str, + mime_type: str = "application/octet-stream", + caption: str = "", +) -> DeliveryResult: + from googleapiclient.http import MediaIoBaseUpload + + try: + media = MediaIoBaseUpload( + io.BytesIO(file_data), + mimetype=mime_type, + resumable=True, + ) + body = {"text": caption or " "} + result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute() + return DeliveryResult( + success=True, + message_id=result.get("name", ""), + ) + except Exception as e: + logger.error(f"upload_file failed: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def download_media(chat_service: Resource, file_id: str) -> bytes: + try: + file_info = json.loads(file_id) + except (json.JSONDecodeError, TypeError): + raise DeliveryFailedError(f"Invalid file_id format: {file_id}") + + resource_name = file_info.get("name", file_info.get("resource_name", "")) + if not resource_name: + raise DeliveryFailedError("file_id missing attachment resource name") + + try: + attachment = chat_service.spaces().messages().attachments().get(name=resource_name).execute() + except Exception as e: + raise DeliveryFailedError(f"Failed to get attachment metadata: {e}") + + drive_data = attachment.get("driveDataRef", {}) + if drive_data and drive_data.get("driveFileId"): + try: + return await _download_drive_file(chat_service, drive_data["driveFileId"]) + except Exception as e: + raise DeliveryFailedError(f"Drive file download failed (file_id={drive_data.get('driveFileId')}): {e}") + + try: + creds = _get_credentials(chat_service) + token = await _get_access_token(creds) + + import httpx + + url = f"https://chat.googleapis.com/v1/{resource_name}?alt=media" + async with httpx.AsyncClient(timeout=httpx.Timeout(60)) as client: + resp = await client.get( + url, + headers={"Authorization": f"Bearer {token}"}, + follow_redirects=True, + ) + if resp.status_code == 200: + content_length = int(resp.headers.get("content-length", 0)) + if content_length > _MAX_DOWNLOAD_SIZE_BYTES: + raise DeliveryFailedError( + f"File too large: {content_length} bytes (max {_MAX_DOWNLOAD_SIZE_BYTES})" + ) + return resp.content + raise DeliveryFailedError(f"Media download returned HTTP {resp.status_code}") + except DeliveryFailedError: + raise + except Exception as e: + logger.warning(f"Direct media download failed, falling back: {e}") + + raise DeliveryFailedError(f"Unable to download media for attachment: {resource_name}") + + +async def _download_drive_file(chat_service: Resource, drive_file_id: str) -> bytes: + from googleapiclient.discovery import build + from googleapiclient.http import MediaIoBaseDownload + + creds = _get_credentials(chat_service) + drive_service = build("drive", "v3", credentials=creds, cache_discovery=False) + + request = drive_service.files().get_media(fileId=drive_file_id) + buf = io.BytesIO() + downloader = MediaIoBaseDownload(buf, request) + done = False + while not done: + _, done = downloader.next_chunk() + return buf.getvalue() + + +def _get_credentials(chat_service: Resource): + http = getattr(chat_service, "_http", None) + if http is None: + raise DeliveryFailedError("No HTTP transport on chat_service") + creds = getattr(http, "credentials", None) + if creds is None: + raise DeliveryFailedError("No credentials on chat_service HTTP transport") + return creds + + +async def _get_access_token(credentials) -> str: + from google.auth.transport.requests import Request as GARequest + + if credentials.valid: + return credentials.token + credentials.refresh(GARequest()) + return credentials.token diff --git a/backend/package/yuxi/channels/adapters/googlechat/mentions.py b/backend/package/yuxi/channels/adapters/googlechat/mentions.py new file mode 100644 index 00000000..6c754ac5 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/mentions.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from yuxi.channels.models import MentionsInfo + + +def parse_mentions(message: dict, bot_user: str = "") -> MentionsInfo: + annotations = message.get("annotations", []) + mentioned_ids = [ + a.get("userMention", {}).get("user", {}).get("name", "") for a in annotations if a.get("type") == "USER_MENTION" + ] + is_at_bot = False + for a in annotations: + if a.get("type") != "USER_MENTION": + continue + um = a.get("userMention", {}) + if um.get("type") == "BOT": + is_at_bot = True + break + if bot_user and um.get("user", {}).get("name", "") == bot_user: + is_at_bot = True + break + return MentionsInfo( + mentioned_user_ids=mentioned_ids, + is_bot_mentioned=is_at_bot, + ) diff --git a/backend/package/yuxi/channels/adapters/googlechat/msg_cache.py b/backend/package/yuxi/channels/adapters/googlechat/msg_cache.py new file mode 100644 index 00000000..69ab371a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/msg_cache.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import time +from collections import OrderedDict +from typing import Any + +from yuxi.utils.logging_config import logger + +_MAX_CACHE_SIZE = 128 +_MAX_AGE_S = 3600 + + +class SentMessageCache: + def __init__(self, max_size: int = _MAX_CACHE_SIZE, max_age_s: int = _MAX_AGE_S): + self._cache: OrderedDict[str, tuple[float, dict[str, Any]]] = OrderedDict() + self._max_size = max_size + self._max_age_s = max_age_s + + def put(self, msg_id: str, metadata: dict[str, Any] | None = None) -> None: + now = time.monotonic() + if len(self._cache) >= self._max_size: + self._cache.popitem(last=False) + self._cache[msg_id] = (now, metadata or {}) + + def get(self, msg_id: str) -> dict[str, Any] | None: + entry = self._cache.get(msg_id) + if entry is None: + return None + ts, meta = entry + if time.monotonic() - ts > self._max_age_s: + self._cache.pop(msg_id, None) + return None + return meta + + def remove(self, msg_id: str) -> None: + self._cache.pop(msg_id, None) + + def update(self, msg_id: str, metadata: dict[str, Any]) -> None: + entry = self._cache.get(msg_id) + if entry is None: + return + ts, existing = entry + existing.update(metadata) + self._cache[msg_id] = (ts, existing) + + def size(self) -> int: + return len(self._cache) + + def clear(self) -> None: + self._cache.clear() + + def expired_count(self) -> int: + now = time.monotonic() + expired = 0 + for _msg_id, (ts, _) in self._cache.items(): + if now - ts > self._max_age_s: + expired += 1 + return expired + + +def track_sent_message( + cache: SentMessageCache, + msg_id: str, + chat_id: str = "", + message_type: str = "text", +) -> None: + cache.put(msg_id, {"chat_id": chat_id, "message_type": message_type, "sent_at": time.monotonic()}) + logger.debug(f"Sent message cached: {msg_id}") + + +def update_sent_message_status( + cache: SentMessageCache, + msg_id: str, + status: str, + metadata: dict[str, Any] | None = None, +) -> None: + update_data = {"status": status} + if metadata: + update_data.update(metadata) + cache.update(msg_id, update_data) + logger.debug(f"Sent message status updated: {msg_id} -> {status}") diff --git a/backend/package/yuxi/channels/adapters/googlechat/normalizer.py b/backend/package/yuxi/channels/adapters/googlechat/normalizer.py new file mode 100644 index 00000000..33fb198d --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/normalizer.py @@ -0,0 +1,249 @@ +from __future__ import annotations + +import json + +from yuxi.channels.models import ( + Attachment, + ChannelIdentity, + ChannelMessage, + ChannelType, + ChatType, + EventType, + MessageType, +) + +from .mentions import parse_mentions +from .slash_commands import extract_command + + +def map_event_type(event_str: str) -> EventType: + _map = { + "MESSAGE": EventType.MESSAGE_RECEIVED, + "MESSAGE_UPDATE": EventType.MESSAGE_UPDATED, + "MESSAGE_DELETE": EventType.MESSAGE_DELETED, + "ADDED_TO_SPACE": EventType.BOT_ADDED, + "REMOVED_FROM_SPACE": EventType.BOT_REMOVED, + "CARD_CLICKED": EventType.CARD_ACTION, + } + return _map.get(event_str, EventType.MESSAGE_RECEIVED) + + +def normalize_inbound( + channel_id: str, + channel_type: ChannelType, + raw_payload: dict, + bot_user: str = "", +) -> ChannelMessage: + event = raw_payload.get("event", {}) + event_type_str = event.get("type", "") + message = event.get("message", {}) + space = event.get("space", {}) + user = event.get("user", {}) + sender = message.get("sender", {}) + + event_type = map_event_type(event_type_str) + + space_name = space.get("name", "") + space_type = space.get("spaceType", "SPACE") + chat_type = _resolve_chat_type(space_type, message, raw_payload, event_type) + + thread_key = message.get("thread", {}).get("threadKey", "") + if thread_key: + chat_type = ChatType.THREAD + chat_id = f"{space_name}/threads/{thread_key}" + else: + chat_id = space_name + + mentions = parse_mentions(message, bot_user) + + content = _extract_content(message, event_type, raw_payload) + msg_type = map_msg_type(message, event_type) + attachments = extract_attachments(message) + argument_text = _extract_argument_text(message) + context = _build_context(channel_id, event, message, space, user, sender) + route_envelope = resolve_route_envelope(chat_type, space_name, user.get("name", "")) + + command, command_args = extract_command(content) + + return ChannelMessage( + identity=ChannelIdentity( + channel_id=channel_id, + channel_type=channel_type, + channel_user_id=user.get("name", ""), + channel_chat_id=chat_id, + channel_message_id=message.get("name", ""), + ), + chat_type=chat_type, + event_type=event_type, + message_type=MessageType.COMMAND if command else msg_type, + content=content, + attachments=attachments, + mentions=mentions, + metadata={ + "space_name": space_name, + "space_type": space_type, + "thread_key": thread_key, + "sender_name": sender.get("name", ""), + "sender_type": sender.get("type", ""), + "argument_text": argument_text, + "context": context, + "route_envelope": route_envelope, + "slash_command": command, + "slash_args": command_args, + }, + ) + + +def is_bot_message(message: dict) -> bool: + sender = message.get("sender", {}) + return sender.get("type", "") == "BOT" + + +def _extract_argument_text(message: dict) -> str: + annotations = message.get("annotations", []) + for annotation in annotations: + if annotation.get("type") == "SLASH_COMMAND": + return message.get("argumentText", "") + return "" + + +def _build_context( + channel_id: str, + event: dict, + message: dict, + space: dict, + user: dict, + sender: dict, +) -> dict: + return { + "channel": channel_id, + "accountId": event.get("accountId", ""), + "messageId": message.get("name", ""), + "from": sender.get("name", ""), + "sender": sender, + "conversation": space, + "reply": message.get("thread", {}), + } + + +def _resolve_chat_type( + space_type: str, + message: dict, + raw_payload: dict, + event_type: EventType, +) -> ChatType: + if space_type == "DIRECT_MESSAGE": + return ChatType.DIRECT + if raw_payload.get("is_card_clicked"): + return ChatType.SPACE + if event_type == EventType.CARD_ACTION: + return ChatType.SPACE + return ChatType.SPACE + + +def is_forwarded_message(message: dict) -> bool: + return bool(message.get("retentionSettings")) or bool(message.get("lastUpdateTime") and message.get("createTime")) + + +def detect_message_edit(message: dict) -> bool: + create_time = message.get("createTime", "") + update_time = message.get("lastUpdateTime", "") + if create_time and update_time and create_time != update_time: + return True + return False + + +def resolve_route_envelope(chat_type: ChatType, space_name: str, user_name: str) -> dict: + if chat_type == ChatType.DIRECT: + user_id = user_name.removeprefix("users/") if user_name.startswith("users/") else user_name + return { + "peer_kind": "direct", + "peer_id": user_name, + "peer_key": f"direct:{user_id}", + "route": { + "session_key": f"direct:{user_id}", + "content": {"type": "direct", "id": user_name}, + }, + } + space_id = space_name.removeprefix("spaces/") if space_name.startswith("spaces/") else space_name + return { + "peer_kind": "group", + "peer_id": space_name, + "peer_key": f"space:{space_id}", + "route": { + "session_key": f"space:{space_id}", + "content": {"type": "space", "id": space_name}, + }, + } + + +def map_msg_type(message: dict, event_type: EventType) -> MessageType: + if event_type == EventType.CARD_ACTION: + return MessageType.CARD + + attachments = message.get("attachment", message.get("attachments", [])) + if attachments: + first = attachments[0] if attachments else {} + content_type = first.get("contentType", "") + if content_type.startswith("image/"): + return MessageType.IMAGE + if content_type.startswith("video/"): + return MessageType.VIDEO + if content_type.startswith("audio/"): + return MessageType.AUDIO + if content_type: + return MessageType.FILE + + text = message.get("text", "") + if message.get("cardsV2") or message.get("cards_v2"): + return MessageType.CARD + if text: + return MessageType.TEXT + + return MessageType.TEXT + + +def extract_attachments(message: dict) -> list[Attachment]: + attachments = message.get("attachment", message.get("attachments", [])) + if not attachments: + return [] + + result: list[Attachment] = [] + for att in attachments: + name = att.get("name", "") + content_name = att.get("contentName", "") + content_type = att.get("contentType", "") + + att_type = "file" + if content_type.startswith("image/"): + att_type = "image" + elif content_type.startswith("video/"): + att_type = "video" + elif content_type.startswith("audio/"): + att_type = "audio" + + result.append( + Attachment( + type=att_type, + filename=content_name, + mime_type=content_type, + file_id=_encode_file_id(name), + metadata={"attachment_name": name}, + ) + ) + return result + + +def _encode_file_id(attachment_name: str) -> str: + return json.dumps({"name": attachment_name}) + + +def _extract_content( + message: dict, + event_type: EventType, + raw_payload: dict, +) -> str: + if event_type == EventType.CARD_ACTION: + action = raw_payload.get("action", {}) + return json.dumps(action) + return message.get("text", "") diff --git a/backend/package/yuxi/channels/adapters/googlechat/pairing.py b/backend/package/yuxi/channels/adapters/googlechat/pairing.py new file mode 100644 index 00000000..503128e1 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/pairing.py @@ -0,0 +1,49 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.channels.models import DeliveryResult +from yuxi.utils.logging_config import logger + +_PAIRING_CHALLENGE_MESSAGE = ( + "{agent_name} Approval Required\n\n" + "To continue, send the following:\n" + "```\n/approve {user_id}\n```\n" + "This confirmation is required to pair with this bot account." +) + + +async def send_pairing_challenge( + chat_service: Resource, + space_name: str, + user_id: str, + agent_name: str = "Bot", +) -> DeliveryResult: + content = _PAIRING_CHALLENGE_MESSAGE.format( + agent_name=agent_name, + user_id=user_id, + ) + body = {"text": content} + try: + result = chat_service.spaces().messages().create(parent=space_name, body=body).execute() + msg_id = result.get("name", "") + logger.info(f"Pairing challenge sent to {space_name}: user_id={user_id}") + return DeliveryResult(success=True, message_id=msg_id) + except Exception as e: + logger.warning(f"Failed to send pairing challenge to {space_name}: {e}") + return DeliveryResult(success=False, error=str(e)) + + +async def check_pairing_approval( + chat_service: Resource, + space_name: str, + user_id: str, + message_text: str, +) -> bool: + approved = f"/approve {user_id}" in message_text.strip().lower() + if approved: + logger.info(f"Pairing approved for {space_name}: user_id={user_id}") + return approved diff --git a/backend/package/yuxi/channels/adapters/googlechat/policy.py b/backend/package/yuxi/channels/adapters/googlechat/policy.py new file mode 100644 index 00000000..46cd1c50 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/policy.py @@ -0,0 +1,185 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum +from typing import Any + +from yuxi.channels.models import ChatType, ChannelMessage +from yuxi.utils.logging_config import logger + + +class DmPolicy(StrEnum): + OPEN = "open" + PAIRING = "pairing" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +class GroupPolicy(StrEnum): + OPEN = "open" + ALLOWLIST = "allowlist" + DISABLED = "disabled" + + +DM_POLICY_VALUES = frozenset(p.value for p in DmPolicy) +GROUP_POLICY_VALUES = frozenset(p.value for p in GroupPolicy) + + +@dataclass +class PerGroupConfig: + require_mention: bool = False + enabled: bool = True + users: list[str] = field(default_factory=list) + system_prompt: str = "" + + +@dataclass +class GoogleChatPolicy: + dm_policy: DmPolicy = DmPolicy.OPEN + group_policy: GroupPolicy = GroupPolicy.OPEN + allow_from: list[str] = field(default_factory=list) + group_allow_from: list[str] = field(default_factory=list) + require_mention: bool = False + allow_bots: bool = False + bot_user: str = "" + groups: dict[str, PerGroupConfig] = field(default_factory=dict) + dangerously_allow_name_matching: bool = False + + @classmethod + def from_config(cls, config: dict[str, Any] | None) -> GoogleChatPolicy: + if not config: + return cls() + + dm_raw = str(config.get("dm_policy", config.get("dmPolicy", "open"))).strip().lower() + dm_policy = DmPolicy(dm_raw) if dm_raw in DM_POLICY_VALUES else DmPolicy.OPEN + + group_raw = str(config.get("group_policy", config.get("groupPolicy", "open"))).strip().lower() + group_policy = GroupPolicy(group_raw) if group_raw in GROUP_POLICY_VALUES else GroupPolicy.OPEN + + allow_from = _normalize_list(config.get("allowFrom", config.get("allow_from", []))) + group_allow_from = _normalize_list(config.get("groupAllowFrom", config.get("group_allow_from", []))) + require_mention = bool(config.get("require_mention", config.get("requireMention", False))) + allow_bots = bool(config.get("allow_bots", config.get("allowBots", False))) + bot_user = str(config.get("bot_user", config.get("botUser", ""))).strip() + groups = _parse_per_group_config(config.get("groups", {})) + dangerously_allow_name_matching = bool( + config.get("dangerously_allow_name_matching", config.get("dangerouslyAllowNameMatching", False)) + ) + + return cls( + dm_policy=dm_policy, + group_policy=group_policy, + allow_from=allow_from, + group_allow_from=group_allow_from, + require_mention=require_mention, + allow_bots=allow_bots, + bot_user=bot_user, + groups=groups, + dangerously_allow_name_matching=dangerously_allow_name_matching, + ) + + def check_dm_access(self, user_id: str) -> bool: + if self.dm_policy == DmPolicy.DISABLED: + return False + if self.dm_policy == DmPolicy.OPEN: + return True + if self.dm_policy == DmPolicy.PAIRING: + return True + if self.dm_policy == DmPolicy.ALLOWLIST: + return self._match_allowlist(user_id, self.allow_from) + return False + + def check_group_access(self, space_id: str) -> bool: + per_group = self._resolve_per_group(space_id) + if per_group is not None and not per_group.enabled: + return False + if self.group_policy == GroupPolicy.DISABLED: + return False + if self.group_policy == GroupPolicy.OPEN: + return True + if self.group_policy == GroupPolicy.ALLOWLIST: + return self._match_allowlist(space_id, self.group_allow_from) + return False + + def check_inbound(self, channel_msg: ChannelMessage) -> bool: + chat_type = channel_msg.chat_type + + if chat_type == ChatType.DIRECT: + user_id = channel_msg.identity.channel_user_id + return self.check_dm_access(user_id) + + space_id = channel_msg.metadata.get("space_name", "") + if not self.check_group_access(space_id): + return False + + per_group = self._resolve_per_group(space_id) + + if per_group is not None and per_group.users: + user_id = channel_msg.identity.channel_user_id + if user_id not in per_group.users: + logger.debug(f"Group message filtered: user {user_id} not in per-group users whitelist") + return False + + require_mention = self.require_mention + if per_group is not None: + require_mention = per_group.require_mention + + if require_mention and chat_type != ChatType.DIRECT: + mentions = channel_msg.mentions + if mentions and not mentions.is_bot_mentioned: + logger.debug("Group message filtered: bot not mentioned") + return False + + return True + + def get_per_group_system_prompt(self, space_id: str) -> str: + per_group = self._resolve_per_group(space_id) + if per_group and per_group.system_prompt: + return per_group.system_prompt + return "" + + def _resolve_per_group(self, space_id: str) -> PerGroupConfig | None: + if not space_id or not self.groups: + return None + return self.groups.get(space_id) + + def _match_allowlist(self, target: str, allowlist: list[str]) -> bool: + if not allowlist: + return False + if "*" in allowlist: + return True + if target in allowlist: + return True + if self.dangerously_allow_name_matching and "@" in target: + email_local = target.split("@")[0].lower() + for entry in allowlist: + if "@" in entry and entry.split("@")[0].lower() == email_local: + logger.warning( + f"Dangerously matched target '{target}' to allowlist entry '{entry}' via email local-part" + ) + return True + return False + + +def _normalize_list(raw: Any) -> list[str]: + if isinstance(raw, str): + return [x.strip() for x in raw.split(",") if x.strip()] + if isinstance(raw, (list, tuple)): + return [str(x).strip() for x in raw if x] + return [] + + +def _parse_per_group_config(groups_raw: Any) -> dict[str, PerGroupConfig]: + if not isinstance(groups_raw, dict): + return {} + result: dict[str, PerGroupConfig] = {} + for space_name, cfg in groups_raw.items(): + if not isinstance(cfg, dict): + continue + result[space_name] = PerGroupConfig( + require_mention=bool(cfg.get("require_mention", cfg.get("requireMention", False))), + enabled=bool(cfg.get("enabled", True)), + users=_normalize_list(cfg.get("users", [])), + system_prompt=str(cfg.get("system_prompt", cfg.get("systemPrompt", ""))), + ) + return result diff --git a/backend/package/yuxi/channels/adapters/googlechat/proxy.py b/backend/package/yuxi/channels/adapters/googlechat/proxy.py new file mode 100644 index 00000000..ff5e470a --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/proxy.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import os +from typing import Any + +from yuxi.utils.logging_config import logger + + +def resolve_proxy_config(config: dict[str, Any] | None = None) -> dict[str, str | None]: + proxy_config: dict[str, str | None] = { + "http": None, + "https": None, + "no_proxy": None, + } + + if config: + proxy_config["http"] = config.get("httpProxy", config.get("HTTP_PROXY")) + proxy_config["https"] = config.get("httpsProxy", config.get("HTTPS_PROXY")) + proxy_config["no_proxy"] = config.get("noProxy", config.get("NO_PROXY")) + + if not proxy_config["http"]: + proxy_config["http"] = os.getenv("HTTP_PROXY") + if not proxy_config["https"]: + proxy_config["https"] = os.getenv("HTTPS_PROXY") + if not proxy_config["no_proxy"]: + proxy_config["no_proxy"] = os.getenv("NO_PROXY") + + if proxy_config["http"] and not proxy_config["https"]: + proxy_config["https"] = proxy_config["http"] + + return proxy_config + + +def build_proxies_dict(proxy_config: dict[str, str | None]) -> dict[str, str] | None: + proxies: dict[str, str] = {} + if proxy_config.get("http"): + proxies["http"] = proxy_config["http"] + if proxy_config.get("https"): + proxies["https"] = proxy_config["https"] + if proxy_config.get("no_proxy"): + proxies["no_proxy"] = proxy_config["no_proxy"] + + if proxies: + logger.debug(f"Proxy configured: https={proxies.get('https')}") + return proxies + return None + + +def build_google_auth_request(proxy_config: dict[str, str | None] | None = None) -> Any: + try: + from google.auth.transport.requests import Request as GARequest + except ImportError: + return None + + proxies = build_proxies_dict(proxy_config or {}) + if not proxies: + return GARequest() + + try: + result = GARequest() + result.session.proxies.update(proxies) + return result + except Exception: + logger.debug("Failed to apply proxy to Google Auth request, using default") + return GARequest() + + +def resolve_tls_config(config: dict[str, Any] | None = None) -> dict[str, str | None]: + tls_config: dict[str, str | None] = { + "cert": None, + "key": None, + } + + if config: + tls_config["cert"] = config.get("cert", config.get("tlsCert")) + tls_config["key"] = config.get("key", config.get("tlsKey")) + + if not tls_config["cert"]: + tls_config["cert"] = os.getenv("GOOGLE_CHAT_TLS_CERT") + if not tls_config["key"]: + tls_config["key"] = os.getenv("GOOGLE_CHAT_TLS_KEY") + + return tls_config + + +def has_tls_config(tls_config: dict[str, str | None]) -> bool: + return bool(tls_config.get("cert")) diff --git a/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py b/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py new file mode 100644 index 00000000..a34f6263 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/pubsub_.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +import json +import base64 + +from yuxi.utils.logging_config import logger + + +def decode_pubsub_push(body: dict) -> dict | None: + message = body.get("message", {}) + data_b64 = message.get("data", "") + + if not data_b64: + return None + + try: + decoded = base64.b64decode(data_b64).decode("utf-8") + return json.loads(decoded) + except Exception as e: + logger.warning(f"Failed to decode Pub/Sub push data: {e}") + return None diff --git a/backend/package/yuxi/channels/adapters/googlechat/secret_contract.py b/backend/package/yuxi/channels/adapters/googlechat/secret_contract.py new file mode 100644 index 00000000..3b89aada --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/secret_contract.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + +_SECRET_CONTRACTS: dict[str, dict[str, Any]] = { + "serviceAccount": { + "name": "serviceAccount", + "description": "Google Chat service account JSON credentials (inline)", + "required": True, + "sensitive": True, + "validate": "_validate_service_account_json", + }, + "serviceAccountFile": { + "name": "serviceAccountFile", + "description": "Path to Google Chat service account JSON key file", + "required": False, + "sensitive": True, + }, + "serviceAccountRef": { + "name": "serviceAccountRef", + "description": "Secret reference to Google Chat service account credentials", + "required": False, + "sensitive": True, + }, + "webhookSecret": { + "name": "webhookSecret", + "description": "Webhook secret for Pub/Sub push endpoint (if applicable)", + "required": False, + "sensitive": True, + }, +} + + +def get_secret_contracts() -> dict[str, dict[str, Any]]: + return dict(_SECRET_CONTRACTS) + + +def register_secret_contract(name: str, contract: dict[str, Any]) -> None: + _SECRET_CONTRACTS[name] = contract + logger.debug(f"Registered secret contract: {name}") + + +def get_required_secrets() -> list[str]: + return [name for name, contract in _SECRET_CONTRACTS.items() if contract.get("required")] + + +def get_sensitive_fields() -> set[str]: + return {name for name, contract in _SECRET_CONTRACTS.items() if contract.get("sensitive")} + + +def is_sensitive_field(field_name: str) -> bool: + return field_name in get_sensitive_fields() or field_name.endswith("_key") or field_name.endswith("_secret") diff --git a/backend/package/yuxi/channels/adapters/googlechat/send.py b/backend/package/yuxi/channels/adapters/googlechat/send.py new file mode 100644 index 00000000..babf0b5c --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/send.py @@ -0,0 +1,314 @@ +from __future__ import annotations + +import asyncio +import io +import random +from typing import TYPE_CHECKING, Any + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.channels.exceptions import ( + ChannelAuthenticationError, + ChannelRateLimitError, + DeliveryFailedError, +) +from yuxi.channels.models import DeliveryResult +from yuxi.utils.logging_config import logger + +_MAX_RETRIES = 5 +_MAX_BACKOFF_S = 64 +_AUTH_401_MAX_BACKOFF_S = 300 +_AUTH_401_COOLDOWN: dict[str, float] = {} + + +def _check_401_cooldown(account_id: str = "default") -> bool: + now = __import__("time").monotonic() + cooldown_until = _AUTH_401_COOLDOWN.get(account_id, 0) + if now < cooldown_until: + remaining = cooldown_until - now + logger.warning(f"Auth 401 cooldown active for {remaining:.0f}s") + return False + return True + + +def _apply_401_cooldown(account_id: str = "default", backoff_s: float = 60.0) -> None: + now = __import__("time").monotonic() + capped = min(backoff_s, _AUTH_401_MAX_BACKOFF_S) + _AUTH_401_COOLDOWN[account_id] = now + capped + logger.info(f"Auth 401 backoff applied: {capped:.0f}s cooldown") + + +def _extend_401_cooldown(account_id: str = "default") -> None: + now = __import__("time").monotonic() + existing = _AUTH_401_COOLDOWN.get(account_id, 0) + remaining = max(existing - now, 0) + extension = min(remaining * 2 + 30, _AUTH_401_MAX_BACKOFF_S) + _AUTH_401_COOLDOWN[account_id] = now + extension + logger.warning(f"Auth 401 cooldown extended to {extension:.0f}s") + + +def _classify_error(e: Exception, account_id: str = "default") -> None: + try: + from googleapiclient.errors import HttpError + + if isinstance(e, HttpError): + status = e.resp.status + if status == 429: + raise ChannelRateLimitError() + if status in (401, 403): + reason = _extract_rate_limit_reason(e) + if reason: + raise ChannelRateLimitError() + if status == 401: + _extend_401_cooldown(account_id) + raise ChannelAuthenticationError(str(e)) + raise DeliveryFailedError(str(e)) from e + except ImportError: + raise DeliveryFailedError(str(e)) from None + raise DeliveryFailedError(str(e)) from e + + +def _extract_rate_limit_reason(e: Exception) -> bool: + try: + content = e.content if hasattr(e, "content") else b"" + if isinstance(content, bytes): + content = content.decode("utf-8", errors="replace") + if isinstance(content, str) and ("rateLimitExceeded" in content or "userRateLimitExceeded" in content): + return True + except Exception: + pass + return False + + +def _backoff_with_jitter(retry_count: int) -> float: + delay = min((2**retry_count) + random.uniform(0, 1), _MAX_BACKOFF_S) + return delay + + +async def _retry_with_backoff(func, *args, account_id: str = "default", **kwargs) -> DeliveryResult: + last_error = None + for attempt in range(_MAX_RETRIES): + try: + return await func(*args, **kwargs) + except ChannelRateLimitError as e: + last_error = e + delay = _backoff_with_jitter(attempt) + logger.warning(f"Rate limited (attempt {attempt + 1}/{_MAX_RETRIES}), backing off {delay:.1f}s") + await asyncio.sleep(delay) + except (ChannelAuthenticationError, DeliveryFailedError): + raise + except Exception as e: + last_error = e + if attempt < _MAX_RETRIES - 1: + delay = _backoff_with_jitter(attempt) + logger.warning(f"Request failed (attempt {attempt + 1}/{_MAX_RETRIES}): {e}, retrying in {delay:.1f}s") + await asyncio.sleep(delay) + else: + break + + error_msg = str(last_error) if last_error else "Max retries exceeded" + logger.error(f"send failed after {_MAX_RETRIES} retries: {error_msg}") + return DeliveryResult(success=False, error=error_msg) + + +async def send_message( + chat_service: Resource, + chat_id: str, + body: dict[str, Any], + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + result = chat_service.spaces().messages().create(parent=chat_id, body=body).execute() + return DeliveryResult(success=True, message_id=result.get("name", "")) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def update_message( + chat_service: Resource, + message_id: str, + body: dict[str, Any], + update_mask: str = "text", + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + result = chat_service.spaces().messages().update(name=message_id, updateMask=update_mask, body=body).execute() + return DeliveryResult(success=True, message_id=result.get("name", message_id)) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def delete_message( + chat_service: Resource, + message_id: str, + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + chat_service.spaces().messages().delete(name=message_id).execute() + return DeliveryResult(success=True) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def send_media( + chat_service: Resource, + chat_id: str, + media_url: str, + caption: str = "", + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + body: dict[str, Any] = {"text": caption or " "} + body["cards_v2"] = [{"card": {"sections": [{"widgets": [{"image": {"imageUrl": media_url}}]}]}}] + result = chat_service.spaces().messages().create(parent=chat_id, body=body).execute() + return DeliveryResult(success=True, message_id=result.get("name", "")) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def send_reaction( + chat_service: Resource, + message_id: str, + emoji: str, + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + body = {"emoji": {"unicode": emoji}} + (chat_service.spaces().messages().reactions().create(parent=message_id, body=body).execute()) + return DeliveryResult(success=True) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def list_reactions( + chat_service: Resource, + message_id: str, +) -> list[dict[str, Any]]: + try: + result = chat_service.spaces().messages().reactions().list(parent=message_id).execute() + return result.get("reactions", []) + except Exception as e: + logger.warning(f"list_reactions failed for {message_id}: {e}") + return [] + + +async def delete_reaction( + chat_service: Resource, + reaction_id: str, + account_id: str = "default", +) -> DeliveryResult: + async def _do_send(): + chat_service.spaces().messages().reactions().delete(name=reaction_id).execute() + return DeliveryResult(success=True) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def upload_file_message( + chat_service: Resource, + chat_id: str, + file_data: bytes, + filename: str, + mime_type: str = "application/octet-stream", + caption: str = "", + account_id: str = "default", +) -> DeliveryResult: + from googleapiclient.http import MediaIoBaseUpload + + async def _do_send(): + media = MediaIoBaseUpload( + io.BytesIO(file_data), + mimetype=mime_type, + resumable=True, + ) + body: dict[str, Any] = {"text": caption or " "} + result = chat_service.spaces().messages().create(parent=chat_id, body=body, media_body=media).execute() + return DeliveryResult(success=True, message_id=result.get("name", "")) + + try: + return await _retry_with_backoff(_do_send, account_id=account_id) + except (ChannelAuthenticationError, DeliveryFailedError, ChannelRateLimitError) as e: + return DeliveryResult(success=False, error=str(e)) + except Exception as e: + _classify_error(e, account_id) + return DeliveryResult(success=False, error=str(e)) + + +async def upload_image_message( + chat_service: Resource, + chat_id: str, + image_data: bytes, + filename: str = "image.png", + caption: str = "", + account_id: str = "default", +) -> DeliveryResult: + return await upload_file_message( + chat_service, chat_id, image_data, filename, "image/png", caption, account_id=account_id + ) + + +async def find_direct_message_space( + chat_service: Resource, + user_name: str, +) -> str | None: + try: + result = chat_service.spaces().findDirectMessage(query=user_name).execute() + return result.get("name", "") + except Exception as e: + logger.warning(f"findDirectMessage failed for user {user_name}: {e}") + return None + + +async def resolve_outbound_space( + chat_service: Resource, + target: str, +) -> str | None: + if target.startswith("spaces/"): + try: + result = chat_service.spaces().get(name=target).execute() + return result.get("name", "") + except Exception: + return target + + dm_result = await find_direct_message_space(chat_service, target) + if dm_result: + return dm_result + + return None diff --git a/backend/package/yuxi/channels/adapters/googlechat/session.py b/backend/package/yuxi/channels/adapters/googlechat/session.py new file mode 100644 index 00000000..9e3fbe2d --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/session.py @@ -0,0 +1,9 @@ +from __future__ import annotations + + +def build_thread_id(agent_id: str, space_name: str, user_email: str = "") -> str: + if space_name.startswith("spaces/"): + space_id = space_name.removeprefix("spaces/") + return f"agent:{agent_id}:googlechat:space:{space_id}" + + return f"agent:main:googlechat:dm:{user_email}" diff --git a/backend/package/yuxi/channels/adapters/googlechat/setup.py b/backend/package/yuxi/channels/adapters/googlechat/setup.py new file mode 100644 index 00000000..9c8557a0 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/setup.py @@ -0,0 +1,339 @@ +from __future__ import annotations + +from typing import Any + + +_SETUP_GUIDE = """ +Google Chat Setup Guide +======================= + +1. Go to Google Cloud Console → APIs & Services → Enable APIs + Enable the "Google Chat API" + +2. Go to IAM & Admin → Service Accounts + Create a new service account or use an existing one + Generate a JSON key and download it + +3. Go to Google Chat API → Configuration + Set up the bot: + - App name: Your bot name + - Avatar URL: (optional) + - Description: Your bot description + - Functionality: Enable "Receive messages" and "Join spaces and group conversations" + - Connection settings: + * App URL: https://your-domain.com/api/webhook/googlechat + * (Or use Pub/Sub for production) + +4. Configure the adapter with your service account credentials + using one of the supported methods (file, inline, or environment variable). + +5. Optional: Set audienceType to "project-number" if using project-number auth, + and configure the audience accordingly. + +6. Configure security policies: + - dmPolicy: "open" | "pairing" | "allowlist" | "disabled" + - groupPolicy: "open" | "allowlist" | "disabled" + - allowFrom: list of users/spaces allowed to interact + - requireMention: require @mention in group messages +""" + + +class SetupWizard: + def __init__(self, adapter=None): + self._adapter = adapter + self._steps: list[dict[str, Any]] = [] + self._current_step = 0 + self._results: dict[str, Any] = {} + + @property + def current_step(self) -> dict[str, Any] | None: + if 0 <= self._current_step < len(self._steps): + return self._steps[self._current_step] + return None + + def start(self) -> dict[str, Any]: + self._results = {} + self._current_step = 0 + self._steps = [ + self._step_credential_mode(), + self._step_credential_input(), + self._step_audience_config(), + self._step_policy_config(), + self._step_review(), + ] + return {"step": self._current_step, "steps": len(self._steps), "data": self.current_step} + + def next(self, answer: Any = None) -> dict[str, Any]: + if self._current_step > 0 and answer is not None: + self._results[self._steps[self._current_step - 1]["id"]] = answer + + self._current_step += 1 + if self._current_step >= len(self._steps): + return {"step": self._current_step, "done": True, "config": self._build_config()} + + return {"step": self._current_step, "steps": len(self._steps), "data": self.current_step} + + def _step_credential_mode(self) -> dict[str, Any]: + return { + "id": "credential_mode", + "question": "How would you like to provide service account credentials?", + "type": "select", + "options": [ + { + "id": "env_file", + "label": "Environment (GOOGLE_SERVICE_ACCOUNT_FILE)", + "description": ( + "Set the path to a service account JSON key file " + "via the GOOGLE_SERVICE_ACCOUNT_FILE environment variable" + ), + }, + { + "id": "env_json", + "label": "Environment (GOOGLE_CHAT_SERVICE_ACCOUNT)", + "description": ( + "Paste the full service account JSON content " + "into the GOOGLE_CHAT_SERVICE_ACCOUNT environment variable" + ), + }, + { + "id": "file", + "label": "File (serviceAccountFile)", + "description": "Specify the path to a service account JSON key file on disk in the config", + }, + { + "id": "inline", + "label": "Inline (serviceAccount)", + "description": "Paste the full service account JSON content directly into the config", + }, + { + "id": "secret_ref", + "label": "Secret Reference (serviceAccountRef)", + "description": "Reference a secret via the secret management system", + }, + ], + } + + def _step_credential_input(self) -> dict[str, Any]: + mode = self._results.get("credential_mode", "") + if mode == "file": + return { + "id": "credential_value", + "question": "Please provide the path to your Google Cloud service account JSON key file:", + "type": "text", + "required": True, + "hint": "e.g., /etc/secrets/google-service-account.json or ~/.config/gcp-sa-key.json", + } + if mode == "inline": + return { + "id": "credential_value", + "question": "Please paste the full content of your Google Cloud service account JSON key:", + "type": "text", + "required": True, + "hint": ( + "This file can be downloaded from Google Cloud Console " + "→ IAM & Admin → Service Accounts → Create Key → JSON" + ), + } + if mode == "secret_ref": + return { + "id": "credential_value", + "question": "Please provide the secret reference identifier:", + "type": "text", + "required": True, + "hint": "e.g., myapp/gcp/googlechat-sa-key", + } + return { + "id": "credential_value", + "question": ( + "Credential will be read from environment variables. " + "Ensure GOOGLE_CHAT_SERVICE_ACCOUNT or " + "GOOGLE_SERVICE_ACCOUNT_FILE is set." + ), + "type": "info", + "required": False, + } + + def _step_audience_config(self) -> dict[str, Any]: + return { + "id": "audience_config", + "question": "Configure audience validation for webhook JWT:", + "type": "form", + "fields": [ + { + "id": "audience_type", + "question": "Audience type:", + "type": "select", + "options": [ + {"id": "app-url", "label": "App URL (recommended)", "description": "Match the app's HTTPS URL"}, + { + "id": "project-number", + "label": "Project Number", + "description": "Match the GCP project number", + }, + ], + "default": "app-url", + }, + { + "id": "audience", + "question": "Audience value:", + "type": "text", + "hint": "For app-url: your HTTPS app URL. For project-number: your GCP project number.", + }, + ], + } + + def _step_policy_config(self) -> dict[str, Any]: + return { + "id": "policy_config", + "question": "Configure access policies:", + "type": "form", + "fields": [ + { + "id": "dm_policy", + "question": "DM Policy:", + "type": "select", + "options": [ + {"id": "open", "label": "Open", "description": "Any user can DM the bot"}, + {"id": "pairing", "label": "Pairing", "description": "Users must complete a pairing challenge"}, + {"id": "allowlist", "label": "Allowlist", "description": "Only users in allowFrom can DM"}, + {"id": "disabled", "label": "Disabled", "description": "No DM access"}, + ], + "default": "open", + }, + { + "id": "group_policy", + "question": "Group Policy:", + "type": "select", + "options": [ + {"id": "open", "label": "Open", "description": "Bot responds in all groups"}, + { + "id": "allowlist", + "label": "Allowlist", + "description": "Only spaces in groupAllowFrom are active", + }, + {"id": "disabled", "label": "Disabled", "description": "No group access"}, + ], + "default": "open", + }, + { + "id": "require_mention", + "question": "Require @mention in groups?", + "type": "boolean", + "default": False, + }, + ], + } + + def _step_review(self) -> dict[str, Any]: + return { + "id": "review", + "question": "Review configuration before saving:", + "type": "review", + "summary": self._build_config(), + } + + def _build_config(self) -> dict[str, Any]: + config: dict[str, Any] = {} + mode = self._results.get("credential_mode", "") + credential_value = self._results.get("credential_value", "") + + if mode == "inline" and credential_value: + config["serviceAccount"] = credential_value + elif mode == "file" and credential_value: + config["serviceAccountFile"] = credential_value + elif mode == "secret_ref" and credential_value: + config["serviceAccountRef"] = credential_value + + audience = self._results.get("audience_config", {}) + if isinstance(audience, dict): + if audience.get("audience_type"): + config["audienceType"] = audience["audience_type"] + if audience.get("audience"): + config["audience"] = audience["audience"] + + policy = self._results.get("policy_config", {}) + if isinstance(policy, dict): + if policy.get("dm_policy"): + config["dmPolicy"] = policy["dm_policy"] + if policy.get("group_policy"): + config["groupPolicy"] = policy["group_policy"] + if policy.get("require_mention") is not None: + config["requireMention"] = policy["require_mention"] + + config["enabled"] = True + return config + + +def wizard_prompt_credential_mode() -> dict[str, Any]: + return { + "step": "select_credential_mode", + "question": "How would you like to provide service account credentials?", + "options": [ + { + "id": "env_file", + "label": "Environment (GOOGLE_SERVICE_ACCOUNT_FILE)", + "description": ( + "Set the path to a service account JSON key file " + "via the GOOGLE_SERVICE_ACCOUNT_FILE environment variable" + ), + }, + { + "id": "env_json", + "label": "Environment (GOOGLE_CHAT_SERVICE_ACCOUNT)", + "description": ( + "Paste the full service account JSON content " + "into the GOOGLE_CHAT_SERVICE_ACCOUNT environment variable" + ), + }, + { + "id": "file", + "label": "File (serviceAccountFile)", + "description": "Specify the path to a service account JSON key file on disk", + }, + { + "id": "inline", + "label": "Inline (serviceAccount)", + "description": "Paste the full service account JSON content directly into the config", + }, + ], + } + + +def wizard_prompt_file_path() -> dict[str, Any]: + return { + "step": "provide_credential_path", + "question": "Please provide the path to your Google Cloud service account JSON key file:", + "required": True, + } + + +def wizard_prompt_service_account_json() -> dict[str, Any]: + return { + "step": "provide_credential_json", + "question": ( + "Please paste the full content of your Google Cloud service account " + "JSON key (or set the environment variable directly):" + ), + "required": True, + "hint": ( + "This file can be downloaded from Google Cloud Console " + "→ IAM & Admin → Service Accounts → Create Key → JSON" + ), + } + + +def wizard_validate_service_account(cred_json: dict) -> dict[str, Any]: + errors: list[str] = [] + if cred_json.get("type") != "service_account": + errors.append("Invalid credential type: expected 'service_account'") + if not cred_json.get("private_key"): + errors.append("Missing 'private_key' field") + if not cred_json.get("client_email"): + errors.append("Missing 'client_email' field") + if not cred_json.get("token_uri"): + errors.append("Missing 'token_uri' field") + return {"valid": len(errors) == 0, "errors": errors} + + +def get_setup_guide() -> str: + return _SETUP_GUIDE.strip() diff --git a/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py b/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py new file mode 100644 index 00000000..4ae435c3 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/slash_commands.py @@ -0,0 +1,33 @@ +from __future__ import annotations + + +SLASH_COMMAND_MAP: dict[str, str] = { + "/reset": "清除当前会话上下文，重新开始对话", + "/history": "查看当前会话的对话历史摘要", + "/context": "查看当前会话的上下文信息", + "/summary": "生成当前会话的总结", + "/help": "显示可用命令列表", + "/status": "查看 Bot 状态", +} + + +def extract_command(content: str) -> tuple[str | None, str]: + stripped = content.strip() + if not stripped.startswith("/"): + return None, content + + parts = stripped.split(maxsplit=1) + command = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + if command in SLASH_COMMAND_MAP: + return command, args + + return None, content + + +def get_command_help() -> str: + lines = ["可用命令："] + for cmd, desc in SLASH_COMMAND_MAP.items(): + lines.append(f" {cmd} - {desc}") + return "\n".join(lines) diff --git a/backend/package/yuxi/channels/adapters/googlechat/ssrf.py b/backend/package/yuxi/channels/adapters/googlechat/ssrf.py new file mode 100644 index 00000000..aab2d08e --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/ssrf.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +from typing import Any + +from yuxi.utils.logging_config import logger + +_UNSAFE_TRANSPORT_FIELDS = frozenset( + { + "agent", + "cert", + "cert_file", + "key", + "key_file", + "dispatcher", + "proxy", + "session", + } +) + +_AUTH_RESPONSE_MAX_BYTES = 1 * 1024 * 1024 + + +def sanitize_google_auth_init(kwargs: dict[str, Any]) -> dict[str, Any]: + cleaned: dict[str, Any] = {} + removed: list[str] = [] + for key, value in kwargs.items(): + if key in _UNSAFE_TRANSPORT_FIELDS: + removed.append(key) + continue + cleaned[key] = value + if removed: + logger.warning(f"SSRF guard: removed unsafe transport fields: {removed}") + return cleaned + + +def sanitize_request_kwargs(kwargs: dict[str, Any]) -> dict[str, Any]: + cleaned: dict[str, Any] = {} + removed: list[str] = [] + for key, value in kwargs.items(): + if key in _UNSAFE_TRANSPORT_FIELDS: + removed.append(key) + continue + cleaned[key] = value + if removed: + logger.warning(f"SSRF guard: removed unsafe request fields: {removed}") + return cleaned + + +def apply_ssrf_guard() -> dict[str, Any]: + return { + "agent": None, + "cert": None, + "cert_file": None, + "key": None, + "key_file": None, + } + + +def check_auth_response_size(data: bytes) -> bytes: + if len(data) > _AUTH_RESPONSE_MAX_BYTES: + logger.warning( + f"Google Auth response exceeds size limit: {len(data)} bytes > {_AUTH_RESPONSE_MAX_BYTES} bytes, truncating" + ) + return data[:_AUTH_RESPONSE_MAX_BYTES] + return data + + +def check_auth_response_text(text: str) -> str: + data = text.encode("utf-8", errors="replace") + truncated = check_auth_response_size(data) + return truncated.decode("utf-8", errors="replace") diff --git a/backend/package/yuxi/channels/adapters/googlechat/streaming.py b/backend/package/yuxi/channels/adapters/googlechat/streaming.py new file mode 100644 index 00000000..ff2f114f --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/streaming.py @@ -0,0 +1,190 @@ +from __future__ import annotations + +import time +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from googleapiclient.discovery import Resource + +from yuxi.channels.models import DeliveryResult +from yuxi.utils.logging_config import logger + +from .send import update_message, send_message, delete_message + +DEFAULT_UPDATE_INTERVAL_MS = 800 +DEFAULT_COALESCE_MIN_CHARS = 1500 +DEFAULT_COALESCE_IDLE_MS = 1000 + + +class StreamManager: + def __init__( + self, + update_interval_ms: int = DEFAULT_UPDATE_INTERVAL_MS, + coalesce_min_chars: int = DEFAULT_COALESCE_MIN_CHARS, + coalesce_idle_ms: int = DEFAULT_COALESCE_IDLE_MS, + ): + self._messages: dict[str, str] = {} + self._texts: dict[str, str] = {} + self._last_update: dict[str, float] = {} + self._update_interval_ms = update_interval_ms + self._coalesce_min_chars = coalesce_min_chars + self._coalesce_idle_ms = coalesce_idle_ms + + self._typing_messages: dict[str, str] = {} + self._typing_replaced: set[str] = set() + + @property + def update_interval_ms(self) -> int: + return self._update_interval_ms + + def has_pending(self, chat_id: str) -> bool: + return chat_id in self._messages + + def get_message_name(self, chat_id: str) -> str | None: + return self._messages.get(chat_id) + + def get_accumulated_text(self, chat_id: str) -> str: + return self._texts.get(chat_id, "") + + def register_message(self, chat_id: str, message_name: str, text: str) -> None: + self._messages[chat_id] = message_name + self._texts[chat_id] = text + self._last_update[chat_id] = time.monotonic() + + def register_first_chunk(self, chat_id: str, message_name: str, text: str) -> None: + self._messages[chat_id] = message_name + self._texts[chat_id] = text + self._last_update[chat_id] = time.monotonic() + + typing_id = self._typing_messages.pop(chat_id, None) + if typing_id: + self._typing_replaced.add(chat_id) + logger.debug(f"Stream: replaced typing message {typing_id} with first chunk {message_name}") + + def append_text(self, chat_id: str, chunk: str) -> str: + current = self._texts.get(chat_id, "") + current += chunk + self._texts[chat_id] = current + return current + + def should_update(self, chat_id: str) -> bool: + last = self._last_update.get(chat_id, 0) + elapsed_ms = (time.monotonic() - last) * 1000 + + if elapsed_ms >= self._update_interval_ms: + return True + + if self._coalesce_idle_ms > 0: + idle_threshold = max(self._coalesce_idle_ms, self._update_interval_ms) + if elapsed_ms >= idle_threshold: + new_chars_since_last = 0 + return new_chars_since_last >= self._coalesce_min_chars + + return False + + def mark_update(self, chat_id: str) -> None: + self._last_update[chat_id] = time.monotonic() + + async def create_typing_message( + self, + chat_service: Resource, + chat_id: str, + bot_name: str = "Bot", + ) -> str | None: + if chat_id in self._typing_messages: + return self._typing_messages[chat_id] + + try: + body = {"text": f"*{bot_name} is typing...*"} + result = await send_message(chat_service, chat_id, body) + if result.success and result.message_id: + self._typing_messages[chat_id] = result.message_id + return result.message_id + except Exception as e: + logger.warning(f"Failed to create typing message in {chat_id}: {e}") + + return None + + async def clear_typing_message( + self, + chat_service: Resource, + chat_id: str, + ) -> None: + msg_id = self._typing_messages.pop(chat_id, None) + if msg_id and chat_id not in self._typing_replaced: + try: + await delete_message(chat_service, msg_id) + except Exception as e: + logger.warning(f"Failed to delete typing message {msg_id}: {e}") + self._typing_replaced.discard(chat_id) + + async def handle_error( + self, + chat_service: Resource, + chat_id: str, + ) -> None: + await self.clear_typing_message(chat_service, chat_id) + self._messages.pop(chat_id, None) + self._texts.pop(chat_id, None) + self._last_update.pop(chat_id, None) + + async def finalize( + self, + chat_service: Resource, + chat_id: str, + ) -> DeliveryResult: + await self.clear_typing_message(chat_service, chat_id) + + message_name = self._messages.pop(chat_id, None) + text = self._texts.pop(chat_id, None) + self._last_update.pop(chat_id, None) + + if not message_name or text is None: + return DeliveryResult(success=False, error="No pending stream message") + + return await update_message(chat_service, message_name, {"text": text}) + + async def send_update( + self, + chat_service: Resource, + chat_id: str, + finished: bool = False, + ) -> DeliveryResult | None: + message_name = self._messages.get(chat_id) + text = self._texts.get(chat_id) + if not message_name or text is None: + return None + + if chat_id in self._typing_messages and text: + typing_id = self._typing_messages.pop(chat_id, None) + if typing_id: + try: + await delete_message(chat_service, typing_id) + except Exception: + pass + self._typing_replaced.add(chat_id) + + body = {"text": text if finished else text + " ⏳"} + result = await update_message(chat_service, message_name, body) + + if result.success: + if finished: + self._messages.pop(chat_id, None) + self._texts.pop(chat_id, None) + self._last_update.pop(chat_id, None) + await self.clear_typing_message(chat_service, chat_id) + else: + self._last_update[chat_id] = time.monotonic() + + return result + + def clear(self) -> None: + self._messages.clear() + self._texts.clear() + self._last_update.clear() + self._typing_messages.clear() + self._typing_replaced.clear() + + @property + def pending_count(self) -> int: + return len(self._messages) diff --git a/backend/package/yuxi/channels/adapters/googlechat/target.py b/backend/package/yuxi/channels/adapters/googlechat/target.py new file mode 100644 index 00000000..dd8e6f08 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/target.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +import re +from typing import Any + + +_CHANNEL_PREFIXES = ("googlechat:", "gchat:", "google-chat:") + + +def normalize_googlechat_target(target: str) -> dict[str, Any]: + if not target: + return {"raw": target, "type": "unknown"} + + target = _strip_channel_prefix(target) + + if "@" in target and not target.startswith(("spaces/", "users/")): + target = _convert_email_to_user(target) + + if target.startswith("spaces/"): + parts = target.removeprefix("spaces/").split("/") + space_id = parts[0] + thread_key = parts[2] if len(parts) > 2 and parts[1] == "threads" else None + return { + "raw": target, + "type": "thread" if thread_key else "space", + "space_id": space_id, + "space_name": f"spaces/{space_id}", + "thread_key": thread_key, + "full_target": target, + } + + if target.startswith("users/"): + user_id = target.removeprefix("users/") + cleaned = user_id.removeprefix("user:") + return { + "raw": target, + "type": "user", + "user_id": cleaned, + "user_name": f"users/{cleaned}", + "full_target": target, + } + + if target.startswith("user:"): + user_id = target.removeprefix("user:") + return { + "raw": target, + "type": "user", + "user_id": user_id, + "user_name": f"users/{user_id}", + "full_target": f"users/{user_id}", + } + + return {"raw": target, "type": "unknown"} + + +def is_googlechat_user_target(target: str) -> bool: + target = _strip_channel_prefix(target) + return target.startswith("users/") or target.startswith("user:") + + +def is_googlechat_space_target(target: str) -> bool: + target = _strip_channel_prefix(target) + return target.startswith("spaces/") + + +def resolve_targets(inputs: list[str], kind: str | None = None) -> list[dict[str, Any]]: + results: list[dict[str, Any]] = [] + + for target in inputs: + normalized = normalize_googlechat_target(target) + target_type = normalized.get("type", "unknown") + + if kind and target_type != kind: + continue + + results.append(normalized) + + return results + + +def _strip_channel_prefix(target: str) -> str: + for prefix in _CHANNEL_PREFIXES: + if target.lower().startswith(prefix): + return target[len(prefix) :] + return target + + +def _convert_email_to_user(email: str) -> str: + email_match = re.match(r"^[\w.+-]+@[\w-]+\.[\w.-]+$", email) + if email_match: + return f"users/{email}" + return f"users/{email}" diff --git a/backend/package/yuxi/channels/adapters/googlechat/threads.py b/backend/package/yuxi/channels/adapters/googlechat/threads.py new file mode 100644 index 00000000..6a5b9711 --- /dev/null +++ b/backend/package/yuxi/channels/adapters/googlechat/threads.py @@ -0,0 +1,18 @@ +from __future__ import annotations + + +def parse_thread_key(message: dict) -> str | None: + return message.get("thread", {}).get("threadKey") + + +def extract_thread_metadata(message: dict) -> dict: + thread = message.get("thread", {}) + return { + "thread_key": thread.get("threadKey"), + "thread_name": thread.get("name"), + "is_thread_root": not bool(thread.get("threadKey")), + } + + +def is_thread_message(message: dict) -> bool: + return bool(message.get("thread", {}).get("threadKey"))