Fix network-ops tool wrapper bugs from test analysis

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:

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(

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

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(