Fix S3 ArtifactStore auth issue

docs/book/component-guide/artifact-stores/s3.md

@@ -182,7 +182,7 @@ This method has some advantages over the implicit authentication method:
 * you don't need to care about enabling your other stack components (orchestrators, step operators, and model deployers) to have access to the artifact store through IAM roles and policies
 * you can combine the S3 artifact store with other stack components that are not running in AWS
 
-> **Note**: When you create the IAM user for your AWS access key, please remember to grant the created IAM user permissions to read and write to your S3 bucket (i.e. at a minimum: `s3:PutObject`, `s3:GetObject`, `s3:ListBucket`, `s3:DeleteObject`)
+> **Note**: When you create the IAM user for your AWS access key, please remember to grant the created IAM user permissions to read and write to your S3 bucket (i.e. at a minimum: `s3:PutObject`, `s3:GetObject`, `s3:ListBucket`, `s3:DeleteObject`, `s3:GetBucketVersioning`, `s3:ListBucketVersions`, `s3:DeleteObjectVersion`)
 
 After having set up the IAM user and generated the access key, as described in the [AWS documentation](https://aws.amazon.com/premiumsupport/knowledge-center/create-access-key/), you can register the S3 Artifact Store as follows:
 

docs/book/how-to/auth-management/README.md

@@ -225,13 +225,16 @@ The configured credentials must have at least the following AWS IAM permissions
 associated with the ARNs of S3 buckets that the connector will be allowed to    
 access (e.g. arn:aws:s3:::* and arn:aws:s3:::*/* represent all the available S3 
 buckets).                                                                       
-
- • s3:ListBucket                                                                
- • s3:GetObject                                                                 
- • s3:PutObject                                                                 
- • s3:DeleteObject                                                              
- • s3:ListAllMyBuckets                                                          
-
+
+ • s3:ListBucket
+ • s3:GetObject
+ • s3:PutObject
+ • s3:DeleteObject
+ • s3:ListAllMyBuckets
+ • s3:GetBucketVersioning
+ • s3:ListBucketVersions
+ • s3:DeleteObjectVersion
+
 If set, the resource name must identify an S3 bucket using one of the following 
 formats:                                                                        
 

docs/book/how-to/auth-management/aws-service-connector.md

@@ -72,6 +72,9 @@ The configured credentials must have at least the following [AWS IAM permissions
 * `s3:PutObject`
 * `s3:DeleteObject`
 * `s3:ListAllMyBuckets`
+* `s3:GetBucketVersioning`
+* `s3:ListBucketVersions`
+* `s3:DeleteObjectVersion`
 
 {% hint style="info" %}
 If you are using the [AWS IAM role](aws-service-connector.md#aws-iam-role), [Session Token](aws-service-connector.md#aws-session-token), or [Federation Token](aws-service-connector.md#aws-federation-token) authentication methods, you don't have to worry too much about restricting the permissions of the AWS credentials that you use to access the AWS cloud resources. These authentication methods already support [automatically generating temporary tokens](best-security-practices.md#generating-temporary-and-down-scoped-credentials) with permissions down-scoped to the minimum required to access the target resource.

docs/book/how-to/stack-deployment/deploy-a-cloud-stack.md

@@ -320,6 +320,9 @@ following AWS permissions in your AWS account:
   * s3:GetObject
   * s3:PutObject
   * s3:DeleteObject
+  * s3:GetBucketVersioning
+  * s3:ListBucketVersions
+  * s3:DeleteObjectVersion
 * ECR Repository:
   * ecr:DescribeRepositories
   * ecr:ListRepositories

infra/aws/aws-ecr-s3-sagemaker.yaml

@@ -119,6 +119,9 @@ Resources:
                   - 's3:GetObject'
                   - 's3:PutObject'
                   - 's3:DeleteObject'
+                  - 's3:GetBucketVersioning'
+                  - 's3:ListBucketVersions'
+                  - 's3:DeleteObjectVersion'
                 Resource:
                   - !Sub '${S3Bucket.Arn}'
                   - !Sub '${S3Bucket.Arn}/*'

src/zenml/integrations/aws/service_connectors/aws_service_connector.py

@@ -522,6 +522,9 @@ class AWSAuthenticationMethods(StrEnum):
 - `s3:PutObject`
 - `s3:DeleteObject`
 - `s3:ListAllMyBuckets`
+- `s3:GetBucketVersioning`
+- `s3:ListBucketVersions`
+- `s3:DeleteObjectVersion`
 
 If set, the resource name must identify an S3 bucket using one of the following
 formats:
@@ -795,6 +798,9 @@ def _get_iam_policy(
                             "s3:PutObject",
                             "s3:DeleteObject",
                             "s3:ListAllMyBuckets",
+                            "s3:GetBucketVersioning",
+                            "s3:ListBucketVersions",
+                            "s3:DeleteObjectVersion",
                         ],
                         "Resource": resource,
                     },

src/zenml/integrations/s3/artifact_stores/s3_artifact_store.py

@@ -13,10 +13,12 @@
 #  permissions and limitations under the License.
 """Implementation of the S3 Artifact Store."""
 
+from contextlib import contextmanager
 from typing import (
     Any,
     Callable,
     Dict,
+    Generator,
     Iterable,
     List,
     Optional,
@@ -27,6 +29,7 @@
 
 import boto3
 import s3fs
+from botocore.exceptions import ClientError
 from fsspec.asyn import FSTimeoutError, sync, sync_wrapper
 
 from zenml.artifact_stores import BaseArtifactStore
@@ -130,17 +133,19 @@ def __init__(
             **kwargs: Additional keyword arguments.
         """
         super().__init__(*args, **kwargs)
+        self._boto3_bucket_holder = None
 
         # determine bucket versioning status
-        s3 = boto3.resource("s3")
-        bucket = s3.Bucket(self.config.bucket)
-        versioning = bucket.Versioning()
-        if versioning.status == "Enabled":
-            self.is_versioned = True
-            logger.warning(
-                f"The artifact store bucket `{self.config.bucket}` is versioned, "
-                "this may slow down logging process significantly."
-            )
+        versioning = self._boto3_bucket.Versioning()
+        with self._shield_lack_of_versioning_permissions(
+            "s3:GetBucketVersioning"
+        ):
+            if versioning.status == "Enabled":
+                self.is_versioned = True
+                logger.warning(
+                    f"The artifact store bucket `{self.config.bucket}` is versioned, "
+                    "this may slow down logging process significantly."
+                )
 
     @property
     def config(self) -> S3ArtifactStoreConfig:
@@ -467,10 +472,51 @@ def _remove_previous_file_versions(self, path: PathType) -> None:
             if isinstance(path, bytes):
                 path = path.decode()
             _, prefix = split_s3_path(path)
-            s3 = boto3.resource("s3")
-            bucket = s3.Bucket(self.config.bucket)
-            for version in bucket.object_versions.filter(Prefix=prefix):
-                if not version.is_latest:
-                    version.delete()
+            with self._shield_lack_of_versioning_permissions(
+                "s3:ListBucketVersions"
+            ):
+                for version in self._boto3_bucket.object_versions.filter(
+                    Prefix=prefix
+                ):
+                    if not version.is_latest:
+                        with self._shield_lack_of_versioning_permissions(
+                            "s3:DeleteObjectVersion"
+                        ):
+                            version.delete()
+
+    @property
+    def _boto3_bucket(self) -> Any:
+        """Get the boto3 bucket object.
+
+        Returns:
+            The boto3 bucket object.
+        """
+        if self._boto3_bucket_holder and not self.connector_has_expired():
+            return self._boto3_bucket_holder
+
+        key, secret, token, region = self.get_credentials()
+        s3 = boto3.resource(
+            "s3",
+            aws_access_key_id=key,
+            aws_secret_access_key=secret,
+            aws_session_token=token,
+            region_name=region,
+        )
+        self._boto3_bucket_holder = s3.Bucket(self.config.bucket)
+        return self._boto3_bucket_holder
 
-        return
+    @contextmanager
+    def _shield_lack_of_versioning_permissions(
+        self, auth_missing: str
+    ) -> Generator[Any, None, None]:
+        try:
+            yield
+        except ClientError as e:
+            if "not authorized" in e.args[0] and auth_missing in e.args[0]:
+                logger.warning(
+                    "Your AWS Connector is lacking critical Versioning permissions. "
+                    f"Please check that `{auth_missing}` is granted."
+                    "This is needed to remove previous versions of log files from your "
+                    "Artifact Store bucket."
+                )
+                self.is_versioned = False

src/zenml/stack_deployments/aws_stack_deployment.py

@@ -139,6 +139,9 @@ def permissions(cls) -> Dict[str, List[str]]:
                 "s3:GetObject",
                 "s3:PutObject",
                 "s3:DeleteObject",
+                "s3:GetBucketVersioning",
+                "s3:ListBucketVersions",
+                "s3:DeleteObjectVersion",
             ],
             "ECR Repository": [
                 "ecr:DescribeRepositories",