ENG-8830 fix(network-ops): make certipy stdin confirmation uniform and hang-safe

capabilities/network-ops/capability.yaml

@@ -1,6 +1,6 @@
 schema: 1
 name: network-ops
-version: "2.2.0"
+version: "2.2.1"
 description: >
   Network operations and Active Directory exploitation. Multi-agent
   pipeline for autonomous red teaming with Nmap scanning, Netexec

capabilities/network-ops/tests/test_tool_fixes.py

@@ -297,6 +297,127 @@ async def test_certipy_find_builds_auth_correctly(self):
             assert "-domain" not in cmd
 
 
+# ===================================================================
+# Certipy: uniform, hang-safe stdin confirmation across the toolset
+# ===================================================================
+
+
+class TestCertipyConfirmInput:
+    """Every certipy_* tool should auto-confirm interactive prompts by default
+    and close stdin, so no sibling tool can hang on a prompt until timeout, and
+    an explicit `input` always overrides."""
+
+    # Assert the literal "y" the subprocess must receive, not Certipy._CONFIRM_INPUT:
+    # comparing to the module constant would stay green if someone set it to "" or
+    # None, which (via execute()'s `if proc.stdin and input:` guard) would stop
+    # closing stdin and reintroduce the very prompt-hang these tests guard against.
+    CONFIRM = "y"
+
+    def _make_certipy(self):
+        c = certipy_mod.Certipy()
+        c.certipy_cmd = "certipy"
+        c.timeout = 10
+        return c
+
+    def _input_kwarg(self, mock_exec):
+        """Extract the `input` passed to execute, positional or keyword."""
+        assert mock_exec.await_count == 1
+        return mock_exec.call_args.kwargs["input"]
+
+    @pytest.mark.asyncio
+    async def test_direct_generic_tools_default_to_confirm(self):
+        """account/auth/ca/cert/req/template send the confirm default to stdin
+        when no input is given (none inherit the parent's stdin)."""
+        for method in [
+            "certipy_account",
+            "certipy_auth",
+            "certipy_ca",
+            "certipy_cert",
+            "certipy_req",
+            "certipy_template",
+        ]:
+            c = self._make_certipy()
+            with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+                mock_exec.return_value = "output"
+                await getattr(c, method)(args=[])
+                assert self._input_kwarg(mock_exec) == self.CONFIRM, f"{method} should auto-confirm by default"
+
+    @pytest.mark.asyncio
+    async def test_passthrough_and_find_default_to_confirm(self):
+        """The generic passthrough and certipy_find route through certipy() and
+        inherit the same confirm default."""
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy(action="find", args=[], target="10.0.0.1")
+            assert self._input_kwarg(mock_exec) == self.CONFIRM
+
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy_find(target="10.0.0.1")
+            assert self._input_kwarg(mock_exec) == self.CONFIRM
+
+    @pytest.mark.asyncio
+    async def test_relay_does_not_receive_confirm_stdin(self):
+        """The relay action is a streaming listener that never reads stdin; the
+        confirm default must not leak into it (it bypasses execute())."""
+        c = self._make_certipy()
+        with (
+            patch.object(certipy_mod, "stream_listener", new_callable=AsyncMock) as mock_relay,
+            patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec,
+        ):
+            mock_relay.return_value = "output"
+            await c.certipy(action="relay", args=["-target", "http://ca"], target="1.2.3.4")
+            mock_exec.assert_not_awaited()
+            mock_relay.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_high_level_tools_default_to_confirm(self):
+        """The high-level helpers no longer hardcode stdin; they inherit the
+        uniform confirm default via certipy()."""
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy_request_certificate(
+                target="10.0.0.1",
+                ca_server="ca.corp.local",
+                ca_name="corp-CA",
+                template="User",
+                username="Admin",
+                domain="corp.local",
+            )
+            assert self._input_kwarg(mock_exec) == self.CONFIRM
+
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy_certificate_auth(pfx_file="admin.pfx", target="10.0.0.1")
+            assert self._input_kwarg(mock_exec) == self.CONFIRM
+
+    @pytest.mark.asyncio
+    async def test_explicit_input_overrides_default(self):
+        """A caller-supplied non-empty input is passed through unchanged."""
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy_cert(args=["-export"], input="n")
+            assert self._input_kwarg(mock_exec) == "n"
+
+    @pytest.mark.asyncio
+    async def test_empty_input_still_confirms(self):
+        """An empty string falls back to the confirm default rather than being
+        forwarded as-is. This is deliberate: execute() only closes stdin when the
+        input is truthy, so forwarding "" would leave stdin open and reintroduce
+        the prompt-hang. Callers decline by sending an explicit answer (e.g. "n"),
+        not an empty string."""
+        c = self._make_certipy()
+        with patch.object(certipy_mod, "execute", new_callable=AsyncMock) as mock_exec:
+            mock_exec.return_value = "output"
+            await c.certipy_cert(args=["-export"], input="")
+            assert self._input_kwarg(mock_exec) == self.CONFIRM
+
+
 # ===================================================================
 # Impacket: _build_ntlmrelayx_args
 # ===================================================================

capabilities/network-ops/tools/certipy.py

@@ -21,6 +21,20 @@
     "pkcs#12",
 ]
 
