remote: Extend S3 remote support for custom ACL

Besides AWS boto3 extra_arg `ACL` (`x-amz-acl` header), there four more
that could be used to grant specific permissions to specific grantees on
the uploaded object.
These headers are:
  - `x-amz-grant-full-control` with boto3's extra_arg `GrantFullControl`
  - `x-amz-grant-read` with boto3's extra_arg `GrantRead`
  - `x-amz-grant-read-acp` with boto3's extra_arg `GrantReadACP`
  - `x-amz-grant-write-acp` with boto3's extra_arg `GrantWriteACP`

As `ACL` (`x-amz-acl` header) is limited to be used with predefined canned ACLs,
adding support for the aformentioned headers will help in applying
more customized persmissions solutions according to different project
needs.

References:
  - https://docs.aws.amazon.com/AmazonS3/latest/dev/acl-overview.html#permissions
  - https://boto3.amazonaws.com/v1/documentation/api/latest/reference/customizations/s3.html#boto3.s3.transfer.S3Transfer.ALLOWED_UPLOAD_ARGS
  - https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObjectAcl.html

Fixes #(2930).

dvc/config.py

@@ -171,6 +171,10 @@ class Config(object):  # pylint: disable=too-many-instance-attributes
     SECTION_AWS_USE_SSL = "use_ssl"
     SECTION_AWS_SSE = "sse"
     SECTION_AWS_ACL = "acl"
+    SECTION_AWS_GRANT_READ = "grant_read"
+    SECTION_AWS_GRANT_READ_ACP = "grant_read_acp"
+    SECTION_AWS_GRANT_WRITE_ACP = "grant_write_acp"
+    SECTION_AWS_GRANT_FULL_CONTROL = "grant_full_control"
 
     # gcp specific options
     SECTION_GCP_CREDENTIALPATH = CREDENTIALPATH
@@ -210,6 +214,10 @@ class Config(object):  # pylint: disable=too-many-instance-attributes
         Optional(SECTION_AWS_USE_SSL, default=True): Bool,
         SECTION_AWS_SSE: str,
         SECTION_AWS_ACL: str,
+        SECTION_AWS_GRANT_READ: str,
+        SECTION_AWS_GRANT_READ_ACP: str,
+        SECTION_AWS_GRANT_WRITE_ACP: str,
+        SECTION_AWS_GRANT_FULL_CONTROL: str,
         SECTION_GCP_PROJECTNAME: str,
         SECTION_CACHE_TYPE: supported_cache_type,
         Optional(SECTION_CACHE_PROTECTED, default=False): Bool,

dvc/remote/s3.py

@@ -53,6 +53,8 @@ class RemoteS3(RemoteBASE):
         if self.acl:
             self.extra_args["ACL"] = self.acl
 
+        self._append_aws_grants_to_extra_args(config)
+
         shared_creds = config.get(Config.SECTION_AWS_CREDENTIALPATH)
         if shared_creds:
             os.environ.setdefault("AWS_SHARED_CREDENTIALS_FILE", shared_creds)
@@ -302,3 +304,56 @@ class RemoteS3(RemoteBASE):
                 continue
 
             yield path_info.replace(path=fname)
+
+    def _append_aws_grants_to_extra_args(self, config):
+        # Keys for extra_args can be one of the following list:
+        # https://boto3.amazonaws.com/v1/documentation/api/latest/reference/customizations/s3.html#boto3.s3.transfer.S3Transfer.ALLOWED_UPLOAD_ARGS
+        """
+          ALLOWED_UPLOAD_ARGS = [
+            'ACL', 'CacheControl', 'ContentDisposition', 'ContentEncoding',
+            'ContentLanguage', 'ContentType', 'Expires', 'GrantFullControl',
+            'GrantRead', 'GrantReadACP', 'GrantWriteACP', 'Metadata',
+            'RequestPayer', 'ServerSideEncryption', 'StorageClass',
+            'SSECustomerAlgorithm', 'SSECustomerKey', 'SSECustomerKeyMD5',
+            'SSEKMSKeyId', 'WebsiteRedirectLocation'
+          ]
+        """
+        overwriting_acl_grantees = False
+
+        self.grant_full_control = config.get(
+            Config.SECTION_AWS_GRANT_FULL_CONTROL, ""
+        )
+        if self.grant_full_control:
+            self.extra_args["GrantFullControl"] = self.grant_full_control
+            overwriting_acl_grantees = True
+
+        self.grant_read = config.get(Config.SECTION_AWS_GRANT_READ, "")
+        if self.grant_read:
+            self.extra_args["GrantRead"] = self.grant_read
+            overwriting_acl_grantees = True
+
+        self.grant_read_acp = config.get(Config.SECTION_AWS_GRANT_READ_ACP, "")
+        if self.grant_read_acp:
+            self.extra_args["GrantReadACP"] = self.grant_read_acp
+            overwriting_acl_grantees = True
+
+        self.grant_write_acp = config.get(
+            Config.SECTION_AWS_GRANT_WRITE_ACP, ""
+        )
+        if self.grant_write_acp:
+            self.extra_args["GrantWriteACP"] = self.grant_write_acp
+            overwriting_acl_grantees = True
+
+        if overwriting_acl_grantees:
+            overwriting_fields = [
+                Config.SECTION_AWS_GRANT_FULL_CONTROL,
+                Config.SECTION_AWS_GRANT_READ,
+                Config.SECTION_AWS_GRANT_READ_ACP,
+                Config.SECTION_AWS_GRANT_WRITE_ACP,
+            ]
+            warning_message = [
+                f"Using one of the [{', '.join(overwriting_fields)}] will",
+                "overwrite default ACL Grantees list. Default ACL Grantees",
+                "list include Owner with FULL_CONTROL permissions.",
+            ]
+            logger.warning(" ".join(warning_message))