Python Scripting

pip install boto3 schedule

# Option A — AWS CLI (recommended for local use)
aws configure
# AWS Access Key ID:     AKIA...
# AWS Secret Access Key: xxxxxxxx
# Default region:        ap-south-1
# Default output format: json

# Option B — Environment variables (for CI/CD or Docker)
export AWS_ACCESS_KEY_ID=AKIA...
export AWS_SECRET_ACCESS_KEY=xxxxxxxx
export AWS_DEFAULT_REGION=ap-south-1

# Option C — IAM Role (best for EC2/Lambda — no credentials needed)
# Attach an IAM role with the right permissions to the instance running this script

# Tag an instance via AWS CLI
aws ec2 create-tags \
  --resources i-0abc123def456789 \
  --tags Key=AutoStop,Value=true

# Or tag multiple instances at once
aws ec2 create-tags \
  --resources i-0abc123def456789 i-0def456789abc123 \
  --tags Key=AutoStop,Value=true

# Verify the tag
aws ec2 describe-tags \
  --filters "Name=resource-id,Values=i-0abc123def456789"

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "EC2ScheduleControl",
      "Effect": "Allow",
      "Action": [
        "ec2:DescribeInstances",
        "ec2:DescribeTags",
        "ec2:StartInstances",
        "ec2:StopInstances"
      ],
      "Resource": "*"
    }
  ]
}

# ec2_scheduler.py
"""
EC2 Auto Stop/Start Scheduler
Stops instances tagged AutoStop=true at 8 PM.
Starts them at 8 AM.
"""

import boto3
import schedule
import time
import logging
from datetime import datetime
from botocore.exceptions import ClientError, NoCredentialsError

# ── Logging setup ────────────────────────────────────────────────
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(),                    # Print to console
        logging.FileHandler("ec2_scheduler.log"),   # Also write to file
    ],
)
log = logging.getLogger(__name__)

# ── Configuration ─────────────────────────────────────────────────
REGION        = "ap-south-1"   # Change to your AWS region
TAG_KEY       = "AutoStop"     # The tag key we look for
TAG_VALUE     = "true"         # The tag value we look for
STOP_TIME     = "20:00"        # 8 PM — 24-hour format
START_TIME    = "08:00"        # 8 AM — 24-hour format


# ── boto3 client ──────────────────────────────────────────────────
def get_ec2_client():
    """
    Create an EC2 client for the specified region.

    boto3.client() creates a low-level service client.
    - 'ec2'     : the AWS service name
    - region_name: which AWS region to connect to
    Credentials are picked up automatically from:
      1. Environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY)
      2. ~/.aws/credentials file (set by `aws configure`)
      3. IAM role attached to the EC2/Lambda running this script
    """
    return boto3.client("ec2", region_name=REGION)


# ── Helper: find tagged instances ─────────────────────────────────
def get_tagged_instances(ec2, desired_state: str) -> list[dict]:
    """
    Return EC2 instances that have AutoStop=true AND are in the desired_state.

    Parameters
    ----------
    ec2           : boto3 EC2 client
    desired_state : 'running' to find instances to stop,
                    'stopped' to find instances to start

    How describe_instances works
    ----------------------------
    - Filters is a list of dicts with 'Name' and 'Values' keys.
    - 'tag:AutoStop'     matches the tag KEY named AutoStop.
    - 'instance-state-name' filters by the current lifecycle state.
    - The API returns a paginated response — Reservations is the top-level list.
      Each Reservation can contain multiple Instances (from a single launch command).
    """
    try:
        response = ec2.describe_instances(
            Filters=[
                {
                    "Name":   f"tag:{TAG_KEY}",    # Filter by tag key
                    "Values": [TAG_VALUE],          # Tag value must be 'true'
                },
                {
                    "Name":   "instance-state-name",
                    "Values": [desired_state],      # Only running or stopped
                },
            ]
        )
    except NoCredentialsError:
        log.error("AWS credentials not found. Run 'aws configure' or set env vars.")
        return []
    except ClientError as e:
        log.error(f"AWS API error: {e.response['Error']['Message']}")
        return []

    # Flatten the nested Reservations → Instances structure into a flat list
    instances = [
        instance
        for reservation in response["Reservations"]
        for instance in reservation["Instances"]
    ]

    return instances


# ── Helper: get a human-readable instance name ─────────────────────
def get_instance_name(instance: dict) -> str:
    """
    Extract the 'Name' tag value from the instance's Tags list.

    Tags is a list of {'Key': '...', 'Value': '...'} dicts.
    We use next() with a default so it doesn't crash if 'Name' tag is absent.
    """
    tags = instance.get("Tags", [])
    name = next((t["Value"] for t in tags if t["Key"] == "Name"), "Unnamed")
    return name


