From 68abd128e6f2cd239a50578391d18f6c65bdfcb0 Mon Sep 17 00:00:00 2001 From: TristanInSec Date: Sat, 16 May 2026 16:17:22 -0400 Subject: [PATCH 01/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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/13] 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