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 вещи
Файл учетной записи службы. Это файл JSON, который вы получаете после создания учетной записи службы.
ОБЛАСТИ. Это указывает, к какому API Google вам нужно получить доступ. В моем случае мне нужно было получить доступ к API групп Google. Следовательно, объем, который мне нужен, был: "https://www.googleapis.com/auth/admin.directory.group"
вы можете найти список всех областей здесь: https://developers.google.com/identity/protocols/googlescopes
- Электронный адрес необходимого пользователя. Это адрес электронной почты пользователя, которого ваша учетная запись службы хочет выдать (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']
Вы можете узнать больше о других способах здесь.