merge dev

This commit is contained in:
rookiestar28
2026-02-12 03:56:33 +08:00
25 changed files with 4861 additions and 75 deletions
+63 -5
View File
@@ -25,11 +25,48 @@ It is designed to make **ComfyUI a reliable automation target** with an explicit
<details>
<summary><strong>Sprint A: closes out with five concrete reliability and security improvements</strong></summary>
- Configuration save/apply now returns explicit apply metadata, so callers can see what was actually applied, what requires restart, and which effective provider/model is active.
- The Settings update flow adds defensive guards against stale or partial state, reducing accidental overwrites.
- Provider/model precedence is now deterministic across save, test, and chat paths, and prevents model contamination when switching providers.
- In localhost convenience mode (no admin token configured), chat requests enforce same-origin CSRF protection: same-origin requests are allowed, cross-origin requests are denied.
- Model-list fetching now uses a bounded in-memory cache keyed by provider and base URL, with a 5-minute TTL and LRU eviction cap to improve responsiveness and stability.
- Configuration save/apply now returns explicit apply metadata, so callers can see what was actually applied, what requires restart, and which effective provider/model is active.
- The Settings update flow adds defensive guards against stale or partial state, reducing accidental overwrites.
- Provider/model precedence is now deterministic across save, test, and chat paths, and prevents model contamination when switching providers.
- In localhost convenience mode (no admin token configured), chat requests enforce same-origin CSRF protection: same-origin requests are allowed, cross-origin requests are denied.
- Model-list fetching now uses a bounded in-memory cache keyed by provider and base URL, with a 5-minute TTL and LRU eviction cap to improve responsiveness and stability.
</details>
<details>
<summary><strong>Sprint B: ships security doctor diagnostics, registry quarantine gates, and constrained transforms defaults</strong></summary>
- Added the Security Doctor surface (`GET /openclaw/security/doctor`) for operator-focused security posture checks across endpoint exposure, token boundaries, SSRF posture, state-dir permissions, redaction drift, runtime mode, feature flags, and API key posture.
- Added optional remote pack registry quarantine controls with explicit lifecycle states, SHA256 integrity verification, bounded local persistence, and per-entry audit trail; this path remains disabled by default and fail-closed.
- Added optional constrained transform execution with trusted-directory + integrity pinning, timeout and output-size caps, and bounded chain execution semantics; transforms remain disabled by default and mapping-only behavior remains intact unless explicitly enabled.
</details>
<details>
<summary><strong>Settings contract, frontend graceful degradation, and provider drift governance</strong></summary>
- Enforced a strict settings write contract with schema-coerced values and explicit unknown-key rejection, reducing save/apply regressions across ComfyUI variants.
- Hardened frontend behavior to degrade safely when optional routes or runtime capabilities are unavailable, with clearer recovery hints instead of brittle failures.
- Added provider alias/deprecation governance and normalization coverage to reduce preset drift as upstream model IDs and endpoint shapes evolve.
</details>
<details>
<summary><strong>Mapping v1, job event stream, and operator doctor</strong></summary>
- Added webhook mapping engine v1 with declarative field mapping + type coercion, enabling external payload normalization without custom adapter code paths.
- Added real-time job event stream support via SSE (`/openclaw/events/stream`) with bounded buffering and polling fallback (`/openclaw/events`) for compatibility.
- Added Operator Doctor diagnostics tooling for runtime/deployment checks (Python/Node environment, state-dir posture, and contract readiness signals).
</details>
<details>
<summary><strong>Security doctor, registry quarantine, and constrained transforms</strong></summary>
- Added Security Doctor diagnostics surface (`GET /openclaw/security/doctor`) for operator-focused security posture checks and guarded remediation flow.
- Added optional remote registry quarantine lifecycle controls with integrity verification, bounded local persistence, and explicit trust/audit gates.
- Added optional constrained transform execution with integrity pinning, timeout/output caps, and bounded chain semantics; default posture remains disabled/fail-closed.
</details>
---
@@ -220,6 +257,17 @@ Notes:
- `POST /openclaw/webhook/validate` — dry-run render (no queue submission; includes render budgets + warnings)
- `POST /openclaw/webhook/submit` — full pipeline: auth → normalize → idempotency → render → submit to queue
**Payload Mapping (F40)**:
- Submit arbitrary payloads (GitHub, Discord, etc.) by adding `X-Webhook-Mapping-Profile: github_push` (or `discord_message`).
- The internal engine maps fields to the canonical schema before validation.
**Job Events (R71)**:
- `GET /openclaw/events/stream` — SSE endpoint for real-time job lifecycle events (queued, running, completed, failed).
- `GET /openclaw/events` — JSON polling fallback.
- Supports `Last-Event-ID` header to resume streams without data loss.
Request schema (minimal):
```json
@@ -410,6 +458,16 @@ Notes:
- If your pack folder name is not `comfyui-openclaw`, the smoke script may need `OPENCLAW_PACK_IMPORT_NAME=your-folder-name`.
- If imports fail with a `services.*` module error, check for name collisions with other custom nodes and prefer package-relative imports.
### Operator Doctor (R72)
Run the built-in diagnostic tool to verify environment readiness (libraries, permissions, contract files):
```bash
python scripts/operator_doctor.py
# Or check JSON output:
python scripts/operator_doctor.py --json
```
### Webhooks return `403 auth_not_configured`
Set webhook auth env vars (see “Quick Start”) and restart ComfyUI.
+188
View File
@@ -0,0 +1,188 @@
"""
R71 — Job Event Stream Endpoint.
SSE (Server-Sent Events) endpoint for real-time job lifecycle delivery,
plus a JSON polling fallback endpoint.
Routes:
GET /openclaw/events/stream — SSE (text/event-stream)
GET /openclaw/events — JSON polling fallback
"""
from __future__ import annotations
import asyncio
import json
import logging
from typing import Any, Dict
try:
from aiohttp import web # type: ignore
except ModuleNotFoundError: # pragma: no cover
web = None # type: ignore
if __package__ and "." in __package__:
from ..services.access_control import require_observability_access
from ..services.job_events import get_job_event_store
from ..services.metrics import metrics
from ..services.rate_limit import check_rate_limit
else: # pragma: no cover
from services.access_control import require_observability_access # type: ignore
from services.job_events import get_job_event_store # type: ignore
from services.metrics import metrics # type: ignore
from services.rate_limit import check_rate_limit # type: ignore
logger = logging.getLogger("ComfyUI-OpenClaw.api.events")
# SSE keep-alive interval (seconds)
SSE_KEEPALIVE_SEC = 15
# Maximum SSE connection duration (seconds) — prevents zombie connections
SSE_MAX_DURATION_SEC = 300 # 5 minutes
async def events_stream_handler(request: web.Request) -> web.StreamResponse:
"""
GET /openclaw/events/stream
SSE endpoint for job lifecycle events.
Supports Last-Event-ID for resume.
Access control parity with observability endpoints.
"""
if web is None:
raise RuntimeError("aiohttp not available")
# Rate limit
if not check_rate_limit(request, "events"):
return web.json_response(
{"ok": False, "error": "rate_limit_exceeded"},
status=429,
headers={"Retry-After": "60"},
)
# Access control (same as logs/tail)
denied = require_observability_access(request)
if denied:
return denied
store = get_job_event_store()
# Parse Last-Event-ID for resume
last_seq = 0
last_event_id = request.headers.get("Last-Event-ID", "").strip()
if last_event_id:
try:
last_seq = int(last_event_id)
except ValueError:
pass
# Optional prompt_id filter
prompt_id = request.query.get("prompt_id")
# Set up SSE response
response = web.StreamResponse(
status=200,
headers={
"Content-Type": "text/event-stream",
"Cache-Control": "no-cache",
"Connection": "keep-alive",
"X-Accel-Buffering": "no",
},
)
await response.prepare(request)
metrics.inc("events_sse_connections")
import time
start_time = time.time()
last_keepalive = time.time()
try:
while True:
# Check max duration
if time.time() - start_time > SSE_MAX_DURATION_SEC:
break
events = get_job_event_store().events_since(
last_seq=last_seq,
limit=50,
prompt_id=prompt_id,
)
if events:
for evt in events:
await response.write(evt.to_sse().encode("utf-8"))
last_seq = evt.seq
else:
# Send keep-alive header only if interval exceeded
now = time.time()
if now - last_keepalive > SSE_KEEPALIVE_SEC:
await response.write(b": keepalive\n\n")
last_keepalive = now
# Poll interval (1s latency is acceptable for job events)
await asyncio.sleep(1)
except (ConnectionError, asyncio.CancelledError):
pass
finally:
metrics.inc("events_sse_disconnections")
return response
async def events_poll_handler(request: web.Request) -> web.Response:
"""
GET /openclaw/events
JSON polling fallback for job events.
Query params:
- since: sequence number to resume from (default 0)
- prompt_id: optional filter
- limit: max events to return (default 50, max 200)
"""
if web is None:
raise RuntimeError("aiohttp not available")
# Rate limit
if not check_rate_limit(request, "events"):
return web.json_response(
{"ok": False, "error": "rate_limit_exceeded"},
status=429,
headers={"Retry-After": "60"},
)
# Access control
denied = require_observability_access(request)
if denied:
return denied
store = get_job_event_store()
# Parse query params
try:
since = int(request.query.get("since", "0"))
except ValueError:
since = 0
prompt_id = request.query.get("prompt_id")
try:
limit = max(1, min(int(request.query.get("limit", "50")), 200))
except ValueError:
limit = 50
events = store.events_since(
last_seq=since,
limit=limit,
prompt_id=prompt_id,
)
return web.json_response(
{
"ok": True,
"events": [e.to_dict() for e in events],
"latest_seq": store.latest_seq(),
}
)
+24
View File
@@ -21,9 +21,11 @@ PACK_NAME = PACK_VERSION = PACK_START_TIME = LOG_FILE = get_api_key = None # ty
metrics = tail_log = require_observability_access = check_rate_limit = trace_store = None # type: ignore
webhook_handler = webhook_submit_handler = webhook_validate_handler = capabilities_handler = preflight_handler = None # type: ignore
config_get_handler = config_put_handler = llm_test_handler = llm_models_handler = llm_chat_handler = None # type: ignore
security_doctor_handler = None # type: ignore # S30
templates_list_handler = None # type: ignore
secrets_status_handler = secrets_put_handler = secrets_delete_handler = None # type: ignore
list_checkpoints_handler = create_checkpoint_handler = get_checkpoint_handler = delete_checkpoint_handler = None # type: ignore
events_stream_handler = events_poll_handler = None # type: ignore # R71
redact_text = None # type: ignore
if web is not None:
@@ -45,12 +47,14 @@ if web is not None:
llm_models_handler,
llm_test_handler,
)
from ..api.events import events_poll_handler, events_stream_handler # R71
from ..api.preflight_handler import inventory_handler, preflight_handler
from ..api.secrets import (
secrets_delete_handler,
secrets_put_handler,
secrets_status_handler,
)
from ..api.security_doctor import security_doctor_handler # S30
from ..api.templates import templates_list_handler
from ..api.webhook import webhook_handler
from ..api.webhook_submit import webhook_submit_handler
@@ -87,12 +91,17 @@ if web is not None:
llm_models_handler,
llm_test_handler,
)
from api.events import ( # R71 # type: ignore
events_poll_handler,
events_stream_handler,
)
from api.preflight_handler import inventory_handler, preflight_handler
from api.secrets import (
secrets_delete_handler,
secrets_put_handler,
secrets_status_handler,
)
from api.security_doctor import security_doctor_handler # type: ignore # S30
from api.templates import templates_list_handler
from api.webhook import webhook_handler
from api.webhook_submit import webhook_submit_handler
@@ -504,11 +513,26 @@ def register_routes(server) -> None:
secrets_status_handler,
), # S25: Secret status (no values)
("PUT", f"{prefix}/secrets", secrets_put_handler), # S25: Save secret
(
"GET",
f"{prefix}/events/stream",
events_stream_handler,
), # R71: SSE event stream
(
"GET",
f"{prefix}/events",
events_poll_handler,
), # R71: JSON polling fallback
(
"DELETE",
f"{prefix}/secrets/{{provider}}",
secrets_delete_handler,
), # S25: Clear secret
(
"GET",
f"{prefix}/security/doctor",
security_doctor_handler,
), # S30: Security Doctor diagnostics
]
for method, path, handler in core_routes:
+103
View File
@@ -0,0 +1,103 @@
"""
S30 Security Doctor API handler.
GET /openclaw/security/doctor — Run security posture diagnostics.
Admin-only. JSON or human-readable output.
"""
from __future__ import annotations
import json
import logging
try:
from aiohttp import web
except ImportError: # pragma: no cover
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:
pass
class Response:
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
if __package__ and "." in __package__:
from ..services.access_control import require_admin_token
from ..services.rate_limit import check_rate_limit
from ..services.security_doctor import run_security_doctor
else: # pragma: no cover (test-only)
from services.access_control import require_admin_token # type: ignore
from services.rate_limit import check_rate_limit # type: ignore
from services.security_doctor import run_security_doctor # type: ignore
logger = logging.getLogger("ComfyUI-OpenClaw.api.security_doctor")
async def security_doctor_handler(request: web.Request) -> web.Response:
"""
GET /openclaw/security/doctor
Run security diagnostics. Admin-only.
Query params:
- format=json|text (default: json)
- remediate=1 (optional, run safe remediations)
- apply=1 (optional, actually apply remediations instead of dry-run)
"""
if web is None:
raise RuntimeError("aiohttp not available")
# S17: Rate limit
if not check_rate_limit(request, "admin"):
return web.json_response(
{"ok": False, "error": "Rate limit exceeded"}, status=429
)
# Admin boundary
allowed, err = require_admin_token(request)
if not allowed:
return web.json_response(
{"ok": False, "error": err or "Unauthorized"}, status=403
)
try:
fmt = (request.query.get("format") or "json").lower()
remediate = request.query.get("remediate", "").lower() in ("1", "true", "yes")
apply_mode = request.query.get("apply", "").lower() in ("1", "true", "yes")
report = run_security_doctor(
remediate=remediate,
dry_run=not apply_mode,
)
if fmt == "text":
return web.Response(
text=report.to_human(),
content_type="text/plain",
status=200,
)
return web.json_response(
{"ok": True, "report": report.to_dict()},
status=200,
)
except Exception as e:
logger.exception("Security doctor failed")
return web.json_response({"ok": False, "error": str(e)}, status=500)
+48
View File
@@ -20,6 +20,7 @@ if __package__ and "." in __package__:
from ..services.callback_delivery import start_callback_watch
from ..services.execution_budgets import BudgetExceededError
from ..services.idempotency_store import IdempotencyStore
from ..services.job_events import JobEventType, get_job_event_store
from ..services.metrics import metrics
from ..services.queue_submit import submit_prompt
from ..services.rate_limit import check_rate_limit
@@ -27,11 +28,13 @@ if __package__ and "." in __package__:
from ..services.trace import get_effective_trace_id
from ..services.trace_store import trace_store
from ..services.webhook_auth import require_auth
from ..services.webhook_mapping import apply_mapping, resolve_profile # F40
else: # pragma: no cover (test-only import mode)
from models.schemas import MAX_BODY_SIZE, WebhookJobRequest
from services.callback_delivery import start_callback_watch # type: ignore
from services.execution_budgets import BudgetExceededError # type: ignore
from services.idempotency_store import IdempotencyStore # type: ignore
from services.job_events import JobEventType, get_job_event_store # type: ignore
from services.metrics import metrics # type: ignore
from services.queue_submit import submit_prompt # type: ignore
from services.rate_limit import check_rate_limit # type: ignore
@@ -39,6 +42,10 @@ else: # pragma: no cover (test-only import mode)
from services.trace import get_effective_trace_id # type: ignore
from services.trace_store import trace_store # type: ignore
from services.webhook_auth import require_auth # type: ignore
from services.webhook_mapping import ( # F40 # type: ignore
apply_mapping,
resolve_profile,
)
logger = logging.getLogger("ComfyUI-OpenClaw.api.webhook_submit")
@@ -125,6 +132,30 @@ async def webhook_submit_handler(request: web.Request) -> web.Response:
trace_id = get_effective_trace_id(request.headers, data)
data["trace_id"] = trace_id
# F40: Payload Mapping Engine
# 1. Resolve profile
mapping_profile = resolve_profile(request.headers)
# 2. Apply mapping if profile found
if mapping_profile:
try:
# Log usage of mapping profile for observability
logger.info(
f"Applying mapping profile '{mapping_profile.id}' to request (trace: {trace_id})"
)
mapped_data, mapping_warnings = apply_mapping(mapping_profile, data)
# If trace_id was not in source but resolved from headers, re-inject it
if "trace_id" not in mapped_data:
mapped_data["trace_id"] = trace_id
data = mapped_data
for w in mapping_warnings:
logger.warning(f"Mapping warning (trace: {trace_id}): {w}")
except ValueError as e:
metrics.inc("webhook_denied")
return safe_error_response(400, "mapping_error", str(e))
# Validate against schema
try:
job_request = WebhookJobRequest.from_dict(data)
@@ -197,6 +228,22 @@ async def webhook_submit_handler(request: web.Request) -> web.Response:
prompt_id, callback_config, trace_id=trace_id
)
# R71: Emit QUEUED event
if prompt_id:
try:
get_job_event_store().emit(
JobEventType.QUEUED,
prompt_id=prompt_id,
trace_id=trace_id,
data={
"source": "webhook",
"template_id": template_id,
"job_id": job_id,
},
)
except Exception:
pass
metrics.inc("webhook_requests_executed")
return web.json_response(
{
@@ -204,6 +251,7 @@ async def webhook_submit_handler(request: web.Request) -> web.Response:
"deduped": False,
"prompt_id": prompt_id,
"trace_id": trace_id,
"mapped": bool(mapping_profile), # F40
"number": result.get("number"),
"callback_scheduled": bool(callback_config),
}
+26 -3
View File
@@ -43,6 +43,7 @@ if __package__ and "." in __package__:
from ..services.templates import get_template_service
from ..services.trace import get_effective_trace_id
from ..services.webhook_auth import require_auth
from ..services.webhook_mapping import apply_mapping, resolve_profile # F40
else: # pragma: no cover (test-only import mode)
from models.schemas import MAX_BODY_SIZE, WebhookJobRequest
from services.execution_budgets import ( # type: ignore
@@ -54,6 +55,10 @@ else: # pragma: no cover (test-only import mode)
from services.templates import get_template_service # type: ignore
from services.trace import get_effective_trace_id # type: ignore
from services.webhook_auth import require_auth # type: ignore
from services.webhook_mapping import ( # F40 # type: ignore
apply_mapping,
resolve_profile,
)
logger = logging.getLogger("ComfyUI-OpenClaw.api.webhook_validate")
@@ -137,6 +142,19 @@ async def webhook_validate_handler(request: web.Request) -> web.Response:
trace_id = get_effective_trace_id(request.headers, data)
data["trace_id"] = trace_id
# F40: Payload Mapping Engine
# 1. Resolve profile
mapping_profile = resolve_profile(request.headers)
mapping_warnings = []
# 2. Apply mapping if profile found
if mapping_profile:
try:
data, mapping_warnings = apply_mapping(mapping_profile, data)
except ValueError as e:
metrics.inc("webhook_denied")
return _safe_error_response(400, "mapping_error", str(e))
# Schema validation
try:
job_request = WebhookJobRequest.from_dict(data)
@@ -181,8 +199,8 @@ async def webhook_validate_handler(request: web.Request) -> web.Response:
headers={"Retry-After": str(e.retry_after)},
)
# Warnings: unresolved placeholders
warnings: List[str] = []
# Warnings: unresolved placeholders + mapping warnings
warnings: List[str] = mapping_warnings
unresolved: List[str] = []
try:
workflow_json = json.dumps(workflow, ensure_ascii=False, separators=(",", ":"))
@@ -197,7 +215,11 @@ async def webhook_validate_handler(request: web.Request) -> web.Response:
# Redact normalized response (security: may contain secrets)
try:
from services.redaction import redact_json
# Import discipline: attempt package-relative first
if __package__ and "." in __package__:
from ..services.redaction import redact_json
else:
from services.redaction import redact_json
safe_normalized = redact_json(normalized)
except ImportError:
@@ -209,6 +231,7 @@ async def webhook_validate_handler(request: web.Request) -> web.Response:
return web.json_response(
{
"ok": True,
"mapped": bool(mapping_profile), # F40: Indicate if mapping occurred
"trace_id": trace_id,
"template_id": template_id,
"normalized": safe_normalized, # Redacted for security
+1 -1
View File
@@ -1,7 +1,7 @@
[project]
name = "comfyui-openclaw"
description = "Your own personal AIGC Factory. Any picture. Any reel. The Comfy way.©️"
version = "0.2.5"
version = "0.2.6"
license = {text = "MIT"}
readme = "README.md"
requires-python = ">=3.10"
+23
View File
@@ -0,0 +1,23 @@
#!/usr/bin/env python
"""
R72 — Operator Doctor CLI convenience script.
Usage:
python scripts/operator_doctor.py
python scripts/operator_doctor.py --json
python scripts/operator_doctor.py --pack-root /path/to/pack
"""
import os
import sys
# Ensure pack root is on sys.path for imports
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PACK_ROOT = os.path.dirname(SCRIPT_DIR)
if PACK_ROOT not in sys.path:
sys.path.insert(0, PACK_ROOT)
from services.operator_doctor import main
if __name__ == "__main__":
main()
+68 -13
View File
@@ -18,12 +18,23 @@ function Require-Cmd($cmd) {
}
}
function Invoke-Checked {
param(
[Parameter(Mandatory = $true)][string]$Label,
[Parameter(Mandatory = $true)][scriptblock]$Command
)
& $Command
if ($LASTEXITCODE -ne 0) {
throw "[tests] ERROR: $Label failed with exit code $LASTEXITCODE"
}
}
Require-Cmd node
Require-Cmd npm
# Prefer project-local virtualenv to avoid global PATH / cache conflicts on Windows.
$venvPython = Join-Path $root ".venv\Scripts\python.exe"
if (-not (Test-Path $venvPython)) {
function New-ProjectVenv {
Write-Host "[tests] Creating project venv at $root\.venv ..."
if (Get-Command py -ErrorAction SilentlyContinue) {
& py -3 -m venv .venv
@@ -34,26 +45,70 @@ if (-not (Test-Path $venvPython)) {
}
}
function Test-VenvPython {
param([string]$PythonExe)
if (-not (Test-Path $PythonExe)) {
return $false
}
try {
& $PythonExe -c "import sys; print(sys.executable)" | Out-Null
return $true
} catch {
return $false
}
}
function Test-VenvCfgWindowsCompatible {
$cfg = Join-Path $root ".venv\pyvenv.cfg"
if (-not (Test-Path $cfg)) {
return $false
}
try {
$content = Get-Content $cfg -Raw
# WSL/Linux-built venvs typically contain POSIX home paths (e.g. /usr/bin)
if ($content -match "home\s*=\s*/") {
return $false
}
return $true
} catch {
return $false
}
}
if (-not (Test-Path $venvPython)) {
New-ProjectVenv
} elseif (-not (Test-VenvCfgWindowsCompatible)) {
Write-Host "[tests] WARN: existing .venv was created from non-Windows interpreter; recreating ..."
Remove-Item -Recurse -Force ".venv"
New-ProjectVenv
} elseif (-not (Test-VenvPython -PythonExe $venvPython)) {
Write-Host "[tests] WARN: existing .venv is invalid for current OS/interpreter; recreating ..."
Remove-Item -Recurse -Force ".venv"
New-ProjectVenv
}
if (-not (Test-VenvPython -PythonExe $venvPython)) {
throw "[tests] ERROR: project venv python is not runnable: $venvPython"
}
$hasPreCommit = $true
try {
& $venvPython -m pre_commit --version | Out-Null
} catch {
& $venvPython -m pre_commit --version | Out-Null
if ($LASTEXITCODE -ne 0) {
$hasPreCommit = $false
}
if (-not $hasPreCommit) {
Write-Host "[tests] Installing pre-commit into project venv ..."
& $venvPython -m pip install -U pip pre-commit
Invoke-Checked "pip install pre-commit" { & $venvPython -m pip install -U pip pre-commit }
}
$hasAiohttp = $true
try {
& $venvPython -c "import aiohttp" | Out-Null
} catch {
& $venvPython -c "import aiohttp" | Out-Null
if ($LASTEXITCODE -ne 0) {
$hasAiohttp = $false
}
if (-not $hasAiohttp) {
Write-Host "[tests] Installing aiohttp into project venv ..."
& $venvPython -m pip install aiohttp
Invoke-Checked "pip install aiohttp" { & $venvPython -m pip install aiohttp }
}
# Ensure Node >= 18
@@ -72,16 +127,16 @@ if ($nodeMajor -lt 18) {
Write-Host "[tests] Node version: $(node -v)"
Write-Host "[tests] 1/4 detect-secrets"
& $venvPython -m pre_commit run detect-secrets --all-files
Invoke-Checked "detect-secrets" { & $venvPython -m pre_commit run detect-secrets --all-files }
Write-Host "[tests] 2/4 pre-commit all hooks"
& $venvPython -m pre_commit run --all-files --show-diff-on-failure
Invoke-Checked "pre-commit all hooks" { & $venvPython -m pre_commit run --all-files --show-diff-on-failure }
Write-Host "[tests] 3/4 backend unit tests"
$env:MOLTBOT_STATE_DIR = "$root\moltbot_state\_local_unit"
& $venvPython scripts\run_unittests.py --start-dir tests --pattern "test_*.py"
Invoke-Checked "backend unit tests" { & $venvPython scripts\run_unittests.py --start-dir tests --pattern "test_*.py" }
Write-Host "[tests] 4/4 frontend E2E"
npm test
Invoke-Checked "frontend E2E" { npm test }
Write-Host "[tests] PASS"
+41
View File
@@ -10,6 +10,7 @@ from typing import Any, Dict, Optional, Set
from .async_utils import run_in_thread
from .comfyui_history import extract_images, fetch_history, get_job_status
from .job_events import JobEventType, get_job_event_store # R71
from .metrics import metrics
from .safe_io import SSRFError, safe_request_json
from .trace_store import trace_store
@@ -94,6 +95,15 @@ async def _watch_and_deliver(
if history_item is None:
logger.warning(f"[Callback] Job {prompt_id} never completed (timed out)")
metrics.inc("callback_timeout")
try:
get_job_event_store().emit(
JobEventType.FAILED,
prompt_id=prompt_id,
trace_id=trace_id or "",
data={"reason": "timeout"},
)
except Exception:
pass
return
# R25: Record completion
@@ -102,6 +112,21 @@ async def _watch_and_deliver(
except Exception:
pass
# R71: Emit lifecycle event (COMPLETED vs ERROR)
status = get_job_status(history_item)
try:
event_type = (
JobEventType.COMPLETED if status == "completed" else JobEventType.FAILED
)
get_job_event_store().emit(
event_type,
prompt_id=prompt_id,
trace_id=trace_id or "",
data={"status": status},
)
except Exception:
pass
# Extract outputs
images = extract_images(history_item) if history_item else []
@@ -133,6 +158,13 @@ async def _watch_and_deliver(
"delivered",
{"host": (url.split("/")[2] if "/" in url else url)},
)
# R71: Emit delivery success
get_job_event_store().emit(
JobEventType.CALLBACK_SENT,
prompt_id=prompt_id,
trace_id=trace_id or "",
data={"target": url},
)
except Exception:
pass
metrics.inc("callback_success")
@@ -148,3 +180,12 @@ async def _watch_and_deliver(
logger.error(f"[Callback] All retries failed for {prompt_id}")
metrics.inc("callback_failed")
try:
get_job_event_store().emit(
JobEventType.CALLBACK_FAILED,
prompt_id=prompt_id,
trace_id=trace_id or "",
data={"target": url, "reason": "max_retries_exceeded"},
)
except Exception:
pass
+4
View File
@@ -45,5 +45,9 @@ def get_capabilities() -> dict:
# R70/F39/R73: Settings contract + UX degradation + Provider governance
"settings_contract": True,
"provider_governance": True,
# F40/R71/R72: Webhook mapping + Job events + Operator doctor
"webhook_mapping": True,
"job_events": True,
"operator_doctor": True,
},
}
+582
View File
@@ -0,0 +1,582 @@
"""
F42 — Constrained Transform Engine.
Optional, auditable, opt-in execution model for advanced webhook payload transforms.
Transform modules execute ONLY from trusted directories with integrity pinning.
Runtime enforces strict limits:
- Timeout per transform
- Output size cap
- CPU/memory budget (best-effort)
- No arbitrary network/filesystem access
- Bounded audit schema for each transform stage
Default posture: DISABLED. Requires OPENCLAW_ENABLE_TRANSFORMS=1.
"""
from __future__ import annotations
import hashlib
import importlib
import importlib.util
import json
import logging
import os
import signal
import sys
import threading
import time
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Set
logger = logging.getLogger("ComfyUI-OpenClaw.services.constrained_transforms")
# ---------------------------------------------------------------------------
# Feature gate
# ---------------------------------------------------------------------------
_FEATURE_FLAG = "OPENCLAW_ENABLE_TRANSFORMS"
def is_transforms_enabled() -> bool:
"""Check if constrained transforms are enabled (default: OFF)."""
val = os.environ.get(_FEATURE_FLAG, "").strip().lower()
return val in ("1", "true", "yes", "on")
# ---------------------------------------------------------------------------
# Runtime limits
# ---------------------------------------------------------------------------
DEFAULT_TRANSFORM_TIMEOUT_SEC = 5
DEFAULT_MAX_OUTPUT_BYTES = 64 * 1024 # 64KB
DEFAULT_MAX_TRANSFORMS_PER_REQUEST = 5
MAX_TRANSFORM_MODULE_SIZE_BYTES = 50 * 1024 # 50KB — prevent loading huge scripts
@dataclass
class TransformLimits:
"""Runtime limits for transform execution."""
timeout_sec: float = DEFAULT_TRANSFORM_TIMEOUT_SEC
max_output_bytes: int = DEFAULT_MAX_OUTPUT_BYTES
max_transforms_per_request: int = DEFAULT_MAX_TRANSFORMS_PER_REQUEST
@classmethod
def from_env(cls) -> "TransformLimits":
"""Load limits from environment variables."""
def _env_int(key: str, default: int) -> int:
try:
return int(os.environ.get(key, str(default)))
except (ValueError, TypeError):
return default
def _env_float(key: str, default: float) -> float:
try:
return float(os.environ.get(key, str(default)))
except (ValueError, TypeError):
return default
return cls(
timeout_sec=_env_float(
"OPENCLAW_TRANSFORM_TIMEOUT", DEFAULT_TRANSFORM_TIMEOUT_SEC
),
max_output_bytes=_env_int(
"OPENCLAW_TRANSFORM_MAX_OUTPUT", DEFAULT_MAX_OUTPUT_BYTES
),
max_transforms_per_request=_env_int(
"OPENCLAW_TRANSFORM_MAX_PER_REQUEST", DEFAULT_MAX_TRANSFORMS_PER_REQUEST
),
)
# ---------------------------------------------------------------------------
# Transform result
# ---------------------------------------------------------------------------
class TransformStatus(str, Enum):
SUCCESS = "success"
ERROR = "error"
TIMEOUT = "timeout"
DENIED = "denied"
SKIPPED = "skipped"
@dataclass
class TransformResult:
"""Result of a single transform execution."""
transform_id: str
status: str # TransformStatus.value
output: Optional[Dict[str, Any]] = None
error: str = ""
duration_ms: float = 0.0
output_bytes: int = 0
audit: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"transform_id": self.transform_id,
"status": self.status,
"duration_ms": round(self.duration_ms, 2),
"output_bytes": self.output_bytes,
}
if self.output is not None:
d["output"] = self.output
if self.error:
d["error"] = self.error
if self.audit:
d["audit"] = self.audit
return d
# ---------------------------------------------------------------------------
# Transform registry (trusted modules)
# ---------------------------------------------------------------------------
@dataclass
class TrustedTransform:
"""A registered, integrity-pinned transform module."""
id: str
label: str
module_path: str # Absolute path to .py module
sha256: str # Integrity hash of the module file
description: str = ""
trusted_source: str = "" # Who published this transform
registered_at: float = 0.0
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
class TransformRegistryError(Exception):
"""Error in transform registry operations."""
pass
class TransformRegistry:
"""
Manages trusted transform modules with integrity pinning.
Transforms can only be loaded from explicitly trusted directories.
Each module is pinned by its SHA256 hash at registration time.
"""
def __init__(self, state_dir: str, trusted_dirs: Optional[List[str]] = None):
self._state_dir = state_dir
self._registry_dir = os.path.join(state_dir, "transforms")
self._index_path = os.path.join(self._registry_dir, "registry.json")
self._transforms: Dict[str, TrustedTransform] = {}
# Trusted directories where transform modules can live
self._trusted_dirs: Set[str] = set()
if trusted_dirs:
for d in trusted_dirs:
resolved = str(Path(d).resolve())
self._trusted_dirs.add(resolved)
os.makedirs(self._registry_dir, exist_ok=True)
self._load()
def _load(self) -> None:
"""Load transform registry from disk."""
if not os.path.exists(self._index_path):
self._transforms = {}
return
try:
with open(self._index_path, "r", encoding="utf-8") as f:
data = json.load(f)
self._transforms = {}
for tid, tdata in data.items():
self._transforms[tid] = TrustedTransform(**tdata)
except Exception as e:
logger.error(f"Failed to load transform registry: {e}")
self._transforms = {}
def _save(self) -> None:
"""Persist transform registry to disk."""
try:
data = {k: v.to_dict() for k, v in self._transforms.items()}
tmp_path = self._index_path + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
os.replace(tmp_path, self._index_path)
except Exception as e:
logger.error(f"Failed to save transform registry: {e}")
@staticmethod
def _compute_sha256(file_path: str) -> str:
"""Compute SHA256 hash of a file."""
h = hashlib.sha256()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
h.update(chunk)
return h.hexdigest()
def _is_in_trusted_dir(self, module_path: str) -> bool:
"""Check if a module path is inside a trusted directory."""
resolved = str(Path(module_path).resolve())
for trusted in self._trusted_dirs:
if resolved.startswith(trusted + os.sep) or resolved == trusted:
return True
return False
def register_transform(
self,
transform_id: str,
module_path: str,
*,
label: str = "",
description: str = "",
trusted_source: str = "",
) -> TrustedTransform:
"""
Register a transform module with integrity pinning.
The module must be in a trusted directory and within size limits.
"""
if not is_transforms_enabled():
raise TransformRegistryError(
f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
)
abs_path = str(Path(module_path).resolve())
# Security: must be in trusted directory
if not self._is_in_trusted_dir(abs_path):
raise TransformRegistryError(
f"Module path is not in a trusted directory: {abs_path}"
)
if not os.path.isfile(abs_path):
raise TransformRegistryError(f"Module file not found: {abs_path}")
# Size check
file_size = os.path.getsize(abs_path)
if file_size > MAX_TRANSFORM_MODULE_SIZE_BYTES:
raise TransformRegistryError(
f"Module exceeds size limit ({file_size} > {MAX_TRANSFORM_MODULE_SIZE_BYTES})"
)
# Must be a .py file
if not abs_path.endswith(".py"):
raise TransformRegistryError("Only .py modules are allowed as transforms")
sha256 = self._compute_sha256(abs_path)
transform = TrustedTransform(
id=transform_id,
label=label or transform_id,
module_path=abs_path,
sha256=sha256,
description=description,
trusted_source=trusted_source,
registered_at=time.time(),
)
self._transforms[transform_id] = transform
self._save()
logger.info(f"F42: Registered transform '{transform_id}' from {abs_path}")
return transform
def unregister_transform(self, transform_id: str) -> bool:
"""Remove a transform from the registry."""
if not is_transforms_enabled():
raise TransformRegistryError(
f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable."
)
if transform_id not in self._transforms:
return False
del self._transforms[transform_id]
self._save()
logger.info(f"F42: Unregistered transform '{transform_id}'")
return True
def get_transform(self, transform_id: str) -> Optional[TrustedTransform]:
"""Get a registered transform by ID."""
return self._transforms.get(transform_id)
def list_transforms(self) -> List[TrustedTransform]:
"""List all registered transforms."""
return list(self._transforms.values())
def verify_integrity(self, transform_id: str) -> bool:
"""Verify that a registered transform's file hasn't been modified."""
transform = self._transforms.get(transform_id)
if not transform:
return False
if not os.path.isfile(transform.module_path):
return False
actual_hash = self._compute_sha256(transform.module_path)
return actual_hash == transform.sha256
# ---------------------------------------------------------------------------
# Constrained executor
# ---------------------------------------------------------------------------
class TransformTimeoutError(Exception):
"""Raised when a transform exceeds its timeout budget."""
pass
class TransformExecutor:
"""
Executes registered transforms with strict runtime constraints.
Enforces:
- Timeout per transform
- Output size cap
- Integrity verification before execution
- No network/filesystem access (best-effort: module is pre-vetted)
- Audit events for each stage
"""
def __init__(
self,
registry: TransformRegistry,
limits: Optional[TransformLimits] = None,
):
self._registry = registry
self._limits = limits or TransformLimits.from_env()
def execute_transform(
self,
transform_id: str,
input_data: Dict[str, Any],
*,
trace_id: str = "",
) -> TransformResult:
"""
Execute a single registered transform within constraints.
The transform module must export a `transform(input_data: dict) -> dict` function.
"""
if not is_transforms_enabled():
return TransformResult(
transform_id=transform_id,
status=TransformStatus.DENIED.value,
error=f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable.",
)
transform = self._registry.get_transform(transform_id)
if not transform:
return TransformResult(
transform_id=transform_id,
status=TransformStatus.ERROR.value,
error=f"Transform '{transform_id}' not found in registry",
)
# Verify integrity before execution
if not self._registry.verify_integrity(transform_id):
return TransformResult(
transform_id=transform_id,
status=TransformStatus.DENIED.value,
error="Integrity verification failed — module may have been modified",
audit={"reason": "integrity_check_failed", "trace_id": trace_id},
)
# Execute with timeout
start_time = time.monotonic()
result_holder: Dict[str, Any] = {}
error_holder: Dict[str, str] = {}
def _run_transform():
try:
# Load the module dynamically
spec = importlib.util.spec_from_file_location(
f"_transform_{transform_id}", transform.module_path
)
if not spec or not spec.loader:
error_holder["error"] = "Failed to load transform module"
return
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module) # type: ignore
# Must export a transform() function
transform_fn = getattr(module, "transform", None)
if not callable(transform_fn):
error_holder["error"] = (
"Module does not export a 'transform(input_data)' function"
)
return
# Execute the transform
output = transform_fn(input_data)
if not isinstance(output, dict):
error_holder["error"] = (
f"Transform must return a dict, got {type(output).__name__}"
)
return
result_holder["output"] = output
except Exception as e:
error_holder["error"] = str(e)
# Run in a thread with timeout
thread = threading.Thread(target=_run_transform, daemon=True)
thread.start()
thread.join(timeout=self._limits.timeout_sec)
elapsed_ms = (time.monotonic() - start_time) * 1000
if thread.is_alive():
return TransformResult(
transform_id=transform_id,
status=TransformStatus.TIMEOUT.value,
error=f"Transform exceeded timeout ({self._limits.timeout_sec}s)",
duration_ms=elapsed_ms,
audit={"timeout_sec": self._limits.timeout_sec, "trace_id": trace_id},
)
if error_holder:
return TransformResult(
transform_id=transform_id,
status=TransformStatus.ERROR.value,
error=error_holder.get("error", "Unknown error"),
duration_ms=elapsed_ms,
audit={"trace_id": trace_id},
)
output = result_holder.get("output", {})
# Check output size
try:
output_json = json.dumps(output, default=str)
output_bytes = len(output_json.encode("utf-8"))
except Exception:
output_bytes = 0
if output_bytes > self._limits.max_output_bytes:
return TransformResult(
transform_id=transform_id,
status=TransformStatus.ERROR.value,
error=f"Output exceeds size limit ({output_bytes} > {self._limits.max_output_bytes})",
duration_ms=elapsed_ms,
output_bytes=output_bytes,
audit={"trace_id": trace_id},
)
return TransformResult(
transform_id=transform_id,
status=TransformStatus.SUCCESS.value,
output=output,
duration_ms=elapsed_ms,
output_bytes=output_bytes,
audit={"trace_id": trace_id},
)
def execute_chain(
self,
transform_ids: List[str],
input_data: Dict[str, Any],
*,
trace_id: str = "",
) -> List[TransformResult]:
"""
Execute a chain of transforms sequentially.
Output of each transform becomes input for the next.
Chain stops on first error/timeout/denial.
"""
if not is_transforms_enabled():
return [
TransformResult(
transform_id="chain",
status=TransformStatus.DENIED.value,
error=f"Transforms disabled. Set {_FEATURE_FLAG}=1 to enable.",
)
]
if len(transform_ids) > self._limits.max_transforms_per_request:
return [
TransformResult(
transform_id="chain",
status=TransformStatus.DENIED.value,
error=f"Transform chain exceeds limit ({len(transform_ids)} > {self._limits.max_transforms_per_request})",
)
]
results: List[TransformResult] = []
current_data = input_data
for tid in transform_ids:
result = self.execute_transform(tid, current_data, trace_id=trace_id)
results.append(result)
if result.status != TransformStatus.SUCCESS.value:
# Stop chain on failure
break
# Pass output as input to next transform
if result.output:
current_data = result.output
return results
# ---------------------------------------------------------------------------
# Module-level convenience
# ---------------------------------------------------------------------------
_registry: Optional[TransformRegistry] = None
_executor: Optional[TransformExecutor] = None
def get_transform_registry() -> TransformRegistry:
"""Get or create the global transform registry."""
global _registry
if _registry is None:
try:
from .state_dir import get_state_dir
state_dir = get_state_dir()
except ImportError:
try:
from services.state_dir import get_state_dir
state_dir = get_state_dir()
except ImportError:
state_dir = os.path.join(
os.path.dirname(os.path.dirname(__file__)), "data"
)
# Default trusted directory: pack-local transforms dir
pack_root = Path(__file__).resolve().parent.parent
trusted_dirs = [str(pack_root / "data" / "transforms")]
# Allow additional trusted dirs from env
extra = os.environ.get("OPENCLAW_TRANSFORM_TRUSTED_DIRS", "")
if extra:
for d in extra.split(os.pathsep):
d = d.strip()
if d:
trusted_dirs.append(d)
_registry = TransformRegistry(state_dir, trusted_dirs=trusted_dirs)
return _registry
def get_transform_executor() -> TransformExecutor:
"""Get or create the global transform executor."""
global _executor
if _executor is None:
_executor = TransformExecutor(get_transform_registry())
return _executor
+210
View File
@@ -0,0 +1,210 @@
"""
R71 — Job Event Stream.
Bounded in-memory event store for job lifecycle transitions.
Provides an SSE endpoint for real-time job status delivery and
a JSON fallback endpoint for polling clients.
Events are derived from queue submission, history polling, and
callback delivery without patching ComfyUI core.
Design:
- Ring-buffer event store with configurable max capacity.
- Each event has a monotonic sequence ID for SSE `id:` field.
- Clients can resume from `Last-Event-ID` header.
- Access control parity with existing observability endpoints.
"""
from __future__ import annotations
import json
import logging
import os
import threading
import time
from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
logger = logging.getLogger("ComfyUI-OpenClaw.services.job_events")
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
MAX_EVENT_BUFFER = int(os.environ.get("OPENCLAW_JOB_EVENT_BUFFER_SIZE", "500"))
EVENT_TTL_SEC = int(os.environ.get("OPENCLAW_JOB_EVENT_TTL_SEC", "600")) # 10 minutes
# ---------------------------------------------------------------------------
# Event types
# ---------------------------------------------------------------------------
class JobEventType(Enum):
"""Lifecycle events for prompt/job tracking."""
QUEUED = "queued" # Job submitted to ComfyUI queue
RUNNING = "running" # Job execution started
COMPLETED = "completed" # Job finished successfully
FAILED = "failed" # Job failed with error
CANCELLED = "cancelled" # Job was cancelled
CALLBACK_SENT = "callback_sent" # Callback delivery succeeded
CALLBACK_FAILED = "callback_failed" # Callback delivery failed
# ---------------------------------------------------------------------------
# Event data class
# ---------------------------------------------------------------------------
@dataclass
class JobEvent:
"""A single job lifecycle event."""
seq: int # Monotonic sequence number (SSE id)
event_type: str # JobEventType.value
prompt_id: str
trace_id: str = ""
timestamp: float = 0.0
data: Dict[str, Any] = field(default_factory=dict)
def __post_init__(self) -> None:
if self.timestamp == 0.0:
self.timestamp = time.time()
def to_sse(self) -> str:
"""Format as an SSE event string."""
payload = {
"event_type": self.event_type,
"prompt_id": self.prompt_id,
"trace_id": self.trace_id,
"timestamp": self.timestamp,
"data": self.data,
}
lines = [
f"id: {self.seq}",
f"event: {self.event_type}",
f"data: {json.dumps(payload, separators=(',', ':'))}",
"",
"",
]
return "\n".join(lines)
def to_dict(self) -> Dict[str, Any]:
"""Serialise for JSON polling responses."""
return {
"seq": self.seq,
"event_type": self.event_type,
"prompt_id": self.prompt_id,
"trace_id": self.trace_id,
"timestamp": self.timestamp,
"data": self.data,
}
# ---------------------------------------------------------------------------
# Bounded event store (ring buffer)
# ---------------------------------------------------------------------------
class JobEventStore:
"""
Thread-safe bounded ring-buffer for job events.
Supports:
- emit(): add events
- events_since(seq): retrieve events after a given sequence ID
- SSE client resume via Last-Event-ID
"""
def __init__(self, max_size: int = MAX_EVENT_BUFFER) -> None:
self._lock = threading.Lock()
self._events: List[JobEvent] = []
self._max_size = max_size
self._seq_counter = 0
def emit(
self,
event_type: JobEventType,
prompt_id: str,
trace_id: str = "",
data: Optional[Dict[str, Any]] = None,
) -> JobEvent:
"""Record a new job event and return it."""
with self._lock:
self._seq_counter += 1
evt = JobEvent(
seq=self._seq_counter,
event_type=event_type.value,
prompt_id=prompt_id,
trace_id=trace_id,
data=data or {},
)
self._events.append(evt)
# Evict oldest if over capacity
if len(self._events) > self._max_size:
self._events = self._events[-self._max_size :]
return evt
def events_since(
self,
last_seq: int = 0,
limit: int = 100,
prompt_id: Optional[str] = None,
) -> List[JobEvent]:
"""
Return events with seq > last_seq, optionally filtered by prompt_id.
Returns at most `limit` events (oldest first).
"""
now = time.time()
with self._lock:
results = []
for evt in self._events:
if evt.seq <= last_seq:
continue
if now - evt.timestamp > EVENT_TTL_SEC:
continue
if prompt_id and evt.prompt_id != prompt_id:
continue
results.append(evt)
if len(results) >= limit:
break
return results
def latest_seq(self) -> int:
"""Return the latest sequence number."""
with self._lock:
return self._seq_counter
@property
def size(self) -> int:
with self._lock:
return len(self._events)
def clear(self) -> None:
"""Clear all events (used in tests)."""
with self._lock:
self._events.clear()
self._seq_counter = 0
# ---------------------------------------------------------------------------
# Singleton
# ---------------------------------------------------------------------------
_store: Optional[JobEventStore] = None
def get_job_event_store() -> JobEventStore:
"""Get or create the global job event store."""
global _store
if _store is None:
_store = JobEventStore()
return _store
def reset_job_event_store() -> None:
"""Reset the global store (test utility)."""
global _store
_store = None
+520
View File
@@ -0,0 +1,520 @@
"""
R72 — Operator Doctor CLI.
One-command diagnostics for deployment readiness and runtime health.
Read-only checks only; no auto-remediation.
Checks:
- Release-gate: required contract files, feature-flag policy, route health
- Runtime: .venv usage, Python/Node versions, Windows pre-commit/cache pitfalls
- Config/Token: state-dir permissions, token posture, env key presence
Usage:
python -m services.operator_doctor
python scripts/operator_doctor.py
"""
from __future__ import annotations
import importlib
import json
import os
import platform
import shutil
import subprocess
import sys
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional
# ---------------------------------------------------------------------------
# Result types
# ---------------------------------------------------------------------------
class Severity(Enum):
PASS = "pass"
WARN = "warn"
FAIL = "fail"
SKIP = "skip"
@dataclass
class CheckResult:
"""Result of a single diagnostic check."""
name: str
severity: str # Severity.value
message: str
detail: str = ""
remediation: str = ""
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"name": self.name,
"severity": self.severity,
"message": self.message,
}
if self.detail:
d["detail"] = self.detail
if self.remediation:
d["remediation"] = self.remediation
return d
@dataclass
class DoctorReport:
"""Aggregated diagnostic report."""
checks: List[CheckResult] = field(default_factory=list)
environment: Dict[str, str] = field(default_factory=dict)
summary: Dict[str, int] = field(default_factory=dict)
def add(self, result: CheckResult) -> None:
self.checks.append(result)
def build_summary(self) -> None:
counts: Dict[str, int] = {"pass": 0, "warn": 0, "fail": 0, "skip": 0}
for c in self.checks:
counts[c.severity] = counts.get(c.severity, 0) + 1
self.summary = counts
@property
def has_failures(self) -> bool:
return any(c.severity == Severity.FAIL.value for c in self.checks)
def to_dict(self) -> Dict[str, Any]:
self.build_summary()
return {
"environment": self.environment,
"checks": [c.to_dict() for c in self.checks],
"summary": self.summary,
}
def to_human(self) -> str:
"""Human-readable report output."""
self.build_summary()
lines: List[str] = []
lines.append("=" * 60)
lines.append(" OpenClaw Operator Doctor Report")
lines.append("=" * 60)
lines.append("")
# Environment
lines.append("Environment:")
for k, v in self.environment.items():
lines.append(f" {k}: {v}")
lines.append("")
# Checks grouped by severity
for sev in [Severity.FAIL, Severity.WARN, Severity.PASS, Severity.SKIP]:
checks = [c for c in self.checks if c.severity == sev.value]
if not checks:
continue
icon = {"pass": "", "warn": "", "fail": "", "skip": ""}[sev.value]
lines.append(f" [{icon}] {sev.value.upper()} ({len(checks)})")
for c in checks:
lines.append(f" {c.name}: {c.message}")
if c.detail:
lines.append(f" Detail: {c.detail}")
if c.remediation:
lines.append(f" Fix: {c.remediation}")
lines.append("")
# Summary
total = sum(self.summary.values())
lines.append("-" * 60)
lines.append(
f" Total: {total} | "
f"Pass: {self.summary.get('pass', 0)} | "
f"Warn: {self.summary.get('warn', 0)} | "
f"Fail: {self.summary.get('fail', 0)} | "
f"Skip: {self.summary.get('skip', 0)}"
)
lines.append("=" * 60)
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Pack root detection
# ---------------------------------------------------------------------------
def _get_pack_root() -> Path:
"""Detect the ComfyUI-OpenClaw pack root directory."""
# Try relative to this file
this_dir = Path(__file__).resolve().parent
candidate = this_dir.parent
if (candidate / "ROADMAP.md").exists():
return candidate
# Fallback to cwd
return Path.cwd()
# ---------------------------------------------------------------------------
# Individual checks
# ---------------------------------------------------------------------------
def check_python_version(report: DoctorReport) -> None:
ver = sys.version_info
report.environment["python"] = f"{ver.major}.{ver.minor}.{ver.micro}"
if ver.major == 3 and ver.minor >= 10:
report.add(
CheckResult(
name="python_version",
severity=Severity.PASS.value,
message=f"Python {ver.major}.{ver.minor}.{ver.micro}",
)
)
else:
report.add(
CheckResult(
name="python_version",
severity=Severity.FAIL.value,
message=f"Python {ver.major}.{ver.minor}.{ver.micro} (need >=3.10)",
remediation="Install Python 3.10 or later.",
)
)
def check_node_version(report: DoctorReport) -> None:
node = shutil.which("node")
if not node:
report.add(
CheckResult(
name="node_version",
severity=Severity.WARN.value,
message="Node.js not found on PATH",
remediation="Install Node.js 18+ for frontend E2E tests.",
)
)
return
try:
out = subprocess.check_output([node, "--version"], text=True, timeout=5).strip()
report.environment["node"] = out
major = int(out.lstrip("v").split(".")[0])
if major >= 18:
report.add(
CheckResult(
name="node_version",
severity=Severity.PASS.value,
message=f"Node.js {out}",
)
)
else:
report.add(
CheckResult(
name="node_version",
severity=Severity.FAIL.value,
message=f"Node.js {out} (need >=18)",
remediation="Upgrade to Node.js 18 or later.",
)
)
except Exception as e:
report.add(
CheckResult(
name="node_version",
severity=Severity.WARN.value,
message=f"Could not determine Node.js version: {e}",
)
)
def check_venv(report: DoctorReport) -> None:
in_venv = sys.prefix != sys.base_prefix
report.environment["in_venv"] = str(in_venv)
if in_venv:
report.add(
CheckResult(
name="venv_active",
severity=Severity.PASS.value,
message="Running inside a virtual environment",
)
)
else:
report.add(
CheckResult(
name="venv_active",
severity=Severity.WARN.value,
message="Not running inside a virtual environment",
remediation="Use a project-local .venv for isolation.",
)
)
def check_contract_files(report: DoctorReport, pack_root: Path) -> None:
"""Check that required release-gate contract files exist."""
required_files = [
"docs/release/api_contract.md",
"docs/release/config_secrets_contract.md",
"docs/release/compatibility_matrix.md",
"docs/release/support_policy.md",
"docs/release/ci_regression_policy.md",
"RELEASE_CHECKLIST.md",
"SECURITY.md",
"tests/TEST_SOP.md",
]
for rel_path in required_files:
full = pack_root / rel_path
if full.exists():
report.add(
CheckResult(
name=f"contract_file:{rel_path}",
severity=Severity.PASS.value,
message=f"Found: {rel_path}",
)
)
else:
report.add(
CheckResult(
name=f"contract_file:{rel_path}",
severity=Severity.FAIL.value,
message=f"Missing required contract file: {rel_path}",
remediation=f"Create or restore {rel_path}.",
)
)
def check_state_dir(report: DoctorReport) -> None:
"""Check state directory accessibility."""
state_dir = os.environ.get("MOLTBOT_STATE_DIR") or os.environ.get(
"OPENCLAW_STATE_DIR"
)
if not state_dir:
report.add(
CheckResult(
name="state_dir",
severity=Severity.PASS.value,
message="Using default state directory (user data dir)",
)
)
return
p = Path(state_dir)
if not p.exists():
report.add(
CheckResult(
name="state_dir",
severity=Severity.WARN.value,
message=f"State dir does not exist: {state_dir}",
remediation="The directory will be created on first run.",
)
)
elif not os.access(str(p), os.W_OK):
report.add(
CheckResult(
name="state_dir",
severity=Severity.FAIL.value,
message=f"State dir not writable: {state_dir}",
remediation="Check file permissions.",
)
)
else:
report.add(
CheckResult(
name="state_dir",
severity=Severity.PASS.value,
message=f"State dir OK: {state_dir}",
)
)
def check_token_posture(report: DoctorReport) -> None:
"""Check admin/observability token configuration."""
admin_token = os.environ.get("OPENCLAW_ADMIN_TOKEN") or os.environ.get(
"MOLTBOT_ADMIN_TOKEN"
)
obs_token = os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN") or os.environ.get(
"MOLTBOT_OBSERVABILITY_TOKEN"
)
if admin_token:
report.add(
CheckResult(
name="admin_token",
severity=Severity.PASS.value,
message="Admin token configured",
)
)
else:
report.add(
CheckResult(
name="admin_token",
severity=Severity.WARN.value,
message="No admin token — loopback-only convenience mode",
detail="Remote admin access is denied by default.",
)
)
if obs_token:
report.add(
CheckResult(
name="observability_token",
severity=Severity.PASS.value,
message="Observability token configured",
)
)
else:
report.add(
CheckResult(
name="observability_token",
severity=Severity.WARN.value,
message="No observability token — loopback-only",
)
)
def check_pre_commit(report: DoctorReport) -> None:
"""Check pre-commit availability."""
pre_commit = shutil.which("pre-commit")
if pre_commit:
report.add(
CheckResult(
name="pre_commit",
severity=Severity.PASS.value,
message="pre-commit found on PATH",
)
)
return
# Try as Python module
try:
importlib.import_module("pre_commit")
report.add(
CheckResult(
name="pre_commit",
severity=Severity.PASS.value,
message="pre-commit available as Python module",
)
)
except ImportError:
report.add(
CheckResult(
name="pre_commit",
severity=Severity.WARN.value,
message="pre-commit not found",
remediation="pip install pre-commit (required for SOP validation).",
)
)
def check_core_imports(report: DoctorReport) -> None:
"""Verify core service modules can be imported."""
modules = [
"services.runtime_config",
"services.capabilities",
"services.webhook_auth",
"services.templates",
"services.llm_client",
"services.metrics",
]
for mod_name in modules:
try:
importlib.import_module(mod_name)
report.add(
CheckResult(
name=f"import:{mod_name}",
severity=Severity.PASS.value,
message=f"OK: {mod_name}",
)
)
except Exception as e:
report.add(
CheckResult(
name=f"import:{mod_name}",
severity=Severity.FAIL.value,
message=f"Import failed: {mod_name}",
detail=str(e),
remediation=f"Check for missing dependencies or circular imports.",
)
)
def check_os_environment(report: DoctorReport) -> None:
"""Record OS environment info."""
report.environment["os"] = platform.system()
report.environment["os_version"] = platform.version()
report.environment["arch"] = platform.machine()
if platform.system() == "Windows":
# Check for common Windows pitfalls
long_path = os.environ.get("MSYS_NO_PATHCONV")
report.add(
CheckResult(
name="windows_env",
severity=Severity.PASS.value,
message="Windows environment detected",
detail=f"Architecture: {platform.machine()}",
)
)
# ---------------------------------------------------------------------------
# Main runner
# ---------------------------------------------------------------------------
def run_doctor(pack_root: Optional[Path] = None) -> DoctorReport:
"""Run all diagnostic checks and return a report."""
report = DoctorReport()
if pack_root is None:
pack_root = _get_pack_root()
report.environment["pack_root"] = str(pack_root)
# Run checks
check_os_environment(report)
check_python_version(report)
check_node_version(report)
check_venv(report)
check_pre_commit(report)
check_state_dir(report)
check_token_posture(report)
check_contract_files(report, pack_root)
check_core_imports(report)
report.build_summary()
return report
# ---------------------------------------------------------------------------
# CLI entrypoint
# ---------------------------------------------------------------------------
def main() -> None:
"""CLI entrypoint for operator doctor."""
import argparse
parser = argparse.ArgumentParser(
description="OpenClaw Operator Doctor — deployment readiness diagnostics"
)
parser.add_argument(
"--json",
action="store_true",
help="Output machine-readable JSON instead of human-readable text",
)
parser.add_argument(
"--pack-root",
type=str,
default=None,
help="Override pack root directory detection",
)
args = parser.parse_args()
pack_root = Path(args.pack_root) if args.pack_root else None
report = run_doctor(pack_root)
if args.json:
print(json.dumps(report.to_dict(), indent=2))
else:
print(report.to_human())
sys.exit(1 if report.has_failures else 0)
if __name__ == "__main__":
main()
+389
View File
@@ -0,0 +1,389 @@
"""
F41 — Registry Quarantine Service.
Remote pack registry sync with quarantine/trust gates.
All remote registry features are disabled by default (fail-closed).
Features:
- Remote registry metadata fetch with signature/hash/provenance verification
- Quarantine state: imported packs must be explicitly activated
- Audit records for fetch/verify/quarantine/activate/rollback actions
- Explicit operator action required for all state transitions
Default posture: DISABLED. Requires OPENCLAW_ENABLE_REGISTRY_SYNC=1.
"""
from __future__ import annotations
import hashlib
import json
import logging
import os
import time
from dataclasses import asdict, dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional
logger = logging.getLogger("ComfyUI-OpenClaw.services.registry_quarantine")
# ---------------------------------------------------------------------------
# Feature gate
# ---------------------------------------------------------------------------
_FEATURE_FLAG = "OPENCLAW_ENABLE_REGISTRY_SYNC"
def is_registry_sync_enabled() -> bool:
"""Check if remote registry sync is enabled (default: OFF)."""
val = os.environ.get(_FEATURE_FLAG, "").strip().lower()
return val in ("1", "true", "yes", "on")
# ---------------------------------------------------------------------------
# Quarantine states
# ---------------------------------------------------------------------------
class QuarantineState(str, Enum):
"""Pack quarantine lifecycle states."""
FETCHED = "fetched" # Downloaded but not verified
VERIFIED = "verified" # Integrity verified but not activated
QUARANTINED = "quarantined" # Held for operator review
ACTIVATED = "activated" # Approved and ready for use
REJECTED = "rejected" # Explicitly rejected by operator
ROLLED_BACK = "rolled_back" # Previously activated, now rolled back
# ---------------------------------------------------------------------------
# Registry entry
# ---------------------------------------------------------------------------
@dataclass
class RegistryEntry:
"""A pack entry in the registry quarantine system."""
name: str
version: str
source_url: str = ""
state: str = QuarantineState.FETCHED.value
sha256: str = ""
signature: str = "" # Optional signature for provenance
provenance: str = "" # Author/publisher provenance info
fetched_at: float = 0.0
verified_at: float = 0.0
activated_at: float = 0.0
rejected_at: float = 0.0
rejection_reason: str = ""
audit_trail: List[Dict[str, Any]] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "RegistryEntry":
trail = data.pop("audit_trail", [])
entry = cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
entry.audit_trail = trail
return entry
def add_audit(self, action: str, detail: str = "") -> None:
"""Append an audit record to this entry's trail."""
self.audit_trail.append(
{
"action": action,
"timestamp": time.time(),
"detail": detail,
}
)
# ---------------------------------------------------------------------------
# Registry quarantine store
# ---------------------------------------------------------------------------
class RegistryQuarantineError(Exception):
"""Error in registry quarantine operations."""
pass
class RegistryQuarantineStore:
"""
Manages the quarantine lifecycle for remote pack registry entries.
All state is persisted to a JSON file in the state directory.
All operations require the feature flag to be ON.
"""
MAX_ENTRIES = 200 # Cap total registry entries
def __init__(self, state_dir: str):
self._state_dir = state_dir
self._quarantine_dir = os.path.join(state_dir, "registry", "quarantine")
self._index_path = os.path.join(self._quarantine_dir, "index.json")
self._entries: Dict[str, RegistryEntry] = {}
os.makedirs(self._quarantine_dir, exist_ok=True)
self._load()
def _entry_key(self, name: str, version: str) -> str:
return f"{name}@{version}"
def _load(self) -> None:
"""Load registry index from disk."""
if not os.path.exists(self._index_path):
self._entries = {}
return
try:
with open(self._index_path, "r", encoding="utf-8") as f:
data = json.load(f)
self._entries = {}
for key, entry_data in data.items():
try:
self._entries[key] = RegistryEntry.from_dict(entry_data)
except Exception as e:
logger.warning(f"Skipping corrupt registry entry {key}: {e}")
except Exception as e:
logger.error(f"Failed to load registry index: {e}")
self._entries = {}
def _save(self) -> None:
"""Persist registry index to disk."""
try:
data = {k: v.to_dict() for k, v in self._entries.items()}
tmp_path = self._index_path + ".tmp"
with open(tmp_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
f.write("\n")
os.replace(tmp_path, self._index_path)
except Exception as e:
logger.error(f"Failed to save registry index: {e}")
raise RegistryQuarantineError(f"Failed to persist registry state: {e}")
def _require_enabled(self) -> None:
"""Fail-closed if feature is not enabled."""
if not is_registry_sync_enabled():
raise RegistryQuarantineError(
f"Remote registry sync is disabled. Set {_FEATURE_FLAG}=1 to enable."
)
# ----- Public API -----
def register_fetch(
self,
name: str,
version: str,
source_url: str,
sha256: str,
*,
signature: str = "",
provenance: str = "",
) -> RegistryEntry:
"""
Register a newly fetched pack in quarantine.
The pack enters FETCHED state and requires explicit verification + activation.
"""
self._require_enabled()
if len(self._entries) >= self.MAX_ENTRIES:
raise RegistryQuarantineError(
f"Registry entry limit reached ({self.MAX_ENTRIES}). "
"Remove old entries before adding new ones."
)
key = self._entry_key(name, version)
entry = RegistryEntry(
name=name,
version=version,
source_url=source_url,
state=QuarantineState.FETCHED.value,
sha256=sha256,
signature=signature,
provenance=provenance,
fetched_at=time.time(),
)
entry.add_audit("fetch", f"Fetched from {source_url}")
self._entries[key] = entry
self._save()
logger.info(f"F41: Registered pack {key} in quarantine (FETCHED)")
return entry
def verify_integrity(
self,
name: str,
version: str,
actual_sha256: str,
) -> bool:
"""
Verify a fetched pack's integrity against its registered hash.
Transitions: FETCHED → VERIFIED (on success) or QUARANTINED (on failure).
"""
self._require_enabled()
key = self._entry_key(name, version)
entry = self._entries.get(key)
if not entry:
raise RegistryQuarantineError(f"No registry entry for {key}")
if entry.state not in (
QuarantineState.FETCHED.value,
QuarantineState.QUARANTINED.value,
):
raise RegistryQuarantineError(
f"Cannot verify pack in state '{entry.state}' (must be fetched or quarantined)"
)
if actual_sha256 == entry.sha256:
entry.state = QuarantineState.VERIFIED.value
entry.verified_at = time.time()
entry.add_audit("verify", "Integrity check passed")
self._save()
logger.info(f"F41: Pack {key} integrity verified")
return True
else:
entry.state = QuarantineState.QUARANTINED.value
entry.add_audit(
"verify_failed",
f"Hash mismatch: expected {entry.sha256}, got {actual_sha256}",
)
self._save()
logger.warning(f"F41: Pack {key} integrity FAILED — moved to quarantine")
return False
def activate(self, name: str, version: str) -> RegistryEntry:
"""
Activate a verified pack for use.
Requires explicit operator action. Only VERIFIED packs can be activated.
"""
self._require_enabled()
key = self._entry_key(name, version)
entry = self._entries.get(key)
if not entry:
raise RegistryQuarantineError(f"No registry entry for {key}")
if entry.state != QuarantineState.VERIFIED.value:
raise RegistryQuarantineError(
f"Cannot activate pack in state '{entry.state}' (must be verified first)"
)
entry.state = QuarantineState.ACTIVATED.value
entry.activated_at = time.time()
entry.add_audit("activate", "Operator-approved activation")
self._save()
logger.info(f"F41: Pack {key} activated")
return entry
def reject(self, name: str, version: str, reason: str = "") -> RegistryEntry:
"""Reject a quarantined or fetched pack."""
self._require_enabled()
key = self._entry_key(name, version)
entry = self._entries.get(key)
if not entry:
raise RegistryQuarantineError(f"No registry entry for {key}")
if entry.state == QuarantineState.ACTIVATED.value:
raise RegistryQuarantineError(
"Cannot reject an activated pack — use rollback first."
)
entry.state = QuarantineState.REJECTED.value
entry.rejected_at = time.time()
entry.rejection_reason = reason
entry.add_audit("reject", reason or "Operator rejected")
self._save()
logger.info(f"F41: Pack {key} rejected")
return entry
def rollback(self, name: str, version: str, reason: str = "") -> RegistryEntry:
"""Roll back a previously activated pack."""
self._require_enabled()
key = self._entry_key(name, version)
entry = self._entries.get(key)
if not entry:
raise RegistryQuarantineError(f"No registry entry for {key}")
if entry.state != QuarantineState.ACTIVATED.value:
raise RegistryQuarantineError(
f"Cannot rollback pack in state '{entry.state}' (must be activated)"
)
entry.state = QuarantineState.ROLLED_BACK.value
entry.add_audit("rollback", reason or "Operator-initiated rollback")
self._save()
logger.info(f"F41: Pack {key} rolled back")
return entry
def get_entry(self, name: str, version: str) -> Optional[RegistryEntry]:
"""Get a single registry entry."""
key = self._entry_key(name, version)
return self._entries.get(key)
def list_entries(
self,
*,
state_filter: Optional[str] = None,
) -> List[RegistryEntry]:
"""List all registry entries, optionally filtered by state."""
entries = list(self._entries.values())
if state_filter:
entries = [e for e in entries if e.state == state_filter]
return entries
def remove_entry(self, name: str, version: str) -> bool:
"""Remove a registry entry (must be rejected or rolled_back)."""
self._require_enabled()
key = self._entry_key(name, version)
entry = self._entries.get(key)
if not entry:
return False
if entry.state not in (
QuarantineState.REJECTED.value,
QuarantineState.ROLLED_BACK.value,
):
raise RegistryQuarantineError(
f"Cannot remove entry in state '{entry.state}' — reject or rollback first."
)
del self._entries[key]
self._save()
logger.info(f"F41: Removed registry entry {key}")
return True
# ---------------------------------------------------------------------------
# Module-level convenience
# ---------------------------------------------------------------------------
_store: Optional[RegistryQuarantineStore] = None
def get_quarantine_store() -> RegistryQuarantineStore:
"""Get or create the global quarantine store singleton."""
global _store
if _store is None:
try:
from .state_dir import get_state_dir
state_dir = get_state_dir()
except ImportError:
try:
from services.state_dir import get_state_dir
state_dir = get_state_dir()
except ImportError:
state_dir = os.path.join(
os.path.dirname(os.path.dirname(__file__)), "data"
)
_store = RegistryQuarantineStore(state_dir)
return _store
+12 -7
View File
@@ -13,14 +13,16 @@ logger = logging.getLogger("ComfyUI-OpenClaw.services.runtime_config")
# R70: Settings schema registry (type coercion + unknown-key rejection)
try:
from .settings_schema import (
coerce_dict as _schema_coerce,
get_schema_map,
is_registered as _schema_registered,
)
from .settings_schema import coerce_dict as _schema_coerce
from .settings_schema import get_schema_map
from .settings_schema import is_registered as _schema_registered
except ImportError:
try:
from services.settings_schema import coerce_dict as _schema_coerce, get_schema_map, is_registered as _schema_registered # type: ignore
from services.settings_schema import (
coerce_dict as _schema_coerce, # type: ignore
)
from services.settings_schema import get_schema_map
from services.settings_schema import is_registered as _schema_registered
except ImportError:
# Fail-open: no schema enforcement if module missing
def _schema_coerce(updates): # type: ignore
@@ -349,7 +351,10 @@ def validate_config_update(updates: Dict[str, Any]) -> Tuple[Dict[str, Any], lis
valid_providers = set(list_providers())
except ImportError:
try:
from services.providers.catalog import list_providers, normalize_provider_id # type: ignore
from services.providers.catalog import ( # type: ignore
list_providers,
normalize_provider_id,
)
val = normalize_provider_id(val)
valid_providers = set(list_providers())
+918
View File
@@ -0,0 +1,918 @@
"""
S30 ComfyUI-Aware Security Doctor.
Deploy-time and runtime security diagnostics specific to ComfyUI extension operations.
Read-only checks by default; optional guarded remediation for safe/local actions only.
Checks:
- Endpoint exposure: detect non-loopback access without token
- Token boundaries: admin vs observability token posture
- SSRF posture: callback_url / base_url allowlist wildcard misuse
- State-dir permissions: writable, world-readable checks
- Redaction drift: verify redaction patterns cover known sensitive keys
- ComfyUI runtime mode: Desktop/portable/venv compatibility
- Feature flag posture: high-risk features default-off check
Usage:
from services.security_doctor import run_security_doctor
report = run_security_doctor()
"""
from __future__ import annotations
import ipaddress
import json
import logging
import os
import platform
import stat
import sys
from dataclasses import asdict, dataclass, field
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Set
logger = logging.getLogger("ComfyUI-OpenClaw.services.security_doctor")
# ---------------------------------------------------------------------------
# Severity + Result types (reuse operator_doctor patterns)
# ---------------------------------------------------------------------------
class SecuritySeverity(str, Enum):
PASS = "pass"
WARN = "warn"
FAIL = "fail"
SKIP = "skip"
INFO = "info"
@dataclass
class SecurityCheckResult:
"""Result of a single security diagnostic check."""
name: str
severity: str # SecuritySeverity.value
message: str
category: str = "" # e.g. "endpoint", "token", "ssrf", "state_dir", "redaction"
detail: str = ""
remediation: str = ""
def to_dict(self) -> Dict[str, Any]:
d: Dict[str, Any] = {
"name": self.name,
"severity": self.severity,
"message": self.message,
"category": self.category,
}
if self.detail:
d["detail"] = self.detail
if self.remediation:
d["remediation"] = self.remediation
return d
@dataclass
class SecurityReport:
"""Aggregated security diagnostic report."""
checks: List[SecurityCheckResult] = field(default_factory=list)
environment: Dict[str, str] = field(default_factory=dict)
summary: Dict[str, int] = field(default_factory=dict)
remediation_applied: List[str] = field(default_factory=list)
def add(self, result: SecurityCheckResult) -> None:
self.checks.append(result)
def build_summary(self) -> None:
counts: Dict[str, int] = {}
for c in self.checks:
counts[c.severity] = counts.get(c.severity, 0) + 1
self.summary = counts
@property
def has_failures(self) -> bool:
return any(c.severity == SecuritySeverity.FAIL.value for c in self.checks)
@property
def risk_score(self) -> int:
"""Compute a simple risk score: FAIL=10, WARN=3, INFO/PASS/SKIP=0."""
score = 0
for c in self.checks:
if c.severity == SecuritySeverity.FAIL.value:
score += 10
elif c.severity == SecuritySeverity.WARN.value:
score += 3
return score
def to_dict(self) -> Dict[str, Any]:
self.build_summary()
return {
"environment": self.environment,
"checks": [c.to_dict() for c in self.checks],
"summary": self.summary,
"risk_score": self.risk_score,
"remediation_applied": self.remediation_applied,
}
def to_human(self) -> str:
"""Human-readable security report."""
self.build_summary()
lines: List[str] = []
lines.append("=" * 64)
lines.append(" OpenClaw Security Doctor Report")
lines.append("=" * 64)
lines.append("")
# Environment
lines.append("Environment:")
for k, v in self.environment.items():
lines.append(f" {k}: {v}")
lines.append("")
# Group by category
categories: Dict[str, List[SecurityCheckResult]] = {}
for c in self.checks:
cat = c.category or "general"
categories.setdefault(cat, []).append(c)
for cat, checks in categories.items():
lines.append(f" [{cat.upper()}]")
for c in checks:
icon = {
"pass": "",
"warn": "",
"fail": "",
"skip": "",
"info": "",
}.get(c.severity, "?")
lines.append(f" [{icon}] {c.name}: {c.message}")
if c.detail:
lines.append(f" Detail: {c.detail}")
if c.remediation:
lines.append(f" Fix: {c.remediation}")
lines.append("")
# Risk score
lines.append("-" * 64)
lines.append(f" Risk Score: {self.risk_score}")
total = sum(self.summary.values())
lines.append(
f" Total: {total} | "
f"Fail: {self.summary.get('fail', 0)} | "
f"Warn: {self.summary.get('warn', 0)} | "
f"Pass: {self.summary.get('pass', 0)} | "
f"Skip: {self.summary.get('skip', 0)}"
)
if self.remediation_applied:
lines.append(f" Remediations applied: {len(self.remediation_applied)}")
for r in self.remediation_applied:
lines.append(f" - {r}")
lines.append("=" * 64)
return "\n".join(lines)
# ---------------------------------------------------------------------------
# Pack root detection
# ---------------------------------------------------------------------------
def _get_pack_root() -> Path:
"""Detect the ComfyUI-OpenClaw pack root directory."""
this_dir = Path(__file__).resolve().parent
candidate = this_dir.parent
if (candidate / "pyproject.toml").exists():
return candidate
return Path.cwd()
# ---------------------------------------------------------------------------
# Security checks — Endpoint exposure
# ---------------------------------------------------------------------------
def check_endpoint_exposure(report: SecurityReport) -> None:
"""Check if endpoints are exposed without token protection."""
admin_token = os.environ.get("OPENCLAW_ADMIN_TOKEN") or os.environ.get(
"MOLTBOT_ADMIN_TOKEN"
)
obs_token = os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN") or os.environ.get(
"MOLTBOT_OBSERVABILITY_TOKEN"
)
if not admin_token and not obs_token:
report.add(
SecurityCheckResult(
name="endpoint_exposure",
severity=SecuritySeverity.WARN.value,
message="No admin or observability tokens configured — loopback-only mode",
category="endpoint",
detail="All admin/observability endpoints require loopback access.",
remediation="Set OPENCLAW_ADMIN_TOKEN and OPENCLAW_OBSERVABILITY_TOKEN for remote deployments.",
)
)
elif not admin_token:
report.add(
SecurityCheckResult(
name="admin_token_missing",
severity=SecuritySeverity.WARN.value,
message="No admin token — config/secrets endpoints in convenience mode",
category="endpoint",
remediation="Set OPENCLAW_ADMIN_TOKEN for production deployments.",
)
)
else:
report.add(
SecurityCheckResult(
name="admin_token_set",
severity=SecuritySeverity.PASS.value,
message="Admin token configured",
category="endpoint",
)
)
if obs_token:
report.add(
SecurityCheckResult(
name="observability_token_set",
severity=SecuritySeverity.PASS.value,
message="Observability token configured",
category="endpoint",
)
)
# ---------------------------------------------------------------------------
# Security checks — Token boundaries
# ---------------------------------------------------------------------------
def check_token_boundaries(report: SecurityReport) -> None:
"""Verify admin and observability tokens are distinct."""
admin_token = (
os.environ.get("OPENCLAW_ADMIN_TOKEN")
or os.environ.get("MOLTBOT_ADMIN_TOKEN")
or ""
).strip()
obs_token = (
os.environ.get("OPENCLAW_OBSERVABILITY_TOKEN")
or os.environ.get("MOLTBOT_OBSERVABILITY_TOKEN")
or ""
).strip()
if admin_token and obs_token and admin_token == obs_token:
report.add(
SecurityCheckResult(
name="token_reuse",
severity=SecuritySeverity.FAIL.value,
message="Admin and observability tokens are identical — privilege confusion risk",
category="token",
remediation="Use distinct tokens for admin and observability access.",
)
)
elif admin_token and obs_token:
report.add(
SecurityCheckResult(
name="token_separation",
severity=SecuritySeverity.PASS.value,
message="Admin and observability tokens are distinct",
category="token",
)
)
# Check token strength (minimum length)
for label, token in [("admin", admin_token), ("observability", obs_token)]:
if token and len(token) < 16:
report.add(
SecurityCheckResult(
name=f"{label}_token_weak",
severity=SecuritySeverity.WARN.value,
message=f"{label.title()} token is short ({len(token)} chars) — consider longer tokens",
category="token",
remediation=f"Use a {label} token of at least 16 characters.",
)
)
# ---------------------------------------------------------------------------
# Security checks — SSRF posture
# ---------------------------------------------------------------------------
def check_ssrf_posture(report: SecurityReport) -> None:
"""Check callback/base_url configurations for SSRF risk indicators."""
# Check OPENCLAW_CALLBACK_ALLOWLIST for wildcard abuse
callback_allowlist = os.environ.get(
"OPENCLAW_CALLBACK_ALLOWLIST", ""
) or os.environ.get("MOLTBOT_CALLBACK_ALLOWLIST", "")
if callback_allowlist:
hosts = [h.strip() for h in callback_allowlist.split(",") if h.strip()]
if "*" in hosts or "*.com" in hosts or "*.net" in hosts:
report.add(
SecurityCheckResult(
name="callback_wildcard",
severity=SecuritySeverity.FAIL.value,
message="Callback allowlist contains overly broad wildcards",
category="ssrf",
detail=f"Allowlist: {callback_allowlist}",
remediation="Use specific hostnames instead of wildcards in callback allowlists.",
)
)
else:
report.add(
SecurityCheckResult(
name="callback_allowlist",
severity=SecuritySeverity.PASS.value,
message=f"Callback allowlist configured with {len(hosts)} host(s)",
category="ssrf",
)
)
# Check base_url configuration
try:
from .state_dir import get_state_dir
config_path = os.path.join(get_state_dir(), "config.json")
except Exception:
try:
from services.state_dir import get_state_dir
config_path = os.path.join(get_state_dir(), "config.json")
except Exception:
config_path = None
if config_path and os.path.exists(config_path):
try:
with open(config_path, "r", encoding="utf-8") as f:
cfg = json.load(f)
base_url = cfg.get("base_url", "")
if base_url:
# Check for private IP in base_url
try:
from urllib.parse import urlparse
parsed = urlparse(base_url)
host = parsed.hostname or ""
try:
ip = ipaddress.ip_address(host)
if ip.is_private and not ip.is_loopback:
report.add(
SecurityCheckResult(
name="base_url_private_ip",
severity=SecuritySeverity.WARN.value,
message=f"LLM base_url points to private IP ({host})",
category="ssrf",
remediation="Ensure this is an intentional local LLM setup.",
)
)
except ValueError:
pass # hostname, not IP — OK
except Exception:
pass
except Exception:
pass
report.add(
SecurityCheckResult(
name="ssrf_posture",
severity=SecuritySeverity.PASS.value,
message="SSRF posture check completed",
category="ssrf",
)
)
# ---------------------------------------------------------------------------
# Security checks — State directory permissions
# ---------------------------------------------------------------------------
def check_state_dir_permissions(report: SecurityReport) -> None:
"""Check state directory for unsafe permissions."""
try:
from .state_dir import get_state_dir
state_dir = get_state_dir()
except Exception:
try:
from services.state_dir import get_state_dir
state_dir = get_state_dir()
except Exception:
state_dir = os.environ.get("OPENCLAW_STATE_DIR") or os.environ.get(
"MOLTBOT_STATE_DIR"
)
if not state_dir:
report.add(
SecurityCheckResult(
name="state_dir_perms",
severity=SecuritySeverity.SKIP.value,
message="State directory not configured — using defaults",
category="state_dir",
)
)
return
p = Path(state_dir)
if not p.exists():
report.add(
SecurityCheckResult(
name="state_dir_exists",
severity=SecuritySeverity.INFO.value,
message=f"State dir does not exist yet: {state_dir}",
category="state_dir",
)
)
return
# Writable check
if not os.access(str(p), os.W_OK):
report.add(
SecurityCheckResult(
name="state_dir_writable",
severity=SecuritySeverity.FAIL.value,
message=f"State dir not writable: {state_dir}",
category="state_dir",
remediation="Check file permissions on the state directory.",
)
)
return
# Platform-specific permission checks
if platform.system() != "Windows":
try:
st = os.stat(str(p))
mode = st.st_mode
# Check for world-readable or world-writable
if mode & stat.S_IROTH:
report.add(
SecurityCheckResult(
name="state_dir_world_readable",
severity=SecuritySeverity.WARN.value,
message="State directory is world-readable",
category="state_dir",
detail=f"Permissions: {oct(mode)}",
remediation="Run: chmod 700 " + state_dir,
)
)
if mode & stat.S_IWOTH:
report.add(
SecurityCheckResult(
name="state_dir_world_writable",
severity=SecuritySeverity.FAIL.value,
message="State directory is world-writable — critical security risk",
category="state_dir",
detail=f"Permissions: {oct(mode)}",
remediation="Run: chmod 700 " + state_dir,
)
)
except Exception:
pass
# Check for secrets files with open permissions
secrets_file = p / "secrets.json"
if secrets_file.exists() and platform.system() != "Windows":
try:
st = os.stat(str(secrets_file))
if st.st_mode & (stat.S_IROTH | stat.S_IWOTH):
report.add(
SecurityCheckResult(
name="secrets_file_perms",
severity=SecuritySeverity.FAIL.value,
message="Secrets file has world-accessible permissions",
category="state_dir",
remediation="Run: chmod 600 " + str(secrets_file),
)
)
except Exception:
pass
report.add(
SecurityCheckResult(
name="state_dir_check",
severity=SecuritySeverity.PASS.value,
message=f"State directory permissions OK: {state_dir}",
category="state_dir",
)
)
# ---------------------------------------------------------------------------
# Security checks — Redaction drift
# ---------------------------------------------------------------------------
def check_redaction_drift(report: SecurityReport) -> None:
"""Verify that redaction patterns cover expected sensitive keys."""
try:
from .redaction import SENSITIVE_KEYS
except ImportError:
try:
from services.redaction import SENSITIVE_KEYS
except ImportError:
report.add(
SecurityCheckResult(
name="redaction_module",
severity=SecuritySeverity.SKIP.value,
message="Redaction module not available",
category="redaction",
)
)
return
# Expected minimum set of sensitive keys
expected_keys = {
"api_key",
"password",
"secret",
"token",
"authorization",
"private_key",
}
missing = expected_keys - SENSITIVE_KEYS
if missing:
report.add(
SecurityCheckResult(
name="redaction_coverage",
severity=SecuritySeverity.WARN.value,
message=f"Redaction missing expected sensitive keys: {missing}",
category="redaction",
remediation="Update services/redaction.py SENSITIVE_KEYS to include missing keys.",
)
)
else:
report.add(
SecurityCheckResult(
name="redaction_coverage",
severity=SecuritySeverity.PASS.value,
message=f"Redaction covers all {len(expected_keys)} expected sensitive keys",
category="redaction",
)
)
# ---------------------------------------------------------------------------
# Security checks — ComfyUI runtime mode
# ---------------------------------------------------------------------------
def check_comfyui_runtime(report: SecurityReport) -> None:
"""Check ComfyUI runtime mode compatibility."""
in_venv = sys.prefix != sys.base_prefix
report.environment["in_venv"] = str(in_venv)
report.environment["os"] = platform.system()
if not in_venv:
report.add(
SecurityCheckResult(
name="venv_isolation",
severity=SecuritySeverity.WARN.value,
message="Not running in a virtual environment — shared system packages risk",
category="runtime",
remediation="Use a project-local .venv for dependency isolation.",
)
)
# Check for ComfyUI Desktop indicators
desktop_indicators = [
os.environ.get("COMFYUI_DESKTOP"),
os.environ.get("ELECTRON_RUN_AS_NODE"),
]
if any(desktop_indicators):
report.environment["runtime_mode"] = "desktop"
report.add(
SecurityCheckResult(
name="desktop_mode",
severity=SecuritySeverity.INFO.value,
message="ComfyUI Desktop mode detected",
category="runtime",
detail="Desktop mode may restrict file access and network behavior.",
)
)
else:
report.environment["runtime_mode"] = "standard"
# Check Python version for security support
ver = sys.version_info
if ver.major == 3 and ver.minor < 10:
report.add(
SecurityCheckResult(
name="python_security",
severity=SecuritySeverity.WARN.value,
message=f"Python {ver.major}.{ver.minor} may lack security patches",
category="runtime",
remediation="Upgrade to Python 3.10+ for active security support.",
)
)
# ---------------------------------------------------------------------------
# Security checks — Feature flag posture
# ---------------------------------------------------------------------------
# High-risk feature flags that should be OFF by default
HIGH_RISK_FLAGS = {
"OPENCLAW_ENABLE_REMOTE_ADMIN": "Remote admin access",
"OPENCLAW_ENABLE_BRIDGE": "Sidecar bridge",
"OPENCLAW_ENABLE_TRANSFORMS": "Constrained transforms (F42)",
"OPENCLAW_ENABLE_REGISTRY_SYNC": "Remote registry sync (F41)",
"MOLTBOT_DEV_MODE": "Development mode (auth bypass)",
}
def check_feature_flags(report: SecurityReport) -> None:
"""Check that high-risk features are not accidentally enabled."""
enabled_flags = []
for env_key, label in HIGH_RISK_FLAGS.items():
val = os.environ.get(env_key, "").strip().lower()
if val in ("1", "true", "yes", "on"):
enabled_flags.append(f"{env_key} ({label})")
if enabled_flags:
report.add(
SecurityCheckResult(
name="high_risk_flags",
severity=SecuritySeverity.WARN.value,
message=f"{len(enabled_flags)} high-risk feature flag(s) enabled",
category="feature_flags",
detail="; ".join(enabled_flags),
remediation="Disable high-risk flags unless explicitly required for your deployment.",
)
)
else:
report.add(
SecurityCheckResult(
name="high_risk_flags",
severity=SecuritySeverity.PASS.value,
message="All high-risk features disabled (default-off)",
category="feature_flags",
)
)
# ---------------------------------------------------------------------------
# Security checks — API key posture
# ---------------------------------------------------------------------------
def check_api_key_posture(report: SecurityReport) -> None:
"""Check API key configuration for common issues."""
api_key = (
os.environ.get("OPENCLAW_LLM_API_KEY")
or os.environ.get("MOLTBOT_LLM_API_KEY")
or os.environ.get("CLAWDBOT_LLM_API_KEY")
or ""
)
if not api_key:
report.add(
SecurityCheckResult(
name="api_key_present",
severity=SecuritySeverity.INFO.value,
message="No LLM API key in environment — may use stored key or local LLM",
category="api_key",
)
)
else:
# Never log the key — just check properties
if len(api_key) < 10:
report.add(
SecurityCheckResult(
name="api_key_length",
severity=SecuritySeverity.WARN.value,
message="LLM API key appears unusually short",
category="api_key",
remediation="Verify the API key is complete and valid.",
)
)
else:
report.add(
SecurityCheckResult(
name="api_key_present",
severity=SecuritySeverity.PASS.value,
message="LLM API key configured via environment",
category="api_key",
)
)
# ---------------------------------------------------------------------------
# Guarded remediation — safe/local-only actions
# ---------------------------------------------------------------------------
SAFE_REMEDIATIONS = {
"tighten_state_dir": "Set state directory permissions to owner-only (chmod 700/600)",
"tighten_secrets_file": "Set secrets file permissions to owner-only (chmod 600)",
}
def apply_guarded_remediation(
report: SecurityReport,
action: str,
*,
dry_run: bool = True,
) -> bool:
"""
Apply a safe, predefined remediation.
Only allows predefined safe actions (permissions tightening).
No external command execution. No arbitrary file mutation.
Args:
report: The security report to append results to.
action: One of the SAFE_REMEDIATIONS keys.
dry_run: If True, only report what would be done.
Returns:
True if remediation was applied (or would be applied in dry_run).
"""
if action not in SAFE_REMEDIATIONS:
report.add(
SecurityCheckResult(
name=f"remediation:{action}",
severity=SecuritySeverity.FAIL.value,
message=f"Unknown remediation action: {action}",
category="remediation",
)
)
return False
if platform.system() == "Windows":
report.add(
SecurityCheckResult(
name=f"remediation:{action}",
severity=SecuritySeverity.SKIP.value,
message=f"Remediation '{action}' not supported on Windows (use ACLs manually)",
category="remediation",
)
)
return False
try:
from .state_dir import get_state_dir
state_dir = get_state_dir()
except Exception:
try:
from services.state_dir import get_state_dir
state_dir = get_state_dir()
except Exception:
state_dir = None
if not state_dir:
return False
if action == "tighten_state_dir":
target = state_dir
target_mode = 0o700
elif action == "tighten_secrets_file":
target = os.path.join(state_dir, "secrets.json")
target_mode = 0o600
else:
return False
if not os.path.exists(target):
return False
if dry_run:
report.add(
SecurityCheckResult(
name=f"remediation:{action}",
severity=SecuritySeverity.INFO.value,
message=f"[DRY RUN] Would set {target} to {oct(target_mode)}",
category="remediation",
)
)
return True
try:
os.chmod(target, target_mode)
report.remediation_applied.append(
f"{action}: set {target} to {oct(target_mode)}"
)
report.add(
SecurityCheckResult(
name=f"remediation:{action}",
severity=SecuritySeverity.PASS.value,
message=f"Applied: set {target} to {oct(target_mode)}",
category="remediation",
)
)
return True
except Exception as e:
report.add(
SecurityCheckResult(
name=f"remediation:{action}",
severity=SecuritySeverity.FAIL.value,
message=f"Remediation failed: {e}",
category="remediation",
)
)
return False
# ---------------------------------------------------------------------------
# Main runner
# ---------------------------------------------------------------------------
def run_security_doctor(
*,
remediate: bool = False,
dry_run: bool = True,
) -> SecurityReport:
"""
Run all security diagnostic checks and return a report.
Args:
remediate: If True, apply safe remediations after scanning.
dry_run: If True (default), only report what would be remediated.
"""
report = SecurityReport()
pack_root = _get_pack_root()
report.environment["pack_root"] = str(pack_root)
report.environment["scan_mode"] = (
"read-only" if not remediate else ("dry-run" if dry_run else "remediate")
)
# Run all checks
check_endpoint_exposure(report)
check_token_boundaries(report)
check_ssrf_posture(report)
check_state_dir_permissions(report)
check_redaction_drift(report)
check_comfyui_runtime(report)
check_feature_flags(report)
check_api_key_posture(report)
# Optional guarded remediation
if remediate:
# Identify failing checks that have safe remediations
for check in report.checks:
if check.severity == SecuritySeverity.FAIL.value:
if "state_dir" in check.name and "world" in check.message.lower():
if "secret" in check.name:
apply_guarded_remediation(
report, "tighten_secrets_file", dry_run=dry_run
)
else:
apply_guarded_remediation(
report, "tighten_state_dir", dry_run=dry_run
)
report.build_summary()
return report
# ---------------------------------------------------------------------------
# CLI entrypoint
# ---------------------------------------------------------------------------
def main() -> None:
"""CLI entrypoint for security doctor."""
import argparse
parser = argparse.ArgumentParser(
description="OpenClaw Security Doctor — security posture diagnostics"
)
parser.add_argument(
"--json",
action="store_true",
help="Output machine-readable JSON instead of human-readable text",
)
parser.add_argument(
"--remediate",
action="store_true",
help="Apply safe remediations (permissions tightening only)",
)
parser.add_argument(
"--dry-run",
action="store_true",
default=True,
help="Only report what would be remediated (default: True)",
)
parser.add_argument(
"--apply",
action="store_true",
help="Actually apply remediations (requires --remediate)",
)
args = parser.parse_args()
dry_run = not args.apply
report = run_security_doctor(remediate=args.remediate, dry_run=dry_run)
if args.json:
print(json.dumps(report.to_dict(), indent=2))
else:
print(report.to_human())
sys.exit(1 if report.has_failures else 0)
if __name__ == "__main__":
main()
+51 -22
View File
@@ -14,6 +14,7 @@ logger = logging.getLogger("ComfyUI-OpenClaw.services.settings_schema")
class SettingType(Enum):
"""Supported setting value types."""
STRING = "string"
INT = "int"
FLOAT = "float"
@@ -25,6 +26,7 @@ class SettingType(Enum):
@dataclass
class SettingDef:
"""Definition for a single registered setting key."""
key: str
type: SettingType
default: Any
@@ -85,6 +87,7 @@ def get_schema_map() -> Dict[str, dict]:
# ──────────────────────────── Coercion ────────────────────────────
def coerce_value(key: str, raw: Any) -> Tuple[Any, Optional[str]]:
"""
Coerce *raw* to the registered type for *key*.
@@ -136,9 +139,7 @@ def _coerce(defn: SettingDef, raw: Any) -> Any:
if defn.type == SettingType.ENUM:
s = str(raw).strip()
if defn.enum_values and s not in defn.enum_values:
raise ValueError(
f"'{s}' not in allowed values: {defn.enum_values}"
)
raise ValueError(f"'{s}' not in allowed values: {defn.enum_values}")
return s
if defn.type == SettingType.LIST_STRING:
@@ -176,43 +177,71 @@ def coerce_dict(updates: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:
# ──────────────────────────── Bootstrap defaults ────────────────────────────
def _register_defaults() -> None:
"""Register built-in OpenClaw setting definitions."""
_defs = [
SettingDef(
key="provider", type=SettingType.STRING, default="openai",
description="LLM provider ID", category="llm",
key="provider",
type=SettingType.STRING,
default="openai",
description="LLM provider ID",
category="llm",
),
SettingDef(
key="model", type=SettingType.STRING, default="gpt-4o-mini",
description="LLM model ID", category="llm",
key="model",
type=SettingType.STRING,
default="gpt-4o-mini",
description="LLM model ID",
category="llm",
),
SettingDef(
key="base_url", type=SettingType.STRING, default="",
description="Custom base URL (empty = provider default)", category="llm",
key="base_url",
type=SettingType.STRING,
default="",
description="Custom base URL (empty = provider default)",
category="llm",
),
SettingDef(
key="timeout_sec", type=SettingType.INT, default=120,
min_val=5, max_val=300,
description="LLM request timeout in seconds", category="llm",
key="timeout_sec",
type=SettingType.INT,
default=120,
min_val=5,
max_val=300,
description="LLM request timeout in seconds",
category="llm",
),
SettingDef(
key="max_retries", type=SettingType.INT, default=3,
min_val=0, max_val=10,
description="Max LLM retry attempts", category="llm",
key="max_retries",
type=SettingType.INT,
default=3,
min_val=0,
max_val=10,
description="Max LLM retry attempts",
category="llm",
),
SettingDef(
key="fallback_models", type=SettingType.LIST_STRING, default=[],
description="Failover model list (comma-separated)", category="llm",
key="fallback_models",
type=SettingType.LIST_STRING,
default=[],
description="Failover model list (comma-separated)",
category="llm",
),
SettingDef(
key="fallback_providers", type=SettingType.LIST_STRING, default=[],
description="Failover provider list (comma-separated)", category="llm",
key="fallback_providers",
type=SettingType.LIST_STRING,
default=[],
description="Failover provider list (comma-separated)",
category="llm",
),
SettingDef(
key="max_failover_candidates", type=SettingType.INT, default=3,
min_val=1, max_val=5,
description="Max failover candidates", category="llm",
key="max_failover_candidates",
type=SettingType.INT,
default=3,
min_val=1,
max_val=5,
description="Max failover candidates",
category="llm",
),
]
for d in _defs:
+408
View File
@@ -0,0 +1,408 @@
"""
F40 Webhook Mapping Engine v1.
Schema-first, deterministic payload mapping for external webhook sources.
Maps diverse incoming payloads into the canonical WebhookJobRequest format
without executing arbitrary code.
Design:
- Mapping profiles are declarative JSON/dict configurations.
- Each profile defines sourcetarget field paths + optional coercion + defaults.
- Profiles are matched by source identifier or explicit header.
- Unknown fields are dropped (safe-by-default), not passed through.
"""
from __future__ import annotations
import copy
import logging
import os
import re
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Dict, List, Optional, Tuple
logger = logging.getLogger("ComfyUI-OpenClaw.services.webhook_mapping")
# ---------------------------------------------------------------------------
# Coercion types
# ---------------------------------------------------------------------------
class CoercionType(Enum):
"""Supported field coercion types for mapping."""
STRING = "string"
INT = "int"
FLOAT = "float"
BOOL = "bool"
JSON = "json"
PASSTHROUGH = "passthrough"
# ---------------------------------------------------------------------------
# Field mapping rule
# ---------------------------------------------------------------------------
@dataclass
class FieldMapping:
"""A single source→target field mapping rule."""
source_path: str # dot-notation path in source payload, e.g. "data.repo.name"
target_path: str # dot-notation path in target payload, e.g. "inputs.repo_name"
coercion: CoercionType = CoercionType.PASSTHROUGH
default: Any = None # used when source_path is missing
required: bool = False # if True, mapping fails when source is absent
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "FieldMapping":
coercion = CoercionType(data.get("coercion", "passthrough"))
return cls(
source_path=data["source_path"],
target_path=data["target_path"],
coercion=coercion,
default=data.get("default"),
required=data.get("required", False),
)
# ---------------------------------------------------------------------------
# Mapping profile
# ---------------------------------------------------------------------------
MAX_FIELD_MAPPINGS = 50 # Prevent DoS via oversized profiles
@dataclass
class MappingProfile:
"""
A declarative mapping profile that transforms an external webhook
payload into the canonical WebhookJobRequest shape.
"""
id: str
label: str
description: str = ""
# Fixed values injected into the target regardless of source
defaults: Dict[str, Any] = field(default_factory=dict)
# Ordered field mapping rules
field_mappings: List[FieldMapping] = field(default_factory=list)
# Source identifier match pattern (matched against X-Webhook-Source header)
source_pattern: Optional[str] = None
@classmethod
def from_dict(cls, data: Dict[str, Any]) -> "MappingProfile":
raw_mappings = data.get("field_mappings", [])
if len(raw_mappings) > MAX_FIELD_MAPPINGS:
raise ValueError(
f"Too many field_mappings ({len(raw_mappings)}), max {MAX_FIELD_MAPPINGS}"
)
mappings = [FieldMapping.from_dict(m) for m in raw_mappings]
return cls(
id=data["id"],
label=data.get("label", data["id"]),
description=data.get("description", ""),
defaults=data.get("defaults", {}),
field_mappings=mappings,
source_pattern=data.get("source_pattern"),
)
# ---------------------------------------------------------------------------
# Path utilities (safe, deterministic, no eval)
# ---------------------------------------------------------------------------
def _resolve_path(obj: Any, path: str) -> Tuple[bool, Any]:
"""
Resolve a dot-notation path against a nested dict.
Returns (found, value).
Array indexing via [N] is supported for simple cases.
"""
parts = re.split(r"\.|(?=\[)", path)
current = obj
for part in parts:
if not part:
continue
# Array index: [0], [1], etc.
idx_match = re.match(r"^\[(\d+)\]$", part)
if idx_match:
idx = int(idx_match.group(1))
if isinstance(current, list) and 0 <= idx < len(current):
current = current[idx]
else:
return False, None
elif isinstance(current, dict):
if part in current:
current = current[part]
else:
return False, None
else:
return False, None
return True, current
def _set_path(obj: Dict[str, Any], path: str, value: Any) -> None:
"""Set a value at a dot-notation path, creating intermediate dicts as needed."""
parts = path.split(".")
current = obj
for part in parts[:-1]:
if part not in current or not isinstance(current[part], dict):
current[part] = {}
current = current[part]
current[parts[-1]] = value
# ---------------------------------------------------------------------------
# Coercion engine
# ---------------------------------------------------------------------------
def _coerce_value(value: Any, coercion: CoercionType) -> Any:
"""Coerce value to the target type. Raises ValueError on failure."""
if value is None:
return None
if coercion == CoercionType.PASSTHROUGH:
return value
if coercion == CoercionType.STRING:
return str(value)
if coercion == CoercionType.INT:
return int(value)
if coercion == CoercionType.FLOAT:
return float(value)
if coercion == CoercionType.BOOL:
if isinstance(value, bool):
return value
if isinstance(value, str):
return value.lower() in ("true", "1", "yes", "on")
return bool(value)
if coercion == CoercionType.JSON:
import json
if isinstance(value, str):
return json.loads(value)
return value # already parsed
return value
# ---------------------------------------------------------------------------
# Apply mapping profile
# ---------------------------------------------------------------------------
def apply_mapping(
profile: MappingProfile,
source_payload: Dict[str, Any],
) -> Tuple[Dict[str, Any], List[str]]:
"""
Apply a mapping profile to a source payload.
Returns:
(mapped_payload, warnings)
mapped_payload is a dict shaped like WebhookJobRequest fields.
warnings is a list of non-fatal issues encountered during mapping.
"""
result: Dict[str, Any] = copy.deepcopy(profile.defaults)
warnings: List[str] = []
for fm in profile.field_mappings:
found, value = _resolve_path(source_payload, fm.source_path)
if not found:
if fm.required:
raise ValueError(
f"Required source field '{fm.source_path}' not found in payload"
)
if fm.default is not None:
_set_path(result, fm.target_path, copy.deepcopy(fm.default))
else:
warnings.append(
f"Optional source field '{fm.source_path}' not found, skipped"
)
continue
try:
coerced = _coerce_value(value, fm.coercion)
except (ValueError, TypeError) as e:
raise ValueError(
f"Coercion failed for '{fm.source_path}'"
f"'{fm.target_path}' ({fm.coercion.value}): {e}"
)
_set_path(result, fm.target_path, coerced)
return result, warnings
# ---------------------------------------------------------------------------
# Built-in mapping profiles for common webhook sources
# ---------------------------------------------------------------------------
BUILTIN_PROFILES: Dict[str, MappingProfile] = {}
def _register_builtin_profiles() -> None:
"""Register built-in mapping profiles for common webhook shapes."""
# 1. GitHub webhook (push event)
BUILTIN_PROFILES["github_push"] = MappingProfile(
id="github_push",
label="GitHub Push Event",
description="Maps GitHub push webhook to a template trigger",
source_pattern="github",
defaults={
"version": 1,
"profile_id": "default",
},
field_mappings=[
FieldMapping(
source_path="repository.full_name",
target_path="inputs.repo_name",
coercion=CoercionType.STRING,
),
FieldMapping(
source_path="ref",
target_path="inputs.ref",
coercion=CoercionType.STRING,
),
FieldMapping(
source_path="head_commit.message",
target_path="inputs.commit_message",
coercion=CoercionType.STRING,
required=False,
),
FieldMapping(
source_path="sender.login",
target_path="inputs.actor",
coercion=CoercionType.STRING,
default="unknown",
),
],
)
# 2. Discord webhook (simple message)
BUILTIN_PROFILES["discord_message"] = MappingProfile(
id="discord_message",
label="Discord Message",
description="Maps Discord webhook message to template inputs",
source_pattern="discord",
defaults={
"version": 1,
"profile_id": "default",
},
field_mappings=[
FieldMapping(
source_path="content",
target_path="inputs.requirements",
coercion=CoercionType.STRING,
required=True,
),
FieldMapping(
source_path="author.username",
target_path="inputs.actor",
coercion=CoercionType.STRING,
default="discord_user",
),
],
)
# 3. Generic / passthrough (minimal mapping)
BUILTIN_PROFILES["generic"] = MappingProfile(
id="generic",
label="Generic Webhook",
description="Minimal passthrough mapping — expects near-canonical payload shape",
source_pattern=None,
defaults={"version": 1},
field_mappings=[
FieldMapping(
source_path="template_id",
target_path="template_id",
coercion=CoercionType.STRING,
required=True,
),
FieldMapping(
source_path="profile_id",
target_path="profile_id",
coercion=CoercionType.STRING,
default="default",
),
FieldMapping(
source_path="inputs",
target_path="inputs",
coercion=CoercionType.PASSTHROUGH,
default={},
),
FieldMapping(
source_path="job_id",
target_path="job_id",
coercion=CoercionType.STRING,
),
FieldMapping(
source_path="trace_id",
target_path="trace_id",
coercion=CoercionType.STRING,
),
FieldMapping(
source_path="callback",
target_path="callback",
coercion=CoercionType.PASSTHROUGH,
),
],
)
_register_builtin_profiles()
# ---------------------------------------------------------------------------
# Profile resolution
# ---------------------------------------------------------------------------
def resolve_profile(
headers: Optional[Dict[str, str]] = None,
source_hint: Optional[str] = None,
) -> Optional[MappingProfile]:
"""
Resolve a mapping profile from request metadata.
Priority:
1. Explicit header: X-Webhook-Mapping-Profile
2. Source hint header: X-Webhook-Source
3. source_hint argument
4. None (caller should fall back to canonical parsing)
"""
# 1. Explicit profile selection
if headers:
explicit = headers.get("X-Webhook-Mapping-Profile", "").strip()
if explicit and explicit in BUILTIN_PROFILES:
return BUILTIN_PROFILES[explicit]
# 2. Source-based matching
source = None
if headers:
source = headers.get("X-Webhook-Source", "").strip().lower()
if not source and source_hint:
source = source_hint.lower()
if source:
for profile in BUILTIN_PROFILES.values():
if profile.source_pattern and profile.source_pattern in source:
return profile
return None
def get_available_profiles() -> List[Dict[str, str]]:
"""Return list of available mapping profiles (for diagnostics/docs)."""
return [
{
"id": p.id,
"label": p.label,
"description": p.description,
"source_pattern": p.source_pattern or "",
}
for p in BUILTIN_PROFILES.values()
]
+120
View File
@@ -0,0 +1,120 @@
"""
Tests for F40 Webhook Mapping Engine.
"""
import unittest
from services.webhook_mapping import (
BUILTIN_PROFILES,
CoercionType,
FieldMapping,
MappingProfile,
_resolve_path,
_set_path,
apply_mapping,
resolve_profile,
)
class TestWebhookMapping(unittest.TestCase):
def test_path_utilities(self):
# Resolve
data = {"a": {"b": [10, 20]}, "c": 30}
self.assertEqual(_resolve_path(data, "a.b[1]"), (True, 20))
self.assertEqual(_resolve_path(data, "a.b[99]"), (False, None))
self.assertEqual(_resolve_path(data, "x.y"), (False, None))
self.assertEqual(_resolve_path(data, "a.b"), (True, [10, 20]))
# Set
target = {}
_set_path(target, "x.y.z", 100)
self.assertEqual(target, {"x": {"y": {"z": 100}}})
def test_apply_mapping_success(self):
profile = MappingProfile(
id="test",
label="Test",
defaults={"version": 1, "profile_id": "p1"},
field_mappings=[
FieldMapping("user.name", "inputs.user"),
FieldMapping("user.age", "inputs.age", coercion=CoercionType.INT),
FieldMapping("active", "inputs.is_active", coercion=CoercionType.BOOL),
],
)
source = {
"user": {"name": "Alice", "age": "25"},
"active": "yes",
}
mapped, warnings = apply_mapping(profile, source)
self.assertEqual(mapped["version"], 1)
self.assertEqual(mapped["profile_id"], "p1")
self.assertEqual(mapped["inputs"]["user"], "Alice")
self.assertEqual(mapped["inputs"]["age"], 25)
self.assertTrue(mapped["inputs"]["is_active"])
self.assertEqual(warnings, [])
def test_apply_mapping_missing_required(self):
profile = MappingProfile(
id="test",
label="Test",
field_mappings=[
FieldMapping("required_field", "target", required=True),
],
)
with self.assertRaises(ValueError) as cm:
apply_mapping(profile, {})
self.assertIn("required_field", str(cm.exception))
def test_apply_mapping_coercion_failure(self):
profile = MappingProfile(
id="test",
label="Test",
field_mappings=[
FieldMapping("val", "target", coercion=CoercionType.INT),
],
)
with self.assertRaises(ValueError) as cm:
apply_mapping(profile, {"val": "not-an-int"})
self.assertIn("Coercion failed", str(cm.exception))
def test_resolve_profile(self):
# Header match
p = resolve_profile({"X-Webhook-Mapping-Profile": "github_push"})
self.assertIsNotNone(p)
self.assertEqual(p.id, "github_push")
# Source hint match (header)
p = resolve_profile({"X-Webhook-Source": "Discord"})
self.assertIsNotNone(p)
self.assertEqual(p.id, "discord_message")
# No match
p = resolve_profile({})
self.assertIsNone(p)
def test_github_push_builtin(self):
profile = BUILTIN_PROFILES["github_push"]
payload = {
"repository": {"full_name": "user/repo"},
"ref": "refs/heads/main",
"sender": {"login": "dev"},
}
mapped, _ = apply_mapping(profile, payload)
self.assertEqual(mapped["inputs"]["repo_name"], "user/repo")
self.assertEqual(mapped["inputs"]["ref"], "refs/heads/main")
self.assertEqual(mapped["inputs"]["actor"], "dev")
def test_discord_message_builtin(self):
profile = BUILTIN_PROFILES["discord_message"]
payload = {
"content": "!generate cat",
"author": {"username": "user1"},
}
mapped, _ = apply_mapping(profile, payload)
self.assertEqual(mapped["inputs"]["requirements"], "!generate cat")
self.assertEqual(mapped["inputs"]["actor"], "user1")
if __name__ == "__main__":
unittest.main()
+15 -11
View File
@@ -106,11 +106,13 @@ class TestSettingsSchema(unittest.TestCase):
"""coerce_dict should handle valid + invalid keys together."""
from services.settings_schema import coerce_dict
coerced, errors = coerce_dict({
"provider": "openai",
"timeout_sec": "30",
"unknown_key": "foo",
})
coerced, errors = coerce_dict(
{
"provider": "openai",
"timeout_sec": "30",
"unknown_key": "foo",
}
)
self.assertEqual(len(errors), 1)
self.assertIn("unknown_key", errors[0])
self.assertEqual(coerced["provider"], "openai")
@@ -126,12 +128,14 @@ class TestSettingsSchema(unittest.TestCase):
register_setting,
)
register_setting(SettingDef(
key="custom_flag",
type=SettingType.BOOL,
default=False,
description="Test custom flag",
))
register_setting(
SettingDef(
key="custom_flag",
type=SettingType.BOOL,
default=False,
description="Test custom flag",
)
)
self.assertTrue(is_registered("custom_flag"))
val, err = coerce_value("custom_flag", "true")
+87
View File
@@ -0,0 +1,87 @@
"""
Tests for R71 Job Event Stream (SSE).
"""
import unittest
from services.job_events import (
JobEvent,
JobEventStore,
JobEventType,
get_job_event_store,
)
class TestJobEventStore(unittest.TestCase):
def setUp(self):
self.store = JobEventStore(max_size=5)
def test_emit_and_retrieve(self):
store = self.store
# Emit a few events
store.emit(JobEventType.QUEUED, "p1", "t1")
store.emit(JobEventType.RUNNING, "p1", "t1")
last_evt = store.emit(JobEventType.COMPLETED, "p1", "t1")
# Verify sequence
self.assertEqual(last_evt.seq, 3)
self.assertEqual(store.latest_seq(), 3)
self.assertEqual(store.size, 3)
# Retrieve all
events = store.events_since(0)
self.assertEqual(len(events), 3)
self.assertEqual(events[0].event_type, "queued")
self.assertEqual(events[2].event_type, "completed")
def test_buffer_rotation(self):
store = self.store
# Emit 6 events (capacity 5)
for i in range(1, 7):
store.emit(JobEventType.QUEUED, f"p{i}")
# Should drop the first one (seq 1)
self.assertEqual(store.size, 5)
self.assertEqual(store.latest_seq(), 6)
events = store.events_since(0)
self.assertEqual(len(events), 5)
self.assertEqual(events[0].seq, 2) # First event is now seq=2
self.assertEqual(events[-1].seq, 6)
def test_events_since_filter(self):
store = self.store
store.emit(JobEventType.QUEUED, "p1")
store.emit(JobEventType.QUEUED, "p2")
store.emit(JobEventType.QUEUED, "p1")
# Filter by prompt_id
p1_events = store.events_since(0, prompt_id="p1")
self.assertEqual(len(p1_events), 2)
self.assertEqual(p1_events[0].prompt_id, "p1")
self.assertEqual(p1_events[1].prompt_id, "p1")
def test_sse_format(self):
evt = JobEvent(
seq=123,
event_type="test",
prompt_id="p1",
trace_id="t1",
timestamp=1000.0,
data={"foo": "bar"},
)
sse = evt.to_sse()
expected_lines = [
"id: 123",
"event: test",
'data: {"event_type":"test","prompt_id":"p1","trace_id":"t1","timestamp":1000.0,"data":{"foo":"bar"}}',
"",
"",
]
self.assertEqual(sse, "\n".join(expected_lines))
if __name__ == "__main__":
unittest.main()
+61
View File
@@ -0,0 +1,61 @@
"""
Tests for R72 Operator Doctor.
"""
import unittest
from pathlib import Path
from services.operator_doctor import (
CheckResult,
DoctorReport,
check_python_version,
check_state_dir,
check_venv,
)
class TestOperatorDoctor(unittest.TestCase):
def test_report_structure(self):
report = DoctorReport()
report.add(CheckResult("c1", "pass", "ok"))
report.add(CheckResult("c2", "fail", "bad"))
d = report.to_dict()
self.assertEqual(d["summary"]["pass"], 1)
self.assertEqual(d["summary"]["fail"], 1)
self.assertTrue(report.has_failures)
def test_check_python_version(self):
report = DoctorReport()
check_python_version(report)
# Should detect current version (which is > 3.10)
self.assertEqual(report.checks[-1].severity, "pass")
def test_check_venv(self):
report = DoctorReport()
check_venv(report)
# Result depends on environment, but should always produce a check
self.assertIn(report.checks[-1].name, ["venv_active"])
def test_check_state_dir_missing(self):
report = DoctorReport()
# Mock env vars?? No, just call directly.
# But check_state_dir reads os.environ.
import os
orig = os.environ.get("MOLTBOT_STATE_DIR")
try:
os.environ["MOLTBOT_STATE_DIR"] = "/tmp/does-not-exist-123"
check_state_dir(report)
last = report.checks[-1]
self.assertEqual(last.name, "state_dir")
self.assertEqual(last.severity, "warn")
finally:
if orig:
os.environ["MOLTBOT_STATE_DIR"] = orig
else:
del os.environ["MOLTBOT_STATE_DIR"]
if __name__ == "__main__":
unittest.main()
+23 -13
View File
@@ -94,7 +94,10 @@ class TestProviderGovernanceInfo(unittest.TestCase):
def test_governance_info_complete(self):
"""get_provider_governance_info should return entries for all catalog providers."""
from services.providers.catalog import PROVIDER_CATALOG, get_provider_governance_info
from services.providers.catalog import (
PROVIDER_CATALOG,
get_provider_governance_info,
)
info = get_provider_governance_info()
for pid in PROVIDER_CATALOG:
@@ -166,6 +169,7 @@ class TestR73InValidateConfig(unittest.TestCase):
self.assertEqual(sanitized["provider"], "gemini")
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
def test_local_alias_with_localhost_base_url_accepted(self):
"""provider=local (alias for lmstudio) + localhost URL should be accepted."""
import shutil
@@ -181,10 +185,12 @@ class TestR73InValidateConfig(unittest.TestCase):
):
from services.runtime_config import validate_config_update
sanitized, errors = validate_config_update({
"provider": "local",
"base_url": "http://127.0.0.1:1234",
})
sanitized, errors = validate_config_update(
{
"provider": "local",
"base_url": "http://127.0.0.1:1234",
}
)
self.assertEqual(len(errors), 0, f"Unexpected errors: {errors}")
self.assertEqual(sanitized["provider"], "lmstudio")
self.assertEqual(sanitized["base_url"], "http://127.0.0.1:1234")
@@ -206,10 +212,12 @@ class TestR73InValidateConfig(unittest.TestCase):
):
from services.runtime_config import validate_config_update
sanitized, errors = validate_config_update({
"provider": "local",
"base_url": "http://localhost:1234",
})
sanitized, errors = validate_config_update(
{
"provider": "local",
"base_url": "http://localhost:1234",
}
)
self.assertEqual(len(errors), 0, f"Unexpected errors: {errors}")
self.assertEqual(sanitized["provider"], "lmstudio")
@@ -231,10 +239,12 @@ class TestR73InValidateConfig(unittest.TestCase):
):
from services.runtime_config import validate_config_update
sanitized, errors = validate_config_update({
"provider": "chatgpt",
"base_url": "", # empty = use default
})
sanitized, errors = validate_config_update(
{
"provider": "chatgpt",
"base_url": "", # empty = use default
}
)
self.assertEqual(len(errors), 0, f"Unexpected errors: {errors}")
self.assertEqual(sanitized["provider"], "openai")
finally:
@@ -0,0 +1,876 @@
"""
Tests for S30, F41, F42 Security Doctor, Registry Quarantine, and Constrained Transforms.
Covers:
- S30: diagnostics finding coverage and severity classifications
- F41: registry signature/hash/provenance and quarantine flow tests
- F42: transform runtime constraint and denial-path tests
- Regression: default-off behavior for F41/F42
- Regression: mapping-only (F40) remains functional when transforms disabled
"""
import hashlib
import json
import os
import shutil
import tempfile
import unittest
# ---------------------------------------------------------------------------
# S30 — Security Doctor Tests
# ---------------------------------------------------------------------------
class TestSecurityDoctor(unittest.TestCase):
"""S30: Security diagnostic checks."""
def test_run_security_doctor_returns_report(self):
from services.security_doctor import run_security_doctor
report = run_security_doctor()
self.assertIsNotNone(report)
self.assertIsInstance(report.checks, list)
self.assertTrue(len(report.checks) > 0, "Should produce at least one check")
def test_report_to_dict(self):
from services.security_doctor import run_security_doctor
report = run_security_doctor()
d = report.to_dict()
self.assertIn("checks", d)
self.assertIn("summary", d)
self.assertIn("risk_score", d)
self.assertIn("environment", d)
def test_report_to_human(self):
from services.security_doctor import run_security_doctor
report = run_security_doctor()
human = report.to_human()
self.assertIn("Security Doctor", human)
self.assertIn("Risk Score", human)
def test_check_categories_present(self):
from services.security_doctor import run_security_doctor
report = run_security_doctor()
categories = {c.category for c in report.checks}
# At least these categories should appear
self.assertTrue(
categories & {"endpoint", "ssrf", "redaction", "runtime", "feature_flags"}
)
def test_no_secrets_in_output(self):
"""Verify that security doctor output never contains actual secrets."""
from services.security_doctor import run_security_doctor
# Set some fake env vars
old_env = {}
for key in ("OPENCLAW_ADMIN_TOKEN", "OPENCLAW_LLM_API_KEY"):
old_env[key] = os.environ.get(key)
try:
os.environ["OPENCLAW_ADMIN_TOKEN"] = "test-secret-token-12345678"
os.environ["OPENCLAW_LLM_API_KEY"] = "sk-test-key-abcdefghij"
report = run_security_doctor()
output = json.dumps(report.to_dict())
human = report.to_human()
# Must not leak the actual token values
self.assertNotIn("test-secret-token-12345678", output)
self.assertNotIn("sk-test-key-abcdefghij", output)
self.assertNotIn("test-secret-token-12345678", human)
self.assertNotIn("sk-test-key-abcdefghij", human)
finally:
for key, val in old_env.items():
if val is None:
os.environ.pop(key, None)
else:
os.environ[key] = val
def test_token_reuse_detection(self):
"""S30: detect identical admin and observability tokens."""
from services.security_doctor import SecurityReport, check_token_boundaries
old_env = {}
for key in ("OPENCLAW_ADMIN_TOKEN", "OPENCLAW_OBSERVABILITY_TOKEN"):
old_env[key] = os.environ.get(key)
try:
os.environ["OPENCLAW_ADMIN_TOKEN"] = "same-token-value-1234"
os.environ["OPENCLAW_OBSERVABILITY_TOKEN"] = "same-token-value-1234"
report = SecurityReport()
check_token_boundaries(report)
names = [c.name for c in report.checks]
self.assertIn("token_reuse", names)
reuse = next(c for c in report.checks if c.name == "token_reuse")
self.assertEqual(reuse.severity, "fail")
finally:
for key, val in old_env.items():
if val is None:
os.environ.pop(key, None)
else:
os.environ[key] = val
def test_weak_token_warning(self):
"""S30: warn about short tokens."""
from services.security_doctor import SecurityReport, check_token_boundaries
old = os.environ.get("OPENCLAW_ADMIN_TOKEN")
try:
os.environ["OPENCLAW_ADMIN_TOKEN"] = "short"
report = SecurityReport()
check_token_boundaries(report)
weak = [c for c in report.checks if "weak" in c.name]
self.assertTrue(len(weak) > 0)
finally:
if old is None:
os.environ.pop("OPENCLAW_ADMIN_TOKEN", None)
else:
os.environ["OPENCLAW_ADMIN_TOKEN"] = old
def test_feature_flags_default_off(self):
"""S30: all high-risk flags should be off by default."""
from services.security_doctor import SecurityReport, check_feature_flags
# Clear high-risk flags
old_env = {}
flags = [
"OPENCLAW_ENABLE_REMOTE_ADMIN",
"OPENCLAW_ENABLE_BRIDGE",
"OPENCLAW_ENABLE_TRANSFORMS",
"OPENCLAW_ENABLE_REGISTRY_SYNC",
"MOLTBOT_DEV_MODE",
]
for f in flags:
old_env[f] = os.environ.get(f)
os.environ.pop(f, None)
try:
report = SecurityReport()
check_feature_flags(report)
flags_check = next(c for c in report.checks if c.name == "high_risk_flags")
self.assertEqual(flags_check.severity, "pass")
finally:
for key, val in old_env.items():
if val is None:
os.environ.pop(key, None)
else:
os.environ[key] = val
def test_redaction_drift_check(self):
"""S30: verify redaction coverage passes."""
from services.security_doctor import SecurityReport, check_redaction_drift
report = SecurityReport()
check_redaction_drift(report)
coverage = next(
(c for c in report.checks if c.name == "redaction_coverage"), None
)
self.assertIsNotNone(coverage)
self.assertEqual(coverage.severity, "pass")
def test_guarded_remediation_unknown_action(self):
"""S30: unknown remediation actions are rejected."""
from services.security_doctor import SecurityReport, apply_guarded_remediation
report = SecurityReport()
result = apply_guarded_remediation(report, "unknown_action")
self.assertFalse(result)
def test_risk_score_calculation(self):
"""S30: risk score is computed correctly."""
from services.security_doctor import SecurityCheckResult, SecurityReport
report = SecurityReport()
report.add(SecurityCheckResult(name="a", severity="fail", message="x"))
report.add(SecurityCheckResult(name="b", severity="warn", message="x"))
report.add(SecurityCheckResult(name="c", severity="pass", message="x"))
self.assertEqual(report.risk_score, 13) # 10 + 3 + 0
# ---------------------------------------------------------------------------
# F41 — Registry Quarantine Tests
# ---------------------------------------------------------------------------
class TestRegistryQuarantine(unittest.TestCase):
"""F41: Registry quarantine flow tests."""
def setUp(self):
self.test_dir = tempfile.mkdtemp()
self.old_flag = os.environ.get("OPENCLAW_ENABLE_REGISTRY_SYNC")
os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"] = "1"
def tearDown(self):
shutil.rmtree(self.test_dir)
if self.old_flag is None:
os.environ.pop("OPENCLAW_ENABLE_REGISTRY_SYNC", None)
else:
os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"] = self.old_flag
def test_default_off(self):
"""F41: registry sync is disabled by default."""
from services.registry_quarantine import is_registry_sync_enabled
old = os.environ.pop("OPENCLAW_ENABLE_REGISTRY_SYNC", None)
try:
self.assertFalse(is_registry_sync_enabled())
finally:
if old is not None:
os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"] = old
def test_disabled_operations_fail_closed(self):
"""F41: operations fail-closed when feature is disabled."""
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
old = os.environ.pop("OPENCLAW_ENABLE_REGISTRY_SYNC", None)
try:
store = RegistryQuarantineStore(self.test_dir)
with self.assertRaises(RegistryQuarantineError):
store.register_fetch("test", "1.0.0", "https://example.com", "abc123")
finally:
if old is not None:
os.environ["OPENCLAW_ENABLE_REGISTRY_SYNC"] = old
def test_full_quarantine_lifecycle(self):
"""F41: fetch → verify → activate lifecycle."""
from services.registry_quarantine import (
QuarantineState,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
# Fetch
entry = store.register_fetch(
"my-pack", "1.0.0", "https://example.com/pack.zip", "abc123"
)
self.assertEqual(entry.state, QuarantineState.FETCHED.value)
self.assertEqual(len(entry.audit_trail), 1)
# Verify (success)
ok = store.verify_integrity("my-pack", "1.0.0", "abc123")
self.assertTrue(ok)
entry = store.get_entry("my-pack", "1.0.0")
self.assertEqual(entry.state, QuarantineState.VERIFIED.value)
# Activate
entry = store.activate("my-pack", "1.0.0")
self.assertEqual(entry.state, QuarantineState.ACTIVATED.value)
def test_verify_failure_quarantines(self):
"""F41: hash mismatch moves pack to quarantine."""
from services.registry_quarantine import (
QuarantineState,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch(
"bad-pack", "1.0.0", "https://example.com", "expected_hash"
)
ok = store.verify_integrity("bad-pack", "1.0.0", "wrong_hash")
self.assertFalse(ok)
entry = store.get_entry("bad-pack", "1.0.0")
self.assertEqual(entry.state, QuarantineState.QUARANTINED.value)
def test_cannot_activate_unverified(self):
"""F41: cannot activate a pack that hasn't been verified."""
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("test-pack", "1.0.0", "https://example.com", "hash")
with self.assertRaises(RegistryQuarantineError):
store.activate("test-pack", "1.0.0")
def test_reject_flow(self):
"""F41: reject a quarantined pack."""
from services.registry_quarantine import (
QuarantineState,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("bad-pack", "1.0.0", "https://example.com", "hash")
entry = store.reject("bad-pack", "1.0.0", "Suspicious provenance")
self.assertEqual(entry.state, QuarantineState.REJECTED.value)
self.assertEqual(entry.rejection_reason, "Suspicious provenance")
def test_rollback_flow(self):
"""F41: rollback a previously activated pack."""
from services.registry_quarantine import (
QuarantineState,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("pack", "1.0.0", "https://example.com", "h")
store.verify_integrity("pack", "1.0.0", "h")
store.activate("pack", "1.0.0")
entry = store.rollback("pack", "1.0.0", "Security concern")
self.assertEqual(entry.state, QuarantineState.ROLLED_BACK.value)
def test_cannot_rollback_non_activated(self):
"""F41: cannot rollback a pack that isn't activated."""
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("pack", "1.0.0", "https://example.com", "h")
with self.assertRaises(RegistryQuarantineError):
store.rollback("pack", "1.0.0")
def test_entry_limit(self):
"""F41: enforce max entries limit."""
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.MAX_ENTRIES = 3 # Override for test
for i in range(3):
store.register_fetch(f"pack-{i}", "1.0.0", "https://example.com", f"h{i}")
with self.assertRaises(RegistryQuarantineError):
store.register_fetch("pack-overflow", "1.0.0", "https://example.com", "hx")
def test_persistence(self):
"""F41: entries persist across store instances."""
from services.registry_quarantine import RegistryQuarantineStore
store1 = RegistryQuarantineStore(self.test_dir)
store1.register_fetch("persist-test", "1.0.0", "https://example.com", "abc")
# Create new store instance pointing to same dir
store2 = RegistryQuarantineStore(self.test_dir)
entry = store2.get_entry("persist-test", "1.0.0")
self.assertIsNotNone(entry)
self.assertEqual(entry.name, "persist-test")
def test_list_with_filter(self):
"""F41: list entries with state filter."""
from services.registry_quarantine import (
QuarantineState,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("a", "1.0", "https://example.com", "h1")
store.register_fetch("b", "1.0", "https://example.com", "h2")
store.verify_integrity("b", "1.0", "h2")
fetched = store.list_entries(state_filter=QuarantineState.FETCHED.value)
self.assertEqual(len(fetched), 1)
self.assertEqual(fetched[0].name, "a")
verified = store.list_entries(state_filter=QuarantineState.VERIFIED.value)
self.assertEqual(len(verified), 1)
self.assertEqual(verified[0].name, "b")
def test_remove_entry_requires_terminal_state(self):
"""F41: can only remove rejected/rolled_back entries."""
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("pack", "1.0", "https://example.com", "h")
# Should fail — still in FETCHED state
with self.assertRaises(RegistryQuarantineError):
store.remove_entry("pack", "1.0")
def test_audit_trail_accumulates(self):
"""F41: audit trail grows with each action."""
from services.registry_quarantine import RegistryQuarantineStore
store = RegistryQuarantineStore(self.test_dir)
store.register_fetch("audit-test", "1.0", "https://example.com", "h")
store.verify_integrity("audit-test", "1.0", "h")
store.activate("audit-test", "1.0")
entry = store.get_entry("audit-test", "1.0")
self.assertEqual(len(entry.audit_trail), 3)
actions = [a["action"] for a in entry.audit_trail]
self.assertEqual(actions, ["fetch", "verify", "activate"])
# ---------------------------------------------------------------------------
# F42 — Constrained Transform Tests
# ---------------------------------------------------------------------------
class TestConstrainedTransforms(unittest.TestCase):
"""F42: Constrained transform execution tests."""
def setUp(self):
self.test_dir = tempfile.mkdtemp()
self.transforms_dir = os.path.join(self.test_dir, "transforms")
os.makedirs(self.transforms_dir, exist_ok=True)
self.old_flag = os.environ.get("OPENCLAW_ENABLE_TRANSFORMS")
os.environ["OPENCLAW_ENABLE_TRANSFORMS"] = "1"
def tearDown(self):
shutil.rmtree(self.test_dir)
if self.old_flag is None:
os.environ.pop("OPENCLAW_ENABLE_TRANSFORMS", None)
else:
os.environ["OPENCLAW_ENABLE_TRANSFORMS"] = self.old_flag
def _write_transform(self, filename: str, code: str) -> str:
"""Write a transform module and return its path."""
path = os.path.join(self.transforms_dir, filename)
with open(path, "w", encoding="utf-8") as f:
f.write(code)
return path
def test_default_off(self):
"""F42: transforms are disabled by default."""
from services.constrained_transforms import is_transforms_enabled
old = os.environ.pop("OPENCLAW_ENABLE_TRANSFORMS", None)
try:
self.assertFalse(is_transforms_enabled())
finally:
if old is not None:
os.environ["OPENCLAW_ENABLE_TRANSFORMS"] = old
def test_disabled_execution_denied(self):
"""F42: transform execution is denied when disabled."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
old = os.environ.pop("OPENCLAW_ENABLE_TRANSFORMS", None)
try:
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
executor = TransformExecutor(registry)
result = executor.execute_transform("any-id", {})
self.assertEqual(result.status, TransformStatus.DENIED.value)
finally:
if old is not None:
os.environ["OPENCLAW_ENABLE_TRANSFORMS"] = old
def test_register_and_execute_simple(self):
"""F42: register and execute a simple transform."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
# Write a simple transform
path = self._write_transform(
"double_value.py",
'def transform(data):\n return {"result": data.get("value", 0) * 2}\n',
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("double", path, label="Double Value")
executor = TransformExecutor(registry)
result = executor.execute_transform("double", {"value": 21})
self.assertEqual(result.status, TransformStatus.SUCCESS.value)
self.assertEqual(result.output, {"result": 42})
self.assertGreaterEqual(result.duration_ms, 0)
def test_untrusted_directory_rejected(self):
"""F42: modules from untrusted directories are rejected."""
from services.constrained_transforms import (
TransformRegistry,
TransformRegistryError,
)
untrusted = tempfile.mkdtemp()
try:
path = os.path.join(untrusted, "evil.py")
with open(path, "w") as f:
f.write('def transform(d): return {"evil": True}\n')
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
with self.assertRaises(TransformRegistryError):
registry.register_transform("evil", path)
finally:
shutil.rmtree(untrusted)
def test_non_py_rejected(self):
"""F42: only .py files are allowed."""
from services.constrained_transforms import (
TransformRegistry,
TransformRegistryError,
)
path = os.path.join(self.transforms_dir, "transform.js")
with open(path, "w") as f:
f.write("module.exports = {};\n")
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
with self.assertRaises(TransformRegistryError):
registry.register_transform("js-transform", path)
def test_integrity_verification(self):
"""F42: integrity check detects tampered modules."""
from services.constrained_transforms import TransformRegistry
path = self._write_transform(
"integrity.py", 'def transform(d): return {"ok": True}\n'
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("integrity-test", path)
# Integrity passes initially
self.assertTrue(registry.verify_integrity("integrity-test"))
# Tamper with the file
with open(path, "w") as f:
f.write('def transform(d): return {"hacked": True}\n')
# Integrity fails after tampering
self.assertFalse(registry.verify_integrity("integrity-test"))
def test_tampered_execution_denied(self):
"""F42: execution is denied if integrity check fails."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
path = self._write_transform(
"tamper.py", 'def transform(d): return {"ok": True}\n'
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("tamper-test", path)
# Tamper
with open(path, "w") as f:
f.write('def transform(d): return {"evil": True}\n')
executor = TransformExecutor(registry)
result = executor.execute_transform("tamper-test", {})
self.assertEqual(result.status, TransformStatus.DENIED.value)
self.assertIn("integrity", result.error.lower())
def test_timeout_enforcement(self):
"""F42: transforms that exceed timeout are killed."""
from services.constrained_transforms import (
TransformExecutor,
TransformLimits,
TransformRegistry,
TransformStatus,
)
path = self._write_transform(
"slow.py",
"import time\ndef transform(d):\n time.sleep(10)\n return {}\n",
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("slow", path)
limits = TransformLimits(timeout_sec=0.5, max_output_bytes=65536)
executor = TransformExecutor(registry, limits)
result = executor.execute_transform("slow", {})
self.assertEqual(result.status, TransformStatus.TIMEOUT.value)
def test_output_size_cap(self):
"""F42: output exceeding size limit is rejected."""
from services.constrained_transforms import (
TransformExecutor,
TransformLimits,
TransformRegistry,
TransformStatus,
)
path = self._write_transform(
"big_output.py",
'def transform(d):\n return {"data": "x" * 100000}\n',
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("big", path)
limits = TransformLimits(timeout_sec=5, max_output_bytes=1024) # 1KB limit
executor = TransformExecutor(registry, limits)
result = executor.execute_transform("big", {})
self.assertEqual(result.status, TransformStatus.ERROR.value)
self.assertIn("size", result.error.lower())
def test_non_dict_return_rejected(self):
"""F42: transforms must return dict."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
path = self._write_transform(
"bad_return.py", 'def transform(d): return "not a dict"\n'
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("bad-return", path)
executor = TransformExecutor(registry)
result = executor.execute_transform("bad-return", {})
self.assertEqual(result.status, TransformStatus.ERROR.value)
self.assertIn("dict", result.error.lower())
def test_missing_transform_function(self):
"""F42: modules without transform() function fail."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
path = self._write_transform("no_func.py", "x = 42\n")
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("no-func", path)
executor = TransformExecutor(registry)
result = executor.execute_transform("no-func", {})
self.assertEqual(result.status, TransformStatus.ERROR.value)
self.assertIn("transform", result.error.lower())
def test_chain_execution(self):
"""F42: sequential chain execution."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
self._write_transform(
"add_one.py",
'def transform(d):\n return {"value": d.get("value", 0) + 1}\n',
)
self._write_transform(
"double.py",
'def transform(d):\n return {"value": d.get("value", 0) * 2}\n',
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform(
"add1", os.path.join(self.transforms_dir, "add_one.py")
)
registry.register_transform(
"dbl", os.path.join(self.transforms_dir, "double.py")
)
executor = TransformExecutor(registry)
results = executor.execute_chain(["add1", "dbl"], {"value": 5})
self.assertEqual(len(results), 2)
self.assertEqual(results[0].status, TransformStatus.SUCCESS.value)
self.assertEqual(results[0].output["value"], 6)
self.assertEqual(results[1].status, TransformStatus.SUCCESS.value)
self.assertEqual(results[1].output["value"], 12)
def test_chain_stops_on_error(self):
"""F42: chain stops on first error."""
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
self._write_transform("fail.py", 'def transform(d): raise ValueError("boom")\n')
self._write_transform("ok.py", 'def transform(d): return {"ok": True}\n')
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform(
"fail", os.path.join(self.transforms_dir, "fail.py")
)
registry.register_transform("ok", os.path.join(self.transforms_dir, "ok.py"))
executor = TransformExecutor(registry)
results = executor.execute_chain(["fail", "ok"], {})
self.assertEqual(len(results), 1)
self.assertEqual(results[0].status, TransformStatus.ERROR.value)
def test_chain_limit_enforcement(self):
"""F42: chain exceeding max transforms limit is denied."""
from services.constrained_transforms import (
TransformExecutor,
TransformLimits,
TransformRegistry,
TransformStatus,
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
limits = TransformLimits(max_transforms_per_request=2)
executor = TransformExecutor(registry, limits)
results = executor.execute_chain(["a", "b", "c"], {})
self.assertEqual(len(results), 1)
self.assertEqual(results[0].status, TransformStatus.DENIED.value)
def test_module_size_limit(self):
"""F42: oversized modules are rejected."""
from services.constrained_transforms import (
MAX_TRANSFORM_MODULE_SIZE_BYTES,
TransformRegistry,
TransformRegistryError,
)
path = self._write_transform(
"huge.py", "x = 1\n" * (MAX_TRANSFORM_MODULE_SIZE_BYTES // 6 + 100)
)
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
with self.assertRaises(TransformRegistryError):
registry.register_transform("huge", path)
def test_unregister(self):
"""F42: unregistering removes from registry."""
from services.constrained_transforms import TransformRegistry
path = self._write_transform("removable.py", "def transform(d): return {}\n")
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("removable", path)
self.assertIsNotNone(registry.get_transform("removable"))
result = registry.unregister_transform("removable")
self.assertTrue(result)
self.assertIsNone(registry.get_transform("removable"))
def test_list_transforms(self):
"""F42: list registered transforms."""
from services.constrained_transforms import TransformRegistry
self._write_transform("t1.py", "def transform(d): return {}\n")
self._write_transform("t2.py", "def transform(d): return {}\n")
registry = TransformRegistry(self.test_dir, [self.transforms_dir])
registry.register_transform("t1", os.path.join(self.transforms_dir, "t1.py"))
registry.register_transform("t2", os.path.join(self.transforms_dir, "t2.py"))
transforms = registry.list_transforms()
self.assertEqual(len(transforms), 2)
ids = {t.id for t in transforms}
self.assertEqual(ids, {"t1", "t2"})
# ---------------------------------------------------------------------------
# Regression: default-off behavior
# ---------------------------------------------------------------------------
class TestDefaultOffRegression(unittest.TestCase):
"""Verify F41/F42 are disabled by default and fail closed."""
def setUp(self):
self.saved_env = {}
for key in ("OPENCLAW_ENABLE_REGISTRY_SYNC", "OPENCLAW_ENABLE_TRANSFORMS"):
self.saved_env[key] = os.environ.get(key)
os.environ.pop(key, None)
def tearDown(self):
for key, val in self.saved_env.items():
if val is None:
os.environ.pop(key, None)
else:
os.environ[key] = val
def test_registry_sync_off(self):
from services.registry_quarantine import is_registry_sync_enabled
self.assertFalse(is_registry_sync_enabled())
def test_transforms_off(self):
from services.constrained_transforms import is_transforms_enabled
self.assertFalse(is_transforms_enabled())
def test_registry_fail_closed(self):
from services.registry_quarantine import (
RegistryQuarantineError,
RegistryQuarantineStore,
)
store = RegistryQuarantineStore(tempfile.mkdtemp())
with self.assertRaises(RegistryQuarantineError):
store.register_fetch("x", "1.0", "https://example.com", "h")
def test_transform_execution_fail_closed(self):
from services.constrained_transforms import (
TransformExecutor,
TransformRegistry,
TransformStatus,
)
registry = TransformRegistry(tempfile.mkdtemp(), [])
executor = TransformExecutor(registry)
result = executor.execute_transform("x", {})
self.assertEqual(result.status, TransformStatus.DENIED.value)
# ---------------------------------------------------------------------------
# Regression: F40 mapping-only still works without transforms
# ---------------------------------------------------------------------------
class TestF40MappingOnlyRegression(unittest.TestCase):
"""Verify F40 mapping engine works when F42 transforms are disabled."""
def test_mapping_works_without_transforms(self):
"""F40 mapping-only mode remains functional."""
try:
from services.webhook_mapping import (
CoercionType,
FieldMapping,
MappingProfile,
apply_mapping,
)
except ImportError:
self.skipTest("webhook_mapping not available")
profile = MappingProfile(
id="test-profile",
label="Test",
field_mappings=[
FieldMapping(
source_path="data.msg",
target_path="prompt",
coercion=CoercionType.STRING,
),
],
)
result, warnings = apply_mapping(profile, {"data": {"msg": "hello"}})
self.assertEqual(result.get("prompt"), "hello")
if __name__ == "__main__":
unittest.main()