From 8f3e3828b7b46af1d3ff07a7555e81e39f7906cf Mon Sep 17 00:00:00 2001 From: Thomas Q Brady Date: Thu, 15 Oct 2026 17:37:24 -0400 Subject: [PATCH] fix(network-ops): make certipy stdin confirmation uniform and hang-safe (ENG-8830) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The certipy_* tools were split on auto-answering certipy's interactive prompts: certipy_auth/cert/req (and the high-level request/auth helpers) piped "y" to stdin, while certipy_account/ca/find/template passed input=None. The None path leaves stdin inherited, so when certipy prompts (e.g. overwrite a saved PFX/ccache) the tool blocks until the 20s command timeout and surfaces as a timeout failure. Make every certipy_* tool default to the same confirmation via a single, documented _CONFIRM_INPUT constant, so stdin is always written and closed and no tool can hang on a prompt. Behavior stays overridable through the existing family-wide `input` parameter (pass "n" to decline); an empty string falls back to the confirm default because execute() only closes stdin for a truthy input. The relay action is exempt — it is a streaming listener that never reads stdin. Bumps network-ops to 2.1.11 and adds regression tests covering the uniform default, explicit/empty overrides, and the relay exemption. --- capabilities/network-ops/capability.yaml | 2 +- .../network-ops/tests/test_tool_fixes.py | 121 ++++++++++++++++++ capabilities/network-ops/tools/certipy.py | 67 +++++++--- 3 files changed, 170 insertions(+), 20 deletions(-) diff --git a/capabilities/network-ops/capability.yaml b/capabilities/network-ops/capability.yaml index 2f64d8d..9c680e5 100644 --- a/capabilities/network-ops/capability.yaml +++ b/capabilities/network-ops/capability.yaml @@ -1,6 +1,6 @@ schema: 1 name: network-ops -version: "2.1.10" +version: "2.1.11" description: > Network operations and Active Directory exploitation. Multi-agent pipeline for autonomous red teaming with Nmap scanning, Netexec diff --git a/capabilities/network-ops/tests/test_tool_fixes.py b/capabilities/network-ops/tests/test_tool_fixes.py index 5849c36..9bf0a2c 100644 --- a/capabilities/network-ops/tests/test_tool_fixes.py +++ b/capabilities/network-ops/tests/test_tool_fixes.py @@ -296,6 +296,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 # =================================================================== diff --git a/capabilities/network-ops/tools/certipy.py b/capabilities/network-ops/tools/certipy.py index e27990c..26f7431 100644 --- a/capabilities/network-ops/tools/certipy.py +++ b/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)