Find Untagged Long-Running EC2 Instances & Alert via SES

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["ec2:DescribeInstances"],
      "Resource": "*"
    },
    {
      "Effect": "Allow",
      "Action": ["ses:SendEmail"],
      "Resource": "arn:aws:ses:us-east-1:123456789012:identity/[email protected]"
    }
  ]
}

import boto3
from datetime import datetime, timezone, timedelta


def find_untagged_long_running_instances(region: str = "us-east-1") -> list[dict]:
    """
    Returns a list of violation dicts for instances that have been
    running > 7 days without a 'Project' tag.
    """
    ec2 = boto3.client("ec2", region_name=region)
    ses = boto3.client("ses", region_name=region)

    # datetime.now(timezone.utc) returns a timezone-aware UTC datetime.
    # EC2 instance["LaunchTime"] is also timezone-aware (UTC).
    # Both must be timezone-aware to subtract them without a TypeError.
    threshold = datetime.now(timezone.utc) - timedelta(days=7)
    violations = []

    # get_paginator ensures we get ALL instances even if there are > 1000
    paginator = ec2.get_paginator("describe_instances")
    for page in paginator.paginate(
        Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
    ):
        for reservation in page["Reservations"]:
            for instance in reservation["Instances"]:
                launch_time = instance["LaunchTime"]   # timezone-aware UTC datetime

                # Convert tags list to a dict for O(1) lookup
                # instance.get("Tags", []) returns [] if no tags — avoids KeyError
                tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}

                # Check BOTH conditions: old enough AND missing required tag
                if launch_time < threshold and "Project" not in tags:
                    violations.append({
                        "InstanceId":    instance["InstanceId"],
                        "LaunchTime":    launch_time.strftime("%Y-%m-%d %H:%M UTC"),
                        "InstanceType":  instance["InstanceType"],
                        "Name":          tags.get("Name", "Unnamed"),
                        "Owner":         tags.get("Owner", "Unknown"),
                        "PrivateIp":     instance.get("PrivateIpAddress", "N/A"),
                        "RunningDays":   (datetime.now(timezone.utc) - launch_time).days,
                    })

    if violations:
        send_violation_email(ses, violations)
    else:
        print("All running instances are compliant.")

    return violations


def send_violation_email(ses_client, violations: list[dict]) -> None:
    """
    ses_client.send_email() requires:
    - Source: a verified sender email address or domain
    - Destination.ToAddresses: list of recipient emails
    - Message.Subject.Data: email subject string
    - Message.Body.Text.Data: plain-text body
      (Use Body.Html.Data for HTML-formatted emails)

    SES is a regional service — the client must match the region
    where your email identities are verified.
    """
    # Build a human-readable table for the email body
    rows = "\n".join([
        (f"  [{i+1}] {v['InstanceId']} | {v['Name']} | "
         f"{v['InstanceType']} | Running {v['RunningDays']} days | "
         f"Owner: {v['Owner']} | IP: {v['PrivateIp']}")
        for i, v in enumerate(violations)
    ])

    body = f"""
AWS Compliance Alert: Untagged Long-Running EC2 Instances
==========================================================

The following {len(violations)} EC2 instance(s) have been running for more
than 7 days without a required 'Project' tag:

{rows}

Required Action:
  1. Add the 'Project' tag to identify the billing owner.
  2. Terminate the instance if it is no longer needed.
  3. Instances without a Project tag will be stopped automatically
     after 14 days (policy enforcement pending).

This is an automated message from the Cloud Governance bot.
Run script: governance/find_untagged_instances.py
    """

    ses_client.send_email(
        Source="[email protected]",          # Must be SES-verified
        Destination={
            "ToAddresses": ["[email protected]"],
            "CcAddresses": ["[email protected]"],
        },
        Message={
            "Subject": {
                "Data": f"⚠️ {len(violations)} Untagged EC2 Instance(s) Detected",
                "Charset": "UTF-8",
            },
            "Body": {
                "Text": {
                    "Data": body,
                    "Charset": "UTF-8",
                }
            },
        },
        # Optional: ReplyToAddresses, ReturnPath, Tags for SES tracking
    )
    print(f"Alert email sent for {len(violations)} violation(s).")


# ── Lambda entry point (trigger via EventBridge daily) ────────────
def lambda_handler(event, context):
    violations = find_untagged_long_running_instances(region="us-east-1")
    return {
        "statusCode": 200,
        "violations_found": len(violations),
        "instances": [v["InstanceId"] for v in violations],
    }


# ── Local run ─────────────────────────────────────────────────────
if __name__ == "__main__":
    result = find_untagged_long_running_instances(region="ap-south-1")
    print(f"\nTotal violations: {len(result)}")
    for v in result:
        print(f"  - {v['InstanceId']} ({v['Name']}) running {v['RunningDays']} days")

def enforce_stop_policy(violations: list[dict], stop_after_days: int = 14):
    """
    Stop (not terminate) instances that have been running more than
    stop_after_days without a Project tag.
    Termination should require a human decision.
    """
    ec2 = boto3.client("ec2")
    to_stop = [v for v in violations if v["RunningDays"] >= stop_after_days]

    if not to_stop:
        return

    ids = [v["InstanceId"] for v in to_stop]
    ec2.stop_instances(InstanceIds=ids)
    print(f"Auto-stopped {len(ids)} instances: {ids}")

ec2 = boto3.client("ec2", region_name=region)
ses = boto3.client("ses", region_name=region)

threshold = datetime.now(timezone.utc) - timedelta(days=7)

paginator = ec2.get_paginator("describe_instances")
for page in paginator.paginate(
    Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
):

for reservation in page["Reservations"]:
    for instance in reservation["Instances"]:
        launch_time = instance["LaunchTime"]

tags = {t["Key"]: t["Value"] for t in instance.get("Tags", [])}

if launch_time < threshold and "Project" not in tags:
    violations.append({...})

"RunningDays": (datetime.now(timezone.utc) - launch_time).days,

rows = "\n".join([
    f"  [{i+1}] {v['InstanceId']} | {v['Name']} | ..."
    for i, v in enumerate(violations)
])

ses_client.send_email(
    Source="[email protected]",
    Destination={
        "ToAddresses": ["[email protected]"],
        "CcAddresses": ["[email protected]"],
    },
    Message={
        "Subject": {"Data": f"⚠️ {len(violations)} Untagged EC2 Instance(s)", "Charset": "UTF-8"},
        "Body": {"Text": {"Data": body, "Charset": "UTF-8"}},
    },
)

def lambda_handler(event, context):
    violations = find_untagged_long_running_instances(region="us-east-1")
    return {
        "statusCode": 200,
        "violations_found": len(violations),
        "instances": [v["InstanceId"] for v in violations],
    }