Creating a Pre-Signed URL Using a Lambda Function

pre-signed-url-python-lambda.py

import json
import boto3
from botocore.client import Config
import os
from datetime import datetime
from datetime import timedelta
import logging

# Configure simple logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# Constants
S3_CONFIG = Config(signature_version="s3v4")

def format_duration(seconds):
    """Convert seconds to a human-readable duration."""
    if seconds < 60:
        return f"{seconds} seconds"
    elif seconds < 3600:
        return f"{seconds // 60} minutes"
    elif seconds < 86400:
        return f"{seconds // 3600} hours"
    else:
        days = seconds // 86400
        return f"{days} days"

def lambda_handler(event, context):
    try:
        # Environment variables
        env_vars = {
            "access_key": os.environ["YOUR_USER_ACCESS_KEY"],
            "secret_key": os.environ["YOUR_USER_SECRET_KEY"],
            "bucket_name": os.environ["YOUR_USER_BUCKET_NAME"],
            "object_name": os.environ["YOUR_USER_OBJECT_KEY_NAME"],
            "expiration_seconds": int(os.environ["EXPIRATION_TIME"])
        }

        current_time = datetime.utcnow()
        json_filename = f"presigned_url_{current_time.strftime('%Y%m%d%H%M%S')}.json"

        # Initialize S3 client with specific configuration
        client = boto3.client(
            "s3",
            config=S3_CONFIG,
            aws_access_key_id=env_vars["access_key"],
            aws_secret_access_key=env_vars["secret_key"],
            region_name='us-east-1'  # Specify the region
        )

        # Generate presigned URL for the original object
        object_presigned_url = client.generate_presigned_url(
            "get_object",
            Params={"Bucket": env_vars["bucket_name"], "Key": env_vars["object_name"]},
            ExpiresIn=env_vars["expiration_seconds"]
        )

        # Create metadata JSON content
        json_content = {
            "original_object": {
                "bucket": env_vars["bucket_name"],
                "key": env_vars["object_name"],
                "presigned_url": object_presigned_url,
                "url_created_at": current_time.isoformat(),
                "url_expires_at": (current_time + timedelta(seconds=env_vars["expiration_seconds"])).isoformat(),
                "expiration": format_duration(env_vars["expiration_seconds"])
            }
        }

        # Upload JSON file to S3
        client.put_object(
            Bucket=env_vars["bucket_name"],
            Key=json_filename,
            Body=json.dumps(json_content, indent=2),
            ContentType="application/json"
        )

        # Generate presigned URL for the JSON file with inline content disposition
        json_presigned_url = client.generate_presigned_url(
            "get_object",
            Params={
                "Bucket": env_vars["bucket_name"],
                "Key": json_filename,
                "ResponseContentDisposition": "inline"
            },
            ExpiresIn=300  # 5 minutes
        )

        # Log the essential information
        logger.info(f"""
        1. Metadata JSON:
        {json_presigned_url}

        2. Original Object (expires in {format_duration(env_vars['expiration_seconds'])}):
        {object_presigned_url}
        """)

        return {
            "statusCode": 200,
            "body": json.dumps({
                "status": "success",
                "urls": {
                    "metadata_json": {
                        "url": json_presigned_url,
                        "expires_in": "5 minutes"
                    },
                    "original_object": {
                        "url": object_presigned_url,
                        "expires_in": format_duration(env_vars["expiration_seconds"])
                    }
                }
            }, indent=2),
            "headers": {"Content-Type": "application/json"}
        }

    except Exception as e:
        logger.error(f"Error occurred: {str(e)}")
        return {
            "statusCode": 500,
            "body": json.dumps({
                "status": "error",
                "message": "Error generating URLs",
                "error": str(e)
            }, indent=2),
            "headers": {"Content-Type": "application/json"}
        }