DEV Community: Daffa Rabbani

Ransomware-Resilient Disaster Recovery ⚡♻

Daffa Rabbani — Wed, 16 Sep 2026 14:20:27 +0000

🔴 The Incident That Shook a Nation

On June 20, 2024, the Indonesian government faced its worst digital nightmare in history.

Brain Cipher ransomware — a new variant of LockBit 3.0 — successfully encrypted the entire infrastructure of the Temporary National Data Center (PDNS), operated by the Ministry of Communication and Information Technology.

The result?

❌ 230+ government services PARALYZED
❌ Airports forced to rely on MANUAL processes
❌ Ransom demand: USD 8,000,000
❌ Data that had a backup: < 2%
❌ Recovery time: WEEKS

And the root cause wasn't sophisticated malware — it was having no backup at all.

"The most sophisticated ransomware in the world is powerless against a well-tested backup."

This article rebuilds that scenario as a resilient reference architecture on AWS — complete with code, runbooks, and cost calculations.

🏗️ The Architecture We'll Build

📋 AWS Services Used

Service	Role
AWS Backup	Centralized backup orchestration, cross-region copy
S3 Object Lock (WORM)	Immutable backups that can't be deleted or modified
S3 Vault Lock	Policy-level immutability for the AWS Backup vault
AWS Elastic DRS	Continuous replication, failover in under 1 hour
Amazon GuardDuty	ML-based, real-time threat detection
AWS Security Hub	Finding aggregation, compliance scoring
AWS KMS	End-to-end encryption for all backup data
AWS CloudTrail	100% API action audit log, forensic-ready
AWS Config	Compliance rules, drift detection
EventBridge + Lambda	Automated response & isolation

⚙️ Step-by-Step Implementation

Step	AWS Module / Service	Target Region	Core Objective & Key Deliverable
STEP 1	AWS KMS (Customer Managed Keys)	Jakarta & Singapore	End-to-end symmetric encryption with annual key rotation in both regions
STEP 2	Amazon S3 Object Lock (WORM Bucket)	Jakarta (`ap-southeast-3`)	Immutable storage bucket (Compliance Mode 30 days + Glacier Lifecycle)
STEP 3	AWS Backup Vault & Vault Lock	Jakarta & Singapore	Anti-tamper/anti-deletion backup vaults (Compliance Mode) across regions
STEP 4	AWS Backup Plan & Resource Assignment	Jakarta → Singapore	Automated backup policies (Hourly, Daily, Monthly) & cross-region copy
STEP 5	Amazon GuardDuty & SNS Alerting	Jakarta (`ap-southeast-3`)	Real-time threat detection & instant email notification (Severity ≥ 7)
STEP 6	AWS Security Hub & Insights	Jakarta (`ap-southeast-3`)	Centralized compliance dashboard (CIS & FSBP) + Ransomware Risk Insight
STEP 7	AWS CloudTrail (Multi-Region)	Multi-Region (`ap-southeast-3` & `ap-southeast-1`)	Immutable audit trail (Read/Write Management + S3/Lambda Data Events)
STEP 8	AWS Elastic Disaster Recovery (DRS)	Jakarta → Singapore	Block-level continuous disk replication (RPO < 1 min, RTO < 1 hr)

Primary Region: ap-southeast-3 (Asia Pacific – Jakarta)
Recovery Region: ap-southeast-1 (Asia Pacific – Singapore)

STEP 1 — Create KMS Encryption Keys

1A. KMS Key for Jakarta Region (Primary)

Open AWS Management Console → confirm the region in the top-right corner: Asia Pacific (Jakarta) ap-southeast-3.
In the top search bar, type KMS → click Key Management Service.
Click the Create key button (orange).
Fill in the key configuration:

Parameter	Value	Description & Purpose
Key type	Symmetric	Industry-standard AES-256 encryption
Key usage	Encrypt and decrypt	Used for automated backup and recovery operations
Regionality	Single-Region key	Isolated key dedicated to Jakarta region

Click Next.
Set Alias: pdns-backup-key.
Set Description: PDNS Backup Encryption Key - Jakarta Primary.
Click Next → Next (leave Key Administrators & Key Permissions default) → click Finish.
Record the Key ARN that appears in the reference table below. Example: arn:aws:kms:ap-southeast-3:123456789012:key/xxxx-xxxx-xxxx-xxxx

1B. Enable Key Rotation

On the detail page of the newly created pdns-backup-key, click the Key rotation tab.
Check the box: Automatically rotate this KMS key every year.
Click Save.

1C. KMS Key for Singapore Region (Recovery)

Switch region on the top-right dropdown → select Asia Pacific (Singapore) ap-southeast-1.
Repeat steps 1A and 1B with these settings:
- Alias: pdns-recovery-key
- Description: PDNS Recovery Encryption Key - Singapore Recovery
- Key rotation: Automatically enabled (Enabled)
Record the Key ARN for Singapore region.

KMS Key Reference Table

Record your generated Key ARNs in the following reference table:

Key Alias	Target Region	Primary Purpose	Key ARN (Copy from Console)
`alias/pdns-backup-key`	`ap-southeast-3` (Jakarta)	Encrypts S3 WORM, Primary Backup Vault, CloudTrail	`arn:aws:kms:ap-southeast-3:[ACCOUNT-ID]:key/...`
`alias/pdns-recovery-key`	`ap-southeast-1` (Singapore)	Encrypts Singapore Recovery Vault, DRS Staging Volumes	`arn:aws:kms:ap-southeast-1:[ACCOUNT-ID]:key/...`

STEP 2 — Create S3 Bucket with Object Lock (WORM)

⚠️ IMPORTANT: S3 Object Lock can only be enabled when a bucket is first created. It cannot be added after the bucket exists.

2A. Create Bucket in Jakarta

Confirm active region: ap-southeast-3 (Jakarta).
Search for S3 in the top search bar → click S3.
Click the Create bucket button.
Fill in the bucket configuration:

Parameter	Value	Security Rationale
Bucket name	`pdns-immutable-backup-[your-account-number]`	Must be globally unique across AWS
AWS Region	Asia Pacific (Jakarta) ap-southeast-3	Primary production region
Object Ownership	ACLs disabled (recommended)	Access controlled exclusively via IAM policies
Block Public Access	*Block all* public access** (All checkboxes checked)	Completely prevents unauthorized public exposure
Bucket Versioning	Enable	Mandatory technical prerequisite for Object Lock
Default Encryption	SSE-KMS	Server-side encryption with Customer Managed Key (CMK)
KMS Key	Select `pdns-backup-key`	Uses Jakarta CMK created in Step 1
Bucket Key	Enable	Reduces KMS API call costs by up to 99%
Object Lock	Enable ✅	Enables WORM (Write Once, Read Many) capability

Under Advanced settings → Object Lock, check Enable and acknowledge the confirmation dialog.
Click Create bucket.

2B. Configure Default Retention (Compliance Mode)

Click the name of the newly created bucket (pdns-immutable-backup-...).
Go to the Properties tab.
Scroll down to Object Lock → click Edit.
Fill in the default retention settings:

Parameter	Value	Security Impact
Default retention	Enable	Automatically applied to all newly stored objects
Default retention mode	Compliance ⚠️	Absolute protection — objects cannot be deleted by anyone, including Root and AWS Support
Retention period	30 Days	Minimum 30-day immutability window for ransomware defense

Click Save changes.

🔴 Critical Note: In Compliance Mode, no user (not even the root account or AWS Support) can delete objects or overwrite them until the 30-day retention period has expired.

2C. Add Lifecycle Rule (Auto-transition to Glacier)

In the same bucket, click the Management tab.
Click the Create lifecycle rule button.
Fill in the lifecycle configuration:

Parameter	Value	Optimization Purpose
Lifecycle rule name	`move-to-glacier-after-90-days`	Policy identifier
Filter scope	Apply to all objects in the bucket	Covers all historical backup archives
Lifecycle rule actions	Check: Transition current versions of objects between storage classes	Moves older objects to colder tiers
Storage class transition	Glacier Flexible Retrieval	Low-cost cold archival storage
Days after object creation	90	Automatic migration after 90 days

Acknowledge the cost transition dialog if prompted → click Create rule.

STEP 3 — AWS Backup Vault + Vault Lock

3A. Create Backup Vault in Jakarta

Confirm active region: ap-southeast-3 (Jakarta).
Search for AWS Backup in the search bar → click AWS Backup.
In the left navigation menu, click Backup vaults.
Click Create backup vault.
Fill in the configuration:

Parameter	Value	Description
Backup vault name	`pdns-immutable-vault`	Primary backup vault in Jakarta
Encryption key	Select `pdns-backup-key`	Uses Jakarta KMS key from Step 1
Tags	Key: `Environment`, Value: `Production`	Environment metadata classification

Click Create backup vault.

3B. Enable Vault Lock

⚠️ There is a 72-hour grace period. During the first 3 days, Vault Lock can still be modified or deleted. After that period, it becomes PERMANENT and cannot be deleted or bypassed by anyone, including AWS.

On the pdns-immutable-vault page, click the Vault Lock tab.
Click Configure Vault Lock.
Fill in the parameters:

Parameter	Value	Security Boundary
Vault Lock mode	Compliance	Permanent compliance mode (WORM Vault)
Min retention period	7 Days	Snapshots cannot be deleted before 7 days
Max retention period	365 Days	Upper bound for snapshot retention in the vault
Grace period	72 Hours (Default)	Evaluation period before the lock becomes irreversible

Type confirm in the confirmation dialog.
Click Apply Vault Lock.

3C. Create Recovery Vault in Singapore

Switch region to ap-southeast-1 (Singapore).
Navigate to AWS Backup → Backup vaults → Create backup vault.
Fill in the recovery vault settings:

Parameter	Value	Description
Target Region	Asia Pacific (Singapore) ap-southeast-1	Isolated disaster recovery region (Air-gapped)
Backup vault name	`pdns-recovery-vault`	Destination vault for cross-region copies
Encryption key	Select `pdns-recovery-key`	Independently encrypted with Singapore CMK
Vault Lock mode	Compliance (Min 7 days, Max 365 days)	Anti-ransomware protection in recovery region

Click Create backup vault.
Repeat step 3B (Vault Lock) on this Singapore vault using the same parameters (Compliance Mode, Min 7 days, Max 365 days).

STEP 4 — Backup Plan (Automated Backup Schedule)

4A. Create IAM Role for Backup

Search for IAM in the search bar → click IAM.
In the left menu, click Roles → click Create role.
Select Trusted entity type: AWS service.
Under Use case, scroll down → select Backup → click Next.
In the Add permissions section, search and check both policies:
- ✅ AWSBackupServiceRolePolicyForBackup
- ✅ AWSBackupServiceRolePolicyForRestores
Click Next.
Enter Role name: AWSBackupServiceRole-PDNS.
Click Create role.

4B. Create Backup Plan

Switch region back to ap-southeast-3 (Jakarta).
In AWS Backup → left menu → Backup plans → click Create backup plan.
Select Build a new plan.
Enter Backup plan name: PDNS-Ransomware-Resilient-Plan.

4C. Add Rule 1: Hourly Backup (Critical Services)

Click the Add backup rule button.
Configure the hourly backup rule:

Parameter	Value	Purpose
Rule name	`Hourly-Critical-Services`	High-frequency protection for critical workloads
Backup vault	`pdns-immutable-vault`	Local immutable vault in Jakarta
Backup frequency	Hourly (Every 1 hour)	Ensures ultra-low RPO (< 1 hour)
Backup window	Start within 1 hour	Execution tolerance window
Transition to cold storage	Never	Retained in warm storage for rapid restores
Retention period	7 Days	Local retention duration

Under Copy to destination:
- Click Add copy
- Destination region: Asia Pacific (Singapore)
- Destination vault: pdns-recovery-vault
- Retention period: 30 Days
Click Save rule.

4D. Add Rule 2: Daily Backup

Click Add backup rule again.
Configure the daily backup rule:

Parameter	Value	Purpose
Rule name	`Daily-All-Resources`	Routine daily protection for all resources
Backup vault	`pdns-immutable-vault`	Primary Jakarta vault
Backup frequency	Daily	Executed once per day
Backup window	01:00 UTC	Off-peak execution window
Transition to cold storage	After 90 days	Automated migration to cold storage
Retention period	1 Year (365 days)	Annual compliance retention

Click Save rule.

4E. Add Rule 3: Monthly Long-Term

Click Add backup rule again.
Configure the monthly long-term backup rule:

Parameter	Value	Purpose
Rule name	`Monthly-Long-Term`	Long-term regulatory compliance archive
Backup vault	`pdns-immutable-vault`	Primary Jakarta vault
Backup frequency	Monthly	First day of every month
Transition to cold storage	After 30 days	Rapid transition to cold storage for cost efficiency
Retention period	7 Years	Meets statutory compliance & audit retention rules

Click Save rule → click Create plan.

4F. Assign Resources to Backup Plan

After the plan is created, click Assign resources.
Fill in the resource assignment form:

Parameter	Value	Assignment Logic
Resource assignment name	`All-Production-Resources`	Resource group assignment identifier
IAM role	`AWSBackupServiceRole-PDNS`	Dedicated backup service role from Step 4A
Resource selection	Include specific resource types	Target workload selection
Resource types	Check: EC2, RDS, EFS, DynamoDB, EBS	Core compute and database assets
Refine selection using tags	Key: `Environment`, Value: `Production`	All resources with this tag are automatically protected

Click Assign resources.

⚠️ Operational Notice: Ensure all production EC2 instances, RDS databases, and EBS volumes in Jakarta carry the tag: Environment=Production.

STEP 5 — Amazon GuardDuty (Threat Detection)

5A. Enable GuardDuty in Jakarta

Confirm active region: ap-southeast-3 (Jakarta).
Search for GuardDuty in the search bar → click Amazon GuardDuty.
Click the Get Started button.
On the welcome page, click Enable GuardDuty.

5B. Enable Protection Plans

After GuardDuty is active:

In the left menu, click Protection plans.
Enable each protection plan individually by clicking Enable:

Protection Plan	Target Status	Ransomware Defense Capability
S3 Protection	Enabled ✅	Detects suspicious mass reading or unauthorized API activity on S3
Malware Protection for EC2	Enabled ✅	Automatically scans EBS volumes upon detecting suspicious activities
RDS Protection	Enabled ✅	Identifies anomalous logins and brute-force patterns on RDS databases
Lambda Protection	Enabled ✅	Detects malicious code executions on serverless functions

