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

Agent: Make ExploiterResultData a dataclass instead of a named tuple 

and modify HostExploiter and the SSH exploiter accordingly
This commit is contained in:
Shreya Malviya 2022-02-23 18:24:54 +05:30
parent 2a8186928d
commit e993998432
3 changed files with 53 additions and 61 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
monkey/infection_monkey/exploit/HostExploiter.py
View File

@ -85,14 +85,9 @@ class HostExploiter:
return result return result
def pre_exploit(self): def pre_exploit(self):
self.exploit_result = { self.exploit_result = ExploiterResultData(
"exploitation_success": False, os=self.host.os.get("type"), info=self.exploit_info, attempts=self.exploit_attempts
"propagation_success": False, )
"os": self.host.os.get("type"),
"info": self.exploit_info,
"attempts": self.exploit_attempts,
"error_message": "",
}
self.set_start_time() self.set_start_time()
def post_exploit(self): def post_exploit(self):
@ -115,13 +110,3 @@ class HostExploiter:
""" """
powershell = True if "powershell" in cmd.lower() else False powershell = True if "powershell" in cmd.lower() else False
self.exploit_info["executed_cmds"].append({"cmd": cmd, "powershell": powershell}) self.exploit_info["executed_cmds"].append({"cmd": cmd, "powershell": powershell})
def return_exploit_result_data(self) -> ExploiterResultData:
return ExploiterResultData(
self.exploit_result["exploitation_success"],
self.exploit_result["propagation_success"],
self.exploit_result["os"],
self.exploit_result["info"],
self.exploit_result["attempts"],
self.exploit_result["error_message"],
)

76
monkey/infection_monkey/exploit/sshexec.py
View File

