GitHub-Proxy/Cura
1
0
Fork 0
mirror of https://git.mirrors.martin98.com/https://github.com/Ultimaker/Cura synced 2025-08-14 05:15:52 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #10901 from Ultimaker/CURA-8539_oauth_via_httprequestmanager

Browse Source 
OAuth via HttpRequestManager
This commit is contained in:
Jelle Spijker 2021-11-23 13:59:45 +01:00 committed by GitHub
commit 8427555d6a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 374 additions and 272 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

49
cura/API/Account.py
View File

@ -1,15 +1,15 @@
# Copyright (c) 2018 Ultimaker B.V. # Copyright (c) 2021 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher. # Cura is released under the terms of the LGPLv3 or higher.
from datetime import datetime
from typing import Any, Optional, Dict, TYPE_CHECKING, Callable
from datetime import datetime
from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlot, pyqtProperty, QTimer, Q_ENUMS from PyQt5.QtCore import QObject, pyqtSignal, pyqtSlot, pyqtProperty, QTimer, Q_ENUMS
from typing import Any, Optional, Dict, TYPE_CHECKING, Callable
from UM.Logger import Logger from UM.Logger import Logger
from UM.Message import Message from UM.Message import Message
from UM.i18n import i18nCatalog from UM.i18n import i18nCatalog
from cura.OAuth2.AuthorizationService import AuthorizationService from cura.OAuth2.AuthorizationService import AuthorizationService
from cura.OAuth2.Models import OAuth2Settings from cura.OAuth2.Models import OAuth2Settings, UserProfile
from cura.UltimakerCloud import UltimakerCloudConstants from cura.UltimakerCloud import UltimakerCloudConstants
if TYPE_CHECKING: if TYPE_CHECKING:
@ -46,6 +46,9 @@ class Account(QObject):
loginStateChanged = pyqtSignal(bool) loginStateChanged = pyqtSignal(bool)
"""Signal emitted when user logged in or out""" """Signal emitted when user logged in or out"""
userProfileChanged = pyqtSignal()
"""Signal emitted when new account information is available."""
additionalRightsChanged = pyqtSignal("QVariantMap") additionalRightsChanged = pyqtSignal("QVariantMap")
"""Signal emitted when a users additional rights change""" """Signal emitted when a users additional rights change"""
@ -71,13 +74,14 @@ class Account(QObject):
self._application = application self._application = application
self._new_cloud_printers_detected = False self._new_cloud_printers_detected = False
self._error_message = None # type: Optional[Message] self._error_message: Optional[Message] = None
self._logged_in = False self._logged_in = False
self._user_profile: Optional[UserProfile] = None
self._additional_rights: Dict[str, Any] = {} self._additional_rights: Dict[str, Any] = {}
self._sync_state = SyncState.IDLE self._sync_state = SyncState.IDLE
self._manual_sync_enabled = False self._manual_sync_enabled = False
self._update_packages_enabled = False self._update_packages_enabled = False
self._update_packages_action = None # type: Optional[Callable] self._update_packages_action: Optional[Callable] = None
self._last_sync_str = "-" self._last_sync_str = "-"
self._callback_port = 32118 self._callback_port = 32118
@ -103,7 +107,7 @@ class Account(QObject):
self._update_timer.setSingleShot(True) self._update_timer.setSingleShot(True)
self._update_timer.timeout.connect(self.sync) self._update_timer.timeout.connect(self.sync)
self._sync_services = {} # type: Dict[str, int] self._sync_services: Dict[str, int] = {}
"""contains entries "service_name" : SyncState""" """contains entries "service_name" : SyncState"""
def initialize(self) -> None: def initialize(self) -> None:
@ -196,12 +200,17 @@ class Account(QObject):
self._logged_in = logged_in self._logged_in = logged_in
self.loginStateChanged.emit(logged_in) self.loginStateChanged.emit(logged_in)
if logged_in: if logged_in:
self._authorization_service.getUserProfile(self._onProfileChanged)
self._setManualSyncEnabled(False) self._setManualSyncEnabled(False)
self._sync() self._sync()
else: else:
if self._update_timer.isActive(): if self._update_timer.isActive():
self._update_timer.stop() self._update_timer.stop()
def _onProfileChanged(self, profile: Optional[UserProfile]) -> None:
self._user_profile = profile
self.userProfileChanged.emit()
def _sync(self) -> None: def _sync(self) -> None:
"""Signals all sync services to start syncing """Signals all sync services to start syncing
@ -243,32 +252,28 @@ class Account(QObject):
return return
self._authorization_service.startAuthorizationFlow(force_logout_before_login) self._authorization_service.startAuthorizationFlow(force_logout_before_login)
@pyqtProperty(str, notify=loginStateChanged) @pyqtProperty(str, notify = userProfileChanged)
def userName(self): def userName(self):
user_profile = self._authorization_service.getUserProfile() if not self._user_profile:
if not user_profile: return ""
return None return self._user_profile.username
return user_profile.username
@pyqtProperty(str, notify = loginStateChanged) @pyqtProperty(str, notify = userProfileChanged)
def profileImageUrl(self): def profileImageUrl(self):
user_profile = self._authorization_service.getUserProfile() if not self._user_profile:
if not user_profile: return ""
return None return self._user_profile.profile_image_url
return user_profile.profile_image_url
@pyqtProperty(str, notify=accessTokenChanged) @pyqtProperty(str, notify=accessTokenChanged)
def accessToken(self) -> Optional[str]: def accessToken(self) -> Optional[str]:
return self._authorization_service.getAccessToken() return self._authorization_service.getAccessToken()
@pyqtProperty("QVariantMap", notify = loginStateChanged) @pyqtProperty("QVariantMap", notify = userProfileChanged)
def userProfile(self) -> Optional[Dict[str, Optional[str]]]: def userProfile(self) -> Optional[Dict[str, Optional[str]]]:
"""None if no user is logged in otherwise the logged in user as a dict containing containing user_id, username and profile_image_url """ """None if no user is logged in otherwise the logged in user as a dict containing containing user_id, username and profile_image_url """
if not self._user_profile:
user_profile = self._authorization_service.getUserProfile()
if not user_profile:
return None return None
return user_profile.__dict__ return self._user_profile.__dict__
@pyqtProperty(str, notify=lastSyncDateTimeChanged) @pyqtProperty(str, notify=lastSyncDateTimeChanged)
def lastSyncDateTime(self) -> str: def lastSyncDateTime(self) -> str:

177
cura/OAuth2/AuthorizationHelpers.py
View File

@ -1,18 +1,19 @@
# Copyright (c) 2021 Ultimaker B.V. # Copyright (c) 2021 Ultimaker B.V.
# Cura is released under the terms of the LGPLv3 or higher. # Cura is released under the terms of the LGPLv3 or higher.
from datetime import datetime
import json
import secrets
from hashlib import sha512
from base64 import b64encode from base64 import b64encode
from typing import Optional from datetime import datetime
import requests from hashlib import sha512
from PyQt5.QtNetwork import QNetworkReply
from UM.i18n import i18nCatalog import secrets
from UM.Logger import Logger from typing import Callable, Optional
import urllib.parse
from cura.OAuth2.Models import AuthenticationResponse, UserProfile, OAuth2Settings from cura.OAuth2.Models import AuthenticationResponse, UserProfile, OAuth2Settings
from UM.i18n import i18nCatalog
from UM.Logger import Logger
from UM.TaskManagement.HttpRequestManager import HttpRequestManager # To download log-in tokens.
catalog = i18nCatalog("cura") catalog = i18nCatalog("cura")
TOKEN_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S" TOKEN_TIMESTAMP_FORMAT = "%Y-%m-%d %H:%M:%S"
@ -30,14 +31,13 @@ class AuthorizationHelpers:
return self._settings return self._settings
def getAccessTokenUsingAuthorizationCode(self, authorization_code: str, verification_code: str) -> "AuthenticationResponse": def getAccessTokenUsingAuthorizationCode(self, authorization_code: str, verification_code: str, callback: Callable[[AuthenticationResponse], None]) -> None:
"""Request the access token from the authorization server. """
Request the access token from the authorization server.
:param authorization_code: The authorization code from the 1st step. :param authorization_code: The authorization code from the 1st step.
:param verification_code: The verification code needed for the PKCE extension. :param verification_code: The verification code needed for the PKCE extension.
:return: An AuthenticationResponse object. :param callback: Once the token has been obtained, this function will be called with the response.
""" """
data = { data = {
"client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "", "client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "",
"redirect_uri": self._settings.CALLBACK_URL if self._settings.CALLBACK_URL is not None else "", "redirect_uri": self._settings.CALLBACK_URL if self._settings.CALLBACK_URL is not None else "",
@ -46,18 +46,21 @@ class AuthorizationHelpers:
"code_verifier": verification_code, "code_verifier": verification_code,
"scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "", "scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "",
} }
try: headers = {"Content-type": "application/x-www-form-urlencoded"}
return self.parseTokenResponse(requests.post(self._token_url, data = data)) # type: ignore HttpRequestManager.getInstance().post(
except requests.exceptions.ConnectionError as connection_error: self._token_url,
return AuthenticationResponse(success = False, err_message = f"Unable to connect to remote server: {connection_error}") data = urllib.parse.urlencode(data).encode("UTF-8"),
headers_dict = headers,
callback = lambda response: self.parseTokenResponse(response, callback),
error_callback = lambda response, _: self.parseTokenResponse(response, callback)
)
def getAccessTokenUsingRefreshToken(self, refresh_token: str) -> "AuthenticationResponse": def getAccessTokenUsingRefreshToken(self, refresh_token: str, callback: Callable[[AuthenticationResponse], None]) -> None:
"""Request the access token from the authorization server using a refresh token. """
Request the access token from the authorization server using a refresh token.
:param refresh_token: :param refresh_token: A long-lived token used to refresh the authentication token.
:return: An AuthenticationResponse object. :param callback: Once the token has been obtained, this function will be called with the response.
""" """
Logger.log("d", "Refreshing the access token for [%s]", self._settings.OAUTH_SERVER_URL) Logger.log("d", "Refreshing the access token for [%s]", self._settings.OAUTH_SERVER_URL)
data = { data = {
"client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "", "client_id": self._settings.CLIENT_ID if self._settings.CLIENT_ID is not None else "",
@ -66,75 +69,99 @@ class AuthorizationHelpers:
"refresh_token": refresh_token, "refresh_token": refresh_token,
"scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "", "scope": self._settings.CLIENT_SCOPES if self._settings.CLIENT_SCOPES is not None else "",
} }
try: headers = {"Content-type": "application/x-www-form-urlencoded"}
return self.parseTokenResponse(requests.post(self._token_url, data = data)) # type: ignore HttpRequestManager.getInstance().post(
except requests.exceptions.ConnectionError: self._token_url,
return AuthenticationResponse(success = False, err_message = "Unable to connect to remote server") data = urllib.parse.urlencode(data).encode("UTF-8"),
except OSError as e: headers_dict = headers,
return AuthenticationResponse(success = False, err_message = "Operating system is unable to set up a secure connection: {err}".format(err = str(e))) callback = lambda response: self.parseTokenResponse(response, callback),
error_callback = lambda response, _: self.parseTokenResponse(response, callback)
)
@staticmethod def parseTokenResponse(self, token_response: QNetworkReply, callback: Callable[[AuthenticationResponse], None]) -> None:
def parseTokenResponse(token_response: requests.models.Response) -> "AuthenticationResponse":
"""Parse the token response from the authorization server into an AuthenticationResponse object. """Parse the token response from the authorization server into an AuthenticationResponse object.
:param token_response: The JSON string data response from the authorization server. :param token_response: The JSON string data response from the authorization server.
:return: An AuthenticationResponse object. :return: An AuthenticationResponse object.
""" """
token_data = HttpRequestManager.readJSON(token_response)
token_data = None
try:
token_data = json.loads(token_response.text)
except ValueError:
Logger.log("w", "Could not parse token response data: %s", token_response.text)
if not token_data: if not token_data:
return AuthenticationResponse(success = False, err_message = catalog.i18nc("@message", "Could not read response.")) callback(AuthenticationResponse(success = False, err_message = catalog.i18nc("@message", "Could not read response.")))
return
if token_response.status_code not in (200, 201): if token_response.error() != QNetworkReply.NetworkError.NoError:
return AuthenticationResponse(success = False, err_message = token_data["error_description"]) callback(AuthenticationResponse(success = False, err_message = token_data["error_description"]))
return
return AuthenticationResponse(success=True, callback(AuthenticationResponse(success = True,
token_type=token_data["token_type"], token_type = token_data["token_type"],
access_token=token_data["access_token"], access_token = token_data["access_token"],
refresh_token=token_data["refresh_token"], refresh_token = token_data["refresh_token"],
expires_in=token_data["expires_in"], expires_in = token_data["expires_in"],
scope=token_data["scope"], scope = token_data["scope"],
received_at=datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT)) received_at = datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT)))
return
def parseJWT(self, access_token: str) -> Optional["UserProfile"]: def checkToken(self, access_token: str, success_callback: Optional[Callable[[UserProfile], None]] = None, failed_callback: Optional[Callable[[], None]] = None) -> None:
"""Calls the authentication API endpoint to get the token data. """Calls the authentication API endpoint to get the token data.
The API is called asynchronously. When a response is given, the callback is called with the user's profile.
:param access_token: The encoded JWT token. :param access_token: The encoded JWT token.
:return: Dict containing some profile data. :param success_callback: When a response is given, this function will be called with a user profile. If None,
there will not be a callback.
:param failed_callback: When the request failed or the response didn't parse, this function will be called.
""" """
try:
check_token_url = "{}/check-token".format(self._settings.OAUTH_SERVER_URL) check_token_url = "{}/check-token".format(self._settings.OAUTH_SERVER_URL)
Logger.log("d", "Checking the access token for [%s]", check_token_url) Logger.log("d", "Checking the access token for [%s]", check_token_url)
token_request = requests.get(check_token_url, headers = { headers = {
"Authorization": "Bearer {}".format(access_token) "Authorization": f"Bearer {access_token}"
}) }
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout): HttpRequestManager.getInstance().get(
# Connection was suddenly dropped. Nothing we can do about that. check_token_url,
Logger.logException("w", "Something failed while attempting to parse the JWT token") headers_dict = headers,
return None callback = lambda reply: self._parseUserProfile(reply, success_callback, failed_callback),
if token_request.status_code not in (200, 201): error_callback = lambda _, _2: failed_callback() if failed_callback is not None else None
Logger.log("w", "Could not retrieve token data from auth server: %s", token_request.text)
return None
user_data = token_request.json().get("data")
if not user_data or not isinstance(user_data, dict):
Logger.log("w", "Could not parse user data from token: %s", user_data)
return None
return UserProfile(
user_id = user_data["user_id"],
username = user_data["username"],
profile_image_url = user_data.get("profile_image_url", ""),
organization_id = user_data.get("organization", {}).get("organization_id"),
subscriptions = user_data.get("subscriptions", [])
) )
def _parseUserProfile(self, reply: QNetworkReply, success_callback: Optional[Callable[[UserProfile], None]], failed_callback: Optional[Callable[[], None]] = None) -> None:
"""
Parses the user profile from a reply to /check-token.
If the response is valid, the callback will be called to return the user profile to the caller.
:param reply: A network reply to a request to the /check-token URL.
:param success_callback: A function to call once a user profile was successfully obtained.
:param failed_callback: A function to call if parsing the profile failed.
"""
if reply.error() != QNetworkReply.NetworkError.NoError:
Logger.warning(f"Could not access account information. QNetworkError {reply.errorString()}")
if failed_callback is not None:
failed_callback()
return
profile_data = HttpRequestManager.getInstance().readJSON(reply)
if profile_data is None or "data" not in profile_data:
Logger.warning("Could not parse user data from token.")
if failed_callback is not None:
failed_callback()
return
profile_data = profile_data["data"]
required_fields = {"user_id", "username"}
if "user_id" not in profile_data or "username" not in profile_data:
Logger.warning(f"User data missing required field(s): {required_fields - set(profile_data.keys())}")
if failed_callback is not None:
failed_callback()
return
if success_callback is not None:
success_callback(UserProfile(
user_id = profile_data["user_id"],
username = profile_data["username"],
profile_image_url = profile_data.get("profile_image_url", ""),
organization_id = profile_data.get("organization", {}).get("organization_id"),
subscriptions = profile_data.get("subscriptions", [])
))
@staticmethod @staticmethod
def generateVerificationCode(code_length: int = 32) -> str: def generateVerificationCode(code_length: int = 32) -> str:
"""Generate a verification code of arbitrary length. """Generate a verification code of arbitrary length.