# ── STOP action ───────────────────────────────────────────────────
def stop_instances():
    """
    Find all running instances tagged AutoStop=true and stop them.
    Called automatically at STOP_TIME (8 PM).
    """
    log.info("=" * 55)
    log.info(f"[STOP JOB] Starting at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    ec2 = get_ec2_client()
    instances = get_tagged_instances(ec2, desired_state="running")

    if not instances:
        log.info("[STOP JOB] No running instances with AutoStop=true found.")
        return

    # Extract just the instance IDs — that's what start/stop APIs need
    instance_ids = [inst["InstanceId"] for inst in instances]

    log.info(f"[STOP JOB] Found {len(instance_ids)} instance(s) to stop:")
    for inst in instances:
        log.info(f"  - {inst['InstanceId']}  ({get_instance_name(inst)})")

    try:
        """
        ec2.stop_instances() sends a stop signal to each instance.
        - InstanceIds: list of instance ID strings (e.g. ['i-0abc...', 'i-0def...'])
        - The API call is asynchronous — it returns immediately.
          The instance transitions: running → stopping → stopped.
        - DryRun=True can be used to test without actually stopping.
        - StopInstances does NOT terminate (delete) the instance.
          Data on the EBS root volume is preserved.
        """
        response = ec2.stop_instances(InstanceIds=instance_ids)

        # Log the new state reported by AWS for each instance
        for item in response["StoppingInstances"]:
            prev  = item["PreviousState"]["Name"]
            curr  = item["CurrentState"]["Name"]
            iid   = item["InstanceId"]
            log.info(f"  ✓ {iid}: {prev} → {curr}")

        log.info(f"[STOP JOB] Stop signal sent to {len(instance_ids)} instance(s).")

    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        error_msg  = e.response["Error"]["Message"]
        log.error(f"[STOP JOB] Failed to stop instances. Code: {error_code} — {error_msg}")
        # Common errors:
        # UnsupportedOperation: instance store-backed instances can't be stopped
        # IncorrectInstanceState: instance is already stopping/terminated


# ── START action ──────────────────────────────────────────────────
def start_instances():
    """
    Find all stopped instances tagged AutoStop=true and start them.
    Called automatically at START_TIME (8 AM).
    """
    log.info("=" * 55)
    log.info(f"[START JOB] Starting at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")

    ec2 = get_ec2_client()
    instances = get_tagged_instances(ec2, desired_state="stopped")

    if not instances:
        log.info("[START JOB] No stopped instances with AutoStop=true found.")
        return

    instance_ids = [inst["InstanceId"] for inst in instances]

    log.info(f"[START JOB] Found {len(instance_ids)} instance(s) to start:")
    for inst in instances:
        log.info(f"  - {inst['InstanceId']}  ({get_instance_name(inst)})")

    try:
        """
        ec2.start_instances() boots stopped EBS-backed instances.
        - InstanceIds: list of instance ID strings.
        - The API call is asynchronous.
          The instance transitions: stopped → pending → running.
        - A new public IP is assigned (unless an Elastic IP is attached).
        - Instance store data is NOT preserved across stop/start cycles.
          EBS data IS preserved.
        """
        response = ec2.start_instances(InstanceIds=instance_ids)

        for item in response["StartingInstances"]:
            prev  = item["PreviousState"]["Name"]
            curr  = item["CurrentState"]["Name"]
            iid   = item["InstanceId"]
            log.info(f"  ✓ {iid}: {prev} → {curr}")

        log.info(f"[START JOB] Start signal sent to {len(instance_ids)} instance(s).")

    except ClientError as e:
        error_code = e.response["Error"]["Code"]
        error_msg  = e.response["Error"]["Message"]
        log.error(f"[START JOB] Failed to start instances. Code: {error_code} — {error_msg}")


# ── Scheduler setup ───────────────────────────────────────────────
def setup_schedule():
    """
    Register stop and start jobs with the schedule library.

    schedule.every().day.at("HH:MM") registers a recurring daily job.
    - Times use 24-hour format.
    - The scheduler runs in the LOCAL timezone of the machine.
      If you need UTC or a specific timezone, use pytz:
          import pytz
          tz = pytz.timezone("Asia/Kolkata")
          now = datetime.now(tz)
    """
    schedule.every().day.at(STOP_TIME).do(stop_instances)
    schedule.every().day.at(START_TIME).do(start_instances)

    log.info(f"Scheduler active — Stop: {STOP_TIME} | Start: {START_TIME}")
    log.info(f"Region: {REGION} | Tag filter: {TAG_KEY}={TAG_VALUE}")
    log.info("Waiting for next scheduled job... (Ctrl+C to exit)")


# ── Entry point ───────────────────────────────────────────────────
if __name__ == "__main__":
    setup_schedule()

    """
    schedule.run_pending() checks if any registered job is due and runs it.
    It does NOT block — it returns immediately if no job is due.

    We wrap it in an infinite loop with time.sleep(60) so we check
    every 60 seconds. This is lightweight (almost no CPU usage while sleeping).

    The loop runs forever until:
    - You press Ctrl+C (raises KeyboardInterrupt)
    - The process is killed (systemd, Docker, etc.)
    """
    try:
        while True:
            schedule.run_pending()
            time.sleep(60)   # Check every minute — fine-grained enough for HH:MM scheduling
    except KeyboardInterrupt:
        log.info("Scheduler stopped by user.")

# Run and watch the logs
python ec2_scheduler.py

# 2025-01-20 07:59:00  INFO     Scheduler active — Stop: 20:00 | Start: 08:00
# 2025-01-20 07:59:00  INFO     Region: ap-south-1 | Tag filter: AutoStop=true
# 2025-01-20 08:00:00  INFO     ═══════════════════════════════════════════════════════
# 2025-01-20 08:00:00  INFO     [START JOB] Starting at 2025-01-20 08:00:00
# 2025-01-20 08:00:00  INFO     Found 3 instance(s) to start:
# 2025-01-20 08:00:00  INFO       - i-0abc123  (dev-api-server)
# 2025-01-20 08:00:00  INFO       - i-0def456  (staging-db)
# 2025-01-20 08:00:00  INFO       - i-0ghi789  (test-worker)
# 2025-01-20 08:00:01  INFO       ✓ i-0abc123: stopped → pending
# 2025-01-20 08:00:01  INFO       ✓ i-0def456: stopped → pending
# 2025-01-20 08:00:01  INFO       ✓ i-0ghi789: stopped → pending

# /etc/systemd/system/ec2-scheduler.service
[Unit]
Description=EC2 Auto Stop/Start Scheduler
After=network.target

[Service]
Type=simple
User=ubuntu
WorkingDirectory=/opt/ec2-scheduler
ExecStart=/opt/ec2-scheduler/venv/bin/python ec2_scheduler.py
Restart=always
RestartSec=10
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target

sudo systemctl daemon-reload
sudo systemctl enable ec2-scheduler
sudo systemctl start ec2-scheduler
sudo systemctl status ec2-scheduler
journalctl -u ec2-scheduler -f   # Follow live logs

# Edit crontab
crontab -e

# Add these two lines:
# Stop at 8 PM (20:00) every day
0 20 * * * /usr/bin/python3 /opt/ec2-scheduler/ec2_scheduler_once.py stop >> /var/log/ec2-stop.log 2>&1

# Start at 8 AM (08:00) every day
0 8  * * * /usr/bin/python3 /opt/ec2-scheduler/ec2_scheduler_once.py start >> /var/log/ec2-start.log 2>&1

# ec2_scheduler_once.py — for cron (runs once, stops/starts, then exits)
import sys
from ec2_scheduler import start_instances, stop_instances

if __name__ == "__main__":
    action = sys.argv[1] if len(sys.argv) > 1 else "stop"
    if action == "stop":
        stop_instances()
    elif action == "start":
        start_instances()

# lambda_handler.py — deploy this as a Lambda function
import boto3
from botocore.exceptions import ClientError

TAG_KEY   = "AutoStop"
TAG_VALUE = "true"
REGION    = "ap-south-1"

def lambda_handler(event, context):
    """
    EventBridge triggers this Lambda on a cron schedule.
    event['action'] is set by the EventBridge rule's Input field.
    """
    action = event.get("action", "stop")
    ec2 = boto3.client("ec2", region_name=REGION)

    state = "running" if action == "stop" else "stopped"
    response = ec2.describe_instances(
        Filters=[
            {"Name": f"tag:{TAG_KEY}", "Values": [TAG_VALUE]},
            {"Name": "instance-state-name", "Values": [state]},
        ]
    )

    instance_ids = [
        inst["InstanceId"]
        for r in response["Reservations"]
        for inst in r["Instances"]
    ]

    if not instance_ids:
        return {"status": "no instances found", "action": action}

    if action == "stop":
        ec2.stop_instances(InstanceIds=instance_ids)
    else:
        ec2.start_instances(InstanceIds=instance_ids)

    return {
        "status": "success",
        "action": action,
        "instances": instance_ids,
        "count": len(instance_ids),
    }

# EventBridge rules (AWS Console or CLI)
# Stop rule — every day at 8 PM UTC
aws events put-rule \
  --name ec2-stop-rule \
  --schedule-expression "cron(0 14 * * ? *)" \
  --state ENABLED

# Start rule — every day at 8 AM UTC
aws events put-rule \
  --name ec2-start-rule \
  --schedule-expression "cron(0 2 * * ? *)" \
  --state ENABLED

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler("ec2_scheduler.log"),
    ],
)

