auth.token

Google Cloud auth via service account file

Attributes

CustomFileError

GCE_METADATA_BASE

GCE_METADATA_HEADERS

GCE_ENDPOINT_PROJECT

GCE_ENDPOINT_TOKEN

GCE_ENDPOINT_ID_TOKEN

GCLOUD_ENDPOINT_GENERATE_ACCESS_TOKEN

GCLOUD_ENDPOINT_GENERATE_ID_TOKEN

REFRESH_HEADERS

Classes

Type

Create a collection of name/value pairs.

TokenResponse

BaseToken

GCP auth token base class.

Token

GCP OAuth 2.0 access token.

IapToken

An OpenID Connect ID token for a single IAP-secured service.

Functions

get_service_data(service)

Get the service data dictionary for the current auth method.

Module Contents

auth.token.CustomFileError: Any
auth.token.GCE_METADATA_BASE = 'http://metadata.google.internal/computeMetadata/v1'
auth.token.GCE_METADATA_HEADERS
auth.token.GCE_ENDPOINT_PROJECT = 'http://metadata.google.internal/computeMetadata/v1/project/project-id'
auth.token.GCE_ENDPOINT_TOKEN = 'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/token?recur...
auth.token.GCE_ENDPOINT_ID_TOKEN = 'http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?au...
auth.token.GCLOUD_ENDPOINT_GENERATE_ACCESS_TOKEN = 'https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{service_account}:generateAc...
auth.token.GCLOUD_ENDPOINT_GENERATE_ID_TOKEN = 'https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/{service_account}:generateIdToken'
auth.token.REFRESH_HEADERS
class auth.token.Type(*args, **kwds)

Bases: enum.Enum

Create a collection of name/value pairs.

Example enumeration:

>>> class Color(Enum):
...     RED = 1
...     BLUE = 2
...     GREEN = 3

Access them by:

  • attribute access:

    >>> Color.RED
    <Color.RED: 1>
    
  • value lookup:

    >>> Color(1)
    <Color.RED: 1>
    
  • name lookup:

    >>> Color['RED']
    <Color.RED: 1>
    

Enumerations can be iterated over, and know how many members they have:

>>> len(Color)
3
>>> list(Color)
[<Color.RED: 1>, <Color.BLUE: 2>, <Color.GREEN: 3>]

Methods can be added to enumerations, and members can have their own attributes – see the documentation for details.

AUTHORIZED_USER = 'authorized_user'
GCE_METADATA = 'gce_metadata'
SERVICE_ACCOUNT = 'service_account'
auth.token.get_service_data(service)

Get the service data dictionary for the current auth method.

This method is meant to match the official google.auth.default() method (or rather, the subset relevant to our use-case). Things such as the precedence order of various approaches MUST be maintained. It was last updated to match the following commit:

https://github.com/googleapis/google-auth-library-python/blob/6c1297c4d69ba40a8b9392775c17411253fcd73b/google/auth/_default.py#L504

Parameters:

service (Optional[Union[str, IO[AnyStr]]])

Return type:

Dict[str, Any]

class auth.token.TokenResponse
value: str
expires_in: int
class auth.token.BaseToken(service_file=None, session=None)

GCP auth token base class.

Parameters:
  • service_file (Optional[Union[str, IO[AnyStr]]])

  • session (Optional[requests.Session])

__metaclass__
service_data
session
access_token: str | None = None
access_token_duration = 0
access_token_acquired_at
acquiring: asyncio.Future[Any] | None = None
async get_project()
Return type:

Optional[str]

async get()
Return type:

Optional[str]

async ensure_token()
Return type:

None

abstract refresh(*, timeout)
Async:

Parameters:

timeout (int)

Return type:

TokenResponse

async acquire_access_token(timeout=10)
Parameters:

timeout (int)

Return type:

None

async close()
Return type:

None

async __aenter__()
Return type:

BaseToken

async __aexit__(*args)
Parameters:

args (Any)

Return type:

None

class auth.token.Token(service_file=None, session=None, scopes=None, target_principal=None, delegates=None)

Bases: BaseToken

GCP OAuth 2.0 access token.

Parameters:
  • service_file (Optional[Union[str, IO[AnyStr]]])

  • session (Optional[requests.Session])

  • scopes (Optional[List[str]])

  • target_principal (Optional[str])

  • delegates (Optional[List[str]])

default_token_ttl = 3600
scopes = ''
target_principal = None
delegates = None
async _refresh_authorized_user(timeout)
Parameters:

timeout (int)

Return type:

TokenResponse

async _refresh_gce_metadata(timeout)
Parameters:

timeout (int)

Return type:

TokenResponse

async _refresh_service_account(timeout)
Parameters:

timeout (int)

Return type:

TokenResponse

async _impersonate(token, *, timeout)
Parameters:
Return type:

TokenResponse

async refresh(*, timeout)
Parameters:

timeout (int)

Return type:

TokenResponse

class auth.token.IapToken(app_uri, service_file=None, session=None, impersonating_service_account=None)

Bases: BaseToken

An OpenID Connect ID token for a single IAP-secured service.

Parameters:
  • app_uri (str)

  • service_file (Optional[Union[str, IO[AnyStr]]])

  • session (Optional[requests.Session])

  • impersonating_service_account (Optional[str])

default_token_ttl = 3600
app_uri
service_account = None
async _get_iap_client_id(*, timeout)

Fetch the IAP client ID from the service URI.

If not logged in already, then we parse the OAuth redirect location to get the client ID. The redirect location is a header of the form:

For more details, see the GCP docs for programmatic IAP access: https://cloud.google.com/iap/docs/authentication-howto

Parameters:

timeout (int)

Return type:

str

async _refresh_authorized_user(iap_client_id, timeout)

Fetch IAP ID token by impersonating a service account.

https://cloud.google.com/iap/docs/authentication-howto#obtaining_an_oidc_token_in_all_other_cases

Parameters:
  • iap_client_id (str)

  • timeout (int)

Return type:

TokenResponse

async _refresh_gce_metadata(iap_client_id, timeout)

Fetch IAP ID token from the GCE metadata servers.

Note: The official documentation states that the URI be used for the audience but this is not the case. The typical audience value must be used as in other flavours of ID token fetching.

https://cloud.google.com/docs/authentication/get-id-token#metadata-server

Parameters:
  • iap_client_id (str)

  • timeout (int)

Return type:

TokenResponse

async _refresh_service_account(iap_client_id, timeout)
Parameters:
  • iap_client_id (str)

  • timeout (int)

Return type:

TokenResponse

async refresh(*, timeout)
Parameters:

timeout (int)

Return type:

TokenResponse