Merge pull request #1444 from Pennyw0rth/fix-path-traversals

Fix all kinds of security issues

AGENTS.md

@@ -36,6 +36,7 @@
 # Filesystem and source conventions
 
 - Assume that the underlying operating system and its storage are fully secure. Files and other data produced by NetExec may intentionally contain readable plaintext credentials; do not add encryption, redaction, restrictive permission handling, or permission warnings unless explicitly requested.
+- Before opening, creating, writing, renaming, moving, or deleting a file or directory, sanitize every path component derived from untrusted data and verify that the resulting path cannot traverse outside its intended base directory.
 - Place every file and directory produced by NetExec under `NXC_PATH` by default, using `TMP_PATH` for temporary artifacts. An explicit user-provided output path or task requirement may override this default.
 - `NXC_PATH` defaults to `~/.nxc` but may be overridden by the environment. Import and use `NXC_PATH` or its derived path constants instead of hardcoding `~/.nxc`.
 - Place imports at the top of files unless explicitly instructed otherwise.

nxc/connection.py

@@ -14,6 +14,7 @@ from socket import AF_UNSPEC, SOCK_DGRAM, IPPROTO_IP, AI_CANONNAME, getaddrinfo
 
 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 @@ class connection:
 
             # 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/bloodhound.py

@@ -43,7 +43,7 @@ def add_user_bh(user, domain, logger, config):
             with driver.session().begin_transaction() as tx:
                 for user_info in users_owned:
                     distinguished_name = "".join([f"DC={dc}," for dc in user_info["domain"].split(".")]).rstrip(",")
-                    domain_query = tx.run(f"MATCH (d:Domain) WHERE d.distinguishedname STARTS WITH '{distinguished_name}' RETURN d").data()
+                    domain_query = tx.run("MATCH (d:Domain) WHERE d.distinguishedname STARTS WITH $distinguished_name RETURN d", distinguished_name=distinguished_name).data()
                     if not domain_query:
                         logger.debug(f"Domain {user_info['domain']} not found in BloodHound. Falling back to domainless query.")
                         _add_without_domain(user_info, tx, logger)
@@ -68,14 +68,14 @@ def _add_with_domain(user_info, domain, tx, logger):
         user_owned = f"{user_info['username']}@{domain}"
         account_type = "User"
 
-    result = tx.run(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) RETURN c").data()
+    result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) RETURN c", user_owned=user_owned).data()
 
     if len(result) == 0:
         logger.fail("Account not found in the BloodHound database.")
         return
     if result[0]["c"].get("owned") in (False, None):
         logger.debug(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) SET c.owned=True RETURN c.name AS name")
-        result = tx.run(f"MATCH (c:{account_type} {{name:'{user_owned}'}}) SET c.owned=True RETURN c.name AS name").data()[0]
+        result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) SET c.owned=True RETURN c.name AS name", user_owned=user_owned).data()[0]
         logger.highlight(f"Node {result['name']} successfully set as owned in BloodHound")
 
 
@@ -87,7 +87,7 @@ def _add_without_domain(user_info, tx, logger):
         user_owned = user_info["username"]
         account_type = "User"
 
-    result = tx.run(f"MATCH (c:{account_type}) WHERE c.name STARTS WITH '{user_owned}' RETURN c").data()
+    result = tx.run(f"MATCH (c:{account_type}) WHERE c.name STARTS WITH $user_owned RETURN c", user_owned=user_owned).data()
 
     if len(result) == 0:
         logger.fail("Account not found in the BloodHound database.")
@@ -97,5 +97,5 @@ def _add_without_domain(user_info, tx, logger):
         return
     elif result[0]["c"].get("owned") in (False, None):
         logger.debug(f"MATCH (c:{account_type} {{name:'{result[0]['c']['name']}'}}) SET c.owned=True RETURN c.name AS name")
