CloudWatchDog is a zero-trust security engine that detects and auto-remediates dangerous AWS security-group modifications in real time — before the attack window opens.
Opening port 22 or 3389 to 0.0.0.0/0 exposes infrastructure to the entire internet instantly.
Manual audits and periodic scans create a dangerous window where attackers can exploit misconfigurations.
Real-time detection and auto-remediation close the exposure window to under 5 seconds.
CloudTrail continuously logs every AuthorizeSecurityGroupIngress API call across the account.
EventBridge rule isolates security-group ingress modifications from all other API traffic.
Lambda parses ipPermissions for port 22/3389 exposure to public CIDRs (0.0.0.0/0, ::/0).
Non-compliant rules are immediately revoked via ec2:RevokeSecurityGroupIngress.
Markdown-formatted Slack alert with user identity, security group ID, and remediation status.
EventBridge triggers within seconds of any security-group modification.
Revokes only the non-compliant rule — all other rules remain untouched.
Detects both IPv4 (0.0.0.0/0) and IPv6 (::/0) public CIDR violations.
Catches broad ranges like 1-1024 that overlap with SSH/RDP ports.
Lambda role scoped to exactly 3 permissions — no wildcards, no admin.
Full pytest suite with zero AWS credentials required for local testing.
def find_noncompliant_rules(ip_permissions):
"""Identify ingress rules exposing dangerous ports."""
violations = []
for perm in ip_permissions:
from_port = perm.get("fromPort", 0)
to_port = perm.get("toPort", 0)
# Check SSH (22) or RDP (3389) overlap
port_overlap = any(
from_port <= p <= to_port
for p in {22, 3389}
)
if not port_overlap: continue
# Find public CIDR ranges
public_ipv4 = [
r for r in perm.get("ipRanges", {}).get("items", [])
if r.get("cidrIp") == "0.0.0.0/0"
]
if public_ipv4:
violations.append(build_revocation(perm))
return violations
def revoke_rules(ec2_client, group_id, violations):
"""Revoke non-compliant ingress rules via EC2 API."""
for rule in violations:
ec2_client.revoke_security_group_ingress(
GroupId=group_id,
IpPermissions=[rule],
)
return [{"status": "REVOKED"} for _ in violations]resource "aws_cloudwatch_event_rule" "sg_monitor" {
name = "cloudwatchdog-sg-monitor"
event_pattern = jsonencode({
source = ["aws.ec2"]
detail-type = ["AWS API Call via CloudTrail"]
detail = {
eventSource = ["ec2.amazonaws.com"]
eventName = ["AuthorizeSecurityGroupIngress"]
}
})
}
resource "aws_iam_role_policy" "remediator" {
# Principle of Least Privilege: only 3 actions
policy = jsonencode({
Statement = [{
Action = [
"ec2:RevokeSecurityGroupIngress",
"ec2:DescribeSecurityGroups",
]
Resource = "*"
Condition = {
StringEquals = {
"aws:RequestedRegion" = var.aws_region
}
}
}]
})
}class TestLambdaHandler:
"""End-to-end handler tests with mocked AWS."""
def test_remediates_malicious_ssh(self, event, mock_ec2):
result = lambda_handler(event, ec2_client=mock_ec2)
assert result["action"] == "REMEDIATED"
assert result["violations_found"] == 1
mock_ec2.revoke_security_group_ingress \
.assert_called_once()
def test_ignores_safe_event(self, safe_event, mock_ec2):
result = lambda_handler(safe_event, ec2_client=mock_ec2)
assert result["action"] == "COMPLIANT"
mock_ec2.revoke_security_group_ingress \
.assert_not_called()
# Result: 27 passed, 0 failed, 0 skippedCloudWatchDog/
├── terraform/
│ ├── main.tf # IAM, Lambda, EventBridge rule & target
│ ├── variables.tf # 9 configurable inputs with validation
│ └── outputs.tf # Resource ARNs and identifiers
├── src/
│ └── lambda_remediator.py # Core remediation engine (Boto3)
├── tests/
│ ├── mock_events.json # 3 realistic CloudTrail event fixtures
│ └── test_remediator.py # 27 unit & integration tests
└── README.md # Architecture docs & deployment guide