How to Use AWS Security Hub 2026 and GCP SCC to Manage Multi-Cloud Compliance

# Provider configuration for AWS Security Hub 2026 setup
# Requires Terraform 1.7+ and AWS provider v5.42+ (adds Security Hub 2026 API support)
terraform {
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.42"
    }
  }
  required_version = "~> 1.7"
}

# Configure AWS provider with region and credentials (use env vars for prod)
provider "aws" {
  region = var.aws_region
  # Credentials are pulled from AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY env vars
  # Never hardcode credentials in Terraform files
}

# Variables for AWS configuration
variable "aws_region" {
  type        = string
  description = "AWS region to deploy Security Hub resources"
  default     = "us-east-1"
}

variable "security_hub_admin_account_id" {
  type        = string
  description = "AWS account ID designated as Security Hub administrator"
}

# Enable AWS Security Hub 2026 with CIS 1.4.0 standard
resource "aws_securityhub_account" "main" {
  enable_default_standards = false # We enable CIS 1.4.0 explicitly
  auto_enable_controls     = true
  # Security Hub 2026 specific: enable ML-based false positive filtering
  feature_options {
    ml_classification = "ENABLED"
  }
}

# Enable CIS AWS Foundations Benchmark v1.4.0 (supported natively in Security Hub 2026)
resource "aws_securityhub_standards_subscription" "cis_140" {
  depends_on = [aws_securityhub_account.main]
  standards_arn = "arn:aws:securityhub:${var.aws_region}::standards/cis-aws-foundations-benchmark/v/1.4.0"
}

# Create S3 bucket to export Security Hub findings (normalized to Parquet for downstream processing)
resource "aws_s3_bucket" "security_hub_findings" {
  bucket        = "org-multi-cloud-compliance-hub-findings-${data.aws_caller_identity.current.account_id}"
  force_destroy = true # Only for demo; use lifecycle rules for prod
  tags = {
    Purpose = "Multi-cloud compliance findings storage"
  }
}

# Block public access to S3 bucket
resource "aws_s3_bucket_public_access_block" "findings_block" {
  bucket = aws_s3_bucket.security_hub_findings.id
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

# IAM role to allow Security Hub to write findings to S3
resource "aws_iam_role" "security_hub_s3_write" {
  name = "SecurityHubS3WriteRole"
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = "sts:AssumeRole"
        Effect = "Allow"
        Principal = {
          Service = "securityhub.amazonaws.com"
        }
      }
    ]
  })
}

# Attach policy to IAM role for S3 write access
resource "aws_iam_role_policy" "security_hub_s3_policy" {
  name = "SecurityHubS3WritePolicy"
  role = aws_iam_role.security_hub_s3_write.id
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Action = [
          "s3:PutObject",
          "s3:PutObjectAcl"
        ]
        Effect   = "Allow"
        Resource = "${aws_s3_bucket.security_hub_findings.arn}/*"
      }
    ]
  })
}

# Get current AWS account ID for dynamic resource naming
data "aws_caller_identity" "current" {}

# Outputs for downstream integration
output "security_hub_arn" {
  value = aws_securityhub_account.main.arn
}

output "findings_s3_bucket_name" {
  value = aws_s3_bucket.security_hub_findings.id
}

# Provider configuration for GCP Security Command Center v3.4 setup
# Requires Terraform 1.7+ and Google provider v5.32+ (adds SCC v3.4 API support)
terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 5.32"
    }
  }
  required_version = "~> 1.7"
}

# Configure GCP provider with project and credentials (use env vars for prod)
provider "google" {
  project = var.gcp_project_id
  region  = var.gcp_region
  # Credentials are pulled from GOOGLE_APPLICATION_CREDENTIALS env var
  # Never hardcode service account keys in Terraform files
}

# Variables for GCP configuration
variable "gcp_project_id" {
  type        = string
  description = "GCP project ID to deploy SCC resources"
}

