Python / boto3

Upload file with locally-calculated checksum & checking it later

def checksum(filename_or_fopen):
    get_sha = lambda f: reduce(lambda sha, b: (sha.update(b), sha)[1], iter(lambda: f.read(2**16), b""), hashlib.sha256())

    sha = None
    if isinstance(filename_or_fopen, str):
        with open(filename_or_fopen, "rb") as f: 
            sha = get_sha(f)
    elif isinstance(filename_or_fopen,  io.IOBase):
        sha = get_sha(filename_or_fopen)
    else:
        raise ValueError("Must be a path or file-like object")
    return sha.digest()

# This one is used by AWS - part_size is from `ObjectParts` below
def checksum_by_parts(filename, part_size):
    final_sha = hashlib.sha256()
    with open(filename, "rb") as f: 
        for chunk in iter(lambda: f.read(part_size), b""): 
            sha = hashlib.sha256()
            sha.update(chunk)
            final_sha.update(sha.digest())
    return final_sha.digest()

p = "local_path"
s256 = checksum(p)
with Path(p).open("rb") as f:
    s3_client.put_object(
        Bucket="<bucket>", 
        Key="<path in bucket>",
        ChecksumSHA256=b64encode(s256).decode("utf-8"),
        Body=f
    )

# Retrieving the checksum later:
attrs = s3_client.get_object_attributes(Bucket="<bucket>", 
    Key="<path in bucket>", 
    ObjectAttributes=["Checksum", "ObjectParts"]
    )
attrs

S3

Server Side encryption with custom Key - permission errors

When you get Access Denied on a bucket which has SSE enabled (especially with Custom Managed Key stored in KMS), check the following:

1. Bucket policy - there can be policies that deny some request, e.g. denyUnencryptedObjectUploads
2. Make sure that the requests (whether via REST or Python or whatever) have appropriate headers set. Required headers for SSE with CMS:
   • x-amz-server-side-encryption
   • x-amz-server-side-encryption-aws-kms-key-id

They can be set directly in boto3 or with OpenDAL config.