def get_ec2_client():
    return boto3.client("ec2", region_name=REGION)

response = ec2.describe_instances(
    Filters=[
        {"Name": f"tag:{TAG_KEY}", "Values": [TAG_VALUE]},
        {"Name": "instance-state-name", "Values": [desired_state]},
    ]
)

tags = instance.get("Tags", [])
name = next((t["Value"] for t in tags if t["Key"] == "Name"), "Unnamed")

response = ec2.stop_instances(InstanceIds=instance_ids)
for item in response["StoppingInstances"]:
    prev = item["PreviousState"]["Name"]
    curr = item["CurrentState"]["Name"]

response = ec2.start_instances(InstanceIds=instance_ids)
for item in response["StartingInstances"]:
    prev = item["PreviousState"]["Name"]   # "stopped"
    curr = item["CurrentState"]["Name"]    # "pending"

schedule.every().day.at(STOP_TIME).do(stop_instances)
schedule.every().day.at(START_TIME).do(start_instances)

while True:
    schedule.run_pending()
    time.sleep(60)

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Action": [
      "ec2:DescribeInstances",
      "ec2:CreateSnapshot",
      "ec2:DescribeSnapshots",
      "ec2:DeleteSnapshot"
    ],
    "Resource": "*"
  }]
}

import boto3
from datetime import datetime, timezone, timedelta
import logging

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")


