#!/usr/bin/env python3
"""Active Directory Vulnerability Assessment agent - parses PingCastle XML
reports and performs LDAP checks to assess AD security posture against
common vulnerability categories."""

import argparse
import json
import xml.etree.ElementTree as ET
from collections import Counter
from datetime import datetime
from pathlib import Path

try:
    from ldap3 import Server, Connection, ALL, SUBTREE
except ImportError:
    Connection = None


def parse_pingcastle_report(xml_path: str) -> dict:
    """Parse PingCastle XML health check report."""
    tree = ET.parse(xml_path)
    root = tree.getroot()

    scores = {}
    for score_elem in root.iter("GlobalScore"):
        scores["global"] = int(score_elem.text) if score_elem.text else 0
    for cat in ("StaleObjectsScore", "PrivilegiedGroupScore",
                "TrustScore", "AnomalyScore"):
        elem = root.find(f".//{cat}")
        if elem is not None and elem.text:
            scores[cat.replace("Score", "").lower()] = int(elem.text)

    risks = []
    for rule in root.iter("HealthcheckRiskRule"):
        rationale = rule.find("Rationale")
        category = rule.find("Category")
        points = rule.find("Points")
        risks.append({
            "rule": rule.find("RiskId").text if rule.find("RiskId") is not None else "",
            "category": category.text if category is not None else "",
            "points": int(points.text) if points is not None and points.text else 0,
            "rationale": rationale.text if rationale is not None else "",
        })

    return {"scores": scores, "risks": risks}


def check_password_policy_ldap(server_url: str, username: str, password: str) -> list[dict]:
    """Check domain password policy via LDAP."""
    if Connection is None:
        return [{"error": "ldap3 not installed"}]
    srv = Server(server_url, get_info=ALL, use_ssl=True)
    conn = Connection(srv, user=username, password=password, auto_bind=True)
    base_dn = conn.server.info.other.get("defaultNamingContext", [""])[0]

    conn.search(base_dn, "(objectClass=domain)", search_scope=SUBTREE,
                attributes=["minPwdLength", "lockoutThreshold", "pwdHistoryLength",
                             "maxPwdAge", "minPwdAge"])
    findings = []
    if conn.entries:
        entry = conn.entries[0]
        min_len = int(str(entry.minPwdLength)) if hasattr(entry, "minPwdLength") else 0
        lockout = int(str(entry.lockoutThreshold)) if hasattr(entry, "lockoutThreshold") else 0
        history = int(str(entry.pwdHistoryLength)) if hasattr(entry, "pwdHistoryLength") else 0

        if min_len < 12:
            findings.append({"check": "min_password_length", "value": min_len,
                             "severity": "high", "detail": f"Min length {min_len} < 12"})
        if lockout == 0:
            findings.append({"check": "account_lockout", "value": lockout,
                             "severity": "critical", "detail": "No account lockout policy"})
        if history < 12:
            findings.append({"check": "password_history", "value": history,
                             "severity": "medium", "detail": f"History {history} < 12"})
    conn.unbind()
    return findings


def check_krbtgt_age(server_url: str, username: str, password: str) -> list[dict]:
    """Check krbtgt account password age."""
    if Connection is None:
        return []
    srv = Server(server_url, get_info=ALL, use_ssl=True)
    conn = Connection(srv, user=username, password=password, auto_bind=True)
    base_dn = conn.server.info.other.get("defaultNamingContext", [""])[0]

    conn.search(base_dn, "(&(objectClass=user)(sAMAccountName=krbtgt))",
                search_scope=SUBTREE, attributes=["pwdLastSet"])
    findings = []
    if conn.entries:
        pwd_set = conn.entries[0].pwdLastSet.value
        if pwd_set:
            age_days = (datetime.utcnow() - pwd_set.replace(tzinfo=None)).days
            if age_days > 180:
                findings.append({
                    "check": "krbtgt_password_age",
                    "value": age_days,
                    "severity": "critical",
                    "detail": f"krbtgt password is {age_days} days old (reset recommended every 180 days)",
                })
    conn.unbind()
    return findings


def assess_pingcastle_risks(risks: list[dict]) -> list[dict]:
    """Convert PingCastle risk rules into standardized findings."""
    findings = []
    for risk in risks:
        severity = "critical" if risk["points"] >= 50 else "high" if risk["points"] >= 20 else "medium" if risk["points"] >= 5 else "low"
        findings.append({
            "type": f"pingcastle_{risk['rule']}",
            "severity": severity,
            "category": risk["category"],
            "points": risk["points"],
            "detail": risk["rationale"],
        })
    return findings


