pss.py

#!/usr/bin/env python3
import argparse
import logging as logger
import os
import re
import shutil
import sys
from pathlib import Path

import colorama
import netifaces
from colorama import Fore
from prompt_toolkit import PromptSession
from prompt_toolkit.auto_suggest import AutoSuggestFromHistory
from prompt_toolkit.completion import WordCompleter
from prompt_toolkit.formatted_text import HTML
from prompt_toolkit.history import FileHistory, InMemoryHistory
from prompt_toolkit.styles import Style

import plugins
import pssutil.transformers
from pssmods.pssdns import PssDnsServer
from pssmods.pssftp import PssFtpServer
from pssmods.psshttp import PssHttpServer
from pssmods.psssmb import PssSmbServer
from pssutil.cmdutils import CmdHandler, CmdLenValidator, ensure_folder
from pssutil.coloredformatter import ColoredFormatter
from pssutil.nestedwordcompleter import NestedWordCompleter
from pssutil.payloadmanager import PayloadManager, get_all_files

VERSION = "0.2.3"


def build_config_path(config_dir: Path, name: str) -> Path:
    if not config_dir.exists():
        os.makedirs(config_dir)

    # path may start with a "/", this is intended,
    # beware of file overwriting
    config_path = Path(config_dir, name)

    if not config_path.parent.exists():
        os.makedirs(config_path)

    return config_path


def change_serve_folder(pm: PayloadManager, folder: Path, smb_server: PssSmbServer, ftp_server: PssFtpServer):
    if folder.exists():
        logger.info(f'Serving folder changed to "{Fore.LIGHTYELLOW_EX}{folder.absolute()}{Fore.RESET}"')

        # prevent the output of smb debug stuff
        prev_logging_level = logger.getLogger().level
        logger.getLogger().setLevel(logger.WARNING)

        # http serve folder is set via config
        pm.cfg.serve_folder = folder
        pm.cfg.tmpfolder = Path(folder, "tmp")
        pm.update_serve_folder_watchdog()

        if pm.cfg.uploads_use_serve_folder:
            pm.cfg.uploads_folder = Path(pm.cfg.serve_folder, "uploads")

        if smb_server:
            smb_server.set_folder(folder)

        if ftp_server:
            ftp_server.set_folder(folder)

        logger.getLogger().setLevel(prev_logging_level)
    else:
        logger.warning(f'Folder "{Fore.LIGHTYELLOW_EX}{folder.absolute()}{Fore.RESET}" does not exist')


def print_help():
    print(
        f""">> {Fore.LIGHTYELLOW_EX}Basic{Fore.RESET}
   - exit                      : Shutdown server and exit.
   - save {{NAME}}               : Save current options and shell selection. No name to save as default.
   - load {{NAME}}               : Load saved options and shell selection. No name to load default.
   - set {{NAME}} {{VALUE}} (s)    : Set the options value.
   - serve {{VALUE}}      (x)    : Set the serving folder.
   - http (copy)               : Print/copy the http server address.
   - smb  (copy)               : Print/copy the smb server address.
   - ftp  (copy)               : Print/copy the ftp server address.
   - pty  (copy, execute, tmux): Generate/copy python pty shell upgrade command.
   - nc   (copy, execute, tmux): Generate/copy netcat listener command.
   - ncat (copy, execute, tmux): Generate/copy ncat listener command.
   - msfl (copy, execute, tmux): Generate/copy metasploit multi/handler listener command.
   - msfv {{PAYLOAD}} {{FORMAT}}   : Generate a msfvenom executable.

>> {Fore.LIGHTYELLOW_EX}Shells{Fore.RESET} (sh, shells)
   - list          (l): List all loaded shells.
   - select {{NAME}} (s): Select a shell payload.
   - options       (o): Show options for the selected shell.

>> {Fore.LIGHTYELLOW_EX}Launchers{Fore.RESET} (l, launchers)
   - list                (l)   : List all loaded launchers.
   - generate {{LAUNCHER}} (copy): Generate/copy a launcher command.

>> {Fore.LIGHTYELLOW_EX}Stagers{Fore.RESET} (st, stagers)
   - list         (l): List all loaded stagers.
   - build {{NAME}} (b): Print stager with shell.

>> {Fore.LIGHTYELLOW_EX}Transformers{Fore.RESET} (t, transformers)
   - list           (l): List all loaded transformers.
   - disableall     (x): Disable all transformers.
   - enable {{NAME}}  (e): Enable a transformer.
   - disable {{NAME}} (d): Disable a transformer."""
    )


