hermes-agent/scripts/import_cybersecurity_skills.py

#!/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()