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.
  """GCP Service Connector.
  
  The GCP Service Connector implements various authentication methods for GCP
  services:
  
  - Explicit GCP service account key
  
  """
  
  import base64
  import datetime
  import json
  import os
  import re
  import shutil
  import subprocess
  import tempfile
  from typing import Any, Dict, List, Optional, Tuple
  
  import google.api_core.exceptions
  import google.auth
  import google.auth.exceptions
  import requests
  from google.auth import aws as gcp_aws
  from google.auth import external_account as gcp_external_account
  from google.auth import (
      impersonated_credentials as gcp_impersonated_credentials,
  )
  from google.auth._default import (
      _AWS_SUBJECT_TOKEN_TYPE,
      _get_external_account_credentials,
  )
  from google.auth.transport.requests import Request
  from google.cloud import artifactregistry_v1, container_v1, storage
  from google.cloud.location import locations_pb2
  from google.oauth2 import credentials as gcp_credentials
  from google.oauth2 import service_account as gcp_service_account
  from pydantic import Field, field_validator, model_validator
  
  from zenml.constants import (
      DOCKER_REGISTRY_RESOURCE_TYPE,
      KUBERNETES_CLUSTER_RESOURCE_TYPE,
  )
  from zenml.exceptions import AuthorizationException
  from zenml.integrations.gcp import (
      GCP_CONNECTOR_TYPE,
      GCP_RESOURCE_TYPE,
      GCS_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.pydantic_utils import before_validator_handler
  from zenml.utils.secret_utils import PlainSerializedSecretStr
  from zenml.utils.time_utils import to_utc_timezone, utc_now
  
  logger = get_logger(__name__)
  
  GKE_KUBE_API_TOKEN_EXPIRATION = 60
  DEFAULT_IMPERSONATE_TOKEN_EXPIRATION = 3600  # 1 hour
  GCP_SESSION_EXPIRATION_BUFFER = 15  # 15 minutes
  
  
  class GCPUserAccountCredentials(AuthenticationConfig):
      """GCP user account credentials."""
  
      user_account_json: PlainSerializedSecretStr = Field(
          title="GCP User Account Credentials JSON optionally base64 encoded.",
      )
  
      generate_temporary_tokens: bool = Field(
          default=True,
          title="Generate temporary OAuth 2.0 tokens",
          description="Whether to generate temporary OAuth 2.0 tokens from the "
          "user account credentials JSON. If set to False, the connector will "
          "distribute the user account credentials JSON to clients instead.",
      )
  
      @model_validator(mode="before")
      @classmethod
      @before_validator_handler
      def validate_user_account_dict(
          cls, data: Dict[str, Any]
      ) -> Dict[str, Any]:
          """Convert the user account credentials to JSON if given in dict format.
  
          Args:
              data: The configuration values.
  
          Returns:
              The validated configuration values.
  
          Raises:
              ValueError: If the user account credentials JSON is invalid.
          """
          user_account_json = data.get("user_account_json")
          if isinstance(user_account_json, dict):
              data["user_account_json"] = json.dumps(data["user_account_json"])
          elif isinstance(user_account_json, str):
              # Check if the user account JSON is base64 encoded and decode it
              if re.match(r"^[A-Za-z0-9+/=]+$", user_account_json):
                  try:
                      data["user_account_json"] = base64.b64decode(
                          user_account_json
                      ).decode("utf-8")
                  except Exception as e:
                      raise ValueError(
                          f"Failed to decode base64 encoded user account JSON: {e}"
                      )
          return data
  
      @field_validator("user_account_json")
      @classmethod
      def validate_user_account_json(
          cls, value: PlainSerializedSecretStr
      ) -> PlainSerializedSecretStr:
          """Validate the user account credentials JSON.
  
          Args:
              value: The user account credentials JSON.
  
          Returns:
              The validated user account credentials JSON.
  
          Raises:
              ValueError: If the user account credentials JSON is invalid.
          """
          try:
              user_account_info = json.loads(value.get_secret_value())
          except json.JSONDecodeError as e:
              raise ValueError(
                  f"GCP user account credentials is not a valid JSON: {e}"
              )
  
          # Check that all fields are present
          required_fields = [
              "type",
              "refresh_token",
              "client_secret",
              "client_id",
          ]
          # Compute missing fields
          missing_fields = set(required_fields) - set(user_account_info.keys())
          if missing_fields:
              raise ValueError(
                  f"GCP user account credentials JSON is missing required "
                  f"fields: {', '.join(list(missing_fields))}"
              )
  
          if user_account_info["type"] != "authorized_user":
              raise ValueError(
                  "The JSON does not contain GCP user account credentials. The "
                  f'"type" field is set to {user_account_info["type"]} '
                  "instead of 'authorized_user'."
              )
  
          return value
  
  
  class GCPServiceAccountCredentials(AuthenticationConfig):
      """GCP service account credentials."""
  
      service_account_json: PlainSerializedSecretStr = Field(
          title="GCP Service Account Key JSON optionally base64 encoded.",
      )
  
      generate_temporary_tokens: bool = Field(
          default=True,
          title="Generate temporary OAuth 2.0 tokens",
          description="Whether to generate temporary OAuth 2.0 tokens from the "
          "service account key JSON. If set to False, the connector will "
          "distribute the service account key JSON to clients instead.",
      )
  
      @model_validator(mode="before")
      @classmethod
      @before_validator_handler
      def validate_service_account_dict(
          cls, data: Dict[str, Any]
      ) -> Dict[str, Any]:
          """Convert the service account credentials to JSON if given in dict format.
  
          Args:
              data: The configuration values.
  
          Returns:
              The validated configuration values.
  
          Raises:
              ValueError: If the service account credentials JSON is invalid.
          """
          service_account_json = data.get("service_account_json")
          if isinstance(service_account_json, dict):
              data["service_account_json"] = json.dumps(
                  data["service_account_json"]
              )
          elif isinstance(service_account_json, str):
              # Check if the service account JSON is base64 encoded and decode it
              if re.match(r"^[A-Za-z0-9+/=]+$", service_account_json):
                  try:
                      data["service_account_json"] = base64.b64decode(
                          service_account_json
                      ).decode("utf-8")
                  except Exception as e:
                      raise ValueError(
                          f"Failed to decode base64 encoded service account JSON: {e}"
                      )
  
          return data
  
      @field_validator("service_account_json")
      @classmethod
      def validate_service_account_json(
          cls, value: PlainSerializedSecretStr
      ) -> PlainSerializedSecretStr:
          """Validate the service account credentials JSON.
  
          Args:
              value: The service account credentials JSON.
  
          Returns:
              The validated service account credentials JSON.
  
          Raises:
              ValueError: If the service account credentials JSON is invalid.
          """
          try:
              service_account_info = json.loads(value.get_secret_value())
          except json.JSONDecodeError as e:
              raise ValueError(
                  f"GCP service account credentials is not a valid JSON: {e}"
              )
  
          # Check that all fields are present
          required_fields = [
              "type",
              "project_id",
              "private_key_id",
              "private_key",
              "client_email",
              "client_id",
              "auth_uri",
              "token_uri",
              "auth_provider_x509_cert_url",
              "client_x509_cert_url",
          ]
          # Compute missing fields
          missing_fields = set(required_fields) - set(
              service_account_info.keys()
          )
          if missing_fields:
              raise ValueError(
                  f"GCP service account credentials JSON is missing required "
                  f"fields: {', '.join(list(missing_fields))}"
              )
  
          if service_account_info["type"] != "service_account":
              raise ValueError(
                  "The JSON does not contain GCP service account credentials. "
                  f'The "type" field is set to {service_account_info["type"]} '
                  "instead of 'service_account'."
              )
  
          return value
  
  
  class GCPExternalAccountCredentials(AuthenticationConfig):
      """GCP external account credentials."""
  
      external_account_json: PlainSerializedSecretStr = Field(
          title="GCP External Account JSON optionally base64 encoded.",
      )
  
      generate_temporary_tokens: bool = Field(
          default=True,
          title="Generate temporary OAuth 2.0 tokens",
          description="Whether to generate temporary OAuth 2.0 tokens from the "
          "external account key JSON. If set to False, the connector will "
          "distribute the external account JSON to clients instead.",
      )
  
      @model_validator(mode="before")
      @classmethod
      @before_validator_handler
      def validate_service_account_dict(
          cls, data: Dict[str, Any]
      ) -> Dict[str, Any]:
          """Convert the external account credentials to JSON if given in dict format.
  
          Args:
              data: The configuration values.
  
          Returns:
              The validated configuration values.
  
          Raises:
              ValueError: If the external account credentials JSON is invalid.
          """
          external_account_json = data.get("external_account_json")
          if isinstance(external_account_json, dict):
              data["external_account_json"] = json.dumps(
                  data["external_account_json"]
              )
          elif isinstance(external_account_json, str):
              # Check if the external account JSON is base64 encoded and decode it
              if re.match(r"^[A-Za-z0-9+/=]+$", external_account_json):
                  try:
                      data["external_account_json"] = base64.b64decode(
                          external_account_json
                      ).decode("utf-8")
                  except Exception as e:
                      raise ValueError(
                          f"Failed to decode base64 encoded external account JSON: {e}"
                      )
  
          return data
  
      @field_validator("external_account_json")
      @classmethod
      def validate_external_account_json(
          cls, value: PlainSerializedSecretStr
      ) -> PlainSerializedSecretStr:
          """Validate the external account credentials JSON.
  
          Args:
              value: The external account credentials JSON.
  
          Returns:
              The validated external account credentials JSON.
  
          Raises:
              ValueError: If the external account credentials JSON is invalid.
          """
          try:
              external_account_info = json.loads(value.get_secret_value())
          except json.JSONDecodeError as e:
              raise ValueError(
                  f"GCP external account credentials is not a valid JSON: {e}"
              )
  
          # Check that all fields are present
          required_fields = [
              "type",
              "subject_token_type",
              "token_url",
          ]
          # Compute missing fields
          missing_fields = set(required_fields) - set(
              external_account_info.keys()
          )
          if missing_fields:
              raise ValueError(
                  f"GCP external account credentials JSON is missing required "
                  f"fields: {', '.join(list(missing_fields))}"
              )
  
          if external_account_info["type"] != "external_account":
              raise ValueError(
                  "The JSON does not contain GCP external account credentials. "
                  f'The "type" field is set to {external_account_info["type"]} '
                  "instead of 'external_account'."
              )
  
          return value
  
  
  class GCPOAuth2Token(AuthenticationConfig):
      """GCP OAuth 2.0 token credentials."""
  
      token: PlainSerializedSecretStr = Field(
          title="GCP OAuth 2.0 Token",
      )
  
  
  class GCPBaseConfig(AuthenticationConfig):
      """GCP base configuration."""
  
      @property
      def gcp_project_id(self) -> str:
          """Get the GCP project ID.
  
          This method must be implemented by subclasses to ensure that the GCP
          project ID is always available.
  
          Raises:
              NotImplementedError: If the method is not implemented.
          """
          raise NotImplementedError
  
  
  class GCPBaseProjectIDConfig(GCPBaseConfig):
      """GCP base configuration with included project ID."""
  
      project_id: str = Field(
          title="GCP Project ID where the target resource is located.",
      )
  
      @property
      def gcp_project_id(self) -> str:
          """Get the GCP project ID.
  
          Returns:
              The GCP project ID.
          """
          return self.project_id
  
  
  class GCPUserAccountConfig(GCPBaseProjectIDConfig, GCPUserAccountCredentials):
      """GCP user account configuration."""
  
  
  class GCPServiceAccountConfig(GCPBaseConfig, GCPServiceAccountCredentials):
      """GCP service account configuration."""
  
      project_id: Optional[str] = None
  
      @property
      def gcp_project_id(self) -> str:
          """Get the GCP project ID.
  
          When a service account JSON is provided, the project ID can be extracted
          from it instead of being provided explicitly.
  
          Returns:
              The GCP project ID.
          """
          if self.project_id is None:
              self.project_id = json.loads(
                  self.service_account_json.get_secret_value()
              )["project_id"]
              # Guaranteed by the field validator
              assert self.project_id is not None
  
          return self.project_id
  
  
  class GCPExternalAccountConfig(
      GCPBaseProjectIDConfig, GCPExternalAccountCredentials
  ):
      """GCP external account configuration."""
  
  
  class GCPOAuth2TokenConfig(GCPBaseProjectIDConfig, GCPOAuth2Token):
      """GCP OAuth 2.0 configuration."""
  
      service_account_email: Optional[str] = Field(
          default=None,
          title="GCP Service Account Email",
          description="The email address of the service account that signed the "
          "token. If not provided, the token is assumed to be issued for a user "
          "account.",
      )
  
  
  class GCPServiceAccountImpersonationConfig(GCPServiceAccountConfig):
      """GCP service account impersonation configuration."""
  
      target_principal: str = Field(
          title="GCP Service Account Email to impersonate",
      )
  
  
  class GCPAuthenticationMethods(StrEnum):
      """GCP Authentication methods."""
  
      IMPLICIT = "implicit"
      USER_ACCOUNT = "user-account"
      SERVICE_ACCOUNT = "service-account"
      EXTERNAL_ACCOUNT = "external-account"
      OAUTH2_TOKEN = "oauth2-token"
      IMPERSONATION = "impersonation"
  
  
  try:
      from google.auth.aws import _DefaultAwsSecurityCredentialsSupplier
  
      class ZenMLAwsSecurityCredentialsSupplier(
          _DefaultAwsSecurityCredentialsSupplier  # type: ignore[misc]
      ):
          """An improved version of the GCP external account credential supplier for AWS.
  
          The original GCP external account credential supplier only provides
          rudimentary support for extracting AWS credentials from environment
          variables or the AWS metadata service. This version improves on that by
          using the boto3 library itself (if available), which uses the entire range
          of implicit authentication features packed into it.
  
          Without this improvement, `sts.AssumeRoleWithWebIdentity` authentication is
          not supported for EKS pods and the EC2 attached role credentials are
          used instead (see: https://medium.com/@derek10cloud/gcp-workload-identity-federation-doesnt-yet-support-eks-irsa-in-aws-a3c71877671a).
          """
  
          def get_aws_security_credentials(
              self, context: Any, request: Any
          ) -> gcp_aws.AwsSecurityCredentials:
              """Get the security credentials from the local environment.
  
              This method is a copy of the original method from the
              `google.auth.aws._DefaultAwsSecurityCredentialsSupplier` class. It has
              been modified to use the boto3 library to extract the AWS credentials
              from the local environment.
  
              Args:
                  context: The context to use to get the security credentials.
                  request: The request to use to get the security credentials.
  
              Returns:
                  The AWS temporary security credentials.
              """
              try:
                  import boto3
  
                  session = boto3.Session()
                  credentials = session.get_credentials()
                  if credentials is not None:
                      creds = credentials.get_frozen_credentials()
                      return gcp_aws.AwsSecurityCredentials(
                          creds.access_key,
                          creds.secret_key,
                          creds.token,
                      )
              except ImportError:
                  pass
  
              logger.debug(
                  "Failed to extract AWS credentials from the local environment "
                  "using the boto3 library. Falling back to the original "
                  "implementation."
              )
  
              return super().get_aws_security_credentials(context, request)
  
          def get_aws_region(self, context: Any, request: Any) -> str:
              """Get the AWS region from the local environment.
  
              This method is a copy of the original method from the
              `google.auth.aws._DefaultAwsSecurityCredentialsSupplier` class. It has
              been modified to use the boto3 library to extract the AWS
              region from the local environment.
  
              Args:
                  context: The context to use to get the security credentials.
                  request: The request to use to get the security credentials.
  
              Returns:
                  The AWS region.
              """
              try:
                  import boto3
  
                  session = boto3.Session()
                  if session.region_name:
                      return session.region_name  # type: ignore[no-any-return]
              except ImportError:
                  pass
  
              logger.debug(
                  "Failed to extract AWS region from the local environment "
                  "using the boto3 library. Falling back to the original "
                  "implementation."
              )
  
              return super().get_aws_region(  # type: ignore[no-any-return]
                  context, request
              )
  
  except ImportError:
      # The `google.auth.aws._DefaultAwsSecurityCredentialsSupplier`
      # class has been introduced in the `google-auth` library version 2.29.0.
      # Before that, the AWS logic was part of the `google.auth.awsCredentials`
      # class itself.
      ZenMLAwsSecurityCredentialsSupplier = None  # type: ignore[assignment,misc]
      pass
  
  
  class ZenMLGCPAWSExternalAccountCredentials(gcp_aws.Credentials):  # type: ignore[misc]
      """An improved version of the GCP external account credential for AWS.
  
      The original GCP external account credential only provides rudimentary
      support for extracting AWS credentials from environment variables or the
      AWS metadata service. This version improves on that by using the boto3
      library itself (if available), which uses the entire range of implicit
      authentication features packed into it.
  
      Without this improvement, `sts.AssumeRoleWithWebIdentity` authentication is
      not supported for EKS pods and the EC2 attached role credentials are
      used instead (see: https://medium.com/@derek10cloud/gcp-workload-identity-federation-doesnt-yet-support-eks-irsa-in-aws-a3c71877671a).
  
      IMPORTANT: subclassing this class only works with the `google-auth` library
      version lower than 2.29.0. Starting from version 2.29.0, the AWS logic
      has been moved to a separate `google.auth.aws._DefaultAwsSecurityCredentialsSupplier`
      class that can be subclassed instead and supplied as the
      `aws_security_credentials_supplier` parameter to the
      `google.auth.aws.Credentials` class.
      """
  
      def _get_security_credentials(
          self, request: Any, imdsv2_session_token: Any
      ) -> Dict[str, Any]:
          """Get the security credentials from the local environment.
  
          This method is a copy of the original method from the
          `google.auth._default` module. It has been modified to use the boto3
          library to extract the AWS credentials from the local environment.
  
          Args:
              request: The request to use to get the security credentials.
              imdsv2_session_token: The IMDSv2 session token to use to get the
                  security credentials.
  
          Returns:
              The AWS temporary security credentials.
          """
          try:
              import boto3
  
              session = boto3.Session()
              credentials = session.get_credentials()
              if credentials is not None:
                  creds = credentials.get_frozen_credentials()
                  return {
                      "access_key_id": creds.access_key,
                      "secret_access_key": creds.secret_key,
                      "security_token": creds.token,
                  }
          except ImportError:
              pass
  
          logger.debug(
              "Failed to extract AWS credentials from the local environment "
              "using the boto3 library. Falling back to the original "
              "implementation."
          )
  
          return super()._get_security_credentials(  # type: ignore[no-any-return]
              request, imdsv2_session_token
          )
  
  
  GCP_SERVICE_CONNECTOR_TYPE_SPEC = ServiceConnectorTypeModel(
      name="GCP Service Connector",
      connector_type=GCP_CONNECTOR_TYPE,
      description="""
  The ZenML GCP Service Connector facilitates the authentication and access to
  managed GCP services and resources. These encompass a range of resources,
  including GCS buckets, GCR container repositories and GKE clusters. The
  connector provides support for various authentication methods, including GCP
  user accounts, service accounts, short-lived OAuth 2.0 tokens and implicit
  authentication.
  
  To ensure heightened security measures, this connector always issues short-lived
  OAuth 2.0 tokens to clients instead of long-lived credentials unless explicitly
  configured to do otherwise. Furthermore, it includes automatic configuration and
  detection of credentials locally configured through the GCP CLI.
  
  This connector serves as a general means of accessing any GCP service by issuing
  OAuth 2.0 credential objects to clients. Additionally, the connector can handle
  specialized authentication for GCS, Docker and Kubernetes Python clients. It
  also allows for the configuration of local Docker and Kubernetes CLIs.
  
  The GCP Service Connector is part of the GCP 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-gcp]"` installs only prerequisites for the GCP
  Service Connector Type
  * `zenml integration install gcp` installs the entire GCP ZenML integration
  
  It is not required to install and set up [the GCP CLI](https://cloud.google.com/sdk/gcloud)
  on your local machine to use the GCP Service Connector to link Stack Components
  to GCP 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/gcp/gcp.png",
      emoji=":blue_circle:",
      auth_methods=[
          AuthenticationMethodModel(
              name="GCP Implicit Authentication",
              auth_method=GCPAuthenticationMethods.IMPLICIT,
              description="""
  Implicit authentication to GCP services using [Application Default Credentials](https://cloud.google.com/docs/authentication/provide-credentials-adc).
  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 (`GOOGLE_APPLICATION_CREDENTIALS`)
  - local ADC credential files set up by running `gcloud auth application-default
  login` (e.g. `~/.config/gcloud/application_default_credentials.json`).
  - GCP service account attached to the resource where the ZenML server is running.
  Only works when running the ZenML server on a GCP resource with an service
  account attached to it or when using Workload Identity (e.g. GKE cluster).
  
  This is the quickest and easiest way to authenticate to GCP 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 those set up on your machine (i.e. by running
  `gcloud auth application-default login` or setting the
  `GOOGLE_APPLICATION_CREDENTIALS` environment variable to point to a service
  account key JSON file).
  - when connected to a ZenML server, this method only works if the ZenML server
  is deployed in GCP and will use the service account attached to the GCP resource
  where the ZenML server is running (e.g. an GKE cluster). The service account
  permissions may need to be adjusted to allow listing and accessing/describing
  the GCP resources that the connector is configured to access.
  
  Note that the discovered credentials inherit the full set of permissions of the
  local GCP CLI credentials or service account attached to the ZenML server GCP
  workload. Depending on the extent of those permissions, this authentication
  method might not be suitable for production use, as it can lead to accidental
  privilege escalation. Instead, it is recommended to use the Service Account Key
  or Service Account Impersonation authentication methods to restrict the
  permissions that are granted to the connector clients.
  
  To find out more about Application Default Credentials,
  [see the GCP ADC documentation](https://cloud.google.com/docs/authentication/provide-credentials-adc).
  
  A GCP project is required and the connector may only be used to access GCP
  resources in the specified project. When used remotely in a GCP workload, the
  configured project has to be the same as the project of the attached service
  account.
  """,
              config_class=GCPBaseProjectIDConfig,
          ),
          AuthenticationMethodModel(
              name="GCP User Account",
              auth_method=GCPAuthenticationMethods.USER_ACCOUNT,
              description="""
  Use a GCP user account and its credentials to authenticate to GCP services.
  
  This method requires GCP user account credentials like those generated by
  the `gcloud auth application-default login` command.
  
  By default, the GCP connector generates temporary OAuth 2.0 tokens from the user
  account credentials and distributes them to clients. The tokens have a limited
  lifetime of 1 hour. This behavior can be disabled by setting the
  `generate_temporary_tokens` configuration option to `False`, in which case, the
  connector will distribute the user account credentials JSON to clients instead
  (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 are granted the full set of permissions
  of the GCP user account. For production, it is recommended to use the GCP
  Service Account or GCP Service Account Impersonation authentication methods.
  
  A GCP project is required and the connector may only be used to access GCP
  resources in the specified project.
  
  If you already have the local GCP CLI set up with these credentials, they will
  be automatically picked up when auto-configuration is used.
  """,
              config_class=GCPUserAccountConfig,
          ),
          AuthenticationMethodModel(
              name="GCP Service Account",
              auth_method=GCPAuthenticationMethods.SERVICE_ACCOUNT,
              description="""
  Use a GCP service account and its credentials to authenticate to GCP services.
  This method requires a [GCP service account](https://cloud.google.com/iam/docs/service-account-overview)
  and [a service account key JSON](https://cloud.google.com/iam/docs/service-account-creds#key-types)
  created for it.
  
  By default, the GCP connector generates temporary OAuth 2.0 tokens from the
  service account credentials and distributes them to clients. The tokens have a
  limited lifetime of 1 hour. This behavior can be disabled by setting the
  `generate_temporary_tokens` configuration option to `False`, in which case, the
  connector will distribute the service account credentials JSON to clients
  instead (not recommended).
  
  A GCP project is required and the connector may only be used to access GCP
  resources in the specified project. If the `project_id` is not provided, the
  connector will use the one extracted from the service account key JSON.
  
  If you already have the GOOGLE_APPLICATION_CREDENTIALS environment variable
  configured to point to a service account key JSON file, it will be automatically
  picked up when auto-configuration is used.
  """,
              config_class=GCPServiceAccountConfig,
          ),
          AuthenticationMethodModel(
              name="GCP External Account",
              auth_method=GCPAuthenticationMethods.EXTERNAL_ACCOUNT,
              description="""
  Use [GCP workload identity federation](https://cloud.google.com/iam/docs/workload-identity-federation)
  to authenticate to GCP services using AWS IAM credentials, Azure Active
  Directory credentials or generic OIDC tokens.
  
  This authentication method only requires a GCP workload identity external
  account JSON file that only contains the configuration for the external account
  without any sensitive credentials. It 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 GCP service account instead.
  
  This authentication method can be used to authenticate to GCP services using
  credentials from other cloud providers or identity providers. When used with
  workloads running on AWS or Azure, it involves automatically picking up
  credentials from the AWS IAM or Azure AD identity associated with the workload
  and using them to authenticate to GCP services. This means that the result
  depends on the environment where the ZenML server is deployed and is thus not
  fully reproducible.
  
  By default, the GCP connector generates temporary OAuth 2.0 tokens from the
  external account credentials and distributes them to clients. The tokens have a
  limited lifetime of 1 hour. This behavior can be disabled by setting the
  `generate_temporary_tokens` configuration option to `False`, in which case, the
  connector will distribute the external account credentials JSON to clients
  instead (not recommended).
  
  A GCP project is required and the connector may only be used to access GCP
  resources in the specified project. This project must be the same as the one
  for which the external account was configured.
  
  