Authorization code

Since authorization code process needs the user to accept the access to its data by the application, the library starts locally a http server. You may put the host part of the redirect_uri parameter in your hosts file pointing to your loop-back address. The server waits a GET requests with the code as a query parameter.

Getting a couple of access token may be done like this:

scopes = ['scope_1', 'scope_2']

service_information = ServiceInformation('https://authorization-server/oauth/authorize',
                                         'https://token-server/oauth/token',
                                         'client_id',
                                         'client_secret',
                                          scopes)
manager = CredentialManager(service_information,
                            proxies=dict(http='http://localhost:3128', https='http://localhost:3128'))
redirect_uri = 'http://somewhere.io:8080/oauth/code'

# Builds the authorization url and starts the local server according to the redirect_uri parameter
url = manager.init_authorize_code_process(redirect_uri, 'state_test')
_logger.info('Open this url in your browser\n%s', url)

code = manager.wait_and_terminate_authorize_code_process()
# From this point the http server is opened on 8080 port and wait to receive a single GET request
# All you need to do is open the url and the process will go on
# (as long you put the host part of your redirect uri in your host file)
# when the server gets the request with the code (or error) in its query parameters
_logger.debug('Code got = %s', code)
manager.init_with_authorize_code(redirect_uri, code)
_logger.debug('Access got = %s', manager._access_token)
# Here access and refresh token may be used with self.refresh_token

Authorization code with Proof Key for Code Exchange (PKCE)

In case you can generate a couple of code verifier and code challenge as follows:

import base64
import hashlib
import logging
import secrets
from typing import Tuple

def generate_sha256_pkce(length: int) -> Tuple[str, str]:
    if not (43 <= length <= 128):
        raise Exception("Invalid length: " % str(length))
    verifier = secrets.token_urlsafe(length)
    encoded = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode('ascii')).digest())
    challenge = encoded.decode('ascii')[:-1]
    return verifier, challenge

Then you can init authorization code workflow as follows

code_verifier, code_challenge = generate_sha256_pkce(64)
url = manager.init_authorize_code_process(redirect_uri, 'state_test',
                                          code_challenge=code_challenge,
                                          code_challenge_method="S256")

or either generate the url

url = manager.generate_authorize_url(redirect_uri, 'state_test',
                                     code_challenge=code_challenge,
                                     code_challenge_method="S256")

And once you obtains the code exchange it as follows

manager.init_with_authorize_code(redirect_uri, code, code_verifier=code_verifier)

User credentials

Getting a couple of access and refresh token is much easier:

scopes = ['scope_1', 'scope_2']

service_information = ServiceInformation('https://authorization-server/oauth/authorize',
                                         'https://token-server/oauth/token',
                                         'client_id',
                                         'client_secret',
                                          scopes)
manager = CredentialManager(service_information,
                            proxies=dict(http='http://localhost:3128', https='http://localhost:3128'))
manager.init_with_user_credentials('login', 'password')
_logger.debug('Access got = %s', manager._access_token)
# Here access and refresh token may be used

Client credentials

You can also get a token with client credentials process

manager = CredentialManager(service_information,
                            proxies=dict(http='http://localhost:3128', https='http://localhost:3128'))
manager.init_with_client_credentials()
# here application admin operation may be called

Refresh token

Provided that you kept a previous refresh_token, you can initiate your credential manager with it:

manager = CredentialManager(service_information,
                            proxies=dict(http='http://localhost:3128', https='http://localhost:3128'))
manager.init_with_token('my saved refreshed token')

Token expiration

CredentialManager class handle token expiration by calling the CredentialManager._is_token_expired static method. This implementation is not accurate for all OAuth server implementation. You’d better extend CredentialManager class and override _is_token_expired method.

Read other fields from token response

CredentialManager can be subclassed to handle other token response fields such as id_token in OpenId protocol.

class OpenIdCredentialManager(CredentialManager):
    def __init__(self, service_information, proxies=None):
        super(OpenIdCredentialManager, self).__init__(service_information, proxies)
        self.id_token = None

    def _process_token_response(self,  token_response, refresh_token_mandatory):
        id_token = token_response.get('id_token')
        OpenIdCredentialManager._check_id(id_token)
        super(OpenIdCredentialManager, self)._process_token_response(token_response, refresh_token_mandatory)
        self.id_token = id_token

    @staticmethod
    def _check_id(id_token):
        # check that open id token is valid
        pass