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.
不过我不明白其中的道理。
根据文档,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 :
Note: While cron jobs can use URL paths restricted with
login: admin
, they cannot use URL paths restricted withlogin: 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.
不过我不明白其中的道理。