@clawhub-trent-ai-release-2bdf428e5b
Assess your Agent deployment against security risks using Trent.
---
name: trent-openclaw-security
description: Assess your Agent deployment against security risks using Trent.
version: 1.2.0
homepage: https://trent.ai
user-invocable: true
tags:
- trentai
- trent
- trentclaw
- security
- assessment
- threat-modeling
metadata:
openclaw:
requires:
env:
- TRENT_API_KEY
optionalEnv:
- TRENT_CHAT_API_URL
- TRENT_AGENT_API_URL
- OPENCLAW_WORKSPACE
primaryEnv: TRENT_API_KEY
---
# Trent OpenClaw Security Assessment
Audit your OpenClaw deployment for security risks. Identifies misconfigurations,
chained attack paths, and provides severity-rated findings with fixes.
## Setup
All tools are bundled — no external installer needed.
Set the `TRENT_API_KEY` environment variable. Get a key at https://trent.ai/openclaw/
## Instructions
This audit runs in three phases. Run them in order.
### Phase 1 — Configuration Audit
Collect metadata and send to Trent for analysis:
```python
from openclaw_trent.openclaw_config.collector import collect_openclaw_metadata
from openclaw_trent.lib.audit_prompt import build_audit_prompt
from openclaw_trent.lib import trent_client
metadata = collect_openclaw_metadata()
message = build_audit_prompt(metadata)
response = trent_client.chat(message=message)
```
Save `response["thread_id"]` for Phase 3.
Present findings grouped by severity (see "Present results" below).
Summarize: "Phase 1 complete. N findings from configuration analysis.
Phase 2 will scan your skills for deeper analysis — I'll show you exactly
what would be uploaded before anything is sent. Ready to continue?"
Optional: specify a custom config path:
```python
from pathlib import Path
metadata = collect_openclaw_metadata(openclaw_path=Path("/path/to/openclaw/config"))
```
### Phase 2 — Skill Upload
Scan the workspace first (nothing is uploaded yet):
```python
from openclaw_trent.lib.package_skills import scan_workspace
skills = scan_workspace()
```
Present what was found and how it will be protected. Example:
> I found N skills in your workspace:
>
> | Skill | Type | Size |
> |---|---|---|
> | skill-name | installed-skill | 12KB |
>
> Before upload, each skill is packaged with its source code and metadata
> (name, version, dependencies). Files like .env, .pem, .key, and .db are
> excluded, and secrets in standard formats (API keys, tokens, AWS credentials,
> connection strings) are automatically redacted locally. If you use custom
> secret formats, keep them in environment variables rather than hard-coded
> in skill files.
>
> Ready to upload?
Use the `secrets_redacted` field — if any skills had secrets redacted,
mention which ones in the table or below it.
**Wait for the user to confirm before uploading.**
After user confirms, upload:
```python
from openclaw_trent.lib.upload_skills import upload_packaged_skills
upload_summary = upload_packaged_skills(skills)
```
Present the upload summary:
- How many skills were uploaded, skipped (unchanged), failed, or too large
- List each skill by name and status
If all uploads failed, report the errors and stop. Otherwise proceed.
Summarize: "Phase 2 complete. N skills uploaded. Proceeding to deep skill analysis..."
### Phase 3 — Deep Skill Analysis
Analyse each uploaded skill using the thread ID from Phase 1:
```python
from openclaw_trent.lib.prompts import build_per_skill_analysis_prompt
from openclaw_trent.lib import trent_client
thread_id = "<THREAD_ID from Phase 1>"
for skill in upload_summary["skills"]:
if skill["status"] in ("uploaded", "skipped"):
prompt = build_per_skill_analysis_prompt(skill)
result = trent_client.chat(message=prompt, thread_id=thread_id)
```
Each request uses the Phase 1 thread ID so the advisor has full
context from the configuration audit.
Present the deep analysis results alongside the Phase 1 findings.
### Inspect system context separately
To view the system analysis data without running a full audit:
```python
from openclaw_trent.lib.system_analyzer import collect_system_analysis
import json
result = collect_system_analysis()
print(json.dumps(result, indent=2))
```
This returns channel configuration and installed skill names.
Useful for debugging or verifying what data is sent.
### Present results
Format findings grouped by severity:
- **CRITICAL**: Immediate action required
- **HIGH**: Fix soon
- **MEDIUM**: Recommended improvement
- **LOW**: Minor hardening
For each finding show: the risk, where it was found, and the exact fix.
Highlight **chained attack paths** — where multiple settings combine to create worse outcomes.
Present recommended config changes as a diff snippet for the user to review
and apply manually. Do **not** modify any system files directly.
## When to use
- User asks "Is my setup secure?" or "audit my config"
- After changes to OpenClaw configuration, new plugins, or new MCP servers
FILE:scripts/openclaw_trent/__init__.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""OpenClaw security audit plugin powered by Trent AppSec Advisor."""
__version__ = "1.1.1"
FILE:scripts/openclaw_trent/lib/__init__.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Self-contained library modules bundled with the openclaw-trent skill.
All modules in this package use only Python stdlib — no third-party deps required.
"""
FILE:scripts/openclaw_trent/lib/audit_prompt.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Builds the structured prompt for OpenClaw security audits."""
import json
def build_audit_prompt(metadata: dict, system_info: dict | None = None) -> str:
"""Build the structured prompt for the chat endpoint.
The prompt provides metadata with clear markers. The backend chat agent
detects the markers and delegates to a specialized OpenClaw audit sub-agent
that has access to the OpenClaw security knowledge base for trust-model-aware,
false-positive-filtered analysis.
Args:
metadata: OpenClaw configuration metadata from collector.
system_info: Optional system analysis data (e.g. installed skill names).
"""
lines = [
"You are performing a security audit of an OpenClaw deployment.",
"The following is METADATA ONLY from the user's ~/.openclaw/ directory.",
"Secret values have been redacted and marked as [REDACTED].",
"Do NOT ask for actual secrets — they are intentionally withheld.",
"",
"Analyze this OpenClaw configuration for security risks.",
"",
"--- BEGIN OPENCLAW METADATA ---",
json.dumps(metadata, indent=2, default=str),
"--- END OPENCLAW METADATA ---",
]
if system_info is not None:
lines.extend(
[
"",
"The following is system-level context for the deployment environment.",
"",
"--- BEGIN SYSTEM ANALYSIS ---",
json.dumps(system_info, indent=2, default=str),
"--- END SYSTEM ANALYSIS ---",
]
)
return "\n".join(lines)
FILE:scripts/openclaw_trent/lib/package_skills.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""
package_skills.py
Scans the OpenClaw workspace, identifies custom code and installed skills,
packages each as a .skill file (zip), and prints a JSON summary.
All text files are scanned for secrets before packaging — detected secrets
are replaced with [REDACTED] in the ZIP copy. Original files are never modified.
Discovery rules:
- workspace root: any dir or file that looks like code (heuristic-based)
- workspace/skills/: recursively finds skill dirs (contain SKILL.md)
- Skips itself, venv/node_modules, empty dirs, and known config-only files
"""
import hashlib
import json
import logging
import os
import pathlib
import re
import sys
import zipfile
from openclaw_trent.openclaw_config.secret_redactor import (
REDACTED_MARKER,
SECRET_KEY_PATTERNS,
SECRET_VALUE_PATTERNS,
)
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
WORKSPACE = pathlib.Path(
os.environ.get("OPENCLAW_WORKSPACE", pathlib.Path.home() / ".openclaw/workspace")
)
OUTPUT_DIR = WORKSPACE
# Dirs that are never code
SKIP_DIRS = {
"venv",
"__pycache__",
".git",
"node_modules",
".clawhub",
".learnings",
".openclaw",
}
# Root-level names that are workspace config, not user code
WORKSPACE_CONFIG_NAMES = {
"AGENTS.md",
"SOUL.md",
"USER.md",
"TOOLS.md",
"IDENTITY.md",
"HEARTBEAT.md",
"MEMORY.md",
"memory",
"main.sqlite",
"package_skills.py", # this script itself
}
# File extensions considered "code"
CODE_EXTENSIONS = {
".py",
".sh",
".js",
".ts",
".rb",
".go",
".rs",
".java",
".cpp",
".c",
".cs",
".php",
".swift",
".kt",
".r",
".lua",
}
# Min number of code files for a directory to be considered a project
MIN_CODE_FILES = 1
# Max file size for text redaction (skip redaction for very large files)
MAX_REDACT_FILE_SIZE = 10 * 1024 * 1024 # 10MB
# File extensions that are never included in ZIPs (may contain secrets)
EXCLUDED_EXTENSIONS = {
".pyc",
".pyo", # compiled Python
".db",
".sqlite", # databases
".pkl",
".pickle", # serialized objects
".key",
".pem",
".p12",
".pfx",
".jks", # crypto keys/certs
".env", # environment files (secrets.env, prod.env, etc.)
}
# Filenames that are never included in ZIPs (may contain secrets)
EXCLUDED_FILENAMES = {
".env",
".env.local",
".env.production",
".env.development",
"credentials.json",
"service-account.json",
}
# Context-aware key=value pattern: lines like `API_KEY = "sk-..."` or `token: ghp_...`
# Handles three forms:
# key = "value" (double-quoted)
# key = 'value' (single-quoted)
# key = value (unquoted)
_KEY_VALUE_RE = re.compile(
r"""
(?:^|[\s,;{(]) # line start or separator
(?P<key>[\w.-]+) # key name
\s*[:=]\s* # separator (= or :)
(?:
"(?P<dq>[^"]{8,})" # double-quoted value
| '(?P<sq>[^']{8,})' # single-quoted value
| (?P<uq>[^\s"',;)}{]{8,}) # unquoted value
)
""",
re.VERBOSE | re.MULTILINE,
)
# ---------------------------------------------------------------------------
# Secret redaction for file contents
# ---------------------------------------------------------------------------
def redact_file_content(content: str) -> tuple[str, int]:
"""Redact secrets from text file content.
Uses two strategies:
1. Value-format patterns (sk-..., ghp_..., AKIA..., connection strings)
2. Context-aware key=value detection (api_key = "...", token: "...")
Returns (redacted_content, redaction_count).
"""
count = 0
# Strategy 1: Replace known secret value patterns
for pattern in SECRET_VALUE_PATTERNS:
matches = list(pattern.finditer(content))
if matches:
count += len(matches)
content = pattern.sub(REDACTED_MARKER, content)
# Strategy 2: Context-aware key=value pairs
def _replace_key_value(m: re.Match) -> str:
nonlocal count
key = m.group("key")
# Pick whichever capture group matched (double-quoted, single-quoted, or unquoted)
value = m.group("dq") or m.group("sq") or m.group("uq")
if value is None:
return m.group(0)
# Check if the key name suggests a secret
if any(p.search(key) for p in SECRET_KEY_PATTERNS):
if value != REDACTED_MARKER:
count += 1
return m.group(0).replace(value, REDACTED_MARKER)
return m.group(0)
content = _KEY_VALUE_RE.sub(_replace_key_value, content)
return content, count
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def should_skip_dir(name: str) -> bool:
return name in SKIP_DIRS or name.startswith(".")
def is_code_file(path: pathlib.Path) -> bool:
return path.is_file() and path.suffix.lower() in CODE_EXTENSIONS
def count_code_files(directory: pathlib.Path) -> int:
"""Recursively count code files, skipping venv-like dirs."""
count = 0
for root, dirs, files in os.walk(directory):
dirs[:] = [d for d in dirs if not should_skip_dir(d)]
count += sum(1 for f in files if pathlib.Path(f).suffix.lower() in CODE_EXTENSIONS)
return count
def dir_is_empty_or_data_only(directory: pathlib.Path) -> bool:
"""True if dir has no code files at all."""
return count_code_files(directory) == 0
def parse_skill_frontmatter(skill_md: pathlib.Path) -> dict:
"""
Parse YAML-ish front-matter from SKILL.md (--- block at top).
Returns dict with 'name' and 'description' if found.
"""
if not skill_md.exists():
return {}
content = skill_md.read_text(encoding="utf-8", errors="replace")
if not content.startswith("---"):
return {}
end = content.find("---", 3)
if end < 0:
return {}
fm_text = content[3:end]
result = {}
# Handle multi-line description (> or | block scalars, or plain)
# Collect all lines after 'description:' until next key
lines = fm_text.splitlines()
i = 0
while i < len(lines):
line = lines[i]
m = re.match(r"^(\w[\w-]*):\s*(.*)", line)
if m:
key = m.group(1).lower()
val = m.group(2).strip()
if val in (">", "|", ""):
# Multi-line: collect indented continuation lines
parts = []
i += 1
while i < len(lines):
next_line = lines[i]
if next_line.startswith(" ") or next_line.startswith("\t"):
parts.append(next_line.strip())
i += 1
else:
break
result[key] = " ".join(parts)
continue
else:
result[key] = val.strip("'\"")
i += 1
return result
def read_meta(skill_dir: pathlib.Path) -> dict:
meta_path = skill_dir / "_meta.json"
if meta_path.exists():
try:
return json.loads(meta_path.read_text())
except Exception:
pass
return {}
def file_hash(path: pathlib.Path) -> str:
h = hashlib.sha256()
h.update(path.read_bytes())
return h.hexdigest()[:12]
def _add_file_to_zip(
zf: zipfile.ZipFile,
fp: pathlib.Path,
arcname: pathlib.PurePosixPath | str,
workspace_root: pathlib.Path | None = None,
) -> int:
"""Add a single file to a ZIP with secret redaction.
Skips symlinks and files that resolve outside the workspace root.
Excludes dangerous file types entirely. Text files are redacted.
Binary files are added as-is (secrets in binaries are rare in skill code).
Returns the number of secrets redacted.
"""
# Skip symlinks — prevent traversal outside workspace
if fp.is_symlink():
logger.warning("Skipped symlink %s", arcname)
return 0
# Ensure resolved path is within the workspace root
if workspace_root is not None:
try:
resolved = fp.resolve()
if not resolved.is_relative_to(workspace_root.resolve()):
logger.warning("Skipped %s (resolves outside workspace)", arcname)
return 0
except (OSError, RuntimeError):
return 0
# Exclude file types and filenames that commonly contain secrets
if fp.suffix.lower() in EXCLUDED_EXTENSIONS or fp.name.lower() in EXCLUDED_FILENAMES:
logger.debug("Excluded %s (dangerous extension)", arcname)
return 0
file_size = fp.stat().st_size
# Refuse to package files too large to redact safely
if file_size > MAX_REDACT_FILE_SIZE:
logger.warning("Excluded %s — too large for safe redaction (%d bytes)", arcname, file_size)
return 0
# Try to read as text and redact
try:
content = fp.read_text(encoding="utf-8")
redacted, count = redact_file_content(content)
if count > 0:
logger.info("Redacted %d secret(s) in %s", count, arcname)
zf.writestr(str(arcname), redacted)
return count
except (UnicodeDecodeError, ValueError):
# Binary file — add as-is
zf.write(fp, arcname)
return 0
def zip_directory(source_dir: pathlib.Path, output_path: pathlib.Path) -> tuple[int, int]:
"""ZIP a directory with secret redaction.
Returns (file_size, total_redactions).
"""
total_redactions = 0
with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zf:
for root, dirs, files in os.walk(source_dir):
dirs[:] = [d for d in dirs if not should_skip_dir(d)]
for file in sorted(files):
fp = pathlib.Path(root) / file
arcname = fp.relative_to(source_dir.parent)
total_redactions += _add_file_to_zip(zf, fp, arcname, workspace_root=source_dir)
return output_path.stat().st_size, total_redactions
def zip_file(source_file: pathlib.Path, output_path: pathlib.Path) -> tuple[int, int]:
"""ZIP a single file with secret redaction.
Returns (file_size, total_redactions).
"""
total_redactions = 0
with zipfile.ZipFile(output_path, "w", zipfile.ZIP_DEFLATED) as zf:
total_redactions += _add_file_to_zip(zf, source_file, source_file.name)
return output_path.stat().st_size, total_redactions
def find_skill_dirs_recursive(base: pathlib.Path) -> list[pathlib.Path]:
"""
Recursively find directories containing SKILL.md under base.
Stops descending once a SKILL.md is found (a skill won't contain another skill).
Skips symlinked directories and dirs that resolve outside base.
"""
found = []
base_resolved = base.resolve()
for entry in sorted(base.iterdir()):
if not entry.is_dir() or should_skip_dir(entry.name):
continue
if entry.is_symlink():
logger.warning("Skipped symlinked directory %s", entry.name)
continue
try:
if not entry.resolve().is_relative_to(base_resolved):
logger.warning("Skipped %s (resolves outside base)", entry.name)
continue
except (OSError, RuntimeError):
continue
if (entry / "SKILL.md").exists():
found.append(entry)
else:
found.extend(find_skill_dirs_recursive(entry))
return found
# ---------------------------------------------------------------------------
# Main scan
# ---------------------------------------------------------------------------
def scan_workspace(workspace: pathlib.Path | None = None) -> list[dict]:
ws = workspace or WORKSPACE
output_dir = ws
results = []
packaged_paths = set()
# -----------------------------------------------------------------------
# 1. Skills directory — recursive discovery
# -----------------------------------------------------------------------
skills_dir = ws / "skills"
if skills_dir.is_dir():
for skill_dir in find_skill_dirs_recursive(skills_dir):
# slug: filesystem directory name — stable, unique, used to name the .skill archive.
slug = skill_dir.name
skill_file = output_dir / f"{slug}.skill"
already_exists = skill_file.exists()
meta = read_meta(skill_dir)
fm = parse_skill_frontmatter(skill_dir / "SKILL.md")
size, redactions = zip_directory(skill_dir, skill_file)
packaged_paths.add(skill_dir)
results.append(
{
# name: human-readable label from SKILL.md frontmatter; falls back to slug.
# IMPORTANT: upload_skills.py uses this as the backend document identifier,
# and Phase 3 prompts reference the skill by this value — keep them in sync.
"name": fm.get("name", slug),
"slug": slug,
"type": "installed-skill",
"description": fm.get("description", "(no description)"),
"source": str(skill_dir.relative_to(ws)),
"skill_file": skill_file.name,
"skill_file_path": str(skill_file),
"skill_size_bytes": size,
"secrets_redacted": redactions > 0,
"version": meta.get("version"),
"origin": "clawhub" if meta.get("ownerId") else "custom",
"status": "re-packaged" if already_exists else "newly-packaged",
}
)
# -----------------------------------------------------------------------
# 2. Workspace root — skill dirs and code projects/files
# -----------------------------------------------------------------------
for entry in sorted(ws.iterdir()):
# Skip config, hidden, already-processed, .skill outputs
if entry.name in WORKSPACE_CONFIG_NAMES:
continue
if entry.name.startswith("."):
continue
if entry.suffix == ".skill":
continue
if entry in packaged_paths:
continue
# slug: filesystem entry name (dir name or file stem) — used to name the .skill archive.
slug = entry.stem if entry.is_file() else entry.name
skill_file = output_dir / f"{slug}.skill"
already_exists = skill_file.exists()
if entry.is_dir():
# Skip dirs with no code files
if dir_is_empty_or_data_only(entry):
continue
# Skip the skills/ dir itself (handled above)
if entry.name == "skills":
continue
is_skill = (entry / "SKILL.md").exists()
fm = parse_skill_frontmatter(entry / "SKILL.md") if is_skill else {}
meta = read_meta(entry) if is_skill else {}
size, redactions = zip_directory(entry, skill_file)
results.append(
{
# name: human-readable label from SKILL.md frontmatter; falls back to slug.
# IMPORTANT: upload_skills.py uses this as the backend document identifier,
# and Phase 3 prompts reference the skill by this value — keep them in sync.
"name": fm.get("name", slug),
"slug": slug,
"type": "skill-directory" if is_skill else "code-project",
"description": fm.get("description", "(no SKILL.md)"),
"source": str(entry.relative_to(ws)),
"skill_file": skill_file.name,
"skill_file_path": str(skill_file),
"skill_size_bytes": size,
"secrets_redacted": redactions > 0,
"version": meta.get("version"),
"origin": "clawhub" if meta.get("ownerId") else "custom",
"status": "re-packaged" if already_exists else "newly-packaged",
}
)
elif entry.is_file() and is_code_file(entry):
size, redactions = zip_file(entry, skill_file)
results.append(
{
# name == slug for standalone scripts: no SKILL.md means no frontmatter name.
"name": slug,
"slug": slug,
"type": "standalone-script",
"description": "(single file)",
"source": str(entry.relative_to(ws)),
"skill_file": skill_file.name,
"skill_file_path": str(skill_file),
"skill_size_bytes": size,
"secrets_redacted": redactions > 0,
"version": None,
"origin": "custom",
"status": "re-packaged" if already_exists else "newly-packaged",
}
)
return results
def main():
print(f"Scanning workspace: {WORKSPACE}", file=sys.stderr)
results = scan_workspace()
print(json.dumps(results, indent=2))
newly = sum(1 for r in results if r["status"] == "newly-packaged")
repkg = sum(1 for r in results if r["status"] == "re-packaged")
print(f"\n{len(results)} item(s): {newly} new, {repkg} re-packaged.", file=sys.stderr)
if __name__ == "__main__":
main()
FILE:scripts/openclaw_trent/lib/prompts.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Prompt builders for OpenClaw skill security analysis."""
def build_skill_analysis_prompt(upload_summary: dict) -> str:
"""Build a chat prompt requesting analysis of uploaded skills."""
skills = upload_summary.get("skills", [])
summary = upload_summary.get("summary", {})
uploaded_skills = [s for s in skills if s.get("status") in ("uploaded", "skipped")]
# s['name'] is the human-readable label registered with the backend in Phase 2
# (passed as the `name` argument to prepare_document_upload in upload_skills.py).
# It must match so the backend can correlate these prompts with the uploaded documents.
skill_list = "\n".join(f"- {s['name']} ({s['type']})" for s in uploaded_skills)
total = summary.get("uploaded", 0) + summary.get("skipped", 0)
return (
f"I have uploaded {total} OpenClaw skill packages for security analysis. "
f"Please analyse them and return the results.\n\n"
f"Uploaded skills:\n{skill_list}\n\n"
f"Focus on: prompt injection vectors, tool permission escalation, "
f"data exfiltration paths, secrets in skill code, unsafe subprocess calls, "
f"and chained attack paths between skills.\n"
f"Group findings by severity (CRITICAL, HIGH, MEDIUM, LOW)."
)
def build_per_skill_analysis_prompt(skill: dict) -> str:
"""Build a chat prompt requesting analysis of a single uploaded skill.
skill['name'] must match the document name registered with the backend in Phase 2
(i.e., the `name` passed to prepare_document_upload() in upload_skills.py).
Both use the human-readable label from SKILL.md frontmatter (not the filesystem slug).
"""
return (
f"Please analyse the uploaded OpenClaw skill package '{skill['name']}' "
f"(type: {skill['type']}) for security risks.\n\n"
f"Focus on: prompt injection vectors, tool permission escalation, "
f"data exfiltration paths, secrets in skill code, unsafe subprocess calls, "
f"and chained attack paths with other skills in this deployment.\n"
f"Group findings by severity (CRITICAL, HIGH, MEDIUM, LOW)."
)
FILE:scripts/openclaw_trent/lib/system_analyzer.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Collects OpenClaw deployment context for security audits.
Gathers installed skill names from the filesystem.
All output is redacted for secrets before emission.
Channel configuration is read from openclaw.json by the collector module.
"""
import json
import logging
from pathlib import Path
from typing import Any
from openclaw_trent import __version__
from openclaw_trent.openclaw_config.secret_redactor import SecretRedactor
logger = logging.getLogger(__name__)
DEFAULT_OPENCLAW_PATH = Path.home() / ".openclaw"
def detect_skills(openclaw_path: Path | None = None) -> dict[str, Any]:
"""List installed skills from managed and workspace directories.
Only collects directory names — does not read file contents.
"""
base_path = (openclaw_path or DEFAULT_OPENCLAW_PATH).expanduser()
skills: list[dict[str, str]] = []
errors: list[str] = []
skill_dirs = [
("managed", base_path / "skills"),
("workspace", base_path / "workspace" / "skills"),
]
for source, skills_dir in skill_dirs:
if not skills_dir.is_dir():
continue
try:
for entry in sorted(skills_dir.iterdir()):
if entry.is_dir() and not entry.name.startswith("."):
# Use "slug" (not "name") — entry.name is the filesystem directory name,
# which is the slug. The human-readable name from SKILL.md frontmatter
# is NOT available here because detect_skills() does not read file contents.
skills.append({"slug": entry.name, "source": source})
except OSError as e:
errors.append(f"Cannot read {skills_dir}: {e}")
managed_count = sum(1 for s in skills if s["source"] == "managed")
workspace_count = sum(1 for s in skills if s["source"] == "workspace")
result: dict[str, Any] = {
"skills": skills,
"managed_count": managed_count,
"workspace_count": workspace_count,
}
if errors:
result["errors"] = errors
return result
def collect_system_analysis(openclaw_path: Path | None = None) -> dict[str, Any]:
"""Collect OpenClaw deployment context with secret redaction.
Gathers installed skill names. Channel configuration is available
in openclaw.json via the collector module.
Applies SecretRedactor to the full result before returning.
"""
errors: list[str] = []
skills_info = detect_skills(openclaw_path=openclaw_path)
if skills_info.get("errors"):
errors.extend(skills_info.pop("errors"))
result: dict[str, Any] = {
"schema_version": "3.0",
"skills": skills_info,
"errors": errors,
}
# Apply secret redaction
redactor = SecretRedactor()
result = redactor.redact(result)
result["redacted_paths"] = redactor.redacted_paths
return result
def main() -> None:
"""CLI entry point for system analysis."""
import argparse
parser = argparse.ArgumentParser(
prog="trent-openclaw-sysinfo",
description="Collect OpenClaw deployment context for security audit.",
)
parser.add_argument(
"--version",
action="version",
version=f"%(prog)s {__version__}",
)
parser.add_argument(
"--path",
type=str,
default=None,
help="Path to OpenClaw config directory (default: ~/.openclaw)",
)
args = parser.parse_args()
path_arg = Path(args.path) if args.path else None
result = collect_system_analysis(openclaw_path=path_arg)
print(json.dumps(result, indent=2, default=str))
if __name__ == "__main__":
main()
FILE:scripts/openclaw_trent/lib/trent_client.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Minimal Trent API client using stdlib only (no httpx, no third-party deps).
Handles:
- API key auth from TRENT_API_KEY env var
- Streaming SSE chat endpoint
- Presigned S3 upload (prepare + PUT)
"""
import json
import os
import urllib.error
import urllib.request
from openclaw_trent import __version__
# ---------------------------------------------------------------------------
# Config
# ---------------------------------------------------------------------------
_DEFAULT_CHAT_URL = "https://chat.trent.ai"
_DEFAULT_AGENT_URL = "https://api.trent.ai"
def _get_chat_url() -> str:
return os.environ.get("TRENT_CHAT_API_URL") or _DEFAULT_CHAT_URL
def _get_agent_url() -> str:
return os.environ.get("TRENT_AGENT_API_URL") or _DEFAULT_AGENT_URL
def _get_api_key() -> str | None:
"""Read API key from TRENT_API_KEY env var."""
key = os.environ.get("TRENT_API_KEY", "").strip()
return key if key else None
def _get_auth_header() -> str:
key = _get_api_key()
if key:
return key
raise RuntimeError("No Trent API key found. Set the TRENT_API_KEY environment variable.")
# ---------------------------------------------------------------------------
# Chat (streaming SSE)
# ---------------------------------------------------------------------------
def chat(
message: str,
context: str | None = None,
thread_id: str | None = None,
output_file: str | None = None,
) -> dict:
"""Send a chat message to Trent API using streaming SSE.
Streams chunks to a temp file as they arrive so that partial results
survive if the process is killed by a sandbox timeout. The file path
is included in the response as ``output_file`` so the caller (or the
agent) can read it back.
Args:
message: The chat message to send.
context: Optional context string.
thread_id: Optional thread ID for conversation continuity.
output_file: Optional path to write streaming output to.
Defaults to a temp file in /tmp.
Returns:
dict with keys: content, thread_id, output_file, error (optional).
"""
import tempfile
auth_header = _get_auth_header()
payload = json.dumps(
{
"message": message,
"context": context,
"thread_id": thread_id,
"stream": True,
"client_info": {
"client_type": "openclaw-skill",
"client_version": __version__,
},
}
).encode()
headers = {
"Authorization": auth_header,
"Content-Type": "application/json",
"Accept": "text/event-stream",
}
req = urllib.request.Request(
f"{_get_chat_url()}/v1/chat",
data=payload,
headers=headers,
method="POST",
)
# Write chunks to a file as they arrive — survives sandbox SIGTERM
out_path = output_file or tempfile.mktemp(prefix="trent_chat_", suffix=".json")
content_chunks: list[str] = []
returned_thread_id: str | None = thread_id
try:
with urllib.request.urlopen(req, timeout=300) as resp, open(out_path, "w") as out:
for raw_line in resp:
line = raw_line.decode("utf-8").rstrip("\r\n")
if not line:
continue
if not line.startswith("data: "):
continue
data_str = line[6:]
if data_str.strip() == "[DONE]":
break
try:
data = json.loads(data_str)
chunk = None
if "content" in data:
chunk = data["content"]
elif "delta" in data and "content" in data["delta"]:
chunk = data["delta"]["content"]
if chunk:
content_chunks.append(chunk)
out.write(chunk)
out.flush()
if "thread_id" in data:
returned_thread_id = data["thread_id"]
except json.JSONDecodeError:
continue
# Write final result as structured JSON
result = {
"content": "".join(content_chunks),
"thread_id": returned_thread_id,
}
out.seek(0)
out.truncate()
out.write(json.dumps(result, indent=2))
except urllib.error.HTTPError as e:
body = ""
try:
body = e.read().decode("utf-8", errors="replace")
except Exception:
pass
if e.code == 401:
return {
"content": "API key rejected (expired or revoked). "
"Generate a new key at https://app.trent.ai.",
"error": True,
"output_file": out_path,
}
return {"content": f"API error {e.code}: {body}", "error": True, "output_file": out_path}
except TimeoutError:
return {
"content": "Request timed out. Please try again.",
"error": True,
"output_file": out_path,
}
except Exception as e:
return {"content": f"An error occurred: {e}", "error": True, "output_file": out_path}
return {
"content": "".join(content_chunks),
"thread_id": returned_thread_id,
"output_file": out_path,
}
# ---------------------------------------------------------------------------
# Document upload (presigned S3)
# ---------------------------------------------------------------------------
def _api_request(method: str, endpoint: str, json_data: dict | None = None) -> dict:
"""Make an authenticated JSON request to the Trent agent API."""
auth_header = _get_auth_header()
url = f"{_get_agent_url()}/v1/humber-agent{endpoint}"
payload = json.dumps(json_data).encode() if json_data is not None else None
headers: dict[str, str] = {
"Authorization": auth_header,
"Content-Type": "application/json",
}
req = urllib.request.Request(url, data=payload, headers=headers, method=method)
with urllib.request.urlopen(req, timeout=60) as resp:
return json.loads(resp.read().decode())
def prepare_document_upload(
name: str,
doc_type: str,
doc_format: str,
digest: str | None = None,
) -> dict:
"""Prepare a presigned S3 URL for document upload."""
body: dict = {"name": name, "type": doc_type, "format": doc_format}
if digest:
body["digest"] = digest
return _api_request("POST", "/documents/upload", json_data=body)
def upload_content_to_presigned_url(
upload_url: str,
content: bytes,
content_type: str = "application/zip",
) -> None:
"""PUT content to a presigned S3 URL (no auth header — creds in query params)."""
req = urllib.request.Request(
upload_url,
data=content,
headers={"Content-Type": content_type},
method="PUT",
)
try:
with urllib.request.urlopen(req, timeout=120):
pass # 2xx — success
except urllib.error.HTTPError as e:
body = ""
try:
body = e.read().decode("utf-8", errors="replace")
except Exception:
pass
raise RuntimeError(f"S3 upload failed: HTTP {e.code} — {body}") from e
FILE:scripts/openclaw_trent/lib/upload_skills.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Skill upload logic using stdlib-only trent_client.
Packages installed skills and custom code from the OpenClaw workspace
as .skill ZIP archives and uploads them to Trent via presigned S3 URLs
for deep security analysis.
"""
import hashlib
import logging
from collections.abc import Callable
from pathlib import Path
from openclaw_trent.lib import trent_client
logger = logging.getLogger(__name__)
SOURCE_NAME = "HumberAgent OpenClaw Skill Upload"
MAX_SKILL_SIZE_BYTES = 50 * 1024 * 1024 # 50MB client-side limit
def _compute_sha256(content: bytes) -> str:
"""Compute SHA256 digest in the format expected by the backend."""
return f"sha256:{hashlib.sha256(content).hexdigest()}"
def upload_packaged_skills(
skills: list[dict],
on_progress: Callable[[str, int, int], None] | None = None,
) -> dict:
"""Upload .skill files via presigned S3 URLs.
Args:
skills: List of skill dicts from ``scan_workspace()``.
on_progress: Optional ``(message, progress, total)`` callback.
Returns:
Upload summary dict with ``source``, ``skills``, and ``summary`` keys.
"""
if not skills:
return {
"source": SOURCE_NAME,
"skills": [],
"summary": {
"total": 0,
"uploaded": 0,
"skipped": 0,
"failed": 0,
"too_large": 0,
},
"message": "No skills or custom code found in workspace.",
}
results: list[dict] = []
counts = {"uploaded": 0, "skipped": 0, "failed": 0, "too_large": 0}
for i, skill in enumerate(skills):
# skill_name: human-readable label from SKILL.md frontmatter (or slug if no frontmatter).
# This is the document identifier used with the backend API — Phase 3 prompts reference
# skills by this value, so it must be consistent end-to-end.
skill_name = skill["name"]
# skill_slug: filesystem directory/file name — unique on disk, used only for local
# logging and result dict keys. NOT used as the backend document identifier.
skill_slug = skill["slug"]
skill_path = Path(skill["skill_file_path"])
skill_size = skill["skill_size_bytes"]
progress = 10 + int((i / len(skills)) * 80)
if on_progress:
on_progress(f"Uploading {skill_name} ({i + 1}/{len(skills)})...", progress, 100)
# Check client-side size limit
if skill_size > MAX_SKILL_SIZE_BYTES:
size_mb = skill_size / (1024 * 1024)
limit_mb = MAX_SKILL_SIZE_BYTES / (1024 * 1024)
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "too_large",
"size_bytes": skill_size,
"s3_url": None,
"error": f"Skill too large: {size_mb:.1f}MB (limit: {limit_mb:.0f}MB)",
}
)
counts["too_large"] += 1
continue
try:
file_bytes = skill_path.read_bytes()
except OSError as e:
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "failed",
"size_bytes": skill_size,
"s3_url": None,
"error": f"Cannot read skill file: {e}",
}
)
counts["failed"] += 1
continue
digest = _compute_sha256(file_bytes)
try:
upload_info = trent_client.prepare_document_upload(
# Use skill_name (human-readable label) as the document identifier so that
# Phase 3 chat prompts — which reference skill['name'] — match what the
# backend has stored. Using skill_slug here would cause a name mismatch
# when slug ≠ name (e.g. dir "trentclaw" vs SKILL.md name "Trent OpenClaw…").
name=skill_name,
doc_type="openclaw_config",
doc_format="zip",
digest=digest,
)
except Exception as e:
logger.warning("Failed to prepare upload for %s: %s", skill_slug, e)
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "failed",
"size_bytes": skill_size,
"s3_url": None,
"error": f"Upload preparation failed: {e}",
}
)
counts["failed"] += 1
continue
# Check server-side size limit
max_size = upload_info.get("max_size_bytes")
if max_size and skill_size > max_size:
size_mb = skill_size / (1024 * 1024)
limit_mb = max_size / (1024 * 1024)
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "too_large",
"size_bytes": skill_size,
"s3_url": None,
"error": f"Exceeds server limit: {size_mb:.1f}MB (max: {limit_mb:.1f}MB)",
}
)
counts["too_large"] += 1
continue
s3_url = upload_info.get("s3_url")
if upload_info.get("skipped_upload"):
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "skipped",
"size_bytes": skill_size,
"s3_url": s3_url,
"error": None,
}
)
counts["skipped"] += 1
continue
try:
trent_client.upload_content_to_presigned_url(
upload_url=upload_info["upload_url"],
content=file_bytes,
content_type="application/zip",
)
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "uploaded",
"size_bytes": skill_size,
"s3_url": s3_url,
"error": None,
}
)
counts["uploaded"] += 1
except Exception as e:
logger.warning("Failed to upload %s: %s", skill_slug, e)
results.append(
{
"name": skill_name,
"slug": skill_slug,
"type": skill["type"],
"status": "failed",
"size_bytes": skill_size,
"s3_url": s3_url,
"error": f"S3 upload failed: {e}",
}
)
counts["failed"] += 1
uploaded = counts["uploaded"]
skipped = counts["skipped"]
failed = counts["failed"]
too_large = counts["too_large"]
if on_progress:
on_progress(
f"Upload complete: {uploaded} uploaded, {skipped} skipped, "
f"{failed} failed, {too_large} too large",
100,
100,
)
return {
"source": SOURCE_NAME,
"skills": results,
"summary": {
"total": len(skills),
**counts,
},
}
FILE:scripts/openclaw_trent/openclaw_config/__init__.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""OpenClaw configuration collection and secret redaction."""
from openclaw_trent.openclaw_config.collector import collect_openclaw_metadata
__all__ = ["collect_openclaw_metadata"]
FILE:scripts/openclaw_trent/openclaw_config/collector.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Collects OpenClaw deployment metadata for security audit.
Reads ~/.openclaw/ and extracts ONLY metadata — never actual secret values.
Secret values are redacted locally before any data leaves the machine.
"""
import json
import logging
import stat
from pathlib import Path
from typing import Any
from openclaw_trent.openclaw_config.secret_redactor import SecretRedactor
logger = logging.getLogger(__name__)
MAX_CONFIG_FILE_SIZE = 5 * 1024 * 1024 # 5 MB
MAX_SKILL_FILE_SIZE = 1 * 1024 * 1024 # 1 MB
MAX_TOTAL_READ_SIZE = 50 * 1024 * 1024 # 50 MB
MAX_SKILL_COUNT = 100
DEFAULT_OPENCLAW_PATH = Path.home() / ".openclaw"
def _is_safe_path(path: Path, base_path: Path) -> bool:
"""Validate path is within base directory and not a symlink."""
try:
resolved = path.resolve()
except (OSError, RuntimeError):
return False
if not resolved.is_relative_to(base_path.resolve()):
logger.warning("Path %s is outside allowed directory %s", path, base_path)
return False
if path.is_symlink():
logger.warning("Path %s is a symlink, refusing to read", path)
return False
return True
def _safe_read_file(
path: Path,
base_path: Path,
max_size: int,
total_tracker: dict,
) -> str | None:
"""Safely read a file with size limits and symlink checks.
Returns None on failure (missing, symlink, too large, etc.).
"""
if not _is_safe_path(path, base_path):
return None
if not path.is_file():
return None
try:
file_size = path.stat().st_size
except OSError:
return None
if file_size > max_size:
logger.warning("File %s exceeds size limit (%d > %d)", path, file_size, max_size)
return None
if total_tracker["bytes"] + file_size > MAX_TOTAL_READ_SIZE:
logger.warning("Total read size limit exceeded, skipping further reads")
return None
try:
content = path.read_text(encoding="utf-8")
total_tracker["bytes"] += len(content)
return content
except (OSError, UnicodeDecodeError) as e:
logger.warning("Failed to read %s: %s", path, e)
return None
def _get_file_permissions(path: Path) -> dict[str, Any] | None:
"""Get file permission metadata."""
try:
st = path.stat()
mode = stat.S_IMODE(st.st_mode)
return {
"mode_octal": oct(mode),
"owner_read_only": mode in (0o600, 0o400),
"world_readable": bool(mode & stat.S_IROTH),
"world_writable": bool(mode & stat.S_IWOTH),
}
except OSError:
return None
def _parse_json(content: str) -> dict | None:
"""Parse JSON config file."""
try:
return json.loads(content)
except json.JSONDecodeError:
return None
def _parse_yaml_frontmatter(content: str) -> dict[str, Any] | None:
"""Extract YAML frontmatter from a SKILL.md file using simple parsing."""
if not content.startswith("---"):
return None
parts = content.split("---", 2)
if len(parts) < 3:
return None
result: dict[str, Any] = {}
for line in parts[1].strip().splitlines():
if ":" in line:
key, _, value = line.partition(":")
key = key.strip()
value = value.strip().strip("'\"")
if key and value:
result[key] = value
return result if result else None
def collect_openclaw_metadata(
openclaw_path: Path | None = None,
) -> dict[str, Any]:
"""Collect OpenClaw deployment metadata with secret redaction.
Reads ~/.openclaw/ and extracts ONLY metadata — never actual secret values.
Designed to be called from both the MCP tool (async context) and the CLI
(sync context).
Args:
openclaw_path: Override path to OpenClaw config directory.
Defaults to ~/.openclaw.
Returns:
dict with keys: config, skills, workspace, file_permissions,
redacted_paths, errors, collection_path.
On directory not found: {"error": True, "message": "..."}.
"""
base_path = (openclaw_path or DEFAULT_OPENCLAW_PATH).expanduser()
if not base_path.is_dir():
return {
"error": True,
"message": f"OpenClaw config directory not found: {base_path}",
"path_checked": str(base_path),
}
redactor = SecretRedactor()
total_tracker = {"bytes": 0}
errors: list[str] = []
result: dict[str, Any] = {}
# --- 1. Read openclaw.json ---
config_file = base_path / "openclaw.json"
config_content = _safe_read_file(config_file, base_path, MAX_CONFIG_FILE_SIZE, total_tracker)
if config_content is not None:
raw_config = _parse_json(config_content)
if raw_config is not None:
result["config"] = redactor.redact(raw_config)
else:
errors.append("Invalid JSON in openclaw.json")
result["config"] = None
else:
result["config"] = None
if config_file.exists():
errors.append("Could not read openclaw.json (symlink, too large, or permission denied)")
# --- 2. Read skills ---
skills: list[dict[str, Any]] = []
skills_dir = base_path / "skills"
if skills_dir.is_dir() and _is_safe_path(skills_dir, base_path):
skill_count = 0
try:
entries = sorted(skills_dir.iterdir())
except OSError:
entries = []
for skill_entry in entries:
if skill_count >= MAX_SKILL_COUNT:
errors.append(f"Skill count limit reached ({MAX_SKILL_COUNT}), skipping remaining")
break
if not skill_entry.is_dir():
continue
skill_count += 1
skill_file = skill_entry / "SKILL.md"
skill_info: dict[str, Any] = {"name": skill_entry.name}
if skill_file.exists():
content = _safe_read_file(skill_file, base_path, MAX_SKILL_FILE_SIZE, total_tracker)
if content is not None:
frontmatter = _parse_yaml_frontmatter(content)
if frontmatter:
skill_info["frontmatter"] = redactor.redact(frontmatter)
# Include section headings only, not full content
skill_info["sections"] = [
line.lstrip("# ").strip()
for line in content.split("\n")
if line.startswith("#")
]
else:
skill_info["read_error"] = True
else:
skill_info["has_skill_file"] = False
skills.append(skill_info)
result["skills"] = skills
# --- 3. Check workspace ---
workspace_dir = base_path / "workspace"
workspace_info: dict[str, Any] = {"exists": workspace_dir.is_dir()}
if workspace_dir.is_dir() and _is_safe_path(workspace_dir, base_path):
for fname in ("SOUL.md", "AGENTS.md", "MEMORY.md", "TOOLS.md", "IDENTITY.md"):
key = fname.lower().replace(".", "_") + "_exists"
workspace_info[key] = (workspace_dir / fname).is_file()
workspace_info["permissions"] = _get_file_permissions(workspace_dir)
result["workspace"] = workspace_info
# --- 4. Check file permissions ---
permissions: dict[str, Any] = {}
for name, path in [
("openclaw.json", config_file),
("config_directory", base_path),
]:
if path.exists():
permissions[name] = _get_file_permissions(path)
result["file_permissions"] = permissions
# --- 5. Metadata ---
result["redacted_paths"] = redactor.redacted_paths
result["errors"] = errors
result["collection_path"] = str(base_path)
return result
FILE:scripts/openclaw_trent/openclaw_config/secret_redactor.py
# Copyright (c) 2025-2026 Trent AI. All rights reserved.
# Licensed under the Trent AI Proprietary License.
"""Secret redaction for OpenClaw configuration metadata.
Recursively traverses nested data structures and replaces detected secrets
with a [REDACTED] marker. Uses two-layer detection: key-name patterns
(e.g., api_key, password) and value-format patterns (e.g., sk-..., ghp_...).
"""
import re
from typing import Any
# Key names that suggest the value is a secret (case-insensitive)
SECRET_KEY_PATTERNS = [
re.compile(p, re.IGNORECASE)
for p in [
r"(api[_-]?key|apikey|api[_-]?secret)",
r"(^token$|access[_-]?token|auth[_-]?token|bearer[_-]?token|refresh[_-]?token)",
r"(password|passwd|pwd)",
r"(secret|credential)",
r"(private[_-]?key|secret[_-]?key|signing[_-]?key|encryption[_-]?key)",
r"(client[_-]?secret)",
r"(webhook[_-]?secret|webhook[_-]?token)",
r"(database[_-]?url|db[_-]?url|connection[_-]?string)",
r"(aws[_-]?access|aws[_-]?secret)",
]
]
# Value formats that indicate secrets regardless of key name
SECRET_VALUE_PATTERNS = [
re.compile(p)
for p in [
r"sk-[a-zA-Z0-9]{20,}", # OpenAI-style
r"sk-ant-[a-zA-Z0-9\-]{20,}", # Anthropic
r"sk-proj-[a-zA-Z0-9]{20,}", # OpenAI project keys
r"xoxb-[0-9]+-[0-9]+-[a-zA-Z0-9]+", # Slack bot tokens
r"xoxp-[0-9]+-[0-9]+-[0-9]+-[a-f0-9]+", # Slack user tokens
r"ghp_[a-zA-Z0-9]{36,}", # GitHub PATs
r"gho_[a-zA-Z0-9]{36,}", # GitHub OAuth
r"github_pat_[a-zA-Z0-9_]{20,}", # GitHub fine-grained PATs
r"AKIA[0-9A-Z]{16}", # AWS access keys
r"https?://[^:]+:[^@]+@", # URLs with embedded credentials
r"postgresql://[^:]+:[^@]+@", # PostgreSQL connection strings
r"mongodb(\+srv)?://[^:]+:[^@]+@", # MongoDB connection strings
r"mysql://[^:]+:[^@]+@", # MySQL connection strings
r"redis://[^:]+:[^@]+@", # Redis connection strings
]
]
REDACTED_MARKER = "[REDACTED]"
class SecretRedactor:
"""Recursively redacts secrets from nested data structures.
Tracks all redacted JSON paths for audit trail.
"""
def __init__(self) -> None:
self.redacted_paths: list[str] = []
def is_secret_key(self, key: str) -> bool:
"""Check if a key name suggests its value is a secret."""
return any(p.search(key) for p in SECRET_KEY_PATTERNS)
def is_secret_value(self, value: str) -> bool:
"""Check if a string value matches known secret formats."""
if not isinstance(value, str) or len(value) < 8:
return False
return any(p.search(value) for p in SECRET_VALUE_PATTERNS)
def redact(self, data: Any, path: str = "") -> Any:
"""Recursively redact secrets from a data structure.
Args:
data: The data to redact (dict, list, or scalar).
path: Dot-separated JSON path for audit trail.
Returns:
A copy of the data with secrets replaced by [REDACTED].
"""
if isinstance(data, dict):
result = {}
for key, value in data.items():
current_path = f"{path}.{key}" if path else key
if self.is_secret_key(str(key)):
if isinstance(value, str) and value.strip():
result[key] = REDACTED_MARKER
self.redacted_paths.append(current_path)
else:
# Empty/null/non-string values are not secrets
result[key] = self.redact(value, current_path)
elif isinstance(value, str) and self.is_secret_value(value):
result[key] = REDACTED_MARKER
self.redacted_paths.append(current_path)
else:
result[key] = self.redact(value, current_path)
return result
elif isinstance(data, list):
return [self.redact(item, f"{path}[{i}]") for i, item in enumerate(data)]
elif isinstance(data, str) and self.is_secret_value(data):
self.redacted_paths.append(path)
return REDACTED_MARKER
else:
return data