admin/ComfyUI-OpenClaw
1
0
Fork 0
mirror of https://github.com/rookiestar28/ComfyUI-OpenClaw.git synced 2026-08-14 00:48:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(config): decompose API handlers

Browse Source
This commit is contained in:
rookiestar28
2026-07-11 08:14:32 +08:00
parent fdda687141
commit 746d9a1352
11 changed files with 1258 additions and 828 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/config.py
+74 -798
View File
@@ -166,6 +166,29 @@ except Exception:
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,
@@ -258,6 +281,46 @@ except ImportError:
]
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,
@@ -272,73 +335,8 @@ async def config_get_handler(request: web.Request) -> web.Response:
Returns effective config, sources, and provider catalog.
Enforced by S14 Access Control.
"""
if web is None:
raise RuntimeError("aiohttp not available")
# S14: Access Control
allowed, error = require_observability_access(request)
if not allowed:
return web.json_response({"ok": False, "error": error}, status=403)
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
return build_rate_limit_response(
request,
"admin",
web_module=web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = resolve_token_info(request)
try:
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": False, "error": e.code, "message": str(e)},
status=403,
)
except Exception as e:
logger.error("Error getting config (error_type=%s)", type(e).__name__)
return web.json_response(
{
"ok": False,
"error": "config_read_failed",
},
status=500,
)
# CRITICAL: owned implementation performs require_observability_access before reads.
return await config_get_response(request, _handler_dependencies())
@endpoint_metadata(
@@ -359,212 +357,9 @@ async def llm_models_handler(request: web.Request) -> web.Response:
- loopback-only unless OPENCLAW_ALLOW_REMOTE_ADMIN=1
- SSRF policy enforced via LLM egress controls, including scoped private-network allowance
"""
if web is None:
raise RuntimeError("aiohttp not available")
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
return build_rate_limit_response(
request,
"admin",
web_module=web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = resolve_token_info(request)
try:
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,
)
# 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(tenant_id=tenant.tenant_id)
try:
target = resolve_model_list_target(
provider_override,
effective,
tenant.tenant_id,
)
except ValueError as e:
return web.json_response(
{"ok": False, "error": str(e)},
status=400,
)
except TypeError as e:
return web.json_response(
{"ok": False, "error": str(e)},
status=400,
)
# R60: Check bounded TTL+LRU cache
cached_entry = _cache_get(target.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": target.provider,
"models": models,
"cached": True,
}
)
# 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 target.requires_api_key and not target.api_key:
return web.json_response(
{
"ok": False,
"error": f"No API key configured for provider '{target.provider}'.",
},
status=400,
)
# SSRF policy
try:
controls = get_llm_egress_controls(
target.provider,
target.base_url,
allow_private_network=target.allow_private_network,
)
validate_model_list_target(
target,
controls,
allow_insecure_base_url=_llm_insecure_override_enabled(),
)
except Exception as e:
return web.json_response(
{"ok": False, "error": _format_llm_ssrf_error(e)},
status=403,
)
# Fetch /models
try:
try:
from ..services.safe_io import SSRFError
except ImportError:
from services.safe_io import SSRFError # type: ignore
models = fetch_remote_model_list(
target,
controls,
pack_version=PACK_VERSION,
allow_insecure_base_url=_llm_insecure_override_enabled(),
)
return web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"provider": target.provider,
"models": models,
"cached": False,
}
)
except SSRFError as e:
return web.json_response(
{"ok": False, "error": _format_llm_ssrf_error(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 = get_stale_cached_models(target.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": target.provider,
"models": models,
"cached": True,
"warning": warning,
}
)
return web.json_response(
{"ok": False, "error": f"Upstream error: {str_e}"}, status=502
)
raise
except Exception as e:
stale = get_stale_cached_models(target.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": target.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,
)
# 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(
@@ -580,182 +375,8 @@ 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+).
"""
if web is None:
raise RuntimeError("aiohttp not available")
# 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
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
return build_rate_limit_response(
request,
"admin",
web_module=web,
error="Rate limit exceeded",
include_ok=True,
)
# R99/S46: resolve identity context for non-repudiation audits.
token_info = resolve_token_info(request)
# Still enforce admin requirement (which checks hierarchy)
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 "admin_token_required"},
request=request,
)
return web.json_response(
{
"ok": False,
"error": err or "Unauthorized",
},
status=403,
)
# S13: Optional loopback check
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"):
# Use S14 is_loopback which handles ipv6/mapped
remote = get_client_ip(request)
if not is_loopback(remote):
emit_audit_event(
action="config.update",
target="config.json",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": "remote_admin_denied", "remote": remote},
request=request,
)
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,
)
try:
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):
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=403,
details={"reason": e.code},
request=request,
)
return web.json_response(
{"ok": False, "error": e.code, "message": str(e)},
status=403,
)
# CRITICAL: owned implementation performs require_admin_token( before mutation.
return await config_put_response(request, _handler_dependencies())
@endpoint_metadata(
@@ -771,215 +392,8 @@ async def llm_test_handler(request: web.Request) -> web.Response:
POST /moltbot/llm/test
Tests LLM connection. Protected by admin boundary (S13) + CSRF (S26+).
"""
if web is None:
raise RuntimeError("aiohttp not available")
try:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
try:
# IMPORTANT: use package-relative import in ComfyUI runtime.
# CRITICAL: Missing this import causes NameError in provider error handling.
from ..services.provider_errors import ProviderHTTPError
except ImportError:
from services.provider_errors import ProviderHTTPError # type: ignore
# 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
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
return build_rate_limit_response(
request,
"admin",
web_module=web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = resolve_token_info(request)
# S13: Validate admin boundary
allowed, err = require_admin_token(request)
if not allowed:
emit_audit_event(
action="llm.test_connection",
target="llm",
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,
)
try:
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 = {}
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 (TypeError, ValueError, OverflowError):
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 (TypeError, ValueError, OverflowError):
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": True,
"tenant_id": tenant.tenant_id,
"message": "Connection successful",
"response": result["text"].strip(),
"provider": client.provider,
"model": client.model,
}
)
emit_audit_event(
action="llm.test_connection",
target=f"{client.provider}:{client.model}",
outcome="error",
token_info=token_info,
status_code=500,
details={
"tenant_id": tenant.tenant_id,
"provider": client.provider,
"model": client.model,
"error": "Empty response",
},
request=request,
)
return web.json_response(
{
"ok": False,
"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.error("LLM test failed (error_type=%s)", type(e).__name__)
emit_audit_event(
action="llm.test_connection",
target="llm",
outcome="error",
token_info=token_info,
status_code=500,
details={"error": "llm_test_failed"},
request=request,
)
return web.json_response(
{
"ok": False,
"error": "llm_test_failed",
},
status=500,
)
# CRITICAL: owned implementation performs require_admin_token( before provider access.
return await llm_test_response(request, _handler_dependencies())
@endpoint_metadata(
@@ -996,143 +410,5 @@ async def llm_chat_handler(request: web.Request) -> web.Response:
Run a simple chat completion using server-side LLM config + keys.
This endpoint is intended for the connector; no prompt content is logged.
"""
if web is None:
raise RuntimeError("aiohttp not available")
try:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
try:
# IMPORTANT: use package-relative import in ComfyUI runtime.
# CRITICAL: Missing this import causes NameError in provider error handling.
from ..services.provider_errors import ProviderHTTPError
except ImportError:
from services.provider_errors import ProviderHTTPError # type: ignore
# S28: CSRF protection for convenience mode (no admin token configured)
admin_token_configured = bool(get_admin_token())
resp = require_same_origin_if_no_token(request, admin_token_configured)
if resp:
return resp
# S17: Rate Limit
if not check_rate_limit(request, "admin"):
return build_rate_limit_response(
request,
"admin",
web_module=web,
error="Rate limit exceeded",
include_ok=True,
)
# 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(
{
"ok": False,
"error": err or "Unauthorized",
},
status=403,
)
try:
body = await request.json()
except Exception:
body = {}
if not isinstance(body, dict):
return web.json_response(
{"ok": False, "error": "Expected JSON object body"},
status=400,
)
system = body.get("system") if isinstance(body.get("system"), str) else ""
user_message = (
body.get("user_message")
if isinstance(body.get("user_message"), str)
else body.get("message") if isinstance(body.get("message"), str) else ""
)
temperature = (
body.get("temperature")
if isinstance(body.get("temperature"), (int, float))
else 0.7
)
max_tokens = (
body.get("max_tokens") if isinstance(body.get("max_tokens"), int) else 1024
)
if not user_message:
return web.json_response(
{"ok": False, "error": "missing_user_message"},
status=400,
)
# S29: Debug-level structured log — metadata only, never raw prompt content.
logger.debug(
"llm_chat: has_system=%s msg_len=%d temperature=%.2f max_tokens=%d",
bool(system),
len(user_message),
temperature,
max_tokens,
)
try:
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,
)
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}
)
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(
{"ok": False, "error": str(e)},
status=400,
)
except ProviderHTTPError as e:
# IMPORTANT (recurring support issue):
# Do not swallow provider errors into a generic "llm_request_failed" without context.
# The connector can safely surface *redacted* provider messages (no prompt content)
# so users can fix misconfiguration (401/403/429, SSRF allowlist, etc.) quickly.
payload = {
"ok": False,
"error": f"{e.provider} HTTP {e.status_code}: {e.message}",
"provider": e.provider,
"status_code": e.status_code,
}
if getattr(e, "retry_after", None):
payload["retry_after"] = e.retry_after
return web.json_response(payload, status=e.status_code)
except Exception as e:
# S29: classify only; preserve the explicit redaction marker for log consumers.
logger.warning(
"LLM chat request failed: ***REDACTED*** (error_type=%s)",
type(e).__name__,
)
return web.json_response(
{"ok": False, "error": "llm_request_failed"},
status=500,
)
# CRITICAL: owned implementation performs require_admin_token( before provider access.
return await llm_chat_response(request, _handler_dependencies())
api/config_llm_handlers.py
+283
View File
@@ -0,0 +1,283 @@
"""Owned LLM connection-test and chat handler implementations."""
from __future__ import annotations
from typing import Any
from .config_projection_handlers import ConfigHandlerDependencies
async def llm_test_response(request: Any, deps: ConfigHandlerDependencies) -> Any:
"""Run the existing tenant-scoped, audited LLM connection test."""
if deps.web is None:
raise RuntimeError("aiohttp not available")
try:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
admin_token_configured = bool(deps.get_admin_token())
response = deps.require_same_origin_if_no_token(request, admin_token_configured)
if response:
return response
if not deps.check_rate_limit(request, "admin"):
return deps.build_rate_limit_response(
request,
"admin",
web_module=deps.web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = deps.resolve_token_info(request)
allowed, error = deps.require_admin_token(request)
if not allowed:
deps.emit_audit_event(
action="llm.test_connection",
target="llm",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": error or "unauthorized"},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": error or "Unauthorized"}, status=403
)
try:
with deps.request_tenant_scope(
request=request, token_info=token_info, allow_default_when_missing=True
) as tenant:
try:
body = await request.json()
if body is None:
body = {}
except Exception:
body = {}
if body and not isinstance(body, dict):
return deps.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 (TypeError, ValueError, OverflowError):
return deps.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 (TypeError, ValueError, OverflowError):
return deps.web.json_response(
{"ok": False, "error": "max_retries must be an integer"},
status=400,
)
client = deps.llm_client(
provider=provider,
base_url=base_url,
model=model,
timeout=timeout_sec,
max_retries=max_retries,
)
result = await run_in_thread(
client.complete,
system="You are a test assistant.",
user_message="Respond with exactly: OK",
max_tokens=10,
)
if result and "text" in result:
deps.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 deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"message": "Connection successful",
"response": result["text"].strip(),
"provider": client.provider,
"model": client.model,
}
)
deps.emit_audit_event(
action="llm.test_connection",
target=f"{client.provider}:{client.model}",
outcome="error",
token_info=token_info,
status_code=500,
details={
"tenant_id": tenant.tenant_id,
"provider": client.provider,
"model": client.model,
"error": "Empty response",
},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": "Empty or invalid response from LLM"}
)
except deps.tenant_boundary_error as exc:
deps.emit_audit_event(
action="llm.test_connection",
target="llm",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": exc.code},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": exc.code, "message": str(exc)}, status=403
)
except Exception as exc:
deps.logger.error("LLM test failed (error_type=%s)", type(exc).__name__)
deps.emit_audit_event(
action="llm.test_connection",
target="llm",
outcome="error",
token_info=token_info,
status_code=500,
details={"error": "llm_test_failed"},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": "llm_test_failed"}, status=500
)
async def llm_chat_response(request: Any, deps: ConfigHandlerDependencies) -> Any:
"""Run server-side tenant-scoped chat without logging prompt content."""
if deps.web is None:
raise RuntimeError("aiohttp not available")
try:
from ..services.async_utils import run_in_thread
except ImportError:
from services.async_utils import run_in_thread
try:
from ..services.provider_errors import ProviderHTTPError
except ImportError:
from services.provider_errors import ProviderHTTPError
admin_token_configured = bool(deps.get_admin_token())
response = deps.require_same_origin_if_no_token(request, admin_token_configured)
if response:
return response
if not deps.check_rate_limit(request, "admin"):
return deps.build_rate_limit_response(
request,
"admin",
web_module=deps.web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = deps.resolve_token_info(request)
allowed, error = deps.require_admin_token(request)
if not allowed:
return deps.web.json_response(
{"ok": False, "error": error or "Unauthorized"}, status=403
)
try:
body = await request.json()
except Exception:
body = {}
if not isinstance(body, dict):
return deps.web.json_response(
{"ok": False, "error": "Expected JSON object body"}, status=400
)
system = body.get("system") if isinstance(body.get("system"), str) else ""
user_message = (
body.get("user_message")
if isinstance(body.get("user_message"), str)
else body.get("message") if isinstance(body.get("message"), str) else ""
)
temperature = (
body.get("temperature")
if isinstance(body.get("temperature"), (int, float))
else 0.7
)
max_tokens = (
body.get("max_tokens") if isinstance(body.get("max_tokens"), int) else 1024
)
if not user_message:
return deps.web.json_response(
{"ok": False, "error": "missing_user_message"}, status=400
)
deps.logger.debug(
"llm_chat: has_system=%s msg_len=%d temperature=%.2f max_tokens=%d",
bool(system),
len(user_message),
temperature,
max_tokens,
)
try:
with deps.request_tenant_scope(
request=request, token_info=token_info, allow_default_when_missing=True
) as tenant:
client = deps.llm_client()
def _run():
return client.complete(
system=system,
user_message=user_message,
temperature=temperature,
max_tokens=max_tokens,
)
result = await run_in_thread(_run)
text = result.get("text") or "" if isinstance(result, dict) else ""
return deps.web.json_response(
{"ok": True, "tenant_id": tenant.tenant_id, "text": text}
)
except deps.tenant_boundary_error as exc:
return deps.web.json_response(
{"ok": False, "error": exc.code, "message": str(exc)}, status=403
)
except ValueError as exc:
return deps.web.json_response({"ok": False, "error": str(exc)}, status=400)
except ProviderHTTPError as exc:
payload = {
"ok": False,
"error": f"{exc.provider} HTTP {exc.status_code}: {exc.message}",
"provider": exc.provider,
"status_code": exc.status_code,
}
if getattr(exc, "retry_after", None):
payload["retry_after"] = exc.retry_after
return deps.web.json_response(payload, status=exc.status_code)
except Exception as exc:
deps.logger.warning(
"LLM chat request failed: ***REDACTED*** (error_type=%s)",
type(exc).__name__,
)
return deps.web.json_response(
{"ok": False, "error": "llm_request_failed"}, status=500
)
api/config_model_handlers.py
+182
View File
@@ -0,0 +1,182 @@
"""Owned remote model-discovery handler implementation."""
from __future__ import annotations
import os
from typing import Any
from .config_projection_handlers import ConfigHandlerDependencies
async def llm_models_response(request: Any, deps: ConfigHandlerDependencies) -> Any:
"""Serve tenant-isolated bounded provider model discovery."""
if deps.web is None:
raise RuntimeError("aiohttp not available")
if not deps.check_rate_limit(request, "admin"):
return deps.build_rate_limit_response(
request,
"admin",
web_module=deps.web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = deps.resolve_token_info(request)
try:
with deps.request_tenant_scope(
request=request, token_info=token_info, allow_default_when_missing=True
) as tenant:
allowed, error = deps.require_admin_token(request)
if not allowed:
deps.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": error or "unauthorized",
},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": error or "Unauthorized"}, status=403
)
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 deps.is_loopback_client(remote):
return deps.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 = deps.get_effective_config(tenant_id=tenant.tenant_id)
try:
target = deps.resolve_model_list_target(
provider_override, effective, tenant.tenant_id
)
except (TypeError, ValueError) as exc:
return deps.web.json_response(
{"ok": False, "error": str(exc)}, status=400
)
cached_entry = deps.model_cache_get(target.cache_key)
if cached_entry:
_timestamp, models = cached_entry
if isinstance(models, list):
return deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"provider": target.provider,
"models": models,
"cached": True,
}
)
# CRITICAL: local providers intentionally work without API keys.
if target.requires_api_key and not target.api_key:
return deps.web.json_response(
{
"ok": False,
"error": f"No API key configured for provider '{target.provider}'.",
},
status=400,
)
try:
controls = deps.get_llm_egress_controls(
target.provider,
target.base_url,
allow_private_network=target.allow_private_network,
)
deps.validate_model_list_target(
target,
controls,
allow_insecure_base_url=deps.llm_insecure_override_enabled(),
)
except Exception as exc:
return deps.web.json_response(
{"ok": False, "error": deps.format_llm_ssrf_error(exc)},
status=403,
)
try:
try:
from ..services.safe_io import SSRFError
except ImportError:
from services.safe_io import SSRFError
models = deps.fetch_remote_model_list(
target,
controls,
pack_version=deps.pack_version,
allow_insecure_base_url=deps.llm_insecure_override_enabled(),
)
return deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"provider": target.provider,
"models": models,
"cached": False,
}
)
except SSRFError as exc:
return deps.web.json_response(
{"ok": False, "error": deps.format_llm_ssrf_error(exc)},
status=403,
)
except RuntimeError as exc:
error_text = str(exc)
if "HTTP" in error_text:
stale = deps.get_stale_cached_models(target.cache_key)
if stale:
_timestamp, models = stale
warning = f"Using cached list (refresh failed: {error_text})"
return deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"provider": target.provider,
"models": models,
"cached": True,
"warning": warning,
}
)
return deps.web.json_response(
{"ok": False, "error": f"Upstream error: {error_text}"},
status=502,
)
raise
except Exception as exc:
stale = deps.get_stale_cached_models(target.cache_key)
if stale:
deps.logger.warning(
"Model list refresh failed, serving cached list: %s", exc
)
_timestamp, models = stale
warning = f"Using cached list (refresh failed: {exc!s})"
return deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"provider": target.provider,
"models": models,
"cached": True,
"warning": warning,
}
)
deps.logger.exception("Failed to fetch model list")
return deps.web.json_response(
{"ok": False, "error": str(exc)}, status=500
)
except deps.tenant_boundary_error as exc:
return deps.web.json_response(
{"ok": False, "error": exc.code, "message": str(exc)}, status=403
)
api/config_projection_handlers.py
+244
View File
@@ -0,0 +1,244 @@
"""Owned config projection and mutation handler implementations."""
from __future__ import annotations
import json
import os
from dataclasses import dataclass
from typing import Any
@dataclass(frozen=True)
class ConfigHandlerDependencies:
web: Any
logger: Any
provider_catalog: Any
pack_version: Any
require_observability_access: Any
require_admin_token: Any
require_same_origin_if_no_token: Any
resolve_token_info: Any
emit_audit_event: Any
check_rate_limit: Any
build_rate_limit_response: Any
get_client_ip: Any
is_loopback: Any
get_admin_token: Any
get_apply_semantics: Any
get_effective_config: Any
get_llm_egress_controls: Any
get_runtime_guardrails: Any
get_settings_schema: Any
is_loopback_client: Any
update_config: Any
tenant_boundary_error: Any
request_tenant_scope: Any
runtime_only_code: Any
payload_contains_runtime_guardrails: Any
model_cache_get: Any
format_llm_ssrf_error: Any
llm_insecure_override_enabled: Any
fetch_remote_model_list: Any
get_stale_cached_models: Any
resolve_model_list_target: Any
validate_model_list_target: Any
llm_client: Any
async def config_get_response(request: Any, deps: ConfigHandlerDependencies) -> Any:
"""Return the tenant-scoped effective configuration projection."""
if deps.web is None:
raise RuntimeError("aiohttp not available")
allowed, error = deps.require_observability_access(request)
if not allowed:
return deps.web.json_response({"ok": False, "error": error}, status=403)
if not deps.check_rate_limit(request, "admin"):
return deps.build_rate_limit_response(
request,
"admin",
web_module=deps.web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = deps.resolve_token_info(request)
try:
with deps.request_tenant_scope(
request=request, token_info=token_info, allow_default_when_missing=True
) as tenant:
effective, sources = deps.get_effective_config(tenant_id=tenant.tenant_id)
guardrails = deps.get_runtime_guardrails()
if guardrails.get("status") != "ok":
deps.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 deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"config": effective,
"sources": sources,
"runtime_guardrails": guardrails,
"providers": deps.provider_catalog,
"schema": deps.get_settings_schema(),
"write_enabled": True,
}
)
except deps.tenant_boundary_error as exc:
return deps.web.json_response(
{"ok": False, "error": exc.code, "message": str(exc)}, status=403
)
except Exception as exc:
deps.logger.error("Error getting config (error_type=%s)", type(exc).__name__)
return deps.web.json_response(
{"ok": False, "error": "config_read_failed"}, status=500
)
async def config_put_response(request: Any, deps: ConfigHandlerDependencies) -> Any:
"""Validate and atomically apply tenant-scoped non-secret config updates."""
if deps.web is None:
raise RuntimeError("aiohttp not available")
admin_token_configured = bool(deps.get_admin_token())
response = deps.require_same_origin_if_no_token(request, admin_token_configured)
if response:
return response
if not deps.check_rate_limit(request, "admin"):
return deps.build_rate_limit_response(
request,
"admin",
web_module=deps.web,
error="Rate limit exceeded",
include_ok=True,
)
token_info = deps.resolve_token_info(request)
allowed, error = deps.require_admin_token(request)
if not allowed:
deps.emit_audit_event(
action="config.update",
target="config.json",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": error or "admin_token_required"},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": error or "Unauthorized"}, status=403
)
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 = deps.get_client_ip(request)
if not deps.is_loopback(remote):
deps.emit_audit_event(
action="config.update",
target="config.json",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": "remote_admin_denied", "remote": remote},
request=request,
)
return deps.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,
)
try:
with deps.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 deps.web.json_response(
{"ok": False, "error": "Invalid JSON body"}, status=400
)
if deps.payload_contains_runtime_guardrails(body):
deps.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": deps.runtime_only_code,
},
request=request,
)
return deps.web.json_response(
{
"ok": False,
"error": "runtime_guardrails are runtime-only (ENV-driven) and cannot be persisted via /config",
"code": deps.runtime_only_code,
},
status=400,
)
updates = body.get("llm", body)
if not isinstance(updates, dict):
return deps.web.json_response(
{"ok": False, "error": "Expected object with config fields"},
status=400,
)
success, errors = deps.update_config(updates, tenant_id=tenant.tenant_id)
deps.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 deps.web.json_response(
{"ok": False, "errors": errors}, status=400
)
effective, sources = deps.get_effective_config(tenant_id=tenant.tenant_id)
apply_info = deps.get_apply_semantics(list(updates.keys()))
return deps.web.json_response(
{
"ok": True,
"tenant_id": tenant.tenant_id,
"config": effective,
"sources": sources,
"apply": apply_info,
}
)
except deps.tenant_boundary_error as exc:
deps.emit_audit_event(
action="config.update",
target="config.json",
outcome="deny",
token_info=token_info,
status_code=403,
details={"reason": exc.code},
request=request,
)
return deps.web.json_response(
{"ok": False, "error": exc.code, "message": str(exc)}, status=403
)
scripts/verify_api_config_contract.py
+154
View File
@@ -0,0 +1,154 @@
"""Verify the frozen R221 API config facade and governance contract."""
from __future__ import annotations
import argparse
import hashlib
import inspect
import json
import sys
from pathlib import Path
from typing import Any
ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
CONTRACT_PATH = ROOT / "tests" / "api_config_contract_r221.json"
def _canonical_json(value: Any) -> str:
return json.dumps(value, indent=2, sort_keys=True, ensure_ascii=False) + "\n"
def _digest(value: Any) -> str:
return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()
def _metadata(handler: Any) -> dict[str, Any]:
from services.endpoint_manifest import get_metadata
meta = get_metadata(handler)
if meta is None:
raise RuntimeError(f"missing endpoint metadata for {handler.__name__}")
return {
"auth": meta.auth_tier.value,
"risk": meta.risk_tier.value,
"plane": meta.route_plane.value if meta.route_plane else None,
"summary": meta.summary,
"description": meta.description,
"audit": meta.audit_action,
"scopes": list(meta.required_scopes),
}
def build_contract() -> dict[str, Any]:
from api import config
handlers = (
"config_get_handler",
"llm_models_handler",
"config_put_handler",
"llm_test_handler",
"llm_chat_handler",
)
patch_seams = (
"web",
"logger",
"require_observability_access",
"require_admin_token",
"require_same_origin_if_no_token",
"check_rate_limit",
"build_rate_limit_response",
"resolve_token_info",
"emit_audit_event",
"request_tenant_scope",
"get_effective_config",
"get_runtime_guardrails",
"get_settings_schema",
"update_config",
"get_apply_semantics",
"get_admin_token",
"payload_contains_runtime_guardrails",
"get_llm_egress_controls",
"is_loopback_client",
"get_client_ip",
"resolve_model_list_target",
"validate_model_list_target",
"fetch_remote_model_list",
"get_stale_cached_models",
"_cache_get",
"_format_llm_ssrf_error",
"_llm_insecure_override_enabled",
"LLMClient",
)
schema = config.get_settings_schema()
route_contract = ROOT / "tests" / "api_route_contract_r220.json"
openapi = ROOT / "docs" / "openapi.yaml"
return {
"schema_version": 1,
"facade_signatures": {
name: str(inspect.signature(getattr(config, name))) for name in handlers
},
"facade_metadata": {
name: _metadata(getattr(config, name)) for name in handlers
},
"patch_seams": list(patch_seams),
"provider_catalog": config.PROVIDER_CATALOG,
"allowed_llm_keys": sorted(config.ALLOWED_LLM_KEYS),
"model_cache": {
"max_entries": config._MODEL_LIST_MAX_ENTRIES,
"ttl_sec": config._MODEL_LIST_TTL_SEC,
"exported_cache_type": type(config._MODEL_LIST_CACHE).__name__,
},
"settings_schema_sha256": _digest(schema),
"apply_semantics": {
"provider": config.get_apply_semantics(["provider"]),
"model": config.get_apply_semantics(["model"]),
"base_url": config.get_apply_semantics(["base_url"]),
},
"owned_response_matrices": {
"config": [
"tests.test_s66_api_config_guardrails",
"tests.test_r53_apply_semantics",
"tests.security.test_r99_sensitive_contract",
"tests.test_r219_exception_boundary_phase2",
],
"models": [
"tests.test_api_model_list",
"tests.test_r60_model_cache",
"tests.test_r123_real_backend_model_list_lane",
"tests.test_r155_exception_fidelity",
"tests.test_llm_default_allowlist",
],
"llm": [
"tests.test_s28s29_chat_csrf_redaction",
"tests.test_r219_exception_boundary_phase2",
],
},
"r220_route_contract_sha256": hashlib.sha256(
route_contract.read_bytes()
).hexdigest(),
"openapi_sha256": hashlib.sha256(openapi.read_bytes()).hexdigest(),
}
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--write-baseline", action="store_true")
args = parser.parse_args()
actual = build_contract()
if args.write_baseline:
CONTRACT_PATH.write_text(_canonical_json(actual), encoding="utf-8")
print(f"API-CONFIG-CONTRACT-WRITTEN: {CONTRACT_PATH}")
return 0
expected = json.loads(CONTRACT_PATH.read_text(encoding="utf-8"))
if actual != expected:
print("API-CONFIG-CONTRACT-FAIL: frozen config/facade contract drifted")
return 1
print("API-CONFIG-CONTRACT-PASS")
return 0
if __name__ == "__main__":
raise SystemExit(main())
tests/api_config_contract_r221.json
+205
View File
@@ -0,0 +1,205 @@
{
"allowed_llm_keys": [
"allow_private_network",
"base_url",
"fallback_models",
"fallback_providers",
"max_failover_candidates",
"max_retries",
"model",
"provider",
"timeout_sec"
],
"apply_semantics": {
"base_url": {
"applied_now": [
"base_url"
],
"notes": [],
"restart_required": []
},
"model": {
"applied_now": [
"model"
],
"notes": [],
"restart_required": []
},
"provider": {
"applied_now": [
"provider"
],
"notes": [],
"restart_required": []
}
},
"facade_metadata": {
"config_get_handler": {
"audit": "config.read",
"auth": "obs",
"description": "Returns effective config, sources, and provider catalog.",
"plane": "admin",
"risk": "low",
"scopes": [],
"summary": "Get configuration"
},
"config_put_handler": {
"audit": "config.update",
"auth": "admin",
"description": "Updates non-secret LLM config.",
"plane": "admin",
"risk": "high",
"scopes": [],
"summary": "Update configuration"
},
"llm_chat_handler": {
"audit": "llm.chat_completion",
"auth": "admin",
"description": "Run a simple chat completion using server-side LLM config.",
"plane": "admin",
"risk": "medium",
"scopes": [],
"summary": "Chat completion"
},
"llm_models_handler": {
"audit": "llm.list_models",
"auth": "admin",
"description": "Fetch a remote model list (best-effort) for OpenAI-compatible providers.",
"plane": "admin",
"risk": "low",
"scopes": [],
"summary": "List remote models"
},
"llm_test_handler": {
"audit": "llm.test_connection",
"auth": "admin",
"description": "Tests LLM connection using provided or stored credentials.",
"plane": "admin",
"risk": "medium",
"scopes": [],
"summary": "Test LLM connection"
}
},
"facade_signatures": {
"config_get_handler": "(request: 'web.Request') -> 'web.Response'",
"config_put_handler": "(request: 'web.Request') -> 'web.Response'",
"llm_chat_handler": "(request: 'web.Request') -> 'web.Response'",
"llm_models_handler": "(request: 'web.Request') -> 'web.Response'",
"llm_test_handler": "(request: 'web.Request') -> 'web.Response'"
},
"model_cache": {
"exported_cache_type": "OrderedDict",
"max_entries": 16,
"ttl_sec": 600
},
"openapi_sha256": "7997883a91ca7b3e3058d3c299512947ca634d92ddd1aa4c06f0d33539325f33",
"owned_response_matrices": {
"config": [
"tests.test_s66_api_config_guardrails",
"tests.test_r53_apply_semantics",
"tests.security.test_r99_sensitive_contract",
"tests.test_r219_exception_boundary_phase2"
],
"llm": [
"tests.test_s28s29_chat_csrf_redaction",
"tests.test_r219_exception_boundary_phase2"
],
"models": [
"tests.test_api_model_list",
"tests.test_r60_model_cache",
"tests.test_r123_real_backend_model_list_lane",
"tests.test_r155_exception_fidelity",
"tests.test_llm_default_allowlist"
]
},
"patch_seams": [
"web",
"logger",
"require_observability_access",
"require_admin_token",
"require_same_origin_if_no_token",
"check_rate_limit",
"build_rate_limit_response",
"resolve_token_info",
"emit_audit_event",
"request_tenant_scope",
"get_effective_config",
"get_runtime_guardrails",
"get_settings_schema",
"update_config",
"get_apply_semantics",
"get_admin_token",
"payload_contains_runtime_guardrails",
"get_llm_egress_controls",
"is_loopback_client",
"get_client_ip",
"resolve_model_list_target",
"validate_model_list_target",
"fetch_remote_model_list",
"get_stale_cached_models",
"_cache_get",
"_format_llm_ssrf_error",
"_llm_insecure_override_enabled",
"LLMClient"
],
"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": "Gemini (OpenAI-compat)",
"requires_key": true
},
{
"id": "groq",
"label": "Groq",
"requires_key": true
},
{
"id": "deepseek",
"label": "DeepSeek",
"requires_key": true
},
{
"id": "xai",
"label": "xAI",
"requires_key": true
},
{
"id": "ollama",
"label": "Ollama (Local)",
"requires_key": false
},
{
"id": "lmstudio",
"label": "LM Studio (Local)",
"requires_key": false
},
{
"id": "antigravity_proxy",
"label": "Antigravity Claude Proxy (Local)",
"requires_key": false
},
{
"id": "custom",
"label": "Custom",
"requires_key": true
}
],
"r220_route_contract_sha256": "17c804dc80f35ddf774e5a941ab4fd4404f55f1e37a8cafd3b3a9238a5e12c9e",
"schema_version": 1,
"settings_schema_sha256": "e129472bd8b4fb81181c2a3169ed6177757cb54050276646eea70052007bd10b"
}
tests/exception_boundary_policy.json
+15 -8
View File
@@ -133,24 +133,31 @@
}
]
},
"api/config.py": {
"api/config_projection_handlers.py": {
"coverage": "selected_scopes",
"selected_scopes": [
"config_get_handler",
"llm_test_handler",
"llm_chat_handler"
"config_get_response"
],
"broad_catches": [
{
"scope": "config_get_handler",
"scope": "config_get_response",
"expected_count": 1,
"classification": "allowed_boundary_guard",
"reason": "The admin read boundary returns a fixed content-free 500 code after tenant-specific failures.",
"regression_owner": "tests/test_r219_exception_boundary_phase2.py",
"review_after": "2027-01-11"
},
}
]
},
"api/config_llm_handlers.py": {
"coverage": "selected_scopes",
"selected_scopes": [
"llm_test_response",
"llm_chat_response"
],
"broad_catches": [
{
"scope": "llm_test_handler",
"scope": "llm_test_response",
"expected_count": 2,
"classification": "allowed_boundary_guard",
"reason": "Request decoding and final LLM test translation retain existing response shape with fixed errors.",
@@ -158,7 +165,7 @@
"review_after": "2027-01-11"
},
{
"scope": "llm_chat_handler",
"scope": "llm_chat_response",
"expected_count": 2,
"classification": "allowed_boundary_guard",
"reason": "Chat request decoding and final server-error translation are public route boundaries.",
tests/static_analysis_policy.json
+1 -15
View File
@@ -124,7 +124,7 @@
"path": "api/config.py",
"code": "unused-ignore",
"message": "Unused \"type: ignore\" comment",
"count": 6
"count": 3
},
{
"tool": "mypy",
@@ -2492,20 +2492,6 @@
"message": "`typing.Dict` is deprecated, use `dict` instead",
"count": 1
},
{
"tool": "ruff",
"path": "api/config.py",
"code": "F401",
"message": "`services.provider_errors.ProviderHTTPError` imported but unused",
"count": 1
},
{
"tool": "ruff",
"path": "api/config.py",
"code": "RUF010",
"message": "Use explicit conversion flag",
"count": 1
},
{
"tool": "ruff",
"path": "api/connector_contracts.py",
tests/test_r155_exception_fidelity.py
+13 -1
View File
@@ -88,7 +88,10 @@ class TestR155ExceptionFidelity(unittest.TestCase):
)
def test_api_config_runtime_error_preserves_original_traceback_line(self):
import inspect
from api.config import llm_models_handler
from api.config_model_handlers import llm_models_response
request = MagicMock()
request.query = {}
@@ -113,8 +116,17 @@ class TestR155ExceptionFidelity(unittest.TestCase):
lambda: self._run_async(llm_models_handler(request))
)
source, start_line = inspect.getsourcelines(llm_models_response)
call_offset = next(
index
for index, line in enumerate(source)
if "models = deps.fetch_remote_model_list(" in line
)
self._assert_traceback_contains_frame(
tb, "api/config.py", 494, "models = fetch_remote_model_list("
tb,
"api/config_model_handlers.py",
start_line + call_offset,
"models = deps.fetch_remote_model_list(",
)
def _run_async(self, coro):
tests/test_r219_exception_boundary_phase2.py
+7 -6
View File
@@ -38,7 +38,8 @@ class TestPolicyV2(unittest.TestCase):
"api/route_orchestration.py",
"connector/router.py",
"services/route_bootstrap.py",
"api/config.py",
"api/config_projection_handlers.py",
"api/config_llm_handlers.py",
"connector/platforms/slack_webhook.py",
"connector/platforms/feishu_webhook.py",
},
@@ -92,9 +93,9 @@ class TestPolicyV2(unittest.TestCase):
)
stale_scope = copy.deepcopy(policy)
stale_scope["selected_modules"]["api/config.py"]["selected_scopes"].append(
"removed_scope"
)
stale_scope["selected_modules"]["api/config_projection_handlers.py"][
"selected_scopes"
].append("removed_scope")
self.assertTrue(
any(
"selected scope has no broad catch" in item
@@ -261,8 +262,8 @@ class TestExpectedParserBoundaries(unittest.TestCase):
def test_config_numeric_parsers_have_no_broad_catch(self):
from scripts.verify_exception_boundary_policy import iter_broad_catches
catches = list(iter_broad_catches(ROOT / "api" / "config.py"))
llm_test = [catch for catch in catches if catch.scope == "llm_test_handler"]
catches = list(iter_broad_catches(ROOT / "api" / "config_llm_handlers.py"))
llm_test = [catch for catch in catches if catch.scope == "llm_test_response"]
self.assertEqual(len(llm_test), 2)
tests/test_r221_api_config_decomposition.py
+80
View File
@@ -0,0 +1,80 @@
"""Contract-first tests for R221 API config hotspot decomposition."""
from __future__ import annotations
import importlib.util
import inspect
import json
import unittest
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
def _load_verifier():
path = ROOT / "scripts" / "verify_api_config_contract.py"
spec = importlib.util.spec_from_file_location("r221_config_contract", path)
if spec is None or spec.loader is None:
raise RuntimeError("unable to load R221 config contract verifier")
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module
class TestR221ApiConfigDecomposition(unittest.TestCase):
def test_frozen_config_facade_and_outer_contract_matches_byte_for_byte(self):
verifier = _load_verifier()
fixture = (ROOT / "tests" / "api_config_contract_r221.json").read_text(
encoding="utf-8"
)
self.assertEqual(verifier._canonical_json(verifier.build_contract()), fixture)
def test_all_established_patch_seams_remain_on_facade(self):
from api import config
verifier = _load_verifier()
contract = verifier.build_contract()
missing = [
name for name in contract["patch_seams"] if not hasattr(config, name)
]
self.assertEqual(missing, [])
def test_owned_modules_are_substantive_one_way_boundaries(self):
from api import (
config_llm_handlers,
config_model_handlers,
config_projection_handlers,
)
expected = {
config_projection_handlers: ("config_get_response", "config_put_response"),
config_model_handlers: ("llm_models_response",),
config_llm_handlers: ("llm_test_response", "llm_chat_response"),
}
for module, functions in expected.items():
source = inspect.getsource(module)
self.assertNotIn("import api.config", source)
self.assertNotIn("from api import config", source)
self.assertNotIn("from . import config", source)
for name in functions:
self.assertGreater(
len(inspect.getsource(getattr(module, name)).splitlines()), 20
)
def test_r220_route_and_openapi_digests_remain_frozen(self):
verifier = _load_verifier()
expected = json.loads(
(ROOT / "tests" / "api_config_contract_r221.json").read_text(
encoding="utf-8"
)
)
actual = verifier.build_contract()
self.assertEqual(
actual["r220_route_contract_sha256"],
expected["r220_route_contract_sha256"],
)
self.assertEqual(actual["openapi_sha256"], expected["openapi_sha256"])
if __name__ == "__main__":
unittest.main()