-        result = tx.run(f"MATCH (c:{account_type} {{name:'{result[0]['c']['name']}'}}) SET c.owned=True RETURN c.name AS name").data()[0]
+        result = tx.run(f"MATCH (c:{account_type} {{name: $user_owned}}) SET c.owned=True RETURN c.name AS name", user_owned=result[0]["c"]["name"]).data()[0]
         logger.highlight(f"Node {result['name']} successfully set as owned in BloodHound")

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,91 @@ 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]}_"
+
+    # 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)
+    if (
+        not sanitized
+        or not normalized
+        or len(sanitized.encode("utf-8")) > 253
+        or normalized in (".", "..")
+        or normalized.endswith(".")
+        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 @@ import contextlib
 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, control, and Unicode-equivalent metacharacters.
+    unsafe_characters = '<>:"/\\|?*'  # Portable filename 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 minikerberos.common.ccache import CCACHE
 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 @@ class NXCModule:
         """
         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 @@ class NXCModule:
         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 @@ class NXCModule:
         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 @@ class NXCModule:
                         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 datetime import datetime
 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 @@ class NXCModule:
             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 @@ class NXCModule:
 
             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/hash_spider.py

@@ -79,26 +79,26 @@ def process_creds(context, connection, credentials_data, dbconnection, cursor, d
         if result["password"] is not None:
             context.log.highlight(f"Found a cleartext password for: {username}:{password}. Adding to the DB and marking user as owned in BH.")
             cursor.execute(
-                "UPDATE admin_users SET password = ? WHERE username LIKE '" + username + "%'",
-                [password],
+                "UPDATE admin_users SET password = ? WHERE upper(substr(username, 1, ?)) = ?",
+                [password, len(username), username.upper()],
             )
             username = f"{username.upper()}@{domain.upper()}"
             dbconnection.commit()
             session = driver.session()
-            session.run('MATCH (u) WHERE (u.name = "' + username + '") SET u.owned=True RETURN u,u.name,u.owned')
+            session.run("MATCH (u) WHERE u.name = $username SET u.owned=True RETURN u,u.name,u.owned", username=username)
         if nthash == "aad3b435b51404eeaad3b435b51404ee" or nthash == "31d6cfe0d16ae931b73c59d7e0c089c0":
             context.log.fail(f"Hash for {username} is expired.")
         elif username not in found_users and nthash is not None:
             context.log.highlight(f"Found hashes for: '{username}:{nthash}'. Adding them to the DB and marking user as owned in BH.")
             found_users.append(username)
             cursor.execute(
-                "UPDATE admin_users SET hash = ? WHERE username LIKE '" + username + "%'",
-                [nthash],
+                "UPDATE admin_users SET hash = ? WHERE upper(substr(username, 1, ?)) = ?",
+                [nthash, len(username), username.upper()],
             )
             dbconnection.commit()
             username = f"{username.upper()}@{domain.upper()}"
             session = driver.session()
-            session.run('MATCH (u) WHERE (u.name = "' + username + '") SET u.owned=True RETURN u,u.name,u.owned')
+            session.run("MATCH (u) WHERE u.name = $username SET u.owned=True RETURN u,u.name,u.owned", username=username)
             path_to_da = session.run("MATCH p=shortestPath((n)-[*1..]->(m)) WHERE n.owned=true AND m.name=~ '.*DOMAIN ADMINS.*' RETURN p")
             paths = list(path_to_da.data())
 
@@ -118,12 +118,12 @@ def initial_run(connection, cursor):
     password = getattr(connection, "password", "")
     nthash = getattr(connection, "nthash", "")
     cursor.execute(
-        "UPDATE admin_users SET password = ? WHERE username LIKE '" + username + "%'",
-        [password],
+        "UPDATE admin_users SET password = ? WHERE upper(substr(username, 1, ?)) = ?",
+        [password, len(username), username.upper()],
     )
     cursor.execute(
-        "UPDATE admin_users SET hash = ? WHERE username LIKE '" + username + "%'",
-        [nthash],
+        "UPDATE admin_users SET hash = ? WHERE upper(substr(username, 1, ?)) = ?",
+        [nthash, len(username), username.upper()],
     )
 
 
@@ -188,7 +188,7 @@ class NXCModule:
         )
         if session.smb_session is None:
             context.log.fail("Couldn't connect to remote host. Password likely expired/changed. Removing from DB.")
-            cursor.execute(f"UPDATE admin_users SET hash = NULL WHERE username LIKE '{username}'")
+            cursor.execute("UPDATE admin_users SET hash = NULL WHERE username = ? COLLATE NOCASE", [f"{username}@{domain_name}"])
             return False
         dumper = Dumper(session, timeout=10, time_between_commands=7).load(self.method)
         if dumper is None:
@@ -236,7 +236,7 @@ class NXCModule:
         for user in compromised_users:
             for pc in admin_access:
                 if user[0] in pc[1]:
-                    cursor.execute(f"SELECT * FROM pc_and_admins WHERE pc_name = '{pc[0]}' AND dumped NOT LIKE 'TRUE'")
+                    cursor.execute("SELECT * FROM pc_and_admins WHERE pc_name = ? AND dumped NOT LIKE 'TRUE'", [pc[0]])
                     more_to_dump = cursor.fetchall()
                     if len(more_to_dump) > 0:
                         context.log.display(f"User {user[0]} has more access to {pc[0]}. Attempting to dump.")
@@ -247,7 +247,7 @@ class NXCModule:
                         connection.nthash = user[1]
                         try:
                             self.run_lsassy(context, connection, cursor)
-                            cursor.execute("UPDATE pc_and_admins SET dumped = 'TRUE' WHERE pc_name LIKE '" + pc[0] + "%'")
+                            cursor.execute("UPDATE pc_and_admins SET dumped = 'TRUE' WHERE pc_name = ?", [pc[0]])
 
                             process_creds(
                                 context,

nxc/modules/mssql_dumper.py

@@ -4,6 +4,7 @@ import os
 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 @@ class NXCModule:
                         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)

nxc/modules/nanodump.py

@@ -10,6 +10,7 @@ import tempfile
 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 @@ class NXCModule:
             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:

nxc/modules/notepad++.py

@@ -1,7 +1,9 @@
 from io import BytesIO
 from os import makedirs
-from os.path import join, abspath
+from os.path import join
+from pathlib import Path
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 
 
@@ -36,9 +38,12 @@ class NXCModule:
                             context.log.highlight(f"C:\\{file_path}")
                             for line in file_content.splitlines():
                                 context.log.highlight(f"\t{line}")
-                            filename = f"{connection.host}_{directory.get_longname()}_notepad_backup_{found}.txt"
+                            filename = sanitize_path_component(f"{connection.host}_{directory.get_longname()}_notepad_backup_{found}.txt")
                             export_path = join(NXC_PATH, "modules", "notepad++")
-                            path = abspath(join(export_path, filename))
+                            path = Path(export_path, filename).resolve()
+                            if Path(export_path).resolve() not in path.parents:
+                                context.log.fail(f"Path traversal detected in {filename!r}, skipping")
+                                continue
                             makedirs(export_path, exist_ok=True)
                             try:
                                 with open(path, "w+") as file:

nxc/modules/notepad.py

@@ -1,9 +1,11 @@
 from io import BytesIO
 from os import makedirs
-from os.path import join, abspath
+from os.path import join
+from pathlib import Path
 import re
 import time
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 from impacket.smbconnection import SessionError
 
@@ -161,9 +163,12 @@ class NXCModule:
                                     content_lines.append(string)  # Store the string value only
 
                             # Save to file
-                            filename = f"{connection.host}_{directory.get_longname()}_notepad_tabstate_{found}.txt"
+                            filename = sanitize_path_component(f"{connection.host}_{directory.get_longname()}_notepad_tabstate_{found}.txt")
                             export_path = join(NXC_PATH, "modules", "notepad")
-                            path = abspath(join(export_path, filename))
+                            path = Path(export_path, filename).resolve()
+                            if Path(export_path).resolve() not in path.parents:
+                                context.log.fail(f"Path traversal detected in {filename!r}, skipping")
+                                continue
                             makedirs(export_path, exist_ok=True)
 
                             with open(path, "w+") as output_file:

nxc/modules/obsolete.py

@@ -1,7 +1,9 @@
 #!/usr/bin/env python3
 
 from datetime import datetime, timedelta
+from os.path import join
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 import socket
 
@@ -71,7 +73,7 @@ class NXCModule:
 
         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:

nxc/modules/powershell_history.py

@@ -1,6 +1,8 @@
 from os import makedirs
-from os.path import join, abspath
+from os.path import join
+from pathlib import Path
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 from io import BytesIO
 
@@ -46,9 +48,12 @@ class NXCModule:
                             for line in file_content.splitlines():
                                 context.log.highlight(f"\t{line}")
                             if self.export:
-                                filename = f"{connection.host}_{directory.get_longname()}_powershell_history.txt"
+                                filename = sanitize_path_component(f"{connection.host}_{directory.get_longname()}_powershell_history.txt")
                                 export_path = join(NXC_PATH, "modules", "powershell_history")
-                                path = abspath(join(export_path, filename))
+                                path = Path(export_path, filename).resolve()
+                                if Path(export_path).resolve() not in path.parents:
+                                    context.log.fail(f"Path traversal detected in {filename!r}, skipping")
+                                    continue
                                 makedirs(export_path, exist_ok=True)
                                 try:
                                     with open(path, "w+") as file:

nxc/modules/pre2k.py

@@ -1,9 +1,11 @@
 import os
+from pathlib import Path
 from impacket.krb5.kerberosv5 import getKerberosTGT
 from impacket.krb5.ccache import CCache
 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 +56,7 @@ class NXCModule:
                 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")
 
@@ -137,7 +139,10 @@ class NXCModule:
         try:
             ccache = CCache()
             ccache.fromTGT(ticket, sessionKey, sessionKey)
-            ccache_filename = os.path.join(ccache_base_dir, f"{username}.ccache")
+            ccache_filename = Path(ccache_base_dir, sanitize_path_component(f"{username}.ccache")).resolve()
+            if Path(ccache_base_dir).resolve() not in ccache_filename.parents:
+                context.log.fail(f"Path traversal detected in ticket filename for {username!r}, skipping")
+                return
             ccache.saveFile(ccache_filename)
             context.log.info(f"Saved ticket in {ccache_filename}")
         except Exception as e:

nxc/modules/putty.py

@@ -6,8 +6,10 @@ from impacket.dcerpc.v5.rpcrt import DCERPCException
 from impacket.dcerpc.v5 import rrp
 from impacket.examples.secretsdump import RemoteOperations
 from os import makedirs
+from pathlib import Path
 from nxc.helpers.logger import highlight
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 import re
 
@@ -160,7 +162,10 @@ class NXCModule:
                 makedirs(f"{NXC_PATH}/modules/PuTTY", exist_ok=True)
                 share = session["private_key_path"].split(":")[0] + "$"
                 file_path = session["private_key_path"].split(":")[1]
-                download_path = f"{NXC_PATH}/modules/PuTTY/putty_{session['user']}_{session['session_name']}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.sec"
+                download_path = Path(NXC_PATH, "modules", "PuTTY", sanitize_path_component(f"putty_{session['user']}_{session['session_name']}_{datetime.now().strftime('%Y-%m-%d_%H%M%S')}.sec")).resolve()
+                if Path(NXC_PATH, "modules", "PuTTY").resolve() not in download_path.parents:
+                    self.context.log.fail(f"Path traversal detected in PuTTY session {session['session_name']!r}, skipping")
+                    continue
 
                 buf = BytesIO()
                 with open(download_path, "wb") as file:

nxc/modules/recyclebin.py

@@ -1,7 +1,9 @@
 from os import makedirs
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
-from os.path import join, abspath
+from os.path import join
+from pathlib import Path
 from impacket.dcerpc.v5 import rrp
 from impacket.dcerpc.v5.rrp import DCERPCSessionError
 from impacket.examples.secretsdump import RemoteOperations
@@ -73,8 +75,11 @@ class NXCModule:
 
                                     # Formatting the destination filename
                                     file_path = path.split("$")[-1].replace("/", "_")
-                                    filename = f"{connection.host}_{username if username else sid_directory.get_longname()}_recyclebin_{file_path}"
-                                    dest_path = abspath(join(export_path, filename))
+                                    filename = sanitize_path_component(f"{connection.host}_{username if username else sid_directory.get_longname()}_recyclebin_{file_path}")
+                                    dest_path = Path(export_path, filename).resolve()
+                                    if Path(export_path).resolve() not in dest_path.parents:
+                                        context.log.fail(f"Path traversal detected in {filename!r}, skipping")
+                                        continue
                                     try:
                                         with open(dest_path, "wb+") as file:
                                             connection.conn.getFile("C$", path, file.write)

nxc/modules/snipped.py

@@ -1,8 +1,9 @@
 import ntpath
 import os
 from os.path import join, getsize, exists
-from pathlib import PurePosixPath
+from pathlib import Path
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 
 
@@ -85,11 +86,10 @@ class NXCModule:
                                 continue
 
                             remote_file_path = ntpath.join(screenshot_path, remote_file_name)
-                            # replace \\ with underscores and ignore absolute path or path traversal attempts
-                            clean_screenshot_path = "_".join(p for p in PurePosixPath(screenshot_path.replace("\\", "/")).parts if p not in ("..", ".", "/"))
-                            clean_file = "_".join(p for p in PurePosixPath(remote_file_name.replace("\\", "/")).parts if p not in ("..", ".", "/"))
-
-                            local_file_path = join(user_output_dir, f"{clean_screenshot_path}_{clean_file}")
+                            local_file_path = Path(user_output_dir, sanitize_path_component(f"{screenshot_path}_{remote_file_name}")).resolve()
+                            if Path(user_output_dir).resolve() not in local_file_path.parents:
+                                context.log.fail(f"Path traversal detected in {remote_file_path!r}, skipping")
+                                continue
                             context.log.debug(f"{local_file_path=}")
 
                             try:

nxc/modules/spider_plus.py

@@ -2,9 +2,10 @@ import json
 import errno
 from os.path import abspath, join, exists, splitext, getsize
 from os import makedirs, remove, stat
-from pathlib import Path, PurePosixPath
+from pathlib import Path
 import time
 from nxc.helpers.misc import CATEGORY
+from nxc.helpers.path import sanitize_path_component
 from nxc.paths import NXC_PATH
 from nxc.protocols.smb.remotefile import RemoteFile
 from impacket.smb3structs import FILE_READ_DATA
@@ -172,13 +173,13 @@ class SMBSpiderPlus:
     def get_file_save_path(self, remote_file):
         r"""Processes the remote file path to extract the filename and the folder path where the file should be saved locally.
 
-        Creates a PurePosixPath and replaces UNC parts, then cleans it of any path traversal (see issue #1120)
+        Sanitizes each local path component and verifies containment in the output folder.
         """
         self.logger.debug(f"Remote file: {remote_file}")
-        raw_path = PurePosixPath(remote_file._RemoteFile__share, remote_file._RemoteFile__fileName.replace("\\", "/"))
-        self.logger.debug(f"Raw path: {remote_file}")
-        clean_parts = [p for p in raw_path.parts if p not in ("..", ".")]
-        resolved = Path(self.output_folder).joinpath(self.host, *clean_parts)
+        clean_parts = [sanitize_path_component(part) for part in remote_file._RemoteFile__fileName.replace("\\", "/").split("/") if part]
+        resolved = Path(self.output_folder).joinpath(sanitize_path_component(self.host), sanitize_path_component(remote_file._RemoteFile__share), *clean_parts).resolve()
+        if Path(self.output_folder).resolve() not in resolved.parents:
+            raise ValueError(f"Path traversal detected in {remote_file!s}")
         self.logger.debug(f"Resolved path: {resolved}")
         return str(resolved.parent), resolved.name
 
@@ -308,7 +309,12 @@ class SMBSpiderPlus:
             return
 
         # Check if the file is already downloaded and up-to-date.
-        file_dir, file_name = self.get_file_save_path(remote_file)
+        try:
+            file_dir, file_name = self.get_file_save_path(remote_file)
+        except ValueError as e:
+            self.logger.fail(str(e))
+            self.stats["num_get_fail"] += 1
+            return
         download_path = join(file_dir, file_name)
         needs_update_flag = False
         if exists(download_path):
@@ -324,7 +330,7 @@ class SMBSpiderPlus:
         try:
             self.logger.info(f'Downloading file "{file_path}" => "{download_path}".')
             remote_file.open_file()
-            self.save_file(remote_file, share_name)
+            self.save_file(remote_file, share_name, download_path)
             remote_file.close()
             download_success = True
         except SessionError as e:
@@ -341,7 +347,7 @@ class SMBSpiderPlus:
         else:
             self.stats["num_get_fail"] += 1
 
-    def save_file(self, remote_file, share_name):
+    def save_file(self, remote_file, share_name, download_path):
         """Reads the `remote_file` in chunks using the `read_chunk` method.
 
         Each chunk is then written to the local file until the entire file is saved.
@@ -350,12 +356,9 @@ class SMBSpiderPlus:
         # Reset the remote_file to point to the beginning of the file.
         remote_file.seek(0, 0)
 
-        folder, filename = self.get_file_save_path(remote_file)
-        download_path = join(folder, filename)
-
         # Create the subdirectories based on the share name and file path.
-        self.logger.debug(f"Creating folder '{folder}'")
-        make_dirs(folder)
+        self.logger.debug(f"Creating folder '{Path(download_path).parent}'")
+        make_dirs(Path(download_path).parent)
 
         try:
             with open(download_path, "wb") as fd:

nxc/modules/user-desc.py

@@ -3,6 +3,7 @@ from datetime import datetime
 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 @@ class NXCModule:
 
     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}'")

nxc/protocols/ldap.py

@@ -41,7 +41,7 @@ from impacket.ntlm import getNTLMSSPType1
 from nxc.config import process_secret, host_info_colors
 from nxc.connection import connection
 from nxc.helpers.bloodhound import add_user_bh
-from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument
+from nxc.helpers.misc import get_bloodhound_info, convert, d2b, parse_argument, sanitize_dns
 from nxc.logger import NXCAdapter
 from nxc.protocols.ldap.bloodhound import BloodHound, resolve_collection_methods
 from nxc.protocols.ldap.gmsa import MSDS_MANAGEDPASSWORD_BLOB
@@ -226,13 +226,13 @@ class ldap(connection):
         except Exception as e:
             self.logger.fail(f"Failed to enumerate host info for {self.host}, error: {e!s}")
 
-        self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}")
-        self.target = target
-        self.targetDomain = target_domain
+        self.target = sanitize_dns(target or self.host, self.logger)
+        self.targetDomain = sanitize_dns(target_domain or (target.split(".", 1)[1] if "." in target else target), self.logger)
         self.baseDN = base_dn
