#!/usr/bin/env python3
"""Privilege escalation detection agent for Windows and Linux endpoints.

Detects token manipulation, UAC bypass, sudo abuse, kernel exploits, and
unquoted service paths by analyzing process creation and security logs.
"""

import argparse
import json
import re
from datetime import datetime

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

WINDOWS_PRIVESC_PATTERNS = [
    (r"eventvwr\.exe|fodhelper\.exe|computerdefaults\.exe", "T1548.002", "CRITICAL", "UAC Bypass"),
    (r"whoami\s+/priv", "T1033", "MEDIUM", "Privilege enumeration"),
    (r"sc\s+(config|create).*binpath", "T1543.003", "HIGH", "Service binary modification"),
    (r"potato.*exploit|juicypotato|sweetpotato|godpotato", "T1134.001", "CRITICAL", "Token impersonation exploit"),
    (r"printspoofer|efspotato", "T1134.001", "CRITICAL", "Named pipe impersonation"),
    (r"schtasks.*\/ru.*system", "T1053.005", "HIGH", "Scheduled task as SYSTEM"),
    (r"reg\s+add.*ImagePath", "T1574.011", "HIGH", "Service registry modification"),
]

LINUX_PRIVESC_PATTERNS = [
    (r"sudo\s+-l|sudo\s+--list", "T1548.003", "MEDIUM", "Sudo enumeration"),
    (r"find.*-perm.*4000|find.*-perm.*/u=s", "T1548.001", "MEDIUM", "SUID binary search"),
    (r"chmod\s+[u+]?s|chmod\s+4\d{3}", "T1548.001", "HIGH", "SUID bit set"),
    (r"linpeas|linenum|linux-exploit-suggester", "T1046", "HIGH", "Privesc enumeration tool"),
    (r"pkexec|CVE-2021-4034|pwnkit", "T1068", "CRITICAL", "Kernel exploit"),
    (r"dirty.*pipe|CVE-2022-0847", "T1068", "CRITICAL", "Kernel exploit"),
]


def analyze_evtx(filepath):
    if evtx is None:
        return {"error": "python-evtx not installed: pip install python-evtx"}
    findings = []
    with evtx.Evtx(filepath) as log:
        for record in log.records():
            xml = record.xml()
            event_id_match = re.search(r'<EventID[^>]*>(\d+)</EventID>', xml)
            if not event_id_match:
                continue
            event_id = int(event_id_match.group(1))
            if event_id not in (1, 4688, 4672):
                continue
            cmdline = re.search(r'<Data Name="CommandLine">([^<]+)', xml)
            image = re.search(r'<Data Name="Image">([^<]+)', xml)
            time_match = re.search(r'SystemTime="([^"]+)"', xml)
            cmd = cmdline.group(1) if cmdline else ""
            proc = image.group(1) if image else ""
            text = f"{cmd} {proc}"
            for pattern, mitre, severity, desc in WINDOWS_PRIVESC_PATTERNS:
                if re.search(pattern, text, re.IGNORECASE):
                    findings.append({
                        "event_id": event_id,
                        "timestamp": time_match.group(1) if time_match else "",
                        "command": cmd[:200], "technique": desc,
                        "mitre": mitre, "severity": severity,
                    })
            if event_id == 4672:
                privs = re.search(r'<Data Name="PrivilegeList">([^<]+)', xml)
                if privs and "SeDebugPrivilege" in privs.group(1):
                    findings.append({
                        "event_id": 4672,
                        "timestamp": time_match.group(1) if time_match else "",
                        "technique": "SeDebugPrivilege assigned",
                        "mitre": "T1134", "severity": "HIGH",
                    })
    return findings


def analyze_text_log(filepath):
    findings = []
    all_patterns = WINDOWS_PRIVESC_PATTERNS + LINUX_PRIVESC_PATTERNS
    with open(filepath, "r", encoding="utf-8", errors="replace") as f:
        for num, line in enumerate(f, 1):
            for pattern, mitre, severity, desc in all_patterns:
                if re.search(pattern, line, re.IGNORECASE):
                    findings.append({
                        "line": num, "technique": desc,
                        "mitre": mitre, "severity": severity,
                        "excerpt": line.strip()[:200],
                    })
    return findings


def main():
    parser = argparse.ArgumentParser(description="Privilege Escalation Detector")
    parser.add_argument("--evtx-file", help="Sysmon or Security EVTX file")
    parser.add_argument("--text-log", help="Text log to scan")
    args = parser.parse_args()
    results = {"timestamp": datetime.utcnow().isoformat() + "Z", "findings": []}
    if args.evtx_file:
        r = analyze_evtx(args.evtx_file)
        if isinstance(r, dict):
            results.update(r)
        else:
            results["findings"].extend(r)
    if args.text_log:
        results["findings"].extend(analyze_text_log(args.text_log))
    results["total_findings"] = len(results["findings"])
    print(json.dumps(results, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Privilege Escalation Detection - Analyzes logs for T1134 indicators."""

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

DETECTION_PATTERNS = [
    r'potato',
    r'PrintSpoofer',
    r'JuicyPotato',
    r'fodhelper',
    r'eventvwr',
    r'sdclt',
    r'ComputerDefaults',
]

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": "T1134",
        "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"[*] Privilege Escalation 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 = "detecting_privilege_"
    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"# Privilege Escalation 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="Privilege Escalation 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="./detecting_privi_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()