From 68abd128e6f2cd239a50578391d18f6c65bdfcb0 Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Sat, 16 May 2026 16:17:22 -0400 Subject: [PATCH 01/21] Sanitize NTLM hostname to prevent path traversal and DoS Strip non-alphanumeric characters (except hyphens and dots) from server-provided NTLM hostname before use in file paths or content. Prevents: - Path traversal via ../ in hostname (file creation outside ~/.nxc/logs/) - DoS via null byte (ValueError crash in open()) - DoS via { characters (KeyError crash in str.format()) - Newline injection in --generate-hosts-file output - Affects: SMB, RDP, VNC, WinRM, MSSQL credential dump and screenshot paths --- nxc/connection.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/nxc/connection.py b/nxc/connection.py index f20d05ddaa..b4926a0472 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -1,5 +1,6 @@ from datetime import datetime import os +import re import random import sys import contextlib @@ -245,6 +246,7 @@ def proto_flow(self): else: self.logger.debug("Created connection object") self.enum_host_info() + self.hostname = re.sub(r'[^\w\-.]', '_', self.hostname) # Construct the output file template using os.path.join for OS compatibility base_log_dir = os.path.join(NXC_PATH, "logs") From 9c293a8eb9c70af105b4f1f280b264419e86827c Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Mon, 18 May 2026 14:42:53 -0400 Subject: [PATCH 02/21] Warn when NTLM hostname is sanitized Log a warning showing the original and sanitized hostname so users are alerted to potential non-compliant implementations or rogue servers. --- nxc/connection.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/nxc/connection.py b/nxc/connection.py index b4926a0472..17ff3cfc87 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -246,7 +246,10 @@ def proto_flow(self): else: self.logger.debug("Created connection object") self.enum_host_info() - self.hostname = re.sub(r'[^\w\-.]', '_', self.hostname) + sanitized = re.sub(r'[^\w\-.]', '_', self.hostname) + if sanitized != self.hostname: + self.logger.warning(f"Hostname contains invalid characters (received: {self.hostname!r}), sanitized to: {sanitized!r}") + self.hostname = sanitized # Construct the output file template using os.path.join for OS compatibility base_log_dir = os.path.join(NXC_PATH, "logs") From ccf36b492e7253d4a3ecc9c9edce3b9586a0f0ff Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Sat, 23 May 2026 15:52:44 -0400 Subject: [PATCH 03/21] Use display() for sanitization notice so it shows at default verbosity --- nxc/connection.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nxc/connection.py b/nxc/connection.py index 17ff3cfc87..76a1392ee4 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -248,7 +248,7 @@ def proto_flow(self): self.enum_host_info() sanitized = re.sub(r'[^\w\-.]', '_', self.hostname) if sanitized != self.hostname: - self.logger.warning(f"Hostname contains invalid characters (received: {self.hostname!r}), sanitized to: {sanitized!r}") + self.logger.display(f"Hostname contains invalid characters (received: {self.hostname!r}), sanitized to: {sanitized!r}") self.hostname = sanitized # Construct the output file template using os.path.join for OS compatibility From c20060a64a11aa08f697344bcb58aa63b8acf4cb Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Tue, 2 Jun 2026 16:49:46 -0400 Subject: [PATCH 04/21] Refactor hostname sanitization into helpers/misc.py Move the hostname sanitization regex from an inline check in connection.py into a reusable sanitize_hostname() function in nxc/helpers/misc.py, and apply it at the source in each protocol's enum_host_info where hostnames are received from NTLM/server data. Protocols covered: SMB, WinRM, WMI, MSSQL, RDP. Addresses review feedback on PR #1243. --- nxc/connection.py | 5 ----- nxc/helpers/misc.py | 14 ++++++++++++++ nxc/protocols/mssql.py | 4 ++-- nxc/protocols/rdp.py | 3 ++- nxc/protocols/smb.py | 2 ++ nxc/protocols/winrm.py | 3 ++- nxc/protocols/wmi.py | 3 ++- 7 files changed, 24 insertions(+), 10 deletions(-) diff --git a/nxc/connection.py b/nxc/connection.py index 76a1392ee4..f20d05ddaa 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -1,6 +1,5 @@ from datetime import datetime import os -import re import random import sys import contextlib @@ -246,10 +245,6 @@ def proto_flow(self): else: self.logger.debug("Created connection object") self.enum_host_info() - sanitized = re.sub(r'[^\w\-.]', '_', self.hostname) - if sanitized != self.hostname: - self.logger.display(f"Hostname contains invalid characters (received: {self.hostname!r}), sanitized to: {sanitized!r}") - self.hostname = sanitized # Construct the output file template using os.path.join for OS compatibility base_log_dir = os.path.join(NXC_PATH, "logs") diff --git a/nxc/helpers/misc.py b/nxc/helpers/misc.py index dd9dbe8705..68ca41edf1 100755 --- a/nxc/helpers/misc.py +++ b/nxc/helpers/misc.py @@ -26,6 +26,20 @@ def gen_random_string(length=10): return "".join(random.sample(string.ascii_letters, int(length))) +_HOSTNAME_SANITIZE_RE = re.compile(r"[^\w\-.]") + + +def sanitize_hostname(hostname, logger=None): + """Strip characters from a server-provided hostname that could cause path + traversal, newline injection, or format-string issues when used in file + paths or output content. Logs a warning when the value is modified. + """ + sanitized = _HOSTNAME_SANITIZE_RE.sub("_", hostname) + if sanitized != hostname and logger: + logger.display(f"Hostname contained invalid characters (received: {hostname!r}), sanitized to: {sanitized!r}") + return sanitized + + def validate_ntlm(data): allowed = re.compile(r"^[0-9a-f]{32}", re.IGNORECASE) return bool(allowed.match(data)) diff --git a/nxc/protocols/mssql.py b/nxc/protocols/mssql.py index b7edc248ec..8127bf3d7b 100755 --- a/nxc/protocols/mssql.py +++ b/nxc/protocols/mssql.py @@ -6,7 +6,7 @@ from nxc.config import process_secret, host_info_colors from nxc.connection import connection from nxc.connection import requires_admin -from nxc.helpers.misc import gen_random_string +from nxc.helpers.misc import gen_random_string, sanitize_hostname from nxc.logger import NXCAdapter from nxc.helpers.bloodhound import add_user_bh from nxc.helpers.negotiate_parser import parse_challenge, login7_integrated_auth_error_message @@ -138,7 +138,7 @@ def enum_host_info(self): if challenge.startswith(b"NTLMSSP\x00"): ntlm_info = parse_challenge(challenge) self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = ntlm_info["hostname"] + self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: diff --git a/nxc/protocols/rdp.py b/nxc/protocols/rdp.py index fdbdd923c4..a84dda304e 100644 --- a/nxc/protocols/rdp.py +++ b/nxc/protocols/rdp.py @@ -8,6 +8,7 @@ from impacket.krb5.ccache import CCache from nxc.connection import connection +from nxc.helpers.misc import sanitize_hostname from nxc.helpers.bloodhound import add_user_bh from nxc.logger import NXCAdapter from nxc.config import host_info_colors, process_secret @@ -142,7 +143,7 @@ def create_conn_obj(self): pass else: self.domain = info_domain["dnsdomainname"] - self.hostname = info_domain["computername"] + self.hostname = sanitize_hostname(info_domain["computername"], self.logger) self.server_os = info_domain["os_guess"] + " Build " + str(info_domain["os_build"]) self.logger.extra["hostname"] = self.hostname break diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index 6c5618464e..431ad051f8 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -6,6 +6,7 @@ import ipaddress from Cryptodome.Hash import MD4 from textwrap import dedent +from nxc.helpers.misc import sanitize_hostname from impacket.smbconnection import SMBConnection, SessionError from impacket.smb import SMB_DIALECT @@ -200,6 +201,7 @@ def enum_host_info(self): self.hostname = dns_hostname else: self.hostname = self.conn.getServerName() + self.hostname = sanitize_hostname(self.hostname, self.logger) self.targetDomain = self.conn.getServerDNSDomainName() if not self.targetDomain: # Not sure if that can even happen but now we are safe self.targetDomain = self.hostname diff --git a/nxc/protocols/winrm.py b/nxc/protocols/winrm.py index 50bd0010fa..3f758b8c9f 100644 --- a/nxc/protocols/winrm.py +++ b/nxc/protocols/winrm.py @@ -8,6 +8,7 @@ import xml.etree.ElementTree as ET from pypsrp.wsman import NAMESPACES +from nxc.helpers.misc import sanitize_hostname from pypsrp.client import Client from pypsrp.powershell import PSDataStreams from termcolor import colored @@ -68,7 +69,7 @@ def enum_host_info(self): return False self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = ntlm_info["hostname"] + self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname diff --git a/nxc/protocols/wmi.py b/nxc/protocols/wmi.py index c5e5af0dea..845adc0702 100644 --- a/nxc/protocols/wmi.py +++ b/nxc/protocols/wmi.py @@ -2,6 +2,7 @@ from io import StringIO from nxc.helpers.negotiate_parser import parse_challenge +from nxc.helpers.misc import sanitize_hostname from nxc.config import process_secret from nxc.connection import connection, dcom_FirewallChecker, requires_admin from nxc.logger import NXCAdapter @@ -130,7 +131,7 @@ def enum_host_info(self): bindResp = MSRPCBindAck(response.getData()) ntlm_info = parse_challenge(bindResp["auth_data"]) self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = ntlm_info["hostname"] + self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: From 09f3f823c64333e3c4506f068c47c38e7eb08b35 Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Sat, 6 Jun 2026 16:19:38 -0400 Subject: [PATCH 05/21] Address review: make logger required, use .fail() for sanitization warnings --- nxc/helpers/misc.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/nxc/helpers/misc.py b/nxc/helpers/misc.py index 68ca41edf1..2aa98d138f 100755 --- a/nxc/helpers/misc.py +++ b/nxc/helpers/misc.py @@ -29,14 +29,14 @@ def gen_random_string(length=10): _HOSTNAME_SANITIZE_RE = re.compile(r"[^\w\-.]") -def sanitize_hostname(hostname, logger=None): +def sanitize_hostname(hostname, logger): """Strip characters from a server-provided hostname that could cause path traversal, newline injection, or format-string issues when used in file paths or output content. Logs a warning when the value is modified. """ sanitized = _HOSTNAME_SANITIZE_RE.sub("_", hostname) - if sanitized != hostname and logger: - logger.display(f"Hostname contained invalid characters (received: {hostname!r}), sanitized to: {sanitized!r}") + if sanitized != hostname: + logger.fail(f"Hostname contained invalid characters (received: {hostname!r}), sanitized to: {sanitized!r}") return sanitized From e63a43c1f0080f7019ba6901fc36ebdb5301e73c Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Sun, 20 Sep 2026 08:04:08 -0400 Subject: [PATCH 06/21] Improve DNS check to be compliant with referenced RFCs --- nxc/helpers/misc.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/nxc/helpers/misc.py b/nxc/helpers/misc.py index 2aa98d138f..6f9a659115 100755 --- a/nxc/helpers/misc.py +++ b/nxc/helpers/misc.py @@ -26,18 +26,18 @@ def gen_random_string(length=10): return "".join(random.sample(string.ascii_letters, int(length))) -_HOSTNAME_SANITIZE_RE = re.compile(r"[^\w\-.]") - - -def sanitize_hostname(hostname, logger): - """Strip characters from a server-provided hostname that could cause path - traversal, newline injection, or format-string issues when used in file - paths or output content. Logs a warning when the value is modified. - """ - sanitized = _HOSTNAME_SANITIZE_RE.sub("_", hostname) - if sanitized != hostname: - logger.fail(f"Hostname contained invalid characters (received: {hostname!r}), sanitized to: {sanitized!r}") - return sanitized +def sanitize_dns(hostname, logger): + """Check that the hostname is compliant with DNS naming conventions and sanitize it if necessary.""" + # As defined in: https://learn.microsoft.com/en-us/troubleshoot/windows-server/active-directory/naming-conventions-for-computer-domain-site-ou + # and RFCs 952, 1123 this should restrict DNS names (including the hostname) to the following regex + # Taken from https://stackoverflow.com/a/2063247 + DNS_REGEX = re.compile(r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{,63}(? Date: Sun, 20 Sep 2026 08:11:15 -0400 Subject: [PATCH 07/21] Rename sanitizing function --- nxc/protocols/mssql.py | 4 ++-- nxc/protocols/rdp.py | 4 ++-- nxc/protocols/smb.py | 10 +++++----- nxc/protocols/winrm.py | 4 ++-- nxc/protocols/wmi.py | 4 ++-- 5 files changed, 13 insertions(+), 13 deletions(-) diff --git a/nxc/protocols/mssql.py b/nxc/protocols/mssql.py index d367cdcf83..c65f6f8c32 100755 --- a/nxc/protocols/mssql.py +++ b/nxc/protocols/mssql.py @@ -7,7 +7,7 @@ from nxc.config import process_secret, host_info_colors from nxc.connection import connection from nxc.connection import requires_admin -from nxc.helpers.misc import gen_random_string, sanitize_hostname +from nxc.helpers.misc import gen_random_string, sanitize_dns from nxc.logger import NXCAdapter from nxc.helpers.bloodhound import add_user_bh from nxc.helpers.negotiate_parser import parse_challenge, login7_integrated_auth_error_message @@ -156,7 +156,7 @@ def enum_host_info(self): if challenge.startswith(b"NTLMSSP\x00"): ntlm_info = parse_challenge(challenge) self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) + self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: diff --git a/nxc/protocols/rdp.py b/nxc/protocols/rdp.py index a84dda304e..698fdae2cd 100644 --- a/nxc/protocols/rdp.py +++ b/nxc/protocols/rdp.py @@ -8,7 +8,7 @@ from impacket.krb5.ccache import CCache from nxc.connection import connection -from nxc.helpers.misc import sanitize_hostname +from nxc.helpers.misc import sanitize_dns from nxc.helpers.bloodhound import add_user_bh from nxc.logger import NXCAdapter from nxc.config import host_info_colors, process_secret @@ -143,7 +143,7 @@ def create_conn_obj(self): pass else: self.domain = info_domain["dnsdomainname"] - self.hostname = sanitize_hostname(info_domain["computername"], self.logger) + self.hostname = sanitize_dns(info_domain["computername"], self.logger) self.server_os = info_domain["os_guess"] + " Build " + str(info_domain["os_build"]) self.logger.extra["hostname"] = self.hostname break diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index 48c02b1570..e6af66fa7e 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -9,7 +9,7 @@ from nxc.helpers.path import sanitize_filename from Cryptodome.Hash import MD4 from textwrap import dedent -from nxc.helpers.misc import sanitize_hostname +from nxc.helpers.misc import sanitize_dns from impacket.smbconnection import SMBConnection, SessionError from impacket.smb import SMB_DIALECT @@ -210,12 +210,12 @@ def enum_host_info(self): # Try to get hostname with getServerDNSHostName as getServerName is truncated to 15 chars dns_hostname = self.conn.getServerDNSHostName().upper() if dns_hostname and "." in dns_hostname: - self.hostname = dns_hostname.split(".")[0] + hostname = dns_hostname.split(".")[0] elif dns_hostname: - self.hostname = dns_hostname + hostname = dns_hostname else: - self.hostname = self.conn.getServerName() - self.hostname = sanitize_hostname(self.hostname, self.logger) + hostname = self.conn.getServerName() + self.hostname = sanitize_dns(hostname, self.logger) self.targetDomain = self.conn.getServerDNSDomainName() if not self.targetDomain: # Not sure if that can even happen but now we are safe self.targetDomain = self.hostname diff --git a/nxc/protocols/winrm.py b/nxc/protocols/winrm.py index 5619f88a28..b17126d3f2 100644 --- a/nxc/protocols/winrm.py +++ b/nxc/protocols/winrm.py @@ -8,7 +8,7 @@ import xml.etree.ElementTree as ET from pypsrp.wsman import NAMESPACES -from nxc.helpers.misc import sanitize_hostname +from nxc.helpers.misc import sanitize_dns from pypsrp.client import Client from pypsrp.powershell import PSDataStreams from termcolor import colored @@ -70,7 +70,7 @@ def enum_host_info(self): return False self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) + self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname diff --git a/nxc/protocols/wmi.py b/nxc/protocols/wmi.py index 434845be84..b2dcf430dc 100644 --- a/nxc/protocols/wmi.py +++ b/nxc/protocols/wmi.py @@ -2,7 +2,7 @@ from io import StringIO from nxc.helpers.negotiate_parser import parse_challenge -from nxc.helpers.misc import sanitize_hostname +from nxc.helpers.misc import sanitize_dns from nxc.config import process_secret from nxc.connection import connection, dcom_FirewallChecker, requires_admin from nxc.logger import NXCAdapter @@ -138,7 +138,7 @@ def enum_host_info(self): bindResp = MSRPCBindAck(response.getData()) ntlm_info = parse_challenge(bindResp["auth_data"]) self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_hostname(ntlm_info["hostname"], self.logger) + self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: From c375b3bb8ce44a34b7cd6ca89e54153679ab5020 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Wed, 23 Sep 2026 14:52:09 -0400 Subject: [PATCH 08/21] Add DNS sanitization function --- nxc/helpers/misc.py | 104 +++++++++++++++++++++++++++++++++++++++----- 1 file changed, 93 insertions(+), 11 deletions(-) diff --git a/nxc/helpers/misc.py b/nxc/helpers/misc.py index 6f9a659115..e026e43bff 100755 --- a/nxc/helpers/misc.py +++ b/nxc/helpers/misc.py @@ -1,9 +1,12 @@ +from contextlib import suppress from enum import Enum +import hashlib import random import string import re import inspect import os +from unicodedata import normalize from termcolor import colored from ipaddress import ip_address from nxc.logger import nxc_logger @@ -26,18 +29,97 @@ def gen_random_string(length=10): return "".join(random.sample(string.ascii_letters, int(length))) -def sanitize_dns(hostname, logger): - """Check that the hostname is compliant with DNS naming conventions and sanitize it if necessary.""" - # As defined in: https://learn.microsoft.com/en-us/troubleshoot/windows-server/active-directory/naming-conventions-for-computer-domain-site-ou - # and RFCs 952, 1123 this should restrict DNS names (including the hostname) to the following regex - # Taken from https://stackoverflow.com/a/2063247 - DNS_REGEX = re.compile(r"^(?![0-9]+$)(?!-)[a-zA-Z0-9-]{,63}(?:"/\\|?*{}[]=#;\'' # Portable paths, format strings, hosts files, and krb5.conf + sanitized = "".join( + character if character.isprintable() + and not character.isspace() + and character not in unsafe_characters + and all( + normalized_character.isprintable() + and not normalized_character.isspace() + and normalized_character not in unsafe_characters + for normalized_character in normalize("NFKC", character) + ) + else "_" + for character in hostname + ) + + # Neutralize traversal segments and Windows-trimmed trailing dots. + if normalize("NFKC", sanitized) in (".", ".."): + sanitized = "_" * len(sanitized) + if sanitized.endswith(".") or normalize("NFKC", sanitized[-1]).endswith("."): + sanitized = f"{sanitized[:-1]}_" + + # Avoid Windows device names when the result is used as a filename. + normalized_stem = normalize("NFKC", sanitized).split(".", 1)[0].upper() + if normalized_stem in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} or re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem): + sanitized = f"_{sanitized}" + + # Bound filename length while retaining a stable identifier for long names. + if len(sanitized.encode("utf-8")) > 253: + suffix = f"_{hashlib.sha256(hostname.encode('utf-8', errors='surrogatepass')).hexdigest()[:12]}" + byte_length = 0 + truncated = [] + for character in sanitized: + character_length = len(character.encode("utf-8")) + if byte_length + character_length > 253 - len(suffix): + break + truncated.append(character) + byte_length += character_length + sanitized = f"{''.join(truncated)}{suffix}" + + # Fail closed if a future change violates any output invariant. + normalized = normalize("NFKC", sanitized) + normalized_stem = normalized.split(".", 1)[0].upper() + if ( + not sanitized + or not normalized + or len(sanitized.encode("utf-8")) > 253 + or normalized in (".", "..") + or normalized.endswith(".") + or normalized_stem in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} + or re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem) + or any( + not character.isprintable() + or character.isspace() + or character in unsafe_characters + or any( + not normalized_character.isprintable() + or normalized_character.isspace() + or normalized_character in unsafe_characters + for normalized_character in normalize("NFKC", character) + ) + for character in sanitized + ) + ): + sanitized = "_" + + # Report changed input without allowing logging failures to affect safety. + if sanitized != hostname: + received = ascii(hostname[:256]) + result = ascii(sanitized) + if len(hostname) > 256 or len(received) > 256: + received = f"{received[:253]}..." + if len(result) > 256: + result = f"{result[:253]}..." + if logger is not None: + with suppress(Exception): + logger.fail(f"Unsafe hostname or domain received: {received}; using {result}") + return sanitized def validate_ntlm(data): From a995133e833af4252112be0728bd754b579c244e Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Wed, 23 Sep 2026 14:57:19 -0400 Subject: [PATCH 09/21] Add path sanitizer --- nxc/helpers/path.py | 87 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 87 insertions(+) diff --git a/nxc/helpers/path.py b/nxc/helpers/path.py index a4dd96029e..c050546ff3 100644 --- a/nxc/helpers/path.py +++ b/nxc/helpers/path.py @@ -1,4 +1,91 @@ +import hashlib from pathlib import PurePosixPath +import re +from unicodedata import normalize + + +def sanitize_path_component(name, max_bytes=255): + """Return one portable, bounded path component from untrusted input.""" + if max_bytes < 14: + raise ValueError("max_bytes must be at least 14") + + try: + name = name.decode("utf-8", errors="surrogateescape") if isinstance(name, bytes) else str(name) if name is not None else "" + except Exception: + name = "" + if not name: + return "_" + + unsafe_characters = '<>:"/\\|?*{}' # Portable filename and format-string metacharacters + sanitized = "".join( + character if character.isprintable() + and character not in unsafe_characters + and all( + normalized_character.isprintable() and normalized_character not in unsafe_characters + for normalized_character in normalize("NFKC", character) + ) + else "_" + for character in name + ) + + if normalize("NFKC", sanitized) in (".", ".."): + sanitized = "_" * len(sanitized) + while sanitized and ( + sanitized[-1] == "." + or sanitized[-1].isspace() + or normalize("NFKC", sanitized[-1]).endswith(".") + or any(character.isspace() for character in normalize("NFKC", sanitized[-1])) + ): + sanitized = f"{sanitized[:-1]}_" + + normalized_stem = normalize("NFKC", sanitized).split(".", 1)[0].rstrip(" ").upper() + if normalized_stem in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} or re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem): + sanitized = f"_{sanitized}" + + if len(sanitized.encode("utf-8")) > max_bytes: + digest = f"_{hashlib.sha256(name.encode('utf-8', errors='surrogatepass')).hexdigest()[:12]}" + extension = "" + extension_index = sanitized.rfind(".") + if ( + extension_index > 0 + and len(sanitized[extension_index:].encode("utf-8")) <= 32 + and len(sanitized[extension_index:].encode("utf-8")) <= max_bytes - len(digest) + ): + extension = sanitized[extension_index:] + sanitized = sanitized[:extension_index] + byte_length = 0 + truncated = [] + for character in sanitized: + character_length = len(character.encode("utf-8")) + if byte_length + character_length > max_bytes - len(digest) - len(extension.encode("utf-8")): + break + truncated.append(character) + byte_length += character_length + sanitized = f"{''.join(truncated)}{digest}{extension}" + + normalized = normalize("NFKC", sanitized) + normalized_stem = normalized.split(".", 1)[0].rstrip(" ").upper() + if ( + not sanitized + or not normalized + or len(sanitized.encode("utf-8")) > max_bytes + or normalized in (".", "..") + or normalized.endswith(".") + or normalized[-1].isspace() + or normalized_stem in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} + or re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem) + or any( + not character.isprintable() + or character in unsafe_characters + or any( + not normalized_character.isprintable() or normalized_character in unsafe_characters + for normalized_character in normalize("NFKC", character) + ) + for character in sanitized + ) + ): + return "_" + return sanitized def sanitize_filename(name: str) -> str: From 23be541d0a9388d273ff4f8deb251c80035be300 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Wed, 23 Sep 2026 15:02:47 -0400 Subject: [PATCH 10/21] Add comments --- nxc/helpers/path.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/nxc/helpers/path.py b/nxc/helpers/path.py index c050546ff3..b2dd40f242 100644 --- a/nxc/helpers/path.py +++ b/nxc/helpers/path.py @@ -6,9 +6,11 @@ def sanitize_path_component(name, max_bytes=255): """Return one portable, bounded path component from untrusted input.""" + # Reserve enough space for one character and the collision-resistant suffix. if max_bytes < 14: raise ValueError("max_bytes must be at least 14") + # Always provide a safe fallback, even for missing or unstringable input. try: name = name.decode("utf-8", errors="surrogateescape") if isinstance(name, bytes) else str(name) if name is not None else "" except Exception: @@ -16,6 +18,7 @@ def sanitize_path_component(name, max_bytes=255): if not name: return "_" + # Replace path, formatting, control, and Unicode-equivalent metacharacters. unsafe_characters = '<>:"/\\|?*{}' # Portable filename and format-string metacharacters sanitized = "".join( character if character.isprintable() @@ -28,6 +31,7 @@ def sanitize_path_component(name, max_bytes=255): for character in name ) + # Neutralize traversal-only names and Windows-trimmed trailing dots or spaces. if normalize("NFKC", sanitized) in (".", ".."): sanitized = "_" * len(sanitized) while sanitized and ( @@ -38,10 +42,12 @@ def sanitize_path_component(name, max_bytes=255): ): sanitized = f"{sanitized[:-1]}_" + # Avoid Windows device names, including names followed by an extension. normalized_stem = normalize("NFKC", sanitized).split(".", 1)[0].rstrip(" ").upper() if normalized_stem in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} or re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem): sanitized = f"_{sanitized}" + # Bound the byte length with a stable digest while preserving short extensions. if len(sanitized.encode("utf-8")) > max_bytes: digest = f"_{hashlib.sha256(name.encode('utf-8', errors='surrogatepass')).hexdigest()[:12]}" extension = "" @@ -63,6 +69,7 @@ def sanitize_path_component(name, max_bytes=255): byte_length += character_length sanitized = f"{''.join(truncated)}{digest}{extension}" + # Fail closed if a future change violates any output invariant. normalized = normalize("NFKC", sanitized) normalized_stem = normalized.split(".", 1)[0].rstrip(" ").upper() if ( From c812041b171765806e5cd271fecdba2d7264c780 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Wed, 23 Sep 2026 15:22:05 -0400 Subject: [PATCH 11/21] Apply path sanitization --- nxc/connection.py | 3 ++- nxc/helpers/pfx.py | 5 +++-- nxc/modules/certipy-find.py | 14 ++++++++------ nxc/modules/enum_dns.py | 3 ++- nxc/modules/get-network.py | 5 +++-- nxc/modules/get_netconnections.py | 3 ++- nxc/modules/mssql_dumper.py | 5 +++-- nxc/modules/nanodump.py | 3 ++- nxc/modules/obsolete.py | 6 ++++-- nxc/modules/pre2k.py | 3 ++- nxc/modules/user-desc.py | 3 ++- nxc/protocols/rdp.py | 7 +++++-- nxc/protocols/smb.py | 6 +++--- nxc/protocols/vnc.py | 3 ++- 14 files changed, 43 insertions(+), 26 deletions(-) diff --git a/nxc/connection.py b/nxc/connection.py index f575dc97cf..d619ffef7f 100755 --- a/nxc/connection.py +++ b/nxc/connection.py @@ -14,6 +14,7 @@ from nxc.config import pwned_label from nxc.helpers.logger import highlight +from nxc.helpers.path import sanitize_path_component from nxc.loaders.moduleloader import ModuleLoader, ModuleOptionsError from nxc.logger import nxc_logger, NXCAdapter from nxc.context import Context @@ -251,7 +252,7 @@ def proto_flow(self): # Construct the output file template using os.path.join for OS compatibility base_log_dir = os.path.join(NXC_PATH, "logs") - filename_pattern = f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}".replace(":", "-") + filename_pattern = sanitize_path_component(f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}", max_bytes=220) self.output_file_template = os.path.join(base_log_dir, "{output_folder}", filename_pattern) # Default output filename for logs self.output_filename = os.path.join(base_log_dir, filename_pattern) diff --git a/nxc/helpers/pfx.py b/nxc/helpers/pfx.py index 941ca237fb..4ccdbfcfa4 100644 --- a/nxc/helpers/pfx.py +++ b/nxc/helpers/pfx.py @@ -71,6 +71,7 @@ from impacket.krb5.ccache import CCache as impacket_CCache from nxc.paths import NXC_PATH +from nxc.helpers.path import sanitize_path_component from nxc.logger import nxc_logger @@ -530,8 +531,8 @@ def pfx_auth(self): return False username = self.args.username[0] - basename = f"{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}-{username}.ccache" - log_ccache = os.path.normpath(os.path.expanduser(f"{NXC_PATH}/logs/{basename}")) + basename = sanitize_path_component(f"{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}-{username}.ccache") + log_ccache = os.path.normpath(os.path.expanduser(os.path.join(NXC_PATH, "logs", basename))) # Request a TGT with the cert data req = ini.build_asreq(self.domain, username) diff --git a/nxc/modules/certipy-find.py b/nxc/modules/certipy-find.py index ef6884cfbc..df74173989 100644 --- a/nxc/modules/certipy-find.py +++ b/nxc/modules/certipy-find.py @@ -2,12 +2,14 @@ import json import socket from os import makedirs +from os.path import join from certipy.commands.find import Find from certipy.lib.target import Target, DnsResolver from certipy.lib.formatting import pretty_print from datetime import datetime from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH @@ -34,7 +36,7 @@ def options(self, context, module_options): """ self.vuln = True self.enabled = False - self.output_path = f"{NXC_PATH}/modules/certipy-find" + self.output_path = join(NXC_PATH, "modules", "certipy-find") self.json = False self.csv = False self.text = False @@ -119,9 +121,9 @@ def on_login(self, context, connection): if self.json or self.csv or self.text: makedirs(self.output_path, exist_ok=True) - filename = f"certipy_{connection.hostname}_{connection.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}".replace(":", "-") + filename = sanitize_path_component(f"certipy_{connection.hostname}_{connection.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}", max_bytes=220) if self.json: - with open(f"{self.output_path}/{filename}.json", "w") as f: + with open(join(self.output_path, f"{filename}.json"), "w") as f: json.dump( output, f, @@ -131,10 +133,10 @@ def on_login(self, context, connection): if self.csv: template_output = finder.get_template_output_for_csv(output) ca_output = finder.get_ca_output_for_csv(output) - with open(f"{self.output_path}/{filename}-templates.csv", "w") as f: + with open(join(self.output_path, f"{filename}-templates.csv"), "w") as f: f.write(template_output) - with open(f"{self.output_path}/{filename}-cas.csv", "w") as f: + with open(join(self.output_path, f"{filename}-cas.csv"), "w") as f: f.write(ca_output) if self.text: - with open(f"{self.output_path}/{filename}.txt", "w") as f: + with open(join(self.output_path, f"{filename}.txt"), "w") as f: pretty_print(output, print_func=lambda x: f.write(x + "\n")) diff --git a/nxc/modules/enum_dns.py b/nxc/modules/enum_dns.py index cb295a6abc..7c1c4bed2a 100644 --- a/nxc/modules/enum_dns.py +++ b/nxc/modules/enum_dns.py @@ -1,6 +1,7 @@ from datetime import datetime from nxc.helpers.logger import write_log from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH @@ -63,6 +64,6 @@ def on_admin_login(self, context, connection): context.log.highlight("\t" + d) data += "\t" + d + "\n" - log_name = f"DNS-Enum-{connection.host}-{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log" + log_name = sanitize_path_component(f"DNS-Enum-{connection.host}-{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log") write_log(data, log_name) context.log.display(f"Saved raw output to {NXC_PATH}/logs/{log_name}") diff --git a/nxc/modules/get-network.py b/nxc/modules/get-network.py index dcbb9188c9..f9fadfa0f2 100644 --- a/nxc/modules/get-network.py +++ b/nxc/modules/get-network.py @@ -7,8 +7,9 @@ from struct import unpack from impacket.structure import Structure -from os.path import expanduser +from os.path import expanduser, join from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH from nxc.parsers.ldap_results import parse_result_attributes @@ -141,7 +142,7 @@ def on_login(self, context, connection): outdata = [x for x in outdata if not (x["value"] in seen_ips or seen_ips.add(x["value"]))] context.log.highlight(f"Found {len(outdata)} records") - path = expanduser(f"{NXC_PATH}/logs/{connection.domain}_network_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log") + path = expanduser(join(NXC_PATH, "logs", sanitize_path_component(f"{connection.domain}_network_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log"))) with open(path, "w") as outfile: for row in outdata: if self.showhosts: diff --git a/nxc/modules/get_netconnections.py b/nxc/modules/get_netconnections.py index facefd18d6..dc646b50fa 100755 --- a/nxc/modules/get_netconnections.py +++ b/nxc/modules/get_netconnections.py @@ -1,6 +1,7 @@ from datetime import datetime from nxc.helpers.logger import write_log from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH import json @@ -30,6 +31,6 @@ def on_admin_login(self, context, connection): data.append(cards) - log_name = f"network-connections-{connection.host}-{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log" + log_name = sanitize_path_component(f"network-connections-{connection.host}-{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.log") write_log(json.dumps(data), log_name) context.log.display(f"Saved raw output to {NXC_PATH}/logs/{log_name}") diff --git a/nxc/modules/mssql_dumper.py b/nxc/modules/mssql_dumper.py index f52473ae6e..61dbc25d47 100644 --- a/nxc/modules/mssql_dumper.py +++ b/nxc/modules/mssql_dumper.py @@ -4,6 +4,7 @@ from pathlib import Path import re from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH @@ -139,8 +140,8 @@ def on_login(self, context, connection): context.log.fail(f"Regex scan failed for {db_name}.{table_name}: {e}") if self.save and all_results: - filename = f"{connection.hostname}_{connection.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}.json" - file_path = Path(f"{NXC_PATH}/modules/mssql-dumper/{filename}").resolve() + filename = sanitize_path_component(f"{connection.hostname}_{connection.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}.json") + file_path = (Path(NXC_PATH) / "modules" / "mssql-dumper" / filename).resolve() os.makedirs(file_path.parent, exist_ok=True) with open(file_path, "w") as f: json.dump(all_results, f, indent=2) diff --git a/nxc/modules/nanodump.py b/nxc/modules/nanodump.py index 3e79889ce7..eb5da55a54 100644 --- a/nxc/modules/nanodump.py +++ b/nxc/modules/nanodump.py @@ -10,6 +10,7 @@ from datetime import datetime from nxc.helpers.bloodhound import add_user_bh from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.protocols.mssql.mssqlexec import MSSQLEXEC @@ -158,7 +159,7 @@ def on_admin_login(self, context, connection): return else: self.context.log.display(f"Copying {nano_log_name} to host") - filename = os.path.join(self.dir_result, f"{self.connection.hostname}_{self.connection.os_arch}_{self.connection.domain}.log") + filename = os.path.join(self.dir_result, sanitize_path_component(f"{self.connection.hostname}_{self.connection.os_arch}_{self.connection.domain}.log")) if self.context.protocol == "smb": with open(filename, "wb+") as dump_file: try: diff --git a/nxc/modules/obsolete.py b/nxc/modules/obsolete.py index 27525cc25c..2d96cb2c1c 100644 --- a/nxc/modules/obsolete.py +++ b/nxc/modules/obsolete.py @@ -1,7 +1,9 @@ #!/usr/bin/env python3 from datetime import datetime, timedelta -from nxc.helpers.misc import CATEGORY +from os.path import join +from nxc.helpers.misc import CATEGORY, sanitize_dns +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH import socket @@ -71,7 +73,7 @@ def on_login(self, context, connection): if answers: obsolete_hosts_count = len(answers) - filename = f"{NXC_PATH}/logs/{connection.domain}.obsoletehosts.txt" + filename = join(NXC_PATH, "logs", sanitize_path_component(f"{connection.domain}.obsoletehosts.txt")) context.log.display(f"{obsolete_hosts_count} Obsolete hosts will be saved to {filename}") with open(filename, "w") as f: for dns_hostname, ip_address, os, pwd_last_set_readable in answers: diff --git a/nxc/modules/pre2k.py b/nxc/modules/pre2k.py index 9e8087bd7e..5a845f5a27 100644 --- a/nxc/modules/pre2k.py +++ b/nxc/modules/pre2k.py @@ -4,6 +4,7 @@ from impacket.krb5.types import Principal from impacket.krb5 import constants from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.parsers.ldap_results import parse_result_attributes from nxc.paths import NXC_PATH @@ -54,7 +55,7 @@ def on_login(self, context, connection): context.log.debug(f"Added computer: {computer['sAMAccountName']}") # Save computers to file - domain_dir = os.path.join(f"{NXC_PATH}/modules/pre2k", connection.domain) + domain_dir = os.path.join(NXC_PATH, "modules", "pre2k", sanitize_path_component(connection.domain)) output_file_pre2k = os.path.join(domain_dir, "precreated_computers.txt") output_file_non_pre2k = os.path.join(domain_dir, "non_precreated_computers.txt") diff --git a/nxc/modules/user-desc.py b/nxc/modules/user-desc.py index a43e001596..3b2167b9db 100644 --- a/nxc/modules/user-desc.py +++ b/nxc/modules/user-desc.py @@ -3,6 +3,7 @@ from impacket.ldap import ldap, ldapasn1 from impacket.ldap.ldap import LDAPSearchError from nxc.helpers.misc import CATEGORY +from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH @@ -91,7 +92,7 @@ def on_login(self, context, connection): def create_log_file(self, host, time): """Create a log file for dumping user descriptions.""" - logfile = f"UserDesc-{host}-{time}.log" + logfile = sanitize_path_component(f"UserDesc-{host}-{time}.log") logfile = Path(NXC_PATH).joinpath(logfile) self.context.log.info(f"Creating log file '{logfile}'") diff --git a/nxc/protocols/rdp.py b/nxc/protocols/rdp.py index 698fdae2cd..bbb8d6cfea 100644 --- a/nxc/protocols/rdp.py +++ b/nxc/protocols/rdp.py @@ -9,6 +9,7 @@ from nxc.connection import connection from nxc.helpers.misc import sanitize_dns +from nxc.helpers.path import sanitize_path_component from nxc.helpers.bloodhound import add_user_bh from nxc.logger import NXCAdapter from nxc.config import host_info_colors, process_secret @@ -590,7 +591,8 @@ async def screen(self): await asyncio.sleep(5) if self.conn is not None and self.conn.desktop_buffer_has_data is True: buffer = self.conn.get_desktop_buffer(VIDEO_FORMAT.PIL) - filename = await Path(f"{NXC_PATH}/screenshots/{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.png").expanduser() + filename_stem = sanitize_path_component(f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}", max_bytes=251) + filename = await (Path(NXC_PATH) / "screenshots" / f"{filename_stem}.png").expanduser() buffer.save(filename, "png") self.logger.highlight(f"Screenshot saved {filename}") except Exception as e: @@ -618,7 +620,8 @@ async def nla_screen(self): await asyncio.sleep(int(self.args.screentime)) if self.conn is not None and self.conn.desktop_buffer_has_data is True: buffer = self.conn.get_desktop_buffer(VIDEO_FORMAT.PIL) - filename = await Path(f"{NXC_PATH}/screenshots/{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.png").expanduser() + filename_stem = sanitize_path_component(f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}", max_bytes=251) + filename = await (Path(NXC_PATH) / "screenshots" / f"{filename_stem}.png").expanduser() buffer.save(filename, "png") self.logger.highlight(f"NLA Screenshot saved {filename}") return diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index e6af66fa7e..2e9b3a845e 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -6,7 +6,7 @@ import ipaddress from pathlib import Path -from nxc.helpers.path import sanitize_filename +from nxc.helpers.path import sanitize_filename, sanitize_path_component from Cryptodome.Hash import MD4 from textwrap import dedent from nxc.helpers.misc import sanitize_dns @@ -2168,10 +2168,10 @@ def download_file(self, share_name, remote_path, dest_file, access_mode=FILE_REA def get_file_single(self, remote_path, download_path, silent=False): share_name = self.args.share + if self.args.append_host: + download_path = sanitize_path_component(f"{self.hostname}-{remote_path}") if not silent: self.logger.display(f"Copying '{remote_path}' to '{download_path}'") - if self.args.append_host: - download_path = f"{self.hostname}-{remote_path}" with open(download_path, "wb+") as file: if self.download_file(share_name, remote_path, file.write): if not silent: diff --git a/nxc/protocols/vnc.py b/nxc/protocols/vnc.py index a0096f2723..fc0e1c5866 100644 --- a/nxc/protocols/vnc.py +++ b/nxc/protocols/vnc.py @@ -7,6 +7,7 @@ from nxc.config import host_info_colors from nxc.connection import connection from nxc.helpers.logger import highlight +from nxc.helpers.path import sanitize_path_component from nxc.logger import NXCAdapter from nxc.paths import NXC_PATH from aardwolf.commons.target import RDPTarget @@ -160,7 +161,7 @@ async def screen(self): await asyncio.sleep(int(self.args.screentime)) if self.conn is not None and self.conn.desktop_buffer_has_data is True: buffer = self.conn.get_desktop_buffer(VIDEO_FORMAT.PIL) - filename = await Path(f"{NXC_PATH}/screenshots/{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.png").expanduser() + filename = await (Path(NXC_PATH) / "screenshots" / sanitize_path_component(f"{self.hostname}_{self.host}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.png")).expanduser() buffer.save(filename, "png") self.logger.highlight(f"Screenshot saved {filename}") From 88c4963fc69b81bedb2ab549d67d831a91a42c89 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 13:17:17 -0400 Subject: [PATCH 12/21] Sanitize manual NTLM challenge parsing --- nxc/helpers/negotiate_parser.py | 4 ++++ nxc/protocols/mssql.py | 7 +++++-- nxc/protocols/rdp.py | 6 +++--- nxc/protocols/winrm.py | 7 +++++-- nxc/protocols/wmi.py | 9 ++++++--- 5 files changed, 23 insertions(+), 10 deletions(-) diff --git a/nxc/helpers/negotiate_parser.py b/nxc/helpers/negotiate_parser.py index 9460efc2fc..f7830ee5c8 100644 --- a/nxc/helpers/negotiate_parser.py +++ b/nxc/helpers/negotiate_parser.py @@ -12,6 +12,7 @@ def parse_challenge(challange): target_info = { "hostname": None, + "dns_hostname": None, "domain": None, "os_version": None } @@ -20,6 +21,9 @@ def parse_challenge(challange): if av_pairs[ntlm.NTLMSSP_AV_HOSTNAME] is not None: with contextlib.suppress(Exception): target_info["hostname"] = av_pairs[ntlm.NTLMSSP_AV_HOSTNAME][1].decode("utf-16le") + if av_pairs[ntlm.NTLMSSP_AV_DNS_HOSTNAME] is not None: + with contextlib.suppress(Exception): + target_info["dns_hostname"] = av_pairs[ntlm.NTLMSSP_AV_DNS_HOSTNAME][1].decode("utf-16le") if av_pairs[ntlm.NTLMSSP_AV_DNS_DOMAINNAME] is not None: with contextlib.suppress(Exception): target_info["domain"] = av_pairs[ntlm.NTLMSSP_AV_DNS_DOMAINNAME][1].decode("utf-16le") diff --git a/nxc/protocols/mssql.py b/nxc/protocols/mssql.py index c65f6f8c32..0c0fb25b68 100755 --- a/nxc/protocols/mssql.py +++ b/nxc/protocols/mssql.py @@ -155,8 +155,11 @@ def enum_host_info(self): else: if challenge.startswith(b"NTLMSSP\x00"): ntlm_info = parse_challenge(challenge) - self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) + dns_hostname = ntlm_info["dns_hostname"] or "" + hostname = ntlm_info["hostname"] or dns_hostname.split(".", 1)[0] or self.host + domain = ntlm_info["domain"] or (dns_hostname.split(".", 1)[1] if "." in dns_hostname else self.host) + self.hostname = sanitize_dns(hostname, self.logger) + self.targetDomain = self.domain = sanitize_dns(domain, self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: diff --git a/nxc/protocols/rdp.py b/nxc/protocols/rdp.py index bbb8d6cfea..22f9280e6f 100644 --- a/nxc/protocols/rdp.py +++ b/nxc/protocols/rdp.py @@ -143,9 +143,9 @@ def create_conn_obj(self): except Exception: pass else: - self.domain = info_domain["dnsdomainname"] - self.hostname = sanitize_dns(info_domain["computername"], self.logger) - self.server_os = info_domain["os_guess"] + " Build " + str(info_domain["os_build"]) + self.hostname = sanitize_dns(info_domain.get("computername") or self.host, self.logger) + self.domain = sanitize_dns(info_domain.get("dnsdomainname") or self.host, self.logger) + self.server_os = f"{info_domain.get('os_guess', 'Unknown')} Build {info_domain.get('os_build', 'Unknown')}" self.logger.extra["hostname"] = self.hostname break diff --git a/nxc/protocols/winrm.py b/nxc/protocols/winrm.py index b17126d3f2..1edf2dbc53 100644 --- a/nxc/protocols/winrm.py +++ b/nxc/protocols/winrm.py @@ -69,8 +69,11 @@ def enum_host_info(self): self.no_ntlm = True return False - self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) + dns_hostname = ntlm_info["dns_hostname"] or "" + hostname = ntlm_info["hostname"] or dns_hostname.split(".", 1)[0] or self.host + domain = ntlm_info["domain"] or (dns_hostname.split(".", 1)[1] if "." in dns_hostname else self.host) + self.hostname = sanitize_dns(hostname, self.logger) + self.targetDomain = self.domain = sanitize_dns(domain, self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname diff --git a/nxc/protocols/wmi.py b/nxc/protocols/wmi.py index b2dcf430dc..328bec44e1 100644 --- a/nxc/protocols/wmi.py +++ b/nxc/protocols/wmi.py @@ -137,12 +137,15 @@ def enum_host_info(self): response = MSRPCHeader(buffer) bindResp = MSRPCBindAck(response.getData()) ntlm_info = parse_challenge(bindResp["auth_data"]) - self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = sanitize_dns(ntlm_info["hostname"], self.logger) + dns_hostname = ntlm_info["dns_hostname"] or "" + hostname = ntlm_info["hostname"] or dns_hostname.split(".", 1)[0] or self.host + domain = ntlm_info["domain"] or (dns_hostname.split(".", 1)[1] if "." in dns_hostname else self.host) + self.hostname = sanitize_dns(hostname, self.logger) + self.targetDomain = self.domain = sanitize_dns(domain, self.logger) self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: - self.hostname = self.host + self.hostname = sanitize_dns(self.host, self.logger) if self.args.local_auth: self.domain = self.hostname if self.args.domain: From 25076be9df9adcc53e24cad9c6f7bf55bc1b64b5 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 13:17:26 -0400 Subject: [PATCH 13/21] Add tests --- tests/test_sanitize.py | 263 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 263 insertions(+) create mode 100644 tests/test_sanitize.py diff --git a/tests/test_sanitize.py b/tests/test_sanitize.py new file mode 100644 index 0000000000..2752ebac75 --- /dev/null +++ b/tests/test_sanitize.py @@ -0,0 +1,263 @@ +from pathlib import Path, PureWindowsPath +import re +from unicodedata import normalize +from unittest.mock import patch + +import pytest +from impacket import ntlm +from impacket.ldap import ldapasn1 as ldapasn1_impacket + +from nxc.helpers.misc import sanitize_dns +from nxc.helpers.negotiate_parser import parse_challenge +from nxc.helpers.path import sanitize_path_component +from nxc.parsers.ldap_results import parse_result_attributes + + +class Logger: + def __init__(self): + self.messages = [] + + def fail(self, message): + self.messages.append(message) + + +class RaisingLogger: + def fail(self, message): + raise RuntimeError(message) + + +class Unstringable: + def __str__(self): + raise ValueError + + +class Challenge: + fields = {} + + def __getitem__(self, key): + return {"TargetInfoFields": b"x", "TargetInfoFields_len": 1}[key] + + +class AVPairs: + def __init__(self, pairs): + self.pairs = pairs + + def __getitem__(self, key): + return self.pairs.get(key) + + +@pytest.mark.parametrize( + "hostname", + [ + "server", + "SRV-01", + "3com", + "123", + "host.example.com", + "host_name", + "münchen.example", + "-odd-", + "name$@!%^&(),+~`", + "a" * 253, + ], +) +def test_sanitize_dns_preserves_safe_and_noncompliant_names(hostname): + logger = Logger() + assert sanitize_dns(hostname, logger) == hostname + assert logger.messages == [] + + +@pytest.mark.parametrize( + ("hostname", "expected"), + [ + (None, "_"), + ("", "_"), + (".", "_"), + ("..", "__"), + ("host.", "host_"), + ("host name", "host_name"), + ("../../pwn", ".._.._pwn"), + (r"..\..\pwn", ".._.._pwn"), + ("/tmp/pwn", "_tmp_pwn"), + (r"C:\temp", "C__temp"), + ("{output_folder}", "_output_folder_"), + ("host\uff0fname", "host_name"), + (b"server", "server"), + ], +) +def test_sanitize_dns_returns_safe_strings(hostname, expected): + assert sanitize_dns(hostname, Logger()) == expected + + +def test_sanitize_dns_replaces_config_and_control_characters(): + logger = Logger() + sanitized = sanitize_dns("host name\n\x00\x1b#comment;[section]={value}'\"", logger) + assert sanitized == "host_name____comment__section___value___" + assert len(logger.messages) == 1 + assert "\n" not in logger.messages[0] + assert "\x00" not in logger.messages[0] + assert "\x1b" not in logger.messages[0] + assert r"\n" in logger.messages[0] + assert r"\x00" in logger.messages[0] + assert r"\x1b" in logger.messages[0] + + +@pytest.mark.parametrize("name", ["CON", "NUL.txt", "PRN", "AUX.log", "COM1", "LPT9.txt", "COM¹.txt", "CONIN$", "CONOUT$"]) +def test_sanitize_dns_neutralizes_windows_device_names(name): + assert sanitize_dns(name, Logger()).startswith("_") + + +def test_sanitize_dns_bounds_long_values_with_stable_hash(): + name = "é" * 200 + sanitized = sanitize_dns(name, Logger()) + assert len(sanitized.encode("utf-8")) <= 253 + assert sanitized == sanitize_dns(name, Logger()) + assert sanitized != name + assert len(sanitized.rsplit("_", 1)[1]) == 12 + + +def test_sanitize_dns_replaces_malformed_bytes(): + assert sanitize_dns(b"host\xffname", Logger()) == "host_name" + + +def test_sanitize_dns_always_returns_a_string(): + assert sanitize_dns(Unstringable(), Logger()) == "_" + assert sanitize_dns("../host", None) == ".._host" + assert sanitize_dns("../host", RaisingLogger()) == ".._host" + + +@pytest.mark.parametrize( + "hostname", + ["\n", "host\n", "host\x00name", "host\u202ename", "../../x", r"..\..\x", "{output_folder}", "host name", "\uff23\uff2f\uff2e"], +) +def test_sanitize_dns_postconditions(hostname): + sanitized = sanitize_dns(hostname, Logger()) + normalized = normalize("NFKC", sanitized) + normalized_stem = normalized.split(".", 1)[0].upper() + assert isinstance(sanitized, str) + assert sanitized + assert len(sanitized.encode("utf-8")) <= 253 + assert all(character.isprintable() and not character.isspace() for character in sanitized) + assert not any(character in '<>:"/\\|?*{}[]=#;\'' for character in sanitized) + assert all( + normalized_character.isprintable() + and not normalized_character.isspace() + and normalized_character not in '<>:"/\\|?*{}[]=#;\'' + for character in sanitized + for normalized_character in normalize("NFKC", character) + ) + assert normalized not in (".", "..") + assert not normalized.endswith(".") + assert normalized_stem not in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} + assert re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem) is None + assert sanitize_dns(sanitized, Logger()) == sanitized + + +def test_parse_challenge_keeps_dns_and_netbios_names_distinct(): + pairs = AVPairs({ + ntlm.NTLMSSP_AV_HOSTNAME: (0, "NETBIOS_NAME".encode("utf-16le")), + ntlm.NTLMSSP_AV_DNS_HOSTNAME: (0, "server.example.com".encode("utf-16le")), + ntlm.NTLMSSP_AV_DNS_DOMAINNAME: (0, "example.com".encode("utf-16le")), + }) + with patch("nxc.helpers.negotiate_parser.ntlm.NTLMAuthChallenge", return_value=Challenge()), patch("nxc.helpers.negotiate_parser.ntlm.AV_PAIRS", return_value=pairs): + result = parse_challenge(b"challenge") + + assert result["hostname"] == "NETBIOS_NAME" + assert result["dns_hostname"] == "server.example.com" + assert result["domain"] == "example.com" + + +def test_parse_challenge_handles_missing_names(): + with patch("nxc.helpers.negotiate_parser.ntlm.NTLMAuthChallenge", return_value=Challenge()), patch("nxc.helpers.negotiate_parser.ntlm.AV_PAIRS", return_value=AVPairs({})): + result = parse_challenge(b"challenge") + + assert result["hostname"] is None + assert result["dns_hostname"] is None + assert result["domain"] is None + + +def test_ldap_dns_hostname_is_sanitized_at_parse_boundary(): + entry = ldapasn1_impacket.SearchResultEntry() + entry["objectName"] = "" + entry["attributes"][0]["type"] = "dNSHostName" + entry["attributes"][0]["vals"][0] = "../../evil\n" + + with patch("nxc.parsers.ldap_results.nxc_logger", Logger()): + result = parse_result_attributes([entry]) + + assert result == [{"dNSHostName": ".._.._evil_"}] + + +@pytest.mark.parametrize( + ("name", "expected"), + [ + ("CON", "_CON"), + ("NUL.txt", "_NUL.txt"), + ("../pwn", ".._pwn"), + (r"..\pwn", ".._pwn"), + ("{hostname}", "_hostname_"), + ("host\u202ename", "host_name"), + ("münchen.example", "münchen.example"), + ("CON .txt", "_CON .txt"), + ], +) +def test_sanitize_path_component_handles_cross_platform_names(name, expected): + assert sanitize_path_component(name) == expected + + +def test_sanitize_path_component_replaces_malformed_bytes(): + assert sanitize_path_component(b"host\xffname") == "host_name" + + +@pytest.mark.parametrize("name", ["../../x", r"C:\x", "\uff23\uff2f\uff2e.txt", "name. ", "x\uff0fy", "{output_folder}"]) +def test_sanitize_path_component_postconditions(name): + sanitized = sanitize_path_component(name) + normalized = normalize("NFKC", sanitized) + normalized_stem = normalized.split(".", 1)[0].rstrip(" ").upper() + assert sanitized + assert len(sanitized.encode("utf-8")) <= 255 + assert all(character.isprintable() for character in sanitized) + assert not any(character in '<>:"/\\|?*{}' for character in sanitized) + assert all( + normalized_character.isprintable() and normalized_character not in '<>:"/\\|?*{}' + for character in sanitized + for normalized_character in normalize("NFKC", character) + ) + assert normalized not in (".", "..") + assert not normalized.endswith(".") + assert not normalized[-1].isspace() + assert normalized_stem not in {"CON", "PRN", "AUX", "NUL", "CLOCK$", "CONIN$", "CONOUT$"} + assert re.fullmatch(r"(?:COM|LPT)[1-9]", normalized_stem) is None + assert sanitize_path_component(sanitized) == sanitized + + +def test_sanitize_path_component_bounds_long_names_and_preserves_extension(): + sanitized = sanitize_path_component(f"{'a' * 300}.log") + assert len(sanitized.encode("utf-8")) <= 255 + assert sanitized.endswith(".log") + assert len(sanitized.rsplit("_", 1)[1].removesuffix(".log")) == 12 + + +def test_sanitize_path_component_keeps_paths_inside_base(tmp_path): + sanitized = sanitize_path_component("../../outside") + output = tmp_path / sanitized + assert output.parent == tmp_path + assert PureWindowsPath("C:/base", sanitized).parent == PureWindowsPath("C:/base") + + +def test_sanitize_path_component_is_safe_in_output_template(tmp_path): + template = str(Path(tmp_path) / "{output_folder}" / sanitize_path_component("{hostname}")) + assert template.format(output_folder="sam") == str(Path(tmp_path) / "sam" / "_hostname_") + + +def test_sanitize_path_component_honors_custom_budget(): + sanitized = sanitize_path_component("a" * 300, max_bytes=220) + assert len(sanitized.encode("utf-8")) <= 220 + assert sanitized == sanitize_path_component("a" * 300, max_bytes=220) + + +def test_sanitize_path_component_drops_extension_that_exceeds_budget(): + sanitized = sanitize_path_component(f"{'a' * 30}.abcdefghijklmnopqrstuvwx", max_bytes=20) + assert sanitized != "_" + assert len(sanitized.encode("utf-8")) <= 20 + assert sanitize_path_component(sanitized, max_bytes=20) == sanitized From 736ca36c560bef4b1d9e37e1f39f3c8e53665e29 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 14:20:31 -0400 Subject: [PATCH 14/21] Add AGENTS.md instructions --- AGENTS.md | 1 + 1 file changed, 1 insertion(+) diff --git a/AGENTS.md b/AGENTS.md index 21f4d620d7..cfe68186d5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,6 +36,7 @@ # Filesystem and source conventions - Assume that the underlying operating system and its storage are fully secure. Files and other data produced by NetExec may intentionally contain readable plaintext credentials; do not add encryption, redaction, restrictive permission handling, or permission warnings unless explicitly requested. +- Before opening, creating, writing, renaming, moving, or deleting a file or directory, sanitize every path component derived from untrusted data and verify that the resulting path cannot traverse outside its intended base directory. - Place every file and directory produced by NetExec under `NXC_PATH` by default, using `TMP_PATH` for temporary artifacts. An explicit user-provided output path or task requirement may override this default. - `NXC_PATH` defaults to `~/.nxc` but may be overridden by the environment. Import and use `NXC_PATH` or its derived path constants instead of hardcoding `~/.nxc`. - Place imports at the top of files unless explicitly instructed otherwise. From 150f323e90fe2e1048b3bd3146a259676a2e4ab4 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 16:17:08 -0400 Subject: [PATCH 15/21] Add DNS sanitization --- nxc/modules/obsolete.py | 2 +- nxc/protocols/ldap.py | 13 ++++++------- nxc/protocols/ldap/resolution.py | 4 ++++ nxc/protocols/winrm.py | 3 +-- nxc/protocols/wmi.py | 2 +- 5 files changed, 13 insertions(+), 11 deletions(-) diff --git a/nxc/modules/obsolete.py b/nxc/modules/obsolete.py index 2d96cb2c1c..e9602bc509 100644 --- a/nxc/modules/obsolete.py +++ b/nxc/modules/obsolete.py @@ -2,7 +2,7 @@ from datetime import datetime, timedelta from os.path import join -from nxc.helpers.misc import CATEGORY, sanitize_dns +from nxc.helpers.misc import CATEGORY from nxc.helpers.path import sanitize_path_component from nxc.paths import NXC_PATH import socket diff --git a/nxc/protocols/ldap.py b/nxc/protocols/ldap.py index 3ac2c00515..00526e278c 100644 --- a/nxc/protocols/ldap.py +++ b/nxc/protocols/ldap.py @@ -41,7 +41,7 @@ from nxc.config import process_secret, host_info_colors from nxc.connection import connection from nxc.helpers.bloodhound import add_user_bh -from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument +from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument, sanitize_dns from nxc.logger import NXCAdapter from nxc.protocols.ldap.bloodhound import BloodHound, resolve_collection_methods from nxc.protocols.ldap.gmsa import MSDS_MANAGEDPASSWORD_BLOB @@ -226,13 +226,13 @@ def enum_host_info(self): except Exception as e: self.logger.fail(f"Failed to enumerate host info for {self.host}, error: {e!s}") - self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}") - self.target = target - self.targetDomain = target_domain + self.target = sanitize_dns(target or self.host, self.logger) + self.targetDomain = sanitize_dns(target_domain or (target.split(".", 1)[1] if "." in target else target), self.logger) self.baseDN = base_dn + self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}") # Parse hostname and remoteName - self.hostname = self.target.split(".")[0].upper() if "." in self.target else self.target + self.hostname = sanitize_dns(self.target.split(".", 1)[0].upper() or self.host, self.logger) self.remoteName = self.target # Parse NTLM challenge @@ -940,8 +940,7 @@ def resolve_and_display_hostname(name, domain_name=None): resp_parse = parse_result_attributes(resp) for item in resp_parse: if "dNSHostName" in item: # Get dNSHostName attribute - name = item["dNSHostName"] - resolve_and_display_hostname(name) + resolve_and_display_hostname(item["dNSHostName"]) # Find all trusted domains self.logger.info("Enumerating Trusted Domains...") diff --git a/nxc/protocols/ldap/resolution.py b/nxc/protocols/ldap/resolution.py index 9175e0cc82..c7993eddba 100644 --- a/nxc/protocols/ldap/resolution.py +++ b/nxc/protocols/ldap/resolution.py @@ -5,6 +5,7 @@ from impacket.ldap import ldap as ldap_impacket from impacket.ldap import ldapasn1 as ldapasn1_impacket +from nxc.helpers.misc import sanitize_dns from nxc.parsers.ldap_results import parse_result_attributes from nxc.logger import nxc_logger @@ -18,6 +19,7 @@ def get_resolution(self): target = "" target_domain = "" base_dn = "" + machine_name = "" try: ldap_url = f"ldap://{self.host}" nxc_logger.info(f"Connecting to {ldap_url} with no baseDN") @@ -60,5 +62,7 @@ def get_resolution(self): else: nxc_logger.error(f"Error getting ldap info {e}") + machine_name = sanitize_dns(machine_name, nxc_logger) + target_domain = sanitize_dns(target_domain, nxc_logger) nxc_logger.debug(f"Target: {machine_name}.{target_domain}; target_domain: {target_domain}; base_dn: {base_dn}") return machine_name, target_domain diff --git a/nxc/protocols/winrm.py b/nxc/protocols/winrm.py index 1edf2dbc53..3439b28582 100644 --- a/nxc/protocols/winrm.py +++ b/nxc/protocols/winrm.py @@ -8,7 +8,6 @@ import xml.etree.ElementTree as ET from pypsrp.wsman import NAMESPACES -from nxc.helpers.misc import sanitize_dns from pypsrp.client import Client from pypsrp.powershell import PSDataStreams from termcolor import colored @@ -20,7 +19,7 @@ from nxc.connection import connection from nxc.helpers.bloodhound import add_user_bh from nxc.helpers.dpapi import DPAPITriage -from nxc.helpers.misc import gen_random_string +from nxc.helpers.misc import gen_random_string, sanitize_dns from nxc.helpers.negotiate_parser import parse_challenge from nxc.logger import NXCAdapter diff --git a/nxc/protocols/wmi.py b/nxc/protocols/wmi.py index 328bec44e1..214a28d517 100644 --- a/nxc/protocols/wmi.py +++ b/nxc/protocols/wmi.py @@ -145,7 +145,7 @@ def enum_host_info(self): self.server_os = ntlm_info["os_version"] self.logger.extra["hostname"] = self.hostname else: - self.hostname = sanitize_dns(self.host, self.logger) + self.hostname = self.host if self.args.local_auth: self.domain = self.hostname if self.args.domain: From 5e312533137d1e04cea261f623a0c61fcbd3a868 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 16:19:11 -0400 Subject: [PATCH 16/21] Sanitize SQL statements --- nxc/helpers/bloodhound.py | 10 +++++----- nxc/modules/hash_spider.py | 24 ++++++++++++------------ 2 files changed, 17 insertions(+), 17 deletions(-) diff --git a/nxc/helpers/bloodhound.py b/nxc/helpers/bloodhound.py index b625473357..ebc0f7725c 100644 --- a/nxc/helpers/bloodhound.py +++ b/nxc/helpers/bloodhound.py @@ -43,7 +43,7 @@ def add_user_bh(user, domain, logger, config): with driver.session().begin_transaction() as tx: for user_info in users_owned: distinguished_name = "".join([f"DC={dc}," for dc in user_info["domain"].split(".")]).rstrip(",") - domain_query = tx.run(f"MATCH (d:Domain) WHERE d.distinguishedname STARTS WITH '{distinguished_name}' RETURN d").data() + domain_query = tx.run("MATCH (d:Domain) WHERE d.distinguishedname STARTS WITH $distinguished_name RETURN d", distinguished_name=distinguished_name).data() if not domain_query: logger.debug(f"Domain {user_info['domain']} not found in BloodHound. Falling back to domainless query.") _add_without_domain(user_info, tx, logger) @@ -68,14 +68,14 @@ def _add_with_domain(user_info, domain, tx, logger): user_owned = f"{user_info['username']}@{domain}" account_type = "User" - result = tx.run(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) RETURN c").data() + result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) RETURN c", user_owned=user_owned).data() if len(result) == 0: logger.fail("Account not found in the BloodHound database.") return if result[0]["c"].get("owned") in (False, None): logger.debug(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) SET c.owned=True RETURN c.name AS name") - result = tx.run(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) SET c.owned=True RETURN c.name AS name").data()[0] + result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) SET c.owned=True RETURN c.name AS name", user_owned=user_owned).data()[0] logger.highlight(f"Node {result['name']} successfully set as owned in BloodHound") @@ -87,7 +87,7 @@ def _add_without_domain(user_info, tx, logger): user_owned = user_info["username"] account_type = "User" - result = tx.run(f"MATCH (c:{account_type}) WHERE c.name STARTS WITH '{user_owned}' RETURN c").data() + result = tx.run(f"MATCH (c:{account_type}) WHERE c.name STARTS WITH $user_owned RETURN c", user_owned=user_owned).data() if len(result) == 0: logger.fail("Account not found in the BloodHound database.") @@ -97,5 +97,5 @@ def _add_without_domain(user_info, tx, logger): return elif result[0]["c"].get("owned") in (False, None): logger.debug(f"MATCH (c:{account_type} {{name:'{result[0]['c']['name']}'}}) SET c.owned=True RETURN c.name AS name") - result = tx.run(f"MATCH (c:{account_type} {{name:'{result[0]['c']['name']}'}}) SET c.owned=True RETURN c.name AS name").data()[0] + result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) SET c.owned=True RETURN c.name AS name", user_owned=result[0]["c"]["name"]).data()[0] logger.highlight(f"Node {result['name']} successfully set as owned in BloodHound") diff --git a/nxc/modules/hash_spider.py b/nxc/modules/hash_spider.py index dc50bd92e3..69a9947b07 100644 --- a/nxc/modules/hash_spider.py +++ b/nxc/modules/hash_spider.py @@ -79,26 +79,26 @@ def process_creds(context, connection, credentials_data, dbconnection, cursor, d if result["password"] is not None: context.log.highlight(f"Found a cleartext password for: {username}:{password}. Adding to the DB and marking user as owned in BH.") cursor.execute( - "UPDATE admin_users SET password = ? WHERE username LIKE '" + username + "%'", - [password], + "UPDATE admin_users SET password = ? WHERE upper(substr(username, 1, ?)) = ?", + [password, len(username), username.upper()], ) username = f"{username.upper()}@{domain.upper()}" dbconnection.commit() session = driver.session() - session.run('MATCH (u) WHERE (u.name = "' + username + '") SET u.owned=True RETURN u,u.name,u.owned') + session.run("MATCH (u) WHERE u.name = $username SET u.owned=True RETURN u,u.name,u.owned", username=username) if nthash == "aad3b435b51404eeaad3b435b51404ee" or nthash == "31d6cfe0d16ae931b73c59d7e0c089c0": context.log.fail(f"Hash for {username} is expired.") elif username not in found_users and nthash is not None: context.log.highlight(f"Found hashes for: '{username}:{nthash}'. Adding them to the DB and marking user as owned in BH.") found_users.append(username) cursor.execute( - "UPDATE admin_users SET hash = ? WHERE username LIKE '" + username + "%'", - [nthash], + "UPDATE admin_users SET hash = ? WHERE upper(substr(username, 1, ?)) = ?", + [nthash, len(username), username.upper()], ) dbconnection.commit() username = f"{username.upper()}@{domain.upper()}" session = driver.session() - session.run('MATCH (u) WHERE (u.name = "' + username + '") SET u.owned=True RETURN u,u.name,u.owned') + session.run("MATCH (u) WHERE u.name = $username SET u.owned=True RETURN u,u.name,u.owned", username=username) path_to_da = session.run("MATCH p=shortestPath((n)-[*1..]->(m)) WHERE n.owned=true AND m.name=~ '.*DOMAIN ADMINS.*' RETURN p") paths = list(path_to_da.data()) @@ -118,12 +118,12 @@ def initial_run(connection, cursor): password = getattr(connection, "password", "") nthash = getattr(connection, "nthash", "") cursor.execute( - "UPDATE admin_users SET password = ? WHERE username LIKE '" + username + "%'", - [password], + "UPDATE admin_users SET password = ? WHERE upper(substr(username, 1, ?)) = ?", + [password, len(username), username.upper()], ) cursor.execute( - "UPDATE admin_users SET hash = ? WHERE username LIKE '" + username + "%'", - [nthash], + "UPDATE admin_users SET hash = ? WHERE upper(substr(username, 1, ?)) = ?", + [nthash, len(username), username.upper()], ) @@ -236,7 +236,7 @@ def spider_pcs(self, context, connection, cursor, dbconnection, driver): for user in compromised_users: for pc in admin_access: if user[0] in pc[1]: - cursor.execute(f"SELECT * FROM pc_and_admins WHERE pc_name = '{pc[0]}' AND dumped NOT LIKE 'TRUE'") + cursor.execute("SELECT * FROM pc_and_admins WHERE pc_name = ? AND dumped NOT LIKE 'TRUE'", [pc[0]]) more_to_dump = cursor.fetchall() if len(more_to_dump) > 0: context.log.display(f"User {user[0]} has more access to {pc[0]}. Attempting to dump.") @@ -247,7 +247,7 @@ def spider_pcs(self, context, connection, cursor, dbconnection, driver): connection.nthash = user[1] try: self.run_lsassy(context, connection, cursor) - cursor.execute("UPDATE pc_and_admins SET dumped = 'TRUE' WHERE pc_name LIKE '" + pc[0] + "%'") + cursor.execute("UPDATE pc_and_admins SET dumped = 'TRUE' WHERE pc_name = ?", [pc[0]]) process_creds( context, From 31ca909e4ef7a4a7f4e0220eab9611e6a1a7ac6a Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 16:19:28 -0400 Subject: [PATCH 17/21] Add tests --- tests/test_sanitize.py | 218 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 214 insertions(+), 4 deletions(-) diff --git a/tests/test_sanitize.py b/tests/test_sanitize.py index 2752ebac75..3f14ffaf3b 100644 --- a/tests/test_sanitize.py +++ b/tests/test_sanitize.py @@ -1,5 +1,8 @@ +import ntpath +from importlib.util import module_from_spec, spec_from_file_location from pathlib import Path, PureWindowsPath import re +from types import SimpleNamespace from unicodedata import normalize from unittest.mock import patch @@ -7,6 +10,7 @@ from impacket import ntlm from impacket.ldap import ldapasn1 as ldapasn1_impacket +from nxc.helpers.bloodhound import _add_with_domain, _add_without_domain from nxc.helpers.misc import sanitize_dns from nxc.helpers.negotiate_parser import parse_challenge from nxc.helpers.path import sanitize_path_component @@ -16,10 +20,26 @@ class Logger: def __init__(self): self.messages = [] + self.extra = {} def fail(self, message): self.messages.append(message) + def debug(self, message): + self.messages.append(message) + + def highlight(self, message): + self.messages.append(message) + + def display(self, message): + self.messages.append(message) + + def success(self, message): + self.messages.append(message) + + def error(self, message): + self.messages.append(message) + class RaisingLogger: def fail(self, message): @@ -46,6 +66,103 @@ def __getitem__(self, key): return self.pairs.get(key) +class QueryResult: + def __init__(self, data): + self.result = data + + def data(self): + return self.result + + +class Transaction: + def __init__(self, responses): + self.responses = responses + self.calls = [] + + def run(self, query, **parameters): + self.calls.append((query, parameters)) + return QueryResult(self.responses.pop(0)) + + +class SMBItem: + def __init__(self, name, directory=False): + self.name = name + self.directory = directory + + def get_longname(self): + return self.name + + def is_directory(self): + return self.directory + + +class SMBConnection: + def __init__(self, name): + self.item = SMBItem(name) + self.remote_path = None + + def listPath(self, share, path): + return [self.item] + + def getFile(self, share, remote_path, callback, shareAccessMode=None): + self.remote_path = remote_path + callback(b"content") + + +class RecursiveSMBConnection: + def __init__(self): + self.remote_path = None + + def listPath(self, share, path): + if path == ntpath.join("root", "*"): + return [SMBItem(r"C:\outside", directory=True)] + return [SMBItem("proof.txt")] + + def getFile(self, share, remote_path, callback, shareAccessMode=None): + self.remote_path = remote_path + callback(b"content") + + +class EnumerationSMBConnection: + def getSMBServer(self): + return SimpleNamespace(get_socket=lambda: SimpleNamespace(getsockname=lambda: ("192.0.2.10", 445))) + + def login(self, username, password): + return None + + def getServerDNSHostName(self): + return "HOST/../../evil.example" + + def getServerName(self): + return "NETBIOS" + + def getServerDNSDomainName(self): + return "example\n#injected" + + def getServerOS(self): + return "Windows 11" + + def getServerOSMajor(self): + return 10 + + def getServerOSMinor(self): + return 0 + + def getServerOSBuild(self): + return 26100 + + def logoff(self): + return None + + +@pytest.fixture(scope="module") +def smb_class(): + spec = spec_from_file_location("sanitize_test_smb_protocol", Path(__file__).parents[1] / "nxc" / "protocols" / "smb.py") + module = module_from_spec(spec) + spec.loader.exec_module(module) + return module.smb + + @pytest.mark.parametrize( "hostname", [ @@ -176,16 +293,109 @@ def test_parse_challenge_handles_missing_names(): assert result["domain"] is None -def test_ldap_dns_hostname_is_sanitized_at_parse_boundary(): +def test_ldap_dns_hostname_remains_raw_at_parse_boundary(): entry = ldapasn1_impacket.SearchResultEntry() entry["objectName"] = "" entry["attributes"][0]["type"] = "dNSHostName" entry["attributes"][0]["vals"][0] = "../../evil\n" - with patch("nxc.parsers.ldap_results.nxc_logger", Logger()): - result = parse_result_attributes([entry]) + result = parse_result_attributes([entry]) - assert result == [{"dNSHostName": ".._.._evil_"}] + assert result == [{"dNSHostName": "../../evil\n"}] + + +@pytest.mark.parametrize( + ("function", "user_info", "domain", "first_result", "expected"), + [ + (_add_with_domain, {"username": "HOST' OR 1=1$"}, "EXAMPLE' OR 1=1", [{"c": {"owned": False}}], "HOST' OR 1=1.EXAMPLE' OR 1=1"), + (_add_without_domain, {"username": "USER' OR 1=1"}, None, [{"c": {"owned": False, "name": "USER' OR 1=1@EXAMPLE"}}], "USER' OR 1=1"), + ], +) +def test_bloodhound_uses_parameters_for_untrusted_names(function, user_info, domain, first_result, expected): + transaction = Transaction([first_result, [{"name": expected}]]) + if domain is None: + function(user_info, transaction, Logger()) + else: + function(user_info, domain, transaction, Logger()) + + assert len(transaction.calls) == 2 + assert all(expected not in query for query, _ in transaction.calls) + assert transaction.calls[0][1]["user_owned"] == expected + assert transaction.calls[1][1]["user_owned"] in (expected, "USER' OR 1=1@EXAMPLE") + + +def test_smb_hosts_file_sanitizes_at_sink_without_mutating_values(tmp_path, smb_class): + connection = smb_class.__new__(smb_class) + connection.host = "192.0.2.1" + connection.hostname = "../../HOST\n" + connection.targetDomain = "[evil]\n" + connection.domain = connection.targetDomain + connection.signing = False + connection.smbv1 = False + connection.no_ntlm = False + connection.is_guest = False + connection.isdc = False + connection.null_auth = False + connection.server_os = "Windows" + connection.os_arch = 64 + connection.logger = Logger() + connection.args = SimpleNamespace(generate_hosts_file=str(tmp_path / "hosts"), generate_krb5_file=None) + + result = connection.print_host_info() + + assert result == (connection.host, "../../HOST\n", "[evil]\n") + assert (tmp_path / "hosts").read_text().splitlines() == ["192.0.2.1 .._.._HOST_._evil__ .._.._HOST_"] + + +def test_smb_enum_host_info_sanitizes_remote_names(smb_class): + connection = smb_class.__new__(smb_class) + connection.conn = EnumerationSMBConnection() + connection.host = "192.0.2.10" + connection.hostname = connection.host + connection.domain = None + connection.no_ntlm = False + connection.isdc = False + connection.os_arch = 0 + connection.smbv1 = False + connection.kerberos = False + connection.kdcHost = "192.0.2.10" + connection.logger = Logger() + connection.db = SimpleNamespace(add_host=lambda *args: None) + connection.args = SimpleNamespace(generate_hosts_file=None, generate_krb5_file=None, domain=None, use_kcache=False, local_auth=False) + connection.is_host_dc = lambda aggressive_check=False: None + connection._is_signing_required = lambda: False + connection.get_os_arch = lambda: 64 + + connection.enum_host_info() + + assert connection.hostname == "HOST_" + assert connection.targetDomain == "example__injected" + + +def test_smb_download_sanitizes_only_the_local_path(tmp_path, smb_class): + remote_name = r"..\..\escape.txt" + connection = smb_class.__new__(smb_class) + connection.conn = SMBConnection(remote_name) + connection.logger = Logger() + connection.args = SimpleNamespace(share="SHARE", append_host=False) + + connection.download_folder("root", str(tmp_path), silent=True) + + assert connection.conn.remote_path == ntpath.join("root", remote_name) + assert (tmp_path / ".._.._escape.txt").read_bytes() == b"content" + assert not (tmp_path.parent / "escape.txt").exists() + + +def test_smb_download_sanitizes_recursive_directory_components(tmp_path, smb_class): + connection = smb_class.__new__(smb_class) + connection.conn = RecursiveSMBConnection() + connection.logger = Logger() + connection.args = SimpleNamespace(share="SHARE", append_host=False) + + connection.download_folder("root", str(tmp_path), recursive=True, silent=True) + + assert connection.conn.remote_path == ntpath.join(r"C:\outside", "proof.txt") + assert (tmp_path / "C_" / "outside" / "proof.txt").read_bytes() == b"content" @pytest.mark.parametrize( From 654f3c12d87d8fcf67bc710ebca7ae811f0aa1c1 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 16:25:00 -0400 Subject: [PATCH 18/21] Sanitize SMB input --- nxc/protocols/smb.py | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index 2e9b3a845e..f40eeca12c 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -6,7 +6,7 @@ import ipaddress from pathlib import Path -from nxc.helpers.path import sanitize_filename, sanitize_path_component +from nxc.helpers.path import sanitize_path_component from Cryptodome.Hash import MD4 from textwrap import dedent from nxc.helpers.misc import sanitize_dns @@ -208,17 +208,15 @@ def enum_host_info(self): # self.targetDomain is the attribute which gets displayed as host domain if not self.no_ntlm: # Try to get hostname with getServerDNSHostName as getServerName is truncated to 15 chars - dns_hostname = self.conn.getServerDNSHostName().upper() + dns_hostname = self.conn.getServerDNSHostName() if dns_hostname and "." in dns_hostname: - hostname = dns_hostname.split(".")[0] + hostname = dns_hostname.split(".", 1)[0] elif dns_hostname: hostname = dns_hostname else: hostname = self.conn.getServerName() self.hostname = sanitize_dns(hostname, self.logger) - self.targetDomain = self.conn.getServerDNSDomainName() - if not self.targetDomain: # Not sure if that can even happen but now we are safe - self.targetDomain = self.hostname + self.targetDomain = sanitize_dns(self.conn.getServerDNSDomainName() or self.hostname, self.logger) else: try: # If we know the host is a DC we can still get the hostname over LDAP if NTLM is not available From bde37055bcb27ec54f4f296aace1e271a73079a4 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 16:35:52 -0400 Subject: [PATCH 19/21] Sanitize SMB download folder --- nxc/protocols/smb.py | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index f40eeca12c..3e09b4a442 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -2202,10 +2202,18 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= filtered_items = [item for item in items if item.get_longname() not in [".", ".."]] - # create local directory structure regardless of content; download empty folders by default - # change the Windows path to Linux and then join it with the base directory to get our actual save path - relative_path = os.path.join(*folder.replace(base_dir or folder, "").lstrip("\\").split("\\")) - local_folder_path = os.path.join(dest, relative_path) + # Create a safe local directory structure while retaining the raw remote path for SMB. + try: + relative_path = ntpath.relpath(folder, base_dir or folder) + except ValueError: + relative_path = folder + relative_parts = [] if relative_path == "." else [sanitize_path_component(part) for part in relative_path.split("\\") if part] + local_folder_path = os.path.join(dest, *relative_parts) + destination_path = Path(dest).resolve() + resolved_folder = Path(local_folder_path).resolve() + if resolved_folder != destination_path and destination_path not in resolved_folder.parents: + self.logger.fail(f"Path traversal detected in '{folder}', skipping") + return if not filtered_items and ignore_empty: self.logger.debug(f"Skipping empty folder '{folder}'") @@ -2217,10 +2225,7 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= self.logger.display(f"Created empty directory '{local_folder_path}'") for item in filtered_items: - item_name = sanitize_filename(item.get_longname()) - if not item_name: - self.logger.fail(f"Path traversal detected in '{item.get_longname()}', skipping") - continue + item_name = sanitize_path_component(item.get_longname()) dir_path = ntpath.normpath(ntpath.join(folder, item_name)) self.logger.debug(f"Parsing item: {item_name}, {dir_path}") @@ -2232,7 +2237,7 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= local_file_path = os.path.join(local_folder_path, item_name) # Defense-in-depth: verify path stays under destination resolved = Path(local_file_path).resolve() - if not str(resolved).startswith(str(Path(dest).resolve()) + os.sep): + if destination_path not in resolved.parents: self.logger.fail(f"Path traversal detected in '{item_name}', skipping") continue self.logger.debug(f"{dest=} {remote_file_path=} {relative_path=} {local_folder_path=} {local_file_path=}") From 87d34add5f2ca19b75a22e4f5234e1bf6987426e Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 17:04:50 -0400 Subject: [PATCH 20/21] Preserve remote filename while sanitizing local file name --- nxc/protocols/smb.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index 3e09b4a442..bd07e202c4 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -2225,15 +2225,16 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= self.logger.display(f"Created empty directory '{local_folder_path}'") for item in filtered_items: - item_name = sanitize_path_component(item.get_longname()) - dir_path = ntpath.normpath(ntpath.join(folder, item_name)) - self.logger.debug(f"Parsing item: {item_name}, {dir_path}") + remote_item_name = item.get_longname() + item_name = sanitize_path_component(remote_item_name) + dir_path = ntpath.normpath(ntpath.join(folder, remote_item_name)) + self.logger.debug(f"Parsing item: {remote_item_name!r}, {dir_path!r}") if item.is_directory() and recursive: - self.logger.debug(f"Found new directory to parse: {dir_path}") + self.logger.debug(f"Found new directory to parse: {dir_path!r}") self.download_folder(dir_path, dest, recursive, silent, base_dir or folder, ignore_empty) elif not item.is_directory(): - remote_file_path = ntpath.join(folder, item_name) + remote_file_path = ntpath.join(folder, remote_item_name) local_file_path = os.path.join(local_folder_path, item_name) # Defense-in-depth: verify path stays under destination resolved = Path(local_file_path).resolve() @@ -2245,7 +2246,7 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= try: self.get_file_single(remote_file_path, local_file_path, silent) except FileNotFoundError: - self.logger.fail(f"Error downloading file '{remote_file_path}' due to file not found (probably a race condition between listing and downloading)") + self.logger.fail(f"Error downloading file {remote_file_path!r} due to file not found (probably a race condition between listing and downloading)") def get_folder(self): recursive = self.args.recursive From a47d07e005b0d228f6e8ccb7ebfc5a12364f7658 Mon Sep 17 00:00:00 2001 From: Alexander Neff Date: Thu, 24 Sep 2026 17:05:18 -0400 Subject: [PATCH 21/21] Preserve remote filename while sanitizing local file name --- nxc/protocols/smb.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/nxc/protocols/smb.py b/nxc/protocols/smb.py index bd07e202c4..d62fee06d2 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -2246,7 +2246,7 @@ def download_folder(self, folder, dest, recursive=False, silent=False, base_dir= try: self.get_file_single(remote_file_path, local_file_path, silent) except FileNotFoundError: - self.logger.fail(f"Error downloading file {remote_file_path!r} due to file not found (probably a race condition between listing and downloading)") + self.logger.fail(f"Error downloading file '{remote_file_path!r}' due to file not found (probably a race condition between listing and downloading)") def get_folder(self): recursive = self.args.recursive