In the previous three article of the multi-cloud identity federation series we discussed about access token, identity token, how to differentiate them and how to exchange service identity between Google Cloud and Azure without exposing your keys and secrets. If you don’t know want I am referring to, make sure to catch up with the links above.

Not let’s see the Python implementation for your production applications, first from Azure environment to impersonate Google Cloud service account, then from GCP to impersonate an Azure App Registration.

1. From Azure environment : impersonate GCP service account

Generate the Azure access token

Use the Azure Identity library to generate an access token

You can use this option if your application is running in a compute instance that have access to the Azure Instance Metadata Service (IMDS). It’s the recommended method because you do not have to store secrets in the instance or in environment variables, but it’s not always available depending on your use case.

Note: azure.identity module is part of azure-identity package.

from azure.identity import DefaultAzureCredential
from azure.identity import AzureAuthorityHosts

default_credential = DefaultAzureCredential(
    authority=AzureAuthorityHosts.AZURE_PUBLIC_CLOUD
)
azure_access_token = default_credential.get_token(
    scopes=f"{os.environ['APPLICATION_ID']}.default"
)

b. Use the MSAL library with your client_id and client_secret

If you do not have access to IMDS, you can always expose the CLIENT_SECRET and CLIENT_ID (App ID) in the environment variables of your application or most preferably store and retrieve them in a Key Vault. You can then use the MSAL ConfidentialClientApplication to get your App Registration access token.

from msal import ConfidentialClientApplication

CLIENT_SECRET = os.environ["CLIENT_SECRET"]
TENANT_ID = os.environ["TENANT_ID"]
APPLICATION_ID = os.environ["APPLICATION_ID"]

app = ConfidentialClientApplication(
    client_id=APPLICATION_ID,
    client_credential=CLIENT_SECRET,
    authority=f"{AzureAuthorityHosts.AZURE_PUBLIC_CLOUD}/{TENANT_ID}"
)
azure_access_token = app.acquire_token_for_client(
    scopes=f"{APPLICATION_ID}/.default"
)["access_token"]

Use the Google’s STS Client to get a federated token via the Workload Identity Federation

The second step of the token exchange process is to request a short-lived token to Google STS API. Make sure to understand the parameters detailed in the article Part 2.

from google.oauth2.sts import Client
from google.auth.transport.requests import Request

GCP_PROJECT_NUMBER = os.environ["PROJECT_NUMER"]
GCP_PROJECT_ID = os.environ["GCP_PROJECT_ID"]
POOL_ID = os.environ["POOL_ID"]
PROVIDER_ID = os.environ["PROVIDER_ID"]

sts_client = Client(token_exchange_endpoint="https://sts.googleapis.com/v1/token")
response = sts_client.exchange_token(
    request=Request(),
    audience=f"//iam.googleapis.com/projects/{GCP_PROJECT_NUMBER}/locations/global/workloadIdentityPools/{POOL_ID}/providers/{PROVIDER_ID}",
    grant_type="urn:ietf:params:oauth:grant-type:token-exchange",
    subject_token=azure_access_token,
    scopes=["https://www.googleapis.com/auth/cloud-platform"],
    subject_token_type="urn:ietf:params:oauth:token-type:jwt",
    requested_token_type="urn:ietf:params:oauth:token-type:access_token"
)
sts_access_token = response["access_token"]

Impersonate the target service account with STS token

When you have your STS token (federated token) you can finally impersonate the target service account (assuming you gave the correct role to your Workload Identity PrincipalSet)

Create the target credential object

from google.oauth2.credentials import Credentials
from google.auth import impersonated_credentials

TARGET_SERVICE_ACCOUNT = os.environ["TARGET_SERVICE_ACCOUNT"]

sts_credentials = Credentials(token=sts_access_token)

credentials = impersonated_credentials.Credentials(
  source_credentials=sts_credentials,
  target_principal=TARGET_SERVICE_ACCOUNT,
  target_scopes = ["https://www.googleapis.com/auth/cloud-platform"],
  lifetime=500
)
credentials.refresh(Request())

Call your Google API (here BigQuery) from Azure environment

Now that the token exchange process is over, you can request any API that the target service account have access to by using the corresponding Client library (here it’s BigQuery).

from google.cloud import bigquery

client = bigquery.Client(credentials=credentials, project=GCP_PROJECT_ID)

