War Story: We Replaced AWS IAM with Vault 1.16 and Cut Our Permission Error Rate by 60% for 500+ Developers

# Terraform configuration for HashiCorp Vault 1.16 AWS Auth Backend with EKS Workload Identity
# Provider versions pinned to ensure reproducibility
terraform {
  required_version = ">= 1.6.0"
  required_providers {
    vault = {
      source  = "hashicorp/vault"
      version = "~> 3.18.0" # Vault 1.16 compatible provider
    }
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.20.0"
    }
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.23.0"
    }
  }
}

# Configure Vault provider using local token (for CI/CD, use approle)
provider "vault" {
  address = "https://vault.internal.example.com:8200"
  token   = var.vault_root_token # In prod, use approle auth, not root token
}

# Configure AWS provider for IAM role creation
provider "aws" {
  region = var.aws_region
  default_tags {
    tags = {
      ManagedBy = "terraform"
      Project   = "vault-iam-migration"
    }
  }
}

# Create IAM role for Vault to assume when validating EKS service accounts
resource "aws_iam_role" "vault_aws_auth" {
  name = "vault-aws-auth-role-${var.env}"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Principal = {
          Service = "ec2.amazonaws.com" # Vault runs on EC2, so it assumes this role
        }
        Action = "sts:AssumeRole"
        Condition = {
          StringEquals = {
            "aws:RequestedRegion" = var.aws_region
          }
        }
      }
    ]
  })

  tags = {
    Purpose = "Allow Vault to validate EKS workload identities"
  }
}

# Attach policy to Vault IAM role for reading EKS service account issuer metadata
resource "aws_iam_role_policy" "vault_eks_read" {
  name = "vault-eks-read-policy-${var.env}"
  role = aws_iam_role.vault_aws_auth.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "eks:DescribeCluster",
          "iam:GetOpenIDConnectProvider",
          "iam:ListOpenIDConnectProviders"
        ]
        Resource = "*" # Restrict to specific clusters in prod
      }
    ]
  })
}

# Configure Vault AWS auth backend for EKS workload identity
resource "vault_aws_auth_backend" "eks" {
  path = "aws-eks-${var.env}"
  type = "aws"

  # Enable workload identity federation for EKS
  identity_token_audience = "vault.io"
  identity_token_ttl     = 600 # 10 minutes, matches EKS service account token TTL

  # Error handling: fail if backend already exists with different config
  lifecycle {
    prevent_destroy = false # Allow destruction in non-prod envs
  }
}

# Configure EKS OIDC provider as a trusted identity source in Vault
resource "vault_aws_auth_backend_config_identity" "eks_oidc" {
  backend = vault_aws_auth_backend.eks.path
  oidc_discovery_url = "https://oidc.eks.${var.aws_region}.amazonaws.com/id/${var.eks_cluster_id}"
  oidc_discovery_ca_pem = var.eks_oidc_ca_pem

  # Validate OIDC issuer on every login to prevent forgery
  disable_remount = true
}

# Create Vault role mapping EKS service accounts to AWS IAM permissions
resource "vault_aws_auth_backend_role" "developer_deploy" {
  backend = vault_aws_auth_backend.eks.path
  role    = "developer-deploy-${var.env}"

  # Bound EKS service account: only allow sa "deploy-sa" in "default" namespace
  bound_service_account_names      = ["deploy-sa"]
  bound_service_account_namespaces = ["default", "staging", "prod"]

  # AWS IAM permissions to grant: limited to deploy S3 buckets and ECR push
  aws_role_arns = [aws_iam_role.eks_deploy_role.arn]
  ttl           = 900 # 15 minute credential TTL
  max_ttl       = 3600 # 1 hour max

  # Error handling: require explicit service account annotation
  bound_claims = {
    "eks.amazonaws.com/role-arn" = aws_iam_role.eks_deploy_role.arn
  }
}