28
cura/OAuth2/AuthorizationRequestHandler.py
View File

@ -2,6 +2,7 @@
# Cura is released under the terms of the LGPLv3 or higher. # Cura is released under the terms of the LGPLv3 or higher.
from http.server import BaseHTTPRequestHandler from http.server import BaseHTTPRequestHandler
from threading import Lock # To turn an asynchronous call synchronous.
from typing import Optional, Callable, Tuple, Dict, Any, List, TYPE_CHECKING from typing import Optional, Callable, Tuple, Dict, Any, List, TYPE_CHECKING
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
@ -14,6 +15,7 @@ if TYPE_CHECKING:
catalog = i18nCatalog("cura") catalog = i18nCatalog("cura")
class AuthorizationRequestHandler(BaseHTTPRequestHandler): class AuthorizationRequestHandler(BaseHTTPRequestHandler):
"""This handler handles all HTTP requests on the local web server. """This handler handles all HTTP requests on the local web server.
@ -24,11 +26,11 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
super().__init__(request, client_address, server) super().__init__(request, client_address, server)
# These values will be injected by the HTTPServer that this handler belongs to. # These values will be injected by the HTTPServer that this handler belongs to.
self.authorization_helpers = None # type: Optional[AuthorizationHelpers] self.authorization_helpers: Optional[AuthorizationHelpers] = None
self.authorization_callback = None # type: Optional[Callable[[AuthenticationResponse], None]] self.authorization_callback: Optional[Callable[[AuthenticationResponse], None]] = None
self.verification_code = None # type: Optional[str] self.verification_code: Optional[str] = None
self.state = None # type: Optional[str] self.state: Optional[str] = None
# CURA-6609: Some browser seems to issue a HEAD instead of GET request as the callback. # CURA-6609: Some browser seems to issue a HEAD instead of GET request as the callback.
def do_HEAD(self) -> None: def do_HEAD(self) -> None:
@ -70,13 +72,23 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
if state != self.state: if state != self.state:
token_response = AuthenticationResponse( token_response = AuthenticationResponse(
success = False, success = False,
err_message=catalog.i18nc("@message", err_message = catalog.i18nc("@message", "The provided state is not correct.")
"The provided state is not correct.")
) )
elif code and self.authorization_helpers is not None and self.verification_code is not None: elif code and self.authorization_helpers is not None and self.verification_code is not None:
token_response = AuthenticationResponse(
success = False,
err_message = catalog.i18nc("@message", "Timeout when authenticating with the account server.")
)
# If the code was returned we get the access token. # If the code was returned we get the access token.
token_response = self.authorization_helpers.getAccessTokenUsingAuthorizationCode( lock = Lock()
code, self.verification_code) lock.acquire()
def callback(response: AuthenticationResponse) -> None:
nonlocal token_response
token_response = response
lock.release()
self.authorization_helpers.getAccessTokenUsingAuthorizationCode(code, self.verification_code, callback)
lock.acquire(timeout = 60) # Block thread until request is completed (which releases the lock). If not acquired, the timeout message stays.
elif self._queryGet(query, "error_code") == "user_denied": elif self._queryGet(query, "error_code") == "user_denied":
# Otherwise we show an error message (probably the user clicked "Deny" in the auth dialog). # Otherwise we show an error message (probably the user clicked "Deny" in the auth dialog).