5C. Configure Finding Frequency

In the left menu, click Settings.
Under Finding export options → Updated finding frequency.
Select: Update CWE every 15 minutes (accelerates finding notifications to EventBridge).
Click Save.

5D. Create SNS Topic for Alerts

Search for SNS in the search bar → click Simple Notification Service.
In the left menu, click Topics → click Create topic.
Configure the topic:

Parameter	Value	Description
Type	Standard	Standard publish/subscribe messaging
Name	`pdns-security-alerts`	Security alert notification topic

Click Create topic.
On the new topic page, click Create subscription.
Configure the subscription:

Parameter	Value	Description
Protocol	Email	Notifications sent via email
Endpoint	`security@yourdomain.com`	Target SecOps team email address

Click Create subscription.
Check your inbox → click Confirm subscription on the email received from AWS Notifications.

5E. Create EventBridge Rule: GuardDuty HIGH → Alert

Search for EventBridge in the search bar → click Amazon EventBridge.
In the left menu, click Rules → click Create rule.
Fill in the rule parameters:

Parameter	Value	Description
Name	`GuardDuty-High-Severity-Alert`	EventBridge rule name
Description	`Alert when GuardDuty detects HIGH/CRITICAL threat`	Functional description
Event bus	default	Standard account event bus
Rule type	Rule with an event pattern	Triggers actions based on incoming event patterns

Click Next.
Under Event pattern → select Custom pattern (JSON editor).
Paste the following JSON pattern:

{
  "source": ["aws.guardduty"],
  "detail-type": ["GuardDuty Finding"],
  "detail": {
    "severity": [{ "numeric": [">=", 7] }]
  }
}

Click Next.
Under Target 1:
- Target type: AWS service
- Select a target: SNS topic
- Topic: select pdns-security-alerts
Click Next → Next → click Create rule.

✅ Now every GuardDuty finding with severity ≥ 7 will be automatically dispatched to your email within seconds.

STEP 6 — AWS Security Hub

6A. Enable Security Hub

Search for Security Hub in the search bar → click Security Hub.
Click Go to Security Hub / Enable Security Hub.
Ensure the following compliance standards are checked:
- ✅ AWS Foundational Security Best Practices v1.0.0
- ✅ CIS AWS Foundations Benchmark v1.4.0
- ✅ Enable AWS Config (Security Hub requires AWS Config for resource evaluations)
Click Enable Security Hub.

6B. Integrate with GuardDuty

In Security Hub, click Integrations in the left menu.
Type GuardDuty in the search box.
Click Accept findings on the Amazon GuardDuty integration card.
Confirm acceptance.

6C. Setup Ransomware Insight

In the left menu, click Insights → click Create insight.
Configure the custom insight:

Parameter	Value	Description
Insight name	`Ransomware-Risk-Indicators`	Custom security analytical filter
Filter 1	Product name = `GuardDuty`	Focuses on GuardDuty intelligence
Filter 2	Severity label = `HIGH`, `CRITICAL`	Filters for severe and fatal threats
Filter 3	Workflow status = `NEW`	Displays active unaddressed incidents
Group by	Resource ID	Aggregates findings per impacted asset

Click Create insight.

STEP 7 — AWS CloudTrail (Audit Log)

7A. Create CloudTrail Trail

Search for CloudTrail in the search bar → open CloudTrail.
In the left menu, click Trails → click Create trail.
Fill in the trail parameters:

Parameter	Value	Forensic & Security Rationale
Trail name	`pdns-full-audit-trail`	Master audit trail name
Storage location	Create new S3 bucket	Dedicated S3 bucket for audit records
Trail log bucket name	`pdns-cloudtrail-logs-[account-number]`	Unique audit bucket name
Log file SSE-KMS encryption	Enabled	Encrypts all stored log archives
AWS KMS alias	Select `pdns-backup-key`	Uses Jakarta CMK
Log file validation	Enabled ✅	Generates cryptographic hashes to detect log tampering
CloudWatch Logs	Enabled	Streams events to CloudWatch for near real-time alerting
Log group name	`/aws/cloudtrail/pdns-audit`	Dedicated CloudWatch log group
Multi-region trail	Yes ✅	Captures API actions across all AWS regions

Click Next.

7B. Select Events to Log

Under Events:
- Check: ✅ Management events (Read + Write)
- Check: ✅ Data events
Under Data events, click Add data event type:
- Data event type: S3 → All current and future S3 buckets → Read + Write
Click Add data event type again:
- Data event type: Lambda → All current and future functions
Click Next → review settings → click Create trail.

STEP 8 — AWS Elastic Disaster Recovery (DRS)

8A. Initialize DRS in Recovery Region

Switch region to ap-southeast-1 (Singapore).
Search for Elastic Disaster Recovery in the search bar → click AWS Elastic Disaster Recovery.
Click Set up Elastic Disaster Recovery / Get started.
Configure Default replication settings:

Parameter	Value	Technical Rationale
Staging area subnet	Public Subnet in `ap-southeast-1b`	Subnet where lightweight replication servers run
Security groups	`pdns-drs-replication-sg`	Security group governing replication traffic
Instance type	`t3.small`	Cost-effective instance for replication servers
EBS encryption	Custom → select `pdns-recovery-key`	Ensures encrypted staging storage in DR region
Create public IP	Yes	Required for cross-region replication traffic over the internet

💡 Architectural Recommendation: Before initializing DRS, create a dedicated VPC in Singapore named pdns-recovery-vpc (10.10.0.0/16) with 1 Public Subnet in AZ ap-southeast-1b and an active Internet Gateway.

Create Security Group pdns-drs-replication-sg in Singapore with:
- Inbound: Port TCP 1500 from 0.0.0.0/0 (incoming block-level data replication from Jakarta)
- Outbound: All traffic (Default)
Click Next.

8B. Configure Point-in-Time (PIT) Recovery

Under Point-in-time recovery:
- Click Add rule
- Snapshot retention (in days): 7 days
Click Next → click Create.

8C. Install DRS Agent on Production Server

DRS Agent installation is conducted via AWS Systems Manager Session Manager (no SSH port 22 required).

DRS Replication Topology

[Jakarta - ap-southeast-3]              [Singapore - ap-southeast-1]
┌──────────────────────────┐            ┌──────────────────────────┐
│ PRODUCTION EC2           │            │ DRS Replication Server    │
│ ← Install DRS Agent      │─replicate─►│ (created AUTOMATICALLY    │
│   on this server         │  (Port 1500)│  by AWS DRS, not manually) │
└──────────────────────────┘            └──────────────────────────┘
       SOURCE (Origin)                          TARGET (Recovery)

Step 1 — Create Production EC2 in Jakarta (if not yet exists)

Confirm region: ap-southeast-3 (Jakarta).
Go to EC2 → Instances → Launch instances.
Fill in the instance configuration:

Parameter	Value	Critical Note
Name	`pdns-prod-server-01`	Production instance identifier
AMI	Amazon Linux 2023	SSM Agent pre-installed
Instance type	t3.medium	⚠️ Minimum t3.medium — t3.micro (1 GiB RAM) is insufficient for DRS Agent
Storage (EBS)	20 GiB	⚠️ Minimum 20 GiB — 8 GiB is too small for OS + Agent download
Key pair	Proceed without key pair	Access via Session Manager (no SSH required)
Auto-assign public IP	Enable	Required for internet connectivity
IAM instance profile	`EC2-SSM-SessionManager-Role`	Attach role created in Step 2 below
Tags	Key: `Environment`, Value: `Production`	Required for automated backup by the Backup Plan

Click Launch instance.

Step 2 — Create IAM Role for EC2 (SSM + DRS)

Go to IAM → Roles → Create role.
Select AWS service → Use case: EC2 → click Next.
Check BOTH of the following policies:
- ✅ AmazonSSMManagedInstanceCore (Session Manager access)
- ✅ AWSElasticDisasterRecoveryAgentInstallationPolicy (DRS agent installation permissions)
Set role name: EC2-SSM-SessionManager-Role → click Create role.

Step 3 — Attach IAM Role to EC2 Instance

In Jakarta console, open EC2 → Instances.
Check pdns-prod-server-01 → click Actions → Security → Modify IAM role.
Select EC2-SSM-SessionManager-Role → click Update IAM role.

Step 4 — Configure Outbound Security Group (Ports 443 & 1500)

The DRS Agent requires 2 outbound ports from Jakarta to the internet:

Target Port	Protocol	Purpose
443	HTTPS	Communication to DRS Control Plane API
1500	TCP	Block-level disk data replication to Singapore

On instance pdns-prod-server-01, go to the Security tab → click the Security Group name.
Open Outbound rules → click Edit outbound rules → click Add rule:
- Type: Custom TCP
- Port range: 1500
- Destination: 0.0.0.0/0
- Description: DRS Data Replication to Singapore
Click Save rules.

Step 5 — Add KMS Key Permission for DRS

Switch region to ap-southeast-1 (Singapore).
Open KMS → click pdns-recovery-key → Key policy tab → click Edit.
Add the following statement inside the Statement array before the closing ]:

{
  "Sid": "Allow DRS Service to use this key",
  "Effect": "Allow",
  "Principal": {
    "AWS": "arn:aws:iam::[YOUR-ACCOUNT-ID]:role/aws-service-role/drs.amazonaws.com/AWSServiceRoleForElasticDisasterRecovery"
  },
  "Action": [
    "kms:Encrypt",
    "kms:Decrypt",
    "kms:GenerateDataKey",
    "kms:GenerateDataKeyWithoutPlaintext",
    "kms:DescribeKey",
    "kms:CreateGrant",
    "kms:ReEncrypt*"
  ],
  "Resource": "*"
}

(Replace [YOUR-ACCOUNT-ID] with your 12-digit AWS account ID). Click Save changes.

Step 6 — Verify Instance Readiness in Fleet Manager

Open Systems Manager → Fleet Manager in region ap-southeast-3 (Jakarta).
Wait 3–5 minutes after attaching the IAM role.
Refresh the page until pdns-prod-server-01 appears with status: Online ✅.

Possible Cause	Recommended Solution
IAM Role not yet attached	Verify and attach `EC2-SSM-SessionManager-Role` in EC2 Security tab
SSM Agent not running	Restart instance from EC2 console
Instance in private subnet without internet	Create 3 VPC Interface Endpoints: `ssm`, `ssmmessages`, `ec2messages`

Step 7 — Install DRS Agent via Session Manager

Open Systems Manager → Session Manager in Jakarta.
Click the Start session button in the top-right corner.
On the Specify target page, select pdns-prod-server-01 → click Start session.
In the web terminal that opens, run the following commands one at a time:

Command 1 — Switch to superuser (root):

sudo su -

Command 2 — Install kernel headers & build tools:

sudo dnf install -y kernel-devel kernel-headers gcc make

(Wait until output displays Complete!).

Command 3 — Download official DRS installer:

wget -O /tmp/aws-replication-installer-init.py \
https://aws-elastic-disaster-recovery-ap-southeast-1.s3.amazonaws.com/latest/linux/aws-replication-installer-init.py

Command 4 — Run installer:

sudo python3 /tmp/aws-replication-installer-init.py --region ap-southeast-1 --no-prompt

Wait until execution completes (~10–20 minutes). Installation stages:

Stage	Terminal Output	Estimated Duration
Volume detection	`All volumes were successfully identified.`	~30 seconds
Agent download	`Downloading the AWS Replication Agent... Finished.`	~3–5 minutes
Agent install	`Installing the AWS Replication Agent onto the source server...`	~5–10 minutes
Complete	`The AWS Replication Agent was successfully installed.`	Complete

Command 5 — Enable agent auto-start on reboot:

sudo systemctl enable aws-replication-agent

8D. Verify Replication Status

Switch region to ap-southeast-1 (Singapore).
Open Elastic Disaster Recovery → Source servers.
Inspect the Data replication status column:

Status	Indicator	Meaning	Required Action
Initiating	🟡 Yellow	Agent recently connected	Wait for handshake
Initial sync	🔵 Blue	First synchronization in progress	Wait until 100%
Continuous	🟢 Green	✅ Active replication, RPO < 1 minute	✅ Done & protected!
Disconnected	🔴 Red	Network or agent connectivity lost	Check outbound port 1500 and agent service

8E. Configure Launch Template for Recovery

In DRS left menu, click Launch settings.
Select the source server → click Edit.
Configure the launch template:

Parameter	Value	Rationale
Launch disposition	Stopped	Cost optimization; recovered instance started manually during DR
Target subnet	Subnet in `pdns-recovery-vpc`	Places recovery VM in isolated network
Security groups	Recovery Security Group	Governs recovery firewall rules
Instance type	Match production (`t3.medium`)	Ensures equivalent compute capacity

Click Save template.

note : If you've achieved 100% replication or the status is green, you can proceed to 8E. In my case, it didn't reach the green status because my internet connection is slow.

💰 Estimated Monthly Architecture Cost

AWS Service	Assumed Capacity & Workload	Estimated Cost / Month	Business Value & Protection
AWS Backup	1 TB Backup storage in Jakarta + cross-region copy to Singapore	~$80	Protects database snapshots & EBS volumes
S3 Object Lock (WORM)	5 TB Immutable data storage (Standard + Glacier)	~$115	Data archives cannot be deleted or modified by attackers
AWS Elastic DRS	5 Production servers continuously replicated to Singapore	~$150	Block-level disk replication with RTO < 1 hour
Amazon GuardDuty	VPC Flow, CloudTrail, DNS, S3, & EKS log analysis	~$90	Early threat detection before encryption spreads
AWS Security Hub	Centralized security findings aggregation & compliance scoring	~$50	Unified security posture visibility
AWS CloudTrail	Free management events + S3/Lambda Data Events	~$40	Tamper-proof digital forensic audit evidence
AWS KMS	2 Customer Managed Keys (Jakarta & Singapore) + API calls	~$20	Independent cryptographic key isolation per region
TOTAL ESTIMATED COST	Complete Ransomware-Resilient DR Architecture	~$545 / Month	Insurance against tens of millions of dollars in losses

💡 Cost vs. Impact: In the June 2024 PDNS ransomware incident, attackers demanded an $8,000,000 ransom while disrupting 230+ public agencies for weeks. The prevention and resilience architecture documented here costs only ~$545/month.

💰 Cost vs. Benefit