+        self.logger.debug(f"Target: {target}; target_domain: {target_domain}; base_dn: {base_dn}")
 
         # Parse hostname and remoteName
-        self.hostname = self.target.split(".")[0].upper() if "." in self.target else self.target
+        self.hostname = sanitize_dns(self.target.split(".", 1)[0].upper() or self.host, self.logger)
         self.remoteName = self.target
 
         # Parse NTLM challenge
@@ -940,8 +940,7 @@ class ldap(connection):
         resp_parse = parse_result_attributes(resp)
         for item in resp_parse:
             if "dNSHostName" in item:  # Get dNSHostName attribute
-                name = item["dNSHostName"]
-                resolve_and_display_hostname(name)
+                resolve_and_display_hostname(item["dNSHostName"])
 
         # Find all trusted domains
         self.logger.info("Enumerating Trusted Domains...")

nxc/protocols/ldap/resolution.py

@@ -5,6 +5,7 @@ from OpenSSL.SSL import SysCallError
 from impacket.ldap import ldap as ldap_impacket
 from impacket.ldap import ldapasn1 as ldapasn1_impacket
 
+from nxc.helpers.misc import sanitize_dns
 from nxc.parsers.ldap_results import parse_result_attributes
 from nxc.logger import nxc_logger
 
@@ -18,6 +19,7 @@ class LDAPResolution:
         target = ""
         target_domain = ""
         base_dn = ""