150
cura/OAuth2/AuthorizationService.py
View File

@ -3,10 +3,9 @@
import json import json
from datetime import datetime, timedelta from datetime import datetime, timedelta
from typing import Optional, TYPE_CHECKING, Dict from typing import Callable, Dict, Optional, TYPE_CHECKING, Union
from urllib.parse import urlencode, quote_plus from urllib.parse import urlencode, quote_plus
import requests.exceptions
from PyQt5.QtCore import QUrl from PyQt5.QtCore import QUrl
from PyQt5.QtGui import QDesktopServices from PyQt5.QtGui import QDesktopServices
@ -16,7 +15,7 @@ from UM.Signal import Signal
from UM.i18n import i18nCatalog from UM.i18n import i18nCatalog
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers, TOKEN_TIMESTAMP_FORMAT from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers, TOKEN_TIMESTAMP_FORMAT
from cura.OAuth2.LocalAuthorizationServer import LocalAuthorizationServer from cura.OAuth2.LocalAuthorizationServer import LocalAuthorizationServer
from cura.OAuth2.Models import AuthenticationResponse from cura.OAuth2.Models import AuthenticationResponse, BaseModel
i18n_catalog = i18nCatalog("cura") i18n_catalog = i18nCatalog("cura")
@ -26,6 +25,7 @@ if TYPE_CHECKING:
MYCLOUD_LOGOFF_URL = "https://account.ultimaker.com/logoff?utm_source=cura&utm_medium=software&utm_campaign=change-account-before-adding-printers" MYCLOUD_LOGOFF_URL = "https://account.ultimaker.com/logoff?utm_source=cura&utm_medium=software&utm_campaign=change-account-before-adding-printers"
class AuthorizationService: class AuthorizationService:
"""The authorization service is responsible for handling the login flow, storing user credentials and providing """The authorization service is responsible for handling the login flow, storing user credentials and providing
account information. account information.
@ -43,12 +43,13 @@ class AuthorizationService:
self._settings = settings self._settings = settings
self._auth_helpers = AuthorizationHelpers(settings) self._auth_helpers = AuthorizationHelpers(settings)
self._auth_url = "{}/authorize".format(self._settings.OAUTH_SERVER_URL) self._auth_url = "{}/authorize".format(self._settings.OAUTH_SERVER_URL)
self._auth_data = None # type: Optional[AuthenticationResponse] self._auth_data: Optional[AuthenticationResponse] = None
self._user_profile = None # type: Optional["UserProfile"] self._user_profile: Optional["UserProfile"] = None
self._preferences = preferences self._preferences = preferences
self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True) self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True)
self._currently_refreshing_token = False # Whether we are currently in the process of refreshing auth. Don't make new requests while busy.
self._unable_to_get_data_message = None # type: Optional[Message] self._unable_to_get_data_message: Optional[Message] = None
self.onAuthStateChanged.connect(self._authChanged) self.onAuthStateChanged.connect(self._authChanged)
@ -62,69 +63,80 @@ class AuthorizationService:
if self._preferences: if self._preferences:
self._preferences.addPreference(self._settings.AUTH_DATA_PREFERENCE_KEY, "{}") self._preferences.addPreference(self._settings.AUTH_DATA_PREFERENCE_KEY, "{}")
def getUserProfile(self) -> Optional["UserProfile"]: def getUserProfile(self, callback: Optional[Callable[[Optional["UserProfile"]], None]] = None) -> None:
"""Get the user profile as obtained from the JWT (JSON Web Token). """
Get the user profile as obtained from the JWT (JSON Web Token).
If the JWT is not yet parsed, calling this will take care of that. If the JWT is not yet checked and parsed, calling this will take care of that.
:param callback: Once the user profile is obtained, this function will be called with the given user profile. If
:return: UserProfile if a user is logged in, None otherwise. the profile fails to be obtained, this function will be called with None.
See also: :py:method:`cura.OAuth2.AuthorizationService.AuthorizationService._parseJWT` See also: :py:method:`cura.OAuth2.AuthorizationService.AuthorizationService._parseJWT`
""" """
if self._user_profile:
# We already obtained the profile. No need to make another request for it.
if callback is not None:
callback(self._user_profile)
return
if not self._user_profile:
# If no user profile was stored locally, we try to get it from JWT. # If no user profile was stored locally, we try to get it from JWT.
try: def store_profile(profile: Optional["UserProfile"]) -> None:
self._user_profile = self._parseJWT() if profile is not None:
except requests.exceptions.ConnectionError: self._user_profile = profile
# Unable to get connection, can't login. if callback is not None:
Logger.logException("w", "Unable to validate user data with the remote server.") callback(profile)
return None elif self._auth_data:
# If there is no user profile from the JWT, we have to log in again.
if not self._user_profile and self._auth_data: Logger.warning("The user profile could not be loaded. The user must log in again!")
# If there is still no user profile from the JWT, we have to log in again.
Logger.log("w", "The user profile could not be loaded. The user must log in again!")
self.deleteAuthData() self.deleteAuthData()
return None if callback is not None:
callback(None)
else:
if callback is not None:
callback(None)
return self._user_profile self._parseJWT(callback = store_profile)
def _parseJWT(self) -> Optional["UserProfile"]: def _parseJWT(self, callback: Callable[[Optional["UserProfile"]], None]) -> None:
"""Tries to parse the JWT (JSON Web Token) data, which it does if all the needed data is there. """
Tries to parse the JWT (JSON Web Token) data, which it does if all the needed data is there.
:return: UserProfile if it was able to parse, None otherwise. :param callback: A function to call asynchronously once the user profile has been obtained. It will be called
with `None` if it failed to obtain a user profile.
""" """
if not self._auth_data or self._auth_data.access_token is None: if not self._auth_data or self._auth_data.access_token is None:
# If no auth data exists, we should always log in again. # If no auth data exists, we should always log in again.
Logger.log("d", "There was no auth data or access token") Logger.debug("There was no auth data or access token")
return None callback(None)
return
try: # When we checked the token we may get a user profile. This callback checks if that is a valid one and tries to refresh the token if it's not.
user_data = self._auth_helpers.parseJWT(self._auth_data.access_token) def check_user_profile(user_profile: Optional["UserProfile"]) -> None:
except AttributeError: if user_profile:
# THis might seem a bit double, but we get crash reports about this (CURA-2N2 in sentry) # If the profile was found, we call it back immediately.
Logger.log("d", "There was no auth data or access token") callback(user_profile)
return None return
if user_data:
# If the profile was found, we return it immediately.
return user_data
# The JWT was expired or invalid and we should request a new one. # The JWT was expired or invalid and we should request a new one.
if self._auth_data.refresh_token is None: if self._auth_data is None or self._auth_data.refresh_token is None:
Logger.log("w", "There was no refresh token in the auth data.") Logger.warning("There was no refresh token in the auth data.")
return None callback(None)
self._auth_data = self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token) return
if not self._auth_data or self._auth_data.access_token is None:
Logger.log("w", "Unable to use the refresh token to get a new access token.") def process_auth_data(auth_data: AuthenticationResponse) -> None:
# The token could not be refreshed using the refresh token. We should login again. if auth_data.access_token is None:
return None Logger.warning("Unable to use the refresh token to get a new access token.")
# Ensure it gets stored as otherwise we only have it in memory. The stored refresh token has been deleted callback(None)
# from the server already. Do not store the auth_data if we could not get new auth_data (eg due to a return
# network error), since this would cause an infinite loop trying to get new auth-data # Ensure it gets stored as otherwise we only have it in memory. The stored refresh token has been
if self._auth_data.success: # deleted from the server already. Do not store the auth_data if we could not get new auth_data (e.g.
self._storeAuthData(self._auth_data) # due to a network error), since this would cause an infinite loop trying to get new auth-data.
return self._auth_helpers.parseJWT(self._auth_data.access_token) if auth_data.success:
self._storeAuthData(auth_data)
self._auth_helpers.checkToken(auth_data.access_token, callback, lambda: callback(None))
self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token, process_auth_data)
self._auth_helpers.checkToken(self._auth_data.access_token, check_user_profile, lambda: check_user_profile(None))
def getAccessToken(self) -> Optional[str]: def getAccessToken(self) -> Optional[str]:
"""Get the access token as provided by the response data.""" """Get the access token as provided by the response data."""
@ -149,14 +161,21 @@ class AuthorizationService:
if self._auth_data is None or self._auth_data.refresh_token is None: if self._auth_data is None or self._auth_data.refresh_token is None:
Logger.log("w", "Unable to refresh access token, since there is no refresh token.") Logger.log("w", "Unable to refresh access token, since there is no refresh token.")
return return
response = self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token)
def process_auth_data(response: AuthenticationResponse) -> None:
if response.success: if response.success:
self._storeAuthData(response) self._storeAuthData(response)
self.onAuthStateChanged.emit(logged_in = True) self.onAuthStateChanged.emit(logged_in = True)
else: else:
Logger.log("w", "Failed to get a new access token from the server.") Logger.warning("Failed to get a new access token from the server.")
self.onAuthStateChanged.emit(logged_in = False) self.onAuthStateChanged.emit(logged_in = False)
if self._currently_refreshing_token:
Logger.debug("Was already busy refreshing token. Do not start a new request.")
return
self._currently_refreshing_token = True
self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token, process_auth_data)
def deleteAuthData(self) -> None: def deleteAuthData(self) -> None:
"""Delete the authentication data that we have stored locally (eg; logout)""" """Delete the authentication data that we have stored locally (eg; logout)"""
@ -244,21 +263,23 @@ class AuthorizationService:
preferences_data = json.loads(self._preferences.getValue(self._settings.AUTH_DATA_PREFERENCE_KEY)) preferences_data = json.loads(self._preferences.getValue(self._settings.AUTH_DATA_PREFERENCE_KEY))
if preferences_data: if preferences_data:
self._auth_data = AuthenticationResponse(**preferences_data) self._auth_data = AuthenticationResponse(**preferences_data)
# Also check if we can actually get the user profile information. # Also check if we can actually get the user profile information.
user_profile = self.getUserProfile() def callback(profile: Optional["UserProfile"]) -> None:
if user_profile is not None: if profile is not None:
self.onAuthStateChanged.emit(logged_in = True) self.onAuthStateChanged.emit(logged_in = True)
Logger.log("d", "Auth data was successfully loaded") Logger.debug("Auth data was successfully loaded")
else: else:
if self._unable_to_get_data_message is not None: if self._unable_to_get_data_message is not None:
self._unable_to_get_data_message.hide() self._unable_to_get_data_message.show()
else:
self._unable_to_get_data_message = Message(i18n_catalog.i18nc("@info", self._unable_to_get_data_message = Message(i18n_catalog.i18nc("@info",
"Unable to reach the Ultimaker account server."), "Unable to reach the Ultimaker account server."),
title = i18n_catalog.i18nc("@info:title", "Warning"), title = i18n_catalog.i18nc("@info:title", "Log-in failed"),
message_type = Message.MessageType.ERROR) message_type = Message.MessageType.ERROR)
Logger.log("w", "Unable to load auth data from preferences") Logger.warning("Unable to get user profile using auth data from preferences.")
self._unable_to_get_data_message.show() self._unable_to_get_data_message.show()
self.getUserProfile(callback)
except (ValueError, TypeError): except (ValueError, TypeError):
Logger.logException("w", "Could not load auth data from preferences") Logger.logException("w", "Could not load auth data from preferences")
@ -271,8 +292,9 @@ class AuthorizationService:
return return
self._auth_data = auth_data self._auth_data = auth_data
self._currently_refreshing_token = False
if auth_data: if auth_data:
self._user_profile = self.getUserProfile() self.getUserProfile()
self._preferences.setValue(self._settings.AUTH_DATA_PREFERENCE_KEY, json.dumps(auth_data.dump())) self._preferences.setValue(self._settings.AUTH_DATA_PREFERENCE_KEY, json.dumps(auth_data.dump()))
else: else:
Logger.log("d", "Clearing the user profile") Logger.log("d", "Clearing the user profile")

