Source snapshot of github.com/Scottcjn/trashclaw@6dce9ed

53b2ac9 verified 13 days ago

10.5 kB

# SPDX-License-Identifier: MIT
"""
Env Audit Plugin for TrashClaw
Scan a directory for common security misconfigurations:
  - .env files committed to git
  - Exposed secrets (API keys, passwords, tokens in source)
  - World-readable private keys
  - Sensitive files without .gitignore coverage
Plugin contract:
  TOOL_DEF = dict with name, description, parameters (OpenAI function schema)
  run(**kwargs) -> str  (called with the parsed arguments)
"""

import os
import re
import stat

TOOL_DEF = {
    "name": "env_audit",
    "description": "Scan a directory for common security misconfigurations: .env in git, exposed secrets, world-readable keys, missing .gitignore entries.",
    "parameters": {
        "type": "object",
        "properties": {
            "path": {
                "type": "string",
                "description": "Directory to scan (default: current directory)"
            },
            "deep": {
                "type": "boolean",
                "description": "If true, scan file contents for secrets (slower). Default: true."
            }
        },
        "required": []
    }
}

# Filenames that should never be committed
SENSITIVE_FILES = [
    ".env",
    ".env.local",
    ".env.production",
    ".env.staging",
    ".env.development",
    "credentials.json",
    "service-account.json",
    "secrets.yaml",
    "secrets.yml",
    ".netrc",
    ".npmrc",  # can contain tokens
    ".pypirc",  # can contain tokens
]

# Private key file patterns
KEY_PATTERNS = [
    "*.pem",
    "*.key",
    "*.p12",
    "*.pfx",
    "id_rsa",
    "id_ed25519",
    "id_ecdsa",
    "id_dsa",
]

# Regex patterns for secrets in file contents
SECRET_PATTERNS = [
    (r'(?i)(?:api[_-]?key|apikey)\s*[:=]\s*["\']?[A-Za-z0-9_\-]{16,}', "API key"),
    (r'(?i)(?:secret|password|passwd|pwd)\s*[:=]\s*["\']?[^\s"\']{8,}', "Password/Secret"),
    (r'(?i)(?:aws_access_key_id|aws_secret_access_key)\s*[:=]\s*["\']?[A-Za-z0-9/+=]{16,}', "AWS credential"),
    (r'(?i)bearer\s+[A-Za-z0-9_\-\.]{20,}', "Bearer token"),
    (r'ghp_[A-Za-z0-9]{36}', "GitHub PAT"),
    (r'sk-[A-Za-z0-9]{20,}', "OpenAI/Stripe secret key"),
    (r'(?i)(?:private[_-]?key)\s*[:=]\s*["\']?[A-Za-z0-9_\-/+=]{16,}', "Private key value"),
    (r'-----BEGIN (?:RSA |EC |DSA |OPENSSH )?PRIVATE KEY-----', "Private key block"),
]

# Extensions to scan for secrets
SCANNABLE_EXTENSIONS = {
    ".py", ".js", ".ts", ".jsx", ".tsx", ".rb", ".go", ".java",
    ".yaml", ".yml", ".json", ".toml", ".ini", ".cfg", ".conf",
    ".sh", ".bash", ".zsh", ".env", ".properties", ".xml",
    ".tf", ".hcl", ".dockerfile",
}

# Directories to skip
SKIP_DIRS = {
    ".git", "node_modules", "__pycache__", ".venv", "venv",
    "env", ".tox", ".mypy_cache", ".pytest_cache", "dist",
    "build", ".eggs", "*.egg-info",
}

MAX_FILE_SIZE = 512 * 1024  # 512KB max per file for content scanning


def _is_git_tracked(filepath: str, repo_root: str) -> bool:
    """Check if a file is tracked by git (would be committed)."""
    try:
        import subprocess
        result = subprocess.run(
            ["git", "ls-files", "--error-unmatch", filepath],
            cwd=repo_root, capture_output=True, text=True, timeout=5
        )
        return result.returncode == 0
    except Exception:
        return False  # Can't determine; assume not tracked


