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

Multi-Cloud identity federation explained (5 Part Series)

1 Part 1. Access token vs ID token
2 Part 2. Token exchange from Azure to GCP
3 Part 3. Token exchange from GCP to Azure
4 Part 4. Implement token exchange between Azure and GCP in Python
5 Part 5. Provision Azure resources with Terraform from GCP with token exchange

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.


<span>from</span> <span>azure.identity</span> <span>import</span> <span>DefaultAzureCredential</span>
<span>from</span> <span>azure.identity</span> <span>import</span> <span>AzureAuthorityHosts</span>
<span>default_credential</span> <span>=</span> <span>DefaultAzureCredential</span><span>(</span>
    <span>authority</span><span>=</span><span>AzureAuthorityHosts</span><span>.</span><span>AZURE_PUBLIC_CLOUD</span>
<span>)</span>
<span>azure_access_token</span> <span>=</span> <span>default_credential</span><span>.</span><span>get_token</span><span>(</span>
    <span>scopes</span><span>=</span><span>f</span><span>"</span><span>{</span><span>os</span><span>.</span><span>environ</span><span>[</span><span>'APPLICATION_ID'</span><span>]</span><span>}</span><span>.default"</span>
<span>)</span>
<span>from</span> <span>azure.identity</span> <span>import</span> <span>DefaultAzureCredential</span>
<span>from</span> <span>azure.identity</span> <span>import</span> <span>AzureAuthorityHosts</span>

<span>default_credential</span> <span>=</span> <span>DefaultAzureCredential</span><span>(</span>
    <span>authority</span><span>=</span><span>AzureAuthorityHosts</span><span>.</span><span>AZURE_PUBLIC_CLOUD</span>
<span>)</span>
<span>azure_access_token</span> <span>=</span> <span>default_credential</span><span>.</span><span>get_token</span><span>(</span>
    <span>scopes</span><span>=</span><span>f</span><span>"</span><span>{</span><span>os</span><span>.</span><span>environ</span><span>[</span><span>'APPLICATION_ID'</span><span>]</span><span>}</span><span>.default"</span>
<span>)</span>
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"
)

Enter fullscreen mode Exit fullscreen mode

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.


<span>from</span> <span>msal</span> <span>import</span> <span>ConfidentialClientApplication</span>
<span>CLIENT_SECRET</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"CLIENT_SECRET"</span><span>]</span>
<span>TENANT_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"TENANT_ID"</span><span>]</span>
<span>APPLICATION_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"APPLICATION_ID"</span><span>]</span>
<span>app</span> <span>=</span> <span>ConfidentialClientApplication</span><span>(</span>
    <span>client_id</span><span>=</span><span>APPLICATION_ID</span><span>,</span>
    <span>client_credential</span><span>=</span><span>CLIENT_SECRET</span><span>,</span>
    <span>authority</span><span>=</span><span>f</span><span>"</span><span>{</span><span>AzureAuthorityHosts</span><span>.</span><span>AZURE_PUBLIC_CLOUD</span><span>}</span><span>/</span><span>{</span><span>TENANT_ID</span><span>}</span><span>"</span>
<span>)</span>
<span>azure_access_token</span> <span>=</span> <span>app</span><span>.</span><span>acquire_token_for_client</span><span>(</span>
    <span>scopes</span><span>=</span><span>f</span><span>"</span><span>{</span><span>APPLICATION_ID</span><span>}</span><span>/.default"</span>
<span>)[</span><span>"access_token"</span><span>]</span>
<span>from</span> <span>msal</span> <span>import</span> <span>ConfidentialClientApplication</span>

<span>CLIENT_SECRET</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"CLIENT_SECRET"</span><span>]</span>
<span>TENANT_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"TENANT_ID"</span><span>]</span>
<span>APPLICATION_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"APPLICATION_ID"</span><span>]</span>

