Step-by-Step: Set Up Zero-Trust Security with HashiCorp Vault 1.16 and AWS IAM Identity Center 2026-02

# Provider configuration for AWS and Vault
tterraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.36" # Matches AWS provider support for IAM Identity Center 2026-02
    }
    vault = {
      source  = "hashicorp/vault"
      version = "~> 4.0" # Supports Vault 1.16 OIDC workload identity plugin
    }
  }
}

# Configure AWS provider with region matching IAM Identity Center deployment
provider "aws" {
  region = "us-east-1"
  # Assume role for deployment if using temporary credentials
  assume_role {
    role_arn = var.deployment_role_arn != "" ? var.deployment_role_arn : null
  }
}

# Configure Vault provider with root token or approle
provider "vault" {
  address = var.vault_addr
  token   = var.vault_root_token
  # Add TLS config for production Vault deployments
  ca_cert_file = var.vault_ca_cert != "" ? var.vault_ca_cert : null
}

# Enable AWS IAM Identity Center 2026-02 with ABAC support
resource "aws_ssoadmin_instance" "main" {
  name = "vault-zero-trust-sso-instance"
  # 2026-02 release adds native ABAC for attribute mapping
  tags = {
    Purpose = "Zero-Trust Vault Federation"
    Version = "2026-02"
  }
}

# Create OIDC trust provider for Vault 1.16
resource "aws_iam_openid_connect_provider" "vault" {
  url             = var.vault_oidc_issuer_url
  client_id_list  = ["vault.workload.identity"] # Matches Vault 1.16 plugin client ID
  thumbprint_list = [var.vault_oidc_thumbprint] # Get from Vault issuer TLS cert

  tags = {
    Name = "Vault-1.16-OIDC-Provider"
  }
}

# Create IAM role for Vault to assume when verifying identities
resource "aws_iam_role" "vault_oidc_role" {
  name = "Vault-OIDC-Verification-Role"

  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRoleWithWebIdentity"
        Effect = "Allow"
        Principal = {
          Federated = aws_iam_openid_connect_provider.vault.arn
        }
        Condition = {
          StringEquals = {
            "${aws_iam_openid_connect_provider.vault.url}:sub" = "system:serviceaccount:vault:workload-identity"
          }
        }
      }
    ]
  })

  tags = {
    Purpose = "Vault OIDC Identity Verification"
  }
}

# Attach read-only policy for IAM Identity Center attribute lookup
resource "aws_iam_role_policy_attachment" "vault_sso_read" {
  role       = aws_iam_role.vault_oidc_role.name
  policy_arn = "arn:aws:iam::aws:policy/AWSReadOnlyAccess" # Restrict to SSO-specific actions in prod
}

# Enable Vault OIDC auth method with IAM Identity Center config
resource "vault_jwt_auth_backend" "iam_identity_center" {
  path               = "oidc-iam-identity-center"
  type               = "oidc"
  oidc_discovery_url = "https://identitycenter.amazonaws.com/${aws_ssoadmin_instance.main.id}"
  oidc_client_id     = var.vault_oidc_client_id
  oidc_client_secret = var.vault_oidc_client_secret
  bound_issuer       = "https://identitycenter.amazonaws.com/${aws_ssoadmin_instance.main.id}"
  # Vault 1.16 specific: enable workload identity plugin
  plugin_name        = "vault-oidc-workload-identity-plugin"
  plugin_version     = "1.16.0"

  tune {
    max_ttl     = "1h"
    default_ttl = "15m"
  }
}

# Variable definitions
variable "deployment_role_arn" {
  type        = string
  description = "ARN of IAM role to assume for deployment"
  default     = ""
}

variable "vault_addr" {
  type        = string
  description = "Address of the Vault 1.16 instance"

  validation {
    condition     = can(regex("^https?://", var.vault_addr))
    error_message = "Vault address must start with http:// or https://."
  }
}

