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