+        machine_name = ""
         try:
             ldap_url = f"ldap://{self.host}"
             nxc_logger.info(f"Connecting to {ldap_url} with no baseDN")
@@ -60,5 +62,7 @@ class LDAPResolution:
             else:
                 nxc_logger.error(f"Error getting ldap info {e}")
 
+        machine_name = sanitize_dns(machine_name, nxc_logger)
+        target_domain = sanitize_dns(target_domain, nxc_logger)
         nxc_logger.debug(f"Target: {machine_name}.{target_domain}; target_domain: {target_domain}; base_dn: {base_dn}")
         return machine_name, target_domain

nxc/protocols/mssql.py

@@ -7,7 +7,7 @@ from termcolor import colored
 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 @@ class mssql(connection):
         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:

nxc/protocols/rdp.py

@@ -8,6 +8,8 @@ from termcolor import colored
 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 @@ class rdp(connection):
                     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
 
@@ -590,7 +592,8 @@ class rdp(connection):
             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 +621,8 @@ class rdp(connection):
                 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

nxc/protocols/smb.py

@@ -6,9 +6,10 @@ import struct
 import ipaddress
 from pathlib import Path
 
-from nxc.helpers.path import sanitize_filename
+from nxc.helpers.path import 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
@@ -207,16 +208,15 @@ class smb(connection):
         # self.targetDomain is the attribute which gets displayed as host domain
         if not self.no_ntlm:
             # Try to get hostname with getServerDNSHostName as getServerName is truncated to 15 chars
