python - Cron 作业无法使用 admin_required 装饰器访问 url

Question

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

在app.yaml中我定义了这个:

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

View :

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)

AssertionError: 200 != 401

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

Answer 1

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

然而，真正的问题是，当 cron 访问该 URL 时，装饰器没有发现任何人“登录”。 https://cloud.google.com/appengine/docs/python/config/cron#Python_app_yaml_Securing_URLs_for_cron 暗示了这一点(尽管它肯定应该更清楚/明确地记录下来!) :

Note: While cron jobs can use URL paths restricted with login: admin, they cannot use URL paths restricted with login: required.

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

Requests from the Cron Service will also contain a HTTP header:

X-AppEngine-Cron: true

The X-AppEngine-Cron header is set internally by Google App Engine. If your request handler finds this header it can trust that the request is a cron request. If the header is present in an external user request to your app, it is stripped, except for requests from logged in administrators of the application, who are allowed to set the header for testing purposes.

因此，您的装饰器必须检查 self.request 处的 header - 如果找到 X-AppEngine-Cron: true，则必须让请求通过，否则它可以继续执行您现在正在执行的检查。

我不太确定您应该如何最好地在您选择的网络框架中获取请求的 header ，您没有提到这一点，但如果它是例如 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

应该可以解决问题。