From 84ca8197e3c546ce907555fa172cbb6d48f993de Mon Sep 17 00:00:00 2001 From: mkultraWasHere Date: Tue, 7 Jul 2026 12:47:11 -0400 Subject: [PATCH 1/4] fix(network-ops): fix tool wrapper bugs found in cross-session test analysis Address confirmed tool bugs from three test sessions (482 total calls, 62 errors). Fixes silent data loss in SMB listing, broken multi-group netexec queries, naive arg splitting in SharpView, and impacket script discovery failing on shell wrapper scripts. Co-Authored-By: Claude Opus 4.6 (1M context) --- capabilities/network-ops/tools/impacket.py | 73 ++++++++++++++++++--- capabilities/network-ops/tools/netexec.py | 68 ++++++++++++------- capabilities/network-ops/tools/sharpview.py | 3 +- capabilities/network-ops/tools/smbclient.py | 55 +++++++++++++--- 4 files changed, 158 insertions(+), 41 deletions(-) diff --git a/capabilities/network-ops/tools/impacket.py b/capabilities/network-ops/tools/impacket.py index a8a3a30..d135b67 100644 --- a/capabilities/network-ops/tools/impacket.py +++ b/capabilities/network-ops/tools/impacket.py @@ -1,3 +1,4 @@ +import re import shutil import sys from pathlib import Path @@ -6,26 +7,82 @@ from dreadnode.agents.tools import Toolset, tool_method from dreadnode.tools.execute import execute +from loguru import logger + + +def _is_python_script(path: Path) -> bool: + """Check if a file is a real Python script (not a shell wrapper).""" + try: + with open(path, "rb") as f: + first_line = f.readline(256) + # Shell wrappers start with #!/bin/bash or #!/bin/sh + if first_line.startswith(b"#!") and ( + b"/bin/bash" in first_line + or b"/bin/sh" in first_line + or b"/usr/bin/env bash" in first_line + or b"/usr/bin/env sh" in first_line + ): + return False + return True + except OSError: + return False + + +def _extract_real_path_from_wrapper(wrapper_path: Path) -> Path | None: + """ + Extract the real script path from a shell wrapper. + + These wrappers typically contain: + exec python3 /real/path/to/script.py "$@" + """ + try: + content = wrapper_path.read_text() + except (OSError, UnicodeDecodeError): + return None + + # Match: exec python[3] /path/to/script.py "$@" + # or: exec /path/to/python /path/to/script.py "$@" + match = re.search(r"exec\s+\S*python\S*\s+(\S+\.py)", content) + if match: + real_path = Path(match.group(1)) + if real_path.is_file(): + return real_path.parent + + return None + def _get_impacket_script_path() -> Path: """ Auto-discover the impacket scripts directory. - Always returns a directory path so that scripts are invoked via - sys.executable, bypassing potentially broken shebangs in wrapper - scripts (e.g. pipx or apt wrappers pointing to a missing venv). + Returns a directory containing real Python impacket scripts (not + shell wrappers) so they can be invoked via sys.executable. Tries multiple common locations: - 1. Global PATH (pipx install): resolve the found script's directory + 1. Global PATH: resolve the found script's directory, following + shell wrappers to the real scripts if needed 2. pip-installed: site-packages/impacket/examples/ 3. apt-installed: /usr/share/doc/python3-impacket/examples/ """ - # Check if scripts are on PATH (pipx / manual install) and resolve - # the directory so we can invoke via sys.executable instead of - # relying on the wrapper's shebang. + # Check if scripts are on PATH (pipx / manual install) found = shutil.which("secretsdump.py") if found is not None: - return Path(found).resolve().parent + found_path = Path(found).resolve() + if _is_python_script(found_path): + return found_path.parent + + # PATH entry is a shell wrapper — try to extract the real + # script path (e.g. "exec python3 /real/path/script.py") + real_dir = _extract_real_path_from_wrapper(found_path) + if real_dir is not None: + logger.debug( + f"Impacket wrapper at {found_path} points to real scripts at {real_dir}" + ) + return real_dir + + logger.warning( + f"Impacket script at {found_path} is a shell wrapper, skipping PATH discovery" + ) # Try to find impacket in site-packages (pip install) try: diff --git a/capabilities/network-ops/tools/netexec.py b/capabilities/network-ops/tools/netexec.py index 0551ded..fc569e1 100644 --- a/capabilities/network-ops/tools/netexec.py +++ b/capabilities/network-ops/tools/netexec.py @@ -63,9 +63,13 @@ async def netexec( cmd.extend(["-d", domain]) if username: - cmd.extend(["-u", *(username if isinstance(username, list) else [username])]) + cmd.extend( + ["-u", *(username if isinstance(username, list) else [username])] + ) if password: - cmd.extend(["-p", *(password if isinstance(password, list) else [password])]) + cmd.extend( + ["-p", *(password if isinstance(password, list) else [password])] + ) if hash: cmd.extend(["-H", *(hash if isinstance(hash, list) else [hash])]) @@ -185,17 +189,26 @@ async def netexec_smb_enum_group_members( kerberos: Kerberos ccache file path or AES key for authentication. local_auth: Use local authentication (`--local-auth`). Mutually exclusive with `domain`. """ - return await self.netexec( - "smb", - targets, - args=["--groups", *groups], - username=username, - password=password, - domain=domain, - hash=hash, - kerberos=kerberos, - local_auth=local_auth, - ) + # netexec --groups only accepts one group at a time, so we loop + # and join results when multiple groups are requested. + results: list[str] = [] + for group in groups: + try: + output = await self.netexec( + "smb", + targets, + args=["--groups", group], + username=username, + password=password, + domain=domain, + hash=hash, + kerberos=kerberos, + local_auth=local_auth, + ) + results.append(output) + except Exception as e: + results.append(f"[error querying group '{group}': {e}]") + return "\n".join(results) @tool_method(catch=True, variants=["specialized", "all"]) async def netexec_smb_enum_shares( @@ -323,16 +336,25 @@ async def netexec_ldap_enum_group_members( hash: A single NTLM hash, a list of hashes, or a path to a hash file. kerberos: Kerberos ccache file path or AES key for authentication. """ - return await self.netexec( - "ldap", - targets, - args=["--groups", *groups], - username=username, - password=password, - domain=domain, - hash=hash, - kerberos=kerberos, - ) + # netexec --groups only accepts one group at a time, so we loop + # and join results when multiple groups are requested. + results: list[str] = [] + for group in groups: + try: + output = await self.netexec( + "ldap", + targets, + args=["--groups", group], + username=username, + password=password, + domain=domain, + hash=hash, + kerberos=kerberos, + ) + results.append(output) + except Exception as e: + results.append(f"[error querying group '{group}': {e}]") + return "\n".join(results) @tool_method(catch=True, variants=["specialized", "all"]) async def netexec_asreproast( diff --git a/capabilities/network-ops/tools/sharpview.py b/capabilities/network-ops/tools/sharpview.py index 04d2e1d..eb056f3 100644 --- a/capabilities/network-ops/tools/sharpview.py +++ b/capabilities/network-ops/tools/sharpview.py @@ -1,3 +1,4 @@ +import shlex import typing as t from dreadnode import Config @@ -36,7 +37,7 @@ async def _execute(self, method: str, method_args: str = "") -> str: return await self.apollo.sharpview(method=method, method_args=method_args) args = ["SharpView.exe", method] if method_args: - args.extend(method_args.split()) + args.extend(shlex.split(method_args)) return await execute(args, timeout=self.timeout) @tool_method(catch=True, variants=["all"]) diff --git a/capabilities/network-ops/tools/smbclient.py b/capabilities/network-ops/tools/smbclient.py index ece9c16..f6a96dd 100644 --- a/capabilities/network-ops/tools/smbclient.py +++ b/capabilities/network-ops/tools/smbclient.py @@ -1,3 +1,5 @@ +import asyncio + from dreadnode.agents.tools import Toolset, tool_method from dreadnode.tools.execute import execute from loguru import logger @@ -34,16 +36,51 @@ async def smb_list_files( smb_command = f"recurse ON; ls {path}" logger.info(f"Recursively listing files in {share_path}\\{path}") - return await execute( - [ - "smbclient", - share_path, - "-U", - f"{username}%{password}", - "-c", - smb_command, - ] + + # smbclient returns non-zero when any subdirectory is inaccessible + # during recursive listing, even when most of the tree enumerates + # successfully. We capture stdout regardless of exit code to preserve + # partial output. + timeout = 120 + proc = await asyncio.create_subprocess_exec( + "smbclient", + share_path, + "-U", + f"{username}%{password}", + "-c", + smb_command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + start_new_session=True, ) + try: + stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except TimeoutError: + proc.kill() + await proc.wait() + raise TimeoutError( + f"smbclient timed out after {timeout}s listing {share_path}\\{path}" + ) + output = stdout.decode(errors="replace") + errors = stderr.decode(errors="replace") + + if proc.returncode != 0 and output.strip(): + if errors.strip(): + logger.warning( + f"smbclient exited {proc.returncode} with partial output. Errors: {errors.strip()}" + ) + return ( + f"{output}\n\n[smbclient warnings — some paths may be inaccessible]\n{errors}" + if errors.strip() + else output + ) + + if proc.returncode != 0: + raise RuntimeError( + f"Command failed ({proc.returncode}):\n{errors or output}" + ) + + return output @tool_method(catch=True) async def smb_download_file( From 1650928c1268ea44117dd193711e640a6b7dbcf8 Mon Sep 17 00:00:00 2001 From: mkultraWasHere Date: Tue, 7 Jul 2026 13:01:09 -0400 Subject: [PATCH 2/4] fix(network-ops): address PR review feedback - Always append warning banner in smb_list_files when exit is non-zero, even without stderr output - Log full exceptions server-side in netexec group queries instead of exposing raw exception text (may contain credentials) in tool output - Add binary file detection to _is_python_script() for robustness Co-Authored-By: Claude Opus 4.6 (1M context) --- capabilities/network-ops/tools/impacket.py | 5 ++++- capabilities/network-ops/tools/netexec.py | 10 ++++++---- capabilities/network-ops/tools/smbclient.py | 9 ++++----- 3 files changed, 14 insertions(+), 10 deletions(-) diff --git a/capabilities/network-ops/tools/impacket.py b/capabilities/network-ops/tools/impacket.py index d135b67..833300a 100644 --- a/capabilities/network-ops/tools/impacket.py +++ b/capabilities/network-ops/tools/impacket.py @@ -11,10 +11,13 @@ def _is_python_script(path: Path) -> bool: - """Check if a file is a real Python script (not a shell wrapper).""" + """Check if a file is a real Python script (not a shell wrapper or binary).""" try: with open(path, "rb") as f: first_line = f.readline(256) + # Binary files (ELF, etc.) are not Python + if b"\x00" in first_line: + return False # Shell wrappers start with #!/bin/bash or #!/bin/sh if first_line.startswith(b"#!") and ( b"/bin/bash" in first_line diff --git a/capabilities/network-ops/tools/netexec.py b/capabilities/network-ops/tools/netexec.py index fc569e1..e77d9c1 100644 --- a/capabilities/network-ops/tools/netexec.py +++ b/capabilities/network-ops/tools/netexec.py @@ -206,8 +206,9 @@ async def netexec_smb_enum_group_members( local_auth=local_auth, ) results.append(output) - except Exception as e: - results.append(f"[error querying group '{group}': {e}]") + except Exception: + logger.exception(f"Failed to query SMB group '{group}'") + results.append(f"[error querying group '{group}']") return "\n".join(results) @tool_method(catch=True, variants=["specialized", "all"]) @@ -352,8 +353,9 @@ async def netexec_ldap_enum_group_members( kerberos=kerberos, ) results.append(output) - except Exception as e: - results.append(f"[error querying group '{group}': {e}]") + except Exception: + logger.exception(f"Failed to query LDAP group '{group}'") + results.append(f"[error querying group '{group}']") return "\n".join(results) @tool_method(catch=True, variants=["specialized", "all"]) diff --git a/capabilities/network-ops/tools/smbclient.py b/capabilities/network-ops/tools/smbclient.py index f6a96dd..5c384a0 100644 --- a/capabilities/network-ops/tools/smbclient.py +++ b/capabilities/network-ops/tools/smbclient.py @@ -69,11 +69,10 @@ async def smb_list_files( logger.warning( f"smbclient exited {proc.returncode} with partial output. Errors: {errors.strip()}" ) - return ( - f"{output}\n\n[smbclient warnings — some paths may be inaccessible]\n{errors}" - if errors.strip() - else output - ) + warning = "\n\n[smbclient exited non-zero — listing may be incomplete]" + if errors.strip(): + warning += f"\n{errors}" + return f"{output}{warning}" if proc.returncode != 0: raise RuntimeError( From 1b562ddd3875a1645f6999e7e75ff9bf4bfb641d Mon Sep 17 00:00:00 2001 From: mkultraWasHere Date: Tue, 7 Jul 2026 13:17:11 -0400 Subject: [PATCH 3/4] fix(network-ops): address second round of PR feedback - Use posix=False in shlex.split for SharpView args to preserve backslashes in AD-style domain\user arguments - Fix docstring for _extract_real_path_from_wrapper to match return type Co-Authored-By: Claude Opus 4.6 (1M context) --- capabilities/network-ops/tools/impacket.py | 4 +++- capabilities/network-ops/tools/sharpview.py | 5 ++++- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/capabilities/network-ops/tools/impacket.py b/capabilities/network-ops/tools/impacket.py index 833300a..dbcdc52 100644 --- a/capabilities/network-ops/tools/impacket.py +++ b/capabilities/network-ops/tools/impacket.py @@ -33,10 +33,12 @@ def _is_python_script(path: Path) -> bool: def _extract_real_path_from_wrapper(wrapper_path: Path) -> Path | None: """ - Extract the real script path from a shell wrapper. + Extract the real script directory from a shell wrapper. These wrappers typically contain: exec python3 /real/path/to/script.py "$@" + + Returns the parent directory of the real script, or None. """ try: content = wrapper_path.read_text() diff --git a/capabilities/network-ops/tools/sharpview.py b/capabilities/network-ops/tools/sharpview.py index eb056f3..f72ee0e 100644 --- a/capabilities/network-ops/tools/sharpview.py +++ b/capabilities/network-ops/tools/sharpview.py @@ -37,7 +37,10 @@ async def _execute(self, method: str, method_args: str = "") -> str: return await self.apollo.sharpview(method=method, method_args=method_args) args = ["SharpView.exe", method] if method_args: - args.extend(shlex.split(method_args)) + # posix=False preserves backslashes (common in AD args like + # domain\user) but keeps surrounding quotes — strip them. + tokens = shlex.split(method_args, posix=False) + args.extend(t.strip('"').strip("'") for t in tokens) return await execute(args, timeout=self.timeout) @tool_method(catch=True, variants=["all"]) From 6cc0e883790117b38480d6777a3103eec74dfe30 Mon Sep 17 00:00:00 2001 From: mkultraWasHere Date: Tue, 7 Jul 2026 13:21:57 -0400 Subject: [PATCH 4/4] fix(network-ops): address third round of PR feedback - Rename generator variable to avoid shadowing typing import in sharpview - Strip quotes from extracted script paths in impacket wrapper parser Co-Authored-By: Claude Opus 4.6 (1M context) --- capabilities/network-ops/tools/impacket.py | 2 +- capabilities/network-ops/tools/sharpview.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/capabilities/network-ops/tools/impacket.py b/capabilities/network-ops/tools/impacket.py index dbcdc52..4e6aef3 100644 --- a/capabilities/network-ops/tools/impacket.py +++ b/capabilities/network-ops/tools/impacket.py @@ -49,7 +49,7 @@ def _extract_real_path_from_wrapper(wrapper_path: Path) -> Path | None: # or: exec /path/to/python /path/to/script.py "$@" match = re.search(r"exec\s+\S*python\S*\s+(\S+\.py)", content) if match: - real_path = Path(match.group(1)) + real_path = Path(match.group(1).strip("'\"")) if real_path.is_file(): return real_path.parent diff --git a/capabilities/network-ops/tools/sharpview.py b/capabilities/network-ops/tools/sharpview.py index f72ee0e..1945dd5 100644 --- a/capabilities/network-ops/tools/sharpview.py +++ b/capabilities/network-ops/tools/sharpview.py @@ -40,7 +40,7 @@ async def _execute(self, method: str, method_args: str = "") -> str: # posix=False preserves backslashes (common in AD args like # domain\user) but keeps surrounding quotes — strip them. tokens = shlex.split(method_args, posix=False) - args.extend(t.strip('"').strip("'") for t in tokens) + args.extend(token.strip('"').strip("'") for token in tokens) return await execute(args, timeout=self.timeout) @tool_method(catch=True, variants=["all"])