def generate_report(xml_path: str = None, server_url: str = None,
                    username: str = None, password: str = None) -> dict:
    """Run all assessments and build consolidated report."""
    findings = []
    scores = {}

    if xml_path:
        pc_data = parse_pingcastle_report(xml_path)
        scores = pc_data["scores"]
        findings.extend(assess_pingcastle_risks(pc_data["risks"]))

    if server_url and username and password:
        findings.extend(check_password_policy_ldap(server_url, username, password))
        findings.extend(check_krbtgt_age(server_url, username, password))

    severity_counts = Counter(f.get("severity", "info") for f in findings)
    return {
        "report": "ad_vulnerability_assessment",
        "generated_at": datetime.utcnow().isoformat() + "Z",
        "pingcastle_scores": scores,
        "total_findings": len(findings),
        "severity_summary": dict(severity_counts),
        "findings": findings,
    }


def main():
    parser = argparse.ArgumentParser(description="AD Vulnerability Assessment Agent")
    parser.add_argument("--pingcastle-xml", help="PingCastle XML report file")
    parser.add_argument("--server", help="LDAP server URL for live checks")
    parser.add_argument("--username", help="Domain username")
    parser.add_argument("--password", help="Password")
    parser.add_argument("--output", help="Output JSON file path")
    args = parser.parse_args()

    if not args.pingcastle_xml and not args.server:
        parser.error("At least --pingcastle-xml or --server is required")

    report = generate_report(args.pingcastle_xml, args.server, args.username, args.password)
    output = json.dumps(report, indent=2)
    if args.output:
        Path(args.output).write_text(output, encoding="utf-8")
        print(f"Report written to {args.output}")
    else:
        print(output)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""Active Directory Vulnerability Assessment Analyzer.

