""" S25: Secrets Management API Admin-gated endpoints for managing server-side secrets (API keys, tokens). Security model: - Requires valid Admin Token - Loopback-only unless OPENCLAW_ALLOW_REMOTE_ADMIN=1 - Never returns actual secret values - Redacts request bodies containing secrets Endpoints: - GET /openclaw/secrets/status: Secret configuration status (no values) - PUT /openclaw/secrets: Save API key to server store - DELETE /openclaw/secrets/{provider}: Clear provider secret """ import logging import os from typing import Optional from aiohttp import web # Import discipline: # - ComfyUI runtime: package-relative imports only. # - Unit tests: allow top-level fallbacks. if __package__ and "." in __package__: from ..models.schemas import MAX_BODY_SIZE from ..services.access_control import require_admin_token from ..services.audit import audit_secret_delete, audit_secret_write from ..services.csrf_protection import require_same_origin_if_no_token from ..services.metrics import metrics from ..services.rate_limit import check_rate_limit from ..services.request_ip import get_client_ip from ..services.runtime_config import get_admin_token, is_loopback_client from ..services.secret_store import get_secret_store else: # pragma: no cover (test-only import mode) from models.schemas import MAX_BODY_SIZE # type: ignore from services.access_control import require_admin_token # type: ignore from services.audit import audit_secret_delete # type: ignore from services.audit import audit_secret_write from services.csrf_protection import require_same_origin_if_no_token # type: ignore from services.metrics import metrics # type: ignore from services.rate_limit import check_rate_limit # type: ignore from services.request_ip import get_client_ip # type: ignore from services.runtime_config import get_admin_token # type: ignore from services.runtime_config import is_loopback_client from services.secret_store import get_secret_store # type: ignore logger = logging.getLogger("ComfyUI-OpenClaw.api.secrets") _TRUTHY = ("1", "true", "yes", "on") def _deny_if_remote_admin_not_allowed(request: web.Request) -> Optional[web.Response]: allow_remote = ( os.environ.get("OPENCLAW_ALLOW_REMOTE_ADMIN") or os.environ.get("MOLTBOT_ALLOW_REMOTE_ADMIN") or "" ).lower() if allow_remote in _TRUTHY: return None remote = request.remote or "" if not is_loopback_client(remote): return web.json_response( { "ok": False, "error": "remote_admin_denied", "message": "Remote admin access denied. Set OPENCLAW_ALLOW_REMOTE_ADMIN=1 (or legacy MOLTBOT_ALLOW_REMOTE_ADMIN=1) to allow.", }, status=403, ) return None def _require_admin(request: web.Request) -> Optional[web.Response]: allowed, error = require_admin_token(request) if not allowed: return web.json_response( {"ok": False, "error": error or "Unauthorized"}, status=403 ) return None def _rate_limit_admin(request: web.Request) -> Optional[web.Response]: if check_rate_limit(request, "admin"): return None metrics.increment("rate_limit_exceeded") return web.json_response( {"ok": False, "error": "rate_limit_exceeded"}, status=429, headers={"Retry-After": "60"}, ) async def secrets_status_handler(request: web.Request) -> web.Response: """ GET /openclaw/secrets/status Returns secret configuration status (NO ACTUAL VALUES). Security: - Admin-gated - Rate-limited """ resp = _deny_if_remote_admin_not_allowed(request) if resp: return resp resp = _require_admin(request) if resp: return resp resp = _rate_limit_admin(request) if resp: return resp # Get status (no secret values) store = get_secret_store() status = store.get_status() return web.json_response( { "ok": True, "secrets": status, "warning": "Secrets stored server-side. Recommended for localhost-only usage.", } ) async def secrets_put_handler(request: web.Request) -> web.Response: """ PUT /openclaw/secrets Save API key to server store. Body: {"provider": "openai"|"anthropic"|"generic", "api_key": ""} Security: - Admin-gated - Rate-limited - CSRF-protected (same-origin if no token) - Request body NEVER logged """ # S26+: CSRF protection for convenience mode admin_token_configured = bool(get_admin_token()) resp = require_same_origin_if_no_token(request, admin_token_configured) if resp: return resp resp = _deny_if_remote_admin_not_allowed(request) if resp: return resp resp = _require_admin(request) if resp: return resp resp = _rate_limit_admin(request) if resp: return resp if request.content_length and request.content_length > MAX_BODY_SIZE: return web.json_response( {"ok": False, "error": "payload_too_large"}, status=413 ) # Parse body (do NOT log) try: body = await request.json() except Exception: return web.json_response({"ok": False, "error": "invalid_json"}, status=400) # Validate required fields provider = body.get("provider") api_key = body.get("api_key") if not provider or not isinstance(provider, str): return web.json_response( { "ok": False, "error": "missing_provider", "message": "Field 'provider' is required (string)", }, status=400, ) provider = provider.strip().lower() if not provider or len(provider) > 64: return web.json_response({"ok": False, "error": "invalid_provider"}, status=400) if not api_key or not isinstance(api_key, str): return web.json_response( { "ok": False, "error": "missing_api_key", "message": "Field 'api_key' is required (non-empty string)", }, status=400, ) # Size check if len(api_key) > 4096: # Reasonable key size limit return web.json_response( { "ok": False, "error": "api_key_too_long", "message": "API key exceeds maximum length (4KB)", }, status=400, ) # Save to store actor_ip = get_client_ip(request) try: store = get_secret_store() store.set_secret(provider, api_key) # Never log the secret value logger.info(f"S25: Saved secret for provider '{provider}' via UI") # S26+: Audit event (no secrets) audit_secret_write(actor_ip, provider, ok=True) return web.json_response( {"ok": True, "message": f"Secret saved for provider '{provider}'"} ) except Exception as e: logger.error(f"S25: Failed to save secret: {e}") audit_secret_write(actor_ip, provider, ok=False, error=str(e)) return web.json_response( {"ok": False, "error": "save_failed", "message": str(e)}, status=500 ) async def secrets_delete_handler(request: web.Request) -> web.Response: """ DELETE /openclaw/secrets/{provider} Clear provider secret. Security: - Admin-gated - Rate-limited - CSRF-protected (same-origin if no token) """ # S26+: CSRF protection for convenience mode admin_token_configured = bool(get_admin_token()) resp = require_same_origin_if_no_token(request, admin_token_configured) if resp: return resp resp = _deny_if_remote_admin_not_allowed(request) if resp: return resp resp = _require_admin(request) if resp: return resp resp = _rate_limit_admin(request) if resp: return resp # Get provider from path provider = request.match_info.get("provider") if not provider: return web.json_response({"ok": False, "error": "missing_provider"}, status=400) provider = provider.strip().lower() if not provider or len(provider) > 64: return web.json_response({"ok": False, "error": "invalid_provider"}, status=400) # Clear from store actor_ip = get_client_ip(request) try: store = get_secret_store() removed = store.clear_secret(provider) if removed: logger.info(f"S25: Cleared secret for provider '{provider}' via UI") audit_secret_delete(actor_ip, provider, ok=True) return web.json_response( {"ok": True, "message": f"Secret cleared for provider '{provider}'"} ) else: audit_secret_delete(actor_ip, provider, ok=False, error="not_found") return web.json_response( { "ok": False, "error": "not_found", "message": f"No secret found for provider '{provider}'", }, status=404, ) except Exception as e: logger.error(f"S25: Failed to clear secret: {e}") audit_secret_delete(actor_ip, provider, ok=False, error=str(e)) return web.json_response( {"ok": False, "error": "clear_failed", "message": str(e)}, status=500 )