# IAM role that Vault will assume to grant to EKS service accounts
resource "aws_iam_role" "eks_deploy_role" {
  name = "eks-deploy-role-${var.env}"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Principal = {
          AWS = aws_iam_role.vault_aws_auth.arn # Only Vault can assume this role
        }
        Action = "sts:AssumeRole"
        Condition = {
          StringEquals = {
            "aws:RequestedRegion" = var.aws_region
          }
        }
      }
    ]
  })
}

# Policy for EKS deploy role: minimal permissions for CI/CD
resource "aws_iam_role_policy" "eks_deploy_policy" {
  name = "eks-deploy-policy-${var.env}"
  role = aws_iam_role.eks_deploy_role.id

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:PutObject",
          "s3:GetObject",
          "ecr:PutImage",
          "ecr:InitiateLayerUpload"
        ]
        Resource = [
          "arn:aws:s3:::example-deploy-bucket/*",
          "arn:aws:ecr:${var.aws_region}:${var.aws_account_id}:repository/example-app/*"
        ]
      }
    ]
  })
}

# Variable definitions
variable "vault_root_token" {
  type        = string
  sensitive   = true
  description = "Vault root token (use approle in prod)"
}

variable "aws_region" {
  type    = string
  default = "us-east-1"
}

variable "env" {
  type    = string
  default = "staging"
}

variable "eks_cluster_id" {
  type = string
}

variable "eks_oidc_ca_pem" {
  type = string
}

variable "aws_account_id" {
  type = string
}

// vault-aws-creds.go: Fetch dynamic AWS IAM credentials from Vault 1.16 for EKS workloads
// Build: go build -o vault-aws-creds vault-aws-creds.go
// Requires: Vault 1.16+ with AWS auth backend configured, EKS service account with Vault annotation
package main

import (
    "encoding/json"
    "fmt"
    "log"
    "os"
    "time"

    vault "github.com/hashicorp/vault/api"
)

// CredentialResponse represents the dynamic AWS credentials returned by Vault
type CredentialResponse struct {
    AccessKeyID     string    `json:"access_key"`
    SecretAccessKey string    `json:"secret_key"`
    SessionToken    string    `json:"security_token"`
    LeaseID         string    `json:"lease_id"`
    LeaseDuration   int       `json:"lease_duration"`
    Renewable       bool      `json:"renewable"`
}

