使用JWT触发Google Cloud功能

发布于 2025-01-23 02:15:55 字数 2514 浏览 4 评论 0原文

我正在尝试使用HTTP请求和服务帐户触发Google云功能。我遵循以下步骤：

创建一个服务帐户
下载JSON密钥文件
创建一个需要身份验证的Google Cloud函数（这是一个仅返回“ Hello World！”的测试功能，
我在“云函数权限”中添加了我的服务帐户，其中包括角色云功能如果我使用此服务帐户（GCLOUD auth print-nidentity-token）生成身份令牌，并使用它来触发我的

云功能（授权：bearer $ id_token在标题中），则可以按预期工作。但是现在我希望能够在生产中做到这一点。因此，我所做的就是生成一个JWT，使用它来获得访问令牌并使用此访问令牌，就像我对身份令牌一样，但我得到401未经授权的响应。

这是我的代码（Python），

import jwt
import time
import requests

# These elements are found in the service account JSON file (step 2)
PRIVATE_KEY_ID_FROM_JSON = "123ab4cdefghi56jk789123456789f123456m7n"
PRIVATE_KEY_FROM_JSON = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
SERVICE_ACCOUNT_EMAIL = "ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com"

# Create JWT
iat = int(time.time())
exp = iat + 3600

payload = {
    'iss': SERVICE_ACCOUNT_EMAIL,
    'sub': SERVICE_ACCOUNT_EMAIL,
    'aud': "https://www.googleapis.com/oauth2/v4/token",
    'target_audience ': 'https://REGION-PROJECT_ID.cloudfunctions.net/FUNCTION_NAME',
    'iat': iat,
    'exp': exp,
    "scope": "https://www.googleapis.com/auth/cloud-platform"
}
additional_headers = {'kid': PRIVATE_KEY_ID_FROM_JSON}

signed_jwt = jwt.encode(
    payload,
    PRIVATE_KEY_FROM_JSON,
    headers=additional_headers,
    algorithm='RS256'
)

# Use JWT to get access token
resp = requests.post(
    f"https://www.googleapis.com/oauth2/v4/token"
    f"?grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&assertion={signed_jwt}",
    headers={
        "Authorization": f"Bearer {signed_jwt}",
        "Content-Type": "application/x-www-form-urlencoded"
    }
)

access_token = resp.json()["access_token"]

# Trigger Cloud Function using this access token
resp_cf = requests.post(
    "https://europe-west1-cuure-265415.cloudfunctions.net/Test_service_account",
    headers={
        "Authorization": f"Bearer {access_token}"
    }
)

print(resp_cf.content)

结果是

b'\n<html><head>\n<meta http-equiv="content-type" content="text/html;charset=utf-8">\n<title>401 Unauthorized</title>\n</head>\n<body text=#000000 bgcolor=#ffffff>\n<h1>Error: Unauthorized</h1>\n<h2>Your client does not have permission to the requested URL <code>/CLOUD FUNCTION NAME</code>.</h2>\n<h2></h2>\n</body></html>\n'

我不想使用python库用服务帐户来验证我的请求，因为我的目的是能够在一个无代码开发平台Bubble.io中重现此请求。

有人知道我在做什么错吗？

原文

I am trying to trigger a Google cloud function using an HTTP request and a service account.
I followed these steps :

