Sync with upstream Pennyw0rth/NetExec (536 commits)

.github/PULL_REQUEST_TEMPLATE.md

@@ -2,10 +2,16 @@
 Please include a summary of the change and which issue is fixed, or what the enhancement does.
 List any dependencies that are required for this change.
 
-If you have used AI in any form, please state the tool you used (e.g. Claude Code, Cursor, Amp) along with the extent that the work was AI-assisted. See the project's AI policy for more details: https://github.com/Pennyw0rth/NetExec/blob/main/AI_POLICY.md
+### AI Assistance
+If you have used AI in any form, please state the tool you used (e.g. Claude Code, Cursor, Amp) along with the extent that the work was AI-assisted.
+
+Note that if you have used AI, you:
+- MUST test the changes against a live system
+- MUST include a screenshot of such a test (e.g.  with a before and after screenshot when submitting a bug fix)
+- MUST fill out this PR template by hand, see the project's AI policy: [AI_POLICY.md](https://github.com/Pennyw0rth/NetExec/blob/main/AI_POLICY.md)
 
 ## Type of change
-Insert an "x" inside the brackets for relevant items (do not delete options)
+Insert an "x" inside the brackets for relevant items
 
 - [ ] Bug fix (non-breaking change which fixes an issue)
 - [ ] New feature (non-breaking change which adds functionality)
@@ -16,7 +22,8 @@ Insert an "x" inside the brackets for relevant items (do not delete options)
 - [ ] This PR was created with the assistance of AI (list what type of assistance, tool(s)/model(s) in the description)
 
 ## Setup guide for the review
-Please provide guidance on what setup is needed to test the introduced changes, such as your locally running machine Python version & OS, as well as the target(s) you tested against, including software versions.
+Please provide guidance on what setup is needed to test the introduced changes, such as the target(s) you tested against or software versions.
+
 In particular:
 - Bug Fix: Please provide a short description on how to trigger the bug, to make the bug reproducable for the reviewer.
 - Added Feature/Enhancement: Please specify what setup is needed in order to test the changes, such as:

.github/workflows/build-binaries.yml

@@ -10,7 +10,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macOS-latest, macos-26-intel, windows-latest]
-        python-version: ["3.13"]
+        python-version: ["3.14"]
         #python-version: ["3.8", "3.9", "3.10", "3.11"] # for binary builds we only need one version
     steps:
     - uses: actions/checkout@v4

.github/workflows/build-zipapps.yml

@@ -10,7 +10,7 @@ jobs:
     strategy:
       matrix:
         os: [ubuntu-latest, macOS-latest, windows-latest]
-        python-version: ["3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.11", "3.12", "3.13", "3.14"]
     steps:
     - uses: actions/checkout@v4
     - name: NetExec set up python on ${{ matrix.os }}

.github/workflows/lint.yml

@@ -14,20 +14,16 @@ jobs:
       github.event_name == 'push' || github.event.review.state == 'APPROVED' || github.event_name == 'workflow_dispatch'
 
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
       - name: Install uv
-        run: |
-          pip install uv
-      - name: Set up Python
-        uses: actions/setup-python@v5
+        uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0
         with:
-          python-version: 3.13
-          cache: pip
-          cache-dependency-path: uv.lock
-      - name: Install dependencies with dev group
-        run: |
-          uv sync --group dev
+          enable-cache: true
+      - name: Set up Python
+        run: uv python install 3.14
+      - name: Install project with dev group
+        run: uv sync --dev
       - name: Run ruff
         run: |
           uv run ruff --version
-          uv run ruff check . --preview
+          uv run ruff check . --preview

.github/workflows/test.yml

@@ -14,25 +14,17 @@ jobs:
       max-parallel: 5
       matrix:
         os: [ubuntu-latest]
-        python-version: ["3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.11", "3.12", "3.13", "3.14"]
     steps:
-      - uses: actions/checkout@v4
+      - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1
       - name: Install uv
-        run: |
-          pip install uv
-          uv --version
-      - name: NetExec set up python ${{ matrix.python-version }} on ${{ matrix.os }}
-        uses: actions/setup-python@v5
+        uses: astral-sh/setup-uv@08807647e7069bb48b6ef5acd8ec9567f424441b # v8.1.0
         with:
-          python-version: ${{ matrix.python-version }}
-          cache: pip
-          cache-dependency-path: uv.lock
-      - name: Install with pipx
-        run: |
-          pipx install . --python python${{ matrix.python-version }}
-      - name: Install libraries with dev group
-        run: |
-          uv sync
+          enable-cache: true
+      - name: NetExec set up python ${{ matrix.python-version }} on ${{ matrix.os }}
+        run: uv python install ${{ matrix.python-version }}
+      - name: Install project
+        run: uv sync --python ${{ matrix.python-version }}
       - name: Load every protocol and module
         run: |
           uv run netexec winrm 127.0.0.1