variable "vault_root_token" {
  type        = string
  description = "Root token for Vault initial configuration"
  sensitive   = true
}

variable "vault_ca_cert" {
  type        = string
  description = "Path to Vault CA cert for TLS validation"
  default     = ""
}

variable "vault_oidc_issuer_url" {
  type        = string
  description = "OIDC issuer URL for Vault (e.g., https://vault.example.com:8200)"
}

variable "vault_oidc_thumbprint" {
  type        = string
  description = "TLS thumbprint of Vault OIDC issuer"
}

variable "vault_oidc_client_id" {
  type        = string
  description = "OIDC client ID registered in IAM Identity Center"
}

variable "vault_oidc_client_secret" {
  type        = string
  description = "OIDC client secret from IAM Identity Center"
  sensitive   = true
}

package main

import (
    "context"
    "crypto/tls"
    "fmt"
    "log"
    "os"
    "strings"
    "time"

    "github.com/hashicorp/vault/api" // Vault 1.16 client library
    "github.com/aws/aws-sdk-go-v2/aws"
    "github.com/aws/aws-sdk-go-v2/config"
    "github.com/aws/aws-sdk-go-v2/service/ssoadmin"
    "github.com/golang-jwt/jwt/v5" // JWT validation for OIDC tokens
)

// VaultOIDCValidator validates OIDC tokens issued by IAM Identity Center via Vault 1.16
type VaultOIDCValidator struct {
    vaultClient  *api.Client
    ssoClient    *ssoadmin.Client
    issuerURL    string
    expectedAud  string
}

// NewValidator initializes a new OIDC validator with Vault and AWS SSO clients
func NewValidator(vaultAddr, vaultToken, issuerURL, expectedAud string) (*VaultOIDCValidator, error) {
    // Configure Vault client with TLS and error handling
    vaultConfig := api.DefaultConfig()
    vaultConfig.Address = vaultAddr
    // Skip TLS verify only for testing; enforce in production
    if os.Getenv("VAULT_SKIP_VERIFY") == "true" {
        vaultConfig.HttpClient.Transport = &http.Transport{
            TLSClientConfig: &tls.Config{InsecureSkipVerify: true},
        }
    }
    vaultClient, err := api.NewClient(vaultConfig)
    if err != nil {
        return nil, fmt.Errorf("failed to create Vault client: %w", err)
    }
    vaultClient.SetToken(vaultToken)

    // Configure AWS SSO client for IAM Identity Center 2026-02
    awsCfg, err := config.LoadDefaultConfig(context.TODO(),
        config.WithRegion("us-east-1"),
        config.WithSharedConfigProfile("vault-deploy"),
    )
    if err != nil {
        return nil, fmt.Errorf("failed to load AWS config: %w", err)
    }
    ssoClient := ssoadmin.NewFromConfig(awsCfg)

    return &VaultOIDCValidator{
        vaultClient: vaultClient,
        ssoClient:   ssoClient,
        issuerURL:   issuerURL,
        expectedAud: expectedAud,
    }, nil
}

