Anthropic-Cybersecurity-Skills/skills/analyzing-linux-kernel-rootkits/scripts/agent.py

#!/usr/bin/env python3
"""Linux Kernel Rootkit Detection Agent - analyzes memory dumps with Volatility3 and live system with rkhunter."""

import json
import argparse
import logging
import subprocess
import os
from datetime import datetime

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def run_vol3_plugin(memory_dump, plugin, isf_url=None):
    """Run a Volatility3 Linux plugin and return parsed output."""
    cmd = ["vol", "-f", memory_dump, plugin, "-r", "json"]
    if isf_url:
        cmd.extend(["--isf", isf_url])
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=600)
    try:
        return json.loads(result.stdout) if result.stdout else []
    except json.JSONDecodeError:
        logger.error("Volatility3 %s output parse failed", plugin)
        return []


def check_syscall_hooks(memory_dump, isf_url=None):
    """Detect hooked system calls using linux.check_syscall."""
    results = run_vol3_plugin(memory_dump, "linux.check_syscall.Check_syscall", isf_url)
    hooked = []
    for entry in results:
        row = entry.get("__children", [entry]) if isinstance(entry, dict) else [entry]
        for item in row:
            symbol = item.get("Symbol", item.get("symbol", ""))
            module = item.get("Module", item.get("module", ""))
            if module and module != "kernel":
                hooked.append({
                    "syscall_number": item.get("Index", item.get("index", "")),
                    "expected_handler": symbol,
                    "actual_module": module,
                    "severity": "critical",
                    "indicator": "syscall_hook",
                })
    return hooked


def detect_hidden_modules(memory_dump, isf_url=None):
    """Detect hidden kernel modules using cross-view analysis."""
    lsmod_results = run_vol3_plugin(memory_dump, "linux.lsmod.Lsmod", isf_url)
    hidden_results = run_vol3_plugin(memory_dump, "linux.hidden_modules.Hidden_modules", isf_url)
    lsmod_names = set()
    for entry in lsmod_results:
        name = entry.get("Name", entry.get("name", ""))
        if name:
            lsmod_names.add(name)
    hidden = []
    for entry in hidden_results:
        name = entry.get("Name", entry.get("name", ""))
        if name:
            hidden.append({
                "module_name": name,
                "in_lsmod": name in lsmod_names,
                "severity": "critical",
                "indicator": "hidden_kernel_module",
                "detail": f"Module '{name}' hidden from standard listing",
            })
    return hidden


def check_idt_hooks(memory_dump, isf_url=None):
    """Check Interrupt Descriptor Table for hooks."""
    results = run_vol3_plugin(memory_dump, "linux.check_idt.Check_idt", isf_url)
    hooked = []
    for entry in results:
        module = entry.get("Module", entry.get("module", ""))
        if module and module != "kernel":
            hooked.append({
                "interrupt": entry.get("Index", ""),
                "handler_module": module,
                "severity": "critical",
                "indicator": "idt_hook",
            })
    return hooked


def run_rkhunter():
    """Run rkhunter rootkit scanner on live system."""
    cmd = ["rkhunter", "--check", "--skip-keypress", "--report-warnings-only", "--nocolors"]
    result = subprocess.run(cmd, capture_output=True, text=True, timeout=300)
    findings = []
    for line in result.stdout.split("\n"):
        line = line.strip()
        if "Warning:" in line or "[ Warning ]" in line:
            findings.append({
                "tool": "rkhunter",
                "finding": line.replace("Warning:", "").strip(),
                "severity": "high",
            })
    return findings


def check_proc_sys_discrepancy():
    """Compare /proc/modules with /sys/module for hidden modules."""
    findings = []
    proc_modules = set()
    sys_modules = set()
    try:
        with open("/proc/modules") as f:
            for line in f:
                proc_modules.add(line.split()[0])
    except (FileNotFoundError, PermissionError):
        return findings
    try:
        sys_modules = set(os.listdir("/sys/module"))
    except (FileNotFoundError, PermissionError):
        return findings
    only_in_sys = sys_modules - proc_modules
    for mod in only_in_sys:
        if not os.path.exists(f"/sys/module/{mod}/initstate"):
            continue
        findings.append({
            "module": mod, "indicator": "proc_sys_discrepancy",
            "severity": "high",
            "detail": f"Module '{mod}' in /sys/module but missing from /proc/modules",
        })
    return findings


def generate_report(syscall_hooks, hidden_mods, idt_hooks, rkhunter_findings, proc_findings, source):
    all_findings = syscall_hooks + hidden_mods + idt_hooks + rkhunter_findings + proc_findings
    critical = sum(1 for f in all_findings if f.get("severity") == "critical")
    return {
        "timestamp": datetime.utcnow().isoformat(),
        "analysis_source": source,
        "syscall_hooks": syscall_hooks,
        "hidden_modules": hidden_mods,
        "idt_hooks": idt_hooks,
        "rkhunter_warnings": rkhunter_findings,
        "proc_sys_discrepancies": proc_findings,
        "total_findings": len(all_findings),
        "critical_findings": critical,
        "rootkit_detected": critical > 0,
    }


def main():
    parser = argparse.ArgumentParser(description="Linux Kernel Rootkit Detection Agent")
    parser.add_argument("--memory-dump", help="Path to Linux memory dump for Volatility3 analysis")
    parser.add_argument("--isf-url", help="Volatility3 ISF symbol table URL")
    parser.add_argument("--live-scan", action="store_true", help="Run rkhunter + /proc analysis on live system")
    parser.add_argument("--output", default="rootkit_detection_report.json")
    args = parser.parse_args()

    syscall_hooks, hidden_mods, idt_hooks = [], [], []
    rkhunter_findings, proc_findings = [], []
    source = "none"
    if args.memory_dump:
        source = f"memory_dump:{args.memory_dump}"
        syscall_hooks = check_syscall_hooks(args.memory_dump, args.isf_url)
        hidden_mods = detect_hidden_modules(args.memory_dump, args.isf_url)
        idt_hooks = check_idt_hooks(args.memory_dump, args.isf_url)
    if args.live_scan:
        source = "live_system" if source == "none" else source + "+live_system"
        rkhunter_findings = run_rkhunter()
        proc_findings = check_proc_sys_discrepancy()
    report = generate_report(syscall_hooks, hidden_mods, idt_hooks, rkhunter_findings, proc_findings, source)
    with open(args.output, "w") as f:
        json.dump(report, f, indent=2, default=str)
    logger.info("Rootkit scan: %d findings (%d critical), rootkit detected: %s",
                report["total_findings"], report["critical_findings"], report["rootkit_detected"])
    print(json.dumps(report, indent=2, default=str))


if __name__ == "__main__":
    main()