|
4 | 4 | from typing import List, Set |
5 | 5 |
|
6 | 6 | import google.api_core.exceptions |
7 | | -from google.auth import load_credentials_from_dict |
| 7 | +from google.auth import load_credentials_from_dict, load_credentials_from_file |
8 | 8 | from google.cloud import compute_v1, container_v1, iam_admin_v1, storage |
9 | 9 | from google.oauth2 import service_account |
10 | 10 |
|
@@ -33,9 +33,32 @@ def load_credentials(): |
33 | 33 | # to determine if the credentials are stored as a file or not before |
34 | 34 | # reading them |
35 | 35 | if credentials.endswith(".json"): |
36 | | - loaded_credentials = service_account.Credentials.from_service_account_file( |
37 | | - credentials, scopes=scopes |
38 | | - ) |
| 36 | + # Read the file to determine credential type |
| 37 | + with open(credentials, "r") as f: |
| 38 | + cred_data = json.load(f) |
| 39 | + |
| 40 | + # Check if this is a traditional service account vs workload identity federation |
| 41 | + if cred_data.get("type") == "service_account": |
| 42 | + # Traditional service account JSON format: |
| 43 | + # {"type": "service_account", "project_id": "...", "private_key_id": "...", |
| 44 | + # "private_key": "...", "client_email": "...", "client_id": "...", |
| 45 | + # "auth_uri": "https://accounts.google.com/o/oauth2/auth", |
| 46 | + # "token_uri": "https://oauth2.googleapis.com/token", |
| 47 | + # "auth_provider_x509_cert_url": "...", "client_x509_cert_url": "..."} |
| 48 | + # See: https://cloud.google.com/iam/docs/keys-create-delete#creating |
| 49 | + loaded_credentials = service_account.Credentials.from_service_account_file( |
| 50 | + credentials, scopes=scopes |
| 51 | + ) |
| 52 | + else: |
| 53 | + # Workload identity federation or other external account types: |
| 54 | + # {"type": "external_account", "audience": "//iam.googleapis.com/...", |
| 55 | + # "subject_token_type": "urn:ietf:params:oauth:token-type:jwt", |
| 56 | + # "token_url": "https://sts.googleapis.com/v1/token", |
| 57 | + # "credential_source": {...}, "service_account_impersonation_url": "..."} |
| 58 | + # See: https://google.aip.dev/auth/4117 |
| 59 | + loaded_credentials, _ = load_credentials_from_file( |
| 60 | + credentials, scopes=scopes |
| 61 | + ) |
39 | 62 | else: |
40 | 63 | loaded_credentials, _ = load_credentials_from_dict( |
41 | 64 | json.loads(credentials), scopes=scopes |
|
0 commit comments