admin / Synapse-NetscanXi

"""
v7 post-scan auditors: SSL/TLS certificate expiry tracking and Active Directory
/ Domain Controller auditing.

Both are pure functions over the parsed host dict (port_details[].scripts and
host_scripts). They turn raw NSE script output into structured findings that the
dashboard and compliance engine consume. No network access happens here.
"""

import re
from datetime import datetime, timezone

# Days-before-expiry thresholds for certificate status.
CERT_EXPIRING_DAYS = 30

# Weak signature / key signals in an ssl-cert output.
_WEAK_CERT_SIGNALS = ("md5", "sha1with", "1024 bit", "512 bit", "md2")

# nmap ssl-cert prints "Not valid after:  2025-01-31T23:59:59"
_NOT_AFTER_RE = re.compile(r"Not valid after:\s*([0-9T:\-\+ ]+)", re.IGNORECASE)
_NOT_BEFORE_RE = re.compile(r"Not valid before:\s*([0-9T:\-\+ ]+)", re.IGNORECASE)
_SUBJECT_RE = re.compile(r"Subject:\s*(.+)")
_ISSUER_RE = re.compile(r"Issuer:\s*(.+)")
_CN_RE = re.compile(r"commonName=([^/\n,]+)")


def _parse_dt(raw):
    raw = (raw or "").strip()
    for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S%z"):
        try:
            dt = datetime.strptime(raw.split("+")[0].strip()
                                   if fmt == "%Y-%m-%dT%H:%M:%S" else raw, fmt)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return dt
        except ValueError:
            continue
    return None


def _cert_from_script(output, port_label, now=None):
    now = now or datetime.now(timezone.utc)
    after_m = _NOT_AFTER_RE.search(output)
    if not after_m:
        return None
    not_after = _parse_dt(after_m.group(1))
    if not not_after:
        return None
    before_m = _NOT_BEFORE_RE.search(output)
    not_before = _parse_dt(before_m.group(1)) if before_m else None

    days_left = (not_after - now).days

    subj = ""
    cn = _CN_RE.search(output)
    if cn:
        subj = cn.group(1).strip()
    else:
        sm = _SUBJECT_RE.search(output)
        if sm:
            subj = sm.group(1).strip()
    issuer = ""
    im = _ISSUER_RE.search(output)
    if im:
        issuer = im.group(1).strip()

    low = output.lower()
    weak = any(sig in low for sig in _WEAK_CERT_SIGNALS)

    if days_left < 0:
        status = "expired"
    elif weak:
        status = "weak"
    elif days_left <= CERT_EXPIRING_DAYS:
        status = "expiring"
    else:
        status = "ok"

    return {
        "port": port_label,
        "subject": subj,
        "issuer": issuer,
        "not_before": not_before.isoformat() if not_before else None,
        "not_after": not_after.isoformat(),
        "days_left": days_left,
        "weak": weak,
        "status": status,
    }


def certificate_findings(host, now=None):
    """Extract TLS certificate records from ssl-cert script output across ports."""
    certs = []
    for pd in host.get("port_details", []):
        port_label = f"{pd.get('port','?')}/{pd.get('protocol','tcp')}"
        for sc in pd.get("scripts", []):
            if sc.get("id") == "ssl-cert":
                rec = _cert_from_script(sc.get("output", ""), port_label, now=now)
                if rec:
                    certs.append(rec)
    # worst status first for display
    order = {"expired": 0, "weak": 1, "expiring": 2, "ok": 3}
    certs.sort(key=lambda c: order.get(c["status"], 9))
    return certs


# ---------------------------------------------------------------------------
# Active Directory / Domain Controller auditing
# ---------------------------------------------------------------------------
# Ports that, taken together, strongly indicate a Domain Controller.
_DC_PORTS = {88, 389, 636, 3268, 3269, 464}


def _open_ports(host):
    nums = set()
    for pd in host.get("port_details", []):
        try:
            nums.add(int(pd.get("port")))
        except (TypeError, ValueError):
            pass
    return nums