def setup_argparser():
    script_dir = Path(__file__).parent
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-S", "--serve", default="/opt/pss/tools", dest="serve_folder", help="Serve folder for the servers", type=str, required=True
    )
    parser.add_argument("--no-servers", default=False, dest="no_servers", help="Disable all servers", action="store_true")

    parser.add_argument("--http-host", default="0.0.0.0", dest="http_host", help="Listen address for the HTTP server", type=str)
    parser.add_argument("--http-port", default="80", dest="http_port", help="Listen port for the HTTP server", type=int)
    parser.add_argument("--no-http", default=False, dest="no_http", help="Disable HTTP server", action="store_true")
    parser.add_argument("--no-uploads", default=True, dest="accept_uploads", help="Disable HTTP file upload support", action="store_false")
    parser.add_argument(
        "-U", "--upload-folder", default=None, dest="uploads_folder", help="Folder to store files uploaded via HTTP POST", type=str
    )

    parser.add_argument("--smb-host", default="0.0.0.0", dest="smb_host", help="Listen address for the SMB server", type=str)
    parser.add_argument("--smb-port", default="445", dest="smb_port", help="Listen port for the SMB server", type=int)
    parser.add_argument("--smb-john-file", default=None, dest="jtr_folder", help="Log NetNTLMv2 hashes to a file", type=str)
    parser.add_argument("--no-smb", default=False, dest="no_smb", help="Disable SMB server", action="store_true")
    parser.add_argument("--no-smb2", default=False, dest="no_smb2", help="Disable SMB2 support", action="store_true")

    parser.add_argument("--ftp-host", default="0.0.0.0", dest="ftp_host", help="Listen address for the FTP server", type=str)
    parser.add_argument("--ftp-port", default="21", dest="ftp_port", help="Listen port for the FTP server", type=int)
    parser.add_argument("--no-ftp", default=False, dest="no_ftp", help="Disable FTP server", action="store_true")

    parser.add_argument("--dns-host", default="0.0.0.0", dest="dns_host", help="Listen address for the DNS server", type=str)
    parser.add_argument("--dns-port", default="53", dest="dns_port", help="Listen port for the DNS server", type=int)
    parser.add_argument("--no-dns", default=False, dest="no_dns", help="Disable DNS server", action="store_true")

    parser.add_argument("--lhost", default="", dest="lhost", help="Default LHOST value", type=str)
    parser.add_argument("--lport", default="4711", dest="lport", help="Default LPORT value", type=str)
    parser.add_argument("--shellbin", default="/bin/bash", dest="shellbin", help="Default shell binary to use", type=str)

    parser.add_argument(
        "--shells-file", default=str(Path(script_dir, "shells.json")), dest="shells_file", help="JSON file containing shells", type=str
    )
    parser.add_argument(
        "--launchers-file",
        default=str(Path(script_dir, "launchers.json")),
        dest="launchers_file",
        help="JSON file containing launchers",
        type=str,
    )
    parser.add_argument(
        "--stagers-file", default=str(Path(script_dir, "stagers.json")), dest="stagers_file", help="JSON file containing stagers", type=str
    )
    parser.add_argument(
        "--no-history", default=False, dest="no_history", help='Disable history saving in file "~/.psshistory"', action="store_true"
    )
    parser.add_argument(
        "--no-plugins",
        default=False,
        dest="no_plugins",
        help='Disable plugin system as it makes your system vulnerable"',
        action="store_true",
    )
    parser.add_argument(
        "--show-logger-name", default=False, dest="show_logger_name", help="Show the current logger's name", action="store_true"
    )
    return parser.parse_args()


