(self, request, call_next)
| 29 | super().__init__(app) |
| 30 | |
| 31 | async def dispatch(self, request, call_next): |
| 32 | |
| 33 | if self.is_options(request) or whiteUtils.is_whitelisted(request.url.path): |
| 34 | return await call_next(request) |
| 35 | assistantTokenKey = settings.ASSISTANT_TOKEN_KEY |
| 36 | assistantToken = request.headers.get(assistantTokenKey) |
| 37 | askToken = request.headers.get("X-SQLBOT-ASK-TOKEN") |
| 38 | trans = await get_i18n(request) |
| 39 | if askToken: |
| 40 | validate_pass, data = await self.validateAskToken(askToken, trans) |
| 41 | if validate_pass: |
| 42 | request.state.current_user = data |
| 43 | return await call_next(request) |
| 44 | message = trans('i18n_permission.authenticate_invalid', msg = data) |
| 45 | return JSONResponse(message, status_code=401, headers={"Access-Control-Allow-Origin": "*"}) |
| 46 | #if assistantToken and assistantToken.lower().startswith("assistant "): |
| 47 | if assistantToken: |
| 48 | validator: tuple[any] = await self.validateAssistant(assistantToken, trans) |
| 49 | if validator[0]: |
| 50 | request.state.current_user = validator[1] |
| 51 | if request.state.current_user and trans.lang: |
| 52 | request.state.current_user.language = trans.lang |
| 53 | request.state.assistant = validator[2] |
| 54 | origin = request.headers.get("X-SQLBOT-HOST-ORIGIN") or get_origin_from_referer(request) |
| 55 | if origin and validator[2]: |
| 56 | request.state.assistant.request_origin = origin |
| 57 | return await call_next(request) |
| 58 | message = trans('i18n_permission.authenticate_invalid', msg = validator[1]) |
| 59 | return JSONResponse(message, status_code=401, headers={"Access-Control-Allow-Origin": "*"}) |
| 60 | #validate pass |
| 61 | tokenkey = settings.TOKEN_KEY |
| 62 | token = request.headers.get(tokenkey) |
| 63 | validate_pass, data = await self.validateToken(token, trans) |
| 64 | if validate_pass: |
| 65 | request.state.current_user = data |
| 66 | return await call_next(request) |
| 67 | |
| 68 | message = trans('i18n_permission.authenticate_invalid', msg = data) |
| 69 | return JSONResponse(message, status_code=401, headers={"Access-Control-Allow-Origin": "*"}) |
| 70 | |
| 71 | def is_options(self, request: Request): |
| 72 | return request.method == "OPTIONS" |
nothing calls this directly
no test coverage detected