-            dns_hostname = self.conn.getServerDNSHostName().upper()
+            dns_hostname = self.conn.getServerDNSHostName()
             if dns_hostname and "." in dns_hostname:
-                self.hostname = dns_hostname.split(".")[0]
+                hostname = dns_hostname.split(".", 1)[0]
             elif dns_hostname:
-                self.hostname = dns_hostname
+                hostname = dns_hostname
             else:
-                self.hostname = self.conn.getServerName()
-            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
+                hostname = self.conn.getServerName()
+            self.hostname = sanitize_dns(hostname, self.logger)
+            self.targetDomain = sanitize_dns(self.conn.getServerDNSDomainName() or self.hostname, self.logger)
         else:
             try:
                 # If we know the host is a DC we can still get the hostname over LDAP if NTLM is not available
@@ -2166,10 +2166,10 @@ class smb(connection):
 
     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:
@@ -2202,10 +2202,18 @@ class smb(connection):
 
         filtered_items = [item for item in items if item.get_longname() not in [".", ".."]]
 
-        # create local directory structure regardless of content; download empty folders by default
-        # change the Windows path to Linux and then join it with the base directory to get our actual save path
-        relative_path = os.path.join(*folder.replace(base_dir or folder, "").lstrip("\\").split("\\"))
-        local_folder_path = os.path.join(dest, relative_path)
+        # Create a safe local directory structure while retaining the raw remote path for SMB.
+        try:
+            relative_path = ntpath.relpath(folder, base_dir or folder)
+        except ValueError:
+            relative_path = folder
+        relative_parts = [] if relative_path == "." else [sanitize_path_component(part) for part in relative_path.split("\\") if part]
+        local_folder_path = os.path.join(dest, *relative_parts)
+        destination_path = Path(dest).resolve()
+        resolved_folder = Path(local_folder_path).resolve()
+        if resolved_folder != destination_path and destination_path not in resolved_folder.parents:
+            self.logger.fail(f"Path traversal detected in '{folder}', skipping")
+            return
 
         if not filtered_items and ignore_empty:
             self.logger.debug(f"Skipping empty folder '{folder}'")
