auth

This library implements various methods for working with the Google IAM / auth APIs. This includes authenticating for the purpose of using other Google APIs, managing service accounts and public keys, URL-signing blobs, etc.

Installation

$ pip install --upgrade gcloud-aio-auth

Usage

from gcloud.aio.auth import IamClient
from gcloud.aio.auth import IapToken
from gcloud.aio.auth import Token


client = IamClient()
pubkeys = await client.list_public_keys()

iap_token = IapToken('https://your.service.url.com')
print(await iap_token.get())

token = Token()
print(await token.get())

```

The IapToken constructor accepts the following optional arguments:

  • service_file: path to a service account, authorized user file, or any other application credentials. Alternatively, you can pass a file-like object, like an io.StringIO instance, in case your credentials are not stored in a file but in memory. If omitted, will attempt to find one on your path or fallback to generating a token from GCE metadata.

  • session: an aiohttp.ClientSession instance to be used for all requests. If omitted, a default session will be created. If you use the default session, you may be interested in using IapToken() as a context manager (async with IapToken(..) as token:) or explicitly calling the IapToken.close() method to ensure the session is cleaned up appropriately.

  • impersonating_service_account: an optional string denoting a GCP service account which takes the form of an email address. Only valid (and required!) for authentication with a project’s authorized users. Impersonating a service account is required when generating an ID token in this case.

The Token constructor accepts the following optional arguments:

  • service_file: path to a service account authorized user file, or any other application credentials. Alternatively, you can pass a file-like object, like an io.StringIO instance, in case your credentials are not stored in a file but in memory. If omitted, will attempt to find one on your path or fallback to generating a token from GCE metadata.

  • session: an aiohttp.ClientSession instance to be used for all requests. If omitted, a default session will be created. If you use the default session, you may be interested in using Token() as a context manager (async with Token(..) as token:) or explicitly calling the Token.close() method to ensure the session is cleaned up appropriately.

  • scopes: an optional list of GCP scopes for which to generate our token. Only valid (and required!) for service account authentication.

  • target_principal: The service account to generate the access token for. The iam.serviceAccounts.getAccessToken permission on that service account is required.

  • delegates: The sequence of service accounts in a delegation chain. This field is required for delegated requests. Each service account must be granted the roles/iam.serviceAccountTokenCreator role on its next service account in the chain. The last service account in the chain must be granted the roles/iam.serviceAccountTokenCreator role on the service account that is specified in the target_principal.

CLI

This project can also be used to help you manually authenticate to test GCP routes, eg. we can list our project’s uptime checks with a tool such as curl:

# using default application credentials
curl       -H "Authorization: Bearer $(python3 -c 'from gcloud.rest.auth import Token; print(Token().get())')"       "https://monitoring.googleapis.com/v3/projects/PROJECT_ID/uptimeCheckConfigs"

# using a service account (make sure to provide a scope!)
export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service.json
curl       -H "Authorization: Bearer $(python3 -c 'from gcloud.rest.auth import Token; print(Token(scopes=["'"https://www.googleapis.com/auth/cloud-platform"'"]).get())')"       "https://monitoring.googleapis.com/v3/projects/PROJECT_ID/uptimeCheckConfigs"

# using legacy account credentials
export GOOGLE_APPLICATION_CREDENTIALS=~/.config/gcloud/legacy_credentials/EMAIL@DOMAIN.TLD/adc.json
curl       -H "Authorization: Bearer $(python3 -c 'from gcloud.rest.auth import Token; print(Token().get())')"       "https://monitoring.googleapis.com/v3/projects/PROJECT_ID/uptimeCheckConfigs"

Similarly it can be used to quickly test your IAP-secured endpoints:

# using default application credentials
curl       -H "Authorization: Bearer $(python3 -c 'from gcloud.rest.auth import IapToken; print(IapToken(APP_URL, impersonating_service_account=SA))')"       APP_URL

Submodules

Package Contents

Classes

IamClient

IapToken

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

Token

GCP OAuth 2.0 access token.

Functions

decode(payload)

Modified Base64 for URL variants exist, where the + and / characters of

encode(payload)

Modified Base64 for URL variants exist, where the + and / characters of

Attributes

BUILD_GCLOUD_REST

__version__

auth.BUILD_GCLOUD_REST
class auth.IamClient(service_file=None, session=None, token=None)
Parameters:

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

  • session (Optional[requests.Session]) –

  • token (Optional[auth.token.Token]) –

property service_account_email: str | None
Return type:

Optional[str]

async headers()
Return type:

Dict[str, str]

async get_public_key(key_id=None, key=None, service_account_email=None, project=None, session=None, timeout=10)
Parameters:

  • key_id (Optional[str]) –

  • key (Optional[str]) –

  • service_account_email (Optional[str]) –

  • project (Optional[str]) –

  • session (Optional[requests.Session]) –

  • timeout (int) –

Return type:

Dict[str, str]

async list_public_keys(service_account_email=None, project=None, session=None, timeout=10)
Parameters:

  • service_account_email (Optional[str]) –

  • project (Optional[str]) –

  • session (Optional[requests.Session]) –

  • timeout (int) –

Return type:

List[Dict[str, str]]

async sign_blob(payload, service_account_email=None, delegates=None, session=None, timeout=10)
Parameters:

  • payload (Optional[Union[str, bytes]]) –

  • service_account_email (Optional[str]) –

  • delegates (Optional[List[str]]) –

  • session (Optional[requests.Session]) –

  • timeout (int) –

Return type:

Dict[str, str]

async close()
Return type:

None

async __aenter__()
Return type:

IamClient

async __aexit__(*args)
Parameters:

args (Any) –

Return type:

None

class auth.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
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

class auth.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
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

auth.decode(payload)

Modified Base64 for URL variants exist, where the + and / characters of standard Base64 are respectively replaced by - and _.

See https://en.wikipedia.org/wiki/Base64#URL_applications

Parameters:

payload (str) –

Return type:

bytes

auth.encode(payload)

Modified Base64 for URL variants exist, where the + and / characters of standard Base64 are respectively replaced by - and _.

See https://en.wikipedia.org/wiki/Base64#URL_applications

Parameters:

payload (Union[bytes, str]) –

Return type:

bytes

auth.__version__