8.7 KiB
| title | sidebar_label |
|---|---|
| Python library: authentication | Authentication |
The Authentication part of your library is responsible for acquiring authentication and for making authenticated requests. It should not be aware of what is in the requests.
Authentication comes in many forms, but it generally boils down to that each request is accompanied by an authorization header which contains an access token. The access token is generally a string of random numbers/letters.
Your library should be able to acquire the authentication tokens, update them if necessary and use the authentication to make requests. It should not offer features to store the authentication data.
Because authentication is going to be stored by the developer, it is important that you return the authentication to the developer in a format that can be JSON serializable. A dict with primitive types (str, float, int) is recommended.
If your API can be served from multiple locations, your authentication class should allow the developer to pass in the location of the API.
Async example
Python allows developers to write code that is either synchronous or asynchronous (via asyncio). Home Assistant is written in async, but is able to work with synchronous libraries too. We prefer async libraries.
If you are writing a library in async, we recommend that you use aiohttp. It's a modern and mature HTTP library and is easy to use.
from aiohttp import ClientSession, ClientResponse
class Auth:
"""Class to make authenticated requests."""
def __init__(self, websession: ClientSession, host: str, access_token: str):
"""Initialize the auth."""
self.websession = websession
self.host = host
self.access_token = access_token
async def request(self, method: str, path: str, **kwargs) -> ClientResponse:
"""Make a request."""
headers = kwargs.get("headers")
if headers is None:
headers = {}
else:
headers = dict(headers)
headers["authorization"] = self.access_token
return await self.websession.request(
method, f"{self.host}/{path}", **kwargs, headers=headers,
)
To use this class, you will need to create an aiohttp ClientSession and pass it together with the API info to the constructor.
import asyncio
import aiohttp
from my_package import Auth
async def main():
async with aiohttp.ClientSession() as session:
auth = Auth(session, "http://example.com/api", "secret_access_token")
# This will fetch data from http://example.com/api/lights
resp = await auth.request("get", "lights")
print("HTTP response status code", resp.status)
print("HTTP response JSON content", await resp.json())
asyncio.run(main())
Sync example
import requests
class Auth:
"""Class to make authenticated requests."""
def __init__(self, host: str, access_token: str):
"""Initialize the auth."""
self.host = host
self.access_token = access_token
def request(self, method: str, path: str, **kwargs) -> requests.Response:
"""Make a request."""
headers = kwargs.get("headers")
if headers is None:
headers = {}
else:
headers = dict(headers)
headers["authorization"] = self.access_token
return requests.request(
method, f"{self.host}/{path}", **kwargs, headers=headers,
)
To use this class, construct the class with the API info.
from my_package import Auth
auth = Auth("http://example.com/api", "secret_access_token")
# This will fetch data from http://example.com/api/lights
resp = auth.request("get", "lights")
print("HTTP response status code", resp.status_code)
print("HTTP response JSON content", resp.json())
OAuth2
OAuth2 is a standardized version of an authentication schema leveraging refresh and access tokens. The access token expires within a short period of time after being issued. The refresh token can be used to acquire new access tokens.
Refreshing access tokens relies on a client ID and secret, which might be held by an external service. We need to structure the authentication class to be able to allow the developer to implement their own token refresh logic.
Home Assistant ships with the Home Assistant Cloud Account Linking service, a free cloud service to allow users to quickly connect accounts using OAuth2. Home Assistant has easy to use tools built-in to allow users to configure OAuth2-based integrations. For more info, read here. These built-in tools work best if your library is implemented like the examples below.
Async example
from abc import ABC, abstractmethod
class AbstractAuth(ABC):
"""Abstract class to make authenticated requests."""
def __init__(self, websession: ClientSession, host: str):
"""Initialize the auth."""
self.websession = websession
self.host = host
@abstractmethod
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
async def request(self, method, url, **kwargs) -> ClientResponse:
"""Make a request."""
headers = kwargs.get("headers")
if headers is None:
headers = {}
else:
headers = dict(headers)
access_token = await self.async_get_access_token()
headers["authorization"] = f"Bearer {access_token}"
return await self.websession.request(
method, f"{self.host}/{url}", **kwargs, headers=headers,
)
Now the developer that is using your library will have to implement the abstract method for getting the access token. Let's assume that the developer has their own token manager class.
from my_package import AbstractAuth
class Auth(AbstractAuth):
def __init__(self, websession: ClientSession, host: str, token_manager):
"""Initialize the auth."""
super().__init__(websession, host)
self.token_manager = token_manager
async def async_get_access_token(self) -> str:
"""Return a valid access token."""
if self.token_manager.is_token_valid():
return self.token_manager.access_token
await self.token_manager.fetch_access_token()
await self.token_manager.save_access_token()
return self.token_manager.access_token
Sync example
If you are using requests, we recommend that you use the requests_oauthlib package. Below is an example that works with a local client ID and secret but also allows outsourcing token fetching to Home Assistant.
from typing import Optional, Union, Callable, Dict
from requests import Response
from requests_oauthlib import OAuth2Session
from oauthlib.oauth2 import TokenExpiredError
class Auth:
def __init__(
self,
host: str,
token: Optional[Dict[str, str]] = None,
client_id: str = None,
client_secret: str = None,
token_updater: Optional[Callable[[str], None]] = None,
):
self.host = host
self.client_id = client_id
self.client_secret = client_secret
self.token_updater = token_updater
extra = {"client_id": self.client_id, "client_secret": self.client_secret}
self._oauth = OAuth2Session(
auto_refresh_kwargs=extra,
client_id=client_id,
token=token,
token_updater=token_updater,
)
def refresh_tokens(self) -> Dict[str, Union[str, int]]:
"""Refresh and return new tokens."""
token = self._oauth.refresh_token(f"{self.host}/auth/token")
if self.token_updater is not None:
self.token_updater(token)
return token
def request(self, method: str, path: str, **kwargs) -> Response:
"""Make a request.
We don't use the built-in token refresh mechanism of OAuth2 session because
we want to allow overriding the token refresh logic.
"""
url = f"{self.host}/{path}"
try:
return getattr(self._oauth, method)(url, **kwargs)
except TokenExpiredError:
self._oauth.token = self.refresh_tokens()
return getattr(self._oauth, method)(url, **kwargs)
A developer will now be able to override the refresh token function to route it via their own external service.
from my_package import AbstractAuth
class Auth(AbstractAuth):
def refresh_tokens(self) -> Dict[str, Union[str, int]]:
"""Refresh and return new tokens."""
self.token_manager.fetch_access_token()
self.token_manager.save_access_token()
return self.token_manager.access_token