class EBSSnapshotManager:
    def __init__(self, region="us-east-1", retention_days=30):
        """
        boto3.client("ec2") creates a low-level EC2 service client.
        region_name targets the correct AWS region.
        retention_days controls how long snapshots are kept.
        """
        self.ec2 = boto3.client("ec2", region_name=region)
        self.retention_days = retention_days

    # ── Step 1: Discover volumes ─────────────────────────────────────
    def get_running_instance_volumes(self) -> list[dict]:
        """
        get_paginator("describe_instances") returns a paginator object.
        Paginators automatically handle the NextToken loop so you never
        miss instances when you have more than 1,000 results.

        paginator.paginate() yields one page dict at a time.
        Each page["Reservations"] is a list of Reservation objects.
        Each Reservation["Instances"] is a list of Instance objects.

        BlockDeviceMappings lists every EBS volume attached to the instance.
        Each mapping has "DeviceName" (e.g., /dev/xvda) and "Ebs.VolumeId".
        """
        volumes = []
        paginator = self.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"]:
                    instance_id = instance["InstanceId"]

                    # Extract the Name tag — default to instance ID if absent
                    instance_name = next(
                        (tag["Value"] for tag in instance.get("Tags", [])
                         if tag["Key"] == "Name"),
                        instance_id,
                    )

                    for mapping in instance.get("BlockDeviceMappings", []):
                        volumes.append({
                            "volume_id":     mapping["Ebs"]["VolumeId"],
                            "instance_id":   instance_id,
                            "instance_name": instance_name,
                            "device":        mapping["DeviceName"],  # e.g. /dev/xvda
                        })
        return volumes

    # ── Step 2: Create snapshots ─────────────────────────────────────
    def create_snapshots(self, volumes: list[dict]) -> list[str]:
        """
        create_snapshot() initiates an async snapshot.
        The snapshot state starts as "pending" and transitions to "completed".

        TagSpecifications lets you tag the snapshot at creation time
        (atomic — avoids a separate create_tags call).

        ResourceType: "snapshot" tells AWS these tags belong to the snapshot,
        not the source volume.
        """
        created = []
        timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d-%H%M")

        for vol in volumes:
            try:
                response = self.ec2.create_snapshot(
                    VolumeId=vol["volume_id"],
                    Description=(
                        f"Auto-backup {vol['instance_name']} "
                        f"{vol['device']} {timestamp}"
                    ),
                    TagSpecifications=[{
                        "ResourceType": "snapshot",
                        "Tags": [
                            {"Key": "Name",       "Value": f"auto-snap-{vol['instance_name']}-{timestamp}"},
                            {"Key": "AutoBackup", "Value": "true"},   # Used by cleanup filter
                            {"Key": "InstanceId", "Value": vol["instance_id"]},
                            {"Key": "CreatedAt",  "Value": timestamp},
                        ],
                    }],
                )
                snapshot_id = response["SnapshotId"]
                created.append(snapshot_id)
                logger.info(
                    f"Created snapshot {snapshot_id} for volume "
                    f"{vol['volume_id']} ({vol['instance_name']} {vol['device']})"
                )
            except Exception as e:
                logger.error(f"Failed to snapshot {vol['volume_id']}: {e}")

        return created

    # ── Step 3: Delete old snapshots ─────────────────────────────────
    def delete_old_snapshots(self) -> list[str]:
        """
        describe_snapshots with OwnerIds=["self"] only returns YOUR snapshots
        (not public ones — important to avoid accidental deletions).

        The tag filter Filters=[{"Name":"tag:AutoBackup","Values":["true"]}]
        ensures we only touch snapshots this script created, never manual ones.

        snapshot["StartTime"] is timezone-aware (UTC) so we compare against
        datetime.now(timezone.utc) — mixing naive/aware datetimes raises TypeError.
        """
        cutoff = datetime.now(timezone.utc) - timedelta(days=self.retention_days)
        deleted = []
        paginator = self.ec2.get_paginator("describe_snapshots")

        for page in paginator.paginate(
            Filters=[{"Name": "tag:AutoBackup", "Values": ["true"]}],
            OwnerIds=["self"],
        ):
            for snapshot in page["Snapshots"]:
                if snapshot["StartTime"] < cutoff:
                    try:
                        self.ec2.delete_snapshot(SnapshotId=snapshot["SnapshotId"])
                        deleted.append(snapshot["SnapshotId"])
                        age_days = (datetime.now(timezone.utc) - snapshot["StartTime"]).days
                        logger.info(
                            f"Deleted snapshot {snapshot['SnapshotId']} "
                            f"(age: {age_days} days)"
                        )
                    except Exception as e:
                        # Snapshot may be in use by an AMI — skip gracefully
                        logger.error(
                            f"Could not delete {snapshot['SnapshotId']}: {e}"
                        )

        return deleted

    # ── Orchestrator ──────────────────────────────────────────────────
    def run(self) -> dict:
        logger.info("Starting EBS snapshot backup cycle...")
        volumes = self.get_running_instance_volumes()
        logger.info(f"Found {len(volumes)} volumes across running instances")

        created = self.create_snapshots(volumes)
        deleted = self.delete_old_snapshots()

        result = {
            "volumes_found":      len(volumes),
            "snapshots_created":  len(created),
            "snapshots_deleted":  len(deleted),
            "created_ids":        created,
            "deleted_ids":        deleted,
        }
        logger.info(f"Backup cycle complete: {result}")
        return result