<span>app</span> <span>=</span> <span>ConfidentialClientApplication</span><span>(</span>
    <span>client_id</span><span>=</span><span>APPLICATION_ID</span><span>,</span>
    <span>client_credential</span><span>=</span><span>CLIENT_SECRET</span><span>,</span>
    <span>authority</span><span>=</span><span>f</span><span>"</span><span>{</span><span>AzureAuthorityHosts</span><span>.</span><span>AZURE_PUBLIC_CLOUD</span><span>}</span><span>/</span><span>{</span><span>TENANT_ID</span><span>}</span><span>"</span>
<span>)</span>
<span>azure_access_token</span> <span>=</span> <span>app</span><span>.</span><span>acquire_token_for_client</span><span>(</span>
    <span>scopes</span><span>=</span><span>f</span><span>"</span><span>{</span><span>APPLICATION_ID</span><span>}</span><span>/.default"</span>
<span>)[</span><span>"access_token"</span><span>]</span>
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"]

Enter fullscreen mode Exit fullscreen mode

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.


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

<span>GCP_PROJECT_NUMBER</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"PROJECT_NUMER"</span><span>]</span>
<span>GCP_PROJECT_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"GCP_PROJECT_ID"</span><span>]</span>
<span>POOL_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"POOL_ID"</span><span>]</span>
<span>PROVIDER_ID</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"PROVIDER_ID"</span><span>]</span>

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

Enter fullscreen mode Exit fullscreen mode

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


<span>from</span> <span>google.oauth2.credentials</span> <span>import</span> <span>Credentials</span>
<span>from</span> <span>google.auth</span> <span>import</span> <span>impersonated_credentials</span>
<span>TARGET_SERVICE_ACCOUNT</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"TARGET_SERVICE_ACCOUNT"</span><span>]</span>
<span>sts_credentials</span> <span>=</span> <span>Credentials</span><span>(</span><span>token</span><span>=</span><span>sts_access_token</span><span>)</span>
<span>credentials</span> <span>=</span> <span>impersonated_credentials</span><span>.</span><span>Credentials</span><span>(</span>
  <span>source_credentials</span><span>=</span><span>sts_credentials</span><span>,</span>
  <span>target_principal</span><span>=</span><span>TARGET_SERVICE_ACCOUNT</span><span>,</span>
  <span>target_scopes</span> <span>=</span> <span>[</span><span>"https://www.googleapis.com/auth/cloud-platform"</span><span>],</span>
  <span>lifetime</span><span>=</span><span>500</span>
<span>)</span>
<span>credentials</span><span>.</span><span>refresh</span><span>(</span><span>Request</span><span>())</span>
<span>from</span> <span>google.oauth2.credentials</span> <span>import</span> <span>Credentials</span>
<span>from</span> <span>google.auth</span> <span>import</span> <span>impersonated_credentials</span>

<span>TARGET_SERVICE_ACCOUNT</span> <span>=</span> <span>os</span><span>.</span><span>environ</span><span>[</span><span>"TARGET_SERVICE_ACCOUNT"</span><span>]</span>

<span>sts_credentials</span> <span>=</span> <span>Credentials</span><span>(</span><span>token</span><span>=</span><span>sts_access_token</span><span>)</span>

<span>credentials</span> <span>=</span> <span>impersonated_credentials</span><span>.</span><span>Credentials</span><span>(</span>
  <span>source_credentials</span><span>=</span><span>sts_credentials</span><span>,</span>
  <span>target_principal</span><span>=</span><span>TARGET_SERVICE_ACCOUNT</span><span>,</span>
  <span>target_scopes</span> <span>=</span> <span>[</span><span>"https://www.googleapis.com/auth/cloud-platform"</span><span>],</span>
  <span>lifetime</span><span>=</span><span>500</span>
<span>)</span>
<span>credentials</span><span>.</span><span>refresh</span><span>(</span><span>Request</span><span>())</span>
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())

Enter fullscreen mode Exit fullscreen mode

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).


