Lesson 7: Python for Platform Engineers, OPA Harnesses & AWS Lambda Auto-Remediation

🧠 The Concept (Explain Like I’m 5)

Imagine you are the chief operations inspector for a massive international airport:

• The Confusion (Rego vs. Python): Many engineers ask: “Can I write OPA policies in Python?” The answer is no. OPA uses Rego, a specialized query language designed purely for assertions (like a metal detector scanning bags). Rego cannot make network calls, send emails, or modify databases—by design, it only returns allow or deny.
• Where Python Enters (The Ultimate Cloud Glue): If Rego is the metal detector, Python is the security officer and the automated conveyor belt system:
  1. Python takes the passenger’s luggage (Terraform plan tfplan.json), feeds it into the OPA metal detector, parses the results, and writes the formal security incident report.
  2. If an unauthorized passenger somehow sneaks through onto the tarmac (an engineer creates an unencrypted S3 bucket out-of-band in the AWS Console), an AWS Lambda function written in Python intercepts the intruder, locks down the bucket, and alerts the security operations team in milliseconds!

🏢 The Enterprise Context (Why Platform Architects Care)

1. The “Platform Python” Paradigm: Platform Architects do not build consumer web apps or solve abstract algorithmic puzzles in Python. You write Systems & Cloud Automation Python:
   • Parsing complex nested JSON states (tfplan.json, Kubernetes admission payloads).
   • Interacting with cloud control planes using boto3 (the AWS SDK).
   • Building automated serverless auto-remediation loops via AWS Lambda.
   • Orchestrating CI/CD quality gates and policy runners.
2. Shift-Left vs. Shift-Right (The Complete Governance Loop):
   • Shift-Left (Pre-Apply in Git): A Python test harness executes OPA Rego against tfplan.json. If violations occur, the Pull Request is automatically blocked.
   • Shift-Right (Post-Apply in Cloud): If someone bypasses Git and uses the AWS Console to open an S3 bucket or create an open Security Group, AWS EventBridge triggers a Python Lambda function that instantly revokes the violation within 3 seconds, enforcing zero-trust compliance.

🗺️ Visual Architecture: The Python Policy & Auto-Remediation Loop

flowchart TD
    subgraph ShiftLeft ["Shift-Left: Git Pull Request Phase"]
        Dev["👨‍💻 Engineer PR"] --> TFPlan["terraform show -json > tfplan.json"]
        TFPlan --> PyHarness["<b>Python OPA Harness (evaluate_policy.py)</b><br/>• Ingests tfplan.json<br/>• Invokes OPA REST / CLI API<br/>• Formats PR markdown comments"]
        PyHarness --> OPACheck{"OPA Rego Engine"}
        OPACheck -->|"Violation Detected"| BlockPR["❌ Hard Fail PR (Exit Code 1)"]
        OPACheck -->|"Compliant"| MergePR["✅ Approved for Deployment"]
    end

    subgraph ShiftRight ["Shift-Right: Real-Time Event-Driven Remediation"]
        DriftUser["⚠️ Rogue Out-of-Band Console Action<br/>(Creates Unencrypted S3 Bucket)"] --> CloudTrail["AWS CloudTrail (API Audit Log)"]
        CloudTrail --> EB["Amazon EventBridge (Pattern Match)"]
        EB --> Lambda["<b>Python 3.12 AWS Lambda Function</b><br/>• Boto3 SDK Client<br/>• Enables KMS SSE Encryption<br/>• Attaches Public Access Block"]
        Lambda --> S3Bucket["🗄️ Remediated S3 Bucket (Secure) ✅"]
        Lambda --> Slack["📢 Real-Time SOC2 Slack Alert"]
    end

🔬 Core Language Primitives: The “Platform Python” Toolkit

You only need 20% of Python’s features to execute 80% of enterprise platform engineering. Focus relentlessly on these patterns:

1. Safe Nested Dictionary Navigation (.get())

Cloud JSON documents (tfplan.json, Kubernetes specs) have deeply nested structures where keys may be missing or None.

# ❌ THE JUNIOR ANTI-PATTERN: Crashes with KeyError or TypeError
bucket_name = plan["resource_changes"][0]["change"]["after"]["bucket"]

# ✅ THE ENTERPRISE PATTERN: Safe navigation with default fallbacks
resource_changes = plan.get("resource_changes", [])
for resource in resource_changes:
    change = resource.get("change", {})
    after = change.get("after") or {}
    bucket_name = after.get("bucket", "unknown-bucket")

