Active Directory Security Auditor

#!/usr/bin/env python3
"""
Active Directory Security Auditor

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/active-directory-auditor

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "ldap3>=2.9" "impacket>=0.11" "colorama>=0.4"
"""

import argparse
import json
import os
import sys
import datetime
from typing import Optional

try:
    from ldap3 import Server, Connection, ALL, NTLM, SUBTREE
    from ldap3.core.exceptions import LDAPException
except ImportError:
    print("[!] Missing dependency: pip install ldap3")
    sys.exit(1)

try:
    from colorama import init, Fore, Style
    init(autoreset=True)
except ImportError:
    class Fore:
        RED = YELLOW = GREEN = CYAN = WHITE = RESET = ""
    class Style:
        BRIGHT = RESET_ALL = ""

UAC_DONT_REQUIRE_PREAUTH     = 0x400000
UAC_PASSWORD_NEVER_EXPIRES   = 0x10000
UAC_ENCRYPTED_TEXT_PWD       = 0x80
UAC_ACCOUNT_DISABLED         = 0x2

PRIVILEGED_GROUPS = [
    "Domain Admins", "Enterprise Admins", "Schema Admins",
    "Backup Operators", "Account Operators", "Print Operators", "Server Operators",
]


def connect_ldap(dc_ip: str, domain: str, username: str, password: str) -> Optional[object]:
    """Authenticated NTLM LDAP bind to Domain Controller."""
    try:
        server = Server(dc_ip, get_info=ALL, connect_timeout=10)
        conn = Connection(
            server,
            user=f"{domain}\\{username}",
            password=password,
            authentication=NTLM,
            auto_bind=True,
        )
        return conn
    except LDAPException as e:
        print(f"{Fore.RED}[-] LDAP connection failed: {e}")
        return None


def domain_to_dn(domain: str) -> str:
    return ",".join(f"DC={p}" for p in domain.split("."))


def get_uac_flag(entry, flag: int) -> bool:
    try:
        return bool(int(entry.userAccountControl.value) & flag)
    except Exception:
        return False


def days_since(ts) -> Optional[int]:
    """
    Age in days of an LDAP timestamp.

    ldap3 returns timezone-aware datetimes in UTC. Keep them aware and compare
    against an aware now() rather than stripping tzinfo: mixing aware and naive
    datetimes raises TypeError, and the except below would swallow it and
    return None for every account - making every password look as though it had
    no recorded age.
    """
    if ts is None:
        return None
    try:
        if not isinstance(ts, datetime.datetime):
            return None
        if ts.tzinfo is None:
            ts = ts.replace(tzinfo=datetime.timezone.utc)
        return (datetime.datetime.now(datetime.timezone.utc) - ts).days
    except (TypeError, ValueError, OverflowError):
        return None


def find_kerberoastable(conn, base_dn: str) -> list:
    """Users with ServicePrincipalName — TGS hash can be requested and cracked offline."""
    findings = []
    search_f = (
        "(&(objectClass=user)(objectCategory=person)(servicePrincipalName=*)"
        "(!(objectClass=computer))(!(userAccountControl:1.2.840.113556.1.4.803:=2)))"
    )
    conn.search(base_dn, search_f, SUBTREE,
                attributes=["sAMAccountName", "distinguishedName", "servicePrincipalName",
                             "pwdLastSet", "memberOf", "userAccountControl"])
    for e in conn.entries:
        spns    = list(e.servicePrincipalName.values) if e.servicePrincipalName else []
        pwd_age = days_since(e.pwdLastSet.value if e.pwdLastSet else None)
        groups  = [g.split(",")[0].replace("CN=", "") for g in (e.memberOf.values if e.memberOf else [])]
        in_da   = any(g in ("Domain Admins", "Enterprise Admins") for g in groups)
        findings.append({
            "username": str(e.sAMAccountName),
            "dn": str(e.distinguishedName),
            "spns": spns,
            "pwd_age_days": pwd_age,
            "member_of": groups,
            "in_privileged_group": in_da,
            "risk": "CRITICAL" if in_da else "HIGH",
        })
    return findings


