feat(security): complete wave2 hardening with S12 tool runner, S30 wave2 doctor checks, and R78 auth coverage for tools routes

README.md

@@ -22,6 +22,33 @@ It is designed to make **ComfyUI a reliable automation target** with an explicit
 
 ## Latest Updates - Click to expand
 
+<details>
+<summary><strong>Security Hardening: Observability/Auth boundaries, transform isolation, integrity checks, and safe tooling controls</strong></summary>
+
+- Delivered observability tier hardening with explicit sensitivity split:
+  - Public-safe: `/openclaw/health`
+  - Observability token: `/openclaw/config`, `/openclaw/events`, `/openclaw/events/stream`
+  - Admin-only: `/openclaw/logs/tail`, `/openclaw/trace/{prompt_id}`, `/openclaw/secrets/status`, `/openclaw/security/doctor`
+- Delivered constrained transform isolation hardening:
+  - process-boundary execution via `TransformProcessRunner`
+  - timeout/output caps and network-deny worker posture
+  - feature-gated default-off behavior for safer rollout
+- Delivered approval/checkpoint integrity hardening:
+  - canonical JSON + SHA-256 integrity envelopes
+  - tamper detection and fail-closed handling on integrity violations
+  - migration-safe loading behavior for legacy persistence files
+- Delivered external tooling execution policy:
+  - allowlist-driven tool definitions (`data/tools_allowlist.json`)
+  - strict argument validation, bounded timeout/output, and redacted output handling
+  - gated by `OPENCLAW_ENABLE_EXTERNAL_TOOLS` plus admin access policy
+- Extended security doctor coverage with wave-2 checks:
+  - validates transform isolation posture
+  - reports external tooling posture
+  - verifies integrity module availability
+- Auth-coverage contract tests were updated to include new tool routes and prevent future route-auth drift regressions.
+
+</details>
+
 <details>
 <summary><strong>Sprint A: closes out with five concrete reliability and security improvements</strong></summary>
 

api/routes.py

@@ -56,6 +56,7 @@ if web is not None:
         )
         from ..api.security_doctor import security_doctor_handler  # S30
         from ..api.templates import templates_list_handler
+        from ..api.tools import tools_list_handler, tools_run_handler  # S12
         from ..api.webhook import webhook_handler
         from ..api.webhook_submit import webhook_submit_handler
         from ..api.webhook_validate import webhook_validate_handler
@@ -106,6 +107,7 @@ if web is not None:
         )
         from api.security_doctor import security_doctor_handler  # type: ignore  # S30
         from api.templates import templates_list_handler
+        from api.tools import tools_list_handler, tools_run_handler  # S12
         from api.webhook import webhook_handler
         from api.webhook_submit import webhook_submit_handler
         from api.webhook_validate import webhook_validate_handler
@@ -538,6 +540,16 @@ def register_routes(server) -> None:
                 f"{prefix}/security/doctor",
                 security_doctor_handler,
             ),  # S30: Security Doctor diagnostics
+            (
+                "GET",
+                f"{prefix}/tools",
+                tools_list_handler,
+            ),  # S12: List allowed tools
+            (
+                "POST",
+                f"{prefix}/tools/{{name}}/run",
+                tools_run_handler,
+            ),  # S12: Execute tool (admin only)
         ]
 
         for method, path, handler in core_routes:

api/tools.py