2. Comprehensions for Cloud Resource Filtering

Extracting specific non-compliant resources in a single readable line:

# Filter all S3 buckets marked for creation or update
s3_creations = [
    r for r in plan.get("resource_changes", [])
    if r.get("type") == "aws_s3_bucket"
    and any(action in ["create", "update"] for action in r.get("change", {}).get("actions", []))
]

3. Resilient AWS SDK Handling (botocore.exceptions.ClientError)

Never catch a generic except Exception. In production cloud automation, catch AWS-specific errors and inspect error codes:

import boto3
from botocore.exceptions import ClientError
import logging

logger = logging.getLogger(__name__)

s3_client = boto3.client("s3")

try:
    s3_client.put_public_access_block(
        Bucket="corp-financial-records",
        PublicAccessBlockConfiguration={
            "BlockPublicAcls": True,
            "IgnorePublicAcls": True,
            "BlockPublicPolicy": True,
            "RestrictPublicBuckets": True
        }
    )
except ClientError as e:
    error_code = e.response["Error"]["Code"]
    if error_code == "NoSuchBucket":
        logger.error("Bucket does not exist!")
    elif error_code == "AccessDenied":
        logger.critical("Lambda execution role lacks s3:PutPublicAccessBlock permissions!")
    else:
        raise

💻 Production Manifest 1: The Python OPA Evaluation Harness (evaluate_policy.py)

This production script runs in CI/CD. It parses tfplan.json, queries the OPA evaluation engine, structures violations, and sets proper CI exit codes:

#!/usr/bin/env python3
"""
evaluate_policy.py - Enterprise Python OPA Policy Harness
Ingests a Terraform JSON execution plan, evaluates it against OPA Rego policies,
and outputs structured results for GitHub Actions PR commenting.
"""

import json
import subprocess
import sys
from pathlib import Path
from typing import Dict, List, Any

def load_json_file(file_path: Path) -> Dict[str, Any]:
    """Loads and validates JSON file existence and structure."""
    if not file_path.is_file():
        print(f"❌ Error: Required plan file '{file_path}' does not exist.", file=sys.stderr)
        sys.exit(2)
    
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            return json.load(f)
    except json.JSONDecodeError as exc:
        print(f"❌ Error: Failed to parse '{file_path}' as valid JSON: {exc}", file=sys.stderr)
        sys.exit(2)

def run_opa_evaluation(policy_dir: Path, plan_file: Path) -> List[str]:
    """
    Executes OPA binary as a sub-process and queries the deny rule set.
    Equally applicable to HTTP OPA Daemon (http://localhost:8181/v1/data/...).
    """
    cmd = [
        "opa", "eval",
        "--data", str(policy_dir),
        "--input", str(plan_file),
        "data.terraform.security.deny",
        "--format", "json"
    ]

    result = subprocess.run(cmd, capture_output=True, text=True, check=False)
    
    if result.returncode != 0:
        print(f"❌ OPA execution error:\n{result.stderr}", file=sys.stderr)
        sys.exit(2)

    try:
        opa_response = json.loads(result.stdout)
        # OPA returns: {"result": [{"expressions": [{"value": ["Violation 1", ...], ...}]}]}
        expressions = opa_response.get("result", [{}])[0].get("expressions", [])
        if expressions:
            violations = expressions[0].get("value", [])
            return list(violations)
        return []
    except (json.JSONDecodeError, IndexError, KeyError) as e:
        print(f"❌ Unexpected OPA output format: {e}", file=sys.stderr)
        sys.exit(2)

def main() -> None:
    plan_path = Path("tfplan.json")
    policies_path = Path("policies")

    print(f"🔍 [Shift-Left] Evaluating '{plan_path}' against OPA policies in '{policies_path}'...")
    
    violations = run_opa_evaluation(policies_path, plan_path)

    if violations:
        print("\n" + "=" * 80)
        print("❌ POLICY VIOLATIONS DETECTED (PULL REQUEST BLOCKED)")
        print("=" * 80)
        for idx, violation in enumerate(violations, start=1):
            print(f"  {idx}. {violation}")
        print("=" * 80 + "\n")
        
        # Write GitHub Step Summary markdown if running inside GitHub Actions
        summary_file = Path("policy_summary.md")
        with open(summary_file, "w", encoding="utf-8") as f:
            f.write("### ❌ Policy-as-Code Violations Detected\n\n")
            for violation in violations:
                f.write(f"- ⛔ **{violation}**\n")
            f.write("\n*Please remediate the issues above in your Terraform code before merging.*")
            
        sys.exit(1) # Halt CI/CD pipeline!

    print("✅ All Policy-as-Code checks passed! Zero violations detected.")
    sys.exit(0)