def execute_os_cmd(cmd):
    try:
        os_cmd = "".join(cmd[1:])

        try:
            if len(os_cmd) == 0:
                os_cmd = os.environ.get("SHELL", "/bin/bash")

            os.system(os_cmd)
        except Exception as ex:
            logger.warning(f'Failed to execute: "{Fore.LIGHTRED_EX}{os_cmd}{Fore.RESET}": {ex}')
    except Exception as ex:
        logger.warning(f"Failed parse command: {ex}")


def add_pss_commands(pm, cmd, smb_server, ftp_server):
    cmd.add(["?", "help"], lambda _: print_help())
    cmd.add(["exit"], lambda _: exit(0))
    cmd.add(["clear"], lambda _: os.system("cls" if os.name == "nt" else "clear"))

    def toggle_http_dump_headers(args):
        pm.cfg.dump_headers = bool(int(args[0]) if len(args) > 0 else (int(pm.cfg.dump_headers) ^ 1))
        logger.info(f"http_dump_headers: {pm.cfg.dump_headers}")

    cmd.add(["http_dump_headers"], toggle_http_dump_headers)

    def clean_folder(path):
        for root, dirs, files in os.walk(path):
            for f in files:
                os.unlink(Path(root, f))
            for d in dirs:
                shutil.rmtree(Path(root, d))

    cleanup_handler = cmd.add(["cleanup"])
    cleanup_handler.add(["uploads"], lambda _: clean_folder(pm.cfg.uploads_folder))
    cleanup_handler.add(["temp"], lambda _: clean_folder(Path(pm.cfg.serve_folder, "tmp")))

    config_dir = Path(Path(__file__).parent, "configs")
    cmd.add(["save"], lambda args: pm.save_config(build_config_path(config_dir, args[0] if len(args) > 0 else "default")))
    cmd.add(["load"], lambda args: pm.load_config(build_config_path(config_dir, args[0] if len(args) > 0 else "default")))

    def cmd_set(args):
        pm.cvars[args[0]] = " ".join(args[1:])

    cmd.add(
        ["set"],
        CmdLenValidator(cmd_set, 2, ["cvar_name", "value"]),
        {
            "LHOST": netifaces.interfaces(),
            "LPORT": ["25", "80", "443", "1299", "1337", "4444", "4711"],
            "SHELLBIN": ["/bin/sh", "/bin/bash", "/bin/ash", "/bin/zsh", "powershell.exe", "cmd.exe"],
        },
    )
    cmd.add(
        ["serve"],
        CmdLenValidator(lambda args: change_serve_folder(pm, Path(" ".join(args[0:])), smb_server, ftp_server), 1, ["serve_folder"]),
    )


def add_transformer_commands(pm, cmd, transformers_set):
    transformers_handler = cmd.add(["transformer"])

    def cmd_transformers_list(args):
        for transformer in pm.cfg.transformers.keys():
            is_enabled = pm.cfg.transformers[transformer]["enabled"]
            color = Fore.LIGHTGREEN_EX if is_enabled else Fore.LIGHTRED_EX
            print(f"[{color}{'X' if is_enabled else '-'}{Fore.RESET}] {transformer}")

    def cmd_transformers_disableall(args):
        for transformer in pm.cfg.transformers.keys():
            is_enabled = pm.cfg.transformers[transformer]["enabled"]

            if is_enabled:
                pm.cfg.transformers[transformer]["enabled"] = False
                logger.info(f"{transformer} {Fore.LIGHTRED_EX}disabled{Fore.RESET}")

    transformers_handler.add(["+", "enable"], lambda args: pm.tranformer_switch(args[0], True), transformers_set)
    transformers_handler.add(["-", "disable"], lambda args: pm.tranformer_switch(args[0], False), transformers_set)
    transformers_handler.add(["disableall"], cmd_transformers_disableall)
    transformers_handler.add(["list"], cmd_transformers_list)


def add_stager_commands(pm, cmd):
    stagers_handler = cmd.add(["stager"])
    stagers_handler.add(["build"], lambda args: print(pm.build_payload(args[0].lower())), pm.cfg.stagers)
    stagers_handler.add(["list"], lambda _: [print(stager) for stager in pm.cfg.stagers])


