(self, token: Optional[str], trans: I18n)
| 116 | return False, e |
| 117 | |
| 118 | async def validateToken(self, token: Optional[str], trans: I18n): |
| 119 | if not token: |
| 120 | return False, f"Miss Token[{settings.TOKEN_KEY}]!" |
| 121 | schema, param = get_authorization_scheme_param(token) |
| 122 | if schema.lower() != "bearer": |
| 123 | return False, f"Token schema error!" |
| 124 | try: |
| 125 | payload = jwt.decode( |
| 126 | param, settings.SECRET_KEY, algorithms=[security.ALGORITHM] |
| 127 | ) |
| 128 | token_data = TokenPayload(**payload) |
| 129 | with Session(engine) as session: |
| 130 | session_user = await get_user_info(session = session, user_id = token_data.id) |
| 131 | if not session_user: |
| 132 | message = trans('i18n_not_exist', msg = trans('i18n_user.account')) |
| 133 | raise Exception(message) |
| 134 | session_user = UserInfoDTO.model_validate(session_user) |
| 135 | if session_user.status != 1: |
| 136 | message = trans('i18n_login.user_disable', msg = trans('i18n_concat_admin')) |
| 137 | raise Exception(message) |
| 138 | if not session_user.oid or session_user.oid == 0: |
| 139 | message = trans('i18n_login.no_associated_ws', msg = trans('i18n_concat_admin')) |
| 140 | raise Exception(message) |
| 141 | return True, session_user |
| 142 | except Exception as e: |
| 143 | msg = str(e) |
| 144 | SQLBotLogUtil.exception(f"Token validation error: {msg}") |
| 145 | if 'expired' in msg: |
| 146 | return False, jwt.ExpiredSignatureError(trans('i18n_permission.token_expired')) |
| 147 | return False, e |
| 148 | |
| 149 | |
| 150 | async def validateAssistant(self, assistantToken: Optional[str], trans: I18n) -> tuple[any]: |
no test coverage detected