2
plugins/Toolbox/src/CloudSync/CloudPackageChecker.py
View File

@ -66,7 +66,7 @@ class CloudPackageChecker(QObject):
self._application.getHttpRequestManager().get(url, self._application.getHttpRequestManager().get(url,
callback = self._onUserPackagesRequestFinished, callback = self._onUserPackagesRequestFinished,
error_callback = self._onUserPackagesRequestFinished, error_callback = self._onUserPackagesRequestFinished,
timeout=10, timeout = 10,
scope = self._scope) scope = self._scope)
def _onUserPackagesRequestFinished(self, reply: "QNetworkReply", error: Optional["QNetworkReply.NetworkError"] = None) -> None: def _onUserPackagesRequestFinished(self, reply: "QNetworkReply", error: Optional["QNetworkReply.NetworkError"] = None) -> None:

40
tests/API/TestAccount.py
View File

@ -80,46 +80,6 @@ def test_errorLoginState(application):
account._onLoginStateChanged(False, "OMGZOMG!") account._onLoginStateChanged(False, "OMGZOMG!")
account.loginStateChanged.emit.called_with(False) account.loginStateChanged.emit.called_with(False)
def test_userName(user_profile):
account = Account(MagicMock())
mocked_auth_service = MagicMock()
account._authorization_service = mocked_auth_service
mocked_auth_service.getUserProfile = MagicMock(return_value = user_profile)
assert account.userName == "username!"
mocked_auth_service.getUserProfile = MagicMock(return_value=None)
assert account.userName is None
def test_profileImageUrl(user_profile):
account = Account(MagicMock())
mocked_auth_service = MagicMock()
account._authorization_service = mocked_auth_service
mocked_auth_service.getUserProfile = MagicMock(return_value = user_profile)
assert account.profileImageUrl == "profile_image_url!"
mocked_auth_service.getUserProfile = MagicMock(return_value=None)
assert account.profileImageUrl is None
def test_userProfile(user_profile):
account = Account(MagicMock())
mocked_auth_service = MagicMock()
account._authorization_service = mocked_auth_service
mocked_auth_service.getUserProfile = MagicMock(return_value=user_profile)
returned_user_profile = account.userProfile
assert returned_user_profile["username"] == "username!"
assert returned_user_profile["profile_image_url"] == "profile_image_url!"
assert returned_user_profile["user_id"] == "user_id!"
mocked_auth_service.getUserProfile = MagicMock(return_value=None)
assert account.userProfile is None
def test_sync_success(): def test_sync_success():
account = Account(MagicMock()) account = Account(MagicMock())