def add_shell_commands(pm, cmd):
    shells_handler = cmd.add(["shell"])

    def cmd_shells_build(args):
        shell_name = args[0].lower() if len(args) > 0 else pm.cvars.SHELL
        shell = pm.try_get_shell(shell_name)

        if shell:
            print(pm.cvars.insert_all(pssutil.transformers.ITransformer.apply_enabled(pm.cfg.transformers, shell.payload, pm.cvars)))
        else:
            logger.warning(
                f'No shell found with name "{Fore.LIGHTRED_EX}{shell_name}{Fore.RESET}". If you want to directly inject something, set the SHELL option manually using "set SHELL ..."'
            )

    def cmd_shells_set(args):
        shell = args[0].lower()

        if pm.try_get_shell(shell):
            pm.cvars.SHELL = shell
        else:
            logger.warning(
                f'No shell found with name "{Fore.LIGHTRED_EX}{shell}{Fore.RESET}". If you want to directly inject something, set the SHELL option manually using "set SHELL ..."'
            )

    def cmd_shells_options(args):
        shell = args[0].lower() if len(args) > 0 else pm.cvars.SHELL
        logger.info(f'Selected shell "{Fore.LIGHTYELLOW_EX}{shell}{Fore.RESET}"')

        if shell:
            for cvar in pm.get_cvars_shell(shell):
                val = pm.cvars[cvar]

                if val:
                    logger.info(f'{cvar} = "{Fore.LIGHTYELLOW_EX}{val}{Fore.RESET}"')
                else:
                    logger.info(f"{cvar} = {Fore.LIGHTRED_EX}Not set{Fore.RESET}")

    shells_handler.add(["build"], cmd_shells_build, pm.shell_names)
    shells_handler.add(["list"], lambda _: [print(shell) for shell in pm.cfg.shells])
    shells_handler.add(["set"], cmd_shells_set, pm.shell_names)
    shells_handler.add(["options"], cmd_shells_options, pm.shell_names)


def add_launcher_commands(pm, cmd, smb_server):
    serve_files_completer = WordCompleter(
        lambda: pm.serve_files,
        ignore_case=True,
        match_middle=True,
        pattern=re.compile(r"([a-zA-Z0-9_\.\-]+|[^a-zA-Z0-9_\.\-\s]+)"),
    )

    launchers_handler = cmd.add(["launcher"])

    def cmd_launchers_generate(args):
        launcher, sub_sub_args = CmdHandler.split(args, True)
        print(pm.launcher_generate(launcher, " ".join(sub_sub_args), smb_server.default_share))

    launchers_handler.add(
        ["generate"],
        CmdLenValidator(cmd_launchers_generate, 1, ["launcher"]),
        {x: serve_files_completer for x in pm.launcher_names},
    )
    launchers_handler.add(["list"], lambda _: [print(launcher) for launcher in pm.cfg.launchers])


def add_smb_command(pm, cmd, smb_server):
    smb_handler = cmd.add(["samba"])
    smb_handler_jtr = smb_handler.add(["john"])

    def cmd_smb_jtr_on(args):
        jtr_path = Path(pm.cfg.serve_folder, "smb-hashes.txt")
        smb_server.set_jtr_logfolder(jtr_path)
        logger.info(f'SMB hash output set to: "{Fore.LIGHTYELLOW_EX}{jtr_path}{Fore.RESET}"')

    smb_handler_jtr.add(["on"], cmd_smb_jtr_on)

    def cmd__smb_jtr_off(args):
        smb_server.set_jtr_logfolder(None)
        logger.info(f'SMB hash output "{Fore.LIGHTRED_EX}disabled{Fore.RESET}"')

    smb_handler_jtr.add(["off"], cmd__smb_jtr_off)

    def cmd__smb_jtr_set(args):
        jtr_path = Path(args[0])

        if jtr_path.exists():
            smb_server.set_jtr_logfolder(args[0])
            logger.info(f'SMB hash output set to: "{Fore.LIGHTYELLOW_EX}{jtr_path}{Fore.RESET}"')
        else:
            logger.warning(f'Path does not exist: "{Fore.LIGHTRED_EX}{jtr_path}{Fore.RESET}"')

    smb_handler_jtr.add(["set"], cmd__smb_jtr_set)


