Python: как я могу создать GoogleCredentials, используя определенного пользователя вместо get_application_default()

Я обновляю скрипт для вызова конечных точек Google Cloud, защищенных OAuth2. В предыдущей версии предполагалось, что один пользователь предварительно аутентифицирован gcloud auth login и, таким образом, смог использовать значение по умолчанию:

credentials = GoogleCredentials.get_application_default()
http = credentials.authorize(http)

Однако теперь я должен выполнить некоторые вызовы как пользователь A, а некоторые как пользователь B. Я могу выполнить эти шаги в оболочке для генерации токенов доступа, но я бы предпочел сделать это непосредственно в программе:

gcloud auth login user_A@email.com
gcloud auth print-access-token user_A@email.com

Есть ли способ сгенерировать два Credentials значения для двух разных писем без необходимости запуска каких-либо команд оболочки?

1 ответ

Мой случай немного отличается, но, надеюсь, вы сможете найти эквивалент для своего варианта использования. Я использую сервисные аккаунты для аутентификации.

Для успешной аутентификации вам нужны 3 вещи

  1. Файл учетной записи службы. Это файл JSON, который вы получаете после создания учетной записи службы.

  2. ОБЛАСТИ. Это указывает, к какому API Google вам нужно получить доступ. В моем случае мне нужно было получить доступ к API групп Google. Следовательно, объем, который мне нужен, был: "https://www.googleapis.com/auth/admin.directory.group"

вы можете найти список всех областей здесь: https://developers.google.com/identity/protocols/googlescopes

  1. Электронный адрес необходимого пользователя. Это адрес электронной почты пользователя, которого ваша учетная запись службы хочет выдать (user_A@email.com). В моем случае мне понадобилась электронная почта администратора G Suite, поскольку у них есть доступ только к группам Google.

Ниже приведен эквивалентный код Python:

from google.oauth2 import service_account
from googleapiclient.discovery import build

SCOPES = ["https://www.googleapis.com/auth/admin.directory.group"]
SERVICE_ACCOUNT_FILE = "my-service-account.json"

credentials = service_account.Credentials.from_service_account_file(
        SERVICE_ACCOUNT_FILE,
        scopes=SCOPES,
        subject="admin@example.com")

admin_service = build("admin", "directory_v1", credentials=credentials)
group = admin_service.groups().list(domain="example.com").execute()
print("groups list: ", group)

Теперь в соответствии с вашим вопросом, если вы хотите выдать себя за другого пользователя. Тогда вы можете использовать

new_user_email = "user2@example.com"
user2_credentials = credentials.with_subject(new_user_email)
new_service = build("admin", "directory_v1", credentials=user2_credentials)
# use this new_service to call required APIs

Вот ссылка на документы для аутентификации с использованием служебных учетных записей для библиотеки Python: https://google-auth.readthedocs.io/en/latest/reference/google.oauth2.service_account.html

Вот ссылка на то, как обычно происходит поток Oauth для учетных записей служб: https://github.com/googleapis/google-api-python-client/blob/master/docs/oauth-server.md

Надеюсь, это вам помогло.

PS - Это мой первый ответ на stackru, поэтому, пожалуйста, расскажите мне, как я мог бы улучшить свой ответ.

Вы, вероятно, хотите использовать

gcloud auth application-default login
gcloud auth application-default print-access-token

вместо gcloud auth login,

Но если вы используете учетные данные gcloud (не по умолчанию для приложения), обратите внимание, что

gcloud auth login

это интерактивная команда. Вы выбираете пользователя для входа, как в браузере, а не в командной строке.

Вы можете предварительно войти в систему заранее, а затем использовать учетные данные, которые вы хотите. Например:

gcloud auth login --account user_A@email.com
gcloud auth login --account user_B@email.com

это добавляет учетные данные в хранилище учетных данных (обратите внимание, что здесь --account используется только для проверки, чтобы убедиться, что выбранный вами веб-поток совпадает с запросом здесь). Вы можете увидеть все доступные учетные данные, запустив

gcloud auth list

Тогда вы можете использовать конкретную учетную запись по требованию

gcloud auth print-access-token --account user_A@email.com
gcloud auth print-access-token --account user_B@email.com

Обратите внимание, что print-access-token - недокументированная команда, и вы должны использовать ее только для отладки.

Несколько более продвинутая возможность - использовать конфигурации

gcloud config configurations list

Вы можете создавать новые путем

gcloud config configurations create A
gcloud config set account user_A@email.com
gcloud config set project project_A

gcloud config configurations create B
gcloud config set account user_B@email.com
gcloud config set project project_B

тогда вы можете сделать

gcloud auth print-access-token --configuration A
gcloud auth print-access-token --configuration B

дополнительное преимущество в том, что не только вы можете настроить учетную запись, но и другие атрибуты, такие как проект, вычислительная зона и т. д.

Насколько я понимаю, вы хотите делать запросы к ресурсам, защищенным OAuth2.0, и указывать учетную запись "вызывающего абонента" внутри кода Python.

Вот хорошая функция, которую я лично использую (у нее больше, чем вам нужно, поскольку она работает по-разному в зависимости от того, откуда она вызывается). Вам просто нужно дать ему идентификатор клиента учетной записи, которую вы хотите использовать для звонка.