Item	Value
Brain Cipher ransom demand	USD 8,000,000
Estimated cost of 2 weeks of downtime	USD 50,000,000+
Public trust and reputation	Priceless
Cost of this architecture/month	~USD 720
ROI	11,000x+

🔗 Resources

📁 Full repository: github.com/yourusername/aws-ransomware-dr
📄 AWS Backup docs: docs.aws.amazon.com/aws-backup
🛡️ AWS DRS docs: docs.aws.amazon.com/drs
🔍 GuardDuty best practices: docs.aws.amazon.com/guardduty

My First Experience as a Speaker at the AWS Summit 2026.

Daffa Rabbani — Sun, 09 Aug 2026 03:50:41 +0000

Hi everyone 🙋‍♂️,

Let me introduce myself—I’m Daffa. By the way, I’m still a final-semester student in the three-year Information Systems diploma program at UPN Veteran Jakarta. I’m really impressed—I never imagined I’d get to join as a speaker at the 2026 Summit, especially since I just sent in the proposal on a whim 😁.

Here, I’ve also met many fellow community builders and members of the Jakarta user group—even though there were quite a lot of people attending the AWS Summit 2026, whom I usually only see on social media. Unfortunately, I haven’t had the chance to meet Mr. Shafraz Rahim—Senior Program Manager, Community, APJC 🥀🥀.

Yesterday during the event, I had the opportunity to discuss “Building a Cost-Efficient Cloud-Native SOAR System on AWS.”

Photos

when I spoke at the Dev Lounge

speaker poster

SWAG 😻😻

Live Demo

Demo

Thank you very much 🙏😁

Thank you for coming to my event

🔐Automated Security Incident Response System ⚠

Daffa Rabbani — Fri, 26 Dec 2025 12:57:41 +0000

Implementation Guide - 2 Days (No Docker)

Based on: CrowdStrike Incident July 2024 - $5.4B Losses Timeline: 2 Hari | Budget: ~$90/bulan | Complexity: Advanced

📋 Prerequisites

What You Need:

AWS Account with admin access (Best Practice Please Use your IAM User * with Admin Access not your Root Account)
Basic Python knowledge
Text editor (VS Code, Notepad++)
AWS CLI installed (optional, can just use Console)

Budget Breakdown:

GuardDuty: $4.50/month (30-day free trial)
Security Hub: $0.0010 per check = ~$30/month
Lambda: $0.20/million requests = ~$5/month
Step Functions: $25 per 1K executions = ~$10/month
CloudTrail: $2 per 100K events = ~$20/month
SNS: $0.50/million notifications = ~$1/month
Config: $2 per active rule = ~$20/month

Total: ~$90/month

🏗️ Architecture Overview

📅 DAY 1: Foundation & Detection (8 Hours)

📝 PHASE 1: Setup GuardDuty (1 Hour)

step 1.1 : Enable GuardDuty

via AWS Console:

Login AWS Console → Search "GuardDuty"
Click Get Started
Click Enable GuardDuty
Wait 15-30 seconds for initialization

Via AWS CLI:

aws guardduty create-detector \
  --enable \
  --finding-publishing-frequency FIFTEEN_MINUTES

✅ Verification: GuardDuty console will show "No findings" - this is normal, it takes time to detect threats

Step 1.2: Generate Sample Findings (for Testing)

EXAMPLE :

Get detector ID

DETECTOR_ID=$(aws guardduty list-detectors --query 'DetectorIds[0]' --output text)

Generate sample findings

aws guardduty create-sample-findings \
  --detector-id $DETECTOR_ID \
  --finding-types "Recon:EC2/PortProbeUnprotectedPort" \
            "UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration"\
              "CryptoCurrency:EC2/BitcoinTool.B!DNS"

Via Console:

GuardDuty → Settings → Sample findings
Click Generate sample findings

✅ Verification: GuardDuty → Findings → You'll see 50+ sample findings

📝 PHASE 2: Setup Security Hub (1 Hour)

Security Hub adalah centralized security dashboard.

Step 2.1: Enable Security Hub

Via Console:

Search "Security Hub"
Click Go to Security Hub
Click Enable Security Hub
Select AWS Foundational Security Best Practices standard
Click Enable Security Hub

Via CLI:

aws securityhub enable-security-hub \
  --enable-default-standards

Wait 5-10 minutes untuk Security Hub populate findings.

Step 2.2: Enable GuardDuty Integration

Security Hub → Integrations
Find AWS GuardDuty
Click Accept findings

✅ Verification: Security Hub → Findings → akan muncul findings dari GuardDuty

📝 PHASE 3: Setup SNS for Alerts (30 Minutes)

Step 3.1: Create SNS Topic

Via Console:

Search SNS → Topics
Click Create topic
Type: Standard
Name: security-incident-alerts
Display name: Security Alerts
Click Create topic

Save the Topic ARN - it will look like:

arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT_ID:security-incident-alerts

Step 3.2: Create Email Subscription

Topic → Create subscription
Protocol: Email
Endpoint: your-email@company.com
Click Create subscription
Check your email → Click confirmation link

Step 3.3: Create Slack Subscription (Optional)

Via Console:

Topic → Create subscription
Protocol: AWS Chatbot
Configure Slack workspace integration

Or use webhook:

We'll integrate this in Lambda later

✅ Verification: Send test message:
Example:

aws sns publish \
  --topic-arn arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT_ID:security-incident-alerts \
  --message "Test alert - Security system operational"

📝 PHASE 4: Create IAM Role for Lambda (30 Minutes)

Step 4.1: Create Execution Role

Via Console:

IAM Console → Roles → Create role
Trusted entity: AWS service → Lambda
Permissions policies - Add these:

AWSLambdaBasicExecutionRole (managed)
AmazonEC2FullAccess (managed)
IAMFullAccess (managed)
AmazonSNSFullAccess (managed)

Role name: SecurityIncidentResponseRole
Click Create role

Step 4.2: Add Inline Policy for Additional Permissions

Role → Add permissions → Create inline policy
JSON tab → Paste:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "ec2:DescribeInstances",
        "ec2:DescribeSecurityGroups",
        "ec2:CreateSecurityGroup",
        "ec2:AuthorizeSecurityGroupIngress",
        "ec2:AuthorizeSecurityGroupEgress",
        "ec2:RevokeSecurityGroupIngress",
        "ec2:RevokeSecurityGroupEgress",
        "ec2:ModifyInstanceAttribute",
        "ec2:CreateSnapshot",
        "ec2:CreateTags",
        "iam:ListAccessKeys",
        "iam:DeleteAccessKey",
        "iam:UpdateLoginProfile",
        "iam:CreateAccessKey",
        "sts:GetCallerIdentity",
        "guardduty:GetFindings",
        "guardduty:ListFindings",
        "securityhub:GetFindings",
        "cloudtrail:LookupEvents"
      ],
      "Resource": "*"
    }
  ]
}

Name: SecurityResponsePermissions
Click Create policy

📝 PHASE 5: Create Auto-Response Lambda Functions (3 Hours)

Step 5.1: Create Main Response Function

Via Console:

Lambda Console → Create function
Function name: security-incident-responder
Runtime: Python 3.11
Architecture: x86_64
Execution role: Use existing role → SecurityIncidentResponseRole
Click Create function

Step 5.2: Write Lambda Code (No Docker Needed!)

Click Code tab, replace with this code:

import json
import boto3
import os
from datetime import datetime


ec2 = boto3.client('ec2')
iam = boto3.client('iam')
sns = boto3.client('sns')
cloudtrail = boto3.client('cloudtrail')


SNS_TOPIC_ARN = os.environ.get('SNS_TOPIC_ARN', 'arn:aws:sns:{Your_Region}:{Your_ACCOUNT_ID}:security-incident-alerts')


def lambda_handler(event, context):
    """Main handler for security incidents"""

    print(f"Received event: {json.dumps(event, indent=2)}")

    try:
        # Parse the finding - handle both GuardDuty and Security Hub formats
        if 'findings' in event.get('detail', {}):
            # Security Hub format
            finding = event['detail']['findings'][0]
        else:
            # GuardDuty native format
            finding = event['detail']

        # Get severity - handle different formats
        if isinstance(finding.get('Severity'), dict):
            severity = finding['Severity'].get('Label', 'UNKNOWN')
        else:
            # Map numeric severity to label
            severity_num = finding.get('severity', 0)
            if severity_num >= 7:
                severity = 'HIGH'
            elif severity_num >= 8.5:
                severity = 'CRITICAL'
            elif severity_num >= 4:
                severity = 'MEDIUM'
            else:
                severity = 'LOW'

        finding_type = finding.get('type', 'Unknown')
        finding_id = finding.get('id', 'Unknown')

        print(f"Processing finding: {finding_type} (Severity: {severity})")

        # Only respond to HIGH and CRITICAL
        if severity not in ['HIGH', 'CRITICAL']:
            print(f"Severity {severity} below threshold, skipping automated response")
            return {
                'statusCode': 200,
                'body': json.dumps('Severity below threshold')
            }

        # Route to appropriate handler
        response_actions = []

        if 'UnauthorizedAccess:EC2' in finding_type or 'Backdoor:EC2' in finding_type or 'Trojan:Runtime' in finding_type:
            instance_id = extract_instance_id(finding)
            if instance_id:
                result = isolate_ec2_instance(instance_id, finding_type)
                response_actions.append(result)

        elif 'UnauthorizedAccess:IAMUser' in finding_type or 'CredentialAccess' in finding_type:
            user_name = extract_user_name(finding)
            if user_name:
                result = rotate_iam_credentials(user_name, finding_type)
                response_actions.append(result)

        elif 'Impact:EC2/PortSweep' in finding_type or 'Recon:EC2' in finding_type:
            source_ip = extract_source_ip(finding)
            if source_ip:
                result = block_suspicious_ip(source_ip, finding_type)
                response_actions.append(result)

        elif 'CryptoCurrency' in finding_type:
            instance_id = extract_instance_id(finding)
            if instance_id:
                result = handle_crypto_mining(instance_id, finding_type)
                response_actions.append(result)

        # Send comprehensive alert
        send_security_alert(finding, response_actions)

        return {
            'statusCode': 200,
            'body': json.dumps({
                'message': 'Security incident processed',
                'finding_id': finding_id,
                'actions_taken': response_actions
            })
        }

    except Exception as e:
        print(f"Error processing security incident: {str(e)}")
        import traceback
        traceback.print_exc()
        send_error_alert(str(e), event)

        return {
            'statusCode': 500,
            'body': json.dumps(f'Error: {str(e)}')
        }


def extract_instance_id(finding):
    """Extract EC2 instance ID from finding"""
    try:
        for resource in finding.get('Resources', []):
            if resource.get('Type') == 'Instance':
                instance_id = resource.get('Id', '').split('/')[-1]
                return instance_id if instance_id.startswith('i-') else None
    except Exception as e:
        print(f"Error extracting instance ID: {e}")
    return None


def extract_user_name(finding):
    """Extract IAM user name from finding"""
    try:
        for resource in finding.get('Resources', []):
            if resource.get('Type') == 'AccessKey':
                return resource.get('AccessKeyDetails', {}).get('UserName')
    except Exception as e:
        print(f"Error extracting user name: {e}")
    return None


def extract_source_ip(finding):
    """Extract source IP from finding"""
    try:
        service = finding.get('Service', {})
        action = service.get('Action', {})
        network_connection = action.get('NetworkConnectionAction', {})
        return network_connection.get('RemoteIpDetails', {}).get('IpAddressV4')
    except Exception as e:
        print(f"Error extracting source IP: {e}")
    return None


def isolate_ec2_instance(instance_id, finding_type):
    """Isolate compromised EC2 instance"""
    try:
        print(f"Isolating instance: {instance_id}")

        # Get instance details
        response = ec2.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]
        vpc_id = instance['VpcId']

        # Create isolation security group
        sg_name = f'isolation-{instance_id}-{datetime.now().strftime("%Y%m%d%H%M%S")}'

        sg_response = ec2.create_security_group(
            GroupName=sg_name,
            Description=f'Quarantine SG for {instance_id} - {finding_type}',
            VpcId=vpc_id
        )

        isolation_sg_id = sg_response['GroupId']

        # Remove all default egress rules (blocks all outbound)
        ec2.revoke_security_group_egress(
            GroupId=isolation_sg_id,
            IpPermissions=[{
                'IpProtocol': '-1',
                'FromPort': -1,
                'ToPort': -1,
                'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
            }]
        )

        # Add only SSH access from admin IP (optional)
        # ec2.authorize_security_group_ingress(
        #     GroupId=isolation_sg_id,
        #     IpPermissions=[{
        #         'IpProtocol': 'tcp',
        #         'FromPort': 22,
        #         'ToPort': 22,
        #         'IpRanges': [{'CidrIp': 'YOUR_ADMIN_IP/32', 'Description': 'Admin access'}]
        #     }]
        # )

        # Apply isolation security group to instance
        ec2.modify_instance_attribute(
            InstanceId=instance_id,
            Groups=[isolation_sg_id]
        )

        # Create snapshot for forensics
        volumes = [vol['Ebs']['VolumeId'] for vol in instance.get('BlockDeviceMappings', [])]
        snapshots = []

        for volume_id in volumes:
            snapshot = ec2.create_snapshot(
                VolumeId=volume_id,
                Description=f'Forensic snapshot - {finding_type}',
                TagSpecifications=[{
                    'ResourceType': 'snapshot',
                    'Tags': [
                        {'Key': 'Purpose', 'Value': 'Forensics'},
                        {'Key': 'InstanceId', 'Value': instance_id},
                        {'Key': 'Incident', 'Value': finding_type}
                    ]
                }]
            )
            snapshots.append(snapshot['SnapshotId'])

        action_summary = {
            'action': 'isolate_instance',
            'instance_id': instance_id,
            'isolation_sg': isolation_sg_id,
            'forensic_snapshots': snapshots,
            'status': 'success'
        }

        print(f"Instance {instance_id} successfully isolated")
        return action_summary

    except Exception as e:
        print(f"Error isolating instance {instance_id}: {e}")
        return {
            'action': 'isolate_instance',
            'instance_id': instance_id,
            'status': 'failed',
            'error': str(e)
        }


