SRE Paranoia Gates on AWS Lambda-Backed Production Readiness Enforcement Using CodePipeline, DynamoDB, and CloudWatch.

import boto3
import json
from datetime import datetime, timezone, timedelta
from agentsre.sre_paranoia_gates import (
SREParanoiaGateRunner,
gate_rollback_readiness,
gate_in_flight_traffic,
gate_degraded_load_test,
gate_blast_radius_survivability,
gate_error_budget,
GateResult
)

codepipeline = boto3.client('codepipeline', region_name='us-east-1')
dynamodb = boto3.resource('dynamodb', region_name='us-east-1')
cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')
sns = boto3.client('sns', region_name='us-east-1')
s3 = boto3.client('s3', region_name='us-east-1')

GATE_HISTORY_TABLE = 'SREParanoiaGateHistory'
SNS_TOPIC = 'arn:aws:sns:us-east-1:ACCOUNT_ID:sre-paranoia-alerts'

def build_deployment_context(service_name: str,
pipeline_execution_id: str) -> dict:
"""
Build gate evaluation context from AWS service state.

Pulls live data from:
- DynamoDB: rollback drill history
- ECS: task definition for graceful shutdown config
- S3: load test results artifact
- Service Catalog: blast radius definition
- CloudWatch: live SLO error budget metric
"""
# Rollback drill history from DynamoDB
drill_table = dynamodb.Table('RollbackDrillHistory')
drill_response = drill_table.query(
KeyConditionExpression='service_name = :svc',
ExpressionAttributeValues={':svc': service_name},
ScanIndexForward=False,
Limit=1
)
last_drill = drill_response['Items'][0] if drill_response['Items'] else {}

# ECS task definition for deployment config
ecs = boto3.client('ecs', region_name='us-east-1')
task_defs = ecs.list_task_definitions(
familyPrefix=service_name, sort='DESC', maxResults=1
)
task_def = {}
if task_defs.get('taskDefinitionArns'):
td = ecs.describe_task_definition(
taskDefinition=task_defs['taskDefinitionArns'][0]
)
container = td['taskDefinition']['containerDefinitions'][0]
task_def = {
'graceful_shutdown_seconds': td['taskDefinition'].get('stopTimeout', 0),
'connection_draining_enabled': container.get('essential', False)
}

# Load test artifact from S3
load_test_config = {}
try:
s3_response = s3.get_object(
Bucket='deployment-artifacts',
Key=f'{service_name}/load-test-results/latest.json'
)
load_test_config = json.loads(s3_response['Body'].read())
except Exception:
load_test_config = {
'tested_at_peak_load': False,
'tested_with_degraded_dependency': False
}

# Blast radius from Service Catalog / ARO registry
aro_table = dynamodb.Table('AgentReliabilityOwnership')
try:
aro = aro_table.get_item(Key={'agent_id': service_name})
blast = json.loads(aro.get('Item', {}).get('blast_radius', '[]'))
cannot_touch = json.loads(
aro.get('Item', {}).get('cannot_touch', '[]')
)
blast_radius = {
'downstream_service_count': len(blast),
'contains_payment_path': any(
'payment' in s.lower() for s in blast
),
'single_engineer_containable': len(blast) <= 4,
'circuit_breaker_configured': True
}
except Exception:
blast_radius = {
'downstream_service_count': 10,
'contains_payment_path': True,
'single_engineer_containable': False,
'circuit_breaker_configured': False
}

# Live error budget from CloudWatch
end = datetime.now(timezone.utc)
start = end - timedelta(hours=1)
budget_response = cloudwatch.get_metric_statistics(
Namespace='AgentSRE/ErrorBudget',
MetricName='ErrorBudgetRemaining',
Dimensions=[{'Name': 'ServiceName', 'Value': service_name}],
StartTime=start, EndTime=end,
Period=3600, Statistics=['Minimum']
)
budget_pts = budget_response.get('Datapoints', [])
budget_remaining = budget_pts[0]['Minimum'] if budget_pts else 100.0

return {
"rollback": {
"last_tested_at": last_drill.get('tested_at'),
"p95_duration_minutes": last_drill.get('p95_minutes', 99)
},
"deployment": {
**task_def,
"schema_backward_compatible": True
},
"load_testing": load_test_config,
"blast_radius": blast_radius,
"slo": {"error_budget_remaining_pct": budget_remaining},
"is_critical_change": False
}