Create a service account
Download JSON key file
Create a Google Cloud function that requires authentication (this is a test function that just returns "Hello world !"
In the Cloud function permissions I added my service account with the role Cloud Functions Invoker

If I generate an identity token with this service account (gcloud auth print-identity-token) and use it to trigger my Cloud function (with Authorization: Bearer $ID_TOKEN in headers) it works as expected.
But now I want to be able to do it in production. So what I did is generate a JWT, use it to get an access token and use this access token just like I did with the identity token but I get 401 Unauthorized response.

Here is my code (Python)

import jwt
import time
import requests

# These elements are found in the service account JSON file (step 2)
PRIVATE_KEY_ID_FROM_JSON = "123ab4cdefghi56jk789123456789f123456m7n"
PRIVATE_KEY_FROM_JSON = "-----BEGIN PRIVATE KEY-----\n...\n-----END PRIVATE KEY-----"
SERVICE_ACCOUNT_EMAIL = "ACCOUNT_NAME@PROJECT_ID.iam.gserviceaccount.com"

# Create JWT
iat = int(time.time())
exp = iat + 3600

payload = {
    'iss': SERVICE_ACCOUNT_EMAIL,
    'sub': SERVICE_ACCOUNT_EMAIL,
    'aud': "https://www.googleapis.com/oauth2/v4/token",
    'target_audience ': 'https://REGION-PROJECT_ID.cloudfunctions.net/FUNCTION_NAME',
    'iat': iat,
    'exp': exp,
    "scope": "https://www.googleapis.com/auth/cloud-platform"
}
additional_headers = {'kid': PRIVATE_KEY_ID_FROM_JSON}

signed_jwt = jwt.encode(
    payload,
    PRIVATE_KEY_FROM_JSON,
    headers=additional_headers,
    algorithm='RS256'
)

# Use JWT to get access token
resp = requests.post(
    f"https://www.googleapis.com/oauth2/v4/token"
    f"?grant_type=urn:ietf:params:oauth:grant-type:jwt-bearer&assertion={signed_jwt}",
    headers={
        "Authorization": f"Bearer {signed_jwt}",
        "Content-Type": "application/x-www-form-urlencoded"
    }
)

access_token = resp.json()["access_token"]

# Trigger Cloud Function using this access token
resp_cf = requests.post(
    "https://europe-west1-cuure-265415.cloudfunctions.net/Test_service_account",
    headers={
        "Authorization": f"Bearer {access_token}"
    }
)

print(resp_cf.content)

The result is the following

b'\n<html><head>\n<meta http-equiv="content-type" content="text/html;charset=utf-8">\n<title>401 Unauthorized</title>\n</head>\n<body text=#000000 bgcolor=#ffffff>\n<h1>Error: Unauthorized</h1>\n<h2>Your client does not have permission to the requested URL <code>/CLOUD FUNCTION NAME</code>.</h2>\n<h2></h2>\n</body></html>\n'

I don't want to use Python libraries to authenticate my request with the service account because my aim is to be able to reproduce this in Bubble.io, a no code development platform.

Anyone knows what I'm doing wrong?

分享到QQ

分享到微博

如果你对这篇内容有疑问，欢迎到本站社区发帖提问参与讨论，获取更多帮助，或者扫码二维码加入 Web 技术交流群。

发布评论

需要登录才能够评论，你可以免费注册一个本站的账号。

耶耶耶 2025-01-30 02:15:55

如答案：

有三种类型的Oauth令牌。刷新令牌，访问令牌，
身份令牌。您正在尝试使用oauth访问令牌而不是
一个oauth身份令牌。为了授权，Google云功能
在授权中使用oauth身份令牌：携带者http标头。

您还可以参考答案在其中解释了Google签名的ID代币的自我签名JWT的地方，可以很好地解释。

您可以使用IAM角色管理对HTTP云功能的访问。你
将要分配角色/云函数。
服务帐户，并让呼叫者提供OAuth的凭据
授权标题中的功能。

回复收藏 0 原文

为你拒绝所有暧昧 2025-01-30 02:15:55

我注意到的一件事是云功能第二代不使用这种类型的端点'https://region-project_id.cloudfunctions.net/function_name'。 GCP生成了该功能的URL

回复收藏 0 原文

故乡的云 2025-01-30 02:15:55

我正在发布对我有用的解决方案。

为了获取ID令牌而不是访问令牌（以便再次在ADC或Google Enviornment或本地机器外面获取/请求您的另一个Google服务（如Cloud Funtion）。
这是您可能使用的代码。

import google.auth
import google.auth.transport.requests
from google.oauth2 import service_account
import requests

credentials = service_account.IDTokenCredentials.from_service_account_file('./secret.json', target_audience="https://us-central1-whatsapp-marekting-patchup.cloudfunctions.net/deploying_functions")
auth_req = google.auth.transport.requests.Request()
credentials.refresh(auth_req)
print(credentials.token)

现在，要测试此令牌以请求和身份验证的功能，您可以通过

# Replace with your Cloud Function URL
function_url = f"https://project-region-projectid.cloudfunctions.net/myFunction"

# Generate access token
access_token = credentials.token

# Optional: Prepare data to send to the function
data = {}  # Replace with your actual data (if applicable)

# Set headers with the access token
headers = {"Authorization": f"Bearer {access_token}"}

# Send the request with headers
response = requests.post(function_url, headers=headers, json=data)  # Adjust verb and data format (GET, PUT, etc.)

# Process the response from your Cloud Function
if response.status_code == 200:
    print(response.text)  # Assuming response is of text format (adjust parsing if needed)
else:
    print(f"Error: {response.status_code}")
    print(response)

I am posting the solution which worked for me.

To fetch the ID token rather than access token (for again fetching/requesting your another google service like cloud funtion) outside the ADC or google enviornment or in local machine.
Here is the code which you may use.

import google.auth
import google.auth.transport.requests
from google.oauth2 import service_account
import requests

credentials = service_account.IDTokenCredentials.from_service_account_file('./secret.json', target_audience="https://us-central1-whatsapp-marekting-patchup.cloudfunctions.net/deploying_functions")
auth_req = google.auth.transport.requests.Request()
credentials.refresh(auth_req)
print(credentials.token)

now to test this token to request and authenticated function you can do so by

# Replace with your Cloud Function URL
function_url = f"https://project-region-projectid.cloudfunctions.net/myFunction"

# Generate access token
access_token = credentials.token

# Optional: Prepare data to send to the function
data = {}  # Replace with your actual data (if applicable)

# Set headers with the access token
headers = {"Authorization": f"Bearer {access_token}"}

# Send the request with headers
response = requests.post(function_url, headers=headers, json=data)  # Adjust verb and data format (GET, PUT, etc.)

# Process the response from your Cloud Function
if response.status_code == 200:
    print(response.text)  # Assuming response is of text format (adjust parsing if needed)
else:
    print(f"Error: {response.status_code}")
    print(response)

回复收藏 0 原文

~没有更多了~