Audit Openclaw Security

{
  "fs.state_dir.perms_world_writable": {
    "severity": "critical",
    "category": "filesystem",
    "why": "Other users or processes can modify the full OpenClaw state directory.",
    "primary_fix": "filesystem perms on ~/.openclaw",
    "auto_fix": true
  },
  "fs.config.perms_writable": {
    "severity": "critical",
    "category": "filesystem",
    "why": "Other users can change auth, tool policy, or routing config.",
    "primary_fix": "filesystem perms on ~/.openclaw/openclaw.json",
    "auto_fix": true
  },
  "fs.config.perms_world_readable": {
    "severity": "critical",
    "category": "filesystem",
    "why": "The config may leak tokens or other sensitive settings.",
    "primary_fix": "filesystem perms on the active config file",
    "auto_fix": true
  },
  "gateway.bind_no_auth": {
    "severity": "critical",
    "category": "network",
    "why": "The Gateway is remotely reachable without shared-secret auth.",
    "primary_fix": "gateway.bind, gateway.auth.*",
    "auto_fix": false
  },
  "gateway.loopback_no_auth": {
    "severity": "critical",
    "category": "network",
    "why": "A reverse-proxied loopback listener may become effectively unauthenticated.",
    "primary_fix": "gateway.auth.*, reverse-proxy setup",
    "auto_fix": false
  },
  "gateway.http.no_auth": {
    "severity": "warn/critical",
    "category": "network",
    "why": "HTTP API endpoints are reachable while `auth.mode=\"none\"`.",
    "primary_fix": "gateway.auth.mode, gateway.http.endpoints.*",
    "auto_fix": false
  },
  "gateway.tools_invoke_http.dangerous_allow": {
    "severity": "warn/critical",
    "category": "tools",
    "why": "Dangerous tools are reachable through HTTP API paths.",
    "primary_fix": "gateway.tools.allow",
    "auto_fix": false
  },
  "gateway.nodes.allow_commands_dangerous": {
    "severity": "warn/critical",
    "category": "nodes",
    "why": "High-impact node commands are reachable remotely.",
    "primary_fix": "gateway.nodes.allowCommands",
    "auto_fix": false
  },
  "gateway.tailscale_funnel": {
    "severity": "critical",
    "category": "network",
    "why": "Tailscale Funnel makes the Gateway publicly reachable.",
    "primary_fix": "gateway.tailscale.mode",
    "auto_fix": false
  },
  "gateway.control_ui.allowed_origins_required": {
    "severity": "critical",
    "category": "control-ui",
    "why": "A non-loopback Control UI is missing an explicit browser-origin allowlist.",
    "primary_fix": "gateway.controlUi.allowedOrigins",
    "auto_fix": false
  },
  "gateway.control_ui.host_header_origin_fallback": {
    "severity": "warn/critical",
    "category": "control-ui",
    "why": "Host-header origin fallback weakens DNS rebinding protections.",
    "primary_fix": "gateway.controlUi.dangerouslyAllowHostHeaderOriginFallback",
    "auto_fix": false
  },
  "gateway.control_ui.insecure_auth": {
    "severity": "warn",
    "category": "control-ui",
    "why": "Control UI insecure-auth compatibility mode is enabled.",
    "primary_fix": "gateway.controlUi.allowInsecureAuth",
    "auto_fix": false
  },
  "gateway.control_ui.device_auth_disabled": {
    "severity": "critical",
    "category": "control-ui",
    "why": "Device-identity checks are disabled entirely.",
    "primary_fix": "gateway.controlUi.dangerouslyDisableDeviceAuth",
    "auto_fix": false
  },
  "gateway.real_ip_fallback_enabled": {
    "severity": "warn/critical",
    "category": "reverse-proxy",
    "why": "Trusting X-Real-IP can enable source-IP spoofing via proxy misconfiguration.",
    "primary_fix": "gateway.allowRealIpFallback, gateway.trustedProxies",
    "auto_fix": false
  },
  "discovery.mdns_full_mode": {
    "severity": "warn/critical",
    "category": "discovery",
    "why": "mDNS full mode advertises extra metadata on the local network.",
    "primary_fix": "discovery.mdns.mode, gateway.bind",
    "auto_fix": false
  },
  "config.insecure_or_dangerous_flags": {
    "severity": "warn",
    "category": "config",
    "why": "One or more insecure or dangerous flags are enabled.",
    "primary_fix": "see finding detail",
    "auto_fix": false
  },
  "hooks.token_too_short": {
    "severity": "warn",
    "category": "hooks",
    "why": "The hook ingress token is easier to brute-force.",
    "primary_fix": "hooks.token",
    "auto_fix": false
  },
  "hooks.request_session_key_enabled": {
    "severity": "warn/critical",
    "category": "hooks",
    "why": "External callers can choose session keys, creating persistence or collision risk.",
    "primary_fix": "hooks.allowRequestSessionKey",
    "auto_fix": false
  },
  "hooks.request_session_key_prefixes_missing": {
    "severity": "warn/critical",
    "category": "hooks",
    "why": "There is no bound on externally supplied session-key shapes.",
    "primary_fix": "hooks.allowedSessionKeyPrefixes",
    "auto_fix": false
  },
  "logging.redact_off": {
    "severity": "warn",
    "category": "logging",
    "why": "Sensitive values can leak into logs and status output.",
    "primary_fix": "logging.redactSensitive",
    "auto_fix": true
  },
  "sandbox.docker_config_mode_off": {
    "severity": "warn",
    "category": "sandbox",
    "why": "Docker sandbox config is present but sandboxing is inactive.",
    "primary_fix": "agents.*.sandbox.mode",
    "auto_fix": false
  },
  "sandbox.dangerous_network_mode": {
    "severity": "critical",
    "category": "sandbox",
    "why": "The Docker sandbox uses host or namespace-join networking.",
    "primary_fix": "agents.*.sandbox.docker.network",
    "auto_fix": false
  },
  "tools.exec.host_sandbox_no_sandbox_defaults": {
    "severity": "warn",
    "category": "tools",
    "why": "Default `exec host=sandbox` resolves to host exec when sandboxing is off.",
    "primary_fix": "tools.exec.host, agents.defaults.sandbox.mode",
    "auto_fix": false
  },
  "tools.exec.host_sandbox_no_sandbox_agents": {
    "severity": "warn",
    "category": "tools",
    "why": "Per-agent `exec host=sandbox` resolves to host exec when sandboxing is off.",
    "primary_fix": "agents.list[].tools.exec.host, agents.list[].sandbox.mode",
    "auto_fix": false
  },
  "tools.exec.safe_bins_interpreter_unprofiled": {
    "severity": "warn",
    "category": "tools",
    "why": "Interpreter/runtime bins are allowed without explicit safe profiles.",
    "primary_fix": "tools.exec.safeBins, tools.exec.safeBinProfiles, agents.list[].tools.exec.*",
    "auto_fix": false
  },
  "skills.workspace.symlink_escape": {
    "severity": "warn",
    "category": "skills",
    "why": "A skill path in the workspace resolves outside the workspace root.",
    "primary_fix": "workspace skills/** filesystem state",
    "auto_fix": false
  },
  "security.exposure.open_groups_with_elevated": {
    "severity": "critical",
    "category": "exposure",
    "why": "Open groups plus elevated tools create a high-impact prompt-injection path.",
    "primary_fix": "channels.*.groupPolicy, tools.elevated.*",
    "auto_fix": false
  },
  "security.exposure.open_groups_with_runtime_or_fs": {
    "severity": "critical/warn",
    "category": "exposure",
    "why": "Open groups can reach command or file tools without sandbox/workspace guards.",
    "primary_fix": "channels.*.groupPolicy, tools.profile/deny, tools.fs.workspaceOnly, agents.*.sandbox.mode",
    "auto_fix": false
  },
  "security.trust_model.multi_user_heuristic": {
    "severity": "warn",
    "category": "trust-model",
    "why": "The config looks multi-user without matching isolation and tool hardening.",
    "primary_fix": "sandbox.mode, tool deny, workspace scoping",
    "auto_fix": false
  },
  "tools.profile_minimal_overridden": {
    "severity": "warn",
    "category": "tools",
    "why": "Per-agent overrides bypass a minimal global profile.",
    "primary_fix": "agents.list[].tools.profile",
    "auto_fix": false
  },
  "plugins.tools_reachable_permissive_policy": {
    "severity": "warn",
    "category": "plugins",
    "why": "Extension tools are reachable in permissive contexts.",
    "primary_fix": "tools.profile plus tool allow/deny",
    "auto_fix": false
  },
  "models.small_params": {
    "severity": "critical/info",
    "category": "model",
    "why": "Small or weak models plus tool access increase prompt-injection risk.",
    "primary_fix": "model choice plus sandbox/tool policy",
    "auto_fix": false
  }
}