# Here my TARGET_SERVICE_ACCOUNT has bigquery.jobUser role.
query = "SELECT CURRENT_DATE() as date"
query_job = client.query(query)  # Make an API request.

print("The query data:")
for row in query_job:
    print(row["date"])
# It works !

2. From Google Cloud : impersonate Azure App

Let’s see the Python implementation from the other perspective : impersonate an Azure App from GCP environment. This process is detailed in the Part 3 of the series. Make sure to read it to understand the process.

Implement a python Credential class from TokenCredential

Most of Microsoft client libraries can take a Credential instance as argument. Even if most of the time it’s a DefaultAzureCredential or ConfidentialClientApplication, you can create your own by implementing the TokenCredential interface. The class must implement the get_token method, that is called by the client library when authenticating.

Here we first perform the Google ID token generation by querying the Google Metadata Server, then we use the ConfidentialClientApplication with the ID token as client_assertion to get the federated token.

from azure.core.credentials import TokenCredential, AccessToken
from msal import ConfidentialClientApplication
from google.auth.transport.requests import Request
import time

class GoogleAssertionCredential(TokenCredential):

    def __init__(self, azure_client_id, azure_tenant_id, azure_authority_host):
        # create a confidential client application
        self.app = ConfidentialClientApplication(
            azure_client_id,
            client_credential={
                'client_assertion': self._get_google_id_token()
            },
            authority=f"{azure_authority_host}{azure_tenant_id}"
        )

    def _get_google_id_token(self) -> str:
                """Request an ID token to the Metadata Server"""
        response = Request()(
            f"{GOOGLE_METADATA_API}/instance/service-accounts/default/identity",
                        f"?audience=api://AzureADTokenExchange",
            method="GET",
            headers={"Metadata-Flavor": "Google"},
          )
                return response.data.decode("utf-8")

    def get_token(
        self,
        *scopes: str,
        claims: Optional[str] = None,
        tenant_id: Optional[str] = None,
        **kwargs: Any
    ) -> AccessToken:
        # get the token using the application
        token = self.app.acquire_token_for_client(scopes)
        if 'error' in token:
            raise Exception(token['error_description'])
        expires_on = time.time() + token['expires_in']
        # return an access token with the token string and expiration time
        return AccessToken(token['access_token'], int(expires_on))

Note: the token generation with Metadata Server will only work on an app deployed on GCP. If you want to test locally, you can use a service account file.

credentials = IDTokenCredentials.from_service_account_file(
    GOOGLE_APPLICATION_CREDENTIALS,
    target_audience="api://AzureADTokenExchange",
)
credentials.refresh(Request())
return credentials.token

Instantiate the GoogleAssertionCredential and query final Azure API

Finally you can request any API the Azure App registration have access to, to get your work done. Just instantiate the GoogleAssertionCredential with your target Azure App CLIENT_ID & TENANT_ID, and pass it to the client library (here it’s BlobServiceClient, assuming that the App registration have Contributor role in the Azure Storage Account)

CLIENT_ID = os.environ["CLIENT_ID"]
TENANT_ID = os.environ["TENANT_ID"]

creds = GoogleAssertionCredential(
    azure_client_id=CLIENT_ID,
    azure_tenant_id=TENANT_ID,
    azure_authority_host=AzureAuthorityHosts.AZURE_PUBLIC_CLOUD
)

STORAGE_ACCOUNT = os.environ["STORAGE_ACCOUNT"]
CONTAINER = os.environ["CONTAINER"]
# Here the App registration is Contributor of the Azure storage account
blob_service_client = BlobServiceClient(f"https://{STORAGE_ACCOUNT}.blob.core.windows.net", credential=creds)
container_client = blob_service_client.get_container_client(container=CONTAINER)
for blob in container_client.list_blob_names():
    print(blob)
        # It works !

We just saw how to concretely impersonate service identities between Google Cloud and Azure in your production with Python. Keep it mind the good practices :

no secret storage if no need to, there is the Metadata Server in both clouds
use the correct audience or scope for just what you need to do, so if the token leaks the thief will only be able to use it for the target service before the token expires (less than 1 hour)

We will see in the next and final part of this multi-cloud series how to exchange token using Terraform to create Azure resource from Google Cloud Build.

Part 4. Implement token exchange between Azure and GCP in Python