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

Cron job fails accessing url with admin_required decorator

根据文档,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)

AssertionError: 200 != 401

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

单元测试的 self.client.get 毫无疑问不会一路返回到 app.yaml 进行路由——所以如果你删除 app-level 检查你就不足为奇了在装饰器中做,它让 non-admin 用户通过。

然而,真正的问题是装饰器没有找到任何人 "logged in",而 cron 正在击中那个 URL。这在 https://cloud.google.com/appengine/docs/python/config/cron#Python_app_yaml_Securing_URLs_for_cron :

处有所暗示(尽管它肯定应该有更多 clearly/explicitly 记录!)

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

这表明服务基础设施不会通过检查当前logged-in用户来验证cron请求,因为它会发现none。相反,它依赖于请求中的 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

应该可以解决问题。

flask docs 好像说你不能这样装饰你的方法:

Decorating Views

Since the view class itself is not the view function that is added to the routing system it does not make much sense to decorate the class itself. Instead you either have to decorate the return value of as_view() by hand:

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)

Starting with Flask 0.8 there is also an alternative way where you can specify a list of decorators to apply in the class declaration:

class UserAPI(MethodView):
    decorators = [user_required]

Due to the implicit self from the caller’s perspective you cannot use regular view decorators on the individual methods of the view however, keep this in mind.

不过我不明白其中的道理。