Logs in using OAuth's auth code and PKCE flow, intended for when the user is on a device with a web browser and the ability to initialize the callback server
| 207 | |
| 208 | |
| 209 | class SameDeviceLoginTokenFetcher(BaseLoginTokenFetcher): |
| 210 | """ |
| 211 | Logs in using OAuth's auth code and PKCE flow, |
| 212 | intended for when the user is on a device with a web browser and the |
| 213 | ability to initialize the callback server |
| 214 | """ |
| 215 | |
| 216 | def __init__( |
| 217 | self, |
| 218 | client, |
| 219 | auth_code_fetcher, |
| 220 | on_pending_authorization, |
| 221 | private_key=None, |
| 222 | base_uri_builder=None, |
| 223 | ): |
| 224 | super().__init__( |
| 225 | client, |
| 226 | on_pending_authorization, |
| 227 | private_key=private_key, |
| 228 | base_uri_builder=base_uri_builder, |
| 229 | ) |
| 230 | self._auth_code_fetcher = auth_code_fetcher |
| 231 | |
| 232 | def fetch_token(self): |
| 233 | register_feature_id('LOGIN_SAME_DEVICE') |
| 234 | authorization_uri = self._get_authorization_uri( |
| 235 | client_id=CLIENT_ID[LoginType.SAME_DEVICE], |
| 236 | redirect_uri=self._auth_code_fetcher.redirect_uri_with_port(), |
| 237 | expected_state=self._expected_state, |
| 238 | code_challenge=self._code_challenge, |
| 239 | ) |
| 240 | |
| 241 | # Open/display the link, then block until the redirect uri is hit |
| 242 | self._on_pending_authorization( |
| 243 | **self._get_browser_handler_args(authorization_uri) |
| 244 | ) |
| 245 | auth_code, state = self._auth_code_fetcher.get_auth_code_and_state() |
| 246 | |
| 247 | if auth_code is None: |
| 248 | raise LoginAuthorizationCodeError( |
| 249 | error_msg='Failed to retrieve an authorization code.' |
| 250 | ) |
| 251 | |
| 252 | # The state we get back from the redirect is just a string, so |
| 253 | # cast our original UUID before comparing |
| 254 | if state != str(self._expected_state): |
| 255 | raise LoginAuthorizationCodeError( |
| 256 | error_msg=f'State parameter {state} does not match expected value {self._expected_state}.' |
| 257 | ) |
| 258 | |
| 259 | return self._exchange_auth_code_for_access_token( |
| 260 | client_id=CLIENT_ID[LoginType.SAME_DEVICE], |
| 261 | auth_code=auth_code, |
| 262 | redirect_uri=self._auth_code_fetcher.redirect_uri_with_port(), |
| 263 | ) |
| 264 | |
| 265 | |
| 266 | class CrossDeviceLoginTokenFetcher(BaseLoginTokenFetcher): |
no outgoing calls