func main() {
    // Initialize Vault client with default config (reads VAULT_ADDR, VAULT_TOKEN from env)
    config := vault.DefaultConfig()
    client, err := vault.NewClient(config)
    if err != nil {
        log.Fatalf("Failed to initialize Vault client: %v", err)
    }

    // Validate Vault connection by checking seal status
    sealStatus, err := client.Sys().SealStatus()
    if err != nil {
        log.Fatalf("Failed to connect to Vault: %v", err)
    }
    if sealStatus.Sealed {
        log.Fatal("Vault is sealed, cannot fetch credentials")
    }

    // Get EKS workload identity token from projected service account volume
    // EKS mounts the OIDC token at this path by default
    tokenPath := "/var/run/secrets/eks.amazonaws.com/serviceaccount/token"
    jwtToken, err := os.ReadFile(tokenPath)
    if err != nil {
        log.Fatalf("Failed to read EKS service account token: %v", err)
    }

    // Login to Vault using AWS auth backend with EKS workload identity
    // Path matches the aws-eks-staging backend we configured in Terraform
    loginPath := "auth/aws-eks-staging/login"
    loginData := map[string]interface{}{
        "role":                "developer-deploy-staging", // Matches Vault role name
        "jwt":                 string(jwtToken),
        "service_account_name": "deploy-sa",
        "service_account_namespace": "staging",
    }

    loginSecret, err := client.Auth().Login(loginPath, loginData)
    if err != nil {
        log.Fatalf("Vault login failed: %v", err)
    }
    if loginSecret == nil || loginSecret.Auth == nil {
        log.Fatal("Vault login returned empty auth response")
    }

    // Set the client token to the login token for subsequent requests
    client.SetToken(loginSecret.Auth.ClientToken)
    fmt.Printf("Successfully logged in to Vault, token TTL: %d seconds\n", loginSecret.Auth.TTL)

    // Fetch dynamic AWS IAM credentials from Vault
    // Path: aws-eks-staging/sts/developer-deploy-staging (matches role name)
    credPath := "aws-eks-staging/sts/developer-deploy-staging"
    credSecret, err := client.Logical().Read(credPath)
    if err != nil {
        log.Fatalf("Failed to fetch AWS credentials: %v", err)
    }
    if credSecret == nil || credSecret.Data == nil {
        log.Fatal("Vault returned empty credential response")
    }

    // Parse credential response
    var creds CredentialResponse
    credBytes, err := json.Marshal(credSecret.Data)
    if err != nil {
        log.Fatalf("Failed to marshal credential data: %v", err)
    }
    if err := json.Unmarshal(credBytes, &creds); err != nil {
        log.Fatalf("Failed to unmarshal credentials: %v", err)
    }

    // Validate credential TTL is within expected bounds (15 minutes max)
    if creds.LeaseDuration > 900 {
        log.Fatalf("Credential TTL %d exceeds maximum allowed 900 seconds", creds.LeaseDuration)
    }

    // Print credentials (in prod, inject into environment or AWS SDK config)
    fmt.Printf("Fetched AWS Credentials:\n")
    fmt.Printf("Access Key ID: %s\n", creds.AccessKeyID)
    fmt.Printf("Secret Access Key: %s\n", "***REDACTED***") // Never log secrets
    fmt.Printf("Session Token: %s\n", "***REDACTED***")
    fmt.Printf("Lease ID: %s\n", creds.LeaseID)
    fmt.Printf("Lease Duration: %d seconds\n", creds.LeaseDuration)
    fmt.Printf("Renewable: %v\n", creds.Renewable)

    // Renew lease if renewable (optional, for long-running workloads)
    if creds.Renewable {
        renewData := map[string]interface{}{
            "lease_id": creds.LeaseID,
        }
        renewSecret, err := client.Sys().RenewLease(creds.LeaseID, creds.LeaseDuration)
        if err != nil {
            log.Printf("Warning: Failed to renew lease: %v", err)
        } else {
            fmt.Printf("Renewed lease, new TTL: %d seconds\n", renewSecret.Auth.TTL)
        }
    }

    // Clean up: revoke lease when done (for short-lived workloads)
    defer func() {
        if err := client.Sys().Revoke(creds.LeaseID); err != nil {
            log.Printf("Warning: Failed to revoke lease: %v", err)
        } else {
            fmt.Println("Successfully revoked credential lease")
        }
    }()

    // Simulate using credentials for an AWS SDK call
    // In real code, inject these into aws.Config
    fmt.Println("Credentials ready for AWS SDK use")
    time.Sleep(10 * time.Second) // Simulate workload execution
}

# audit_permission_errors.py: Analyze Datadog audit logs to compare pre/post Vault migration error rates
# Requires: datadog-api-client-v2, pandas, matplotlib
# Usage: python audit_permission_errors.py --start 2023-07-01 --end 2023-12-31 --output report.png
import argparse
import json
import sys
from datetime import datetime, timedelta
from typing import Dict, List

from datadog_api_client import ApiClient, Configuration
from datadog_api_client.v2.api.logs_api import LogsApi
from datadog_api_client.v2.model.logs_list_request import LogsListRequest
from datadog_api_client.v2.model.logs_list_request_page import LogsListRequestPage
from datadog_api_client.v2.model.logs_query_filter import LogsQueryFilter
from datadog_api_client.v2.model.logs_sort import LogsSort
import pandas as pd
import matplotlib.pyplot as plt

# Datadog query for AWS AccessDenied errors
ACCESS_DENIED_QUERY = """
aws.iam.error_code:AccessDenied AND service:ecs OR service:eks
"""

