ComfyUI-OpenClaw/api/config.py

"""
Config API handlers (R21/S13/F20).
Provides GET/PUT /moltbot/config and optional /moltbot/llm/test.
"""

from __future__ import annotations

import json
import logging

if __package__ and "." in __package__:
    from ..services.import_fallback import import_attrs_dual, import_module_dual
else:
    from services.import_fallback import (  # type: ignore
        import_attrs_dual,
        import_module_dual,
    )

try:
    (PACK_VERSION,) = import_attrs_dual(
        __package__,
        "..config",
        "config",
        ("PACK_VERSION",),
    )
except ImportError:  # pragma: no cover
    PACK_VERSION = "0.1.0"

try:
    from aiohttp import web
except ImportError:  # pragma: no cover (optional for unit tests)
    # CRITICAL test/CI fallback:
    # Some CI/unit environments intentionally run without aiohttp installed.
    # Keep this module importable by providing a minimal `web` shim used by
    # handler tests (json_response/status/body), while production keeps real aiohttp.
    class _MockResponse:
        def __init__(
            self, payload: dict, status: int = 200, headers: dict | None = None
        ):
            self.status = status
            self.headers = headers or {}
            self.body = json.dumps(payload).encode("utf-8")

    class _MockWeb:
        _IS_MOCKWEB = True

        class Request:  # pragma: no cover - typing shim only
            pass

        class Response:  # pragma: no cover - typing shim only
            pass

        @staticmethod
        def json_response(
            payload: dict, status: int = 200, headers: dict | None = None
        ):
            return _MockResponse(payload, status=status, headers=headers)

    web = _MockWeb()  # type: ignore

# Import discipline:
# - In real ComfyUI runtimes, this pack is loaded as a package and must use package-relative imports.
# - In unit tests, modules may be imported as top-level (e.g., `api.*`), so we allow top-level fallbacks.
(
    is_loopback,
    require_admin_token,
    require_observability_access,
    resolve_token_info,
) = import_attrs_dual(
    __package__,
    "..services.access_control",
    "services.access_control",
    (
        "is_loopback",
        "require_admin_token",
        "require_observability_access",
        "resolve_token_info",
    ),
)
(emit_audit_event,) = import_attrs_dual(
    __package__,
    "..services.audit",
    "services.audit",
    ("emit_audit_event",),
)

try:
    (require_same_origin_if_no_token,) = import_attrs_dual(
        __package__,
        "..services.csrf_protection",
        "services.csrf_protection",
        ("require_same_origin_if_no_token",),
    )
except Exception:
    # CRITICAL test/CI fallback (DO NOT replace with a direct import):
    # Some unit-test environments import `api.config` without aiohttp installed.
    # `services.csrf_protection` imports aiohttp at module load, which can raise
    # ModuleNotFoundError and break unrelated tests (`test_r53`, `test_r60`).
    # Keep import-time behavior resilient by using a no-op guard in that case.
    def require_same_origin_if_no_token(*_args, **_kwargs):  # type: ignore
        return None