#!/usr/bin/env bash
set -euo pipefail

usage() {
  cat <<'EOF'
Collect mostly read-only OpenClaw security diagnostics.

Usage:
  collect_openclaw_audit.sh --out DIR

The script writes a timestamped folder inside DIR:
  DIR/openclaw-audit-<UTC timestamp>/

Safety:
- Does not run any --fix operations.
- Avoids copying credential files.
- Collects shareable CLI diagnostics plus basic host/network metadata.
- Review outputs before sharing externally.

Examples:
  bash "{baseDir}/scripts/collect_openclaw_audit.sh" --out ./openclaw-audit
EOF
}

OUT_DIR=""

while [[ $# -gt 0 ]]; do
  case "$1" in
    --out)
      OUT_DIR="${2:-}"
      shift 2
      ;;
    -h|--help)
      usage
      exit 0
      ;;
    *)
      echo "Unknown arg: $1" >&2
      usage >&2
      exit 2
      ;;
  esac
done

if [[ -z "${OUT_DIR}" ]]; then
  echo "Missing --out <dir>" >&2
  usage >&2
  exit 2
fi

TS="$(date -u +"%Y%m%dT%H%M%SZ")"
ROOT="${OUT_DIR%/}/openclaw-audit-${TS}"
mkdir -p "${ROOT}"

log() { echo "[collect] $*"; }

write_note() {
  local name="$1"; shift
  printf '%s
' "$*" > "${ROOT}/${name}.txt"
}