@@ -0,0 +1,103 @@
+"""
+S12: API Handlers for External Tools.
+Protected by Admin Token and Feature Flag.
+"""
+
+import json
+import logging
+
+from aiohttp import web
+
+try:
+    from ..services.access_control import require_admin_token
+    from ..services.tool_runner import get_tool_runner, is_tools_enabled
+except ImportError:
+    from services.access_control import require_admin_token
+    from services.tool_runner import get_tool_runner, is_tools_enabled
+
+logger = logging.getLogger("ComfyUI-OpenClaw.api.tools")
+
+
+async def tools_list_handler(request: web.Request) -> web.Response:
+    """
+    GET /openclaw/tools
+    List allowed external tools.
+    Requires: Admin Token.
+    """
+    if not is_tools_enabled():
+        return web.json_response(
+            {"ok": False, "error": "External tooling is disabled (feature flag off)."},
+            status=404,  # Not Found or Forbidden? 404 implies feature doesn't exist.
+        )
+
+    # Admin check
+    allowed, error = require_admin_token(request)
+    if not allowed:
+        return web.json_response({"ok": False, "error": error}, status=403)
+
+    runner = get_tool_runner()
+    tools = runner.list_tools()
+
+    return web.json_response({"ok": True, "tools": tools})
+
+
+async def tools_run_handler(request: web.Request) -> web.Response:
+    """
+    POST /openclaw/tools/{name}/run
+    Execute an external tool.
+    Body: {"args": {"arg1": "val1", ...}}
+    Requires: Admin Token.
+    """
+    if not is_tools_enabled():
+        return web.json_response(
+            {"ok": False, "error": "External tooling is disabled."}, status=404
+        )
+
+    # Admin check
+    allowed, error = require_admin_token(request)
+    if not allowed:
+        return web.json_response({"ok": False, "error": error}, status=403)
+
+    tool_name = request.match_info.get("name")
+    if not tool_name:
+        return web.json_response(
+            {"ok": False, "error": "Tool name required"}, status=400
+        )
+
+    try:
+        body = await request.json()
+    except json.JSONDecodeError:
+        return web.json_response(
+            {"ok": False, "error": "Invalid JSON body"}, status=400
+        )
+
+    args = body.get("args", {})
+    if not isinstance(args, dict):
+        return web.json_response(
+            {"ok": False, "error": "'args' must be a dictionary"}, status=400
+        )
+
+    runner = get_tool_runner()
+    result = runner.execute_tool(tool_name, args)
+
+    if not result.success:
+        return web.json_response(
+            {
+                "ok": False,
+                "tool": tool_name,
+                "error": result.error,
+                "output": result.output,  # Redacted output might contain useful error info
+                "exit_code": result.exit_code,
+                "duration_ms": result.duration_ms,
+            },
+            status=500 if result.error else 400,
+        )  # 500 for runtime error, 400 for validation?
+
+    return web.json_response(
+        {
+            "ok": True,
+            "tool": tool_name,
+            "output": result.output,
+            "duration_ms": result.duration_ms,
+        }
+    )

data/tools_allowlist.json

@@ -0,0 +1,17 @@
+{
+    "tools": [
+        {
+            "name": "example_echo",
+            "description": "Example tool that echoes input",
+            "command": [
+                "echo",
+                "{message}"
+            ],
+            "args": {
+                "message": "^[a-zA-Z0-9_ -]+$"
+            },
+            "timeout_sec": 5,
+            "max_output_bytes": 1024
+        }
+    ]
+}

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "comfyui-openclaw"
 description = "Your own personal AIGC Factory. Any picture. Any reel. The Comfy way.©️"
-version = "0.2.6"
+version = "0.2.8"
 license = {text = "MIT"}
 readme = "README.md"
 requires-python = ">=3.10"

services/security_doctor.py