if __name__ == "__main__":
    main()

💻 Production Manifest 2: AWS Lambda Auto-Remediator (s3_auto_remediator.py)

When an unauthorized engineer manually creates an unencrypted S3 bucket in the AWS Console, AWS EventBridge triggers this Python Lambda function to enforce encryption and public access blocks automatically in under 3 seconds:

"""
s3_auto_remediator.py - Serverless Cloud Governance Lambda
Triggered by Amazon EventBridge on AWS CloudTrail 'CreateBucket' API calls.
Enforces AES256 KMS default encryption and blocks public access.
"""

import json
import logging
import os
import boto3
from botocore.exceptions import ClientError

logger = logging.getLogger()
logger.setLevel(logging.INFO)

s3_client = boto3.client("s3")
KMS_MASTER_KEY_ID = os.environ.get("KMS_KEY_ARN", "alias/aws/s3")

def lambda_handler(event, context):
    logger.info("Received CloudTrail event: %s", json.dumps(event))

    # 1. Parse EventBridge / CloudTrail event payload safely
    detail = event.get("detail", {})
    event_name = detail.get("eventName")
    request_params = detail.get("requestParameters", {})
    bucket_name = request_params.get("bucketName")

    if not bucket_name:
        logger.warning("No bucket name found in event parameters. Skipping.")
        return {"status": "SKIPPED", "reason": "No bucketName"}

    logger.info("Processing S3 Bucket: '%s' for Event: '%s'", bucket_name, event_name)

    # 2. Enforce S3 Default Server-Side Encryption (KMS / AES256)
    try:
        s3_client.put_bucket_encryption(
            Bucket=bucket_name,
            ServerSideEncryptionConfiguration={
                "Rules": [
                    {
                        "ApplyServerSideEncryptionByDefault": {
                            "SSEAlgorithm": "aws:kms",
                            "KMSMasterKeyId": KMS_MASTER_KEY_ID
                        },
                        "BucketKeyEnabled": True # FinOps: Cuts KMS API request costs by 99%!
                    }
                ]
            }
        )
        logger.info("✅ Applied KMS default encryption to bucket '%s'", bucket_name)
    except ClientError as e:
        logger.error("Failed to enforce encryption on bucket '%s': %s", bucket_name, e)
        raise

    # 3. Enforce S3 Block Public Access (All 4 flags ON)
    try:
        s3_client.put_public_access_block(
            Bucket=bucket_name,
            PublicAccessBlockConfiguration={
                "BlockPublicAcls": True,
                "IgnorePublicAcls": True,
                "BlockPublicPolicy": True,
                "RestrictPublicBuckets": True
            }
        )
        logger.info("✅ Enabled full Public Access Block on bucket '%s'", bucket_name)
    except ClientError as e:
        logger.error("Failed to block public access on bucket '%s': %s", bucket_name, e)
        raise

    return {
        "status": "REMEDIATED",
        "bucket": bucket_name,
        "encryption": "aws:kms",
        "public_access_blocked": True
    }

Production IAM Least-Privilege Role for the Lambda (lambda_iam.tf):

resource "aws_iam_policy" "lambda_s3_remediator" {
  name        = "S3GovernanceAutoRemediatorPolicy"
  description = "Allows Lambda to enforce encryption and public block on S3"

  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:PutEncryptionConfiguration",
          "s3:PutBucketPublicAccessBlock",
          "s3:GetBucketLocation"
        ]
        Resource = "arn:aws:s3:::*"
      },
      {
        Effect = "Allow"
        Action = [
          "kms:DescribeKey",
          "kms:GenerateDataKey"
        ]
        Resource = "*"
      }
    ]
  })
}

💻 Production Manifest 3: Testing Platform Python Locally with moto (test_remediator.py)

In enterprise CI/CD pipelines, your Python Lambda code must have automated unit tests that run without requiring real AWS credentials. We use moto to mock AWS in-memory:

"""
test_remediator.py - Unit Tests using Moto (In-Memory AWS Mock)
Run with: pytest test_remediator.py -v
"""

import os
import boto3
import pytest
from moto import mock_aws
from s3_auto_remediator import lambda_handler

