Source code for fortrace.utility.console_applications.metasploit_console
import os
import time
import pexpect
from pexpect import replwrap
from fortrace.utility.console_applications.console_application import (
GenericConsoleApplication,
)
from fortrace.utility.console_applications.text_editor.text_editor import (
GenericConsoleTextEditor,
)
from fortrace.utility.console_applications.text_editor.text_editor_factory import (
get_console_text_editor,
)
from fortrace.utility.string_filtering import ansi_escape, unicode_control_escape
[docs]
class MetasploitConsole(GenericConsoleApplication):
"""Class to handle a MetasploitConsole opened in a VirshConsole session."""
def __init__(self, pty: replwrap.REPLWrapper):
"""Create a MetasploitConsole object.
Args:
pty: handle to VirshConsole's pty attribute
"""
super().__init__(pty, "Metasploit Console")
self.__closed = False # FIXME: this seems to be an unused parameter
self._pty.prompt = "$$ "
self._pty.continuation_prompt = "~~ "
self._virsh.sendline("set PromptChar $$")
self._virsh.expect_exact("$$") # echo of set command
self._virsh.expect_exact("$$ ") # real prompt
[docs]
def close(self):
self._pty.prompt = self._parent_prompt.prompt
self._pty.continuation_prompt = self._parent_prompt.continuation_prompt
try:
self._pty.run_command(
"exit -y", 10
) # give Meterpreter some time to shut down
except pexpect.exceptions.ExceptionPexpect:
pass
self.__closed = True
[docs]
def run_command(self, command: str, timeout: int | None = None) -> str | list[str]:
"""Run a command in the metasploit console.
Args:
command: command to be run
timeout: optional timeout (useful if command takes longer)
Returns:
If the output is a single line a string is returned, else a list of strings or an empty list
"""
stdout = self._pty.run_command(command, timeout)
stdout = stdout.splitlines()
stdout = [
unicode_control_escape(ansi_escape(line)).strip() for line in stdout if line
] # TODO: ansi_escape seems superflous
if (
len(stdout) == 3
): # output is like: command, output, prompt --> we only want the output
return stdout[1]
else:
return stdout[1:-1] # omit command and beginning of msf prompt
[docs]
def edit(self, file: os.PathLike) -> GenericConsoleTextEditor:
"""Open/Create a file with the VIM editor from within the Metasploit console
Args:
file: path to the file
Returns:
handle to console text editor
"""
self._virsh.sendline(f"edit {file}")
return get_console_text_editor(
"vi", self._pty
) # TODO: since vim is used here, type hint can be updated
[docs]
def open_shell(self) -> "ShellSession":
"""Open a shell in MetasploitConsole.
Use this method if the metasploit run command opens a shell without a prompt.
Returns:
ShellSession object
"""
self._pty.run_command("run", None)
return ShellSession(self._pty)
[docs]
def open_meterpreter_session(self, session_id: int) -> "MeterpreterSession":
"""Start a Meterpreter session.
Args:
session_id: ID of already created Meterpreter session
Returns:
opened Meterpreter session
"""
self._pty.prompt = MeterpreterSession.PROMPT
self._pty.run_command(f"sessions -i {session_id}")
return MeterpreterSession(self._pty, session_id)
[docs]
class MeterpreterSession(GenericConsoleApplication):
"""Class to handle an opened Meterpreter session."""
PROMPT = "\x1b[4mmeterpreter\x1b[0m > "
def __init__(self, pty: replwrap.REPLWrapper, session_id: id):
"""Create a MeterpreterSession object.
Args:
pty: handle to VirshConsole's pty attribute
session_id: ID of this Meterpreter session
"""
super().__init__(pty, "Meterpreter session")
self._session_id = session_id
self._pty.prompt = (
MeterpreterSession.PROMPT
) # beginning of prompt is underscore(meterpreter)
[docs]
def open_shell(self) -> "ShellSession":
"""Open a Shell from within MeterpreterSession.
Returns:
handle to opened shell
"""
# shell opened by meterpreter has no prompt, thus pty won't work
self._virsh.sendline("shell")
return ShellSession(self._pty)
[docs]
def close(self):
"""Close the Meterpreter session."""
# TODO: replace with self._parent_prompt
self._pty.prompt = "$$ "
self._pty.continuation_prompt = "~~ "
self._pty.run_command("exit") # background session and return to msf console
[docs]
def suspend(self):
"""Suspend the Meterpreter session to the background.
Can be opened again later with the same ID.
"""
self._pty.prompt = "$$ "
self._pty.continuation_prompt = "~~ "
self._pty.run_command("bg") # background session and return to msf console
[docs]
def run_command(self, command: str, timeout: int | None = None) -> str | list[str]:
"""Run a command inside the Meterpreter.
Args:
command: command to be run
timeout: maximum time to wait for output
Returns:
The output without the entered command.
"""
stdout = self._pty.run_command(command, timeout)
stdout = ansi_escape(stdout).splitlines()
stdout = [
unicode_control_escape(line).strip() for line in stdout if line
] # FIXME: there are no unicode chars but empty lines
return stdout[1:]
[docs]
def edit(self, file: os.PathLike) -> GenericConsoleTextEditor:
"""Edit a file in the Meterpreter.
Args:
file: file to be edited
Returns:
handle to opened text editor
"""
self._virsh.sendline(f"edit {file}")
return get_console_text_editor(
"vi", self._pty
) # TODO: same as above with type hint
[docs]
class ShellSession(GenericConsoleApplication):
"""Class to be used when dealing with an opened shell session without a prompt."""
def __init__(self, pty: replwrap.REPLWrapper):
"""Create a new ShellSession
Args:
pty: VirshConsole's pty attribute
"""
super().__init__(pty, "Meterpreter Shell Session")
# skip three following outputs
self._virsh.readline() # shell command
self._virsh.readline() # Process XXXX created
self._virsh.readline() # Channel X created
[docs]
def run_command(
self, command: str, sleep: int | None = None, expect: str | None = None
) -> str | list[str]:
"""Run a command in a shell opened from within a Meterpreter session.
There is no prompt, so we need to read everything or need to know what to expect.
Args:
command: the command to run
sleep: an estimate by the user how long the command will take.
None, if the result is there immediately
expect: Some string to expect, if the command has finished.
None to read until timeout exception
Returns:
a list of strings for multiline outputs,
a single string for single line output,
an empty list for no output
"""
self._virsh.sendline(command)
if sleep:
time.sleep(sleep)
self._virsh.readline() # skip entered command
stdout = []
if expect is not None:
self._virsh.expect_exact(expect)
stdout = self._virsh.before.splitlines() # TODO: right one?
else:
try:
while True:
stdout.append(
unicode_control_escape(ansi_escape(self._virsh.readline()))
)
except pexpect.exceptions.ExceptionPexpect:
pass
if len(stdout) == 1:
return stdout[0]
else:
return stdout
[docs]
def close(self):
"""Close the Shell session"""
self._virsh.sendline("exit")
self._virsh.expect_exact(MeterpreterSession.PROMPT)
[docs]
def background(self):
"""Background the shell session, so it can be used again later."""
self._virsh.sendcontrol("z") # background shell session
self._virsh.expect_exact(MeterpreterSession.PROMPT)