variable "gcp_region" {
  type        = string
  description = "GCP region to deploy SCC resources"
  default     = "us-central1"
}

# Enable GCP Security Command Center v3.4 API
resource "google_project_service" "scc_api" {
  service = "securitycenter.googleapis.com"
  disable_on_destroy = false # Keep API enabled even if Terraform destroys resources
}

# Enable CIS GCP Foundations Benchmark v1.4.0 in SCC v3.4
resource "google_scc_organization_security_policy" "cis_140" {
  depends_on = [google_project_service.scc_api]
  organization = var.gcp_organization_id
  display_name = "CIS-GCP-1.4.0-Policy"
  policy = jsonencode({
    admission_rules = []
    # SCC v3.4 native CIS 1.4.0 module configuration
    cis_benchmark_policy = {
      benchmark_version = "1.4.0"
      enabled           = true
    }
  })
}

# Create BigQuery dataset to export SCC findings (normalized to SQL for downstream processing)
resource "google_bigquery_dataset" "scc_findings" {
  dataset_id    = "multi_cloud_compliance_findings"
  project       = var.gcp_project_id
  friendly_name = "Multi-Cloud Compliance Findings"
  description   = "Stores normalized GCP SCC v3.4 findings for unified compliance dashboard"
  location      = var.gcp_region
  # Set default table expiration to 90 days for compliance retention
  default_table_expiration_ms = 7776000000 # 90 days in ms
}

# Service account for SCC to write findings to BigQuery
resource "google_service_account" "scc_bigquery_write" {
  account_id   = "scc-bigquery-write"
  display_name = "SCC BigQuery Write Service Account"
  project      = var.gcp_project_id
}

# Grant BigQuery Data Owner role to service account for findings export
resource "google_project_iam_member" "scc_bigquery_role" {
  project = var.gcp_project_id
  role    = "roles/bigquery.dataOwner"
  member  = "serviceAccount:${google_service_account.scc_bigquery_write.email}"
}

# Configure SCC v3.4 to export findings to BigQuery
resource "google_scc_project_notification_config" "bigquery_export" {
  config_id    = "scc-bigquery-export"
  project      = var.gcp_project_id
  location     = "global"
  description  = "Export SCC findings to BigQuery for unified compliance processing"
  pubsub_topic = google_pubsub_topic.scc_findings_topic.id
  # Filter to only export high and critical severity findings
  streaming_config = {
    filter = "severity: HIGH OR severity: CRITICAL"
  }
}

# Pub/Sub topic to intermediate SCC findings before BigQuery load
resource "google_pubsub_topic" "scc_findings_topic" {
  name    = "scc-findings-export-topic"
  project = var.gcp_project_id
}

# Cloud Function to load Pub/Sub messages to BigQuery (handles normalization)
resource "google_cloudfunctions2_function" "scc_to_bigquery" {
  name        = "scc-findings-to-bigquery"
  location    = var.gcp_region
  project     = var.gcp_project_id
  description = "Normalizes SCC v3.4 findings and loads to BigQuery"

  build_config {
    runtime     = "python312"
    entry_point = "load_to_bigquery"
    source {
      storage_source {
        bucket = google_storage_bucket.function_source.name
        object = google_storage_bucket_object.function_zip.name
      }
    }
  }

  service_config {
    service_account_email = google_service_account.scc_bigquery_write.email
    timeout_seconds      = 60
  }

  event_trigger {
    trigger_region = var.gcp_region
    event_type     = "google.cloud.pubsub.topic.v1.messagePublished"
    pubsub_topic   = google_pubsub_topic.scc_findings_topic.id
  }
}

# GCS bucket to store Cloud Function source code
resource "google_storage_bucket" "function_source" {
  name     = "org-multi-cloud-scc-function-source-${var.gcp_project_id}"
  location = var.gcp_region
}

# Upload function source code zip to GCS
resource "google_storage_bucket_object" "function_zip" {
  name   = "scc-to-bigquery.zip"
  bucket = google_storage_bucket.function_source.name
  source = path.module + "/function-source/scc-to-bigquery.zip" # Local zip file path
}

