python - 如何将对 google app engine flask 端点的访问限制为仅应用程序代码(或 app engine 服务帐户)

Question

目前正在使用 python 3.7 和 flask 框架在 App Engine 标准环境中构建应用程序。我需要安排一些任务，这些任务需要应用定期运行多个敏感端点。

我想将对这些端点的访问限制为应用程序本身，以防止(非管理员)用户访问这些端点。在 Python 2 版本的应用程序引擎中，可以通过指定 login: admin 在 app.yaml 文件中，如下所示:

# app.yaml for google app engine standard env python 2

handlers:

  - url: /this_is/my_protected/endpoint
    script: main.app
    login: admin

但是，在应用引擎环境的 Python 3.7 化身中，这不再可能。

我知道可能需要在我的 Flask 应用程序的 main.py 文件中进行身份验证，但我不确定从哪里开始。我已经有 firebase 身份验证工作，并且该应用程序正在对多个面向用户的端点的用户进行身份验证。但是我不确定如何验证我自己的应用程序引擎应用程序(或者可能是服务帐户)以运行它自己的几个端点。我已尝试查看文档，但它们要么很少，要么就是找不到我需要的信息。

有没有一种简单的方法可以实现这一点？

Answer 1

正如评论中所建议的，这是我的简化(简单化？)解决方案，使谷歌应用引擎中的特定 Flask 端点只能由应用程序代码或应用引擎服务帐户访问。答案基于有关 validating cron requests 的文档和 validating task requests .

基本上，我们编写一个装饰器来验证 X-Appengine-Cron: true 是否在 header 中(暗示端点是由您的代码调用的，而不是远程用户).如果未找到 header ，则我们不会运行 protected 函数。

# python
# main.py

from flask import Flask, request, redirect, render_template

app = Flask(__name__)

# Define the decorator to protect your end points
def validate_cron_header(protected_function):
    def cron_header_validator_wrapper(*args, **kwargs):
        # https://cloud.google.com/appengine/docs/standard/python3/scheduling-jobs-with-cron-yaml#validating_cron_requests
        header = request.headers.get('X-Appengine-Cron')
        # If you are validating a TASK request from a TASK QUEUE instead of a CRON request, then use 'X-Appengine-TaskName' instead of 'X-Appengine-Cron'
        # example:
        # header = request.headers.get('X-Appengine-TaskName')
        # Other possible headers to check can be found here: https://cloud.google.com/tasks/docs/creating-appengine-handlers#reading_app_engine_task_request_headers

        # If the header does not exist, then don't run the protected function
        if not header:
            # here you can raise an error, redirect to a page, etc.
            return redirect("/")

        # Run and return the protected function
        return protected_function(*args, **kwargs)

    # The line below is necessary to allow the use of the wrapper on multiple endpoints
    # https://stackoverflow.com/a/42254713
    cron_header_validator_wrapper.__name__ = protected_function.__name__
    return cron_header_validator_wrapper


@app.route("/example/protected/handler")
@validate_cron_header
def a_protected_handler():
    # Run your code here
    your_response_or_error_etc = "text"
    return your_response_or_error_etc


@app.route("/yet/another/example/protected/handler/<myvar>")
@validate_cron_header
def another_protected_handler(some_var=None):
    # Run your code here
    return render_template("my_sample_template", some_var=some_var)