(self, jwt: str)
| 102 | await self.user_service.update(uid=user.id, data={"status": UserStatus.ACTIVE}) |
| 103 | |
| 104 | async def refresh(self, jwt: str) -> TokenBearer: |
| 105 | token = self.hash_service.verify_token(token=jwt) |
| 106 | if token is None: |
| 107 | raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_INVALID, message="Unauthorized!") |
| 108 | if token.token_type != TokenType.REFRESH: |
| 109 | raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_TYPE_INVALID, message="Unauthorized!") |
| 110 | |
| 111 | user = await self.user_service.get_by_id(uid=int(token.subject)) |
| 112 | if user is None: |
| 113 | raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_USER_NOT_FOUND, message="Unauthorized!") |
| 114 | if user.status != UserStatus.ACTIVE: |
| 115 | raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_USER_STATUS_NOT_ACTIVE, message="Unauthorized!") |
| 116 | |
| 117 | if user.session != token.session: |
| 118 | raise UnauthorizedException(error_no=ErrorNo.REFRESH_TOKEN_USER_SESSION_INVALID, message="Unauthorized!") |
| 119 | |
| 120 | user = await self.user_service.update(uid=user.id, data={"session": self.hash_service.random_string()}) |
| 121 | |
| 122 | return self.hash_service.create_token_bearer(user=user) |
no test coverage detected