3D-printing/Cura
1
0
Fork 0
mirror of https://github.com/Ultimaker/Cura.git synced 2025-07-07 15:07:28 -06:00
Code Issues Projects Releases Packages Wiki Activity Actions 8

Changed documentation style to doxygen

Browse source 
CURA-5744
This commit is contained in:
Jaime van Kessel 2018-09-27 11:03:17 +02:00
parent d5dbf91a4f
commit 1c8804ff2c
5 changed files with 62 additions and 97 deletions
Download patch file Download diff file Expand all files Collapse all files

54
cura/OAuth2/AuthorizationHelpers.py
View file

@ -13,25 +13,22 @@ from UM.Logger import Logger
from cura.OAuth2.Models import AuthenticationResponse, UserProfile, OAuth2Settings
# Class containing several helpers to deal with the authorization flow.
class AuthorizationHelpers:
"""Class containing several helpers to deal with the authorization flow."""
def __init__(self, settings: "OAuth2Settings") -> None:
self._settings = settings
self._token_url = "{}/token".format(self._settings.OAUTH_SERVER_URL)
@property
# The OAuth2 settings object.
def settings(self) -> "OAuth2Settings":
"""Get the OAuth2 settings object."""
return self._settings
# Request the access token from the authorization server.
# \param authorization_code: The authorization code from the 1st step.
# \param verification_code: The verification code needed for the PKCE extension.
# \return: An AuthenticationResponse object.
def getAccessTokenUsingAuthorizationCode(self, authorization_code: str, verification_code: str)-> "AuthenticationResponse":
"""
Request the access token from the authorization server.
:param authorization_code: The authorization code from the 1st step.
:param verification_code: The verification code needed for the PKCE extension.
:return: An AuthenticationResponse object.
"""
return self.parseTokenResponse(requests.post(self._token_url, data={
"client_id": self._settings.CLIENT_ID,
"redirect_uri": self._settings.CALLBACK_URL,
@ -41,12 +38,10 @@ class AuthorizationHelpers:
"scope": self._settings.CLIENT_SCOPES
}))
# Request the access token from the authorization server using a refresh token.
# \param refresh_token:
# \return: An AuthenticationResponse object.
def getAccessTokenUsingRefreshToken(self, refresh_token: str) -> AuthenticationResponse:
"""
Request the access token from the authorization server using a refresh token.
:param refresh_token:
:return: An AuthenticationResponse object.
"""
return self.parseTokenResponse(requests.post(self._token_url, data={
"client_id": self._settings.CLIENT_ID,
"redirect_uri": self._settings.CALLBACK_URL,
@ -56,12 +51,10 @@ class AuthorizationHelpers:
}))
@staticmethod
# Parse the token response from the authorization server into an AuthenticationResponse object.
# \param token_response: The JSON string data response from the authorization server.
# \return: An AuthenticationResponse object.
def parseTokenResponse(token_response: requests.models.Response) -> AuthenticationResponse:
"""
Parse the token response from the authorization server into an AuthenticationResponse object.
:param token_response: The JSON string data response from the authorization server.
:return: An AuthenticationResponse object.
"""
token_data = None
try:
@ -82,12 +75,10 @@ class AuthorizationHelpers:
expires_in=token_data["expires_in"],
scope=token_data["scope"])
# Calls the authentication API endpoint to get the token data.
# \param access_token: The encoded JWT token.
# \return: Dict containing some profile data.
def parseJWT(self, access_token: str) -> Optional["UserProfile"]:
"""
Calls the authentication API endpoint to get the token data.
:param access_token: The encoded JWT token.
:return: Dict containing some profile data.
"""
token_request = requests.get("{}/check-token".format(self._settings.OAUTH_SERVER_URL), headers = {
"Authorization": "Bearer {}".format(access_token)
})
@ -105,20 +96,15 @@ class AuthorizationHelpers:
)
@staticmethod
# Generate a 16-character verification code.
# \param code_length: How long should the code be?
def generateVerificationCode(code_length: int = 16) -> str:
"""
Generate a 16-character verification code.
:param code_length:
:return:
"""
return "".join(random.choice("0123456789ABCDEF") for i in range(code_length))
@staticmethod
# Generates a base64 encoded sha512 encrypted version of a given string.
# \param verification_code:
# \return: The encrypted code in base64 format.
def generateVerificationCodeChallenge(verification_code: str) -> str:
"""
Generates a base64 encoded sha512 encrypted version of a given string.
:param verification_code:
:return: The encrypted code in base64 format.
"""
encoded = sha512(verification_code.encode()).digest()
return b64encode(encoded, altchars = b"_-").decode()

23
cura/OAuth2/AuthorizationRequestHandler.py
View file

