无法执行云函数触发不允许未经身份验证的调用的 HTTP 触发的云函数？
Posted
技术标签:
【中文标题】无法执行云函数触发不允许未经身份验证的调用的 HTTP 触发的云函数？【英文标题】：Unable to perform Cloud Function trigger a HTTP triggered Cloud Function that doesn't allow unauthenticated invocations? 【发布时间】：2020-07-18 05:24:25 【问题描述】：我有一种情况，我正在尝试创建两个云功能，即 CF1 和 CF2，并且我有一个云调度程序。两个云功能都启用了经过身份验证的调用。我的流程是 Cloud Scheduler 将触发 CF1。 CF1 完成后，CF1 将触发 CF2 作为 http 调用。我已推荐 Cannot invoke Google Cloud Function from GCP Scheduler 从 Cloud Scheduler 访问经过身份验证的 CF1 并能够访问 CF1。但是从 CF1 访问 CF2 时出现问题。 CF1 不会触发 CF2，也不会给出任何错误消息。从另一个经过身份验证的 Cloud Function 访问经过身份验证的 Cloud Function 时，我们是否需要遵循任何其他技术。
CF1代码：
import json
import logging
from requests_futures.sessions import FuturesSession
def main(request):
# To read parameter values from request (url arguments or Json body).
raw_request_data = request.data
string_request_data = raw_request_data.decode("utf-8")
request_json: dict = json.loads(string_request_data)
request_args = request.args
if request_json and 'cf2_endpoint' in request_json:
cf2_endpoint = request_json['cf2_endpoint']
elif request_args and 'cf2_endpoint' in request_args:
cf2_endpoint = request_args['cf2_endpoint']
else:
cf2_endpoint = 'Invalid endpoint for CF2'
logger = logging.getLogger('test')
try:
session = FuturesSession()
session.get("".format(cf2_endpoint))
logger.info("First cloud function executed successfully.")
except RuntimeError:
logger.error("Exception occurred ".format(RuntimeError))
CF2代码：
import logging
def main(request):
logger = logging.getLogger('test')
logger.info("second cloud function executed successfully.")
当前输出日志：
First cloud function executed successfully.
预期输出日志：
First cloud function executed successfully.
second cloud function executed successfully.
注意：如果我对这两个云功能使用未经身份验证的访问，则相同的流程正在工作。
【问题讨论】：
【参考方案1】：这里发生了两件事：
-
您没有完全正确地使用
request-futures。由于请求是异步发出的，因此您需要在函数隐式返回之前阻止结果，否则它可能会在您的 HTTP 请求完成之前返回（尽管它可能在此示例中）：
session = FuturesSession()
future = session.get("".format(cf2_endpoint))
resp = future.result() # Block on the request completing
-
您对第二个函数发出的请求实际上不是经过身份验证的请求。默认情况下，来自云函数的出站请求未经过身份验证。如果您查看上面的实际响应，您会看到：
>>> resp.status_code
403
>>> resp.content
b'\n<html><head>\n<meta http-equiv="content-type" content="text/html;charset=utf-8">\n<title>403 Forbidden</title>\n</head>\n<body text=#000000 bgcolor=#ffffff>\n<h1>Error: Forbidden</h1>\n<h2>Your client does not have permission to get URL <code>/function_two</code> from this server.</h2>\n<h2></h2>\n</body></html>\n'
您可以跳过很多麻烦来正确验证此请求，如文档中所述：https://cloud.google.com/functions/docs/securing/authenticating#function-to-function
但是，更好的选择是将您的第二个函数设为“后台”函数，并改为通过从第一个函数发布的 PubSub 消息调用它：
from google.cloud import pubsub
publisher = pubsub.PublisherClient()
topic_name = 'projects/project_id/topics/topic'.format(
project_id=<your project id>,
topic='MY_TOPIC_NAME', # Set this to something appropriate.
)
def function_one(request):
message = b'My first message!'
publisher.publish(topic_name, message)
def function_two(event, context):
message = event['data'].decode('utf-8')
print(message)
只要您的函数具有发布 PubSub 消息的权限，就可以避免对 HTTP 请求添加授权，并确保至少一次交付。
【讨论】：
【参考方案2】：Google Cloud Function 提供 REST API interface 包括 call 方法，可用于另一个 Cloud Function HTTP 调用。 虽然文档 mention using Google-provided client libraries 仍然没有关于 Python 上的 Cloud Function 的文档。
相反，您需要使用通用的 Google API 客户端库。 [这是python之一].3
使用这种方法的主要困难可能是对身份验证过程的理解。 通常，您需要提供两件事来构建客户端服务： 凭证和范围。
获取凭据的最简单方法是在应用程序默认凭据 (ADC) 库上进行中继。正确的文档是：
-
https://cloud.google.com/docs/authentication/production
https://github.com/googleapis/google-api-python-client/blob/master/docs/auth.md
获取范围的地方是每个 REST API 函数文档页面。 喜欢，OAuth scope: https://www.googleapis.com/auth/cloud-platform
调用“hello-world”云函数的完整代码示例如下。 运行前：
-
在您的项目中在 GCP 上创建默认 Cloud Function。
-
注意您部署函数的project_id、函数名称、位置。
如果您要在 Cloud Function 环境（例如本地）之外调用函数，请根据上述文档设置环境变量 GOOGLE_APPLICATION_CREDENTIALS
如果您实际上是从另一个 Cloud Function 调用，则根本不需要配置凭据。
from googleapiclient.discovery import build
from googleapiclient.discovery_cache.base import Cache
import google.auth
import pprint as pp
def get_cloud_function_api_service():
class MemoryCache(Cache):
_CACHE =
def get(self, url):
return MemoryCache._CACHE.get(url)
def set(self, url, content):
MemoryCache._CACHE[url] = content
scopes = ['https://www.googleapis.com/auth/cloud-platform']
# If the environment variable GOOGLE_APPLICATION_CREDENTIALS is set,
# ADC uses the service account file that the variable points to.
#
# If the environment variable GOOGLE_APPLICATION_CREDENTIALS isn't set,
# ADC uses the default service account that Compute Engine, Google Kubernetes Engine, App Engine, Cloud Run,
# and Cloud Functions provide
#
# see more on https://cloud.google.com/docs/authentication/production
credentials, project_id = google.auth.default(scopes)
service = build('cloudfunctions', 'v1', credentials=credentials, cache=MemoryCache())
return service
google_api_service = get_cloud_function_api_service()
name = 'projects/project_id/locations/us-central1/functions/function-1'
body =
'data': ' "message": "It is awesome, you are develop on Stack Overflow language!"' # json passed as a string
result_call = google_api_service.projects().locations().functions().call(name=name, body=body).execute()
pp.pprint(result_call)
# expected out out is:
# 'executionId': '3h4c8cb1kwe2', 'result': 'It is awesome, you are develop on Stack Overflow language!'
【讨论】：
以上是关于无法执行云函数触发不允许未经身份验证的调用的 HTTP 触发的云函数？的主要内容，如果未能解决你的问题，请参考以下文章
Google Cloud Functions Deploy“允许未经身份验证的调用......”
Firebase + Flutter - 云函数 onCall 导致 Android 应用出现“未经身份验证”错误