Building a Zero-Trust DevSecOps Agent with Open Policy Agent (OPA), Terraform, and Amazon Bedrock Guardrails

2. Open Policy Agent (OPA) & Rego Rule Engine Architecture

Open Policy Agent evaluates JSON data documents against declarative Rego policy rules. When applied to Terraform execution plans, OPA parses terraform show -json outputs to verify compliance before state changes touch cloud endpoints:

package terraform.security

default allow = false

# Rule 1: Deny S3 buckets without server-side encryption
deny[msg] {
    resource := input.resource_changes[_]
    resource.type == "aws_s3_bucket"
    not resource.change.after.server_side_encryption_configuration
    msg := sprintf("SECURITY VIOLATION: S3 Bucket '%v' must enable server-side encryption.", [resource.name])
}

# Rule 2: Deny ingress security groups with 0.0.0.0/0 on SSH port 22
deny[msg] {
    resource := input.resource_changes[_]
    resource.type == "aws_security_group"
    ingress := resource.change.after.ingress[_]
    ingress.from_port <= 22
    ingress.to_port >= 22
    cidr := ingress.cidr_blocks[_]
    cidr == "0.0.0.0/0"
    msg := sprintf("SECURITY VIOLATION: Security Group '%v' permits unrestricted SSH access from 0.0.0.0/0.", [resource.name])
}

allow {
    count(deny) == 0
}

3. Amazon Bedrock Guardrails & Content Filtering

While OPA validates generated infrastructure state, Amazon Bedrock Guardrails protect the LLM inference loop itself. Guardrails filter out prompt injection attacks, mask exposed PII or secret API keys, and block unsafe commands before the model generates response payloads.

• Sensitive Data Masking: Redacts AWS secret access keys, database passwords, and JWT tokens in real-time.
• Denied Topics Filter: Rejects requests attempting to disable CloudTrail, modify GuardDuty configs, or bypass logging.
• Prompt Attack Resistance: Prevents jailbreaking techniques attempting to bypass corporate security policies.

4. Parsing Terraform JSON Plans & Static AST Validation

When an AI agent drafts a Terraform patch, the DevSecOps pipeline executes a dry-run terraform plan -out=tfplan followed by terraform show -json tfplan. The resulting JSON document is passed directly into the embedded OPA evaluator. If any Rego deny statements trigger, the build fails instantly and feedback is returned to the agent for self-correction.

5. Production Python Implementation: DevSecOps OPA Agent

Below is a complete Python script illustrating how an autonomous DevSecOps agent generates Terraform HCL code via Amazon Bedrock and validates the plan against an OPA policy engine before opening a pull request:

import json
import subprocess
import boto3

class DevSecOpsAgent:
    def __init__(self):
        self.bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')
        self.model_id = 'anthropic.claude-3-5-sonnet-20240620-v1:0'

    def evaluate_opa_policy(self, plan_json_path: str, rego_policy_path: str) -> bool:
        cmd = ["opa", "eval", "--data", rego_policy_path, "--input", plan_json_path, "data.terraform.security.deny"]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            print("❌ OPA Execution Error:", result.stderr)
            return False

        output = json.loads(result.stdout)
        violations = output.get("result", [{}])[0].get("expressions", [{}])[0].get("value", [])

        if violations:
            print("🚨 DevSecOps OPA Security Violations Found:")
            for v in violations:
                print(f"  - {v}")
            return False

        print("✓ All Open Policy Agent (OPA) Security Checks Passed Cleanly.")
        return True

    def generate_secure_iac(self, prompt: str):
        print(f"🤖 Generating Secure Terraform IaC via Bedrock for prompt: '{prompt}'...")
        messages = [{"role": "user", "content": [{"text": prompt}]}]
        res = self.bedrock.converse(
            modelId=self.model_id,
            messages=messages,
            inferenceConfig={"temperature": 0.1, "maxTokens": 1024}
        )
        return res['output']['message']['content'][0]['text']

if __name__ == '__main__':
    agent = DevSecOpsAgent()
    hcl = agent.generate_secure_iac("Draft secure AWS S3 bucket module with encryption and block public access.")
    print("✓ Secure HCL Drafted Successfully.")