Allow accessing frigate with authentication

custom_components/frigate/__init__.py

@@ -25,7 +25,13 @@
 )
 from homeassistant.components.switch import DOMAIN as SWITCH_DOMAIN
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.const import ATTR_MODEL, CONF_HOST, CONF_URL
+from homeassistant.const import (
+    ATTR_MODEL,
+    CONF_HOST,
+    CONF_PASSWORD,
+    CONF_URL,
+    CONF_USERNAME,
+)
 from homeassistant.core import HomeAssistant, callback, valid_entity_id
 from homeassistant.exceptions import ConfigEntryNotReady
 from homeassistant.helpers import device_registry as dr, entity_registry as er
@@ -196,6 +202,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
     client = FrigateApiClient(
         str(entry.data.get(CONF_URL)),
         async_get_clientsession(hass),
+        str(entry.data.get(CONF_USERNAME)),
+        str(entry.data.get(CONF_PASSWORD)),
     )
     coordinator = FrigateDataUpdateCoordinator(hass, client=client)
     await coordinator.async_config_entry_first_refresh()

custom_components/frigate/api.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+import datetime
 import logging
 import socket
 from typing import Any, cast
@@ -11,6 +12,8 @@
 import async_timeout
 from yarl import URL
 
+from homeassistant.auth import jwt_wrapper
+
 TIMEOUT = 10
 
 _LOGGER: logging.Logger = logging.getLogger(__name__)
@@ -31,10 +34,19 @@ class FrigateApiClientError(Exception):
 class FrigateApiClient:
     """Frigate API client."""
 
-    def __init__(self, host: str, session: aiohttp.ClientSession) -> None:
+    def __init__(
+        self,
+        host: str,
+        session: aiohttp.ClientSession,
+        username: str | None = None,
+        password: str | None = None,
+    ) -> None:
         """Construct API Client."""
         self._host = host
         self._session = session
+        self._username = username
+        self._password = password
+        self._token_data: dict[str, Any] = {}
 
     async def async_get_version(self) -> str:
         """Get data from the API."""
@@ -216,27 +228,98 @@ async def async_get_recordings(
         )
         return cast(dict[str, Any], result) if decode_json else result
 
+    async def _get_token(self) -> None:
+        """
+        Obtain a new JWT token using the provided username and password.
+        Sends a POST request to the login endpoint and extracts the token
+        and expiration date from the response headers.
+        """
+        response = await self.api_wrapper(
+            method="post",
+            url=str(URL(self._host) / "api/login"),
+            data={"user": self._username, "password": self._password},
+            decode_json=False,
+            login_request=True,
+        )
+
+        set_cookie_header = response.headers.get("Set-Cookie", "")
+        if not set_cookie_header:
+            raise KeyError("Missing Set-Cookie header in response")
+
+        for cookie_prop in set_cookie_header.split(";"):
+            cookie_prop = cookie_prop.strip()
+            if cookie_prop.startswith("frigate_token="):
+                jwt_token = cookie_prop.split("=", 1)[1]
+                self._token_data["token"] = jwt_token
+                try:
+                    decoded_token = jwt_wrapper.unverified_hs256_token_decode(jwt_token)
+                except Exception as e:
+                    raise ValueError(f"Failed to decode JWT token: {e}")
+                exp_timestamp = decoded_token.get("exp")
+                if not exp_timestamp:
+                    raise KeyError("JWT is missing 'exp' claim")
+                self._token_data["expires"] = datetime.datetime.fromtimestamp(
+                    exp_timestamp, datetime.UTC
+                )
+                break
+        else:
+            raise KeyError("Missing 'frigate_token' in Set-Cookie header")
+
+    async def _refresh_token_if_needed(self) -> None:
+        """
+        Refresh the JWT token if it is expired or about to expire.
+        """
+        if "expires" not in self._token_data:
+            await self._get_token()
+            return
+
+        current_time = datetime.datetime.now(datetime.UTC)
+        if current_time >= self._token_data["expires"]:  # Compare UTC-aware datetimes
+            await self._get_token()
+
+    async def _get_auth_headers(self) -> dict[str, str]:
+        """
+        Get headers for API requests, including the JWT token if available.
+        Ensures that the token is refreshed if needed.
+        """
+        headers = {}
+
+        if self._username and self._password:
+            await self._refresh_token_if_needed()
+
+            if "token" in self._token_data:
+                headers["Authorization"] = f"Bearer {self._token_data['token']}"
+
+        return headers
+
     async def api_wrapper(
         self,
         method: str,
         url: str,
         data: dict | None = None,
         headers: dict | None = None,
         decode_json: bool = True,
+        login_request: bool = False,
     ) -> Any:
         """Get information from the API."""
         if data is None:
             data = {}
         if headers is None:
             headers = {}
 