+# Certipy's interactive subcommands prompt for confirmation (e.g. overwriting a
+# saved PFX/ccache). Agents run non-interactively, so an unanswered prompt would
+# block on stdin until the command hits its timeout. Every certipy_* tool passes
+# this as the default stdin so behavior is uniform across the toolset and stdin
+# is always closed (the prompt gets a definitive answer, never a hang). Callers
+# pass an explicit `input` to answer differently (e.g. "n").
+#
+# This is exposed as the family-wide `input: str | None` stdin parameter rather
+# than a dedicated `confirm: bool`, to stay uniform with the sibling tools
+# (impacket, bloodyad, coercion, krbrelayx), which all take `input` and guard
+# the same prompt-hang this way. The `relay` action is exempt: it is a streaming
+# listener that never reads stdin.
+_CONFIRM_INPUT = "y"
+
 
 def _detect_certipy_command() -> str:
     """Detect whether certipy is available as 'certipy' (pip) or 'certipy-ad' (Kali)."""
@@ -108,7 +122,9 @@ async def certipy(
             relay_timeout: Only for action="relay": seconds to keep the listener up
                 waiting for a relayed certificate before returning (defaults to the
                 toolset's relay_timeout). Ignored for every other action.
-            input: Optional input string to send to the command's standard input.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
                 Ignored for action="relay", which does not read from stdin.
         """
         cmd = [self.certipy_cmd, action]
@@ -140,7 +156,7 @@ async def certipy(
                 env=None,
             )
 
-        return await execute(cmd, timeout=self.timeout, input=input)
+        return await execute(cmd, timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     # Specialized methods
 
@@ -220,7 +236,6 @@ async def certipy_request_certificate(
             domain=domain,
             password=password,
             nt_hash=nt_hash,
-            input="y",
         )
 
     @tool_method(catch=True, variants=["specialized", "all"])
@@ -247,7 +262,6 @@ async def certipy_certificate_auth(
             username=username,
             domain=domain,
             target=target,
-            input="y",
         )
 
     # General methods
@@ -318,12 +332,14 @@ async def certipy_account(self, args: list[str], input: str | None = None) -> st
 
         Args:
             args: List of arguments for the command.
-            input: Optional input string to pass to the command's stdin.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "account", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "account", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     @tool_method(catch=True, variants=["generic", "all"])
-    async def certipy_auth(self, args: list[str], input: str | None = "y") -> str:
+    async def certipy_auth(self, args: list[str], input: str | None = None) -> str:
         """
         Execute a certipy-ad auth command.
 
@@ -364,9 +380,11 @@ async def certipy_auth(self, args: list[str], input: str | None = "y") -> str:
 
         Args:
             args: List of arguments for the command.
-            input: Optional input string to pass to the command's stdin.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "auth", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "auth", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     @tool_method(catch=True, variants=["generic", "all"])
     async def certipy_ca(self, args: list[str], input: str | None = None) -> str:
@@ -447,12 +465,14 @@ async def certipy_ca(self, args: list[str], input: str | None = None) -> str:
 
         Args:
             args: List of arguments for the command.
-            input: Optional input string to pass to the command's stdin.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "ca", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "ca", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     @tool_method(catch=True, variants=["generic", "all"])
-    async def certipy_cert(self, args: list[str], input: str | None = "y") -> str:
+    async def certipy_cert(self, args: list[str], input: str | None = None) -> str:
         r"""
         Execute a certipy-ad cert command.
 
@@ -479,9 +499,11 @@ async def certipy_cert(self, args: list[str], input: str | None = "y") -> str:
 
         Args:
             args: List of arguments for the command.
-            input: Optional input string to pass to the command's stdin.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "cert", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "cert", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     @tool_method(catch=True, variants=["generic", "all"])
     async def certipy_find(
@@ -539,7 +561,9 @@ async def certipy_find(
             domain: The domain name (combined with username as user@domain).
             password: The password for authentication.
             nt_hash: The NTLM hash for authentication.
-            input: Optional input string to pass to the command.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
         return await self.certipy(
             action="find",
@@ -553,7 +577,7 @@ async def certipy_find(
         )
 
     @tool_method(catch=True, variants=["generic", "all"])
-    async def certipy_req(self, args: list[str], input: str | None = "y") -> str:
+    async def certipy_req(self, args: list[str], input: str | None = None) -> str:
         r"""
         Execute a certipy-ad req command.
 
@@ -643,8 +667,11 @@ async def certipy_req(self, args: list[str], input: str | None = "y") -> str:
 
         Args:
             args: List of arguments for the command.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "req", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "req", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)
 
     @tool_method(catch=True, variants=["generic", "all"])
     async def certipy_template(self, args: list[str], input: str | None = None) -> str:
@@ -713,6 +740,8 @@ async def certipy_template(self, args: list[str], input: str | None = None) -> s
 
         Args:
             args: List of arguments for the command.
-            input: Optional input string to pass to the command's stdin.
+            input: Optional stdin for the command. The default (and any empty
+                value) auto-confirms certipy's interactive prompts with "y";
+                pass an explicit answer such as "n" to respond differently.
         """
-        return await execute([self.certipy_cmd, "template", *args], timeout=self.timeout, input=input)
+        return await execute([self.certipy_cmd, "template", *args], timeout=self.timeout, input=input or _CONFIRM_INPUT)