# ── Lambda entry point ────────────────────────────────────────────
def lambda_handler(event, context):
    """Deploy as Lambda + EventBridge cron(0 1 * * ? *) to run at 1 AM UTC daily."""
    manager = EBSSnapshotManager(retention_days=30)
    return manager.run()


# ── Local / CLI entry point ───────────────────────────────────────
if __name__ == "__main__":
    manager = EBSSnapshotManager(region="ap-south-1", retention_days=30)
    result = manager.run()
    print(result)

# 1. Zip the script
zip ebs_backup.zip ebs_snapshot_manager.py

# 2. Create Lambda function
aws lambda create-function \
  --function-name EBSSnapshotManager \
  --runtime python3.12 \
  --role arn:aws:iam::123456789012:role/LambdaEC2BackupRole \
  --handler ebs_snapshot_manager.lambda_handler \
  --zip-file fileb://ebs_backup.zip \
  --timeout 300

# 3. Add daily EventBridge trigger (1 AM UTC)
aws events put-rule \
  --name DailyEBSBackup \
  --schedule-expression "cron(0 1 * * ? *)" \
  --state ENABLED

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")

def __init__(self, region="us-east-1", retention_days=30):
    self.ec2 = boto3.client("ec2", region_name=region)
    self.retention_days = retention_days

paginator = self.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"]:
        instance_id = instance["InstanceId"]

instance_name = next(
    (tag["Value"] for tag in instance.get("Tags", []) if tag["Key"] == "Name"),
    instance_id,
)

for mapping in instance.get("BlockDeviceMappings", []):
    volumes.append({
        "volume_id": mapping["Ebs"]["VolumeId"],
        "device":    mapping["DeviceName"],
    })

response = self.ec2.create_snapshot(
    VolumeId=vol["volume_id"],
    Description=f"Auto-backup {vol['instance_name']} {vol['device']} {timestamp}",
    TagSpecifications=[{
        "ResourceType": "snapshot",
        "Tags": [
            {"Key": "AutoBackup", "Value": "true"},
            ...
        ],
    }],
)

cutoff = datetime.now(timezone.utc) - timedelta(days=self.retention_days)

paginator = self.ec2.get_paginator("describe_snapshots")
for page in paginator.paginate(
    Filters=[{"Name": "tag:AutoBackup", "Values": ["true"]}],
    OwnerIds=["self"],
):

if snapshot["StartTime"] < cutoff:
    self.ec2.delete_snapshot(SnapshotId=snapshot["SnapshotId"])

volumes = self.get_running_instance_volumes()
created = self.create_snapshots(volumes)
deleted = self.delete_old_snapshots()

def lambda_handler(event, context):
    manager = EBSSnapshotManager(retention_days=30)
    return manager.run()

if __name__ == "__main__":
    manager = EBSSnapshotManager(region="ap-south-1", retention_days=30)

{
  "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],
    }

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": ["s3:PutObject", "s3:GetObject", "s3:HeadObject"],
      "Resource": "arn:aws:s3:::my-company-data/*"
    },
    {
      "Effect": "Allow",
      "Action": ["kms:GenerateDataKey", "kms:Decrypt"],
      "Resource": "arn:aws:kms:us-east-1:123456789012:key/abc-123"
    }
  ]
}

import boto3
import os
import hashlib
import json
from pathlib import Path
from datetime import datetime
from botocore.exceptions import ClientError