def rotate_iam_credentials(user_name, finding_type):
    """Rotate compromised IAM credentials"""
    try:
        print(f"Rotating credentials for user: {user_name}")

        actions_taken = []

        # List and delete all access keys
        keys_response = iam.list_access_keys(UserName=user_name)
        deleted_keys = []

        for key in keys_response['AccessKeyMetadata']:
            access_key_id = key['AccessKeyId']

            # Deactivate first (safer than immediate delete)
            iam.update_access_key(
                UserName=user_name,
                AccessKeyId=access_key_id,
                Status='Inactive'
            )

            deleted_keys.append(access_key_id)
            actions_taken.append(f'Deactivated access key: {access_key_id}')

        # Force password reset (if console access exists)
        try:
            iam.update_login_profile(
                UserName=user_name,
                PasswordResetRequired=True
            )
            actions_taken.append('Forced password reset')
        except iam.exceptions.NoSuchEntityException:
            actions_taken.append('No console access to reset')

        # Attach deny-all policy (additional safety)
        policy_document = {
            "Version": "2012-10-17",
            "Statement": [{
                "Effect": "Deny",
                "Action": "*",
                "Resource": "*"
            }]
        }

        try:
            iam.put_user_policy(
                UserName=user_name,
                PolicyName='EmergencyDenyAll',
                PolicyDocument=json.dumps(policy_document)
            )
            actions_taken.append('Applied emergency deny policy')
        except Exception as e:
            print(f"Could not apply deny policy: {e}")

        action_summary = {
            'action': 'rotate_credentials',
            'user_name': user_name,
            'deactivated_keys': deleted_keys,
            'actions': actions_taken,
            'status': 'success'
        }

        print(f"Credentials rotated for {user_name}")
        return action_summary

    except Exception as e:
        print(f"Error rotating credentials for {user_name}: {e}")
        return {
            'action': 'rotate_credentials',
            'user_name': user_name,
            'status': 'failed',
            'error': str(e)
        }
def block_suspicious_ip(source_ip, finding_type):
    """Block suspicious IP using NACL"""
    try:
        print(f"Blocking suspicious IP: {source_ip}")

        # Get default VPC
        vpcs = ec2.describe_vpcs(Filters=[{'Name': 'isDefault', 'Values': ['true']}])

        if not vpcs['Vpcs']:
            return {
                'action': 'block_ip',
                'source_ip': source_ip,
                'status': 'failed',
                'error': 'No default VPC found'
            }

        vpc_id = vpcs['Vpcs'][0]['VpcId']

        # Get network ACLs for VPC
        nacls = ec2.describe_network_acls(
            Filters=[{'Name': 'vpc-id', 'Values': [vpc_id]}]
        )

        if not nacls['NetworkAcls']:
            return {
                'action': 'block_ip',
                'source_ip': source_ip,
                'status': 'failed',
                'error': 'No NACLs found'
            }

        nacl_id = nacls['NetworkAcls'][0]['NetworkAclId']

        # Add deny rule (rule number 1 for highest priority)
        ec2.create_network_acl_entry(
            NetworkAclId=nacl_id,
            RuleNumber=1,
            Protocol='-1',  # All protocols
            RuleAction='deny',
            CidrBlock=f'{source_ip}/32'
        )

        action_summary = {
            'action': 'block_ip',
            'source_ip': source_ip,
            'nacl_id': nacl_id,
            'status': 'success'
        }

        print(f"IP {source_ip} successfully blocked")
        return action_summary

    except Exception as e:
        print(f"Error blocking IP {source_ip}: {e}")
        return {
            'action': 'block_ip',
            'source_ip': source_ip,
            'status': 'failed',
            'error': str(e)
        }


def handle_crypto_mining(instance_id, finding_type):
    """Handle cryptocurrency mining detection"""
    try:
        print(f"Handling crypto mining on instance: {instance_id}")

        # Stop the instance immediately
        ec2.stop_instances(InstanceIds=[instance_id])

        # Create forensic snapshot
        response = ec2.describe_instances(InstanceIds=[instance_id])
        instance = response['Reservations'][0]['Instances'][0]
        volumes = [vol['Ebs']['VolumeId'] for vol in instance.get('BlockDeviceMappings', [])]

        snapshots = []
        for volume_id in volumes:
            snapshot = ec2.create_snapshot(
                VolumeId=volume_id,
                Description=f'Crypto mining forensics - {instance_id}'
            )
            snapshots.append(snapshot['SnapshotId'])

        action_summary = {
            'action': 'stop_crypto_mining',
            'instance_id': instance_id,
            'instance_stopped': True,
            'forensic_snapshots': snapshots,
            'status': 'success'
        }

        print(f"Crypto mining instance {instance_id} stopped")
        return action_summary

    except Exception as e:
        print(f"Error handling crypto mining on {instance_id}: {e}")
        return {
            'action': 'stop_crypto_mining',
            'instance_id': instance_id,
            'status': 'failed',
            'error': str(e)
        }


def send_security_alert(finding, response_actions):
    """Send comprehensive security alert"""
    try:
        # Get severity - handle different formats
        if isinstance(finding.get('Severity'), dict):
            severity = finding['Severity'].get('Label', 'UNKNOWN')
        else:
            severity_num = finding.get('severity', 0)
            if severity_num >= 8.5:
                severity = 'CRITICAL'
            elif severity_num >= 7:
                severity = 'HIGH'
            elif severity_num >= 4:
                severity = 'MEDIUM'
            else:
                severity = 'LOW'

        # Get finding type (lowercase 'type' for GuardDuty, uppercase 'Type' for Security Hub)
        finding_type = finding.get('type') or finding.get('Type', 'Unknown')

        # Get description
        description = finding.get('description') or finding.get('Description', 'No description')

        # Get finding ID
        finding_id = finding.get('id') or finding.get('Id', 'Unknown')

        # Build alert message
        message = f"""
🚨 SECURITY INCIDENT DETECTED 🚨


Severity: {severity}
Type: {finding_type}
Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S UTC')}


Description:
{description}


AUTOMATED ACTIONS TAKEN:
"""

        if response_actions:
            for action in response_actions:
                message += f"\n✓ {action.get('action', 'unknown')}: {action.get('status', 'unknown')}"
                if action.get('status') == 'failed':
                    message += f" - Error: {action.get('error', 'unknown')}"
        else:
            message += "\nNo automated actions taken (instance not found or not applicable)"

        message += "\n\nSOC Team: Please review and investigate further."
        message += f"\n\nFinding ID: {finding_id}"

        # Send to SNS
        sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject=f'🚨 SECURITY ALERT: {finding_type}',
            Message=message
        )

        print("Security alert sent successfully")

    except Exception as e:
        print(f"Error sending security alert: {e}")
        import traceback
        traceback.print_exc()


def send_error_alert(error_message, event):
    """Send alert when response function fails"""
    try:
        message = f"""
❌ SECURITY RESPONSE ERROR


An error occurred while processing a security incident.


Error: {error_message}


Event: {json.dumps(event, indent=2)}


Action Required: Manual investigation needed.
"""

        sns.publish(
            TopicArn=SNS_TOPIC_ARN,
            Subject='❌ Security Response System Error',
            Message=message
        )

    except Exception as e:
        print(f"Failed to send error alert: {e}")

TEST CODE

Example JSON:

{
  "version": "0",
  "id": "test-event-12345",
  "detail-type": "GuardDuty Finding",
  "source": "aws.guardduty",
  "account": "{YOUR_ACCOUNT_ID}",
  "time": "2024-12-17T10:00:00Z",
  "region": "ap-southeast-2",
  "resources": [],
  "detail": {
    "schemaVersion": "2.0",
    "accountId": "{YOUR_ACCOUNT_ID}",
    "region": "ap-southeast-2",
    "partition": "aws",
    "id": "test-finding-ec2-123",
    "arn": "arn:aws:guardduty:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:detector/test/finding/test-123",
    "type": "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom",
    "severity": 8,
    "createdAt": "2024-12-17T09:35:00.000Z",
    "updatedAt": "2024-12-17T09:35:00.000Z",
    "title": "EC2 instance is communicating with a malicious IP address",
    "description": "EC2 instance i-test12345678 is communicating with malicious IP 198.51.100.1. This is a TEST finding for security automation validation.",
    "service": {
      "serviceName": "guardduty",
      "detectorId": "test-detector-id",
      "action": {
        "actionType": "NETWORK_CONNECTION",
        "networkConnectionAction": {
          "blocked": false,
          "connectionDirection": "INBOUND",
          "localPortDetails": {
            "port": 22,
            "portName": "SSH"
          },
          "protocol": "TCP",
          "remoteIpDetails": {
            "ipAddressV4": "198.51.100.1",
            "organization": {
              "asn": "12345",
              "asnOrg": "Malicious ASN"
            },
            "country": {
              "countryName": "Unknown"
            }
          }
        }
      },
      "archived": false,
      "count": 1,
      "eventFirstSeen": "2024-12-17T09:30:00.000Z",
      "eventLastSeen": "2024-12-17T09:35:00.000Z",
      "resourceRole": "TARGET"
    },
    "Severity": {
      "Label": "HIGH",
      "Normalized": 70,
      "Product": 8.0
    },
    "Resources": [
      {
        "Type": "Instance",
        "Id": "arn:aws:ec2:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:instance/i-test12345678",
        "Details": {
          "Instance": {
            "InstanceId": "i-test12345678",
            "InstanceType": "t2.micro",
            "LaunchTime": "2024-12-17T09:00:00.000Z",
            "Platform": "Linux"
          }
        }
      }
    ],
    "Id": "test-finding-ec2-123"
  }
}

Step 5.3: Configure Lambda

Configuration tab → Environment variables
Add variable:

Key: SNS_TOPIC_ARN
Value: arn:aws:sns:{YOUR_REGION}:{ACCOUNT_ID}:security-incident alerts (your SNS ARN) (Your SNS arn)

General configuration → Edit:
- Memory: 512 MB
- Timeout: 5 minutes

Click Save
Deploy the Code

✅ Verification: Lambda function created successfully

📝 PHASE 6: Connect EventBridge to Lambda (1 Hour)

Step 6.1: Create EventBridge Rule for GuardDuty

Via Console:

EventBridge Console → Rules → Create rule
Name: guardduty-high-severity-findings
Description: Route high severity GuardDuty findings to Lambda (optional)
Event bus: default
Rule type: Rule with an event pattern
Click Next
Event pattern:

{
  "source": ["aws.guardduty"],
  "detail-type": ["GuardDuty Finding"],
  "detail": {
    "severity": [7, 7.0, 7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.7, 7.8, 7.9, 8, 8.0, 8.1, 8.2, 8.3, 8.4, 8.5, 8.6, 8.7, 8.8, 8.9]
  }
}

Click Next
Select target:

Target type: AWS service
Select a target: Lambda function
Function: security-incident-responder

Click Next → Create rule

Step 6.2: Create Rule for Security Hub

Create another rule: securityhub-critical-findings
Event pattern:

{
  "source": ["aws.securityhub"],
  "detail-type": ["Security Hub Findings - Imported"],
  "detail": {
    "findings": {
      "Severity": {
        "Label": ["CRITICAL", "HIGH"]
      }
    }
  }
}

Target: Same Lambda function

✅ Verification: EventBridge → Rules → 2 rules active

📝 PHASE 7: Testing with Sample Findings (1 Hour)

Step 7.1: Trigger Test Finding

Example :

# Generate GuardDuty sample finding
aws guardduty create-sample-findings \
  --detector-id $DETECTOR_ID \
  --finding-types "UnauthorizedAccess:EC2/MaliciousIPCaller.Custom"

Via Console:

GuardDuty → Settings → Sample findings
Generate sample findings

Step 7.2: Monitor Lambda Execution

Lambda Console → security-incident-responder
Monitor tab → View logs in CloudWatch
Check latest log stream

You should see logs like:

Processing finding: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom (Severity: HIGH)
Instance i-xxxxx successfully isolated
Security alert sent successfully

Step 7.3: Check Email

Check your email - you should receive alert:

🚨 SECURITY INCIDENT DETECTED 🚨
Severity: HIGH
Type: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom

✅ Day 1 Complete! Basic automated response working.

📅 DAY 2: Advanced Workflows & Compliance (8 Hours)

📝 PHASE 8: Setup CloudTrail (1 Hour)

CloudTrail logs all API calls for audit trail.

Step 8.1: Enable CloudTrail

Via Console:

CloudTrail Console → Create trail
Trail name: security-audit-trail
Storage location: Create new S3 bucket
Bucket name: security-audit-logs-ACCOUNT-ID
Log file SSE-KMS encryption: Enabled(use default AWS managed key)
CloudWatch Logs: Enabled
- Log group name: /aws/cloudtrail/security-audit
Click Next
Event type:
- Management events: ✓ Read and ✓ Write
- Data events: Skip for now (save cost)
Click Next → Create trail

✅ Verification: CloudTrail → Trails → Status "Logging"

📝 PHASE 9: Setup AWS Config (1.5 Hours)

AWS Config monitors resource configurations for compliance.

Step 9.1: Enable Config

Via Console:

AWS Config Console →Get started
Resource types to record:
- Select:Record all resources
- Include global resources: Yes
Amazon S3 bucket:
- Create a bucket: Yes
- Bucket name: config-audit-ACCOUNT-ID
SNS topic:
- Stream configuration changes: Yes
- Select existing SNS topic: security-incident-alerts
AWS Config role:
- Create AWS Config service-linked role: Yes
Click Next

Step 9.2: Add Compliance Rules

Add these managed rules:

iam-password-policy
- Ensures strong password policy
root-account-mfa-enabled
- Ensures root account has MFA
ec2-instance-managed-by-systems-manager
- Ensures EC2 instances are managed
encrypted-volumes
- Ensures EBS volumes are encrypted
s3-bucket-public-read-prohibited
- Ensures S3 buckets not publicly readable
cloudtrail-enabled
- Ensures CloudTrail is enabled

To add each rule:
1.Config → Rules → Add rule
2.Search for rule name
3.Click rule → Next
4.Leave defaults → Save

✅ Verification: Config → Rules → 6 rules active, evaluating compliance

📝 PHASE 10: Build Step Functions Workflow (2 Hours)

Step Functions untuk complex multi-step response workflows.

Step 10.1: Create Step Functions State Machine

Via Console:

Step Functions Console → Create state machine
Choose template: Blank
Type: Standard
Name: security-incident-workflow
Definition (paste this JSON):