# Variables for GCP org
variable "gcp_organization_id" {
  type        = string
  description = "GCP organization ID for SCC policy attachment"
}

# Outputs for downstream integration
output "scc_findings_bigquery_dataset" {
  value = google_bigquery_dataset.scc_findings.dataset_id
}

output "scc_service_account_email" {
  value = google_service_account.scc_bigquery_write.email
}

# Unified Multi-Cloud Compliance Normalizer
# Python 3.12+ required, dependencies: boto3, google-cloud-bigquery, pandas, pyarrow, sqlalchemy
import os
import json
import logging
from typing import List, Dict, Any
import boto3
from google.cloud import bigquery
import pandas as pd
from sqlalchemy import create_engine, text
from datetime import datetime

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

# Common compliance finding schema (normalized across AWS and GCP)
COMMON_SCHEMA = {
    "finding_id": str,
    "source": str, # "AWS_SECURITY_HUB" or "GCP_SCC"
    "account_id": str,
    "resource_arn": str,
    "resource_type": str,
    "control_id": str, # CIS control ID e.g., "CIS-1.1"
    "control_title": str,
    "severity": str, # CRITICAL, HIGH, MEDIUM, LOW
    "status": str, # ACTIVE, RESOLVED, SUPPRESSED
    "first_observed": datetime,
    "last_observed": datetime,
    "description": str,
    "remediation_steps": str
}