run_cmd() {
  local name="$1"; shift
  local file="${ROOT}/${name}.txt"
  log "Running: $*"
  {
    echo "$ $*"
    "$@"
  } > "${file}" 2>&1 || {
    echo "[warn] command failed (continuing): $*" >> "${file}"
    return 0
  }
}

run_cmd_maybe_sudo() {
  local name="$1"; shift
  local file="${ROOT}/${name}.txt"
  if command -v sudo >/dev/null 2>&1; then
    log "Running (sudo -n): $*"
    {
      echo "$ sudo -n $*"
      sudo -n "$@"
    } > "${file}" 2>&1 || {
      echo "[info] sudo not available without password (skipped): $*" >> "${file}"
      return 0
    }
  else
    echo "[info] sudo not installed; skipped: $*" > "${file}"
  fi
}

# Host basics
run_cmd "host_whoami" whoami
run_cmd "host_uname" uname -a

if command -v sw_vers >/dev/null 2>&1; then
  run_cmd "host_sw_vers" sw_vers
  if [[ -x /usr/libexec/ApplicationFirewall/socketfilterfw ]]; then
    run_cmd "macos_firewall_state" /usr/libexec/ApplicationFirewall/socketfilterfw --getglobalstate
    run_cmd "macos_firewall_stealth" /usr/libexec/ApplicationFirewall/socketfilterfw --getstealthmode
  fi
  run_cmd "macos_filevault" fdesetup status
fi

if [[ -f /etc/os-release ]]; then
  run_cmd "host_os_release" cat /etc/os-release
fi

# Network listeners
if command -v lsof >/dev/null 2>&1; then
  run_cmd "net_lsof_listen" lsof -nP -iTCP -sTCP:LISTEN
elif command -v ss >/dev/null 2>&1; then
  run_cmd "net_ss_listen" ss -ltnp
elif command -v netstat >/dev/null 2>&1; then
  run_cmd "net_netstat_listen" netstat -anv
fi

# Linux firewall snapshot (best-effort)
if [[ -f /etc/os-release ]]; then
  run_cmd_maybe_sudo "linux_ufw_status" ufw status verbose
  run_cmd_maybe_sudo "linux_nft_ruleset" nft list ruleset
  run_cmd_maybe_sudo "linux_iptables_rules" iptables -S
fi

# Docker / container clues
if command -v docker >/dev/null 2>&1; then
  run_cmd "docker_ps" docker ps --format 'table {{.Names}}	{{.Image}}	{{.Ports}}'
  run_cmd "docker_compose_ps" docker compose ps
  run_cmd "docker_port_openclaw_gateway_18789" docker port openclaw-gateway 18789
fi

if command -v podman >/dev/null 2>&1; then
  run_cmd "podman_ps" podman ps --format 'table {{.Names}}	{{.Image}}	{{.Ports}}'
fi

if ! command -v openclaw >/dev/null 2>&1; then
  write_note "openclaw_missing" "openclaw not found on PATH; collected host-level data only."
  (
    cd "${ROOT}"
    find . -maxdepth 1 -type f -printf '%f\n' | sort > manifest.txt
  )
  log "openclaw missing; collected host info only"
  exit 0
fi

# OpenClaw core diagnostics
run_cmd "openclaw_version" openclaw --version
run_cmd "openclaw_status_all" openclaw status --all
run_cmd "openclaw_status_deep" openclaw status --deep
run_cmd "openclaw_doctor" openclaw doctor
run_cmd "openclaw_gateway_status" openclaw gateway status
run_cmd "openclaw_gateway_probe_json" openclaw gateway probe --json
run_cmd "openclaw_channels_status_probe" openclaw channels status --probe
run_cmd "openclaw_health_json" openclaw health --json
run_cmd "openclaw_security_audit_json" openclaw security audit --json
run_cmd "openclaw_security_audit_deep_json" openclaw security audit --deep --json

# Backup readiness (read-only dry runs)
run_cmd "openclaw_backup_create_dry_run_json" openclaw backup create --dry-run --json
run_cmd "openclaw_backup_only_config_dry_run_json" openclaw backup create --only-config --dry-run --json

# Safe targeted config reads
run_cmd "openclaw_config_gateway_bind" openclaw config get gateway.bind
run_cmd "openclaw_config_gateway_auth_mode" openclaw config get gateway.auth.mode
run_cmd "openclaw_config_gateway_auth_allow_tailscale" openclaw config get gateway.auth.allowTailscale
run_cmd "openclaw_config_gateway_controlui_allowed_origins" openclaw config get gateway.controlUi.allowedOrigins
run_cmd "openclaw_config_gateway_trusted_proxies" openclaw config get gateway.trustedProxies
run_cmd "openclaw_config_gateway_allow_real_ip_fallback" openclaw config get gateway.allowRealIpFallback
run_cmd "openclaw_config_discovery_mdns_mode" openclaw config get discovery.mdns.mode
run_cmd "openclaw_config_session_dm_scope" openclaw config get session.dmScope
run_cmd "openclaw_config_tools_profile" openclaw config get tools.profile
run_cmd "openclaw_config_tools_fs_workspace_only" openclaw config get tools.fs.workspaceOnly
run_cmd "openclaw_config_tools_exec_security" openclaw config get tools.exec.security
run_cmd "openclaw_config_tools_elevated_enabled" openclaw config get tools.elevated.enabled
run_cmd "openclaw_config_channels_defaults_dm_policy" openclaw config get channels.defaults.dmPolicy
run_cmd "openclaw_config_channels_defaults_group_policy" openclaw config get channels.defaults.groupPolicy
run_cmd "openclaw_config_logging_redact_sensitive" openclaw config get logging.redactSensitive