def _all_scripts(host):
    """Yield (port_label, script dict) for both port and host scripts."""
    for pd in host.get("port_details", []):
        label = f"{pd.get('port','?')}/{pd.get('protocol','tcp')}"
        for sc in pd.get("scripts", []):
            yield label, sc
    for sc in host.get("host_scripts", []):
        yield "host", sc


def ad_audit(host):
    """Identify a Domain Controller and report directory/auth hardening findings.

    Returns {is_dc, domain, forest, findings:[...]} or {is_dc: False} when the
    host shows no AD/DC characteristics.
    """
    open_ports = _open_ports(host)
    dc_signals = open_ports & _DC_PORTS
    scripts = {sid: out for _, sc in _all_scripts(host)
               for sid, out in [(sc.get("id", ""), sc.get("output", "") or "")]}

    has_ad_scripts = any(k in scripts for k in
                         ("ldap-rootdse", "smb-os-discovery", "smb-protocols"))
    is_dc = bool(dc_signals) or has_ad_scripts
    if not is_dc:
        return {"is_dc": False, "findings": []}

    findings = []
    domain = forest = ""

    osd = scripts.get("smb-os-discovery", "")
    if osd:
        dm = re.search(r"Domain name:\s*(\S+)", osd)
        fm = re.search(r"Forest name:\s*(\S+)", osd)
        if dm:
            domain = dm.group(1).strip()
        if fm:
            forest = fm.group(1).strip()

    rootdse = scripts.get("ldap-rootdse", "")
    if rootdse and not domain:
        dn = re.search(r"(?:defaultNamingContext|rootDomainNamingContext)[^A-Za-z]*"
                       r"((?:DC=[^,\n]+,?)+)", rootdse)
        if dn:
            domain = dn.group(1).strip()

    # SMBv1 still enabled -> serious hardening gap.
    protocols = scripts.get("smb-protocols", "")
    if re.search(r"\bSMBv1\b|NT LM 0\.12", protocols):
        findings.append({
            "id": "smbv1-enabled",
            "title": "SMBv1 enabled on domain controller",
            "severity": "critical",
            "detail": "Legacy SMBv1 is enabled; it is insecure and exploitable "
                      "(e.g. EternalBlue). Disable it on all DCs.",
        })

    # SMB signing not required -> NTLM relay risk on a DC.
    for sid in ("smb-security-mode", "smb2-security-mode"):
        out = scripts.get(sid, "")
        if out and ("not required" in out.lower() or "disabled" in out.lower()):
            findings.append({
                "id": f"smb-signing-{sid}",
                "title": "SMB signing not required",
                "severity": "high",
                "detail": "Message signing is not enforced; domain authentication "
                          "is exposed to NTLM relay attacks. Require SMB signing.",
            })
            break

    # Anonymous/again-exposed LDAP root DSE readable without auth is expected,
    # but flag if LDAP (cleartext 389) is the only directory access available.
    if 389 in open_ports and 636 not in open_ports:
        findings.append({
            "id": "ldap-no-tls",
            "title": "LDAP exposed without LDAPS (636)",
            "severity": "warn",
            "detail": "Directory queries available over cleartext LDAP (389) with "
                      "no LDAPS (636). Enable LDAP over TLS and consider requiring it.",
        })

    # Kerberos username enumeration possible.
    if scripts.get("krb5-enum-users"):
        if re.search(r"@", scripts["krb5-enum-users"]):
            findings.append({
                "id": "krb5-user-enum",
                "title": "Kerberos username enumeration possible",
                "severity": "warn",
                "detail": "Valid usernames can be enumerated via Kerberos "
                          "pre-auth responses. Restrict and monitor; enforce "
                          "lockout and strong passwords.",
            })

    return {
        "is_dc": True,
        "domain": domain,
        "forest": forest,
        "findings": findings,
    }