feat: enable verification of certs via network.tls_verify and private CA chains with network.tls_ca_path (#16)

Signed-off-by: Christopher Klein <ckl@dreitier.com>

README.md

@@ -71,6 +71,8 @@ With the default configuration, [a person entry](https://www.home-assistant.io/i
 | `claims.groups`            | `string` | No       | `groups`                     | The claim to use to obtain the user's group(s). |
 | `roles.admin`            | `string` | No       | `admins`                     | Group name to require for users to get the 'admin' role in Home Assistant. Defaults to 'admins', the default group name for admins in Authentik. Doesn't do anything if no groups claim is found in your token. |
 | `roles.user`            | `string` | No       |                     | Group name to require for users to get the 'user' role in Home Assistant. Defaults to giving all users this role, unless configured. |
+| `network.tls_verify`         | `boolean` | No       | `true`                     | Verify TLS certificate. You may want to set this set to `false` when testing locally. |
+| `network.tls_ca_path`            | `string` | No       |                       | Path to file containing a private certificate authority chain. |
 
 #### Example: Migrating from HA username/password users to OIDC users
 If you already have users created within Home Assistant and would like to re-use the current user profile for your OIDC login, you can (temporarily) enable `features.automatic_user_linking`, with the following config (example):
@@ -88,6 +90,26 @@ Upon login, OIDC users will then automatically be linked to the HA user with the
 > [!IMPORTANT]
 > It's recommended to only enable this temporarily as it may pose a security risk. Any OIDC user with a username corresponding to a user in Home Assistant can get access to that user, and it's existing rights (admin), even if MFA is currently enabled for that account. After you have migrated your users (and linked OIDC to all existing accounts) you can disable the feature and keep using the linked users.
 
+#### Example: Using a private certificate authority
+If you use a private certificate authority to secure your OIDC provider (e.g. Keycloak), your CA must be able to be used by this component. Otherwise you will receive a certificate error (`[SSL: CERTIFICATE_VERIFY_FAILED]`) when connecting to the OIDC provider.
+You can either make the CA known to the entire operating system or configure only this component to use the CA. If you only want to let this component know your CA, you can specify it via `network.tls_ca_path`:
+
+```yaml
+auth_oidc:
+    network:
+        tls_ca_path: /path/to/private-ca.pem
+```
+
+If you want to deactivate the validation of the certificates for test purposes, you can do this via `network.tls_verify: false`:
+
+```yaml
+auth_oidc:
+    network:
+        tls_verify: false
+```
+
+In productive use, however, you should set `network.tls_verify` to `true`.
+
 ## Development
 This project uses the Rye package manager for development. You can find installation instructions here: https://rye.astral.sh/guide/installation/.
 Start by installing the dependencies using `rye sync` and then point your editor towards the environment created in the `.venv` directory.

custom_components/auth_oidc/__init__.py

@@ -19,6 +19,7 @@ from .config import (
     FEATURES,
     CLAIMS,
     ROLES,
+    NETWORK,
 )
 
 # pylint: enable=useless-import-alias
@@ -55,6 +56,7 @@ async def async_setup(hass: HomeAssistant, config):
     scope = "openid profile groups"
 
     oidc_client = oidc_client = OIDCClient(
+        hass=hass,
         discovery_url=my_config.get(DISCOVERY_URL),
         client_id=my_config.get(CLIENT_ID),
         scope=scope,
@@ -63,6 +65,7 @@ async def async_setup(hass: HomeAssistant, config):
         features=my_config.get(FEATURES, {}),
         claims=my_config.get(CLAIMS, {}),
         roles=my_config.get(ROLES, {}),
+        network=my_config.get(NETWORK, {}),
     )
 
     # Register the views

custom_components/auth_oidc/config.py

@@ -19,6 +19,10 @@ ROLES = "roles"
 ROLE_ADMINS = "admin"
 ROLE_USERS = "user"
 
+NETWORK = "network"
+NETWORK_TLS_VERIFY = "tls_verify"
+NETWORK_TLS_CA_PATH = "tls_ca_path"
+
 DEFAULT_TITLE = "OpenID Connect (SSO)"
 
 DOMAIN = "auth_oidc"
@@ -78,6 +82,17 @@ CONFIG_SCHEMA = vol.Schema(
                         vol.Optional(ROLE_ADMINS): vol.Coerce(str),
                     }
                 ),
+                # Network options
+                vol.Optional(NETWORK): vol.Schema(
+                    {
+                        # Verify x509 certificates provided when starting TLS connections
+                        vol.Optional(NETWORK_TLS_VERIFY, default=True): vol.Coerce(
+                            bool
+                        ),
+                        # Load custom certificate chain for private CAs
+                        vol.Optional(NETWORK_TLS_CA_PATH): vol.Coerce(str),
+                    }
+                ),
             }
         )
     },