# Supply-chain visibility
run_cmd "openclaw_skills_eligible_json" openclaw skills list --eligible --json
run_cmd "openclaw_plugins_list_json" openclaw plugins list --json

# State directory metadata (do not copy secrets)
STATE_DIR="${OPENCLAW_STATE_DIR:-$HOME/.openclaw}"
if [[ -d "${STATE_DIR}" ]]; then
  run_cmd "openclaw_state_ls" ls -la "${STATE_DIR}"
  if command -v stat >/dev/null 2>&1; then
    run_cmd "openclaw_state_stat" stat "${STATE_DIR}" "${STATE_DIR}/openclaw.json"
  fi
fi

(
  cd "${ROOT}"
  find . -maxdepth 1 -type f -printf '%f
' | sort > manifest.txt
)

log "Done. Output: ${ROOT}"

#!/usr/bin/env python3
"""Redact an OpenClaw config file for safer sharing.

The active OpenClaw config is often JSON5-like (`~/.openclaw/openclaw.json` by
default), so this script tries strict JSON first, then an optional JSON5 parser,
then falls back to regex-based redaction on the raw text.

What it redacts (best-effort):
- Secret-like keys: token, password, secret, api key, cookie, session key, etc.
- String values that strongly resemble long secrets or JWT-style tokens.
- Query-string secrets in URLs such as ?token=... or ?access_token=...

Examples:
  python3 "{baseDir}/scripts/redact_openclaw_config.py" ~/.openclaw/openclaw.json > openclaw.json.redacted
  cat ~/.openclaw/openclaw.json | python3 "{baseDir}/scripts/redact_openclaw_config.py" - > openclaw.json.redacted

Always review the redacted output before sharing it.
"""

from __future__ import annotations

import argparse
import json
import re
import sys
from typing import Any, Callable, Optional

SENSITIVE_KEY_RE = re.compile(
    r"(token|password|secret|api[_-]?key|apikey|client[_-]?secret|private[_-]?key|session[_-]?key|cookie|bearer|access[_-]?token|refresh[_-]?token)\b",
    re.IGNORECASE,
)

JWT_LIKE_RE = re.compile(r"^[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+$")
HEXISH_RE = re.compile(r"^[A-Fa-f0-9]{32,}$")
ALNUMISH_RE = re.compile(r"^[A-Za-z0-9_\-]{24,}$")

URL_QS_SECRET_RE = re.compile(
    r"(?P<prefix>[?&](?:token|password|api[_-]?key|apikey|key|access_token|refresh_token|session[_-]?key)=)(?P<val>[^&\s#]+)",
    re.IGNORECASE,
)


def mask(s: str) -> str:
    s = s or ""
    if len(s) <= 8:
        return "***"
    return f"{s[:4]}…{s[-4:]}"


def looks_secret(s: str) -> bool:
    if len(s) < 24:
        return False
    if s.startswith(("http://", "https://", "/", "./", "../", "~/")):
        return False
    return bool(JWT_LIKE_RE.match(s) or HEXISH_RE.match(s) or ALNUMISH_RE.match(s))


def redact_string(s: str) -> str:
    if looks_secret(s):
        return mask(s)
    return URL_QS_SECRET_RE.sub(lambda m: m.group("prefix") + mask(m.group("val")), s)


def redact_obj(obj: Any) -> Any:
    if isinstance(obj, dict):
        out: dict[str, Any] = {}
        for key, value in obj.items():
            skey = str(key)
            if SENSITIVE_KEY_RE.search(skey):
                if isinstance(value, str):
                    out[skey] = redact_string(value)
                else:
                    out[skey] = "***"
            else:
                out[skey] = redact_obj(value)
        return out

    if isinstance(obj, list):
        return [redact_obj(item) for item in obj]

    if isinstance(obj, str):
        return redact_string(obj)

    return obj


def try_json5_parser() -> Optional[Callable[[str], Any]]:
    for module_name in ("json5", "pyjson5"):
        try:
            module = __import__(module_name)
            loads = getattr(module, "loads", None)
            if callable(loads):
                return loads  # type: ignore[return-value]
        except Exception:
            continue
    return None