@ -12,12 +12,9 @@ if TYPE_CHECKING:
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
# This handler handles all HTTP requests on the local web server.
# It also requests the access token for the 2nd stage of the OAuth flow.
class AuthorizationRequestHandler(BaseHTTPRequestHandler):
"""
This handler handles all HTTP requests on the local web server.
It also requests the access token for the 2nd stage of the OAuth flow.
"""
def __init__(self, request, client_address, server) -> None:
super().__init__(request, client_address, server)
@ -27,8 +24,6 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
self.verification_code = None # type: Optional[str]
def do_GET(self) -> None:
"""Entry point for GET requests"""
# Extract values from the query string.
parsed_url = urlparse(self.path)
query = parse_qs(parsed_url.query)
@ -52,12 +47,10 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
# 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)
# Handler for the callback URL redirect.
# \param query: Dict containing the HTTP query parameters.
# \return: HTTP ResponseData containing a success page to show to the user.
def _handleCallback(self, query: Dict[Any, List]) -> Tuple[ResponseData, Optional[AuthenticationResponse]]:
"""
Handler for the callback URL redirect.
:param query: Dict containing the HTTP query parameters.
:return: HTTP ResponseData containing a success page to show to the user.
"""
code = self._queryGet(query, "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.
@ -88,12 +81,11 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
), token_response
@staticmethod
# Handle all other non-existing server calls.
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.")
def _sendHeaders(self, status: "ResponseStatus", content_type: str, redirect_uri: str = None) -> None:
"""Send out the headers"""
self.send_response(status.code, status.message)
self.send_header("Content-type", content_type)
if redirect_uri:
@ -101,10 +93,9 @@ class AuthorizationRequestHandler(BaseHTTPRequestHandler):
self.end_headers()
def _sendData(self, data: bytes) -> None:
"""Send out the data"""
self.wfile.write(data)
@staticmethod
# Convenience Helper for getting values from a pre-parsed query string
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]

16
cura/OAuth2/AuthorizationRequestServer.py
View file

@ -8,21 +8,19 @@ if TYPE_CHECKING:
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
# The authorization request callback handler server.
# This subclass is needed to be able to pass some data to the request handler.
# This cannot be done on the request handler directly as the HTTPServer creates an instance of the handler after
# init.
class AuthorizationRequestServer(HTTPServer):
"""
The authorization request callback handler server.
This subclass is needed to be able to pass some data to the request handler.
This cannot be done on the request handler directly as the HTTPServer creates an instance of the handler after init.
"""
# Set the authorization helpers instance on the request handler.
def setAuthorizationHelpers(self, authorization_helpers: "AuthorizationHelpers") -> None:
"""Set the authorization helpers instance on the request handler."""
self.RequestHandlerClass.authorization_helpers = authorization_helpers # type: ignore
# Set the authorization callback on the request handler.
def setAuthorizationCallback(self, authorization_callback: Callable[["AuthenticationResponse"], Any]) -> None:
"""Set the authorization callback on the request handler."""
self.RequestHandlerClass.authorization_callback = authorization_callback # type: ignore
# Set the verification code on the request handler.
def setVerificationCode(self, verification_code: str) -> None:
"""Set the verification code on the request handler."""
self.RequestHandlerClass.verification_code = verification_code # type: ignore

38
cura/OAuth2/AuthorizationService.py
View file