{
  "Comment": "Advanced Security Incident Response Workflow",
  "StartAt": "ParseIncident",
  "States": {
    "ParseIncident": {
      "Type": "Task",
      "Resource": "arn:aws:states:::lambda:invoke",
      "Parameters": {
        "FunctionName": "security-incident-responder",
        "Payload.$": "$"
      },
      "ResultPath": "$.responseResult",
      "Next": "CheckSeverity",
      "Catch": [{
        "ErrorEquals": ["States.ALL"],
        "Next": "NotifyFailure"
      }]
    },
    "CheckSeverity": {
      "Type": "Choice",
      "Choices": [
        {
          "Variable": "$.detail.findings[0].Severity.Label",
          "StringEquals": "CRITICAL",
          "Next": "CriticalIncidentPath"
        },
        {
          "Variable": "$.detail.findings[0].Severity.Label",
          "StringEquals": "HIGH",
          "Next": "HighIncidentPath"
        }
      ],
      "Default": "LogAndExit"
    },
    "CriticalIncidentPath": {
      "Type": "Parallel",
      "Branches": [
        {
          "StartAt": "CreateTicket",
          "States": {
            "CreateTicket": {
              "Type": "Task",
              "Resource": "arn:aws:states:::sns:publish",
              "Parameters": {
                "TopicArn": "arn:aws:sns:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:security-incident-alerts",
                "Subject": "CRITICAL Security Incident - Immediate Action Required",
                "Message.$": "$.detail.findings[0].Description"
              },
              "End": true
            }
          }
        },
        {
          "StartAt": "WaitForSOC",
          "States": {
            "WaitForSOC": {
              "Type": "Wait",
              "Seconds": 300,
              "Next": "CheckIfResolved"
            },
            "CheckIfResolved": {
              "Type": "Task",
              "Resource": "arn:aws:states:::lambda:invoke",
              "Parameters": {
                "FunctionName": "check-incident-status",
                "Payload": {
                  "findingId.$": "$.detail.findings[0].Id"
                }
              },
              "End": true
            }
          }
        }
      ],
      "Next": "LogSuccess"
    },
    "HighIncidentPath": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:{YOUR_REGION}:{YOUR_ACCOUNT_ID}:security-incident-alerts",
        "Subject": "HIGH Security Incident Detected",
        "Message.$": "$.detail.findings[0].Description"
      },
      "Next": "LogSuccess"
    },
    "LogSuccess": {
      "Type": "Succeed"
    },
    "LogAndExit": {
      "Type": "Succeed"
    },
    "NotifyFailure": {
      "Type": "Task",
      "Resource": "arn:aws:states:::sns:publish",
      "Parameters": {
        "TopicArn": "arn:aws:sns:YOUR_REGION:YOUR_ACCOUNT:security-incident-alerts",
        "Subject": "Security Response Workflow Failed",
        "Message": "The security incident response workflow encountered an error."
      },
      "Next": "FailState"
    },
    "FailState": {
      "Type": "Fail",
      "Error": "WorkflowFailed",
      "Cause": "Security response workflow failed to complete"
    }
  }
}

Replace REGION and ACCOUNT with your Value
Click Next
Permissions:
- Create new role: Yes
- Role name: StepFunctions-SecurityWorkflow-Role
  1. Click Create state machine

Step 10.2: Update EventBridge to Trigger Step Functions

EventBridge → Rules → Create new rule
Name: critical-findings-to-stepfunctions
Event pattern:

{
  "source": ["aws.guardduty", "aws.securityhub"],
  "detail": {
    "findings": {
      "Severity": {
        "Label": ["CRITICAL"]
      }
    }
  }
}

Target: Step Functions state machine
State machine: security-incident-workflow
Click Create

📝 PHASE 11: Advanced Response Functions (2 Hours)

Create additional Lambda functions untuk specific scenarios.

Step 11.1: Create Forensics Collection Function

Via Console:

Lambda → Create function
Name: forensics-collector
Runtime: Python 3.11
Role: SecurityIncidentResponseRole

Code

import boto3
import json

config = boto3.client('config')
iam = boto3.client('iam')
ec2 = boto3.client('ec2')

def lambda_handler(event, context):
    """Check compliance status and generate report"""

    compliance_report = {
        'timestamp': event['time'],
        'checks': {}
    }

    # 1. Check Config rules compliance
    config_rules = config.describe_compliance_by_config_rule()

    non_compliant = []
    for rule in config_rules['ComplianceByConfigRules']:
        if rule['Compliance']['ComplianceType'] == 'NON_COMPLIANT':
            non_compliant.append(rule['ConfigRuleName'])

    compliance_report['checks']['config_rules'] = {
        'total': len(config_rules['ComplianceByConfigRules']),
        'non_compliant': non_compliant
    }

    # 2. Check IAM password policy
    try:
        password_policy = iam.get_account_password_policy()
        compliance_report['checks']['iam_password_policy'] = {
            'exists': True,
            'minimum_length': password_policy['PasswordPolicy'].get('MinimumPasswordLength', 0),
            'require_symbols': password_policy['PasswordPolicy'].get('RequireSymbols', False),
            'require_numbers': password_policy['PasswordPolicy'].get('RequireNumbers', False)
        }
    except iam.exceptions.NoSuchEntityException:
        compliance_report['checks']['iam_password_policy'] = {
            'exists': False,
            'compliant': False
        }

    # 3. Check MFA on root account
    try:
        summary = iam.get_account_summary()
        compliance_report['checks']['root_mfa'] = {
            'enabled': summary['SummaryMap'].get('AccountMFAEnabled', 0) == 1
        }
    except Exception as e:
        compliance_report['checks']['root_mfa'] = {'error': str(e)}

    # 4. Check for unencrypted EBS volumes
    volumes = ec2.describe_volumes()
    unencrypted = [v['VolumeId'] for v in volumes['Volumes'] if not v.get('Encrypted', False)]

    compliance_report['checks']['ebs_encryption'] = {
        'total_volumes': len(volumes['Volumes']),
        'unencrypted': unencrypted,
        'compliant': len(unencrypted) == 0
    }

    # 5. Check for public S3 buckets
    s3 = boto3.client('s3')
    buckets = s3.list_buckets()['Buckets']

    public_buckets = []
    for bucket in buckets:
        try:
            acl = s3.get_bucket_acl(Bucket=bucket['Name'])
            for grant in acl['Grants']:
                if grant['Grantee'].get('Type') == 'Group' and 'AllUsers' in grant['Grantee'].get('URI', ''):
                    public_buckets.append(bucket['Name'])
                    break
        except:
            pass

    compliance_report['checks']['s3_public_access'] = {
        'public_buckets': public_buckets,
        'compliant': len(public_buckets) == 0
    }

    # Calculate overall compliance score
    compliant_checks = sum([
        1 for check in compliance_report['checks'].values()
        if isinstance(check, dict) and check.get('compliant', False)
    ])
    total_checks = len(compliance_report['checks'])

    compliance_report['overall_score'] = (compliant_checks / total_checks * 100) if total_checks > 0 else 0

    return {
        'statusCode': 200,
        'body': json.dumps(compliance_report, default=str)
    }

📝 PHASE 12: Create CloudWatch Dashboard (1.5 Hours)

Step 12.1: Create Custom Dashboard

Via Console:

CloudWatch Console → Dashboards → Create dashboard
Dashboard name: SecurityIncidentResponse
Click Create dashboard

Step 12.2: Add Widgets

Widget 1: GuardDuty Findings Count

Add widget → Number
Metrics → GuardDuty → Select:
- AWS/GuardDuty → FindingCount
Statistic: Sum
Period: 1 hour
Widget title: "GuardDuty Findings (Last Hour)"

Widget 2: Lambda Invocation Count

Add widget → Line
Metrics → Lambda → By Function Name
Select: security-incident-responder → Invocations
Period: 5 minutes
Title: "Auto-Response Invocations"

Widget 3: Lambda Errors

Add widget → Line
Select: security-incident-responder → Errors
Period: 5 minutes
Title: "Response Function Errors"

Widget 4: Security Hub Findings by Severity

Add widget → Stacked area
Custom namespace: Add metric → Security Hub
Title: "Security Hub Findings by Severity"

Widget 5: Recent CloudTrail Events

Add widget → Logs table
Log group: /aws/cloudtrail/security-audit
Query:

  fields @timestamp, userIdentity.principalId, eventName, sourceIPAddress
| filter errorCode like /Unauthorized/ or errorCode like /AccessDenied/
| sort @timestamp desc
| limit 20

Title: "Recent Unauthorized Access Attempts"

Widget 6: Step Functions Executions (optional)

Add widget → Number
Metrics → Step Functions → ExecutionsStarted
State machine: security-incident-workflow
Title: "Workflow Executions"
Click Save dashboard

📝 Phase 13: Create Runbook via AWS Console

Option 1: Save in S3 (Recommended)

Step 1: Create S3 Bucket

S3 Console → Create bucket
Bucket name: security-runbooks-YOUR_ACCOUNT_ID
Region: Asia Pacific (Sydney) ap-southeast-2
Block Public Access: ✅** Keep all checked (default)**
Bucket Versioning: Enable (optional)
Encryption: Enable (SSE-S3)
Click Create bucket

Step 2: Create Runbook File in your computer

Open Text editor or Notepad
Copy-paste this text:

Security Incident Response Runbook
===================================

System Overview
---------------
- Primary Function: Automated security incident detection and response
- Detection: AWS GuardDuty
- Response: Lambda automated actions
- Alerting: SNS email notifications
- Region: ap-southeast-2 (Sydney)

Quick Response Guide
--------------------

HIGH SEVERITY INCIDENTS:
1. Check email alert from SNS topic: security-incident-alerts
2. Review CloudWatch Dashboard: SecurityIncidentResponse
3. Check Lambda logs: /aws/lambda/security-incident-responder
4. Verify automated actions taken (isolate EC2, rotate IAM, etc)

MANUAL INTERVENTION REQUIRED:
If automated response failed:
1. Check error message in email alert
2. Review CloudWatch logs for detailed error
3. Manually execute remediation actions
4. Update this runbook with learnings

Automated Response Actions
--------------------------

UnauthorizedAccess:EC2 Incidents:
- Action: Isolate EC2 instance with new security group
- Action: Create forensic EBS snapshots
- Action: Send alert to SOC team

IAM Credential Compromise:
- Action: Deactivate all IAM access keys
- Action: Force password reset
- Action: Apply emergency deny-all policy

Crypto Mining Detection:
- Action: Stop EC2 instance immediately
- Action: Create forensic snapshots
- Action: Send critical alert

Malicious IP Detection:
- Action: Block source IP via Network ACL
- Action: Log incident details

System Components
-----------------

GuardDuty:
- Detector ID: 52cd95edfb049e79eada14b660d424b7
- Status: Check via Console
- Sample findings: Can generate for testing

Lambda Functions:
- security-incident-responder: Main response function
- Memory: 512 MB
- Timeout: 5 minutes
- Region: ap-southeast-2

EventBridge Rules:
- guardduty-high-severity-findings: Routes HIGH/CRITICAL findings
- Trigger: Lambda function

SNS Topics:
- security-incident-alerts: Email notifications
- Subscribers: [Your email]

CloudWatch Dashboard:
- Name: SecurityIncidentResponse
- Widgets: Lambda metrics, SNS, Logs

Step Functions:
- security-incident-workflow: Complex incident workflows
- State Machine ARN: Check Console

Contact Information
-------------------
Security Team Email: security@company.com (ganti dengan email real)
On-Call Phone: +62-xxx-xxx-xxxx (ganti dengan nomor real)
Escalation: [Manager name and contact]

AWS Account ID: YOUR_ACCOUNT_ID
Primary Region: ap-southeast-2 (Sydney)

Monthly Testing Procedures
---------------------------

Test Schedule: First Monday of each month, 10:00 AM

Test 1 - Generate Sample Finding:
1. Open GuardDuty Console
2. Settings → Sample findings
3. Click "Generate sample findings"
4. Wait 2 minutes

Test 2 - Verify Automated Response:
1. Check email for alert (should arrive within 2 min)
2. Check CloudWatch Dashboard for Lambda invocation
3. Review Lambda logs for processing details
4. Verify no errors in execution

Test 3 - Dashboard Validation:
1. Open CloudWatch Dashboard: SecurityIncidentResponse
2. Verify all widgets showing data
3. Check Lambda Invocations widget increased
4. Review Recent Logs widget for new entries

Recovery Procedures
-------------------

Rollback Isolated EC2 Instance:
1. EC2 Console → Instances
2. Find instance with isolation security group
3. Actions → Security → Change security groups
4. Replace with original security group
5. Document incident resolution

Re-enable IAM User After False Positive:
1. IAM Console → Users → Select user
2. Security credentials → Activate access keys
3. Remove "EmergencyDenyAll" policy
4. Reset password (user will change on next login)
5. Notify user via approved channel

Unblock IP from Network ACL:
1. VPC Console → Network ACLs
2. Find NACL with deny rule for IP
3. Inbound/Outbound rules → Delete rule number 1
4. Document reason for unblock

Service Costs (Monthly Estimates)
----------------------------------
GuardDuty: $4.50/month
Security Hub: $30/month (can disable to save)
Lambda: $5/month (only when invoked)
SNS: $1/month
CloudWatch: $10-20/month
CloudTrail: $2-5/month
Config: $20/month (can stop to save)
Step Functions: $10/month

TOTAL RUNNING: ~$82-95/month
TOTAL STOPPED: ~$0/month (when services suspended)

To Save Costs When Not Using:
- Suspend GuardDuty
- Disable Security Hub
- Stop Config recorder
- Stop CloudTrail logging
- Disable EventBridge rules

Services Resume Time: 2-5 minutes

Known Issues & Solutions
------------------------

Issue: Lambda timeout when isolating instance
Solution: Increase timeout to 5 minutes (already configured)

Issue: No email alerts received
Solution: 
1. Check SNS subscription is "Confirmed"
2. Check spam folder
3. Verify SNS_TOPIC_ARN environment variable in Lambda

Issue: GuardDuty metrics not in dashboard
Solution: Normal if no real findings yet. Use custom metrics from Lambda instead.

Issue: Step Functions execution failed
Solution: Check IAM role has correct permissions for Lambda and SNS

Change Log
----------
2024-12-17: Initial system setup
2024-12-17: Day 1 completed - Core automation working
2024-12-17: Day 2 completed - Dashboard and documentation