def publish_gate_metrics(service_name: str, result: dict) -> None:
"""Publish gate pass/fail counts to CloudWatch for trend tracking."""
cloudwatch.put_metric_data(
Namespace='SRE/ParanoiaGates',
MetricData=[
{
'MetricName': 'GatesPassed',
'Dimensions': [{'Name': 'ServiceName', 'Value': service_name}],
'Value': float(result['gates_passed']),
'Unit': 'Count'
},
{
'MetricName': 'GatesFailed',
'Dimensions': [{'Name': 'ServiceName', 'Value': service_name}],
'Value': float(result['gates_failed']),
'Unit': 'Count'
},
{
'MetricName': 'DeploymentApproved',
'Dimensions': [{'Name': 'ServiceName', 'Value': service_name}],
'Value': 1.0 if result['deployment_approved'] else 0.0,
'Unit': 'None'
}
]
)

def store_gate_history(service_name: str,
pipeline_id: str,
result: dict) -> None:
"""Log gate evaluation to DynamoDB for audit and trend analysis."""
table = dynamodb.Table(GATE_HISTORY_TABLE)
table.put_item(Item={
'service_name': service_name,
'evaluated_at': result['evaluated_at'],
'pipeline_execution_id': pipeline_id,
'deployment_approved': result['deployment_approved'],
'gates_passed': result['gates_passed'],
'gates_failed': result['gates_failed'],
'gates_warned': result['gates_warned'],
'gate_results': json.dumps(result['results']),
'ttl': int(
(datetime.now(timezone.utc) + timedelta(days=90)).timestamp()
)
})

def lambda_handler(event: dict, context) -> dict:
"""
SRE Paranoia Gate Lambda — invoked as CodePipeline action.

Evaluates all five gates against live AWS service state.
On pass: signals CodePipeline to continue.
On fail: signals failure, alerts SRE owner via SNS.
All results logged to DynamoDB and CloudWatch.
"""
# Extract CodePipeline context
job_id = event.get('CodePipeline.job', {}).get('id', 'local-test')
job_data = event.get('CodePipeline.job', {}).get('data', {})

user_params = json.loads(
job_data.get('actionConfiguration', {})
.get('configuration', {})
.get('UserParameters', '{}')
)
service_name = user_params.get('service_name', 'unknown-service')
sre_owner = user_params.get('sre_owner', 'platform-sre')
pipeline_id = job_data.get('pipelineContext', {}).get('pipelineExecutionId', 'test')

# Build context from live AWS state
deploy_context = build_deployment_context(service_name, pipeline_id)

# Run all five SRE Paranoia Gates
runner = SREParanoiaGateRunner(service_name, sre_owner)
runner.register(gate_rollback_readiness)
runner.register(gate_in_flight_traffic)
runner.register(gate_degraded_load_test)
runner.register(gate_blast_radius_survivability)
runner.register(gate_error_budget)

result = runner.run_all(deploy_context)

# Store history and publish metrics
store_gate_history(service_name, pipeline_id, result)
publish_gate_metrics(service_name, result)

if not result['deployment_approved']:
# Alert SRE team
failed_gates = [
r for r in result['results'] if r['result'] == 'fail'
]
sns.publish(
TopicArn=SNS_TOPIC,
Subject=f"[SRE Gate BLOCKED] {service_name} — {len(failed_gates)} gate(s) failed",
Message=json.dumps({
"service": service_name,
"pipeline_execution_id": pipeline_id,
"failed_gates": failed_gates,
"decision": result['decision']
}, indent=2)
)

# Fail the CodePipeline action
if job_id != 'local-test':
codepipeline.put_job_failure_result(
jobId=job_id,
failureDetails={
'type': 'JobFailed',
'message': result['decision'],
'externalExecutionId': pipeline_id
}
)
return {'statusCode': 422, 'body': json.dumps(result)}

# Success — continue pipeline
if job_id != 'local-test':
codepipeline.put_job_success_result(jobId=job_id)

return {'statusCode': 200, 'body': json.dumps(result)}