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

Question

根据文档，应允许 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 作业：

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 != 401

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

Answer 1

flask docs 似乎在说你不能这样装饰你的方法：

装饰视图

由于视图类本身不是视图 添加到路由系统的功能并没有多大作用 感觉来装饰班级本身。相反，您要么必须 手工修饰 as_view() 的返回值：

def user_required(f):
    """Checks whether user is logged in or raises error 401."""
    def decorator(*args, **kwargs):
        if not g.user:
            abort(401)
        return f(*args, **kwargs)
    return decorator

view = user_required(UserAPI.as_view('users'))
app.add_url_rule('/users/', view_func=view)

从 Flask 0.8 开始，您还可以使用另一种方法 指定要在类声明中应用的装饰器列表：

class UserAPI(MethodView):
    decorators = [user_required]

由于从调用者的角度来看隐含的自我，你不能使用 然而，视图的各个方法上的常规视图装饰器， 请记住这一点。

不过，我不明白其中的原因。

Answer 2

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

然而，真正的问题是，当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

应该可以解决问题。