从字典而不是文件路径设置[“GOOGLE_APPLICATION_CREDENTIALS”]

Question

我正在尝试从字典设置环境变量，但在连接时出现错误。

#service account pulls in airflow variable that contains the json dict with service_account credentials

service_account = Variable.get('google_cloud_credentials')

os.environ["GOOGLE_APPLICATION_CREDENTIALS"]=str(service_account)

错误

    PermissionDeniedError: Error executing an HTTP request: HTTP response code 403 with body '<?xml version='1.0' encoding='UTF-8'?><Error><Code>AccessDenied</Code><Message>Access denied.</Message><Details>Anonymous caller does not have storage.objects.get access to the Google Cloud Storage object.</Details></Error>'

阅读时，如果我使用并指向文件，则没有问题。

os.environ["GOOGLE_APPLICATION_CREDENTIALS"]=/file/path/service_account.json

我想知道有没有办法将 dict 对象转换为类似操作系统路径的对象？我不想将 json 文件存储在容器上，并且airflow/google 文档根本不清楚。

Answer 1

有一种方法可以做到这一点，但谷歌文档是可怕。因此，我编写了 a Github gist 来记录我和同事 (Imee Cuison) 开发的安全使用密钥的方法。示例代码如下：

import json
from google.oauth2.service_account import Credentials
from google.cloud import secretmanager

def access_secret(project_id:str, secret_id:str, version_id:str="latest")->str:
    """Return the secret in string format"""
    
    # Create the Secret Manager client.
    client = secretmanager.SecretManagerServiceClient()

    # Build the resource name of the secret version.
    name = f"projects/{project_id}/secrets/{secret_id}/versions/{version_id}"

    # Access the secret version.
    response = client.access_secret_version(name=name)

    # Return the decoded payload.
    return response.payload.data.decode('UTF-8')


def get_credentials_from_token(token:str)->Credentials:
    """Given an authentication token, return a Credentials object"""
    
    credential_dict = json.loads(secret_payload)
    return Credentials.from_service_account_info(credential_dict)


credentials_secret = access_secret("my_project", "my_secret")
creds = get_credentials_from_token(credentials_secret)
# And now you can use the `creds` Credentials object to authenticate to an API

Answer 2

将服务帐户放入存储库并不是一个好的做法。作为最佳实践；您需要在应用程序中使用从默认谷歌身份验证传播的身份验证。

例如，使用 Google Cloud Kubernetes，您可以使用以下 python 代码：

from google.cloud.container_v1 import ClusterManagerClient   
credentials, project = google.auth.default(
    scopes=['https://www.googleapis.com/auth/cloud-platform', ])
credentials.refresh(google.auth.transport.requests.Request())
cluster_manager = ClusterManagerClient(credentials=credentials)

Answer 3

1. （可选）将 JSON 文件编码为 base64 字符串以确保兼容性
2. 将base64字符串放入环境变量
3. 从环境变量加载base64字符串（例如

   os.environ["BASE64_ENCODED_CREDENTIALS_JSON"]

   ）
4. 验证：

from google.cloud import storage
from google.oauth2 import service_account

gcp_json_credentials_dict = json.loads(os.environ["BASE64_ENCODED_CREDENTIALS_JSON"])
credentials = service_account.Credentials.from_service_account_info(gcp_json_credentials_dict)
client = storage.Client(project=gcp_json_credentials_dict['project_id'], credentials=credentials)

Answer 4

Python stringio 包允许您创建由字符串支持的类文件对象，但这在这里没有帮助，因为此环境变量的使用者期望的是文件路径，而不是类文件对象。我认为不可能做你想做的事。您是否有理由不想将凭据放入文件中？