class S3Syncer:
    def __init__(self, bucket: str, prefix: str, kms_key_id: str, region: str = "us-east-1"):
        """
        boto3.client("s3") — low-level S3 client used for fine-grained control.
        We choose client (not resource) here because we need ExtraArgs
        for KMS encryption in upload_file().

        kms_key_id can be:
          - Key ID:  "abc-123-..."
          - Key ARN: "arn:aws:kms:us-east-1:123456789012:key/abc-123"
          - Alias:   "alias/my-data-key"
        """
        self.s3 = boto3.client("s3", region_name=region)
        self.bucket = bucket
        self.prefix = prefix.rstrip("/")
        self.kms_key_id = kms_key_id
        self.manifest: list[dict] = []

    # ── MD5 hash ──────────────────────────────────────────────────
    def get_file_md5(self, filepath: Path) -> str:
        """
        Compute the MD5 hash of a local file in 8 KB chunks.
        Reading in chunks prevents loading multi-GB files into RAM.
        iter(lambda: f.read(8192), b"") calls the lambda repeatedly
        until it returns the sentinel value b"" (empty bytes = EOF).
        """
        md5 = hashlib.md5()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                md5.update(chunk)
        return md5.hexdigest()

    # ── Deduplication check ───────────────────────────────────────
    def file_exists_in_s3(self, s3_key: str, local_md5: str) -> bool:
        """
        head_object() fetches only the metadata of an S3 object (no body).
        It's cheap (~1 ms) compared to downloading the file for comparison.

        S3 ETag for non-multipart uploads equals the MD5 of the object.
        For multipart uploads the ETag is different — this check works
        for files < 5 GB (single-part) which covers most use cases.

        ClientError with code "404" means the object doesn't exist yet.
        Any other error (403 Forbidden, etc.) is re-raised.
        """
        try:
            response = self.s3.head_object(Bucket=self.bucket, Key=s3_key)
            s3_etag = response.get("ETag", "").strip('"')
            return s3_etag == local_md5
        except ClientError as e:
            if e.response["Error"]["Code"] == "404":
                return False
            raise

    # ── Single file upload ────────────────────────────────────────
    def upload_file(self, local_path: Path, s3_key: str) -> str:
        """
        upload_file() streams the file from disk to S3 using multipart
        upload automatically for files > 8 MB. ExtraArgs are passed
        through to the underlying PutObject / CreateMultipartUpload API.

        ServerSideEncryption: "aws:kms" instructs S3 to encrypt with KMS.
        SSEKMSKeyId:           which KMS key to use (defaults to the S3
                               service key if omitted).
        Metadata:              arbitrary key-value pairs stored alongside
                               the object — useful for integrity checks
                               without downloading the file.
        """
        local_md5 = self.get_file_md5(local_path)

        if self.file_exists_in_s3(s3_key, local_md5):
            print(f"  [SKIP]   {s3_key} (unchanged)")
            return "skipped"

        self.s3.upload_file(
            Filename=str(local_path),
            Bucket=self.bucket,
            Key=s3_key,
            ExtraArgs={
                "ServerSideEncryption": "aws:kms",
                "SSEKMSKeyId":          self.kms_key_id,
                "Metadata":             {"md5checksum": local_md5},
            },
        )
        print(f"  [UPLOAD] {s3_key}  ({local_path.stat().st_size:,} bytes)")
        return "uploaded"

    # ── Directory sync ────────────────────────────────────────────
    def sync_directory(self, local_dir: str) -> dict:
        """
        Path.rglob("*") recursively yields all files and directories.
        is_file() filters out directories.
        relative_to() strips the local base path so we get just the
        relative path for the S3 key.
        .replace("\\\\", "/") normalises Windows backslashes to forward
        slashes for S3 keys.
        """
        local_dir_path = Path(local_dir)
        stats = {"uploaded": 0, "skipped": 0, "failed": 0}

        for file_path in local_dir_path.rglob("*"):
            if not file_path.is_file():
                continue

            relative = file_path.relative_to(local_dir_path)
            s3_key = f"{self.prefix}/{relative}".replace("\\", "/")

            try:
                result = self.upload_file(file_path, s3_key)
                stats[result] += 1
                self.manifest.append({
                    "local_path": str(file_path),
                    "s3_key":     s3_key,
                    "s3_uri":     f"s3://{self.bucket}/{s3_key}",
                    "size_bytes": file_path.stat().st_size,
                    "status":     result,
                    "timestamp":  datetime.utcnow().isoformat() + "Z",
                })
            except Exception as e:
                print(f"  [ERROR]  {file_path}: {e}")
                stats["failed"] += 1
                self.manifest.append({
                    "local_path": str(file_path),
                    "s3_key":     s3_key,
                    "status":     "failed",
                    "error":      str(e),
                    "timestamp":  datetime.utcnow().isoformat() + "Z",
                })

        return stats

    # ── Save manifest ─────────────────────────────────────────────
    def save_manifest(self, local_output: str = "manifest.json") -> None:
        """
        Write the manifest locally and also upload it to S3
        so there's a persistent audit trail of every sync run.
        The timestamp in the key prevents overwriting previous manifests.
        """
        with open(local_output, "w") as f:
            json.dump(self.manifest, f, indent=2)

        manifest_key = (
            f"{self.prefix}/_manifests/"
            f"manifest_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
        )
        self.s3.upload_file(
            Filename=local_output,
            Bucket=self.bucket,
            Key=manifest_key,
            ExtraArgs={
                "ServerSideEncryption": "aws:kms",
                "SSEKMSKeyId":          self.kms_key_id,
            },
        )
        print(f"\nManifest uploaded → s3://{self.bucket}/{manifest_key}")


