Sync Local Directory to S3 with KMS Encryption & Manifest

{
  "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, ...)