// ValidateToken checks if an OIDC token is valid and issued by trusted IAM Identity Center
func (v *VaultOIDCValidator) ValidateToken(tokenString string) (jwt.MapClaims, error) {
    // Parse token with validation for issuer, audience, expiry
    token, err := jwt.Parse(tokenString, func(token *jwt.Token) (interface{}, error) {
        // Validate signing method is RS256 (used by Vault 1.16 OIDC plugin)
        if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
            return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
        }
        // Get signing key from Vault OIDC discovery endpoint
        // In production, cache this key to avoid repeated lookups
        keyResp, err := v.vaultClient.Logical().Read("auth/oidc-iam-identity-center/oidc/discovery/keys")
        if err != nil {
            return nil, fmt.Errorf("failed to get OIDC signing keys: %w", err)
        }
        if keyResp == nil || keyResp.Data == nil {
            return nil, fmt.Errorf("no OIDC signing keys found")
        }
        // Extract public key from JWKS response (simplified for example)
        keys := keyResp.Data["keys"].([]interface{})
        for _, k := range keys {
            keyMap := k.(map[string]interface{})
            if keyMap["kid"] == token.Header["kid"] {
                // Convert JWK to RSA public key (omitted for brevity; use jwt-go's jwk package in prod)
                return nil, fmt.Errorf("JWK to RSA conversion not implemented in example")
            }
        }
        return nil, fmt.Errorf("signing key not found for kid: %v", token.Header["kid"])
    })

    if err != nil {
        return nil, fmt.Errorf("token validation failed: %w", err)
    }

    // Check token claims
    claims, ok := token.Claims.(jwt.MapClaims)
    if !ok || !token.Valid {
        return nil, fmt.Errorf("invalid token claims")
    }

    // Validate issuer matches IAM Identity Center instance
    if claims["iss"] != v.issuerURL {
        return nil, fmt.Errorf("invalid issuer: got %s, expected %s", claims["iss"], v.issuerURL)
    }

    // Validate audience matches Vault client ID
    aud, ok := claims["aud"].(string)
    if !ok || !strings.Contains(aud, v.expectedAud) {
        return nil, fmt.Errorf("invalid audience: got %s, expected %s", aud, v.expectedAud)
    }

    // Check token expiry
    exp, ok := claims["exp"].(float64)
    if !ok || time.Now().Unix() > int64(exp) {
        return nil, fmt.Errorf("token expired")
    }

    // Validate IAM Identity Center user/group attributes via SSO API
    userID, ok := claims["sub"].(string)
    if !ok {
        return nil, fmt.Errorf("no subject claim in token")
    }
    // Call AWS SSO to verify user exists (rate-limited in production)
    _, err = v.ssoClient.DescribeUser(context.TODO(), &ssoadmin.DescribeUserInput{
        IdentityStoreId: aws.String(os.Getenv("SSO_IDENTITY_STORE_ID")),
        UserId:          aws.String(userID),
    })
    if err != nil {
        return nil, fmt.Errorf("user %s not found in IAM Identity Center: %w", userID, err)
    }

    return claims, nil
}

func main() {
    // Read required environment variables with error handling
    vaultAddr := os.Getenv("VAULT_ADDR")
    if vaultAddr == "" {
        log.Fatal("VAULT_ADDR environment variable is required")
    }
    vaultToken := os.Getenv("VAULT_ROOT_TOKEN")
    if vaultToken == "" {
        log.Fatal("VAULT_ROOT_TOKEN environment variable is required")
    }
    issuerURL := os.Getenv("OIDC_ISSUER_URL")
    if issuerURL == "" {
        log.Fatal("OIDC_ISSUER_URL environment variable is required")
    }
    expectedAud := os.Getenv("OIDC_EXPECTED_AUD")
    if expectedAud == "" {
        log.Fatal("OIDC_EXPECTED_AUD environment variable is required")
    }
    tokenToValidate := os.Getenv("OIDC_TOKEN")
    if tokenToValidate == "" {
        log.Fatal("OIDC_TOKEN environment variable is required")
    }

    // Initialize validator
    validator, err := NewValidator(vaultAddr, vaultToken, issuerURL, expectedAud)
    if err != nil {
        log.Fatalf("Failed to initialize validator: %v", err)
    }

    // Validate token
    claims, err := validator.ValidateToken(tokenToValidate)
    if err != nil {
        log.Fatalf("Token validation failed: %v", err)
    }

    fmt.Printf("Token validated successfully. Claims: %+v\n", claims)
}

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

import boto3
from hvac import Client as VaultClient # HVAC v2.0+ supports Vault 1.16
from hvac.exceptions import VaultError