def text_fallback_redact(raw: str) -> str:
    redacted = URL_QS_SECRET_RE.sub(lambda m: m.group("prefix") + mask(m.group("val")), raw)

    quoted_kv_re = re.compile(
        r"(?P<key>\b[\w.-]*(?:token|password|secret|api[_-]?key|apikey|client[_-]?secret|private[_-]?key|session[_-]?key|cookie|bearer|access[_-]?token|refresh[_-]?token)[\w.-]*\b)"
        r"(?P<ws>\s*:\s*)"
        r"(?P<val>(?:\"[^\"]*\"|'[^']*'))",
        re.IGNORECASE,
    )

    def repl_quoted(match: re.Match[str]) -> str:
        key = match.group("key")
        ws = match.group("ws")
        val = match.group("val")
        if val.startswith('"'):
            inner = val[1:-1]
            return f'{key}{ws}"{mask(inner)}"'
        inner = val[1:-1]
        return f"{key}{ws}'{mask(inner)}'"

    redacted = quoted_kv_re.sub(repl_quoted, redacted)

    bare_kv_re = re.compile(
        r"(?P<key>\b[\w.-]*(?:token|password|secret|api[_-]?key|apikey|client[_-]?secret|private[_-]?key|session[_-]?key|cookie|bearer|access[_-]?token|refresh[_-]?token)[\w.-]*\b)"
        r"(?P<ws>\s*:\s*)"
        r"(?P<val>[A-Za-z0-9._-]{12,})",
        re.IGNORECASE,
    )

    def repl_bare(match: re.Match[str]) -> str:
        value = match.group("val")
        redacted_value = mask(value) if looks_secret(value) else value
        return f"{match.group('key')}{match.group('ws')}{redacted_value}"

    return bare_kv_re.sub(repl_bare, redacted)


def load_raw(path: str) -> str:
    if path == "-":
        return sys.stdin.read()
    with open(path, "r", encoding="utf-8", errors="replace") as handle:
        return handle.read()


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Redact an OpenClaw config file before sharing it.",
        epilog="The tool prefers structured JSON/JSON5 redaction when possible and falls back to raw-text redaction otherwise.",
    )
    parser.add_argument(
        "path",
        nargs="?",
        default="-",
        help="Path to the config file, or '-' to read from stdin.",
    )
    parser.add_argument(
        "--text-fallback-only",
        action="store_true",
        help="Skip JSON/JSON5 parsing and redact the raw text directly.",
    )
    args = parser.parse_args()

    raw = load_raw(args.path)

    if args.text_fallback_only:
        sys.stdout.write(text_fallback_redact(raw))
        if not raw.endswith("\n"):
            sys.stdout.write("\n")
        return

    parsed = False
    obj: Any = None

    try:
        obj = json.loads(raw)
        parsed = True
    except Exception:
        parsed = False

    if not parsed:
        json5_loads = try_json5_parser()
        if json5_loads is not None:
            try:
                obj = json5_loads(raw)
                parsed = True
            except Exception:
                parsed = False

    if parsed:
        json.dump(redact_obj(obj), sys.stdout, indent=2, ensure_ascii=False)
        sys.stdout.write("\n")
        return

    sys.stdout.write(text_fallback_redact(raw))
    if not raw.endswith("\n"):
        sys.stdout.write("\n")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Render a Markdown report from collected OpenClaw audit artefacts.

The renderer is intentionally tolerant:
- reads capture files produced by collect_openclaw_audit.sh
- extracts JSON from CLI captures when possible
- merges audit findings from regular + deep runs
- enriches findings with assets/openclaw_checkid_map.json
- fills in environment clues from targeted config-get captures

Usage:
  python3 "{baseDir}/scripts/render_report.py" --input ./openclaw-audit --output ./openclaw-security-report.md