@@ -947,6 +947,111 @@ def apply_guarded_remediation(
         return False
 
 
+# ---------------------------------------------------------------------------
+# Security checks — Wave 2 Hardening (S35, S12, R77)
+# ---------------------------------------------------------------------------
+
+
+def check_hardening_wave2(report: SecurityReport) -> None:
+    """Verify Security Hardening Wave 2 status."""
+
+    # 1. S35 Transform Isolation
+    try:
+        from .constrained_transforms import get_transform_executor
+        from .transform_common import is_transforms_enabled
+        from .transform_runner import TransformProcessRunner
+
+        if not is_transforms_enabled():
+            report.add(
+                SecurityCheckResult(
+                    name="s35_isolation",
+                    severity=SecuritySeverity.SKIP.value,
+                    message="Transforms disabled (feature flag off)",
+                    category="wave2",
+                )
+            )
+        else:
+            executor = get_transform_executor()
+            if isinstance(executor, TransformProcessRunner):
+                report.add(
+                    SecurityCheckResult(
+                        name="s35_isolation",
+                        severity=SecuritySeverity.PASS.value,
+                        message="S35: Process isolation active",
+                        category="wave2",
+                    )
+                )
+            else:
+                report.add(
+                    SecurityCheckResult(
+                        name="s35_isolation",
+                        severity=SecuritySeverity.FAIL.value,
+                        message="S35: Process isolation NOT active (using thread/unsafe executor)",
+                        category="wave2",
+                        detail=f"Current executor: {type(executor)}",
+                        remediation="Ensure TransformProcessRunner is used.",
+                    )
+                )
+
+    except ImportError:
+        report.add(
+            SecurityCheckResult(
+                name="s35_isolation",
+                severity=SecuritySeverity.FAIL.value,
+                message="S35: Modules not importable",
+                category="wave2",
+            )
+        )
+
+    # 2. S12 Tooling (Opt-in)
+    try:
+        from .tool_runner import is_tools_enabled
+
+        if is_tools_enabled():
+            report.add(
+                SecurityCheckResult(
+                    name="s12_tooling",
+                    severity=SecuritySeverity.WARN.value,
+                    message="S12: External tooling ENABLED (admin-only)",
+                    category="wave2",
+                    detail="Ensure tools_allowlist.json is strict.",
+                )
+            )
+        else:
+            report.add(
+                SecurityCheckResult(
+                    name="s12_tooling",
+                    severity=SecuritySeverity.PASS.value,
+                    message="S12: External tooling disabled (safe default)",
+                    category="wave2",
+                )
+            )
+    except ImportError:
+        pass
+
+    # 3. R77 Integrity (Existence Check)
+    try:
+        from .integrity import load_verified
+
+        report.add(
+            SecurityCheckResult(
+                name="r77_integrity",
+                severity=SecuritySeverity.PASS.value,
+                message="R77: Integrity module loaded",
+                category="wave2",
+            )
+        )
+    except ImportError:
+        report.add(
+            SecurityCheckResult(
+                name="r77_integrity",
+                severity=SecuritySeverity.FAIL.value,
+                message="R77: Integrity module missing",
+                category="wave2",
+            )
+        )
+
+
 # ---------------------------------------------------------------------------
 # Main runner
 # ---------------------------------------------------------------------------
@@ -982,6 +1087,7 @@ def run_security_doctor(
     check_feature_flags(report)
     check_api_key_posture(report)
     check_connector_security_posture(report)  # S32
+    check_hardening_wave2(report)  # Wave 2
 
     # Optional guarded remediation
     if remediate:

services/tool_runner.py

@@ -0,0 +1,269 @@
+"""
+S12: External Tool Runner.
+Implements a secure, allowlist-based execution environment for external CLI tools.
+Enforces strict argument validation, templating (no shell injection), timeouts, and output limits.
+"""
+
+import json
+import logging
+import os
+import re
+import shlex
+import subprocess
+import time
+from dataclasses import dataclass, field
+from typing import Any, Dict, List, Optional
+
+from .redaction import redact_text
+
+logger = logging.getLogger("ComfyUI-OpenClaw.services.tool_runner")
+
+# Default limits
+DEFAULT_TOOL_TIMEOUT_SEC = 30
+DEFAULT_TOOL_MAX_OUTPUT_BYTES = 64 * 1024  # 64KB
+
+
+def is_tools_enabled() -> bool:
+    """Check if external tooling is enabled (Opt-in)."""
+    return os.environ.get("OPENCLAW_ENABLE_EXTERNAL_TOOLS", "false").lower() in (
+        "true",
+        "1",
+        "yes",
+        "on",
+    )
+
+
+@dataclass
+class ToolResult:
+    """Result of a tool execution."""
+
+    tool_name: str
+    success: bool
+    output: str  # stdout + stderr (redacted)
+    duration_ms: float
+    error: Optional[str] = None
+    exit_code: Optional[int] = None
+
+
+@dataclass
+class ToolDefinition:
+    """Definition of an allowed tool."""
+
+    name: str
+    command_template: List[str]  # e.g. ["git", "log", "-n", "{limit}"]
+    allowed_args: Dict[str, str]  # regex patterns for each arg key
+    timeout_sec: int = DEFAULT_TOOL_TIMEOUT_SEC
+    max_output_bytes: int = DEFAULT_TOOL_MAX_OUTPUT_BYTES
+    description: str = ""
+
+    def validate_args(self, args: Dict[str, str]) -> None:
+        """Validate provided arguments against regex patterns."""
+        # 1. Strict check: reject unknown arguments
+        unknown_args = set(args.keys()) - set(self.allowed_args.keys())
+        if unknown_args:
+            raise ValueError(
+                f"Unknown arguments provided: {unknown_args}. Allowed: {list(self.allowed_args.keys())}"
+            )
+
+        # 2. Regex check
+        for key, pattern in self.allowed_args.items():
+            if key in args:
+                val = str(args[key])
+                if not re.fullmatch(pattern, val):
+                    raise ValueError(
+                        f"Argument '{key}' validation failed: '{val}' does not match pattern '{pattern}'"
+                    )
+
+
+class ToolRunner:
+    """
+    Secure runner for external tools.
+    """
+
+    def __init__(self, config_path: Optional[str] = None):
+        self._tools: Dict[str, ToolDefinition] = {}
+        self._config_path = config_path or os.environ.get("OPENCLAW_TOOLS_CONFIG_PATH")
+        if not self._config_path:
+            # Default to data/tools_allowlist.json (shipped default)
+            try:
+                from config import DATA_DIR
+
+                self._config_path = os.path.join(DATA_DIR, "tools_allowlist.json")
+            except ImportError:
+                # Fallback for unconnected tests
+                self._config_path = "data/tools_allowlist.json"
+
+        self.reload_config()
+
+    def reload_config(self):
+        """Load tools configuration from disk."""
+        self._tools = {}
+        if not os.path.exists(self._config_path):
+            logger.info(
+                f"No tools config found at {self._config_path}. Tool runner invalid/empty."
+            )
+            return
+
+        try:
+            with open(self._config_path, "r", encoding="utf-8") as f:
+                data = json.load(f)
+
+            for tool_data in data.get("tools", []):
+                try:
+                    tool = ToolDefinition(
+                        name=tool_data["name"],
+                        command_template=tool_data["command"],
+                        allowed_args=tool_data.get("args", {}),
+                        timeout_sec=tool_data.get(
+                            "timeout_sec", DEFAULT_TOOL_TIMEOUT_SEC
+                        ),
+                        max_output_bytes=tool_data.get(
+                            "max_output_bytes", DEFAULT_TOOL_MAX_OUTPUT_BYTES
+                        ),
+                        description=tool_data.get("description", ""),
+                    )
+                    self._tools[tool.name] = tool
+                except Exception as e:
+                    logger.warning(f"Skipping invalid tool definition: {e}")
+
+            logger.info(
+                f"S12: Loaded {len(self._tools)} allowed tools from {self._config_path}"
+            )
+
+        except Exception as e:
+            logger.error(f"Failed to load tools config: {e}")
+
+    def list_tools(self) -> List[Dict[str, Any]]:
+        """Return metadata of allowed tools."""
+        return [
+            {
+                "name": t.name,
+                "description": t.description,
+                "args": list(t.allowed_args.keys()),
+            }
+            for t in self._tools.values()
+        ]
+
+    def _sanitize_env(self) -> Dict[str, str]:
+        """Create a sanitized environment (blocklist approach)."""
+        env = os.environ.copy()
+        for key in list(env.keys()):
+            upper_key = key.upper()
+            if (
+                "TOKEN" in upper_key
+                or "SECRET" in upper_key
+                or "KEY" in upper_key
+                or "PASSWORD" in upper_key
+            ):
+                del env[key]
+        return env
+
+    def execute_tool(self, tool_name: str, arguments: Dict[str, str]) -> ToolResult:
+        """
+        Execute a named tool with validated arguments.
+        """
+        start_time = time.monotonic()
+
+        tool = self._tools.get(tool_name)
+        if not tool:
+            return ToolResult(
+                tool_name=tool_name,
+                success=False,
+                output="",
+                duration_ms=0,
+                error=f"Tool '{tool_name}' not allowed or not found.",
+            )
+
+        try:
+            # 1. Validate Arguments
+            tool.validate_args(arguments)
+
+            # 2. Build Command
+            cmd = []
+            for part in tool.command_template:
+                # Use format(), args are validated strings
+                try:
+                    rendered = part.format(**arguments)
+                    cmd.append(rendered)
+                except KeyError as e:
+                    # This happens if template needs {arg} but it wasn't provided (and regex validation passed implicitly if arg was optional?)
+                    # Actually regex validation iterates over Allowed, checks if in Args.
+                    # But template needs specific args.
+                    # If allowed_args has 'limit' but args doesn't, and template needs {limit}...
+                    raise ValueError(
+                        f"Missing required argument for tool template: {e}"
+                    )
+
+            logger.info(f"S12: Executing tool '{tool_name}'")
+            # Log redacted command just in case
+            logger.debug(f"Command: {redact_text(str(cmd))}")
+
+            # 3. Execute
+            # S12: Capability restriction - strict timeout, limited output
+            # S12: Env Sanitization
+            clean_env = self._sanitize_env()
+
+            proc = subprocess.run(
+                cmd,
+                capture_output=True,
+                text=True,
+                timeout=tool.timeout_sec,
+                check=False,
+                env=clean_env,
+            )
+
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+
+            # 4. Process Output
+            raw_output = proc.stdout + proc.stderr
+            # Enforce size limit
+            if len(raw_output.encode("utf-8")) > tool.max_output_bytes:
+                raw_output = raw_output[: tool.max_output_bytes] + "... [TRUNCATED]"
+
+            # S12: Redact output
+            clean_output = redact_text(raw_output)
+
+            success = proc.returncode == 0
+
+            return ToolResult(
+                tool_name=tool_name,
+                success=success,
+                output=clean_output,
+                duration_ms=elapsed_ms,
+                exit_code=proc.returncode,
+                error=(
+                    None if success else f"Process exited with code {proc.returncode}"
+                ),
+            )
+
+        except subprocess.TimeoutExpired:
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+            logger.warning(f"Tool '{tool_name}' timed out after {tool.timeout_sec}s")
+            return ToolResult(
+                tool_name=tool_name,
+                success=False,
+                output="",
+                duration_ms=elapsed_ms,
+                error="Execution timed out",
+            )
+        except Exception as e:
+            elapsed_ms = (time.monotonic() - start_time) * 1000
+            logger.error(f"Error executing tool '{tool_name}': {e}")
+            return ToolResult(
+                tool_name=tool_name,
+                success=False,
+                output="",
+                duration_ms=elapsed_ms,
+                error=str(e),
+            )
+
+
+# Global singleton
+_runner = None
+
+
+def get_tool_runner() -> ToolRunner:
+    global _runner
+    if _runner is None:
+        _runner = ToolRunner()
+    return _runner

tests/test_s12_tool_runner.py

@@ -0,0 +1,90 @@
+"""
+S12 Tool Runner Tests.
+"""
+
+import json
+import os
+import unittest
+from unittest.mock import MagicMock, patch
+
+from services.tool_runner import ToolDefinition, ToolResult, ToolRunner
+
+
+class TestS12ToolRunner(unittest.TestCase):
+
+    def setUp(self):
+        self.runner = ToolRunner(config_path="dummy_path")  # Won't load anything real
+
+        # Manually inject a test tool
+        self.tool = ToolDefinition(
+            name="test_echo",
+            command_template=["echo", "{msg}"],
+            allowed_args={"msg": "^[a-z0-9]+$"},
+            timeout_sec=1,
+            max_output_bytes=100,
+        )
+        self.runner._tools["test_echo"] = self.tool
+
+    def test_validate_args_success(self):
+        self.tool.validate_args({"msg": "hello"})
+
+    def test_validate_args_failure_regex(self):
+        with self.assertRaises(ValueError):
+            self.tool.validate_args({"msg": "BAD@KEY"})
+
+    def test_validate_args_failure_unknown(self):
+        with self.assertRaises(ValueError):
+            self.tool.validate_args({"msg": "hello", "unknown": "val"})
+
+    @patch("subprocess.run")
+    def test_execute_success(self, mock_run):
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        mock_proc.stdout = "hello\n"
+        mock_proc.stderr = ""
+        mock_run.return_value = mock_proc
+
+        result = self.runner.execute_tool("test_echo", {"msg": "hello"})
+
+        self.assertTrue(result.success)
+        self.assertEqual(result.output, "hello\n")
+        self.assertEqual(result.exit_code, 0)
+
+        # Verify call args
+        args, kwargs = mock_run.call_args
+        self.assertEqual(args[0], ["echo", "hello"])
+
+    @patch("subprocess.run")
+    def test_env_sanitization(self, mock_run):
+        mock_proc = MagicMock()
+        mock_proc.returncode = 0
+        mock_proc.stdout = ""
+        mock_proc.stderr = ""
+        mock_run.return_value = mock_proc
+
+        with patch.dict(
+            os.environ, {"SECRET_KEY": "fail", "TOKEN_X": "fail", "SAFE_VAR": "ok"}
+        ):
+            self.runner.execute_tool("test_echo", {"msg": "hello"})
+
+            args, kwargs = mock_run.call_args
+            env_used = kwargs["env"]
+
+            self.assertIn("SAFE_VAR", env_used)
+            self.assertNotIn("SECRET_KEY", env_used)
+            self.assertNotIn("TOKEN_X", env_used)
+
+    @patch("subprocess.run")
+    def test_timeout(self, mock_run):
+        import subprocess
+
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd=["echo"], timeout=1)
+
+        result = self.runner.execute_tool("test_echo", {"msg": "hello"})
+
+        self.assertFalse(result.success)
+        self.assertEqual(result.error, "Execution timed out")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_s30_security_doctor_wave2.py

@@ -0,0 +1,79 @@
+"""
+S30 Security Doctor Wave 2 Tests.
+"""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from services.security_doctor import (
+    SecurityReport,
+    SecuritySeverity,
+    check_hardening_wave2,
+)
+
+
+class TestS30Wave2(unittest.TestCase):
+
+    def setUp(self):
+        self.report = SecurityReport()
+
+    @patch("services.transform_common.is_transforms_enabled")
+    def test_s35_disabled(self, mock_enabled):
+        mock_enabled.return_value = False
+        check_hardening_wave2(self.report)
+
+        # Verify result
+        # Check list of check names content
+        names = [c.name for c in self.report.checks]
+        self.assertIn("s35_isolation", names)
+
+        check = next(c for c in self.report.checks if c.name == "s35_isolation")
+        self.assertEqual(check.severity, SecuritySeverity.SKIP.value)
+
+    @patch("services.transform_common.is_transforms_enabled")
+    @patch("services.constrained_transforms.get_transform_executor")
+    def test_s35_active(self, mock_executor, mock_enabled):
+        mock_enabled.return_value = True
+
+        # Mock class matching TransformProcessRunner
+        # Ensure we can import it to compare types
+        from services.transform_runner import TransformProcessRunner
+
+        # Determine the real type or mock it?
+        # The code checks `isinstance(executor, TransformProcessRunner)`.
+        # So we need mock_executor.return_value to be an instance of TransformProcessRunner.
+        # Create a real instance or a mock spec?
+        # A mock with spec should satisfy isinstance if spec is the class for some mock libs, but safer is:
+        executor_instance = MagicMock(spec=TransformProcessRunner)
+        # However, isinstance(mock, Class) returns True only if spec=Class is set AND the mock library handles it.
+        # unittest.mock.MagicMock(spec=Class) DOES satisfy isinstance check.
+        mock_executor.return_value = executor_instance
+
+        check_hardening_wave2(self.report)
+
+        check = next(c for c in self.report.checks if c.name == "s35_isolation")
+        # If passed:
+        if check.severity != SecuritySeverity.PASS.value:
+            print(f"Check message: {check.message}")
+
+        self.assertEqual(check.severity, SecuritySeverity.PASS.value)
+
+    @patch("services.tool_runner.is_tools_enabled")
+    def test_s12_enabled(self, mock_enabled):
+        mock_enabled.return_value = True
+        check_hardening_wave2(self.report)
+
+        check = next(c for c in self.report.checks if c.name == "s12_tooling")
+        self.assertEqual(check.severity, SecuritySeverity.WARN.value)
+
+    @patch("services.tool_runner.is_tools_enabled")
+    def test_s12_disabled(self, mock_enabled):
+        mock_enabled.return_value = False
+        check_hardening_wave2(self.report)
+        check = next((c for c in self.report.checks if c.name == "s12_tooling"), None)
+        if check:
+            self.assertEqual(check.severity, SecuritySeverity.PASS.value)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_s33r76r78_auth_hardening.py

@@ -57,6 +57,8 @@ AUTH_CLASS_BY_ROUTE = {
     ("GET", "/events/stream"): "observability",
     ("GET", "/events"): "observability",
     ("GET", "/security/doctor"): "admin",
+    ("GET", "/tools"): "admin",
+    ("POST", "/tools/{name}/run"): "admin",
 }
 
 OPTIONAL_SUFFIX_PREFIXES = (

verify_s30_doctor.py

@@ -0,0 +1,33 @@
+"""
+Verify S30 Security Doctor output.
+"""
+
+import os
+import sys
+
+# Add project root to path
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from services.security_doctor import run_security_doctor
+
+
+def main():
+    print("Running Security Doctor...")
+    # Enable tools/transforms for full check
+    os.environ["OPENCLAW_ENABLE_EXTERNAL_TOOLS"] = "true"
+    os.environ["OPENCLAW_ENABLE_TRANSFORMS"] = "true"
+    os.environ["OPENCLAW_ADMIN_TOKEN"] = "test-admin"
+    os.environ["OPENCLAW_OBSERVABILITY_TOKEN"] = "test-obs"
+
+    report = run_security_doctor(remediate=False)
+    print(report.to_human())
+
+    if report.has_failures:
+        print("\nFAILURE: Security Doctor reported failures.")
+        sys.exit(1)
+    else:
+        print("\nSUCCESS: Security Doctor reported no failures.")
+
+
+if __name__ == "__main__":
+    main()