Process the token refresh request.
(self, data: dict[str, str])
| 137 | return secrets.token_urlsafe(32) |
| 138 | |
| 139 | async def _process_token_refresh(self, data: dict[str, str]) -> web.Response: |
| 140 | """Process the token refresh request.""" |
| 141 | refresh_token = data.get("refresh_token") |
| 142 | |
| 143 | if not refresh_token: |
| 144 | return web.json_response({"error": "refresh_token required"}, status=400) |
| 145 | |
| 146 | # Hash the refresh token to look it up |
| 147 | refresh_token_hash = hashlib.sha256(refresh_token.encode()).hexdigest() |
| 148 | |
| 149 | if refresh_token_hash not in self.refresh_tokens_db: |
| 150 | return web.json_response({"error": "Invalid refresh token"}, status=401) |
| 151 | |
| 152 | user_data = self.refresh_tokens_db[refresh_token_hash] |
| 153 | |
| 154 | # Generate new access token |
| 155 | access_token = self.generate_access_token() |
| 156 | expires_in = 300 # 5 minutes for demo |
| 157 | |
| 158 | # Store the access token with expiry |
| 159 | token_hash = hashlib.sha256(access_token.encode()).hexdigest() |
| 160 | self.tokens_db[token_hash] = { |
| 161 | "user_id": user_data["user_id"], |
| 162 | "username": user_data["username"], |
| 163 | "expires_at": time.time() + expires_in, |
| 164 | "issued_at": time.time(), |
| 165 | } |
| 166 | |
| 167 | # Clean up expired tokens periodically |
| 168 | current_time = time.time() |
| 169 | self.tokens_db = { |
| 170 | k: v |
| 171 | for k, v in self.tokens_db.items() |
| 172 | if isinstance(v["expires_at"], float) and v["expires_at"] > current_time |
| 173 | } |
| 174 | |
| 175 | return web.json_response( |
| 176 | { |
| 177 | "access_token": access_token, |
| 178 | "token_type": "Bearer", |
| 179 | "expires_in": expires_in, |
| 180 | } |
| 181 | ) |
| 182 | |
| 183 | async def handle_token_refresh(self, request: web.Request) -> web.Response: |
| 184 | """Handle token refresh requests.""" |
no test coverage detected