Parses PingCastle and BloodHound output to generate consolidated
AD security assessment reports with prioritized remediation actions.
"""

import argparse
import csv
import json
import re
import sys
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path


RISK_WEIGHTS = {
    "kerberoastable_admin": 10,
    "unconstrained_delegation": 9,
    "as_rep_roastable": 8,
    "password_never_expires_admin": 8,
    "adminsdholder_modified": 8,
    "dcsync_non_dc": 9,
    "gpo_abuse_path": 7,
    "stale_admin_account": 6,
    "ldap_signing_disabled": 6,
    "ntlm_not_restricted": 5,
    "excessive_domain_admins": 7,
    "sid_history_present": 6,
    "trust_sid_filtering_disabled": 7,
    "unsupported_os_dc": 9,
    "password_policy_weak": 5,
}


def parse_pingcastle_xml(xml_path):
    """Parse PingCastle HTML/XML health check report."""
    findings = []
    try:
        tree = ET.parse(xml_path)
        root = tree.getroot()
        for risk in root.iter("RiskRule"):
            finding = {
                "source": "PingCastle",
                "category": risk.findtext("Category", ""),
                "rule_id": risk.findtext("RiskId", ""),
                "title": risk.findtext("Rationale", ""),
                "description": risk.findtext("Detail", ""),
                "points": int(risk.findtext("Points", "0")),
            }
            findings.append(finding)
    except ET.ParseError:
        print(f"[-] Could not parse PingCastle XML: {xml_path}")
        print("    Try exporting PingCastle results as XML format")
    return findings


def parse_bloodhound_json(json_path):
    """Parse BloodHound CE exported data for critical findings."""
    findings = []
    with open(json_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    if isinstance(data, dict):
        nodes = data.get("nodes", [])
        edges = data.get("edges", [])
    elif isinstance(data, list):
        nodes = data
        edges = []
    else:
        return findings

    for node in nodes:
        props = node.get("properties", node.get("Properties", {}))
        kind = node.get("kind", node.get("label", ""))

        if kind == "User":
            if props.get("hasspn", False) and props.get("admincount", False):
                findings.append({
                    "source": "BloodHound",
                    "category": "Kerberos",
                    "rule_id": "kerberoastable_admin",
                    "title": f"Kerberoastable admin account: {props.get('name', 'unknown')}",
                    "description": f"User {props.get('name')} has SPN set and is in admin group",
                    "risk_weight": RISK_WEIGHTS["kerberoastable_admin"],
                })
            if props.get("dontreqpreauth", False):
                findings.append({
                    "source": "BloodHound",
                    "category": "Kerberos",
                    "rule_id": "as_rep_roastable",
                    "title": f"AS-REP roastable account: {props.get('name', 'unknown')}",
                    "description": f"User {props.get('name')} has Kerberos pre-auth disabled",
                    "risk_weight": RISK_WEIGHTS["as_rep_roastable"],
                })
            if props.get("pwdneverexpires", False) and props.get("admincount", False):
                findings.append({
                    "source": "BloodHound",
                    "category": "Privileged Accounts",
                    "rule_id": "password_never_expires_admin",
                    "title": f"Admin with non-expiring password: {props.get('name', 'unknown')}",
                    "description": f"Privileged user {props.get('name')} has password set to never expire",
                    "risk_weight": RISK_WEIGHTS["password_never_expires_admin"],
                })

        elif kind == "Computer":
            if props.get("unconstraineddelegation", False):
                name = props.get("name", "unknown")
                if "DC" not in name.upper():
                    findings.append({
                        "source": "BloodHound",
                        "category": "Kerberos",
                        "rule_id": "unconstrained_delegation",
                        "title": f"Unconstrained delegation on non-DC: {name}",
                        "description": f"Computer {name} has unconstrained delegation enabled",
                        "risk_weight": RISK_WEIGHTS["unconstrained_delegation"],
                    })

    return findings


def consolidate_findings(pingcastle_findings, bloodhound_findings):
    """Merge and deduplicate findings from multiple tools."""
    all_findings = pingcastle_findings + bloodhound_findings

    for finding in all_findings:
        if "risk_weight" not in finding:
            points = finding.get("points", 0)
            if points >= 30:
                finding["risk_weight"] = 10
            elif points >= 20:
                finding["risk_weight"] = 8
            elif points >= 10:
                finding["risk_weight"] = 6
            elif points >= 5:
                finding["risk_weight"] = 4
            else:
                finding["risk_weight"] = 2

        rule_id = finding.get("rule_id", "")
        if rule_id in RISK_WEIGHTS:
            finding["risk_weight"] = max(finding["risk_weight"], RISK_WEIGHTS[rule_id])

    all_findings.sort(key=lambda f: f.get("risk_weight", 0), reverse=True)
    return all_findings


def generate_report(findings, output_path):
    """Generate consolidated AD assessment report."""
    report = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "total_findings": len(findings),
        "critical": len([f for f in findings if f.get("risk_weight", 0) >= 9]),
        "high": len([f for f in findings if 7 <= f.get("risk_weight", 0) < 9]),
        "medium": len([f for f in findings if 4 <= f.get("risk_weight", 0) < 7]),
        "low": len([f for f in findings if f.get("risk_weight", 0) < 4]),
        "findings": findings,
        "categories": {},
    }

    for f in findings:
        cat = f.get("category", "Other")
        if cat not in report["categories"]:
            report["categories"][cat] = 0
        report["categories"][cat] += 1

    with open(output_path, "w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2)

    print(f"[+] AD Assessment Report: {output_path}")
    print(f"    Total findings: {report['total_findings']}")
    print(f"    Critical: {report['critical']}")
    print(f"    High: {report['high']}")
    print(f"    Medium: {report['medium']}")
    print(f"    Low: {report['low']}")
    print(f"    Categories: {json.dumps(report['categories'], indent=6)}")
    return report


def main():
    parser = argparse.ArgumentParser(description="AD Vulnerability Assessment Analyzer")
    parser.add_argument("--pingcastle", help="PingCastle XML report path")
    parser.add_argument("--bloodhound", help="BloodHound JSON export path")
    parser.add_argument("--output", default="ad_assessment_report.json", help="Output report path")
    args = parser.parse_args()

    pc_findings = []
    bh_findings = []

    if args.pingcastle:
        print(f"[*] Parsing PingCastle report: {args.pingcastle}")
        pc_findings = parse_pingcastle_xml(args.pingcastle)
        print(f"    Found {len(pc_findings)} PingCastle findings")

    if args.bloodhound:
        print(f"[*] Parsing BloodHound data: {args.bloodhound}")
        bh_findings = parse_bloodhound_json(args.bloodhound)
        print(f"    Found {len(bh_findings)} BloodHound findings")

    if not pc_findings and not bh_findings:
        print("[-] No input files provided. Use --pingcastle and/or --bloodhound")
        parser.print_help()
        sys.exit(1)

    findings = consolidate_findings(pc_findings, bh_findings)
    generate_report(findings, args.output)


if __name__ == "__main__":
    main()