Мобильная версияВ настоящее время я использую Google Cloud Workload Identity Federation с AWS для создания подписанного JWT для сервисного аккаунта, как показано ниже.
Код: Выделить всё
...
def generate_jwt(sts_token: str, jwt_payload: str):
logger.debug("STS token is: %s", sts_token)
headers = {
"Content-Type": "text/json; charset=utf-8",
"Authorization": "Bearer " + sts_token,
}
try:
response = requests.post(
f"https://iamcredentials.googleapis.com/v1/projects/-/serviceAccounts/[email protected]:signJwt",
headers=headers,
data=jwt_payload,
)
logger.debug("JWT raw response is %s".format(response))
jwt_response = response.json()
logger.info(jwt_response)
return jwt_response.get("signedJwt", "")
except Exception as e:
logger.error(
f"There was an error in getting the JWT response. gcp_auth_token_error={e}"
)
raise HTTPException(status_code=500, detail={"error": "GCP_AUTH_TOKEN_FAILURE"})
def get_gcp_auth_token():
project_number = "123456789"
pool_id = "aws-pool"
provider_id = "aws-pool-provider"
jwt_payload = '{"payload":"{\\"sub\\":\\"[email protected]\\",\\"iat\\":1686268053,\\"aud\\":\\"https://google.gateway.example"\\",\\"iss\\":\\"[email protected]\\",\\"email\\":\\"[email protected]\\"}"}'
subject_token = generate_subject_token(project_number, pool_id, provider_id)
sts_token = generate_sts_token(subject_token, project_number, pool_id, provider_id)
jwt = generate_jwt(sts_token, jwt_payload)
return jwt
Код: Выделить всё
swagger: "2.0"
info:
title: "Test API"
description: "Random descipriotn"
version: 1.0.0
schemes:
- "https"
consumes:
- "application/json"
produces:
- "application/json"
securityDefinitions:
google_jwt:
type: "oauth2"
flow: "implicit"
authorizationUrl: ""
x-google-issuer: "[email protected]"
x-google-jwks_uri: "https://www.googleapis.com/service_accounts/v1/jwk/[email protected]"
x-google-audiences: "https://test-gateway-123456789.uc.gateway.dev"
security:
- google_jwt: []
paths:
"/serve":
post:
summary: "Serve endpoint for feature value fetching"
description: "Serve endpoint for feature value fetching"
parameters:
- name: body
in: body
required: true
schema:
type: object
properties:
location_id:
type: string
project_id:
type: string
featureonlinestore_name:
type: string
featureview_name:
type: string
entity_id:
type: string
required:
- location_id
- project_id
- featureonlinestore_name
- featureview_name
- entity_id
x-google-backend:
address: "https://test-gateway-123456789.uc.gateway.dev"
operationId: serve
responses:
'200':
description: Successful response
'400':
description: Bad request
'401':
description: Unauthorized
'500':
description: Internal server error
security:
- google_jwt: []
Код: Выделить всё
import requests
import json
import functions_framework
@functions_framework.http
def fetch_feature_values(request):
jwt_token = request.headers.get('Authorization')
print(f"JWT: {jwt_token}")
if not jwt_token:
return json.dumps({"error": "Missing JWT token"}), 401, {'Content-Type': 'application/json'}
request_json = request.get_json(silent=True)
location_id = request_json.get('location_id')
print(f"Location id: {location_id}")
project_id = request_json.get('project_id')
print(f"Project id: {project_id}")
featureonlinestore_name = request_json.get('featureonlinestore_name')
print(f"Feature online store name: {featureonlinestore_name}")
featureview_name = request_json.get('featureview_name')
print(f"Featureview Name: {featureview_name}")
entity_id = request_json.get('entity_id')
print(f"Entity id: {entity_id}")
data_format = request_json.get('data_format', 'JSON')
url = f"https://{location_id}-aiplatform.googleapis.com/v1/projects/{project_id}/locations/{location_id}/featureOnlineStores/{featureonlinestore_name}/featureViews/{featureview_name}:fetchFeatureValues"
print(url)
payload = {
"data_key": str(entity_id),
"data_format": data_format
}
try:
headers = {"Authorization": jwt_token}
response = requests.post(url, json=payload, headers=headers)
print(response)
if response.status_code == 200:
feature_values = response.json()
return json.dumps(feature_values), 200, {'Content-Type': 'application/json'}
else:
error_message = f"Error fetching feature values. Status code: {response.status_code}"
return json.dumps({"error": error_message}), response.status_code, {'Content-Type': 'application/json'}
except requests.exceptions.RequestException as e:
error_message = f"Error making the request: {str(e)}"
return json.dumps({"error": error_message}), 500, {'Content-Type': 'application/json'}
I've seen articles explaining to issue a Google ID Token, but also some conflicting information that the JWT signed in the notebook should suffice (clearly not). Is anybody able to provide any insights as to why this JWT may not be sufficient and how to successfully hit this endpoint?
Sidenote, it may not matter, but I am unable to hit the Function endpoint directly from Postman with the issued JWT as well, not just the Gateway -> Function path.
Would this kind of solution be what i'm looking for, or would I need to have two separate identity token generated in the notebook and pass that information over with the gateway as well to send to the function.
Источник: https://stackoverflow.com/questions/781 ... authorized
