Fixes Home Assistant error about re-creating HTTP sessions (#22)

* Bump to 0.5.1

* Prevent HA errors about HTTP session left open

README.md

@@ -69,6 +69,10 @@ With the default configuration, [a person entry](https://www.home-assistant.io/i
 | `claims.display_name`      | `string` | No       | `name`                     | The claim to use to obtain the display name.
 | `claims.username`         | `string` | No       | `preferred_username`                     | The claim to use to obtain the username.
 | `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):
@@ -86,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

@@ -18,6 +18,8 @@ from .config import (
     ID_TOKEN_SIGNING_ALGORITHM,
     FEATURES,
     CLAIMS,
+    ROLES,
+    NETWORK,
 )
 
 # pylint: enable=useless-import-alias
@@ -50,10 +52,11 @@ async def async_setup(hass: HomeAssistant, config):
 
     _LOGGER.info("Registered OIDC provider")
 
-    # We only use openid & profile, never email
-    scope = "openid profile"
+    # We only use openid, profile & groups, never email
+    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,
@@ -61,6 +64,8 @@ async def async_setup(hass: HomeAssistant, config):
         id_token_signing_alg=my_config.get(ID_TOKEN_SIGNING_ALGORITHM),
         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

@@ -15,6 +15,13 @@ CLAIMS = "claims"
 CLAIMS_DISPLAY_NAME = "display_name"
 CLAIMS_USERNAME = "username"
 CLAIMS_GROUPS = "groups"
+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)"
 
@@ -63,6 +70,29 @@ CONFIG_SCHEMA = vol.Schema(
                         vol.Optional(CLAIMS_GROUPS): vol.Coerce(str),
                     }
                 ),
+                # Determine which specific group values will be mapped to which roles
+                # Optional, defaults user = null, admin = 'admins'
+                # If user role is set, users that do not have either will be rejected!
+                vol.Optional(ROLES): vol.Schema(
+                    {
+                        # Which group name should we use to assign the user role?
+                        vol.Optional(ROLE_USERS): vol.Coerce(str),
+                        # What group name should we use to assign the admin role?
+                        # Defaults to admins
+                        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/endpoints/callback.py

@@ -52,5 +52,15 @@ class OIDCCallbackView(HomeAssistantView):
             )
             return web.Response(text=view_html, content_type="text/html")
 
+        if user_details.get("role") == "invalid":
+            view_html = await get_view(
+                "error",
+                {
+                    "error": "User is not in the correct group to access Home Assistant, "
+                    + "contact your administrator!",
+                },
+            )
+            return web.Response(text=view_html, content_type="text/html")
+
         code = await self.oidc_provider.async_save_user_info(user_details)
         return web.HTTPFound(get_url("/auth/oidc/finish?code=" + code))

custom_components/auth_oidc/endpoints/finish.py

@@ -46,9 +46,9 @@ class OIDCFinishView(HomeAssistantView):
                 # Set a cookie to enable autologin on only the specific path used
                 # for the POST request, with all strict parameters set
                 # This cookie should not be read by any Javascript or any other paths.
-                # It can be really short lifetime as we redirect immediately (15 seconds)
+                # It can be really short lifetime as we redirect immediately (5 seconds)
                 "set-cookie": "auth_oidc_code="
                 + code
-                + "; Path=/auth/login_flow; SameSite=Strict; HttpOnly; Max-Age=15",
+                + "; Path=/auth/login_flow; SameSite=Strict; HttpOnly; Max-Age=5",
             },
         )

custom_components/auth_oidc/manifest.json

@@ -19,5 +19,5 @@
     "jinja2>=3.1.4",
     "bcrypt>=4.2.0"
   ],
-  "version": "0.4.1"
+  "version": "0.5.1"
 }

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 (
@@ -15,6 +18,10 @@ from .config import (
     CLAIMS_DISPLAY_NAME,
     CLAIMS_USERNAME,
     CLAIMS_GROUPS,
+    ROLE_ADMINS,
+    ROLE_USERS,
+    NETWORK_TLS_VERIFY,
+    NETWORK_TLS_CA_PATH,
 )
 
 _LOGGER = logging.getLogger(__name__)
@@ -51,7 +58,18 @@ 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):
+    # HTTP session to be used
+    http_session: aiohttp.ClientSession = None
+
+    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
@@ -67,11 +85,25 @@ 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)
+
+    def __del__(self):
+        """Cleanup the HTTP session."""
+
+        # HA never seems to run this, but it's good practice to close the session
+        if self.http_session:
+            _LOGGER.debug("Closing HTTP session")
+            self.http_session.close()
 
     def _base64url_encode(self, value: str) -> str:
         """Uses base64url encoding on a given string"""
@@ -81,13 +113,40 @@ class OIDCClient:
         """Generates a random URL safe string (base64_url encoded)"""
         return self._base64url_encode(os.urandom(length))
 
+    async def _get_http_session(self) -> aiohttp.ClientSession:
+        """Create or get the existing client session with custom networking/TLS options"""
+        if self.http_session is not None:
+            return self.http_session
+
+        _LOGGER.debug(
+            "Creating HTTP session provider with options: "
+            + "verify certificates: %r, custom CA file: %s",
+            self.tls_verify,
+            self.tls_ca_path,
+        )
+
+        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
+
+        self.http_session = aiohttp.ClientSession(
+            connector=aiohttp.TCPConnector(**tcp_connector_args)
+        )
+        return self.http_session
+
     async def _fetch_discovery_document(self):
         """Fetches discovery document from the given URL."""
         try:
-            async with aiohttp.ClientSession() as session:
-                async with session.get(self.discovery_url) as response:
-                    response.raise_for_status()
-                    return await response.json()
+            session = await self._get_http_session()
+
+            async with session.get(self.discovery_url) as response:
+                response.raise_for_status()
+                return await response.json()
         except aiohttp.ClientResponseError as e:
             if e.status == 404:
                 _LOGGER.warning(
@@ -100,10 +159,11 @@ class OIDCClient:
     async def _get_jwks(self, jwks_uri):
         """Fetches JWKS from the given URL."""
         try:
-            async with aiohttp.ClientSession() as session:
-                async with session.get(jwks_uri) as response:
-                    response.raise_for_status()
-                    return await response.json()
+            session = await self._get_http_session()
+
+            async with session.get(jwks_uri) as response:
+                response.raise_for_status()
+                return await response.json()
         except aiohttp.ClientResponseError as e:
             _LOGGER.warning("Error fetching JWKS: %s - %s", e.status, e.message)
             raise OIDCJWKSInvalid from e
@@ -111,10 +171,11 @@ class OIDCClient:
     async def _make_token_request(self, token_endpoint, query_params):
         """Performs the token POST call"""
         try:
-            async with aiohttp.ClientSession() as session:
-                async with session.post(token_endpoint, data=query_params) as response:
-                    response.raise_for_status()
-                    return await response.json()
+            session = await self._get_http_session()
+
+            async with session.post(token_endpoint, data=query_params) as response:
+                response.raise_for_status()
+                return await response.json()
         except aiohttp.ClientResponseError as e:
             if e.status == 400:
                 _LOGGER.warning(
@@ -356,6 +417,20 @@ class OIDCClient:
 
             # TODO: If the configured claims are not present in id_token, we should fetch userinfo
 
+            # Get and parse groups (to check if it's an array)
+            groups = id_token.get(self.groups_claim, [])
+            if not isinstance(groups, list):
+                _LOGGER.warning("Groups claim is not a list, using empty list instead.")
+                groups = []
+
+            # Assign role if user has the required groups
+            role = "invalid"
+            if self.user_role in groups or self.user_role is None:
+                role = "system-users"
+
+            if self.admin_role in groups:
+                role = "system-admin"
+
             # Create a user details dict based on the contents of the id_token & userinfo
             data: UserDetails = {
                 # Subject Identifier. A locally unique and never reassigned identifier within the
@@ -371,8 +446,8 @@ class OIDCClient:
                 "display_name": id_token.get(self.display_name_claim),
                 # Username, configurable
                 "username": id_token.get(self.username_claim),
-                # Groups, configurable
-                "groups": id_token.get(self.groups_claim),
+                # Role
+                "role": role,
             }
 
             # Log which details were obtained for debugging

custom_components/auth_oidc/provider.py

@@ -259,14 +259,11 @@ class OpenIDAuthProvider(AuthProvider):
         sub = credentials.data["sub"]
         meta = self._user_meta.get(sub, {})
 
-        groups = meta.get("groups", [])
-
-        # TODO: Allow setting which group is for admins
-        group = "system-admin" if "admins" in groups else "system-users"
+        role = meta.get("role")
         return UserMeta(
             name=meta.get("display_name"),
             is_active=True,
-            group=group,
+            group=role,
             local_only=False,
         )
 

custom_components/auth_oidc/types.py

@@ -1,7 +1,9 @@
 """Generic data types"""
 
-
 # Dict class to give a type to the user details
+from typing import Literal
+
+
 class UserDetails(dict):
     """User details representation"""
 
@@ -12,5 +14,5 @@ class UserDetails(dict):
     # Preferred username for the user, will be used when first generating the account
     # or to link the account on first login
     username: str
-    # Groups that the user has, if any are sent from the OIDC provider
-    groups: list[str]
+    # Home Assistant role to assign to this user
+    role: Literal["system-admin", "system-users", "invalid"]

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "hass-oidc-auth"
-version = "0.4.1"
+version = "0.5.1"
 description = "OIDC component for Home Assistant"
 authors = [
     { name = "Christiaan Goossens", email = "contact@christiaangoossens.nl" }