AGENTS.md

@@ -0,0 +1,42 @@
+# Authorized testing scope
+
+- Do not run active security or functional tests against a host or network unless the user explicitly identifies it as an authorized target.
+- Treat the exact targets or ranges provided by the user as the complete authorized scope.
+- Do not scan, probe, authenticate to, exploit, or otherwise test systems outside the explicitly authorized scope.
+- Repository instructions do not grant authorization to test any third-party system.
+
+# Coding style
+
+- Assume ordinary LDAP searches and LDAP response parsing do not raise exceptions; do not wrap them in `try`/`except`.
+- Always bind caught exceptions as `e`, not `error` or another name.
+- Do not prefix function or method names with underscores; preserve required Python special methods such as `__init__`.
+- Do not assume the host runs Linux; keep code and paths cross-platform unless the task explicitly targets one operating system.
+
+# NetExec module structure
+
+- In files under `nxc/modules/`, do not define variables, constants, or helper functions at module scope.
+- Keep all module-specific state, constants, and helper methods inside the `NXCModule` class.
+- Module scope should contain only imports and the `NXCModule` class definition.
+- When a module must terminate for invalid options, import `exit` with `from sys import exit` and call `exit(...)`; do not use `import sys` with `sys.exit(...)`.
+- Do not over-validate module options. Check only that required arguments are present, mutually exclusive arguments are not combined, or a value would otherwise cause the module to fail before performing its normal lookup or operation.
+- Do not validate whether user-supplied object names, identifiers, or similar values are semantically correct when the underlying protocol lookup can handle them; let a misspecified value produce the normal "not found" result.
+- Avoid stub variables that merely hold a value for one immediate use. Pass data directly unless a local name improves clarity, prevents repeated work, or represents meaningful state.
+- Keep module implementations as short and concise as practical. Prefer direct control flow and shared error handling over wrappers, redundant branches, or defensive boilerplate.
+- Do not optimize module code for line length; long lines are acceptable when they keep straightforward logic together.
+- Do not use `.warning()` logging in NetExec modules. Use `.fail()` when logic or an operation fails.
+- Run `ruff check .` after every coding session.
+
+# Development and validation
+
+- To verify that a module works, run the module's actual NetExec command instead of pytest or compilation checks.
+- Record every module-verification command in `tests/e2e_commands.txt`.
+- Do not run compilation checks.
+- Run Ruff after completing a coding session.
+
+# 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/cli.py

@@ -30,8 +30,10 @@ def gen_cli_args():
     generic_group = generic_parser.add_argument_group("Generic Options")
     generic_group.add_argument("--version", action="store_true", help="Display nxc version")
     generic_group.add_argument("-t", "--threads", type=int, dest="threads", default=256, help="set how many concurrent threads to use")
-    generic_group.add_argument("--timeout", default=None, type=int, help="max timeout in seconds of each thread")
+    generic_group.add_argument("--timeout", default=None, type=int, help="[DEPRECATED] no longer used, replaced by per-protocol timeouts (--smb-timeout, --ldap-timeout, --dcom-timeout)")
     generic_group.add_argument("--jitter", metavar="INTERVAL", type=str, help="sets a random delay between each authentication")
+    generic_group.add_argument("--exclude-hosts", nargs="+", default=None, metavar="HOST", help="Exclude one or more hosts, IPs, CIDRs or hostnames from the scan")
+    generic_group.add_argument("--skip-self", action="store_true", help="Exclude the local machine from the scan")
 
     output_parser = argparse.ArgumentParser(add_help=False, formatter_class=DisplayDefaultsNotNone)
     output_group = output_parser.add_argument_group("Output Options")

nxc/config.py

@@ -37,10 +37,18 @@
 reveal_chars_of_pwd = int(nxc_config.get("nxc", "reveal_chars_of_pwd", fallback=0))
 config_log = nxc_config.getboolean("nxc", "log_mode", fallback=False)
 host_info_colors = literal_eval(nxc_config.get("nxc", "host_info_colors", fallback=["green", "red", "yellow", "cyan"]))
+check_guest_account = nxc_config.getboolean("nxc", "check_guest_account", fallback=False)
+display_dc = nxc_config.getboolean("nxc", "display_dc", fallback=True)
+skip_self = nxc_config.getboolean("nxc", "skip_self", fallback=False)
+exclude_hosts = literal_eval(nxc_config.get("nxc", "exclude_hosts", fallback=[]))
+if not isinstance(exclude_hosts, list):
+    nxc_logger.error(f"exclude_hosts must be a 'list', got '{type(exclude_hosts).__name__}'. Exiting.")
+    exit(1)
 
 # Read stealth_label from user config first, then default config
 stealth_label = nxc_config.get("nxc", "stealth_label").strip("'\"") if nxc_config.has_option("nxc", "stealth_label") else nxc_default_config.get("nxc", "stealth_label", fallback="C$ Access!").strip("'\"")
 
