Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions src/attackmate/executors/shell/shellexecutor.py
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,13 @@ def popen_noninteractive(self, proc: subprocess.Popen, cmd: bytes, timeout=None)
proc.kill()
output, error = proc.communicate()
output += error
return output.decode()
# Command output is arbitrary bytes, not guaranteed UTF-8. A strict decode
# raises UnicodeDecodeError, which nothing between here and main() catches -
# so one undecodable byte terminates the whole playbook rather than failing
# this step. Replace undecodable bytes instead: output is used for logging,
# error_if matching and save-to-file, none of which need a lossless
# round-trip.
return output.decode(errors='replace')

def popen_interactive(
self, proc: subprocess.Popen, cmd: bytes, timeout: int = 5, read: bool = True
Expand All @@ -96,7 +102,8 @@ def popen_interactive(
outline += tmp
begin = datetime.now() # reset timer when data comes

return outline.decode()
# Same reasoning as popen_noninteractive: never let output bytes end the run.
return outline.decode(errors='replace')

async def _exec_cmd(self, command: ShellCommand) -> Result:
try:
Expand Down
48 changes: 48 additions & 0 deletions test/units/test_shellexecutor.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,3 +99,51 @@ async def test_execution_of_hex_command(shell_executor):

result = await shell_executor._exec_cmd(command)
assert result.stdout == 'id\n'


@pytest.mark.asyncio
async def test_non_utf8_output_does_not_raise(mock_popen, shell_executor):
"""A command emitting a non-UTF-8 byte must not terminate the playbook.

0xc9 without a continuation byte is not valid UTF-8. A strict decode raises
UnicodeDecodeError, which nothing catches between _exec_cmd and main().
"""
mock_open_proc, mock_popen_instance = mock_popen
mock_popen_instance.communicate.return_value = (b'ok\xc9done', b'')
with patch.object(shell_executor, 'open_proc', mock_open_proc):

command = ShellCommand(
type='shell',
cmd="printf 'ok\\311done'",
bin=False,
)
result = await shell_executor._exec_cmd(command)

assert result.stdout == 'ok�done'


def test_popen_interactive_non_utf8_output_does_not_raise(shell_executor):
"""The interactive read path decodes separately and needs the same treatment."""
reads = [b'ok\xc9done']

def fake_read(_stdout):
return reads.pop(0) if reads else b''

mock_popen_instance = MagicMock()
with patch.object(shell_executor, 'non_block_read', side_effect=fake_read):
output = shell_executor.popen_interactive(mock_popen_instance, b'cmd\n', timeout=1)

assert output == 'ok�done'


@pytest.mark.asyncio
async def test_execution_of_command_with_non_utf8_output(shell_executor):
"""End to end: /bin/sh printf has no \\xHH, so the octal escape is the portable one."""
command = ShellCommand(
type='shell',
cmd="printf 'ok\\311done'",
bin=False,
)

result = await shell_executor._exec_cmd(command)
assert result.stdout == 'ok�done'
Loading