def fetch_logs(start: datetime, end: datetime, query: str) -> List[Dict]:
    """Fetch AccessDenied logs from Datadog between start and end dates."""
    configuration = Configuration()
    configuration.api_key["apiKeyAuth"] = os.environ.get("DD_API_KEY")
    configuration.application_key["appKeyAuth"] = os.environ.get("DD_APP_KEY")

    if not configuration.api_key["apiKeyAuth"] or not configuration.application_key["appKeyAuth"]:
        raise ValueError("DD_API_KEY and DD_APP_KEY environment variables must be set")

    with ApiClient(configuration) as api_client:
        api_instance = LogsApi(api_client)
        logs = []
        page = LogsListRequestPage(limit=1000)
        filter_params = LogsQueryFilter(
            query=query,
            from_=int(start.timestamp() * 1000),
            to_=int(end.timestamp() * 1000),
        )
        request = LogsListRequest(
            filter=filter_params,
            sort=LogsSort("timestamp:asc"),
            page=page,
        )

        # Paginate through all logs (Datadog returns max 1000 per request)
        while True:
            response = api_instance.list_logs(request)
            if not response.data:
                break
            logs.extend([log.to_dict() for log in response.data])
            if len(response.data) < 1000:
                break
            # Update page cursor for next request
            request.page.cursor = response.meta.page.cursor

        return logs

def calculate_error_rate(logs: List[Dict], total_deployments: int) -> float:
    """Calculate permission error rate as percentage of total deployments."""
    if total_deployments == 0:
        raise ValueError("Total deployments cannot be zero")
    error_count = len(logs)
    return (error_count / total_deployments) * 100

def generate_report(pre_logs: List[Dict], post_logs: List[Dict], output_path: str):
    """Generate bar chart comparing pre and post migration error rates."""
    # Get total deployments from our CI/CD system (hardcoded for example, pull from API in prod)
    pre_total_deployments = 12400  # Q2 2023: 3 months of deployments
    post_total_deployments = 13100 # Q4 2023: 3 months of deployments

    pre_error_rate = calculate_error_rate(pre_logs, pre_total_deployments)
    post_error_rate = calculate_error_rate(post_logs, post_total_deployments)

    # Plot results
    labels = ["Pre-Vault (AWS IAM Only)", "Post-Vault (1.16 Migration)"]
    rates = [pre_error_rate, post_error_rate]
    colors = ["#ff6b6b", "#51cf66"]

    fig, ax = plt.subplots(figsize=(10, 6))
    bars = ax.bar(labels, rates, color=colors)
    ax.set_ylabel("Permission Error Rate (%)")
    ax.set_title("AWS IAM Permission Error Rate: Pre vs Post Vault 1.16 Migration")
    ax.set_ylim(0, max(rates) * 1.2)

    # Add value labels on top of bars
    for bar in bars:
        height = bar.get_height()
        ax.text(
            bar.get_x() + bar.get_width() / 2,
            height + 0.5,
            f"{height:.2f}%",
            ha="center",
            va="bottom",
        )

    # Add reduction annotation
    reduction = ((pre_error_rate - post_error_rate) / pre_error_rate) * 100
    ax.text(
        0.5,
        0.9,
        f"Reduction: {reduction:.1f}%",
        transform=ax.transAxes,
        ha="center",
        fontsize=12,
        bbox=dict(boxstyle="round", facecolor="wheat", alpha=0.5),
    )

    plt.tight_layout()
    plt.savefig(output_path)
    print(f"Report saved to {output_path}")
    print(f"Pre-migration error rate: {pre_error_rate:.2f}%")
    print(f"Post-migration error rate: {post_error_rate:.2f}%")
    print(f"Reduction: {reduction:.1f}%")

