如何用Python实现AWS KMS服务端加密文件并优化S3上传代码？

Hey there! Let's fix up your code to meet all your requirements—proper file validation (existence + non-zero size), AWS KMS server-side encryption for S3 uploads, and cleaner, more maintainable code structure.

First, let's outline the key issues in your original code: #

• Overuse of global variables: Makes code hard to test, debug, and reuse
• Incomplete file checks: Only verifies file existence, not that it's non-zero size
• Broken S3 upload logic: Passes a string File instead of actual file content to the S3 Body parameter
• Missing KMS encryption: No parameters configured to enable server-side encryption with AWS KMS
• Platform-specific paths: Uses hardcoded \\ which fails on macOS/Linux
• Coupled functions: readstatus directly calls movefiles, reducing flexibility
• Glob pattern mishandling: Tries to open a glob pattern as a single file, which will fail if multiple files match

Optimized Code #

import os
import sys
import boto3
import glob
from botocore.client import Config
import configparser
from typing import Dict

def load_config(config_path: str, section: str) -> Dict[str, str]:
    """Load and validate configuration from the specified config file and section."""
    config = configparser.ConfigParser()
    if not config.read(config_path):
        raise FileNotFoundError(f"Config file {config_path} not found or unreadable")
    
    # Load global access credentials and bucket info
    config_data = {
        "ACCESS_KEY_ID": config.get("ACCESS", "ACCESS_KEY_ID"),
        "ACCESS_SECRET_KEY": config.get("ACCESS", "ACCESS_SECRET_KEY"),
        "BUCKET_NAME": config.get("ACCESS", "BUCKET_NAME")
    }
    
    # Load section-specific settings
    config_data.update({
        "SRC_DIR": config.get(section, "SRC_DIR"),
        "FILENAME": config.get(section, "FILENAME"),
        "TARGET_DIR": config.get(section, "TARGET_DIR")
    })
    
    # Ensure all required keys are present
    required_keys = ["ACCESS_KEY_ID", "ACCESS_SECRET_KEY", "BUCKET_NAME", "SRC_DIR", "FILENAME", "TARGET_DIR"]
    for key in required_keys:
        if key not in config_data:
            raise ValueError(f"Missing required config key: {key}")
    
    return config_data

def validate_file(file_path: str) -> None:
    """Validate file exists and is non-zero size; raise exceptions if not."""
    if not os.path.exists(file_path):
        raise FileNotFoundError(f"File does not exist: {file_path}")
    
    if os.path.getsize(file_path) == 0:
        raise ValueError(f"File is empty (0 bytes): {file_path}")

def upload_to_s3_with_kms(
    file_path: str,
    bucket_name: str,
    s3_key: str,
    aws_access_key: str,
    aws_secret_key: str,
    kms_key_id: str = None
) -> None:
    """Upload a file to S3 with AWS KMS server-side encryption."""
    s3 = boto3.resource(
        "s3",
        aws_access_key_id=aws_access_key,
        aws_secret_access_key=aws_secret_key,
        config=Config(signature_version="s3v4")
    )
    
    # Configure KMS encryption parameters
    encryption_args = {"ServerSideEncryption": "aws:kms"}
    if kms_key_id:
        encryption_args["SSEKMSKeyId"] = kms_key_id
    
    # Upload the file in binary mode
    with open(file_path, "rb") as file_obj:
        s3.Bucket(bucket_name).put_object(
            Key=s3_key,
            Body=file_obj,
            **encryption_args
        )
    print(f"✅ Successfully uploaded {file_path} to s3://{bucket_name}/{s3_key} with KMS encryption")

if __name__ == "__main__":
    if len(sys.argv) != 2:
        print("Usage: python script.py <config_section_name>")
        sys.exit(1)
    
    try:
        # Load configuration
        config = load_config("CONFIG_AIRBILLING.conf", sys.argv[1])
        
        # Find all matching files using glob pattern
        file_pattern = f"{config['FILENAME']}*.txt"
        full_pattern = os.path.join(config["SRC_DIR"], file_pattern)
        matching_files = glob.glob(full_pattern)
        
        if not matching_files:
            raise FileNotFoundError(f"No files matching pattern '{file_pattern}' found in {config['SRC_DIR']}")
        
        # Process first matching file (loop through matching_files to handle all)
        target_file = matching_files[0]
        
        # Validate the file meets requirements
        validate_file(target_file)
        
        # Build S3 key (use forward slashes for S3 compatibility)
        s3_key = os.path.join(config["TARGET_DIR"], os.path.basename(target_file)).replace("\\", "/")
        
        # Upload with KMS encryption (replace with your KMS key ID if using a custom key)
        # Omit kms_key_id to use the default S3 KMS key
        upload_to_s3_with_kms(
            file_path=target_file,
            bucket_name=config["BUCKET_NAME"],
            s3_key=s3_key,
            aws_access_key=config["ACCESS_KEY_ID"],
            aws_secret_key=config["ACCESS_SECRET_KEY"],
            kms_key_id="your-kms-key-arn-or-id-here"
        )
        
    except Exception as e:
        print(f"❌ Error: {str(e)}")
        sys.exit(1)

Key Improvements Explained: #

1. No Global Variables: Configuration and file paths are passed as function parameters, making the code modular and testable.
2. Comprehensive File Validation:
   • Checks if the file exists
   • Verifies the file size is not zero bytes
   • Properly handles glob patterns to find matching files
3. KMS Encryption Implementation:
   • Uses ServerSideEncryption='aws:kms' to enable KMS server-side encryption
   • Supports custom KMS keys via the SSEKMSKeyId parameter (omit this to use the default S3 KMS key)
4. Cross-Platform Compatibility: Uses os.path.join instead of hardcoded slashes to work on Windows, macOS, and Linux.
5. Single-Responsibility Functions: Each function handles one task (config loading, file validation, S3 upload), making the code easier to understand and modify.
6. Clear Error Handling: Specific exceptions are raised for different failure cases, with descriptive messages to simplify debugging.
7. Correct S3 Upload Logic: Reads the file in binary mode and passes the file object to S3, ensuring the actual content is uploaded.

内容的提问来源于stack exchange，提问作者user138770