diff --git a/capabilities/network-ops/tools/impacket.py b/capabilities/network-ops/tools/impacket.py index a8a3a30..4e6aef3 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,87 @@ 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 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 + 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 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() + 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).strip("'\"")) + 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..e77d9c1 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,27 @@ 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: + 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"]) async def netexec_smb_enum_shares( @@ -323,16 +337,26 @@ 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: + 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"]) async def netexec_asreproast( diff --git a/capabilities/network-ops/tools/sharpview.py b/capabilities/network-ops/tools/sharpview.py index 04d2e1d..1945dd5 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,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(method_args.split()) + # 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(token.strip('"').strip("'") for token in tokens) 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..5c384a0 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,50 @@ 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()}" + ) + 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( + f"Command failed ({proc.returncode}):\n{errors or output}" + ) + + return output @tool_method(catch=True) async def smb_download_file(