(LLMClient,) = import_attrs_dual(
    __package__,
    "..services.llm_client",
    "services.llm_client",
    ("LLMClient",),
)
(check_rate_limit, build_rate_limit_response) = import_attrs_dual(
    __package__,
    "..services.rate_limit",
    "services.rate_limit",
    ("check_rate_limit", "build_rate_limit_response"),
)
(get_client_ip,) = import_attrs_dual(
    __package__,
    "..services.request_ip",
    "services.request_ip",
    ("get_client_ip",),
)
(
    ALLOWED_LLM_KEYS,
    get_admin_token,
    get_apply_semantics,
    get_effective_config,
    get_llm_egress_controls,
    get_runtime_guardrails,
    get_settings_schema,
    is_loopback_client,
    update_config,
) = import_attrs_dual(
    __package__,
    "..services.runtime_config",
    "services.runtime_config",
    (
        "ALLOWED_LLM_KEYS",
        "get_admin_token",
        "get_apply_semantics",
        "get_effective_config",
        "get_llm_egress_controls",
        "get_runtime_guardrails",
        "get_settings_schema",
        "is_loopback_client",
        "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,
) = import_attrs_dual(
    __package__,
    "..services.runtime_guardrails",
    "services.runtime_guardrails",
    ("CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN", "payload_contains_runtime_guardrails"),
)

logger = logging.getLogger("ComfyUI-OpenClaw.api.config")

(
    ConfigHandlerDependencies,
    config_get_response,
    config_put_response,
) = import_attrs_dual(
    __package__,
    "..api.config_projection_handlers",
    "api.config_projection_handlers",
    ("ConfigHandlerDependencies", "config_get_response", "config_put_response"),
)
(llm_models_response,) = import_attrs_dual(
    __package__,
    "..api.config_model_handlers",
    "api.config_model_handlers",
    ("llm_models_response",),
)
(llm_chat_response, llm_test_response) = import_attrs_dual(
    __package__,
    "..api.config_llm_handlers",
    "api.config_llm_handlers",
    ("llm_chat_response", "llm_test_response"),
)

(
    _MODEL_LIST_CACHE,
    _MODEL_LIST_MAX_ENTRIES,
    _MODEL_LIST_TTL_SEC,
    _build_model_cache_key,
    _cache_get,
    _cache_put,
    _extract_models_from_payload,
    _format_llm_ssrf_error,
    _get_llm_allowed_hosts,
    _llm_insecure_override_enabled,
    fetch_remote_model_list,
    get_stale_cached_models,
    resolve_model_list_target,
    validate_model_list_target,
) = import_attrs_dual(
    __package__,
    "..services.llm_model_list",
    "services.llm_model_list",
    (
        "_MODEL_LIST_CACHE",
        "_MODEL_LIST_MAX_ENTRIES",
        "_MODEL_LIST_TTL_SEC",
        "build_model_cache_key",
        "cache_get",
        "cache_put",
        "extract_models_from_payload",
        "format_llm_ssrf_error",
        "get_llm_allowed_hosts",
        "llm_insecure_override_enabled",
        "fetch_remote_model_list",
        "get_stale_cached_models",
        "resolve_model_list_target",
        "validate_model_list_target",
    ),
)


# S14/R98 / R64: Import Endpoint Metadata
(
    AuthTier,
    RiskTier,
    RoutePlane,
    endpoint_metadata,
) = import_attrs_dual(
    __package__,
    "..services.endpoint_manifest",
    "services.endpoint_manifest",
    ("AuthTier", "RiskTier", "RoutePlane", "endpoint_metadata"),
)


# Provider catalog for UI dropdown (R16 dynamic)
PROVIDER_CATALOG = []

try:
    raw_catalog_module = import_module_dual(
        __package__,
        "..services.providers.catalog",
        "services.providers.catalog",
    )
    RAW_CATALOG = raw_catalog_module.PROVIDER_CATALOG

    for pid, info in RAW_CATALOG.items():
        PROVIDER_CATALOG.append(
            {
                "id": pid,
                "label": info.name,
                "requires_key": info.env_key_name is not None,
            }
        )
    # Ensure custom is present if not in catalog (though it is)
    if not any(p["id"] == "custom" for p in PROVIDER_CATALOG):
        PROVIDER_CATALOG.append(
            {"id": "custom", "label": "Custom OpenAI-compatible", "requires_key": True}
        )
except ImportError:
    # Fallback if catalog module missing
    PROVIDER_CATALOG = [
        {"id": "openai", "label": "OpenAI", "requires_key": True},
        {"id": "anthropic", "label": "Anthropic", "requires_key": True},
        {"id": "openrouter", "label": "OpenRouter", "requires_key": True},
        {"id": "gemini", "label": "Google Gemini", "requires_key": True},
        {"id": "groq", "label": "Groq", "requires_key": True},
        {"id": "deepseek", "label": "DeepSeek", "requires_key": True},
        {"id": "xai", "label": "xAI (Grok)", "requires_key": True},
        {"id": "ollama", "label": "Ollama (Local)", "requires_key": False},
        {"id": "lmstudio", "label": "LM Studio (Local)", "requires_key": False},
        {"id": "custom", "label": "Custom OpenAI-compatible", "requires_key": True},
    ]


def _handler_dependencies():
    """Capture established facade patch seams for owned config handlers."""

    return ConfigHandlerDependencies(
        web=web,
        logger=logger,
        provider_catalog=PROVIDER_CATALOG,
        pack_version=PACK_VERSION,
        require_observability_access=require_observability_access,
        require_admin_token=require_admin_token,
        require_same_origin_if_no_token=require_same_origin_if_no_token,
        resolve_token_info=resolve_token_info,
        emit_audit_event=emit_audit_event,
        check_rate_limit=check_rate_limit,
        build_rate_limit_response=build_rate_limit_response,
        get_client_ip=get_client_ip,
        is_loopback=is_loopback,
        get_admin_token=get_admin_token,
        get_apply_semantics=get_apply_semantics,
        get_effective_config=get_effective_config,
        get_llm_egress_controls=get_llm_egress_controls,
        get_runtime_guardrails=get_runtime_guardrails,
        get_settings_schema=get_settings_schema,
        is_loopback_client=is_loopback_client,
        update_config=update_config,
        tenant_boundary_error=TenantBoundaryError,
        request_tenant_scope=request_tenant_scope,
        runtime_only_code=CODE_RUNTIME_ONLY_PERSIST_FORBIDDEN,
        payload_contains_runtime_guardrails=payload_contains_runtime_guardrails,
        model_cache_get=_cache_get,
        format_llm_ssrf_error=_format_llm_ssrf_error,
        llm_insecure_override_enabled=_llm_insecure_override_enabled,
        fetch_remote_model_list=fetch_remote_model_list,
        get_stale_cached_models=get_stale_cached_models,
        resolve_model_list_target=resolve_model_list_target,
        validate_model_list_target=validate_model_list_target,
        llm_client=LLMClient,
    )


@endpoint_metadata(
    auth=AuthTier.OBSERVABILITY,
    risk=RiskTier.LOW,
    summary="Get configuration",
    description="Returns effective config, sources, and provider catalog.",
    audit="config.read",
    plane=RoutePlane.ADMIN,
)
async def config_get_handler(request: web.Request) -> web.Response:
    """
    GET /moltbot/config
    Returns effective config, sources, and provider catalog.
    Enforced by S14 Access Control.
    """
    # CRITICAL: owned implementation performs require_observability_access before reads.
    return await config_get_response(request, _handler_dependencies())


@endpoint_metadata(
    auth=AuthTier.ADMIN,
    risk=RiskTier.LOW,  # Read-only external fetch, but admin-gated
    summary="List remote models",
    description="Fetch a remote model list (best-effort) for OpenAI-compatible providers.",
    audit="llm.list_models",
    plane=RoutePlane.ADMIN,
)
async def llm_models_handler(request: web.Request) -> web.Response:
    """
    GET /openclaw/llm/models (legacy: /moltbot/llm/models)
    Fetch a remote model list (best-effort) for OpenAI-compatible providers.

    Security:
    - admin boundary
    - loopback-only unless OPENCLAW_ALLOW_REMOTE_ADMIN=1
    - SSRF policy enforced via LLM egress controls, including scoped private-network allowance
    """
    # CRITICAL: owned implementation performs require_admin_token( before network access.
    # CRITICAL S65: fetch_remote_model_list remains the safe_request_json egress owner.
    return await llm_models_response(request, _handler_dependencies())


@endpoint_metadata(
    auth=AuthTier.ADMIN,
    risk=RiskTier.HIGH,
    summary="Update configuration",
    description="Updates non-secret LLM config.",
    audit="config.update",
    plane=RoutePlane.ADMIN,
)
async def config_put_handler(request: web.Request) -> web.Response:
    """
    PUT /moltbot/config
    Updates non-secret LLM config. Protected by admin boundary (S13) + CSRF (S26+).
    """
    # CRITICAL: owned implementation performs require_admin_token( before mutation.
    return await config_put_response(request, _handler_dependencies())


@endpoint_metadata(
    auth=AuthTier.ADMIN,
    risk=RiskTier.MEDIUM,
    summary="Test LLM connection",
    description="Tests LLM connection using provided or stored credentials.",
    audit="llm.test_connection",
    plane=RoutePlane.ADMIN,
)
async def llm_test_handler(request: web.Request) -> web.Response:
    """
    POST /moltbot/llm/test
    Tests LLM connection. Protected by admin boundary (S13) + CSRF (S26+).
    """
    # CRITICAL: owned implementation performs require_admin_token( before provider access.
    return await llm_test_response(request, _handler_dependencies())


@endpoint_metadata(
    auth=AuthTier.ADMIN,
    risk=RiskTier.MEDIUM,  # Consumed by connector, costs money/tokens
    summary="Chat completion",
    description="Run a simple chat completion using server-side LLM config.",
    audit="llm.chat_completion",
    plane=RoutePlane.ADMIN,
)
async def llm_chat_handler(request: web.Request) -> web.Response:
    """
    POST /openclaw/llm/chat (legacy: /moltbot/llm/chat)
    Run a simple chat completion using server-side LLM config + keys.
    This endpoint is intended for the connector; no prompt content is logged.
    """
    # CRITICAL: owned implementation performs require_admin_token( before provider access.
    return await llm_chat_response(request, _handler_dependencies())