def find_asrep_roastable(conn, base_dn: str) -> list:
    """Accounts with DONT_REQUIRE_PREAUTH — crackable AS-REP hash without knowing password."""
    findings = []
    search_f = (
        f"(&(objectClass=user)(objectCategory=person)"
        f"(userAccountControl:1.2.840.113556.1.4.803:={UAC_DONT_REQUIRE_PREAUTH}))"
    )
    conn.search(base_dn, search_f, SUBTREE,
                attributes=["sAMAccountName", "distinguishedName", "lastLogon",
                             "userAccountControl", "memberOf"])
    for e in conn.entries:
        logon_age = days_since(e.lastLogon.value if e.lastLogon else None)
        groups    = [g.split(",")[0].replace("CN=", "") for g in (e.memberOf.values if e.memberOf else [])]
        in_da     = any(g in ("Domain Admins", "Enterprise Admins") for g in groups)
        disabled  = get_uac_flag(e, UAC_ACCOUNT_DISABLED)
        findings.append({
            "username": str(e.sAMAccountName),
            "dn": str(e.distinguishedName),
            "last_logon_days_ago": logon_age,
            "account_enabled": not disabled,
            "member_of": groups,
            "in_privileged_group": in_da,
            "risk": "CRITICAL" if in_da else "HIGH",
        })
    return findings


def find_stale_privileged(conn, base_dn: str, stale_days: int = 90) -> list:
    """Privileged accounts not logged in for stale_days or more."""
    findings = []
    for group_name in PRIVILEGED_GROUPS:
        conn.search(base_dn, f"(&(objectClass=group)(sAMAccountName={group_name}))",
                    SUBTREE, attributes=["member"])
        if not conn.entries:
            continue
        for member_dn in (conn.entries[0].member.values if conn.entries[0].member else []):
            conn.search(member_dn, "(objectClass=user)",
                        attributes=["sAMAccountName", "lastLogon", "userAccountControl"])
            if not conn.entries:
                continue
            user = conn.entries[0]
            logon_age = days_since(user.lastLogon.value if user.lastLogon else None)
            if logon_age is None or logon_age >= stale_days:
                findings.append({
                    "username": str(user.sAMAccountName),
                    "dn": member_dn,
                    "privileged_group": group_name,
                    "last_logon_days_ago": logon_age,
                    "risk": "HIGH",
                })
    return findings


def find_password_never_expires(conn, base_dn: str) -> list:
    """Enabled accounts with Password Never Expires flag set."""
    findings = []
    search_f = (
        f"(&(objectClass=user)(objectCategory=person)"
        f"(userAccountControl:1.2.840.113556.1.4.803:={UAC_PASSWORD_NEVER_EXPIRES})"
        f"(!(userAccountControl:1.2.840.113556.1.4.803:=2)))"
    )
    conn.search(base_dn, search_f, SUBTREE,
                attributes=["sAMAccountName", "distinguishedName"])
    for e in conn.entries:
        findings.append({
            "username": str(e.sAMAccountName),
            "dn": str(e.distinguishedName),
            "risk": "MEDIUM",
        })
    return findings


def find_privileged_groups(conn, base_dn: str) -> dict:
    """Enumerate members of all high-privilege groups."""
    results = {}
    for group_name in PRIVILEGED_GROUPS:
        conn.search(base_dn, f"(&(objectClass=group)(sAMAccountName={group_name}))",
                    SUBTREE, attributes=["member"])
        if not conn.entries:
            continue
        members = [
            {"cn": m.split(",")[0].replace("CN=", ""), "dn": m}
            for m in (conn.entries[0].member.values if conn.entries[0].member else [])
        ]
        results[group_name] = members
    return results


def find_reversible_encryption(conn, base_dn: str) -> list:
    """Accounts storing password with reversible encryption (cleartext equivalent)."""
    findings = []
    search_f = (
        f"(&(objectClass=user)(objectCategory=person)"
        f"(userAccountControl:1.2.840.113556.1.4.803:={UAC_ENCRYPTED_TEXT_PWD}))"
    )
    conn.search(base_dn, search_f, SUBTREE,
                attributes=["sAMAccountName", "distinguishedName"])
    for e in conn.entries:
        findings.append({
            "username": str(e.sAMAccountName),
            "dn": str(e.distinguishedName),
            "risk": "HIGH",
        })
    return findings


def find_adminsdholder_issues(conn, base_dn: str) -> list:
    """AdminSDHolder-protected objects (adminCount=1) that appear stale or disabled."""
    findings = []
    conn.search(base_dn, "(&(objectClass=user)(adminCount=1))", SUBTREE,
                attributes=["sAMAccountName", "distinguishedName", "memberOf",
                             "userAccountControl", "lastLogon"])
    for e in conn.entries:
        disabled  = get_uac_flag(e, UAC_ACCOUNT_DISABLED)
        logon_age = days_since(e.lastLogon.value if e.lastLogon else None)
        groups    = [g.split(",")[0].replace("CN=", "") for g in (e.memberOf.values if e.memberOf else [])]
        if disabled or (logon_age is not None and logon_age > 90):
            findings.append({
                "username": str(e.sAMAccountName),
                "dn": str(e.distinguishedName),
                "admin_count_flag": True,
                "account_enabled": not disabled,
                "last_logon_days_ago": logon_age,
                "member_of": groups,
                "risk": "MEDIUM",
            })
    return findings


