ergosteur
d2ff9d2617
Every table cell fed from directory content (descriptions, sAMAccountName, OS strings, DNs, object classes) was interpolated into the Markdown report raw. python-markdown doesn't escape inline HTML by default, so a directory-controlled value like an OU description containing <script>... would render live once the report was converted to HTML with md_to_html.py -- and the underlying data is attacker-influenceable, not just operator-authored. Added md_escape() (Python) / ConvertTo-MdSafe (PowerShell), applied at every table-row interpolation in both scripts. Escapes &, <, > to HTML entities and | plus embedded newlines to keep the table structure intact. Raw JSON dumps are left untouched -- this only affects the Markdown/HTML presentation layer. Verified end-to-end with <script>, <img onerror=...>, embedded &, and embedded | payloads across every affected table in both scripts; confirmed no live tags reach the rendered HTML and no double-escaping occurs.
604 lines
22 KiB
Python
604 lines
22 KiB
Python
#!/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 re
|
|
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 md_escape(value):
|
|
"""Escape a value before it goes into a Markdown table cell.
|
|
|
|
AD attributes (descriptions, sAMAccountName, OS strings, DNs, ...) are
|
|
directory content, not report-generated text -- they can contain
|
|
anything a writer or an attacker put there, including raw HTML. Markdown
|
|
doesn't escape inline HTML by default, so an unescaped '<script>' in an
|
|
OU description would render live when the report is viewed as HTML.
|
|
Also neutralizes '|' and embedded newlines, which would otherwise
|
|
corrupt the table row itself.
|
|
"""
|
|
if value is None:
|
|
return ""
|
|
text = str(value)
|
|
text = text.replace("&", "&").replace("<", "<").replace(">", ">")
|
|
text = text.replace("|", "\\|")
|
|
text = re.sub(r"\s*[\r\n]+\s*", " ", text).strip()
|
|
return text
|
|
|
|
|
|
def rfc3339(dt):
|
|
"""Format a datetime as RFC 3339 (UTC, second precision, 'Z' suffix).
|
|
|
|
Falls back to str(dt) for values ldap3 didn't parse into a datetime
|
|
(e.g. if a server's schema doesn't trigger its generalizedTime formatter).
|
|
"""
|
|
if dt is None:
|
|
return ""
|
|
if not isinstance(dt, datetime.datetime):
|
|
return str(dt)
|
|
if dt.tzinfo is not None:
|
|
dt = dt.astimezone(datetime.timezone.utc)
|
|
return dt.strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
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": rfc3339(e.whenCreated.value) 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": rfc3339(last_logon) if last_logon else None,
|
|
"never_logged_on": never_logged_on,
|
|
"stale": stale,
|
|
"pwd_last_set": rfc3339(pwd_last_set) if pwd_last_set else None,
|
|
"created": rfc3339(e.whenCreated.value) 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": rfc3339(last_logon) if last_logon else None,
|
|
"stale": stale,
|
|
"created": rfc3339(e.whenCreated.value) 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": rfc3339(e.whenCreated.value) 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 compute_stats(data, args):
|
|
ous = data["ous"]
|
|
users = data["users"]
|
|
computers = data["computers"]
|
|
groups = data["groups"]
|
|
|
|
max_depth = max((o["depth"] for o in ous), default=0)
|
|
|
|
total_users = len(users)
|
|
disabled = sum(1 for u in users if u["disabled"])
|
|
stats = {
|
|
"total_users": total_users,
|
|
"enabled": total_users - disabled,
|
|
"disabled": 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"]),
|
|
}
|
|
|
|
total_comp = len(computers)
|
|
comp_disabled = sum(1 for c in computers if c["disabled"])
|
|
comp_stats = {
|
|
"total": total_comp,
|
|
"disabled": comp_disabled,
|
|
"stale": sum(1 for c in computers if c["stale"]),
|
|
"os_counter": Counter(c["os"] or "Unknown" for c in computers),
|
|
}
|
|
|
|
total_groups = len(groups)
|
|
security_groups = sum(1 for g in groups if g["type"] == "Security")
|
|
group_stats = {
|
|
"total": total_groups,
|
|
"security": security_groups,
|
|
"distribution": total_groups - security_groups,
|
|
"empty": sum(1 for g in groups if g["empty"]),
|
|
"scope_counter": Counter(g["scope"] for g in groups),
|
|
}
|
|
|
|
return {
|
|
"max_ou_depth": max_depth,
|
|
"users": stats,
|
|
"computers": comp_stats,
|
|
"groups": group_stats,
|
|
}
|
|
|
|
|
|
def build_executive_summary(data, stats, args):
|
|
users = data["users"]
|
|
computers = data["computers"]
|
|
groups = data["groups"]
|
|
u = stats["users"]
|
|
c = stats["computers"]
|
|
g = stats["groups"]
|
|
|
|
# Each finding: severity, title, notes, column labels, matching object rows.
|
|
findings = []
|
|
|
|
def user_rows(pred):
|
|
return [
|
|
(x["sam"], x["last_logon"] or "", x["dn"])
|
|
for x in sorted(
|
|
(x for x in users if pred(x)),
|
|
key=lambda x: x["sam"].lower(),
|
|
)
|
|
]
|
|
|
|
USER_COLS = ("Account", "Last Logon", "DN")
|
|
|
|
if u["pwd_not_required"]:
|
|
findings.append(("Critical", "User accounts allowing blank passwords",
|
|
"PASSWD_NOTREQD flag set; remove unless there is a specific reason",
|
|
USER_COLS, user_rows(lambda x: x["pwd_not_required"])))
|
|
if u["no_preauth"]:
|
|
findings.append(("Critical", "AS-REP roastable accounts (Kerberos pre-auth disabled)",
|
|
"Offline password cracking risk; re-enable pre-auth unless required",
|
|
USER_COLS, user_rows(lambda x: x["kerberos_preauth_disabled"])))
|
|
if u["trusted_deleg"]:
|
|
findings.append(("Critical", "Accounts trusted for unconstrained delegation",
|
|
"High-value targets for credential theft; move to constrained/no delegation",
|
|
USER_COLS, user_rows(lambda x: x["trusted_for_delegation"])))
|
|
if u["locked"]:
|
|
findings.append(("High", "Currently locked-out user accounts",
|
|
"May indicate attack activity or stale service credentials",
|
|
USER_COLS, user_rows(lambda x: x["locked"])))
|
|
if u["stale"]:
|
|
findings.append(("Medium", f"Stale enabled user accounts (>{args.stale_days}d inactive)",
|
|
"Candidates for disable/offboarding review",
|
|
USER_COLS, user_rows(lambda x: x["stale"])))
|
|
if c["stale"]:
|
|
findings.append(("Medium", f"Stale enabled computer accounts (>{args.stale_days}d inactive)",
|
|
"Likely decommissioned hardware still trusted in the domain",
|
|
("Computer", "Last Logon", "DN"),
|
|
[(x["sam"], x["last_logon"] or "", x["dn"])
|
|
for x in sorted((x for x in computers if x["stale"]), key=lambda x: x["sam"].lower())]))
|
|
if g["empty"]:
|
|
findings.append(("Medium", "Empty security/distribution groups",
|
|
"Cleanup candidates ahead of OU/group reorg",
|
|
("Group", "Type", "DN"),
|
|
[(x["sam"], x["type"], x["dn"])
|
|
for x in sorted((x for x in groups if x["empty"]), key=lambda x: x["sam"].lower())]))
|
|
if u["pwd_never_expires"]:
|
|
findings.append(("Medium", "Accounts with password-never-expires set",
|
|
"Review against password policy; exempt only where justified",
|
|
USER_COLS, user_rows(lambda x: x["pwd_never_expires"])))
|
|
if u["never_logged_on"]:
|
|
findings.append(("Low", "Enabled accounts that have never logged on",
|
|
"Possibly unused/orphaned provisioning; verify before disabling",
|
|
USER_COLS, user_rows(lambda x: x["never_logged_on"])))
|
|
if u["admin_count_flagged"]:
|
|
findings.append(("Info", "Accounts with adminCount=1 (current or former privileged)",
|
|
"SDProp-protected ACLs persist even after privilege is removed; review membership",
|
|
USER_COLS, user_rows(lambda x: x["admin_count"])))
|
|
|
|
severity_order = {"Critical": 0, "High": 1, "Medium": 2, "Low": 3, "Info": 4}
|
|
findings.sort(key=lambda f: severity_order[f[0]])
|
|
|
|
lines = []
|
|
a = lines.append
|
|
a("## Executive Summary")
|
|
a("")
|
|
a(f"- Users: {u['total_users']} total ({u['enabled']} enabled, {u['disabled']} disabled)")
|
|
a(f"- Computers: {c['total']} total ({c['disabled']} disabled)")
|
|
a(f"- Groups: {g['total']} total ({g['security']} security, {g['distribution']} distribution)")
|
|
a(f"- Max OU nesting depth: {stats['max_ou_depth']}")
|
|
a("")
|
|
if findings:
|
|
a("### Risk & Cleanup Findings")
|
|
a("")
|
|
a("| Severity | Finding | Count | Notes |")
|
|
a("|---|---|---|---|")
|
|
for severity, title, notes, cols, rows in findings:
|
|
a(f"| {severity} | {md_escape(title)} | {len(rows)} | {md_escape(notes)} |")
|
|
a("")
|
|
|
|
a("### Finding Detail Lists")
|
|
a("")
|
|
for severity, title, notes, cols, rows in findings:
|
|
a(f"#### [{severity}] {title}")
|
|
a("")
|
|
a(f"| {md_escape(cols[0])} | {md_escape(cols[1])} | {md_escape(cols[2])} |")
|
|
a("|---|---|---|")
|
|
for col0, col1, dn in rows:
|
|
a(f"| {md_escape(col0)} | {md_escape(col1)} | {md_escape(dn)} |")
|
|
a("")
|
|
else:
|
|
a("No notable risk or cleanup findings surfaced by this audit's checks.")
|
|
a("")
|
|
|
|
return "\n".join(lines)
|
|
|
|
|
|
def build_report(data, args):
|
|
ous = data["ous"]
|
|
users = data["users"]
|
|
computers = data["computers"]
|
|
groups = data["groups"]
|
|
obj_counts = data["object_counts"]
|
|
|
|
stats = compute_stats(data, args)
|
|
|
|
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: {rfc3339(datetime.datetime.utcnow())}")
|
|
a(f"- Stale-account threshold: {args.stale_days} days of inactivity")
|
|
a(f"")
|
|
|
|
a(build_executive_summary(data, stats, args))
|
|
|
|
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"| {md_escape(cls)} | {count} |")
|
|
a("")
|
|
|
|
a("## Organizational Units")
|
|
a("")
|
|
a(f"- Total OUs: {len(ous)}")
|
|
a(f"- Maximum nesting depth: {stats['max_ou_depth']}")
|
|
a("")
|
|
a("| OU DN | Depth | Description |")
|
|
a("|---|---|---|")
|
|
for o in sorted(ous, key=lambda x: x["dn"]):
|
|
a(f"| {md_escape(o['dn'])} | {o['depth']} | {md_escape(o['description'])} |")
|
|
a("")
|
|
|
|
a("## Users")
|
|
a("")
|
|
u = stats["users"]
|
|
total_users, enabled, disabled = u["total_users"], u["enabled"], u["disabled"]
|
|
locked = u["locked"]
|
|
pwd_never_expires = u["pwd_never_expires"]
|
|
pwd_not_required = u["pwd_not_required"]
|
|
never_logged_on = u["never_logged_on"]
|
|
stale = u["stale"]
|
|
admin_count_flagged = u["admin_count_flagged"]
|
|
trusted_deleg = u["trusted_deleg"]
|
|
no_preauth = u["no_preauth"]
|
|
|
|
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("")
|
|
|
|
a("## Computers")
|
|
a("")
|
|
c = stats["computers"]
|
|
total_comp, comp_disabled, comp_stale = c["total"], c["disabled"], c["stale"]
|
|
os_counter = c["os_counter"]
|
|
|
|
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"| {md_escape(os_name)} | {count} |")
|
|
a("")
|
|
|
|
a("## Groups")
|
|
a("")
|
|
gs = stats["groups"]
|
|
total_groups, security_groups, distribution_groups = gs["total"], gs["security"], gs["distribution"]
|
|
empty_groups = gs["empty"]
|
|
scope_counter = gs["scope_counter"]
|
|
|
|
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"| {md_escape(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"| {md_escape(g['sam'])} | {md_escape(g['type'])} | {md_escape(g['scope'])} | {g['member_count']} |")
|
|
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()
|