content: #  Copyright (c) ZenML GmbH 2023. All Rights Reserved.
  #
  #  Licensed under the Apache License, Version 2.0 (the "License");
  #  you may not use this file except in compliance with the License.
  #  You may obtain a copy of the License at:
  #
  #       https://www.apache.org/licenses/LICENSE-2.0
  #
  #  Unless required by applicable law or agreed to in writing, software
  #  distributed under the License is distributed on an "AS IS" BASIS,
  #  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
  #  or implied. See the License for the specific language governing
  #  permissions and limitations under the License.
  """AWS Service Connector.
  
  The AWS Service Connector implements various authentication methods for AWS
  services:
  
  - Explicit AWS secret key (access key, secret key)
  - Explicit  AWS STS tokens (access key, secret key, session token)
  - IAM roles (i.e. generating temporary STS tokens on the fly by assuming an
  IAM role)
  - IAM user federation tokens
  - STS Session tokens
  
  """
  
  import base64
  import datetime
  import json
  import os
  import re
  from typing import Any, Dict, List, Optional, Tuple, cast
  
  import boto3
  from aws_profile_manager import Common  # type: ignore[import-untyped]
  from botocore.client import BaseClient
  from botocore.exceptions import BotoCoreError, ClientError
  from botocore.signers import RequestSigner
  from pydantic import Field
  
  from zenml.constants import (
      DOCKER_REGISTRY_RESOURCE_TYPE,
      KUBERNETES_CLUSTER_RESOURCE_TYPE,
  )
  from zenml.exceptions import AuthorizationException
  from zenml.integrations.aws import (
      AWS_CONNECTOR_TYPE,
      AWS_RESOURCE_TYPE,
      S3_RESOURCE_TYPE,
  )
  from zenml.logger import get_logger
  from zenml.models import (
      AuthenticationMethodModel,
      ResourceTypeModel,
      ServiceConnectorTypeModel,
  )
  from zenml.service_connectors.docker_service_connector import (
      DockerAuthenticationMethods,
      DockerConfiguration,
      DockerServiceConnector,
  )
  from zenml.service_connectors.service_connector import (
      AuthenticationConfig,
      ServiceConnector,
  )
  from zenml.utils.enum_utils import StrEnum
  from zenml.utils.secret_utils import PlainSerializedSecretStr
  from zenml.utils.time_utils import utc_now_tz_aware
  
  logger = get_logger(__name__)
  
  EKS_KUBE_API_TOKEN_EXPIRATION = 15  # 15 minutes
  DEFAULT_IAM_ROLE_TOKEN_EXPIRATION = 3600  # 1 hour
  DEFAULT_STS_TOKEN_EXPIRATION = 43200  # 12 hours
  BOTO3_SESSION_EXPIRATION_BUFFER = 15  # 15 minutes
  
  
  class AWSSecretKey(AuthenticationConfig):
      """AWS secret key credentials."""
  
      aws_access_key_id: PlainSerializedSecretStr = Field(
          title="AWS Access Key ID",
          description="An AWS access key ID associated with an AWS account or IAM user.",
      )
      aws_secret_access_key: PlainSerializedSecretStr = Field(
          title="AWS Secret Access Key",
      )
  
  
  class STSToken(AWSSecretKey):
      """AWS STS token."""
  
      aws_session_token: PlainSerializedSecretStr = Field(
          title="AWS Session Token",
      )
  
  
  class AWSBaseConfig(AuthenticationConfig):
      """AWS base configuration."""
  
      region: str = Field(
          title="AWS Region",
      )
      endpoint_url: Optional[str] = Field(
          default=None,
          title="AWS Endpoint URL",
      )
  
  
  class AWSSessionPolicy(AuthenticationConfig):
      """AWS session IAM policy configuration."""
  
      policy_arns: Optional[List[str]] = Field(
          default=None,
          title="ARNs of the IAM managed policies that you want to use as a "
          "managed session policy. The policies must exist in the same account "
          "as the IAM user that is requesting temporary credentials.",
      )
      policy: Optional[str] = Field(
          default=None,
          title="An IAM policy in JSON format that you want to use as an inline "
          "session policy",
      )
  
  
  class AWSImplicitConfig(AWSBaseConfig, AWSSessionPolicy):
      """AWS implicit configuration."""
  
      profile_name: Optional[str] = Field(
          default=None,
          title="AWS Profile Name",
      )
      role_arn: Optional[str] = Field(
          default=None,
          title="Optional AWS IAM Role ARN to assume",
      )
  
  
  class AWSSecretKeyConfig(AWSBaseConfig, AWSSecretKey):
      """AWS secret key authentication configuration."""
  
  
  class STSTokenConfig(AWSBaseConfig, STSToken):
      """AWS STS token authentication configuration."""
  
  
  class IAMRoleAuthenticationConfig(AWSSecretKeyConfig, AWSSessionPolicy):
      """AWS IAM authentication config."""
  
      role_arn: str = Field(
          title="AWS IAM Role ARN",
      )
  
  
  class SessionTokenAuthenticationConfig(AWSSecretKeyConfig):
      """AWS session token authentication config."""
  
  
  class FederationTokenAuthenticationConfig(
      AWSSecretKeyConfig, AWSSessionPolicy
  ):
      """AWS federation token authentication config."""
  
  
  class AWSAuthenticationMethods(StrEnum):
      """AWS Authentication methods."""
  
      IMPLICIT = "implicit"
      SECRET_KEY = "secret-key"
      STS_TOKEN = "sts-token"
      IAM_ROLE = "iam-role"
      SESSION_TOKEN = "session-token"
      FEDERATION_TOKEN = "federation-token"
  
  
  AWS_SERVICE_CONNECTOR_TYPE_SPEC = ServiceConnectorTypeModel(
      name="AWS Service Connector",
      connector_type=AWS_CONNECTOR_TYPE,
      description="""
  The ZenML AWS Service Connector facilitates the authentication and access to
  managed AWS services and resources. These encompass a range of resources,
  including S3 buckets, ECR repositories, and EKS clusters. The connector
  provides support for various authentication methods, including explicit
  long-lived AWS secret keys, IAM roles, short-lived STS tokens and implicit
  authentication.
  
  To ensure heightened security measures, this connector also enables the
  generation of temporary STS security tokens that are scoped down to the
  minimum permissions necessary for accessing the intended resource. Furthermore,
  it includes automatic configuration and detection of credentials locally
  configured through the AWS CLI.
  
  This connector serves as a general means of accessing any AWS service by
  issuing pre-authenticated boto3 sessions to clients. Additionally, the connector
  can handle specialized authentication for S3, Docker and Kubernetes Python
  clients. It also allows for the configuration of local Docker and Kubernetes
  CLIs.
  
  The AWS Service Connector is part of the AWS ZenML integration. You can either
  install the entire integration or use a pypi extra to install it independently
  of the integration:
  
  * `pip install "zenml[connectors-aws]"` installs only prerequisites for the AWS
  Service Connector Type
  * `zenml integration install aws` installs the entire AWS ZenML integration
  
  It is not required to [install and set up the AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)
  on your local machine to use the AWS Service Connector to link Stack Components
  to AWS resources and services. However, it is recommended to do so if you are
  looking for a quick setup that includes using the auto-configuration Service
  Connector features.
  """,
      supports_auto_configuration=True,
      logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/connectors/aws/aws.png",
      emoji=":large_orange_diamond:",
      auth_methods=[
          AuthenticationMethodModel(
              name="AWS Implicit Authentication",
              auth_method=AWSAuthenticationMethods.IMPLICIT,
              description="""
  Implicit authentication to AWS services using environment variables, local
  configuration files or IAM roles. This authentication method doesn't require
  any credentials to be explicitly configured. It automatically discovers and uses
  credentials from one of the following sources:
  
  - environment variables (AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY,
  AWS_SESSION_TOKEN, AWS_DEFAULT_REGION)
  - local configuration files [set up through the AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/cli-chap-configure.html)
  (~/aws/credentials, ~/.aws/config)
  - IAM roles for Amazon EC2, ECS, EKS, Lambda, etc. Only works when running
  the ZenML server on an AWS resource with an IAM role attached to it.
  
  This is the quickest and easiest way to authenticate to AWS services. However,
  the results depend on how ZenML is deployed and the environment where it is used
  and is thus not fully reproducible:
  
  - when used with the default local ZenML deployment or a local ZenML server, the
  credentials are the same as those used by the AWS CLI or extracted from local
  environment variables.
  - when connected to a ZenML server, this method only works if the ZenML server
  is deployed in AWS and will use the IAM role attached to the AWS resource where
  the ZenML server is running (e.g. an EKS cluster). The IAM role permissions may
  need to be adjusted to allows listing and accessing/describing the AWS resources
  that the connector is configured to access.
  
  An IAM role may optionally be specified to be assumed by the connector on top of
  the implicit credentials. This is only possible when the implicit credentials
  have permissions to assume the target IAM role. Configuring an IAM role has all
  the advantages of the AWS IAM Role authentication method plus the added benefit
  of not requiring any explicit credentials to be configured and stored:
  
  * The connector will generate temporary STS tokens upon request by
  [calling the AssumeRole STS API](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_assumerole).
  * Allows implementing a two layer authentication scheme that keeps the set of
  permissions associated with implicit credentials down to the bare minimum and
  grants permissions to the privilege-bearing IAM role instead.
  * One or more optional [IAM session policies](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html#policies_session)
  may also be configured to further restrict the permissions of the generated
  STS tokens. If not specified, IAM session policies are automatically configured
  for the generated STS tokens to restrict them to the minimum set of permissions
  required to access the target resource. Refer to the documentation for each
  supported Resource Type for the complete list of AWS permissions automatically
  granted to the generated STS tokens.
  * The default expiration period for generated STS tokens is 1 hour with a
  minimum of 15 minutes up to the maximum session duration setting configured for
  the IAM role (default is 1 hour). If you need longer-lived tokens, you can
  configure the IAM role to use a higher maximum expiration value (up to 12 hours)
  or use the AWS Federation Token or AWS Session Token authentication methods.
  
  Note that the discovered credentials inherit the full set of permissions of the
  local AWS client configuration, environment variables or attached AWS IAM role.
  Depending on the extent of those permissions, this authentication method might
  not be recommended for production use, as it can lead to accidental privilege
  escalation. It is recommended to also configure an IAM role when using
  the implicit authentication method, or to use the AWS IAM Role, AWS Session
  Token or AWS Federation Token authentication methods instead to limit the
  validity and/or permissions of the credentials being issued to connector
  clients.
  
  If you need to access an EKS kubernetes cluster with this authentication method,
  please be advised that the EKS cluster's aws-auth ConfigMap may need to be
  manually configured to allow authentication with the implicit IAM user or role
  picked up by the Service Connector. For more information,
  [see this documentation](https://docs.aws.amazon.com/eks/latest/userguide/add-user-role.html).
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  """,
              min_expiration_seconds=900,  # 15 minutes
              default_expiration_seconds=DEFAULT_IAM_ROLE_TOKEN_EXPIRATION,  # 1 hour
              config_class=AWSImplicitConfig,
          ),
          AuthenticationMethodModel(
              name="AWS Secret Key",
              auth_method=AWSAuthenticationMethods.SECRET_KEY,
              description="""
  Long-lived AWS credentials consisting of an AWS access key ID and secret access
  key associated with an AWS IAM user or AWS account root user (not recommended).
  
  This method is preferred during development and testing due to its simplicity
  and ease of use. It is not recommended as a direct authentication method for
  production use cases because the clients have direct access to long-lived
  credentials and are granted the full set of permissions of the IAM user or AWS
  account root user associated with the credentials. For production, it is
  recommended to use the AWS IAM Role, AWS Session Token or AWS Federation Token
  authentication method instead.
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  
  If you already have the local AWS CLI set up with these credentials, they will
  be automatically picked up when auto-configuration is used.
  """,
              config_class=AWSSecretKeyConfig,
          ),
          AuthenticationMethodModel(
              name="AWS STS Token",
              auth_method=AWSAuthenticationMethods.STS_TOKEN,
              description="""
  Uses temporary STS tokens explicitly configured by the user or auto-configured
  from a local environment. This method has the major limitation that the user
  must regularly generate new tokens and update the connector configuration as STS
  tokens expire. On the other hand, this method is ideal in cases where the
  connector only needs to be used for a short period of time, such as sharing
  access temporarily with someone else in your team.
  
  Using other authentication methods like IAM role, Session Token or Federation
  Token will automatically generate and refresh STS tokens for clients upon
  request.
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  """,
              config_class=STSTokenConfig,
          ),
          AuthenticationMethodModel(
              name="AWS IAM Role",
              auth_method=AWSAuthenticationMethods.IAM_ROLE,
              description="""
  Generates temporary STS credentials by assuming an AWS IAM role.
  The connector needs to be configured with the IAM role to be assumed accompanied
  by an AWS secret key associated with an IAM user or an STS token associated with
  another IAM role. The IAM user or IAM role must have permissions to assume the
  target IAM role. The connector will generate temporary STS tokens upon
  request by [calling the AssumeRole STS API](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_assumerole).
  
  This authentication method still requires credentials to be explicitly
  configured. If your ZenML server is running in AWS and you're looking for an
  alternative that uses implicit credentials while at the same time benefits from
  all the security advantages of assuming an IAM role, you should use the implicit
  authentication method with a configured IAM role instead.
  
  The best practice implemented with this authentication scheme is to keep the set
  of permissions associated with the primary IAM user or IAM role down to the bare
  minimum and grant permissions to the privilege bearing IAM role instead.
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  
  One or more optional [IAM session policies](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html#policies_session)
  may also be configured to further restrict the permissions of the generated STS
  tokens. If not specified, IAM session policies are automatically configured for
  the generated STS tokens to restrict them to the minimum set of permissions
  required to access the target resource. Refer to the documentation for each
  supported Resource Type for the complete list of AWS permissions automatically
  granted to the generated STS tokens.
  
  The default expiration period for generated STS tokens is 1 hour with a minimum
  of 15 minutes up to the maximum session duration setting configured for the IAM
  role (default is 1 hour). If you need longer-lived tokens, you can configure the
  IAM role to use a higher maximum expiration value (up to 12 hours) or use the
  AWS Federation Token or AWS Session Token authentication methods.
  
  For more information on IAM roles and the AssumeRole AWS API, see
  [the official AWS documentation on the subject](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_assumerole).
  
  For more information about the difference between this method and the AWS
  Federation Token authentication method,
  [consult this AWS documentation page](https://aws.amazon.com/blogs/security/understanding-the-api-options-for-securely-delegating-access-to-your-aws-account/). 
  """,
              min_expiration_seconds=900,  # 15 minutes
              default_expiration_seconds=DEFAULT_IAM_ROLE_TOKEN_EXPIRATION,  # 1 hour
              config_class=IAMRoleAuthenticationConfig,
          ),
          AuthenticationMethodModel(
              name="AWS Session Token",
              auth_method=AWSAuthenticationMethods.SESSION_TOKEN,
              description="""
  Generates temporary session STS tokens for IAM users.
  The connector needs to be configured with an AWS secret key associated with an
  IAM user or AWS account root user (not recommended). The connector will generate
  temporary STS tokens upon request by calling [the GetSessionToken STS API](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_getsessiontoken).
  
  These STS tokens have an expiration period longer that those issued through the
  AWS IAM Role authentication method and are more suitable for long-running
  processes that cannot automatically re-generate credentials upon expiration.
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  
  The default expiration period for generated STS tokens is 12 hours with a
  minimum of 15 minutes and a maximum of 36 hours. Temporary credentials obtained
  by using the AWS account root user credentials (not recommended) have a maximum
  duration of 1 hour.
  
  As a precaution, when long-lived credentials (i.e. AWS Secret Keys) are detected
  on your environment by the Service Connector during auto-configuration, this
  authentication method is automatically chosen instead of the AWS Secret Key
  authentication method alternative.
  
  Generated STS tokens inherit the full set of permissions of the IAM user or
  AWS account root user that is calling the GetSessionToken API. Depending on your
  security needs, this may not be suitable for production use, as it can lead to
  accidental privilege escalation. Instead, it is recommended to use the AWS
  Federation Token or AWS IAM Role authentication methods to restrict the
  permissions of the generated STS tokens.
  
  For more information on session tokens and the GetSessionToken AWS API, see:
  [the official AWS documentation on the subject](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_getsessiontoken).
  """,
              min_expiration_seconds=900,  # 15 minutes
              max_expiration_seconds=129600,  # 36 hours
              default_expiration_seconds=DEFAULT_STS_TOKEN_EXPIRATION,  # 12 hours
              config_class=SessionTokenAuthenticationConfig,
          ),
          AuthenticationMethodModel(
              name="AWS Federation Token",
              auth_method=AWSAuthenticationMethods.FEDERATION_TOKEN,
              description="""
  Generates temporary STS tokens for federated users.
  The connector needs to be configured with an AWS secret key associated with an
  IAM user or AWS account root user (not recommended). The IAM user must have
  permissions to call [the GetFederationToken STS API](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_getfederationtoken)
  (i.e. allow the `sts:GetFederationToken` action on the `*` IAM resource). The
  connector will generate temporary STS tokens upon request by calling the
  GetFederationToken STS API.
  
  These STS tokens have an expiration period longer that those issued through the AWS IAM Role
  authentication method and are more suitable for long-running processes that
  cannot automatically re-generate credentials upon expiration.
  
  An AWS region is required and the connector may only be used to access AWS
  resources in the specified region.
  
  One or more optional [IAM session policies](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html#policies_session)
  may also be configured to further restrict the permissions of the generated STS
  tokens. If not specified, IAM session policies are automatically configured for
  the generated STS tokens to restrict them to the minimum set of permissions
  required to access the target resource. Refer to the documentation for each
  supported Resource Type for the complete list of AWS permissions automatically
  granted to the generated STS tokens.
  
  If this authentication method is used with the generic AWS resource type, a
  session policy MUST be explicitly specified, otherwise the generated STS tokens
  will not have any permissions.
  
  The default expiration period for generated STS tokens is 12 hours with a
  minimum of 15 minutes and a maximum of 36 hours. Temporary credentials obtained
  by using the AWS account root user credentials (not recommended) have a maximum
  duration of 1 hour.
  
  If you need to access an EKS kubernetes cluster with this authentication method,
  please be advised that the EKS cluster's aws-auth ConfigMap may need to be
  manually configured to allow authentication with the federated user. For more
  information, [see this documentation](https://docs.aws.amazon.com/eks/latest/userguide/add-user-role.html).
  
  For more information on user federation tokens, session policies and the
  GetFederationToken AWS API, see [the official AWS documentation on the subject](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_credentials_temp_request.html#api_getfederationtoken).
  
  For more information about the difference between this method and the AWS
  IAM Role authentication method,
  [consult this AWS documentation page](https://aws.amazon.com/blogs/security/understanding-the-api-options-for-securely-delegating-access-to-your-aws-account/).
  """,
              min_expiration_seconds=900,  # 15 minutes
              max_expiration_seconds=129600,  # 36 hours
              default_expiration_seconds=DEFAULT_STS_TOKEN_EXPIRATION,  # 12 hours
              config_class=FederationTokenAuthenticationConfig,
          ),
      ],
      resource_types=[
          ResourceTypeModel(
              name="Generic AWS resource",
              resource_type=AWS_RESOURCE_TYPE,
              description="""
  Multi-purpose AWS resource type. It allows Stack Components to use the connector
  to connect to any AWS service. When used by Stack Components, they are provided
  a generic Python boto3 session instance pre-configured with AWS credentials.
  This session can then be used to create boto3 clients for any particular AWS
  service.
  
  This generic AWS resource type is meant to be used with Stack Components that
  are not represented by other, more specific resource type, like S3 buckets,
  Kubernetes clusters or Docker registries. It should be accompanied by a matching
  set of AWS permissions that allow access to the set of remote resources required
  by the client(s).
  
  The resource name represents the AWS region that the connector is authorized to
  access.
  """,
              auth_methods=AWSAuthenticationMethods.values(),
              # Don't request an AWS specific resource instance ID, given that
              # the connector provides a generic boto3 session instance.
              supports_instances=False,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/connectors/aws/aws.png",
              emoji=":large_orange_diamond:",
          ),
          ResourceTypeModel(
              name="AWS S3 bucket",
              resource_type=S3_RESOURCE_TYPE,
              description="""
  Allows users to connect to S3 buckets. When used by Stack Components, they
  are provided a pre-configured boto3 S3 client instance.
  
  The configured credentials must have at least the following
  [AWS IAM permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html)
  associated with the [ARNs of S3 buckets](https://docs.aws.amazon.com/AmazonS3/latest/userguide/s3-arn-format.html)
  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:GetBucketVersioning`
  - `s3:ListBucketVersions`
  - `s3:DeleteObjectVersion`
  
  If set, the resource name must identify an S3 bucket using one of the following
  formats:
  
  - S3 bucket URI (canonical resource name): `s3://{bucket-name}`
  - S3 bucket ARN: `arn:aws:s3:::{bucket-name}`
  - S3 bucket name: `{bucket-name}`
  """,
              auth_methods=AWSAuthenticationMethods.values(),
              # Request an S3 bucket to be configured in the
              # connector or provided by the consumer
              supports_instances=True,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/artifact_store/aws.png",
              emoji=":package:",
          ),
          ResourceTypeModel(
              name="AWS EKS Kubernetes cluster",
              resource_type=KUBERNETES_CLUSTER_RESOURCE_TYPE,
              description="""
  Allows users to access an EKS cluster as a standard Kubernetes cluster
  resource. When used by Stack Components, they are provided a
  pre-authenticated python-kubernetes client instance.
  
  The configured credentials must have at least the following
  [AWS IAM permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html)
  associated with the [ARNs of EKS clusters](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html)
  that the connector will be allowed to access (e.g.
  `arn:aws:eks:{region}:{account}:cluster/*` represents all the EKS clusters
  available in the target AWS region).
  
  - `eks:ListClusters`
  - `eks:DescribeCluster`
  
  In addition to the above permissions, if the credentials are not associated
  with the same IAM user or role that created the EKS cluster, the IAM principal
  must be manually added to the EKS cluster's `aws-auth` ConfigMap, otherwise the
  Kubernetes client will not be allowed to access the cluster's resources. This
  makes it more challenging to use the AWS Implicit and AWS Federation Token
  authentication methods for this resource. For more information,
  [see this documentation](https://docs.aws.amazon.com/eks/latest/userguide/add-user-role.html).
  
  If set, the resource name must identify an EKS cluster using one of the
  following formats:
  
  - EKS cluster name (canonical resource name): `{cluster-name}`
  - EKS cluster ARN: `arn:aws:eks:{region}:{account}:cluster/{cluster-name}`
  
  EKS cluster names are region scoped. The connector can only be used to access
  EKS clusters in the AWS region that it is configured to use.
  """,
              auth_methods=AWSAuthenticationMethods.values(),
              # Request an EKS cluster name to be configured in the
              # connector or provided by the consumer
              supports_instances=True,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/orchestrator/kubernetes.png",
              emoji=":cyclone:",
          ),
          ResourceTypeModel(
              name="AWS ECR container registry",
              resource_type=DOCKER_REGISTRY_RESOURCE_TYPE,
              description="""
  Allows users to access one or more ECR repositories as a standard Docker
  registry resource. When used by Stack Components, they are provided a
  pre-authenticated python-docker client instance.
  
  The configured credentials must have at least the following
  [AWS IAM permissions](https://docs.aws.amazon.com/IAM/latest/UserGuide/access_policies.html)
  associated with the [ARNs of one or more ECR repositories](https://docs.aws.amazon.com/IAM/latest/UserGuide/reference-arns.html)
  that the connector will be allowed to access (e.g.
  `arn:aws:ecr:{region}:{account}:repository/*` represents all the ECR
  repositories available in the target AWS region).
  
  - `ecr:DescribeRegistry`
  - `ecr:DescribeRepositories`
  - `ecr:ListRepositories`
  - `ecr:BatchGetImage`
  - `ecr:DescribeImages`
  - `ecr:BatchCheckLayerAvailability`
  - `ecr:GetDownloadUrlForLayer`
  - `ecr:InitiateLayerUpload`
  - `ecr:UploadLayerPart`
  - `ecr:CompleteLayerUpload`
  - `ecr:PutImage`
  - `ecr:GetAuthorizationToken`
  
  This resource type is not scoped to a single ECR repository. Instead,
  a connector configured with this resource type will grant access to all the
  ECR repositories that the credentials are allowed to access under the configured
  AWS region (i.e. all repositories under the Docker registry URL
  `https://{account-id}.dkr.ecr.{region}.amazonaws.com`).
  
  The resource name associated with this resource type uniquely identifies an ECR
  registry using one of the following formats (the repository name is ignored,
  only the registry URL/ARN is used):
              
  - ECR repository URI (canonical resource name):
  `[https://]{account}.dkr.ecr.{region}.amazonaws.com[/{repository-name}]`
  - ECR repository ARN: `arn:aws:ecr:{region}:{account-id}:repository[/{repository-name}]`
  
  ECR repository names are region scoped. The connector can only be used to access
  ECR repositories in the AWS region that it is configured to use.
  """,
              auth_methods=AWSAuthenticationMethods.values(),
              # Does not support instances, given that the connector
              # provides access to all permitted ECR repositories under the
              # same ECR registry in the configured AWS region.
              supports_instances=False,
              logo_url="https://public-flavor-logos.s3.eu-central-1.amazonaws.com/container_registry/docker.png",
              emoji=":whale:",
          ),
      ],
  )
  
  
  class AWSServiceConnector(ServiceConnector):
      """AWS service connector."""
  
      config: AWSBaseConfig
  
      _account_id: Optional[str] = None
      _session_cache: Dict[
          Tuple[str, Optional[str], Optional[str]],
          Tuple[boto3.Session, Optional[datetime.datetime]],
      ] = {}
  
      @classmethod
      def _get_connector_type(cls) -> ServiceConnectorTypeModel:
          """Get the service connector type specification.
  
          Returns:
              The service connector type specification.
          """
          return AWS_SERVICE_CONNECTOR_TYPE_SPEC
  
      @property
      def account_id(self) -> str:
          """Get the AWS account ID.
  
          Returns:
              The AWS account ID.
  
          Raises:
              AuthorizationException: If the AWS account ID could not be
                  determined.
          """
          if self._account_id is None:
              logger.debug("Getting account ID from AWS...")
              try:
                  session, _ = self.get_boto3_session(self.auth_method)
                  sts_client = session.client("sts")
                  response = sts_client.get_caller_identity()
              except (ClientError, BotoCoreError) as e:
                  raise AuthorizationException(
                      f"Failed to fetch the AWS account ID: {e}"
                  ) from e
  
              self._account_id = response["Account"]
  
          return self._account_id
  
      def get_boto3_session(
          self,
          auth_method: str,
          resource_type: Optional[str] = None,
          resource_id: Optional[str] = None,
      ) -> Tuple[boto3.Session, Optional[datetime.datetime]]:
          """Get a boto3 session for the specified resource.
  
          Args:
              auth_method: The authentication method to use.
              resource_type: The resource type to get a boto3 session for.
              resource_id: The resource ID to get a boto3 session for.
  
          Returns:
              A boto3 session for the specified resource and its expiration
              timestamp, if applicable.
          """
          # We maintain a cache of all sessions to avoid re-authenticating
          # multiple times for the same resource
          key = (auth_method, resource_type, resource_id)
          if key in self._session_cache:
              session, expires_at = self._session_cache[key]
              if expires_at is None:
                  return session, None
  
              # Refresh expired sessions
              now = utc_now_tz_aware()
              expires_at = expires_at.replace(tzinfo=datetime.timezone.utc)
              # check if the token expires in the near future
              if expires_at > now + datetime.timedelta(
                  minutes=BOTO3_SESSION_EXPIRATION_BUFFER
              ):
                  return session, expires_at
  
          logger.debug(
              f"Creating boto3 session for auth method '{auth_method}', "
              f"resource type '{resource_type}' and resource ID "
  