Multi-Cloud Zero Trust Security Platform
Build an enterprise security platform spanning AWS and GCP with zero static credentials.
Introduction
30 Second Summary
When organizations spread their workloads across multiple clouds, a single compromised credential can unlock resources in both environments simultaneously. Zero Trust architecture solves this by eliminating implicit trust between systems and verifying every request, every time.
In this project, you will build an enterprise-grade multi-cloud security platform spanning AWS and GCP with encrypted private networking, keyless cross-cloud authentication, automated threat detection and response, policy-as-code governance, and unified observability.
What You'll Build
A complete security platform that connects AWS and GCP via HA VPN, authenticates workloads without static credentials using Workload Identity Federation, orchestrates incident response with Step Functions and Python Lambda functions, and enforces compliance with OPA policies and Prowler scanning.
By the end of this project, you'll have:
- A Terraform monorepo managing 50+ resources across AWS and GCP with remote state, locking, and modular architecture (networking, identity, detection, orchestration, governance, observability modules).
- A Python automation toolkit with Lambda functions for incident response, Cloud Functions for cross-cloud event bridging, CLI tools for audit querying, compliance report generation, and diagrams-as-code.
- A fully automated security orchestration pipeline that detects threats via GuardDuty and Security Command Center, correlates cross-cloud events, triggers multi-step remediation, and logs everything to a queryable audit trail.
- Secret Mission: a cross-cloud honeypot system with a Python canary token generator that deploys decoy resources in both clouds and triggers the Step Functions orchestrator when honeypots are touched.
Are there any prerequisites?
This is an advanced project. You need an AWS account with CLI configured, a GCP account with billing enabled and gcloud CLI configured, Terraform installed, Python 3 installed, Git installed, and a GitHub account.
You should be comfortable with basic Terraform resource definitions, Python scripting, and navigating both cloud consoles.
Build the Terraform Monorepo and Python Toolkit
Your tools are installed and ready. Now you need a structured home for the infrastructure code and automation scripts that will power the entire security platform.
Enterprise multi-cloud projects demand a monorepo pattern where all Terraform modules, Python scripts, and configuration live in a single repository. In this step, you will scaffold that structure, build a shared Python utilities package, and wire up dual remote state backends so your infrastructure state is safely stored in both clouds.
In this step, get ready to:
- Initialize the Terraform monorepo with six infrastructure modules.
- Create a shared Python utilities package with client factories and error handling.
- Configure dual remote state backends (S3 and GCS) with locking and encryption.
Initialize the Terraform monorepo structure
A monorepo groups related infrastructure modules together so you can version, test, and deploy them as a unit. Each module handles one security domain.
- Switch back to your terminal (with the virtual environment active) and create the module directories and scripts folder by running these commands:
mkdir -p modules/networking modules/identity modules/detection \
modules/orchestration modules/governance modules/observability \
scripts/utils- Add placeholder files so Git tracks the empty module directories by running:
touch modules/networking/.gitkeep \
modules/identity/.gitkeep \
modules/detection/.gitkeep \
modules/orchestration/.gitkeep \
modules/governance/.gitkeep \
modules/observability/.gitkeep- Confirm the structure exists by running:
ls modules/You should see six directories listed: detection, governance, identity, networking, observability, and orchestration.
Why six modules?
Each module maps to one pillar of enterprise security: network perimeter, identity federation, threat detection, incident orchestration, policy governance, and operational observability. Keeping them separate means you can develop, test, and apply changes to one pillar without risking the others.
Create the Python requirements and shared utilities
Your virtual environment already has the Python packages installed. Now you will create a requirements.txt to pin them and build a shared utilities package that every script in the project will import.
- Create the requirements.txt file in your project root by running:
cat > requirements.txt << 'EOF'
boto3
google-cloud-storage
google-auth
requests
rich
diagrams
EOFWhat does this file do?
The requirements.txt documents every Python dependency the project needs. Anyone cloning the repo can recreate your environment with a single pip install -r requirements.txt command.
- Create the logging setup module by running:
# scripts/utils/logging_setup.py
"""Common logging configuration for security platform scripts."""
import logging
import sys
def setup_logging(name: str, level: str = "INFO") -> logging.Logger:
"""Configure and return a logger with consistent formatting."""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper()))
# Console handler with structured output
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(getattr(logging, level.upper()))
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
handler.setFormatter(formatter)
# Avoid duplicate handlers on repeated calls
if not logger.handlers:
logger.addHandler(handler)
return loggerWhat does this code do?
Every script in the project will call setup_logging(__name__) to get a pre-configured logger. The consistent timestamp and severity format makes it easy to correlate events across scripts during incident response testing.
- Save scripts/utils/logging_setup.py.
- Create the error handling module by running:
# scripts/utils/error_handling.py
"""Common error handling utilities for the security platform."""
import sys
import traceback
from .logging_setup import setup_logging
logger = setup_logging(__name__)
class SecurityPlatformError(Exception):
"""Base exception for security platform operations."""
def __init__(self, message: str, service: str = None, operation: str = None):
self.service = service
self.operation = operation
super().__init__(message)
def handle_error(error: Exception, context: str = ""):
"""Log error details and return a structured error response."""
error_info = {
"error_type": type(error).__name__,
"message": str(error),
"context": context,
"traceback": traceback.format_exc(),
}
logger.error(f"[{context}] {type(error).__name__}: {error}")
return error_info
def exit_on_error(error: Exception, context: str = ""):
"""Log the error and exit with a non-zero status code."""
handle_error(error, context)
sys.exit(1)What does this code do?
The SecurityPlatformError class gives you a project-wide base exception that carries the service name and operation context. The handle_error function captures structured error data for later analysis. The exit_on_error function terminates scripts cleanly after logging.
- Save scripts/utils/error_handling.py.
- Create the AWS client factory by running:
# scripts/utils/aws_client.py
"""AWS client factory using boto3."""
import boto3
from botocore.config import Config
from .logging_setup import setup_logging
logger = setup_logging(__name__)
def get_aws_client(service: str, region: str = "us-east-1", **kwargs):
"""Create and return a configured boto3 client."""
config = Config(
retries={"max_attempts": 3, "mode": "adaptive"},
connect_timeout=5,
read_timeout=10,
)
logger.info(f"Creating AWS client for {service} in {region}")
return boto3.client(
service,
region_name=region,
config=config,
**kwargs
)
def get_aws_resource(service: str, region: str = "us-east-1", **kwargs):
"""Create and return a configured boto3 resource."""
config = Config(
retries={"max_attempts": 3, "mode": "adaptive"},
connect_timeout=5,
read_timeout=10,
)
return boto3.resource(
service,
region_name=region,
config=config,
**kwargs
)What does this code do?
Instead of calling boto3.client() directly in every script, you call get_aws_client("ec2"). This centralizes retry logic (adaptive mode with 3 attempts), timeouts, and logging so every AWS call behaves consistently across the project.
- Save scripts/utils/aws_client.py.
- Create the GCP client factory by running:
# scripts/utils/gcp_client.py
"""GCP client factory using google-auth."""
import google.auth
from google.auth import default as get_default_credentials
from google.cloud import storage
from .logging_setup import setup_logging
logger = setup_logging(__name__)
def get_gcp_credentials(scopes=None):
"""Get default GCP credentials with optional scope."""
if scopes is None:
scopes = ["https://www.googleapis.com/auth/cloud-platform"]
credentials, project = get_default_credentials(scopes=scopes)
logger.info(f"Authenticated to GCP project: {project}")
return credentials, project
def get_gcs_client():
"""Create and return a configured GCS storage client."""
credentials, project = get_gcp_credentials()
return storage.Client(
project=project,
credentials=credentials
)What does this code do?
The get_gcp_credentials function wraps Google's Application Default Credentials flow with logging. The get_gcs_client function returns a ready-to-use Cloud Storage client. Later steps will add more GCP client helpers here as you integrate additional services.
- Save scripts/utils/gcp_client.py.
- Create the package init file that exposes the public API by running:
# scripts/utils/__init__.py
"""Shared utilities package for the Zero Trust Security Platform."""
from .logging_setup import setup_logging
from .aws_client import get_aws_client
from .gcp_client import get_gcp_client
from .error_handling import handle_error, SecurityPlatformErrorWhat does this code do?
The __init__.py re-exports the most-used functions so other scripts can import with a short path like from scripts.utils import get_aws_client instead of reaching into individual modules.
- Save scripts/utils/__init__.py.
- Verify the package structure by running:
ls scripts/utils/You should see five files: __init__.py, aws_client.py, error_handling.py, gcp_client.py, and logging_setup.py.
Import errors when testing?
If you try importing and get a ModuleNotFoundError, make sure your virtual environment is active. The relative imports (from .logging_setup) only work when Python recognizes scripts/utils as a package.
help me debug my Python import error for the utils package
Configure dual remote state backends and providers
Terraform state tracks every resource you create. For a multi-cloud project, you need state stored securely in both clouds with locking to prevent concurrent modifications.
You will create an S3 bucket with a DynamoDB lock table for AWS state, and a GCS bucket for GCP state. Then you will write the Terraform configuration that references them.
- Create the S3 state bucket and DynamoDB lock table by running these commands (replace the bucket name with something globally unique):
aws s3api create-bucket \
--bucket [[S3_STATE_BUCKET="your-unique-s3-state-bucket-name"]] \
--region us-east-1
aws s3api put-bucket-versioning \
--bucket [[S3_STATE_BUCKET="your-unique-s3-state-bucket-name"]] \
--versioning-configuration Status=Enabled
aws dynamodb create-table \
--table-name zero-trust-platform-tflock \
--attribute-definitions AttributeName=LockID,AttributeType=S \
--key-schema AttributeName=LockID,KeyType=HASH \
--billing-mode PAY_PER_REQUEST \
--region us-east-1What do these commands do?
The first command creates the S3 bucket that will hold your Terraform state file. The second enables versioning so you can recover previous state if something goes wrong.
The third command creates a DynamoDB table with a LockID partition key. Terraform uses this table to lock state during plan and apply operations, preventing two engineers from modifying infrastructure simultaneously.
- Create the GCS state bucket by running (replace the bucket name with something globally unique):
gcloud storage buckets create gs://[[GCS_STATE_BUCKET="your-unique-gcs-state-bucket-name"]] \
--location=us-central1 \
--uniform-bucket-level-access
gcloud storage buckets update gs://[[GCS_STATE_BUCKET="your-unique-gcs-state-bucket-name"]] \
--versioningWhat do these commands do?
The first command creates a GCS bucket with uniform bucket-level access (the recommended security model). The second enables object versioning for state recovery. The GCS backend supports state locking natively without needing a separate table.
Bucket name already exists?
Both S3 and GCS bucket names must be globally unique across all accounts. If you get an error about the name being taken, append your initials or a short random string to the bucket name.
help me resolve a bucket naming conflict for my Terraform state backend
- Create the backend.tf file in your project root with the following content:
# backend.tf
# Primary backend: S3 for AWS state with DynamoDB locking
terraform {
backend "s3" {
bucket = "[[S3_STATE_BUCKET=\"your-unique-s3-state-bucket-name\"]]"
key = "global/terraform.tfstate"
region = "us-east-1"
dynamodb_table = "zero-trust-platform-tflock"
encrypt = true
}
}
# GCS backend for GCP state (used in separate workspace or via
# terraform_remote_state data source for cross-cloud references).
# To use GCS as primary instead, replace the block above with:
#
# terraform {
# backend "gcs" {
# bucket = "your-unique-gcs-state-bucket-name"
# prefix = "global/terraform.tfstate"
# }
# }Why not two active backends?
Terraform only allows one backend per configuration. The standard multi-cloud pattern uses S3 as the primary backend for the root module and references GCP state via terraform_remote_state data sources or separate workspaces. The GCS bucket you created will store state for GCP-focused modules in later sessions.
- Save backend.tf.
- Create the main.tf file with provider configurations and module declarations:
# main.tf
# Provider and module configuration for the Zero Trust Security Platform
terraform {
required_version = ">= 1.0"
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 6.0"
}
google = {
source = "hashicorp/google"
version = "~> 7.0"
}
}
}
# AWS provider configuration
provider "aws" {
region = "us-east-1"
}
# Google Cloud provider configuration
provider "google" {
project = [[GCP_PROJECT_ID="your-gcp-project-id"]]
region = "us-central1"
}
# Security domain modules
module "networking" {
source = "./modules/networking"
}
module "identity" {
source = "./modules/identity"
}
module "detection" {
source = "./modules/detection"
}
module "orchestration" {
source = "./modules/orchestration"
}
module "governance" {
source = "./modules/governance"
}
module "observability" {
source = "./modules/observability"
}What does this code do?
The required_providers block pins the AWS provider to the ~> 6.0 constraint (any 6.x release) and the Google provider to ~> 7.0 (any 7.x release). This allows patch updates while preventing breaking major version changes.
Each module block points to a local directory. Right now they only contain .gitkeep files. You will populate them with real resources in upcoming sessions.
- Save main.tf.
- Initialize Terraform to download providers and validate your backend by running:
terraform initYou should see output confirming the AWS and Google providers were installed, and that the S3 backend was successfully configured.
Backend initialization failed?
If Terraform reports that the S3 bucket does not exist, double-check that the bucket name in backend.tf matches the bucket you created with the AWS CLI. Names are case-sensitive and must match exactly.
If you see a DynamoDB error, confirm the table zero-trust-platform-tflock was created in us-east-1 and has a partition key named LockID of type String.
help me debug my Terraform backend initialization error
✔️ Awesome, I've got everything!
Great. Double check you have saved both backend.tf and main.tf, and that terraform init completed without errors.
ⓧ I'd like to double check the full code
Here is the complete backend.tf:
# backend.tf
# Primary backend: S3 for AWS state with DynamoDB locking
terraform {
backend "s3" {
bucket = "[[S3_STATE_BUCKET=\"your-unique-s3-state-bucket-name\"]]"
key = "global/terraform.tfstate"
region = "us-east-1"
dynamodb_table = "zero-trust-platform-tflock"
encrypt = true
}
}
# GCS backend for GCP state (used in separate workspace or via
# terraform_remote_state data source for cross-cloud references).
# To use GCS as primary instead, replace the block above with:
#
# terraform {
# backend "gcs" {
# bucket = "your-unique-gcs-state-bucket-name"
# prefix = "global/terraform.tfstate"
# }
# }Here is the complete main.tf:
# main.tf
# Provider and module configuration for the Zero Trust Security Platform
terraform {
required_version = ">= 1.0"
required_providers {
aws = {
source = "hashicorp/aws"
version = "~> 6.0"
}
google = {
source = "hashicorp/google"
version = "~> 7.0"
}
}
}
# AWS provider configuration
provider "aws" {
region = "us-east-1"
}
# Google Cloud provider configuration
provider "google" {
project = [[GCP_PROJECT_ID="your-gcp-project-id"]]
region = "us-central1"
}
# Security domain modules
module "networking" {
source = "./modules/networking"
}
module "identity" {
source = "./modules/identity"
}
module "detection" {
source = "./modules/detection"
}
module "orchestration" {
source = "./modules/orchestration"
}
module "governance" {
source = "./modules/governance"
}
module "observability" {
source = "./modules/observability"
}Here is the complete requirements.txt:
boto3
google-cloud-storage
google-auth
requests
rich
diagramsHere is the complete scripts/utils/__init__.py:
# scripts/utils/__init__.py
"""Shared utilities package for the Zero Trust Security Platform."""
from .logging_setup import setup_logging
from .aws_client import get_aws_client
from .gcp_client import get_gcp_client
from .error_handling import handle_error, SecurityPlatformErrorHere is the complete scripts/utils/logging_setup.py:
# scripts/utils/logging_setup.py
"""Common logging configuration for security platform scripts."""
import logging
import sys
def setup_logging(name: str, level: str = "INFO") -> logging.Logger:
"""Configure and return a logger with consistent formatting."""
logger = logging.getLogger(name)
logger.setLevel(getattr(logging, level.upper()))
# Console handler with structured output
handler = logging.StreamHandler(sys.stdout)
handler.setLevel(getattr(logging, level.upper()))
formatter = logging.Formatter(
"%(asctime)s - %(name)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
handler.setFormatter(formatter)
# Avoid duplicate handlers on repeated calls
if not logger.handlers:
logger.addHandler(handler)
return loggerHere is the complete scripts/utils/error_handling.py:
# scripts/utils/error_handling.py
"""Common error handling utilities for the security platform."""
import sys
import traceback
from .logging_setup import setup_logging
logger = setup_logging(__name__)
class SecurityPlatformError(Exception):
"""Base exception for security platform operations."""
def __init__(self, message: str, service: str = None, operation: str = None):
self.service = service
self.operation = operation
super().__init__(message)
def handle_error(error: Exception, context: str = ""):
"""Log error details and return a structured error response."""
error_info = {
"error_type": type(error).__name__,
"message": str(error),
"context": context,
"traceback": traceback.format_exc(),
}
logger.error(f"[{context}] {type(error).__name__}: {error}")
return error_info
def exit_on_error(error: Exception, context: str = ""):
"""Log the error and exit with a non-zero status code."""
handle_error(error, context)
sys.exit(1)Here is the complete scripts/utils/aws_client.py:
# scripts/utils/aws_client.py
"""AWS client factory using boto3."""
import boto3
from botocore.config import Config
from .logging_setup import setup_logging
logger = setup_logging(__name__)
def get_aws_client(service: str, region: str = "us-east-1", **kwargs):
"""Create and return a configured boto3 client."""
config = Config(
retries={"max_attempts": 3, "mode": "adaptive"},
connect_timeout=5,
read_timeout=10,
)
logger.info(f"Creating AWS client for {service} in {region}")
return boto3.client(
service,
region_name=region,
config=config,
**kwargs
)
def get_aws_resource(service: str, region: str = "us-east-1", **kwargs):
"""Create and return a configured boto3 resource."""
config = Config(
retries={"max_attempts": 3, "mode": "adaptive"},
connect_timeout=5,
read_timeout=10,
)
return boto3.resource(
service,
region_name=region,
config=config,
**kwargs
)Here is the complete scripts/utils/gcp_client.py:
# scripts/utils/gcp_client.py
"""GCP client factory using google-auth."""
import google.auth
from google.auth import default as get_default_credentials
from google.cloud import storage
from .logging_setup import setup_logging
logger = setup_logging(__name__)
def get_gcp_credentials(scopes=None):
"""Get default GCP credentials with optional scope."""
if scopes is None:
scopes = ["https://www.googleapis.com/auth/cloud-platform"]
credentials, project = get_default_credentials(scopes=scopes)
logger.info(f"Authenticated to GCP project: {project}")
return credentials, project
def get_gcs_client():
"""Create and return a configured GCS storage client."""
credentials, project = get_gcp_credentials()
return storage.Client(
project=project,
credentials=credentials
)Your monorepo skeleton is in place with modular Terraform structure, a Python utilities package, and remote state safely stored in both clouds. Next up, you will fill the networking module with a cross-cloud VPN connecting AWS and GCP over encrypted private tunnels.
Deploy Cross-Cloud Private Networking
Your Terraform monorepo and Python toolkit are ready. Now you need a private, encrypted communication channel between your AWS and GCP environments.
In this step, you will deploy virtual private clouds in both AWS and GCP, then link them together using an HA VPN with BGP dynamic routing. This gives your security platform a private backbone that never touches the public internet.
In this step, get ready to:
- Deploy an AWS VPC and a GCP VPC with private subnets using Terraform.
- Configure an HA VPN gateway with Cloud Router and 4 IPsec tunnels using IKEv2.
- Write and run a Python connectivity validation script that outputs a formatted health report.
Deploy AWS VPC and GCP VPC with Terraform
Both clouds need isolated network spaces before you can connect them. You will create an AWS VPC with CIDR 10.0.0.0/16 and a GCP VPC with CIDR 172.16.0.0/16.
- Create the file modules/networking/variables.tf by adding the following code:
variable "aws_vpc_cidr" {
description = "CIDR block for the AWS VPC"
type = string
default = "10.0.0.0/16"
}
variable "gcp_vpc_cidr" {
description = "CIDR block for the GCP VPC subnet"
type = string
default = "172.16.0.0/16"
}
variable "aws_bgp_asn" {
description = "BGP ASN for the AWS Virtual Private Gateway"
type = number
default = 64512
}
variable "gcp_bgp_asn" {
description = "BGP ASN for the GCP Cloud Router"
type = number
default = 65001
}
variable "tunnel_count" {
description = "Number of VPN tunnels (4 for 99.99% SLA)"
type = number
default = 4
}What does this code do?
These variables parameterize the networking module so you can reuse it across environments. The BGP ASNs follow the standard convention for cross-cloud VPN: GCP Cloud Router uses ASN 65001 and AWS Virtual Private Gateway uses ASN 64512 (both in the private ASN range 64512-65534).
The tunnel_count defaults to 4 because GCP HA VPN requires a minimum of four tunnels for its 99.99% availability SLA.
- Save modules/networking/variables.tf.
- Create the file modules/networking/main.tf and add the AWS VPC and subnet resources:
# --- AWS VPC and Private Subnets ---
resource "aws_vpc" "main" {
cidr_block = var.aws_vpc_cidr
enable_dns_support = true
enable_dns_hostnames = true
tags = {
Name = "zero-trust-vpc"
}
}
resource "aws_subnet" "private_a" {
vpc_id = aws_vpc.main.id
cidr_block = "10.0.1.0/24"
availability_zone = "us-east-1a"
tags = {
Name = "private-subnet-a"
}
}
resource "aws_subnet" "private_b" {
vpc_id = aws_vpc.main.id
cidr_block = "10.0.2.0/24"
availability_zone = "us-east-1b"
tags = {
Name = "private-subnet-b"
}
}What does this code do?
This creates an AWS VPC spanning the 10.0.0.0/16 address space with two private subnets in separate availability zones. DNS support is enabled so resources inside the VPC can resolve hostnames.
The subnets are private because they have no internet gateway attached. Cross-cloud traffic will flow exclusively through the VPN tunnel.
- Below the AWS resources in modules/networking/main.tf, add the GCP VPC and subnet:
# --- GCP VPC and Custom Subnet ---
resource "google_compute_network" "main" {
name = "zero-trust-vpc"
auto_create_subnetworks = false
}
resource "google_compute_subnetwork" "private" {
name = "private-subnet"
ip_cidr_range = var.gcp_vpc_cidr
region = "us-central1"
network = google_compute_network.main.id
}What does this code do?
GCP VPCs use custom mode (auto_create_subnetworks = false) so you control the exact CIDR ranges. The subnet uses 172.16.0.0/16 in us-central1, giving your GCP workloads a dedicated private address space that does not overlap with AWS.
- Save modules/networking/main.tf.
- Confirm both files exist by running:
ls modules/networking/You should see main.tf and variables.tf in the output.
Don't see the files?
Make sure you created the files inside modules/networking/ and not in the project root. The directory should already exist from Step 1. If not, create it with mkdir -p modules/networking.
Help me debug my networking module file structure.
Configure HA VPN gateway with Cloud Router and IPsec tunnels
With both VPCs in place, you need to connect them. GCP HA VPN provides an encrypted IPsec tunnel with 99.99% availability when configured with four tunnels.
Why HA VPN over Cloud Interconnect?
Cloud Interconnect requires a physical connection order and costs upwards of $1000/month. HA VPN teaches the same concepts (BGP, IPsec, cross-cloud routing) at roughly $0.40/hr and can be torn down immediately after verification.
- In modules/networking/main.tf, add the AWS VPN Gateway and VPN connection resources below the GCP subnet:
# --- AWS Virtual Private Gateway ---
resource "aws_vpn_gateway" "main" {
vpc_id = aws_vpc.main.id
amazon_side_asn = var.aws_bgp_asn
tags = {
Name = "zero-trust-vgw"
}
}
# Pre-shared keys for VPN tunnels (generated, not hardcoded)
resource "random_password" "tunnel_psk" {
count = var.tunnel_count
length = 32
special = false
}
# Customer Gateways represent GCP HA VPN interfaces in AWS
resource "aws_customer_gateway" "gcp" {
count = 2
bgp_asn = var.gcp_bgp_asn
ip_address = google_compute_ha_vpn_gateway.main.vpn_interfaces[count.index].ip_address
type = "ipsec.1"
tags = {
Name = "gcp-cgw-${count.index}"
}
}
# VPN Connections (2 connections x 2 tunnels each = 4 tunnels total)
resource "aws_vpn_connection" "gcp" {
count = 2
vpn_gateway_id = aws_vpn_gateway.main.id
customer_gateway_id = aws_customer_gateway.gcp[count.index].id
type = "ipsec.1"
static_routes_only = false
# IKEv2 with single cipher set to prevent IP fragmentation
tunnel1_ike_versions = ["ikev2"]
tunnel1_phase1_encryption_algorithms = ["AES256"]
tunnel1_phase1_integrity_algorithms = ["SHA2-256"]
tunnel1_phase1_dh_group_numbers = [14]
tunnel1_phase2_encryption_algorithms = ["AES256"]
tunnel1_phase2_integrity_algorithms = ["SHA2-256"]
tunnel1_phase2_dh_group_numbers = [14]
tunnel1_preshared_key = random_password.tunnel_psk[count.index * 2].result
tunnel2_ike_versions = ["ikev2"]
tunnel2_phase1_encryption_algorithms = ["AES256"]
tunnel2_phase1_integrity_algorithms = ["SHA2-256"]
tunnel2_phase1_dh_group_numbers = [14]
tunnel2_phase2_encryption_algorithms = ["AES256"]
tunnel2_phase2_integrity_algorithms = ["SHA2-256"]
tunnel2_phase2_dh_group_numbers = [14]
tunnel2_preshared_key = random_password.tunnel_psk[count.index * 2 + 1].result
tags = {
Name = "gcp-vpn-${count.index}"
}
}Why a single cipher set with IKEv2?
AWS default IKE transform sets include multiple algorithm proposals. This creates large Security Association payloads that cause IP fragmentation. GCP Cloud VPN does not support reassembling fragmented IKE packets.
By specifying exactly one Phase 1 and Phase 2 algorithm (AES256 + SHA2-256 + DH group 14), you keep the SA payload small and prevent fragmentation failures. The random_password resource generates secure pre-shared keys without hardcoding secrets in your Terraform files.
- Continue in modules/networking/main.tf and add the GCP HA VPN Gateway, External VPN Gateway, Cloud Router, tunnels, and BGP peer configuration:
# --- GCP HA VPN Gateway ---
resource "google_compute_ha_vpn_gateway" "main" {
name = "zero-trust-ha-vpn"
region = "us-central1"
network = google_compute_network.main.id
}
# External VPN Gateway represents AWS endpoints in GCP
resource "google_compute_external_vpn_gateway" "aws" {
name = "aws-vpn-gateway"
redundancy_type = "FOUR_IPS_REDUNDANCY"
interface {
id = 0
ip_address = aws_vpn_connection.gcp[0].tunnel1_address
}
interface {
id = 1
ip_address = aws_vpn_connection.gcp[0].tunnel2_address
}
interface {
id = 2
ip_address = aws_vpn_connection.gcp[1].tunnel1_address
}
interface {
id = 3
ip_address = aws_vpn_connection.gcp[1].tunnel2_address
}
}
# Cloud Router for BGP dynamic routing
resource "google_compute_router" "main" {
name = "zero-trust-router"
region = "us-central1"
network = google_compute_network.main.id
bgp {
asn = var.gcp_bgp_asn
}
}
# VPN Tunnels (4 total for 99.99% SLA)
resource "google_compute_vpn_tunnel" "aws" {
count = var.tunnel_count
name = "aws-tunnel-${count.index}"
region = "us-central1"
vpn_gateway = google_compute_ha_vpn_gateway.main.id
vpn_gateway_interface = count.index < 2 ? 0 : 1
peer_external_gateway = google_compute_external_vpn_gateway.aws.id
peer_external_gateway_interface = count.index
shared_secret = random_password.tunnel_psk[count.index].result
router = google_compute_router.main.id
ike_version = 2
}
# Router Interfaces for BGP sessions
resource "google_compute_router_interface" "aws" {
count = var.tunnel_count
name = "aws-interface-${count.index}"
router = google_compute_router.main.name
region = "us-central1"
ip_range = "169.254.${count.index + 1}.1/30"
vpn_tunnel = google_compute_vpn_tunnel.aws[count.index].name
}
# BGP Peers for dynamic route exchange
resource "google_compute_router_peer" "aws" {
count = var.tunnel_count
name = "aws-peer-${count.index}"
router = google_compute_router.main.name
region = "us-central1"
peer_ip_address = "169.254.${count.index + 1}.2"
peer_asn = var.aws_bgp_asn
advertised_route_priority = 100
interface = google_compute_router_interface.aws[count.index].name
}What does this code do?
The GCP HA VPN Gateway has two interfaces. The External VPN Gateway maps the four AWS tunnel endpoints (two per connection) into GCP's model. Each google_compute_vpn_tunnel resource connects to one AWS endpoint using IKE version 2.
The Cloud Router establishes BGP sessions over each tunnel using link-local addresses (169.254.x.x/30). BGP dynamically exchanges routes between the two clouds, so each side automatically learns the other's CIDR without manual route tables.
- Create the file modules/networking/outputs.tf with the following outputs:
output "aws_vpc_id" {
description = "ID of the AWS VPC"
value = aws_vpc.main.id
}
output "gcp_vpc_name" {
description = "Name of the GCP VPC network"
value = google_compute_network.main.name
}
output "vpn_tunnel_ids" {
description = "IDs of the GCP VPN tunnels"
value = google_compute_vpn_tunnel.aws[*].id
}
output "cloud_router_name" {
description = "Name of the GCP Cloud Router"
value = google_compute_router.main.name
}What does this code do?
These outputs expose key resource identifiers from the networking module. Other modules (identity, detection, orchestration) will reference aws_vpc_id and gcp_vpc_name to place their resources inside the correct networks.
- Now update your root main.tf to call the networking module. Find the module "networking" block and replace the placeholder with:
module "networking" {
source = "./modules/networking"
aws_vpc_cidr = "10.0.0.0/16"
gcp_vpc_cidr = "172.16.0.0/16"
aws_bgp_asn = 64512
gcp_bgp_asn = 65001
tunnel_count = 4
}- Save all files, then initialize and apply the networking module by running:
terraform init
terraform plan -target=module.networkingYou should see Terraform planning to create approximately 20 resources across both clouds (VPCs, subnets, gateways, tunnels, and BGP peers).
- Apply the networking infrastructure by running:
terraform apply -target=module.networking- Type yes when prompted to confirm.
The apply will take 3-5 minutes as AWS provisions VPN connections and GCP establishes HA VPN tunnels.
Seeing a provider error?
If Terraform reports a missing random provider, add a required_providers block to your root main.tf that includes hashicorp/random, then re-run terraform init.
Help me fix my Terraform provider configuration for the networking module.
What about VPN costs?
The HA VPN tunnels cost approximately $0.05 per VPN connection-hour on AWS and an hourly charge per tunnel on GCP. Combined, this is roughly $0.40/hr. Tear down the VPN after verification to avoid ongoing charges.
Write a Python connectivity validation script
Infrastructure is deployed, but you need to verify that tunnels are actually up, BGP sessions are established, and routes are being exchanged. A Python script gives you a repeatable health check you can run anytime.
- Install the GCP Compute client library for Python by running:
pip install google-cloud-compute- Create the file scripts/validate_networking.py with the following code:
#!/usr/bin/env python3
"""Cross-cloud VPN connectivity validation script.
Checks VPN tunnel status, BGP session state, and learned routes
from both AWS and GCP, outputting a formatted health report."""
import os
import boto3
from google.cloud import compute_v1
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from utils.logging_setup import get_logger
from utils.aws_client import get_aws_client
from utils.gcp_client import get_gcp_client
logger = get_logger(__name__)
console = Console()
def check_aws_vpn_status(ec2_client):
"""Check AWS VPN connection status and tunnel telemetry."""
response = ec2_client.describe_vpn_connections(
Filters=[{"Name": "tag:Name", "Values": ["gcp-vpn-*"]}]
)
tunnels = []
for connection in response.get("VpnConnections", []):
for tunnel in connection.get("VgwTelemetry", []):
tunnels.append({
"connection_id": connection["VpnConnectionId"],
"outside_ip": tunnel["OutsideIpAddress"],
"status": tunnel["Status"],
"status_message": tunnel.get("StatusMessage", ""),
"accepted_routes": tunnel.get("AcceptedRouteCount", 0),
})
return tunnels
def check_gcp_vpn_status(project_id, region="us-central1"):
"""Check GCP VPN tunnel status and BGP session state."""
tunnels_client = compute_v1.VpnTunnelsClient()
routers_client = compute_v1.RoutersClient()
# Get VPN tunnel statuses
tunnels = []
request = compute_v1.ListVpnTunnelsRequest(
project=project_id, region=region
)
for tunnel in tunnels_client.list(request=request):
if "aws-tunnel" in tunnel.name:
tunnels.append({
"name": tunnel.name,
"status": tunnel.status,
"peer_ip": tunnel.peer_ip,
"ike_version": tunnel.ike_version,
})
# Get BGP session status from Cloud Router
router_status = routers_client.get_router_status(
project=project_id,
region=region,
router="zero-trust-router",
)
bgp_peers = []
for peer in router_status.result.bgp_peer_status:
bgp_peers.append({
"name": peer.name,
"state": peer.status,
"peer_ip": peer.ip_address,
"learned_routes": len(peer.advertised_routes),
})
return tunnels, bgp_peers
def check_learned_routes(project_id, region="us-central1"):
"""Query learned routes from the GCP Cloud Router."""
routers_client = compute_v1.RoutersClient()
router_status = routers_client.get_router_status(
project=project_id,
region=region,
router="zero-trust-router",
)
routes = []
for route in router_status.result.best_routes:
routes.append({
"destination": route.dest_range,
"next_hop": route.next_hop_ip,
"priority": route.priority,
})
return routes
def display_health_report(aws_tunnels, gcp_tunnels, bgp_peers, routes):
"""Display a formatted health report using rich tables."""
console.print(Panel("Cross-Cloud VPN Health Report", style="bold blue"))
# AWS VPN Tunnels table
aws_table = Table(title="AWS VPN Tunnel Status")
aws_table.add_column("Connection ID", style="cyan")
aws_table.add_column("Outside IP", style="white")
aws_table.add_column("Status", style="green")
aws_table.add_column("Accepted Routes", style="yellow")
for tunnel in aws_tunnels:
status_style = "green" if tunnel["status"] == "UP" else "red"
aws_table.add_row(
tunnel["connection_id"],
tunnel["outside_ip"],
f"[{status_style}]{tunnel['status']}[/{status_style}]",
str(tunnel["accepted_routes"]),
)
console.print(aws_table)
# GCP VPN Tunnels table
gcp_table = Table(title="GCP VPN Tunnel Status")
gcp_table.add_column("Tunnel Name", style="cyan")
gcp_table.add_column("Peer IP", style="white")
gcp_table.add_column("Status", style="green")
gcp_table.add_column("IKE Version", style="yellow")
for tunnel in gcp_tunnels:
status_style = "green" if tunnel["status"] == "ESTABLISHED" else "red"
gcp_table.add_row(
tunnel["name"],
tunnel["peer_ip"],
f"[{status_style}]{tunnel['status']}[/{status_style}]",
str(tunnel["ike_version"]),
)
console.print(gcp_table)
# BGP Peers table
bgp_table = Table(title="BGP Session State")
bgp_table.add_column("Peer Name", style="cyan")
bgp_table.add_column("Peer IP", style="white")
bgp_table.add_column("State", style="green")
bgp_table.add_column("Learned Routes", style="yellow")
for peer in bgp_peers:
state_style = "green" if peer["state"] == "UP" else "red"
bgp_table.add_row(
peer["name"],
peer["peer_ip"],
f"[{state_style}]{peer['state']}[/{state_style}]",
str(peer["learned_routes"]),
)
console.print(bgp_table)
# Learned Routes table
routes_table = Table(title="Learned Routes (from GCP Cloud Router)")
routes_table.add_column("Destination", style="cyan")
routes_table.add_column("Next Hop", style="white")
routes_table.add_column("Priority", style="yellow")
for route in routes:
routes_table.add_row(
route["destination"],
route["next_hop"],
str(route["priority"]),
)
console.print(routes_table)
# Summary
total_tunnels = len(aws_tunnels) + len(gcp_tunnels)
up_tunnels = (
sum(1 for t in aws_tunnels if t["status"] == "UP")
+ sum(1 for t in gcp_tunnels if t["status"] == "ESTABLISHED")
)
console.print(
f"\n[bold]Summary:[/bold] {up_tunnels}/{total_tunnels} tunnels healthy"
)
if __name__ == "__main__":
project_id = os.environ.get("GCP_PROJECT_ID", "your-gcp-project-id")
console.print("[bold]Validating cross-cloud VPN connectivity...[/bold]\n")
# Check AWS side
ec2_client = get_aws_client("ec2")
aws_tunnels = check_aws_vpn_status(ec2_client)
# Check GCP side
gcp_tunnels, bgp_peers = check_gcp_vpn_status(project_id)
# Check learned routes
routes = check_learned_routes(project_id)
# Display formatted health report
display_health_report(aws_tunnels, gcp_tunnels, bgp_peers, routes)What does this script do?
The script queries both clouds in parallel. On AWS, it calls describe_vpn_connections to get VgwTelemetry (tunnel UP/DOWN status and accepted route counts). On GCP, it queries the VPN tunnel status and the Cloud Router's BGP peer status plus learned routes.
The rich library formats everything into color-coded tables. Green means healthy, red means down. The summary line gives you a quick glance at overall tunnel health.
- Save scripts/validate_networking.py.
- Set your GCP project ID, then run the validation script:
export GCP_PROJECT_ID=[[GCP_PROJECT_ID="your-gcp-project-id"]]
cd scripts
python validate_networking.pyYou should see four tables: AWS VPN Tunnel Status, GCP VPN Tunnel Status, BGP Session State, and Learned Routes. The summary should report all tunnels healthy.
Seeing tunnels in DOWN state?
VPN tunnels can take 1-2 minutes to establish after Terraform finishes. Wait a moment and re-run the script. If tunnels remain down, verify that both AWS and GCP regions match your configuration and that the pre-shared keys are consistent between both sides.
Help me debug my cross-cloud VPN tunnel connectivity issue.
\u2714\uFE0F Awesome, I've got everything!
Great. Double check you have saved all files in modules/networking/ and that your Terraform apply succeeded with all resources created.
\u24E7 I'd like to double check the full code
Here is the complete modules/networking/variables.tf:
variable "aws_vpc_cidr" {
description = "CIDR block for the AWS VPC"
type = string
default = "10.0.0.0/16"
}
variable "gcp_vpc_cidr" {
description = "CIDR block for the GCP VPC subnet"
type = string
default = "172.16.0.0/16"
}
variable "aws_bgp_asn" {
description = "BGP ASN for the AWS Virtual Private Gateway"
type = number
default = 64512
}
variable "gcp_bgp_asn" {
description = "BGP ASN for the GCP Cloud Router"
type = number
default = 65001
}
variable "tunnel_count" {
description = "Number of VPN tunnels (4 for 99.99% SLA)"
type = number
default = 4
}Here is the complete modules/networking/main.tf:
# --- AWS VPC and Private Subnets ---
resource "aws_vpc" "main" {
cidr_block = var.aws_vpc_cidr
enable_dns_support = true
enable_dns_hostnames = true
tags = {
Name = "zero-trust-vpc"
}
}
resource "aws_subnet" "private_a" {
vpc_id = aws_vpc.main.id
cidr_block = "10.0.1.0/24"
availability_zone = "us-east-1a"
tags = {
Name = "private-subnet-a"
}
}
resource "aws_subnet" "private_b" {
vpc_id = aws_vpc.main.id
cidr_block = "10.0.2.0/24"
availability_zone = "us-east-1b"
tags = {
Name = "private-subnet-b"
}
}
# --- GCP VPC and Custom Subnet ---
resource "google_compute_network" "main" {
name = "zero-trust-vpc"
auto_create_subnetworks = false
}
resource "google_compute_subnetwork" "private" {
name = "private-subnet"
ip_cidr_range = var.gcp_vpc_cidr
region = "us-central1"
network = google_compute_network.main.id
}
# --- AWS Virtual Private Gateway ---
resource "aws_vpn_gateway" "main" {
vpc_id = aws_vpc.main.id
amazon_side_asn = var.aws_bgp_asn
tags = {
Name = "zero-trust-vgw"
}
}
# Pre-shared keys for VPN tunnels (generated, not hardcoded)
resource "random_password" "tunnel_psk" {
count = var.tunnel_count
length = 32
special = false
}
# Customer Gateways represent GCP HA VPN interfaces in AWS
resource "aws_customer_gateway" "gcp" {
count = 2
bgp_asn = var.gcp_bgp_asn
ip_address = google_compute_ha_vpn_gateway.main.vpn_interfaces[count.index].ip_address
type = "ipsec.1"
tags = {
Name = "gcp-cgw-${count.index}"
}
}
# VPN Connections (2 connections x 2 tunnels each = 4 tunnels total)
resource "aws_vpn_connection" "gcp" {
count = 2
vpn_gateway_id = aws_vpn_gateway.main.id
customer_gateway_id = aws_customer_gateway.gcp[count.index].id
type = "ipsec.1"
static_routes_only = false
# IKEv2 with single cipher set to prevent IP fragmentation
tunnel1_ike_versions = ["ikev2"]
tunnel1_phase1_encryption_algorithms = ["AES256"]
tunnel1_phase1_integrity_algorithms = ["SHA2-256"]
tunnel1_phase1_dh_group_numbers = [14]
tunnel1_phase2_encryption_algorithms = ["AES256"]
tunnel1_phase2_integrity_algorithms = ["SHA2-256"]
tunnel1_phase2_dh_group_numbers = [14]
tunnel1_preshared_key = random_password.tunnel_psk[count.index * 2].result
tunnel2_ike_versions = ["ikev2"]
tunnel2_phase1_encryption_algorithms = ["AES256"]
tunnel2_phase1_integrity_algorithms = ["SHA2-256"]
tunnel2_phase1_dh_group_numbers = [14]
tunnel2_phase2_encryption_algorithms = ["AES256"]
tunnel2_phase2_integrity_algorithms = ["SHA2-256"]
tunnel2_phase2_dh_group_numbers = [14]
tunnel2_preshared_key = random_password.tunnel_psk[count.index * 2 + 1].result
tags = {
Name = "gcp-vpn-${count.index}"
}
}
# --- GCP HA VPN Gateway ---
resource "google_compute_ha_vpn_gateway" "main" {
name = "zero-trust-ha-vpn"
region = "us-central1"
network = google_compute_network.main.id
}
# External VPN Gateway represents AWS endpoints in GCP
resource "google_compute_external_vpn_gateway" "aws" {
name = "aws-vpn-gateway"
redundancy_type = "FOUR_IPS_REDUNDANCY"
interface {
id = 0
ip_address = aws_vpn_connection.gcp[0].tunnel1_address
}
interface {
id = 1
ip_address = aws_vpn_connection.gcp[0].tunnel2_address
}
interface {
id = 2
ip_address = aws_vpn_connection.gcp[1].tunnel1_address
}
interface {
id = 3
ip_address = aws_vpn_connection.gcp[1].tunnel2_address
}
}
# Cloud Router for BGP dynamic routing
resource "google_compute_router" "main" {
name = "zero-trust-router"
region = "us-central1"
network = google_compute_network.main.id
bgp {
asn = var.gcp_bgp_asn
}
}
# VPN Tunnels (4 total for 99.99% SLA)
resource "google_compute_vpn_tunnel" "aws" {
count = var.tunnel_count
name = "aws-tunnel-${count.index}"
region = "us-central1"
vpn_gateway = google_compute_ha_vpn_gateway.main.id
vpn_gateway_interface = count.index < 2 ? 0 : 1
peer_external_gateway = google_compute_external_vpn_gateway.aws.id
peer_external_gateway_interface = count.index
shared_secret = random_password.tunnel_psk[count.index].result
router = google_compute_router.main.id
ike_version = 2
}
# Router Interfaces for BGP sessions
resource "google_compute_router_interface" "aws" {
count = var.tunnel_count
name = "aws-interface-${count.index}"
router = google_compute_router.main.name
region = "us-central1"
ip_range = "169.254.${count.index + 1}.1/30"
vpn_tunnel = google_compute_vpn_tunnel.aws[count.index].name
}
# BGP Peers for dynamic route exchange
resource "google_compute_router_peer" "aws" {
count = var.tunnel_count
name = "aws-peer-${count.index}"
router = google_compute_router.main.name
region = "us-central1"
peer_ip_address = "169.254.${count.index + 1}.2"
peer_asn = var.aws_bgp_asn
advertised_route_priority = 100
interface = google_compute_router_interface.aws[count.index].name
}Here is the complete modules/networking/outputs.tf:
output "aws_vpc_id" {
description = "ID of the AWS VPC"
value = aws_vpc.main.id
}
output "gcp_vpc_name" {
description = "Name of the GCP VPC network"
value = google_compute_network.main.name
}
output "vpn_tunnel_ids" {
description = "IDs of the GCP VPN tunnels"
value = google_compute_vpn_tunnel.aws[*].id
}
output "cloud_router_name" {
description = "Name of the GCP Cloud Router"
value = google_compute_router.main.name
}Here is the complete scripts/validate_networking.py:
#!/usr/bin/env python3
"""Cross-cloud VPN connectivity validation script.
Checks VPN tunnel status, BGP session state, and learned routes
from both AWS and GCP, outputting a formatted health report."""
import os
import boto3
from google.cloud import compute_v1
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
from utils.logging_setup import get_logger
from utils.aws_client import get_aws_client
from utils.gcp_client import get_gcp_client
logger = get_logger(__name__)
console = Console()
def check_aws_vpn_status(ec2_client):
"""Check AWS VPN connection status and tunnel telemetry."""
response = ec2_client.describe_vpn_connections(
Filters=[{"Name": "tag:Name", "Values": ["gcp-vpn-*"]}]
)
tunnels = []
for connection in response.get("VpnConnections", []):
for tunnel in connection.get("VgwTelemetry", []):
tunnels.append({
"connection_id": connection["VpnConnectionId"],
"outside_ip": tunnel["OutsideIpAddress"],
"status": tunnel["Status"],
"status_message": tunnel.get("StatusMessage", ""),
"accepted_routes": tunnel.get("AcceptedRouteCount", 0),
})
return tunnels
def check_gcp_vpn_status(project_id, region="us-central1"):
"""Check GCP VPN tunnel status and BGP session state."""
tunnels_client = compute_v1.VpnTunnelsClient()
routers_client = compute_v1.RoutersClient()
# Get VPN tunnel statuses
tunnels = []
request = compute_v1.ListVpnTunnelsRequest(
project=project_id, region=region
)
for tunnel in tunnels_client.list(request=request):
if "aws-tunnel" in tunnel.name:
tunnels.append({
"name": tunnel.name,
"status": tunnel.status,
"peer_ip": tunnel.peer_ip,
"ike_version": tunnel.ike_version,
})
# Get BGP session status from Cloud Router
router_status = routers_client.get_router_status(
project=project_id,
region=region,
router="zero-trust-router",
)
bgp_peers = []
for peer in router_status.result.bgp_peer_status:
bgp_peers.append({
"name": peer.name,
"state": peer.status,
"peer_ip": peer.ip_address,
"learned_routes": len(peer.advertised_routes),
})
return tunnels, bgp_peers
def check_learned_routes(project_id, region="us-central1"):
"""Query learned routes from the GCP Cloud Router."""
routers_client = compute_v1.RoutersClient()
router_status = routers_client.get_router_status(
project=project_id,
region=region,
router="zero-trust-router",
)
routes = []
for route in router_status.result.best_routes:
routes.append({
"destination": route.dest_range,
"next_hop": route.next_hop_ip,
"priority": route.priority,
})
return routes
def display_health_report(aws_tunnels, gcp_tunnels, bgp_peers, routes):
"""Display a formatted health report using rich tables."""
console.print(Panel("Cross-Cloud VPN Health Report", style="bold blue"))
# AWS VPN Tunnels table
aws_table = Table(title="AWS VPN Tunnel Status")
aws_table.add_column("Connection ID", style="cyan")
aws_table.add_column("Outside IP", style="white")
aws_table.add_column("Status", style="green")
aws_table.add_column("Accepted Routes", style="yellow")
for tunnel in aws_tunnels:
status_style = "green" if tunnel["status"] == "UP" else "red"
aws_table.add_row(
tunnel["connection_id"],
tunnel["outside_ip"],
f"[{status_style}]{tunnel['status']}[/{status_style}]",
str(tunnel["accepted_routes"]),
)
console.print(aws_table)
# GCP VPN Tunnels table
gcp_table = Table(title="GCP VPN Tunnel Status")
gcp_table.add_column("Tunnel Name", style="cyan")
gcp_table.add_column("Peer IP", style="white")
gcp_table.add_column("Status", style="green")
gcp_table.add_column("IKE Version", style="yellow")
for tunnel in gcp_tunnels:
status_style = "green" if tunnel["status"] == "ESTABLISHED" else "red"
gcp_table.add_row(
tunnel["name"],
tunnel["peer_ip"],
f"[{status_style}]{tunnel['status']}[/{status_style}]",
str(tunnel["ike_version"]),
)
console.print(gcp_table)
# BGP Peers table
bgp_table = Table(title="BGP Session State")
bgp_table.add_column("Peer Name", style="cyan")
bgp_table.add_column("Peer IP", style="white")
bgp_table.add_column("State", style="green")
bgp_table.add_column("Learned Routes", style="yellow")
for peer in bgp_peers:
state_style = "green" if peer["state"] == "UP" else "red"
bgp_table.add_row(
peer["name"],
peer["peer_ip"],
f"[{state_style}]{peer['state']}[/{state_style}]",
str(peer["learned_routes"]),
)
console.print(bgp_table)
# Learned Routes table
routes_table = Table(title="Learned Routes (from GCP Cloud Router)")
routes_table.add_column("Destination", style="cyan")
routes_table.add_column("Next Hop", style="white")
routes_table.add_column("Priority", style="yellow")
for route in routes:
routes_table.add_row(
route["destination"],
route["next_hop"],
str(route["priority"]),
)
console.print(routes_table)
# Summary
total_tunnels = len(aws_tunnels) + len(gcp_tunnels)
up_tunnels = (
sum(1 for t in aws_tunnels if t["status"] == "UP")
+ sum(1 for t in gcp_tunnels if t["status"] == "ESTABLISHED")
)
console.print(
f"\n[bold]Summary:[/bold] {up_tunnels}/{total_tunnels} tunnels healthy"
)
if __name__ == "__main__":
project_id = os.environ.get("GCP_PROJECT_ID", "your-gcp-project-id")
console.print("[bold]Validating cross-cloud VPN connectivity...[/bold]\n")
# Check AWS side
ec2_client = get_aws_client("ec2")
aws_tunnels = check_aws_vpn_status(ec2_client)
# Check GCP side
gcp_tunnels, bgp_peers = check_gcp_vpn_status(project_id)
# Check learned routes
routes = check_learned_routes(project_id)
# Display formatted health report
display_health_report(aws_tunnels, gcp_tunnels, bgp_peers, routes)Your cross-cloud private network is live. AWS and GCP are communicating over encrypted IPsec tunnels with BGP dynamically exchanging routes. Next up, you will build bidirectional workload identity federation so services in either cloud can authenticate to the other without any static credentials.
Implement Bidirectional Workload Identity Federation
Your VPN tunnels are up and BGP routes are flowing between AWS and GCP. The next challenge is authentication.
In a traditional setup, workloads authenticate to other clouds using static credentials like service account keys or long-lived access keys. These credentials are risky because they never expire and can be leaked. In this step, you will implement Workload Identity Federation to eliminate static credentials entirely. AWS workloads will authenticate to GCP by exchanging short-lived STS tokens, and GCP workloads will assume AWS roles via OIDC federation.
In this step, get ready to:
- Create a GCP Workload Identity Pool and Provider that trusts AWS as an OIDC issuer with attribute conditions scoped to a specific role ARN.
- Deploy a Python Lambda function and a Python Cloud Function that authenticate across clouds without static credentials.
- Write and run a Python integration test that validates the full token exchange flow with timing metrics.
Create the GCP Workload Identity Pool and Provider
Workload Identity Federation lets external identities (like AWS IAM roles) obtain short-lived GCP access tokens without service account keys. You configure a Pool to manage these external identities, and a Provider that defines the trust relationship with AWS.
- Create the identity module directory structure by running these commands:
mkdir -p modules/identity
touch modules/identity/variables.tf modules/identity/main.tf modules/identity/outputs.tf- Confirm the files exist by running:
ls modules/identity/You should see variables.tf, main.tf, and outputs.tf listed.
- Open modules/identity/variables.tf and add the input variables for the identity module:
variable "aws_account_id" {
description = "The AWS account ID allowed to federate"
type = string
}
variable "aws_role_arn" {
description = "The AWS IAM role ARN allowed to authenticate via WIF"
type = string
}
variable "gcp_project_id" {
description = "The GCP project ID for WIF resources"
type = string
}
variable "wif_pool_id" {
description = "The Workload Identity Pool ID"
type = string
default = "aws-wif-pool"
}
variable "wif_provider_id" {
description = "The Workload Identity Pool Provider ID"
type = string
default = "aws-wif-provider"
}What do these variables control?
The aws_account_id scopes federation to your specific AWS account. The aws_role_arn further restricts which IAM role can exchange tokens. This layered approach follows the principle of least privilege for cross-cloud trust.
- Open modules/identity/main.tf and add the WIF Pool, Provider, IAM role, and validation buckets:
# --- GCP Workload Identity Federation ---
# Pool to manage external AWS identities
resource "google_iam_workload_identity_pool" "aws_pool" {
workload_identity_pool_id = var.wif_pool_id
display_name = "AWS WIF Pool"
description = "Pool for AWS workload identity federation"
disabled = false
}
# Provider trusting AWS as identity source
resource "google_iam_workload_identity_pool_provider" "aws_provider" {
workload_identity_pool_id = google_iam_workload_identity_pool.aws_pool.workload_identity_pool_id
workload_identity_pool_provider_id = var.wif_provider_id
display_name = "AWS WIF Provider"
description = "Trusts AWS IAM roles as identity provider"
# Map AWS credential fields to Google Cloud attributes
attribute_mapping = {
"google.subject" = "assertion.arn"
"attribute.aws_account" = "assertion.account"
"attribute.aws_role" = "assertion.arn.contains('assumed-role') ? assertion.arn.extract('{account_arn}assumed-role/') + 'assumed-role/' + assertion.arn.extract('assumed-role/{role_name}/') : assertion.arn"
}
# Only allow the specific AWS role to federate
attribute_condition = "attribute.aws_role == '${var.aws_role_arn}'"
aws {
account_id = var.aws_account_id
}
}
# Service account for AWS workloads to impersonate
resource "google_service_account" "wif_sa" {
account_id = "aws-wif-sa"
display_name = "AWS WIF Service Account"
description = "Service account for federated AWS workloads"
}
# Grant the federated identity permission to impersonate the SA
resource "google_service_account_iam_binding" "wif_sa_binding" {
service_account_id = google_service_account.wif_sa.name
role = "roles/iam.workloadIdentityUser"
members = [
"principalSet://iam.googleapis.com/${google_iam_workload_identity_pool.aws_pool.name}/attribute.aws_role/${var.aws_role_arn}"
]
}
# --- Cross-Cloud Validation Buckets ---
# GCS bucket for AWS-to-GCP validation
resource "google_storage_bucket" "wif_validation" {
name = "${var.gcp_project_id}-wif-validation"
location = "US"
force_destroy = true
uniform_bucket_level_access = true
}
# Grant the WIF service account read access to the bucket
resource "google_storage_bucket_iam_member" "wif_sa_reader" {
bucket = google_storage_bucket.wif_validation.name
role = "roles/storage.objectViewer"
member = "serviceAccount:${google_service_account.wif_sa.email}"
}
# Upload a test object for validation
resource "google_storage_bucket_object" "test_object" {
name = "wif-test.txt"
bucket = google_storage_bucket.wif_validation.name
content = "Cross-cloud WIF validation successful"
}
# --- AWS IAM Role for GCP-to-AWS Federation ---
# OIDC provider in AWS for GCP tokens
resource "aws_iam_openid_connect_provider" "gcp_oidc" {
url = "https://accounts.google.com"
client_id_list = ["sts.amazonaws.com"]
thumbprint_list = []
}
# IAM role that GCP workloads can assume
resource "aws_iam_role" "gcp_wif_role" {
name = "gcp-wif-cross-cloud-role"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Principal = {
Federated = aws_iam_openid_connect_provider.gcp_oidc.arn
}
Action = "sts:AssumeRoleWithWebIdentity"
Condition = {
StringEquals = {
"accounts.google.com:sub" = google_service_account.wif_sa.unique_id
}
}
}]
})
}
# S3 bucket for GCP-to-AWS validation
resource "aws_s3_bucket" "wif_validation" {
bucket = "${var.gcp_project_id}-wif-validation-aws"
force_destroy = true
}
# Policy granting the GCP role read access to S3
resource "aws_iam_role_policy" "gcp_wif_s3_read" {
name = "gcp-wif-s3-read"
role = aws_iam_role.gcp_wif_role.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = ["s3:GetObject", "s3:ListBucket"]
Resource = [
aws_s3_bucket.wif_validation.arn,
"${aws_s3_bucket.wif_validation.arn}/*"
]
}]
})
}
# Upload a test object to S3
resource "aws_s3_object" "test_object" {
bucket = aws_s3_bucket.wif_validation.id
key = "wif-test.txt"
content = "Cross-cloud WIF validation from GCP successful"
}What does this Terraform module do?
- The google_iam_workload_identity_pool creates a container for external identities in GCP.
- The google_iam_workload_identity_pool_provider configures AWS as a trusted identity source, mapping assertion.arn to google.subject and extracting the AWS role into a custom attribute.
- The attribute_condition uses a CEL expression that only accepts tokens from your specific AWS role ARN.
- On the AWS side, an OIDC provider trusts Google accounts, and an IAM role restricts assumption to the specific GCP service account's unique ID.
- Both validation buckets contain test objects for end-to-end verification.
- Open modules/identity/outputs.tf and add the module outputs:
output "wif_pool_name" {
description = "The full resource name of the WIF pool"
value = google_iam_workload_identity_pool.aws_pool.name
}
output "wif_provider_name" {
description = "The full resource name of the WIF provider"
value = google_iam_workload_identity_pool_provider.aws_provider.name
}
output "aws_wif_role_arn" {
description = "The ARN of the AWS role for GCP-to-AWS federation"
value = aws_iam_role.gcp_wif_role.arn
}
output "gcs_bucket_name" {
description = "The GCS bucket name for cross-cloud validation"
value = google_storage_bucket.wif_validation.name
}
output "s3_bucket_name" {
description = "The S3 bucket name for cross-cloud validation"
value = aws_s3_bucket.wif_validation.id
}- Add the identity module to your root main.tf by adding this module block below your existing networking module:
module "identity" {
source = "./modules/identity"
aws_account_id = var.aws_account_id
aws_role_arn = var.aws_role_arn
gcp_project_id = var.gcp_project_id
wif_pool_id = "aws-wif-pool"
wif_provider_id = "aws-wif-provider"
}- Deploy the identity infrastructure by running:
terraform init
terraform plan -out=identity.tfplan
terraform apply identity.tfplanThe plan should show the WIF pool, provider, service account, IAM bindings, OIDC provider, IAM role, and both validation buckets being created.
Seeing permission errors?
- Ensure your GCP account has the roles/iam.workloadIdentityPoolAdmin role to create WIF pools.
- Verify your AWS credentials have permission to create IAM roles and OIDC providers.
- WIF Pools are always created in the global location. If you see a location error, confirm you are not specifying a region for these resources.
Help me troubleshoot WIF pool creation errors
Why Workload Identity Federation instead of service account keys?
Service account keys are static, long-lived credentials that must be rotated manually and can be leaked in code repositories or logs. WIF tokens are short-lived (up to 1 hour), automatically rotated, and never stored on disk.
This is the production pattern used by enterprise organizations to eliminate credential-related breaches across multi-cloud environments.
Deploy Python Lambda and Cloud Function for cross-cloud authentication
With the identity infrastructure in place, you will now write the Python functions that use it. The Lambda function authenticates from AWS to GCP using WIF, and the Cloud Function authenticates from GCP to AWS using sts:AssumeRoleWithWebIdentity.
- Create the Lambda function directory by running:
mkdir -p scripts/lambda/wif_to_gcp- Create scripts/lambda/wif_to_gcp/lambda_function.py with the following code:
"""AWS Lambda: Authenticate to GCP via Workload Identity Federation.
This function uses the Lambda execution role's identity to obtain
a short-lived GCP access token via WIF, then reads from a GCS bucket.
Zero static credentials are used.
"""
import json
import time
import boto3
from google.auth import _default as google_auth_default
from google.auth.transport.requests import Request
from google.cloud import storage
from google.auth import external_account
def get_wif_credentials(wif_provider_name, service_account_email):
"""Exchange AWS STS token for GCP access token via WIF."""
# Build credential configuration for AWS-based external account
credential_config = {
"type": "external_account",
"audience": f"//iam.googleapis.com/{wif_provider_name}",
"subject_token_type": "urn:ietf:params:aws:token-type:aws4_request",
"token_url": "https://sts.googleapis.com/v1/token",
"credential_source": {
"environment_id": "aws1",
"region_url": "http://169.254.169.254/latest/meta-data/placement/availability-zone",
"url": "http://169.254.169.254/latest/meta-data/iam/security-credentials",
"regional_cred_verification_url": "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15"
},
"service_account_impersonation_url": f"https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{service_account_email}:generateAccessToken"
}
# Create credentials from the configuration
credentials = external_account.Credentials.from_info(credential_config)
credentials.refresh(Request())
return credentials
def handler(event, context):
"""Lambda handler: authenticate to GCP and read from GCS."""
start_time = time.time()
# Configuration from environment variables
wif_provider_name = event.get("wif_provider_name")
service_account_email = event.get("service_account_email")
gcs_bucket_name = event.get("gcs_bucket_name")
object_key = event.get("object_key", "wif-test.txt")
results = {
"function": "wif_to_gcp",
"steps": [],
"success": False
}
try:
# Step 1: Exchange AWS identity for GCP token
token_start = time.time()
credentials = get_wif_credentials(wif_provider_name, service_account_email)
token_duration = time.time() - token_start
results["steps"].append({
"step": "token_exchange",
"duration_ms": round(token_duration * 1000, 2),
"status": "success"
})
# Step 2: Read from GCS using federated credentials
read_start = time.time()
client = storage.Client(credentials=credentials)
bucket = client.bucket(gcs_bucket_name)
blob = bucket.blob(object_key)
content = blob.download_as_text()
read_duration = time.time() - read_start
results["steps"].append({
"step": "gcs_read",
"duration_ms": round(read_duration * 1000, 2),
"status": "success",
"content_length": len(content)
})
results["success"] = True
results["total_duration_ms"] = round((time.time() - start_time) * 1000, 2)
except Exception as e:
results["error"] = str(e)
results["total_duration_ms"] = round((time.time() - start_time) * 1000, 2)
return {
"statusCode": 200 if results["success"] else 500,
"body": json.dumps(results, indent=2)
}What does this Lambda do?
- It builds an external_account credential configuration that tells the Google Auth library to exchange the Lambda's AWS identity for a GCP token.
- The token exchange happens via the Google Security Token Service (STS) endpoint.
- Once authenticated, it reads a test object from GCS to prove cross-cloud access works.
- Every step is timed so the integration test can report token exchange latency.
- Create the Cloud Function directory by running:
mkdir -p scripts/cloud_functions/wif_to_aws- Create scripts/cloud_functions/wif_to_aws/main.py with the following code:
"""GCP Cloud Function: Authenticate to AWS via STS AssumeRoleWithWebIdentity.
This function uses its GCP service account identity to obtain
temporary AWS credentials, then reads from an S3 bucket.
Zero static credentials are used.
"""
import json
import time
import boto3
import google.auth
from google.auth.transport.requests import Request
def get_gcp_id_token(target_audience):
"""Get a GCP ID token for the current service account."""
credentials, project = google.auth.default()
# Request an ID token with the AWS audience
from google.oauth2 import id_token
from google.auth.transport.requests import Request as AuthRequest
token = id_token.fetch_id_token(AuthRequest(), target_audience)
return token
def assume_aws_role(role_arn, id_token_value, session_name="gcp-wif-session"):
"""Assume an AWS role using a GCP ID token via STS."""
sts_client = boto3.client("sts", region_name="us-east-1")
response = sts_client.assume_role_with_web_identity(
RoleArn=role_arn,
RoleSessionName=session_name,
WebIdentityToken=id_token_value,
DurationSeconds=3600
)
return response["Credentials"]
def wif_to_aws(request):
"""Cloud Function entry point: authenticate to AWS and read S3."""
start_time = time.time()
# Parse request parameters
request_json = request.get_json(silent=True) or {}
aws_role_arn = request_json.get("aws_role_arn")
s3_bucket_name = request_json.get("s3_bucket_name")
object_key = request_json.get("object_key", "wif-test.txt")
results = {
"function": "wif_to_aws",
"steps": [],
"success": False
}
try:
# Step 1: Get GCP ID token with AWS audience
token_start = time.time()
target_audience = "sts.amazonaws.com"
id_token_value = get_gcp_id_token(target_audience)
token_duration = time.time() - token_start
results["steps"].append({
"step": "get_id_token",
"duration_ms": round(token_duration * 1000, 2),
"status": "success"
})
# Step 2: Assume AWS role with the GCP ID token
assume_start = time.time()
aws_credentials = assume_aws_role(aws_role_arn, id_token_value)
assume_duration = time.time() - assume_start
results["steps"].append({
"step": "assume_role",
"duration_ms": round(assume_duration * 1000, 2),
"status": "success"
})
# Step 3: Read from S3 using temporary credentials
read_start = time.time()
s3_client = boto3.client(
"s3",
aws_access_key_id=aws_credentials["AccessKeyId"],
aws_secret_access_key=aws_credentials["SecretAccessKey"],
aws_session_token=aws_credentials["SessionToken"]
)
response = s3_client.get_object(Bucket=s3_bucket_name, Key=object_key)
content = response["Body"].read().decode("utf-8")
read_duration = time.time() - read_start
results["steps"].append({
"step": "s3_read",
"duration_ms": round(read_duration * 1000, 2),
"status": "success",
"content_length": len(content)
})
results["success"] = True
results["total_duration_ms"] = round((time.time() - start_time) * 1000, 2)
except Exception as e:
results["error"] = str(e)
results["total_duration_ms"] = round((time.time() - start_time) * 1000, 2)
return (json.dumps(results, indent=2), 200 if results["success"] else 500)What does this Cloud Function do?
- It fetches a GCP ID token targeting the sts.amazonaws.com audience.
- It calls AWS STS AssumeRoleWithWebIdentity to exchange the GCP token for temporary AWS credentials.
- It uses those temporary credentials to read from S3, proving bidirectional federation works.
- Each step is timed independently for the integration test report.
- Save both files.
Module import errors in your editor?
Your editor may show import errors for google.auth or boto3 in the function files. This is expected because these functions run in their respective cloud runtimes (Lambda and Cloud Functions) which provide these packages. The imports will resolve at deploy time.
Help me understand why my imports show errors locally for Lambda functions
Write the Python integration test script
The integration test invokes both functions programmatically, validates that cross-cloud access succeeded, and produces a formatted report with timing metrics. This proves the entire WIF pipeline works end-to-end without any static credentials.
- Create scripts/test_wif_integration.py with the following code:
"""Integration test for bidirectional Workload Identity Federation.
Invokes the Lambda (AWS-to-GCP) and Cloud Function (GCP-to-AWS)
programmatically, validates cross-cloud resource access, and reports
the token exchange flow with timing metrics.
"""
import json
import sys
import time
import boto3
import requests
from rich.console import Console
from rich.table import Table
from rich.panel import Panel
console = Console()
def test_aws_to_gcp(lambda_function_name, wif_provider_name,
service_account_email, gcs_bucket_name):
"""Test AWS Lambda authenticating to GCP via WIF."""
console.print("\n[bold cyan]Test 1: AWS -> GCP (Workload Identity Federation)[/bold cyan]")
# Invoke Lambda function
lambda_client = boto3.client("lambda")
payload = {
"wif_provider_name": wif_provider_name,
"service_account_email": service_account_email,
"gcs_bucket_name": gcs_bucket_name,
"object_key": "wif-test.txt"
}
start_time = time.time()
response = lambda_client.invoke(
FunctionName=lambda_function_name,
InvocationType="RequestResponse",
Payload=json.dumps(payload)
)
invocation_duration = time.time() - start_time
# Parse response
response_payload = json.loads(response["Payload"].read())
body = json.loads(response_payload.get("body", "{}"))
return {
"direction": "AWS -> GCP",
"success": body.get("success", False),
"steps": body.get("steps", []),
"total_duration_ms": body.get("total_duration_ms", 0),
"invocation_duration_ms": round(invocation_duration * 1000, 2),
"error": body.get("error")
}
def test_gcp_to_aws(cloud_function_url, aws_role_arn, s3_bucket_name):
"""Test GCP Cloud Function authenticating to AWS via OIDC."""
console.print("\n[bold cyan]Test 2: GCP -> AWS (OIDC Federation)[/bold cyan]")
payload = {
"aws_role_arn": aws_role_arn,
"s3_bucket_name": s3_bucket_name,
"object_key": "wif-test.txt"
}
start_time = time.time()
response = requests.post(cloud_function_url, json=payload, timeout=30)
invocation_duration = time.time() - start_time
body = response.json()
return {
"direction": "GCP -> AWS",
"success": body.get("success", False),
"steps": body.get("steps", []),
"total_duration_ms": body.get("total_duration_ms", 0),
"invocation_duration_ms": round(invocation_duration * 1000, 2),
"error": body.get("error")
}
def print_results(results):
"""Print formatted test results with timing metrics."""
# Summary table
table = Table(title="WIF Integration Test Results")
table.add_column("Direction", style="cyan")
table.add_column("Status", style="green")
table.add_column("Total (ms)", justify="right")
table.add_column("Invocation (ms)", justify="right")
for result in results:
status = "PASS" if result["success"] else "FAIL"
style = "green" if result["success"] else "red"
table.add_row(
result["direction"],
f"[{style}]{status}[/{style}]",
str(result["total_duration_ms"]),
str(result["invocation_duration_ms"])
)
console.print(table)
# Detailed timing for each step
for result in results:
if result["steps"]:
step_table = Table(title=f"Token Exchange Flow: {result['direction']}")
step_table.add_column("Step", style="cyan")
step_table.add_column("Duration (ms)", justify="right")
step_table.add_column("Status")
for step in result["steps"]:
step_style = "green" if step["status"] == "success" else "red"
step_table.add_row(
step["step"],
str(step["duration_ms"]),
f"[{step_style}]{step['status']}[/{step_style}]"
)
console.print(step_table)
# Final verdict
all_passed = all(r["success"] for r in results)
if all_passed:
console.print(Panel(
"[bold green]All tests passed. Bidirectional WIF is operational."
"\nZero static credentials used.[/bold green]",
title="Result"
))
else:
failed = [r for r in results if not r["success"]]
for f in failed:
console.print(f"[bold red]FAILED: {f['direction']} - {f.get('error')}[/bold red]")
return all_passed
def main():
"""Run the full WIF integration test suite."""
if len(sys.argv) < 6:
console.print("[red]Usage: python test_wif_integration.py "
"<lambda_name> <wif_provider_name> <sa_email> "
"<gcs_bucket> <cloud_function_url> <aws_role_arn> "
"<s3_bucket>[/red]")
sys.exit(1)
lambda_name = sys.argv[1]
wif_provider_name = sys.argv[2]
sa_email = sys.argv[3]
gcs_bucket = sys.argv[4]
cloud_function_url = sys.argv[5]
aws_role_arn = sys.argv[6]
s3_bucket = sys.argv[7]
console.print(Panel(
"[bold]Bidirectional Workload Identity Federation Test[/bold]\n"
"Testing cross-cloud authentication without static credentials",
title="WIF Integration Test"
))
results = []
# Test 1: AWS to GCP
result_1 = test_aws_to_gcp(
lambda_name, wif_provider_name, sa_email, gcs_bucket
)
results.append(result_1)
# Test 2: GCP to AWS
result_2 = test_gcp_to_aws(cloud_function_url, aws_role_arn, s3_bucket)
results.append(result_2)
# Print summary
all_passed = print_results(results)
sys.exit(0 if all_passed else 1)
if __name__ == "__main__":
main()What does this test script do?
- It invokes the Lambda function via boto3 and the Cloud Function via an HTTP POST request.
- Both functions return structured JSON with per-step timing data.
- The script uses the rich library to render formatted tables showing the token exchange flow and latency for each direction.
- A final pass/fail verdict confirms zero static credentials were used in either direction.
- Save scripts/test_wif_integration.py.
- Run the integration test after deploying the Lambda and Cloud Function:
python scripts/test_wif_integration.py \
[[LAMBDA_NAME="your-lambda-function-name"]] \
[[WIF_PROVIDER="projects/PROJECT_NUMBER/locations/global/workloadIdentityPools/aws-wif-pool/providers/aws-wif-provider"]] \
[[SA_EMAIL="aws-wif-sa@your-project.iam.gserviceaccount.com"]] \
[[GCS_BUCKET="your-project-wif-validation"]] \
[[CF_URL="https://your-region-your-project.cloudfunctions.net/wif_to_aws"]] \
[[AWS_ROLE_ARN="arn:aws:iam::123456789012:role/gcp-wif-cross-cloud-role"]] \
[[S3_BUCKET="your-project-wif-validation-aws"]]You should see two tables rendered by rich showing PASS for both directions, with timing breakdowns for each token exchange step.
Test failing for one direction?
- If AWS-to-GCP fails, check the attribute_condition in the WIF provider matches the exact role ARN your Lambda uses.
- If GCP-to-AWS fails, verify the OIDC provider's client_id_list includes sts.amazonaws.com and the IAM role's trust policy matches the service account's unique ID.
- Token exchange can take 2-5 seconds on first invocation due to cold starts. Subsequent invocations are faster.
Help me debug my WIF integration test failures
Your cross-cloud identity layer is now fully operational. AWS workloads authenticate to GCP, and GCP workloads authenticate to AWS, with zero static credentials anywhere in the system. Next up, you will build the threat detection infrastructure that monitors both clouds for security events.
Deploy Threat Detection Infrastructure
Your cross-cloud identity layer is working. Workloads in AWS and GCP can authenticate to each other without static credentials.
Now you need eyes on both clouds. Without active threat detection, compromised workloads could move laterally across the VPN tunnel unnoticed. In this session, you will deploy GuardDuty and Security Hub on AWS, configure GCP Security Command Center notifications, and build a Python simulator to validate the detection pipeline end-to-end.
In this step, get ready to:
- Enable GuardDuty with S3 protection and malware scanning plus Security Hub with CIS and FSBP standards.
- Configure GCP Security Command Center finding notifications exported to a Pub/Sub topic.
- Write a Python GuardDuty finding simulator for end-to-end pipeline testing.
Enable GuardDuty and Security Hub using Terraform
GuardDuty is AWS's managed threat detection service. It continuously monitors CloudTrail, VPC Flow Logs, and DNS queries for suspicious activity.
Security Hub aggregates findings from GuardDuty and other services, then measures them against compliance frameworks. Together they form the AWS side of your detection layer.
- Create the detection module directory by running:
mkdir -p modules/detection- Create the variables file for the detection module by running:
touch modules/detection/variables.tf- Add the following input variables to modules/detection/variables.tf:
# Input variables for the detection module
variable "gcp_project_id" {
description = "GCP project ID for SCC notifications"
type = string
}
variable "aws_region" {
description = "AWS region for Security Hub standards ARNs"
type = string
}What do these variables do?
The gcp_project_id variable scopes the SCC notification config to your GCP project. The aws_region variable constructs region-aware Security Hub standards ARNs.
- Create the main detection module Terraform file by adding the following to modules/detection/main.tf:
# --- AWS GuardDuty Detector ---
# Enables threat detection with S3 protection and malware scanning
resource "aws_guardduty_detector" "main" {
enable = true
datasources {
s3_logs {
enable = true
}
malware_protection {
scan_ec2_instance_with_findings {
ebs_volumes {
enable = true
}
}
}
}
tags = {
Module = "detection"
}
}
# --- AWS Security Hub ---
# Enables Security Hub and subscribes to CIS and FSBP standards
resource "aws_securityhub_account" "main" {}
resource "aws_securityhub_standards_subscription" "cis" {
depends_on = [aws_securityhub_account.main]
standards_arn = "arn:aws:securityhub:::ruleset/cis-aws-foundations-benchmark/v/1.2.0"
}
resource "aws_securityhub_standards_subscription" "fsbp" {
depends_on = [aws_securityhub_account.main]
standards_arn = "arn:aws:securityhub:${var.aws_region}::standards/aws-foundational-security-best-practices/v/1.0.0"
}
# --- GCP Security Command Center Notifications ---
# Pub/Sub topic to receive SCC findings
resource "google_pubsub_topic" "scc_findings" {
name = "scc-findings-topic"
project = var.gcp_project_id
}
# SCC notification config exports findings to Pub/Sub
resource "google_scc_v2_project_notification_config" "main" {
config_id = "zero-trust-scc-export"
project = var.gcp_project_id
location = "global"
pubsub_topic = google_pubsub_topic.scc_findings.id
streaming_config {
filter = "state = \"ACTIVE\""
}
}
# --- EventBridge Rule for GuardDuty Findings ---
# Routes GuardDuty findings to downstream targets
resource "aws_cloudwatch_event_rule" "guardduty_findings" {
name = "guardduty-findings-rule"
description = "Route GuardDuty findings to orchestration targets"
event_pattern = jsonencode({
source = ["aws.guardduty"]
detail-type = ["GuardDuty Finding"]
})
tags = {
Module = "detection"
}
}What does this code do?
The aws_guardduty_detector resource enables GuardDuty with S3 data event monitoring and EBS malware scanning for EC2 instances with findings.
The aws_securityhub_account resource activates Security Hub. The two aws_securityhub_standards_subscription resources subscribe to CIS AWS Foundations Benchmark and AWS Foundational Security Best Practices.
On the GCP side, google_scc_v2_project_notification_config streams active SCC findings to a Pub/Sub topic. The aws_cloudwatch_event_rule captures all GuardDuty findings from EventBridge so they can be routed to the orchestration layer you will build next.
- Create the outputs file by adding the following to modules/detection/outputs.tf:
# Detection module outputs
output "guardduty_detector_id" {
description = "GuardDuty detector ID"
value = aws_guardduty_detector.main.id
}
output "securityhub_arn" {
description = "Security Hub account ARN"
value = aws_securityhub_account.main.id
}
output "pubsub_topic_name" {
description = "GCP Pub/Sub topic for SCC findings"
value = google_pubsub_topic.scc_findings.name
}
output "eventbridge_rule_arn" {
description = "EventBridge rule ARN for GuardDuty findings"
value = aws_cloudwatch_event_rule.guardduty_findings.arn
}- Add the detection module to your root main.tf. Open main.tf and add this module block below the existing identity module:
module "detection" {
source = "./modules/detection"
gcp_project_id = var.gcp_project_id
aws_region = var.aws_region
}- Save all files and validate the Terraform configuration by running:
terraform validate- You should see Success! The configuration is valid. in the output.
Seeing a validation error?
Check that you have declared var.gcp_project_id and var.aws_region in your root variables.tf. Also confirm the module source path matches your directory structure exactly.
Help me debug my Terraform validation error for the detection module.
GuardDuty has a 30-day free trial
GuardDuty is free for the first 30 days. After that, charges apply based on the volume of data analyzed. You can disable it before the trial ends to avoid costs.
- Deploy the detection infrastructure by running:
terraform apply -target=module.detection- Type yes when prompted to confirm the apply.
- You should see the GuardDuty detector ID and EventBridge rule ARN in the outputs.
Apply failing on Security Hub?
If Security Hub is already enabled in your account, you may see a conflict error. Import the existing resource with terraform import or remove the aws_securityhub_account resource and add depends_on directly to the standards subscriptions.
Help me resolve my Security Hub Terraform conflict.
Write a Python GuardDuty finding simulator
Real threats are unpredictable. To test your detection pipeline without waiting for actual malicious activity, you need a way to generate findings on demand.
The CreateSampleFindings API in GuardDuty generates realistic sample findings that flow through EventBridge just like real ones.
- Create the simulator script by adding the following to scripts/simulate_guardduty_findings.py:
"""GuardDuty Finding Simulator
Uses the CreateSampleFindings API and a custom EventBridge test event
publisher for end-to-end detection pipeline testing.
"""
import json
import sys
import boto3
from rich.console import Console
from rich.table import Table
console = Console()
# GuardDuty finding types to simulate
SAMPLE_FINDING_TYPES = [
"Recon:EC2/PortProbeUnprotectedPort",
"UnauthorizedAccess:EC2/SSHBruteForce",
"Trojan:EC2/BlackholeTraffic",
"Stealth:S3/ServerAccessLoggingDisabled",
]
def get_detector_id(guardduty_client):
"""Retrieve the GuardDuty detector ID for the current region."""
response = guardduty_client.list_detectors()
detector_ids = response.get("DetectorIds", [])
if not detector_ids:
console.print("[red]No GuardDuty detector found. Deploy the detection module first.[/red]")
sys.exit(1)
return detector_ids[0]
def create_sample_findings(guardduty_client, detector_id):
"""Generate sample findings using the CreateSampleFindings API."""
console.print(f"\n[bold]Creating sample findings for detector: {detector_id}[/bold]")
guardduty_client.create_sample_findings(
DetectorId=detector_id,
FindingTypes=SAMPLE_FINDING_TYPES
)
console.print("[green]Sample findings created successfully.[/green]")
return SAMPLE_FINDING_TYPES
def publish_custom_eventbridge_event(events_client):
"""Publish a custom GuardDuty-like event to EventBridge for testing."""
# Custom test event simulating a high-severity finding
custom_event = {
"version": "0",
"source": "aws.guardduty",
"detail-type": "GuardDuty Finding",
"detail": {
"schemaVersion": "2.0",
"type": "UnauthorizedAccess:IAMUser/MaliciousIPCaller.Custom",
"severity": 8,
"title": "API was invoked from a known malicious IP address.",
"description": "Test event for pipeline validation.",
}
}
console.print("\n[bold]Publishing custom EventBridge test event...[/bold]")
response = events_client.put_events(
Entries=[
{
"Source": "aws.guardduty",
"DetailType": "GuardDuty Finding",
"Detail": json.dumps(custom_event["detail"]),
}
]
)
failed = response.get("FailedEntryCount", 0)
if failed > 0:
console.print(f"[red]Failed to publish {failed} event(s).[/red]")
else:
console.print("[green]Custom EventBridge event published successfully.[/green]")
return response
def display_summary(finding_types):
"""Display a formatted summary of simulated findings."""
table = Table(title="Simulated GuardDuty Findings")
table.add_column("Finding Type", style="cyan")
table.add_column("Status", style="green")
for finding_type in finding_types:
table.add_row(finding_type, "Created")
# Add the custom EventBridge event row
table.add_row("UnauthorizedAccess:IAMUser/MaliciousIPCaller.Custom", "Published")
console.print(table)
if __name__ == "__main__":
console.print("[bold blue]GuardDuty Finding Simulator[/bold blue]")
console.print("Generating sample findings for end-to-end pipeline testing.\n")
# Initialize AWS clients
guardduty_client = boto3.client("guardduty")
events_client = boto3.client("events")
# Step 1: Get detector ID
detector_id = get_detector_id(guardduty_client)
# Step 2: Create sample findings via GuardDuty API
finding_types = create_sample_findings(guardduty_client, detector_id)
# Step 3: Publish custom EventBridge event
publish_custom_eventbridge_event(events_client)
# Step 4: Display summary
display_summary(finding_types)
console.print("\n[bold green]Pipeline test complete.[/bold green]")
console.print("Check CloudWatch Logs and EventBridge to confirm events were routed.")What does this script do?
- It retrieves the GuardDuty detector ID using list_detectors().
- It calls create_sample_findings() with four realistic finding types that GuardDuty publishes to EventBridge.
- It publishes a custom event directly to EventBridge via put_events() for testing downstream routing without waiting for GuardDuty processing delay.
- It displays a formatted summary table using the rich library.
- Save scripts/simulate_guardduty_findings.py and run the simulator:
python scripts/simulate_guardduty_findings.py- You should see a table showing all four sample finding types with status Created and the custom event with status Published.
Seeing "No GuardDuty detector found"?
Make sure terraform apply completed successfully for the detection module. Confirm the detector exists by running aws guardduty list-detectors in your terminal.
Help me debug why my GuardDuty simulator cannot find a detector.
✔️ Awesome, I've got everything!
Great. Double check you have saved all files before moving on.
ⓧ I'd like to double check the full code
Here is the complete detection module and simulator script:
# Input variables for the detection module
variable "gcp_project_id" {
description = "GCP project ID for SCC notifications"
type = string
}
variable "aws_region" {
description = "AWS region for Security Hub standards ARNs"
type = string
}# --- AWS GuardDuty Detector ---
# Enables threat detection with S3 protection and malware scanning
resource "aws_guardduty_detector" "main" {
enable = true
datasources {
s3_logs {
enable = true
}
malware_protection {
scan_ec2_instance_with_findings {
ebs_volumes {
enable = true
}
}
}
}
tags = {
Module = "detection"
}
}
# --- AWS Security Hub ---
# Enables Security Hub and subscribes to CIS and FSBP standards
resource "aws_securityhub_account" "main" {}
resource "aws_securityhub_standards_subscription" "cis" {
depends_on = [aws_securityhub_account.main]
standards_arn = "arn:aws:securityhub:::ruleset/cis-aws-foundations-benchmark/v/1.2.0"
}
resource "aws_securityhub_standards_subscription" "fsbp" {
depends_on = [aws_securityhub_account.main]
standards_arn = "arn:aws:securityhub:${var.aws_region}::standards/aws-foundational-security-best-practices/v/1.0.0"
}
# --- GCP Security Command Center Notifications ---
# Pub/Sub topic to receive SCC findings
resource "google_pubsub_topic" "scc_findings" {
name = "scc-findings-topic"
project = var.gcp_project_id
}
# SCC notification config exports findings to Pub/Sub
resource "google_scc_v2_project_notification_config" "main" {
config_id = "zero-trust-scc-export"
project = var.gcp_project_id
location = "global"
pubsub_topic = google_pubsub_topic.scc_findings.id
streaming_config {
filter = "state = \"ACTIVE\""
}
}
# --- EventBridge Rule for GuardDuty Findings ---
# Routes GuardDuty findings to downstream targets
resource "aws_cloudwatch_event_rule" "guardduty_findings" {
name = "guardduty-findings-rule"
description = "Route GuardDuty findings to orchestration targets"
event_pattern = jsonencode({
source = ["aws.guardduty"]
detail-type = ["GuardDuty Finding"]
})
tags = {
Module = "detection"
}
}# Detection module outputs
output "guardduty_detector_id" {
description = "GuardDuty detector ID"
value = aws_guardduty_detector.main.id
}
output "securityhub_arn" {
description = "Security Hub account ARN"
value = aws_securityhub_account.main.id
}
output "pubsub_topic_name" {
description = "GCP Pub/Sub topic for SCC findings"
value = google_pubsub_topic.scc_findings.name
}
output "eventbridge_rule_arn" {
description = "EventBridge rule ARN for GuardDuty findings"
value = aws_cloudwatch_event_rule.guardduty_findings.arn
}module "detection" {
source = "./modules/detection"
gcp_project_id = var.gcp_project_id
aws_region = var.aws_region
}"""GuardDuty Finding Simulator
Uses the CreateSampleFindings API and a custom EventBridge test event
publisher for end-to-end detection pipeline testing.
"""
import json
import sys
import boto3
from rich.console import Console
from rich.table import Table
console = Console()
# GuardDuty finding types to simulate
SAMPLE_FINDING_TYPES = [
"Recon:EC2/PortProbeUnprotectedPort",
"UnauthorizedAccess:EC2/SSHBruteForce",
"Trojan:EC2/BlackholeTraffic",
"Stealth:S3/ServerAccessLoggingDisabled",
]
def get_detector_id(guardduty_client):
"""Retrieve the GuardDuty detector ID for the current region."""
response = guardduty_client.list_detectors()
detector_ids = response.get("DetectorIds", [])
if not detector_ids:
console.print("[red]No GuardDuty detector found. Deploy the detection module first.[/red]")
sys.exit(1)
return detector_ids[0]
def create_sample_findings(guardduty_client, detector_id):
"""Generate sample findings using the CreateSampleFindings API."""
console.print(f"\n[bold]Creating sample findings for detector: {detector_id}[/bold]")
guardduty_client.create_sample_findings(
DetectorId=detector_id,
FindingTypes=SAMPLE_FINDING_TYPES
)
console.print("[green]Sample findings created successfully.[/green]")
return SAMPLE_FINDING_TYPES
def publish_custom_eventbridge_event(events_client):
"""Publish a custom GuardDuty-like event to EventBridge for testing."""
# Custom test event simulating a high-severity finding
custom_event = {
"version": "0",
"source": "aws.guardduty",
"detail-type": "GuardDuty Finding",
"detail": {
"schemaVersion": "2.0",
"type": "UnauthorizedAccess:IAMUser/MaliciousIPCaller.Custom",
"severity": 8,
"title": "API was invoked from a known malicious IP address.",
"description": "Test event for pipeline validation.",
}
}
console.print("\n[bold]Publishing custom EventBridge test event...[/bold]")
response = events_client.put_events(
Entries=[
{
"Source": "aws.guardduty",
"DetailType": "GuardDuty Finding",
"Detail": json.dumps(custom_event["detail"]),
}
]
)
failed = response.get("FailedEntryCount", 0)
if failed > 0:
console.print(f"[red]Failed to publish {failed} event(s).[/red]")
else:
console.print("[green]Custom EventBridge event published successfully.[/green]")
return response
def display_summary(finding_types):
"""Display a formatted summary of simulated findings."""
table = Table(title="Simulated GuardDuty Findings")
table.add_column("Finding Type", style="cyan")
table.add_column("Status", style="green")
for finding_type in finding_types:
table.add_row(finding_type, "Created")
# Add the custom EventBridge event row
table.add_row("UnauthorizedAccess:IAMUser/MaliciousIPCaller.Custom", "Published")
console.print(table)
if __name__ == "__main__":
console.print("[bold blue]GuardDuty Finding Simulator[/bold blue]")
console.print("Generating sample findings for end-to-end pipeline testing.\n")
# Initialize AWS clients
guardduty_client = boto3.client("guardduty")
events_client = boto3.client("events")
# Step 1: Get detector ID
detector_id = get_detector_id(guardduty_client)
# Step 2: Create sample findings via GuardDuty API
finding_types = create_sample_findings(guardduty_client, detector_id)
# Step 3: Publish custom EventBridge event
publish_custom_eventbridge_event(events_client)
# Step 4: Display summary
display_summary(finding_types)
console.print("\n[bold green]Pipeline test complete.[/bold green]")
console.print("Check CloudWatch Logs and EventBridge to confirm events were routed.")Your detection infrastructure is live. GuardDuty is watching AWS, Security Command Center is watching GCP, and your EventBridge rule is ready to route findings to whatever comes next. In the next session, you will build the Step Functions orchestrator that turns these findings into automated incident response.
Build Security Orchestration with Step Functions
Your detection infrastructure is live. GuardDuty and Security Command Center are watching for threats across both clouds.
Detection without response is just expensive logging. When a threat is detected, your platform needs to automatically isolate compromised resources, preserve forensic evidence, and notify your security team. In this session, you will build an automated incident response orchestrator using Step Functions and Python Lambda functions.
In this step, get ready to:
- Deploy a Step Functions Standard Workflow state machine with ASL JSON defining triage, parallel response branches, audit logging, and resolution.
- Implement five Python Lambda functions for automated incident response actions.
- Create a DynamoDB audit table with a Python CLI query tool for forensic investigation.
Design and deploy the Step Functions state machine
The orchestrator coordinates multi-step incident response using a Standard Workflow. Standard Workflows support the callback pattern for human approval steps and can run for up to a year, making them the right choice for security orchestration.
Why Step Functions Standard over Express
Express Workflows are designed for high-volume, short-duration tasks. They do not support the callback pattern needed for human escalation during incident response.
Standard Workflows support .waitForTaskToken for pausing execution until a human approves an action. You pay per state transition ($0.025 per 1,000) with a free tier of 4,000 transitions per month.
- Create the orchestration module directory by running this command:
mkdir -p modules/orchestration- Create the Lambda function directories for each response action by running:
mkdir -p scripts/lambda/isolate_ec2
mkdir -p scripts/lambda/revoke_iam_keys
mkdir -p scripts/lambda/snapshot_forensics
mkdir -p scripts/lambda/notify_soc
mkdir -p scripts/lambda/audit_logger- Confirm the directories exist by running:
ls modules/orchestration
ls scripts/lambdaYou should see the orchestration directory listed under modules/, and all five Lambda directories listed under scripts/lambda/.
- Create the ASL JSON definition for the state machine by adding the following to modules/orchestration/asl.json:
{
"Comment": "Security incident response orchestrator - triage, parallel containment, audit, resolution",
"StartAt": "Triage",
"States": {
"Triage": {
"Type": "Task",
"Resource": "${IsolateEc2Arn}",
"Comment": "Classify severity and determine response actions",
"ResultPath": "$.triage",
"Next": "ParallelResponse"
},
"ParallelResponse": {
"Type": "Parallel",
"Comment": "Execute containment, forensics, and notification in parallel",
"Branches": [
{
"StartAt": "Containment",
"States": {
"Containment": {
"Type": "Task",
"Resource": "${IsolateEc2Arn}",
"Comment": "Quarantine EC2 and revoke IAM keys",
"End": true
}
}
},
{
"StartAt": "ForensicSnapshot",
"States": {
"ForensicSnapshot": {
"Type": "Task",
"Resource": "${SnapshotForensicsArn}",
"Comment": "Create EBS snapshots for investigation",
"End": true
}
}
},
{
"StartAt": "NotifySOC",
"States": {
"NotifySOC": {
"Type": "Task",
"Resource": "${NotifySocArn}",
"Comment": "Send structured incident details to SOC",
"End": true
}
}
}
],
"ResultPath": "$.parallel_results",
"Next": "AuditLog"
},
"AuditLog": {
"Type": "Task",
"Resource": "${AuditLoggerArn}",
"Comment": "Write incident metadata to DynamoDB audit table",
"ResultPath": "$.audit",
"Next": "Resolution"
},
"Resolution": {
"Type": "Succeed",
"Comment": "Incident response workflow complete"
}
}
}What does this ASL definition do
The Amazon States Language (ASL) JSON defines your workflow as a directed graph of states. Each state has a Type and a Next pointer to the following state.
- The Triage state classifies the incident severity before any action is taken.
- The ParallelResponse state uses the Parallel type to run containment, forensic capture, and SOC notification simultaneously.
- The AuditLog state writes a structured record to DynamoDB after all parallel branches complete.
- The Resolution state marks the workflow as successfully completed.
- The ${...Arn} placeholders are replaced at deploy time by Terraform template substitution.
- Save modules/orchestration/asl.json.
- Now create the Terraform configuration for the orchestration module. Add the following to modules/orchestration/variables.tf:
variable "project_name" {
description = "Project name prefix for resource naming"
type = string
}
variable "aws_region" {
description = "AWS region for deployment"
type = string
}
variable "tags" {
description = "Common tags for all resources"
type = map(string)
default = {}
}- Save modules/orchestration/variables.tf.
- Add the main Terraform resources to modules/orchestration/main.tf:
# DynamoDB audit table for incident tracking
resource "aws_dynamodb_table" "audit_trail" {
name = "${var.project_name}-security-audit"
billing_mode = "PAY_PER_REQUEST"
hash_key = "incident_id"
range_key = "timestamp"
attribute {
name = "incident_id"
type = "S"
}
attribute {
name = "timestamp"
type = "S"
}
attribute {
name = "resource_arn"
type = "S"
}
attribute {
name = "action_type"
type = "S"
}
global_secondary_index {
name = "resource-index"
hash_key = "resource_arn"
range_key = "timestamp"
projection_type = "ALL"
}
global_secondary_index {
name = "action-index"
hash_key = "action_type"
range_key = "timestamp"
projection_type = "ALL"
}
tags = var.tags
}
# SNS topic for SOC notifications
resource "aws_sns_topic" "soc_notifications" {
name = "${var.project_name}-soc-alerts"
tags = var.tags
}
# IAM role for Lambda execution
resource "aws_iam_role" "lambda_execution" {
name = "${var.project_name}-orchestration-lambda"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "lambda.amazonaws.com"
}
}]
})
tags = var.tags
}
# Lambda execution policy - EC2, IAM, EBS, SNS, DynamoDB, CloudWatch Logs
resource "aws_iam_role_policy" "lambda_policy" {
name = "${var.project_name}-lambda-orchestration"
role = aws_iam_role.lambda_execution.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = [
"ec2:ModifyInstanceAttribute",
"ec2:CreateSecurityGroup",
"ec2:AuthorizeSecurityGroupIngress",
"ec2:RevokeSecurityGroupIngress",
"ec2:ModifyNetworkInterfaceAttribute",
"ec2:DescribeInstances",
"ec2:DescribeSecurityGroups",
"ec2:CreateSnapshot",
"ec2:CreateTags",
"ec2:DescribeVolumes"
]
Resource = "*"
},
{
Effect = "Allow"
Action = [
"iam:UpdateAccessKey",
"iam:ListAccessKeys",
"iam:PutUserPolicy"
]
Resource = "*"
},
{
Effect = "Allow"
Action = [
"sns:Publish"
]
Resource = aws_sns_topic.soc_notifications.arn
},
{
Effect = "Allow"
Action = [
"dynamodb:PutItem",
"dynamodb:Query"
]
Resource = [
aws_dynamodb_table.audit_trail.arn,
"${aws_dynamodb_table.audit_trail.arn}/index/*"
]
},
{
Effect = "Allow"
Action = [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
]
Resource = "arn:aws:logs:*:*:*"
}
]
})
}
# Lambda functions for each orchestration step
resource "aws_lambda_function" "isolate_ec2" {
filename = data.archive_file.isolate_ec2.output_path
function_name = "${var.project_name}-isolate-ec2"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.isolate_ec2.output_base64sha256
timeout = 60
tags = var.tags
}
resource "aws_lambda_function" "revoke_iam_keys" {
filename = data.archive_file.revoke_iam_keys.output_path
function_name = "${var.project_name}-revoke-iam-keys"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.revoke_iam_keys.output_base64sha256
timeout = 60
tags = var.tags
}
resource "aws_lambda_function" "snapshot_forensics" {
filename = data.archive_file.snapshot_forensics.output_path
function_name = "${var.project_name}-snapshot-forensics"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.snapshot_forensics.output_base64sha256
timeout = 120
tags = var.tags
}
resource "aws_lambda_function" "notify_soc" {
filename = data.archive_file.notify_soc.output_path
function_name = "${var.project_name}-notify-soc"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.notify_soc.output_base64sha256
timeout = 30
environment {
variables = {
SNS_TOPIC_ARN = aws_sns_topic.soc_notifications.arn
}
}
tags = var.tags
}
resource "aws_lambda_function" "audit_logger" {
filename = data.archive_file.audit_logger.output_path
function_name = "${var.project_name}-audit-logger"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.audit_logger.output_base64sha256
timeout = 30
environment {
variables = {
AUDIT_TABLE_NAME = aws_dynamodb_table.audit_trail.name
}
}
tags = var.tags
}
# Archive files for Lambda deployment packages
data "archive_file" "isolate_ec2" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/isolate_ec2"
output_path = "${path.root}/.build/isolate_ec2.zip"
}
data "archive_file" "revoke_iam_keys" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/revoke_iam_keys"
output_path = "${path.root}/.build/revoke_iam_keys.zip"
}
data "archive_file" "snapshot_forensics" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/snapshot_forensics"
output_path = "${path.root}/.build/snapshot_forensics.zip"
}
data "archive_file" "notify_soc" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/notify_soc"
output_path = "${path.root}/.build/notify_soc.zip"
}
data "archive_file" "audit_logger" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/audit_logger"
output_path = "${path.root}/.build/audit_logger.zip"
}
# IAM role for Step Functions execution
resource "aws_iam_role" "sfn_execution" {
name = "${var.project_name}-sfn-execution"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "states.amazonaws.com"
}
}]
})
tags = var.tags
}
resource "aws_iam_role_policy" "sfn_policy" {
name = "${var.project_name}-sfn-invoke-lambda"
role = aws_iam_role.sfn_execution.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = "lambda:InvokeFunction"
Resource = [
aws_lambda_function.isolate_ec2.arn,
aws_lambda_function.revoke_iam_keys.arn,
aws_lambda_function.snapshot_forensics.arn,
aws_lambda_function.notify_soc.arn,
aws_lambda_function.audit_logger.arn
]
}]
})
}
# Step Functions state machine
resource "aws_sfn_state_machine" "incident_response" {
name = "${var.project_name}-incident-response"
role_arn = aws_iam_role.sfn_execution.arn
type = "STANDARD"
definition = templatefile("${path.module}/asl.json", {
IsolateEc2Arn = aws_lambda_function.isolate_ec2.arn
RevokeIamKeysArn = aws_lambda_function.revoke_iam_keys.arn
SnapshotForensicsArn = aws_lambda_function.snapshot_forensics.arn
NotifySocArn = aws_lambda_function.notify_soc.arn
AuditLoggerArn = aws_lambda_function.audit_logger.arn
})
tags = var.tags
}What does this Terraform configuration do
- The aws_dynamodb_table creates the audit trail table with incident_id as partition key and timestamp as sort key, plus GSIs for querying by resource ARN or action type.
- The aws_sns_topic provides a notification channel for the SOC team.
- Each aws_lambda_function gets its own deployment package built from the scripts/lambda/ directories using archive_file data sources.
- The aws_sfn_state_machine uses templatefile() to inject Lambda ARNs into the ASL JSON at plan time, replacing the ${...Arn} placeholders.
- The Step Functions IAM role trusts states.amazonaws.com and is scoped to invoke only the five orchestration Lambda functions.
- Save modules/orchestration/main.tf.
- Add the module outputs to modules/orchestration/outputs.tf:
output "state_machine_arn" {
description = "ARN of the incident response state machine"
value = aws_sfn_state_machine.incident_response.arn
}
output "dynamodb_table_name" {
description = "Name of the security audit DynamoDB table"
value = aws_dynamodb_table.audit_trail.name
}
output "sns_topic_arn" {
description = "ARN of the SOC notifications SNS topic"
value = aws_sns_topic.soc_notifications.arn
}- Save modules/orchestration/outputs.tf.
- Add the orchestration module block to your root main.tf. Find the existing module blocks and add below them:
module "orchestration" {
source = "./modules/orchestration"
project_name = var.project_name
aws_region = var.aws_region
tags = var.tags
}- Save main.tf.
Seeing a Terraform validation error
If terraform validate fails because Lambda source directories are empty, that is expected. You will create the Python function files in the next substep. Terraform needs the source files to exist before it can plan the archive_file data sources.
help me debug a Terraform validation error with my orchestration module
Implement Python Lambda functions for incident response
Each Lambda function handles one specific response action. Splitting actions into separate functions keeps each one testable, reusable, and independently deployable.
- Create the EC2 isolation function by adding the following to scripts/lambda/isolate_ec2/lambda_function.py:
"""Isolate a compromised EC2 instance by attaching a quarantine security group
and disabling API termination to preserve evidence."""
import json
import logging
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
ec2 = boto3.client("ec2")
def lambda_handler(event, context):
"""Quarantine an EC2 instance for forensic investigation."""
instance_id = event.get("instance_id")
vpc_id = event.get("vpc_id")
incident_id = event.get("incident_id", "unknown")
logger.info(f"Isolating instance {instance_id} for incident {incident_id}")
# Create a quarantine security group that blocks all traffic
sg_name = f"quarantine-{incident_id}"
sg_response = ec2.create_security_group(
GroupName=sg_name,
Description=f"Quarantine SG for incident {incident_id}",
VpcId=vpc_id
)
quarantine_sg_id = sg_response["GroupId"]
# Revoke the default egress rule to block all outbound traffic
ec2.revoke_security_group_egress(
GroupId=quarantine_sg_id,
IpPermissions=[{
"IpProtocol": "-1",
"IpRanges": [{"CidrIp": "0.0.0.0/0"}]
}]
)
# Get current security groups on the instance
instance = ec2.describe_instances(InstanceIds=[instance_id])
original_sgs = [
sg["GroupId"]
for sg in instance["Reservations"][0]["Instances"][0]["SecurityGroups"]
]
# Replace all security groups with quarantine SG
ec2.modify_instance_attribute(
InstanceId=instance_id,
Groups=[quarantine_sg_id]
)
# Disable API termination to prevent accidental deletion
ec2.modify_instance_attribute(
InstanceId=instance_id,
DisableApiTermination={"Value": True}
)
logger.info(f"Instance {instance_id} quarantined with SG {quarantine_sg_id}")
return {
"status": "isolated",
"instance_id": instance_id,
"quarantine_sg_id": quarantine_sg_id,
"original_security_groups": original_sgs,
"incident_id": incident_id
}What does this function do
The isolate function creates a new security group with zero ingress and zero egress rules, then replaces all existing security groups on the instance. This immediately cuts the instance off from the network. Disabling API termination prevents anyone from accidentally destroying the evidence.
- Save scripts/lambda/isolate_ec2/lambda_function.py.
- Create the IAM key revocation function by adding the following to scripts/lambda/revoke_iam_keys/lambda_function.py:
"""Revoke compromised IAM access keys by disabling them and attaching
an explicit deny-all policy to the user."""
import json
import logging
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
iam = boto3.client("iam")
def lambda_handler(event, context):
"""Disable access keys and attach an explicit deny policy."""
username = event.get("username")
access_key_id = event.get("access_key_id")
incident_id = event.get("incident_id", "unknown")
logger.info(f"Revoking keys for user {username}, incident {incident_id}")
# Disable the compromised access key
iam.update_access_key(
UserName=username,
AccessKeyId=access_key_id,
Status="Inactive"
)
# Attach an explicit deny-all policy to block any remaining sessions
deny_policy = json.dumps({
"Version": "2012-10-17",
"Statement": [{
"Effect": "Deny",
"Action": "*",
"Resource": "*"
}]
})
iam.put_user_policy(
UserName=username,
PolicyName=f"incident-deny-all-{incident_id}",
PolicyDocument=deny_policy
)
logger.info(f"Access key {access_key_id} disabled, deny policy attached")
return {
"status": "keys_revoked",
"username": username,
"access_key_id": access_key_id,
"deny_policy_name": f"incident-deny-all-{incident_id}",
"incident_id": incident_id
}Why attach an explicit deny policy
Disabling the access key stops new API calls using that key. However, if the attacker obtained temporary session credentials (via sts:GetSessionToken), those sessions remain active. An explicit deny-all inline policy overrides all other permissions, immediately blocking active sessions too.
- Save scripts/lambda/revoke_iam_keys/lambda_function.py.
- Create the forensic snapshot function by adding the following to scripts/lambda/snapshot_forensics/lambda_function.py:
"""Create EBS snapshots of all volumes attached to a compromised instance,
tagged for forensic investigation."""
import json
import logging
from datetime import datetime, timezone
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
ec2 = boto3.client("ec2")
def lambda_handler(event, context):
"""Snapshot all EBS volumes on a target instance."""
instance_id = event.get("instance_id")
incident_id = event.get("incident_id", "unknown")
timestamp = datetime.now(timezone.utc).isoformat()
logger.info(f"Creating forensic snapshots for {instance_id}")
# Get all volumes attached to the instance
volumes = ec2.describe_volumes(
Filters=[{
"Name": "attachment.instance-id",
"Values": [instance_id]
}]
)
snapshot_ids = []
for volume in volumes["Volumes"]:
volume_id = volume["VolumeId"]
# Create snapshot with forensic tags
snapshot = ec2.create_snapshot(
VolumeId=volume_id,
Description=f"Forensic snapshot - incident {incident_id}",
TagSpecifications=[{
"ResourceType": "snapshot",
"Tags": [
{"Key": "Purpose", "Value": "forensic-investigation"},
{"Key": "IncidentId", "Value": incident_id},
{"Key": "SourceInstance", "Value": instance_id},
{"Key": "SourceVolume", "Value": volume_id},
{"Key": "CaptureTime", "Value": timestamp}
]
}]
)
snapshot_ids.append(snapshot["SnapshotId"])
logger.info(f"Created snapshot {snapshot['SnapshotId']} for {volume_id}")
return {
"status": "snapshots_created",
"instance_id": instance_id,
"snapshot_ids": snapshot_ids,
"volume_count": len(snapshot_ids),
"incident_id": incident_id
}- Save scripts/lambda/snapshot_forensics/lambda_function.py.
- Create the SOC notification function by adding the following to scripts/lambda/notify_soc/lambda_function.py:
"""Publish structured incident details to the SOC notifications SNS topic."""
import json
import logging
import os
from datetime import datetime, timezone
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
sns = boto3.client("sns")
def lambda_handler(event, context):
"""Send structured incident notification to the SOC team."""
incident_id = event.get("incident_id", "unknown")
severity = event.get("severity", "MEDIUM")
resource_arn = event.get("resource_arn", "")
finding_type = event.get("finding_type", "")
timestamp = datetime.now(timezone.utc).isoformat()
topic_arn = os.environ["SNS_TOPIC_ARN"]
# Build structured notification message
message = {
"incident_id": incident_id,
"severity": severity,
"timestamp": timestamp,
"resource_arn": resource_arn,
"finding_type": finding_type,
"status": "response_initiated",
"actions_taken": [
"EC2 instance isolated",
"IAM keys revoked",
"Forensic snapshots captured"
]
}
# Publish to SNS with severity as subject
sns.publish(
TopicArn=topic_arn,
Subject=f"[{severity}] Security Incident {incident_id}",
Message=json.dumps(message, indent=2)
)
logger.info(f"SOC notified for incident {incident_id} ({severity})")
return {
"status": "notified",
"incident_id": incident_id,
"severity": severity,
"topic_arn": topic_arn
}- Save scripts/lambda/notify_soc/lambda_function.py.
- Create the audit logger function by adding the following to scripts/lambda/audit_logger/lambda_function.py:
"""Write incident metadata to the DynamoDB audit table for forensic tracking."""
import json
import logging
import os
from datetime import datetime, timezone
import boto3
logger = logging.getLogger()
logger.setLevel(logging.INFO)
dynamodb = boto3.resource("dynamodb")
def lambda_handler(event, context):
"""Record incident actions in the DynamoDB audit trail."""
table_name = os.environ["AUDIT_TABLE_NAME"]
table = dynamodb.Table(table_name)
incident_id = event.get("incident_id", "unknown")
timestamp = datetime.now(timezone.utc).isoformat()
# Build audit record from the parallel response results
audit_record = {
"incident_id": incident_id,
"timestamp": timestamp,
"resource_arn": event.get("resource_arn", ""),
"action_type": "incident_response",
"severity": event.get("severity", "MEDIUM"),
"actions_taken": json.dumps(event.get("parallel_results", [])),
"status": "completed",
"state_machine_execution": event.get("execution_id", "")
}
# Write to DynamoDB
table.put_item(Item=audit_record)
logger.info(f"Audit record written for incident {incident_id}")
return {
"status": "audit_logged",
"incident_id": incident_id,
"timestamp": timestamp,
"table_name": table_name
}What does the audit logger record
The audit logger captures the full incident context: which resource was affected, what actions were taken, severity level, and the Step Functions execution ID for traceability. Using DynamoDB gives you structured, queryable records with millisecond read latency for forensic analysis.
- Save scripts/lambda/audit_logger/lambda_function.py.
- Validate that Terraform can see all the files by running:
terraform validateYou should see Success! The configuration is valid. if all files are in the correct directories.
Seeing errors about missing files or invalid references
Double-check that each Lambda function file is named lambda_function.py and sits inside its own directory under scripts/lambda/. The archive_file data source expects a source_dir path that contains at least one file.
help me fix Terraform errors related to my Lambda function file paths
Create the DynamoDB audit query CLI tool
During an investigation, your security team needs to quickly search the audit trail by incident ID, affected resource, time window, or action type. A purpose-built CLI tool makes this fast and formatted.
- Create the audit query tool by adding the following to scripts/audit_query.py:
"""CLI tool for querying the DynamoDB security audit trail.
Supports queries by incident ID, resource ARN, time range, and action type.
Outputs formatted tables using rich, with optional JSON export."""
import argparse
import json
import sys
from datetime import datetime, timezone
import boto3
from boto3.dynamodb.conditions import Key, Attr
from rich.console import Console
from rich.table import Table
console = Console()
# Default table name - override with --table flag
DEFAULT_TABLE = "zerotrust-security-audit"
def get_table(table_name):
"""Get a DynamoDB table resource."""
dynamodb = boto3.resource("dynamodb")
return dynamodb.Table(table_name)
def query_by_incident(table, incident_id):
"""Query audit records by incident ID (partition key)."""
response = table.query(
KeyConditionExpression=Key("incident_id").eq(incident_id)
)
return response["Items"]
def query_by_resource(table, resource_arn):
"""Query audit records by resource ARN using GSI."""
response = table.query(
IndexName="resource-index",
KeyConditionExpression=Key("resource_arn").eq(resource_arn)
)
return response["Items"]
def query_by_time_range(table, start_time, end_time):
"""Scan audit records within a time range."""
response = table.scan(
FilterExpression=Attr("timestamp").between(start_time, end_time)
)
return response["Items"]
def query_by_action(table, action_type):
"""Query audit records by action type using GSI."""
response = table.query(
IndexName="action-index",
KeyConditionExpression=Key("action_type").eq(action_type)
)
return response["Items"]
def display_results(items, export_json=False):
"""Display query results as a rich table or export as JSON."""
if export_json:
print(json.dumps(items, indent=2, default=str))
return
if not items:
console.print("[yellow]No records found.[/yellow]")
return
# Build a formatted table
table = Table(title="Security Audit Records")
table.add_column("Incident ID", style="cyan")
table.add_column("Timestamp", style="green")
table.add_column("Resource ARN", style="magenta")
table.add_column("Action", style="yellow")
table.add_column("Severity", style="red")
table.add_column("Status", style="blue")
for item in items:
table.add_row(
item.get("incident_id", ""),
item.get("timestamp", ""),
item.get("resource_arn", "")[:50],
item.get("action_type", ""),
item.get("severity", ""),
item.get("status", "")
)
console.print(table)
console.print(f"\n[dim]Total records: {len(items)}[/dim]")
def main():
"""Parse arguments and execute the appropriate query."""
parser = argparse.ArgumentParser(
description="Query the DynamoDB security audit trail"
)
parser.add_argument(
"--table", default=DEFAULT_TABLE,
help="DynamoDB table name"
)
parser.add_argument(
"--incident", help="Query by incident ID"
)
parser.add_argument(
"--resource", help="Query by resource ARN"
)
parser.add_argument(
"--start", help="Start time for range query (ISO 8601)"
)
parser.add_argument(
"--end", help="End time for range query (ISO 8601)"
)
parser.add_argument(
"--action", help="Query by action type"
)
parser.add_argument(
"--json", action="store_true", dest="export_json",
help="Export results as JSON instead of table"
)
args = parser.parse_args()
table = get_table(args.table)
# Execute the appropriate query based on arguments
if args.incident:
items = query_by_incident(table, args.incident)
elif args.resource:
items = query_by_resource(table, args.resource)
elif args.start and args.end:
items = query_by_time_range(table, args.start, args.end)
elif args.action:
items = query_by_action(table, args.action)
else:
console.print("[red]Provide at least one query filter.[/red]")
console.print("Use --incident, --resource, --start/--end, or --action")
sys.exit(1)
display_results(items, export_json=args.export_json)
if __name__ == "__main__":
main()How the query tool works
- The --incident flag queries the primary key directly for fast single-incident lookups.
- The --resource and --action flags use the Global Secondary Indexes defined in the DynamoDB table.
- The --start and --end flags perform a scan with a filter expression for time-range queries.
- Adding --json exports raw JSON for piping into other tools like jq.
- Save scripts/audit_query.py.
- Verify the CLI tool loads without errors by running:
python scripts/audit_query.py --helpYou should see the argument parser output showing the available flags: --table, --incident, --resource, --start, --end, --action, and --json.
ModuleNotFoundError for rich or boto3
Make sure your Python virtual environment is active. The rich and boto3 packages should already be installed from requirements.txt in Session 1. If not, run pip install -r requirements.txt from your project root.
help me resolve a ModuleNotFoundError when running my audit query script
- Deploy the orchestration infrastructure by running:
terraform plan -target=module.orchestrationReview the plan output. You should see resources for the DynamoDB table, SNS topic, IAM roles, five Lambda functions, and the Step Functions state machine being created.
✔️ Awesome, I've got everything!
Great. Double check you have saved all files before proceeding.
ⓧ I'd like to double check the full code
Here is the complete modules/orchestration/main.tf for reference:
# DynamoDB audit table for incident tracking
resource "aws_dynamodb_table" "audit_trail" {
name = "${var.project_name}-security-audit"
billing_mode = "PAY_PER_REQUEST"
hash_key = "incident_id"
range_key = "timestamp"
attribute {
name = "incident_id"
type = "S"
}
attribute {
name = "timestamp"
type = "S"
}
attribute {
name = "resource_arn"
type = "S"
}
attribute {
name = "action_type"
type = "S"
}
global_secondary_index {
name = "resource-index"
hash_key = "resource_arn"
range_key = "timestamp"
projection_type = "ALL"
}
global_secondary_index {
name = "action-index"
hash_key = "action_type"
range_key = "timestamp"
projection_type = "ALL"
}
tags = var.tags
}
# SNS topic for SOC notifications
resource "aws_sns_topic" "soc_notifications" {
name = "${var.project_name}-soc-alerts"
tags = var.tags
}
# IAM role for Lambda execution
resource "aws_iam_role" "lambda_execution" {
name = "${var.project_name}-orchestration-lambda"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "lambda.amazonaws.com"
}
}]
})
tags = var.tags
}
# Lambda execution policy - EC2, IAM, EBS, SNS, DynamoDB, CloudWatch Logs
resource "aws_iam_role_policy" "lambda_policy" {
name = "${var.project_name}-lambda-orchestration"
role = aws_iam_role.lambda_execution.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [
{
Effect = "Allow"
Action = [
"ec2:ModifyInstanceAttribute",
"ec2:CreateSecurityGroup",
"ec2:AuthorizeSecurityGroupIngress",
"ec2:RevokeSecurityGroupIngress",
"ec2:ModifyNetworkInterfaceAttribute",
"ec2:DescribeInstances",
"ec2:DescribeSecurityGroups",
"ec2:CreateSnapshot",
"ec2:CreateTags",
"ec2:DescribeVolumes"
]
Resource = "*"
},
{
Effect = "Allow"
Action = [
"iam:UpdateAccessKey",
"iam:ListAccessKeys",
"iam:PutUserPolicy"
]
Resource = "*"
},
{
Effect = "Allow"
Action = [
"sns:Publish"
]
Resource = aws_sns_topic.soc_notifications.arn
},
{
Effect = "Allow"
Action = [
"dynamodb:PutItem",
"dynamodb:Query"
]
Resource = [
aws_dynamodb_table.audit_trail.arn,
"${aws_dynamodb_table.audit_trail.arn}/index/*"
]
},
{
Effect = "Allow"
Action = [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
]
Resource = "arn:aws:logs:*:*:*"
}
]
})
}
# Lambda functions for each orchestration step
resource "aws_lambda_function" "isolate_ec2" {
filename = data.archive_file.isolate_ec2.output_path
function_name = "${var.project_name}-isolate-ec2"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.isolate_ec2.output_base64sha256
timeout = 60
tags = var.tags
}
resource "aws_lambda_function" "revoke_iam_keys" {
filename = data.archive_file.revoke_iam_keys.output_path
function_name = "${var.project_name}-revoke-iam-keys"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.revoke_iam_keys.output_base64sha256
timeout = 60
tags = var.tags
}
resource "aws_lambda_function" "snapshot_forensics" {
filename = data.archive_file.snapshot_forensics.output_path
function_name = "${var.project_name}-snapshot-forensics"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.snapshot_forensics.output_base64sha256
timeout = 120
tags = var.tags
}
resource "aws_lambda_function" "notify_soc" {
filename = data.archive_file.notify_soc.output_path
function_name = "${var.project_name}-notify-soc"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.notify_soc.output_base64sha256
timeout = 30
environment {
variables = {
SNS_TOPIC_ARN = aws_sns_topic.soc_notifications.arn
}
}
tags = var.tags
}
resource "aws_lambda_function" "audit_logger" {
filename = data.archive_file.audit_logger.output_path
function_name = "${var.project_name}-audit-logger"
role = aws_iam_role.lambda_execution.arn
handler = "lambda_function.lambda_handler"
runtime = "python3.11"
source_code_hash = data.archive_file.audit_logger.output_base64sha256
timeout = 30
environment {
variables = {
AUDIT_TABLE_NAME = aws_dynamodb_table.audit_trail.name
}
}
tags = var.tags
}
# Archive files for Lambda deployment packages
data "archive_file" "isolate_ec2" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/isolate_ec2"
output_path = "${path.root}/.build/isolate_ec2.zip"
}
data "archive_file" "revoke_iam_keys" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/revoke_iam_keys"
output_path = "${path.root}/.build/revoke_iam_keys.zip"
}
data "archive_file" "snapshot_forensics" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/snapshot_forensics"
output_path = "${path.root}/.build/snapshot_forensics.zip"
}
data "archive_file" "notify_soc" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/notify_soc"
output_path = "${path.root}/.build/notify_soc.zip"
}
data "archive_file" "audit_logger" {
type = "zip"
source_dir = "${path.root}/scripts/lambda/audit_logger"
output_path = "${path.root}/.build/audit_logger.zip"
}
# IAM role for Step Functions execution
resource "aws_iam_role" "sfn_execution" {
name = "${var.project_name}-sfn-execution"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Action = "sts:AssumeRole"
Effect = "Allow"
Principal = {
Service = "states.amazonaws.com"
}
}]
})
tags = var.tags
}
resource "aws_iam_role_policy" "sfn_policy" {
name = "${var.project_name}-sfn-invoke-lambda"
role = aws_iam_role.sfn_execution.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = "lambda:InvokeFunction"
Resource = [
aws_lambda_function.isolate_ec2.arn,
aws_lambda_function.revoke_iam_keys.arn,
aws_lambda_function.snapshot_forensics.arn,
aws_lambda_function.notify_soc.arn,
aws_lambda_function.audit_logger.arn
]
}]
})
}
# Step Functions state machine
resource "aws_sfn_state_machine" "incident_response" {
name = "${var.project_name}-incident-response"
role_arn = aws_iam_role.sfn_execution.arn
type = "STANDARD"
definition = templatefile("${path.module}/asl.json", {
IsolateEc2Arn = aws_lambda_function.isolate_ec2.arn
RevokeIamKeysArn = aws_lambda_function.revoke_iam_keys.arn
SnapshotForensicsArn = aws_lambda_function.snapshot_forensics.arn
NotifySocArn = aws_lambda_function.notify_soc.arn
AuditLoggerArn = aws_lambda_function.audit_logger.arn
})
tags = var.tags
}Your security orchestration platform is deployed. Next up, you will build the cross-cloud event pipeline that routes GCP Security Command Center findings into this Step Functions workflow.
Build the Cross-Cloud Security Event Pipeline
Your detection infrastructure is live in both clouds, and your Step Functions orchestrator can respond to incidents automatically. But right now, each cloud sees threats in isolation.
In this session you will bridge the gap by building a real-time pipeline that moves GCP Security Command Center findings into AWS, normalizes them into a common format, correlates events across both clouds, and triggers the Step Functions orchestrator when something serious appears.
In this step, get ready to:
- Deploy a Python Cloud Function that normalizes GCP SCC findings and forwards them to AWS CloudWatch Logs.
- Create CloudWatch Log Groups with Metric Filters and a cross-cloud event correlator script.
- Wire high-severity findings into Step Functions and validate the full pipeline end-to-end.
Deploy the SCC-to-CloudWatch Cloud Function
GCP Security Command Center publishes findings to a Pub/Sub topic as JSON. Your Cloud Function will receive each finding, normalize it into a structure matching the AWS Security Finding Format (ASFF), then forward it to CloudWatch Logs using the Workload Identity Federation credentials you set up in Session 3.
- Create the Cloud Function directory by running this command:
mkdir -p scripts/cloud_functions/scc_to_cloudwatch- Create scripts/cloud_functions/scc_to_cloudwatch/main.py and paste the following code:
import base64
import json
import os
import uuid
from datetime import datetime, timezone
import boto3
from google.auth import default
# Configuration from environment variables
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
LOG_GROUP_NAME = os.environ.get("LOG_GROUP_NAME", "/security/cross-cloud/gcp-scc")
LOG_STREAM_NAME = os.environ.get("LOG_STREAM_NAME", "scc-findings")
AWS_ROLE_ARN = os.environ.get("AWS_ROLE_ARN")
def get_aws_session():
"""Authenticate to AWS using Workload Identity Federation."""
# Use google-auth default credentials (WIF) to get a GCP token
credentials, project = default()
credentials.refresh(google.auth.transport.requests.Request())
# Exchange GCP token for AWS credentials via STS
sts_client = boto3.client("sts", region_name=AWS_REGION)
response = sts_client.assume_role_with_web_identity(
RoleArn=AWS_ROLE_ARN,
RoleSessionName="scc-to-cloudwatch",
WebIdentityToken=credentials.token,
)
# Create a session with the assumed role credentials
return boto3.Session(
aws_access_key_id=response["Credentials"]["AccessKeyId"],
aws_secret_access_key=response["Credentials"]["SecretAccessKey"],
aws_session_token=response["Credentials"]["SessionToken"],
region_name=AWS_REGION,
)
def normalize_scc_to_asff(scc_finding, gcp_project_id):
"""Normalize a GCP SCC finding into ASFF-compatible JSON."""
finding = scc_finding.get("finding", {})
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
# Map SCC severity to ASFF severity label
severity_map = {
"CRITICAL": "CRITICAL",
"HIGH": "HIGH",
"MEDIUM": "MEDIUM",
"LOW": "LOW",
}
scc_severity = finding.get("severity", "MEDIUM")
asff_severity = severity_map.get(scc_severity, "INFORMATIONAL")
# Build ASFF-compatible finding
asff_finding = {
"SchemaVersion": "2018-10-08",
"Id": f"gcp-scc-{uuid.uuid4()}",
"ProductArn": f"arn:aws:securityhub:{AWS_REGION}::product/gcp/scc",
"GeneratorId": "gcp-security-command-center",
"AwsAccountId": os.environ.get("AWS_ACCOUNT_ID", "000000000000"),
"Types": [f"Software and Configuration Checks/GCP/{finding.get('category', 'Unknown')}"],
"CreatedAt": finding.get("createTime", now),
"UpdatedAt": now,
"Severity": {"Label": asff_severity, "Original": scc_severity},
"Title": f"GCP SCC: {finding.get('category', 'Unknown Finding')}",
"Description": f"GCP SCC finding from project {gcp_project_id}: {finding.get('category', '')}",
"Resources": [{
"Type": "Other",
"Id": finding.get("resourceName", "unknown"),
"Details": {"Other": {
"GcpProject": gcp_project_id,
"FindingName": finding.get("name", ""),
"State": finding.get("state", ""),
"Category": finding.get("category", ""),
}},
}],
"SourceCloud": "GCP",
"OriginalFinding": scc_finding,
}
return asff_finding
def forward_to_cloudwatch(session, asff_finding):
"""Send the normalized finding to AWS CloudWatch Logs."""
logs_client = session.client("logs")
# Ensure log stream exists
try:
logs_client.create_log_stream(
logGroupName=LOG_GROUP_NAME, logStreamName=LOG_STREAM_NAME
)
except logs_client.exceptions.ResourceAlreadyExistsException:
pass
# Put the log event
logs_client.put_log_events(
logGroupName=LOG_GROUP_NAME,
logStreamName=LOG_STREAM_NAME,
logEvents=[{
"timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
"message": json.dumps(asff_finding),
}],
)
def scc_to_cloudwatch(event, context):
"""Cloud Function entry point triggered by Pub/Sub."""
# Decode the Pub/Sub message
pubsub_message = base64.b64decode(event["data"]).decode("utf-8")
scc_notification = json.loads(pubsub_message)
# Extract GCP project from the finding
finding = scc_notification.get("finding", {})
resource_name = finding.get("resourceName", "")
gcp_project_id = os.environ.get("GCP_PROJECT_ID", "unknown")
# Normalize to ASFF format
asff_finding = normalize_scc_to_asff(scc_notification, gcp_project_id)
# Authenticate to AWS via WIF and forward
session = get_aws_session()
forward_to_cloudwatch(session, asff_finding)
print(f"Forwarded SCC finding {finding.get('category', 'unknown')} to CloudWatch")
return "OK"What does this code do?
The Cloud Function receives a Pub/Sub message containing a GCP SCC finding in JSON format (with fields like notificationConfigName, finding.name, finding.category, finding.state, and finding.severity).
It normalizes this into ASFF format, which requires fields like SchemaVersion (always 2018-10-08), Id, ProductArn, Severity, Title, and Resources. This shared format lets AWS Security Hub and your own tools process findings from both clouds identically.
The function then authenticates to AWS using Workload Identity Federation (no static credentials) and writes the normalized finding to a CloudWatch Log Group.
- Save scripts/cloud_functions/scc_to_cloudwatch/main.py.
Cloud Function not forwarding to CloudWatch?
Verify the AWS_ROLE_ARN environment variable matches the IAM role with WIF trust policy from Session 3.
Check that the Cloud Function's service account is bound to the Workload Identity Pool provider as an allowed principal.
help me debug my SCC-to-CloudWatch Cloud Function
Create CloudWatch Log Groups, Metric Filters, and the event correlator
With findings flowing from GCP into AWS, you need somewhere to land them and a way to spot patterns across both clouds. CloudWatch Metric Filters will extract security signals, and a Python correlator script will query CloudWatch Logs Insights to detect the same source IP appearing in both cloud audit trails.
- Create modules/detection/cloudwatch.tf and paste the following Terraform configuration:
# CloudWatch Log Group for normalized GCP SCC findings
resource "aws_cloudwatch_log_group" "gcp_scc_findings" {
name = "/security/cross-cloud/gcp-scc"
retention_in_days = 30
tags = {
Environment = "security"
Source = "gcp-scc"
}
}
# CloudWatch Log Group for AWS-native security events
resource "aws_cloudwatch_log_group" "aws_security_events" {
name = "/security/cross-cloud/aws-events"
retention_in_days = 30
tags = {
Environment = "security"
Source = "aws-native"
}
}
# Metric Filter: high-severity findings from GCP
resource "aws_cloudwatch_log_metric_filter" "gcp_high_severity" {
name = "gcp-high-severity-findings"
pattern = "{ $.Severity.Label = \"HIGH\" || $.Severity.Label = \"CRITICAL\" }"
log_group_name = aws_cloudwatch_log_group.gcp_scc_findings.name
metric_transformation {
name = "GCPHighSeverityFindingCount"
namespace = "Security/CrossCloud"
value = "1"
}
}
# Metric Filter: cross-cloud source IP pattern
resource "aws_cloudwatch_log_metric_filter" "cross_cloud_ip" {
name = "cross-cloud-source-ip"
pattern = "{ $.SourceCloud = \"GCP\" && $.Resources[0].Details.Other.SourceIP EXISTS }"
log_group_name = aws_cloudwatch_log_group.gcp_scc_findings.name
metric_transformation {
name = "CrossCloudIPActivityCount"
namespace = "Security/CrossCloud"
value = "1"
}
}
# EventBridge rule for high-severity cross-cloud findings to Step Functions
resource "aws_cloudwatch_event_rule" "cross_cloud_high_severity" {
name = "cross-cloud-high-severity-to-orchestrator"
description = "Route high-severity cross-cloud findings to Step Functions"
event_pattern = jsonencode({
source = ["custom.security.cross-cloud"]
detail-type = ["Cross-Cloud Security Finding"]
detail = {
severity = ["HIGH", "CRITICAL"]
}
})
}
# EventBridge target: invoke the Step Functions state machine
resource "aws_cloudwatch_event_target" "invoke_orchestrator" {
rule = aws_cloudwatch_event_rule.cross_cloud_high_severity.name
arn = var.state_machine_arn
role_arn = var.eventbridge_role_arn
input_transformer {
input_paths = {
severity = "$.detail.severity"
category = "$.detail.category"
resource = "$.detail.resource_id"
}
input_template = <<EOF
{
"incident_type": "cross_cloud_finding",
"severity": <severity>,
"category": <category>,
"resource_id": <resource>,
"source": "gcp-scc-pipeline"
}
EOF
}
}What does this Terraform do?
Two Log Groups receive security events from different sources: one for normalized GCP findings, one for AWS-native events.
Metric Filters watch the GCP log group for high-severity findings and cross-cloud IP activity, publishing counts to CloudWatch Metrics for dashboarding.
The EventBridge rule matches high-severity cross-cloud findings and routes them to the Step Functions state machine, triggering your automated incident response workflow.
- Save modules/detection/cloudwatch.tf.
- Apply the Terraform changes by running:
terraform plan
terraform apply -auto-approveYou should see Terraform creating the two Log Groups, two Metric Filters, one EventBridge rule, and one EventBridge target.
- Now create the Python event correlator script at scripts/correlate_events.py:
"""Cross-cloud event correlator using CloudWatch Logs Insights.
Queries both AWS and GCP security log groups to identify
the same source IP appearing in both cloud audit trails
within a configurable time window.
"""
import argparse
import json
import time
from datetime import datetime, timezone, timedelta
import boto3
from rich.console import Console
from rich.table import Table
console = Console()
# Log groups to correlate across
GCP_LOG_GROUP = "/security/cross-cloud/gcp-scc"
AWS_LOG_GROUP = "/security/cross-cloud/aws-events"
def run_insights_query(client, log_group, query, start_time, end_time):
"""Execute a CloudWatch Logs Insights query and wait for results."""
response = client.start_query(
logGroupName=log_group,
startTime=int(start_time.timestamp()),
endTime=int(end_time.timestamp()),
queryString=query,
)
query_id = response["queryId"]
# Poll for query completion
while True:
result = client.get_query_results(queryId=query_id)
if result["status"] == "Complete":
return result["results"]
time.sleep(1)
def extract_source_ips(results):
"""Extract source IPs from Logs Insights query results."""
ips = set()
for row in results:
for field in row:
if field["field"] == "source_ip":
ips.add(field["value"])
return ips
def correlate_cross_cloud(hours_back=24, min_occurrences=1):
"""Find IPs appearing in both cloud audit trails within the time window."""
client = boto3.client("logs")
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(hours=hours_back)
console.print(f"[bold]Correlating events from last {hours_back} hours...[/bold]")
# Query GCP findings for source IPs
gcp_query = """
fields @timestamp, @message
| parse @message '"SourceIP":"*"' as source_ip
| filter ispresent(source_ip)
| stats count(*) as occurrences by source_ip
| sort occurrences desc
"""
# Query AWS events for source IPs
aws_query = """
fields @timestamp, @message
| parse @message '"sourceIPAddress":"*"' as source_ip
| filter ispresent(source_ip)
| stats count(*) as occurrences by source_ip
| sort occurrences desc
"""
gcp_results = run_insights_query(client, GCP_LOG_GROUP, gcp_query, start_time, end_time)
aws_results = run_insights_query(client, AWS_LOG_GROUP, aws_query, start_time, end_time)
gcp_ips = extract_source_ips(gcp_results)
aws_ips = extract_source_ips(aws_results)
# Find intersection: IPs in both clouds
cross_cloud_ips = gcp_ips & aws_ips
# Display results
table = Table(title="Cross-Cloud IP Correlation Results")
table.add_column("Source IP", style="red bold")
table.add_column("Seen in GCP", style="cyan")
table.add_column("Seen in AWS", style="yellow")
table.add_column("Risk Level", style="magenta")
for ip in cross_cloud_ips:
risk = "CRITICAL" if ip in gcp_ips and ip in aws_ips else "HIGH"
table.add_row(ip, "Yes", "Yes", risk)
if cross_cloud_ips:
console.print(table)
console.print(
f"\n[bold red]ALERT:[/bold red] {len(cross_cloud_ips)} IP(s) found in both clouds!"
)
else:
console.print("[green]No cross-cloud IP correlation detected.[/green]")
return list(cross_cloud_ips)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Cross-cloud event correlator")
parser.add_argument(
"--hours", type=int, default=24, help="Hours to look back (default: 24)"
)
parser.add_argument(
"--min-occurrences", type=int, default=1, help="Minimum occurrences per IP"
)
args = parser.parse_args()
correlate_cross_cloud(hours_back=args.hours, min_occurrences=args.min_occurrences)What does the correlator do?
The script runs two CloudWatch Logs Insights queries in parallel: one against the GCP findings log group and one against the AWS events log group. It extracts all source IPs from each, computes the intersection, and flags any IP that appears in both cloud audit trails within the configurable time window.
An attacker compromising one cloud often pivots to the other. This cross-cloud IP correlation catches lateral movement that single-cloud tools miss entirely.
- Save scripts/correlate_events.py.
- Test the correlator runs without errors by running:
python scripts/correlate_events.py --hours 1You should see "No cross-cloud IP correlation detected" since no events have flowed yet.
Seeing a ResourceNotFoundException?
This means the CloudWatch Log Group does not exist yet. Make sure your terraform apply completed successfully and created both /security/cross-cloud/gcp-scc and /security/cross-cloud/aws-events Log Groups.
help me debug my CloudWatch Logs Insights correlator query
Write the end-to-end pipeline test
You have all the pipeline components deployed. Now you need a test script that simulates a GCP SCC finding and verifies it flows through the entire chain: Pub/Sub, Cloud Function, CloudWatch, EventBridge, Step Functions, and finally lands as a DynamoDB audit record.
- Create scripts/test_e2e_pipeline.py and paste the following code:
"""End-to-end pipeline test for cross-cloud security event flow.
Simulates a GCP SCC finding flowing through:
Pub/Sub -> Cloud Function -> CloudWatch -> EventBridge -> Step Functions -> DynamoDB
"""
import json
import os
import time
import uuid
from datetime import datetime, timezone
import boto3
from google.cloud import pubsub_v1
from rich.console import Console
from rich.panel import Panel
console = Console()
# Configuration
GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
PUBSUB_TOPIC = os.environ.get("SCC_PUBSUB_TOPIC", "scc-findings-topic")
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
DYNAMODB_TABLE = os.environ.get("DYNAMODB_TABLE", "security-audit-trail")
LOG_GROUP_NAME = "/security/cross-cloud/gcp-scc"
STATE_MACHINE_ARN = os.environ.get("STATE_MACHINE_ARN")
def create_test_scc_finding():
"""Generate a realistic GCP SCC finding notification."""
finding_id = str(uuid.uuid4())
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
return {
"notificationConfigName": f"projects/{GCP_PROJECT_ID}/notificationConfigs/scc-all-active",
"finding": {
"name": f"projects/{GCP_PROJECT_ID}/sources/123456/findings/{finding_id}",
"parent": f"projects/{GCP_PROJECT_ID}/sources/123456",
"resourceName": f"//compute.googleapis.com/projects/{GCP_PROJECT_ID}/zones/us-central1-a/instances/test-vm",
"state": "ACTIVE",
"category": "OPEN_FIREWALL",
"severity": "HIGH",
"eventTime": now,
"createTime": now,
"sourceProperties": {
"SourceIP": "198.51.100.42",
"SeverityLevel": "High",
"Explanation": "Test finding for E2E pipeline validation",
},
},
}
def publish_to_pubsub(finding):
"""Publish the test finding to the SCC Pub/Sub topic."""
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(GCP_PROJECT_ID, PUBSUB_TOPIC)
message_data = json.dumps(finding).encode("utf-8")
future = publisher.publish(topic_path, message_data)
message_id = future.result()
console.print(f" Published to Pub/Sub: message ID {message_id}")
return message_id
def check_cloudwatch_logs(finding_id, timeout=120):
"""Wait for the normalized finding to appear in CloudWatch Logs."""
logs_client = boto3.client("logs", region_name=AWS_REGION)
start_time = time.time()
while time.time() - start_time < timeout:
try:
response = logs_client.filter_log_events(
logGroupName=LOG_GROUP_NAME,
filterPattern=f"\"{finding_id}\"",
limit=1,
)
if response.get("events"):
console.print(" Finding appeared in CloudWatch Logs")
return True
except Exception:
pass
time.sleep(5)
return False
def check_step_functions_execution(timeout=120):
"""Check for a recent Step Functions execution triggered by the finding."""
sfn_client = boto3.client("stepfunctions", region_name=AWS_REGION)
start_time = time.time()
while time.time() - start_time < timeout:
response = sfn_client.list_executions(
stateMachineArn=STATE_MACHINE_ARN,
statusFilter="RUNNING",
maxResults=5,
)
if response.get("executions"):
execution_arn = response["executions"][0]["executionArn"]
console.print(f" Step Functions execution: {execution_arn}")
return execution_arn
time.sleep(5)
return None
def check_dynamodb_audit(timeout=180):
"""Verify the audit record was written to DynamoDB."""
dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
table = dynamodb.Table(DYNAMODB_TABLE)
start_time = time.time()
while time.time() - start_time < timeout:
response = table.scan(
FilterExpression="contains(incident_type, :val)",
ExpressionAttributeValues={":val": "cross_cloud_finding"},
Limit=5,
)
if response.get("Items"):
console.print(" Audit record found in DynamoDB")
return response["Items"][0]
time.sleep(10)
return None
def run_e2e_test():
"""Execute the full end-to-end pipeline test."""
console.print(Panel("[bold]Cross-Cloud Pipeline E2E Test[/bold]", style="blue"))
# Step 1: Create and publish test finding
console.print("\n[bold]1. Publishing test SCC finding to Pub/Sub...[/bold]")
finding = create_test_scc_finding()
finding_id = finding["finding"]["name"].split("/")[-1]
publish_to_pubsub(finding)
# Step 2: Verify CloudWatch ingestion
console.print("\n[bold]2. Waiting for CloudWatch Logs ingestion...[/bold]")
cw_result = check_cloudwatch_logs(finding_id)
assert cw_result, "Finding did not appear in CloudWatch within timeout"
# Step 3: Verify Step Functions triggered
console.print("\n[bold]3. Checking Step Functions execution...[/bold]")
execution = check_step_functions_execution()
if execution:
console.print(" [green]Step Functions triggered successfully[/green]")
else:
console.print(" [yellow]No execution detected (EventBridge may need manual trigger)[/yellow]")
# Step 4: Verify DynamoDB audit trail
console.print("\n[bold]4. Checking DynamoDB audit trail...[/bold]")
audit_record = check_dynamodb_audit()
if audit_record:
console.print(" [green]Audit record confirmed[/green]")
console.print(f" Incident ID: {audit_record.get('incident_id', 'N/A')}")
else:
console.print(" [yellow]No audit record yet (pipeline still processing)[/yellow]")
# Summary
console.print("\n" + "=" * 60)
console.print("[bold green]E2E Pipeline Test Complete[/bold green]")
console.print(f" Pub/Sub publish: PASS")
console.print(f" CloudWatch ingestion: {'PASS' if cw_result else 'FAIL'}")
console.print(f" Step Functions: {'PASS' if execution else 'PENDING'}")
console.print(f" DynamoDB audit: {'PASS' if audit_record else 'PENDING'}")
if __name__ == "__main__":
run_e2e_test()What does the E2E test validate?
The test publishes a realistic SCC finding to Pub/Sub, then polls each downstream stage with a timeout. It verifies the finding flows through every component of the pipeline: the Cloud Function normalizes it, CloudWatch receives it, EventBridge matches the high-severity pattern, Step Functions starts an execution, and the audit_logger Lambda writes the incident to DynamoDB.
This confirms your entire cross-cloud security event pipeline works end-to-end without needing a real threat to trigger it.
- Save scripts/test_e2e_pipeline.py.
- Run the end-to-end test by executing:
python scripts/test_e2e_pipeline.pyYou should see the test publish to Pub/Sub, then confirm CloudWatch ingestion within 2 minutes. The Step Functions and DynamoDB checks may show PENDING if the EventBridge propagation takes longer.
Test timing out at CloudWatch ingestion?
Confirm your Cloud Function is deployed and subscribed to the correct Pub/Sub topic. Check the Cloud Function logs in the GCP Console for any authentication errors with the WIF token exchange.
help me debug why my E2E pipeline test is timing out
✔️ Awesome, I've got everything!
Great. Double check you have saved all files before moving on.
ⓧ I'd like to double check the full code
Here is the full scripts/cloud_functions/scc_to_cloudwatch/main.py:
import base64
import json
import os
import uuid
from datetime import datetime, timezone
import boto3
from google.auth import default
# Configuration from environment variables
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
LOG_GROUP_NAME = os.environ.get("LOG_GROUP_NAME", "/security/cross-cloud/gcp-scc")
LOG_STREAM_NAME = os.environ.get("LOG_STREAM_NAME", "scc-findings")
AWS_ROLE_ARN = os.environ.get("AWS_ROLE_ARN")
def get_aws_session():
"""Authenticate to AWS using Workload Identity Federation."""
# Use google-auth default credentials (WIF) to get a GCP token
credentials, project = default()
credentials.refresh(google.auth.transport.requests.Request())
# Exchange GCP token for AWS credentials via STS
sts_client = boto3.client("sts", region_name=AWS_REGION)
response = sts_client.assume_role_with_web_identity(
RoleArn=AWS_ROLE_ARN,
RoleSessionName="scc-to-cloudwatch",
WebIdentityToken=credentials.token,
)
# Create a session with the assumed role credentials
return boto3.Session(
aws_access_key_id=response["Credentials"]["AccessKeyId"],
aws_secret_access_key=response["Credentials"]["SecretAccessKey"],
aws_session_token=response["Credentials"]["SessionToken"],
region_name=AWS_REGION,
)
def normalize_scc_to_asff(scc_finding, gcp_project_id):
"""Normalize a GCP SCC finding into ASFF-compatible JSON."""
finding = scc_finding.get("finding", {})
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
# Map SCC severity to ASFF severity label
severity_map = {
"CRITICAL": "CRITICAL",
"HIGH": "HIGH",
"MEDIUM": "MEDIUM",
"LOW": "LOW",
}
scc_severity = finding.get("severity", "MEDIUM")
asff_severity = severity_map.get(scc_severity, "INFORMATIONAL")
# Build ASFF-compatible finding
asff_finding = {
"SchemaVersion": "2018-10-08",
"Id": f"gcp-scc-{uuid.uuid4()}",
"ProductArn": f"arn:aws:securityhub:{AWS_REGION}::product/gcp/scc",
"GeneratorId": "gcp-security-command-center",
"AwsAccountId": os.environ.get("AWS_ACCOUNT_ID", "000000000000"),
"Types": [f"Software and Configuration Checks/GCP/{finding.get('category', 'Unknown')}"],
"CreatedAt": finding.get("createTime", now),
"UpdatedAt": now,
"Severity": {"Label": asff_severity, "Original": scc_severity},
"Title": f"GCP SCC: {finding.get('category', 'Unknown Finding')}",
"Description": f"GCP SCC finding from project {gcp_project_id}: {finding.get('category', '')}",
"Resources": [{
"Type": "Other",
"Id": finding.get("resourceName", "unknown"),
"Details": {"Other": {
"GcpProject": gcp_project_id,
"FindingName": finding.get("name", ""),
"State": finding.get("state", ""),
"Category": finding.get("category", ""),
}},
}],
"SourceCloud": "GCP",
"OriginalFinding": scc_finding,
}
return asff_finding
def forward_to_cloudwatch(session, asff_finding):
"""Send the normalized finding to AWS CloudWatch Logs."""
logs_client = session.client("logs")
# Ensure log stream exists
try:
logs_client.create_log_stream(
logGroupName=LOG_GROUP_NAME, logStreamName=LOG_STREAM_NAME
)
except logs_client.exceptions.ResourceAlreadyExistsException:
pass
# Put the log event
logs_client.put_log_events(
logGroupName=LOG_GROUP_NAME,
logStreamName=LOG_STREAM_NAME,
logEvents=[{
"timestamp": int(datetime.now(timezone.utc).timestamp() * 1000),
"message": json.dumps(asff_finding),
}],
)
def scc_to_cloudwatch(event, context):
"""Cloud Function entry point triggered by Pub/Sub."""
# Decode the Pub/Sub message
pubsub_message = base64.b64decode(event["data"]).decode("utf-8")
scc_notification = json.loads(pubsub_message)
# Extract GCP project from the finding
finding = scc_notification.get("finding", {})
resource_name = finding.get("resourceName", "")
gcp_project_id = os.environ.get("GCP_PROJECT_ID", "unknown")
# Normalize to ASFF format
asff_finding = normalize_scc_to_asff(scc_notification, gcp_project_id)
# Authenticate to AWS via WIF and forward
session = get_aws_session()
forward_to_cloudwatch(session, asff_finding)
print(f"Forwarded SCC finding {finding.get('category', 'unknown')} to CloudWatch")
return "OK"Here is the full scripts/correlate_events.py:
"""Cross-cloud event correlator using CloudWatch Logs Insights.
Queries both AWS and GCP security log groups to identify
the same source IP appearing in both cloud audit trails
within a configurable time window.
"""
import argparse
import json
import time
from datetime import datetime, timezone, timedelta
import boto3
from rich.console import Console
from rich.table import Table
console = Console()
# Log groups to correlate across
GCP_LOG_GROUP = "/security/cross-cloud/gcp-scc"
AWS_LOG_GROUP = "/security/cross-cloud/aws-events"
def run_insights_query(client, log_group, query, start_time, end_time):
"""Execute a CloudWatch Logs Insights query and wait for results."""
response = client.start_query(
logGroupName=log_group,
startTime=int(start_time.timestamp()),
endTime=int(end_time.timestamp()),
queryString=query,
)
query_id = response["queryId"]
# Poll for query completion
while True:
result = client.get_query_results(queryId=query_id)
if result["status"] == "Complete":
return result["results"]
time.sleep(1)
def extract_source_ips(results):
"""Extract source IPs from Logs Insights query results."""
ips = set()
for row in results:
for field in row:
if field["field"] == "source_ip":
ips.add(field["value"])
return ips
def correlate_cross_cloud(hours_back=24, min_occurrences=1):
"""Find IPs appearing in both cloud audit trails within the time window."""
client = boto3.client("logs")
end_time = datetime.now(timezone.utc)
start_time = end_time - timedelta(hours=hours_back)
console.print(f"[bold]Correlating events from last {hours_back} hours...[/bold]")
# Query GCP findings for source IPs
gcp_query = """
fields @timestamp, @message
| parse @message '"SourceIP":"*"' as source_ip
| filter ispresent(source_ip)
| stats count(*) as occurrences by source_ip
| sort occurrences desc
"""
# Query AWS events for source IPs
aws_query = """
fields @timestamp, @message
| parse @message '"sourceIPAddress":"*"' as source_ip
| filter ispresent(source_ip)
| stats count(*) as occurrences by source_ip
| sort occurrences desc
"""
gcp_results = run_insights_query(client, GCP_LOG_GROUP, gcp_query, start_time, end_time)
aws_results = run_insights_query(client, AWS_LOG_GROUP, aws_query, start_time, end_time)
gcp_ips = extract_source_ips(gcp_results)
aws_ips = extract_source_ips(aws_results)
# Find intersection: IPs in both clouds
cross_cloud_ips = gcp_ips & aws_ips
# Display results
table = Table(title="Cross-Cloud IP Correlation Results")
table.add_column("Source IP", style="red bold")
table.add_column("Seen in GCP", style="cyan")
table.add_column("Seen in AWS", style="yellow")
table.add_column("Risk Level", style="magenta")
for ip in cross_cloud_ips:
risk = "CRITICAL" if ip in gcp_ips and ip in aws_ips else "HIGH"
table.add_row(ip, "Yes", "Yes", risk)
if cross_cloud_ips:
console.print(table)
console.print(
f"\n[bold red]ALERT:[/bold red] {len(cross_cloud_ips)} IP(s) found in both clouds!"
)
else:
console.print("[green]No cross-cloud IP correlation detected.[/green]")
return list(cross_cloud_ips)
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Cross-cloud event correlator")
parser.add_argument(
"--hours", type=int, default=24, help="Hours to look back (default: 24)"
)
parser.add_argument(
"--min-occurrences", type=int, default=1, help="Minimum occurrences per IP"
)
args = parser.parse_args()
correlate_cross_cloud(hours_back=args.hours, min_occurrences=args.min_occurrences)Here is the full scripts/test_e2e_pipeline.py:
"""End-to-end pipeline test for cross-cloud security event flow.
Simulates a GCP SCC finding flowing through:
Pub/Sub -> Cloud Function -> CloudWatch -> EventBridge -> Step Functions -> DynamoDB
"""
import json
import os
import time
import uuid
from datetime import datetime, timezone
import boto3
from google.cloud import pubsub_v1
from rich.console import Console
from rich.panel import Panel
console = Console()
# Configuration
GCP_PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
PUBSUB_TOPIC = os.environ.get("SCC_PUBSUB_TOPIC", "scc-findings-topic")
AWS_REGION = os.environ.get("AWS_REGION", "us-east-1")
DYNAMODB_TABLE = os.environ.get("DYNAMODB_TABLE", "security-audit-trail")
LOG_GROUP_NAME = "/security/cross-cloud/gcp-scc"
STATE_MACHINE_ARN = os.environ.get("STATE_MACHINE_ARN")
def create_test_scc_finding():
"""Generate a realistic GCP SCC finding notification."""
finding_id = str(uuid.uuid4())
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
return {
"notificationConfigName": f"projects/{GCP_PROJECT_ID}/notificationConfigs/scc-all-active",
"finding": {
"name": f"projects/{GCP_PROJECT_ID}/sources/123456/findings/{finding_id}",
"parent": f"projects/{GCP_PROJECT_ID}/sources/123456",
"resourceName": f"//compute.googleapis.com/projects/{GCP_PROJECT_ID}/zones/us-central1-a/instances/test-vm",
"state": "ACTIVE",
"category": "OPEN_FIREWALL",
"severity": "HIGH",
"eventTime": now,
"createTime": now,
"sourceProperties": {
"SourceIP": "198.51.100.42",
"SeverityLevel": "High",
"Explanation": "Test finding for E2E pipeline validation",
},
},
}
def publish_to_pubsub(finding):
"""Publish the test finding to the SCC Pub/Sub topic."""
publisher = pubsub_v1.PublisherClient()
topic_path = publisher.topic_path(GCP_PROJECT_ID, PUBSUB_TOPIC)
message_data = json.dumps(finding).encode("utf-8")
future = publisher.publish(topic_path, message_data)
message_id = future.result()
console.print(f" Published to Pub/Sub: message ID {message_id}")
return message_id
def check_cloudwatch_logs(finding_id, timeout=120):
"""Wait for the normalized finding to appear in CloudWatch Logs."""
logs_client = boto3.client("logs", region_name=AWS_REGION)
start_time = time.time()
while time.time() - start_time < timeout:
try:
response = logs_client.filter_log_events(
logGroupName=LOG_GROUP_NAME,
filterPattern=f"\"{finding_id}\"",
limit=1,
)
if response.get("events"):
console.print(" Finding appeared in CloudWatch Logs")
return True
except Exception:
pass
time.sleep(5)
return False
def check_step_functions_execution(timeout=120):
"""Check for a recent Step Functions execution triggered by the finding."""
sfn_client = boto3.client("stepfunctions", region_name=AWS_REGION)
start_time = time.time()
while time.time() - start_time < timeout:
response = sfn_client.list_executions(
stateMachineArn=STATE_MACHINE_ARN,
statusFilter="RUNNING",
maxResults=5,
)
if response.get("executions"):
execution_arn = response["executions"][0]["executionArn"]
console.print(f" Step Functions execution: {execution_arn}")
return execution_arn
time.sleep(5)
return None
def check_dynamodb_audit(timeout=180):
"""Verify the audit record was written to DynamoDB."""
dynamodb = boto3.resource("dynamodb", region_name=AWS_REGION)
table = dynamodb.Table(DYNAMODB_TABLE)
start_time = time.time()
while time.time() - start_time < timeout:
response = table.scan(
FilterExpression="contains(incident_type, :val)",
ExpressionAttributeValues={":val": "cross_cloud_finding"},
Limit=5,
)
if response.get("Items"):
console.print(" Audit record found in DynamoDB")
return response["Items"][0]
time.sleep(10)
return None
def run_e2e_test():
"""Execute the full end-to-end pipeline test."""
console.print(Panel("[bold]Cross-Cloud Pipeline E2E Test[/bold]", style="blue"))
# Step 1: Create and publish test finding
console.print("\n[bold]1. Publishing test SCC finding to Pub/Sub...[/bold]")
finding = create_test_scc_finding()
finding_id = finding["finding"]["name"].split("/")[-1]
publish_to_pubsub(finding)
# Step 2: Verify CloudWatch ingestion
console.print("\n[bold]2. Waiting for CloudWatch Logs ingestion...[/bold]")
cw_result = check_cloudwatch_logs(finding_id)
assert cw_result, "Finding did not appear in CloudWatch within timeout"
# Step 3: Verify Step Functions triggered
console.print("\n[bold]3. Checking Step Functions execution...[/bold]")
execution = check_step_functions_execution()
if execution:
console.print(" [green]Step Functions triggered successfully[/green]")
else:
console.print(" [yellow]No execution detected (EventBridge may need manual trigger)[/yellow]")
# Step 4: Verify DynamoDB audit trail
console.print("\n[bold]4. Checking DynamoDB audit trail...[/bold]")
audit_record = check_dynamodb_audit()
if audit_record:
console.print(" [green]Audit record confirmed[/green]")
console.print(f" Incident ID: {audit_record.get('incident_id', 'N/A')}")
else:
console.print(" [yellow]No audit record yet (pipeline still processing)[/yellow]")
# Summary
console.print("\n" + "=" * 60)
console.print("[bold green]E2E Pipeline Test Complete[/bold green]")
console.print(f" Pub/Sub publish: PASS")
console.print(f" CloudWatch ingestion: {'PASS' if cw_result else 'FAIL'}")
console.print(f" Step Functions: {'PASS' if execution else 'PENDING'}")
console.print(f" DynamoDB audit: {'PASS' if audit_record else 'PENDING'}")
if __name__ == "__main__":
run_e2e_test()Your cross-cloud security event pipeline is operational. Findings from GCP now flow automatically into AWS, get normalized, correlated, and trigger your Step Functions orchestrator for high-severity incidents. Next up, you will add policy-as-code governance and compliance scanning to prevent security misconfigurations before they reach production.
Enforce Policy-as-Code and Run Compliance Scans
Your detection and orchestration pipeline catches threats after they happen. Policy-as-code flips the script by preventing insecure infrastructure from being deployed in the first place.
In this session, you will write Rego policies that enforce security baselines across both clouds, run Prowler compliance scans against live infrastructure, and wire everything into a GitHub Actions CI pipeline that blocks non-compliant pull requests automatically.
In this step, get ready to:
- Write Rego policies enforcing security baselines for AWS and GCP, validated by a Python test harness.
- Run Prowler CIS benchmark scans and generate a unified cross-cloud compliance dashboard.
- Create a GitHub Actions workflow with policy checks, Prowler scanning, and a Python CI gate script.
Write Rego policies and validate them with a Python test harness
OPA/Conftest evaluates Terraform plan JSON against Rego policies before any infrastructure is created. Each policy returns a denial message when a resource violates your security baseline.
- Create the policy directories for both clouds by running these commands:
mkdir -p policies/aws policies/gcp- Confirm the directories exist by running:
ls policies/You should see aws and gcp listed in the output.
- Create the first AWS policy to deny public S3 buckets by adding this file:
# Deny S3 buckets with public ACLs
package main
import rego.v1
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_s3_bucket_acl"
resource.change.after.acl == "public-read"
msg := sprintf("S3 bucket ACL '%s' must not be public-read", [resource.address])
}
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_s3_bucket_acl"
resource.change.after.acl == "public-read-write"
msg := sprintf("S3 bucket ACL '%s' must not be public-read-write", [resource.address])
}What does this policy do?
Conftest feeds Terraform plan JSON as input to Rego. The policy iterates over resource_changes looking for S3 bucket ACLs set to public. If found, it emits a denial message with the resource address.
The package main declaration is required for Conftest. The deny contains msg if pattern collects all violations into a set that Conftest reports.
- Create the encryption required policy:
# Require encryption on AWS resources
package main
import rego.v1
# Deny unencrypted EBS volumes
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_ebs_volume"
resource.change.after.encrypted == false
msg := sprintf("EBS volume '%s' must have encryption enabled", [resource.address])
}
# Deny unencrypted S3 buckets (missing server-side encryption)
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_s3_bucket_server_side_encryption_configuration"
count(resource.change.after.rule) == 0
msg := sprintf("S3 bucket '%s' must have server-side encryption configured", [resource.address])
}- Create the no wildcard IAM policy:
# Deny wildcard IAM actions
package main
import rego.v1
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_iam_policy"
policy_doc := json.unmarshal(resource.change.after.policy)
some statement in policy_doc.Statement
statement.Effect == "Allow"
some action in statement.Action
action == "*"
msg := sprintf("IAM policy '%s' must not use wildcard (*) actions", [resource.address])
}- Create the VPC flow logs policy:
# Require VPC flow logs enabled
package main
import rego.v1
# Collect all VPCs in the plan
vpcs contains resource.change.after.vpc_id if {
some resource in input.resource_changes
resource.type == "aws_flow_log"
}
# Deny VPCs without a corresponding flow log
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_vpc"
not resource.address in vpcs
msg := sprintf("VPC '%s' must have flow logs enabled", [resource.address])
}- Create the no open SSH/RDP policy:
# Deny security groups with open SSH or RDP
package main
import rego.v1
# Deny ingress rules allowing SSH (port 22) from anywhere
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_security_group_rule"
resource.change.after.type == "ingress"
resource.change.after.from_port <= 22
resource.change.after.to_port >= 22
some cidr in resource.change.after.cidr_blocks
cidr == "0.0.0.0/0"
msg := sprintf("Security group rule '%s' must not allow SSH from 0.0.0.0/0", [resource.address])
}
# Deny ingress rules allowing RDP (port 3389) from anywhere
deny contains msg if {
some resource in input.resource_changes
resource.type == "aws_security_group_rule"
resource.change.after.type == "ingress"
resource.change.after.from_port <= 3389
resource.change.after.to_port >= 3389
some cidr in resource.change.after.cidr_blocks
cidr == "0.0.0.0/0"
msg := sprintf("Security group rule '%s' must not allow RDP from 0.0.0.0/0", [resource.address])
}Now create the GCP equivalents.
- Create the GCP no public storage policy:
# Deny public GCS buckets
package main
import rego.v1
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_storage_bucket_iam_member"
resource.change.after.member == "allUsers"
msg := sprintf("GCS bucket IAM '%s' must not grant access to allUsers", [resource.address])
}
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_storage_bucket_iam_member"
resource.change.after.member == "allAuthenticatedUsers"
msg := sprintf("GCS bucket IAM '%s' must not grant access to allAuthenticatedUsers", [resource.address])
}- Create the GCP encryption required policy:
# Require encryption on GCP resources
package main
import rego.v1
# Deny compute disks without CMEK encryption
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_compute_disk"
not resource.change.after.disk_encryption_key
msg := sprintf("Compute disk '%s' must have customer-managed encryption key configured", [resource.address])
}- Create the GCP no wildcard IAM policy:
# Deny wildcard GCP IAM bindings
package main
import rego.v1
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_project_iam_binding"
resource.change.after.role == "roles/owner"
msg := sprintf("IAM binding '%s' must not use roles/owner (too broad)", [resource.address])
}
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_project_iam_binding"
resource.change.after.role == "roles/editor"
msg := sprintf("IAM binding '%s' must not use roles/editor (too broad)", [resource.address])
}- Create the GCP VPC flow logs policy:
# Require VPC flow logs on GCP subnets
package main
import rego.v1
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_compute_subnetwork"
not resource.change.after.log_config
msg := sprintf("Subnet '%s' must have VPC flow logs enabled via log_config", [resource.address])
}- Create the GCP no open SSH/RDP policy:
# Deny firewall rules with open SSH/RDP
package main
import rego.v1
# Deny firewall rules allowing SSH from anywhere
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_compute_firewall"
some allow_rule in resource.change.after.allow
some port in allow_rule.ports
port == "22"
some range in resource.change.after.source_ranges
range == "0.0.0.0/0"
msg := sprintf("Firewall rule '%s' must not allow SSH from 0.0.0.0/0", [resource.address])
}
# Deny firewall rules allowing RDP from anywhere
deny contains msg if {
some resource in input.resource_changes
resource.type == "google_compute_firewall"
some allow_rule in resource.change.after.allow
some port in allow_rule.ports
port == "3389"
some range in resource.change.after.source_ranges
range == "0.0.0.0/0"
msg := sprintf("Firewall rule '%s' must not allow RDP from 0.0.0.0/0", [resource.address])
}Why OPA/Conftest over HashiCorp Sentinel?
Sentinel requires Terraform Cloud or Enterprise (paid). OPA/Conftest is open-source, works locally and in GitHub Actions, and is the industry standard for policy-as-code in multi-cloud environments. It is also a more transferable skill that employers value across different infrastructure stacks.
Now write the Python test harness that generates mock Terraform plan JSON and validates each policy.
- Create the test harness script by adding this file:
"""Test harness for Rego policies.
Generates mock Terraform plan JSON and validates each policy
produces expected pass/fail results using Conftest.
"""
import json
import subprocess
import tempfile
import sys
from pathlib import Path
from rich.console import Console
from rich.table import Table
console = Console()
# Mock Terraform plan fixtures for testing policies
TEST_CASES = {
"aws/no_public_storage": {
"should_fail": {
"resource_changes": [{
"address": "aws_s3_bucket_acl.public",
"type": "aws_s3_bucket_acl",
"change": {"after": {"acl": "public-read"}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "aws_s3_bucket_acl.private",
"type": "aws_s3_bucket_acl",
"change": {"after": {"acl": "private"}}
}]
}
},
"aws/encryption_required": {
"should_fail": {
"resource_changes": [{
"address": "aws_ebs_volume.unencrypted",
"type": "aws_ebs_volume",
"change": {"after": {"encrypted": False}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "aws_ebs_volume.encrypted",
"type": "aws_ebs_volume",
"change": {"after": {"encrypted": True}}
}]
}
},
"aws/no_wildcard_iam": {
"should_fail": {
"resource_changes": [{
"address": "aws_iam_policy.wildcard",
"type": "aws_iam_policy",
"change": {"after": {"policy": json.dumps({
"Statement": [{"Effect": "Allow", "Action": ["*"], "Resource": "*"}]
})}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "aws_iam_policy.scoped",
"type": "aws_iam_policy",
"change": {"after": {"policy": json.dumps({
"Statement": [{"Effect": "Allow", "Action": ["s3:GetObject"], "Resource": "arn:aws:s3:::bucket/*"}]
})}}
}]
}
},
"aws/no_open_ssh_rdp": {
"should_fail": {
"resource_changes": [{
"address": "aws_security_group_rule.open_ssh",
"type": "aws_security_group_rule",
"change": {"after": {
"type": "ingress", "from_port": 22, "to_port": 22,
"cidr_blocks": ["0.0.0.0/0"]
}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "aws_security_group_rule.restricted_ssh",
"type": "aws_security_group_rule",
"change": {"after": {
"type": "ingress", "from_port": 22, "to_port": 22,
"cidr_blocks": ["10.0.0.0/8"]
}}
}]
}
},
"gcp/no_public_storage": {
"should_fail": {
"resource_changes": [{
"address": "google_storage_bucket_iam_member.public",
"type": "google_storage_bucket_iam_member",
"change": {"after": {"member": "allUsers"}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "google_storage_bucket_iam_member.private",
"type": "google_storage_bucket_iam_member",
"change": {"after": {"member": "serviceAccount:sa@project.iam.gserviceaccount.com"}}
}]
}
},
"gcp/no_open_ssh_rdp": {
"should_fail": {
"resource_changes": [{
"address": "google_compute_firewall.open_ssh",
"type": "google_compute_firewall",
"change": {"after": {
"allow": [{"ports": ["22"], "protocol": "tcp"}],
"source_ranges": ["0.0.0.0/0"]
}}
}]
},
"should_pass": {
"resource_changes": [{
"address": "google_compute_firewall.restricted_ssh",
"type": "google_compute_firewall",
"change": {"after": {
"allow": [{"ports": ["22"], "protocol": "tcp"}],
"source_ranges": ["10.0.0.0/8"]
}}
}]
}
},
}
def run_conftest(policy_path: str, plan_json: dict) -> dict:
"""Run conftest test against a mock plan JSON file."""
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump(plan_json, f)
f.flush()
result = subprocess.run(
["conftest", "test", "-p", policy_path, f.name, "--output", "json"],
capture_output=True, text=True
)
return {"returncode": result.returncode, "stdout": result.stdout, "stderr": result.stderr}
def main():
"""Run all policy test cases and report results."""
console.print("\n[bold]Running Rego Policy Tests[/bold]\n")
table = Table(title="Policy Test Results")
table.add_column("Policy", style="cyan")
table.add_column("Test Type", style="magenta")
table.add_column("Expected", style="yellow")
table.add_column("Result", style="green")
failures = 0
policy_base = Path("policies")
for policy_name, cases in TEST_CASES.items():
policy_dir = str(policy_base / policy_name.rsplit("/", 1)[0])
policy_file = str(policy_base / f"{policy_name}.rego")
# Test: should_fail case (expect non-zero exit / deny messages)
result = run_conftest(policy_file, cases["should_fail"])
fail_passed = result["returncode"] != 0
status = "PASS" if fail_passed else "FAIL"
if not fail_passed:
failures += 1
table.add_row(policy_name, "should_fail", "DENY", status)
# Test: should_pass case (expect zero exit / no deny messages)
result = run_conftest(policy_file, cases["should_pass"])
pass_passed = result["returncode"] == 0
status = "PASS" if pass_passed else "FAIL"
if not pass_passed:
failures += 1
table.add_row(policy_name, "should_pass", "ALLOW", status)
console.print(table)
if failures > 0:
console.print(f"\n[red bold]{failures} test(s) failed![/red bold]")
sys.exit(1)
else:
console.print("\n[green bold]All policy tests passed![/green bold]")
if __name__ == "__main__":
main()What does this test harness do?
For each Rego policy, the script creates two mock Terraform plan JSON fixtures: one that should trigger a denial and one that should pass cleanly. It runs conftest test against each fixture and checks exit codes to confirm the policy behaves correctly.
The --output json flag returns machine-readable results. A non-zero exit code means Conftest found violations, which is expected for the should_fail cases.
- Run the test harness to validate all policies:
python scripts/test_policies.pyYou should see a table showing all policy tests with PASS status for both should_fail and should_pass cases.
Seeing test failures?
Make sure conftest is on your PATH. Check that each .rego file has package main as its package declaration. Conftest requires this to find the deny rules.
help me debug my conftest policy test failures
Run Prowler CIS benchmark scans and generate a unified report
Prowler scans your live cloud infrastructure against CIS benchmarks. You will run it against both AWS and GCP, then merge the results into a single cross-cloud compliance dashboard.
- Run a Prowler scan against your AWS account by running:
prowler aws -f us-east-1 --output-formats json-ocsf htmlWhat does this command do?
The -f flag restricts the scan to your deployment region, speeding it up significantly. The --output-formats flag requests JSON-OCSF (machine-readable) and HTML (human-readable) reports. Prowler writes output to an output/ directory by default.
- Run a Prowler scan against your GCP project by running:
prowler gcp --project-ids [[GCP_PROJECT_ID="your-gcp-project-id"]] --output-formats json-ocsf html- Confirm both scan outputs exist by running:
ls output/You should see JSON-OCSF and HTML files for both the AWS and GCP scans.
What does Prowler check?
Prowler runs 571 checks across 82 AWS services and 79 checks across 13 GCP services. It covers CIS benchmarks, AWS Foundational Security Best Practices, and more. The JSON-OCSF output uses a standardized schema that your Python aggregator will parse.
Now write a Python report aggregator that merges both scan results into a unified dashboard.
- Create the Prowler report aggregator:
"""Prowler report aggregator.
Parses JSON-OCSF output from AWS and GCP scans, merges findings
into a unified cross-cloud compliance HTML dashboard.
"""
import json
import sys
from pathlib import Path
from datetime import datetime
from rich.console import Console
from rich.table import Table
console = Console()
def load_ocsf_findings(file_path: Path) -> list:
"""Load findings from a Prowler JSON-OCSF file."""
findings = []
with open(file_path, "r") as f:
for line in f:
line = line.strip()
if line:
findings.append(json.loads(line))
return findings
def normalize_finding(finding: dict, cloud: str) -> dict:
"""Normalize a Prowler finding into a common schema."""
return {
"cloud": cloud,
"check_id": finding.get("metadata", {}).get("event_code", "unknown"),
"title": finding.get("finding_info", {}).get("title", "Unknown check"),
"status": finding.get("status", "unknown"),
"severity": finding.get("severity", "unknown"),
"resource": finding.get("resources", [{}])[0].get("uid", "N/A"),
"region": finding.get("resources", [{}])[0].get("region", "global"),
"compliance": finding.get("compliance", {}),
}
def identify_cross_cloud_gaps(aws_findings: list, gcp_findings: list) -> list:
"""Identify controls failing in one cloud but passing in the other."""
# Map check categories to pass/fail status per cloud
aws_status = {}
gcp_status = {}
for f in aws_findings:
category = f["title"]
if category not in aws_status:
aws_status[category] = set()
aws_status[category].add(f["status"])
for f in gcp_findings:
category = f["title"]
if category not in gcp_status:
gcp_status[category] = set()
gcp_status[category].add(f["status"])
gaps = []
# Find common check categories with divergent results
common_categories = set(aws_status.keys()) & set(gcp_status.keys())
for cat in common_categories:
aws_has_fail = "FAIL" in aws_status[cat]
gcp_has_fail = "FAIL" in gcp_status[cat]
if aws_has_fail != gcp_has_fail:
gaps.append({
"category": cat,
"aws_status": "FAIL" if aws_has_fail else "PASS",
"gcp_status": "FAIL" if gcp_has_fail else "PASS",
})
return gaps
def generate_html_dashboard(aws_findings: list, gcp_findings: list, gaps: list, output_path: str):
"""Generate a unified HTML compliance dashboard."""
aws_pass = sum(1 for f in aws_findings if f["status"] == "PASS")
aws_fail = sum(1 for f in aws_findings if f["status"] == "FAIL")
gcp_pass = sum(1 for f in gcp_findings if f["status"] == "PASS")
gcp_fail = sum(1 for f in gcp_findings if f["status"] == "FAIL")
html = f"""<!DOCTYPE html>
<html><head><title>Cross-Cloud Compliance Dashboard</title>
<style>
body {{ font-family: sans-serif; margin: 2rem; }}
.summary {{ display: flex; gap: 2rem; margin-bottom: 2rem; }}
.card {{ padding: 1rem; border: 1px solid #ddd; border-radius: 8px; min-width: 200px; }}
.pass {{ color: #22c55e; }} .fail {{ color: #ef4444; }}
table {{ border-collapse: collapse; width: 100%; margin-top: 1rem; }}
th, td {{ border: 1px solid #ddd; padding: 8px; text-align: left; }}
th {{ background: #f5f5f5; }}
</style></head><body>
<h1>Cross-Cloud Compliance Dashboard</h1>
<p>Generated: {datetime.now().isoformat()}</p>
<div class="summary">
<div class="card"><h3>AWS</h3><p class="pass">Pass: {aws_pass}</p><p class="fail">Fail: {aws_fail}</p></div>
<div class="card"><h3>GCP</h3><p class="pass">Pass: {gcp_pass}</p><p class="fail">Fail: {gcp_fail}</p></div>
</div>
<h2>Cross-Cloud Compliance Gaps</h2>
<table><tr><th>Control Category</th><th>AWS Status</th><th>GCP Status</th></tr>
"""
for gap in gaps:
html += f"<tr><td>{gap['category']}</td><td>{gap['aws_status']}</td><td>{gap['gcp_status']}</td></tr>\n"
html += """</table>
<h2>All Findings (Failures Only)</h2>
<table><tr><th>Cloud</th><th>Check</th><th>Severity</th><th>Resource</th></tr>
"""
all_failures = [f for f in aws_findings + gcp_findings if f["status"] == "FAIL"]
for f in sorted(all_failures, key=lambda x: x["severity"], reverse=True)[:50]:
html += f"<tr><td>{f['cloud']}</td><td>{f['title']}</td><td>{f['severity']}</td><td>{f['resource']}</td></tr>\n"
html += "</table></body></html>"
with open(output_path, "w") as out:
out.write(html)
return output_path
def main():
"""Main entry point for report generation."""
output_dir = Path("output")
# Find the most recent Prowler JSON-OCSF files
aws_files = sorted(output_dir.glob("*aws*ocsf*.json*"))
gcp_files = sorted(output_dir.glob("*gcp*ocsf*.json*"))
if not aws_files:
console.print("[red]No AWS Prowler output found in output/[/red]")
sys.exit(1)
if not gcp_files:
console.print("[red]No GCP Prowler output found in output/[/red]")
sys.exit(1)
console.print(f"[cyan]Loading AWS findings from: {aws_files[-1]}[/cyan]")
raw_aws = load_ocsf_findings(aws_files[-1])
aws_findings = [normalize_finding(f, "AWS") for f in raw_aws]
console.print(f"[cyan]Loading GCP findings from: {gcp_files[-1]}[/cyan]")
raw_gcp = load_ocsf_findings(gcp_files[-1])
gcp_findings = [normalize_finding(f, "GCP") for f in raw_gcp]
# Identify cross-cloud compliance gaps
gaps = identify_cross_cloud_gaps(aws_findings, gcp_findings)
# Print summary table
table = Table(title="Cross-Cloud Compliance Summary")
table.add_column("Metric", style="cyan")
table.add_column("AWS", style="yellow")
table.add_column("GCP", style="blue")
table.add_row("Total Findings", str(len(aws_findings)), str(len(gcp_findings)))
table.add_row("Pass", str(sum(1 for f in aws_findings if f["status"] == "PASS")),
str(sum(1 for f in gcp_findings if f["status"] == "PASS")))
table.add_row("Fail", str(sum(1 for f in aws_findings if f["status"] == "FAIL")),
str(sum(1 for f in gcp_findings if f["status"] == "FAIL")))
table.add_row("Cross-Cloud Gaps", str(len(gaps)), str(len(gaps)))
console.print(table)
# Generate HTML dashboard
dashboard_path = generate_html_dashboard(
aws_findings, gcp_findings, gaps, "output/cross_cloud_compliance.html"
)
console.print(f"\n[green bold]Dashboard generated: {dashboard_path}[/green bold]")
if __name__ == "__main__":
main()What does this report aggregator do?
The script loads JSON-OCSF output from both Prowler scans, normalizes findings into a common schema, identifies controls that are failing in one cloud but passing in the other, and generates a single HTML dashboard with summary statistics, cross-cloud gaps, and the top 50 failures ranked by severity.
- Generate the unified dashboard by running:
python scripts/prowler_report.pyYou should see a summary table in your terminal and a message confirming the HTML dashboard was generated at output/cross_cloud_compliance.html.
No findings files found?
Make sure the Prowler scans completed successfully. Check the output/ directory for files matching the pattern *ocsf*.json. Prowler writes output with timestamps in the filenames, so the glob pattern in the script looks for the most recent file.
help me troubleshoot missing Prowler output files
Create a GitHub Actions CI pipeline with policy gates
The final piece wires everything into CI so that every pull request is automatically checked against your policies and Prowler IaC scanning before any infrastructure change is merged.
- Create the GitHub Actions workflow directory and file:
mkdir -p .github/workflows- Add the security CI workflow:
name: Security CI Pipeline
on:
pull_request:
branches: [main]
jobs:
security-checks:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Setup Terraform
uses: hashicorp/setup-terraform@v3
with:
terraform_version: "1.15.5"
- name: Setup Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
- name: Install dependencies
run: |
pip install prowler rich
# Install conftest
LATEST_VERSION=$(curl -s https://api.github.com/repos/open-policy-agent/conftest/releases/latest | grep tag_name | cut -d '"' -f 4 | sed 's/v//')
wget -q "https://github.com/open-policy-agent/conftest/releases/download/v${LATEST_VERSION}/conftest_${LATEST_VERSION}_Linux_x86_64.tar.gz"
tar xzf conftest_${LATEST_VERSION}_Linux_x86_64.tar.gz
sudo mv conftest /usr/local/bin/
- name: Terraform Init
run: terraform init -backend=false
- name: Terraform Plan
run: |
terraform plan -out=tfplan
terraform show -json tfplan > plan.json
- name: Conftest Policy Validation
run: |
conftest test -p policies/aws plan.json --output json > conftest_aws.json || true
conftest test -p policies/gcp plan.json --output json > conftest_gcp.json || true
- name: Prowler IaC Scan
run: |
prowler iac --scan-path . --frameworks terraform --output-formats json-ocsf > prowler_iac.json || true
- name: CI Gate - Unified Status
run: python scripts/ci_gate.py
env:
CONFTEST_AWS_OUTPUT: conftest_aws.json
CONFTEST_GCP_OUTPUT: conftest_gcp.json
PROWLER_OUTPUT: prowler_iac.jsonWhat does this workflow do?
On every pull request to main, the workflow initializes Terraform (without a real backend), generates a plan JSON, runs Conftest policies against it, runs Prowler IaC scanning against the Terraform files, and passes all outputs to your Python CI gate script for a unified pass/fail verdict.
The || true after each scan ensures the workflow continues even if individual tools find violations. The final ci_gate.py step produces the actual pass/fail status.
- Create the Python CI gate script that aggregates all tool outputs:
"""CI gate script.
Parses terraform plan, conftest, and Prowler outputs to produce
a unified pass/fail status with actionable remediation guidance.
"""
import json
import os
import sys
from pathlib import Path
from rich.console import Console
from rich.table import Table
console = Console()
def load_json_safe(path: str) -> list | dict | None:
"""Safely load a JSON file, return None if not found."""
filepath = Path(path)
if not filepath.exists():
return None
with open(filepath, "r") as f:
content = f.read().strip()
if not content:
return None
return json.loads(content)
def parse_conftest_results(results: list) -> dict:
"""Parse conftest JSON output into structured findings."""
failures = []
warnings = []
if not results:
return {"failures": failures, "warnings": warnings}
for result in results:
if isinstance(result, dict):
for failure in result.get("failures", []):
failures.append(failure.get("msg", "Unknown violation"))
for warning in result.get("warnings", []):
warnings.append(warning.get("msg", "Unknown warning"))
return {"failures": failures, "warnings": warnings}
def parse_prowler_results(results) -> dict:
"""Parse Prowler IaC JSON output into structured findings."""
if not results:
return {"critical": 0, "high": 0, "medium": 0, "low": 0, "findings": []}
findings = results if isinstance(results, list) else [results]
severity_counts = {"critical": 0, "high": 0, "medium": 0, "low": 0}
parsed_findings = []
for finding in findings:
severity = finding.get("severity", "low").lower()
if severity in severity_counts:
severity_counts[severity] += 1
if finding.get("status") == "FAIL":
parsed_findings.append({
"title": finding.get("finding_info", {}).get("title", "Unknown"),
"severity": severity,
"remediation": finding.get("remediation", {}).get("recommendation", "N/A"),
})
return {**severity_counts, "findings": parsed_findings}
def generate_remediation(conftest_failures: list, prowler_findings: list) -> list:
"""Generate actionable remediation guidance."""
guidance = []
for failure in conftest_failures[:10]:
guidance.append(f"[Policy Violation] {failure}")
for finding in prowler_findings[:10]:
guidance.append(
f"[IaC Finding - {finding['severity'].upper()}] "
f"{finding['title']}: {finding['remediation']}"
)
return guidance
def main():
"""Run the CI gate check."""
console.print("\n[bold]Security CI Gate[/bold]\n")
# Load tool outputs from environment variable paths or defaults
conftest_aws_path = os.environ.get("CONFTEST_AWS_OUTPUT", "conftest_aws.json")
conftest_gcp_path = os.environ.get("CONFTEST_GCP_OUTPUT", "conftest_gcp.json")
prowler_path = os.environ.get("PROWLER_OUTPUT", "prowler_iac.json")
# Parse results
aws_conftest = parse_conftest_results(load_json_safe(conftest_aws_path) or [])
gcp_conftest = parse_conftest_results(load_json_safe(conftest_gcp_path) or [])
prowler_results = parse_prowler_results(load_json_safe(prowler_path))
# Build summary table
table = Table(title="CI Security Gate Results")
table.add_column("Check", style="cyan")
table.add_column("Status", style="green")
table.add_column("Details", style="yellow")
all_conftest_failures = aws_conftest["failures"] + gcp_conftest["failures"]
conftest_status = "PASS" if len(all_conftest_failures) == 0 else "FAIL"
table.add_row(
"Conftest Policy",
conftest_status,
f"{len(all_conftest_failures)} violation(s)"
)
# Prowler gate: fail on critical or high findings
prowler_blocking = prowler_results["critical"] + prowler_results["high"]
prowler_status = "PASS" if prowler_blocking == 0 else "FAIL"
table.add_row(
"Prowler IaC",
prowler_status,
f"{prowler_results['critical']} critical, {prowler_results['high']} high, "
f"{prowler_results['medium']} medium, {prowler_results['low']} low"
)
console.print(table)
# Generate remediation guidance if there are failures
if conftest_status == "FAIL" or prowler_status == "FAIL":
console.print("\n[bold red]GATE: FAILED[/bold red]")
console.print("\n[bold]Remediation Guidance:[/bold]")
guidance = generate_remediation(all_conftest_failures, prowler_results["findings"])
for item in guidance:
console.print(f" {item}")
sys.exit(1)
else:
console.print("\n[bold green]GATE: PASSED[/bold green]")
console.print("All security checks passed. Safe to merge.")
if __name__ == "__main__":
main()What does the CI gate decide?
The gate fails the pipeline if any Conftest policy produces a denial OR if Prowler finds critical or high severity IaC issues. Medium and low findings are reported but do not block the merge. The script outputs actionable remediation guidance for every blocking finding.
- Create the governance module to formalize the policy enforcement in your Terraform codebase:
mkdir -p modules/governance- Add the governance module variables file:
# Governance module input variables
variable "project_id" {
description = "GCP project ID for governance scoping"
type = string
}
variable "aws_region" {
description = "AWS region for governance scoping"
type = string
}
variable "policy_path" {
description = "Path to OPA/Conftest policy directory"
type = string
default = "policies"
}
variable "prowler_scan_regions" {
description = "List of regions for Prowler scans"
type = list(string)
default = ["us-east-1"]
}- Add the governance module main configuration:
# Governance module - Policy enforcement hooks and Prowler scan configuration
# This module documents the governance tooling integrated via CI/CD
# OPA/Conftest and Prowler run in the GitHub Actions pipeline, not as Terraform resources
locals {
# Policy enforcement configuration
conftest_policy_dirs = [
"${var.policy_path}/aws",
"${var.policy_path}/gcp",
]
# Prowler scan configuration
prowler_aws_config = {
regions = var.prowler_scan_regions
output_formats = ["json-ocsf", "html"]
}
prowler_gcp_config = {
project_id = var.project_id
output_formats = ["json-ocsf", "html"]
}
# Gate thresholds
ci_gate_config = {
block_on_critical = true
block_on_high = true
block_on_medium = false
block_on_low = false
}
}Now push everything to GitHub to activate the CI pipeline.
- Stage and commit all new governance files by running:
git add policies/ scripts/test_policies.py scripts/prowler_report.py scripts/ci_gate.py .github/ modules/governance/
git commit -m "Add policy-as-code, compliance scanning, and CI pipeline"- Push to GitHub to trigger the workflow by running:
git push origin mainTest the CI pipeline with a PR
To see the full pipeline in action, create a feature branch, introduce an intentionally insecure resource (like a public S3 bucket), and open a pull request. The Security CI Pipeline workflow will run and the gate should block the merge.
Workflow not running?
GitHub Actions workflows only trigger on pull requests to main as configured. Make sure you push to a branch and open a PR against main. Also verify the workflow file is at .github/workflows/security-ci.yml with correct YAML indentation.
help me debug why my GitHub Actions security workflow is not triggering
✔️ Awesome, I've got everything!
Great. Double check you have saved all files and your commit includes the full policies/ directory, all three Python scripts, the GitHub Actions workflow, and the governance module.
ⓧ I'd like to double check the full code
Here is the directory structure you should have after this step:
policies/
aws/
no_public_storage.rego
encryption_required.rego
no_wildcard_iam.rego
vpc_flow_logs.rego
no_open_ssh_rdp.rego
gcp/
no_public_storage.rego
encryption_required.rego
no_wildcard_iam.rego
vpc_flow_logs.rego
no_open_ssh_rdp.rego
scripts/
test_policies.py
prowler_report.py
ci_gate.py
modules/governance/
main.tf
variables.tf
.github/workflows/
security-ci.ymlVerify your policy test harness passes by running python scripts/test_policies.py and confirming all tests show PASS.
Your infrastructure now has automated guardrails. Every pull request is scanned against security baselines before it can reach production. Next up, you will build the observability layer and generate architecture diagrams programmatically.
Add Observability, Diagrams-as-Code, and Portfolio Polish
You have built a fully automated security platform that detects threats, orchestrates incident response, enforces compliance, and correlates events across both clouds. But without a centralized view, operators have to jump between consoles to understand what is happening.
In this session you will add the observability layer that ties everything together. You will build a CloudWatch dashboard (both in Terraform and as a Python script), generate architecture diagrams from code, write a comprehensive README, and create a teardown validator that confirms no orphaned resources linger after cleanup.
In this step, get ready to:
- Deploy a centralized CloudWatch dashboard with cross-cloud security metrics and write a Python dashboard-as-code script.
- Generate architecture diagrams programmatically using the Python diagrams library and write a comprehensive README.
- Write a Python teardown validator that confirms all resources are destroyed after cleanup.
Deploy the observability module and dashboard-as-code script
The observability module creates an aws_cloudwatch_dashboard resource with widgets covering GuardDuty findings, Step Functions execution health, VPN tunnel status, and cross-cloud event pipeline throughput. You will also write a Python script that can recreate or update this dashboard independently of Terraform.
- Create the observability module directory by running this command:
- Create the module directory:
mkdir -p modules/observability- Create modules/observability/variables.tf and add the following input variables:
variable "project_name" {
description = "Project name prefix for resource naming"
type = string
}
variable "aws_region" {
description = "AWS region for CloudWatch resources"
type = string
}
variable "guardduty_detector_id" {
description = "GuardDuty detector ID for metric references"
type = string
}
variable "step_function_arn" {
description = "Step Functions state machine ARN for monitoring"
type = string
}
variable "vpn_connection_ids" {
description = "List of VPN connection IDs for tunnel monitoring"
type = list(string)
}What do these variables control?
Each variable feeds a specific dashboard widget. The guardduty_detector_id powers the threat detection panel, step_function_arn drives the orchestration metrics, and vpn_connection_ids reports tunnel health across all four IPsec tunnels.
- Create modules/observability/main.tf with the dashboard and alarm resources:
# Cross-cloud security metrics dashboard
resource "aws_cloudwatch_dashboard" "security" {
dashboard_name = "${var.project_name}-security-overview"
dashboard_body = jsonencode({
widgets = [
{
type = "metric"
x = 0
y = 0
width = 12
height = 6
properties = {
title = "GuardDuty Findings"
metrics = [
["AWS/GuardDuty", "FindingsCount", "DetectorId", var.guardduty_detector_id]
]
period = 300
stat = "Sum"
region = var.aws_region
}
},
{
type = "metric"
x = 12
y = 0
width = 12
height = 6
properties = {
title = "Step Functions Executions"
metrics = [
["AWS/States", "ExecutionsStarted", "StateMachineArn", var.step_function_arn],
["AWS/States", "ExecutionsFailed", "StateMachineArn", var.step_function_arn],
["AWS/States", "ExecutionsSucceeded", "StateMachineArn", var.step_function_arn]
]
period = 300
stat = "Sum"
region = var.aws_region
}
},
{
type = "metric"
x = 0
y = 6
width = 12
height = 6
properties = {
title = "VPN Tunnel Status"
metrics = [
for id in var.vpn_connection_ids :
["AWS/VPN", "TunnelState", "VpnId", id]
]
period = 60
stat = "Average"
region = var.aws_region
}
},
{
type = "metric"
x = 12
y = 6
width = 12
height = 6
properties = {
title = "Cross-Cloud Events Processed"
metrics = [
["ZeroTrust/CrossCloud", "EventsProcessed", "Pipeline", "SCC-to-CloudWatch"],
["ZeroTrust/CrossCloud", "EventsFailed", "Pipeline", "SCC-to-CloudWatch"]
]
period = 300
stat = "Sum"
region = var.aws_region
}
}
]
})
}
# Alarm for high-severity GuardDuty findings
resource "aws_cloudwatch_metric_alarm" "guardduty_high_severity" {
alarm_name = "${var.project_name}-guardduty-high-severity"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 1
metric_name = "FindingsCount"
namespace = "AWS/GuardDuty"
period = 300
statistic = "Sum"
threshold = 0
alarm_description = "Alert when GuardDuty detects high-severity findings"
dimensions = {
DetectorId = var.guardduty_detector_id
}
}
# Alarm for Step Functions execution failures
resource "aws_cloudwatch_metric_alarm" "stepfunctions_failures" {
alarm_name = "${var.project_name}-sfn-failures"
comparison_operator = "GreaterThanThreshold"
evaluation_periods = 1
metric_name = "ExecutionsFailed"
namespace = "AWS/States"
period = 300
statistic = "Sum"
threshold = 0
alarm_description = "Alert when Step Functions incident response fails"
dimensions = {
StateMachineArn = var.step_function_arn
}
}What does this module create?
The dashboard has four metric widgets arranged in a 2x2 grid. The top row shows GuardDuty findings count and Step Functions execution metrics. The bottom row shows VPN tunnel state and cross-cloud event pipeline throughput.
Two alarms fire when either GuardDuty detects findings or Step Functions executions fail. These alarms surface problems without requiring someone to watch the dashboard continuously.
- Create modules/observability/outputs.tf to export the dashboard name and alarm ARNs:
output "dashboard_name" {
description = "Name of the CloudWatch security dashboard"
value = aws_cloudwatch_dashboard.security.dashboard_name
}
output "alarm_arns" {
description = "ARNs of CloudWatch security alarms"
value = [
aws_cloudwatch_metric_alarm.guardduty_high_severity.arn,
aws_cloudwatch_metric_alarm.stepfunctions_failures.arn
]
}- Add the observability module block to your root main.tf. Find the last module block and add this below it:
module "observability" {
source = "./modules/observability"
project_name = var.project_name
aws_region = var.aws_region
guardduty_detector_id = module.detection.guardduty_detector_id
step_function_arn = module.orchestration.state_machine_arn
vpn_connection_ids = module.networking.vpn_connection_ids
}- Save all three module files and main.tf.
- Deploy the observability module by running:
terraform init
terraform apply -target=module.observability -auto-approve- Confirm the dashboard exists by checking the AWS CloudWatch console. You should see a dashboard named zero-trust-security-overview with four metric widgets.
Dashboard not appearing?
Verify that your detection and orchestration modules export the expected output names (guardduty_detector_id, state_machine_arn, vpn_connection_ids). If Terraform reports an unknown output, check the matching outputs.tf in that module.
help me debug my Terraform observability module deployment
Now write the Python dashboard-as-code script that can create or update this dashboard independently of Terraform.
- Create scripts/dashboard.py with the following code:
#!/usr/bin/env python3
"""Dashboard-as-code: programmatically create/update CloudWatch dashboard."""
import json
import argparse
from utils.logging_setup import get_logger
from utils.aws_client import get_aws_client
logger = get_logger(__name__)
# Dashboard name matches the Terraform-managed resource
DASHBOARD_NAME = "zero-trust-security-overview"
def build_dashboard_body(config):
"""Construct the dashboard JSON from configuration."""
widgets = [
{
"type": "metric",
"x": 0, "y": 0, "width": 12, "height": 6,
"properties": {
"title": "GuardDuty Findings (Last 24h)",
"metrics": [
["AWS/GuardDuty", "FindingsCount",
"DetectorId", config["guardduty_detector_id"]]
],
"period": 3600,
"stat": "Sum",
"region": config["region"]
}
},
{
"type": "metric",
"x": 12, "y": 0, "width": 12, "height": 6,
"properties": {
"title": "Incident Response Executions",
"metrics": [
["AWS/States", "ExecutionsStarted",
"StateMachineArn", config["sfn_arn"]],
["AWS/States", "ExecutionsFailed",
"StateMachineArn", config["sfn_arn"]],
["AWS/States", "ExecutionsSucceeded",
"StateMachineArn", config["sfn_arn"]]
],
"period": 300,
"stat": "Sum",
"region": config["region"]
}
},
{
"type": "metric",
"x": 0, "y": 6, "width": 12, "height": 6,
"properties": {
"title": "VPN Tunnel Health",
"metrics": [
["AWS/VPN", "TunnelState", "VpnId", vpn_id]
for vpn_id in config["vpn_connection_ids"]
],
"period": 60,
"stat": "Average",
"region": config["region"]
}
},
{
"type": "metric",
"x": 12, "y": 6, "width": 12, "height": 6,
"properties": {
"title": "Cross-Cloud Event Pipeline",
"metrics": [
["ZeroTrust/CrossCloud", "EventsProcessed",
"Pipeline", "SCC-to-CloudWatch"],
["ZeroTrust/CrossCloud", "EventsFailed",
"Pipeline", "SCC-to-CloudWatch"]
],
"period": 300,
"stat": "Sum",
"region": config["region"]
}
}
]
return json.dumps({"widgets": widgets})
def deploy_dashboard(config):
"""Create or update the CloudWatch dashboard."""
cw = get_aws_client("cloudwatch")
dashboard_body = build_dashboard_body(config)
# put_dashboard creates or replaces the entire dashboard
response = cw.put_dashboard(
DashboardName=DASHBOARD_NAME,
DashboardBody=dashboard_body
)
# Check for validation errors in the response
if response.get("DashboardValidationMessages"):
for msg in response["DashboardValidationMessages"]:
logger.warning(f"Dashboard validation: {msg['Message']}")
else:
logger.info(f"Dashboard '{DASHBOARD_NAME}' deployed successfully")
return DASHBOARD_NAME
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Deploy CloudWatch security dashboard"
)
parser.add_argument("--region", default="us-east-1",
help="AWS region")
parser.add_argument("--detector-id", required=True,
help="GuardDuty detector ID")
parser.add_argument("--sfn-arn", required=True,
help="Step Functions state machine ARN")
parser.add_argument("--vpn-ids", nargs="+", required=True,
help="VPN connection IDs")
args = parser.parse_args()
config = {
"region": args.region,
"guardduty_detector_id": args.detector_id,
"sfn_arn": args.sfn_arn,
"vpn_connection_ids": args.vpn_ids
}
deploy_dashboard(config)Why maintain both Terraform and a Python script?
Terraform manages the dashboard as infrastructure. The Python script lets operators update dashboard widgets without a full Terraform plan/apply cycle. It uses the boto3 put_dashboard method which creates or replaces the dashboard entirely.
This pattern is common in production SOC teams where dashboards evolve faster than infrastructure and need version-controlled updates outside the IaC pipeline.
- Save scripts/dashboard.py.
Generate architecture diagrams with Python diagrams-as-code
The Python diagrams library generates architecture diagrams programmatically. Each script produces a PNG image that auto-updates when you rerun it after infrastructure changes.
- Create the diagrams directory and output directory by running:
mkdir -p scripts/diagrams
mkdir -p output- Create scripts/diagrams/network_topology.py to generate the network architecture diagram:
#!/usr/bin/env python3
"""Generate network topology architecture diagram."""
from diagrams import Diagram, Cluster, Edge
from diagrams.aws.network import VPC, VPNGateway, PrivateSubnet
from diagrams.gcp.network import VPC as GVPC, VPN, Router
# Generate the cross-cloud network topology
with Diagram("Network Topology - HA VPN with BGP",
filename="output/network_topology", show=False):
with Cluster("AWS (10.0.0.0/16)"):
aws_vpc = VPC("AWS VPC")
aws_subnets = PrivateSubnet("Private Subnets")
aws_vgw = VPNGateway("VPN Gateway\nASN 64512")
aws_vpc >> aws_subnets >> aws_vgw
with Cluster("GCP (172.16.0.0/16)"):
gcp_vpc = GVPC("GCP VPC")
gcp_router = Router("Cloud Router\nASN 65001")
gcp_vpn = VPN("HA VPN Gateway")
gcp_vpc >> gcp_router >> gcp_vpn
# Four IPsec tunnels with BGP dynamic routing
aws_vgw >> Edge(label="4x IPsec/IKEv2\nBGP Dynamic Routing") >> gcp_vpnWhat does this code do?
The diagrams library uses Graphviz to render cloud architecture as PNG images. The Cluster context manager groups nodes visually. The >> operator draws directed edges between components.
Setting show=False prevents the image from auto-opening. The diagram saves to output/network_topology.png for inclusion in the README.
- Create scripts/diagrams/identity_flow.py to visualize the WIF authentication flow:
#!/usr/bin/env python3
"""Generate WIF identity flow architecture diagram."""
from diagrams import Diagram, Cluster, Edge
from diagrams.aws.security import IAM
from diagrams.aws.compute import Lambda
from diagrams.aws.storage import S3
from diagrams.gcp.security import IAP
from diagrams.gcp.compute import Functions
from diagrams.gcp.storage import GCS
# Visualize bidirectional keyless authentication
with Diagram("Bidirectional Workload Identity Federation",
filename="output/identity_flow", show=False):
with Cluster("AWS"):
aws_lambda = Lambda("WIF Lambda")
aws_iam = IAM("IAM Role")
aws_s3 = S3("S3 Bucket")
with Cluster("GCP"):
gcp_wif = IAP("WIF Pool")
gcp_func = Functions("Cloud Function")
gcp_gcs = GCS("GCS Bucket")
# AWS to GCP: Lambda exchanges STS token for GCP access
aws_lambda >> Edge(label="STS Token") >> gcp_wif
gcp_wif >> Edge(label="Short-lived\nAccess Token") >> gcp_gcs
# GCP to AWS: Cloud Function presents OIDC token
gcp_func >> Edge(label="OIDC Token") >> aws_iam
aws_iam >> Edge(label="AssumeRole") >> aws_s3- Create scripts/diagrams/detection_pipeline.py for the cross-cloud detection flow:
#!/usr/bin/env python3
"""Generate detection pipeline architecture diagram."""
from diagrams import Diagram, Cluster, Edge
from diagrams.aws.security import GuardDuty, SecurityHub
from diagrams.aws.integration import Eventbridge
from diagrams.aws.management import Cloudwatch
from diagrams.gcp.security import SecurityCommandCenter
from diagrams.gcp.analytics import PubSub
from diagrams.gcp.compute import Functions
# Show how findings flow from both clouds into unified detection
with Diagram("Cross-Cloud Detection Pipeline",
filename="output/detection_pipeline", show=False):
with Cluster("AWS Detection"):
guardduty = GuardDuty("GuardDuty")
securityhub = SecurityHub("Security Hub")
eventbridge = Eventbridge("EventBridge")
cloudwatch = Cloudwatch("CloudWatch Logs")
with Cluster("GCP Detection"):
scc = SecurityCommandCenter("Security Command\nCenter")
pubsub = PubSub("Pub/Sub")
bridge_func = Functions("SCC-to-CW\nCloud Function")
# AWS native detection flow
guardduty >> securityhub >> eventbridge
# GCP findings bridge into AWS via Cloud Function
scc >> pubsub >> bridge_func
bridge_func >> Edge(label="Normalized\nASFF") >> cloudwatch
cloudwatch >> eventbridge- Create scripts/diagrams/orchestration_workflow.py for the Step Functions incident response flow:
#!/usr/bin/env python3
"""Generate Step Functions orchestration workflow diagram."""
from diagrams import Diagram, Cluster, Edge
from diagrams.aws.integration import StepFunctions, Eventbridge
from diagrams.aws.compute import Lambda
from diagrams.aws.database import Dynamodb
from diagrams.aws.integration import SNS
# Visualize the incident response orchestration
with Diagram("Security Orchestration Workflow",
filename="output/orchestration_workflow", show=False):
eventbridge = Eventbridge("EventBridge\nTrigger")
with Cluster("Step Functions Orchestrator"):
triage = Lambda("Triage\n(Severity)")
with Cluster("Parallel Containment"):
isolate = Lambda("Isolate EC2")
revoke = Lambda("Revoke IAM Keys")
snapshot = Lambda("Snapshot\nForensics")
notify = Lambda("Notify SOC")
audit = Lambda("Audit Logger")
dynamodb = Dynamodb("DynamoDB\nAudit Trail")
sns = SNS("SNS\nAlerts")
# Orchestration flow: triage then parallel containment
eventbridge >> triage
triage >> isolate
triage >> revoke
triage >> snapshot
isolate >> notify
revoke >> notify
snapshot >> notify
notify >> sns
notify >> audit >> dynamodbWhat do these four scripts produce?
Each script generates a PNG diagram in the output/ directory. The four diagrams cover the main architectural pillars: network connectivity, identity federation, threat detection, and incident orchestration.
Because diagrams are code, they stay in version control and automatically reflect architecture changes when you rerun the scripts.
- Generate all four diagrams by running:
cd scripts/diagrams
python network_topology.py
python identity_flow.py
python detection_pipeline.py
python orchestration_workflow.py
cd ../..- Confirm the images were generated by running:
ls output/- You should see four PNG files: network_topology.png, identity_flow.png, detection_pipeline.png, and orchestration_workflow.png.
Seeing a Graphviz error?
The diagrams library requires Graphviz installed as a system dependency. Install it with brew install graphviz (macOS) or sudo apt install graphviz (Linux), then rerun the scripts.
help me fix Graphviz installation for the Python diagrams library
Now create the project README that ties everything together.
- Create README.md at the project root with the following comprehensive documentation:
# Enterprise Zero-Trust Multi-Cloud Security Platform
## Architecture Overview
A production-grade security platform spanning AWS and GCP, implementing all five
pillars of Zero Trust: Identity, Network, Detection, Governance, and Observability.
## Threat Model
| Threat | Mitigation | Module |
|--------|-----------|--------|
| Credential theft | Workload Identity Federation (zero static keys) | identity |
| Lateral movement | VPC isolation + security group quarantine | networking, orchestration |
| Data exfiltration | GuardDuty S3 protection + SCC monitoring | detection |
| Privilege escalation | IAM key revocation + deny policy attachment | orchestration |
| Compliance drift | OPA/Conftest + Prowler CIS scanning | governance |
| Undetected breach | Cross-cloud event correlation | detection |
## Security Rationale
- **No static credentials**: All cross-cloud authentication uses short-lived tokens
via Workload Identity Federation and STS AssumeRoleWithWebIdentity.
- **Defense in depth**: GuardDuty (AWS) + Security Command Center (GCP) provide
independent detection that feeds into unified correlation.
- **Automated response**: Step Functions orchestrates parallel containment within
seconds of detection, eliminating manual toil.
- **Policy-as-code**: OPA/Conftest gates prevent non-compliant infrastructure from
reaching production via CI pipeline enforcement.
## Cost Analysis
| Resource | Cost | Notes |
|----------|------|-------|
| HA VPN (4 tunnels) | ~$0.40/hr | Tear down after verification |
| GuardDuty | Free | 30-day trial |
| Lambda/Cloud Functions | Free | Within free tier |
| Step Functions | Free | 4,000 transitions/month free |
| DynamoDB | Free | 25 GB free tier |
| S3/GCS | ~$0.01 | Test objects only |
| **Total (8hr session)** | **~$3-5** | Tear down VPN same day |
## Deployment Guide
### Prerequisites
- Terraform >= 1.5
- AWS CLI configured with appropriate permissions
- gcloud CLI configured with project access
- Python 3.11+ with virtual environment
### Quick Start
```bash
# Clone and setup
git clone <repo-url>
cd zero-trust-multicloud
python3 -m venv venv && source venv/bin/activate
pip install -r requirements.txt
# Deploy infrastructure
terraform init
terraform plan
terraform apply
# Validate deployment
python scripts/validate_networking.py
python scripts/test_wif_integration.py
python scripts/simulate_guardduty_findings.py
```
### Teardown
```bash
terraform destroy
python scripts/validate_cleanup.py --region us-east-1 --gcp-project <project-id>
```
## Module Structure
```
modules/
networking/ - HA VPN, VPCs, BGP routing
identity/ - WIF pools, IAM roles, cross-cloud auth
detection/ - GuardDuty, SCC, EventBridge, CloudWatch
orchestration/ - Step Functions, Lambda, DynamoDB audit
governance/ - OPA/Conftest, Prowler integration
observability/ - CloudWatch dashboard, alarms
scripts/
lambda/ - Python Lambda functions
cloud_functions/ - Python Cloud Functions
diagrams/ - Architecture diagram generators
utils/ - Shared Python utilities
```What makes a strong security portfolio README?
Employers look for threat models, security rationale, and cost awareness. The threat model shows you think like an attacker. The security rationale demonstrates you chose tools for specific reasons. The cost analysis proves you understand operational constraints.
- Save README.md and all diagram scripts.
Write the teardown validator and finalize the repository
After running terraform destroy, some resources can persist as orphans if Terraform loses track of them. A validator script that queries both cloud APIs gives confidence that nothing billable lingers.
- Create scripts/validate_cleanup.py with the following code:
#!/usr/bin/env python3
"""Teardown validator: confirm all resources destroyed after terraform destroy."""
import argparse
import boto3
from rich.console import Console
from rich.table import Table
from utils.logging_setup import get_logger
from utils.aws_client import get_aws_client
from utils.gcp_client import get_gcp_client
logger = get_logger(__name__)
console = Console()
# Prefix used across all project resources
PROJECT_PREFIX = "zero-trust"
def check_aws_resources(region="us-east-1"):
"""Check for orphaned AWS resources matching project prefix."""
orphaned = []
# Check CloudWatch Log Groups
logs = get_aws_client("logs", region_name=region)
paginator = logs.get_paginator("describe_log_groups")
for page in paginator.paginate(
logGroupNamePrefix=f"/aws/{PROJECT_PREFIX}"
):
for group in page["logGroups"]:
orphaned.append(
("AWS", "CloudWatch Log Group", group["logGroupName"])
)
# Check DynamoDB tables
dynamodb = get_aws_client("dynamodb", region_name=region)
tables = dynamodb.list_tables()["TableNames"]
for table in tables:
if PROJECT_PREFIX in table:
orphaned.append(("AWS", "DynamoDB Table", table))
# Check IAM roles
iam = get_aws_client("iam")
paginator = iam.get_paginator("list_roles")
for page in paginator.paginate():
for role in page["Roles"]:
if PROJECT_PREFIX in role["RoleName"]:
orphaned.append(("AWS", "IAM Role", role["RoleName"]))
# Check Lambda functions
lam = get_aws_client("lambda", region_name=region)
functions = lam.list_functions()["Functions"]
for func in functions:
if PROJECT_PREFIX in func["FunctionName"]:
orphaned.append(
("AWS", "Lambda Function", func["FunctionName"])
)
return orphaned
def check_gcp_resources(project_id):
"""Check for orphaned GCP resources matching project prefix."""
orphaned = []
# Check GCS buckets
storage_client = get_gcp_client("storage")
buckets = storage_client.list_buckets(project=project_id)
for bucket in buckets:
if PROJECT_PREFIX in bucket.name:
orphaned.append(("GCP", "GCS Bucket", bucket.name))
# Check service accounts
from googleapiclient.discovery import build
iam_service = build("iam", "v1")
sa_list = iam_service.projects().serviceAccounts().list(
name=f"projects/{project_id}"
).execute()
for account in sa_list.get("accounts", []):
if PROJECT_PREFIX in account["email"]:
orphaned.append(
("GCP", "Service Account", account["email"])
)
return orphaned
def display_results(orphaned):
"""Display orphaned resources in a formatted table."""
if not orphaned:
console.print(
"\n[bold green]All clear! No orphaned resources found."
"[/bold green]\n"
)
return True
table = Table(title="Orphaned Resources Detected")
table.add_column("Cloud", style="cyan")
table.add_column("Type", style="magenta")
table.add_column("Resource", style="red")
for cloud, rtype, name in orphaned:
table.add_row(cloud, rtype, name)
console.print(table)
console.print(
f"\n[bold red]Found {len(orphaned)} orphaned resource(s)!"
f"[/bold red]"
)
return False
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Validate resource cleanup after terraform destroy"
)
parser.add_argument("--region", default="us-east-1",
help="AWS region to check")
parser.add_argument("--gcp-project", required=True,
help="GCP project ID to check")
args = parser.parse_args()
logger.info("Checking for orphaned resources...")
aws_orphaned = check_aws_resources(args.region)
gcp_orphaned = check_gcp_resources(args.gcp_project)
all_orphaned = aws_orphaned + gcp_orphaned
clean = display_results(all_orphaned)
# Exit with non-zero status if orphans found
exit(0 if clean else 1)What does this script check?
The script scans four AWS resource types (CloudWatch Log Groups, DynamoDB tables, IAM roles, Lambda functions) and two GCP resource types (GCS buckets, service accounts) for anything containing the project prefix.
It uses the rich library for a formatted table output and exits with a non-zero status code if orphans are found. This makes it usable in CI pipelines as a post-destroy validation gate.
- Save scripts/validate_cleanup.py.
- Commit and push all new files to GitHub by running:
git add modules/observability/ scripts/dashboard.py scripts/diagrams/ scripts/validate_cleanup.py README.md output/
git commit -m "Add observability module, diagrams-as-code, README, and cleanup validator"
git push origin main- Confirm the push succeeded by checking your GitHub repository. You should see the new modules/observability/ directory, scripts/diagrams/, scripts/validate_cleanup.py, and README.md in the file listing.
Push rejected?
If you see a rejection due to remote changes, pull first with git pull --rebase origin main then push again. If the CI pipeline is running on push, you may need to wait for it to complete before the next push.
help me resolve a git push rejection for my security platform repository
Your platform now has full observability, version-controlled architecture diagrams, professional documentation, and a safety-net cleanup validator. In the Secret Mission, you will extend this platform with a cross-cloud honeypot system that actively detects attackers.
Secret mission
Cross-Cloud Honeypot System
Your security platform detects threats reactively. What if you could lure attackers into revealing themselves? Build a cross-cloud honeypot system that deploys decoy resources in both AWS and GCP, generates trackable canary tokens, and automatically triggers your Step Functions orchestrator when an intruder takes the bait.
Clean Up Your Resources
Clean Up Your Resources
Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project provisions cloud resources across both AWS and GCP, with HA VPN tunnels being the primary ongoing cost.
Cost warning
The HA VPN tunnels are the main cost driver in this project. GCP charges per tunnel hourly, and AWS charges $0.05 per VPN connection-hour. Combined, the four tunnels and two connections cost approximately $0.40/hr (~$9.60/day, ~$290/month).
All other services (Lambda, Cloud Functions, Step Functions, DynamoDB, S3/GCS, Pub/Sub, EventBridge) operate within free-tier limits at this project's scale. GuardDuty is free for the 30-day trial period. If you leave only the VPN running, tear it down immediately after verifying connectivity.
Resources you used:
- AWS: VPC, VPN Gateway, VPN Connections, Lambda functions (7), Step Functions state machine, DynamoDB table, GuardDuty detector, Security Hub, EventBridge rules, CloudWatch dashboard/alarms/log groups, S3 buckets (including honeypot), IAM roles/policies, SNS topics, CloudTrail data events.
- GCP: VPC, HA VPN Gateway, Cloud Router, Cloud Functions (2), Pub/Sub topics/subscriptions, GCS buckets (including honeypot), Service Accounts, Workload Identity Pool/Provider, Data Access audit logs.
- Local: Project files (Terraform monorepo, Python scripts, Rego policies, GitHub Actions workflow), Python virtual environment.
Keep everything running
No action needed. Choose this if you are still actively building or testing the platform.
- The HA VPN tunnels will continue to cost approximately $0.40/hr (~$9.60/day). If you only need the identity, detection, orchestration, and honeypot infrastructure, consider tearing down just the networking module to save ~$290/month.
- Lambda, Cloud Functions, Step Functions, DynamoDB, and storage remain effectively free at this project's scale.
- GuardDuty is free for the 30-day trial. After that, charges apply based on data volume analyzed.
- Your Python virtual environment, Prowler installation, and local project files remain on your machine.
Pause - I'll come back to this later
Remove the VPN tunnels (the only significant cost) while keeping identity, detection, orchestration, and honeypot infrastructure for continued learning.
- Destroy only the networking module by running this command from your project directory:
terraform destroy -target=module.networking- This removes the AWS VPC, VPN Gateway, VPN Connections, GCP VPC, HA VPN Gateway, Cloud Router, and all four IPsec tunnels.
- The remaining infrastructure (Lambda functions, Step Functions, DynamoDB, GuardDuty, Security Hub, honeypot resources) costs effectively nothing at this scale.
- When you return, run terraform apply to rebuild the networking module and reconnect the clouds.
Delete - I don't want to use this again
Remove all project resources from both clouds and delete local files. This is irreversible.
- Destroy all Terraform-managed resources by running this from your project directory:
terraform destroy- Type yes when prompted to confirm destruction of all resources across both AWS and GCP.
- Run the cleanup validator script to confirm everything is gone:
python scripts/validate_cleanup.py- This script queries both AWS and GCP APIs to confirm all resources have been destroyed, flagging any orphaned resources (CloudWatch Log Groups, DynamoDB tables, IAM roles, service accounts).
- Deactivate your Python virtual environment:
deactivate- Remove your local project files. On macOS/Linux:
cd ~/Desktop
rm -rf zero-trust-multicloudOn Windows:
cd ~/Desktop
rmdir /S /Q zero-trust-multicloud- If the cleanup validator flagged any orphaned resources, manually delete them from the AWS Console or GCP Console following the specific paths the script outputs.
Nice Work!
Nice Work!
Nice work! You have built an enterprise-grade multi-cloud security platform with a Python automation backbone that mirrors what production SOC teams deploy and operate.
What you learned:
- Architected cross-cloud private networking via HA VPN with BGP dynamic routing and implemented bidirectional Workload Identity Federation that eliminated all static credentials across AWS and GCP, validated by Python SDK integration tests.
- Built automated security orchestration with Step Functions coordinating Python Lambda functions for parallel incident response, forensic capture, DynamoDB audit trails, and cross-cloud event correlation from GCP Security Command Center through to AWS alerting.
- Implemented multi-cloud compliance governance with OPA/Conftest policy-as-code gates, Prowler CIS benchmark scanning with a Python report aggregator, and GitHub Actions CI pipelines enforcing security checks on every pull request.
- Secret Mission: Deployed a cross-cloud honeypot system with a Python canary token generator creating trackable fake credentials, decoy resources in both clouds, and wired honeypot access events into the Step Functions orchestrator for automated threat response.
Ready to quiz yourself?