def load_plugins(pm, cmd):
    import importlib

    plugin_folder = Path(plugins.__file__).parent

    if plugin_folder.exists():
        loaded_plugins = []

        for f in get_all_files(plugin_folder):
            if f.suffix == ".py" and f.name != "__init__.py":
                try:
                    yield importlib.import_module(f"plugins.{f.stem}")
                    loaded_plugins.append(f.stem)
                except Exception as ex:
                    logger.warning(f'Failed to load plugin "{Fore.LIGHTRED_EX}{f.stem}{Fore.RESET}": {ex}')

        plugins_str = ""

        for p in sorted(loaded_plugins):
            plugins_str += f'"{Fore.CYAN}{p}{Fore.RESET}", '

        logger.info(f"Loaded plugins: {plugins_str.removesuffix(', ')}")


def try_get_default_netiface():
    netiface_re = [
        re.compile(r"^tun0"),
        re.compile(r"^eth[0-9]+"),
        re.compile(r"^ens[0-9]+"),
        re.compile(r"^enp[0-9]+s[0-9]+"),
    ]

    ifaces = netifaces.interfaces()

    for r in netiface_re:
        for x in ifaces:
            if r.match(x):
                return x
    return None


def running_in_venv():
    return hasattr(sys, "real_prefix") or (hasattr(sys, "base_prefix") and sys.base_prefix != sys.prefix)


