diff --git a/api/approvals.py b/api/approvals.py index b0ead11..2c5d05a 100644 --- a/api/approvals.py +++ b/api/approvals.py @@ -14,6 +14,7 @@ try: from ..services.approvals.service import get_approval_service from ..services.audit import emit_audit_event from ..services.management_query import bounded_scan_collect, normalize_limit_offset + from ..services.tenant_context import TenantBoundaryError, request_tenant_scope from ..services.webhook_auth import AuthError except ImportError: # Fallback for ComfyUI's non-package loader or ad-hoc imports. @@ -25,6 +26,10 @@ except ImportError: bounded_scan_collect, normalize_limit_offset, ) + from services.tenant_context import ( # type: ignore + TenantBoundaryError, + request_tenant_scope, + ) from services.webhook_auth import AuthError logger = logging.getLogger("ComfyUI-OpenClaw.api.approvals") @@ -129,36 +134,52 @@ class ApprovalHandlers: {"error": f"Invalid status: {status_filter}"}, status=400 ) - # Get approvals - # R95: bounded scan window protects API serialization path and keeps - # malformed-record behavior deterministic without swallowing service errors. - scan_cap = max(page.offset + page.limit + 200, page.limit * 10) - approvals = self._service.list_all( - status=status, - limit=min(scan_cap, 5000), - offset=0, - ) - page_result = bounded_scan_collect( - approvals, - skip=page.offset, - take=page.limit, - scan_cap=min(scan_cap, 5000), - serializer=lambda a: a.to_dict(), - ) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + # Get approvals + # R95: bounded scan window protects API serialization path and keeps + # malformed-record behavior deterministic without swallowing service errors. + scan_cap = max(page.offset + page.limit + 200, page.limit * 10) + approvals = self._service.list_all( + status=status, + limit=min(scan_cap, 5000), + offset=0, + tenant_id=tenant.tenant_id, + ) + page_result = bounded_scan_collect( + approvals, + skip=page.offset, + take=page.limit, + scan_cap=min(scan_cap, 5000), + serializer=lambda a: a.to_dict(), + ) - return web.json_response( - { - "approvals": page_result.items, - "count": len(page_result.items), - "pending_count": self._service.count_pending(), - "pagination": { - "limit": page.limit, - "offset": page.offset, - "warnings": page.warnings, - }, - "scan": page_result.to_dict(), - } - ) + return web.json_response( + { + "tenant_id": tenant.tenant_id, + "approvals": page_result.items, + "count": len(page_result.items), + "pending_count": self._service.count_pending( + tenant_id=tenant.tenant_id + ), + "pagination": { + "limit": page.limit, + "offset": page.offset, + "warnings": page.warnings, + }, + "scan": page_result.to_dict(), + } + ) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) async def get_approval(self, request: web.Request) -> web.Response: """GET /moltbot/approvals/{approval_id} - Get a single approval.""" @@ -186,12 +207,28 @@ class ApprovalHandlers: return web.json_response({"error": "Unauthorized"}, status=403) approval_id = request.match_info.get("approval_id", "") - approval = self._service.get(approval_id) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + approval = self._service.get(approval_id, tenant_id=tenant.tenant_id) - if not approval: - return web.json_response({"error": "Approval not found"}, status=404) + if not approval: + return web.json_response( + {"error": "Approval not found"}, status=404 + ) - return web.json_response({"approval": approval.to_dict()}) + return web.json_response( + {"tenant_id": tenant.tenant_id, "approval": approval.to_dict()} + ) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) async def approve_request(self, request: web.Request) -> web.Response: """POST /moltbot/approvals/{approval_id}/approve - Approve and execute request.""" @@ -230,67 +267,88 @@ class ApprovalHandlers: except Exception: pass # No body is fine + token_info = resolve_token_info(request) try: - # First approve the request - approval = self._service.approve(approval_id, actor=actor) - logger.info(f"Approved request: {approval_id}") - - result = { - "approved": True, - "approval": approval.to_dict(), - } - - # Execute if requested and submit_fn is available - if auto_execute and self._submit_fn: - try: - from .triggers import execute_approved_trigger - - exec_result = await execute_approved_trigger( - approval_id=approval_id, - submit_fn=self._submit_fn, - ) - - result["executed"] = True - result["prompt_id"] = exec_result.get("prompt_id") - result["trace_id"] = exec_result.get("trace_id") - - if result.get("prompt_id"): - # NOTE: Persist executed_prompt_id so connector can deliver results - # after UI approvals. Do not remove without updating connector. - try: - self._service.record_execution( - approval_id, - prompt_id=result.get("prompt_id"), - trace_id=result.get("trace_id"), - actor=actor, - ) - except Exception as record_error: - logger.error( - "Failed to record approval execution metadata: " - f"{record_error}" - ) - - logger.info( - f"Executed approved trigger: {approval_id} -> {result.get('prompt_id')}" - ) - - except Exception as exec_error: - logger.error(f"Failed to execute approved trigger: {exec_error}") - result["executed"] = False - result["execution_error"] = str(exec_error) - else: - result["executed"] = False - - self._audit( + with request_tenant_scope( request=request, - action="approvals.approve", - target=approval_id, - outcome="allow", - status_code=200, - details={"executed": result.get("executed", False), "actor": actor}, - ) - return web.json_response(result) + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + # First approve the request + approval = self._service.approve( + approval_id, actor=actor, tenant_id=tenant.tenant_id + ) + logger.info(f"Approved request: {approval_id}") + result = { + "tenant_id": tenant.tenant_id, + "approved": True, + "approval": approval.to_dict(), + } + + # Execute if requested and submit_fn is available + if auto_execute and self._submit_fn: + try: + from .triggers import execute_approved_trigger + + exec_result = await execute_approved_trigger( + approval_id=approval_id, + submit_fn=self._submit_fn, + ) + + result["executed"] = True + result["prompt_id"] = exec_result.get("prompt_id") + result["trace_id"] = exec_result.get("trace_id") + + if result.get("prompt_id"): + # NOTE: Persist executed_prompt_id so connector can deliver results + # after UI approvals. Do not remove without updating connector. + try: + self._service.record_execution( + approval_id, + prompt_id=result.get("prompt_id"), + trace_id=result.get("trace_id"), + actor=actor, + tenant_id=tenant.tenant_id, + ) + except Exception as record_error: + logger.error( + "Failed to record approval execution metadata: " + f"{record_error}" + ) + + logger.info( + f"Executed approved trigger: {approval_id} -> {result.get('prompt_id')}" + ) + + except Exception as exec_error: + logger.error( + f"Failed to execute approved trigger: {exec_error}" + ) + result["executed"] = False + result["execution_error"] = str(exec_error) + else: + result["executed"] = False + + self._audit( + request=request, + action="approvals.approve", + target=approval_id, + outcome="allow", + status_code=200, + details={ + "tenant_id": tenant.tenant_id, + "executed": result.get("executed", False), + "actor": actor, + }, + ) + return web.json_response(result) + + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) except ValueError as e: self._audit( request=request, @@ -337,25 +395,39 @@ class ApprovalHandlers: except Exception: pass + token_info = resolve_token_info(request) try: - approval = self._service.reject(approval_id, actor=actor) - - logger.info(f"Rejected request: {approval_id}") - self._audit( + with request_tenant_scope( request=request, - action="approvals.reject", - target=approval_id, - outcome="allow", - status_code=200, - details={"actor": actor}, - ) - return web.json_response( - { - "rejected": True, - "approval": approval.to_dict(), - } - ) + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + approval = self._service.reject( + approval_id, actor=actor, tenant_id=tenant.tenant_id + ) + logger.info(f"Rejected request: {approval_id}") + self._audit( + request=request, + action="approvals.reject", + target=approval_id, + outcome="allow", + status_code=200, + details={"tenant_id": tenant.tenant_id, "actor": actor}, + ) + return web.json_response( + { + "tenant_id": tenant.tenant_id, + "rejected": True, + "approval": approval.to_dict(), + } + ) + + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) except ValueError as e: self._audit( request=request, diff --git a/api/config.py b/api/config.py index 3ff11b7..bc443d0 100644 --- a/api/config.py +++ b/api/config.py @@ -146,6 +146,15 @@ except Exception: "update_config", ), ) +( + TenantBoundaryError, + request_tenant_scope, +) = import_attrs_dual( + __package__, + "..services.tenant_context", + "services.tenant_context", + ("TenantBoundaryError", "request_tenant_scope"), +) ( CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN, payload_contains_runtime_guardrails, @@ -169,6 +178,13 @@ _MODEL_LIST_TTL_SEC = 600 # 10 minutes _MODEL_LIST_MAX_ENTRIES = 16 +def _build_model_cache_key(provider: str, base_url: str, tenant_id: str) -> tuple: + # S49: keep single-tenant key shape stable while isolating multi-tenant entries. + if str(tenant_id).strip().lower() in ("", "default"): + return (provider, base_url) + return (tenant_id, provider, base_url) + + def _cache_put(key: tuple, models: list) -> None: """Insert into bounded cache, evicting oldest if over cap.""" if key in _MODEL_LIST_CACHE: @@ -277,36 +293,46 @@ async def config_get_handler(request: web.Request) -> web.Response: {"ok": False, "error": "Rate limit exceeded"}, status=429 ) + token_info = resolve_token_info(request) try: - effective, sources = get_effective_config() - guardrails = get_runtime_guardrails() - if guardrails.get("status") != "ok": - token_info = resolve_token_info(request) - emit_audit_event( - action="runtime.guardrails", - target="runtime_guardrails", - outcome="warn", - token_info=token_info, - status_code=200, - details={ - "code": guardrails.get("code"), - "violations": guardrails.get("violations", []), - }, - request=request, - ) + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + effective, sources = get_effective_config(tenant_id=tenant.tenant_id) + guardrails = get_runtime_guardrails() + if guardrails.get("status") != "ok": + emit_audit_event( + action="runtime.guardrails", + target="runtime_guardrails", + outcome="warn", + token_info=token_info, + status_code=200, + details={ + "tenant_id": tenant.tenant_id, + "code": guardrails.get("code"), + "violations": guardrails.get("violations", []), + }, + request=request, + ) + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "config": effective, + "sources": sources, + "runtime_guardrails": guardrails, + "providers": PROVIDER_CATALOG, + # R70: Settings schema for frontend type coercion / validation + "schema": get_settings_schema(), + # Simplified UX: writes are controlled by admin access policy, not a separate env "enable" flag. + "write_enabled": True, + } + ) + except TenantBoundaryError as e: return web.json_response( - { - "ok": True, - "config": effective, - "sources": sources, - "runtime_guardrails": guardrails, - "providers": PROVIDER_CATALOG, - # R70: Settings schema for frontend type coercion / validation - "schema": get_settings_schema(), - # Simplified UX: writes are controlled by admin access policy, not a separate env "enable" flag. - "write_enabled": True, - } + {"ok": False, "error": e.code, "message": str(e)}, + status=403, ) except Exception as e: logger.exception("Error getting config") @@ -408,224 +434,267 @@ async def llm_models_handler(request: web.Request) -> web.Response: {"ok": False, "error": "Rate limit exceeded"}, status=429 ) - # Admin boundary - allowed, err = require_admin_token(request) - if not allowed: - emit_audit_event( - action="config.update", - target="config.json", - outcome="deny", - token_info=token_info, - status_code=403, - details={"reason": err or "unauthorized"}, - request=request, - ) - return web.json_response( - { - "ok": False, - "error": err or "Unauthorized", - }, - status=403, - ) - - # Optional loopback check (match config_put behavior) - import os - - allow_remote = ( - os.environ.get("OPENCLAW_ALLOW_REMOTE_ADMIN") - or os.environ.get("MOLTBOT_ALLOW_REMOTE_ADMIN") - or "" - ).lower() - if allow_remote not in ("1", "true", "yes", "on"): - remote = request.remote or "" - if not is_loopback_client(remote): - return web.json_response( - { - "ok": False, - "error": "Remote admin access denied. Set OPENCLAW_ALLOW_REMOTE_ADMIN=1 (or legacy MOLTBOT_ALLOW_REMOTE_ADMIN=1) to allow.", - }, - status=403, - ) - - provider_override = (request.query.get("provider") or "").strip().lower() - effective, _sources = get_effective_config() - provider = provider_override or (effective.get("provider") or "openai") - - # Resolve Base URL (Runtime config > Catalog Default) - # Allows users to override base_url for standard providers (e.g. self-hosted OpenAI compat) - runtime_base_url = (effective.get("base_url") or "").strip() - + token_info = resolve_token_info(request) try: - from ..services.providers.catalog import ProviderType, get_provider_info - from ..services.providers.keys import get_api_key_for_provider, requires_api_key - except ImportError: - from services.providers.catalog import ProviderType, get_provider_info - from services.providers.keys import get_api_key_for_provider, requires_api_key + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + # Admin boundary + allowed, err = require_admin_token(request) + if not allowed: + emit_audit_event( + action="config.update", + target="config.json", + outcome="deny", + token_info=token_info, + status_code=403, + details={ + "tenant_id": tenant.tenant_id, + "reason": err or "unauthorized", + }, + request=request, + ) + return web.json_response( + { + "ok": False, + "error": err or "Unauthorized", + }, + status=403, + ) - info = get_provider_info(provider) - if not info: - return web.json_response( - {"ok": False, "error": f"Unknown provider: {provider}"}, status=400 - ) + # Optional loopback check (match config_put behavior) + import os - if info.api_type != ProviderType.OPENAI_COMPAT: - return web.json_response( - { - "ok": False, - "error": "Model list is only supported for OpenAI-compatible providers.", - }, - status=400, - ) + allow_remote = ( + os.environ.get("OPENCLAW_ALLOW_REMOTE_ADMIN") + or os.environ.get("MOLTBOT_ALLOW_REMOTE_ADMIN") + or "" + ).lower() + if allow_remote not in ("1", "true", "yes", "on"): + remote = request.remote or "" + if not is_loopback_client(remote): + return web.json_response( + { + "ok": False, + "error": "Remote admin access denied. Set OPENCLAW_ALLOW_REMOTE_ADMIN=1 (or legacy MOLTBOT_ALLOW_REMOTE_ADMIN=1) to allow.", + }, + status=403, + ) - # Priority: Runtime URL -> Info Default - base_url = runtime_base_url if runtime_base_url else info.base_url - if not base_url: - return web.json_response( - { - "ok": False, - "error": f"No base URL configured for provider '{provider}'.", - }, - status=400, - ) + provider_override = (request.query.get("provider") or "").strip().lower() + effective, _sources = get_effective_config(tenant_id=tenant.tenant_id) + provider = provider_override or (effective.get("provider") or "openai") - # R60: Cache key includes provider + base_url to avoid cross-provider staleness. - cache_key = (provider, base_url) + # Resolve Base URL (Runtime config > Catalog Default) + # Allows users to override base_url for standard providers (e.g. self-hosted OpenAI compat) + runtime_base_url = (effective.get("base_url") or "").strip() - # R60: Check bounded TTL+LRU cache - cached_entry = _cache_get(cache_key) - if cached_entry: - _ts, models = cached_entry - if isinstance(models, list): - return web.json_response( - {"ok": True, "provider": provider, "models": models, "cached": True} - ) + try: + from ..services.providers.catalog import ProviderType, get_provider_info + from ..services.providers.keys import ( + get_api_key_for_provider, + requires_api_key, + ) + except ImportError: + from services.providers.catalog import ProviderType, get_provider_info + from services.providers.keys import ( + get_api_key_for_provider, + requires_api_key, + ) - api_key = get_api_key_for_provider(provider) - # CRITICAL: - # Local providers (e.g. ollama/lmstudio) intentionally work without API keys. - # Do not change this gate back to `if not api_key`, or local model-list loading - # will regress with false 400 errors. - if requires_api_key(provider) and not api_key: - return web.json_response( - {"ok": False, "error": f"No API key configured for provider '{provider}'."}, - status=400, - ) + info = get_provider_info(provider) + if not info: + return web.json_response( + {"ok": False, "error": f"Unknown provider: {provider}"}, status=400 + ) - # SSRF policy - try: - try: - from ..services.safe_io import ( - STANDARD_OUTBOUND_POLICY, - validate_outbound_url, - ) - except ImportError: - from services.safe_io import STANDARD_OUTBOUND_POLICY, validate_outbound_url + if info.api_type != ProviderType.OPENAI_COMPAT: + return web.json_response( + { + "ok": False, + "error": "Model list is only supported for OpenAI-compatible providers.", + }, + status=400, + ) - controls = get_llm_egress_controls(provider, base_url) - validate_outbound_url( - base_url, - allow_hosts=controls.get("allow_hosts"), - allow_any_public_host=bool(controls.get("allow_any_public_host")), - allow_loopback_hosts=controls.get("allow_loopback_hosts"), - policy=STANDARD_OUTBOUND_POLICY, - ) - except Exception as e: - return web.json_response( - {"ok": False, "error": f"SSRF policy blocked outbound URL: {e}"}, status=403 - ) + # Priority: Runtime URL -> Info Default + base_url = runtime_base_url if runtime_base_url else info.base_url + if not base_url: + return web.json_response( + { + "ok": False, + "error": f"No base URL configured for provider '{provider}'.", + }, + status=400, + ) - # Fetch /models - try: - try: - from ..services.safe_io import ( - STANDARD_OUTBOUND_POLICY, - SSRFError, - safe_request_json, - ) - except ImportError: - from services.safe_io import ( - STANDARD_OUTBOUND_POLICY, - SSRFError, - safe_request_json, - ) + # R60: Cache key includes provider + base_url to avoid cross-provider staleness. + cache_key = _build_model_cache_key(provider, base_url, tenant.tenant_id) - url = f"{base_url.rstrip('/')}/models" - request_headers = { - "User-Agent": f"ComfyUI-OpenClaw/{PACK_VERSION}", - "Accept": "application/json", - } - if api_key: - request_headers["Authorization"] = f"Bearer {api_key}" + # R60: Check bounded TTL+LRU cache + cached_entry = _cache_get(cache_key) + if cached_entry: + _ts, models = cached_entry + if isinstance(models, list): + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "provider": provider, + "models": models, + "cached": True, + } + ) - # S65: Enforce outbound policy via safe_io - payload = safe_request_json( - method="GET", - url=url, - json_body=None, - headers=request_headers, - timeout_sec=10, - policy=STANDARD_OUTBOUND_POLICY, - allow_hosts=controls.get("allow_hosts"), - allow_any_public_host=bool(controls.get("allow_any_public_host")), - allow_loopback_hosts=controls.get("allow_loopback_hosts"), - ) + api_key = get_api_key_for_provider(provider, tenant_id=tenant.tenant_id) + # CRITICAL: + # Local providers (e.g. ollama/lmstudio) intentionally work without API keys. + # Do not change this gate back to `if not api_key`, or local model-list loading + # will regress with false 400 errors. + if requires_api_key(provider) and not api_key: + return web.json_response( + { + "ok": False, + "error": f"No API key configured for provider '{provider}'.", + }, + status=400, + ) - models = _extract_models_from_payload(payload) + # SSRF policy + try: + try: + from ..services.safe_io import ( + STANDARD_OUTBOUND_POLICY, + validate_outbound_url, + ) + except ImportError: + from services.safe_io import ( # type: ignore + STANDARD_OUTBOUND_POLICY, + validate_outbound_url, + ) - # R60: Insert/update bounded cache - _cache_put(cache_key, models) + controls = get_llm_egress_controls(provider, base_url) + validate_outbound_url( + base_url, + allow_hosts=controls.get("allow_hosts"), + allow_any_public_host=bool(controls.get("allow_any_public_host")), + allow_loopback_hosts=controls.get("allow_loopback_hosts"), + policy=STANDARD_OUTBOUND_POLICY, + ) + except Exception as e: + return web.json_response( + {"ok": False, "error": f"SSRF policy blocked outbound URL: {e}"}, + status=403, + ) + + # Fetch /models + try: + try: + from ..services.safe_io import ( + STANDARD_OUTBOUND_POLICY, + SSRFError, + safe_request_json, + ) + except ImportError: + from services.safe_io import ( # type: ignore + STANDARD_OUTBOUND_POLICY, + SSRFError, + safe_request_json, + ) + + url = f"{base_url.rstrip('/')}/models" + request_headers = { + "User-Agent": f"ComfyUI-OpenClaw/{PACK_VERSION}", + "Accept": "application/json", + } + if api_key: + request_headers["Authorization"] = f"Bearer {api_key}" + + # S65: Enforce outbound policy via safe_io + payload = safe_request_json( + method="GET", + url=url, + json_body=None, + headers=request_headers, + timeout_sec=10, + policy=STANDARD_OUTBOUND_POLICY, + allow_hosts=controls.get("allow_hosts"), + allow_any_public_host=bool(controls.get("allow_any_public_host")), + allow_loopback_hosts=controls.get("allow_loopback_hosts"), + ) + + models = _extract_models_from_payload(payload) + + # R60: Insert/update bounded cache + _cache_put(cache_key, models) - return web.json_response( - {"ok": True, "provider": provider, "models": models, "cached": False} - ) - except SSRFError as e: - return web.json_response( - {"ok": False, "error": f"SSRF policy blocked outbound URL: {e}"}, status=403 - ) - except RuntimeError as e: - # safe_request_json raises RuntimeError for HTTP errors (non-200) contextually - # check if it looks like an HTTP error - str_e = str(e) - if "HTTP" in str_e: - # Fallback: serve stale cache entry (if any) on fetch failure - stale = _MODEL_LIST_CACHE.get(cache_key) - if stale: - _ts, models = stale - warning = f"Using cached list (refresh failed: {str_e})" return web.json_response( { "ok": True, + "tenant_id": tenant.tenant_id, "provider": provider, "models": models, - "cached": True, - "warning": warning, + "cached": False, } ) - return web.json_response( - {"ok": False, "error": f"Upstream error: {str_e}"}, status=502 - ) - raise e + except SSRFError as e: + return web.json_response( + {"ok": False, "error": f"SSRF policy blocked outbound URL: {e}"}, + status=403, + ) + except RuntimeError as e: + # safe_request_json raises RuntimeError for HTTP errors (non-200) contextually + # check if it looks like an HTTP error + str_e = str(e) + if "HTTP" in str_e: + # Fallback: serve stale cache entry (if any) on fetch failure + stale = _MODEL_LIST_CACHE.get(cache_key) + if stale: + _ts, models = stale + warning = f"Using cached list (refresh failed: {str_e})" + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "provider": provider, + "models": models, + "cached": True, + "warning": warning, + } + ) + return web.json_response( + {"ok": False, "error": f"Upstream error: {str_e}"}, status=502 + ) + raise e - except Exception as e: - stale = _MODEL_LIST_CACHE.get(cache_key) - if stale: - # IMPORTANT: - # Test path intentionally injects network failures to verify cache fallback. - # Keep this as warning (no traceback) to avoid noisy false-alarm logs. - logger.warning("Model list refresh failed, serving cached list: %s", e) - _ts, models = stale - warning = f"Using cached list (refresh failed: {str(e)})" - return web.json_response( - { - "ok": True, - "provider": provider, - "models": models, - "cached": True, - "warning": warning, - } - ) - logger.exception("Failed to fetch model list") - return web.json_response({"ok": False, "error": str(e)}, status=500) + except Exception as e: + stale = _MODEL_LIST_CACHE.get(cache_key) + if stale: + # IMPORTANT: + # Test path intentionally injects network failures to verify cache fallback. + # Keep this as warning (no traceback) to avoid noisy false-alarm logs. + logger.warning( + "Model list refresh failed, serving cached list: %s", e + ) + _ts, models = stale + warning = f"Using cached list (refresh failed: {str(e)})" + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "provider": provider, + "models": models, + "cached": True, + "warning": warning, + } + ) + logger.exception("Failed to fetch model list") + return web.json_response({"ok": False, "error": str(e)}, status=500) + except TenantBoundaryError as e: + return web.json_response( + {"ok": False, "error": e.code, "message": str(e)}, + status=403, + ) @endpoint_metadata( @@ -708,88 +777,112 @@ async def config_put_handler(request: web.Request) -> web.Response: ) try: - body = await request.json() - except json.JSONDecodeError: - return web.json_response( - { - "ok": False, - "error": "Invalid JSON body", - }, - status=400, - ) + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + try: + body = await request.json() + except json.JSONDecodeError: + return web.json_response( + { + "ok": False, + "error": "Invalid JSON body", + }, + status=400, + ) - # S66: Runtime guardrails are ENV-driven + runtime-only and must never be - # persisted via config writes (prevents config drift / silent downgrade paths). - if payload_contains_runtime_guardrails(body): + # S66: Runtime guardrails are ENV-driven + runtime-only and must never be + # persisted via config writes (prevents config drift / silent downgrade paths). + if payload_contains_runtime_guardrails(body): + emit_audit_event( + action="config.update", + target="config.json", + outcome="deny", + token_info=token_info, + status_code=400, + details={ + "tenant_id": tenant.tenant_id, + "reason": "runtime_guardrails_runtime_only", + "code": CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN, + }, + request=request, + ) + return web.json_response( + { + "ok": False, + "error": "runtime_guardrails are runtime-only (ENV-driven) and cannot be persisted via /config", + "code": CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN, + }, + status=400, + ) + + # Extract LLM config updates + updates = body.get("llm", body) # Support both { llm: {...} } and {...} + if not isinstance(updates, dict): + return web.json_response( + { + "ok": False, + "error": "Expected object with config fields", + }, + status=400, + ) + + success, errors = update_config(updates, tenant_id=tenant.tenant_id) + + # R99: Standardized Audit Emission + emit_audit_event( + action="config.update", + target="config.json", + outcome="allow" if success else "error", + token_info=token_info, + status_code=200 if success else 400, + details=( + {"tenant_id": tenant.tenant_id, "errors": errors} + if errors + else {"tenant_id": tenant.tenant_id} + ), + request=request, + ) + + if not success: + return web.json_response( + { + "ok": False, + "errors": errors, + }, + status=400, + ) + + # Return updated config + effective, sources = get_effective_config(tenant_id=tenant.tenant_id) + + # R53: Calculate apply semantics + apply_info = get_apply_semantics(list(updates.keys())) + + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "config": effective, + "sources": sources, + "apply": apply_info, + } + ) + except TenantBoundaryError as e: emit_audit_event( action="config.update", target="config.json", outcome="deny", token_info=token_info, - status_code=400, - details={ - "reason": "runtime_guardrails_runtime_only", - "code": CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN, - }, + status_code=403, + details={"reason": e.code}, request=request, ) return web.json_response( - { - "ok": False, - "error": "runtime_guardrails are runtime-only (ENV-driven) and cannot be persisted via /config", - "code": CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN, - }, - status=400, + {"ok": False, "error": e.code, "message": str(e)}, + status=403, ) - # Extract LLM config updates - updates = body.get("llm", body) # Support both { llm: {...} } and {...} - if not isinstance(updates, dict): - return web.json_response( - { - "ok": False, - "error": "Expected object with config fields", - }, - status=400, - ) - - success, errors = update_config(updates) - - # R99: Standardized Audit Emission - emit_audit_event( - action="config.update", - target="config.json", - outcome="allow" if success else "error", - token_info=token_info, - status_code=200 if success else 400, - details={"errors": errors} if errors else None, - request=request, - ) - - if not success: - return web.json_response( - { - "ok": False, - "errors": errors, - }, - status=400, - ) - - # Return updated config - effective, sources = get_effective_config() - - # R53: Calculate apply semantics - apply_info = get_apply_semantics(list(updates.keys())) - - return web.json_response( - { - "ok": True, - "config": effective, - "sources": sources, - "apply": apply_info, - } - ) - @endpoint_metadata( auth=AuthTier.ADMIN, @@ -852,103 +945,111 @@ async def llm_test_handler(request: web.Request) -> web.Response: ) try: - # IMPORTANT (Settings UX / provider mismatch): - # - The Settings UI allows selecting provider/model/base_url without persisting config immediately. - # - If this endpoint only uses effective config, "Test Connection" can misleadingly test the - # previous provider (often "openai") and report: "API key not configured for provider 'openai'" - # even when the UI is set to Gemini and a Gemini key is stored. - # Therefore, accept optional overrides in the JSON body. - # - # Contract: - # - Empty body -> test effective config - # - Body may include: provider, model, base_url, timeout_sec, max_retries - try: - body = await request.json() - if body is None: + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + # IMPORTANT (Settings UX / provider mismatch): + # - The Settings UI allows selecting provider/model/base_url without persisting config immediately. + # - If this endpoint only uses effective config, "Test Connection" can misleadingly test the + # previous provider (often "openai") and report: "API key not configured for provider 'openai'" + # even when the UI is set to Gemini and a Gemini key is stored. + # Therefore, accept optional overrides in the JSON body. + # + # Contract: + # - Empty body -> test effective config + # - Body may include: provider, model, base_url, timeout_sec, max_retries + try: + body = await request.json() + if body is None: + body = {} + except Exception: body = {} - except Exception: - body = {} - if body and not isinstance(body, dict): - return web.json_response( - {"ok": False, "error": "Expected JSON object body (or empty body)"}, - status=400, - ) - - provider = ( - body.get("provider") if isinstance(body.get("provider"), str) else None - ) - model = body.get("model") if isinstance(body.get("model"), str) else None - base_url = ( - body.get("base_url") if isinstance(body.get("base_url"), str) else None - ) - - timeout_val = body.get("timeout_sec") - timeout_sec = None - if ( - isinstance(timeout_val, (int, float, str)) - and str(timeout_val).strip() != "" - ): - try: - timeout_sec = int(timeout_val) - except Exception: + if body and not isinstance(body, dict): return web.json_response( - {"ok": False, "error": "timeout_sec must be an integer"}, + {"ok": False, "error": "Expected JSON object body (or empty body)"}, status=400, ) - retries_val = body.get("max_retries") - max_retries = None - if ( - isinstance(retries_val, (int, float, str)) - and str(retries_val).strip() != "" - ): - try: - max_retries = int(retries_val) - except Exception: + provider = ( + body.get("provider") if isinstance(body.get("provider"), str) else None + ) + model = body.get("model") if isinstance(body.get("model"), str) else None + base_url = ( + body.get("base_url") if isinstance(body.get("base_url"), str) else None + ) + + timeout_val = body.get("timeout_sec") + timeout_sec = None + if ( + isinstance(timeout_val, (int, float, str)) + and str(timeout_val).strip() != "" + ): + try: + timeout_sec = int(timeout_val) + except Exception: + return web.json_response( + {"ok": False, "error": "timeout_sec must be an integer"}, + status=400, + ) + + retries_val = body.get("max_retries") + max_retries = None + if ( + isinstance(retries_val, (int, float, str)) + and str(retries_val).strip() != "" + ): + try: + max_retries = int(retries_val) + except Exception: + return web.json_response( + {"ok": False, "error": "max_retries must be an integer"}, + status=400, + ) + + # Initialize client (uses effective config by default; overrides if provided) + client = LLMClient( + provider=provider, + base_url=base_url, + model=model, + timeout=timeout_sec, + max_retries=max_retries, + ) + + # Run test in a worker thread since LLMClient is sync + result = await run_in_thread( + client.complete, + system="You are a test assistant.", + user_message="Respond with exactly: OK", + max_tokens=10, + ) + + # Check result + if result and "text" in result: + emit_audit_event( + action="llm.test_connection", + target=f"{client.provider}:{client.model}", + outcome="allow", + token_info=token_info, + status_code=200, + details={ + "tenant_id": tenant.tenant_id, + "provider": client.provider, + "model": client.model, + }, + request=request, + ) return web.json_response( - {"ok": False, "error": "max_retries must be an integer"}, - status=400, + { + "ok": True, + "tenant_id": tenant.tenant_id, + "message": "Connection successful", + "response": result["text"].strip(), + "provider": client.provider, + "model": client.model, + } ) - # Initialize client (uses effective config by default; overrides if provided) - client = LLMClient( - provider=provider, - base_url=base_url, - model=model, - timeout=timeout_sec, - max_retries=max_retries, - ) - - # Run test in a worker thread since LLMClient is sync - result = await run_in_thread( - client.complete, - system="You are a test assistant.", - user_message="Respond with exactly: OK", - max_tokens=10, - ) - - # Check result - if result and "text" in result: - emit_audit_event( - action="llm.test_connection", - target=f"{client.provider}:{client.model}", - outcome="allow", - token_info=token_info, - status_code=200, - details={"provider": client.provider, "model": client.model}, - request=request, - ) - return web.json_response( - { - "ok": True, - "message": "Connection successful", - "response": result["text"].strip(), - "provider": client.provider, - "model": client.model, - } - ) - else: emit_audit_event( action="llm.test_connection", target=f"{client.provider}:{client.model}", @@ -956,6 +1057,7 @@ async def llm_test_handler(request: web.Request) -> web.Response: token_info=token_info, status_code=500, details={ + "tenant_id": tenant.tenant_id, "provider": client.provider, "model": client.model, "error": "Empty response", @@ -968,6 +1070,20 @@ async def llm_test_handler(request: web.Request) -> web.Response: "error": "Empty or invalid response from LLM", } ) + except TenantBoundaryError as e: + emit_audit_event( + action="llm.test_connection", + target="llm", + outcome="deny", + token_info=token_info, + status_code=403, + details={"reason": e.code}, + request=request, + ) + return web.json_response( + {"ok": False, "error": e.code, "message": str(e)}, + status=403, + ) except Exception as e: logger.exception("LLM test failed") emit_audit_event( @@ -1030,6 +1146,7 @@ async def llm_chat_handler(request: web.Request) -> web.Response: # NOTE: Keep this server-side. Connector cannot access UI-stored secrets directly. # This endpoint ensures keys are resolved via backend config + secret store. # S13: Validate admin boundary (or loopback if no admin token configured) + token_info = resolve_token_info(request) allowed, err = require_admin_token(request) if not allowed: return web.json_response( @@ -1082,21 +1199,31 @@ async def llm_chat_handler(request: web.Request) -> web.Response: ) try: - client = LLMClient() + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + client = LLMClient() - def _run(): - return client.complete( - system=system, - user_message=user_message, - temperature=temperature, - max_tokens=max_tokens, + def _run(): + return client.complete( + system=system, + user_message=user_message, + temperature=temperature, + max_tokens=max_tokens, + ) + + result = await run_in_thread(_run) + text = "" + if isinstance(result, dict): + text = result.get("text") or "" + return web.json_response( + {"ok": True, "tenant_id": tenant.tenant_id, "text": text} ) - - result = await run_in_thread(_run) - text = "" - if isinstance(result, dict): - text = result.get("text") or "" - return web.json_response({"ok": True, "text": text}) + except TenantBoundaryError as e: + return web.json_response( + {"ok": False, "error": e.code, "message": str(e)}, + status=403, + ) except ValueError as e: # Common: missing API key for selected provider return web.json_response( diff --git a/api/connector_contracts.py b/api/connector_contracts.py index f934a25..a0b4481 100644 --- a/api/connector_contracts.py +++ b/api/connector_contracts.py @@ -24,17 +24,23 @@ except ImportError: # pragma: no cover web = _MockWeb() # type: ignore if __package__ and "." in __package__: - from ..services.access_control import require_admin_token + from ..services.access_control import require_admin_token, resolve_token_info from ..services.connector_installation_registry import ( get_connector_installation_registry, ) from ..services.rate_limit import check_rate_limit + from ..services.tenant_context import TenantBoundaryError, request_tenant_scope else: # pragma: no cover from services.access_control import require_admin_token # type: ignore + from services.access_control import resolve_token_info # type: ignore from services.connector_installation_registry import ( # type: ignore get_connector_installation_registry, ) from services.rate_limit import check_rate_limit # type: ignore + from services.tenant_context import ( # type: ignore + TenantBoundaryError, + request_tenant_scope, + ) if __package__ and "." in __package__: from ..services.endpoint_manifest import ( @@ -82,16 +88,30 @@ async def connector_installations_list_handler(request): platform = request.query.get("platform") workspace_id = request.query.get("workspace_id") status = request.query.get("status") - installations = registry.list_installations( - platform=platform, workspace_id=workspace_id, status=status - ) - return web.json_response( - { - "ok": True, - "installations": [inst.to_public_dict() for inst in installations], - "diagnostics": registry.diagnostics(), - } - ) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + installations = registry.list_installations( + platform=platform, + tenant_id=tenant.tenant_id, + workspace_id=workspace_id, + status=status, + ) + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "installations": [inst.to_public_dict() for inst in installations], + "diagnostics": registry.diagnostics(tenant_id=tenant.tenant_id), + } + ) + except TenantBoundaryError as exc: + return web.json_response( + {"ok": False, "error": exc.code, "message": str(exc)}, + status=403, + ) @endpoint_metadata( @@ -107,12 +127,30 @@ async def connector_installation_get_handler(request): return guard installation_id = request.match_info.get("installation_id", "") registry = get_connector_installation_registry() - installation = registry.get_installation(installation_id) - if installation is None: - return web.json_response({"ok": False, "error": "not_found"}, status=404) - return web.json_response( - {"ok": True, "installation": installation.to_public_dict()} - ) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + installation = registry.get_installation( + installation_id, tenant_id=tenant.tenant_id + ) + if installation is None: + return web.json_response( + {"ok": False, "error": "not_found"}, status=404 + ) + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "installation": installation.to_public_dict(), + } + ) + except TenantBoundaryError as exc: + return web.json_response( + {"ok": False, "error": exc.code, "message": str(exc)}, + status=403, + ) @endpoint_metadata( @@ -134,12 +172,28 @@ async def connector_installation_resolve_handler(request): status=400, ) registry = get_connector_installation_registry() - resolution = registry.resolve_installation(platform, workspace_id) - status_code = 200 if resolution.ok else 409 - return web.json_response( - {"ok": resolution.ok, "resolution": resolution.to_public_dict()}, - status=status_code, - ) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + resolution = registry.resolve_installation( + platform, workspace_id, tenant_id=tenant.tenant_id + ) + status_code = 200 if resolution.ok else 409 + return web.json_response( + { + "ok": resolution.ok, + "tenant_id": tenant.tenant_id, + "resolution": resolution.to_public_dict(), + }, + status=status_code, + ) + except TenantBoundaryError as exc: + return web.json_response( + {"ok": False, "error": exc.code, "message": str(exc)}, + status=403, + ) @endpoint_metadata( @@ -159,11 +213,24 @@ async def connector_installation_audit_handler(request): limit = int(request.query.get("limit") or 100) except Exception: limit = 100 - return web.json_response( - { - "ok": True, - "events": registry.get_audit_trail( - installation_id=installation_id, limit=limit - ), - } - ) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ) as tenant: + return web.json_response( + { + "ok": True, + "tenant_id": tenant.tenant_id, + "events": registry.get_audit_trail( + installation_id=installation_id, + tenant_id=tenant.tenant_id, + limit=limit, + ), + } + ) + except TenantBoundaryError as exc: + return web.json_response( + {"ok": False, "error": exc.code, "message": str(exc)}, + status=403, + ) diff --git a/api/presets.py b/api/presets.py index 3e3377f..9864650 100644 --- a/api/presets.py +++ b/api/presets.py @@ -11,7 +11,7 @@ from typing import Optional from aiohttp import web try: - from ..services.access_control import require_admin_token + from ..services.access_control import require_admin_token, resolve_token_info from ..services.endpoint_manifest import ( AuthTier, RiskTier, @@ -19,9 +19,10 @@ try: endpoint_metadata, ) from ..services.presets import Preset, preset_store + from ..services.tenant_context import TenantBoundaryError, request_tenant_scope except ImportError: # Fallback for ComfyUI's non-package loader or ad-hoc imports. - from services.access_control import require_admin_token + from services.access_control import require_admin_token, resolve_token_info from services.endpoint_manifest import ( AuthTier, RiskTier, @@ -29,6 +30,7 @@ except ImportError: endpoint_metadata, ) from services.presets import Preset, preset_store + from services.tenant_context import TenantBoundaryError, request_tenant_scope logger = logging.getLogger("ComfyUI-OpenClaw.api.presets") @@ -73,9 +75,24 @@ class PresetHandlers: category = request.query.get("category") tag = request.query.get("tag") - - presets = preset_store.list_presets(category=category, tag=tag) - return web.json_response([p.to_dict() for p in presets]) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + presets = preset_store.list_presets( + category=category, + tag=tag, + tenant_id=tenant.tenant_id, + ) + return web.json_response([p.to_dict() for p in presets]) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) @endpoint_metadata( auth=AuthTier.PUBLIC, # Conditionally public @@ -108,11 +125,23 @@ class PresetHandlers: if not preset_id: return web.json_response({"error": "Missing ID"}, status=400) - preset = preset_store.get_preset(preset_id) - if not preset: - return web.json_response({"error": "Not Found"}, status=404) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + preset = preset_store.get_preset(preset_id, tenant_id=tenant.tenant_id) + if not preset: + return web.json_response({"error": "Not Found"}, status=404) - return web.json_response(preset.to_dict()) + return web.json_response(preset.to_dict()) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) @endpoint_metadata( auth=AuthTier.ADMIN, @@ -138,28 +167,40 @@ class PresetHandlers: if not name or not content: return web.json_response({"error": "Name and Content required"}, status=400) + token_info = resolve_token_info(request) try: - # Create object - preset = Preset.new( - name=data["name"], - content=data["content"], - category=data.get("category", "general"), - tags=data.get("tags", []), - ) - - # Milestone E: Schema Validation - try: - preset.validate_content() - except ValueError as e: - return web.json_response( - {"error": f"Validation Error: {str(e)}"}, status=400 + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + # Create object + preset = Preset.new( + name=data["name"], + content=data["content"], + category=data.get("category", "general"), + tags=data.get("tags", []), ) + preset.tenant_id = tenant.tenant_id - # Save - preset_store.save_preset(preset) - logger.info(f"Created preset {preset.id} ({preset.name})") + # Milestone E: Schema Validation + try: + preset.validate_content() + except ValueError as e: + return web.json_response( + {"error": f"Validation Error: {str(e)}"}, status=400 + ) - return web.json_response(preset.to_dict(), status=201) + # Save + preset_store.save_preset(preset) + logger.info(f"Created preset {preset.id} ({preset.name})") + + return web.json_response(preset.to_dict(), status=201) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) except Exception as e: logger.error(f"Failed to create preset: {e}") return web.json_response({"error": str(e)}, status=500) @@ -182,38 +223,50 @@ class PresetHandlers: if not preset_id: return web.json_response({"error": "Missing ID"}, status=400) - preset = preset_store.get_preset(preset_id) - if not preset: - return web.json_response({"error": "Not Found"}, status=404) - + token_info = resolve_token_info(request) try: - data = await request.json() - except Exception: - return web.json_response({"error": "Invalid JSON"}, status=400) + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + preset = preset_store.get_preset(preset_id, tenant_id=tenant.tenant_id) + if not preset: + return web.json_response({"error": "Not Found"}, status=404) - # Update fields - if "name" in data: - preset.name = data["name"] - if "content" in data: - preset.content = data["content"] - if "category" in data: - preset.category = data["category"] - if "tags" in data: - preset.tags = data["tags"] + try: + data = await request.json() + except Exception: + return web.json_response({"error": "Invalid JSON"}, status=400) - # Milestone E: Schema Validation - try: - preset.validate_content() - except ValueError as e: + # Update fields + if "name" in data: + preset.name = data["name"] + if "content" in data: + preset.content = data["content"] + if "category" in data: + preset.category = data["category"] + if "tags" in data: + preset.tags = data["tags"] + + # Milestone E: Schema Validation + try: + preset.validate_content() + except ValueError as e: + return web.json_response( + {"error": f"Validation Error: {str(e)}"}, status=400 + ) + + preset.updated_at = time.time() + preset_store.save_preset(preset) + + return web.json_response(preset.to_dict()) + except TenantBoundaryError as exc: return web.json_response( - {"error": f"Validation Error: {str(e)}"}, status=400 + {"error": exc.code, "message": str(exc)}, + status=403, ) - preset.updated_at = time.time() - preset_store.save_preset(preset) - - return web.json_response(preset.to_dict()) - @endpoint_metadata( auth=AuthTier.ADMIN, risk=RiskTier.HIGH, @@ -232,10 +285,21 @@ class PresetHandlers: if not preset_id: return web.json_response({"error": "Missing ID"}, status=400) - if preset_store.delete_preset(preset_id): - return web.json_response({"ok": True}) - else: - return web.json_response({"error": "Not Found or Failed"}, status=404) + token_info = resolve_token_info(request) + try: + with request_tenant_scope( + request=request, + token_info=token_info, + allow_default_when_missing=True, + ) as tenant: + if preset_store.delete_preset(preset_id, tenant_id=tenant.tenant_id): + return web.json_response({"ok": True}) + return web.json_response({"error": "Not Found or Failed"}, status=404) + except TenantBoundaryError as exc: + return web.json_response( + {"error": exc.code, "message": str(exc)}, + status=403, + ) def register_preset_routes(app: web.Application): diff --git a/api/templates.py b/api/templates.py index 38ca128..793904b 100644 --- a/api/templates.py +++ b/api/templates.py @@ -18,13 +18,24 @@ except ImportError: # pragma: no cover (optional for unit tests) # - ComfyUI runtime: package-relative imports only (prevents collisions with other custom nodes). # - Unit tests: allow top-level fallbacks. if __package__ and "." in __package__: - from ..services.access_control import require_observability_access + from ..services.access_control import ( + require_observability_access, + resolve_token_info, + ) from ..services.rate_limit import check_rate_limit from ..services.templates import get_template_service + from ..services.tenant_context import TenantBoundaryError, request_tenant_scope else: # pragma: no cover (test-only import mode) - from services.access_control import require_observability_access # type: ignore + from services.access_control import ( # type: ignore + require_observability_access, + resolve_token_info, + ) from services.rate_limit import check_rate_limit # type: ignore from services.templates import get_template_service # type: ignore + from services.tenant_context import ( # type: ignore + TenantBoundaryError, + request_tenant_scope, + ) # R98: Endpoint Metadata if __package__ and "." in __package__: @@ -105,33 +116,42 @@ async def templates_list_handler(request: web.Request) -> web.Response: try: svc = get_template_service() - items = [] - # Prefer runtime discovery (file-based templates) so operators don't need - # to maintain a separate allowlist file. - for template_id in svc.get_debug_info().get("discovered_template_ids", []): # type: ignore[call-arg] - cfg = svc.get_template_config(template_id) # type: ignore[arg-type] - if cfg is None: - continue - items.append( - { - "id": template_id, - "allowed_inputs": list(cfg.allowed_inputs or []), - "defaults": dict(cfg.defaults or {}), - } - ) - items.sort(key=lambda x: x["id"]) - resp: dict = {"ok": True, "templates": items, "count": len(items)} + token_info = resolve_token_info(request) + with request_tenant_scope( + request=request, token_info=token_info, allow_default_when_missing=True + ): + items = [] + # Prefer runtime discovery (file-based templates) so operators don't need + # to maintain a separate allowlist file. + for template_id in svc.get_debug_info().get("discovered_template_ids", []): # type: ignore[call-arg] + cfg = svc.get_template_config(template_id) # type: ignore[arg-type] + if cfg is None: + continue + items.append( + { + "id": template_id, + "allowed_inputs": list(cfg.allowed_inputs or []), + "defaults": dict(cfg.defaults or {}), + } + ) + items.sort(key=lambda x: x["id"]) + resp: dict = {"ok": True, "templates": items, "count": len(items)} - # Optional diagnostics. This reveals absolute paths, so keep it opt-in. - debug = request.query.get("debug", "").strip() in ("1", "true", "yes") - if debug: - try: - resp["debug"] = svc.get_debug_info() # type: ignore[attr-defined] - except Exception: - # If TemplateService interface changes, don't break the endpoint. - resp["debug"] = {"error": "debug_info_unavailable"} + # Optional diagnostics. This reveals absolute paths, so keep it opt-in. + debug = request.query.get("debug", "").strip() in ("1", "true", "yes") + if debug: + try: + resp["debug"] = svc.get_debug_info() # type: ignore[attr-defined] + except Exception: + # If TemplateService interface changes, don't break the endpoint. + resp["debug"] = {"error": "debug_info_unavailable"} - return web.json_response(resp) + return web.json_response(resp) + except TenantBoundaryError as e: + return web.json_response( + {"ok": False, "error": e.code, "message": str(e)}, + status=403, + ) except Exception as e: logger.exception("Failed to list templates") return web.json_response({"ok": False, "error": str(e)}, status=500) diff --git a/services/access_control.py b/services/access_control.py index a71b32c..5137c31 100644 --- a/services/access_control.py +++ b/services/access_control.py @@ -9,6 +9,7 @@ import ipaddress import logging import os import uuid +from collections.abc import Mapping from dataclasses import dataclass, field from typing import Dict, List, Optional, Set, Tuple @@ -18,6 +19,12 @@ except ImportError: web = None from .request_ip import get_client_ip +from .tenant_context import ( + DEFAULT_TENANT_ID, + extract_tenant_from_headers, + is_multi_tenant_enabled, + normalize_tenant_id, +) # S46: Scoped RBAC & Tiered Access try: @@ -97,6 +104,7 @@ class TokenInfo: scopes: Set[str] = field(default_factory=set) created_at: float = 0.0 expires_at: Optional[float] = None + tenant_id: str = DEFAULT_TENANT_ID def has_scope(self, required: str) -> bool: """Check if token has scope, supporting wildcards.""" @@ -123,12 +131,17 @@ class TokenRegistry: @classmethod def issue( - cls, role: "AuthTier", scopes: List[str], ttl_seconds: int = 0 + cls, + role: "AuthTier", + scopes: List[str], + ttl_seconds: int = 0, + tenant_id: str = DEFAULT_TENANT_ID, ) -> Tuple[str, TokenInfo]: """Issue a new token.""" secret = f"oc_{role.value}_{uuid.uuid4().hex}" now = datetime.datetime.now().timestamp() expires = (now + ttl_seconds) if ttl_seconds > 0 else None + normalized_tenant = normalize_tenant_id(tenant_id) info = TokenInfo( token_id=f"kid-{uuid.uuid4().hex[:8]}", @@ -136,6 +149,7 @@ class TokenRegistry: scopes=set(scopes), created_at=now, expires_at=expires, + tenant_id=normalized_tenant, ) cls._tokens[secret] = info return secret, info @@ -153,18 +167,42 @@ class TokenRegistry: return cls._tokens.get(secret) +def _header_token_value(headers: Mapping[str, str], key: str) -> str: + value = headers.get(key) + if value is None: + return "" + return str(value).strip() + + +def _resolve_header_tenant(request) -> str: + if not is_multi_tenant_enabled(): + return DEFAULT_TENANT_ID + headers = getattr(request, "headers", None) + if not isinstance(headers, Mapping): + return DEFAULT_TENANT_ID + try: + tenant = extract_tenant_from_headers(headers) + except Exception: + return DEFAULT_TENANT_ID + return tenant or DEFAULT_TENANT_ID + + def resolve_token_info(request) -> Optional[TokenInfo]: """ Resolve the request's authentication token into a TokenInfo object. 1. Check TokenRegistry (Dynamic) 2. Check Environment Variables (Static) """ + headers = getattr(request, "headers", None) + if not isinstance(headers, Mapping): + headers = {} + # Extract token from headers client_token = "" - if request.headers.get("X-OpenClaw-Admin-Token"): - client_token = request.headers.get("X-OpenClaw-Admin-Token") - elif request.headers.get("X-Moltbot-Admin-Token"): - client_token = request.headers.get("X-Moltbot-Admin-Token") + if _header_token_value(headers, "X-OpenClaw-Admin-Token"): + client_token = _header_token_value(headers, "X-OpenClaw-Admin-Token") + elif _header_token_value(headers, "X-Moltbot-Admin-Token"): + client_token = _header_token_value(headers, "X-Moltbot-Admin-Token") try: from .metrics import metrics @@ -175,10 +213,10 @@ def resolve_token_info(request) -> Optional[TokenInfo]: logger.warning( "DEPRECATION WARNING: Legacy header X-Moltbot-Admin-Token used. Please migrate to X-OpenClaw-Admin-Token." ) - elif request.headers.get("X-OpenClaw-Obs-Token"): - client_token = request.headers.get("X-OpenClaw-Obs-Token") - elif request.headers.get("X-Moltbot-Obs-Token"): - client_token = request.headers.get("X-Moltbot-Obs-Token") + elif _header_token_value(headers, "X-OpenClaw-Obs-Token"): + client_token = _header_token_value(headers, "X-OpenClaw-Obs-Token") + elif _header_token_value(headers, "X-Moltbot-Obs-Token"): + client_token = _header_token_value(headers, "X-Moltbot-Obs-Token") try: from .metrics import metrics @@ -190,6 +228,8 @@ def resolve_token_info(request) -> Optional[TokenInfo]: "DEPRECATION WARNING: Legacy header X-Moltbot-Obs-Token used. Please migrate to X-OpenClaw-Obs-Token." ) + request_tenant = _resolve_header_tenant(request) + # 1. Registry Check if client_token: info = TokenRegistry.lookup(client_token) @@ -206,7 +246,12 @@ def resolve_token_info(request) -> Optional[TokenInfo]: if admin_token and client_token: if hmac.compare_digest(client_token, admin_token): - return TokenInfo(token_id="env-admin", role=AuthTier.ADMIN, scopes={"*"}) + return TokenInfo( + token_id="env-admin", + role=AuthTier.ADMIN, + scopes={"*"}, + tenant_id=request_tenant, + ) # Observability obs_token = ( @@ -223,6 +268,7 @@ def resolve_token_info(request) -> Optional[TokenInfo]: token_id="env-obs", role=AuthTier.OBSERVABILITY, scopes={"read:*"}, # S46: Wildcard for Obs + tenant_id=request_tenant, ) # 3. Loopback @@ -230,12 +276,18 @@ def resolve_token_info(request) -> Optional[TokenInfo]: if is_loopback(remote): is_admin_configured = bool(admin_token) if not is_admin_configured: - return TokenInfo(token_id="local-admin", role=AuthTier.ADMIN, scopes={"*"}) + return TokenInfo( + token_id="local-admin", + role=AuthTier.ADMIN, + scopes={"*"}, + tenant_id=request_tenant, + ) else: return TokenInfo( token_id="local-internal", role=AuthTier.INTERNAL, scopes={"internal:call"}, + tenant_id=request_tenant, ) return None diff --git a/services/approvals/models.py b/services/approvals/models.py index d9bc386..5e71c2e 100644 --- a/services/approvals/models.py +++ b/services/approvals/models.py @@ -10,6 +10,17 @@ from datetime import datetime, timezone from enum import Enum from typing import Any, Dict, Optional +try: + from ..tenant_context import DEFAULT_TENANT_ID, normalize_tenant_id +except Exception: # pragma: no cover + DEFAULT_TENANT_ID = "default" + + def normalize_tenant_id(value, *, field_name="tenant_id"): # type: ignore + text = str(value or "").strip().lower() + if not text: + raise ValueError(f"{field_name} must be non-empty") + return text + class ApprovalStatus(str, Enum): """Status of an approval request.""" @@ -57,6 +68,7 @@ class ApprovalRequest: inputs: Dict[str, Any] = field(default_factory=dict) source: ApprovalSource = ApprovalSource.TRIGGER trace_id: Optional[str] = None + tenant_id: str = DEFAULT_TENANT_ID status: ApprovalStatus = ApprovalStatus.PENDING @@ -101,6 +113,8 @@ class ApprovalRequest: if isinstance(self.status, str): self.status = ApprovalStatus(self.status) + self.tenant_id = normalize_tenant_id(self.tenant_id, field_name="tenant_id") + @staticmethod def generate_id() -> str: """Generate a new approval ID.""" @@ -167,6 +181,7 @@ class ApprovalRequest: else self.source ), "trace_id": self.trace_id, + "tenant_id": self.tenant_id, "status": ( self.status.value if isinstance(self.status, ApprovalStatus) @@ -191,6 +206,7 @@ class ApprovalRequest: inputs=data.get("inputs", {}), source=ApprovalSource(data.get("source", "trigger")), trace_id=data.get("trace_id"), + tenant_id=data.get("tenant_id", DEFAULT_TENANT_ID), status=ApprovalStatus(data.get("status", "pending")), requested_at=data.get( "requested_at", datetime.now(timezone.utc).isoformat() diff --git a/services/approvals/service.py b/services/approvals/service.py index c972afb..91bdc9c 100644 --- a/services/approvals/service.py +++ b/services/approvals/service.py @@ -8,6 +8,11 @@ import os from datetime import datetime, timedelta, timezone from typing import Any, Dict, List, Optional +from ..tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + normalize_tenant_id, +) from ..trace import generate_trace_id from .models import ApprovalRequest, ApprovalSource, ApprovalStatus from .storage import get_approval_store @@ -38,6 +43,7 @@ class ApprovalService: inputs: Dict[str, Any], source: ApprovalSource = ApprovalSource.TRIGGER, trace_id: Optional[str] = None, + tenant_id: Optional[str] = None, requested_by: Optional[str] = None, delivery: Optional[Dict[str, Any]] = None, ttl_sec: Optional[int] = None, @@ -65,6 +71,9 @@ class ApprovalService: # Generate IDs approval_id = ApprovalRequest.generate_id() trace_id = trace_id or generate_trace_id() + resolved_tenant = normalize_tenant_id( + tenant_id or get_current_tenant_id() or DEFAULT_TENANT_ID + ) # Calculate expiration ttl = ttl_sec if ttl_sec is not None else DEFAULT_TTL_SEC @@ -77,6 +86,7 @@ class ApprovalService: inputs=inputs, source=source, trace_id=trace_id, + tenant_id=resolved_tenant, status=ApprovalStatus.PENDING, requested_by=requested_by, expires_at=expires_at, @@ -89,20 +99,30 @@ class ApprovalService: raise ValueError(f"Failed to create approval request: {approval_id}") logger.info( - f"Created approval request: {approval_id} (template={template_id}, source={source.value})" + "Created approval request: %s (template=%s, source=%s, tenant=%s)", + approval_id, + template_id, + source.value, + resolved_tenant, ) return request - def get(self, approval_id: str) -> Optional[ApprovalRequest]: + def get( + self, approval_id: str, tenant_id: Optional[str] = None + ) -> Optional[ApprovalRequest]: """Get an approval request by ID.""" - return self._store.get(approval_id) + return self._store.get(approval_id, tenant_id=tenant_id) - def list_pending(self, limit: int = 100) -> List[ApprovalRequest]: + def list_pending( + self, limit: int = 100, tenant_id: Optional[str] = None + ) -> List[ApprovalRequest]: """List pending approval requests.""" # First expire any due requests self._store.expire_due() - pending = self._store.list_by_status(ApprovalStatus.PENDING) + pending = self._store.list_by_status( + ApprovalStatus.PENDING, tenant_id=tenant_id + ) # Sort by requested_at (oldest first) pending.sort(key=lambda x: x.requested_at) @@ -114,14 +134,15 @@ class ApprovalService: status: Optional[ApprovalStatus] = None, limit: int = 100, offset: int = 0, + tenant_id: Optional[str] = None, ) -> List[ApprovalRequest]: """List approval requests with optional status filter.""" self._store.expire_due() if status: - approvals = self._store.list_by_status(status) + approvals = self._store.list_by_status(status, tenant_id=tenant_id) else: - approvals = self._store.list_all() + approvals = self._store.list_all(tenant_id=tenant_id) # Sort by requested_at (newest first for history) approvals.sort(key=lambda x: x.requested_at, reverse=True) @@ -132,6 +153,7 @@ class ApprovalService: self, approval_id: str, actor: Optional[str] = None, + tenant_id: Optional[str] = None, ) -> ApprovalRequest: """ Approve a pending request. @@ -146,7 +168,7 @@ class ApprovalService: Raises: ValueError: If request not found or not pending. """ - request = self._store.get(approval_id) + request = self._store.get(approval_id, tenant_id=tenant_id) if not request: raise ValueError(f"Approval request not found: {approval_id}") @@ -170,6 +192,7 @@ class ApprovalService: self, approval_id: str, actor: Optional[str] = None, + tenant_id: Optional[str] = None, ) -> ApprovalRequest: """ Reject a pending request. @@ -184,7 +207,7 @@ class ApprovalService: Raises: ValueError: If request not found or not pending. """ - request = self._store.get(approval_id) + request = self._store.get(approval_id, tenant_id=tenant_id) if not request: raise ValueError(f"Approval request not found: {approval_id}") @@ -204,6 +227,7 @@ class ApprovalService: prompt_id: str, trace_id: Optional[str] = None, actor: Optional[str] = None, + tenant_id: Optional[str] = None, ) -> ApprovalRequest: """ Record execution metadata after an approval is executed. @@ -211,7 +235,7 @@ class ApprovalService: NOTE: The chat connector relies on executed_prompt_id to deliver images when approvals are done in the UI. Do not remove without updating connector. """ - request = self._store.get(approval_id) + request = self._store.get(approval_id, tenant_id=tenant_id) if not request: raise ValueError(f"Approval request not found: {approval_id}") @@ -233,10 +257,10 @@ class ApprovalService: ) return request - def count_pending(self) -> int: + def count_pending(self, tenant_id: Optional[str] = None) -> int: """Count pending approval requests.""" self._store.expire_due() - return self._store.count_pending() + return self._store.count_pending(tenant_id=tenant_id) # Singleton instance diff --git a/services/approvals/storage.py b/services/approvals/storage.py index a001112..3b2c8c9 100644 --- a/services/approvals/storage.py +++ b/services/approvals/storage.py @@ -11,6 +11,11 @@ from datetime import datetime, timezone from typing import Dict, List, Optional from ..state_dir import get_state_dir +from ..tenant_context import ( + DEFAULT_TENANT_ID, + is_multi_tenant_enabled, + normalize_tenant_id, +) from .models import ApprovalRequest, ApprovalStatus logger = logging.getLogger("ComfyUI-OpenClaw.services.approvals") @@ -129,11 +134,29 @@ class ApprovalStore: self._approvals = load_approvals() self._loaded = True - def get(self, approval_id: str) -> Optional[ApprovalRequest]: + def _match_tenant( + self, approval: ApprovalRequest, tenant_id: Optional[str] + ) -> bool: + if not is_multi_tenant_enabled() or tenant_id is None: + return True + try: + expected = normalize_tenant_id(tenant_id) + except Exception: + expected = DEFAULT_TENANT_ID + return approval.tenant_id == expected + + def get( + self, approval_id: str, tenant_id: Optional[str] = None + ) -> Optional[ApprovalRequest]: """Get an approval by ID.""" with self._lock: self._ensure_loaded() - return self._approvals.get(approval_id) + approval = self._approvals.get(approval_id) + if approval is None: + return None + if not self._match_tenant(approval, tenant_id): + return None + return approval def add(self, approval: ApprovalRequest) -> bool: """Add a new approval.""" @@ -179,19 +202,27 @@ class ApprovalStore: del self._approvals[approval_id] return save_approvals(self._approvals) - def list_all(self) -> List[ApprovalRequest]: + def list_all(self, tenant_id: Optional[str] = None) -> List[ApprovalRequest]: """List all approvals.""" with self._lock: self._ensure_loaded() - return list(self._approvals.values()) + return [ + a for a in self._approvals.values() if self._match_tenant(a, tenant_id) + ] - def list_by_status(self, status: ApprovalStatus) -> List[ApprovalRequest]: + def list_by_status( + self, status: ApprovalStatus, tenant_id: Optional[str] = None + ) -> List[ApprovalRequest]: """List approvals by status.""" with self._lock: self._ensure_loaded() - return [a for a in self._approvals.values() if a.status == status] + return [ + a + for a in self._approvals.values() + if a.status == status and self._match_tenant(a, tenant_id) + ] - def count_pending(self) -> int: + def count_pending(self, tenant_id: Optional[str] = None) -> int: """Count pending approvals.""" with self._lock: self._ensure_loaded() @@ -199,6 +230,7 @@ class ApprovalStore: 1 for a in self._approvals.values() if a.status == ApprovalStatus.PENDING + and self._match_tenant(a, tenant_id) ) def expire_due(self) -> int: diff --git a/services/connector_callback_contract.py b/services/connector_callback_contract.py index bdf6bc3..8884ceb 100644 --- a/services/connector_callback_contract.py +++ b/services/connector_callback_contract.py @@ -28,6 +28,14 @@ except ImportError: get_connector_installation_registry, ) +try: + from .tenant_context import DEFAULT_TENANT_ID, get_current_tenant_id +except ImportError: + from services.tenant_context import ( # type: ignore + DEFAULT_TENANT_ID, + get_current_tenant_id, + ) + logger = logging.getLogger("ComfyUI-OpenClaw.services.connector_callback_contract") DEFAULT_CALLBACK_TIMESTAMP_DRIFT_SEC = 300 @@ -48,6 +56,7 @@ class CallbackDecisionCode(str, Enum): REJECT_AMBIGUOUS_INSTALLATION = "cb_reject_ambiguous_installation" REJECT_INACTIVE_INSTALLATION = "cb_reject_inactive_installation" REJECT_STALE_TOKEN_REF = "cb_reject_stale_token_ref" + REJECT_TENANT_MISMATCH = "cb_reject_tenant_mismatch" REJECT_POLICY_DENIED = "cb_reject_policy_denied" REJECT_INVALID_ENVELOPE = "cb_reject_invalid_envelope" @@ -57,6 +66,7 @@ class CallbackActorContext: is_admin: bool = False is_trusted: bool = False user_id: str = "" + tenant_id: str = DEFAULT_TENANT_ID @dataclass @@ -189,6 +199,8 @@ class ConnectorCallbackContract: code = CallbackDecisionCode.REJECT_INACTIVE_INSTALLATION.value elif reason.startswith("stale_token_ref"): code = CallbackDecisionCode.REJECT_STALE_TOKEN_REF.value + elif reason == "tenant_mismatch": + code = CallbackDecisionCode.REJECT_TENANT_MISMATCH.value else: code = CallbackDecisionCode.REJECT_INVALID_ENVELOPE.value return CallbackDecision(ok=False, decision_code=code, message=reason) @@ -302,7 +314,9 @@ class ConnectorCallbackContract: return decision resolution = self._installation_registry.resolve_installation( - platform, envelope.workspace_id + platform, + envelope.workspace_id, + tenant_id=(actor.tenant_id or get_current_tenant_id()), ) if not resolution.ok or resolution.installation is None: decision = self._map_installation_reject(resolution) diff --git a/services/connector_installation_registry.py b/services/connector_installation_registry.py index 983b477..aeaf344 100644 --- a/services/connector_installation_registry.py +++ b/services/connector_installation_registry.py @@ -14,10 +14,22 @@ try: from .audit import emit_audit_event from .secret_store import SecretStore, get_secret_store from .state_dir import get_state_dir + from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) except ImportError: from services.audit import emit_audit_event # type: ignore from services.secret_store import SecretStore, get_secret_store # type: ignore from services.state_dir import get_state_dir # type: ignore + from services.tenant_context import ( # type: ignore + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) logger = logging.getLogger("ComfyUI-OpenClaw.services.connector_installation_registry") @@ -44,6 +56,7 @@ class ConnectorInstallation: platform: str workspace_id: str installation_id: str + tenant_id: str = DEFAULT_TENANT_ID token_refs: Dict[str, str] = field(default_factory=dict) status: str = InstallationStatus.CREATED.value updated_at: float = field(default_factory=time.time) @@ -54,6 +67,7 @@ class ConnectorInstallation: def to_public_dict(self) -> Dict[str, Any]: return { "platform": self.platform, + "tenant_id": self.tenant_id, "workspace_id": self.workspace_id, "installation_id": self.installation_id, "token_refs": dict(self.token_refs), @@ -71,6 +85,7 @@ class InstallationAuditEvent: action: str installation_id: str platform: str + tenant_id: str workspace_id: str status: str details: Dict[str, Any] = field(default_factory=dict) @@ -81,6 +96,7 @@ class InstallationAuditEvent: "action": self.action, "installation_id": self.installation_id, "platform": self.platform, + "tenant_id": self.tenant_id, "workspace_id": self.workspace_id, "status": self.status, "details": dict(self.details), @@ -132,14 +148,14 @@ class ConnectorInstallationRegistry: return self._normalize_identifier(platform, "platform").lower() def _store_token_refs( - self, installation_id: str, token_values: Dict[str, str] + self, installation_id: str, token_values: Dict[str, str], tenant_id: str ) -> Dict[str, str]: refs: Dict[str, str] = {} for key, value in (token_values or {}).items(): name = self._normalize_identifier(key, "token_name") secret_value = self._normalize_identifier(value, f"token:{name}") ref = f"connector_installation:{installation_id}:{name}" - self._secret_store.set_secret(ref, secret_value) + self._secret_store.set_secret(ref, secret_value, tenant_id=tenant_id) refs[name] = ref return refs @@ -154,6 +170,7 @@ class ConnectorInstallationRegistry: action=action, installation_id=installation.installation_id, platform=installation.platform, + tenant_id=installation.tenant_id, workspace_id=installation.workspace_id, status=installation.status, details=details, @@ -168,6 +185,7 @@ class ConnectorInstallationRegistry: status_code=200, details={ "platform": installation.platform, + "tenant_id": installation.tenant_id, "workspace_id": installation.workspace_id, "status": installation.status, **details, @@ -185,6 +203,7 @@ class ConnectorInstallationRegistry: for item in installations: inst = ConnectorInstallation( platform=item.get("platform", ""), + tenant_id=item.get("tenant_id", DEFAULT_TENANT_ID), workspace_id=item.get("workspace_id", ""), installation_id=item.get("installation_id", ""), token_refs=dict(item.get("token_refs", {}) or {}), @@ -202,6 +221,7 @@ class ConnectorInstallationRegistry: action=item.get("action", "unknown"), installation_id=item.get("installation_id", ""), platform=item.get("platform", ""), + tenant_id=item.get("tenant_id", DEFAULT_TENANT_ID), workspace_id=item.get("workspace_id", ""), status=item.get("status", ""), details=dict(item.get("details", {}) or {}), @@ -228,6 +248,7 @@ class ConnectorInstallationRegistry: self, *, platform: str, + tenant_id: str = DEFAULT_TENANT_ID, workspace_id: str, installation_id: str, token_values: Optional[Dict[str, str]] = None, @@ -238,6 +259,7 @@ class ConnectorInstallationRegistry: ) -> ConnectorInstallation: with self._lock: normalized_platform = self._normalize_platform(platform) + normalized_tenant = normalize_tenant_id(tenant_id) normalized_workspace = self._normalize_identifier( workspace_id, "workspace_id" ) @@ -248,7 +270,9 @@ class ConnectorInstallationRegistry: refs = dict(token_refs or {}) if token_values: refs.update( - self._store_token_refs(normalized_installation, token_values) + self._store_token_refs( + normalized_installation, token_values, normalized_tenant + ) ) existing = self._installations.get(normalized_installation) if existing is None: @@ -259,6 +283,7 @@ class ConnectorInstallationRegistry: refs = dict(existing.token_refs) inst = ConnectorInstallation( platform=normalized_platform, + tenant_id=normalized_tenant, workspace_id=normalized_workspace, installation_id=normalized_installation, token_refs=refs, @@ -273,15 +298,25 @@ class ConnectorInstallationRegistry: self._save() return inst - def get_installation(self, installation_id: str) -> Optional[ConnectorInstallation]: + def get_installation( + self, installation_id: str, tenant_id: Optional[str] = None + ) -> Optional[ConnectorInstallation]: with self._lock: inst = self._installations.get(str(installation_id).strip()) + if ( + inst is not None + and is_multi_tenant_enabled() + and tenant_id is not None + and inst.tenant_id != normalize_tenant_id(tenant_id) + ): + return None return None if inst is None else ConnectorInstallation(**asdict(inst)) def list_installations( self, *, platform: Optional[str] = None, + tenant_id: Optional[str] = None, workspace_id: Optional[str] = None, status: Optional[str] = None, ) -> List[ConnectorInstallation]: @@ -291,6 +326,9 @@ class ConnectorInstallationRegistry: items = [ i for i in items if i.platform == self._normalize_platform(platform) ] + if tenant_id: + normalized_tenant = normalize_tenant_id(tenant_id) + items = [i for i in items if i.tenant_id == normalized_tenant] if workspace_id: items = [ i for i in items if i.workspace_id == str(workspace_id).strip() @@ -300,6 +338,7 @@ class ConnectorInstallationRegistry: items.sort( key=lambda inst: ( inst.platform, + inst.tenant_id, inst.workspace_id, inst.installation_id, ) @@ -349,7 +388,11 @@ class ConnectorInstallationRegistry: if inst is None: raise ValueError(f"Installation not found: {installation_id}") refs = dict(inst.token_refs) - refs.update(self._store_token_refs(inst.installation_id, token_values)) + refs.update( + self._store_token_refs( + inst.installation_id, token_values, inst.tenant_id + ) + ) inst.token_refs = refs inst.status = InstallationStatus.ROTATING.value inst.status_reason = reason @@ -392,7 +435,7 @@ class ConnectorInstallationRegistry: if inst is None: raise ValueError(f"Installation not found: {installation_id}") for ref in list(inst.token_refs.values()): - self._secret_store.clear_secret(ref) + self._secret_store.clear_secret(ref, tenant_id=inst.tenant_id) inst.status = InstallationStatus.UNINSTALLED.value inst.status_reason = reason inst.updated_at = time.time() @@ -402,19 +445,33 @@ class ConnectorInstallationRegistry: return ConnectorInstallation(**asdict(inst)) def resolve_installation( - self, platform: str, workspace_id: str + self, platform: str, workspace_id: str, tenant_id: Optional[str] = None ) -> InstallationResolution: with self._lock: normalized_platform = self._normalize_platform(platform) normalized_workspace = self._normalize_identifier( workspace_id, "workspace_id" ) + normalized_tenant = normalize_tenant_id( + tenant_id or get_current_tenant_id() or DEFAULT_TENANT_ID + ) matches = [ inst for inst in self._installations.values() if inst.platform == normalized_platform and inst.workspace_id == normalized_workspace ] + if is_multi_tenant_enabled(): + tenant_matches = [ + inst for inst in matches if inst.tenant_id == normalized_tenant + ] + if not tenant_matches and matches: + return InstallationResolution( + ok=False, + reject_reason="tenant_mismatch", + audit_code="conn_install.resolve_tenant_mismatch", + ) + matches = tenant_matches eligible = [inst for inst in matches if inst.status in _RESOLVABLE_STATUSES] if len(eligible) > 1: return InstallationResolution( @@ -436,7 +493,7 @@ class ConnectorInstallationRegistry: ) inst = eligible[0] for token_name, ref in inst.token_refs.items(): - if not self._secret_store.get_secret(ref): + 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}", @@ -452,6 +509,7 @@ class ConnectorInstallationRegistry: self, *, installation_id: Optional[str] = None, + tenant_id: Optional[str] = None, limit: int = 100, ) -> List[Dict[str, Any]]: with self._lock: @@ -462,15 +520,24 @@ class ConnectorInstallationRegistry: for event in items if event.installation_id == str(installation_id).strip() ] + if tenant_id: + normalized_tenant = normalize_tenant_id(tenant_id) + items = [ + event for event in items if event.tenant_id == normalized_tenant + ] return [event.to_dict() for event in items[-max(1, min(limit, 500)) :]] - def diagnostics(self) -> Dict[str, Any]: + def diagnostics(self, tenant_id: Optional[str] = None) -> Dict[str, Any]: with self._lock: counts: Dict[str, int] = {} - for inst in self._installations.values(): + 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 return { - "installation_count": len(self._installations), + "installation_count": len(items), "status_counts": counts, "audit_events": len(self._audit_trail), } diff --git a/services/execution_budgets.py b/services/execution_budgets.py index d728bc5..80b9c23 100644 --- a/services/execution_budgets.py +++ b/services/execution_budgets.py @@ -20,6 +20,13 @@ from contextlib import asynccontextmanager from dataclasses import dataclass from typing import Dict, Optional +from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, +) + logger = logging.getLogger("ComfyUI-OpenClaw.services.execution_budgets") # Global concurrency budgets (tuneable via env vars) @@ -28,6 +35,7 @@ DEFAULT_MAX_INFLIGHT_WEBHOOK = 1 DEFAULT_MAX_INFLIGHT_TRIGGER = 1 DEFAULT_MAX_INFLIGHT_SCHEDULER = 1 DEFAULT_MAX_INFLIGHT_BRIDGE = 1 +DEFAULT_MAX_INFLIGHT_PER_TENANT = 1 # Render size budget (512KB default) DEFAULT_MAX_RENDERED_WORKFLOW_BYTES = 512 * 1024 # 512KB @@ -54,7 +62,8 @@ class BudgetConfig: max_inflight_trigger: int max_inflight_scheduler: int max_inflight_bridge: int - max_rendered_workflow_bytes: int + max_inflight_per_tenant: int = DEFAULT_MAX_INFLIGHT_PER_TENANT + max_rendered_workflow_bytes: int = DEFAULT_MAX_RENDERED_WORKFLOW_BYTES def load_budget_config() -> BudgetConfig: @@ -75,6 +84,10 @@ def load_budget_config() -> BudgetConfig: max_inflight_bridge=_get_env_int( "OPENCLAW_MAX_INFLIGHT_SUBMITS_BRIDGE", DEFAULT_MAX_INFLIGHT_BRIDGE ), + max_inflight_per_tenant=_get_env_int( + "OPENCLAW_MAX_INFLIGHT_SUBMITS_PER_TENANT", + DEFAULT_MAX_INFLIGHT_PER_TENANT, + ), max_rendered_workflow_bytes=_get_env_int( "OPENCLAW_MAX_RENDERED_WORKFLOW_BYTES", DEFAULT_MAX_RENDERED_WORKFLOW_BYTES ), @@ -109,6 +122,9 @@ class ExecutionBudgetLimiter: "scheduler": asyncio.Semaphore(self.config.max_inflight_scheduler), "bridge": asyncio.Semaphore(self.config.max_inflight_bridge), } + # S49: optional per-tenant semaphores (created lazily). + self._tenant_semaphores: Dict[str, asyncio.Semaphore] = {} + self._tenant_lock = asyncio.Lock() # Tracking counters (for observability) self._inflight_total = 0 @@ -119,9 +135,25 @@ class ExecutionBudgetLimiter: "bridge": 0, "unknown": 0, } + self._inflight_by_tenant: Dict[str, int] = {} + + async def _get_tenant_semaphore(self, tenant_id: str) -> asyncio.Semaphore: + async with self._tenant_lock: + existing = self._tenant_semaphores.get(tenant_id) + if existing is not None: + return existing + cap = max(1, int(self.config.max_inflight_per_tenant)) + semaphore = asyncio.Semaphore(cap) + self._tenant_semaphores[tenant_id] = semaphore + return semaphore @asynccontextmanager - async def acquire(self, source: str = "unknown", trace_id: Optional[str] = None): + async def acquire( + self, + source: str = "unknown", + trace_id: Optional[str] = None, + tenant_id: Optional[str] = None, + ): """ Acquire concurrency slots for execution (best-effort non-blocking). @@ -146,6 +178,9 @@ class ExecutionBudgetLimiter: source = source.lower() if source else "unknown" if source not in self._source_semaphores: source = "unknown" + tenant = DEFAULT_TENANT_ID + if is_multi_tenant_enabled(): + tenant = normalize_tenant_id(tenant_id or get_current_tenant_id()) # Check global budget (locked check + manual acquire) if self._global_semaphore.locked(): @@ -195,7 +230,27 @@ class ExecutionBudgetLimiter: retry_after=1, ) - # Acquire both semaphores manually (explicit control) + # S49: Optional per-tenant concurrency cap. + tenant_semaphore = None + if is_multi_tenant_enabled(): + tenant_semaphore = await self._get_tenant_semaphore(tenant) + if tenant_semaphore.locked(): + limit = max(1, int(self.config.max_inflight_per_tenant)) + logger.warning( + "Tenant concurrency budget exhausted for tenant=%s (max=%s), denying %s submission (trace_id=%s)", + tenant, + limit, + source, + trace_id, + ) + raise BudgetExceededError( + budget_type="tenant_concurrency", + limit=limit, + source=source, + retry_after=1, + ) + + # Acquire semaphores manually (explicit control) try: await self._global_semaphore.acquire() except Exception: @@ -205,12 +260,23 @@ class ExecutionBudgetLimiter: global_acquired = True try: + tenant_acquired = False + if tenant_semaphore is not None: + try: + await tenant_semaphore.acquire() + tenant_acquired = True + except Exception: + self._global_semaphore.release() + raise + if source_semaphore: try: await source_semaphore.acquire() source_acquired = True except Exception: # Failed to acquire source, release global and re-raise + if tenant_acquired and tenant_semaphore is not None: + tenant_semaphore.release() self._global_semaphore.release() raise else: @@ -221,6 +287,9 @@ class ExecutionBudgetLimiter: self._inflight_by_source[source] = ( self._inflight_by_source.get(source, 0) + 1 ) + self._inflight_by_tenant[tenant] = ( + self._inflight_by_tenant.get(tenant, 0) + 1 + ) logger.debug( f"Acquired budget for {source} (inflight: total={self._inflight_total}, " @@ -233,9 +302,16 @@ class ExecutionBudgetLimiter: # Release and update tracking (always runs) self._inflight_total -= 1 self._inflight_by_source[source] -= 1 + self._inflight_by_tenant[tenant] = max( + 0, self._inflight_by_tenant.get(tenant, 0) - 1 + ) + if self._inflight_by_tenant[tenant] == 0: + self._inflight_by_tenant.pop(tenant, None) if source_acquired and source_semaphore: source_semaphore.release() + if tenant_acquired and tenant_semaphore is not None: + tenant_semaphore.release() self._global_semaphore.release() logger.debug( @@ -252,6 +328,7 @@ class ExecutionBudgetLimiter: return { "total": self._inflight_total, **self._inflight_by_source, + "tenant_count": len(self._inflight_by_tenant), } diff --git a/services/presets/models.py b/services/presets/models.py index 9411bf0..cfdf084 100644 --- a/services/presets/models.py +++ b/services/presets/models.py @@ -8,6 +8,17 @@ import uuid from dataclasses import asdict, dataclass, field from typing import Any, Dict, List, Optional +try: + from ..tenant_context import DEFAULT_TENANT_ID, normalize_tenant_id +except Exception: # pragma: no cover + DEFAULT_TENANT_ID = "default" + + def normalize_tenant_id(value, *, field_name="tenant_id"): # type: ignore + text = str(value or "").strip().lower() + if not text: + raise ValueError(f"{field_name} must be non-empty") + return text + @dataclass class Preset: @@ -21,6 +32,7 @@ class Preset: category: str = "general" # e.g., "prompt", "parameters", "full" tags: List[str] = field(default_factory=list) content: Dict[str, Any] = field(default_factory=dict) + tenant_id: str = DEFAULT_TENANT_ID created_at: float = field(default_factory=time.time) updated_at: float = field(default_factory=time.time) @@ -39,6 +51,7 @@ class Preset: content=content, category=category, tags=tags or [], + tenant_id=DEFAULT_TENANT_ID, ) def to_dict(self) -> Dict[str, Any]: @@ -71,4 +84,9 @@ class Preset: @staticmethod def from_dict(data: Dict[str, Any]) -> "Preset": - return Preset(**data) + if "tenant_id" not in data: + data = dict(data) + data["tenant_id"] = DEFAULT_TENANT_ID + preset = Preset(**data) + preset.tenant_id = normalize_tenant_id(preset.tenant_id, field_name="tenant_id") + return preset diff --git a/services/presets/storage.py b/services/presets/storage.py index ebac7a3..13a16ef 100644 --- a/services/presets/storage.py +++ b/services/presets/storage.py @@ -9,6 +9,11 @@ from pathlib import Path from typing import Dict, List, Optional from ..paths import get_presets_dir +from ..tenant_context import ( + DEFAULT_TENANT_ID, + is_multi_tenant_enabled, + normalize_tenant_id, +) from .models import Preset logger = logging.getLogger("ComfyUI-OpenClaw.services.presets") @@ -25,8 +30,25 @@ class PresetStore: def _get_path(self, preset_id: str) -> Path: return self.storage_dir / f"{preset_id}.json" + def _resolve_tenant_id(self, tenant_id: Optional[str]) -> Optional[str]: + if not is_multi_tenant_enabled(): + return None + try: + return normalize_tenant_id(tenant_id or DEFAULT_TENANT_ID) + except Exception: + return DEFAULT_TENANT_ID + + def _is_visible_to_tenant(self, preset: Preset, tenant_id: Optional[str]) -> bool: + resolved = self._resolve_tenant_id(tenant_id) + if resolved is None: + return True + return preset.tenant_id == resolved + def list_presets( - self, category: Optional[str] = None, tag: Optional[str] = None + self, + category: Optional[str] = None, + tag: Optional[str] = None, + tenant_id: Optional[str] = None, ) -> List[Preset]: """List all presets, optionally filtered.""" presets = [] @@ -35,6 +57,8 @@ class PresetStore: try: p = self._load_file(file_path) if p: + if not self._is_visible_to_tenant(p, tenant_id): + continue if category and p.category != category: continue if tag and tag not in p.tags: @@ -50,15 +74,29 @@ class PresetStore: presets.sort(key=lambda x: x.updated_at, reverse=True) return presets - def get_preset(self, preset_id: str) -> Optional[Preset]: + def get_preset( + self, preset_id: str, tenant_id: Optional[str] = None + ) -> Optional[Preset]: """Get a specific preset.""" path = self._get_path(preset_id) if not path.exists(): return None - return self._load_file(path) + preset = self._load_file(path) + if preset is None: + return None + if not self._is_visible_to_tenant(preset, tenant_id): + return None + return preset def save_preset(self, preset: Preset) -> bool: """Save/Update a preset.""" + try: + preset.tenant_id = normalize_tenant_id( + getattr(preset, "tenant_id", DEFAULT_TENANT_ID), + field_name="tenant_id", + ) + except Exception: + preset.tenant_id = DEFAULT_TENANT_ID path = self._get_path(preset.id) try: with open(path, "w", encoding="utf-8") as f: @@ -69,11 +107,14 @@ class PresetStore: logger.error(f"Failed to save preset {preset.id}: {e}") return False - def delete_preset(self, preset_id: str) -> bool: + def delete_preset(self, preset_id: str, tenant_id: Optional[str] = None) -> bool: """Delete a preset.""" path = self._get_path(preset_id) if not path.exists(): return False + preset = self._load_file(path) + if preset is not None and not self._is_visible_to_tenant(preset, tenant_id): + return False try: path.unlink() logger.info(f"Deleted preset {preset_id}") diff --git a/services/providers/keys.py b/services/providers/keys.py index ea6318d..382d33c 100644 --- a/services/providers/keys.py +++ b/services/providers/keys.py @@ -20,7 +20,9 @@ GENERIC_KEY_NAMES = [ ] -def get_api_key_for_provider(provider: str) -> Optional[str]: +def get_api_key_for_provider( + provider: str, tenant_id: Optional[str] = None +) -> Optional[str]: """ Get API key for a specific provider. @@ -32,7 +34,7 @@ def get_api_key_for_provider(provider: str) -> Optional[str]: Returns None if no key found (acceptable for local providers). """ - key, _source = resolve_provider_secret(provider) + key, _source = resolve_provider_secret(provider, tenant_id=tenant_id) return key @@ -55,7 +57,7 @@ def mask_api_key(key: str) -> str: return f"{key[:4]}...{key[-4:]}" -def get_all_configured_keys() -> dict: +def get_all_configured_keys(tenant_id: Optional[str] = None) -> dict: """ Get a summary of configured keys (masked). Used for diagnostics, never returns actual key values. @@ -70,7 +72,7 @@ def get_all_configured_keys() -> dict: from ..secret_store import get_secret_store store = get_secret_store() - store_status = store.get_status() + store_status = store.get_status(tenant_id=tenant_id) except Exception as e: logger.debug(f"S25: Failed to get secret store status (non-fatal): {e}") @@ -106,7 +108,9 @@ def get_all_configured_keys() -> dict: source = "server_store" if key is None and source is None: - resolved, resolved_source = resolve_provider_secret(provider_id) + resolved, resolved_source = resolve_provider_secret( + provider_id, tenant_id=tenant_id + ) if resolved: key = resolved source = resolved_source @@ -137,7 +141,9 @@ def get_all_configured_keys() -> dict: if generic_key is None and "generic" in store_status: generic_source = "server_store" if generic_key is None and generic_source is None: - resolved, resolved_source = resolve_provider_secret("generic") + resolved, resolved_source = resolve_provider_secret( + "generic", tenant_id=tenant_id + ) if resolved: generic_key = resolved generic_source = resolved_source diff --git a/services/queue_submit.py b/services/queue_submit.py index 03304c8..28481d4 100644 --- a/services/queue_submit.py +++ b/services/queue_submit.py @@ -46,6 +46,11 @@ configure_logger_for_structured_output(logger) import os +try: + from .tenant_context import get_current_tenant_id +except ImportError: + from services.tenant_context import get_current_tenant_id # type: ignore + # ComfyUI internal server URL fallback COMFYUI_URL = ( os.environ.get("OPENCLAW_COMFYUI_URL") @@ -60,6 +65,7 @@ async def submit_prompt( extra_data: Optional[Dict[str, Any]] = None, source: str = "unknown", # R33: Source tracking trace_id: Optional[str] = None, # R33: Trace ID for logging + tenant_id: Optional[str] = None, # S49: tenant context for budget + audit metadata ) -> Dict[str, Any]: """ Submit a prompt workflow to ComfyUI with execution budgets (R33). @@ -85,6 +91,7 @@ async def submit_prompt( "source": source, "trace_id": trace_id, "has_extra_data": bool(extra_data), + "tenant_id": tenant_id or get_current_tenant_id(), }, ) # NOTE: Must try relative import first. In ComfyUI runtime, `services` is not a top-level module. @@ -108,6 +115,10 @@ async def submit_prompt( if extra_data: payload["extra_data"] = extra_data + # S49: keep tenant context in queue metadata for cross-service traceability. + openclaw_extra = payload.setdefault("extra_data", {}).setdefault("openclaw", {}) + openclaw_extra.setdefault("tenant_id", tenant_id or get_current_tenant_id()) + # NOTE: Debug-only full payload logging for troubleshooting mismatched outputs. # Enable with OPENCLAW_DEBUG_PROMPT_PAYLOAD=1. This may include sensitive prompt content. if os.environ.get("OPENCLAW_DEBUG_PROMPT_PAYLOAD", "").strip().lower() in ( @@ -132,7 +143,11 @@ async def submit_prompt( # R33: Acquire concurrency budget limiter = get_limiter() - async with limiter.acquire(source=source, trace_id=trace_id): + async with limiter.acquire( + source=source, + trace_id=trace_id, + tenant_id=tenant_id or get_current_tenant_id(), + ): # Use aiohttp to post to local ComfyUI instance # We assume we are running INSIDE ComfyUI process, but for HTTP access we use localhost # unless we can hook internal server entry point. diff --git a/services/runtime_config.py b/services/runtime_config.py index 1c9a88f..1eccc35 100644 --- a/services/runtime_config.py +++ b/services/runtime_config.py @@ -12,6 +12,35 @@ from urllib.parse import urlparse logger = logging.getLogger("ComfyUI-OpenClaw.services.runtime_config") +# S49: tenant context + namespace-aware config resolution. +try: + from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) +except ImportError: + try: + from services.tenant_context import ( # type: ignore + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) + except ImportError: + DEFAULT_TENANT_ID = "default" + + def get_current_tenant_id(): # type: ignore + return DEFAULT_TENANT_ID + + def is_multi_tenant_enabled(): # type: ignore + return False + + def normalize_tenant_id(value): # type: ignore + return str(value or DEFAULT_TENANT_ID).strip().lower() or DEFAULT_TENANT_ID + + # R70: Settings schema registry (type coercion + unknown-key rejection) try: from .settings_schema import coerce_dict as _schema_coerce @@ -389,6 +418,52 @@ def _get_constraint_range(key: str) -> Tuple[int, int]: return min_val, max_val +def _resolve_active_tenant_id(tenant_id: Optional[str] = None) -> str: + if not is_multi_tenant_enabled(): + return DEFAULT_TENANT_ID + if tenant_id is None: + tenant_id = get_current_tenant_id() + try: + return normalize_tenant_id(tenant_id) + except Exception: + return DEFAULT_TENANT_ID + + +def _allow_tenant_config_fallback() -> bool: + value = ( + os.environ.get("OPENCLAW_MULTI_TENANT_ALLOW_CONFIG_FALLBACK") + or os.environ.get("MOLTBOT_MULTI_TENANT_ALLOW_CONFIG_FALLBACK") + or "0" + ) + return str(value).strip().lower() in ("1", "true", "yes", "on") + + +def _runtime_override_section(tenant_id: Optional[str] = None) -> str: + resolved = _resolve_active_tenant_id(tenant_id) + if resolved == DEFAULT_TENANT_ID: + return "llm" + return f"llm::{resolved}" + + +def _tenant_llm_config_view( + config_blob: Dict[str, Any], tenant_id: str +) -> Dict[str, Any]: + llm_global = config_blob.get("llm", {}) + if tenant_id == DEFAULT_TENANT_ID: + return llm_global if isinstance(llm_global, dict) else {} + + tenants = config_blob.get("tenants", {}) + tenant_cfg = {} + if isinstance(tenants, dict): + tenant_cfg = tenants.get(tenant_id, {}) + tenant_llm = tenant_cfg.get("llm", {}) if isinstance(tenant_cfg, dict) else {} + if isinstance(tenant_llm, dict) and tenant_llm: + return tenant_llm + if _allow_tenant_config_fallback() and isinstance(llm_global, dict): + return llm_global + return {} + + def _load_file_config() -> Dict[str, Any]: """Load config from file if exists.""" if os.path.exists(CONFIG_FILE): @@ -584,12 +659,14 @@ def _normalize_llm_layer_value(key: str, value: Any, source: str) -> Any: return value -def get_runtime_overrides() -> Dict[str, Any]: +def get_runtime_overrides(tenant_id: Optional[str] = None) -> Dict[str, Any]: """Get current in-memory runtime overrides for the LLM section.""" - return _get_runtime_overrides("llm") + return _get_runtime_overrides(_runtime_override_section(tenant_id)) -def set_runtime_overrides(updates: Dict[str, Any]) -> Tuple[bool, list]: +def set_runtime_overrides( + updates: Dict[str, Any], tenant_id: Optional[str] = None +) -> Tuple[bool, list]: """ Set in-memory runtime overrides for LLM config (non-persisted). @@ -598,16 +675,20 @@ def set_runtime_overrides(updates: Dict[str, Any]) -> Tuple[bool, list]: sanitized, errors = validate_config_update(updates) if errors: return False, errors - _set_runtime_overrides("llm", sanitized) + _set_runtime_overrides(_runtime_override_section(tenant_id), sanitized) return True, [] -def clear_runtime_overrides(keys: Optional[List[str]] = None) -> None: +def clear_runtime_overrides( + keys: Optional[List[str]] = None, tenant_id: Optional[str] = None +) -> None: """Clear all runtime overrides (or only selected keys) for LLM config.""" - _clear_runtime_overrides("llm", keys=keys) + _clear_runtime_overrides(_runtime_override_section(tenant_id), keys=keys) -def get_effective_config() -> Tuple[Dict[str, Any], Dict[str, str]]: +def get_effective_config( + tenant_id: Optional[str] = None, +) -> Tuple[Dict[str, Any], Dict[str, str]]: """ Get effective LLM config with precedence: ENV > runtime_override > persisted file > defaults. @@ -615,8 +696,10 @@ def get_effective_config() -> Tuple[Dict[str, Any], Dict[str, str]]: Returns: Tuple of (effective_config, sources) where sources maps each key to its origin. """ - file_config = _load_file_config().get("llm", {}) - runtime_overrides = get_runtime_overrides() + active_tenant = _resolve_active_tenant_id(tenant_id) + file_blob = _load_file_config() + file_config = _tenant_llm_config_view(file_blob, active_tenant) + runtime_overrides = get_runtime_overrides(active_tenant) ordered_keys = list(LLM_KEY_ORDER) + [ k for k in sorted(ALLOWED_LLM_KEYS) if k not in ENV_MAPPINGS @@ -883,7 +966,9 @@ def _merge_config_value(base: Any, patch: Any, key: str = "") -> Any: return patch -def update_config(updates: Dict[str, Any]) -> Tuple[bool, list]: +def update_config( + updates: Dict[str, Any], tenant_id: Optional[str] = None +) -> Tuple[bool, list]: """ Update LLM config, persisting to file. @@ -898,16 +983,32 @@ def update_config(updates: Dict[str, Any]) -> Tuple[bool, list]: if not sanitized: return True, [] # Nothing to update - # R94: Non-destructive merge with existing file config + tenant_id = _resolve_active_tenant_id(tenant_id) + + # R94/S49: Non-destructive merge with existing file config file_config = _load_file_config() - if "llm" not in file_config: - file_config["llm"] = {} + if tenant_id == DEFAULT_TENANT_ID: + if "llm" not in file_config: + file_config["llm"] = {} + target = file_config["llm"] + else: + tenants = file_config.get("tenants") + if not isinstance(tenants, dict): + tenants = {} + file_config["tenants"] = tenants + tenant_cfg = tenants.get(tenant_id) + if not isinstance(tenant_cfg, dict): + tenant_cfg = {} + tenants[tenant_id] = tenant_cfg + if "llm" not in tenant_cfg or not isinstance(tenant_cfg.get("llm"), dict): + tenant_cfg["llm"] = {} + target = tenant_cfg["llm"] for k, v in sanitized.items(): - file_config["llm"][k] = _merge_config_value(file_config["llm"].get(k), v, key=k) + target[k] = _merge_config_value(target.get(k), v, key=k) if _save_file_config(file_config): - logger.info(f"Updated config: {list(sanitized.keys())}") + logger.info("Updated config: %s (tenant=%s)", list(sanitized.keys()), tenant_id) return True, [] else: return False, ["Failed to save config file"] diff --git a/services/secret_providers.py b/services/secret_providers.py index be07bb9..4df9eaf 100644 --- a/services/secret_providers.py +++ b/services/secret_providers.py @@ -30,6 +30,21 @@ try: except ImportError: from services.providers.catalog import get_provider_info # type: ignore +try: + from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) +except ImportError: + from services.tenant_context import ( # type: ignore + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) + logger = logging.getLogger("ComfyUI-OpenClaw.services.secret_providers") _TRUTHY = {"1", "true", "yes", "on"} @@ -57,7 +72,7 @@ def _env_value( class SecretProvider(Protocol): source: str - def get_secret(self, provider: str) -> Optional[str]: ... + def get_secret(self, provider: str, tenant_id: str) -> Optional[str]: ... class EnvSecretProvider: @@ -74,7 +89,7 @@ class EnvSecretProvider: candidates.append(info.env_key_name) return candidates - def get_secret(self, provider: str) -> Optional[str]: + def get_secret(self, provider: str, tenant_id: str) -> Optional[str]: # Provider-specific first for env_name in self._provider_env_candidates(provider): value = os.environ.get(env_name) @@ -119,11 +134,14 @@ class OnePasswordSecretProvider: ).strip() def _item_template(self) -> str: + default_template = "openclaw/{provider}" + if is_multi_tenant_enabled(): + default_template = "openclaw/{tenant}/{provider}" return str( _env_value( "OPENCLAW_1PASSWORD_ITEM_TEMPLATE", "MOLTBOT_1PASSWORD_ITEM_TEMPLATE", - "openclaw/{provider}", + default_template, ) or "" ).strip() @@ -182,16 +200,21 @@ class OnePasswordSecretProvider: "S11: 1Password item template must include '{provider}'; fail-closed." ) return False + if is_multi_tenant_enabled() and "{tenant}" not in template: + logger.warning( + "S49/S11: 1Password item template must include '{tenant}' in multi-tenant mode; fail-closed." + ) + return False return True - def _build_ref(self, provider: str) -> Optional[str]: + def _build_ref(self, provider: str, tenant_id: str) -> Optional[str]: if not _PROVIDER_ID_RE.fullmatch(provider): logger.warning( "S11: Invalid provider id for 1Password lookup; fail-closed." ) return None template = self._item_template() - item = template.format(provider=provider) + item = template.format(provider=provider, tenant=tenant_id) if not item or not _ITEM_RE.fullmatch(item) or ".." in item: logger.warning("S11: 1Password item name is invalid; fail-closed.") return None @@ -237,18 +260,18 @@ class OnePasswordSecretProvider: value = (completed.stdout or "").strip() return value or None - def get_secret(self, provider: str) -> Optional[str]: + def get_secret(self, provider: str, tenant_id: str) -> Optional[str]: if not self.is_available(): return None - provider_ref = self._build_ref(provider) + provider_ref = self._build_ref(provider, tenant_id) if provider_ref: value = self._read_ref(provider_ref, provider) if value: return value if provider != "generic": - generic_ref = self._build_ref("generic") + generic_ref = self._build_ref("generic", tenant_id) if generic_ref: return self._read_ref(generic_ref, "generic") return None @@ -257,7 +280,7 @@ class OnePasswordSecretProvider: class ServerStoreSecretProvider: source = "server_store" - def get_secret(self, provider: str) -> Optional[str]: + def get_secret(self, provider: str, tenant_id: str) -> Optional[str]: try: from .secret_store import get_secret_store except ImportError: @@ -265,11 +288,11 @@ class ServerStoreSecretProvider: try: store = get_secret_store() - value = store.get_secret(provider) + value = store.get_secret(provider, tenant_id=tenant_id) if value: return value if provider != "generic": - return store.get_secret("generic") + return store.get_secret("generic", tenant_id=tenant_id) except Exception as exc: logger.debug( "S25/S11: secret store lookup failed (non-fatal): %s", @@ -287,15 +310,21 @@ def get_secret_providers() -> list[SecretProvider]: ] -def resolve_provider_secret(provider: str) -> tuple[Optional[str], Optional[str]]: +def resolve_provider_secret( + provider: str, tenant_id: Optional[str] = None +) -> tuple[Optional[str], Optional[str]]: """ Resolve a provider secret through the configured provider chain. Returns: (secret, source) where source in {"env","onepassword","server_store"} or None. """ + effective_tenant = normalize_tenant_id( + tenant_id or get_current_tenant_id() or DEFAULT_TENANT_ID + ) + for resolver in get_secret_providers(): - secret = resolver.get_secret(provider) + secret = resolver.get_secret(provider, effective_tenant) if secret: return secret, resolver.source return None, None diff --git a/services/secret_store.py b/services/secret_store.py index d6961fd..7a2bb72 100644 --- a/services/secret_store.py +++ b/services/secret_store.py @@ -22,6 +22,34 @@ import threading from pathlib import Path from typing import Any, Dict, Optional +try: + from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) +except ImportError: + try: + from services.tenant_context import ( # type: ignore + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, + ) + except ImportError: + DEFAULT_TENANT_ID = "default" + + def get_current_tenant_id(): # type: ignore + return DEFAULT_TENANT_ID + + def is_multi_tenant_enabled(): # type: ignore + return False + + def normalize_tenant_id(value): # type: ignore + return str(value or DEFAULT_TENANT_ID).strip().lower() or DEFAULT_TENANT_ID + + try: from .state_dir import get_state_dir except ImportError: @@ -76,6 +104,30 @@ class SecretStore: self._encryption_key = enc._load_or_create_key(self._state_dir) return self._encryption_key + def _resolve_tenant_id(self, tenant_id: Optional[str]) -> str: + if tenant_id is not None: + return normalize_tenant_id(tenant_id) + if is_multi_tenant_enabled(): + try: + return normalize_tenant_id(get_current_tenant_id()) + except Exception: + return DEFAULT_TENANT_ID + return DEFAULT_TENANT_ID + + def _tenant_key(self, provider_id: str, tenant_id: Optional[str] = None) -> str: + tenant = self._resolve_tenant_id(tenant_id) + if tenant == DEFAULT_TENANT_ID: + return provider_id + return f"tenant::{tenant}::{provider_id}" + + def _allow_legacy_fallback(self) -> bool: + value = ( + os.environ.get("OPENCLAW_MULTI_TENANT_ALLOW_LEGACY_SECRET_FALLBACK") + or os.environ.get("MOLTBOT_MULTI_TENANT_ALLOW_LEGACY_SECRET_FALLBACK") + or "0" + ) + return str(value).strip().lower() in ("1", "true", "yes", "on") + def _migrate_legacy(self) -> bool: """Migrate legacy plaintext secrets.json to encrypted format.""" if not self._legacy_path.exists(): @@ -178,7 +230,9 @@ class SecretStore: logger.error(f"S57: Failed to save encrypted store: {e}") raise - def get_secret(self, provider_id: str) -> Optional[str]: + def get_secret( + self, provider_id: str, tenant_id: Optional[str] = None + ) -> Optional[str]: """ Get secret for provider. @@ -188,10 +242,21 @@ class SecretStore: Returns: Secret value or None if not found """ + tenant = self._resolve_tenant_id(tenant_id) + scoped_key = self._tenant_key(provider_id, tenant) with self._lock: - return self._secrets.get(provider_id) + value = self._secrets.get(scoped_key) + if value: + return value + # S49: Optional compatibility fallback (explicit only) when tenant data + # has not been migrated yet. + if tenant != DEFAULT_TENANT_ID and self._allow_legacy_fallback(): + return self._secrets.get(provider_id) + return None - def set_secret(self, provider_id: str, secret: str) -> None: + def set_secret( + self, provider_id: str, secret: str, tenant_id: Optional[str] = None + ) -> None: """ Set secret for provider. @@ -202,14 +267,18 @@ class SecretStore: if not isinstance(secret, str) or not secret.strip(): raise ValueError("Secret must be non-empty string") + tenant = self._resolve_tenant_id(tenant_id) + scoped_key = self._tenant_key(provider_id, tenant) with self._lock: - self._secrets[provider_id] = secret.strip() + self._secrets[scoped_key] = secret.strip() self._save() # Never log secret value - logger.info(f"S25: Set secret for provider '{provider_id}'") + logger.info( + "S25/S49: Set secret for provider '%s' (tenant=%s)", provider_id, tenant + ) - def clear_secret(self, provider_id: str) -> bool: + def clear_secret(self, provider_id: str, tenant_id: Optional[str] = None) -> bool: """ Clear secret for provider. @@ -219,11 +288,17 @@ class SecretStore: Returns: True if secret was removed, False if not found """ + tenant = self._resolve_tenant_id(tenant_id) + scoped_key = self._tenant_key(provider_id, tenant) with self._lock: - if provider_id in self._secrets: - del self._secrets[provider_id] + if scoped_key in self._secrets: + del self._secrets[scoped_key] self._save() - logger.info(f"S25: Cleared secret for provider '{provider_id}'") + logger.info( + "S25/S49: Cleared secret for provider '%s' (tenant=%s)", + provider_id, + tenant, + ) return True return False @@ -241,16 +316,27 @@ class SecretStore: logger.info(f"S25: Cleared all secrets ({count} total)") return count - def get_status(self) -> Dict[str, Dict[str, Any]]: + def get_status(self, tenant_id: Optional[str] = None) -> Dict[str, Dict[str, Any]]: """ Get secret status (NO SECRET VALUES). Returns: Dict of {provider_id: {configured: bool, source: "server_store"}} """ + tenant = self._resolve_tenant_id(tenant_id) + prefix = f"tenant::{tenant}::" with self._lock: status = {} - for provider_id in self._secrets.keys(): + for key in self._secrets.keys(): + provider_id = None + if tenant == DEFAULT_TENANT_ID: + if key.startswith("tenant::"): + continue + provider_id = key + elif key.startswith(prefix): + provider_id = key[len(prefix) :] + if not provider_id: + continue status[provider_id] = {"configured": True, "source": "server_store"} return status diff --git a/services/templates.py b/services/templates.py index 004467a..7d17129 100644 --- a/services/templates.py +++ b/services/templates.py @@ -10,10 +10,16 @@ Loads manifest and renders templates for execution. import copy import logging import os -from dataclasses import dataclass +from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from .safe_io import resolve_under_root, safe_read_json +from .tenant_context import ( + DEFAULT_TENANT_ID, + get_current_tenant_id, + is_multi_tenant_enabled, + normalize_tenant_id, +) logger = logging.getLogger("ComfyUI-OpenClaw.services.templates") @@ -31,6 +37,7 @@ class TemplateConfig: path: str allowed_inputs: List[str] defaults: Dict[str, Any] + tenants: List[str] = field(default_factory=list) class TemplateService: @@ -105,10 +112,25 @@ class TemplateService: self.manifest.clear() for t_id, t_cfg in data.get("templates", {}).items(): + tenants = t_cfg.get("tenants", []) + if not isinstance(tenants, list): + tenants = [] + normalized_tenants: List[str] = [] + for value in tenants: + try: + normalized_tenants.append( + normalize_tenant_id(value, field_name=f"{t_id}.tenants") + ) + except Exception: + logger.warning( + "S49: ignoring invalid template tenant binding for %s", + t_id, + ) self.manifest[t_id] = TemplateConfig( path=t_cfg["path"], allowed_inputs=t_cfg.get("allowed_inputs", []), defaults=t_cfg.get("defaults", {}), + tenants=normalized_tenants, ) try: self._manifest_mtime = os.path.getmtime(self._manifest_abspath) @@ -146,11 +168,26 @@ class TemplateService: def get_template_config(self, template_id: str) -> Optional[TemplateConfig]: self._maybe_reload_manifest() + try: + tenant_id = normalize_tenant_id(get_current_tenant_id()) + except Exception: + tenant_id = DEFAULT_TENANT_ID cfg = self.manifest.get(template_id) if cfg is not None: + if ( + is_multi_tenant_enabled() + and cfg.tenants + and tenant_id not in cfg.tenants + ): + return None return cfg # No manifest entry: treat `.json` as runnable if present. + if is_multi_tenant_enabled(): + # IMPORTANT: in multi-tenant mode, discovery-only templates are hidden by + # default because they cannot express explicit tenant visibility. + return None + rel_path = f"{template_id}.json" try: abs_path = resolve_under_root(self.templates_root, rel_path) @@ -160,7 +197,7 @@ class TemplateService: if not os.path.isfile(abs_path): return None - return TemplateConfig(path=rel_path, allowed_inputs=[], defaults={}) + return TemplateConfig(path=rel_path, allowed_inputs=[], defaults={}, tenants=[]) def render_template( self, template_id: str, inputs: Dict[str, Any] diff --git a/services/tenant_context.py b/services/tenant_context.py new file mode 100644 index 0000000..9834263 --- /dev/null +++ b/services/tenant_context.py @@ -0,0 +1,204 @@ +""" +S49 multi-tenant boundary context. + +Central contract for tenant resolution, validation, and async context propagation. +Default mode remains single-tenant compatible. +""" + +from __future__ import annotations + +import contextlib +import contextvars +import os +import re +from dataclasses import dataclass +from typing import Any, Mapping, Optional + +DEFAULT_TENANT_ID = "default" +TENANT_ID_RE = re.compile(r"^[a-z0-9][a-z0-9._-]{0,63}$") +_TRUTHY = {"1", "true", "yes", "on"} + +_CURRENT_TENANT: contextvars.ContextVar[str] = contextvars.ContextVar( + "openclaw_current_tenant", default=DEFAULT_TENANT_ID +) + + +class TenantBoundaryError(ValueError): + """Raised when tenant context cannot be resolved or verified.""" + + def __init__(self, code: str, message: str): + super().__init__(message) + self.code = code + + +@dataclass(frozen=True) +class TenantContext: + tenant_id: str = DEFAULT_TENANT_ID + source: str = "default" + multi_tenant: bool = False + + +def _is_truthy(value: Optional[str]) -> bool: + if value is None: + return False + return value.strip().lower() in _TRUTHY + + +def is_multi_tenant_enabled() -> bool: + return _is_truthy( + os.environ.get("OPENCLAW_MULTI_TENANT_ENABLED") + or os.environ.get("MOLTBOT_MULTI_TENANT_ENABLED") + or "0" + ) + + +def allow_default_tenant_fallback() -> bool: + return _is_truthy( + os.environ.get("OPENCLAW_MULTI_TENANT_ALLOW_DEFAULT_FALLBACK") + or os.environ.get("MOLTBOT_MULTI_TENANT_ALLOW_DEFAULT_FALLBACK") + or "0" + ) + + +def normalize_tenant_id(tenant_id: Any, *, field_name: str = "tenant_id") -> str: + text = str(tenant_id or "").strip().lower() + if not text: + raise TenantBoundaryError("tenant_invalid", f"{field_name} must be non-empty") + if not TENANT_ID_RE.fullmatch(text): + raise TenantBoundaryError( + "tenant_invalid", + f"{field_name} must match {TENANT_ID_RE.pattern}", + ) + return text + + +def get_tenant_header_names() -> tuple[str, ...]: + configured = ( + os.environ.get("OPENCLAW_TENANT_HEADER") + or os.environ.get("MOLTBOT_TENANT_HEADER") + or "X-OpenClaw-Tenant-Id" + ).strip() + # Keep explicit legacy compatibility fallback. + return tuple( + dict.fromkeys( + [configured, "X-OpenClaw-Tenant-Id", "X-Moltbot-Tenant-Id"] + ).keys() + ) + + +def extract_tenant_from_headers(headers: Mapping[str, Any]) -> Optional[str]: + for key in get_tenant_header_names(): + raw = headers.get(key) if headers else None + if raw is None: + continue + value = str(raw).strip() + if value: + return normalize_tenant_id(value, field_name=key) + return None + + +def resolve_tenant_context( + *, + request: Optional[Any] = None, + token_info: Optional[Any] = None, + allow_default_when_missing: bool = False, +) -> TenantContext: + """ + Resolve tenant context from token + request headers. + + Resolution order (multi-tenant mode): + 1) token_info.tenant_id + 2) request tenant header + mismatch => fail-closed. + """ + multi_tenant = is_multi_tenant_enabled() + if not multi_tenant: + return TenantContext( + tenant_id=DEFAULT_TENANT_ID, + source="single_tenant_mode", + multi_tenant=False, + ) + + token_tenant = None + if token_info is not None and getattr(token_info, "tenant_id", None): + token_tenant = normalize_tenant_id( + getattr(token_info, "tenant_id"), field_name="token_tenant_id" + ) + + header_tenant = None + if request is not None and getattr(request, "headers", None) is not None: + header_tenant = extract_tenant_from_headers(request.headers) + + if token_tenant and header_tenant and token_tenant != header_tenant: + raise TenantBoundaryError( + "tenant_mismatch", + "Tenant mismatch between token context and request header.", + ) + + if token_tenant: + return TenantContext( + tenant_id=token_tenant, + source="token", + multi_tenant=True, + ) + if header_tenant: + return TenantContext( + tenant_id=header_tenant, + source="header", + multi_tenant=True, + ) + + if allow_default_when_missing or allow_default_tenant_fallback(): + return TenantContext( + tenant_id=DEFAULT_TENANT_ID, + source="default_fallback", + multi_tenant=True, + ) + + raise TenantBoundaryError( + "tenant_required", + "Tenant context required in multi-tenant mode.", + ) + + +def get_current_tenant_id() -> str: + if not is_multi_tenant_enabled(): + return DEFAULT_TENANT_ID + tenant_id = _CURRENT_TENANT.get() + if not tenant_id: + return DEFAULT_TENANT_ID + return tenant_id + + +@contextlib.contextmanager +def tenant_scope(tenant_id: str): + """Set tenant context in a contextvar scope (async-safe).""" + normalized = normalize_tenant_id(tenant_id) + token = _CURRENT_TENANT.set(normalized) + try: + yield normalized + finally: + _CURRENT_TENANT.reset(token) + + +@contextlib.contextmanager +def request_tenant_scope( + *, + request: Optional[Any] = None, + token_info: Optional[Any] = None, + allow_default_when_missing: bool = False, +): + """ + Resolve + bind tenant context for current request processing scope. + """ + ctx = resolve_tenant_context( + request=request, + token_info=token_info, + allow_default_when_missing=allow_default_when_missing, + ) + token = _CURRENT_TENANT.set(ctx.tenant_id) + try: + yield ctx + finally: + _CURRENT_TENANT.reset(token) + diff --git a/tests/test_approvals.py b/tests/test_approvals.py index 0b73771..107f249 100644 --- a/tests/test_approvals.py +++ b/tests/test_approvals.py @@ -9,6 +9,7 @@ import time import unittest from datetime import datetime, timedelta, timezone from pathlib import Path +from unittest.mock import patch # Set up test state directory _repo_root = Path(__file__).resolve().parent.parent @@ -293,6 +294,22 @@ class TestApprovalService(unittest.TestCase): self.assertEqual(rejected.status, ApprovalStatus.REJECTED) + def test_multi_tenant_request_isolation(self): + """S49: approval service should deny cross-tenant object access.""" + from services.approvals.service import ApprovalService + + service = ApprovalService() + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + request = service.create_request( + template_id="template_tenant_test", + inputs={}, + tenant_id="tenant-a", + ) + + self.assertIsNone(service.get(request.approval_id, tenant_id="tenant-b")) + with self.assertRaises(ValueError): + service.approve(request.approval_id, tenant_id="tenant-b") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_connector_installation_registry.py b/tests/test_connector_installation_registry.py index 2dc858d..78d79de 100644 --- a/tests/test_connector_installation_registry.py +++ b/tests/test_connector_installation_registry.py @@ -1,6 +1,7 @@ import os import tempfile import unittest +from unittest.mock import patch from services.connector_installation_registry import ( ConnectorInstallationRegistry, @@ -137,6 +138,28 @@ class TestConnectorInstallationRegistry(unittest.TestCase): ).read() self.assertNotIn("xoxb-secret", raw) + def test_multi_tenant_resolve_mismatch_fail_closed(self): + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + self.registry.upsert_installation( + platform="slack", + tenant_id="tenant-a", + workspace_id="T1", + installation_id="inst-tenant", + token_values={"bot_token": "xoxb-tenant"}, + status=InstallationStatus.ACTIVE.value, + ) + + mismatch = self.registry.resolve_installation( + "slack", "T1", tenant_id="tenant-b" + ) + self.assertFalse(mismatch.ok) + self.assertEqual(mismatch.reject_reason, "tenant_mismatch") + + matched = self.registry.resolve_installation( + "slack", "T1", tenant_id="tenant-a" + ) + self.assertTrue(matched.ok) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_execution_budgets.py b/tests/test_execution_budgets.py index 4a03c95..f7e5a06 100644 --- a/tests/test_execution_budgets.py +++ b/tests/test_execution_budgets.py @@ -33,6 +33,7 @@ class TestBudgetConfig(unittest.TestCase): self.assertEqual(config.max_inflight_trigger, 1) self.assertEqual(config.max_inflight_scheduler, 1) self.assertEqual(config.max_inflight_bridge, 1) + self.assertEqual(config.max_inflight_per_tenant, 1) self.assertEqual(config.max_rendered_workflow_bytes, 512 * 1024) @patch.dict( @@ -211,6 +212,37 @@ class TestExecutionBudgetLimiter(unittest.IsolatedAsyncioTestCase): self.assertEqual(stats["total"], 1) self.assertEqual(stats["unknown"], 1) + async def test_tenant_concurrency_cap_in_multi_tenant_mode(self): + """S49: per-tenant concurrency cap should be fail-closed.""" + from services.execution_budgets import BudgetConfig + + config = BudgetConfig( + max_inflight_total=10, + max_inflight_webhook=10, + max_inflight_trigger=10, + max_inflight_scheduler=10, + max_inflight_bridge=10, + max_inflight_per_tenant=1, + max_rendered_workflow_bytes=512 * 1024, + ) + limiter = ExecutionBudgetLimiter(config) + + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + async with limiter.acquire("webhook", trace_id="trc_1", tenant_id="team-a"): + with self.assertRaises(BudgetExceededError) as ctx: + async with limiter.acquire( + "trigger", trace_id="trc_2", tenant_id="team-a" + ): + pass + self.assertEqual(ctx.exception.budget_type, "tenant_concurrency") + + # Different tenant should still be allowed under same global budget. + async with limiter.acquire( + "trigger", trace_id="trc_3", tenant_id="team-b" + ): + stats = limiter.get_stats() + self.assertEqual(stats["total"], 2) + class TestGlobalLimiterSingleton(unittest.TestCase): """Test global limiter singleton.""" diff --git a/tests/test_presets.py b/tests/test_presets.py index fa0b1db..73a5cbc 100644 --- a/tests/test_presets.py +++ b/tests/test_presets.py @@ -7,6 +7,7 @@ import shutil import tempfile import unittest from pathlib import Path +from unittest.mock import patch from services.presets.models import Preset from services.presets.storage import PresetStore @@ -77,6 +78,23 @@ class TestPresetStorage(unittest.TestCase): loaded = store2.get_preset(p.id) self.assertEqual(loaded.name, "Persistent") + def test_multi_tenant_visibility_filter(self): + """S49: preset visibility must be tenant-isolated in multi-tenant mode.""" + p1 = Preset.new("A", {}) + p1.tenant_id = "tenant-a" + p2 = Preset.new("B", {}) + p2.tenant_id = "tenant-b" + self.store.save_preset(p1) + self.store.save_preset(p2) + + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + tenant_a = self.store.list_presets(tenant_id="tenant-a") + self.assertEqual(len(tenant_a), 1) + self.assertEqual(tenant_a[0].tenant_id, "tenant-a") + + self.assertIsNone(self.store.get_preset(p2.id, tenant_id="tenant-a")) + self.assertFalse(self.store.delete_preset(p2.id, tenant_id="tenant-a")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_runtime_config.py b/tests/test_runtime_config.py index 04e7a9b..288f085 100644 --- a/tests/test_runtime_config.py +++ b/tests/test_runtime_config.py @@ -291,6 +291,30 @@ class TestRuntimeConfig(unittest.TestCase): with patch.dict(os.environ, {"OPENCLAW_ADMIN_TOKEN": "newsecret"}): self.assertTrue(validate_admin_token("newsecret")) + def test_multi_tenant_config_namespace_isolation(self): + """S49: persisted/runtime config should isolate tenant branches.""" + from services.runtime_config import get_effective_config, update_config + from services.tenant_context import tenant_scope + + with patch.dict(os.environ, {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with tenant_scope("tenant-a"): + ok, errors = update_config({"model": "model-a"}) + self.assertTrue(ok) + self.assertEqual(errors, []) + effective_a, _ = get_effective_config() + self.assertEqual(effective_a["model"], "model-a") + + with tenant_scope("tenant-b"): + ok, errors = update_config({"model": "model-b"}) + self.assertTrue(ok) + self.assertEqual(errors, []) + effective_b, _ = get_effective_config() + self.assertEqual(effective_b["model"], "model-b") + + with tenant_scope("tenant-a"): + effective_a, _ = get_effective_config() + self.assertEqual(effective_a["model"], "model-a") + if __name__ == "__main__": unittest.main() diff --git a/tests/test_s25_secret_store.py b/tests/test_s25_secret_store.py index 7ad8256..baafcaf 100644 --- a/tests/test_s25_secret_store.py +++ b/tests/test_s25_secret_store.py @@ -17,6 +17,7 @@ import sys import tempfile import unittest from pathlib import Path +from unittest.mock import patch sys.path.insert(0, str(Path(__file__).parent.parent)) @@ -230,6 +231,25 @@ class TestKeyResolution(unittest.TestCase): key = get_api_key_for_provider("openai") self.assertEqual(key, "sk-store-key") + def test_multi_tenant_secret_isolation(self): + """S49: secret store should isolate provider secrets per tenant.""" + from services.secret_store import SecretStore + + store = SecretStore(state_dir=self.test_dir) + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + store.set_secret("openai", "sk-tenant-a", tenant_id="tenant-a") + store.set_secret("openai", "sk-tenant-b", tenant_id="tenant-b") + + self.assertEqual( + store.get_secret("openai", tenant_id="tenant-a"), + "sk-tenant-a", + ) + self.assertEqual( + store.get_secret("openai", tenant_id="tenant-b"), + "sk-tenant-b", + ) + self.assertIsNone(store.get_secret("openai", tenant_id="tenant-c")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_s49_tenant_context.py b/tests/test_s49_tenant_context.py new file mode 100644 index 0000000..3ddc6ae --- /dev/null +++ b/tests/test_s49_tenant_context.py @@ -0,0 +1,61 @@ +import asyncio +import unittest +from unittest.mock import patch + +from services.tenant_context import ( + DEFAULT_TENANT_ID, + TenantBoundaryError, + get_current_tenant_id, + request_tenant_scope, + resolve_tenant_context, + tenant_scope, +) + + +class _Req: + def __init__(self, headers=None): + self.headers = headers or {} + + +class _Token: + def __init__(self, tenant_id): + self.tenant_id = tenant_id + + +class TestTenantContext(unittest.IsolatedAsyncioTestCase): + def test_single_tenant_default(self): + ctx = resolve_tenant_context(request=_Req({"X-OpenClaw-Tenant-Id": "team-a"})) + self.assertEqual(ctx.tenant_id, DEFAULT_TENANT_ID) + + def test_multi_tenant_header_resolution(self): + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + ctx = resolve_tenant_context(request=_Req({"X-OpenClaw-Tenant-Id": "team-a"})) + self.assertEqual(ctx.tenant_id, "team-a") + + def test_multi_tenant_mismatch_fail_closed(self): + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with self.assertRaises(TenantBoundaryError) as exc: + resolve_tenant_context( + request=_Req({"X-OpenClaw-Tenant-Id": "team-b"}), + token_info=_Token("team-a"), + ) + self.assertEqual(exc.exception.code, "tenant_mismatch") + + async def test_async_context_propagation(self): + async def _read_tenant(): + await asyncio.sleep(0) + return get_current_tenant_id() + + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with tenant_scope("team-a"): + value = await _read_tenant() + self.assertEqual(value, "team-a") + + def test_request_scope_sets_current_tenant(self): + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with request_tenant_scope(request=_Req({"X-OpenClaw-Tenant-Id": "team-a"})): + self.assertEqual(get_current_tenant_id(), "team-a") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_templates.py b/tests/test_templates.py index ff9d840..703813b 100644 --- a/tests/test_templates.py +++ b/tests/test_templates.py @@ -13,6 +13,7 @@ from unittest.mock import MagicMock, patch sys.path.append(os.getcwd()) from services.templates import TemplateService +from services.tenant_context import tenant_scope class TestTemplateService(unittest.TestCase): @@ -84,6 +85,34 @@ class TestTemplateService(unittest.TestCase): rendered = self.service.render_template("t2", {"input_any": "hello"}) self.assertEqual(rendered["node1"]["inputs"]["text"], "hello") + def test_multi_tenant_manifest_visibility(self): + """S49: manifest tenant bindings should gate visibility in multi-tenant mode.""" + manifest = { + "version": 1, + "templates": { + "t1": { + "path": "t1.json", + "allowed_inputs": ["input1"], + "tenants": ["tenant-a"], + } + }, + } + with open(os.path.join(self.test_dir, "manifest.json"), "w") as f: + json.dump(manifest, f) + self.service._load_manifest() + + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with tenant_scope("tenant-a"): + self.assertIsNotNone(self.service.get_template_config("t1")) + with tenant_scope("tenant-b"): + self.assertIsNone(self.service.get_template_config("t1")) + + def test_multi_tenant_hides_discovery_only_templates(self): + """S49: discovery-only templates are hidden in multi-tenant mode.""" + with patch.dict("os.environ", {"OPENCLAW_MULTI_TENANT_ENABLED": "1"}): + with tenant_scope("tenant-a"): + self.assertIsNone(self.service.get_template_config("t2")) + if __name__ == "__main__": unittest.main()