def check_password_policy(conn, base_dn: str) -> dict:
    """Retrieve default domain password policy and check CIS benchmarks."""
    conn.search(base_dn, "(objectClass=domain)", SUBTREE,
                attributes=["minPwdLength", "pwdHistoryLength", "lockoutThreshold",
                             "maxPwdAge", "pwdProperties"])
    if not conn.entries:
        return {}
    e = conn.entries[0]
    min_len    = int(e.minPwdLength.value)    if e.minPwdLength    else 0
    history    = int(e.pwdHistoryLength.value) if e.pwdHistoryLength else 0
    lockout    = int(e.lockoutThreshold.value) if e.lockoutThreshold else 0
    complexity = bool(int(e.pwdProperties.value) & 1) if e.pwdProperties else False
    issues = []
    if min_len  < 14:  issues.append(f"Minimum length {min_len} (CIS benchmark: 14+)")
    if history  < 24:  issues.append(f"Password history {history} (CIS benchmark: 24)")
    if lockout == 0:   issues.append("Lockout threshold disabled — unlimited brute force allowed")
    if not complexity: issues.append("Password complexity not enforced")
    return {
        "min_length": min_len, "history": history,
        "lockout_threshold": lockout, "complexity": complexity,
        "issues": issues, "risk": "HIGH" if issues else "OK",
    }


def generate_risk_report(findings: dict) -> dict:
    """Tally risk scores and return overall risk level."""
    counts = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
    for key in ("kerberoastable", "asrep_roastable", "stale_privileged",
                "reversible_encryption", "adminsdholder_issues"):
        for item in findings.get(key, []):
            counts[item.get("risk", "HIGH")] += 1
    if findings.get("password_policy", {}).get("risk") == "HIGH":
        counts["HIGH"] += 1
    overall = ("CRITICAL" if counts["CRITICAL"] else
               "HIGH"     if counts["HIGH"]     else
               "MEDIUM"   if counts["MEDIUM"]   else "LOW")
    return {"summary": counts, "overall_risk": overall}


def pi(label): print(f"{Fore.CYAN}[*] {label}")
def pw(label): print(f"{Fore.YELLOW}[!] {label}")
def pg(label): print(f"{Fore.GREEN}[+] {label}")
def prow(user, value, warn=False):
    c = Fore.RED if warn else Fore.WHITE
    print(f"    {c}{user:<16} {value}")


