synology c2 bucket boto3 working

README.md

Synology C2 S3 Bucket Boto3 Code

This code fixes the chunked encoding error from SYNOLOGY C2 S3 BUCKET in BOTO3

error: boto3.exceptions.S3UploadFailedError: Failed to upload requirements.txt to requirements.txt: An error occurred (NotImplemented) when calling the PutObject operation: Transfering payloads in multiple chunks using aws-chunked is not supported.

We need to disable chunked encoding as c2 synology has no chunked encoding support like aws s3 or recent minio versions.

Note: This was not AI generated

References

• treeverse/lakeFS#2471 (comment)
• vmware-tanzu/velero#8660
• vitessio/vitess#18210

requirements.txt

boto3

s3_config.json

{
    "endpoint_url": "https://region-000.s3.synologyc2.net",
    "bucket_name": "S3_BUCKET",
    "aws_access_key_id": "SYNOLOGY_C2",
    "aws_secret_access_key": "SYNOLOGY_C2"
}

synology_c2_bucket_boto3_working.py

import json

import boto3
from botocore.handlers import disable_signing

# Create a Client with config
s3 = boto3.client('s3', endpoint_url=json.load(open('s3_config.json'))['endpoint_url'],
                  aws_access_key_id=json.load(open('s3_config.json'))[
    'aws_access_key_id'],
    aws_secret_access_key=json.load(open('s3_config.json'))['aws_secret_access_key'])
s3.meta.events.unregister('before-sign.s3.*', disable_signing)


# Clean Chunked Encoding Headers to Fix C2 Compatibility
def header_cleaner(request, **kwargs):
    prohibited_headers = [
        'x-amz-decoded-content-length',
        'x-amz-content-sha256',
        'content-encoding'
    ]
    for hdr in prohibited_headers:
        if hdr in request.headers:
            del request.headers[hdr]

# Remove Chunked Signing
s3.meta.events.register_last('before-sign.s3.*', header_cleaner)

# upload "requirements.txt" to S3 bucket
s3.upload_file(
    Filename='requirements.txt',
    Bucket=json.load(open('s3_config.json'))['bucket_name'],
    Key='requirements.txt'
)

synology_c2_bucket_boto3_working_.py

import json
from pathlib import Path

import boto3
from botocore.client import BaseClient as botocore_client
from botocore.handlers import disable_signing


# Initialize the S3 client without signing requests
def get_s3_client(path: Path | str) -> botocore_client:
    """
    Initialize and return an S3 client with custom configuration.
    This client is configured to not sign requests and to clean up certain headers.
    """
    config_json: dict

    # Ensure the provided path is a valid JSON file
    try:
        with open(path, 'r', encoding='utf-8') as file:
            config_json = json.load(file)
    except FileNotFoundError:
        raise FileNotFoundError(f"Configuration file '{path}' not found.")
    except json.JSONDecodeError:
        raise ValueError(f"Configuration file '{path}' is not a valid JSON.")

    # Load S3 configuration from a JSON file
    s3 = boto3.client('s3', endpoint_url=config_json['endpoint_url'],
                      aws_access_key_id=config_json['aws_access_key_id'],
                      aws_secret_access_key=config_json['aws_secret_access_key'])
    s3.meta.events.unregister('before-sign.s3.*', disable_signing)

    def header_cleaner(request, **kwargs):
        prohibited_headers = [
            'x-amz-decoded-content-length',
            'x-amz-content-sha256',
            'content-encoding'
        ]
        for hdr in prohibited_headers:
            if hdr in request.headers:
                del request.headers[hdr]

    # Register the header cleaner to remove specific headers before signing
    s3.meta.events.register_last('before-sign.s3.*', header_cleaner)
    return s3


# Example usage:
if __name__ == "__main__":
    s3_client = get_s3_client('s3_config.json')
    print("S3 client initialized successfully.")
    # You can now use s3_client to interact with your S3 bucket
    # Example: List buckets
    try:
        response = s3_client.list_buckets()
        print("Buckets available:", [bucket['Name']
              for bucket in response['Buckets']])
    except Exception as e:
        print(f"Error listing buckets: {e}")
    # check if requirements.txt is present try to upload it to s3 bucket
    if Path('requirements.txt').exists():
        try:
            first_bucket = s3_client.list_buckets()['Buckets'][0]['Name']
            s3_client.upload_file('requirements.txt',
                                  first_bucket, 'requirements.txt')
            print("requirements.txt uploaded successfully.")
        except Exception as e:
            print(f"Error uploading requirements.txt: {e}")