custom_components/auth_oidc/oidc_client.py

@@ -5,9 +5,12 @@ import logging
 import os
 import base64
 import hashlib
+import ssl
 from typing import Optional
+from functools import partial
 import aiohttp
 from jose import jwt, jwk
+from homeassistant.core import HomeAssistant
 
 from .types import UserDetails
 from .config import (
@@ -17,6 +20,8 @@ from .config import (
     CLAIMS_GROUPS,
     ROLE_ADMINS,
     ROLE_USERS,
+    NETWORK_TLS_VERIFY,
+    NETWORK_TLS_CA_PATH,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -53,7 +58,15 @@ class OIDCClient:
     # Flows stores the state, code_verifier and nonce of all current flows.
     flows = {}
 
-    def __init__(self, discovery_url: str, client_id: str, scope: str, **kwargs: str):
+    def __init__(
+        self,
+        hass: HomeAssistant,
+        discovery_url: str,
+        client_id: str,
+        scope: str,
+        **kwargs: str,
+    ):
+        self.hass = hass
         self.discovery_url = discovery_url
         self.discovery_document = None
         self.client_id = client_id
@@ -70,13 +83,22 @@ class OIDCClient:
         features = kwargs.get("features")
         claims = kwargs.get("claims")
         roles = kwargs.get("roles")
+        network = kwargs.get("network")
 
-        self.disable_pkce: bool = features.get(FEATURES_DISABLE_PKCE)
+        self.disable_pkce = features.get(FEATURES_DISABLE_PKCE, False)
         self.display_name_claim = claims.get(CLAIMS_DISPLAY_NAME, "name")
         self.username_claim = claims.get(CLAIMS_USERNAME, "preferred_username")
         self.groups_claim = claims.get(CLAIMS_GROUPS, "groups")
         self.user_role = roles.get(ROLE_USERS, None)
         self.admin_role = roles.get(ROLE_ADMINS, "admins")
+        self.tls_verify = network.get(NETWORK_TLS_VERIFY, True)
+        self.tls_ca_path = network.get(NETWORK_TLS_CA_PATH)
+
+        _LOGGER.debug(
+            "OIDC provider network options (verify certificates: %r, custom CA file: %s)",
+            self.tls_verify,
+            self.tls_ca_path,
+        )
 
     def _base64url_encode(self, value: str) -> str:
         """Uses base64url encoding on a given string"""
@@ -86,10 +108,26 @@ class OIDCClient:
         """Generates a random URL safe string (base64_url encoded)"""
         return self._base64url_encode(os.urandom(length))
 
+    async def _create_session(self):
+        """Create a new client session with custom networking/TLS options"""
+        tcp_connector_args = {"verify_ssl": self.tls_verify}
+
+        if self.tls_ca_path:
+            # Move to hass' executor to prevent blocking code inside non-blocking method
+            ssl_context = await self.hass.loop.run_in_executor(
+                None, partial(ssl.create_default_context, cafile=self.tls_ca_path)
+            )
+            tcp_connector_args["ssl"] = ssl_context
+
+        return aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(**tcp_connector_args)
+        )
+
     async def _fetch_discovery_document(self):
         """Fetches discovery document from the given URL."""
         try:
-            async with aiohttp.ClientSession() as session:
+            session = await self._create_session()
+
             async with session.get(self.discovery_url) as response:
                 response.raise_for_status()
                 return await response.json()
@@ -105,7 +143,8 @@ class OIDCClient:
     async def _get_jwks(self, jwks_uri):
         """Fetches JWKS from the given URL."""
         try:
-            async with aiohttp.ClientSession() as session:
+            session = await self._create_session()
+
             async with session.get(jwks_uri) as response:
                 response.raise_for_status()
                 return await response.json()
@@ -116,7 +155,8 @@ class OIDCClient:
     async def _make_token_request(self, token_endpoint, query_params):
         """Performs the token POST call"""
         try:
-            async with aiohttp.ClientSession() as session:
+            session = await self._create_session()
+
             async with session.post(token_endpoint, data=query_params) as response:
                 response.raise_for_status()
                 return await response.json()