# ── Entry point ───────────────────────────────────────────────────
if __name__ == "__main__":
    syncer = S3Syncer(
        bucket="my-company-data",
        prefix="backups/app-data",
        kms_key_id="arn:aws:kms:us-east-1:123456789012:key/abc-123",
        region="us-east-1",
    )

    stats = syncer.sync_directory("/opt/app/data")
    syncer.save_manifest()

    print(f"\nSync complete:")
    print(f"  Uploaded: {stats['uploaded']}")
    print(f"  Skipped:  {stats['skipped']}  (unchanged)")
    print(f"  Failed:   {stats['failed']}")

[
  {
    "local_path": "/opt/app/data/report-2025-01.csv",
    "s3_key": "backups/app-data/report-2025-01.csv",
    "s3_uri": "s3://my-company-data/backups/app-data/report-2025-01.csv",
    "size_bytes": 204800,
    "status": "uploaded",
    "timestamp": "2025-01-20T10:30:00Z"
  },
  {
    "local_path": "/opt/app/data/config.json",
    "s3_key": "backups/app-data/config.json",
    "s3_uri": "s3://my-company-data/backups/app-data/config.json",
    "size_bytes": 1024,
    "status": "skipped",
    "timestamp": "2025-01-20T10:30:01Z"
  }
]

def __init__(self, bucket: str, prefix: str, kms_key_id: str, region: str = "us-east-1"):
    self.s3 = boto3.client("s3", region_name=region)
    self.bucket = bucket
    self.prefix = prefix.rstrip("/")
    self.kms_key_id = kms_key_id
    self.manifest: list[dict] = []

md5 = hashlib.md5()
with open(filepath, "rb") as f:
    for chunk in iter(lambda: f.read(8192), b""):
        md5.update(chunk)
return md5.hexdigest()

response = self.s3.head_object(Bucket=self.bucket, Key=s3_key)
s3_etag = response.get("ETag", "").strip('"')
return s3_etag == local_md5

self.s3.upload_file(
    Filename=str(local_path),
    Bucket=self.bucket,
    Key=s3_key,
    ExtraArgs={
        "ServerSideEncryption": "aws:kms",
        "SSEKMSKeyId": self.kms_key_id,
        "Metadata": {"md5checksum": local_md5},
    },
)

local_dir_path = Path(local_dir)
for file_path in local_dir_path.rglob("*"):
    if not file_path.is_file():
        continue
    relative = file_path.relative_to(local_dir_path)
    s3_key = f"{self.prefix}/{relative}".replace("\\", "/")

with open(local_output, "w") as f:
    json.dump(self.manifest, f, indent=2)

manifest_key = (
    f"{self.prefix}/_manifests/manifest_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.json"
)
self.s3.upload_file(Filename=local_output, Bucket=self.bucket, Key=manifest_key, ...)

import boto3
import json
from botocore.exceptions import ClientError