if __name__ == "__main__":
    colorama.init()
    is_venv = running_in_venv()

    print(
        f">> {Fore.LIGHTYELLOW_EX}Pentest Server Suite{Fore.RESET} v{VERSION} (Python {Fore.LIGHTGREEN_EX}{sys.version.split()[0]}{Fore.RESET}, venv: {Fore.LIGHTGREEN_EX if is_venv else Fore.LIGHTRED_EX}{is_venv}{Fore.RESET})"
    )
    print(f"   [{Fore.LIGHTBLACK_EX}#{Fore.RESET}] https://github.com/Jnnshschl/PentestServerSuite\n")
    print(f'   [{Fore.LIGHTYELLOW_EX}?{Fore.RESET}] Type "help" or "?" to see all available commands.')
    print(f"   [{Fore.LIGHTRED_EX}!{Fore.RESET}] This is software introduces vulnerabilities to your system while running.\n")

    if not is_venv:
        logger.warning(f'Not running in venv, use "{Fore.LIGHTYELLOW_EX}pss.sh{Fore.RESET}" to run.')

    args = setup_argparser()
    ColoredFormatter.init(logger.INFO, args.show_logger_name)

    pm = PayloadManager()

    pm.load_shells_file(args.shells_file)
    logger.info(f"Loaded {Fore.LIGHTYELLOW_EX}{pm.shell_count}{Fore.RESET} Shells")

    pm.load_stagers_file(args.stagers_file)
    logger.info(f"Loaded {Fore.LIGHTYELLOW_EX}{pm.stager_count}{Fore.RESET} Stagers")

    pm.load_launchers_file(args.launchers_file)
    logger.info(f"Loaded {Fore.LIGHTYELLOW_EX}{pm.launcher_count}{Fore.RESET} Launchers")

    # load all tranformers from the transformers.py file
    transformers_set = []

    for name, transformer in pssutil.transformers.get_all():
        pm.transformer_register(transformer)
        transformers_set.append(name)

    logger.info(f"Loaded {Fore.LIGHTYELLOW_EX}{pm.transformer_count}{Fore.RESET} Tranformers")

    # cvar proxy to resolve a network interface to its ip address
    pm.cvars.add_proxy("LHOST", lambda v: netifaces.ifaddresses(v)[netifaces.AF_INET][0]["addr"])

    # set initial cvar values
    pm.cvars.LHOST = try_get_default_netiface()
    pm.cvars.LPORT = args.lport if args.lport else 4711
    pm.cvars.SHELLBIN = args.shellbin if args.shellbin else "/bin/bash"

    serve_folder = Path(args.serve_folder)

    if not ensure_folder(serve_folder):
        logger.warning(f'Failed to create serve folder: "{Fore.LIGHTRED_EX}{serve_folder}{Fore.RESET}"')
        exit(1)

    # start servers in background
    if not args.no_servers:
        http_server = PssHttpServer(args.http_host, int(args.http_port), serve_folder) if not args.no_http else None
        ftp_server = PssFtpServer(args.ftp_host, int(args.ftp_port), serve_folder) if not args.no_ftp else None
        smb_server = PssSmbServer(args.smb_host, int(args.smb_port), serve_folder) if not args.no_smb else None
        dns_server = PssDnsServer(args.dns_host, int(args.dns_port), pm) if not args.no_dns else None

        if args.jtr_folder:
            jtr_folder = Path(args.jtr_folder)

            if jtr_folder.parent.exists():
                smb_server.set_jtr_logfolder(jtr_folder)
            else:
                logger.error(f'John output path invalid: "{Fore.LIGHTRED_EX}{jtr_folder}{Fore.RESET}"')
    else:
        http_server = None
        ftp_server = None
        smb_server = None
        dns_server = None

    change_serve_folder(pm, serve_folder, smb_server, ftp_server)

    # set uploads folder to supplied value, otherwise place stuff in the serve folder
    pm.cfg.accept_uploads = args.accept_uploads

    if pm.cfg.accept_uploads:
        if args.uploads_folder:
            pm.cfg.uploads_folder = Path(args.uploads_folder)
        else:
            pm.cfg.uploads_folder = Path(pm.cfg.serve_folder, "uploads")
            pm.cfg.uploads_use_serve_folder = True

        if not ensure_folder(serve_folder):
            logger.warning(f'Failed to create uploads folder: "{Fore.LIGHTRED_EX}{pm.cfg.uploads_folder}{Fore.RESET}"')
            exit(1)

    # setup prompt
    cmd = CmdHandler()
    add_pss_commands(pm, cmd, smb_server, ftp_server)
    add_launcher_commands(pm, cmd, smb_server)
    add_shell_commands(pm, cmd)
    add_stager_commands(pm, cmd)
    add_transformer_commands(pm, cmd, transformers_set)
    add_smb_command(pm, cmd, smb_server)

    if not args.no_plugins:
        loaded_plugins = load_plugins(pm, cmd)

        for plugin in loaded_plugins:
            if hasattr(plugin, "add_cmd"):
                plugin.add_cmd(pm, cmd)
    else:
        loaded_plugins = None

    def print_toolbar():
        bottom_str = f'SHELL: <style bg="ansibrightyellow">{pm.cvars.SHELL}</style>'
        bottom_str += f' | SHELLBIN: <style bg="ansibrightyellow">{pm.cvars.SHELLBIN}</style>'
        bottom_str += f' | LHOST: <style bg="ansibrightyellow">{pm.cvars.LHOST}</style>'
        bottom_str += f' | LPORT: <style bg="ansibrightyellow">{pm.cvars.LPORT}</style>'
        bottom_str += f' | Serving: "<style bg="ansibrightyellow">{pm.cfg.serve_folder.absolute()}</style>"'
        return HTML(bottom_str)

    history_file = Path(os.path.expanduser("~"), ".psshistory")

    cli_session = PromptSession(
        history=InMemoryHistory() if args.no_history else FileHistory(history_file),
        auto_suggest=AutoSuggestFromHistory(),
        bottom_toolbar=print_toolbar,
        style=Style.from_dict({"pss": "ansibrightyellow", "bottom-toolbar": "#303030 bg:#dddddd"}),
        completer=NestedWordCompleter.from_nested_dict(cmd.tree()),
        complete_in_thread=True,
        complete_while_typing=True,
        message=[("class:pss", "pss"), ("", "> ")],
    )

    while 1:
        try:
            in_str = cli_session.prompt().strip()

            if len(in_str) > 0:
                if in_str.startswith("!"):
                    execute_os_cmd(in_str)
                else:
                    cmd.handle(list(filter(None, in_str.split(" "))))

        except KeyboardInterrupt:
            logger.info(f'Press Ctrl-D or type "{Fore.LIGHTYELLOW_EX}exit{Fore.RESET}" to exit...')
        except EOFError:
            break
        except Exception as e:
            logger.error("%s", str(e), exc_info=True)

    if loaded_plugins:
        for plugin in loaded_plugins:
            if hasattr(plugin, "unload"):
                plugin.unload()

    colorama.deinit()