def _is_gitignored(filepath: str, repo_root: str) -> bool:
    """Check if a file is covered by .gitignore."""
    try:
        import subprocess
        result = subprocess.run(
            ["git", "check-ignore", "-q", filepath],
            cwd=repo_root, capture_output=True, text=True, timeout=5
        )
        return result.returncode == 0
    except Exception:
        return False


def _check_permissions(filepath: str) -> str:
    """Check if file has overly permissive permissions (world-readable)."""
    try:
        st = os.stat(filepath)
        mode = st.st_mode
        if mode & stat.S_IROTH:  # World-readable
            return f"{oct(mode & 0o777)} (world-readable)"
    except Exception:
        pass
    return ""


def _scan_file_contents(filepath: str) -> list:
    """Scan a file for secret patterns. Returns list of (line_no, pattern_name, snippet)."""
    findings = []
    try:
        size = os.path.getsize(filepath)
        if size > MAX_FILE_SIZE:
            return findings
        with open(filepath, "r", errors="replace") as f:
            for line_no, line in enumerate(f, 1):
                for pattern, name in SECRET_PATTERNS:
                    if re.search(pattern, line):
                        # Truncate the line for display
                        snippet = line.strip()
                        if len(snippet) > 100:
                            snippet = snippet[:97] + "..."
                        findings.append((line_no, name, snippet))
                        break  # One finding per line is enough
    except Exception:
        pass
    return findings


def _find_git_root(path: str) -> str:
    """Walk up to find .git directory."""
    current = os.path.abspath(path)
    while current != os.path.dirname(current):
        if os.path.isdir(os.path.join(current, ".git")):
            return current
        current = os.path.dirname(current)
    return ""