@@ -2217,22 +2225,20 @@ class smb(connection):
             self.logger.display(f"Created empty directory '{local_folder_path}'")
 
         for item in filtered_items:
-            item_name = sanitize_filename(item.get_longname())
-            if not item_name:
-                self.logger.fail(f"Path traversal detected in '{item.get_longname()}', skipping")
-                continue
-            dir_path = ntpath.normpath(ntpath.join(folder, item_name))
-            self.logger.debug(f"Parsing item: {item_name}, {dir_path}")
+            remote_item_name = item.get_longname()
+            item_name = sanitize_path_component(remote_item_name)
+            dir_path = ntpath.normpath(ntpath.join(folder, remote_item_name))
+            self.logger.debug(f"Parsing item: {remote_item_name!r}, {dir_path!r}")
 
             if item.is_directory() and recursive:
-                self.logger.debug(f"Found new directory to parse: {dir_path}")
+                self.logger.debug(f"Found new directory to parse: {dir_path!r}")
                 self.download_folder(dir_path, dest, recursive, silent, base_dir or folder, ignore_empty)
             elif not item.is_directory():
-                remote_file_path = ntpath.join(folder, item_name)
+                remote_file_path = ntpath.join(folder, remote_item_name)
                 local_file_path = os.path.join(local_folder_path, item_name)
                 # Defense-in-depth: verify path stays under destination
                 resolved = Path(local_file_path).resolve()
-                if not str(resolved).startswith(str(Path(dest).resolve()) + os.sep):
+                if destination_path not in resolved.parents:
                     self.logger.fail(f"Path traversal detected in '{item_name}', skipping")
                     continue
                 self.logger.debug(f"{dest=} {remote_file_path=} {relative_path=} {local_folder_path=} {local_file_path=}")