def main():
    ap = argparse.ArgumentParser(description="AD Security Auditor — authorized assessments only")
    ap.add_argument("--dc-ip",    required=True)
    ap.add_argument("--domain",   required=True)
    ap.add_argument("--username", required=True)
    ap.add_argument("--output",   default=f"ad_audit_{datetime.date.today()}.json")
    args = ap.parse_args()

    # The password comes from the environment, never from an argument: an
    # argument is readable by every user on the machine in `ps auxww` while the
    # audit runs, and it is written to shell history. A domain credential in a
    # shell history file on a shared jump host is how an audit becomes an
    # incident.
    password = os.environ.get("AD_PASS")
    if not password:
        print("[!] Set AD_PASS in the environment with the password for --username.")
        sys.exit(2)

    pi(f"Active Directory Security Auditor | Domain: {args.domain.upper()}")
    pi(f"Connecting to DC: {args.dc_ip} ...")
    conn = connect_ldap(args.dc_ip, args.domain, args.username, password)
    if not conn:
        sys.exit(1)
    pi("Connected (LDAP/389, NTLM authenticated)")
    base_dn = domain_to_dn(args.domain)
    pi(f"Base DN: {base_dn}")

    results = {}

    kerb = find_kerberoastable(conn, base_dn)
    results["kerberoastable"] = kerb
    pw(f"KERBEROASTABLE ACCOUNTS — {len(kerb)} found (HIGH)")
    for a in kerb:
        spn = ", ".join(a["spns"][:2])
        prow(a["username"], f"SPN: {spn}", warn=a["in_privileged_group"])
        age = a["pwd_age_days"]
        if age:
            print(f"                 Password last set: {age} days ago")
        if a["in_privileged_group"]:
            print(f"    {Fore.RED}    Member of privileged group — CRITICAL RISK")
        print(f"    {Fore.WHITE}    -> Attacker can request TGS ticket offline and crack with hashcat")

    asrep = find_asrep_roastable(conn, base_dn)
    results["asrep_roastable"] = asrep
    pw(f"AS-REP ROASTABLE — {len(asrep)} found (HIGH)")
    for a in asrep:
        logon = f"{a['last_logon_days_ago']}d ago" if a["last_logon_days_ago"] else "never"
        prow(a["username"], f"Pre-auth disabled | Last logon: {logon}", warn=a["in_privileged_group"])
        if a["in_privileged_group"]:
            print(f"    {Fore.RED}    CRITICAL: Dormant privileged account with no Kerberos pre-auth")
        print(f"    {Fore.WHITE}    -> No password needed to obtain crackable AS-REP hash")

    groups = find_privileged_groups(conn, base_dn)
    results["privileged_groups"] = groups
    pw("PRIVILEGED GROUP MEMBERS")
    for gname, members in groups.items():
        print(f"  {Fore.YELLOW}  {gname}: {len(members)} members")
        for m in members[:8]:
            print(f"    {Fore.WHITE}  {m['cn']}")

    stale = find_stale_privileged(conn, base_dn)
    results["stale_privileged"] = stale
    pw(f"STALE PRIVILEGED ACCOUNTS (inactive 90+ days) — {len(stale)} found")
    for a in stale:
        age_s = f"{a['last_logon_days_ago']}d inactive" if a["last_logon_days_ago"] else "never logged in"
        prow(a["username"], f"Group: {a['privileged_group']} | {age_s}", warn=True)

    policy = check_password_policy(conn, base_dn)
    results["password_policy"] = policy
    pw("PASSWORD POLICY")
    if policy.get("issues"):
        for iss in policy["issues"]:
            print(f"    {Fore.RED}x {iss}")
    else:
        pg("Password policy meets CIS benchmarks")

    pne = find_password_never_expires(conn, base_dn)
    results["password_never_expires"] = pne
    if pne:
        pw(f"PASSWORD NEVER EXPIRES — {len(pne)} accounts")
        sample = ", ".join(a["username"] for a in pne[:5])
        print(f"    {Fore.YELLOW}Includes: {sample}{' ...' if len(pne) > 5 else ''}")
        print(f"    {Fore.WHITE}    Service accounts: expected. User accounts: review and justify each.")
    else:
        pg("PASSWORD NEVER EXPIRES — 0 accounts")

    rev = find_reversible_encryption(conn, base_dn)
    results["reversible_encryption"] = rev
    if rev:
        pw(f"REVERSIBLE ENCRYPTION — {len(rev)} accounts (HIGH)")
        for a in rev:
            prow(a["username"], "Reversible password encryption ENABLED", warn=True)
    else:
        pg("REVERSIBLE ENCRYPTION — 0 accounts")

    adm = find_adminsdholder_issues(conn, base_dn)
    results["adminsdholder_issues"] = adm
    if adm:
        pw(f"ADMINSDHOLDER ANOMALIES — {len(adm)} protected objects flagged")
        for a in adm:
            age_s = f"{a['last_logon_days_ago']}d" if a["last_logon_days_ago"] else "never"
            prow(a["username"], f"adminCount=1 | inactive {age_s} | enabled={a['account_enabled']}", warn=True)

    report = generate_risk_report(results)
    results["risk_report"] = report
    results.update({"generated": datetime.datetime.now(datetime.timezone.utc).isoformat(),
                    "domain": args.domain, "dc_ip": args.dc_ip})
    s = report["summary"]
    print(f"\n{Fore.WHITE}{Style.BRIGHT}Summary: "
          f"{Fore.RED}{s['CRITICAL']} CRITICAL{Fore.WHITE} | "
          f"{Fore.YELLOW}{s['HIGH']} HIGH{Fore.WHITE} | "
          f"{Fore.CYAN}{s['MEDIUM']} MEDIUM | "
          f"{Fore.GREEN}{s['LOW']} LOW")
    with open(args.output, "w") as f:
        json.dump(results, f, indent=2, default=str)
    pi(f"Report saved: {args.output}")
    conn.unbind()
    return 2 if s['CRITICAL'] else (1 if s['HIGH'] or s['MEDIUM'] else 0)


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)