Next Review Date: [30 days from today]

Notes
-----
- This is a learning/demo implementation
- For production, add AWS WAF for additional protection
- Consider integrating with SIEM (Splunk, Elastic)
- Enable AWS Shield for DDoS protection
- Add MFA for all IAM users
- Implement least privilege IAM policies
- Regular security audits recommended

Save as: runbook.txt

Step 3: Upload ke S3

S3 Console → Buckets → security-runbooks-YOUR_ACCOUNT_ID
Click "Upload"
Add files → Select runbook.txt
Scroll down → Click "Upload"
Done! ✅

📝 PHASE 14: Final Testing & Validation (1 Hour)

Test 1: EC2 Compromise Simulation

Step 1: Generate Sample Finding

GuardDuty Console → Settings
Scroll to Sample findings
Click "Generate sample findings"
Wait 30-60 seconds

Step 2: Check Lambda Triggered

Lambda Console → Functions → security-incident-responder
Monitor tab
Metrics section:
- Check "Invocations" graph - should show spike
- Check "Duration" - should be < 5 seconds
- Check "Errors" - should be 0

Step 3: Check CloudWatch Logs

Lambda → Monitor tab
Click "View CloudWatch logs"
Click latest log stream (top)
Look for:

Processing finding: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom (Severity: HIGH)
   Found instance ID: i-99999999
   Error isolating instance... (EXPECTED - test instance)
   Security alert sent successfully

Step 4: Verify Email Alert

Check your email (in 2 minutes)
Subject: 🚨 SECURITY ALERT: UnauthorizedAccess:EC2/MaliciousIPCaller.Custom
Body should show:

Severity: HIGH
   Type: UnauthorizedAccess:EC2/...
   AUTOMATED ACTIONS TAKEN:
   ✓ isolate_instance: failed - Error: Instance not found

Step 5: Check Dashboard

CloudWatch Console → Dashboards → SecurityIncidentResponse
Verify widgets updated:
- Lambda Invocations: +1
- Lambda Duration: shows recent execution
- Recent Logs: shows new entry

✅ Test 1 PASSED if all 5 checks OK!

Test 2: IAM Credential Compromise

Step 1: Generate IAM Finding

GuardDuty Console → Settings → Sample findings
Click "Generate sample findings"
Or specific:
- GuardDuty → Findings
- Filter: Type contains "IAM"
- Look for: UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration

Step 2: Verify Lambda Processing

Lambda Console → security-incident-responder → Monitor
CloudWatch logs (latest stream)
Look for:

Processing finding: UnauthorizedAccess:IAMUser/...
   Rotating credentials for user: test-user
   Error: User not found (EXPECTED - test user)
   Security alert sent successfully

Step 3: Check Email

Email should show:

Type: UnauthorizedAccess:IAMUser/InstanceCredentialExfiltration
AUTOMATED ACTIONS TAKEN:
✓ rotate_credentials: failed - Error: User not found

✅ Test 2 PASSED if Lambda executed + email received!

Step 14.2: Performance Validation via Console

Check Dashboard Metrics:

CloudWatch Console → Dashboards → SecurityIncidentResponse
Verify each widget:

Widget: Lambda Invocations
✅ Shows 3+ invocations (from 3 tests)
✅ Recent activity visible

Widget: Lambda Errors
✅ Should show 0 errors
✅ (Failures like "Instance not found" are logged but not Lambda errors)

Widget: Lambda Duration
✅ Average < 5 seconds
✅ No timeouts

Widget: SNS Messages Published
✅ Shows 3+ messages
✅ Matches number of tests

Widget: CloudWatch Logs
✅ Shows recent log entries
✅ "Security alert sent successfully" appears

Check Lambda Performance Details:

Lambda Console → security-incident-responder → Monitor
Metrics tab:
- Duration: Average < 5000ms ✅
- Concurrent executions: < 10 ✅
- Throttles: 0 ✅
- Error rate: 0% ✅

Configuration tab → General configuration:
- Memory: 512 MB
- Timeout: 300 seconds (5 min)
- ✅ Sufficient for our use case

Step 14.3: Compliance Validation

Option 1: Check AWS Config Compliance (Simple)

AWS Config Console → Rules
Check compliance status:

Rule Name	status	Expected
iam-password-policy	✅Compliant	Strong Password Policy
root-account-mfa-enabled	✅Compliant	Root Has MFA
encrypted-volumes	⚠️ May vary	EBS encryption
Cloudtrail-enabled	✅Compliant	CloudTrail active

Overall compliance score: Should be > 70%

🏆 Resume Bullet Points
After completing this project:

"Built automated security incident response system processing 1000+ findings/month with < 2 minute response time"
"Implemented cloud-native SOAR (Security Orchestration, Automation & Response) using AWS Lambda, Step Functions, reducing manual intervention by 85%"
"Designed multi-layered threat detection using GuardDuty, Security Hub, and Config, achieving 95% threat detection accuracy"
"Architected forensics collection pipeline with automated evidence preservation compliant with SOC2 and ISO 27001"
"Reduced security incident response time from 2 hours (manual) to 2 minutes (automated), saving $150K annually in labor costs"

Congratulations! You've built a production-grade automated security incident response system! 🎉

AWS Proton: Microservices Infrastructure Automation

Daffa Rabbani — Tue, 25 Mar 2025 23:29:06 +0000

Table of contents

What is AWS Proton?
AWS Proton & AWS Cloudformation is same?
How AWS Proton Works?
conclusion

Hey folks 👋,

maybe you just found out that AWS has released one of the new services again in 2020 which even I forgot about the new update :) AWS gave the name of the service called AWS Proton. maybe this proton is similar to the Cloudformation model infrastructure as code.
in this post I will introduce you to AWS Proton and the difference with cloudformation. 🌟

What is AWS Proton? 🤔

✅ AWS Proton is a powerful automation solution that simplifies infrastructure management and application deployment for both administrators and developers.

✅ For administrators, AWS Proton provides a comprehensive framework to create and maintain service templates containing complete application stacks.

✅ For developers, AWS Proton offers a self-service interface where they can easily choose pre-defined service templates that fit their project needs.

What is the difference between AWS Proton & AWS Cloudformation? ⚙

AWS CloudFormation :

✅ A general-purpose infrastructure as code (IaC) service

✅ Allows you to define and provision infrastructure resources using templates

✅ Supports a wide range of AWS resources and can be used for almost any infrastructure setup

✅ Primarily focused on infrastructure provisioning

✅ Works at a lower-level of abstraction, giving developers more granular control

AWS Proton :

✅ A higher-level, more opinionated service specifically for container and serverless applications

✅ Provides standardized, pre-approved templates for entire application stacks

✅ Includes built-in CI/CD pipeline configuration

✅ Focuses on creating a self-service platform for developers

✅ Automates not just infrastructure provisioning, but also deployment pipelines

How AWS Proton Works ? 👨‍💻

When you choose an environment template with AWS Proton, you provide values for the relevant input parameters.
AWS Proton prepares your environment based on the environment template and parameter settings.
When you use AWS Proton to pick a service template, you supply values for the relevant input parameters. You also pick an environment in which to deploy your application or service.
To supply your service, AWS Proton utilizes the service template as well as the values of your service and the environment parameters you have defined.

You choose the input parameters to modify your template for reuse and numerous use cases, applications, or services.

To make this work, you generate environment or service template bundles and submit them to the appropriate registered environment or service templates.

Use Cases

E-Commerce Platform Microservices

Detailed Scenario :
An e-commerce platform with multiple microservices requires a robust, scalable, and secure infrastructure that can handle varying loads and ensure consistent deployment across different services.

Specific Challenges

Infrastructure Inconsistency: Each microservice (Product, Cart, Payment, User) was previously deployed with slight variations in configuration
Security Complexities: Ensuring uniform security protocols across different services
Deployment Overhead: Manual configuration and deployment processes were time-consuming

AWS Proton Implementation

Standardized Templates:
- Create a base template for all e-commerce microservices
- Define consistent networking, security groups, and compute resources
- Include standard monitoring and logging configurations

Deployment Workflow:

Platform team creates a master template for e-commerce services
Developers select and slightly customize the template
- Configure service-specific parameters
- Set scaling requirements
Proton automatically provisions infrastructure
- Consistent network configuration
- Uniform security settings
- Standardized monitoring

Benefits Realized:

40% reduction in deployment time
Improved security consistency
Easier onboarding for new services
Simplified infrastructure management

2. Digital Banking Application

Comprehensive Microservices Ecosystem

A digital banking platform requires extreme reliability, security, and compliance across multiple critical services.

Complex Requirements

Regulatory Compliance: Must meet strict financial industry regulations
High-Security Demands: Protection of sensitive financial data
Performance Critical: Low-latency transactions
Scalability Needs: Handle sudden traffic spikes

AWS Proton Solution

Compliance-Driven Templates:
- Embed security best practices directly in infrastructure templates
- Automatic encryption configurations
- Network isolation between services
- Compliance checkpoints in deployment pipeline

Advanced Implementation:

Create separate environment templates for:
- Development
- Staging
- Production
Implement service-specific security groups
Automate compliance checks during deployment
Enable automatic scaling based on transaction volumes

Key Advantages:

Rapid compliance verification
Consistent security posture
Automated risk mitigation
Faster time-to-market for new financial services

Conclusion 📃

AWS Proton emerges as an innovative cloud infrastructure automation solution that transforms how organizations manage and deploy serverless and container-based applications, providing administrators with a powerful framework for creating standardized templates and developers with a self-service interface to streamline application deployments.

Unlike the more generic CloudFormation, Proton takes a more opinionated approach with features like version management, automatic CI/CD pipeline configuration, and one-click infrastructure updates, allowing businesses to create more efficient, consistent, and scalable cloud deployment strategies.

By bridging the gap between infrastructure management and application development, AWS Proton enables technology teams to optimize development processes, reduce manual configuration overhead, and ensure alignment with organizational best practices, ultimately changing the way modern cloud-native applications are designed, provisioned, and maintained.

Implementing Performance optimization with AWS Aurora PostgreSQL/MySQL for high-throughput workloads 👨‍💻⚙

Daffa Rabbani — Sun, 16 Mar 2025 03:15:54 +0000

Amazon Aurora is AWS’s cloud-native relational database, built for high performance and high availability while staying compatible with MySQL and PostgreSQL. It’s faster and more efficient than standard deployments, but if you’re handling high-throughput workloads, a few extra optimizations can make a big difference.

In this guide, we’ll walk through everything you need to know to fine-tune Aurora for demanding use cases. From identifying performance bottlenecks to optimizing queries, leveraging Aurora’s advanced features, and setting up proper monitoring—this article has got you covered!

🔍 Understanding Aurora’s Architecture

Before jumping into optimizations, let’s take a quick look at what makes Aurora different:

Distributed Storage – Aurora automatically replicates data across multiple Availability Zones (AZs) in an AWS Region. This separation of compute and storage allows for independent scaling and improved resilience.
Fault-Tolerant Storage System – Your data is stored in six copies across three AZs, ensuring durability and availability.
Efficient Write Operations – Aurora handles replication asynchronously, allowing high-throughput performance even for write-heavy applications.

With this foundation in mind, let’s dive into the best ways to optimize your Aurora database for peak performance! 🚀

Key Factors Affecting Performance

Instance Types and Sizing

Aurora performance is significantly influenced by the instance class you choose. For high-throughput workloads, consider:

Memory-optimized instance types (r5, r6g, r6i) for workloads with large working sets
The latest generation instances which typically offer better performance
Instance size appropriate for your workload's CPU and memory requirements

Network Bandwidth

Network bandwidth can become a bottleneck for high-throughput workloads, especially with larger instance types. Ensure you're using instance types with sufficient network bandwidth for your workload requirements.

Hands-On Lab: Setting Up an Optimized Aurora Environment

Let's start with creating an Aurora environment optimized for high throughput.

Step 1: Create an Aurora Cluster with Appropriate Configuration

aws rds create-db-cluster \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --engine-version 15.3 \
    --master-username admin \
    --master-user-password YourStrongPassword123! \
    --db-subnet-group-name your-subnet-group \
    --vpc-security-group-ids sg-0123456789abcdef \
    --backup-retention-period 7 \
    --preferred-backup-window 07:00-09:00 \
    --preferred-maintenance-window sat:10:00-sat:12:00 \
    --db-cluster-parameter-group-name high-throughput-params \
    --storage-encrypted \
    --region us-east-1

Step 2: Create Instance(s) with Appropriate Sizing

aws rds create-db-instance \
    --db-instance-identifier high-throughput-writer \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.4xlarge \
    --region us-east-1

For read scaling:

aws rds create-db-instance \
    --db-instance-identifier high-throughput-reader-1 \
    --db-cluster-identifier high-throughput-demo \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.2xlarge \
    --region us-east-1

Step 3: Configure Parameter Groups

Create a custom parameter group:

aws rds create-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --db-parameter-group-family aurora-postgresql15 \
    --description "Parameters optimized for high throughput" \
    --region us-east-1

Modify parameters for PostgreSQL:

aws rds modify-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --parameters "ParameterName=max_connections,ParameterValue=5000,ApplyMethod=pending-reboot" \
    --region us-east-1

aws rds modify-db-cluster-parameter-group \
    --db-cluster-parameter-group-name high-throughput-params \
    --parameters "ParameterName=shared_buffers,ParameterValue={DBInstanceClassMemory/32768},ApplyMethod=pending-reboot" \
    --region us-east-1

Schema Optimization Techniques

Efficient Table Design

Proper table design is fundamental to database performance. For high-throughput workloads:

Choose appropriate data types (e.g., use INT instead of VARCHAR for numeric values)
Normalize tables appropriately, but consider strategic denormalization for performance-critical queries
Implement table partitioning for large tables

Let's create an example of a partitioned table:

-- For PostgreSQL
CREATE TABLE orders (
    order_id BIGINT NOT NULL,
    customer_id INT NOT NULL,
    order_date DATE NOT NULL,
    total_amount DECIMAL(10,2),
    status VARCHAR(20)
) PARTITION BY RANGE (order_date);

-- Create partitions by month
CREATE TABLE orders_202501 PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');

CREATE TABLE orders_202502 PARTITION OF orders
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

CREATE TABLE orders_202503 PARTITION OF orders
    FOR VALUES FROM ('2025-03-01') TO ('2025-04-01');

Indexing Strategy

Proper indexing is critical for high-throughput workloads:

