diff --git a/cura/OAuth2/AuthorizationRequestHandler.py b/cura/OAuth2/AuthorizationRequestHandler.py index d13639c45d..0558db784f 100644 --- a/cura/OAuth2/AuthorizationRequestHandler.py +++ b/cura/OAuth2/AuthorizationRequestHandler.py @@ -1,12 +1,15 @@ # Copyright (c) 2018 Ultimaker B.V. # Cura is released under the terms of the LGPLv3 or higher. -from typing import Optional, Callable, Tuple, Dict, Any, List +from typing import Optional, Callable, Tuple, Dict, Any, List, TYPE_CHECKING from http.server import BaseHTTPRequestHandler from urllib.parse import parse_qs, urlparse from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers -from cura.OAuth2.Models import AuthenticationResponse, ResponseData, HTTP_STATUS, ResponseStatus +from cura.OAuth2.Models import AuthenticationResponse, ResponseData, HTTP_STATUS + +if TYPE_CHECKING: + from cura.OAuth2.Models import ResponseStatus class AuthorizationRequestHandler(BaseHTTPRequestHandler): @@ -15,15 +18,15 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): It also requests the access token for the 2nd stage of the OAuth flow. """ - def __init__(self, request, client_address, server): + def __init__(self, request, client_address, server) -> None: super().__init__(request, client_address, server) # These values will be injected by the HTTPServer that this handler belongs to. - self.authorization_helpers = None # type: AuthorizationHelpers - self.authorization_callback = None # type: Callable[[AuthenticationResponse], None] - self.verification_code = None # type: str + self.authorization_helpers = None # type: Optional[AuthorizationHelpers] + self.authorization_callback = None # type: Optional[Callable[[AuthenticationResponse], None]] + self.verification_code = None # type: Optional[str] - def do_GET(self): + def do_GET(self) -> None: """Entry point for GET requests""" # Extract values from the query string. @@ -44,7 +47,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): # If there is data in the response, we send it. self._sendData(server_response.data_stream) - if token_response: + if token_response and self.authorization_callback is not None: # Trigger the callback if we got a response. # This will cause the server to shut down, so we do it at the very end of the request handling. self.authorization_callback(token_response) @@ -56,7 +59,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): :return: HTTP ResponseData containing a success page to show to the user. """ code = self._queryGet(query, "code") - if code: + if code and self.authorization_helpers is not None and self.verification_code is not None: # If the code was returned we get the access token. token_response = self.authorization_helpers.getAccessTokenUsingAuthorizationCode( code, self.verification_code) @@ -74,6 +77,8 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): success=False, error_message="Something unexpected happened when trying to log in, please try again." ) + if self.authorization_helpers is None: + return ResponseData(), token_response return ResponseData( status=HTTP_STATUS["REDIRECT"], @@ -83,7 +88,7 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): ), token_response @staticmethod - def _handleNotFound() -> "ResponseData": + def _handleNotFound() -> ResponseData: """Handle all other non-existing server calls.""" return ResponseData(status=HTTP_STATUS["NOT_FOUND"], content_type="text/html", data_stream=b"Not found.") @@ -100,6 +105,6 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler): self.wfile.write(data) @staticmethod - def _queryGet(query_data: Dict[Any, List], key: str, default=None) -> Optional[str]: + def _queryGet(query_data: Dict[Any, List], key: str, default: Optional[str]=None) -> Optional[str]: """Helper for getting values from a pre-parsed query string""" return query_data.get(key, [default])[0] diff --git a/cura/OAuth2/AuthorizationRequestServer.py b/cura/OAuth2/AuthorizationRequestServer.py index 270c558167..514a4ab5de 100644 --- a/cura/OAuth2/AuthorizationRequestServer.py +++ b/cura/OAuth2/AuthorizationRequestServer.py @@ -1,8 +1,11 @@ # Copyright (c) 2018 Ultimaker B.V. # Cura is released under the terms of the LGPLv3 or higher. from http.server import HTTPServer +from typing import Callable, Any, TYPE_CHECKING -from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers +if TYPE_CHECKING: + from cura.OAuth2.Models import AuthenticationResponse + from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers class AuthorizationRequestServer(HTTPServer): @@ -14,12 +17,12 @@ class AuthorizationRequestServer(HTTPServer): def setAuthorizationHelpers(self, authorization_helpers: "AuthorizationHelpers") -> None: """Set the authorization helpers instance on the request handler.""" - self.RequestHandlerClass.authorization_helpers = authorization_helpers + self.RequestHandlerClass.authorization_helpers = authorization_helpers # type: ignore - def setAuthorizationCallback(self, authorization_callback) -> None: + def setAuthorizationCallback(self, authorization_callback: Callable[["AuthenticationResponse"], Any]) -> None: """Set the authorization callback on the request handler.""" - self.RequestHandlerClass.authorization_callback = authorization_callback + self.RequestHandlerClass.authorization_callback = authorization_callback # type: ignore def setVerificationCode(self, verification_code: str) -> None: """Set the verification code on the request handler.""" - self.RequestHandlerClass.verification_code = verification_code + self.RequestHandlerClass.verification_code = verification_code # type: ignore diff --git a/cura/OAuth2/AuthorizationService.py b/cura/OAuth2/AuthorizationService.py index 4c66170c32..33ea419ff5 100644 --- a/cura/OAuth2/AuthorizationService.py +++ b/cura/OAuth2/AuthorizationService.py @@ -2,17 +2,18 @@ # Cura is released under the terms of the LGPLv3 or higher. import json import webbrowser -from typing import Optional +from typing import Optional, TYPE_CHECKING from urllib.parse import urlencode -# As this module is specific for Cura plugins, we can rely on these imports. from UM.Logger import Logger from UM.Signal import Signal -# Plugin imports need to be relative to work in final builds. from cura.OAuth2.LocalAuthorizationServer import LocalAuthorizationServer from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers -from cura.OAuth2.Models import OAuth2Settings, AuthenticationResponse, UserProfile +from cura.OAuth2.Models import AuthenticationResponse + +if TYPE_CHECKING: + from cura.OAuth2.Models import UserProfile, OAuth2Settings class AuthorizationService: @@ -32,7 +33,7 @@ class AuthorizationService: self._auth_helpers = AuthorizationHelpers(settings) self._auth_url = "{}/authorize".format(self._settings.OAUTH_SERVER_URL) self._auth_data = None # type: Optional[AuthenticationResponse] - self._user_profile = None # type: Optional[UserProfile] + self._user_profile = None # type: Optional["UserProfile"] self._cura_preferences = preferences self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True) self._loadAuthData() @@ -75,7 +76,6 @@ class AuthorizationService: def getAccessToken(self) -> Optional[str]: """ Get the access token response data. - :return: Dict containing token data. """ if not self.getUserProfile(): # We check if we can get the user profile. @@ -130,7 +130,7 @@ class AuthorizationService: # Start a local web server to receive the callback URL on. self._server.start(verification_code) - def _onAuthStateChanged(self, auth_response: "AuthenticationResponse") -> None: + def _onAuthStateChanged(self, auth_response: AuthenticationResponse) -> None: """Callback method for an authentication flow.""" if auth_response.success: self._storeAuthData(auth_response) @@ -150,7 +150,7 @@ class AuthorizationService: except ValueError as err: Logger.log("w", "Could not load auth data from preferences: %s", err) - def _storeAuthData(self, auth_data: Optional["AuthenticationResponse"] = None) -> None: + def _storeAuthData(self, auth_data: Optional[AuthenticationResponse] = None) -> None: """Store authentication data in preferences and locally.""" self._auth_data = auth_data if auth_data: diff --git a/cura/OAuth2/LocalAuthorizationServer.py b/cura/OAuth2/LocalAuthorizationServer.py index d6a4bf5216..d1d07b5c91 100644 --- a/cura/OAuth2/LocalAuthorizationServer.py +++ b/cura/OAuth2/LocalAuthorizationServer.py @@ -1,22 +1,21 @@ # Copyright (c) 2018 Ultimaker B.V. # Cura is released under the terms of the LGPLv3 or higher. import threading -from http.server import HTTPServer -from typing import Optional, Callable, Any +from typing import Optional, Callable, Any, TYPE_CHECKING -# As this module is specific for Cura plugins, we can rely on these imports. from UM.Logger import Logger -# Plugin imports need to be relative to work in final builds. from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers from cura.OAuth2.AuthorizationRequestServer import AuthorizationRequestServer from cura.OAuth2.AuthorizationRequestHandler import AuthorizationRequestHandler -from cura.OAuth2.Models import AuthenticationResponse + +if TYPE_CHECKING: + from cura.OAuth2.Models import AuthenticationResponse class LocalAuthorizationServer: def __init__(self, auth_helpers: "AuthorizationHelpers", - auth_state_changed_callback: "Callable[[AuthenticationResponse], Any]", + auth_state_changed_callback: Callable[["AuthenticationResponse"], Any], daemon: bool) -> None: """ :param auth_helpers: An instance of the authorization helpers class. @@ -62,7 +61,7 @@ class LocalAuthorizationServer: def stop(self) -> None: """ Stops the web server if it was running. Also deletes the objects. """ - Logger.log("d", "Stopping local web server...") + Logger.log("d", "Stopping local oauth2 web server...") if self._web_server: self._web_server.server_close() diff --git a/cura/OAuth2/Models.py b/cura/OAuth2/Models.py index a6b91cae26..796fdf8746 100644 --- a/cura/OAuth2/Models.py +++ b/cura/OAuth2/Models.py @@ -46,7 +46,7 @@ class ResponseStatus(BaseModel): # Response data template. class ResponseData(BaseModel): - status = None # type: Optional[ResponseStatus] + status = None # type: ResponseStatus data_stream = None # type: Optional[bytes] redirect_uri = None # type: Optional[str] content_type = "text/html" # type: str