class ComplianceNormalizer:
    def __init__(self):
        # Initialize AWS S3 client for Security Hub findings
        self.s3_client = boto3.client(
            "s3",
            region_name=os.getenv("AWS_REGION", "us-east-1"),
            aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
            aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY")
        )
        self.aws_findings_bucket = os.getenv("AWS_FINDINGS_BUCKET")
        if not self.aws_findings_bucket:
            raise ValueError("AWS_FINDINGS_BUCKET environment variable not set")

        # Initialize GCP BigQuery client for SCC findings
        self.bq_client = bigquery.Client(project=os.getenv("GCP_PROJECT_ID"))
        self.bq_dataset = os.getenv("GCP_BQ_DATASET", "multi_cloud_compliance_findings")
        self.gcp_project_id = os.getenv("GCP_PROJECT_ID")
        if not self.gcp_project_id:
            raise ValueError("GCP_PROJECT_ID environment variable not set")

        # Initialize unified database (PostgreSQL for demo, use RDS/Cloud SQL for prod)
        self.db_engine = create_engine(
            os.getenv("UNIFIED_DB_URL", "postgresql://user:pass@localhost:5432/compliance")
        )
        # Create unified findings table if not exists
        self._init_db()

    def _init_db(self):
        """Create unified compliance findings table with common schema"""
        try:
            with self.db_engine.connect() as conn:
                conn.execute(text("""
                    CREATE TABLE IF NOT EXISTS unified_compliance_findings (
                        finding_id VARCHAR(255) PRIMARY KEY,
                        source VARCHAR(50) NOT NULL,
                        account_id VARCHAR(255) NOT NULL,
                        resource_arn TEXT NOT NULL,
                        resource_type VARCHAR(255) NOT NULL,
                        control_id VARCHAR(50) NOT NULL,
                        control_title TEXT NOT NULL,
                        severity VARCHAR(50) NOT NULL,
                        status VARCHAR(50) NOT NULL,
                        first_observed TIMESTAMP NOT NULL,
                        last_observed TIMESTAMP NOT NULL,
                        description TEXT,
                        remediation_steps TEXT,
                        created_at TIMESTAMP DEFAULT NOW()
                    )
                """))
                conn.commit()
                logger.info("Unified compliance database table initialized")
        except Exception as e:
            logger.error(f"Failed to initialize database: {str(e)}")
            raise

    def _normalize_aws_finding(self, raw_finding: Dict[str, Any]) -> Dict[str, Any]:
        """Normalize AWS Security Hub 2026 finding to common schema"""
        try:
            normalized = {
                "finding_id": raw_finding["Id"],
                "source": "AWS_SECURITY_HUB",
                "account_id": raw_finding["AwsAccountId"],
                "resource_arn": raw_finding["Resources"][0]["Id"] if raw_finding.get("Resources") else "",
                "resource_type": raw_finding["Resources"][0]["Type"] if raw_finding.get("Resources") else "",
                "control_id": raw_finding["Compliance"]["SecurityControlId"],
                "control_title": raw_finding["Title"],
                "severity": raw_finding["Severity"]["Label"],
                "status": raw_finding["Compliance"]["Status"],
                "first_observed": datetime.fromisoformat(raw_finding["FirstObservedAt"].replace("Z", "+00:00")),
                "last_observed": datetime.fromisoformat(raw_finding["LastObservedAt"].replace("Z", "+00:00")),
                "description": raw_finding["Description"],
                "remediation_steps": json.dumps(raw_finding.get("Remediation", {}).get("Recommendation", {}))
            }
            return normalized
        except KeyError as e:
            logger.warning(f"Missing key {str(e)} in AWS finding {raw_finding.get('Id')}, skipping")
            return None
        except Exception as e:
            logger.error(f"Failed to normalize AWS finding: {str(e)}")
            return None

    def _normalize_gcp_finding(self, raw_finding: Dict[str, Any]) -> Dict[str, Any]:
        """Normalize GCP SCC v3.4 finding to common schema"""
        try:
            normalized = {
                "finding_id": raw_finding["name"].split("/")[-1],
                "source": "GCP_SCC",
                "account_id": raw_finding["parent"].split("/")[1] if "projects/" in raw_finding["parent"] else "",
                "resource_arn": raw_finding["resourceName"],
                "resource_type": raw_finding["resourceType"],
                "control_id": raw_finding["securityMarks"]["marks"].get("cis_control_id", ""),
                "control_title": raw_finding["category"],
                "severity": raw_finding["severity"],
                "status": "ACTIVE" if raw_finding["state"] == "ACTIVE" else "RESOLVED",
                "first_observed": datetime.fromisoformat(raw_finding["createTime"].replace("Z", "+00:00")),
                "last_observed": datetime.fromisoformat(raw_finding["updateTime"].replace("Z", "+00:00")),
                "description": raw_finding["description"],
                "remediation_steps": json.dumps(raw_finding.get("nextSteps", []))
            }
            return normalized
        except KeyError as e:
            logger.warning(f"Missing key {str(e)} in GCP finding {raw_finding.get('name')}, skipping")
            return None
        except Exception as e:
            logger.error(f"Failed to normalize GCP finding: {str(e)}")
            return None

    def fetch_aws_findings(self) -> List[Dict[str, Any]]:
        """Fetch raw findings from AWS S3 bucket (Parquet format)"""
        findings = []
        try:
            # List all Parquet files in S3 bucket
            paginator = self.s3_client.get_paginator("list_objects_v2")
            for page in paginator.paginate(Bucket=self.aws_findings_bucket, Prefix="security-hub-findings/"):
                for obj in page.get("Contents", []):
                    if obj["Key"].endswith(".parquet"):
                        # Download Parquet file to memory
                        response = self.s3_client.get_object(Bucket=self.aws_findings_bucket, Key=obj["Key"])
                        df = pd.read_parquet(response["Body"])
                        # Convert to list of dicts
                        for _, row in df.iterrows():
                            findings.append(row.to_dict())
            logger.info(f"Fetched {len(findings)} raw AWS Security Hub findings")
            return findings
        except Exception as e:
            logger.error(f"Failed to fetch AWS findings: {str(e)}")
            return []

    def fetch_gcp_findings(self) -> List[Dict[str, Any]]:
        """Fetch raw findings from GCP BigQuery dataset"""
        findings = []
        try:
            query = f"""
                SELECT * FROM `{self.gcp_project_id}.{self.bq_dataset}.scc_findings`
                WHERE update_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
            """
            query_job = self.bq_client.query(query)
            results = query_job.result()
            for row in results:
                findings.append(dict(row))
            logger.info(f"Fetched {len(findings)} raw GCP SCC findings")
            return findings
        except Exception as e:
            logger.error(f"Failed to fetch GCP findings: {str(e)}")
            return []

    def normalize_and_load(self):
        """Main method: fetch, normalize, load to unified DB"""
        # Fetch and normalize AWS findings
        raw_aws = self.fetch_aws_findings()
        normalized_aws = []
        for f in raw_aws:
            norm = self._normalize_aws_finding(f)
            if norm:
                normalized_aws.append(norm)
        logger.info(f"Normalized {len(normalized_aws)} AWS findings")

        # Fetch and normalize GCP findings
        raw_gcp = self.fetch_gcp_findings()
        normalized_gcp = []
        for f in raw_gcp:
            norm = self._normalize_gcp_finding(f)
            if norm:
                normalized_gcp.append(norm)
        logger.info(f"Normalized {len(normalized_gcp)} GCP findings")

        # Load all normalized findings to unified DB
        all_findings = normalized_aws + normalized_gcp
        if all_findings:
            try:
                df = pd.DataFrame(all_findings)
                df.to_sql("unified_compliance_findings", self.db_engine, if_exists="append", index=False)
                logger.info(f"Loaded {len(all_findings)} total findings to unified database")
            except Exception as e:
                logger.error(f"Failed to load findings to DB: {str(e)}")
        else:
            logger.warning("No normalized findings to load")

