Authorising requests

The auth module provides a framework and various implementations for authorising access to a HTTP service, such as a REST API.

Basic usage

All Authoriser implementations can be used in the same way.

import asyncio

from aiorequestful.auth import Authoriser

authoriser = Authoriser(service_name="http service")

async def authorise(auth: Authoriser) -> None:
    headers = await auth.authorise()

    # all the following commands are just aliases for the `authorise` method
    headers = await auth
    headers = await auth()

asyncio.run(authorise(authoriser))

All implementations of Authoriser will accept a service_name which is a just a simple name to refer to the service. This will only be used in logging and exception messages.

The calls in the auth function will handle the authorisation process from start to finish, returning the headers necessary to send authorised requests.

Basic Authorisation

Authoriser with a service using a username and password (if necessary). You may also provide an optional encoding identifier to define how the credentials should be encoded before applying Base64 encoding.

from aiorequestful.auth.basic import BasicAuthoriser

authoriser = BasicAuthoriser(login="username", password="password", encoding="latin1")

`OAuth2Authoriser`

Authorise with a service that implements an OAuth2 authorisation flow.

Note

The OAuth2 framework specification provides many possible flows for authorising depending on the use case. Check which flow the HTTP service you are trying to access allows you to use and use the appropriate one for your use case.

`ClientCredentialsFlow`

Implements the Client Credentials flow as per the OAuth2 specification.

from aiorequestful.auth.oauth2 import ClientCredentialsFlow

authoriser = ClientCredentialsFlow(
    token_request=token_request, response_handler=response_handler, response_tester=response_tester
)

To simplify the creation process, you may use the ClientCredentialsFlow.create() method to automatically generate some of the utility objects from the given parameters.