"""

from __future__ import annotations

import argparse
import json
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple

SEV_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4, "unknown": 9}
ANSI_RE = re.compile(r"\x1b\[[0-9;]*[A-Za-z]")


@dataclass
class Capture:
    name: str
    path: Path
    raw: str
    body: str


def strip_cmd_banner(text: str) -> str:
    text = ANSI_RE.sub("", text)
    lines = text.splitlines()
    if lines and lines[0].startswith("$ "):
        lines = lines[1:]
    return "\n".join(lines).strip()


def load_capture(path: Path) -> Optional[Capture]:
    if not path.exists():
        return None
    raw = path.read_text(encoding="utf-8", errors="replace")
    return Capture(name=path.stem, path=path, raw=raw, body=strip_cmd_banner(raw))


def newest_audit_folder(root: Path) -> Path:
    if root.is_dir():
        kids = [p for p in root.iterdir() if p.is_dir() and p.name.startswith("openclaw-audit-")]
        if kids:
            return sorted(kids, key=lambda p: p.name)[-1]
    return root


def clean_scalar_text(text: str) -> str:
    lines: List[str] = []
    for line in text.splitlines():
        if line.startswith("[warn]") or line.startswith("[info]"):
            continue
        lines.append(line.rstrip())
    return "\n".join(lines).strip()


def load_json_text(txt: str) -> Any:
    body = clean_scalar_text(txt)
    candidate_starts = [i for ch in ("{", "[") for i in [body.find(ch)] if i != -1]
    for idx in sorted(candidate_starts):
        try:
            return json.loads(body[idx:])
        except Exception:
            continue
    return None


def load_json_capture(path: Path) -> Any:
    capture = load_capture(path)
    if capture is None:
        return None
    return load_json_text(capture.body)


def load_json_file(path: Path) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return None


def normalise_sev(raw: str) -> str:
    s = (raw or "").strip().lower()
    if not s:
        return "unknown"
    if "/" in s:
        parts = [p.strip() for p in s.split("/") if p.strip()]
        if "critical" in parts:
            return "critical"
        if "high" in parts:
            return "high"
        if "warn" in parts or "warning" in parts or "medium" in parts:
            return "medium"
        if "low" in parts:
            return "low"
        if "info" in parts:
            return "info"
        return parts[0] if parts else "unknown"
    if s in ("warn", "warning"):
        return "medium"
    if s == "crit":
        return "critical"
    return s if s in SEV_ORDER else s


def extract_findings(obj: Any) -> List[Dict[str, Any]]:
    if obj is None:
        return []
    if isinstance(obj, dict):
        for key in ("findings", "checks", "results", "issues"):
            value = obj.get(key)
            if isinstance(value, list):
                return [item for item in value if isinstance(item, dict)]
        if obj and all(isinstance(v, dict) for v in obj.values()):
            out: List[Dict[str, Any]] = []
            for key, value in obj.items():
                item = dict(value)
                item.setdefault("checkId", key)
                out.append(item)
            return out
    if isinstance(obj, list):
        return [item for item in obj if isinstance(item, dict)]
    return []


def pick_text(finding: Dict[str, Any]) -> str:
    for key in ("title", "summary", "message", "description"):
        value = finding.get(key)
        if isinstance(value, str) and value.strip():
            return value.strip()
    return "(no summary provided)"


def dedupe_findings(findings: Iterable[Dict[str, Any]]) -> List[Dict[str, Any]]:
    seen: set[Tuple[str, str, str]] = set()
    out: List[Dict[str, Any]] = []
    for finding in findings:
        check_id = str(finding.get("checkId") or finding.get("id") or "").strip()
        summary = pick_text(finding)
        sev = normalise_sev(str(finding.get("severity") or finding.get("level") or finding.get("risk") or ""))
        key = (check_id, summary, sev)
        if key in seen:
            continue
        seen.add(key)
        out.append(finding)
    return out


def sort_key(finding: Dict[str, Any]) -> Tuple[int, str, str]:
    sev = normalise_sev(str(finding.get("severity") or finding.get("level") or finding.get("risk") or "unknown"))
    order = SEV_ORDER.get(sev, 8)
    check_id = str(finding.get("checkId") or finding.get("id") or "")
    return (order, check_id, pick_text(finding))


def severity_counts(findings: List[Dict[str, Any]]) -> Dict[str, int]:
    counts: Dict[str, int] = {}
    for finding in findings:
        sev = normalise_sev(str(finding.get("severity") or finding.get("level") or finding.get("risk") or "unknown"))
        counts[sev] = counts.get(sev, 0) + 1
    return counts


def load_checkid_map(skill_root: Path) -> Dict[str, Any]:
    path = skill_root / "assets" / "openclaw_checkid_map.json"
    if path.exists():
        obj = load_json_file(path)
        if isinstance(obj, dict):
            return obj
    return {}


def read_capture_body(in_dir: Path, stem: str) -> str:
    capture = load_capture(in_dir / f"{stem}.txt")
    if capture is None:
        return ""
    return clean_scalar_text(capture.body)


def read_config_value(in_dir: Path, stem: str) -> Optional[Any]:
    body = read_capture_body(in_dir, stem)
    if not body:
        return None
    try:
        return json.loads(body)
    except Exception:
        lowered = body.lower()
        if lowered == "true":
            return True
        if lowered == "false":
            return False
        if lowered == "null":
            return None
        return body


def one_line(value: Any) -> str:
    if value is None:
        return "(unset)"
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, str):
        value = value.strip()
        return value or "(empty)"
    return json.dumps(value, ensure_ascii=False, sort_keys=True)


def detect_os(in_dir: Path) -> str:
    sw_vers = read_capture_body(in_dir, "host_sw_vers")
    if sw_vers:
        product_name = ""
        product_version = ""
        for line in sw_vers.splitlines():
            if ":" not in line:
                continue
            key, value = [part.strip() for part in line.split(":", 1)]
            if key == "ProductName":
                product_name = value
            elif key == "ProductVersion":
                product_version = value
        if product_name or product_version:
            return " ".join(part for part in (product_name, product_version) if part).strip()

    os_release = read_capture_body(in_dir, "host_os_release")
    if os_release:
        for line in os_release.splitlines():
            if line.startswith("PRETTY_NAME="):
                return line.split("=", 1)[1].strip().strip('"')
        for line in os_release.splitlines():
            if line.startswith("NAME="):
                return line.split("=", 1)[1].strip().strip('"')

    uname = read_capture_body(in_dir, "host_uname")
    return uname or "(unknown)"


def parse_version(in_dir: Path) -> str:
    body = read_capture_body(in_dir, "openclaw_version")
    if not body:
        return "(unknown)"
    lines = [line.strip() for line in body.splitlines() if line.strip()]
    return lines[-1] if lines else "(unknown)"


def parse_gateway_runtime(in_dir: Path) -> str:
    body = read_capture_body(in_dir, "openclaw_gateway_status")
    if not body:
        return "(not collected)"
    runtime = ""
    rpc = ""
    for line in body.splitlines():
        line = line.strip()
        if line.lower().startswith("runtime:"):
            runtime = line.split(":", 1)[1].strip()
        elif line.lower().startswith("rpc probe:"):
            rpc = line.split(":", 1)[1].strip()
    if runtime and rpc:
        return f"Runtime: {runtime}; RPC probe: {rpc}"
    if runtime:
        return f"Runtime: {runtime}"
    return body.splitlines()[0].strip()


def build_environment_rows(in_dir: Path) -> List[Tuple[str, str]]:
    rows = [
        ("OS", detect_os(in_dir)),
        ("OpenClaw version", parse_version(in_dir)),
        ("Gateway status", parse_gateway_runtime(in_dir)),
        ("Gateway bind", one_line(read_config_value(in_dir, "openclaw_config_gateway_bind"))),
        ("Gateway auth mode", one_line(read_config_value(in_dir, "openclaw_config_gateway_auth_mode"))),
        ("Gateway auth.allowTailscale", one_line(read_config_value(in_dir, "openclaw_config_gateway_auth_allow_tailscale"))),
        ("Control UI allowed origins", one_line(read_config_value(in_dir, "openclaw_config_gateway_controlui_allowed_origins"))),
        ("Trusted proxies", one_line(read_config_value(in_dir, "openclaw_config_gateway_trusted_proxies"))),
        ("allowRealIpFallback", one_line(read_config_value(in_dir, "openclaw_config_gateway_allow_real_ip_fallback"))),
        ("Discovery mDNS mode", one_line(read_config_value(in_dir, "openclaw_config_discovery_mdns_mode"))),
        ("Session dmScope", one_line(read_config_value(in_dir, "openclaw_config_session_dm_scope"))),
        ("Default DM policy", one_line(read_config_value(in_dir, "openclaw_config_channels_defaults_dm_policy"))),
        ("Default group policy", one_line(read_config_value(in_dir, "openclaw_config_channels_defaults_group_policy"))),
        ("Tools profile", one_line(read_config_value(in_dir, "openclaw_config_tools_profile"))),
        ("FS workspaceOnly", one_line(read_config_value(in_dir, "openclaw_config_tools_fs_workspace_only"))),
        ("Exec security", one_line(read_config_value(in_dir, "openclaw_config_tools_exec_security"))),
        ("Elevated tools enabled", one_line(read_config_value(in_dir, "openclaw_config_tools_elevated_enabled"))),
        ("Logging redactSensitive", one_line(read_config_value(in_dir, "openclaw_config_logging_redact_sensitive"))),
    ]
    docker_ps = read_capture_body(in_dir, "docker_ps")
    if docker_ps:
        rows.append(("Docker present", "yes"))
    return rows


def overall_risk(counts: Dict[str, int]) -> str:
    if counts.get("critical"):
        return "Critical"
    if counts.get("high"):
        return "High"
    if counts.get("medium"):
        return "Moderate"
    if counts.get("low"):
        return "Low"
    if counts.get("info") or counts.get("unknown"):
        return "Informational"
    return "No parsed findings"


def likely_verification_for(check_id: str) -> str:
    if check_id.startswith("gateway.") or check_id.startswith("security.exposure.") or check_id.startswith("discovery."):
        return "`openclaw security audit --deep --json`, `openclaw gateway probe --json`, `openclaw channels status --probe`"
    if check_id.startswith("fs.") or check_id.startswith("logging."):
        return "`openclaw security audit --json` and re-check filesystem permissions"
    if check_id.startswith("sandbox.") or check_id.startswith("tools."):
        return "`openclaw security audit --deep --json` plus targeted `openclaw config get ...`"
    return "`openclaw security audit --deep --json`"


def markdown_escape(text: str) -> str:
    return text.replace("|", "\\|").replace("\n", " ")


def main() -> None:
    parser = argparse.ArgumentParser(description="Render a Markdown report from OpenClaw audit artefacts.")
    parser.add_argument("--input", required=True, help="Audit folder or parent folder produced by collect_openclaw_audit.sh")
    parser.add_argument("--output", required=True, help="Output Markdown file")
    args = parser.parse_args()

    in_path = Path(args.input)
    in_dir = newest_audit_folder(in_path)
    out_path = Path(args.output)

    if not in_dir.exists() or not in_dir.is_dir():
        raise SystemExit(f"Input directory not found: {in_dir}")

    skill_root = Path(__file__).resolve().parents[1]
    check_map = load_checkid_map(skill_root)

    audit_sources = [
        in_dir / "openclaw_security_audit_deep_json.txt",
        in_dir / "openclaw_security_audit_json.txt",
    ]
    findings: List[Dict[str, Any]] = []
    parsed_files: List[str] = []
    for path in audit_sources:
        obj = load_json_capture(path)
        if obj is not None:
            parsed_files.append(path.name)
            findings.extend(extract_findings(obj))

    findings = dedupe_findings(findings)
    findings_sorted = sorted(findings, key=sort_key)
    counts = severity_counts(findings_sorted)
    env_rows = build_environment_rows(in_dir)

    report: List[str] = []
    report.append("# OpenClaw Security Audit Report\n\n")
    report.append("## Executive summary\n\n")
    report.append(f"- **Overall risk rating:** {overall_risk(counts)}\n")
    report.append(f"- **OpenClaw version:** {parse_version(in_dir)}\n")
    report.append(f"- **Audit artefacts folder:** `{in_dir}`\n")
    if parsed_files:
        report.append(f"- **Parsed audit files:** {', '.join(f'`{name}`' for name in parsed_files)}\n")
    else:
        report.append("- **Parsed audit files:** none (review raw artefacts manually)\n")

    if findings_sorted:
        report.append("- **Most urgent issues:**\n")
        for finding in findings_sorted[:3]:
            check_id = str(finding.get("checkId") or finding.get("id") or "(unknown)")
            summary = pick_text(finding)
            sev = normalise_sev(str(finding.get("severity") or finding.get("level") or finding.get("risk") or "unknown"))
            report.append(f"  - `{check_id}` ({sev}): {summary}\n")
    else:
        report.append("- **Most urgent issues:** none parsed automatically; inspect raw audit outputs.\n")
    report.append("\n")

    report.append("## Environment overview\n\n")
    for key, value in env_rows:
        report.append(f"- **{key}:** {value}\n")
    report.append("\n")

    report.append("## Finding counts\n\n")
    if counts:
        for sev in ("critical", "high", "medium", "low", "info", "unknown"):
            if sev in counts:
                report.append(f"- **{sev}:** {counts[sev]}\n")
    else:
        report.append("- No findings parsed.\n")
    report.append("\n")

    report.append("## Findings\n\n")
    if not findings_sorted:
        report.append("No findings were parsed from the collected JSON. Review these files manually:\n\n")
        for path in audit_sources:
            report.append(f"- `{path.name}`\n")
        report.append("\n")
    else:
        report.append("| Severity | Check ID | Summary | Why it matters | Likely fix area | Auto-fix | Verification |\n")
        report.append("|---|---|---|---|---|---:|---|\n")
        for finding in findings_sorted[:400]:
            sev = normalise_sev(str(finding.get("severity") or finding.get("level") or finding.get("risk") or "unknown"))
            check_id = str(finding.get("checkId") or finding.get("id") or "").strip() or "(unknown)"
            summary = markdown_escape(pick_text(finding))
            enrich = check_map.get(check_id, {}) if isinstance(check_map, dict) else {}

            why = ""
            likely_fix = ""
            auto_fix = ""

            if isinstance(enrich, dict):
                why = str(enrich.get("why") or "")
                likely_fix = str(enrich.get("primary_fix") or "")
                if "auto_fix" in enrich:
                    auto_fix = "yes" if enrich.get("auto_fix") else "no"

            for key in ("why", "reason"):
                value = finding.get(key)
                if isinstance(value, str) and value.strip():
                    why = value.strip()
                    break

            for key in ("primaryFix", "fix", "path", "configPath"):
                value = finding.get(key)
                if isinstance(value, str) and value.strip():
                    likely_fix = value.strip()
                    break
                if isinstance(value, dict):
                    maybe_path = value.get("path")
                    if isinstance(maybe_path, str) and maybe_path.strip():
                        likely_fix = maybe_path.strip()
                        break

            if isinstance(finding.get("autoFix"), bool):
                auto_fix = "yes" if finding.get("autoFix") else "no"

            report.append(
                "| {sev} | {check_id} | {summary} | {why} | {fix} | {auto_fix} | {verify} |\n".format(
                    sev=sev,
                    check_id=markdown_escape(check_id),
                    summary=summary,
                    why=markdown_escape(why or "(see finding detail)"),
                    fix=markdown_escape(likely_fix or "(see config / finding detail)"),
                    auto_fix=auto_fix or "",
                    verify=markdown_escape(likely_verification_for(check_id)),
                )
            )
        report.append("\n")

    report.append("## Remediation plan\n\n")
    report.append("### Phase 1 — Stop the bleeding (same day)\n\n")
    report.append("1. Fix any public exposure, missing auth, or open groups combined with dangerous tools.\n")
    report.append("2. Back up the current state before invasive changes: `openclaw backup create --verify`.\n")
    report.append("3. Remove insecure Control UI / reverse-proxy settings and lock down DM/group access.\n\n")

    report.append("### Phase 2 — Reduce blast radius (this week)\n\n")
    report.append("1. Move inbox-facing agents to a conservative tool profile and deny unnecessary groups.\n")
    report.append("2. Tighten workspace, exec, sandbox, node, and plugin trust settings.\n")
    report.append("3. Review writable paths, local skill/plugin sources, transcript retention, and log redaction.\n\n")

    report.append("### Phase 3 — Operationalise (ongoing)\n\n")
    report.append("- Keep OpenClaw and the host OS updated.\n")
    report.append("- Rotate gateway credentials after any suspected leakage.\n")
    report.append("- Re-run `openclaw security audit --deep --json` after major config changes.\n")
    report.append("- Keep an intentional backup/restore routine for config, state, and workspace.\n\n")

    report.append("## Key raw artefacts\n\n")
    for stem in (
        "openclaw_status_all",
        "openclaw_status_deep",
        "openclaw_gateway_status",
        "openclaw_gateway_probe_json",
        "openclaw_channels_status_probe",
        "openclaw_security_audit_json",
        "openclaw_security_audit_deep_json",
        "openclaw_health_json",
    ):
        path = in_dir / f"{stem}.txt"
        if path.exists():
            report.append(f"- `{path.name}`\n")
    report.append("\n")

    report.append("## Residual risk notes\n\n")
    report.append(
        "Even after hardening, any agent that can read untrusted content and call tools carries prompt-injection and social-engineering risk. "
        "Document which surfaces are intentionally open, which tools remain enabled, and what the recovery path is if the Gateway host or credentials are compromised.\n"
    )

    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text("".join(report), encoding="utf-8")


if __name__ == "__main__":
    main()