问题:Cron 作业无法使用 admin_required 装饰器访问 url

根据文档,应允许 Cron 作业访问受管理员保护的视图。但是,如果我在 GET 方法上有@admin_required装饰器,则会收到 302 错误。

在 app.yaml 我定义了这个:

- url: /generator
  script: run.news.app
  login: admin

风景:

class GeneratorView(MethodView):
    @admin_required
    def get(self):
        return 'success', 200

urls.py

app.add_url_rule('/generator', 'generator', view_func=GeneratorView.as_view('generator'))

定时任务:

cron:
- description: Scrape every 3 hours
  url: /generator
  schedule: every 3 hours synchronized

装饰师:

def admin_required(func):
    """Requires App Engine admin credentials"""

    @wraps(func)
    def decorated_view(*args, **kwargs):
        if users.get_current_user():
            if not users.is_current_user_admin():
                abort(401)  # Unauthorized
            return func(*args, **kwargs)
        return redirect(users.create_login_url(request.url))

    return decorated_view

有趣的是,当我删除admin_required装饰器时,由于 app.yaml 中的login: admin,该 url 仍然受到管理员保护。

但是,由于缺少装饰器,我的单元测试未通过授权检查。

def test_generator_fails_as_normal_user(self):
        self.setCurrentUser(u'john@example.com', u'123')
        rv = self.client.get('/generator')
        self.assertEqual(rv.status_code, 401)

断言错误:200 !u003d 401

如果我把装饰器放回去,单元测试会通过,而 cron 作业会失败。有什么建议么?

解答

毫无疑问,单元测试的self.client.get不会一直返回到app.yaml进行路由——因此,如果您删除在装饰器中执行的应用程序级检查,它会让非管理员用户通过,这并不奇怪。

然而,真正的问题是,当cron访问该 URL 时,装饰器没有找到任何“登录”的人。这在https://cloud.google.com/appengine/docs/python/config/cron#Python_app_yaml_Securing_URLs_for _cron:

注意:虽然 cron 作业可以使用受login: admin限制的 URL 路径,但它们不能使用受login: required限制的 URL 路径。

这表明服务基础设施不通过检查当前登录的用户来验证 cron 请求,因为它找不到任何内容。相反,它依赖于请求中的标头:

来自 Cron 服务的请求也将包含一个 HTTP 标头:

X-AppEngine-Cron: true

X-AppEngine-Cron标头由 Google App Engine 内部设置。如果您的请求处理程序找到此标头,它可以相信该请求是 cron 请求。如果标头存在于对您的应用程序的外部用户请求中,则会将其剥离,但来自已登录应用程序管理员的请求除外,他们可以为测试目的设置标头。

所以,你的装饰器必须检查self.request的头文件——如果它找到X-AppEngine-Cron: true,它必须让请求通过,否则它可以继续执行你现在正在做的检查。

我不太确定您应该如何最好地在您选择的 Web 框架中获取请求的标头,您没有提及,但如果它是例如webapp2,那么类似于:

@wraps(func)
def decorated_view(self, *args, **kwargs):
    if self.request.headers.get('X-AppEngine-Cron') == 'true':
        return func(self, *args, **kwargs)
    # continue here with the other checks you do now

应该做的伎俩。