-------------------- Authenticate to the Space and Time network, and store access_token and refresh_token. Returns: bool: Success flag (True/False) indicating the call worked as expected. object: Access_Token if successful, otherwise an error object.
(self)
| 423 | |
| 424 | |
| 425 | def authenticate(self) -> tuple[bool, object]: |
| 426 | """-------------------- |
| 427 | Authenticate to the Space and Time network, and store access_token and refresh_token. |
| 428 | |
| 429 | Returns: |
| 430 | bool: Success flag (True/False) indicating the call worked as expected. |
| 431 | object: Access_Token if successful, otherwise an error object. |
| 432 | """ |
| 433 | if not ((self.user_id and self.private_key) or self.api_key): |
| 434 | raise SxTArgumentError('Must have valid user_id and either api_key or private_key to authenticate.', logger=self.logger) |
| 435 | success = False |
| 436 | |
| 437 | try: |
| 438 | if self.private_key and self.user_id: |
| 439 | |
| 440 | success, response = self.base_api.get_auth_challenge_token(user_id = self.user_id) |
| 441 | if success: |
| 442 | challenge_token = response['authCode'] |
| 443 | signed_challenge_token = self.key_manager.sign_message(challenge_token) |
| 444 | success, response = self.base_api.get_access_token(user_id = self.user_id, |
| 445 | challange_token = challenge_token, |
| 446 | signed_challange_token = signed_challenge_token, |
| 447 | public_key = self.public_key) |
| 448 | |
| 449 | if not success and self.api_key: |
| 450 | success, response = self.base_api.gateway_proxy_auth_apikey(self.api_key) |
| 451 | |
| 452 | # either way, continue on processing tokens |
| 453 | if not success: raise SxTAuthenticationError(str(response), logger=self.logger) |
| 454 | |
| 455 | tokens = response |
| 456 | if len( [v for v in tokens if v in ['accessToken','refreshToken','accessTokenExpires','refreshTokenExpires']] ) < 4: |
| 457 | raise SxTAuthenticationError('Authentication produced incorrect / incomplete output', logger=self.logger) |
| 458 | except SxTAuthenticationError as ex: |
| 459 | return False, [ex] |
| 460 | |
| 461 | self.__settokens__(tokens['accessToken'], tokens['refreshToken'], tokens['accessTokenExpires'], tokens['refreshTokenExpires']) |
| 462 | if self.user_id =='': self.get_user_network_info() # will set user_id if missing |
| 463 | return True, self.access_token |
| 464 | |
| 465 | |
| 466 | def reauthenticate(self) -> str: |