NTDS Hash Filter

#!/usr/bin/env python3
"""Filter user hashes from NetExec/Impacket-style NTDS output.

The default output is one NT hash per line (Hashcat mode 1000).  A summary and
LM-hash warnings are written to stderr so redirecting stdout creates a clean
Hashcat input file.
"""

from __future__ import annotations

import argparse
import glob
import json
import os
import re
import stat
import sys
from dataclasses import dataclass, replace
from pathlib import Path
from typing import Iterable, TextIO


EMPTY_LM = "aad3b435b51404eeaad3b435b51404ee"
HEX32_RE = re.compile(r"^[0-9a-fA-F]{32}$")
ANSI_RE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
RECORD_FIELDS_RE = re.compile(
    r":(?P<rid>\d+):(?P<lm>[^:\s]*):(?P<nt>[^:\s]*)(?=[:\s]|$)"
)
DOMAIN_ACCOUNT_RE = re.compile(r"(?P<account>[^\s:\\]+\\[^:\r\n]+)$")
STATUS_RE = re.compile(r"\bstatus\s*[:=]\s*(enabled|disabled)\b", re.I)
HISTORY_RE = re.compile(r"_history\d+$", re.I)


@dataclass(frozen=True)
class Record:
    account: str
    rid: int
    lm: str
    nt: str
    status: str
    source: str
    line_number: int

    @property
    def base_account(self) -> str:
        return HISTORY_RE.sub("", self.account)

    @property
    def is_machine(self) -> bool:
        return self.base_account.endswith("$")

    @property
    def is_history(self) -> bool:
        return HISTORY_RE.search(self.account) is not None

    @property
    def has_lm(self) -> bool:
        return self.lm != EMPTY_LM


@dataclass
class ParseStats:
    lines: int = 0
    malformed: int = 0
    duplicates: int = 0


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=(
            "Extract human-user NTLM hashes from NTDS output. With no input "
            "argument, piped stdin is read; on a terminal, the newest *.ntds "
            "file in common NetExec log locations is used."
        ),
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""examples:
  %(prog)s domain.ntds > users.ntlm
  %(prog)s - < domain.ntds > users.ntlm
  pbpaste | %(prog)s > users.ntlm
  %(prog)s domain.ntds -f user-ntlm > users_with_names.txt
  %(prog)s '/root/.nxc/logs/**/*.ntds' --all-matches > users.ntlm

Hashcat:
  hashcat -m 1000 users.ntlm WORDLIST
  hashcat -m 1000 --username users_with_names.txt WORDLIST
