GCP - OPA basics

GCP Terraform Infrastructure with OPA Policy Validation

This document demonstrates how to create GCP infrastructure (Compute, Storage, Firewall) using Terraform, validate it with Open Policy Agent (OPA), and run Terraform apply only if policies pass.

1. Directory Structure

GCP_OPA_Project/
│
├── terraform/              # Terraform configuration
│   └── main.tf
├── policy/                 # OPA policies
│   ├── main.rego
│   ├── gcs.rego
│   ├── compute.rego
│   ├── firewall.rego
│   └── iam_gcp.rego
├── plan.json               # Terraform plan in JSON (generated)
└── validate_apply_gcp.py  # Python orchestrator script

2. Terraform Configuration (main.tf)

terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 5.0"
    }
  }
}

provider "google" {
  project     = "YOUR_PROJECT_ID"
  region      = "us-central1"
  zone        = "us-central1-a"
  credentials = file("c:\\test\\credial\\key.json")
}

resource "random_id" "rand" { byte_length = 4 }

resource "google_storage_bucket" "demo_bucket" {
  name     = "my-demo-bucket-${random_id.rand.hex}"
  location = "US"
  storage_class = "STANDARD"
  force_destroy = true
  uniform_bucket_level_access = true
}

resource "google_compute_firewall" "default_allow_ssh" {
  name    = "allow-ssh"
  network = "default"
  allow {
    protocol = "tcp"
    ports    = ["22"]
  }
  source_ranges = ["0.0.0.0/0"]
  target_tags   = ["ssh-allowed"]
}

resource "google_compute_instance" "demo_vm" {
  name         = "demo-vm"
  machine_type = "e2-micro"
  zone         = "us-central1-a"
  tags = ["ssh-allowed"]
  boot_disk {
    initialize_params {
      image = "debian-cloud/debian-11"
      size  = 30
    }
  }
  network_interface {
    network = "default"
    access_config { }
  }
  metadata_startup_script = <<-EOT
    #!/bin/bash
    echo "Hello from Terraform VM" > /var/tmp/startup.txt
  EOT
}

3. OPA Policies

Main Aggregator (policy/main.rego)

package terraform

import data.terraform.gcs
import data.terraform.compute
import data.terraform.firewall
import data.terraform.iam_gcp

deny contains msg if { msg := gcs.deny[_] }
deny contains msg if { msg := compute.deny[_] }
deny contains msg if { msg := firewall.deny[_] }
deny contains msg if { msg := iam_gcp.deny[_] }

GCS Bucket (policy/gcs.rego)

package terraform.gcs
buckets[b] if { some r; b := input.resource_changes[r]; b.type == "google_storage_bucket" }

deny contains msg if { 
  b := buckets[_]; 
  b.change.after.acl == "public-read"; 
  msg := sprintf("GCS bucket %v: acl is public-read", [b.name]) 
}

deny contains msg if { 
  b := buckets[_]; 
  not b.change.after.versioning.enabled; 
  msg := sprintf("GCS bucket %v: versioning not enabled", [b.name]) 
}

deny contains msg if { 
  b := buckets[_]; 
  not b.change.after.encryption; 
  msg := sprintf("GCS bucket %v: encryption not configured", [b.name]) 
}

deny contains msg if { 
  b := buckets[_]; 
  not b.change.after.uniform_bucket_level_access; 
  msg := sprintf("GCS bucket %v: uniform_bucket_level_access not enabled", [b.name]) 
}

deny contains msg if { 
  b := buckets[_]; 
  not b.change.after.labels.Owner; 
  msg := sprintf("GCS bucket %v: missing 'Owner' label", [b.name]) 
}

deny contains msg if { 
  b := buckets[_]; 
  not b.change.after.labels.Environment; 
  msg := sprintf("GCS bucket %v: missing 'Environment' label", [b.name]) 
}

Compute Instance (policy/compute.rego)

package terraform.compute
instances[i] if { some r; i := input.resource_changes[r]; i.type == "google_compute_instance" }
disallowed_types := {"f1-micro","g1-small"}

deny contains msg if { inst := instances[_]; inst.change.after.machine_type in disallowed_types; msg := sprintf("Compute %v: disallowed machine type %v", [inst.name, inst.change.after.machine_type]) }
deny contains msg if { inst := instances[_]; not inst.change.after.service_account; msg := sprintf("Compute %v: missing service_account", [inst.name]) }
deny contains msg if { inst := instances[_]; nic := inst.change.after.network_interface[_]; ac := nic.access_config[_]; ac != null; msg := sprintf("Compute %v: has external IP", [inst.name]) }
deny contains msg if { inst := instances[_]; bd := inst.change.after.boot_disk; bd != null; not bd[0].disk_encryption_key; msg := sprintf("Compute %v: boot disk not encrypted", [inst.name]) }
deny contains msg if { inst := instances[_]; meta := inst.change.after.metadata_startup_script; contains(meta,"curl") & contains(meta,"bash"); msg := sprintf("Compute %v: startup script uses curl|bash pattern", [inst.name]) }
deny contains msg if { inst := instances[_]; not inst.change.after.labels.Environment; msg := sprintf("Compute %v: missing label 'Environment'", [inst.name]) }

