Sanitize NTLM hostname to prevent path traversal and injection

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)

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))

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")

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:

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)

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"))

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}")

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:

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}")

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)