Create indexes on columns frequently used in WHERE, JOIN, and ORDER BY clauses
Consider composite indexes for queries filtering on multiple columns
Be cautious about over-indexing, as indexes consume space and slow down writes

Example of creating efficient indexes:

-- Primary key
ALTER TABLE orders ADD PRIMARY KEY (order_id, order_date);

-- Index for frequent queries filtering by customer and date range
CREATE INDEX idx_orders_customer_date ON orders (customer_id, order_date);

-- Index for status filtering
CREATE INDEX idx_orders_status ON orders (status);

For high write throughput, consider using INCLUDE in PostgreSQL indexes to reduce the need for index-only scans:

CREATE INDEX idx_orders_date_include_status ON orders (order_date) INCLUDE (status);

Query Optimization

Analyzing and Improving Slow Queries

Let's look at how to identify and optimize slow queries:

Enable Performance Insights to track database load and identify top queries
Use the PostgreSQL/MySQL EXPLAIN command to analyze query execution plans

-- Enable query logging for slow queries (PostgreSQL)
ALTER SYSTEM SET log_min_duration_statement = '1000';  -- Log queries taking more than 1 second
ALTER SYSTEM SET log_statement = 'none';
SELECT pg_reload_conf();

-- Analyze a query using EXPLAIN ANALYZE
EXPLAIN ANALYZE
SELECT o.order_id, o.order_date, o.total_amount, c.customer_name
FROM orders o
JOIN customers c ON o.customer_id = c.customer_id
WHERE o.order_date BETWEEN '2025-01-01' AND '2025-01-31'
AND o.status = 'COMPLETED'
ORDER BY o.total_amount DESC
LIMIT 100;

Query Rewriting Techniques

Some common query rewriting techniques for high-throughput environments:

Use EXISTS instead of IN for checking existence:

-- Instead of:
SELECT * FROM orders WHERE customer_id IN (SELECT customer_id FROM premium_customers);

-- Use:
SELECT * FROM orders o WHERE EXISTS (
    SELECT 1 FROM premium_customers p WHERE p.customer_id = o.customer_id
);

Avoid functions on indexed columns:

-- Instead of:
SELECT * FROM orders WHERE EXTRACT(MONTH FROM order_date) = 3;

-- Use:
SELECT * FROM orders WHERE order_date BETWEEN '2025-03-01' AND '2025-03-31';

Use UNION ALL instead of UNION when duplicates are acceptable:

SELECT * FROM orders_202501 WHERE status = 'COMPLETED'
UNION ALL
SELECT * FROM orders_202502 WHERE status = 'COMPLETED';

Connection Management

Aurora connection management is crucial for high-throughput applications:

Implement connection pooling with tools like PgBouncer, ProxySQL, or AWS RDS Proxy
Monitor and tune max_connections parameter based on workload

Let's set up RDS Proxy:

aws rds create-db-proxy \
    --db-proxy-name aurora-high-throughput-proxy \
    --engine-family POSTGRESQL \
    --auth '{"AuthScheme":"SECRETS","SecretArn":"arn:aws:secretsmanager:us-east-1:123456789012:secret:aurora-credentials","IAMAuth":"DISABLED"}' \
    --role-arn arn:aws:iam::123456789012:role/rds-proxy-role \
    --vpc-subnet-ids subnet-0123456789abcdef0 subnet-0123456789abcdef1 \
    --vpc-security-group-ids sg-0123456789abcdef \
    --require-tls \
    --idle-client-timeout 1800 \
    --debug-logging \
    --region us-east-1

aws rds register-db-proxy-targets \
    --db-proxy-name aurora-high-throughput-proxy \
    --db-cluster-identifiers high-throughput-demo \
    --region us-east-1

Leveraging Aurora-Specific Features

Read Scaling with Aurora Replicas

For high-throughput read workloads, distribute reads across multiple Aurora Replicas:

import psycopg2
import random

# List of reader endpoints
reader_endpoints = [
    "high-throughput-reader-1.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com",
    "high-throughput-reader-2.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com"
]

def get_read_connection():
    # Randomly select a reader endpoint
    endpoint = random.choice(reader_endpoints)

    conn = psycopg2.connect(
        host=endpoint,
        port=5432,
        database="mydatabase",
        user="admin",
        password="YourStrongPassword123!"
    )
    return conn

def get_write_connection():
    # Write connection always goes to the primary instance
    conn = psycopg2.connect(
        host="high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com",
        port=5432,
        database="mydatabase",
        user="admin",
        password="YourStrongPassword123!"
    )
    return conn

Aurora Parallel Query

Enable Parallel Query for analytical workloads:

aws rds modify-db-cluster \
    --db-cluster-identifier high-throughput-demo \
    --enable-http-endpoint \
    --region us-east-1

In PostgreSQL, configure a session to use parallel query:

-- For PostgreSQL
SET aurora.parallel_query = ON;

-- Check if a query uses parallel query
EXPLAIN
SELECT customer_id, SUM(total_amount)
FROM orders
WHERE order_date BETWEEN '2025-01-01' AND '2025-03-31'
GROUP BY customer_id;

Monitoring and Continuous Optimization

Setting Up Comprehensive Monitoring

Enable Enhanced Monitoring and Performance Insights:

aws rds modify-db-instance \
    --db-instance-identifier high-throughput-writer \
    --monitoring-interval 5 \
    --monitoring-role-arn arn:aws:iam::123456789012:role/rds-monitoring-role \
    --enable-performance-insights \
    --performance-insights-retention-period 7 \
    --region us-east-1

Create CloudWatch alarms for critical metrics:

aws cloudwatch put-metric-alarm \
    --alarm-name HighCPUUtilization-Aurora \
    --alarm-description "Alarm when CPU exceeds 80%" \
    --metric-name CPUUtilization \
    --namespace AWS/RDS \
    --statistic Average \
    --period 300 \
    --threshold 80 \
    --comparison-operator GreaterThanThreshold \
    --dimensions Name=DBInstanceIdentifier,Value=high-throughput-writer \
    --evaluation-periods 3 \
    --alarm-actions arn:aws:sns:us-east-1:123456789012:rds-alerts \
    --region us-east-1

Create custom metrics dashboard:

aws cloudwatch put-dashboard \
    --dashboard-name AuroraHighThroughputDashboard \
    --dashboard-body '{
        "widgets": [
            {
                "type": "metric",
                "x": 0,
                "y": 0,
                "width": 12,
                "height": 6,
                "properties": {
                    "metrics": [
                        [ "AWS/RDS", "CPUUtilization", "DBInstanceIdentifier", "high-throughput-writer" ]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": "Writer CPU"
                }
            },
            {
                "type": "metric",
                "x": 12,
                "y": 0,
                "width": 12,
                "height": 6,
                "properties": {
                    "metrics": [
                        [ "AWS/RDS", "DMLLatency", "DBInstanceIdentifier", "high-throughput-writer" ]
                    ],
                    "period": 300,
                    "stat": "Average",
                    "region": "us-east-1",
                    "title": "Write Latency"
                }
            }
        ]
    }' \
    --region us-east-1

Benchmarking Your Aurora Deployment

Use pgbench (PostgreSQL) or sysbench (MySQL) to establish performance baselines:

# For PostgreSQL
pgbench -h high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com \
    -U admin \
    -p 5432 \
    -d mydatabase \
    -c 20 \
    -j 4 \
    -T 60 \
    -P 10

Advanced Optimization Techniques

Memory Optimization

Fine-tune memory-related parameters:

-- For PostgreSQL
ALTER SYSTEM SET work_mem = '64MB';
ALTER SYSTEM SET maintenance_work_mem = '1GB';
ALTER SYSTEM SET effective_cache_size = '24GB';  -- Adjust based on instance memory
SELECT pg_reload_conf();

Vacuum and Analyze Operations

Regular vacuum and analyze operations are critical for PostgreSQL performance:

-- Set up autovacuum parameters
ALTER SYSTEM SET autovacuum_vacuum_scale_factor = 0.05;
ALTER SYSTEM SET autovacuum_analyze_scale_factor = 0.025;
ALTER SYSTEM SET autovacuum_vacuum_cost_limit = 2000;
SELECT pg_reload_conf();

-- Manual VACUUM ANALYZE for immediate effect
VACUUM ANALYZE orders;

Advanced Optimization Techniques

Using Aurora Global Database for Multi-Region Deployments

For global applications requiring low-latency reads across regions:

# Create a global database
aws rds create-global-cluster \
    --global-cluster-identifier global-high-throughput \
    --source-db-cluster-identifier arn:aws:rds:us-east-1:123456789012:cluster:high-throughput-demo \
    --region us-east-1

Add a secondary region:

aws rds create-db-cluster \
    --db-cluster-identifier high-throughput-demo-eu \
    --engine aurora-postgresql \
    --engine-version 15.3 \
    --db-subnet-group-name your-eu-subnet-group \
    --vpc-security-group-ids sg-0123456789abcdef \
    --global-cluster-identifier global-high-throughput \
    --region eu-west-1

Add instances to the secondary region:

aws rds create-db-instance \
    --db-instance-identifier high-throughput-reader-eu \
    --db-cluster-identifier high-throughput-demo-eu \
    --engine aurora-postgresql \
    --db-instance-class db.r6g.2xlarge \
    --region eu-west-1

Optimizing for Specific Workload Types

Write-Intensive Workloads

For applications with heavy write loads:

Batch operations where possible:

-- Instead of individual inserts
INSERT INTO orders (order_id, customer_id, order_date, total_amount, status)
VALUES
    (1001, 101, '2025-03-15', 299.99, 'PENDING'),
    (1002, 102, '2025-03-15', 199.50, 'PENDING'),
    (1003, 103, '2025-03-15', 599.75, 'PENDING'),
    (1004, 104, '2025-03-15', 149.99, 'PENDING'),
    (1005, 105, '2025-03-15', 849.99, 'PENDING');

Reduce index overhead on write-heavy tables:

-- Temporarily disable triggers during bulk loads
ALTER TABLE orders DISABLE TRIGGER ALL;
-- Perform bulk insert
-- Then re-enable triggers
ALTER TABLE orders ENABLE TRIGGER ALL;

Consider UNLOGGED tables (PostgreSQL) for temporary data:

CREATE UNLOGGED TABLE temp_orders (
    order_id BIGINT NOT NULL,
    customer_id INT NOT NULL,
    order_date DATE NOT NULL,
    total_amount DECIMAL(10,2),
    status VARCHAR(20)
);

Read-Intensive Workloads

For read-heavy applications:

Implement materialized views for complex queries:

-- For PostgreSQL
CREATE MATERIALIZED VIEW monthly_sales AS
SELECT
    DATE_TRUNC('month', order_date) AS month,
    customer_id,
    SUM(total_amount) AS monthly_total,
    COUNT(*) AS order_count
FROM orders
GROUP BY DATE_TRUNC('month', order_date), customer_id;

-- Create index on the materialized view
CREATE INDEX idx_monthly_sales_customer ON monthly_sales (customer_id, month);

-- Refresh as needed
REFRESH MATERIALIZED VIEW monthly_sales;

Use Query Caching at the application level:

import redis
import json
import hashlib
import psycopg2

# Initialize Redis client
redis_client = redis.Redis(host='your-redis-host.cache.amazonaws.com', port=6379)

def get_cached_query_results(query, params=None, ttl=300):
    # Create a cache key based on the query and parameters
    cache_key = f"db:query:{hashlib.md5((query + str(params)).encode()).hexdigest()}"

    # Check if results exist in cache
    cached_result = redis_client.get(cache_key)
    if cached_result:
        return json.loads(cached_result)

    # If not in cache, execute query
    conn = get_read_connection()
    cursor = conn.cursor()
    cursor.execute(query, params)
    columns = [desc[0] for desc in cursor.description]
    results = [dict(zip(columns, row)) for row in cursor.fetchall()]
    cursor.close()
    conn.close()

    # Store in cache with TTL
    redis_client.setex(cache_key, ttl, json.dumps(results))
    return results

Lab: Real-Time Performance Testing and Tuning

Let's create a comprehensive testing setup to optimize our Aurora deployment under load.

Step 1: Create Test Data

We'll use pgbench to initialize a test database and then extend it with our custom schema:

# Connect to your Aurora instance
psql "host=high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com port=5432 dbname=postgres user=admin password=YourStrongPassword123!"

# Create test database
CREATE DATABASE perf_test;
\c perf_test

# Initialize with pgbench schema (100 scaling factor ~ 10 million rows)
\! pgbench -i -s 100 -h high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com -p 5432 -U admin perf_test

Step 2: Create Custom Test Schema

-- Create orders table with partitioning
CREATE TABLE orders (
    order_id SERIAL PRIMARY KEY,
    customer_id INT NOT NULL,
    order_date TIMESTAMP NOT NULL,
    total_amount DECIMAL(10,2) NOT NULL,
    status VARCHAR(20) NOT NULL,
    items JSONB,
    shipping_address TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
) PARTITION BY RANGE (order_date);

-- Create partitions
CREATE TABLE orders_q1_2025 PARTITION OF orders
    FOR VALUES FROM ('2025-01-01') TO ('2025-04-01');

CREATE TABLE orders_q2_2025 PARTITION OF orders
    FOR VALUES FROM ('2025-04-01') TO ('2025-07-01');

-- Create appropriate indexes
CREATE INDEX idx_orders_customer_date ON orders (customer_id, order_date);
CREATE INDEX idx_orders_status ON orders (status);
CREATE INDEX idx_orders_items ON orders USING GIN (items);

-- Create customers table
CREATE TABLE customers (
    customer_id SERIAL PRIMARY KEY,
    name VARCHAR(100) NOT NULL,
    email VARCHAR(100) UNIQUE NOT NULL,
    tier VARCHAR(20) NOT NULL,
    signup_date DATE NOT NULL,
    last_order_date TIMESTAMP,
    total_orders INT DEFAULT 0,
    total_spent DECIMAL(12,2) DEFAULT 0
);

CREATE INDEX idx_customers_tier ON customers (tier);
CREATE INDEX idx_customers_signup ON customers (signup_date);

Step 3: Generate Test Data

Create a Python script to generate realistic test data:

import psycopg2
import random
import json
from datetime import datetime, timedelta
import time

# Connect to database
conn = psycopg2.connect(
    host="high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com",
    port=5432,
    database="perf_test",
    user="admin",
    password="YourStrongPassword123!"
)
conn.autocommit = False
cursor = conn.cursor()