@@ -2240,7 +2246,7 @@ class smb(connection):
                 try:
                     self.get_file_single(remote_file_path, local_file_path, silent)
                 except FileNotFoundError:
-                    self.logger.fail(f"Error downloading file '{remote_file_path}' due to file not found (probably a race condition between listing and downloading)")
+                    self.logger.fail(f"Error downloading file '{remote_file_path!r}' due to file not found (probably a race condition between listing and downloading)")
 
     def get_folder(self):
         recursive = self.args.recursive

nxc/protocols/vnc.py

@@ -7,6 +7,7 @@ import struct
 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 @@ class vnc(connection):
         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}")
 

nxc/protocols/winrm.py

@@ -19,7 +19,7 @@ from nxc.config import process_secret, host_info_colors
 from nxc.connection import connection
 from nxc.helpers.bloodhound import add_user_bh
 from nxc.helpers.dpapi import DPAPITriage
-from nxc.helpers.misc import gen_random_string
+from nxc.helpers.misc import gen_random_string, sanitize_dns
 from nxc.helpers.negotiate_parser import parse_challenge
 from nxc.logger import NXCAdapter
 
@@ -68,8 +68,11 @@ class winrm(connection):
             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
 

nxc/protocols/wmi.py

@@ -2,6 +2,7 @@ import os
 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,8 +137,11 @@ class wmi(connection):
             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:

tests/test_sanitize.py

