AuthorizationHelpers.py 7.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157
  1. # Copyright (c) 2021 Ultimaker B.V.
  2. # Cura is released under the terms of the LGPLv3 or higher.
  3. from datetime import datetime
  4. import json
  5. import secrets
  6. from hashlib import sha512
  7. from base64 import b64encode
  8. from typing import Optional
  9. import requests
  10. from UM.i18n import i18nCatalog
  11. from UM.Logger import Logger
  12. from cura.OAuth2.Models import AuthenticationResponse, UserProfile, OAuth2Settings
  13. catalog = i18nCatalog("cura")
  14. TOKEN_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
  15. class AuthorizationHelpers:
  16. """Class containing several helpers to deal with the authorization flow."""
  17. def __init__(self, settings: "OAuth2Settings") -> None:
  18. self._settings = settings
  19. self._token_url = "{}/token".format(self._settings.OAUTH_SERVER_URL)
  20. @property
  21. def settings(self) -> "OAuth2Settings":
  22. """The OAuth2 settings object."""
  23. return self._settings
  24. def getAccessTokenUsingAuthorizationCode(self, authorization_code: str, verification_code: str) -> "AuthenticationResponse":
  25. """Request the access token from the authorization server.
  26. :param authorization_code: The authorization code from the 1st step.
  27. :param verification_code: The verification code needed for the PKCE extension.
  28. :return: An AuthenticationResponse object.
  29. """
  30. data = {
  31. "client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "",
  32. "redirect_uri": self._settings.CALLBACK_URL if self._settings.CALLBACK_URL is not None else "",
  33. "grant_type": "authorization_code",
  34. "code": authorization_code,
  35. "code_verifier": verification_code,
  36. "scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "",
  37. }
  38. try:
  39. return self.parseTokenResponse(requests.post(self._token_url, data = data)) # type: ignore
  40. except requests.exceptions.ConnectionError as connection_error:
  41. return AuthenticationResponse(success = False, err_message = f"Unable to connect to remote server: {connection_error}")
  42. def getAccessTokenUsingRefreshToken(self, refresh_token: str) -> "AuthenticationResponse":
  43. """Request the access token from the authorization server using a refresh token.
  44. :param refresh_token:
  45. :return: An AuthenticationResponse object.
  46. """
  47. Logger.log("d", "Refreshing the access token for [%s]", self._settings.OAUTH_SERVER_URL)
  48. data = {
  49. "client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "",
  50. "redirect_uri": self._settings.CALLBACK_URL if self._settings.CALLBACK_URL is not None else "",
  51. "grant_type": "refresh_token",
  52. "refresh_token": refresh_token,
  53. "scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "",
  54. }
  55. try:
  56. return self.parseTokenResponse(requests.post(self._token_url, data = data)) # type: ignore
  57. except requests.exceptions.ConnectionError:
  58. return AuthenticationResponse(success = False, err_message = "Unable to connect to remote server")
  59. except OSError as e:
  60. return AuthenticationResponse(success = False, err_message = "Operating system is unable to set up a secure connection: {err}".format(err = str(e)))
  61. @staticmethod
  62. def parseTokenResponse(token_response: requests.models.Response) -> "AuthenticationResponse":
  63. """Parse the token response from the authorization server into an AuthenticationResponse object.
  64. :param token_response: The JSON string data response from the authorization server.
  65. :return: An AuthenticationResponse object.
  66. """
  67. token_data = None
  68. try:
  69. token_data = json.loads(token_response.text)
  70. except ValueError:
  71. Logger.log("w", "Could not parse token response data: %s", token_response.text)
  72. if not token_data:
  73. return AuthenticationResponse(success = False, err_message = catalog.i18nc("@message", "Could not read response."))
  74. if token_response.status_code not in (200, 201):
  75. return AuthenticationResponse(success = False, err_message = token_data["error_description"])
  76. return AuthenticationResponse(success=True,
  77. token_type=token_data["token_type"],
  78. access_token=token_data["access_token"],
  79. refresh_token=token_data["refresh_token"],
  80. expires_in=token_data["expires_in"],
  81. scope=token_data["scope"],
  82. received_at=datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT))
  83. def parseJWT(self, access_token: str) -> Optional["UserProfile"]:
  84. """Calls the authentication API endpoint to get the token data.
  85. :param access_token: The encoded JWT token.
  86. :return: Dict containing some profile data.
  87. """
  88. try:
  89. check_token_url = "{}/check-token".format(self._settings.OAUTH_SERVER_URL)
  90. Logger.log("d", "Checking the access token for [%s]", check_token_url)
  91. token_request = requests.get(check_token_url, headers = {
  92. "Authorization": "Bearer {}".format(access_token)
  93. })
  94. except (requests.exceptions.ConnectionError, requests.exceptions.Timeout):
  95. # Connection was suddenly dropped. Nothing we can do about that.
  96. Logger.logException("w", "Something failed while attempting to parse the JWT token")
  97. return None
  98. if token_request.status_code not in (200, 201):
  99. Logger.log("w", "Could not retrieve token data from auth server: %s", token_request.text)
  100. return None
  101. user_data = token_request.json().get("data")
  102. if not user_data or not isinstance(user_data, dict):
  103. Logger.log("w", "Could not parse user data from token: %s", user_data)
  104. return None
  105. return UserProfile(
  106. user_id = user_data["user_id"],
  107. username = user_data["username"],
  108. profile_image_url = user_data.get("profile_image_url", ""),
  109. organization_id = user_data.get("organization", {}).get("organization_id"),
  110. subscriptions = user_data.get("subscriptions", [])
  111. )
  112. @staticmethod
  113. def generateVerificationCode(code_length: int = 32) -> str:
  114. """Generate a verification code of arbitrary length.
  115. :param code_length:: How long should the code be in bytes? This should never be lower than 16, but it's probably
  116. better to leave it at 32
  117. """
  118. return secrets.token_hex(code_length)
  119. @staticmethod
  120. def generateVerificationCodeChallenge(verification_code: str) -> str:
  121. """Generates a base64 encoded sha512 encrypted version of a given string.
  122. :param verification_code:
  123. :return: The encrypted code in base64 format.
  124. """
  125. encoded = sha512(verification_code.encode()).digest()
  126. return b64encode(encoded, altchars = b"_-").decode()