Building Guardrail Scripts & Hooks

import re
import sys
import json
from datetime import datetime

# Blocklist: regex patterns for dangerous commands
BLOCKED_PATTERNS = [
    # Terraform
    r"terraform\s+destroy",
    r"terraform\s+apply\s+.*-auto-approve",
    r"terraform\s+apply\s*$",  # apply without plan file

    # Kubernetes
    r"kubectl\s+delete\s+namespace",
    r"kubectl\s+delete\s+--all",
    r"kubectl\s+.*--all-namespaces.*delete",

    # AWS
    r"aws\s+.*\s+delete-",
    r"aws\s+s3\s+rb\s+",
    r"aws\s+s3\s+rm\s+.*--recursive",
    r"aws\s+ec2\s+terminate-instances",
    r"aws\s+rds\s+delete-db",
    r"aws\s+cloudformation\s+delete-stack",

    # Azure
    r"az\s+group\s+delete",
    r"az\s+vm\s+delete",
    r"az\s+storage\s+account\s+delete",

    # GCP
    r"gcloud\s+.*\s+delete",
    r"gcloud\s+projects\s+delete",

    # Git
    r"git\s+push\s+.*--force",
    r"git\s+push\s+-f\s+",
    r"git\s+reset\s+--hard",
    r"git\s+clean\s+-fd",

    # System
    r"rm\s+-rf\s+/",
    r"rm\s+-rf\s+\*",
    r"chmod\s+777",
    r":()\{.*\|.*&",  # fork bomb

    # Database
    r"DROP\s+DATABASE",
    r"DROP\s+TABLE",
    r"TRUNCATE\s+TABLE",
    r"DELETE\s+FROM\s+\w+\s*$",  # DELETE without WHERE
]

# Warning patterns: allowed but logged
WARNING_PATTERNS = [
    r"terraform\s+apply",
    r"kubectl\s+apply",
    r"kubectl\s+delete",
    r"docker\s+system\s+prune",
    r"git\s+push",
]

def check_command(command: str) -> dict:
    """Check a command against guardrail rules."""
    result = {"allowed": True, "warnings": [], "blocked_by": None}

    for pattern in BLOCKED_PATTERNS:
        if re.search(pattern, command, re.IGNORECASE):
            result["allowed"] = False
            result["blocked_by"] = pattern
            log_event("BLOCKED", command, pattern)
            return result

    for pattern in WARNING_PATTERNS:
        if re.search(pattern, command, re.IGNORECASE):
            result["warnings"].append(pattern)
            log_event("WARNING", command, pattern)

    return result

def log_event(level, command, pattern):
    """Log guardrail events for audit trail."""
    entry = {
        "timestamp": datetime.utcnow().isoformat(),
        "level": level,
        "command": command,
        "matched_pattern": pattern
    }
    with open("guardrail-audit.log", "a") as f:
        f.write(json.dumps(entry) + "\n")

if __name__ == "__main__":
    command = " ".join(sys.argv[1:])
    result = check_command(command)

    if not result["allowed"]:
        print(f"BLOCKED: {command}")
        print(f"Matched rule: {result['blocked_by']}")
        sys.exit(1)

    if result["warnings"]:
        print(f"WARNING: Command requires extra caution")
    sys.exit(0)

{
  "hooks": {
    "preToolCall": [
      {
        "matcher": "Bash",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /path/to/guardrail.py \"$TOOL_INPUT\""
          }
        ]
      }
    ]
  }
}

#!/bin/bash
# Pre-commit hook: check for dangerous Terraform changes

# Check for removed lifecycle.prevent_destroy
if git diff --cached | grep -q '^-.*prevent_destroy\s*=\s*true'; then
    echo "❌ BLOCKED: Removing prevent_destroy from a resource."
    echo "   This could allow accidental deletion of protected resources."
    echo "   If intentional, use: git commit --no-verify"
    exit 1
fi

# Check for changes to production tfvars
if git diff --cached --name-only | grep -q 'prod.*\.tfvars'; then
    echo "⚠️ WARNING: Changes to production tfvars detected."
    echo "   Files changed:"
    git diff --cached --name-only | grep 'prod.*\.tfvars'
    read -p "Continue? (y/N): " confirm
    if [[ "$confirm" != "y" ]]; then
        exit 1
    fi
fi