# 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 VaultIAMKeyRotator:
    """Rotates AWS IAM keys via Vault 1.16 dynamic secrets, eliminating long-lived credentials."""

    def __init__(self, vault_addr: str, vault_token: str, aws_secret_path: str = "aws/sts/vault-workloads"):
        self.vault_addr = vault_addr
        self.vault_token = vault_token
        self.aws_secret_path = aws_secret_path
        self.vault_client: Optional[VaultClient] = None
        self.iam_client = None

    def _init_vault_client(self) -> VaultClient:
        """Initialize Vault client with TLS verification and error handling."""
        try:
            client = VaultClient(
                url=self.vault_addr,
                token=self.vault_token,
                verify=os.getenv("VAULT_CA_CERT") is not None # Enforce TLS in production
            )
            # Test connection to Vault
            client.sys.read_health_status()
            logger.info(f"Connected to Vault 1.16 at {self.vault_addr}")
            return client
        except VaultError as e:
            logger.error(f"Failed to connect to Vault: {e}")
            raise
        except Exception as e:
            logger.error(f"Unexpected error initializing Vault client: {e}")
            raise

    def _init_iam_client(self, aws_access_key: str, aws_secret_key: str, aws_session_token: str) -> boto3.client:
        """Initialize AWS IAM client with temporary credentials from Vault."""
        try:
            return boto3.client(
                "iam",
                aws_access_key_id=aws_access_key,
                aws_secret_access_key=aws_secret_key,
                aws_session_token=aws_session_token,
                region_name="us-east-1"
            )
        except Exception as e:
            logger.error(f"Failed to initialize IAM client: {e}")
            raise

    def get_dynamic_iam_credentials(self) -> Dict[str, str]:
        """Retrieve temporary IAM credentials from Vault 1.16 dynamic secrets engine."""
        if not self.vault_client:
            self.vault_client = self._init_vault_client()

        try:
            # Read dynamic AWS credentials from Vault; Vault 1.16 returns 1h TTL by default
            resp = self.vault_client.secrets.aws.read_role_credentials(
                path=self.aws_secret_path,
                mount_point="aws"
            )
            if not resp["data"]:
                raise ValueError("No credentials returned from Vault")

            creds = resp["data"]
            logger.info(f"Retrieved dynamic IAM credentials, TTL: {resp['lease_duration']}s")
            return {
                "access_key": creds["access_key"],
                "secret_key": creds["secret_key"],
                "session_token": creds.get("security_token", ""),
                "lease_id": resp["lease_id"]
            }
        except VaultError as e:
            logger.error(f"Vault error retrieving credentials: {e}")
            raise
        except KeyError as e:
            logger.error(f"Missing key in Vault response: {e}")
            raise

    def rotate_iam_key(self, iam_username: str) -> Dict[str, str]:
        """Rotate the IAM access key for a given user using Vault-issued temporary credentials."""
        # Get temporary credentials from Vault
        temp_creds = self.get_dynamic_iam_credentials()
        # Initialize IAM client with temp creds
        self.iam_client = self._init_iam_client(
            temp_creds["access_key"],
            temp_creds["secret_key"],
            temp_creds["session_token"]
        )

        try:
            # List existing keys for the user
            existing_keys = self.iam_client.list_access_keys(UserName=iam_username)
            logger.info(f"Found {len(existing_keys['AccessKeyMetadata'])} existing keys for {iam_username}")

            # Delete old keys (keep max 2 for rollback)
            for key in existing_keys["AccessKeyMetadata"]:
                if len(existing_keys["AccessKeyMetadata"]) >= 2:
                    logger.info(f"Deleting old key {key['AccessKeyId']} for {iam_username}")
                    self.iam_client.delete_access_key(
                        UserName=iam_username,
                        AccessKeyId=key["AccessKeyId"]
                    )
                    time.sleep(1) # Avoid AWS rate limits

            # Create new access key
            new_key = self.iam_client.create_access_key(UserName=iam_username)
            new_creds = new_key["AccessKey"]
            logger.info(f"Created new access key {new_creds['AccessKeyId']} for {iam_username}")

            # Revoke Vault lease for old credentials
            if temp_creds["lease_id"]:
                self.vault_client.sys.revoke_or_renew_orchestrator(lease_id=temp_creds["lease_id"])
                logger.info(f"Revoked Vault lease {temp_creds['lease_id']}")

            return {
                "new_access_key": new_creds["AccessKeyId"],
                "new_secret_key": new_creds["SecretAccessKey"]
            }
        except Exception as e:
            logger.error(f"Failed to rotate key for {iam_username}: {e}")
            raise