def audit_s3_buckets(region: str = "us-east-1") -> list[dict]:
    """
    s3.list_buckets() returns ALL buckets in the account (globally scoped).
    Unlike most AWS services, S3 buckets don't require a region filter —
    list_buckets() always returns everything regardless of client region.

    We then make per-bucket API calls for each compliance check.
    """
    s3 = boto3.client("s3", region_name=region)
    report = []

    buckets = s3.list_buckets()["Buckets"]
    print(f"Auditing {len(buckets)} S3 buckets...\n")

    for bucket in buckets:
        name = bucket["Name"]
        issues: list[str] = []

        # ── Check 1: Public Access Block ─────────────────────────
        # get_public_access_block() returns four boolean settings:
        # - BlockPublicAcls:       Block new public ACLs
        # - IgnorePublicAcls:      Ignore existing public ACLs
        # - BlockPublicPolicy:     Block new public bucket policies
        # - RestrictPublicBuckets: Restrict access for public-policy buckets
        # ALL four must be True for the bucket to be fully protected.
        try:
            pab = s3.get_public_access_block(Bucket=name)[
                "PublicAccessBlockConfiguration"
            ]
            if not all([
                pab.get("BlockPublicAcls"),
                pab.get("IgnorePublicAcls"),
                pab.get("BlockPublicPolicy"),
                pab.get("RestrictPublicBuckets"),
            ]):
                issues.append("PUBLIC_ACCESS_NOT_FULLY_BLOCKED")
        except ClientError as e:
            if "NoSuchPublicAccessBlockConfiguration" in str(e):
                issues.append("NO_PUBLIC_ACCESS_BLOCK")
            else:
                issues.append(f"PAB_CHECK_ERROR")

        # ── Check 2: Bucket Policy — Public Principal ─────────────
        # get_bucket_policy() returns the policy as a JSON string.
        # We parse it and look for statements where:
        #   Effect = "Allow"  AND  Principal = "*" or {"AWS": "*"}
        # These grant public (anonymous) access to the bucket.
        try:
            raw_policy = s3.get_bucket_policy(Bucket=name)["Policy"]
            policy = json.loads(raw_policy)
            for stmt in policy.get("Statement", []):
                principal = stmt.get("Principal")
                is_public = principal in ("*", {"AWS": "*"})
                if stmt.get("Effect") == "Allow" and is_public:
                    issues.append("PUBLIC_BUCKET_POLICY")
                    break   # One public statement is enough to flag it
        except ClientError as e:
            if "NoSuchBucketPolicy" not in str(e):
                # Real error (e.g., AccessDenied) — note it
                issues.append("POLICY_CHECK_ERROR")
            # NoSuchBucketPolicy means no policy exists — not necessarily bad

        # ── Check 3: Versioning ───────────────────────────────────
        # get_bucket_versioning() can return:
        #   {"Status": "Enabled"}   — versioning on
        #   {"Status": "Suspended"} — was on, now paused
        #   {}                      — never enabled
        # Only "Enabled" is compliant.
        try:
            versioning = s3.get_bucket_versioning(Bucket=name)
            if versioning.get("Status") != "Enabled":
                issues.append("VERSIONING_DISABLED")
        except ClientError:
            issues.append("VERSIONING_CHECK_FAILED")

        # ── Check 4: Default Encryption ───────────────────────────
        # get_bucket_encryption() raises ServerSideEncryptionConfigurationNotFoundError
        # if no default encryption is configured.
        # Valid algorithms: "AES256" (SSE-S3) or "aws:kms" (SSE-KMS).
        try:
            enc = s3.get_bucket_encryption(Bucket=name)
            rules = enc["ServerSideEncryptionConfiguration"]["Rules"]
            algo = rules[0]["ApplyServerSideEncryptionByDefault"]["SSEAlgorithm"]
            if algo not in ("aws:kms", "AES256"):
                issues.append("WEAK_ENCRYPTION")
        except ClientError as e:
            if "ServerSideEncryptionConfigurationNotFoundError" in str(e):
                issues.append("ENCRYPTION_DISABLED")

        # ── Check 5: Access Logging ───────────────────────────────
        # get_bucket_logging() returns {"LoggingEnabled": {...}} if logging
        # is configured, or just {} if disabled.
        # Logging is essential for security investigations and audits.
        try:
            logging_cfg = s3.get_bucket_logging(Bucket=name)
            if "LoggingEnabled" not in logging_cfg:
                issues.append("ACCESS_LOGGING_DISABLED")
        except ClientError:
            issues.append("LOGGING_CHECK_FAILED")

        # ── Record result ─────────────────────────────────────────
        compliant = len(issues) == 0
        report.append({
            "bucket":    name,
            "issues":    issues,
            "compliant": compliant,
        })

        status = "✅" if compliant else "❌"
        issue_str = ", ".join(issues) if issues else "All checks passed"
        print(f"{status} {name}: {issue_str}")

    # ── Summary ───────────────────────────────────────────────────
    non_compliant = [r for r in report if not r["compliant"]]
    print(f"\n{'='*60}")
    print(f"Total Buckets:    {len(report)}")
    print(f"Compliant:        {len(report) - len(non_compliant)}")
    print(f"Non-Compliant:    {len(non_compliant)}")

    if non_compliant:
        print("\nTop Issues:")
        from collections import Counter
        all_issues = [i for r in non_compliant for i in r["issues"]]
        for issue, count in Counter(all_issues).most_common():
            print(f"  {issue}: {count} bucket(s)")

    return report


def save_report(report: list[dict], output: str = "s3_audit_report.json") -> None:
    with open(output, "w") as f:
        json.dump(report, f, indent=2)
    print(f"\nFull report saved to {output}")


def auto_fix_public_access(bucket_name: str, dry_run: bool = True) -> None:
    """
    put_public_access_block() with all four settings = True
    is the AWS-recommended way to make a bucket fully private.
    This is a non-destructive operation — it doesn't delete any objects.
    """
    s3 = boto3.client("s3")
    if dry_run:
        print(f"[DRY-RUN] Would enable Public Access Block on: {bucket_name}")
        return

    s3.put_public_access_block(
        Bucket=bucket_name,
        PublicAccessBlockConfiguration={
            "BlockPublicAcls":       True,
            "IgnorePublicAcls":      True,
            "BlockPublicPolicy":     True,
            "RestrictPublicBuckets": True,
        },
    )
    print(f"✅ Public Access Block enabled on: {bucket_name}")


if __name__ == "__main__":
    report = audit_s3_buckets()
    save_report(report)

    # Auto-fix any bucket with PUBLIC_ACCESS_NOT_FULLY_BLOCKED
    for r in report:
        if "PUBLIC_ACCESS_NOT_FULLY_BLOCKED" in r["issues"]:
            auto_fix_public_access(r["bucket"], dry_run=True)   # Set dry_run=False to apply