# Run terraform fmt check
if git diff --cached --name-only | grep -q '\.tf$'; then
    if ! terraform fmt -check -recursive . > /dev/null 2>&1; then
        echo "❌ Terraform files are not formatted."
        echo "   Run: terraform fmt -recursive ."
        exit 1
    fi
fi

# Run terraform validate
if git diff --cached --name-only | grep -q '\.tf$'; then
    if ! terraform validate > /dev/null 2>&1; then
        echo "❌ Terraform validation failed."
        terraform validate
        exit 1
    fi
fi

echo "✅ Pre-commit checks passed."

#!/bin/bash
# /usr/local/bin/safe-cloud-cli
# Source this in your shell to wrap all cloud CLIs

function aws() {
    local cmd="$*"

    # Block destructive patterns
    if echo "$cmd" | grep -qiE "(delete-|terminate-|remove-|destroy)"; then
        echo "🛑 GUARDRAIL: Destructive AWS command detected"
        echo "   Command: aws $cmd"
        echo "   This command is blocked by agent safety guardrails."
        echo "   Use the CI/CD pipeline for destructive operations."
        return 1
    fi

    # Enforce --dryrun for S3 operations
    if echo "$cmd" | grep -qiE "^s3 (mv|cp|sync|rm)"; then
        if ! echo "$cmd" | grep -q "dryrun"; then
            echo "⚠️ Running with --dryrun first..."
            command aws $cmd --dryrun
            echo ""
            read -p "Execute for real? (y/N): " confirm
            [[ "$confirm" != "y" ]] && return 0
        fi
    fi

    # Pass through to real AWS CLI
    command aws "$@"
}

function gcloud() {
    local cmd="$*"

    if echo "$cmd" | grep -qiE "(delete|destroy|remove)"; then
        echo "🛑 GUARDRAIL: Destructive GCP command detected"
        echo "   Command: gcloud $cmd"
        return 1
    fi

    command gcloud "$@"
}

function az() {
    local cmd="$*"

    if echo "$cmd" | grep -qiE "(delete|remove|purge)"; then
        echo "🛑 GUARDRAIL: Destructive Azure command detected"
        echo "   Command: az $cmd"
        return 1
    fi

    command az "$@"
}

# policy/terraform_safety.rego
package terraform.safety

import future.keywords.in

# Deny any plan that destroys resources
deny[msg] {
    resource := input.resource_changes[_]
    "delete" in resource.change.actions
    msg := sprintf(
        "Destruction of %s.%s is blocked by policy",
        [resource.type, resource.name]
    )
}

# Deny replacing databases (destroy + create)
deny[msg] {
    resource := input.resource_changes[_]
    "delete" in resource.change.actions
    "create" in resource.change.actions
    contains(resource.type, "db")
    msg := sprintf(
        "Replacement of database %s.%s is blocked",
        [resource.type, resource.name]
    )
}

# Deny removing tags from production resources
deny[msg] {
    resource := input.resource_changes[_]
    "update" in resource.change.actions
    before_tags := resource.change.before.tags
    before_tags.env == "production"
    after_tags := resource.change.after.tags
    after_tags.env != "production"
    msg := sprintf(
        "Removing production tag from %s.%s is blocked",
        [resource.type, resource.name]
    )
}

# Warn on any changes to production-tagged resources
warn[msg] {
    resource := input.resource_changes[_]
    resource.change.before.tags.env == "production"
    resource.change.actions != ["no-op"]
    msg := sprintf(
        "Production resource %s.%s is being modified",
        [resource.type, resource.name]
    )
}

#!/bin/bash
# Validate a Terraform plan against OPA policies

# 1. Generate plan as JSON
terraform plan -out=plan.tfplan
terraform show -json plan.tfplan > plan.json

# 2. Evaluate against OPA policies
opa eval \
  --input plan.json \
  --data policy/ \
  --format pretty \
  "data.terraform.safety.deny"

# 3. Check results
VIOLATIONS=$(opa eval --input plan.json --data policy/ \
  --format json "data.terraform.safety.deny" | \
  jq '.result[0].expressions[0].value | length')

if [[ "$VIOLATIONS" -gt 0 ]]; then
    echo "❌ $VIOLATIONS policy violations found. Apply blocked."
    exit 1
fi

echo "✅ No policy violations. Safe to apply."