if __name__ == "__main__":
    try:
        normalizer = ComplianceNormalizer()
        normalizer.normalize_and_load()
        logger.info("Compliance normalization run completed successfully")
    except Exception as e:
        logger.error(f"Normalization run failed: {str(e)}")
        exit(1)

# AWS CLI command to check ML-classified false positive counts
aws securityhub get-findings \
  --filters '{"MLClassification": [{"Value": "false-positive", "Comparison": "EQUALS"}]}' \
  --query "length(Findings)"
# Compare with total findings count:
aws securityhub get-findings --query "length(Findings)"

# GCloud command to list SCC findings with CIS 1.4.0 mapping
gcloud scc findings list \
  --organization=1234567890 \
  --filter "securityMarks.marks.cis_control_id != ''" \
  --format="table(name, cis_control_id, severity, state)"

# Deduplication check in Python normalizer
def _is_duplicate(self, finding: Dict[str, Any]) -> bool:
    try:
        with self.db_engine.connect() as conn:
            result = conn.execute(text("""
                SELECT 1 FROM unified_compliance_findings
                WHERE control_id = :control_id
                AND resource_arn = :resource_arn
                AND status = 'ACTIVE'
                LIMIT 1
            """), {"control_id": finding["control_id"], "resource_arn": finding["resource_arn"]})
            return result.scalar() is not None
    except Exception as e:
        logger.error(f"Deduplication check failed: {str(e)}")
        return False

multi-cloud-compliance-pipeline/
├── terraform/
│   ├── aws/
│   │   ├── main.tf          # AWS Security Hub 2026 Terraform config
│   │   ├── variables.tf
│   │   └── outputs.tf
│   └── gcp/
│       ├── main.tf          # GCP SCC v3.4 Terraform config
│       ├── variables.tf
│       └── outputs.tf
├── src/
│   ├── normalizer.py        # Unified compliance normalizer (Python 3.12)
│   ├── requirements.txt     # Python dependencies
│   └── utils/
│       ├── aws_client.py    # AWS S3 helper functions
│       └── gcp_client.py    # GCP BigQuery helper functions
├── function-source/        # GCP Cloud Function source code
│   └── scc-to-bigquery/
│       ├── main.py
│       └── requirements.txt
├── .env.example            # Example environment variables
├── LICENSE                 # MIT License
└── README.md               # Setup and deployment instructions