Added S3 SSEKMSKeyId config option (#3762)

* Added S3 SSEKMSKeyId config option

* test: unit: s3: add SSEKMSKeyId test

Co-authored-by: Ruslan Kuprieiev <[email protected]>

dvc/config.py

@@ -151,6 +151,7 @@ SCHEMA = {
                     Optional("listobjects", default=False): Bool,
                     Optional("use_ssl", default=True): Bool,
                     "sse": str,
+                    "sse_kms_key_id": str,
                     "acl": str,
                     "grant_read": str,
                     "grant_read_acp": str,

dvc/remote/s3.py

@@ -46,6 +46,10 @@ class S3Remote(BaseRemote):
         if self.sse:
             self.extra_args["ServerSideEncryption"] = self.sse
 
+        self.sse_kms_key_id = config.get("sse_kms_key_id")
+        if self.sse_kms_key_id:
+            self.extra_args["SSEKMSKeyId"] = self.sse_kms_key_id
+
         self.acl = config.get("acl")
         if self.acl:
             self.extra_args["ACL"] = self.acl

tests/unit/remote/test_s3.py

@@ -53,3 +53,8 @@ def test_grants_mutually_exclusive_acl_error(dvc, grants):
 
         with pytest.raises(ConfigError):
             S3Remote(dvc, config)
+
+
+def test_sse_kms_key_id(dvc):
+    remote = S3Remote(dvc, {"url": url, "sse_kms_key_id": "key"})
+    assert remote.extra_args["SSEKMSKeyId"] == "key"