if __name__ == "__main__":
    # Validate required environment variables
    required_vars = ["VAULT_ADDR", "VAULT_ROOT_TOKEN", "IAM_USERNAME"]
    for var in required_vars:
        if not os.getenv(var):
            logger.error(f"Missing required environment variable: {var}")
            sys.exit(1)

    # Initialize rotator
    rotator = VaultIAMKeyRotator(
        vault_addr=os.getenv("VAULT_ADDR"),
        vault_token=os.getenv("VAULT_ROOT_TOKEN"),
        aws_secret_path=os.getenv("VAULT_AWS_PATH", "aws/sts/vault-workloads")
    )

    # Rotate key for target IAM user
    try:
        result = rotator.rotate_iam_key(os.getenv("IAM_USERNAME"))
        logger.info(f"Key rotation successful: {result}")
    except Exception as e:
        logger.error(f"Key rotation failed: {e}")
        sys.exit(1)

# Enable Vault 1.16 OIDC workload identity plugin via CLI
vault auth enable -path=oidc-iam-identity-center oidc
vault write auth/oidc-iam-identity-center/config \
  oidc_discovery_url="https://identitycenter.amazonaws.com/sso-instance-id" \
  plugin_name="vault-oidc-workload-identity-plugin" \
  plugin_version="1.16.0"

# Map IAM Identity Center ABAC attributes to Vault policy via Terraform
resource "vault_jwt_auth_backend_role" "prod_backend" {
  backend        = vault_jwt_auth_backend.iam_identity_center.path
  role_name      = "prod-backend-role"
  token_policies = ["prod-backend-secrets"]

  bound_audiences = ["vault.workload.identity"]
  bound_claims = {
    "aws:PrincipalTag/Environment" = "prod"
    "aws:PrincipalTag/Department"  = "backend"
    "aws:PrincipalTag/Role"        = "engineer"
  }
  user_claim = "sub"
}

# Run Vault 1.16 benchmark for OIDC auth
vault benchmark auth \
  -path=oidc-iam-identity-center \
  -method=oidc \
  -count=1000 \
  -concurrency=10 \
  -token-ttl=15m

vault-iam-identity-center-zero-trust/
├── terraform/ # Infrastructure as code for AWS and Vault setup
│   ├── iam-identity-center.tf # IAM Identity Center and OIDC provider config
│   ├── vault.tf # Vault 1.16 deployment and auth method config
│   ├── variables.tf # Input variables for deployment
│   └── outputs.tf # Deployment outputs (Vault addr, SSO instance ID)
├── go/
│   └── validator/ # OIDC token validator written in Go
│       ├── main.go # Main validator logic
│       ├── go.mod # Go module dependencies
│       └── Dockerfile # Container image for validator
├── python/
│   └── key-rotator/ # IAM key rotator written in Python
│       ├── rotator.py # Main rotator logic
│       ├── requirements.txt # Python dependencies (hvac, boto3)
│       └── Dockerfile # Container image for rotator
├── benchmarks/ # Benchmark results and scripts
│   ├── vault-benchmark-results.json # Auth latency benchmark data
│   └── run-benchmarks.sh # Script to run Vault 1.16 benchmarks
└── README.md # Setup instructions and troubleshooting