📄 Vertex AI Experiments 1.132.x Predictable Bucket Naming

=============================================================================================================================================
    | # Title     : Vertex AI Experiments 1.21.0 to 1.132.x Predictable Bucket Naming Leading to Cross‑Tenant RCE and Model Theft               |
    | # Author    : indoushka                                                                                                                   |
    | # Tested on : windows 11 Fr(Pro) / browser : Mozilla firefox 147.0.4 (64 bits)                                                            |
    | # Vendor    : https://cloud.google.com/vertex-ai                                                                                          |
    =============================================================================================================================================
    
    [+] Summary    :  A vulnerability identified as CVE-2026-2473 affected Google Cloud Vertex AI, specifically the Vertex AI Experiments component, in versions 1.21.0 through 1.132.x (fixed in 1.133.0 and later).
                      The issue stemmed from predictable Cloud Storage bucket naming patterns, enabling a class of attack known as Bucket Squatting. 
    				  Because bucket names in Google Cloud Storage are globally unique, an attacker could pre-create a bucket matching a predictable name before a legitimate tenant did.
    
    This allowed a potential unauthenticated remote attacker to:
    
    Achieve cross-tenant Remote Code Execution (RCE)
    
    Exfiltrate trained models (Model Theft)
    
    Poison experiment artifacts or training outputs
    
    Interfere with experiment tracking pipelines
    
    
    [+] POC   :  
    
    #!/usr/bin/env python3
    
    import google.cloud.storage as storage
    import time
    import sys
    import os
    import json
    import argparse
    from datetime import datetime
    from typing import List, Dict
    import random
    import string
    
    class VertexAIBucketDefensiveScanner:
        """
        Scanner for detecting Bucket Squatting risk
        in Google Cloud Storage used by Vertex AI Experiments by indoushka.
        """
    
        def __init__(self, project_id: str, credentials_path: str):
            self.project_id = project_id
            self.findings = []
    
            try:
                self.client = storage.Client.from_service_account_json(credentials_path)
                print(f"[✓] Connected to GCS project: {project_id}")
            except Exception as e:
                print(f"[✗] Connection failed: {e}")
                sys.exit(1)
    
            self.bucket_patterns = [
                "vertex-ai-experiment-{exp_id}-{timestamp}",
                "vertex-ai-exp-{exp_id}-data-{timestamp}",
                "vertex-{exp_id}-experiment-{random}",
                "ai-platform-{exp_id}-{timestamp}",
                "vertex-ai-{exp_id}-{user_id}",
                "experiment-{exp_id}-bucket-{timestamp}",
                "vertex-{random}-{exp_id}-data"
            ]
    
        def generate_bucket_names(self, exp_id: str, count: int = 5) -> List[str]:
            predicted = []
            timestamp = int(time.time())
    
            for pattern in self.bucket_patterns:
                for i in range(count):
                    random_str = ''.join(random.choices(string.ascii_lowercase + string.digits, k=8))
                    user_id = f"user-{random.randint(1000, 9999)}"
    
                    name = pattern.format(
                        exp_id=exp_id,
                        timestamp=timestamp + i,
                        random=random_str,
                        user_id=user_id
                    )
                    predicted.append(name)
    
            return predicted
    
        def check_bucket_status(self, bucket_name: str) -> Dict:
            """
            Checks if bucket exists and whether accessible
            WITHOUT creating it.
            """
    
            bucket = self.client.bucket(bucket_name)
    
            try:
                exists = bucket.exists()
    
                if not exists:
                    return {
                        "bucket": bucket_name,
                        "status": "AVAILABLE",
                        "risk": "HIGH",
                        "reason": "Predictable name and not registered."
                    }
                try:
                    bucket.reload()
                    owner = bucket.project_number
    
                    if owner != self.client.project:
                        return {
                            "bucket": bucket_name,
                            "status": "EXISTS_EXTERNAL",
                            "risk": "CRITICAL",
                            "reason": "Bucket exists but not owned by this project."
                        }
    
                    return {
                        "bucket": bucket_name,
                        "status": "OWNED",
                        "risk": "LOW",
                        "reason": "Bucket exists and owned by project."
                    }
    
                except Exception:
                    return {
                        "bucket": bucket_name,
                        "status": "EXISTS_NO_ACCESS",
                        "risk": "MEDIUM",
                        "reason": "Bucket exists but metadata not accessible."
                    }
    
            except Exception as e:
                return {
                    "bucket": bucket_name,
                    "status": "ERROR",
                    "risk": "UNKNOWN",
                    "reason": str(e)
                }
    
        def scan_experiment_range(self, start_id: int, end_id: int):
            print(f"[*] Scanning experiment IDs {start_id} → {end_id}")
    
            for exp_id in range(start_id, end_id + 1):
                bucket_names = self.generate_bucket_names(str(exp_id), count=3)
    
                for name in bucket_names:
                    result = self.check_bucket_status(name)
                    self.findings.append(result)
    
                    print(f"[{result['risk']}] {name} → {result['status']}")
    
                    time.sleep(0.2)
    
        def generate_report(self) -> Dict:
            summary = {
                "CRITICAL": len([f for f in self.findings if f["risk"] == "CRITICAL"]),
                "HIGH": len([f for f in self.findings if f["risk"] == "HIGH"]),
                "MEDIUM": len([f for f in self.findings if f["risk"] == "MEDIUM"]),
                "LOW": len([f for f in self.findings if f["risk"] == "LOW"]),
            }
    
            return {
                "timestamp": datetime.now().isoformat(),
                "project_id": self.project_id,
                "summary": summary,
                "total_checked": len(self.findings),
                "findings": self.findings
            }
    
    
    def main():
        parser = argparse.ArgumentParser(
            description="Defensive Scanner - Vertex AI Bucket Squatting Risk"
        )
        parser.add_argument('--project', required=True)
        parser.add_argument('--creds', required=True)
        parser.add_argument('--start', type=int, default=1000)
        parser.add_argument('--end', type=int, default=1010)
    
        args = parser.parse_args()
    
        scanner = VertexAIBucketDefensiveScanner(
            project_id=args.project,
            credentials_path=args.creds
        )
    
        print("=" * 60)
        print("Vertex AI Bucket Squatting Defensive Scanner")
        print("Mode: Detection Only (No Exploitation)")
        print("=" * 60)
    
        scanner.scan_experiment_range(args.start, args.end)
    
        report = scanner.generate_report()
    
        with open("defensive_scan_report.json", "w") as f:
            json.dump(report, f, indent=2)
    
        print("\n[✓] Report saved to defensive_scan_report.json")
    
    
    if __name__ == "__main__":
        exit(main())
    	
    Greetings to :==============================================================================
    jericho * Larry W. Cashdollar * r00t * Yougharta Ghenai * Malvuln (John Page aka hyp3rlinx)|
    ============================================================================================