@ -111,22 +111,22 @@ class SSHExploiter(HostExploiter):
is_open, _ = check_tcp_port(self.host.ip_addr, port) is_open, _ = check_tcp_port(self.host.ip_addr, port)
if not is_open: if not is_open:
self.exploit_result["error_message"] = f"SSH port is closed on {self.host}, skipping" self.exploit_result.error_message = f"SSH port is closed on {self.host}, skipping"
logger.info(self.exploit_result["error_message"]) logger.info(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result
try: try:
ssh = self.exploit_with_ssh_keys(port) ssh = self.exploit_with_ssh_keys(port)
self.exploit_result["exploitation_success"] = True self.exploit_result.exploitation_success = True
except FailedExploitationError: except FailedExploitationError:
try: try:
ssh = self.exploit_with_login_creds(port) ssh = self.exploit_with_login_creds(port)
self.exploit_result["exploitation_success"] = True self.exploit_result.exploitation_success = True
except FailedExploitationError: except FailedExploitationError:
self.exploit_result["error_message"] = "Exploiter SSHExploiter is giving up..." self.exploit_result.error_message = "Exploiter SSHExploiter is giving up..."
logger.debug(self.exploit_result["error_message"]) logger.debug(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result
if not self.host.os.get("type"): if not self.host.os.get("type"):
try: try:
@ -134,20 +134,20 @@ class SSHExploiter(HostExploiter):
uname_os = stdout.read().lower().strip().decode() uname_os = stdout.read().lower().strip().decode()
if "linux" in uname_os: if "linux" in uname_os:
self.host.os["type"] = "linux" self.host.os["type"] = "linux"
self.exploit_result["os"] = "linux" self.exploit_result.os = "linux"
else: else:
self.exploit_result["error_message"] = f"SSH Skipping unknown os: {uname_os}" self.exploit_result.error_message = f"SSH Skipping unknown os: {uname_os}"
if not uname_os: if not uname_os:
logger.error(self.exploit_result["error_message"]) logger.error(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result
except Exception as exc: except Exception as exc:
self.exploit_result[ self.exploit_result.error_message = (
"error_message" f"Error running uname os command on victim {self.host}: ({exc})"
] = f"Error running uname os command on victim {self.host}: ({exc})" )
logger.debug(self.exploit_result["error_message"]) logger.debug(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result
if not self.host.os.get("machine"): if not self.host.os.get("machine"):
try: try:
@ -156,20 +156,20 @@ class SSHExploiter(HostExploiter):
if "" != uname_machine: if "" != uname_machine:
self.host.os["machine"] = uname_machine self.host.os["machine"] = uname_machine
except Exception as exc: except Exception as exc:
self.exploit_result[ self.exploit_result.error_message = (
"error_message" f"Error running uname machine command on victim {self.host}: ({exc})"
] = f"Error running uname machine command on victim {self.host}: ({exc})" )
logger.error(self.exploit_result["error_message"]) logger.error(self.exploit_result.error_message)
src_path = get_target_monkey(self.host) src_path = get_target_monkey(self.host)
if not src_path: if not src_path:
self.exploit_result[ self.exploit_result.error_message = (
"error_message" f"Can't find suitable monkey executable for host {self.host}"
] = f"Can't find suitable monkey executable for host {self.host}" )
logger.info(self.exploit_result["error_message"]) logger.info(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result
try: try:
ftp = ssh.open_sftp() ftp = ssh.open_sftp()
@ -193,10 +193,10 @@ class SSHExploiter(HostExploiter):
) )
ftp.close() ftp.close()
except Exception as exc: except Exception as exc:
self.exploit_result[ self.exploit_result.error_message = (
"error_message" f"Error uploading file into victim {self.host}: ({exc})"
] = f"Error uploading file into victim {self.host}: ({exc})" )
logger.error(self.exploit_result["error_message"]) logger.error(self.exploit_result.error_message)
status = ScanStatus.SCANNED status = ScanStatus.SCANNED
self.telemetry_messenger.send_telemetry( self.telemetry_messenger.send_telemetry(
@ -205,7 +205,7 @@ class SSHExploiter(HostExploiter):
) )
) )
if status == ScanStatus.SCANNED: if status == ScanStatus.SCANNED:
return self.return_exploit_result_data() return self.exploit_result
try: try:
cmdline = "%s %s" % (self.options["dropper_target_path_linux"], MONKEY_ARG) cmdline = "%s %s" % (self.options["dropper_target_path_linux"], MONKEY_ARG)
@ -220,16 +220,16 @@ class SSHExploiter(HostExploiter):
cmdline, cmdline,
) )
self.exploit_result["propagation_success"] = True self.exploit_result.propagation_success = True
ssh.close() ssh.close()
self.add_executed_cmd(cmdline) self.add_executed_cmd(cmdline)
return self.return_exploit_result_data() return self.exploit_result
except Exception as exc: except Exception as exc:
self.exploit_result[ self.exploit_result.error_message = (
"error_message" f"Error running monkey on victim {self.host}: ({exc})"
] = f"Error running monkey on victim {self.host}: ({exc})" )
logger.error(self.exploit_result["error_message"]) logger.error(self.exploit_result.error_message)
return self.return_exploit_result_data() return self.exploit_result

17
monkey/infection_monkey/i_puppet/i_puppet.py
View File

@ -1,8 +1,9 @@
import abc import abc
import threading import threading
from collections import namedtuple from collections import namedtuple
from dataclasses import dataclass
from enum import Enum from enum import Enum
from typing import Dict, List, Sequence from typing import Dict, Iterable, List, Mapping, Sequence
from . import Credentials, PluginType from . import Credentials, PluginType
@ -16,10 +17,16 @@ class UnknownPluginError(Exception):
pass pass
ExploiterResultData = namedtuple( @dataclass
"ExploiterResultData", class ExploiterResultData:
["exploitation_success", "propagation_success", "os", "info", "attempts", "error_message"], exploitation_success: bool = False
) propagation_success: bool = False
os: str = ""
info: Mapping = None
attempts: Iterable = None
error_message: str = ""
PingScanData = namedtuple("PingScanData", ["response_received", "os"]) PingScanData = namedtuple("PingScanData", ["response_received", "os"])
PortScanData = namedtuple("PortScanData", ["port", "status", "banner", "service"]) PortScanData = namedtuple("PortScanData", ["port", "status", "banner", "service"])
FingerprintData = namedtuple("FingerprintData", ["os_type", "os_version", "services"]) FingerprintData = namedtuple("FingerprintData", ["os_type", "os_version", "services"])