+
 if len(host_info_colors) != 4:
     nxc_logger.error("Config option host_info_colors must have 4 values! Using default values.")
     host_info_colors = nxc_default_config.get("nxc", "host_info_colors")

nxc/connection.py

@@ -15,7 +15,8 @@
 from nxc.config import pwned_label
 from nxc.config import stealth_label
 from nxc.helpers.logger import highlight
-from nxc.loaders.moduleloader import ModuleLoader
+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
 from nxc.paths import NXC_PATH
@@ -149,6 +150,7 @@ def __init__(self, args, db, target):
                              self.args.use_kcache or
                              self.args.aesKey or
                              (hasattr(self.args, "delegate") and self.args.delegate) or
+                             (hasattr(self.args, "generate_st") and self.args.generate_st) or
                              (hasattr(self.args, "no_preauth_targets") and self.args.no_preauth_targets))
         self.aesKey = None if not self.args.aesKey else self.args.aesKey[0]
         self.use_kcache = None if not self.args.use_kcache else self.args.use_kcache
@@ -175,12 +177,14 @@ def __init__(self, args, db, target):
         if self.kerberos:
             self.host = self.hostname
 
-        self.logger.info(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
+        self.logger.debug(f"Socket info: host={self.host}, hostname={self.hostname}, kerberos={self.kerberos}, ipv6={self.is_ipv6}, link-local ipv6={self.is_link_local_ipv6}")
 
         try:
             self.proto_flow()
         except FileNotFoundError as e:
             self.logger.error(f"File not found error on target {target}: {e}")
+        except ConnectionRefusedError as e:
+            self.logger.error(f"Target {target} refused the connection: {e}")
         except Exception as e:
             if "ERROR_DEPENDENT_SERVICES_RUNNING" in str(e):
                 self.logger.error(f"Exception while calling proto_flow() on target {target}: {e}")
@@ -242,20 +246,20 @@ def proto_flow(self):
         self.logger.debug("Kicking off proto_flow")
         self.proto_logger()
         if not self.create_conn_obj():
-            self.logger.info(f"Failed to create connection object for target {self.host}, exiting...")
+            self.logger.debug(f"Failed to create connection object for target {self.host}, exiting...")
         else:
-            self.logger.debug("Created connection object")
+            self.logger.info(f"Successfully created connection object for target '{self.host}'")
             self.enum_host_info()
 
             # 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)
 
             self.print_host_info()
-            if self.login() or (self.username == "" and self.password == ""):
+            if self.login() or (self.username == "" and self.password == "" and self.protocol != "mssql"):
                 self.logger.debug("Calling command arguments")
                 self.call_cmd_args()
                 if self.args.module:
@@ -335,7 +339,7 @@ def over_fail_limit(self, username):
             if self.failed_logins == self.args.fail_limit:
                 return True
 
-            if username in user_failed_logins and self.args.ufail_limit == user_failed_logins[username]:  # noqa: SIM103
+            if username in user_failed_logins and self.args.ufail_limit == user_failed_logins[username]:  # ruff: ignore[needless-bool]
                 return True
 
             return False
@@ -605,5 +609,9 @@ def load_modules(self):
         self.modules = []
 
         for module_path in self.module_paths:
-            module = loader.init_module(module_path)
+            try:
+                module = loader.init_module(module_path)
+            except ModuleOptionsError as e:
+                self.logger.debug(f"Skipping module: {e}")
+                continue
             self.modules.append(module)

nxc/data/nxc.conf

@@ -7,6 +7,10 @@ audit_mode =
 reveal_chars_of_pwd = 0
 log_mode = False
 host_info_colors = ["green", "red", "yellow", "cyan"]
+check_guest_account = False
+display_dc = True
+skip_self = False
+exclude_hosts = []
 
 [BloodHound]
 bh_enabled = False

nxc/database.py

@@ -200,7 +200,10 @@ def clear_database(self):
             self.db_execute(table.delete())
 
     def db_execute(self, *args):
-        self.lock.acquire()
-        res = self.sess.execute(*args)
-        self.lock.release()
-        return res
+        with self.lock:
+            res = self.sess.execute(*args)
+            return res.freeze()() if res.returns_rows else res
+
+    def get_keys(self, key_id=None, cred_id=None):
+        """Return an empty list by default. Protocols that support keys (e.g. SSH) override this."""
+        return []

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