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/misc.py b/nxc/helpers/misc.py index dd9dbe8705..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,6 +29,99 @@ def gen_random_string(length=10): return "".join(random.sample(string.ascii_letters, int(length))) +def sanitize_dns(hostname, logger=nxc_logger): + """Return an untrusted hostname or domain as a safe, nonempty string.""" + # Always provide a safe fallback, even for missing or unstringable input. + if hostname is None: + return "_" + + try: + hostname = hostname.decode("utf-8", errors="surrogateescape") if isinstance(hostname, bytes) else str(hostname) + except Exception: + return "_" + if not hostname: + return "_" + + # Replace path, configuration, formatting, whitespace, and control characters. + unsafe_characters = '<>:"/\\|?*{}[]=#;\'' # 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): allowed = re.compile(r"^[0-9a-f]{32}", re.IGNORECASE) return bool(allowed.match(data)) 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/helpers/path.py b/nxc/helpers/path.py index a4dd96029e..b2dd40f242 100644 --- a/nxc/helpers/path.py +++ b/nxc/helpers/path.py @@ -1,4 +1,98 @@ +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.""" + # 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: + name = "" + 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() + 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 + ) + + # Neutralize traversal-only names and Windows-trimmed trailing dots or spaces. + 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]}_" + + # 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 = "" + 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}" + + # Fail closed if a future change violates any output invariant. + 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: 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/mssql.py b/nxc/protocols/mssql.py index 5722ebcc7c..0c0fb25b68 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 +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 @@ -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 = ntlm_info["hostname"] + 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 fdbdd923c4..22f9280e6f 100644 --- a/nxc/protocols/rdp.py +++ b/nxc/protocols/rdp.py @@ -8,6 +8,8 @@ from impacket.krb5.ccache import CCache 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 @@ -141,9 +143,9 @@ def create_conn_obj(self): except Exception: pass else: - self.domain = info_domain["dnsdomainname"] - self.hostname = info_domain["computername"] - 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 @@ -589,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: @@ -617,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 401f5844f4..2e9b3a845e 100755 --- a/nxc/protocols/smb.py +++ b/nxc/protocols/smb.py @@ -6,9 +6,10 @@ 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 from impacket.smbconnection import SMBConnection, SessionError from impacket.smb import SMB_DIALECT @@ -209,11 +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() + 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 @@ -2166,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}") diff --git a/nxc/protocols/winrm.py b/nxc/protocols/winrm.py index 2ea93f4b02..1edf2dbc53 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_dns from pypsrp.client import Client from pypsrp.powershell import PSDataStreams from termcolor import colored @@ -68,8 +69,11 @@ def enum_host_info(self): self.no_ntlm = True return False - self.targetDomain = self.domain = ntlm_info["domain"] - self.hostname = ntlm_info["hostname"] + 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 6b08e419db..328bec44e1 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_dns from nxc.config import process_secret from nxc.connection import connection, dcom_FirewallChecker, requires_admin from nxc.logger import NXCAdapter @@ -136,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 = ntlm_info["hostname"] + 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: 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