def main():
    parser = argparse.ArgumentParser(description="Audit AWS IAM permission errors")
    parser.add_argument("--start", required=True, help="Start date (YYYY-MM-DD)")
    parser.add_argument("--end", required=True, help="End date (YYYY-MM-DD)")
    parser.add_argument("--output", default="error_rate_report.png", help="Output report path")
    args = parser.parse_args()

    try:
        start_date = datetime.strptime(args.start, "%Y-%m-%d")
        end_date = datetime.strptime(args.end, "%Y-%m-%d")
    except ValueError as e:
        print(f"Invalid date format: {e}")
        sys.exit(1)

    # Split date range into pre-migration (before 2023-10-01) and post-migration (after 2023-10-01)
    migration_date = datetime(2023, 10, 1)
    pre_start = start_date
    pre_end = min(end_date, migration_date - timedelta(days=1))
    post_start = max(start_date, migration_date)
    post_end = end_date

    print(f"Fetching pre-migration logs from {pre_start} to {pre_end}...")
    pre_logs = fetch_logs(pre_start, pre_end, ACCESS_DENIED_QUERY)
    print(f"Fetched {len(pre_logs)} pre-migration AccessDenied logs")

    print(f"Fetching post-migration logs from {post_start} to {post_end}...")
    post_logs = fetch_logs(post_start, post_end, ACCESS_DENIED_QUERY)
    print(f"Fetched {len(post_logs)} post-migration AccessDenied logs")

    generate_report(pre_logs, post_logs, args.output)

if __name__ == "__main__":
    main()

# Shadow mode sidecar for EKS pods
apiVersion: v1
kind: Pod
spec:
  containers:
  - name: app
    image: example-app:latest
  - name: vault-shadow
    image: vault:1.16.0
    command: ["sh", "-c"]
    args:
    - |
      while true; do
        # Log IAM request (simulated, in real code use AWS SDK)
        echo "IAM_ALLOW: $(date) service=deploy sa=deploy-sa" >> /var/log/shadow.log
        # Log Vault request
        vault login -method=aws role=developer-deploy-staging > /dev/null 2>&1
        if [ $? -eq 0 ]; then
          echo "VAULT_ALLOW: $(date) service=deploy sa=deploy-sa" >> /var/log/shadow.log
        else
          echo "VAULT_DENY: $(date) service=deploy sa=deploy-sa" >> /var/log/shadow.log
        fi
        sleep 60
      done
    volumeMounts:
    - name: shadow-logs
      mountPath: /var/log
  volumes:
  - name: shadow-logs
    emptyDir: {}

# Cron job that renews Vault leases for long-running workloads
apiVersion: batch/v1
kind: CronJob
metadata:
  name: vault-lease-renewal
spec:
  schedule: "*/10 * * * *" # Run every 10 minutes
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: vault-renew-sa
          containers:
          - name: lease-renew
            image: vault:1.16.0
            command: ["sh", "-c"]
            args:
            - |
              # Get all active leases for this service account
              leases=$(vault list sys/leases/lookup/aws-eks-staging/sts/developer-deploy-staging)
              for lease in $leases; do
                vault lease renew $lease > /dev/null 2>&1
                if [ $? -eq 0 ]; then
                  echo "Renewed lease $lease"
                else
                  echo "Failed to renew lease $lease"
                fi
              done
          restartPolicy: OnFailure

# IAM role cleanup script
import boto3
from hvac import Client as VaultClient

vault_client = VaultClient(url="https://vault.internal.example.com:8200", token="vault-token")
iam_client = boto3.client("iam")

# Get all Vault identity entities
vault_entities = vault_client.identity.entity.list()["data"]["keys"]

# Get all AWS IAM roles
iam_roles = [role["RoleName"] for role in iam_client.list_roles()["Roles"]]

# Find IAM roles with no corresponding Vault entity
orphaned_roles = [role for role in iam_roles if role not in vault_entities]

for role in orphaned_roles:
    # Check if role is in grace period
    tags = iam_client.list_role_tags(RoleName=role)["Tags"]
    grace_period = [tag for tag in tags if tag["Key"] == "GracePeriodEnd"]
    if grace_period:
        end_date = datetime.strptime(grace_period[0]["Value"], "%Y-%m-%d")
        if datetime.now() > end_date:
            iam_client.delete_role(RoleName=role)
            print(f"Deleted orphaned role: {role}")
    else:
        # Add grace period tag
        iam_client.tag_role(
            RoleName=role,
            Tags=[{"Key": "GracePeriodEnd", "Value": (datetime.now() + timedelta(days=7)).strftime("%Y-%m-%d")}]
        )
        print(f"Added grace period to role: {role}")