#!/usr/bin/env python3
"""Agent for detecting service account abuse in Active Directory and cloud environments."""

import argparse
import json
import subprocess
from datetime import datetime, timezone


def query_ad_service_accounts():
    """Query Active Directory for service accounts via PowerShell."""
    ps_cmd = (
        "Get-ADUser -Filter {ServicePrincipalName -ne '$null'} "
        "-Properties ServicePrincipalName,LastLogonDate,Enabled,PasswordLastSet,"
        "PasswordNeverExpires,AdminCount,MemberOf "
        "| Select-Object SamAccountName,Enabled,LastLogonDate,PasswordLastSet,"
        "PasswordNeverExpires,AdminCount,@{N='SPNCount';E={$_.ServicePrincipalName.Count}} "
        "| ConvertTo-Json"
    )
    try:
        result = subprocess.check_output(
            ["powershell", "-NoProfile", "-Command", ps_cmd],
            text=True, errors="replace", timeout=30
        )
        data = json.loads(result) if result.strip().startswith(("[", "{")) else []
        return data if isinstance(data, list) else [data]
    except (subprocess.SubprocessError, json.JSONDecodeError):
        return []


def check_interactive_logons(days=7):
    """Find service accounts with interactive logon events (type 2/10)."""
    ps_cmd = (
        f"Get-WinEvent -FilterHashtable @{{LogName='Security';Id=4624;"
        f"StartTime=(Get-Date).AddDays(-{days})}} "
        "| Where-Object {($_.Properties[8].Value -eq 2 -or $_.Properties[8].Value -eq 10) "
        "-and $_.Properties[5].Value -match 'svc_|service'} "
        "| Select-Object TimeCreated,@{N='Account';E={$_.Properties[5].Value}},"
        "@{N='LogonType';E={$_.Properties[8].Value}},"
        "@{N='SourceIP';E={$_.Properties[18].Value}} "
        "| ConvertTo-Json"
    )
    try:
        result = subprocess.check_output(
            ["powershell", "-NoProfile", "-Command", ps_cmd],
            text=True, errors="replace", timeout=30
        )
        data = json.loads(result) if result.strip() else []
        return data if isinstance(data, list) else [data]
    except (subprocess.SubprocessError, json.JSONDecodeError):
        return []


def analyze_logon_patterns(events):
    """Detect anomalous logon patterns for service accounts."""
    anomalies = []
    account_sources = {}
    for evt in events:
        acct = evt.get("Account", "unknown")
        src = evt.get("SourceIP", "unknown")
        account_sources.setdefault(acct, []).append(src)

    for acct, sources in account_sources.items():
        unique = set(sources)
        if len(unique) > 3:
            anomalies.append({
                "account": acct,
                "issue": "Service account logged in from multiple sources",
                "source_count": len(unique),
                "sources": list(unique)[:10],
            })
    return anomalies


def check_account_risks(accounts):
    """Identify risky service account configurations."""
    risks = []
    for acct in accounts:
        name = acct.get("SamAccountName", "unknown")
        issues = []
        if acct.get("PasswordNeverExpires"):
            issues.append("Password never expires")
        if acct.get("AdminCount") == 1:
            issues.append("Has AdminCount=1 (privileged)")
        if acct.get("Enabled") is False:
            issues.append("Account disabled but has SPNs")
        pw_set = acct.get("PasswordLastSet")
        if pw_set and isinstance(pw_set, str):
            try:
                pw_date = datetime.fromisoformat(pw_set.replace("Z", "+00:00"))
                age = (datetime.now(timezone.utc) - pw_date).days
                if age > 365:
                    issues.append(f"Password {age} days old")
            except ValueError:
                pass
        if issues:
            risks.append({"account": name, "issues": issues, "risk_count": len(issues)})
    return risks


def main():
    parser = argparse.ArgumentParser(
        description="Detect service account abuse in AD environments"
    )
    parser.add_argument("--days", type=int, default=7, help="Lookback days for logon events")
    parser.add_argument("--output", "-o", help="Output JSON report path")
    parser.add_argument("--verbose", "-v", action="store_true")
    args = parser.parse_args()

    print("[*] Service Account Abuse Detection Agent")
    report = {"timestamp": datetime.now(timezone.utc).isoformat(), "findings": {}}

    accounts = query_ad_service_accounts()
    report["findings"]["service_accounts"] = len(accounts)
    print(f"[*] Found {len(accounts)} service accounts with SPNs")

    risks = check_account_risks(accounts)
    report["findings"]["risky_accounts"] = risks
    print(f"[*] Risky accounts: {len(risks)}")

    logons = check_interactive_logons(args.days)
    report["findings"]["interactive_logons"] = len(logons)
    anomalies = analyze_logon_patterns(logons)
    report["findings"]["logon_anomalies"] = anomalies
    print(f"[*] Logon anomalies: {len(anomalies)}")

    total_issues = len(risks) + len(anomalies)
    report["risk_level"] = "CRITICAL" if total_issues >= 5 else "HIGH" if total_issues >= 3 else "MEDIUM" if total_issues > 0 else "LOW"

    if args.output:
        with open(args.output, "w") as f:
            json.dump(report, f, indent=2)
        print(f"[*] Report saved to {args.output}")
    else:
        print(json.dumps(report, indent=2))


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Service Account Abuse Detection - Analyzes logs for T1078.002 indicators."""

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

DETECTION_PATTERNS = [
    r'svc_.*Logon_Type=(2|10)',
    r'service.*interactive',
]

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": "T1078.002",
        "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"[*] Service Account Abuse 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_service_ac"
    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"# Service Account Abuse 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="Service Account Abuse 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_servi_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()