From 52cb09e4bea91f10b49f12842ac9a104d9e130bc Mon Sep 17 00:00:00 2001 From: rookiestar28 Date: Thu, 19 Mar 2026 23:49:49 +0800 Subject: [PATCH] feat: add slack multi-workspace oauth support --- connector/config.py | 39 ++ connector/contract.py | 13 +- connector/platforms/discord_gateway.py | 8 +- connector/platforms/line_webhook.py | 8 +- .../platforms/slack_installation_manager.py | 379 ++++++++++++++++++ connector/platforms/slack_webhook.py | 308 +++++++++++++- connector/platforms/telegram_polling.py | 8 +- connector/platforms/wechat_webhook.py | 8 +- connector/platforms/whatsapp_webhook.py | 8 +- connector/results_poller.py | 132 +++++- connector/router.py | 28 +- docs/connector.md | 42 +- docs/release/api_contract.md | 5 + docs/release/config_secrets_contract.md | 8 + services/connector_installation_registry.py | 80 +++- tests/connector/test_poller.py | 26 +- tests/test_connector_installation_registry.py | 27 +- tests/test_f57_slack_transport_parity.py | 3 +- tests/test_f58_slack_oauth_installations.py | 139 +++++++ 19 files changed, 1203 insertions(+), 66 deletions(-) create mode 100644 connector/platforms/slack_installation_manager.py create mode 100644 tests/test_f58_slack_oauth_installations.py diff --git a/connector/config.py b/connector/config.py index 1c62000..ec8e576 100644 --- a/connector/config.py +++ b/connector/config.py @@ -102,6 +102,23 @@ class ConnectorConfig: slack_reply_in_thread: bool = True slack_mode: str = "events" # F57: events | socket slack_app_token: Optional[str] = None # F57: required in socket mode (xapp-...) + slack_client_id: Optional[str] = None + slack_client_secret: Optional[str] = None + slack_oauth_redirect_uri: Optional[str] = None + slack_oauth_install_path: str = "/slack/install" + slack_oauth_callback_path: str = "/slack/oauth/callback" + slack_oauth_scopes: List[str] = field( + default_factory=lambda: [ + "app_mentions:read", + "channels:history", + "chat:write", + "files:write", + "groups:history", + "im:history", + "mpim:history", + ] + ) + slack_oauth_state_ttl_sec: int = 600 # Privileged Access (ID match across platforms; Telegram Int vs Discord Str handled by router) admin_users: List[str] = field(default_factory=list) @@ -280,6 +297,28 @@ def load_config() -> ConnectorConfig: cfg.slack_reply_in_thread = False cfg.slack_mode = os.environ.get("OPENCLAW_CONNECTOR_SLACK_MODE", "events").lower() cfg.slack_app_token = os.environ.get("OPENCLAW_CONNECTOR_SLACK_APP_TOKEN") + cfg.slack_client_id = os.environ.get("OPENCLAW_CONNECTOR_SLACK_CLIENT_ID") + cfg.slack_client_secret = os.environ.get("OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET") + cfg.slack_oauth_redirect_uri = os.environ.get( + "OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI" + ) + cfg.slack_oauth_install_path = os.environ.get( + "OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH", "/slack/install" + ) + cfg.slack_oauth_callback_path = os.environ.get( + "OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH", "/slack/oauth/callback" + ) + if slack_scopes := os.environ.get("OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES"): + parsed_scopes = [ + scope.strip() for scope in slack_scopes.split(",") if scope.strip() + ] + if parsed_scopes: + cfg.slack_oauth_scopes = parsed_scopes + if slack_oauth_ttl := os.environ.get( + "OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC" + ): + if slack_oauth_ttl.isdigit(): + cfg.slack_oauth_state_ttl_sec = max(60, int(slack_oauth_ttl)) # Admin if admins := os.environ.get("OPENCLAW_CONNECTOR_ADMIN_USERS"): diff --git a/connector/contract.py b/connector/contract.py index 99886d0..8f9961a 100644 --- a/connector/contract.py +++ b/connector/contract.py @@ -4,7 +4,7 @@ Shared data models for request/response. """ from dataclasses import dataclass, field -from typing import List, Optional +from typing import Any, Dict, List, Optional @dataclass @@ -16,6 +16,9 @@ class CommandRequest: message_id: str text: str timestamp: float + workspace_id: str = "" + thread_id: str = "" + metadata: Dict[str, Any] = field(default_factory=dict) @dataclass @@ -42,10 +45,16 @@ class Platform: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[Dict[str, Any]] = None, ): """Send an image to the channel.""" pass - async def send_message(self, channel_id: str, text: str): + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[Dict[str, Any]] = None, + ): """Send a text message to the channel.""" pass diff --git a/connector/platforms/discord_gateway.py b/connector/platforms/discord_gateway.py index 33acc30..a595713 100644 --- a/connector/platforms/discord_gateway.py +++ b/connector/platforms/discord_gateway.py @@ -230,6 +230,7 @@ class DiscordGateway: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[dict] = None, ): """Send image via Discord API.""" if not self.session: @@ -279,7 +280,12 @@ class DiscordGateway: logger.error(f"Discord send_image error: {e}") raise - async def send_message(self, channel_id: str, text: str): + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[dict] = None, + ): """Send text message.""" if not self.session: return diff --git a/connector/platforms/line_webhook.py b/connector/platforms/line_webhook.py index dac49c8..098f262 100644 --- a/connector/platforms/line_webhook.py +++ b/connector/platforms/line_webhook.py @@ -286,6 +286,7 @@ class LINEWebhookServer: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[dict] = None, ): """ Send image via LINE using public URL. @@ -368,7 +369,12 @@ class LINEWebhookServer: err = await resp.text() logger.error(f"LINE image push failed: {resp.status} {err}") - async def send_message(self, channel_id: str, text: str): + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[dict] = None, + ): """Send push message.""" if self._session_invalid: logger.warning("R93: Connector session invalid - blocking outbound message") diff --git a/connector/platforms/slack_installation_manager.py b/connector/platforms/slack_installation_manager.py new file mode 100644 index 0000000..e60702e --- /dev/null +++ b/connector/platforms/slack_installation_manager.py @@ -0,0 +1,379 @@ +from __future__ import annotations + +import json +import logging +import os +import secrets +import threading +import time +from pathlib import Path +from typing import Any, Dict, Optional, Tuple +from urllib.parse import urlencode + +from connector.config import ConnectorConfig + +try: + from services.audit import emit_audit_event + from services.connector_installation_registry import ( + ConnectorInstallation, + ConnectorInstallationRegistry, + InstallationResolution, + InstallationStatus, + get_connector_installation_registry, + ) + from services.secret_store import SecretStore, get_secret_store + from services.state_dir import get_state_dir +except ImportError: # pragma: no cover + from services.audit import emit_audit_event # type: ignore + from services.connector_installation_registry import ( # type: ignore + ConnectorInstallation, + ConnectorInstallationRegistry, + InstallationResolution, + InstallationStatus, + get_connector_installation_registry, + ) + from services.secret_store import SecretStore, get_secret_store # type: ignore + from services.state_dir import get_state_dir # type: ignore + +logger = logging.getLogger(__name__) + +SLACK_AUTHORIZE_URL = "https://slack.com/oauth/v2/authorize" +SLACK_OAUTH_ACCESS_URL = "https://slack.com/api/oauth.v2.access" +SLACK_OAUTH_STATE_FILE = "slack_oauth_states.json" + +_INVALID_TOKEN_ERRORS = frozenset( + {"account_inactive", "invalid_auth", "not_authed", "token_revoked"} +) +_DEGRADED_TOKEN_ERRORS = frozenset({"ratelimited", "request_timeout", "fatal_error"}) + + +def _load_aiohttp(): + try: + import aiohttp # type: ignore + except ModuleNotFoundError: + return None + return aiohttp + + +class SlackInstallationManager: + def __init__( + self, + config: ConnectorConfig, + *, + registry: Optional[ConnectorInstallationRegistry] = None, + secret_store: Optional[SecretStore] = None, + state_dir: Optional[str] = None, + ): + self.config = config + self._state_dir = Path(state_dir or get_state_dir()) + self._state_path = self._state_dir / SLACK_OAUTH_STATE_FILE + self._registry = registry or get_connector_installation_registry( + state_dir=str(self._state_dir) + ) + self._secret_store = secret_store or get_secret_store(str(self._state_dir)) + self._lock = threading.RLock() + self._states: Dict[str, Dict[str, Any]] = {} + self._load_states() + + @property + def oauth_enabled(self) -> bool: + return bool(self.config.slack_client_id and self.config.slack_client_secret) + + def resolve_redirect_uri(self) -> str: + if self.config.slack_oauth_redirect_uri: + return str(self.config.slack_oauth_redirect_uri).strip() + if self.config.public_base_url: + base = self.config.public_base_url.rstrip("/") + path = self.config.slack_oauth_callback_path or "/slack/oauth/callback" + return f"{base}{path}" + return "" + + def can_handle_oauth(self) -> bool: + return self.oauth_enabled and bool(self.resolve_redirect_uri()) + + def _save_states(self) -> None: + self._state_dir.mkdir(parents=True, exist_ok=True) + temp_path = self._state_path.with_suffix(".tmp") + temp_path.write_text(json.dumps(self._states, indent=2), encoding="utf-8") + os.replace(temp_path, self._state_path) + + def _load_states(self) -> None: + if not self._state_path.exists(): + return + try: + data = json.loads(self._state_path.read_text(encoding="utf-8")) + if isinstance(data, dict): + self._states = data + except Exception as exc: + logger.warning("Failed to load Slack OAuth state store: %s", exc) + self._states = {} + self._prune_expired_states() + + def _prune_expired_states(self) -> None: + now = time.time() + ttl = max(60, int(self.config.slack_oauth_state_ttl_sec or 600)) + changed = False + for key, payload in list(self._states.items()): + created_at = float(payload.get("created_at", 0) or 0) + if not created_at or (now - created_at) > ttl: + self._states.pop(key, None) + changed = True + if changed: + self._save_states() + + def issue_install_state(self) -> str: + if not self.can_handle_oauth(): + raise RuntimeError("Slack OAuth flow not configured") + with self._lock: + self._prune_expired_states() + state = secrets.token_urlsafe(32) + self._states[state] = {"created_at": time.time()} + self._save_states() + return state + + def consume_install_state(self, state: str) -> bool: + with self._lock: + self._prune_expired_states() + payload = self._states.pop(str(state or "").strip(), None) + if payload is None: + return False + self._save_states() + return True + + def build_install_url(self, state: str) -> str: + params = { + "client_id": self.config.slack_client_id or "", + "scope": ",".join(self.config.slack_oauth_scopes or []), + "redirect_uri": self.resolve_redirect_uri(), + "state": state, + } + return f"{SLACK_AUTHORIZE_URL}?{urlencode(params)}" + + async def exchange_code(self, code: str) -> Dict[str, Any]: + aiohttp = _load_aiohttp() + if aiohttp is None: + raise RuntimeError("aiohttp required for Slack OAuth exchange") + payload = { + "client_id": self.config.slack_client_id or "", + "client_secret": self.config.slack_client_secret or "", + "code": str(code or "").strip(), + "redirect_uri": self.resolve_redirect_uri(), + } + async with aiohttp.ClientSession() as session: + async with session.post( + SLACK_OAUTH_ACCESS_URL, + data=payload, + timeout=aiohttp.ClientTimeout(total=15), + ) as resp: + data = await resp.json(content_type=None) + if resp.status != 200 or not data.get("ok"): + raise RuntimeError( + f"slack_oauth_exchange_failed:{resp.status}:{data.get('error', 'unknown')}" + ) + return data + + def _normalize_workspace_id(self, payload: Dict[str, Any]) -> str: + workspace_id = ( + (payload.get("team") or {}).get("id") + or payload.get("team_id") + or ( + (payload.get("enterprise") or {}).get("id") + if payload.get("enterprise") + else "" + ) + ) + workspace_id = str(workspace_id or "").strip() + if not workspace_id: + raise ValueError("workspace_id_missing") + return workspace_id + + def installation_id_for_workspace(self, workspace_id: str) -> str: + return f"slack:{str(workspace_id or '').strip()}" + + def metadata_from_oauth_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: + team = dict(payload.get("team", {}) or {}) + enterprise = dict(payload.get("enterprise", {}) or {}) + authed_user = dict(payload.get("authed_user", {}) or {}) + metadata = { + "workspace_name": str(team.get("name", "") or "").strip(), + "enterprise_id": str(enterprise.get("id", "") or "").strip(), + "enterprise_name": str(enterprise.get("name", "") or "").strip(), + "bot_user_id": str(payload.get("bot_user_id", "") or "").strip(), + "app_id": str(payload.get("app_id", "") or "").strip(), + "scope": str(payload.get("scope", "") or "").strip(), + "authed_user_id": str(authed_user.get("id", "") or "").strip(), + "token_type": str(payload.get("token_type", "") or "").strip(), + "transport_mode": self.config.slack_mode, + } + return {key: value for key, value in metadata.items() if value} + + def upsert_from_oauth_payload( + self, payload: Dict[str, Any] + ) -> ConnectorInstallation: + workspace_id = self._normalize_workspace_id(payload) + installation_id = self.installation_id_for_workspace(workspace_id) + token_values = {"bot_token": str(payload.get("access_token", "") or "").strip()} + if self.config.slack_app_token: + token_values["app_token"] = self.config.slack_app_token + if not token_values["bot_token"]: + raise ValueError("bot_token_missing") + + metadata = self.metadata_from_oauth_payload(payload) + existing = self._registry.get_installation(installation_id) + if existing is not None: + rotated = self._registry.rotate_installation_tokens( + installation_id, + token_values, + reason="slack_oauth_reinstall", + ) + inst = self._registry.upsert_installation( + platform="slack", + workspace_id=workspace_id, + installation_id=installation_id, + token_refs=rotated.token_refs, + status=rotated.status, + metadata=metadata, + status_reason="slack_oauth_reinstall", + ) + else: + inst = self._registry.upsert_installation( + platform="slack", + workspace_id=workspace_id, + installation_id=installation_id, + token_values=token_values, + status=InstallationStatus.CREATED.value, + metadata=metadata, + status_reason="slack_oauth_install", + ) + inst = self._registry.activate_installation( + installation_id, reason="slack_oauth_complete" + ) + inst = self._registry.update_installation_health( + installation_id, + health_code="ok", + reason="slack_oauth_complete", + details={"workspace_id": workspace_id}, + ) + emit_audit_event( + action="connector.slack.oauth.install", + target=installation_id, + outcome="allow", + status_code=200, + details={ + "workspace_id": workspace_id, + "workspace_name": metadata.get("workspace_name", ""), + "transport_mode": self.config.slack_mode, + }, + ) + return inst + + def extract_workspace_id(self, payload: Dict[str, Any]) -> str: + if isinstance(payload.get("team_id"), str) and payload.get("team_id"): + return str(payload["team_id"]).strip() + team = payload.get("team") or {} + if isinstance(team, dict) and team.get("id"): + return str(team.get("id")).strip() + authorizations = payload.get("authorizations") or [] + if isinstance(authorizations, list) and authorizations: + workspace_id = str((authorizations[0] or {}).get("team_id", "")).strip() + if workspace_id: + return workspace_id + event = payload.get("event") or {} + workspace_id = str(event.get("team", "") or "").strip() + return workspace_id + + def resolve_workspace_tokens( + self, workspace_id: str + ) -> Tuple[InstallationResolution, Dict[str, str]]: + resolution = self._registry.resolve_installation("slack", workspace_id) + if not resolution.ok or resolution.installation is None: + emit_audit_event( + action="connector.slack.resolve", + target=workspace_id or "unknown_workspace", + outcome="deny", + status_code=409, + details={ + "workspace_id": workspace_id, + "reject_reason": resolution.reject_reason, + "health_code": resolution.health_code, + }, + ) + return resolution, {} + + tokens: Dict[str, str] = {} + for token_name, ref in resolution.installation.token_refs.items(): + secret = self._secret_store.get_secret( + ref, tenant_id=resolution.installation.tenant_id + ) + if secret: + tokens[token_name] = secret + return resolution, tokens + + def bot_user_id_for_installation( + self, installation: Optional[ConnectorInstallation] + ) -> str: + if installation is None: + return "" + return str( + ( + installation.metadata.get("bot_user_id", "") + if installation.metadata + else "" + ) + or "" + ).strip() + + def mark_installation_health( + self, + installation_id: str, + *, + health_code: str, + reason: str, + details: Optional[Dict[str, Any]] = None, + ) -> None: + self._registry.update_installation_health( + installation_id, + health_code=health_code, + reason=reason, + details=details, + ) + + def uninstall_installation(self, installation_id: str, *, reason: str) -> None: + self._registry.uninstall_installation(installation_id, reason=reason) + + def mark_resolution_success(self, installation_id: str, workspace_id: str) -> None: + self._registry.update_installation_health( + installation_id, + health_code="ok", + reason="workspace_resolved", + details={"workspace_id": workspace_id}, + ) + + def classify_error_health(self, error_code: str, status_code: int = 0) -> str: + normalized = str(error_code or "").strip().lower() + if normalized in _INVALID_TOKEN_ERRORS or status_code in (401, 403): + return "invalid_token" + if ( + normalized in _DEGRADED_TOKEN_ERRORS + or status_code == 429 + or status_code >= 500 + ): + return "degraded" + return "degraded" + + def mark_api_error( + self, + installation_id: str, + *, + error_code: str, + status_code: int = 0, + details: Optional[Dict[str, Any]] = None, + ) -> str: + health_code = self.classify_error_health(error_code, status_code=status_code) + self.mark_installation_health( + installation_id, + health_code=health_code, + reason=error_code or f"http_{status_code}", + details=details, + ) + return health_code diff --git a/connector/platforms/slack_webhook.py b/connector/platforms/slack_webhook.py index a54b9b1..e5cd771 100644 --- a/connector/platforms/slack_webhook.py +++ b/connector/platforms/slack_webhook.py @@ -33,12 +33,13 @@ import hmac import json import logging import time -from typing import Any, Dict, Optional +from typing import Any, Dict, Optional, Tuple from ..config import ConnectorConfig from ..contract import CommandRequest, CommandResponse from ..router import CommandRouter from ..security_profile import AllowlistPolicy, ReplayGuard +from .slack_installation_manager import SlackInstallationManager logger = logging.getLogger(__name__) @@ -90,6 +91,12 @@ def _make_json_response(web_mod, data: dict, *, status: int = 200): ) +def _make_redirect_response(web_mod, url: str): + if web_mod is not None: + raise web_mod.HTTPFound(location=url) + return _CompatResponse(status=302, text=url) + + # -- Slack signature verification ------------------------------------------- # Maximum acceptable clock skew for timestamp validation (5 minutes). @@ -172,9 +179,11 @@ class SlackWebhookServer: self._channel_allowlist = AllowlistPolicy( config.slack_allowed_channels, strict=False ) + self._installation_manager = SlackInstallationManager(config) # Bot user ID (resolved on first event or set from config) self._bot_user_id: Optional[str] = None + self._bot_user_ids: Dict[str, str] = {} # ------------------------------------------------------------------ # Lifecycle @@ -186,11 +195,19 @@ class SlackWebhookServer: logger.warning("aiohttp not installed. Skipping Slack adapter.") return - if not self.config.slack_bot_token or not self.config.slack_signing_secret: + if not self.config.slack_signing_secret: logger.info( "Slack adapter disabled " - "(OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN or " - "OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET missing)" + "(OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET missing)" + ) + return + if ( + not self.config.slack_bot_token + and not self._installation_manager.can_handle_oauth() + ): + logger.info( + "Slack adapter disabled " + "(legacy bot token missing and Slack OAuth flow not configured)" ) return @@ -202,6 +219,13 @@ class SlackWebhookServer: self.app = web.Application() self.app.router.add_post(self.config.slack_webhook_path, self.handle_event) + if self._installation_manager.can_handle_oauth(): + self.app.router.add_get( + self.config.slack_oauth_install_path, self.handle_oauth_install + ) + self.app.router.add_get( + self.config.slack_oauth_callback_path, self.handle_oauth_callback + ) self.runner = web.AppRunner(self.app) await self.runner.setup() @@ -220,6 +244,155 @@ class SlackWebhookServer: # Event handler # ------------------------------------------------------------------ + async def handle_oauth_install(self, request): + _, web = _import_aiohttp_web() + if not self._installation_manager.can_handle_oauth(): + return _make_response(web, status=503, text="Slack OAuth not configured") + state = self._installation_manager.issue_install_state() + return _make_redirect_response( + web, self._installation_manager.build_install_url(state) + ) + + async def handle_oauth_callback(self, request): + _, web = _import_aiohttp_web() + if not self._installation_manager.can_handle_oauth(): + return _make_response(web, status=503, text="Slack OAuth not configured") + query = getattr(request, "query", {}) or {} + if query.get("error"): + return _make_response( + web, + status=400, + text=f"Slack OAuth rejected: {query.get('error')}", + ) + state = str(query.get("state", "") or "").strip() + code = str(query.get("code", "") or "").strip() + if not state or not code: + return _make_response(web, status=400, text="Missing OAuth callback fields") + if not self._installation_manager.consume_install_state(state): + return _make_response( + web, status=400, text="Invalid or replayed OAuth state" + ) + try: + payload = await self._installation_manager.exchange_code(code) + installation = self._installation_manager.upsert_from_oauth_payload(payload) + return _make_response( + web, + status=200, + text=( + "Slack installation complete for " + f"{installation.workspace_id} ({installation.installation_id})." + ), + ) + except Exception as exc: + logger.warning("Slack OAuth callback failed: %s", exc) + return _make_response(web, status=502, text=str(exc)) + + def _get_bot_user_id(self, payload: Dict[str, Any], workspace_id: str) -> str: + candidate = "" + if workspace_id and workspace_id in self._bot_user_ids: + return self._bot_user_ids[workspace_id] + if self._bot_user_id: + return self._bot_user_id + authorizations = payload.get("authorizations", []) + if authorizations and isinstance(authorizations, list): + candidate = str((authorizations[0] or {}).get("user_id", "") or "").strip() + if candidate: + self._bot_user_id = candidate + if workspace_id: + self._bot_user_ids[workspace_id] = candidate + return candidate + if workspace_id: + workspace_resolution, _ = ( + self._installation_manager.resolve_workspace_tokens(workspace_id) + ) + candidate = self._installation_manager.bot_user_id_for_installation( + workspace_resolution.installation if workspace_resolution.ok else None + ) + if candidate: + self._bot_user_ids[workspace_id] = candidate + if self._bot_user_id is None: + self._bot_user_id = candidate + return candidate + + def _resolve_workspace_credentials( + self, workspace_id: str + ) -> Tuple[Optional[str], Optional[str], Optional[str]]: + workspace_id = str(workspace_id or "").strip() + if workspace_id: + resolution, tokens = self._installation_manager.resolve_workspace_tokens( + workspace_id + ) + if resolution.ok and resolution.installation is not None: + bot_token = tokens.get("bot_token") + if bot_token: + self._installation_manager.mark_resolution_success( + resolution.installation.installation_id, workspace_id + ) + return ( + resolution.installation.installation_id, + bot_token, + workspace_id, + ) + logger.warning( + "Slack workspace %s resolved without bot token secret", workspace_id + ) + return ( + resolution.installation.installation_id, + None, + workspace_id, + ) + if ( + not self._installation_manager.oauth_enabled + and self.config.slack_bot_token + ): + return (None, self.config.slack_bot_token, workspace_id) + logger.warning( + "Slack workspace resolution failed for %s: %s (%s)", + workspace_id, + resolution.reject_reason, + resolution.health_code, + ) + return (None, None, workspace_id) + if self.config.slack_bot_token: + return (None, self.config.slack_bot_token, "") + return (None, None, workspace_id) + + def _handle_lifecycle_event(self, workspace_id: str, event_type: str) -> None: + installation_id = self._installation_manager.installation_id_for_workspace( + workspace_id + ) + try: + if event_type == "app_uninstalled": + self._installation_manager.mark_installation_health( + installation_id, + health_code="revoked", + reason="slack_app_uninstalled", + details={"workspace_id": workspace_id}, + ) + self._installation_manager.uninstall_installation( + installation_id, reason="slack_app_uninstalled" + ) + elif event_type == "tokens_revoked": + self._installation_manager.mark_installation_health( + installation_id, + health_code="invalid_token", + reason="slack_tokens_revoked", + details={"workspace_id": workspace_id}, + ) + elif event_type == "app_rate_limited": + self._installation_manager.mark_installation_health( + installation_id, + health_code="degraded", + reason="slack_app_rate_limited", + details={"workspace_id": workspace_id}, + ) + except ValueError: + logger.warning( + "Slack lifecycle event for unbound workspace %s (%s)", + workspace_id, + event_type, + ) + async def handle_event(self, request): """POST handler for Slack Events API.""" _, web = _import_aiohttp_web() @@ -273,6 +446,12 @@ class SlackWebhookServer: event = payload.get("event", {}) event_id = payload.get("event_id", "") event_type = event.get("type", "") + workspace_id = self._installation_manager.extract_workspace_id(payload) + + if event_type in ("app_uninstalled", "tokens_revoked", "app_rate_limited"): + if workspace_id: + self._handle_lifecycle_event(workspace_id, event_type) + return # -- Step 5: Replay / dedupe guard -- if not event_id: @@ -285,13 +464,10 @@ class SlackWebhookServer: # -- Step 6: Bot-loop prevention -- # Resolve bot user ID from authorizations or cache. - if self._bot_user_id is None: - auths = payload.get("authorizations", []) - if auths and isinstance(auths, list): - self._bot_user_id = auths[0].get("user_id", "") + bot_user_id = self._get_bot_user_id(payload, workspace_id) sender_id = event.get("user", "") - if sender_id and sender_id == self._bot_user_id: + if sender_id and bot_user_id and sender_id == bot_user_id: return if event.get("bot_id"): @@ -317,11 +493,11 @@ class SlackWebhookServer: is_dm = channel_id.startswith("D") if not is_dm and self.config.slack_require_mention: if event_type != "app_mention": - if self._bot_user_id and f"<@{self._bot_user_id}>" not in text: + if bot_user_id and f"<@{bot_user_id}>" not in text: return - if self._bot_user_id: - text = text.replace(f"<@{self._bot_user_id}>", "").strip() + if bot_user_id: + text = text.replace(f"<@{bot_user_id}>", "").strip() # -- Step 8: Allowlist checks (S67) -- if self._user_allowlist.entries: @@ -345,6 +521,9 @@ class SlackWebhookServer: message_id=event_id, text=text, timestamp=float(message_ts) if message_ts else time.time(), + workspace_id=workspace_id, + thread_id=thread_ts + or (message_ts if self.config.slack_reply_in_thread else ""), ) try: @@ -357,8 +536,11 @@ class SlackWebhookServer: await self._send_reply( channel_id=channel_id, text=resp_text, - thread_ts=thread_ts - or (message_ts if self.config.slack_reply_in_thread else ""), + thread_ts=req.thread_id, + delivery_context={ + "workspace_id": workspace_id, + "thread_id": req.thread_id, + }, ) except Exception as e: logger.exception(f"Error handling Slack event: {e}") @@ -372,6 +554,7 @@ class SlackWebhookServer: channel_id: str, text: str, thread_ts: str = "", + delivery_context: Optional[Dict[str, Any]] = None, ) -> None: """Send a message via Slack Web API (chat.postMessage).""" try: @@ -380,9 +563,22 @@ class SlackWebhookServer: logger.warning("aiohttp not available; cannot send Slack reply") return + ctx = dict(delivery_context or {}) + if not thread_ts: + thread_ts = str(ctx.get("thread_id", "") or "").strip() + installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( + str(ctx.get("workspace_id", "") or "").strip() + ) + if not bot_token: + logger.warning( + "Slack reply dropped: no workspace token available (workspace=%s)", + workspace_id or "legacy", + ) + return + url = "https://slack.com/api/chat.postMessage" headers = { - "Authorization": f"Bearer {self.config.slack_bot_token}", + "Authorization": f"Bearer {bot_token}", "Content-Type": "application/json; charset=utf-8", } payload: Dict[str, Any] = { @@ -402,15 +598,41 @@ class SlackWebhookServer: ) as resp: if resp.status != 200: body = await resp.text() + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=f"http_{resp.status}", + status_code=resp.status, + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + }, + ) logger.warning( f"Slack API error: status={resp.status} body={body[:200]}" ) else: data = await resp.json() if not data.get("ok"): + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=str(data.get("error", "unknown")), + details={ + "workspace_id": workspace_id, + "path": "chat.postMessage", + }, + ) logger.warning( f"Slack API error: {data.get('error', 'unknown')}" ) + elif installation_id: + self._installation_manager.mark_installation_health( + installation_id, + health_code="ok", + reason="chat_post_message_ok", + details={"workspace_id": workspace_id}, + ) except Exception as e: logger.warning(f"Slack reply failed: {e}") @@ -418,9 +640,18 @@ class SlackWebhookServer: # Platform contract: send_message / send_image # ------------------------------------------------------------------ - async def send_message(self, channel_id: str, text: str): + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[Dict[str, Any]] = None, + ): """Platform contract: send text message.""" - await self._send_reply(channel_id=channel_id, text=text) + await self._send_reply( + channel_id=channel_id, + text=text, + delivery_context=delivery_context, + ) async def send_image( self, @@ -428,6 +659,7 @@ class SlackWebhookServer: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[Dict[str, Any]] = None, ): """Platform contract: send image (Slack files.upload).""" try: @@ -436,15 +668,29 @@ class SlackWebhookServer: logger.warning("aiohttp not available; cannot upload Slack image") return + ctx = dict(delivery_context or {}) + thread_ts = str(ctx.get("thread_id", "") or "").strip() + installation_id, bot_token, workspace_id = self._resolve_workspace_credentials( + str(ctx.get("workspace_id", "") or "").strip() + ) + if not bot_token: + logger.warning( + "Slack image dropped: no workspace token available (workspace=%s)", + workspace_id or "legacy", + ) + return + url = "https://slack.com/api/files.upload" headers = { - "Authorization": f"Bearer {self.config.slack_bot_token}", + "Authorization": f"Bearer {bot_token}", } data = _aiohttp.FormData() data.add_field("file", image_data, filename=filename, content_type="image/png") data.add_field("channels", channel_id) if caption: data.add_field("initial_comment", caption) + if thread_ts: + data.add_field("thread_ts", thread_ts) try: async with _aiohttp.ClientSession() as session: @@ -455,12 +701,38 @@ class SlackWebhookServer: timeout=_aiohttp.ClientTimeout(total=30), ) as resp: if resp.status != 200: + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=f"http_{resp.status}", + status_code=resp.status, + details={ + "workspace_id": workspace_id, + "path": "files.upload", + }, + ) logger.warning(f"Slack file upload error: status={resp.status}") else: resp_data = await resp.json() if not resp_data.get("ok"): + if installation_id: + self._installation_manager.mark_api_error( + installation_id, + error_code=str(resp_data.get("error", "unknown")), + details={ + "workspace_id": workspace_id, + "path": "files.upload", + }, + ) logger.warning( f"Slack file upload error: {resp_data.get('error')}" ) + elif installation_id: + self._installation_manager.mark_installation_health( + installation_id, + health_code="ok", + reason="files_upload_ok", + details={"workspace_id": workspace_id}, + ) except Exception as e: logger.warning(f"Slack image upload failed: {e}") diff --git a/connector/platforms/telegram_polling.py b/connector/platforms/telegram_polling.py index a64e9c3..0c5c19e 100644 --- a/connector/platforms/telegram_polling.py +++ b/connector/platforms/telegram_polling.py @@ -182,6 +182,7 @@ class TelegramPolling: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[dict] = None, ): """Send photo via Telegram sendPhoto.""" if not self.session: @@ -205,7 +206,12 @@ class TelegramPolling: except Exception as e: logger.error(f"Telegram send_image error: {e}") - async def send_message(self, channel_id: str, text: str): + async def send_message( + self, + channel_id: str, + text: str, + delivery_context: Optional[dict] = None, + ): """Send text message.""" if not self.session: return diff --git a/connector/platforms/wechat_webhook.py b/connector/platforms/wechat_webhook.py index 0f328b6..1f5b886 100644 --- a/connector/platforms/wechat_webhook.py +++ b/connector/platforms/wechat_webhook.py @@ -680,7 +680,12 @@ class WeChatWebhookServer: # Outbound: Text (Customer Service Message API) # ------------------------------------------------------------------ - async def send_message(self, recipient_openid: str, text: str): + async def send_message( + self, + recipient_openid: str, + text: str, + delivery_context: Optional[dict] = None, + ): """ Send text via WeChat Customer Service Message API. @@ -736,6 +741,7 @@ class WeChatWebhookServer: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[dict] = None, ): """ Send image via WeChat. diff --git a/connector/platforms/whatsapp_webhook.py b/connector/platforms/whatsapp_webhook.py index 63176af..2b8b147 100644 --- a/connector/platforms/whatsapp_webhook.py +++ b/connector/platforms/whatsapp_webhook.py @@ -296,7 +296,12 @@ class WhatsAppWebhookServer: # Outbound: Text # ------------------------------------------------------------------ - async def send_message(self, recipient_id: str, text: str): + async def send_message( + self, + recipient_id: str, + text: str, + delivery_context: Optional[dict] = None, + ): """Send text message via WhatsApp Cloud API.""" if self._session_invalid: logger.warning("R93: Connector session invalid - blocking outbound") @@ -350,6 +355,7 @@ class WhatsAppWebhookServer: image_data: bytes, filename: str = "image.png", caption: Optional[str] = None, + delivery_context: Optional[dict] = None, ): """ Send image via WhatsApp using public media URL. diff --git a/connector/results_poller.py b/connector/results_poller.py index 436623f..e226fd0 100644 --- a/connector/results_poller.py +++ b/connector/results_poller.py @@ -1,7 +1,7 @@ import asyncio import logging import time -from typing import Dict, Optional +from typing import Any, Dict, Optional from .config import ConnectorConfig from .contract import Platform @@ -26,10 +26,10 @@ class ResultsPoller: self.platforms = platforms # map "telegram" -> TelegramPolling, etc. self.queue = ( asyncio.Queue() - ) # (prompt_id, platform_name, channel_id, sender_id) + ) # (prompt_id, platform_name, channel_id, sender_id, delivery_context) self.approval_queue = ( asyncio.Queue() - ) # (approval_id, platform_name, channel_id, sender_id) + ) # (approval_id, platform_name, channel_id, sender_id, delivery_context) self.active_polls = {} # prompt_id -> task self.active_approval_polls = {} # approval_id -> task @@ -55,17 +55,35 @@ class ResultsPoller: logger.info("ResultsPoller stopped.") def track_job( - self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str + self, + prompt_id: str, + platform_name: str, + channel_id: str, + sender_id: str, + delivery_context: Optional[Dict[str, Any]] = None, ): """Enqueue a job for result monitoring.""" if not prompt_id: return logger.info(f"Tracking job {prompt_id} for {platform_name} in {channel_id}") - self.queue.put_nowait((prompt_id, platform_name, channel_id, sender_id)) + self.queue.put_nowait( + ( + prompt_id, + platform_name, + channel_id, + sender_id, + dict(delivery_context or {}), + ) + ) def track_approval( - self, approval_id: str, platform_name: str, channel_id: str, sender_id: str + self, + approval_id: str, + platform_name: str, + channel_id: str, + sender_id: str, + delivery_context: Optional[Dict[str, Any]] = None, ): if not approval_id: return @@ -85,16 +103,28 @@ class ResultsPoller: f"Tracking approval {approval_id} for {platform_name} in {channel_id}" ) self.approval_queue.put_nowait( - (approval_id, platform_name, channel_id, sender_id) + ( + approval_id, + platform_name, + channel_id, + sender_id, + dict(delivery_context or {}), + ) ) async def _job_consumer(self): while True: item = await self.queue.get() try: - prompt_id, platform_name, channel_id, sender_id = item + prompt_id, platform_name, channel_id, sender_id, delivery_context = item task = asyncio.create_task( - self._poll_job(prompt_id, platform_name, channel_id, sender_id) + self._poll_job( + prompt_id, + platform_name, + channel_id, + sender_id, + delivery_context, + ) ) self.active_polls[prompt_id] = task task.add_done_callback( @@ -107,10 +137,20 @@ class ResultsPoller: while True: item = await self.approval_queue.get() try: - approval_id, platform_name, channel_id, sender_id = item + ( + approval_id, + platform_name, + channel_id, + sender_id, + delivery_context, + ) = item task = asyncio.create_task( self._poll_approval( - approval_id, platform_name, channel_id, sender_id + approval_id, + platform_name, + channel_id, + sender_id, + delivery_context, ) ) self.active_approval_polls[approval_id] = task @@ -121,7 +161,12 @@ class ResultsPoller: self.approval_queue.task_done() async def _poll_approval( - self, approval_id: str, platform_name: str, channel_id: str, sender_id: str + self, + approval_id: str, + platform_name: str, + channel_id: str, + sender_id: str, + delivery_context: Optional[Dict[str, Any]] = None, ): start_time = time.time() delay = 2.0 @@ -137,6 +182,7 @@ class ResultsPoller: platform_name, channel_id, f"❌ Approval {approval_id} {status}.", + delivery_context=delivery_context, ) return if status == "approved": @@ -150,7 +196,11 @@ class ResultsPoller: ) if prompt_id: self.track_job( - prompt_id, platform_name, channel_id, sender_id + prompt_id, + platform_name, + channel_id, + sender_id, + delivery_context=delivery_context, ) return except Exception as e: @@ -166,10 +216,16 @@ class ResultsPoller: platform_name, channel_id, f"⚠️ Approval {approval_id} timed out waiting for execution.", + delivery_context=delivery_context, ) async def _poll_job( - self, prompt_id: str, platform_name: str, channel_id: str, sender_id: str + self, + prompt_id: str, + platform_name: str, + channel_id: str, + sender_id: str, + delivery_context: Optional[Dict[str, Any]] = None, ): """Poll history with backoff until complete or timeout.""" start_time = time.time() @@ -185,7 +241,11 @@ class ResultsPoller: if prompt_id in data: job_data = data[prompt_id] await self._deliver_results( - prompt_id, job_data, platform_name, channel_id + prompt_id, + job_data, + platform_name, + channel_id, + delivery_context=delivery_context, ) return except Exception as e: @@ -203,10 +263,16 @@ class ResultsPoller: platform_name, channel_id, f"⚠️ Job {prompt_id} timed out waiting for results.", + delivery_context=delivery_context, ) async def _deliver_results( - self, prompt_id: str, job_data: dict, platform_name: str, channel_id: str + self, + prompt_id: str, + job_data: dict, + platform_name: str, + channel_id: str, + delivery_context: Optional[Dict[str, Any]] = None, ): """Download images and send to platform.""" outputs = job_data.get("outputs", {}) @@ -216,6 +282,7 @@ class ResultsPoller: platform_name, channel_id, f"✅ Job {prompt_id} finished (No output images).", + delivery_context=delivery_context, ) return @@ -231,7 +298,10 @@ class ResultsPoller: if not images_to_send: logger.info(f"Job {prompt_id} finished but no images found.") await self._send_text( - platform_name, channel_id, f"✅ Job {prompt_id} finished (No images)." + platform_name, + channel_id, + f"✅ Job {prompt_id} finished (No images).", + delivery_context=delivery_context, ) return @@ -263,23 +333,43 @@ class ResultsPoller: platform_name, channel_id, f"⚠️ Image {filename} skipped (too large).", + delivery_context=delivery_context, ) continue # Send with error handling try: - await platform.send_image(channel_id, content, filename=filename) + await platform.send_image( + channel_id, + content, + filename=filename, + delivery_context=delivery_context, + ) except Exception as e: logger.error(f"Failed to deliver image to {platform_name}: {e}") # Fallback text await self._send_text( - platform_name, channel_id, f"⚠️ Failed to send image: {filename}" + platform_name, + channel_id, + f"⚠️ Failed to send image: {filename}", + delivery_context=delivery_context, ) - async def _send_text(self, platform_name: str, channel_id: str, text: str): + async def _send_text( + self, + platform_name: str, + channel_id: str, + text: str, + *, + delivery_context: Optional[Dict[str, Any]] = None, + ): platform = self.platforms.get(platform_name) if platform: try: - await platform.send_message(channel_id, text) + await platform.send_message( + channel_id, + text, + delivery_context=dict(delivery_context or {}), + ) except Exception as e: logger.error(f"Failed to send text to {platform_name}: {e}") diff --git a/connector/router.py b/connector/router.py index 51f64f0..d3fef07 100644 --- a/connector/router.py +++ b/connector/router.py @@ -184,6 +184,14 @@ class CommandRouter: def _is_admin(self, user_id: str) -> bool: return str(user_id) in self.config.admin_users + def _delivery_context(self, req: CommandRequest) -> Dict[str, Any]: + context: Dict[str, Any] = {} + if getattr(req, "workspace_id", ""): + context["workspace_id"] = str(req.workspace_id) + if getattr(req, "thread_id", ""): + context["thread_id"] = str(req.thread_id) + return context + def _check_command_authz( self, cmd: str, req: CommandRequest, default_class: CommandClass ) -> Optional[CommandResponse]: @@ -415,14 +423,22 @@ class CommandRouter: # We must start tracking the approval_id so we can map # approval_id -> executed_prompt_id later and auto-deliver images. self.poller.track_approval( - approval_id, req.platform, req.channel_id, req.sender_id + approval_id, + req.platform, + req.channel_id, + req.sender_id, + delivery_context=self._delivery_context(req), ) return CommandResponse(text=msg) else: prompt_id = data.get("prompt_id", "unknown") if self.poller: self.poller.track_job( - prompt_id, req.platform, req.channel_id, req.sender_id + prompt_id, + req.platform, + req.channel_id, + req.sender_id, + delivery_context=self._delivery_context(req), ) return CommandResponse( @@ -547,7 +563,13 @@ class CommandRouter: if self.poller: # Approval request might have come from different flow, but usually user invoking /approve # wants the result. Using current req context is safest assumption for "ChatOps". - self.poller.track_job(pid, req.platform, req.channel_id, req.sender_id) + self.poller.track_job( + pid, + req.platform, + req.channel_id, + req.sender_id, + delivery_context=self._delivery_context(req), + ) elif data.get("executed") is False: msg += "\n(Not Executed)" if err := data.get("execution_error"): diff --git a/docs/connector.md b/docs/connector.md index d9799da..fc2cf4d 100644 --- a/docs/connector.md +++ b/docs/connector.md @@ -27,6 +27,7 @@ OpenClaw now includes a platform-agnostic baseline for multi-workspace connector - `platform`, `workspace_id`, `installation_id`, `token_refs`, `status`, `updated_at` - token material is kept in encrypted server-side secret storage; registry and diagnostics expose token references only - workspace resolution is fail-closed on missing/ambiguous/inactive/stale bindings +- installation diagnostics can also surface stable health states such as `ok`, `invalid_token`, `revoked`, `workspace_unbound`, and `degraded` - interactive callback contract enforces signed envelope checks, timestamp window, payload-hash validation, replay/idempotency guardrails, and command-policy mapping (`public`/`run`/`admin`) with explicit force-approval outcomes for untrusted `run` callbacks Admin diagnostics APIs: @@ -36,6 +37,12 @@ Admin diagnostics APIs: - `GET /openclaw/connector/installations/resolve?platform=&workspace_id=` - `GET /openclaw/connector/installations/audit` +Slack multi-workspace notes: + +- Slack OAuth installs bind one workspace per installation record and persist only encrypted token refs. +- `GET /openclaw/connector/installations/resolve?platform=slack&workspace_id=` returns the fail-closed resolution view for a specific Slack workspace. +- `GET /openclaw/connector/installations` diagnostics may include per-install health metadata plus aggregate `health_counts`. + ### Multi-tenant boundary behavior When backend multi-tenant mode is enabled (`OPENCLAW_MULTI_TENANT_ENABLED=1`): @@ -145,8 +152,15 @@ Set the following environment variables (or put them in a `.env` file if you use *(Requires Inbound Connectivity - see below)* -- `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN`: Bot User OAuth Token (`xoxb-...`). +- `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN`: Optional legacy single-workspace Bot User OAuth Token (`xoxb-...`). When Slack OAuth is configured, per-workspace tokens are resolved from the installation registry instead. - `OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET`: Signing Secret (from App Credentials). +- `OPENCLAW_CONNECTOR_SLACK_CLIENT_ID`: OAuth client ID for multi-workspace installation flow. +- `OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET`: OAuth client secret for multi-workspace installation flow. +- `OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI`: Explicit OAuth callback URL. If omitted, connector derives it from `OPENCLAW_CONNECTOR_PUBLIC_BASE_URL` + callback path. +- `OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH`: Local install route (default `/slack/install`). +- `OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH`: Local callback route (default `/slack/oauth/callback`). +- `OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES`: Comma-separated bot scopes used for install URL generation. +- `OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC`: TTL in seconds for single-use OAuth state tokens (default `600`). - `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS`: Comma-separated user IDs (e.g. `U12345, U67890`). - `OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS`: Comma-separated channel IDs (e.g. `C12345`). - `OPENCLAW_CONNECTOR_SLACK_BIND`: Host to bind (default `127.0.0.1`). @@ -373,14 +387,18 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint - `im:history` (DM support) - `channels:history` (public channel messages) - `groups:history` (private channel messages) - - Click **Install to Workspace**. - - Copy the **Bot User OAuth Token** (`xoxb-...`). + - For legacy single-workspace mode, click **Install to Workspace** and copy the **Bot User OAuth Token** (`xoxb-...`). + - For F58 multi-workspace mode, configure a redirect URL and let OpenClaw handle installs through its OAuth routes. 3. **Configure connector environment variables** ```bash - OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET=your-signing-secret + OPENCLAW_CONNECTOR_SLACK_CLIENT_ID=1234567890.1234567890 + OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET=replace-with-client-secret + OPENCLAW_CONNECTOR_PUBLIC_BASE_URL=https://your-public-host + OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH=/slack/install + OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH=/slack/oauth/callback OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS=U12345,U67890 OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS=C12345 OPENCLAW_CONNECTOR_SLACK_BIND=127.0.0.1 @@ -392,6 +410,7 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint ``` Notes: + - Legacy single-workspace fallback can still set `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-...`; F58 multi-workspace mode no longer requires that token at startup if OAuth install flow is configured. - `OPENCLAW_CONNECTOR_ADMIN_TOKEN` must match server `OPENCLAW_ADMIN_TOKEN` if server-side admin token is enabled. - Slack ingress is fail-closed: invalid/missing signature, stale timestamp, and replayed events are rejected. @@ -400,6 +419,8 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint - Expose local endpoint to public HTTPS (Cloudflare Tunnel/ngrok/reverse proxy): - local upstream: `http://127.0.0.1:8095` - public URL: `https:///slack/events` + - install URL: `https:///slack/install` + - callback URL: `https:///slack/oauth/callback` 5. **Enable Event Subscriptions** - Go to **Event Subscriptions** and enable events. @@ -412,15 +433,17 @@ Slack uses the Events API webhook mode in OpenClaw. You must expose the endpoint - `message.im` 6. **Invite and validate** + - Open `https:///slack/install` and complete the workspace install. - Invite the app to target channels: `/invite @YourBot`. - In channel: `@YourBot /status` (when `OPENCLAW_CONNECTOR_SLACK_REQUIRE_MENTION=true`). - In DM: `/help`. - Verify connector logs show signed ingress accepted and replies delivered. + - Verify `GET /openclaw/connector/installations` shows the Slack workspace binding and health state `ok`. 7. **Security checklist before production** - Keep `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS`/`OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS` restricted. - Keep `OPENCLAW_CONNECTOR_SLACK_REQUIRE_MENTION=true` unless intentionally running command-style channels. - - Rotate Slack bot token/signing secret on incident response. + - Rotate Slack signing secret and OAuth client secret on incident response. - Do not expose connector without HTTPS termination. #### Slack Socket Mode Setup (Optional) @@ -437,9 +460,13 @@ Use Socket Mode when you cannot expose a public HTTPS webhook endpoint. ```bash OPENCLAW_CONNECTOR_SLACK_MODE=socket OPENCLAW_CONNECTOR_SLACK_APP_TOKEN=xapp-your-token - # Bot token + signing secret are still required for parity and safety checks - OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token + # Signing secret remains required for parity/security checks. OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET=your-signing-secret + # Either configure legacy single-workspace token... + OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN=xoxb-your-token + # ...or configure multi-workspace OAuth install flow: + OPENCLAW_CONNECTOR_SLACK_CLIENT_ID=1234567890.1234567890 + OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET=replace-with-client-secret ``` 3. **Start connector** @@ -449,6 +476,7 @@ Use Socket Mode when you cannot expose a public HTTPS webhook endpoint. Notes: - Socket Mode uses outbound WebSocket, so `OPENCLAW_CONNECTOR_SLACK_BIND`, `OPENCLAW_CONNECTOR_SLACK_PORT`, and `OPENCLAW_CONNECTOR_SLACK_PATH` are ignored in this mode. - Startup is fail-closed if `OPENCLAW_CONNECTOR_SLACK_APP_TOKEN` is missing or does not start with `xapp-`. +- In multi-workspace mode, outbound replies still resolve the workspace-specific bot token from the installation registry even though the WebSocket connection itself uses the app-level token. ## Commands diff --git a/docs/release/api_contract.md b/docs/release/api_contract.md index e469b33..c389a6e 100644 --- a/docs/release/api_contract.md +++ b/docs/release/api_contract.md @@ -99,6 +99,11 @@ Assist payload redaction contract: | `GET` | `/connector/installations/resolve` | `/moltbot/connector/installations/resolve` | Admin | Run fail-closed workspace resolution diagnostics (`platform`, `workspace_id`). | | `GET` | `/connector/installations/audit` | `/moltbot/connector/installations/audit` | Admin | List installation lifecycle audit evidence (redacted). | +Connector diagnostics contract notes: +- installation records may expose operator-safe health metadata under `installation.metadata.health` (for example `ok`, `invalid_token`, `revoked`, `degraded`) without exposing token material +- `/connector/installations` diagnostics may include aggregate `health_counts` in addition to lifecycle `status_counts` +- `/connector/installations/resolve` may expose a stable `health_code` alongside the legacy `reject_reason` so clients can distinguish `workspace_unbound` vs token-health failures without parsing status text + ### 1.3C LLM Management & Chat **LLM Base Path**: `/openclaw/llm/` diff --git a/docs/release/config_secrets_contract.md b/docs/release/config_secrets_contract.md index 6b6f96c..c7368eb 100644 --- a/docs/release/config_secrets_contract.md +++ b/docs/release/config_secrets_contract.md @@ -131,12 +131,20 @@ Controls the `connector` sidecar process and outbound delivery. | `OPENCLAW_CONNECTOR_SLACK_BOT_TOKEN` | Slack | Bot OAuth token (`xoxb-*`). | | `OPENCLAW_CONNECTOR_SLACK_SIGNING_SECRET` | Slack | Ingress signature secret. | | `OPENCLAW_CONNECTOR_SLACK_APP_TOKEN` | Slack | Optional Socket Mode app token (`xapp-*`). | +| `OPENCLAW_CONNECTOR_SLACK_CLIENT_ID` | Slack | OAuth client ID for multi-workspace installation flow. | +| `OPENCLAW_CONNECTOR_SLACK_CLIENT_SECRET` | Slack | OAuth client secret for multi-workspace installation flow. | +| `OPENCLAW_CONNECTOR_SLACK_OAUTH_REDIRECT_URI` | Slack | Explicit OAuth callback URL. Falls back to `OPENCLAW_CONNECTOR_PUBLIC_BASE_URL + OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH` when omitted. | +| `OPENCLAW_CONNECTOR_SLACK_OAUTH_INSTALL_PATH` | Slack | Local install route path (default `/slack/install`). | +| `OPENCLAW_CONNECTOR_SLACK_OAUTH_CALLBACK_PATH` | Slack | Local OAuth callback route path (default `/slack/oauth/callback`). | +| `OPENCLAW_CONNECTOR_SLACK_OAUTH_SCOPES` | Slack | Comma-separated bot scopes for install URL generation. | +| `OPENCLAW_CONNECTOR_SLACK_OAUTH_STATE_TTL_SEC` | Slack | TTL for single-use OAuth state tokens (default `600`). | | `OPENCLAW_CONNECTOR_SLACK_ALLOWED_USERS` | Slack | Comma-separated trusted user IDs. | | `OPENCLAW_CONNECTOR_SLACK_ALLOWED_CHANNELS` | Slack | Comma-separated trusted channel IDs. | Connector posture rules: - In strict posture (`OPENCLAW_DEPLOYMENT_PROFILE=public` or `OPENCLAW_RUNTIME_PROFILE=hardened`), active connector platforms without allowlist coverage are fail-closed. - Public deployment profile check surfaces this as `DP-PUBLIC-009`. +- Slack multi-workspace installs persist only encrypted token refs in `connector_installations.json`; raw bot/app tokens remain in encrypted secret storage and must not appear in diagnostics or exported config surfaces. **Delivery & Media:** diff --git a/services/connector_installation_registry.py b/services/connector_installation_registry.py index aeaf344..22303aa 100644 --- a/services/connector_installation_registry.py +++ b/services/connector_installation_registry.py @@ -49,6 +49,7 @@ class InstallationStatus(str, Enum): _RESOLVABLE_STATUSES = frozenset( {InstallationStatus.ACTIVE.value, InstallationStatus.ROTATING.value} ) +_UNRESOLVABLE_HEALTH_CODES = frozenset({"invalid_token", "revoked"}) @dataclass @@ -109,12 +110,14 @@ class InstallationResolution: installation: Optional[ConnectorInstallation] = None reject_reason: str = "" audit_code: str = "" + health_code: str = "" def to_public_dict(self) -> Dict[str, Any]: payload: Dict[str, Any] = { "ok": self.ok, "reject_reason": self.reject_reason, "audit_code": self.audit_code, + "health_code": self.health_code, } if self.installation is not None: payload["installation"] = self.installation.to_public_dict() @@ -147,6 +150,15 @@ class ConnectorInstallationRegistry: def _normalize_platform(self, platform: str) -> str: return self._normalize_identifier(platform, "platform").lower() + def _health_code_for_installation(self, inst: ConnectorInstallation) -> str: + health = dict(inst.metadata.get("health", {}) or {}) + health_code = str(health.get("state", "")).strip().lower() + if health_code: + return health_code + if inst.status == InstallationStatus.REVOKED.value: + return "revoked" + return "" + def _store_token_refs( self, installation_id: str, token_values: Dict[str, str], tenant_id: str ) -> Dict[str, str]: @@ -369,12 +381,18 @@ class ConnectorInstallationRegistry: def activate_installation( self, installation_id: str, reason: str = "" ) -> ConnectorInstallation: - return self._transition( + inst = self._transition( installation_id, InstallationStatus.ACTIVE.value, reason=reason, action="activate", ) + return self.update_installation_health( + installation_id, + health_code="ok", + reason=reason or "activated", + details={"status": inst.status}, + ) def rotate_installation_tokens( self, @@ -410,12 +428,18 @@ class ConnectorInstallationRegistry: def revoke_installation( self, installation_id: str, reason: str = "" ) -> ConnectorInstallation: - return self._transition( + self._transition( installation_id, InstallationStatus.REVOKED.value, reason=reason, action="revoke", ) + return self.update_installation_health( + installation_id, + health_code="revoked", + reason=reason or "revoked", + details={"status": InstallationStatus.REVOKED.value}, + ) def deactivate_installation( self, installation_id: str, reason: str = "" @@ -444,6 +468,37 @@ class ConnectorInstallationRegistry: self._save() return ConnectorInstallation(**asdict(inst)) + def update_installation_health( + self, + installation_id: str, + *, + health_code: str, + reason: str = "", + details: Optional[Dict[str, Any]] = None, + ) -> ConnectorInstallation: + with self._lock: + inst = self._installations.get(str(installation_id).strip()) + if inst is None: + raise ValueError(f"Installation not found: {installation_id}") + metadata = dict(inst.metadata or {}) + metadata["health"] = { + "state": str(health_code or "ok").strip().lower() or "ok", + "reason": str(reason or "").strip(), + "updated_at": time.time(), + "details": dict(details or {}), + } + inst.metadata = metadata + inst.updated_at = time.time() + self._installations[inst.installation_id] = inst + self._audit( + "health_update", + inst, + health_code=metadata["health"]["state"], + reason=metadata["health"]["reason"], + ) + self._save() + return ConnectorInstallation(**asdict(inst)) + def resolve_installation( self, platform: str, workspace_id: str, tenant_id: Optional[str] = None ) -> InstallationResolution: @@ -470,6 +525,7 @@ class ConnectorInstallationRegistry: ok=False, reject_reason="tenant_mismatch", audit_code="conn_install.resolve_tenant_mismatch", + health_code="degraded", ) matches = tenant_matches eligible = [inst for inst in matches if inst.status in _RESOLVABLE_STATUSES] @@ -478,6 +534,7 @@ class ConnectorInstallationRegistry: ok=False, reject_reason="ambiguous_binding", audit_code="conn_install.resolve_ambiguous", + health_code="degraded", ) if not eligible: if not matches: @@ -485,24 +542,39 @@ class ConnectorInstallationRegistry: ok=False, reject_reason="missing_binding", audit_code="conn_install.resolve_missing", + health_code="workspace_unbound", ) + health_code = ( + self._health_code_for_installation(matches[0]) or "degraded" + ) return InstallationResolution( ok=False, reject_reason="inactive_binding", audit_code="conn_install.resolve_inactive", + health_code=health_code, ) inst = eligible[0] + health_code = self._health_code_for_installation(inst) + if health_code in _UNRESOLVABLE_HEALTH_CODES: + return InstallationResolution( + ok=False, + reject_reason="inactive_binding", + audit_code="conn_install.resolve_unhealthy", + health_code=health_code, + ) for token_name, ref in inst.token_refs.items(): if not self._secret_store.get_secret(ref, tenant_id=inst.tenant_id): return InstallationResolution( ok=False, reject_reason=f"stale_token_ref:{token_name}", audit_code="conn_install.resolve_stale_ref", + health_code="degraded", ) return InstallationResolution( ok=True, installation=ConnectorInstallation(**asdict(inst)), audit_code="conn_install.resolve_ok", + health_code=health_code or "ok", ) def get_audit_trail( @@ -530,15 +602,19 @@ class ConnectorInstallationRegistry: def diagnostics(self, tenant_id: Optional[str] = None) -> Dict[str, Any]: with self._lock: counts: Dict[str, int] = {} + health_counts: Dict[str, int] = {} items = list(self._installations.values()) if tenant_id: normalized_tenant = normalize_tenant_id(tenant_id) items = [inst for inst in items if inst.tenant_id == normalized_tenant] for inst in items: counts[inst.status] = counts.get(inst.status, 0) + 1 + health_code = self._health_code_for_installation(inst) or "ok" + health_counts[health_code] = health_counts.get(health_code, 0) + 1 return { "installation_count": len(items), "status_counts": counts, + "health_counts": health_counts, "audit_events": len(self._audit_trail), } diff --git a/tests/connector/test_poller.py b/tests/connector/test_poller.py index 5c93811..a820c6b 100644 --- a/tests/connector/test_poller.py +++ b/tests/connector/test_poller.py @@ -9,11 +9,16 @@ from connector.results_poller import ResultsPoller class MockPlatform(Platform): async def send_image( - self, channel_id, image_data, filename="image.png", caption=None + self, + channel_id, + image_data, + filename="image.png", + caption=None, + delivery_context=None, ): pass - async def send_message(self, channel_id, text): + async def send_message(self, channel_id, text, delivery_context=None): pass @@ -37,7 +42,7 @@ class TestResultsPoller(unittest.TestCase): self.poller.track_job("p-1", "test_plat", "c-1", "u-1") self.assertEqual(self.poller.queue.qsize(), 1) item = self.poller.queue.get_nowait() - self.assertEqual(item, ("p-1", "test_plat", "c-1", "u-1")) + self.assertEqual(item, ("p-1", "test_plat", "c-1", "u-1", {})) @patch("connector.results_poller.time") @patch("connector.results_poller.asyncio.sleep", new_callable=AsyncMock) @@ -62,12 +67,23 @@ class TestResultsPoller(unittest.TestCase): self.client.get_view.return_value = b"image_bytes" - asyncio.run(self.poller._poll_job("p-1", "test_plat", "c-1", "u-1")) + asyncio.run( + self.poller._poll_job( + "p-1", + "test_plat", + "c-1", + "u-1", + {"workspace_id": "T1", "thread_id": "123.456"}, + ) + ) self.assertEqual(self.client.get_history.call_count, 2) self.client.get_view.assert_called_with("f.png", "", "output") self.mock_platform.send_image.assert_called_with( - "c-1", b"image_bytes", filename="f.png" + "c-1", + b"image_bytes", + filename="f.png", + delivery_context={"workspace_id": "T1", "thread_id": "123.456"}, ) @patch("connector.results_poller.time") diff --git a/tests/test_connector_installation_registry.py b/tests/test_connector_installation_registry.py index 78d79de..244f96a 100644 --- a/tests/test_connector_installation_registry.py +++ b/tests/test_connector_installation_registry.py @@ -67,6 +67,7 @@ class TestConnectorInstallationRegistry(unittest.TestCase): res = self.registry.resolve_installation("slack", "missing") self.assertFalse(res.ok) self.assertEqual(res.reject_reason, "missing_binding") + self.assertEqual(res.health_code, "workspace_unbound") def test_resolution_duplicate_binding_fails_closed(self): self.registry.upsert_installation( @@ -113,6 +114,27 @@ class TestConnectorInstallationRegistry(unittest.TestCase): res = self.registry.resolve_installation("slack", "T3") self.assertFalse(res.ok) self.assertEqual(res.reject_reason, "inactive_binding") + self.assertEqual(res.health_code, "revoked") + + def test_invalid_token_health_fails_closed(self): + self.registry.upsert_installation( + platform="slack", + workspace_id="T4", + installation_id="inst-invalid", + token_values={"bot_token": "xoxb-invalid"}, + status=InstallationStatus.ACTIVE.value, + ) + self.registry.update_installation_health( + "inst-invalid", + health_code="invalid_token", + reason="provider_401", + details={"source": "slack_api"}, + ) + + res = self.registry.resolve_installation("slack", "T4") + self.assertFalse(res.ok) + self.assertEqual(res.reject_reason, "inactive_binding") + self.assertEqual(res.health_code, "invalid_token") def test_persistence_reload_and_redaction(self): self.registry.upsert_installation( @@ -131,11 +153,12 @@ class TestConnectorInstallationRegistry(unittest.TestCase): self.assertEqual(len(listed), 1) self.assertEqual(listed[0].installation_id, "inst-persist") - raw = open( + with open( os.path.join(self.state_dir, "connector_installations.json"), "r", encoding="utf-8", - ).read() + ) as fh: + raw = fh.read() self.assertNotIn("xoxb-secret", raw) def test_multi_tenant_resolve_mismatch_fail_closed(self): diff --git a/tests/test_f57_slack_transport_parity.py b/tests/test_f57_slack_transport_parity.py index 46c5d49..3d2e912 100644 --- a/tests/test_f57_slack_transport_parity.py +++ b/tests/test_f57_slack_transport_parity.py @@ -10,6 +10,7 @@ from unittest.mock import AsyncMock, MagicMock sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from connector.config import ConnectorConfig +from connector.contract import CommandResponse from connector.platforms.slack_socket_mode import SlackSocketModeClient @@ -21,7 +22,7 @@ class TestF57SlackTransportParity(unittest.IsolatedAsyncioTestCase): self.config.slack_signing_secret = "secret" self.router = MagicMock() - self.router.handle = AsyncMock() + self.router.handle = AsyncMock(return_value=CommandResponse(text="")) self.client = SlackSocketModeClient(self.config, self.router) async def test_socket_mode_routes_message(self): diff --git a/tests/test_f58_slack_oauth_installations.py b/tests/test_f58_slack_oauth_installations.py new file mode 100644 index 0000000..b0ff68c --- /dev/null +++ b/tests/test_f58_slack_oauth_installations.py @@ -0,0 +1,139 @@ +import os +import sys +import tempfile +import unittest +from unittest.mock import AsyncMock, MagicMock, patch + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) + +from connector.config import ConnectorConfig +from connector.contract import CommandResponse +from connector.platforms.slack_installation_manager import SlackInstallationManager +from connector.platforms.slack_webhook import SlackWebhookServer +from services.connector_installation_registry import ConnectorInstallationRegistry +from services.secret_store import SecretStore + + +class TestF58SlackOAuthInstallations(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self): + self.tmpdir = tempfile.TemporaryDirectory() + self.state_dir = self.tmpdir.name + self.secret_store = SecretStore(state_dir=self.state_dir) + self.registry = ConnectorInstallationRegistry( + state_dir=self.state_dir, + secret_store=self.secret_store, + ) + self.config = ConnectorConfig() + self.config.public_base_url = "https://connector.example.com" + self.config.slack_signing_secret = "signing-secret" + self.config.slack_client_id = "client-id" + self.config.slack_client_secret = "client-secret" + self.config.slack_bot_token = "xoxb-legacy" + self.manager = SlackInstallationManager( + self.config, + registry=self.registry, + secret_store=self.secret_store, + state_dir=self.state_dir, + ) + + async def asyncTearDown(self): + self.tmpdir.cleanup() + + def _oauth_payload(self, token: str, *, workspace_id: str = "T1") -> dict: + return { + "ok": True, + "app_id": "A123", + "access_token": token, + "scope": "chat:write,files:write", + "bot_user_id": "U_BOT_1", + "team": {"id": workspace_id, "name": "Workspace One"}, + "authed_user": {"id": "U_INSTALLER"}, + "token_type": "bot", + } + + async def test_oauth_state_single_use_and_workspace_binding(self): + state = self.manager.issue_install_state() + self.assertTrue(self.manager.consume_install_state(state)) + self.assertFalse(self.manager.consume_install_state(state)) + + inst = self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-first")) + self.assertEqual(inst.workspace_id, "T1") + self.assertEqual(inst.status, "active") + + rotated = self.manager.upsert_from_oauth_payload( + self._oauth_payload("xoxb-rotated") + ) + self.assertEqual(rotated.installation_id, inst.installation_id) + + resolution, tokens = self.manager.resolve_workspace_tokens("T1") + self.assertTrue(resolution.ok) + self.assertEqual(tokens["bot_token"], "xoxb-rotated") + + with open( + os.path.join(self.state_dir, "connector_installations.json"), + "r", + encoding="utf-8", + ) as fh: + raw = fh.read() + self.assertNotIn("xoxb-rotated", raw) + + async def test_workspace_bound_reply_uses_installation_token(self): + self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-workspace")) + + router = MagicMock() + router.handle = AsyncMock(return_value=CommandResponse(text="Done")) + server = SlackWebhookServer(self.config, router) + server._installation_manager = self.manager + + with patch("aiohttp.ClientSession") as mock_session_cls: + mock_session = mock_session_cls.return_value + mock_session.__aenter__.return_value = mock_session + mock_session.post.return_value.__aenter__.return_value.status = 200 + mock_session.post.return_value.__aenter__.return_value.json = AsyncMock( + return_value={"ok": True} + ) + mock_session.post.return_value.__aenter__.return_value.text = AsyncMock( + return_value="OK" + ) + + payload = { + "type": "event_callback", + "team_id": "T1", + "authorizations": [{"user_id": "U_BOT_1", "team_id": "T1"}], + "event_id": "EvF58-1", + "event": { + "type": "message", + "text": "/status", + "user": "U_SENDER", + "channel": "D_DM", + "ts": "1700000000.001", + }, + } + + await server.process_event_payload(payload) + + routed = router.handle.call_args[0][0] + self.assertEqual(routed.workspace_id, "T1") + self.assertEqual(routed.thread_id, "1700000000.001") + headers = mock_session.post.call_args.kwargs["headers"] + self.assertEqual(headers["Authorization"], "Bearer xoxb-workspace") + + async def test_tokens_revoked_event_marks_installation_invalid(self): + self.manager.upsert_from_oauth_payload(self._oauth_payload("xoxb-workspace")) + router = MagicMock() + router.handle = AsyncMock(return_value=CommandResponse(text="Done")) + server = SlackWebhookServer(self.config, router) + server._installation_manager = self.manager + + payload = { + "type": "event_callback", + "team_id": "T1", + "event": { + "type": "tokens_revoked", + }, + } + await server.process_event_payload(payload) + + resolution = self.registry.resolve_installation("slack", "T1") + self.assertFalse(resolution.ok) + self.assertEqual(resolution.health_code, "invalid_token")