@ -38,11 +38,11 @@ class AuthorizationService:
self._server = LocalAuthorizationServer(self._auth_helpers, self._onAuthStateChanged, daemon=True)
self._loadAuthData()
# 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.
# \return UserProfile if a user is logged in, None otherwise.
# \sa _parseJWT
def getUserProfile(self) -> Optional["UserProfile"]:
"""
Get the user data that is stored in the JWT token.
:return: Dict containing some user data.
"""
if not self._user_profile:
# If no user profile was stored locally, we try to get it from JWT.
self._user_profile = self._parseJWT()
@ -52,11 +52,9 @@ class AuthorizationService:
return self._user_profile
# 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.
def _parseJWT(self) -> Optional["UserProfile"]:
"""
Tries to parse the JWT if all the needed data exists.
:return: UserProfile if found, otherwise 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.
return None
@ -74,10 +72,8 @@ class AuthorizationService:
return self._auth_helpers.parseJWT(self._auth_data.access_token)
# Get the access token as provided by the repsonse data.
def getAccessToken(self) -> Optional[str]:
"""
Get the access token response data.
"""
if not self.getUserProfile():
# We check if we can get the user profile.
# If we can't get it, that means the access token (JWT) was invalid or expired.
@ -88,24 +84,22 @@ class AuthorizationService:
return self._auth_data.access_token
# Try to refresh the access token. This should be used when it has expired.
def refreshAccessToken(self) -> None:
"""
Refresh the access token when it expired.
"""
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.")
return
self._storeAuthData(self._auth_helpers.getAccessTokenUsingRefreshToken(self._auth_data.refresh_token))
self.onAuthStateChanged.emit(logged_in=True)
# Delete the authentication data that we have stored locally (eg; logout)
def deleteAuthData(self) -> None:
"""Delete authentication data from preferences and locally."""
self._storeAuthData()
self.onAuthStateChanged.emit(logged_in=False)
if self._auth_data is not None:
self._storeAuthData()
self.onAuthStateChanged.emit(logged_in=False)
# Start the flow to become authenticated. This will start a new webbrowser tap, prompting the user to login.
def startAuthorizationFlow(self) -> None:
"""Start a new OAuth2 authorization flow."""
Logger.log("d", "Starting new OAuth2 flow...")
# Create the tokens needed for the code challenge (PKCE) extension for OAuth2.
@ -131,8 +125,8 @@ class AuthorizationService:
# Start a local web server to receive the callback URL on.
self._server.start(verification_code)
# Callback method for the authentication flow.
def _onAuthStateChanged(self, auth_response: AuthenticationResponse) -> None:
"""Callback method for an authentication flow."""
if auth_response.success:
self._storeAuthData(auth_response)
self.onAuthStateChanged.emit(logged_in=True)
@ -140,8 +134,8 @@ class AuthorizationService:
self.onAuthenticationError.emit(logged_in=False, error_message=auth_response.err_message)
self._server.stop() # Stop the web server at all times.
# Load authentication data from preferences.
def _loadAuthData(self) -> None:
"""Load authentication data from preferences if available."""
self._cura_preferences.addPreference(self._settings.AUTH_DATA_PREFERENCE_KEY, "{}")
try:
preferences_data = json.loads(self._cura_preferences.getValue(self._settings.AUTH_DATA_PREFERENCE_KEY))
@ -151,8 +145,8 @@ class AuthorizationService:
except ValueError:
Logger.logException("w", "Could not load auth data from preferences")
# Store authentication data in preferences.
def _storeAuthData(self, auth_data: Optional[AuthenticationResponse] = None) -> None:
"""Store authentication data in preferences and locally."""
self._auth_data = auth_data
if auth_data:
self._user_profile = self.getUserProfile()

28
cura/OAuth2/LocalAuthorizationServer.py
View file

@ -12,16 +12,17 @@ if TYPE_CHECKING:
from cura.OAuth2.Models import AuthenticationResponse
from cura.OAuth2.AuthorizationHelpers import AuthorizationHelpers
class LocalAuthorizationServer:
# The local LocalAuthorizationServer takes care of the oauth2 callbacks.
# Once the flow is completed, this server should be closed down again by calling stop()
# \param auth_helpers: An instance of the authorization helpers class.
# \param auth_state_changed_callback: A callback function to be called when the authorization state changes.
# \param daemon: Whether the server thread should be run in daemon mode. Note: Daemon threads are abruptly stopped
# at shutdown. Their resources (e.g. open files) may never be released.
def __init__(self, auth_helpers: "AuthorizationHelpers",
auth_state_changed_callback: Callable[["AuthenticationResponse"], Any],
daemon: bool) -> None:
"""
:param auth_helpers: An instance of the authorization helpers class.
:param auth_state_changed_callback: A callback function to be called when the authorization state changes.
:param daemon: Whether the server thread should be run in daemon mode. Note: Daemon threads are abruptly stopped
at shutdown. Their resources (e.g. open files) may never be released.
"""
self._web_server = None # type: Optional[AuthorizationRequestServer]
self._web_server_thread = None # type: Optional[threading.Thread]
self._web_server_port = auth_helpers.settings.CALLBACK_PORT
@ -29,11 +30,9 @@ class LocalAuthorizationServer:
self._auth_state_changed_callback = auth_state_changed_callback
self._daemon = daemon
# Starts the local web server to handle the authorization callback.
# \param verification_code: The verification code part of the OAuth2 client identification.
def start(self, verification_code: str) -> None:
"""
Starts the local web server to handle the authorization callback.
:param verification_code: The verification code part of the OAuth2 client identification.
"""
if self._web_server:
# If the server is already running (because of a previously aborted auth flow), we don't have to start it.
# We still inject the new verification code though.
@ -43,12 +42,10 @@ class LocalAuthorizationServer:
if self._web_server_port is None:
raise Exception("Unable to start server without specifying the port.")
Logger.log("d", "Starting local web server to handle authorization callback on port %s",
self._web_server_port)
Logger.log("d", "Starting local web server to handle authorization callback on port %s", self._web_server_port)
# Create the server and inject the callback and code.
self._web_server = AuthorizationRequestServer(("0.0.0.0", self._web_server_port),
AuthorizationRequestHandler)
self._web_server = AuthorizationRequestServer(("0.0.0.0", self._web_server_port), AuthorizationRequestHandler)
self._web_server.setAuthorizationHelpers(self._auth_helpers)
self._web_server.setAuthorizationCallback(self._auth_state_changed_callback)
self._web_server.setVerificationCode(verification_code)
@ -57,9 +54,8 @@ class LocalAuthorizationServer:
self._web_server_thread = threading.Thread(None, self._web_server.serve_forever, daemon = self._daemon)
self._web_server_thread.start()
# Stops the web server if it was running. It also does some cleanup.
def stop(self) -> None:
""" Stops the web server if it was running. Also deletes the objects. """
Logger.log("d", "Stopping local oauth2 web server...")
if self._web_server: