#!/usr/bin/env python3
"""Agent for hunting shadow copy deletion activity indicating ransomware or anti-forensics."""

import json
import argparse
import re
import xml.etree.ElementTree as ET
from datetime import datetime
from pathlib import Path

try:
    import Evtx.Evtx as evtx
except ImportError:
    evtx = None


SHADOW_PATTERNS = [
    r"vssadmin\s+delete\s+shadows",
    r"vssadmin\.exe.*delete.*shadows",
    r"wmic\s+shadowcopy\s+delete",
    r"Get-WmiObject\s+Win32_ShadowCopy.*Delete",
    r"gwmi\s+Win32_ShadowCopy.*Remove",
    r"bcdedit.*recoveryenabled.*no",
    r"bcdedit.*/set.*bootstatuspolicy\s+ignoreallfailures",
    r"wbadmin\s+delete\s+catalog",
    r"Win32_ShadowCopy.*\.Delete",
    r"powershell.*shadowcopy.*delete",
]

RECOVERY_DISABLE_PATTERNS = [
    r"bcdedit\s+/set\s+\{default\}\s+recoveryenabled\s+no",
    r"reagentc\s+/disable",
    r"vssadmin\s+resize\s+shadowstorage.*maxsize=",
]


def parse_evtx_file(evtx_path):
    """Parse Windows EVTX file for shadow copy deletion events."""
    if evtx is None:
        return []
    events = []
    with evtx.Evtx(evtx_path) as log:
        for record in log.records():
            xml_str = record.xml()
            try:
                root = ET.fromstring(xml_str)
                ns = {"ns": "http://schemas.microsoft.com/win/2004/08/events/event"}
                event_id_el = root.find(".//ns:EventID", ns)
                if event_id_el is None:
                    continue
                event_id = int(event_id_el.text)
                if event_id in (1, 4688, 4698):
                    data_elements = root.findall(".//ns:Data", ns)
                    event_data = {}
                    for d in data_elements:
                        name = d.get("Name", "")
                        event_data[name] = d.text or ""
                    events.append({
                        "event_id": event_id,
                        "timestamp": record.timestamp().isoformat(),
                        "data": event_data,
                    })
            except ET.ParseError:
                continue
    return events


def scan_command_line(cmd_line):
    """Check a command line string against shadow copy deletion patterns."""
    findings = []
    for pattern in SHADOW_PATTERNS:
        if re.search(pattern, cmd_line, re.IGNORECASE):
            findings.append({"pattern": pattern, "severity": "CRITICAL", "category": "shadow_copy_deletion"})
    for pattern in RECOVERY_DISABLE_PATTERNS:
        if re.search(pattern, cmd_line, re.IGNORECASE):
            findings.append({"pattern": pattern, "severity": "HIGH", "category": "recovery_disable"})
    return findings


def hunt_evtx(evtx_path):
    """Hunt for shadow copy deletion in EVTX logs."""
    events = parse_evtx_file(evtx_path)
    results = []
    for event in events:
        cmd = event["data"].get("CommandLine", "") or event["data"].get("TaskContent", "")
        if not cmd:
            cmd = " ".join(event["data"].values())
        matches = scan_command_line(cmd)
        if matches:
            results.append({
                "timestamp": event["timestamp"],
                "event_id": event["event_id"],
                "command_line": cmd[:500],
                "user": event["data"].get("SubjectUserName", event["data"].get("User", "")),
                "computer": event["data"].get("Computer", ""),
                "findings": matches,
            })
    return results


def scan_sysmon_json(log_path):
    """Scan JSON-exported Sysmon logs for shadow copy deletion."""
    results = []
    with open(log_path) as f:
        for line in f:
            try:
                entry = json.loads(line)
            except json.JSONDecodeError:
                continue
            cmd = entry.get("CommandLine", entry.get("command_line", ""))
            image = entry.get("Image", entry.get("process_name", ""))
            matches = scan_command_line(cmd)
            if matches:
                results.append({
                    "timestamp": entry.get("UtcTime", entry.get("timestamp", "")),
                    "image": image,
                    "command_line": cmd[:500],
                    "parent_image": entry.get("ParentImage", ""),
                    "user": entry.get("User", ""),
                    "hostname": entry.get("Computer", entry.get("hostname", "")),
                    "findings": matches,
                })
    return results


def generate_sigma_rule():
    """Generate a Sigma detection rule for shadow copy deletion."""
    return {
        "title": "Shadow Copy Deletion via Vssadmin or WMIC",
        "id": "faa6e1e2-5b4c-4e1a-bb2a-5c1f3e5e3f0a",
        "status": "production",
        "level": "critical",
        "logsource": {"category": "process_creation", "product": "windows"},
        "detection": {
            "selection_vssadmin": {
                "Image|endswith": "\\vssadmin.exe",
                "CommandLine|contains|all": ["delete", "shadows"],
            },
            "selection_wmic": {
                "Image|endswith": "\\wmic.exe",
                "CommandLine|contains|all": ["shadowcopy", "delete"],
            },
            "selection_powershell": {
                "Image|endswith": ["\\powershell.exe", "\\pwsh.exe"],
                "CommandLine|contains": "Win32_ShadowCopy",
            },
            "condition": "selection_vssadmin or selection_wmic or selection_powershell",
        },
        "tags": ["attack.impact", "attack.t1490"],
    }


def main():
    parser = argparse.ArgumentParser(description="Shadow Copy Deletion Hunter")
    parser.add_argument("--evtx", help="Path to EVTX log file")
    parser.add_argument("--json-log", help="Path to JSON Sysmon log")
    parser.add_argument("--output", default="shadow_copy_hunt_report.json")
    parser.add_argument("--action", choices=["hunt_evtx", "hunt_json", "sigma", "full"],
                        default="full")
    args = parser.parse_args()

    report = {"generated_at": datetime.utcnow().isoformat(), "findings": {}}

    if args.action in ("hunt_evtx", "full") and args.evtx:
        results = hunt_evtx(args.evtx)
        report["findings"]["evtx_hits"] = results
        print(f"[+] EVTX shadow copy deletion events: {len(results)}")

    if args.action in ("hunt_json", "full") and args.json_log:
        results = scan_sysmon_json(args.json_log)
        report["findings"]["sysmon_hits"] = results
        print(f"[+] Sysmon JSON shadow copy hits: {len(results)}")

    if args.action in ("sigma", "full"):
        rule = generate_sigma_rule()
        report["findings"]["sigma_rule"] = rule
        print("[+] Sigma rule generated")

    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    print(f"[+] Report saved to {args.output}")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Shadow Copy Deletion Detection - Analyzes logs for T1490 indicators."""

import json, csv, argparse, datetime, re
from collections import defaultdict
from pathlib import Path

DETECTION_PATTERNS = [
    r'vssadmin.*delete',
    r'wmic.*shadowcopy.*delete',
    r'bcdedit.*recoveryenabled.*no',
    r'wbadmin.*delete',
    r'Get-WmiObject.*Win32_ShadowCopy.*Delete',
]

def parse_logs(path):
    p = Path(path)
    if p.suffix == ".json":
        with open(p, encoding="utf-8") as f:
            data = json.load(f)
            return data if isinstance(data, list) else data.get("events", [])
    elif p.suffix == ".csv":
        with open(p, encoding="utf-8-sig") as f:
            return [dict(r) for r in csv.DictReader(f)]
    return []

def analyze_event(event):
    cmd = event.get("CommandLine", event.get("command_line", event.get("ProcessCommandLine", "")))
    content = event.get("Task_Content", event.get("Parameters", event.get("RawEventData", "")))
    search_text = f"{cmd} {content}"
    risk = 0
    indicators = []
    for pattern in DETECTION_PATTERNS:
        if re.search(pattern, search_text, re.IGNORECASE):
            risk += 25
            indicators.append(f"Pattern match: {pattern}")
    if not indicators:
        return None
    risk = min(risk, 100)
    return {
        "technique": "T1490",
        "command_line": cmd[:500] if cmd else content[:500],
        "hostname": event.get("Computer", event.get("DeviceName", event.get("hostname", "unknown"))),
        "user": event.get("User", event.get("AccountName", event.get("UserId", "unknown"))),
        "timestamp": event.get("_time", event.get("timestamp", event.get("UtcTime", event.get("Timestamp", "")))),
        "risk_score": risk,
        "risk_level": "CRITICAL" if risk >= 75 else "HIGH" if risk >= 50 else "MEDIUM" if risk >= 25 else "LOW",
        "indicators": indicators,
    }

def run_hunt(input_path, output_dir):
    print(f"[*] Shadow Copy Deletion Hunt - {datetime.datetime.now().isoformat()}")
    events = parse_logs(input_path)
    findings = [f for f in (analyze_event(e) for e in events) if f]
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    slug = "hunting_for_shadow_c"
    with open(Path(output_dir) / f"{slug}_findings.json", "w", encoding="utf-8") as f:
        json.dump({"hunt_id": f"TH-{datetime.date.today()}", "total_events": len(events), "findings": findings}, f, indent=2)
    with open(Path(output_dir) / "hunt_report.md", "w", encoding="utf-8") as f:
        f.write(f"# Shadow Copy Deletion Hunt Report\n\n")
        f.write(f"**Date**: {datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")
        f.write(f"**Findings**: {len(findings)}\n\n")
        for finding in sorted(findings, key=lambda x: x["risk_score"], reverse=True)[:20]:
            f.write(f"### [{finding['risk_level']}] {finding['technique']}\n")
            f.write(f"- **Host**: {finding['hostname']}\n")
            f.write(f"- **Indicators**: {', '.join(finding['indicators'])}\n\n")
    print(f"[+] {len(findings)} findings written to {output_dir}")

def main():
    p = argparse.ArgumentParser(description="Shadow Copy Deletion Detection")
    sp = p.add_subparsers(dest="cmd")
    h = sp.add_parser("hunt"); h.add_argument("--input", "-i", required=True); h.add_argument("--output", "-o", default="./hunting_for_sha_output")
    sp.add_parser("queries")
    args = p.parse_args()
    if args.cmd == "hunt": run_hunt(args.input, args.output)
    elif args.cmd == "queries":
        print("=== Detection Queries ===")
        print("See references/workflows.md for platform-specific queries")
    else: p.print_help()

if __name__ == "__main__": main()