(self, *args, **kwargs)
| 30 | # Client application should provide redirection in the way it likes |
| 31 | def check_authorization(func, ): |
| 32 | async def wrapper(self, *args, **kwargs): |
| 33 | auth = self.application.auth |
| 34 | authorizer = self.application.authorizer |
| 35 | |
| 36 | login_url = self.get_login_url() |
| 37 | request_path = self.request.path |
| 38 | |
| 39 | login_resource = is_allowed_during_login(request_path, login_url, self) |
| 40 | if login_resource: |
| 41 | return func(self, *args, **kwargs) |
| 42 | |
| 43 | try: |
| 44 | authenticated = await auth.is_authenticated(self) |
| 45 | except (AuthRejectedError, AuthFailureError) as e: |
| 46 | message = 'On-fly auth rejected' |
| 47 | LOGGER.warning(message + ': ' + str(e)) |
| 48 | code = 401 |
| 49 | if isinstance(self, tornado.websocket.WebSocketHandler): |
| 50 | self.close(code=code, reason=message) |
| 51 | return |
| 52 | else: |
| 53 | raise tornado.web.HTTPError(code, message) |
| 54 | |
| 55 | access_allowed = authenticated and authorizer.is_allowed_in_app(identify_user(self)) |
| 56 | |
| 57 | if authenticated and (not access_allowed): |
| 58 | user = identify_user(self) |
| 59 | LOGGER.warning('User ' + user + ' is not allowed') |
| 60 | code = 403 |
| 61 | message = 'Access denied. Please contact system administrator' |
| 62 | if isinstance(self, tornado.websocket.WebSocketHandler): |
| 63 | self.close(code=code, reason=message) |
| 64 | return |
| 65 | else: |
| 66 | if isinstance(self, tornado.web.StaticFileHandler) and request_path.lower().endswith('.html'): |
| 67 | login_url += "?" + urlencode(dict(next=request_path, redirectReason='prohibited')) |
| 68 | redirect_relative(login_url, self) |
| 69 | return |
| 70 | else: |
| 71 | raise tornado.web.HTTPError(code, message) |
| 72 | |
| 73 | if authenticated and access_allowed: |
| 74 | return func(self, *args, **kwargs) |
| 75 | |
| 76 | if not isinstance(self, tornado.web.StaticFileHandler): |
| 77 | message = 'Not authenticated' |
| 78 | code = 401 |
| 79 | LOGGER.warning('%s %s %s: user is not authenticated' % (code, self.request.method, request_path)) |
| 80 | if isinstance(self, tornado.websocket.WebSocketHandler): |
| 81 | self.close(code=code, reason=message) |
| 82 | return |
| 83 | else: |
| 84 | raise tornado.web.HTTPError(code, message) |
| 85 | |
| 86 | login_url += "?" + urlencode(dict(next=request_path)) |
| 87 | |
| 88 | redirect_relative(login_url, self) |
| 89 |
no test coverage detected