<span>from</span> <span>google.cloud</span> <span>import</span> <span>bigquery</span>
<span>client</span> <span>=</span> <span>bigquery</span><span>.</span><span>Client</span><span>(</span><span>credentials</span><span>=</span><span>credentials</span><span>,</span> <span>project</span><span>=</span><span>GCP_PROJECT_ID</span><span>)</span>
<span># Here my TARGET_SERVICE_ACCOUNT has bigquery.jobUser role. </span><span>query</span> <span>=</span> <span>"SELECT CURRENT_DATE() as date"</span>
<span>query_job</span> <span>=</span> <span>client</span><span>.</span><span>query</span><span>(</span><span>query</span><span>)</span>  <span># Make an API request. </span>
<span>print</span><span>(</span><span>"The query data:"</span><span>)</span>
<span>for</span> <span>row</span> <span>in</span> <span>query_job</span><span>:</span>
    <span>print</span><span>(</span><span>row</span><span>[</span><span>"date"</span><span>])</span>
<span># It works ! </span>
<span>from</span> <span>google.cloud</span> <span>import</span> <span>bigquery</span>

<span>client</span> <span>=</span> <span>bigquery</span><span>.</span><span>Client</span><span>(</span><span>credentials</span><span>=</span><span>credentials</span><span>,</span> <span>project</span><span>=</span><span>GCP_PROJECT_ID</span><span>)</span>

<span># Here my TARGET_SERVICE_ACCOUNT has bigquery.jobUser role. </span><span>query</span> <span>=</span> <span>"SELECT CURRENT_DATE() as date"</span>
<span>query_job</span> <span>=</span> <span>client</span><span>.</span><span>query</span><span>(</span><span>query</span><span>)</span>  <span># Make an API request. </span>
<span>print</span><span>(</span><span>"The query data:"</span><span>)</span>
<span>for</span> <span>row</span> <span>in</span> <span>query_job</span><span>:</span>
    <span>print</span><span>(</span><span>row</span><span>[</span><span>"date"</span><span>])</span>
<span># It works ! </span>
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 !

Enter fullscreen mode Exit fullscreen mode

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.