@pytest.fixture
def aws_credentials():
    """Mocked AWS Credentials for moto."""
    os.environ["AWS_ACCESS_KEY_ID"] = "testing"
    os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
    os.environ["AWS_SECURITY_TOKEN"] = "testing"
    os.environ["AWS_SESSION_TOKEN"] = "testing"
    os.environ["AWS_DEFAULT_REGION"] = "us-east-1"
    os.environ["KMS_KEY_ARN"] = "arn:aws:kms:us-east-1:123456789012:key/mock-kms-key"

@mock_aws
def test_lambda_remediates_unencrypted_bucket(aws_credentials):
    # 1. Arrange: Create a raw unencrypted bucket in mock AWS
    s3 = boto3.client("s3", region_name="us-east-1")
    bucket_name = "test-unencrypted-corp-bucket"
    s3.create_bucket(Bucket=bucket_name)

    mock_event = {
        "detail": {
            "eventName": "CreateBucket",
            "requestParameters": {
                "bucketName": bucket_name
            }
        }
    }

    # 2. Act: Invoke the Lambda handler
    result = lambda_handler(mock_event, None)

    # 3. Assert: Verify the function reported remediation
    assert result["status"] == "REMEDIATED"
    assert result["bucket"] == bucket_name

    # 4. Assert: Query mock AWS to prove encryption and public blocks were actually applied!
    enc = s3.get_bucket_encryption(Bucket=bucket_name)
    rules = enc["ServerSideEncryptionConfiguration"]["Rules"]
    assert rules[0]["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"] == "aws:kms"

    pab = s3.get_public_access_block(Bucket=bucket_name)
    config = pab["PublicAccessBlockConfiguration"]
    assert config["BlockPublicAcls"] is True
    assert config["BlockPublicPolicy"] is True

🛠️ Step-by-Step Command Playbook & Failure Modes

1. The Lambda Cold Start vs. Memory Sizing Trap

• Symptom: A Python Lambda function takes 1,800ms to execute on its first invocation (cold start), causing EventBridge or API Gateway to time out.
• Root Cause: AWS allocates CPU power strictly proportional to configured RAM. A Lambda configured with default 128 MB RAM receives only a fraction of a single vCPU core, severely throttling Python import parsing (import boto3).
• Fix: Allocate 512 MB to 1024 MB RAM. The cold start drops from 1,800ms down to 250ms, and because execution time drops proportionally, the total AWS bill often decreases!

2. OPA CI Harness Fail-Open vs. Fail-Closed

• Symptom: During a network hiccup or when tfplan.json is malformed, the Python harness crashes with an unhandled exception, but the CI pipeline reports SUCCESS and merges non-compliant code!
• Fix: Enforce Fail-Closed Design:

  try:
      violations = run_opa_evaluation(policies_path, plan_path)
  except Exception as e:
      logger.critical("Policy evaluation harness crashed: %s", e)
      sys.exit(2) # Non-zero exit code ensures GitHub Actions blocks the merge!

3. Lambda VPC Subnet IP Exhaustion during Burst Invocations

• Symptom: A sudden burst of S3 bucket creation events triggers 500 concurrent Lambda remediation executions. Suddenly, new Lambda invocations and adjacent EKS pods on the same private subnet fail with EC2 ENI allocation failed: Subnet has no available IP addresses.
• Under-the-Hood Root Cause: The Lambda function was placed inside an undersized /26 application subnet (64 IPs) shared with Kubernetes worker nodes. While AWS Lambda uses Hyperplane ENIs to multiplex connections, rapid scaling across multiple security group configurations forced AWS to allocate dozens of secondary private IPs, exhausting the subnet.
• Triage & Remediation Runbook:

  # 1. Check available IP address count in the subnet
  aws ec2 describe-subnets --subnet-ids subnet-0123456789abcdef0 \
    --query "Subnets[*].[SubnetId,CidrBlock,AvailableIpAddressCount]"
  
  # 2. Fix: Place Lambda functions in dedicated, isolated VPC subnets with large CIDR blocks (/20)
  # 3. Ensure all Lambda functions in the VPC share the identical Security Group to maximize Hyperplane ENI reuse.

📝 Staff Architect Interview Scenarios

ec2_client = boto3.client("ec2")
paginator = ec2_client.get_paginator("describe_instances")

# Automatically handles NextToken looping under the hood!
for page in paginator.paginate():
    for reservation in page.get("Reservations", []):
        for instance in reservation.get("Instances", []):
            audit_instance_ebs(instance)