160
tests/TestOAuth2.py
View File

@ -1,9 +1,11 @@
from datetime import datetime # Copyright (c) 2021 Ultimaker B.V.
from unittest.mock import MagicMock, patch # Cura is released under the terms of the LGPLv3 or higher.
import requests from datetime import datetime
from unittest.mock import MagicMock, Mock, patch
from PyQt5.QtGui import QDesktopServices from PyQt5.QtGui import QDesktopServices
from PyQt5.QtNetwork import QNetworkReply
from UM.Preferences import Preferences from UM.Preferences import Preferences
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers, TOKEN_TIMESTAMP_FORMAT from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers, TOKEN_TIMESTAMP_FORMAT
@ -39,6 +41,14 @@ SUCCESSFUL_AUTH_RESPONSE = AuthenticationResponse(
success = True success = True
) )
EXPIRED_AUTH_RESPONSE = AuthenticationResponse(
access_token = "expired",
refresh_token = "beep?",
received_at = datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT),
expires_in = 300, # 5 minutes should be more than enough for testing
success = True
)
NO_REFRESH_AUTH_RESPONSE = AuthenticationResponse( NO_REFRESH_AUTH_RESPONSE = AuthenticationResponse(
access_token = "beep", access_token = "beep",
received_at = datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT), received_at = datetime.now().strftime(TOKEN_TIMESTAMP_FORMAT),
@ -50,12 +60,17 @@ MALFORMED_AUTH_RESPONSE = AuthenticationResponse(success=False)
def test_cleanAuthService() -> None: def test_cleanAuthService() -> None:
# Ensure that when setting up an AuthorizationService, no data is set. """
Ensure that when setting up an AuthorizationService, no data is set.
"""
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize() authorization_service.initialize()
assert authorization_service.getUserProfile() is None
assert authorization_service.getAccessToken() is None
mock_callback = Mock()
authorization_service.getUserProfile(mock_callback)
mock_callback.assert_called_once_with(None)
assert authorization_service.getAccessToken() is None
def test_refreshAccessTokenSuccess(): def test_refreshAccessTokenSuccess():
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
@ -68,34 +83,83 @@ def test_refreshAccessTokenSuccess():
authorization_service.refreshAccessToken() authorization_service.refreshAccessToken()
assert authorization_service.onAuthStateChanged.emit.called_with(True) assert authorization_service.onAuthStateChanged.emit.called_with(True)
def test__parseJWTNoRefreshToken(): def test__parseJWTNoRefreshToken():
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) """
with patch.object(AuthorizationService, "getUserProfile", return_value=UserProfile()): Tests parsing the user profile if there is no refresh token stored, but there is a normal authentication token.
authorization_service._storeAuthData(NO_REFRESH_AUTH_RESPONSE)
assert authorization_service._parseJWT() is None
The request for the user profile using the authentication token should still work normally.
"""
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
with patch.object(AuthorizationService, "getUserProfile", return_value = UserProfile()):
authorization_service._storeAuthData(NO_REFRESH_AUTH_RESPONSE)
mock_callback = Mock() # To log the final profile response.
mock_reply = Mock() # The user profile that the service should respond with.
mock_reply.error = Mock(return_value = QNetworkReply.NetworkError.NoError)
http_mock = Mock()
http_mock.get = lambda url, headers_dict, callback, error_callback: callback(mock_reply)
http_mock.readJSON = Mock(return_value = {"data": {"user_id": "id_ego_or_superego", "username": "Ghostkeeper"}})
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
authorization_service._parseJWT(mock_callback)
mock_callback.assert_called_once()
profile_reply = mock_callback.call_args_list[0][0][0]
assert profile_reply.user_id == "id_ego_or_superego"
assert profile_reply.username == "Ghostkeeper"
def test__parseJWTFailOnRefresh(): def test__parseJWTFailOnRefresh():
"""
Tries to refresh the authentication token using an invalid refresh token. The request should fail.
"""
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
with patch.object(AuthorizationService, "getUserProfile", return_value=UserProfile()): with patch.object(AuthorizationService, "getUserProfile", return_value = UserProfile()):
authorization_service._storeAuthData(SUCCESSFUL_AUTH_RESPONSE) authorization_service._storeAuthData(SUCCESSFUL_AUTH_RESPONSE)
with patch.object(AuthorizationHelpers, "getAccessTokenUsingRefreshToken", return_value=FAILED_AUTH_RESPONSE): mock_callback = Mock() # To log the final profile response.
assert authorization_service._parseJWT() is None mock_reply = Mock() # The response that the request should give, containing an error about it failing to authenticate.
mock_reply.error = Mock(return_value = QNetworkReply.NetworkError.AuthenticationRequiredError) # The reply is 403: Authentication required, meaning the server responded with a "Can't do that, Dave".
http_mock = Mock()
http_mock.get = lambda url, headers_dict, callback, error_callback: callback(mock_reply)
http_mock.post = lambda url, data, headers_dict, callback, error_callback: callback(mock_reply)
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.readJSON", Mock(return_value = {"error_description": "Mock a failed request!"})):
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
authorization_service._parseJWT(mock_callback)
mock_callback.assert_called_once_with(None)
def test__parseJWTSucceedOnRefresh(): def test__parseJWTSucceedOnRefresh():
"""
Tries to refresh the authentication token using a valid refresh token. The request should succeed.
"""
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize() authorization_service.initialize()
with patch.object(AuthorizationService, "getUserProfile", return_value=UserProfile()): with patch.object(AuthorizationService, "getUserProfile", return_value = UserProfile()):
authorization_service._storeAuthData(SUCCESSFUL_AUTH_RESPONSE) authorization_service._storeAuthData(EXPIRED_AUTH_RESPONSE)
with patch.object(AuthorizationHelpers, "getAccessTokenUsingRefreshToken", return_value=SUCCESSFUL_AUTH_RESPONSE): mock_callback = Mock() # To log the final profile response.
with patch.object(AuthorizationHelpers, "parseJWT", MagicMock(return_value = None)) as mocked_parseJWT: mock_reply_success = Mock() # The reply should be a failure when using the expired access token, but succeed when using the refresh token.
authorization_service._parseJWT() mock_reply_success.error = Mock(return_value = QNetworkReply.NetworkError.NoError)
mocked_parseJWT.assert_called_with("beep") mock_reply_failure = Mock()
mock_reply_failure.error = Mock(return_value = QNetworkReply.NetworkError.AuthenticationRequiredError)
http_mock = Mock()
def mock_get(url, headers_dict, callback, error_callback):
if(headers_dict == {"Authorization": "Bearer beep"}):
callback(mock_reply_success)
else:
callback(mock_reply_failure)
http_mock.get = mock_get
http_mock.readJSON = Mock(return_value = {"data": {"user_id": "user_idea", "username": "Ghostkeeper"}})
def mock_refresh(self, refresh_token, callback): # Refreshing gives a valid token.
callback(SUCCESSFUL_AUTH_RESPONSE)
with patch("cura.OAuth2.AuthorizationHelpers.AuthorizationHelpers.getAccessTokenUsingRefreshToken", mock_refresh):
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
authorization_service._parseJWT(mock_callback)
mock_callback.assert_called_once()
profile_reply = mock_callback.call_args_list[0][0][0]
assert profile_reply.user_id == "user_idea"
assert profile_reply.username == "Ghostkeeper"
def test_initialize(): def test_initialize():
original_preference = MagicMock() original_preference = MagicMock()
@ -105,18 +169,29 @@ def test_initialize():
initialize_preferences.addPreference.assert_called_once_with("test/auth_data", "{}") initialize_preferences.addPreference.assert_called_once_with("test/auth_data", "{}")
original_preference.addPreference.assert_not_called() original_preference.addPreference.assert_not_called()
def test_refreshAccessTokenFailed(): def test_refreshAccessTokenFailed():
"""
Test if the authentication is reset once the refresh token fails to refresh access.
"""
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize() authorization_service.initialize()
with patch.object(AuthorizationService, "getUserProfile", return_value=UserProfile()):
def mock_refresh(self, refresh_token, callback): # Refreshing gives a valid token.
callback(FAILED_AUTH_RESPONSE)
mock_reply = Mock() # The response that the request should give, containing an error about it failing to authenticate.
mock_reply.error = Mock(return_value = QNetworkReply.NetworkError.AuthenticationRequiredError) # The reply is 403: Authentication required, meaning the server responded with a "Can't do that, Dave".
http_mock = Mock()
http_mock.get = lambda url, headers_dict, callback, error_callback: callback(mock_reply)
http_mock.post = lambda url, data, headers_dict, callback, error_callback: callback(mock_reply)
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.readJSON", Mock(return_value = {"error_description": "Mock a failed request!"})):
with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
authorization_service._storeAuthData(SUCCESSFUL_AUTH_RESPONSE) authorization_service._storeAuthData(SUCCESSFUL_AUTH_RESPONSE)
authorization_service.onAuthStateChanged.emit = MagicMock() authorization_service.onAuthStateChanged.emit = MagicMock()
with patch.object(AuthorizationHelpers, "getAccessTokenUsingRefreshToken", return_value=FAILED_AUTH_RESPONSE): with patch("cura.OAuth2.AuthorizationHelpers.AuthorizationHelpers.getAccessTokenUsingRefreshToken", mock_refresh):
authorization_service.refreshAccessToken() authorization_service.refreshAccessToken()
assert authorization_service.onAuthStateChanged.emit.called_with(False) assert authorization_service.onAuthStateChanged.emit.called_with(False)
def test_refreshAccesTokenWithoutData(): def test_refreshAccesTokenWithoutData():
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize() authorization_service.initialize()
@ -124,14 +199,6 @@ def test_refreshAccesTokenWithoutData():
authorization_service.refreshAccessToken() authorization_service.refreshAccessToken()
authorization_service.onAuthStateChanged.emit.assert_not_called() authorization_service.onAuthStateChanged.emit.assert_not_called()
def test_userProfileException():
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize()
authorization_service._parseJWT = MagicMock(side_effect=requests.exceptions.ConnectionError)
assert authorization_service.getUserProfile() is None
def test_failedLogin() -> None: def test_failedLogin() -> None:
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.onAuthenticationError.emit = MagicMock() authorization_service.onAuthenticationError.emit = MagicMock()
@ -151,8 +218,7 @@ def test_failedLogin() -> None:
assert authorization_service.getUserProfile() is None assert authorization_service.getUserProfile() is None
assert authorization_service.getAccessToken() is None assert authorization_service.getAccessToken() is None
@patch.object(AuthorizationService, "getUserProfile")
@patch.object(AuthorizationService, "getUserProfile", return_value=UserProfile())
def test_storeAuthData(get_user_profile) -> None: def test_storeAuthData(get_user_profile) -> None:
preferences = Preferences() preferences = Preferences()
authorization_service = AuthorizationService(OAUTH_SETTINGS, preferences) authorization_service = AuthorizationService(OAUTH_SETTINGS, preferences)
@ -170,7 +236,6 @@ def test_storeAuthData(get_user_profile) -> None:
second_auth_service.loadAuthDataFromPreferences() second_auth_service.loadAuthDataFromPreferences()
assert second_auth_service.getAccessToken() == SUCCESSFUL_AUTH_RESPONSE.access_token assert second_auth_service.getAccessToken() == SUCCESSFUL_AUTH_RESPONSE.access_token
@patch.object(LocalAuthorizationServer, "stop") @patch.object(LocalAuthorizationServer, "stop")
@patch.object(LocalAuthorizationServer, "start") @patch.object(LocalAuthorizationServer, "start")
@patch.object(QDesktopServices, "openUrl") @patch.object(QDesktopServices, "openUrl")
@ -188,7 +253,6 @@ def test_localAuthServer(QDesktopServices_openUrl, start_auth_server, stop_auth_
# Ensure that it stopped the server. # Ensure that it stopped the server.
assert stop_auth_server.call_count == 1 assert stop_auth_server.call_count == 1
def test_loginAndLogout() -> None: def test_loginAndLogout() -> None:
preferences = Preferences() preferences = Preferences()
authorization_service = AuthorizationService(OAUTH_SETTINGS, preferences) authorization_service = AuthorizationService(OAUTH_SETTINGS, preferences)
@ -196,8 +260,14 @@ def test_loginAndLogout() -> None:
authorization_service.onAuthStateChanged.emit = MagicMock() authorization_service.onAuthStateChanged.emit = MagicMock()
authorization_service.initialize() authorization_service.initialize()
mock_reply = Mock() # The user profile that the service should respond with.
mock_reply.error = Mock(return_value = QNetworkReply.NetworkError.NoError)
http_mock = Mock()
http_mock.get = lambda url, headers_dict, callback, error_callback: callback(mock_reply)
http_mock.readJSON = Mock(return_value = {"data": {"user_id": "di_resu", "username": "Emanresu"}})
# Let the service think there was a successful response # Let the service think there was a successful response
with patch.object(AuthorizationHelpers, "parseJWT", return_value=UserProfile()): with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
authorization_service._onAuthStateChanged(SUCCESSFUL_AUTH_RESPONSE) authorization_service._onAuthStateChanged(SUCCESSFUL_AUTH_RESPONSE)
# Ensure that the error signal was not triggered # Ensure that the error signal was not triggered
@ -205,7 +275,10 @@ def test_loginAndLogout() -> None:
# Since we said that it went right this time, validate that we got a signal. # Since we said that it went right this time, validate that we got a signal.
assert authorization_service.onAuthStateChanged.emit.call_count == 1 assert authorization_service.onAuthStateChanged.emit.call_count == 1
assert authorization_service.getUserProfile() is not None with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
def callback(profile):
assert profile is not None
authorization_service.getUserProfile(callback)
assert authorization_service.getAccessToken() == "beep" assert authorization_service.getAccessToken() == "beep"
# Check that we stored the authentication data, so next time the user won't have to log in again. # Check that we stored the authentication data, so next time the user won't have to log in again.
@ -214,19 +287,22 @@ def test_loginAndLogout() -> None:
# We're logged in now, also check if logging out works # We're logged in now, also check if logging out works
authorization_service.deleteAuthData() authorization_service.deleteAuthData()
assert authorization_service.onAuthStateChanged.emit.call_count == 2 assert authorization_service.onAuthStateChanged.emit.call_count == 2
assert authorization_service.getUserProfile() is None with patch("UM.TaskManagement.HttpRequestManager.HttpRequestManager.getInstance", MagicMock(return_value = http_mock)):
def callback(profile):
assert profile is None
authorization_service.getUserProfile(callback)
# Ensure the data is gone after we logged out. # Ensure the data is gone after we logged out.
assert preferences.getValue("test/auth_data") == "{}" assert preferences.getValue("test/auth_data") == "{}"
def test_wrongServerResponses() -> None: def test_wrongServerResponses() -> None:
authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences()) authorization_service = AuthorizationService(OAUTH_SETTINGS, Preferences())
authorization_service.initialize() authorization_service.initialize()
with patch.object(AuthorizationHelpers, "parseJWT", return_value=UserProfile()):
authorization_service._onAuthStateChanged(MALFORMED_AUTH_RESPONSE) authorization_service._onAuthStateChanged(MALFORMED_AUTH_RESPONSE)
assert authorization_service.getUserProfile() is None
def callback(profile):
assert profile is None
authorization_service.getUserProfile(callback)
def test__generate_auth_url() -> None: def test__generate_auth_url() -> None:
preferences = Preferences() preferences = Preferences()