<span>from</span> <span>azure.core.credentials</span> <span>import</span> <span>TokenCredential</span><span>,</span> <span>AccessToken</span>
<span>from</span> <span>msal</span> <span>import</span> <span>ConfidentialClientApplication</span>
<span>from</span> <span>google.auth.transport.requests</span> <span>import</span> <span>Request</span>
<span>import</span> <span>time</span>
<span>class</span> <span>GoogleAssertionCredential</span><span>(</span><span>TokenCredential</span><span>):</span>
    <span>def</span> <span>__init__</span><span>(</span><span>self</span><span>,</span> <span>azure_client_id</span><span>,</span> <span>azure_tenant_id</span><span>,</span> <span>azure_authority_host</span><span>):</span>
        <span># create a confidential client application </span>        <span>self</span><span>.</span><span>app</span> <span>=</span> <span>ConfidentialClientApplication</span><span>(</span>
            <span>azure_client_id</span><span>,</span>
            <span>client_credential</span><span>=</span><span>{</span>
                <span>'client_assertion'</span><span>:</span> <span>self</span><span>.</span><span>_get_google_id_token</span><span>()</span>
            <span>},</span>
            <span>authority</span><span>=</span><span>f</span><span>"</span><span>{</span><span>azure_authority_host</span><span>}{</span><span>azure_tenant_id</span><span>}</span><span>"</span>
        <span>)</span>
    <span>def</span> <span>_get_google_id_token</span><span>(</span><span>self</span><span>)</span> <span>-></span> <span>str</span><span>:</span>
                <span>"""Request an ID token to the Metadata Server"""</span>
        <span>response</span> <span>=</span> <span>Request</span><span>()(</span>
            <span>f</span><span>"</span><span>{</span><span>GOOGLE_METADATA_API</span><span>}</span><span>/instance/service-accounts/default/identity"</span><span>,</span>
                        <span>f</span><span>"?audience=api://AzureADTokenExchange"</span><span>,</span>
            <span>method</span><span>=</span><span>"GET"</span><span>,</span>
            <span>headers</span><span>=</span><span>{</span><span>"Metadata-Flavor"</span><span>:</span> <span>"Google"</span><span>},</span>
          <span>)</span>
                <span>return</span> <span>response</span><span>.</span><span>data</span><span>.</span><span>decode</span><span>(</span><span>"utf-8"</span><span>)</span>
    <span>def</span> <span>get_token</span><span>(</span>
        <span>self</span><span>,</span>
        <span>*</span><span>scopes</span><span>:</span> <span>str</span><span>,</span>
        <span>claims</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>None</span><span>,</span>
        <span>tenant_id</span><span>:</span> <span>Optional</span><span>[</span><span>str</span><span>]</span> <span>=</span> <span>None</span><span>,</span>
        <span>**</span><span>kwargs</span><span>:</span> <span>Any</span>
    <span>)</span> <span>-></span> <span>AccessToken</span><span>:</span>
        <span># get the token using the application </span>        <span>token</span> <span>=</span> <span>self</span><span>.</span><span>app</span><span>.</span><span>acquire_token_for_client</span><span>(</span><span>scopes</span><span>)</span>
        <span>if</span> <span>'error'</span> <span>in</span> <span>token</span><span>:</span>
            <span>raise</span> <span>Exception</span><span>(</span><span>token</span><span>[</span><span>'error_description'</span><span>])</span>
        <span>expires_on</span> <span>=</span> <span>time</span><span>.</span><span>time</span><span>()</span> <span>+</span> <span>token</span><span>[</span><span>'expires_in'</span><span>]</span>
        <span># return an access token with the token string and expiration time </span>        <span>return</span> <span>AccessToken</span><span>(</span><span>token</span><span>[</span><span>'access_token'</span><span>],</span> <span>int</span><span>(</span><span>expires_on</span><span>))</span>
<span>from</span> <span>azure.core.credentials</span> <span>import</span> <span>TokenCredential</span><span>,</span> <span>AccessToken</span>
<span>from</span> <span>msal</span> <span>import</span> <span>ConfidentialClientApplication</span>
<span>from</span> <span>google.auth.transport.requests</span> <span>import</span> <span>Request</span>
<span>import</span> <span>time</span>

<span>class</span> <span>GoogleAssertionCredential</span><span>(</span><span>TokenCredential</span><span>):</span>

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

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

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

Enter fullscreen mode Exit fullscreen mode

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.


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

Enter fullscreen mode Exit fullscreen mode

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)


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

<span>creds</span> <span>=</span> <span>GoogleAssertionCredential</span><span>(</span>
    <span>azure_client_id</span><span>=</span><span>CLIENT_ID</span><span>,</span>
    <span>azure_tenant_id</span><span>=</span><span>TENANT_ID</span><span>,</span>
    <span>azure_authority_host</span><span>=</span><span>AzureAuthorityHosts</span><span>.</span><span>AZURE_PUBLIC_CLOUD</span>
<span>)</span>

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

Enter fullscreen mode Exit fullscreen mode

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.

Multi-Cloud identity federation explained (5 Part Series)

原文链接：Part 4. Implement token exchange between Azure and GCP in Python

展开阅读全文

文章版权声明 1、本网站名称：拾光赋
2、本站永久网址：https://www.blogs.ink
3、本网站的文章部分内容可能来源于网络，仅供大家学习与参考，如有侵权，请联系站长QQ：805375623进行删除处理。
4、本站一切资源不代表本站立场，并不代表本站赞同其观点和对其真实性负责。
5、本站一律禁止以任何方式发布或转载任何违法的相关信息，访客发现请向站长举报
6、本站资源大多存储在云盘，如发现链接失效，请联系我们我们会第一时间更新。

THE END