def run(path: str = "", deep: bool = True, **kwargs) -> str:
    scan_dir = os.path.abspath(path or os.getcwd())
    if not os.path.isdir(scan_dir):
        return f"Error: not a directory: {scan_dir}"

    git_root = _find_git_root(scan_dir)
    has_git = bool(git_root)

    issues = []  # (severity, category, filepath, detail)

    # Walk the directory tree
    for root, dirs, files in os.walk(scan_dir):
        # Skip hidden/vendor directories
        dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")]

        rel_root = os.path.relpath(root, scan_dir)

        for fname in files:
            filepath = os.path.join(root, fname)
            rel_path = os.path.relpath(filepath, scan_dir)

            # Check 1: Sensitive files present
            if fname.lower() in [s.lower() for s in SENSITIVE_FILES]:
                if has_git:
                    tracked = _is_git_tracked(rel_path, git_root)
                    ignored = _is_gitignored(rel_path, git_root)
                    if tracked:
                        issues.append(("CRITICAL", "sensitive_file_tracked", rel_path,
                                       f"Sensitive file is tracked by git!"))
                    elif not ignored:
                        issues.append(("WARNING", "sensitive_file_not_ignored", rel_path,
                                       f"Sensitive file exists but is NOT in .gitignore"))
                else:
                    issues.append(("WARNING", "sensitive_file_present", rel_path,
                                   f"Sensitive file found (no git repo to check ignore rules)"))

            # Check 2: Private key files
            for pat in KEY_PATTERNS:
                if pat.startswith("*"):
                    if fname.endswith(pat[1:]):
                        perm = _check_permissions(filepath)
                        if perm:
                            issues.append(("WARNING", "key_permissions", rel_path,
                                           f"Private key is {perm}"))
                        if has_git and _is_git_tracked(rel_path, git_root):
                            issues.append(("CRITICAL", "key_tracked", rel_path,
                                           f"Private key tracked by git!"))
                        break
                elif fname == pat:
                    perm = _check_permissions(filepath)
                    if perm:
                        issues.append(("WARNING", "key_permissions", rel_path,
                                       f"Private key is {perm}"))
                    if has_git and _is_git_tracked(rel_path, git_root):
                        issues.append(("CRITICAL", "key_tracked", rel_path,
                                       f"Private key tracked by git!"))
                    break

            # Check 3: Content scanning for secrets
            if deep:
                _, ext = os.path.splitext(fname)
                if ext.lower() in SCANNABLE_EXTENSIONS or fname.lower() in (".env",):
                    findings = _scan_file_contents(filepath)
                    for line_no, pattern_name, snippet in findings:
                        issues.append(("WARNING", "secret_in_source", rel_path,
                                       f"Line {line_no}: possible {pattern_name}"))

    # Check 4: .gitignore coverage
    if has_git:
        gitignore_path = os.path.join(git_root, ".gitignore")
        if not os.path.exists(gitignore_path):
            issues.append(("INFO", "no_gitignore", ".gitignore",
                           "No .gitignore file found in repository"))
        else:
            try:
                with open(gitignore_path, "r") as f:
                    gitignore_content = f.read()
                recommended = [".env", "*.pem", "*.key", "id_rsa", "credentials.json"]
                missing = [p for p in recommended if p not in gitignore_content]
                if missing:
                    issues.append(("INFO", "gitignore_incomplete", ".gitignore",
                                   f"Missing recommended patterns: {', '.join(missing)}"))
            except Exception:
                pass

    # Format output
    lines = [f"Env Audit: {scan_dir}", "=" * 50]

    if not issues:
        lines.append("")
        lines.append("No issues found. Directory looks clean.")
        return "\n".join(lines)

    # Group by severity
    critical = [i for i in issues if i[0] == "CRITICAL"]
    warnings = [i for i in issues if i[0] == "WARNING"]
    info = [i for i in issues if i[0] == "INFO"]

    if critical:
        lines.append("")
        lines.append(f"CRITICAL ({len(critical)}):")
        for sev, cat, fpath, detail in critical:
            lines.append(f"  [!] {fpath}: {detail}")

    if warnings:
        lines.append("")
        lines.append(f"WARNINGS ({len(warnings)}):")
        for sev, cat, fpath, detail in warnings:
            lines.append(f"  [*] {fpath}: {detail}")

    if info:
        lines.append("")
        lines.append(f"INFO ({len(info)}):")
        for sev, cat, fpath, detail in info:
            lines.append(f"  [i] {fpath}: {detail}")

    lines.append("")
    lines.append(f"Total: {len(critical)} critical, {len(warnings)} warnings, {len(info)} info")

    return "\n".join(lines)

	# SPDX-License-Identifier: MIT
	"""
	Env Audit Plugin for TrashClaw

	Scan a directory for common security misconfigurations:
	- .env files committed to git
	- Exposed secrets (API keys, passwords, tokens in source)
	- World-readable private keys
	- Sensitive files without .gitignore coverage

	Plugin contract:
	TOOL_DEF = dict with name, description, parameters (OpenAI function schema)
	run(**kwargs) -> str (called with the parsed arguments)
	"""

	import os
	import re
	import stat

	TOOL_DEF = {
	"name": "env_audit",
	"description": "Scan a directory for common security misconfigurations: .env in git, exposed secrets, world-readable keys, missing .gitignore entries.",
	"parameters": {
	"type": "object",
	"properties": {
	"path": {
	"type": "string",
	"description": "Directory to scan (default: current directory)"
	},
	"deep": {
	"type": "boolean",
	"description": "If true, scan file contents for secrets (slower). Default: true."
	}
	},
	"required": []
	}
	}

	# Filenames that should never be committed
	SENSITIVE_FILES = [
	".env",
	".env.local",
	".env.production",
	".env.staging",
	".env.development",
	"credentials.json",
	"service-account.json",
	"secrets.yaml",
	"secrets.yml",
	".netrc",
	".npmrc", # can contain tokens
	".pypirc", # can contain tokens
	]

	# Private key file patterns
	KEY_PATTERNS = [
	"*.pem",
	"*.key",
	"*.p12",
	"*.pfx",
	"id_rsa",
	"id_ed25519",
	"id_ecdsa",
	"id_dsa",
	]

	# Regex patterns for secrets in file contents
	SECRET_PATTERNS = [
	(r'(?i)(?:api[_-]?key\|apikey)\s[:=]\s["\']?[A-Za-z0-9_\-]{16,}', "API key"),
	(r'(?i)(?:secret\|password\|passwd\|pwd)\s[:=]\s["\']?[^\s"\']{8,}', "Password/Secret"),
	(r'(?i)(?:aws_access_key_id\|aws_secret_access_key)\s[:=]\s["\']?[A-Za-z0-9/+=]{16,}', "AWS credential"),
	(r'(?i)bearer\s+[A-Za-z0-9_\-\.]{20,}', "Bearer token"),
	(r'ghp_[A-Za-z0-9]{36}', "GitHub PAT"),
	(r'sk-[A-Za-z0-9]{20,}', "OpenAI/Stripe secret key"),
	(r'(?i)(?:private[_-]?key)\s[:=]\s["\']?[A-Za-z0-9_\-/+=]{16,}', "Private key value"),
	(r'-----BEGIN (?:RSA \|EC \|DSA \|OPENSSH )?PRIVATE KEY-----', "Private key block"),
	]

	# Extensions to scan for secrets
	SCANNABLE_EXTENSIONS = {
	".py", ".js", ".ts", ".jsx", ".tsx", ".rb", ".go", ".java",
	".yaml", ".yml", ".json", ".toml", ".ini", ".cfg", ".conf",
	".sh", ".bash", ".zsh", ".env", ".properties", ".xml",
	".tf", ".hcl", ".dockerfile",
	}

	# Directories to skip
	SKIP_DIRS = {
	".git", "node_modules", "__pycache__", ".venv", "venv",
	"env", ".tox", ".mypy_cache", ".pytest_cache", "dist",
	"build", ".eggs", "*.egg-info",
	}

	MAX_FILE_SIZE = 512 * 1024 # 512KB max per file for content scanning


	def _is_git_tracked(filepath: str, repo_root: str) -> bool:
	"""Check if a file is tracked by git (would be committed)."""
	try:
	import subprocess
	result = subprocess.run(
	["git", "ls-files", "--error-unmatch", filepath],
	cwd=repo_root, capture_output=True, text=True, timeout=5
	)
	return result.returncode == 0
	except Exception:
	return False # Can't determine; assume not tracked


	def _is_gitignored(filepath: str, repo_root: str) -> bool:
	"""Check if a file is covered by .gitignore."""
	try:
	import subprocess
	result = subprocess.run(
	["git", "check-ignore", "-q", filepath],
	cwd=repo_root, capture_output=True, text=True, timeout=5
	)
	return result.returncode == 0
	except Exception:
	return False


	def _check_permissions(filepath: str) -> str:
	"""Check if file has overly permissive permissions (world-readable)."""
	try:
	st = os.stat(filepath)
	mode = st.st_mode
	if mode & stat.S_IROTH: # World-readable
	return f"{oct(mode & 0o777)} (world-readable)"
	except Exception:
	pass
	return ""


	def _scan_file_contents(filepath: str) -> list:
	"""Scan a file for secret patterns. Returns list of (line_no, pattern_name, snippet)."""
	findings = []
	try:
	size = os.path.getsize(filepath)
	if size > MAX_FILE_SIZE:
	return findings
	with open(filepath, "r", errors="replace") as f:
	for line_no, line in enumerate(f, 1):
	for pattern, name in SECRET_PATTERNS:
	if re.search(pattern, line):
	# Truncate the line for display
	snippet = line.strip()
	if len(snippet) > 100:
	snippet = snippet[:97] + "..."
	findings.append((line_no, name, snippet))
	break # One finding per line is enough
	except Exception:
	pass
	return findings


	def _find_git_root(path: str) -> str:
	"""Walk up to find .git directory."""
	current = os.path.abspath(path)
	while current != os.path.dirname(current):
	if os.path.isdir(os.path.join(current, ".git")):
	return current
	current = os.path.dirname(current)
	return ""


	def run(path: str = "", deep: bool = True, **kwargs) -> str:
	scan_dir = os.path.abspath(path or os.getcwd())
	if not os.path.isdir(scan_dir):
	return f"Error: not a directory: {scan_dir}"

	git_root = _find_git_root(scan_dir)
	has_git = bool(git_root)

	issues = [] # (severity, category, filepath, detail)

	# Walk the directory tree
	for root, dirs, files in os.walk(scan_dir):
	# Skip hidden/vendor directories
	dirs[:] = [d for d in dirs if d not in SKIP_DIRS and not d.startswith(".")]

	rel_root = os.path.relpath(root, scan_dir)

	for fname in files:
	filepath = os.path.join(root, fname)
	rel_path = os.path.relpath(filepath, scan_dir)

	# Check 1: Sensitive files present
	if fname.lower() in [s.lower() for s in SENSITIVE_FILES]:
	if has_git:
	tracked = _is_git_tracked(rel_path, git_root)
	ignored = _is_gitignored(rel_path, git_root)
	if tracked:
	issues.append(("CRITICAL", "sensitive_file_tracked", rel_path,
	f"Sensitive file is tracked by git!"))
	elif not ignored:
	issues.append(("WARNING", "sensitive_file_not_ignored", rel_path,
	f"Sensitive file exists but is NOT in .gitignore"))
	else:
	issues.append(("WARNING", "sensitive_file_present", rel_path,
	f"Sensitive file found (no git repo to check ignore rules)"))

	# Check 2: Private key files
	for pat in KEY_PATTERNS:
	if pat.startswith("*"):
	if fname.endswith(pat[1:]):
	perm = _check_permissions(filepath)
	if perm:
	issues.append(("WARNING", "key_permissions", rel_path,
	f"Private key is {perm}"))
	if has_git and _is_git_tracked(rel_path, git_root):
	issues.append(("CRITICAL", "key_tracked", rel_path,
	f"Private key tracked by git!"))
	break
	elif fname == pat:
	perm = _check_permissions(filepath)
	if perm:
	issues.append(("WARNING", "key_permissions", rel_path,
	f"Private key is {perm}"))
	if has_git and _is_git_tracked(rel_path, git_root):
	issues.append(("CRITICAL", "key_tracked", rel_path,
	f"Private key tracked by git!"))
	break

	# Check 3: Content scanning for secrets
	if deep:
	_, ext = os.path.splitext(fname)
	if ext.lower() in SCANNABLE_EXTENSIONS or fname.lower() in (".env",):
	findings = _scan_file_contents(filepath)
	for line_no, pattern_name, snippet in findings:
	issues.append(("WARNING", "secret_in_source", rel_path,
	f"Line {line_no}: possible {pattern_name}"))

	# Check 4: .gitignore coverage
	if has_git:
	gitignore_path = os.path.join(git_root, ".gitignore")
	if not os.path.exists(gitignore_path):
	issues.append(("INFO", "no_gitignore", ".gitignore",
	"No .gitignore file found in repository"))
	else:
	try:
	with open(gitignore_path, "r") as f:
	gitignore_content = f.read()
	recommended = [".env", ".pem", ".key", "id_rsa", "credentials.json"]
	missing = [p for p in recommended if p not in gitignore_content]
	if missing:
	issues.append(("INFO", "gitignore_incomplete", ".gitignore",
	f"Missing recommended patterns: {', '.join(missing)}"))
	except Exception:
	pass

	# Format output
	lines = [f"Env Audit: {scan_dir}", "=" * 50]

	if not issues:
	lines.append("")
	lines.append("No issues found. Directory looks clean.")
	return "\n".join(lines)

	# Group by severity
	critical = [i for i in issues if i[0] == "CRITICAL"]
	warnings = [i for i in issues if i[0] == "WARNING"]
	info = [i for i in issues if i[0] == "INFO"]

	if critical:
	lines.append("")
	lines.append(f"CRITICAL ({len(critical)}):")
	for sev, cat, fpath, detail in critical:
	lines.append(f" [!] {fpath}: {detail}")

	if warnings:
	lines.append("")
	lines.append(f"WARNINGS ({len(warnings)}):")
	for sev, cat, fpath, detail in warnings:
	lines.append(f" [*] {fpath}: {detail}")

	if info:
	lines.append("")
	lines.append(f"INFO ({len(info)}):")
	for sev, cat, fpath, detail in info:
	lines.append(f" [i] {fpath}: {detail}")

	lines.append("")
	lines.append(f"Total: {len(critical)} critical, {len(warnings)} warnings, {len(info)} info")

	return "\n".join(lines)