def make_iap_request(url, client_id, method='GET', **kwargs):
    """Makes a request to an application protected by Identity-Aware Proxy.

    Args:
      url: The Identity-Aware Proxy-protected URL to fetch.
      client_id: The client ID used by Identity-Aware Proxy.
      method: The request method to use
              ('GET', 'OPTIONS', 'HEAD', 'POST', 'PUT', 'PATCH', 'DELETE')
      **kwargs: Any of the parameters defined for the request function:
                https://github.com/requests/requests/blob/master/requests/api.py
                If no timeout is provided, it is set to 90 by default.

    Returns:
      The page body, or raises an exception if the page couldn't be retrieved.
    """
    # Set the default timeout, if missing
    if 'timeout' not in kwargs:
        kwargs['timeout'] = 120

    # Figure out what environment we're running in and get some preliminary
    # information about the service account.
    bootstrap_credentials, _ = google.auth.default(
        scopes=[IAM_SCOPE])
    if isinstance(bootstrap_credentials,
                  google.oauth2.credentials.Credentials):
        raise Exception('make_iap_request is only supported for service '
                        'accounts.')
    elif isinstance(bootstrap_credentials,
                    google.auth.app_engine.Credentials):
        requests_toolbelt.adapters.appengine.monkeypatch()

    # For service account's using the Compute Engine metadata service,
    # service_account_email isn't available until refresh is called.
    bootstrap_credentials.refresh(Request())

    signer_email = bootstrap_credentials.service_account_email
    if isinstance(bootstrap_credentials,
                  google.auth.compute_engine.credentials.Credentials):
        # Since the Compute Engine metadata service doesn't expose the service
        # account key, we use the IAM signBlob API to sign instead.
        # In order for this to work:
        #
        # 1. Your VM needs the https://www.googleapis.com/auth/iam scope.
        #    You can specify this specific scope when creating a VM
        #    through the API or gcloud. When using Cloud Console,
        #    you'll need to specify the "full access to all Cloud APIs"
        #    scope. A VM's scopes can only be specified at creation time.
        #
        # 2. The VM's default service account needs the "Service Account Actor"
        #    role. This can be found under the "Project" category in Cloud
        #    Console, or roles/iam.serviceAccountActor in gcloud.
        signer = google.auth.iam.Signer(
            Request(), bootstrap_credentials, signer_email)
    else:
        # A Signer object can sign a JWT using the service account's key.
        signer = bootstrap_credentials.signer

    # Construct OAuth 2.0 service account credentials using the signer
    # and email acquired from the bootstrap credentials.
    service_account_credentials = google.oauth2.service_account.Credentials(
        signer, signer_email, token_uri=OAUTH_TOKEN_URI, additional_claims={
            'target_audience': client_id
        })

    # service_account_credentials gives us a JWT signed by the service
    # account. Next, we use that to obtain an OpenID Connect token,
    # which is a JWT signed by Google.
    google_open_id_connect_token = get_google_open_id_connect_token(
        service_account_credentials)

    # Fetch the Identity-Aware Proxy-protected URL, including an
    # Authorization header containing "Bearer " followed by a
    # Google-issued OpenID Connect token for the service account.
    resp = requests.request(
        method, url,
        headers={'Authorization': 'Bearer {}'.format(
            google_open_id_connect_token)}, **kwargs)
    #time.sleep(50)
    if resp.status_code == 403:
        raise Exception('Service account {} does not have permission to '
                        'access the IAP-protected application.'.format(
                            signer_email))
    elif resp.status_code != 200:
        raise Exception(
            'Bad response from application: {!r} / {!r} / {!r}'.format(
                resp.status_code, resp.headers, resp.text))
    elif rest.status_code == 500:
        time.sleep(90)
        return 'DONE'
    else:
        return resp.text


def get_google_open_id_connect_token(service_account_credentials):
    """Get an OpenID Connect token issued by Google for the service account.

    This function:

      1. Generates a JWT signed with the service account's private key
         containing a special "target_audience" claim.

      2. Sends it to the OAUTH_TOKEN_URI endpoint. Because the JWT in #1
         has a target_audience claim, that endpoint will respond with
         an OpenID Connect token for the service account -- in other words,
         a JWT signed by *Google*. The aud claim in this JWT will be
         set to the value from the target_audience claim in #1.

    For more information, see
    https://developers.google.com/identity/protocols/OAuth2ServiceAccount .
    The HTTP/REST example on that page describes the JWT structure and
    demonstrates how to call the token endpoint. (The example on that page
    shows how to get an OAuth2 access token; this code is using a
    modified version of it to get an OpenID Connect token.)
    """

    service_account_jwt = (
        service_account_credentials._make_authorization_grant_assertion())
    request = google.auth.transport.requests.Request()
    body = {
        'assertion': service_account_jwt,
        'grant_type': google.oauth2._client._JWT_GRANT_TYPE,
    }
    token_response = google.oauth2._client._token_endpoint_request(
        request, OAUTH_TOKEN_URI, body)
    return token_response['id_token']

Вы можете узнать больше о других способах здесь.

Другие вопросы по тегам