authoriser = ClientCredentialsFlow.create(
    service_name="Cool Service",
    token_request_url="https://cool_service.com/api/token",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester

If the service you are accessing requires you send encoded credentials as part of the authorisation process, you may also use the ClientCredentialsFlow.create_with_encoded_credentials() method. This generates utility objects which send the credentials as headers in Base64 encoding i.e.

{'Authorization': 'Basic <base64 encoded client_id:client_secret>'}

authoriser = ClientCredentialsFlow.create_with_encoded_credentials(
    service_name="Cool Service",
    token_request_url="https://cool_service.com/api/token",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester

`AuthorisationCodeFlow`

Implements the Authorization Code flow as per the OAuth2 specification.

from aiorequestful.auth.oauth2 import AuthorisationCodeFlow
from aiorequestful.auth.utils import SocketHandler

user_request = AuthRequest(method="POST", url="https://cool_service.com/authorise")
refresh_request = AuthRequest(method="POST", url="https://cool_service.com/api/token/refresh")
socket_handler = SocketHandler(port=32000, timeout=60)  # the local socket to receive the authorisation code redirect
redirect_uri = "https://myapp.com/authorise"  # the public address of your app

authoriser = AuthorisationCodeFlow(
    user_request=user_request,
    redirect_uri=redirect_uri,
    token_request=token_request,
    refresh_request=refresh_request,
    response_handler=response_handler,
    response_tester=response_tester,
    socket_handler=socket_handler,
)

To simplify the creation process, you may use the AuthorisationCodeFlow.create() method to automatically generate some of the utility objects from the given parameters.

authoriser = AuthorisationCodeFlow.create(
    service_name="Cool Service",
    user_request_url="https://cool_service.com/authorise",
    token_request_url="https://cool_service.com/api/token",
    refresh_request_url="https://cool_service.com/api/token/refresh",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
    scope=["public-user-data", "private-user-data"]
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester
authoriser.redirect_uri = redirect_uri
authoriser.socket_handler = socket_handler

If the service you are accessing requires you send encoded credentials as part of the authorisation process, you may also use the AuthorisationCodeFlow.create_with_encoded_credentials() method. This generates utility objects which send the credentials as headers in Base64 encoding i.e.

{'Authorization': 'Basic <base64 encoded client_id:client_secret>'}

authoriser = AuthorisationCodeFlow.create_with_encoded_credentials(
    service_name="Cool Service",
    user_request_url="https://cool_service.com/authorise",
    token_request_url="https://cool_service.com/api/token",
    refresh_request_url="https://cool_service.com/api/token/refresh",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
    scope=["public-user-data", "private-user-data"]
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester
authoriser.redirect_uri = redirect_uri
authoriser.socket_handler = socket_handler

`AuthorisationCodePKCEFlow`

Implements the Authorization Code with PKCE flow as per the OAuth2 specification.

from aiorequestful.auth.oauth2 import AuthorisationCodePKCEFlow

user_request = AuthRequest(method="POST", url="https://cool_service.com/authorise")
refresh_request = AuthRequest(method="POST", url="https://cool_service.com/api/token/refresh")
socket_handler = SocketHandler(port=32000, timeout=60)  # the local socket to receive the authorisation code redirect
redirect_uri = "https://myapp.com/authorise"  # the public address of your app

authoriser = AuthorisationCodePKCEFlow(
    user_request=user_request,
    redirect_uri=redirect_uri,
    token_request=token_request,
    refresh_request=refresh_request,
    response_handler=response_handler,
    response_tester=response_tester,
    socket_handler=socket_handler,
    pkce_code_length=128,
)

To simplify the creation process, you may use the AuthorisationCodePKCEFlow.create() method to automatically generate some of the utility objects from the given parameters.

authoriser = AuthorisationCodePKCEFlow.create(
    service_name="Cool Service",
    user_request_url="https://cool_service.com/authorise",
    redirect_uri="https://myapp.com/authorise",
    token_request_url="https://cool_service.com/api/token",
    refresh_request_url="https://cool_service.com/api/token/refresh",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
    scope=["public-user-data", "private-user-data"],
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester
authoriser.redirect_uri = redirect_uri
authoriser.socket_handler = socket_handler

If the service you are accessing requires you send encoded credentials as part of the authorisation process, you may also use the AuthorisationCodePKCEFlow.create_with_encoded_credentials() method. This generates utility objects which send the credentials as headers in Base64 encoding i.e.

{'Authorization': 'Basic <base64 encoded client_id:client_secret>'}

authoriser = AuthorisationCodePKCEFlow.create_with_encoded_credentials(
    service_name="Cool Service",
    user_request_url="https://cool_service.com/authorise",
    token_request_url="https://cool_service.com/api/token",
    refresh_request_url="https://cool_service.com/api/token/refresh",
    client_id="<YOUR CLIENT ID>",
    client_secret="<YOUR CLIENT SECRET>",
    scope=["public-user-data", "private-user-data"]
)

# assign additional attributes as necessary
authoriser.response = response_handler
authoriser.tester = response_tester
authoriser.redirect_uri = redirect_uri
authoriser.socket_handler = socket_handler

Utilities

The auth.utils module provides various utilities to aid in automatic authorisation depending on the flow.

`AuthRequest`

Simply a wrapper for an aiohttp request. Stores the kwargs necessary to make a request. This allows the Authoriser to make the request as required.

We may provide the request kwargs on initialisation and execute with AuthRequest.request().

from aiohttp import ClientSession

from aiorequestful.auth.utils import AuthRequest

auth_request = AuthRequest(method="GET", url="http://cool_service.com/authorise")


async def request(request: AuthRequest) -> None:
    async with ClientSession() as session:
        await request(session)
        await request.request(session)  # does the same as above

We may also wish to add temporary parameters to the request should we wish to add sensitive or other temporary information to the request.

async def request_with_temporary_headers(request: AuthRequest, headers: dict[str, str]) -> None:
    with request.enrich_headers(headers):
        async with ClientSession() as session:
            await request(session)

`AuthResponse`

Stores a converted JSON response returned from a service, providing a facade for interacting with the values in the response directly and indirectly.

Crucially, it provides the AuthResponse.token and AuthResponse.headers properties which manage extraction of this information from the response to make an authorised request with.

import tempfile

from aiorequestful.auth.utils import AuthResponse

auth_response = AuthResponse(file_path=f"{tempfile.gettempdir()}/path/to/token/response")

auth_response.load_response_from_file()  # get the response from the file path
auth_response.replace({"access_token": "you_are_authorised", "token_type": "Bearer"})  # force add a new response
auth_response.save_response_to_file()  # save the response to file

auth_response.token  # extract the token
auth_response.headers  # generate the headers
auth_response["token_type"]  # access keys on the response directly

You may also configure additional headers that are required for successful, authorised requests. Equally, if your service does not return the token_type key, you may specify a fallback default too.

auth_response.pop("token_type")

# add additional headers to add to the generated headers to ensure successful requests
auth_response.additional_headers = {"Content-Type": "application/json"}
# set an optional fallback default for the token type if it cannot be found in the response
auth_response.token_prefix_default = "Basic"

assert auth_response.headers == {"Authorization": "Basic you_are_authorised", "Content-Type": "application/json"}

You may also enrich a response to add a UNIX timestamp value for the granted_at and expires_at times for the token.

Warning

To add an expires_at time to the response, the response must return with a value for the expires_in key.

auth_response["expires_in"] = 3600
auth_response.enrich()

auth_response["granted_at"]  # the time at which the token was granted at
auth_response["expires_at"]  # the time at which the token expires

`AuthTester`

Sets up a series of tests to check the validity of a response.

from aiohttp import ClientResponse

from aiorequestful.auth.utils import AuthTester


async def response_test(response: ClientResponse) -> bool:
    return "error" not in response.text()

auth_tester = AuthTester(request=auth_request, response_test=response_test, max_expiry=120)


async def test(tester: AuthTester) -> bool:
    result = await tester.test(response=auth_response)
    result = await tester(response=auth_response)  # does the same as above
    return result

The request is a request that is sent using the headers generated by the AuthResponse.
The response_test tests the response retrieved from this request
The max_expiry is the maximum allowed remaining time on the token in seconds. This is calculated during the test as the difference between the current time and the expires_at time. The test fails when this value is below the max_expiry time.

`SocketHandler`

Handles managing a socket to open to listen for callback data. Useful for authorisation flows which return an authorisation code via a redirect URL.

from aiorequestful.auth.utils import SocketHandler

socket_handler = SocketHandler(port=32000, timeout=60)

with socket_handler as socket_listener:
    data = socket_listener.recv(0)

Writing an `Authoriser`

To implement an Authoriser, you will need to implement the abstract methods as shown below.

class Authoriser(ABC):
    """
    Base interface for authenticating and authorising access to a service over HTTP.

    :param service_name: The service name for which to authorise.
    """

    __slots__ = ("logger", "service_name")

    def __init__(self, service_name: str = _DEFAULT_SERVICE_NAME):
        #: The :py:class:`logging.Logger` for this  object
        self.logger: logging.Logger = logging.getLogger(__name__)

        #: The service name for which to authorise. Currently only used for logging purposes.
        self.service_name = service_name

    @abstractmethod
    async def authorise(self) -> Headers:
        """
        Authenticate and authorise, testing/refreshing/re-authorising as needed.

        :raise AuthoriserError: If the authorisation failed to generate valid a token if needed,
            or if the tests continue to fail despite authorising/re-authorising.
        """
        raise NotImplementedError

    def __call__(self) -> Awaitable[Headers]:
        return self.authorise()

    def __await__(self) -> Generator[Any, None, Headers]:
        return self.authorise().__await__()

As an example, the following implements the BasicAuthoriser.

class BasicAuthoriser(Authoriser):
    """
    Authorise HTTP requests using basic authentication i.e. username + password (optional)

    :param service_name: The service name for which to authorise.
    :param login: The login ID of the credentials.
    :param password: The login password.
    :param encoding: The encoding to apply to credentials when sending requests.
    """

    __slots__ = ("login", "password", "encoding")

    def __init__(
            self,
            login: str,
            password: str = "",
            encoding: str = "latin1",
            service_name: str = _DEFAULT_SERVICE_NAME
    ):
        super().__init__(service_name=service_name)

        #: The login ID of the credentials.
        self.login = login
        #: The login password.
        self.password = password
        #: The encoding to apply to credentials when sending requests.
        self.encoding = encoding

    async def authorise(self) -> Headers:
        credentials = f"{self.login}:{self.password}".encode(self.encoding)
        credentials_encoded = base64.b64encode(credentials).decode(self.encoding)
        return {
            "Authorization": f"Basic {credentials_encoded}"
        }