""",
    )
    parser.add_argument(
        "inputs",
        nargs="*",
        metavar="FILE|DIR|GLOB|-",
        help="NTDS file, directory, glob, or '-' for stdin",
    )
    parser.add_argument(
        "-f",
        "--format",
        choices=("ntlm", "user-ntlm", "pwdump", "json"),
        default="ntlm",
        help="output format (default: ntlm, suitable for Hashcat mode 1000)",
    )
    parser.add_argument(
        "-o",
        "--output",
        metavar="FILE",
        help="write output (mode 0600 on POSIX; inherited ACL on Windows)",
    )
    parser.add_argument(
        "--force", action="store_true", help="allow --output to replace an existing file"
    )
    parser.add_argument(
        "--include-disabled",
        action="store_true",
        help="include accounts explicitly marked Disabled",
    )
    parser.add_argument(
        "--enabled-only",
        action="store_true",
        help="include only accounts explicitly marked Enabled (drops unknown status)",
    )
    parser.add_argument(
        "--include-machines",
        action="store_true",
        help="include accounts whose names end in '$'",
    )
    parser.add_argument(
        "--include-history",
        action="store_true",
        help="include *_historyN password-history records",
    )
    parser.add_argument(
        "--all-matches",
        action="store_true",
        help="when a glob/directory matches several files, process all (default: newest)",
    )
    parser.add_argument(
        "-q", "--quiet", action="store_true", help="suppress summary and LM warnings"
    )
    args = parser.parse_args()
    if args.enabled_only and args.include_disabled:
        parser.error("--enabled-only and --include-disabled are mutually exclusive")
    if args.force and not args.output:
        parser.error("--force requires --output")
    return args


def common_ntds_candidates() -> list[Path]:
    roots = [
        Path("/root/.nxc/logs/ntds"),
        Path("/root/.nxc/logs"),
        Path.home() / ".nxc/logs/ntds",
        Path.home() / ".nxc/logs",
        Path.cwd(),
    ]
    candidates: dict[str, Path] = {}
    for root in roots:
        if not root.is_dir():
            continue
        try:
            for path in root.rglob("*.ntds"):
                if path.is_file():
                    candidates[str(path.resolve())] = path
        except PermissionError:
            continue
    return list(candidates.values())


def expand_inputs(raw_inputs: list[str], all_matches: bool) -> tuple[list[Path], bool]:
    if not raw_inputs:
        if not sys.stdin.isatty():
            return [], True
        candidates = common_ntds_candidates()
        if not candidates:
            raise ValueError(
                "no *.ntds file found in common NetExec locations; pass a file, "
                "pipe text, or use '-' and finish pasted input with Ctrl-D"
            )
        newest = max(candidates, key=lambda p: p.stat().st_mtime)
        return [newest], False

    use_stdin = False
    paths: dict[str, Path] = {}
    for item in raw_inputs:
        if item == "-":
            if use_stdin:
                raise ValueError("stdin ('-') may be specified only once")
            use_stdin = True
            continue

        expanded = [Path(p) for p in glob.glob(os.path.expanduser(item), recursive=True)]
        if not expanded:
            literal = Path(item).expanduser()
            if literal.exists():
                expanded = [literal]
            else:
                raise ValueError(f"input did not match a file or directory: {item}")

        matched: list[Path] = []
        for path in expanded:
            if path.is_dir():
                matched.extend(p for p in path.rglob("*.ntds") if p.is_file())
            elif path.is_file():
                matched.append(path)
        if not matched:
            raise ValueError(f"no *.ntds files found for input: {item}")
        if not all_matches and len(matched) > 1:
            matched = [max(matched, key=lambda p: p.stat().st_mtime)]
        for path in matched:
            paths[str(path.resolve())] = path
    return list(paths.values()), use_stdin


def account_from_prefix(prefix: str) -> str:
    """Recover the account immediately before the :RID:LM:NT boundary."""
    prefix = prefix.strip()
    domain_account = DOMAIN_ACCOUNT_RE.search(prefix)
    if domain_account:
        return domain_account.group("account").strip()

    # NetExec-style prefixes commonly end with a result marker. Keep spaces in
    # an unqualified sAMAccountName instead of taking only its final word.
    for marker in ("[+] ", "[-] ", "[*] "):
        marker_index = prefix.rfind(marker)
        if marker_index >= 0:
            return prefix[marker_index + len(marker) :].strip()
    return prefix


def parse_stream(stream: TextIO, source: str, stats: ParseStats) -> Iterable[Record]:
    for line_number, raw_line in enumerate(stream, 1):
        stats.lines += 1
        line = ANSI_RE.sub("", raw_line).strip()
        if not line:
            continue
        matches = list(RECORD_FIELDS_RE.finditer(line))
        if not matches:
            continue

        match = next(
            (
                candidate
                for candidate in reversed(matches)
                if HEX32_RE.fullmatch(candidate.group("lm"))
                and HEX32_RE.fullmatch(candidate.group("nt"))
            ),
            None,
        )
        if match is None:
            stats.malformed += 1
            continue

        account = account_from_prefix(line[: match.start()])
        if not account:
            stats.malformed += 1
            continue

        lm = match.group("lm").lower()
        nt = match.group("nt").lower()

        status_match = STATUS_RE.search(line)
        status = status_match.group(1).lower() if status_match else "unknown"
        yield Record(
            account=account,
            rid=int(match.group("rid")),
            lm=lm,
            nt=nt,
            status=status,
            source=source,
            line_number=line_number,
        )


def read_records(paths: list[Path], use_stdin: bool) -> tuple[list[Record], ParseStats]:
    stats = ParseStats()
    parsed_records: list[Record] = []

    streams: list[tuple[TextIO, str, bool]] = []
    if use_stdin:
        streams.append((sys.stdin, "<stdin>", False))
    for path in paths:
        try:
            stream = path.open("r", encoding="utf-8", errors="replace")
        except OSError as exc:
            raise ValueError(f"cannot read {path}: {exc}") from exc
        streams.append((stream, str(path), True))

    try:
        for stream, source, _ in streams:
            for record in parse_stream(stream, source, stats):
                parsed_records.append(record)
    finally:
        for stream, _, should_close in streams:
            if should_close:
                stream.close()
    records: list[Record] = []
    positions: dict[tuple[str, int, str, str], int] = {}
    for record in parsed_records:
        key = (record.account.casefold(), record.rid, record.lm, record.nt)
        existing_index = positions.get(key)
        if existing_index is not None:
            stats.duplicates += 1
            existing = records[existing_index]
            if existing.status == "unknown" and record.status != "unknown":
                records[existing_index] = record
            continue
        positions[key] = len(records)
        records.append(record)

    parent_status: dict[str, str] = {}
    for record in records:
        if record.is_history:
            continue
        key = record.account.casefold()
        if key not in parent_status or parent_status[key] == "unknown":
            parent_status[key] = record.status

    records = [
        replace(record, status=parent_status.get(record.base_account.casefold(), record.status))
        if record.is_history and record.status == "unknown"
        else record
        for record in records
    ]
    return records, stats


def select_records(records: list[Record], args: argparse.Namespace) -> list[Record]:
    selected = []
    for record in records:
        if record.is_machine and not args.include_machines:
            continue
        if record.is_history and not args.include_history:
            continue
        if args.enabled_only and record.status != "enabled":
            continue
        if not args.include_disabled and record.status == "disabled":
            continue
        selected.append(record)
    return selected


def render(records: list[Record], output_format: str) -> str:
    if output_format == "ntlm":
        return "".join(f"{record.nt}\n" for record in records)
    if output_format == "user-ntlm":
        return "".join(f"{record.account}:{record.nt}\n" for record in records)
    if output_format == "pwdump":
        return "".join(
            f"{record.account}:{record.rid}:{record.lm}:{record.nt}:::\n"
            for record in records
        )
    payload = [
        {
            "account": record.account,
            "rid": record.rid,
            "lm": record.lm,
            "ntlm": record.nt,
            "status": record.status,
            "source": record.source,
            "line": record.line_number,
        }
        for record in records
    ]
    return json.dumps(payload, indent=2) + "\n"


def open_secure_output(path_text: str, force: bool) -> TextIO:
    path = Path(path_text).expanduser()
    mode = stat.S_IRUSR | stat.S_IWUSR

    try:
        existing = path.lstat()
    except FileNotFoundError:
        existing = None
    except OSError as exc:
        raise ValueError(f"cannot inspect output {path}: {exc}") from exc

    if existing is not None and not force:
        raise ValueError(f"output exists (use --force to replace it): {path}")
    if existing is not None and not stat.S_ISREG(existing.st_mode):
        raise ValueError(f"refusing to replace non-regular output path: {path}")

    no_follow = getattr(os, "O_NOFOLLOW", 0)
    if os.name == "posix" and not no_follow:
        raise ValueError("secure output creation requires O_NOFOLLOW support")

    flags = os.O_WRONLY | no_follow
    if existing is not None:
        flags |= getattr(os, "O_NONBLOCK", 0)
    else:
        flags |= os.O_CREAT | os.O_EXCL

    try:
        fd = os.open(path, flags, mode)
    except FileExistsError as exc:
        raise ValueError(f"output appeared while opening it; retry: {path}") from exc
    except OSError as exc:
        raise ValueError(f"cannot create output {path}: {exc}") from exc

    try:
        opened = os.fstat(fd)
        if not stat.S_ISREG(opened.st_mode):
            raise ValueError(f"refusing to write non-regular output path: {path}")
        if existing is not None:
            if os.name == "posix" and (opened.st_dev, opened.st_ino) != (
                existing.st_dev,
                existing.st_ino,
            ):
                raise ValueError(f"output changed while opening it; retry: {path}")
            os.ftruncate(fd, 0)
        if os.name == "posix":
            os.fchmod(fd, mode)
    except Exception:
        os.close(fd)
        raise

    return os.fdopen(fd, "w", encoding="utf-8")


def report(
    paths: list[Path],
    use_stdin: bool,
    records: list[Record],
    selected: list[Record],
    stats: ParseStats,
    args: argparse.Namespace,
    written: bool,
) -> None:
    human_current = [r for r in records if not r.is_machine and not r.is_history]
    lm_enabled = [r for r in human_current if r.status == "enabled" and r.has_lm]
    lm_unknown = [r for r in human_current if r.status == "unknown" and r.has_lm]
    machines = sum(r.is_machine for r in records)
    history = sum(r.is_history for r in records)
    disabled = sum(
        r.status == "disabled" and not r.is_machine and not r.is_history for r in records
    )
    unknown = sum(
        r.status == "unknown" and not r.is_machine and not r.is_history for r in records
    )

    sources = (["<stdin>"] if use_stdin else []) + [str(path) for path in paths]
    print(f"[+] Sources: {', '.join(sources)}", file=sys.stderr)
    print(
        f"[+] Parsed {len(records)} unique record(s) from {stats.lines} line(s); "
        f"selected {len(selected)} record(s)",
        file=sys.stderr,
    )
    print(
        f"[+] Classified: {machines} $ account(s), {history} history row(s), "
        f"{disabled} disabled human user(s), {unknown} user(s) with unknown status",
        file=sys.stderr,
    )
    if stats.duplicates:
        print(f"[i] Ignored {stats.duplicates} exact duplicate record(s)", file=sys.stderr)
    if stats.malformed:
        print(
            f"[!] Ignored {stats.malformed} pwdump-like row(s) with a non-32-hex LM/NT field",
            file=sys.stderr,
        )

    if lm_enabled:
        names = ", ".join(record.account for record in lm_enabled)
        print(
            f"[!] LM HASH PRESENT on {len(lm_enabled)} enabled human account(s): {names}",
            file=sys.stderr,
        )
    else:
        print("[+] LM hash on explicitly enabled human accounts: none", file=sys.stderr)
    if lm_unknown:
        names = ", ".join(record.account for record in lm_unknown)
        print(
            f"[?] LM hash present on {len(lm_unknown)} account(s) whose enabled state is "
            f"unknown: {names}",
            file=sys.stderr,
        )

    if written:
        destination = args.output if args.output else "stdout"
        suffix = {
            "ntlm": "Hashcat mode 1000",
            "user-ntlm": "Hashcat mode 1000 with --username",
            "pwdump": "pwdump",
            "json": "JSON",
        }[args.format]
        print(
            f"[+] Wrote {len(selected)} {suffix} record(s) to {destination}",
            file=sys.stderr,
        )
        if args.output and os.name != "posix":
            print(
                "[!] Output permissions inherit the Windows ACL; verify access "
                "before storing credential material",
                file=sys.stderr,
            )
    else:
        print("[!] No valid NTDS hash records found; no output written", file=sys.stderr)


def main() -> int:
    args = parse_args()
    try:
        paths, use_stdin = expand_inputs(args.inputs, args.all_matches)
        records, stats = read_records(paths, use_stdin)
        selected = select_records(records, args)
        if not records:
            if not args.quiet:
                report(paths, use_stdin, records, selected, stats, args, written=False)
            return 1
        output_text = render(selected, args.format)
        if args.output:
            with open_secure_output(args.output, args.force) as output:
                output.write(output_text)
        else:
            sys.stdout.write(output_text)
        if not args.quiet:
            report(paths, use_stdin, records, selected, stats, args, written=True)
    except (ValueError, OSError) as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())