我有一個我一直在開發的 Django 應用程式。當我在本地運行它時,它運行完美。當我使用 Cloud Run 在容器中運行它時,出現以下錯誤:
“憑據”物件沒有屬性“id_token”
這是有問題的代碼(有效負載是一個字典物件):
def ProcessPayload(payload):
# Get authorized session credentials
credentials, _ = google.auth.default()
session = AuthorizedSession(credentials)
credentials.refresh(Request(session))
# Process post request
headers = {'Authorization': f'Bearer {credentials.id_token}'}
response = requests.post(URL, json=payload, headers=headers)
在我的本地環境中,重繪 會正確加載所需標頭的正確 id_toled 憑據,但由于某種原因,當代碼部署到 Cloud Run 時,這不起作用。我已將 Cloud 運行實體設定為使用服務帳戶,因此它應該能夠從中獲取憑據。我如何使這項作業?我一直在谷歌搜索,直到我的手指受傷并且沒有找到可行的解決方案。
uj5u.com熱心網友回復:
在計算服務(Compute Engine、Cloud Run、Cloud Functions)下執行代碼時,呼叫元資料服務獲取 OIDC Identity Token。
import requests
METADATA_HEADERS = {'Metadata-Flavor': 'Google'}
METADATA_URL = 'http://metadata.google.internal/computeMetadata/v1/' \
'instance/service-accounts/default/identity?' \
'audience={}'
def fetch_identity_token(audience):
# Construct a URL with the audience and format.
url = METADATA_URL.format(audience)
# Request a token from the metadata server.
r = requests.get(url, headers=METADATA_HEADERS)
r.raise_for_status()
return r.text
def ProcessPayload(payload):
id_token = fetch_identity_token('replace_with_service_url')
# Process post request
headers = {'Authorization': f'Bearer {id_token}'}
response = requests.post(URL, json=payload, headers=headers)
獲取身份令牌的等效curl命令如下所示。您可以從 Compute Engine 實體進行測驗:
curl -H "metadata-flavor: Google" \
http://metadata.google.internal/computeMetadata/v1/instance/service-accounts/default/identity?audience=URL
其中 URL 是您正在呼叫的服務的 URL。
認證服務到服務
我見過這個元資料 URL 快捷方式(用于 Cloud Run),但我還沒有驗證它:
http://metadata/instance/service-accounts/default/identity?audience=URL
uj5u.com熱心網友回復:
因此,經過多次嘗試后,我找到了一個適用于兩個地方的解決方案。非常感謝 Paul Bonser 提出了這個簡單的方法!
import google.auth
from google.auth.transport.requests import AuthorizedSession, Request
from google.oauth2.id_token import fetch_id_token
import requests
def GetIdToken(audience):
credentials, _ = google.auth.default()
session = AuthorizedSession(credentials)
request = Request(session)
credentials.refresh(request)
if hasattr(credentials, "id_token"):
return credentials.id_token
return fetch_id_token(request, audience)
def ProcessPayload(url, payload):
# Get the ID Token
id_token = GetIdToken(url)
# Process post request
headers = {'Authorization': f'Bearer {id_token}'}
response = requests.post(url, json=payload, headers=headers)
轉載請註明出處,本文鏈接:https://www.uj5u.com/yidong/363457.html
上一篇:如何使用Ocelot在.NetCoreAPI網關中實作Windows身份驗證,以便所有下游服務都可以訪問IWindowsPrincipal?