@@ -0,0 +1,445 @@
+import ntpath
+from importlib.util import module_from_spec, spec_from_file_location
+from pathlib import Path, PureWindowsPath
+import re
+from types import SimpleNamespace
+from unicodedata import normalize
+from unittest.mock import patch
+
+import pytest
+from impacket import ntlm
+from impacket.ldap import ldapasn1 as ldapasn1_impacket
+
+from nxc.helpers.bloodhound import _add_with_domain, _add_without_domain
+from nxc.helpers.misc import sanitize_dns
+from nxc.helpers.negotiate_parser import parse_challenge
+from nxc.helpers.path import sanitize_path_component
+from nxc.parsers.ldap_results import parse_result_attributes
+
+
+class Logger:
+    def __init__(self):
+        self.messages = []
+        self.extra = {}
+
+    def fail(self, message):
+        self.messages.append(message)
+
+    def debug(self, message):
+        self.messages.append(message)
+
+    def highlight(self, message):
+        self.messages.append(message)
+
+    def display(self, message):
+        self.messages.append(message)
+
+    def success(self, message):
+        self.messages.append(message)
+
+    def error(self, message):
+        self.messages.append(message)
+
+
+class RaisingLogger:
+    def fail(self, message):
+        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)
+
+
+class QueryResult:
+    def __init__(self, data):
+        self.result = data
+
+    def data(self):
+        return self.result
+
+
+class Transaction:
+    def __init__(self, responses):
+        self.responses = responses
+        self.calls = []
+
+    def run(self, query, **parameters):
+        self.calls.append((query, parameters))
+        return QueryResult(self.responses.pop(0))
+
+
+class SMBItem:
+    def __init__(self, name, directory=False):
+        self.name = name
+        self.directory = directory
+
+    def get_longname(self):
+        return self.name
+
+    def is_directory(self):
+        return self.directory
+
+
+class SMBConnection:
+    def __init__(self, name):
+        self.item = SMBItem(name)
+        self.remote_path = None
+
+    def listPath(self, share, path):
+        return [self.item]
+
+    def getFile(self, share, remote_path, callback, shareAccessMode=None):
+        self.remote_path = remote_path
+        callback(b"content")
+
+
+class RecursiveSMBConnection:
+    def __init__(self):
+        self.remote_path = None
+
+    def listPath(self, share, path):
+        if path == ntpath.join("root", "*"):
+            return [SMBItem(r"C:\outside", directory=True)]
+        return [SMBItem("proof.txt")]
+
+    def getFile(self, share, remote_path, callback, shareAccessMode=None):
+        self.remote_path = remote_path
+        callback(b"content")
+
+
+class EnumerationSMBConnection:
+    def getSMBServer(self):
+        return SimpleNamespace(get_socket=lambda: SimpleNamespace(getsockname=lambda: ("192.0.2.10", 445)))
+
+    def login(self, username, password):
+        return None
+
+    def getServerDNSHostName(self):
+        return "HOST/../../evil.example"
+
+    def getServerName(self):
+        return "NETBIOS"
+
+    def getServerDNSDomainName(self):
+        return "example\n#injected"
+
+    def getServerOS(self):
+        return "Windows 11"
+
+    def getServerOSMajor(self):
+        return 10
+
+    def getServerOSMinor(self):
+        return 0
+
+    def getServerOSBuild(self):
+        return 26100
+
+    def logoff(self):
+        return None
+
+
+@pytest.fixture(scope="module")
+def smb_class():
+    spec = spec_from_file_location("sanitize_test_smb_protocol", Path(__file__).parents[1] / "nxc" / "protocols" / "smb.py")
+    module = module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module.smb
+
+
+@pytest.mark.parametrize(
+    "hostname",
+    [
+        "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$", "aux.example.com", "com1.example.com", "lpt9.example.com"])
+def test_sanitize_dns_preserves_windows_device_names(name):
+    assert sanitize_dns(name, Logger()) == name
+
+
+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)
+    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 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_remains_raw_at_parse_boundary():
+    entry = ldapasn1_impacket.SearchResultEntry()
+    entry["objectName"] = ""
+    entry["attributes"][0]["type"] = "dNSHostName"
+    entry["attributes"][0]["vals"][0] = "../../evil\n"
+
+    result = parse_result_attributes([entry])
+
+    assert result == [{"dNSHostName": "../../evil\n"}]
+
+
+@pytest.mark.parametrize(
+    ("function", "user_info", "domain", "first_result", "expected"),
+    [
+        (_add_with_domain, {"username": "HOST' OR 1=1$"}, "EXAMPLE' OR 1=1", [{"c": {"owned": False}}], "HOST' OR 1=1.EXAMPLE' OR 1=1"),
+        (_add_without_domain, {"username": "USER' OR 1=1"}, None, [{"c": {"owned": False, "name": "USER' OR 1=1@EXAMPLE"}}], "USER' OR 1=1"),
+    ],
+)
+def test_bloodhound_uses_parameters_for_untrusted_names(function, user_info, domain, first_result, expected):
+    transaction = Transaction([first_result, [{"name": expected}]])
+    if domain is None:
+        function(user_info, transaction, Logger())
+    else:
+        function(user_info, domain, transaction, Logger())
+
+    assert len(transaction.calls) == 2
+    assert all(expected not in query for query, _ in transaction.calls)
+    assert transaction.calls[0][1]["user_owned"] == expected
+    assert transaction.calls[1][1]["user_owned"] in (expected, "USER' OR 1=1@EXAMPLE")
+
+
+def test_smb_enum_host_info_sanitizes_remote_names(smb_class):
+    connection = smb_class.__new__(smb_class)
+    connection.conn = EnumerationSMBConnection()
+    connection.host = "192.0.2.10"
+    connection.hostname = connection.host
+    connection.domain = None
+    connection.no_ntlm = False
+    connection.isdc = False
+    connection.os_arch = 0
+    connection.smbv1 = False
+    connection.kerberos = False
+    connection.kdcHost = "192.0.2.10"
+    connection.logger = Logger()
+    connection.db = SimpleNamespace(add_host=lambda *args: None)
+    connection.args = SimpleNamespace(generate_hosts_file=None, generate_krb5_file=None, domain=None, use_kcache=False, local_auth=False)
+    connection.is_host_dc = lambda aggressive_check=False: None
+    connection._is_signing_required = lambda: False
+    connection.get_os_arch = lambda: 64
+
+    connection.enum_host_info()
+
+    assert connection.hostname == "HOST_"
+    assert connection.targetDomain == "example__injected"
+
+
+def test_smb_download_sanitizes_only_the_local_path(tmp_path, smb_class):
+    remote_name = r"..\..\escape.txt"
+    connection = smb_class.__new__(smb_class)
+    connection.conn = SMBConnection(remote_name)
+    connection.logger = Logger()
+    connection.args = SimpleNamespace(share="SHARE", append_host=False)
+
+    connection.download_folder("root", str(tmp_path), silent=True)
+
+    assert connection.conn.remote_path == ntpath.join("root", remote_name)
+    assert (tmp_path / ".._.._escape.txt").read_bytes() == b"content"
+    assert not (tmp_path.parent / "escape.txt").exists()
+
+
+def test_smb_download_sanitizes_recursive_directory_components(tmp_path, smb_class):
+    connection = smb_class.__new__(smb_class)
+    connection.conn = RecursiveSMBConnection()
+    connection.logger = Logger()
+    connection.args = SimpleNamespace(share="SHARE", append_host=False)
+
+    connection.download_folder("root", str(tmp_path), recursive=True, silent=True)
+
+    assert connection.conn.remote_path == ntpath.join(r"C:\outside", "proof.txt")
+    assert (tmp_path / "C_" / "outside" / "proof.txt").read_bytes() == b"content"
+
+
+@pytest.mark.parametrize(
+    ("name", "expected"),
+    [
+        ("CON", "_CON"),
+        ("NUL.txt", "_NUL.txt"),
+        ("../pwn", ".._pwn"),
+        (r"..\pwn", ".._pwn"),
+        ("{hostname}", "{hostname}"),
+        ("report{1}.txt", "report{1}.txt"),
+        ("report_1_.txt", "report_1_.txt"),
+        ("report\uff5b1\uff5d.txt", "report\uff5b1\uff5d.txt"),
+        ("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_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