Why We Replaced AWS IAM with HashiCorp Vault 2.0 for Multi-Cloud – Cut Ops Overhead by 40%

import hvac
import boto3
import logging
import time
import os
import sys
from typing import Dict, Optional

# Configure logging for audit trails
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.StreamHandler(sys.stdout)]
)
logger = logging.getLogger(__name__)

class VaultAWSDynamicSecretProvider:
    \"\"\"Manages dynamic AWS secret retrieval and rotation via HashiCorp Vault 2.0\"\"\"

    def __init__(self, vault_addr: str, vault_token: str, aws_secret_path: str = \"aws/sts/prod-apps\"):
        \"\"\"
        Initialize Vault client and validate connectivity

        Args:
            vault_addr: Full Vault address (e.g., https://vault.example.com:8200)
            vault_token: Vault authentication token (use approle or OIDC in prod)
            aws_secret_path: Vault path for AWS dynamic secrets
        \"\"\"
        self.vault_addr = vault_addr
        self.vault_token = vault_token
        self.aws_secret_path = aws_secret_path
        self.client = None
        self._init_vault_client()

    def _init_vault_client(self) -> None:
        \"\"\"Initialize and validate Vault client connection with retries\"\"\"
        max_retries = 3
        retry_delay = 2  # seconds

        for attempt in range(max_retries):
            try:
                self.client = hvac.Client(url=self.vault_addr, token=self.vault_token)
                if not self.client.is_authenticated():
                    raise hvac.exceptions.Unauthorized(\"Vault token is invalid or expired\")
                logger.info(\"Successfully authenticated to Vault at %s\", self.vault_addr)
                return
            except Exception as e:
                logger.warning(\"Vault init attempt %d failed: %s\", attempt + 1, str(e))
                if attempt < max_retries - 1:
                    time.sleep(retry_delay * (attempt + 1))
                else:
                    logger.error(\"Failed to initialize Vault client after %d attempts\", max_retries)
                    raise

    def get_dynamic_aws_creds(self) -> Optional[Dict]:
        \"\"\"
        Retrieve short-lived dynamic AWS credentials from Vault

        Returns:
            Dict with access_key, secret_key, session_token, lease_id
        \"\"\"
        try:
            response = self.client.secrets.aws.generate_credentials(
                name=self.aws_secret_path.split(\"/\")[-1],  # Extract role name from path
                mount_point=\"/\".join(self.aws_secret_path.split(\"/\")[:-1])  # Extract mount point
            )

            creds = response[\"data\"]
            logger.info(
                \"Retrieved dynamic AWS creds with lease ID %s, ttl %d seconds\",
                response[\"lease_id\"],
                creds.get(\"lease_ttl\", 0)
            )
            return {
                \"access_key\": creds[\"access_key\"],
                \"secret_key\": creds[\"secret_key\"],
                \"session_token\": creds.get(\"security_token\"),
                \"lease_id\": response[\"lease_id\"],
                \"ttl\": creds.get(\"lease_ttl\", 3600)
            }
        except hvac.exceptions.InvalidPath as e:
            logger.error(\"Invalid Vault path %s: %s\", self.aws_secret_path, str(e))
            return None
        except Exception as e:
            logger.error(\"Failed to retrieve AWS creds from Vault: %s\", str(e))
            return None

    def revoke_lease(self, lease_id: str) -> bool:
        \"\"\"Revoke a Vault lease to immediately invalidate credentials\"\"\"
        try:
            self.client.sys.revoke(lease_id)
            logger.info(\"Revoked Vault lease %s\", lease_id)
            return True
        except Exception as e:
            logger.error(\"Failed to revoke lease %s: %s\", lease_id, str(e))
            return False

if __name__ == \"__main__\":
    # Load config from environment variables (never hardcode creds!)
    vault_addr = os.getenv(\"VAULT_ADDR\", \"https://vault.example.com:8200\")
    vault_token = os.getenv(\"VAULT_TOKEN\")
    if not vault_token:
        logger.error(\"VAULT_TOKEN environment variable is required\")
        sys.exit(1)

    provider = VaultAWSDynamicSecretProvider(vault_addr, vault_token)
    creds = provider.get_dynamic_aws_creds()

    if not creds:
        logger.error(\"Failed to retrieve dynamic AWS credentials\")
        sys.exit(1)

    # Test creds with boto3
    try:
        sts_client = boto3.client(
            \"sts\",
            aws_access_key_id=creds[\"access_key\"],
            aws_secret_access_key=creds[\"secret_key\"],
            aws_session_token=creds[\"session_token\"]
        )
        identity = sts_client.get_caller_identity()
        logger.info(\"Successfully assumed role: %s\", identity[\"Arn\"])
    except Exception as e:
        logger.error(\"Boto3 client validation failed: %s\", str(e))
        sys.exit(1)
    finally:
        # Always revoke lease when done (short-lived creds best practice)
        if creds.get(\"lease_id\"):
            provider.revoke_lease(creds[\"lease_id\"])

# Terraform configuration for HashiCorp Vault 2.0 multi-cloud cluster on AWS
# Requires Terraform 1.6+, AWS provider 5.0+, Vault provider 3.0+

terraform {
  required_version = \">= 1.6.0\"
  required_providers {
    aws = {
      source  = \"hashicorp/aws\"
      version = \">= 5.0.0\"
    }
    vault = {
      source  = \"hashicorp/vault\"
      version = \">= 3.0.0\"
    }
    tls = {
      source  = \"hashicorp/tls\"
      version = \">= 4.0.0\"
    }
  }
}

# Configure AWS provider for us-east-1
provider \"aws\" {
  region = \"us-east-1\"
}

# Generate self-signed TLS cert for Vault (use ACM in prod)
resource \"tls_private_key\" \"vault\" {
  algorithm = \"RSA\"
  rsa_bits  = 4096
}

resource \"tls_self_signed_cert\" \"vault\" {
  private_key_pem = tls_private_key.vault.private_key_pem

  subject {
    common_name  = \"vault.example.com\"
    organization = \"Example Corp\"
  }

  validity_period_hours = 8760  # 1 year
  early_renewal_hours   = 720   # Renew 30 days early

  allowed_uses = [
    \"key_encipherment\",
    \"digital_signature\",
    \"server_auth\",
  ]
}

# Store Vault TLS cert in AWS Secrets Manager
resource \"aws_secretsmanager_secret\" \"vault_tls\" {
  name        = \"vault-tls-cert\"
  description = \"TLS certificate for Vault cluster\"
}

resource \"aws_secretsmanager_secret_version\" \"vault_tls\" {
  secret_id     = aws_secretsmanager_secret.vault_tls.id
  secret_string = jsonencode({
    cert = tls_self_signed_cert.vault.cert_pem
    key  = tls_private_key.vault.private_key_pem
  })
}

# Deploy 3-node Vault cluster on EC2 (dev-only, use ECS/EKS in prod)
data \"aws_ami\" \"ubuntu\" {
  most_recent = true
  filter {
    name   = \"name\"
    values = [\"ubuntu/images/hvm-ssd/ubuntu-jammy-22.04-amd64-server-*\"]
  }
  filter {
    name   = \"virtualization-type\"
    values = [\"hvm\"]
  }
  owners = [\"099720109477\"]  # Canonical
}

resource \"aws_instance\" \"vault\" {
  count                  = 3
  ami                    = data.aws_ami.ubuntu.id
  instance_type          = \"t3.medium\"
  subnet_id              = aws_subnet.vault.id
  vpc_security_group_ids = [aws_security_group.vault.id]
  iam_instance_profile   = aws_iam_instance_profile.vault.name

  user_data = <<-EOF
    #!/bin/bash
    set -euo pipefail  # Exit on error, undefined vars

    # Install Vault 2.0.3
    curl -fsSL https://apt.releases.hashicorp.com/gpg | sudo apt-key add -
    sudo apt-add-repository \"deb [arch=amd64] https://apt.releases.hashicorp.com $(lsb_release -cs) main\"
    sudo apt update && sudo apt install vault=2.0.3 -y

    # Configure Vault
    sudo mkdir -p /etc/vault
    sudo tee /etc/vault/config.hcl <<-VAULTCONFIG
      ui = true
      cluster_addr = \"https://${self.private_ip}:8201\"
      api_addr = \"https://${self.private_ip}:8200\"

      listener \"tcp\" {
        address = \"0.0.0.0:8200\"
        tls_cert_file = \"/etc/vault/tls.crt\"
        tls_key_file = \"/etc/vault/tls.key\"
      }

      storage \"raft\" {
        path = \"/var/lib/vault\"
        node_id = \"vault-node-${count.index}\"
        retry_join {
          leader_api_addr = \"https://${aws_instance.vault[0].private_ip}:8200\"
        }
      }
    VAULTCONFIG

    # Copy TLS certs from Secrets Manager
    aws secretsmanager get-secret-value --secret-id vault-tls-cert --query SecretString --output text | \
      jq -r .cert | sudo tee /etc/vault/tls.crt
    aws secretsmanager get-secret-value --secret-id vault-tls-cert --query SecretString --output text | \
      jq -r .key | sudo tee /etc/vault/tls.key
    sudo chown vault:vault /etc/vault/tls.*

    # Start Vault
    sudo systemctl enable vault
    sudo systemctl start vault
  EOF

  tags = {
    Name = \"vault-node-${count.index}\"
    Role = \"vault-server\"
  }
}

# Initialize Vault and configure multi-cloud auth methods
provider \"vault\" {
  address = \"https://${aws_instance.vault[0].public_ip}:8200\"
  token   = var.vault_root_token  # Set via terraform.tfvars after init
}

# Enable AWS auth method
resource \"vault_aws_auth_backend\" \"aws\" {
  path = \"aws\"
  description = \"AWS IAM auth for EC2/EKS workloads\"
}

# Configure AWS auth role for production apps
resource \"vault_aws_auth_backend_role\" \"prod_apps\" {
  backend                         = vault_aws_auth_backend.aws.path
  role                            = \"prod-apps\"
  auth_type                       = \"iam\"
  bound_aws_account_ids           = [var.aws_account_id]
  bound_iam_role_arns             = [\"arn:aws:iam::${var.aws_account_id}:role/prod-app-role\"]
  token_ttl                       = 3600
  token_max_ttl                   = 86400
  token_policies                  = [\"prod-app-policy\"]
}

# Enable GCP auth method for GCP workloads
resource \"vault_gcp_auth_backend\" \"gcp\" {
  path = \"gcp\"
  description = \"GCP IAM auth for GCE/GKE workloads\"
}

# Enable Azure auth method for Azure workloads
resource \"vault_azure_auth_backend\" \"azure\" {
  path = \"azure\"
  description = \"Azure AD auth for VM/aks workloads\"
  tenant_id   = var.azure_tenant_id
  client_id   = var.azure_client_id
  client_secret = var.azure_client_secret
}

# Create Vault policy for production apps
resource \"vault_policy\" \"prod_app\" {
  name = \"prod-app-policy\"
  policy = <<-EOF
    path \"aws/sts/prod-apps\" {
      capabilities = [\"read\"]
    }
    path \"gcp/sts/prod-apps\" {
      capabilities = [\"read\"]
    }
    path \"azure/sts/prod-apps\" {
      capabilities = [\"read\"]
    }
  EOF
}

variable \"vault_root_token\" {
  type        = string
  sensitive   = true
  description = \"Vault root token generated after cluster init\"
}

variable \"aws_account_id\" {
  type = string
  default = \"123456789012\"
}

variable \"azure_tenant_id\" {
  type = string
  sensitive = true
}

variable \"azure_client_id\" {
  type = string
  sensitive = true
}

variable \"azure_client_secret\" {
  type = string
  sensitive = true
}

package main

import (
    \"context\"
    \"encoding/json\"
    \"fmt\"
    \"log\"
    \"os\"
    \"strings\"
    \"time\"

    vault \"github.com/hashicorp/vault/api\"
    \"github.com/aws/aws-sdk-go/aws\"
    \"github.com/aws/aws-sdk-go/aws/session\"
    \"github.com/aws/aws-sdk-go/service/iam\"
)

// StaticCredential represents a legacy static IAM credential to migrate
type StaticCredential struct {
    AccessKeyID string `json:\"access_key_id\"`
    SecretKey   string `json:\"secret_key\"`
    RoleARN     string `json:\"role_arn\"`
    Owner       string `json:\"owner\"`
    LastUsed    time.Time `json:\"last_used\"`
}

// MigrationResult tracks success/failure of credential migration
type MigrationResult struct {
    Credential StaticCredential
    Success    bool
    Error      string
    VaultPath  string
}

func main() {
    // Load configuration from environment
    vaultAddr := os.Getenv(\"VAULT_ADDR\")
    if vaultAddr == \"\" {
        log.Fatal(\"VAULT_ADDR environment variable is required\")
    }
    vaultToken := os.Getenv(\"VAULT_TOKEN\")
    if vaultToken == \"\" {
        log.Fatal(\"VAULT_TOKEN environment variable is required\")
    }
    awsRegion := os.Getenv(\"AWS_REGION\")
    if awsRegion == \"\" {
        awsRegion = \"us-east-1\"
    }

    // Initialize Vault client
    vaultClient, err := vault.NewClient(&vault.Config{
        Address: vaultAddr,
    })
    if err != nil {
        log.Fatalf(\"Failed to create Vault client: %v\", err)
    }
    vaultClient.SetToken(vaultToken)

    // Validate Vault connectivity
    _, err = vaultClient.Sys().Health()
    if err != nil {
        log.Fatalf(\"Vault health check failed: %v\", err)
    }
    log.Println(\"Connected to Vault successfully\")

    // Initialize AWS IAM client
    sess, err := session.NewSession(&aws.Config{
        Region: aws.String(awsRegion),
    })
    if err != nil {
        log.Fatalf(\"Failed to create AWS session: %v\", err)
    }
    iamClient := iam.New(sess)

    // Load static credentials from JSON file (output of legacy IAM audit)
    credsFile := \"static_creds.json\"
    creds, err := loadStaticCredentials(credsFile)
    if err != nil {
        log.Fatalf(\"Failed to load static credentials from %s: %v\", credsFile, err)
    }
    log.Printf(\"Loaded %d static credentials to migrate\", len(creds))

    // Process each credential
    results := make([]MigrationResult, 0, len(creds))
    for _, cred := range creds {
        result := migrateCredential(context.Background(), vaultClient, iamClient, cred)
        results = append(results, result)

        // Rate limit to avoid Vault/AWS throttling
        time.Sleep(500 * time.Millisecond)
    }

    // Write migration results to audit file
    outputFile := \"migration_results.json\"
    output, err := json.MarshalIndent(results, \"\", \"  \")
    if err != nil {
        log.Fatalf(\"Failed to marshal migration results: %v\", err)
    }
    if err := os.WriteFile(outputFile, output, 0644); err != nil {
        log.Fatalf(\"Failed to write migration results to %s: %v\", outputFile, err)
    }
    log.Printf(\"Migration complete. Results written to %s\", outputFile)
}

// loadStaticCredentials reads static IAM credentials from a JSON file
func loadStaticCredentials(path string) ([]StaticCredential, error) {
    data, err := os.ReadFile(path)
    if err != nil {
        return nil, fmt.Errorf(\"read file: %w\", err)
    }

    var creds []StaticCredential
    if err := json.Unmarshal(data, &creds); err != nil {
        return nil, fmt.Errorf(\"unmarshal JSON: %w\", err)
    }
    return creds, nil
}

// migrateCredential migrates a single static IAM credential to Vault dynamic secrets
func migrateCredential(ctx context.Context, vaultClient *vault.Client, iamClient *iam.IAM, cred StaticCredential) MigrationResult {
    result := MigrationResult{
        Credential: cred,
        Success:    false,
    }

    // Step 1: Validate static credential is still active
    _, err := iamClient.GetAccessKey(&iam.GetAccessKeyInput{
        AccessKeyId: aws.String(cred.AccessKeyID),
        UserName:    aws.String(cred.Owner),
    })
    if err != nil {
        result.Error = fmt.Sprintf(\"static credential no longer active: %v\", err)
        return result
    }

    // Step 2: Create Vault AWS role for the credential's IAM role
    vaultRole := fmt.Sprintf(\"migrated-%s\", cred.RoleARN[strings.LastIndex(cred.RoleARN, \"/\")+1:])
    vaultPath := fmt.Sprintf(\"aws/sts/%s\", vaultRole)

    // Write Vault role configuration
    _, err = vaultClient.Logical().Write(\"aws/roles/\"+vaultRole, map[string]interface{}{
        \"credential_type\": \"assumed_role\",
        \"role_arns\":       []string{cred.RoleARN},
        \"default_sts_ttl\": 3600,
        \"max_sts_ttl\":     86400,
    })
    if err != nil {
        result.Error = fmt.Sprintf(\"failed to create Vault role: %v\", err)
        return result
    }

    // Step 3: Deactivate static IAM credential (never delete immediately!)
    _, err = iamClient.UpdateAccessKey(&iam.UpdateAccessKeyInput{
        AccessKeyId: aws.String(cred.AccessKeyID),
        Status:      aws.String(\"Inactive\"),
        UserName:    aws.String(cred.Owner),
    })
    if err != nil {
        result.Error = fmt.Sprintf(\"failed to deactivate static credential: %v\", err)
        return result
    }

    // Step 4: Validate dynamic secret works
    secret, err := vaultClient.Logical().Read(vaultPath)
    if err != nil {
        result.Error = fmt.Sprintf(\"failed to read dynamic secret from Vault: %v\", err)
        return result
    }
    if secret == nil || secret.Data == nil {
        result.Error = \"dynamic secret returned empty data\"
        return result
    }

    result.Success = true
    result.VaultPath = vaultPath
    log.Printf(\"Successfully migrated credential %s to Vault path %s\", cred.AccessKeyID, vaultPath)
    return result
}

# Short snippet: Retrieve dynamic GCP creds via Vault
import hvac
client = hvac.Client(url=\"https://vault.example.com:8200\", token=\"your-token\")
creds = client.secrets.gcp.generate_credentials(name=\"prod-apps\", mount_point=\"gcp\")
print(creds[\"data\"][\"token\"])

# Terraform snippet: Vault OIDC auth group mapping
resource \"vault_identity_group\" \"prod_multi_cloud\" {
  name = \"prod-multi-cloud\"
  policies = [\"aws-prod\", \"gcp-prod\", \"azure-prod\"]
}
resource \"vault_oidc_auth_backend_group\" \"prod\" {
  backend = vault_oidc_auth_backend.okta.path
  group_name = \"prod-multi-cloud\"
  vault_group_id = vault_identity_group.prod_multi_cloud.id
}

# k6 snippet: Load test Vault dynamic AWS secret endpoint
import http from 'k6/http';
import { check } from 'k6';

export default function () {
  const res = http.get('https://vault.example.com:8200/v1/aws/sts/prod-apps', {
    headers: { 'X-Vault-Token': 'your-token' },
  });
  check(res, { 'status is 200': (r) => r.status === 200 });
}