Add AD audit scripts (Python/LDAP and PowerShell/RSAT)
Read-only structural/health audit of Active Directory: OU tree, user and computer account status/hygiene flags, and group breakdown. Two equivalent implementations depending on available access -- raw LDAP via ldap3, or Get-AD* cmdlets via RSAT on a domain-joined machine.
This commit is contained in:
+429
@@ -0,0 +1,429 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Read-only LDAP audit of an Active Directory domain.
|
||||
|
||||
Connects with a basic (non-admin) bind account and enumerates OUs, groups,
|
||||
users, and computers to produce a structural/health report. Designed to work
|
||||
with whatever a standard authenticated-user account can see over LDAP -- no
|
||||
elevated rights required.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import datetime
|
||||
import getpass
|
||||
import json
|
||||
import sys
|
||||
from collections import Counter, defaultdict
|
||||
|
||||
from ldap3 import ALL, SUBTREE, Connection, Server, Tls
|
||||
import ssl
|
||||
|
||||
# --- UserAccountControl bit flags (subset relevant to an audit) ---
|
||||
UAC_ACCOUNTDISABLE = 0x0002
|
||||
UAC_LOCKOUT = 0x0010
|
||||
UAC_PASSWD_NOTREQD = 0x0020
|
||||
UAC_DONT_EXPIRE_PASSWD = 0x10000
|
||||
UAC_SMARTCARD_REQUIRED = 0x40000
|
||||
UAC_TRUSTED_FOR_DELEGATION = 0x80000
|
||||
UAC_NOT_DELEGATED = 0x100000
|
||||
UAC_DONT_REQ_PREAUTH = 0x400000
|
||||
|
||||
FILETIME_EPOCH = datetime.datetime(1601, 1, 1)
|
||||
|
||||
|
||||
def filetime_to_datetime(value):
|
||||
try:
|
||||
v = int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if v == 0 or v == 0x7FFFFFFFFFFFFFFF:
|
||||
return None
|
||||
try:
|
||||
return FILETIME_EPOCH + datetime.timedelta(microseconds=v / 10)
|
||||
except OverflowError:
|
||||
return None
|
||||
|
||||
|
||||
def parse_args():
|
||||
p = argparse.ArgumentParser(description=__doc__)
|
||||
p.add_argument("--host", required=True, help="DC hostname or IP")
|
||||
p.add_argument("--port", type=int, default=None, help="LDAP port (default 389, or 636 with --ssl)")
|
||||
p.add_argument("--base-dn", required=True, help="Search base, e.g. DC=corp,DC=example,DC=com")
|
||||
p.add_argument("--bind-dn", default=None, help="Bind DN (prompted if omitted)")
|
||||
p.add_argument("--ssl", action="store_true", help="Use LDAPS (implicit TLS)")
|
||||
p.add_argument("--starttls", action="store_true", help="Use StartTLS over plain LDAP port")
|
||||
p.add_argument("--no-verify-cert", action="store_true", help="Skip TLS cert verification (self-signed DCs)")
|
||||
p.add_argument("--stale-days", type=int, default=90, help="Days of inactivity to flag an account as stale")
|
||||
p.add_argument("--out-dir", default="reports", help="Directory to write report + raw JSON into")
|
||||
return p.parse_args()
|
||||
|
||||
|
||||
def connect(args):
|
||||
tls = None
|
||||
if args.ssl or args.starttls:
|
||||
validate = ssl.CERT_NONE if args.no_verify_cert else ssl.CERT_REQUIRED
|
||||
tls = Tls(validate=validate)
|
||||
|
||||
port = args.port or (636 if args.ssl else 389)
|
||||
server = Server(args.host, port=port, use_ssl=args.ssl, tls=tls, get_info=ALL)
|
||||
|
||||
bind_dn = args.bind_dn or input("Bind DN (e.g. user@corp.example.com or DOMAIN\\user): ").strip()
|
||||
password = getpass.getpass(f"Password for {bind_dn}: ")
|
||||
|
||||
conn = Connection(server, user=bind_dn, password=password, auto_bind=False)
|
||||
if args.starttls:
|
||||
conn.open()
|
||||
conn.start_tls()
|
||||
if not conn.bind():
|
||||
print(f"Bind failed: {conn.result}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
return conn
|
||||
|
||||
|
||||
def paged_search(conn, base_dn, search_filter, attributes):
|
||||
entries = []
|
||||
conn.search(
|
||||
search_base=base_dn,
|
||||
search_filter=search_filter,
|
||||
search_scope=SUBTREE,
|
||||
attributes=attributes,
|
||||
paged_size=1000,
|
||||
)
|
||||
entries.extend(conn.entries)
|
||||
cookie = conn.result.get("controls", {}).get("1.2.840.113556.1.4.319", {}).get("value", {}).get("cookie")
|
||||
while cookie:
|
||||
conn.search(
|
||||
search_base=base_dn,
|
||||
search_filter=search_filter,
|
||||
search_scope=SUBTREE,
|
||||
attributes=attributes,
|
||||
paged_size=1000,
|
||||
paged_cookie=cookie,
|
||||
)
|
||||
entries.extend(conn.entries)
|
||||
cookie = conn.result.get("controls", {}).get("1.2.840.113556.1.4.319", {}).get("value", {}).get("cookie")
|
||||
return entries
|
||||
|
||||
|
||||
def audit_ous(conn, base_dn):
|
||||
attrs = ["distinguishedName", "ou", "description", "whenCreated"]
|
||||
entries = paged_search(conn, base_dn, "(objectClass=organizationalUnit)", attrs)
|
||||
ous = []
|
||||
for e in entries:
|
||||
ous.append(
|
||||
{
|
||||
"dn": str(e.distinguishedName),
|
||||
"name": str(e.ou) if e.ou else "",
|
||||
"description": str(e.description) if e.description else "",
|
||||
"created": str(e.whenCreated) if e.whenCreated else "",
|
||||
"depth": str(e.distinguishedName).count("OU="),
|
||||
}
|
||||
)
|
||||
return ous
|
||||
|
||||
|
||||
def audit_users(conn, base_dn, stale_days):
|
||||
attrs = [
|
||||
"distinguishedName",
|
||||
"sAMAccountName",
|
||||
"userAccountControl",
|
||||
"lastLogonTimestamp",
|
||||
"pwdLastSet",
|
||||
"whenCreated",
|
||||
"adminCount",
|
||||
"memberOf",
|
||||
"userPrincipalName",
|
||||
]
|
||||
entries = paged_search(conn, base_dn, "(&(objectCategory=person)(objectClass=user))", attrs)
|
||||
|
||||
now = datetime.datetime.utcnow()
|
||||
stale_cutoff = now - datetime.timedelta(days=stale_days)
|
||||
|
||||
users = []
|
||||
for e in entries:
|
||||
uac = int(str(e.userAccountControl)) if e.userAccountControl else 0
|
||||
last_logon = filetime_to_datetime(str(e.lastLogonTimestamp)) if e.lastLogonTimestamp else None
|
||||
pwd_last_set = filetime_to_datetime(str(e.pwdLastSet)) if e.pwdLastSet else None
|
||||
|
||||
disabled = bool(uac & UAC_ACCOUNTDISABLE)
|
||||
pwd_never_expires = bool(uac & UAC_DONT_EXPIRE_PASSWD)
|
||||
pwd_not_required = bool(uac & UAC_PASSWD_NOTREQD)
|
||||
never_logged_on = last_logon is None
|
||||
stale = (not disabled) and (last_logon is not None) and (last_logon < stale_cutoff)
|
||||
|
||||
users.append(
|
||||
{
|
||||
"dn": str(e.distinguishedName),
|
||||
"sam": str(e.sAMAccountName) if e.sAMAccountName else "",
|
||||
"upn": str(e.userPrincipalName) if e.userPrincipalName else "",
|
||||
"disabled": disabled,
|
||||
"locked": bool(uac & UAC_LOCKOUT),
|
||||
"pwd_never_expires": pwd_never_expires,
|
||||
"pwd_not_required": pwd_not_required,
|
||||
"smartcard_required": bool(uac & UAC_SMARTCARD_REQUIRED),
|
||||
"trusted_for_delegation": bool(uac & UAC_TRUSTED_FOR_DELEGATION),
|
||||
"kerberos_preauth_disabled": bool(uac & UAC_DONT_REQ_PREAUTH),
|
||||
"admin_count": bool(e.adminCount and int(str(e.adminCount)) > 0),
|
||||
"last_logon": last_logon.isoformat() if last_logon else None,
|
||||
"never_logged_on": never_logged_on,
|
||||
"stale": stale,
|
||||
"pwd_last_set": pwd_last_set.isoformat() if pwd_last_set else None,
|
||||
"created": str(e.whenCreated) if e.whenCreated else "",
|
||||
"group_count": len(e.memberOf.values) if e.memberOf else 0,
|
||||
}
|
||||
)
|
||||
return users
|
||||
|
||||
|
||||
def audit_computers(conn, base_dn, stale_days):
|
||||
attrs = [
|
||||
"distinguishedName",
|
||||
"sAMAccountName",
|
||||
"userAccountControl",
|
||||
"operatingSystem",
|
||||
"operatingSystemVersion",
|
||||
"lastLogonTimestamp",
|
||||
"whenCreated",
|
||||
]
|
||||
entries = paged_search(conn, base_dn, "(objectCategory=computer)", attrs)
|
||||
|
||||
now = datetime.datetime.utcnow()
|
||||
stale_cutoff = now - datetime.timedelta(days=stale_days)
|
||||
|
||||
computers = []
|
||||
for e in entries:
|
||||
uac = int(str(e.userAccountControl)) if e.userAccountControl else 0
|
||||
last_logon = filetime_to_datetime(str(e.lastLogonTimestamp)) if e.lastLogonTimestamp else None
|
||||
disabled = bool(uac & UAC_ACCOUNTDISABLE)
|
||||
stale = (not disabled) and (last_logon is not None) and (last_logon < stale_cutoff)
|
||||
|
||||
computers.append(
|
||||
{
|
||||
"dn": str(e.distinguishedName),
|
||||
"sam": str(e.sAMAccountName) if e.sAMAccountName else "",
|
||||
"os": str(e.operatingSystem) if e.operatingSystem else "",
|
||||
"os_version": str(e.operatingSystemVersion) if e.operatingSystemVersion else "",
|
||||
"disabled": disabled,
|
||||
"last_logon": last_logon.isoformat() if last_logon else None,
|
||||
"stale": stale,
|
||||
"created": str(e.whenCreated) if e.whenCreated else "",
|
||||
}
|
||||
)
|
||||
return computers
|
||||
|
||||
|
||||
def audit_groups(conn, base_dn):
|
||||
attrs = ["distinguishedName", "sAMAccountName", "groupType", "member", "description", "whenCreated"]
|
||||
entries = paged_search(conn, base_dn, "(objectClass=group)", attrs)
|
||||
|
||||
groups = []
|
||||
for e in entries:
|
||||
gt = int(str(e.groupType)) if e.groupType else 0
|
||||
is_security = bool(gt & 0x80000000)
|
||||
scope_bit = gt & 0x0C
|
||||
if gt & 0x00000002:
|
||||
scope = "DomainLocal"
|
||||
elif gt & 0x00000004:
|
||||
scope = "Global"
|
||||
elif gt & 0x00000008:
|
||||
scope = "Universal"
|
||||
else:
|
||||
scope = "Unknown"
|
||||
|
||||
members = e.member.values if e.member else []
|
||||
groups.append(
|
||||
{
|
||||
"dn": str(e.distinguishedName),
|
||||
"sam": str(e.sAMAccountName) if e.sAMAccountName else "",
|
||||
"type": "Security" if is_security else "Distribution",
|
||||
"scope": scope,
|
||||
"member_count": len(members),
|
||||
"empty": len(members) == 0,
|
||||
"description": str(e.description) if e.description else "",
|
||||
"created": str(e.whenCreated) if e.whenCreated else "",
|
||||
}
|
||||
)
|
||||
return groups
|
||||
|
||||
|
||||
def audit_object_counts(conn, base_dn):
|
||||
attrs = ["objectClass"]
|
||||
entries = paged_search(conn, base_dn, "(objectClass=*)", attrs)
|
||||
counter = Counter()
|
||||
for e in entries:
|
||||
classes = e.objectClass.values if e.objectClass else []
|
||||
# most-specific class is typically the last in the chain
|
||||
leaf = classes[-1] if classes else "unknown"
|
||||
counter[leaf] += 1
|
||||
return counter
|
||||
|
||||
|
||||
def build_report(data, args):
|
||||
ous = data["ous"]
|
||||
users = data["users"]
|
||||
computers = data["computers"]
|
||||
groups = data["groups"]
|
||||
obj_counts = data["object_counts"]
|
||||
|
||||
lines = []
|
||||
a = lines.append
|
||||
a(f"# Active Directory Audit Report")
|
||||
a(f"")
|
||||
a(f"- Base DN: `{args.base_dn}`")
|
||||
a(f"- Domain Controller: `{args.host}`")
|
||||
a(f"- Generated: {datetime.datetime.utcnow().isoformat()}Z")
|
||||
a(f"- Stale-account threshold: {args.stale_days} days of inactivity")
|
||||
a(f"")
|
||||
|
||||
a("## Object Type Counts")
|
||||
a("")
|
||||
a("| Object Class | Count |")
|
||||
a("|---|---|")
|
||||
for cls, count in sorted(obj_counts.items(), key=lambda kv: -kv[1]):
|
||||
a(f"| {cls} | {count} |")
|
||||
a("")
|
||||
|
||||
a("## Organizational Units")
|
||||
a("")
|
||||
a(f"- Total OUs: {len(ous)}")
|
||||
max_depth = max((o["depth"] for o in ous), default=0)
|
||||
a(f"- Maximum nesting depth: {max_depth}")
|
||||
a("")
|
||||
a("| OU DN | Depth | Description |")
|
||||
a("|---|---|---|")
|
||||
for o in sorted(ous, key=lambda x: x["dn"]):
|
||||
a(f"| {o['dn']} | {o['depth']} | {o['description']} |")
|
||||
a("")
|
||||
|
||||
a("## Users")
|
||||
a("")
|
||||
total_users = len(users)
|
||||
disabled = sum(1 for u in users if u["disabled"])
|
||||
enabled = total_users - disabled
|
||||
locked = sum(1 for u in users if u["locked"])
|
||||
pwd_never_expires = sum(1 for u in users if u["pwd_never_expires"])
|
||||
pwd_not_required = sum(1 for u in users if u["pwd_not_required"])
|
||||
never_logged_on = sum(1 for u in users if u["never_logged_on"] and not u["disabled"])
|
||||
stale = sum(1 for u in users if u["stale"])
|
||||
admin_count_flagged = sum(1 for u in users if u["admin_count"])
|
||||
trusted_deleg = sum(1 for u in users if u["trusted_for_delegation"])
|
||||
no_preauth = sum(1 for u in users if u["kerberos_preauth_disabled"])
|
||||
|
||||
a(f"- Total user objects: {total_users}")
|
||||
a(f"- Enabled: {enabled}")
|
||||
a(f"- Disabled: {disabled}")
|
||||
a(f"- Currently locked out: {locked}")
|
||||
a(f"- Password never expires: {pwd_never_expires}")
|
||||
a(f"- Password not required (blank password allowed): **{pwd_not_required}**")
|
||||
a(f"- Enabled but never logged on: {never_logged_on}")
|
||||
a(f"- Stale (enabled, inactive > {args.stale_days}d): {stale}")
|
||||
a(f"- adminCount=1 (protected/privileged, incl. historical): {admin_count_flagged}")
|
||||
a(f"- Trusted for unconstrained delegation: **{trusted_deleg}**")
|
||||
a(f"- Kerberos pre-auth disabled (AS-REP roastable): **{no_preauth}**")
|
||||
a("")
|
||||
|
||||
if stale:
|
||||
a("### Stale user accounts")
|
||||
a("")
|
||||
a("| sAMAccountName | Last Logon | DN |")
|
||||
a("|---|---|---|")
|
||||
for u in sorted((u for u in users if u["stale"]), key=lambda x: x["last_logon"] or ""):
|
||||
a(f"| {u['sam']} | {u['last_logon']} | {u['dn']} |")
|
||||
a("")
|
||||
|
||||
a("## Computers")
|
||||
a("")
|
||||
total_comp = len(computers)
|
||||
comp_disabled = sum(1 for c in computers if c["disabled"])
|
||||
comp_stale = sum(1 for c in computers if c["stale"])
|
||||
os_counter = Counter(c["os"] or "Unknown" for c in computers)
|
||||
|
||||
a(f"- Total computer objects: {total_comp}")
|
||||
a(f"- Disabled: {comp_disabled}")
|
||||
a(f"- Stale (enabled, inactive > {args.stale_days}d): {comp_stale}")
|
||||
a("")
|
||||
a("### OS breakdown")
|
||||
a("")
|
||||
a("| Operating System | Count |")
|
||||
a("|---|---|")
|
||||
for os_name, count in sorted(os_counter.items(), key=lambda kv: -kv[1]):
|
||||
a(f"| {os_name} | {count} |")
|
||||
a("")
|
||||
|
||||
a("## Groups")
|
||||
a("")
|
||||
total_groups = len(groups)
|
||||
security_groups = sum(1 for g in groups if g["type"] == "Security")
|
||||
distribution_groups = total_groups - security_groups
|
||||
empty_groups = sum(1 for g in groups if g["empty"])
|
||||
scope_counter = Counter(g["scope"] for g in groups)
|
||||
|
||||
a(f"- Total groups: {total_groups}")
|
||||
a(f"- Security groups: {security_groups}")
|
||||
a(f"- Distribution groups: {distribution_groups}")
|
||||
a(f"- Empty groups (0 members): {empty_groups}")
|
||||
a("")
|
||||
a("| Scope | Count |")
|
||||
a("|---|---|")
|
||||
for scope, count in scope_counter.most_common():
|
||||
a(f"| {scope} | {count} |")
|
||||
a("")
|
||||
|
||||
largest = sorted(groups, key=lambda g: -g["member_count"])[:15]
|
||||
a("### Largest groups (top 15 by member count)")
|
||||
a("")
|
||||
a("| Group | Type | Scope | Members |")
|
||||
a("|---|---|---|---|")
|
||||
for g in largest:
|
||||
a(f"| {g['sam']} | {g['type']} | {g['scope']} | {g['member_count']} |")
|
||||
a("")
|
||||
|
||||
if empty_groups:
|
||||
a("### Empty groups (candidates for cleanup)")
|
||||
a("")
|
||||
a("| Group | DN |")
|
||||
a("|---|---|")
|
||||
for g in sorted((g for g in groups if g["empty"]), key=lambda x: x["sam"]):
|
||||
a(f"| {g['sam']} | {g['dn']} |")
|
||||
a("")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
conn = connect(args)
|
||||
print("Bound successfully. Enumerating directory (this may take a while on large domains)...")
|
||||
|
||||
data = {}
|
||||
print(" - object type counts...")
|
||||
data["object_counts"] = dict(audit_object_counts(conn, args.base_dn))
|
||||
print(" - organizational units...")
|
||||
data["ous"] = audit_ous(conn, args.base_dn)
|
||||
print(" - users...")
|
||||
data["users"] = audit_users(conn, args.base_dn, args.stale_days)
|
||||
print(" - computers...")
|
||||
data["computers"] = audit_computers(conn, args.base_dn, args.stale_days)
|
||||
print(" - groups...")
|
||||
data["groups"] = audit_groups(conn, args.base_dn)
|
||||
|
||||
conn.unbind()
|
||||
|
||||
import os
|
||||
|
||||
os.makedirs(args.out_dir, exist_ok=True)
|
||||
ts = datetime.datetime.utcnow().strftime("%Y%m%d_%H%M%S")
|
||||
|
||||
raw_path = os.path.join(args.out_dir, f"ad_audit_raw_{ts}.json")
|
||||
with open(raw_path, "w") as f:
|
||||
json.dump(data, f, indent=2, default=str)
|
||||
|
||||
report = build_report(data, args)
|
||||
report_path = os.path.join(args.out_dir, f"ad_audit_report_{ts}.md")
|
||||
with open(report_path, "w") as f:
|
||||
f.write(report)
|
||||
|
||||
print(f"\nDone.\n Raw data: {raw_path}\n Report: {report_path}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user