Fixes #6141. Adds support for SSE-C in s3fs.

Author: ap-kulkarni

src/dvc_objects/fs/implementations/s3.py

@@ -74,6 +74,10 @@ def _load_aws_config_file(self, profile):
         return self._split_s3_config(s3_config)
 
     def _prepare_credentials(self, **config):
+        import base64
+
+        from s3fs.utils import SSEParams
+
         login_info = defaultdict(dict)
 
         # credentials
@@ -98,8 +102,24 @@ def _prepare_credentials(self, **config):
 
         # encryptions
         additional = login_info["s3_additional_kwargs"]
-        additional["ServerSideEncryption"] = config.get("sse")
-        additional["SSEKMSKeyId"] = config.get("sse_kms_key_id")
+        sse_customer_key = None
+        if config.get("sse_customer_key"):
+            if config.get("sse_kms_key_id"):
+                raise ConfigError(
+                    "`sse_kms_key_id` and `sse_customer_key` AWS S3 config "
+                    "options are mutually exclusive"
+                )
+            sse_customer_key = base64.b64decode(config.get("sse_customer_key"))
+        sse_customer_algorithm = config.get("sse_customer_algorithm")
+        if not sse_customer_algorithm and sse_customer_key:
+            sse_customer_algorithm = "AES256"
+        sse_params = SSEParams(
+            server_side_encryption=config.get("sse"),
+            sse_customer_algorithm=sse_customer_algorithm,
+            sse_customer_key=sse_customer_key,
+            sse_kms_key_id=config.get("sse_kms_key_id"),
+        )
+        additional.update(sse_params.to_kwargs())
         additional["ACL"] = config.get("acl")
         for grant_option, grant_key in self._GRANTS.items():
             if config.get(grant_option):