AuthorizationService.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222
  1. # Copyright (c) 2019 Ultimaker B.V.
  2. # Cura is released under the terms of the LGPLv3 or higher.
  3. import json
  4. from datetime import datetime, timedelta
  5. from typing import Optional, TYPE_CHECKING
  6. from urllib.parse import urlencode
  7. import requests.exceptions
  8. from PyQt5.QtCore import QUrl
  9. from PyQt5.QtGui import QDesktopServices
  10. from UM.Logger import Logger
  11. from UM.Message import Message
  12. from UM.Signal import Signal
  13. from cura.OAuth2.LocalAuthorizationServer import LocalAuthorizationServer
  14. from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers, TOKEN_TIMESTAMP_FORMAT
  15. from cura.OAuth2.Models import AuthenticationResponse
  16. from UM.i18n import i18nCatalog
  17. i18n_catalog = i18nCatalog("cura")
  18. if TYPE_CHECKING:
  19. from cura.OAuth2.Models import UserProfile, OAuth2Settings
  20. from UM.Preferences import Preferences
  21. ## The authorization service is responsible for handling the login flow,
  22. # storing user credentials and providing account information.
  23. class AuthorizationService:
  24. # Emit signal when authentication is completed.
  25. onAuthStateChanged = Signal()
  26. # Emit signal when authentication failed.
  27. onAuthenticationError = Signal()
  28. accessTokenChanged = Signal()
  29. def __init__(self, settings: "OAuth2Settings", preferences: Optional["Preferences"] = None) -> None:
  30. self._settings = settings
  31. self._auth_helpers = AuthorizationHelpers(settings)
  32. self._auth_url = "{}/authorize".format(self._settings.OAUTH_SERVER_URL)
  33. self._auth_data = None # type: Optional[AuthenticationResponse]
  34. self._user_profile = None # type: Optional["UserProfile"]
  35. self._preferences = preferences
  36. self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True)
  37. self._unable_to_get_data_message = None # type: Optional[Message]
  38. self.onAuthStateChanged.connect(self._authChanged)
  39. def _authChanged(self, logged_in):
  40. if logged_in and self._unable_to_get_data_message is not None:
  41. self._unable_to_get_data_message.hide()
  42. def initialize(self, preferences: Optional["Preferences"] = None) -> None:
  43. if preferences is not None:
  44. self._preferences = preferences
  45. if self._preferences:
  46. self._preferences.addPreference(self._settings.AUTH_DATA_PREFERENCE_KEY, "{}")
  47. ## Get the user profile as obtained from the JWT (JSON Web Token).
  48. # If the JWT is not yet parsed, calling this will take care of that.
  49. # \return UserProfile if a user is logged in, None otherwise.
  50. # \sa _parseJWT
  51. def getUserProfile(self) -> Optional["UserProfile"]:
  52. if not self._user_profile:
  53. # If no user profile was stored locally, we try to get it from JWT.
  54. try:
  55. self._user_profile = self._parseJWT()
  56. except requests.exceptions.ConnectionError:
  57. # Unable to get connection, can't login.
  58. Logger.logException("w", "Unable to validate user data with the remote server.")
  59. return None
  60. if not self._user_profile and self._auth_data:
  61. # If there is still no user profile from the JWT, we have to log in again.
  62. Logger.log("w", "The user profile could not be loaded. The user must log in again!")
  63. self.deleteAuthData()
  64. return None
  65. return self._user_profile
  66. ## Tries to parse the JWT (JSON Web Token) data, which it does if all the needed data is there.
  67. # \return UserProfile if it was able to parse, None otherwise.
  68. def _parseJWT(self) -> Optional["UserProfile"]:
  69. if not self._auth_data or self._auth_data.access_token is None:
  70. # If no auth data exists, we should always log in again.
  71. Logger.log("d", "There was no auth data or access token")
  72. return None
  73. user_data = self._auth_helpers.parseJWT(self._auth_data.access_token)
  74. if user_data:
  75. # If the profile was found, we return it immediately.
  76. return user_data
  77. # The JWT was expired or invalid and we should request a new one.
  78. if self._auth_data.refresh_token is None:
  79. Logger.log("w", "There was no refresh token in the auth data.")
  80. return None
  81. self._auth_data = self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token)
  82. if not self._auth_data or self._auth_data.access_token is None:
  83. Logger.log("w", "Unable to use the refresh token to get a new access token.")
  84. # The token could not be refreshed using the refresh token. We should login again.
  85. return None
  86. # Ensure it gets stored as otherwise we only have it in memory. The stored refresh token has been deleted
  87. # from the server already.
  88. self._storeAuthData(self._auth_data)
  89. return self._auth_helpers.parseJWT(self._auth_data.access_token)
  90. ## Get the access token as provided by the repsonse data.
  91. def getAccessToken(self) -> Optional[str]:
  92. if self._auth_data is None:
  93. Logger.log("d", "No auth data to retrieve the access_token from")
  94. return None
  95. # Check if the current access token is expired and refresh it if that is the case.
  96. # We have a fallback on a date far in the past for currently stored auth data in cura.cfg.
  97. received_at = datetime.strptime(self._auth_data.received_at, TOKEN_TIMESTAMP_FORMAT) \
  98. if self._auth_data.received_at else datetime(2000, 1, 1)
  99. expiry_date = received_at + timedelta(seconds = float(self._auth_data.expires_in or 0) - 60)
  100. if datetime.now() > expiry_date:
  101. self.refreshAccessToken()
  102. return self._auth_data.access_token if self._auth_data else None
  103. ## Try to refresh the access token. This should be used when it has expired.
  104. def refreshAccessToken(self) -> None:
  105. if self._auth_data is None or self._auth_data.refresh_token is None:
  106. Logger.log("w", "Unable to refresh access token, since there is no refresh token.")
  107. return
  108. response = self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token)
  109. if response.success:
  110. self._storeAuthData(response)
  111. self.onAuthStateChanged.emit(logged_in = True)
  112. else:
  113. Logger.log("w", "Failed to get a new access token from the server.")
  114. self.onAuthStateChanged.emit(logged_in = False)
  115. ## Delete the authentication data that we have stored locally (eg; logout)
  116. def deleteAuthData(self) -> None:
  117. if self._auth_data is not None:
  118. self._storeAuthData()
  119. self.onAuthStateChanged.emit(logged_in = False)
  120. ## Start the flow to become authenticated. This will start a new webbrowser tap, prompting the user to login.
  121. def startAuthorizationFlow(self) -> None:
  122. Logger.log("d", "Starting new OAuth2 flow...")
  123. # Create the tokens needed for the code challenge (PKCE) extension for OAuth2.
  124. # This is needed because the CuraDrivePlugin is a untrusted (open source) client.
  125. # More details can be found at https://tools.ietf.org/html/rfc7636.
  126. verification_code = self._auth_helpers.generateVerificationCode()
  127. challenge_code = self._auth_helpers.generateVerificationCodeChallenge(verification_code)
  128. state = AuthorizationHelpers.generateVerificationCode()
  129. # Create the query string needed for the OAuth2 flow.
  130. query_string = urlencode({
  131. "client_id": self._settings.CLIENT_ID,
  132. "redirect_uri": self._settings.CALLBACK_URL,
  133. "scope": self._settings.CLIENT_SCOPES,
  134. "response_type": "code",
  135. "state": state, # Forever in our Hearts, RIP "(.Y.)" (2018-2020)
  136. "code_challenge": challenge_code,
  137. "code_challenge_method": "S512"
  138. })
  139. # Open the authorization page in a new browser window.
  140. QDesktopServices.openUrl(QUrl("{}?{}".format(self._auth_url, query_string)))
  141. # Start a local web server to receive the callback URL on.
  142. self._server.start(verification_code, state)
  143. ## Callback method for the authentication flow.
  144. def _onAuthStateChanged(self, auth_response: AuthenticationResponse) -> None:
  145. if auth_response.success:
  146. self._storeAuthData(auth_response)
  147. self.onAuthStateChanged.emit(logged_in = True)
  148. else:
  149. self.onAuthenticationError.emit(logged_in = False, error_message = auth_response.err_message)
  150. self._server.stop() # Stop the web server at all times.
  151. ## Load authentication data from preferences.
  152. def loadAuthDataFromPreferences(self) -> None:
  153. if self._preferences is None:
  154. Logger.log("e", "Unable to load authentication data, since no preference has been set!")
  155. return
  156. try:
  157. preferences_data = json.loads(self._preferences.getValue(self._settings.AUTH_DATA_PREFERENCE_KEY))
  158. if preferences_data:
  159. self._auth_data = AuthenticationResponse(**preferences_data)
  160. # Also check if we can actually get the user profile information.
  161. user_profile = self.getUserProfile()
  162. if user_profile is not None:
  163. self.onAuthStateChanged.emit(logged_in = True)
  164. else:
  165. if self._unable_to_get_data_message is not None:
  166. self._unable_to_get_data_message.hide()
  167. self._unable_to_get_data_message = Message(i18n_catalog.i18nc("@info", "Unable to reach the Ultimaker account server."), title = i18n_catalog.i18nc("@info:title", "Warning"))
  168. self._unable_to_get_data_message.show()
  169. except ValueError:
  170. Logger.logException("w", "Could not load auth data from preferences")
  171. ## Store authentication data in preferences.
  172. def _storeAuthData(self, auth_data: Optional[AuthenticationResponse] = None) -> None:
  173. Logger.log("d", "Attempting to store the auth data")
  174. if self._preferences is None:
  175. Logger.log("e", "Unable to save authentication data, since no preference has been set!")
  176. return
  177. self._auth_data = auth_data
  178. if auth_data:
  179. self._user_profile = self.getUserProfile()
  180. self._preferences.setValue(self._settings.AUTH_DATA_PREFERENCE_KEY, json.dumps(vars(auth_data)))
  181. else:
  182. self._user_profile = None
  183. self._preferences.resetPreference(self._settings.AUTH_DATA_PREFERENCE_KEY)
  184. self.accessTokenChanged.emit()