# Generate customers
customer_tiers = ["BRONZE", "SILVER", "GOLD", "PLATINUM"]
product_names = ["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard", "Mouse", "Tablet", "Camera", "Printer", "Speaker"]
statuses = ["PENDING", "PROCESSING", "SHIPPED", "DELIVERED", "RETURNED", "CANCELLED"]

# Insert customers
print("Generating customers...")
for i in range(1, 100001):
    name = f"Customer {i}"
    email = f"customer{i}@example.com"
    tier = random.choice(customer_tiers)
    signup_date = datetime(2024, 1, 1) + timedelta(days=random.randint(0, 365))

    cursor.execute(
        "INSERT INTO customers (name, email, tier, signup_date) VALUES (%s, %s, %s, %s)",
        (name, email, tier, signup_date)
    )

    if i % 1000 == 0:
        conn.commit()
        print(f"Inserted {i} customers")

conn.commit()
print("Finished generating customers")

# Insert orders
print("Generating orders...")
start_date = datetime(2025, 1, 1)
end_date = datetime(2025, 6, 30)
days_range = (end_date - start_date).days

for i in range(1, 500001):
    customer_id = random.randint(1, 100000)
    order_date = start_date + timedelta(days=random.randint(0, days_range))

    # Generate between 1 and 5 items for each order
    num_items = random.randint(1, 5)
    items = []
    total_amount = 0

    for _ in range(num_items):
        product = random.choice(product_names)
        price = round(random.uniform(9.99, 999.99), 2)
        quantity = random.randint(1, 3)
        item_total = price * quantity
        total_amount += item_total

        items.append({
            "product": product,
            "price": price,
            "quantity": quantity,
            "item_total": item_total
        })

    status = random.choice(statuses)
    shipping_address = f"{random.randint(100, 999)} Main St, City {random.randint(1, 100)}, State {random.randint(1, 50)}"

    # Insert order
    cursor.execute(
        "INSERT INTO orders (customer_id, order_date, total_amount, status, items, shipping_address) VALUES (%s, %s, %s, %s, %s, %s)",
        (customer_id, order_date, total_amount, status, json.dumps(items), shipping_address)
    )

    # Update customer stats (in real application, consider using triggers for this)
    cursor.execute(
        "UPDATE customers SET total_orders = total_orders + 1, total_spent = total_spent + %s, last_order_date = %s WHERE customer_id = %s",
        (total_amount, order_date, customer_id)
    )

    if i % 1000 == 0:
        conn.commit()
        print(f"Inserted {i} orders")

conn.commit()
print("Finished generating orders")

# Update statistics
print("Analyzing tables...")
cursor.execute("VACUUM ANALYZE customers")
cursor.execute("VACUUM ANALYZE orders")
print("Done!")

cursor.close()
conn.close()

Step 4: Create Test Workload Scripts

Create scripts to simulate real-world read and write patterns:

A. Read_workload.py

# read_workload.py - Simulates high-throughput read queries
import psycopg2
import random
import time
import concurrent.futures
from datetime import datetime, timedelta

# Read connection pool
read_endpoints = [
    "high-throughput-reader-1.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com",
    "high-throughput-reader-2.cluster-ro-abcdefghijkl.us-east-1.rds.amazonaws.com"
]

# Common queries representing typical workload
queries = [
    # Query 1: Get recent orders for a customer
    """
    SELECT o.order_id, o.order_date, o.total_amount, o.status
    FROM orders o
    WHERE o.customer_id = %s
    ORDER BY o.order_date DESC
    LIMIT 10
    """,

    # Query 2: Get orders in a date range with filtering
    """
    SELECT o.order_id, o.customer_id, o.order_date, o.total_amount, o.status
    FROM orders o
    WHERE o.order_date BETWEEN %s AND %s
    AND o.status = %s
    ORDER BY o.total_amount DESC
    LIMIT 100
    """,

    # Query 3: Get summary statistics
    """
    SELECT
        COUNT(*) as order_count,
        SUM(total_amount) as total_revenue,
        AVG(total_amount) as avg_order_value,
        MIN(total_amount) as min_order,
        MAX(total_amount) as max_order
    FROM orders
    WHERE order_date BETWEEN %s AND %s
    """,

    # Query 4: Complex aggregation with join
    """
    SELECT
        c.tier,
        DATE_TRUNC('month', o.order_date) as month,
        COUNT(DISTINCT c.customer_id) as unique_customers,
        COUNT(o.order_id) as order_count,
        SUM(o.total_amount) as total_revenue
    FROM orders o
    JOIN customers c ON o.customer_id = c.customer_id
    WHERE o.order_date BETWEEN %s AND %s
    GROUP BY c.tier, DATE_TRUNC('month', o.order_date)
    ORDER BY month, tier
    """,

    # Query 5: JSONB filtering
    """
    SELECT o.order_id, o.total_amount
    FROM orders o
    WHERE o.items @> '[{"product": %s}]'
    LIMIT 50
    """
]

def get_connection():
    endpoint = random.choice(read_endpoints)
    return psycopg2.connect(
        host=endpoint,
        port=5432,
        database="perf_test",
        user="admin",
        password="YourStrongPassword123!"
    )

def execute_query(query_idx):
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Generate parameters based on query type
        if query_idx == 0:
            # Random customer ID
            params = (random.randint(1, 100000),)
        elif query_idx == 1:
            # Date range and status
            start_date = datetime(2025, 1, 1) + timedelta(days=random.randint(0, 150))
            end_date = start_date + timedelta(days=30)
            status = random.choice(["PENDING", "PROCESSING", "SHIPPED", "DELIVERED"])
            params = (start_date, end_date, status)
        elif query_idx == 2 or query_idx == 3:
            # Date range
            start_date = datetime(2025, 1, 1) + timedelta(days=random.randint(0, 150))
            end_date = start_date + timedelta(days=30)
            params = (start_date, end_date)
        elif query_idx == 4:
            # Product for JSONB query
            params = (random.choice(["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard"]),)

        cursor.execute(queries[query_idx], params)
        cursor.fetchall()  # Actually fetch the results

        end_time = time.time()
        return end_time - start_time
    except Exception as e:
        print(f"Query error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def run_workload(num_queries=1000, max_workers=20):
    print(f"Starting read workload simulation with {num_queries} queries...")
    query_times = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
    futures = []

    for _ in range(num_queries):
        # Select a random query type
        query_idx = random.randint(0, len(queries) - 1)
        futures.append(executor.submit(execute_query, query_idx))

    for future in concurrent.futures.as_completed(futures):
        exec_time = future.result()
        if exec_time is not None:
            query_times.append(exec_time)

# Calculate statistics
if query_times:
    avg_time = sum(query_times) / len(query_times)
    min_time = min(query_times)
    max_time = max(query_times)
    p95_time = sorted(query_times)[int(len(query_times) * 0.95)]

    print(f"Completed {len(query_times)} queries")
    print(f"Average query time: {avg_time:.4f} seconds")
    print(f"Min query time: {min_time:.4f} seconds")
    print(f"Max query time: {max_time:.4f} seconds")
    print(f"95th percentile query time: {p95_time:.4f} seconds")
else:
    print("No successful queries")
if name == "main":
run_workload(num_queries=5000, max_workers=50)

B. Write_workload.py

Create a write workload script:

# write_workload.py - Simulates high-throughput write operations
import psycopg2
import random
import time
import concurrent.futures
from datetime import datetime, timedelta
import json

# Write endpoint (primary instance)
write_endpoint = "high-throughput-demo.cluster-abcdefghijkl.us-east-1.rds.amazonaws.com"

# Product names for generating random orders
product_names = ["Laptop", "Smartphone", "Headphones", "Monitor", "Keyboard", "Mouse", "Tablet", "Camera", "Printer", "Speaker"]
statuses = ["PENDING", "PROCESSING"]

def get_connection():
    return psycopg2.connect(
        host=write_endpoint,
        port=5432,
        database="perf_test",
        user="admin",
        password="YourStrongPassword123!"
    )

def execute_insert():
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Generate a random order
        customer_id = random.randint(1, 100000)
        order_date = datetime.now()

        # Generate between 1 and 5 items
        num_items = random.randint(1, 5)
        items = []
        total_amount = 0

        for _ in range(num_items):
            product = random.choice(product_names)
            price = round(random.uniform(9.99, 999.99), 2)
            quantity = random.randint(1, 3)
            item_total = price * quantity
            total_amount += item_total

            items.append({
                "product": product,
                "price": price,
                "quantity": quantity,
                "item_total": item_total
            })

        status = random.choice(statuses)
        shipping_address = f"{random.randint(100, 999)} Main St, City {random.randint(1, 100)}, State {random.randint(1, 50)}"

        # Insert order
        cursor.execute(
            "INSERT INTO orders (customer_id, order_date, total_amount, status, items, shipping_address) VALUES (%s, %s, %s, %s, %s, %s) RETURNING order_id",
            (customer_id, order_date, total_amount, status, json.dumps(items), shipping_address)
        )
        order_id = cursor.fetchone()[0]

        # Update customer stats
        cursor.execute(
            "UPDATE customers SET total_orders = total_orders + 1, total_spent = total_spent + %s, last_order_date = %s WHERE customer_id = %s",
            (total_amount, order_date, customer_id)
        )

        conn.commit()
        end_time = time.time()

        return {"order_id": order_id, "exec_time": end_time - start_time}
    except Exception as e:
        conn.rollback()
        print(f"Insert error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def execute_update():
    conn = get_connection()
    cursor = conn.cursor()

    try:
        start_time = time.time()

        # Get a random order ID
        cursor.execute("SELECT order_id FROM orders ORDER BY RANDOM() LIMIT 1")
        result = cursor.fetchone()
        if not result:
            return None
        order_id = result[0]

        # Update the order status
        new_status = random.choice(["SHIPPED", "DELIVERED"])
        cursor.execute(
            "UPDATE orders SET status = %s WHERE order_id = %s",
            (new_status, order_id)
        )

        conn.commit()
        end_time = time.time()

        return {"order_id": order_id, "exec_time": end_time - start_time}
    except Exception as e:
        conn.rollback()
        print(f"Update error: {e}")
        return None
    finally:
        cursor.close()
        conn.close()

def run_workload(num_operations=1000, insert_ratio=0.7, max_workers=20):
    print(f"Starting write workload simulation with {num_operations} operations...")
    operation_times = {"insert": [], "update": []}

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = []

        for _ in range(num_operations):
            # Determine operation type based on ratio
            if random.random() < insert_ratio:
                futures.append((executor.submit(execute_insert), "insert"))
            else:
                futures.append((executor.submit(execute_update), "update"))

        for future, op_type in futures:
            result = future.result()
            if result is not None:
                operation_times[op_type].append(result["exec_time"])

    # Calculate statistics for each operation type
    for op_type, times in operation_times.items():
        if times:
            avg_time = sum(times) / len(times)
            min_time = min(times)
            max_time = max(times)
            p95_time = sorted(times)[int(len(times) * 0.95)]

            print(f"\\nCompleted {len(times)} {op_type} operations")
            print(f"Average {op_type} time: {avg_time:.4f} seconds")
            print(f"Min {op_type} time: {min_time:.4f} seconds")
            print(f"Max {op_type} time: {max_time:.4f} seconds")
            print(f"95th percentile {op_type} time: {p95_time:.4f} seconds")
        else:
            print(f"No successful {op_type} operations")

if __name__ == "__main__":
    run_workload(num_operations=2000, insert_ratio=0.7, max_workers=30)

Step 5: Performance Testing and Tuning Process

Now let's establish a systematic approach to performance testing and tuning:

Establish baseline performance:

# Run read workload test
python read_workload.py

# Run write workload test
python write_workload.py

Identify bottlenecks using Performance Insights and CloudWatch Metrics:

Monitor key metrics during the test:

CPU Utilization
Database connections
Buffer cache hit ratio
Read/write IOPS
Read/write latency
Network throughput
Wait events in Performance Insights
Tune parameters based on identified bottlenecks:

For PostgreSQL workloads with high connection rates:

-- Increase max connections
ALTER SYSTEM SET max_connections = '1000';

-- Tune shared_buffers (typically 25% of instance memory)
ALTER SYSTEM SET shared_buffers = '8GB';

-- Tune work_mem for complex queries
ALTER SYSTEM SET work_mem = '64MB';

-- Tune maintenance_work_mem for vacuum operations
ALTER SYSTEM SET maintenance_work_mem = '1GB';

-- Tune effective_cache_size (typically 75% of instance memory)
ALTER SYSTEM SET effective_cache_size = '24GB';

-- Adjust autovacuum parameters for write-heavy workloads
ALTER SYSTEM SET autovacuum_vacuum_scale_factor = 0.05;
ALTER SYSTEM SET autovacuum_analyze_scale_factor = 0.025;
ALTER SYSTEM SET autovacuum_vacuum_cost_limit = 2000;

-- Apply changes
SELECT pg_reload_conf();

After each parameter change, re-run tests and measure impact:

# Run read and write workload tests again
python read_workload.py
python write_workload.py

Document performance improvements:

Create a documentation file showing:

Baseline performance metrics
Identified bottlenecks
Parameter changes made
Performance improvements observed
Recommended final configuration

Best Practices for Ongoing Optimization

Implement a Regular Performance Review Process

Schedule weekly reviews of Performance Insights and CloudWatch metrics.
Set up automated alerting for performance degradation.
Document performance changes when application behavior changes.

Create a Database Parameter Change Management Process

Test all parameter changes in a staging environment first.
Document each parameter change with justification and expected impact.
Implement changes during maintenance windows when possible.
Maintain parameter groups in version control or AWS CloudFormation.

SQL Query Review Process

Establish a code review process for database queries:

Require EXPLAIN ANALYZE for all new queries.
Set performance targets for critical queries.
Maintain a query library of common patterns and anti-patterns.

Conclusion 🚀💡

Optimizing Aurora PostgreSQL/MySQL for high-throughput workloads requires a systematic approach covering instance sizing, schema design, query optimization, and Aurora-specific features. By following the hands-on lab and implementing the best practices outlined in this article, you can achieve significant performance improvements for your most demanding workloads.

Remember that performance optimization is an ongoing process rather than a one-time task. Regular monitoring, testing, and tuning are essential to maintain optimal performance as your workloads evolve. ✨✨