Firewall (policy/firewall.rego)

package terraform.firewall
fws[f] if { some r; f := input.resource_changes[r]; f.type == "google_compute_firewall" }

deny contains msg if { fw := fws[_]; rule := fw.change.after; rule.allowed[_].protocol=="tcp"; rule.allowed[_].ports[_]=="22"; rule.source_ranges[_]=="0.0.0.0/0"; msg := sprintf("Firewall %v allows SSH 22 from 0.0.0.0/0", [fw.name]) }
deny contains msg if { fw := fws[_]; rule := fw.change.after; rule.allowed[_].protocol=="tcp"; rule.allowed[_].ports[_]=="3389"; rule.source_ranges[_]=="0.0.0.0/0"; msg := sprintf("Firewall %v allows RDP 3389 from 0.0.0.0/0", [fw.name]) }
deny contains msg if { fw := fws[_]; rule := fw.change.after; rule.allowed[_].protocol=="all"; rule.source_ranges[_]=="0.0.0.0/0"; msg := sprintf("Firewall %v allows all traffic from 0.0.0.0/0", [fw.name]) }

IAM / Service Account (policy/iam_gcp.rego)

package terraform.iam_gcp
service_accounts[s] if { some r; s := input.resource_changes[r]; s.type == "google_service_account" }
iam_bindings[b] if { some r; b := input.resource_changes[r]; b.type == "google_project_iam_binding" }

deny contains msg if { sa := service_accounts[_]; not sa.change.after.display_name; msg := sprintf("Service Account %v: missing display_name", [sa.name]) }
deny contains msg if { b := iam_bindings[_]; member := b.change.after.members[_]; member == "allUsers"; msg := sprintf("IAM binding %v grants role %v to allUsers", [b.name, b.change.after.role]) }
deny contains msg if { b := iam_bindings[_]; b.change.after.role == "roles/owner"; msg := sprintf("IAM binding %v uses broad role roles/owner", [b.name]) }

4. Python Orchestrator (validate_apply_gcp.py)

import subprocess, json, os, shutil, sys

TERRAFORM_DIR = "terraform"
PLAN_BIN = "tfplan"
PLAN_JSON = "plan.json"
POLICY_DIR = "policy"

def run(cmd, cwd=None, check=True):
    print("👉", " ".join(cmd))
    proc = subprocess.run(cmd, capture_output=True, text=True, cwd=cwd)
    if proc.returncode != 0 and check:
        print("❌ Command failed:", " ".join(cmd))
        print("STDOUT:", proc.stdout)
        print("STDERR:", proc.stderr)
        sys.exit(proc.returncode)
    return proc.stdout

def terraform_plan_and_show():
    run(["terraform", "init", "-input=false"], cwd=TERRAFORM_DIR)
    run(["terraform", "plan", "-out", PLAN_BIN, "-input=false"], cwd=TERRAFORM_DIR)
    out = run(["terraform", "show", "-json", os.path.join(TERRAFORM_DIR, PLAN_BIN)], cwd=None)
    with open(PLAN_JSON, "w") as f: f.write(out)

def opa_eval():
    opa_path = shutil.which("opa")
    if not opa_path:
        print("❌ opa not found in PATH"); sys.exit(1)
    cmd = [opa_path,"eval","-i",PLAN_JSON,"-d",POLICY_DIR,"--format","json","data.terraform.deny"]
    out = run(cmd)
    return json.loads(out)

def extract_violations(opa_json):
    violations = []
    for item in opa_json.get("result",[]):
        for expr in item.get("expressions",[]):
            val = expr.get("value")
            if isinstance(val,list): violations.extend(val)
    return violations

def terraform_apply():
    run(["terraform","apply","-auto-approve"], cwd=TERRAFORM_DIR)

def main():
    terraform_plan_and_show()
    opa_json = opa_eval()
    violations = extract_violations(opa_json)
    if violations:
        print("\n❌ Policy violations found:")
        for v in violations: print(" -",v)
        print("🚫 Aborting terraform apply.")
        sys.exit(1)
    else:
        print("\n✅ No policy violations. Applying Terraform...")
        terraform_apply()

if __name__=="__main__":
    main()

5. Workflow Summary

1. Put Terraform code in terraform/.
2. Put Rego policies in policy/.
3. Run python validate_apply_gcp.py.
4. The script will generate plan.json, evaluate OPA policies, and abort if violations exist.
5. If clean, it will automatically apply Terraform to create GCP infrastructure.