+        if not login_request:
+            headers.update(await self._get_auth_headers())
+
         try:
             async with async_timeout.timeout(TIMEOUT):
                 func = getattr(self._session, method)
                 if func:
                     response = await func(
                         url, headers=headers, raise_for_status=True, json=data
                     )
+                    response.raise_for_status()
+                    if login_request:
+                        return response
                     if decode_json:
                         return await response.json()
                     return await response.text()
@@ -249,6 +332,27 @@ async def api_wrapper(
             )
             raise FrigateApiClientError from exc
 
+        except aiohttp.ClientResponseError as exc:
+            if exc.status == 401:
+                _LOGGER.error(
+                    "Unauthorized (401) error for URL %s: %s", url, exc.message
+                )
+                raise FrigateApiClientError(
+                    "Unauthorized access - check credentials."
+                ) from exc
+            elif exc.status == 403:
+                _LOGGER.error("Forbidden (403) error for URL %s: %s", url, exc.message)
+                raise FrigateApiClientError(
+                    "Forbidden - insufficient permissions."
+                ) from exc
+            else:
+                _LOGGER.error(
+                    "Client response error (%d) for URL %s: %s",
+                    exc.status,
+                    url,
+                    exc.message,
+                )
+                raise FrigateApiClientError from exc
         except (KeyError, TypeError) as exc:
             _LOGGER.error(
                 "Error parsing information from %s: %s",

custom_components/frigate/config_flow.py

@@ -10,7 +10,7 @@
 from yarl import URL
 
 from homeassistant import config_entries
-from homeassistant.const import CONF_URL
+from homeassistant.const import CONF_PASSWORD, CONF_URL, CONF_USERNAME
 from homeassistant.core import callback
 from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.aiohttp_client import async_create_clientsession
@@ -82,7 +82,12 @@ async def _handle_config_step(
 
         try:
             session = async_create_clientsession(self.hass)
-            client = FrigateApiClient(user_input[CONF_URL], session)
+            client = FrigateApiClient(
+                user_input[CONF_URL],
+                session,
+                user_input.get(CONF_USERNAME),
+                user_input.get(CONF_PASSWORD),
+            )
             await client.async_get_stats()
         except FrigateApiClientError:
             return self._show_config_form(
@@ -122,7 +127,13 @@ def _show_config_form(
                 {
                     vol.Required(
                         CONF_URL, default=user_input.get(CONF_URL, DEFAULT_HOST)
-                    ): str
+                    ): str,
+                    vol.Optional(
+                        CONF_USERNAME, default=user_input.get(CONF_USERNAME, "")
+                    ): str,
+                    vol.Optional(
+                        CONF_PASSWORD, default=user_input.get(CONF_PASSWORD, "")
+                    ): str,
                 }
             ),
             errors=errors,

custom_components/frigate/const.py

@@ -50,6 +50,7 @@
 CONF_MEDIA_BROWSER_ENABLE = "media_browser_enable"
 CONF_NOTIFICATION_PROXY_ENABLE = "notification_proxy_enable"
 CONF_NOTIFICATION_PROXY_EXPIRE_AFTER_SECONDS = "notification_proxy_expire_after_seconds"
+CONF_USERNAME = "username"
 CONF_PASSWORD = "password"
 CONF_PATH = "path"
 CONF_RTSP_URL_TEMPLATE = "rtsp_url_template"

requirements.txt

@@ -6,4 +6,4 @@ homeassistant==2024.12.0
 paho-mqtt
 python-dateutil
 yarl
-hass-web-proxy-lib==0.0.7
+hass-web-proxy-lib==0.0.7