feat: Add support for dynamic proxies URLs

custom_components/hass_proxy/__init__.py

@@ -15,41 +15,45 @@
     from homeassistant.const import Platform
     from homeassistant.core import HomeAssistant
 
-    from .data import HASSProxyData
+    from .data import HASSProxyConfigEntry
 
-from .proxy import async_setup as async_proxy_setup
+from .proxy import async_setup_entry as async_proxy_setup_entry
+from .proxy import async_unload_entry as async_proxy_unload_entry
 
 PLATFORMS: list[Platform] = []
 
 
 # https://developers.home-assistant.io/docs/config_entries_index/#setting-up-an-entry
 async def async_setup_entry(
     hass: HomeAssistant,
-    entry: HASSProxyData,
+    entry: HASSProxyConfigEntry,
 ) -> bool:
     """Set up this integration."""
     LOGGER.info("HASSPROXY Setting up entry %s", entry.entry_id)
 
     await hass.config_entries.async_forward_entry_setups(entry, PLATFORMS)
     entry.async_on_unload(entry.add_update_listener(async_reload_entry))
 
-    await async_proxy_setup(hass)
+    await async_proxy_setup_entry(hass, entry)
 
     return True
 
 
 async def async_unload_entry(
     hass: HomeAssistant,
-    entry: HASSProxyData,
+    entry: HASSProxyConfigEntry,
 ) -> bool:
     """Handle removal of an entry."""
     LOGGER.info("HASSPROXY Unloading entry %s", entry.entry_id)
+
+    await async_proxy_unload_entry(hass, entry)
+
     return await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
 
 
 async def async_reload_entry(
     hass: HomeAssistant,
-    entry: HASSProxyData,
+    entry: HASSProxyConfigEntry,
 ) -> None:
     """Reload config entry."""
     LOGGER.info("HASSPROXY Reloading entry %s", entry.entry_id)

custom_components/hass_proxy/config_flow.py

@@ -9,11 +9,11 @@
 
 from .const import (
     CONF_DYNAMIC_URLS,
-    CONF_SSL_CIPHER_INSECURE,
-    CONF_SSL_CIPHER_INTERMEDIATE,
-    CONF_SSL_CIPHER_MODERN,
-    CONF_SSL_CIPHER_PYTHON_DEFAULT,
     CONF_SSL_CIPHERS,
+    CONF_SSL_CIPHERS_DEFAULT,
+    CONF_SSL_CIPHERS_INSECURE,
+    CONF_SSL_CIPHERS_INTERMEDIATE,
+    CONF_SSL_CIPHERS_MODERN,
     CONF_SSL_VERIFICATION,
     CONF_URL_PATTERNS,
     DEFAULT_OPTIONS,
@@ -38,10 +38,10 @@
         ): selector.SelectSelector(
             selector.SelectSelectorConfig(
                 options=[
-                    CONF_SSL_CIPHER_PYTHON_DEFAULT,
-                    CONF_SSL_CIPHER_MODERN,
-                    CONF_SSL_CIPHER_INTERMEDIATE,
-                    CONF_SSL_CIPHER_INSECURE,
+                    CONF_SSL_CIPHERS_DEFAULT,
+                    CONF_SSL_CIPHERS_MODERN,
+                    CONF_SSL_CIPHERS_INTERMEDIATE,
+                    CONF_SSL_CIPHERS_INSECURE,
                 ],
                 mode=selector.SelectSelectorMode.DROPDOWN,
                 translation_key=CONF_SSL_CIPHERS,

custom_components/hass_proxy/const.py

@@ -1,26 +1,30 @@
 """Constants for hass_proxy."""
 
 from logging import Logger, getLogger
-from typing import Final
-
-from homeassistant.util.ssl import SSLCipherList
+from typing import Final, Literal
 
 DOMAIN: Final = "hass_proxy"
 
 LOGGER: Logger = getLogger(__package__)
 
 CONF_SSL_VERIFICATION: Final = "ssl_verification"
 CONF_SSL_CIPHERS: Final = "ssl_ciphers"
-CONF_SSL_CIPHER_INSECURE: Final = SSLCipherList.INSECURE
-CONF_SSL_CIPHER_MODERN: Final = SSLCipherList.MODERN
-CONF_SSL_CIPHER_INTERMEDIATE: Final = SSLCipherList.INTERMEDIATE
-CONF_SSL_CIPHER_PYTHON_DEFAULT: Final = SSLCipherList.PYTHON_DEFAULT
+
+CONF_SSL_CIPHERS_INSECURE: Final = "insecure"
+CONF_SSL_CIPHERS_MODERN: Final = "modern"
+CONF_SSL_CIPHERS_INTERMEDIATE: Final = "intermediate"
+CONF_SSL_CIPHERS_DEFAULT: Final = "default"
+
+type HASSProxySSLCiphers = Literal["insecure", "modern", "intermediate", "default"]
 
 CONF_DYNAMIC_URLS: Final = "dynamic_urls"
 CONF_URL_PATTERNS: Final = "url_patterns"
 
+SERVICE_CREATE_PROXIED_URL: Final = "create_proxied_url"
+SERVICE_DELETE_PROXIED_URL: Final = "delete_proxied_url"
+
 DEFAULT_OPTIONS: dict[str, str | bool | list[str]] = {
     CONF_SSL_VERIFICATION: True,
     CONF_DYNAMIC_URLS: True,
-    CONF_SSL_CIPHERS: CONF_SSL_CIPHER_PYTHON_DEFAULT,
+    CONF_SSL_CIPHERS: CONF_SSL_CIPHERS_DEFAULT,
 }

custom_components/hass_proxy/data.py

@@ -10,6 +10,17 @@
     from homeassistant.loader import Integration
 
 
+@dataclass
+class DynamicProxiedURL:
+    """A proxied URL."""
+
+    url_pattern: str
+    ssl_verification: bool
+    ssl_ciphers: str
+    open_limit: int
+    time_to_live: int
+
+
 type HASSProxyConfigEntry = ConfigEntry[HASSProxyData]
 
 
@@ -18,3 +29,4 @@ class HASSProxyData:
     """Data for the HASS Proxy integration."""
 
     integration: Integration
+    dynamic_proxied_urls: dict[str, DynamicProxiedURL]

custom_components/hass_proxy/proxy.py

@@ -2,39 +2,150 @@
 
 from __future__ import annotations
 
+import time
 import urllib
+import uuid
 from typing import TYPE_CHECKING, Any
 
 import urlmatch
+import voluptuous as vol
+from homeassistant.core import callback
+from homeassistant.exceptions import ServiceValidationError
+from homeassistant.helpers import config_validation as cv
 from homeassistant.helpers.aiohttp_client import async_get_clientsession
+from homeassistant.loader import async_get_loaded_integration
 from homeassistant.util.ssl import (
+    SSLCipherList,
     client_context,
     client_context_no_verify,
 )
 
 from custom_components.hass_proxy.const import DOMAIN
+from custom_components.hass_proxy.data import (
+    DynamicProxiedURL,
+    HASSProxyConfigEntry,
+    HASSProxyData,
+)
 from custom_components.hass_proxy.proxy_lib import (
-    HASSProxyNotFoundRequestError,
+    HASSProxyLibNotFoundRequestError,
     ProxiedURL,
     ProxyView,
 )
 
-from .const import CONF_SSL_CIPHERS, CONF_SSL_VERIFICATION
+from .const import (
+    CONF_DYNAMIC_URLS,
+    CONF_SSL_CIPHERS,
+    CONF_SSL_CIPHERS_DEFAULT,
+    CONF_SSL_CIPHERS_INSECURE,
+    CONF_SSL_CIPHERS_INTERMEDIATE,
+    CONF_SSL_CIPHERS_MODERN,
+    CONF_SSL_VERIFICATION,
+    SERVICE_CREATE_PROXIED_URL,
+    SERVICE_DELETE_PROXIED_URL,
+)
 
 if TYPE_CHECKING:
     import ssl
     from types import MappingProxyType
 
     import aiohttp
     from aiohttp import web
-    from homeassistant.core import HomeAssistant
+    from homeassistant.core import HomeAssistant, ServiceCall
+
+    from .const import HASSProxySSLCiphers
+
+CREATE_PROXIED_URL_SCHEMA = vol.Schema(
+    {
+        vol.Required("url_pattern"): cv.string,
+        vol.Optional("url_id"): cv.string,
+        vol.Optional("ssl_verification", default=True): cv.boolean,
+        vol.Optional("ssl_ciphers", default=CONF_SSL_CIPHERS_DEFAULT): vol.Any(
+            None,
+            CONF_SSL_CIPHERS_INSECURE,
+            CONF_SSL_CIPHERS_MODERN,
+            CONF_SSL_CIPHERS_INTERMEDIATE,
+            CONF_SSL_CIPHERS_DEFAULT,
+        ),
+        vol.Optional("open_limit", default=1): cv.positive_int,
+        vol.Optional("time_to_live", default=60): cv.positive_int,
+    },
+    required=True,
+)
+
+DELETE_PROXIED_URL_SCHEMA = vol.Schema(
+    {
+        vol.Required("url_id"): cv.string,
+    },
+    required=True,
+)
 
 
-async def async_setup(hass: HomeAssistant) -> None:
-    """Set up the views."""
+class HASSProxyError(Exception):
+    """Exception to indicate a general Proxy error."""
+
+
+class HASSProxyURLIDNotFoundError(HASSProxyError):
+    """Exception to indicate that a URL ID was not found."""
+
+
+@callback
+async def async_setup_entry(hass: HomeAssistant, entry: HASSProxyConfigEntry) -> None:
+    """Set up the proxy entry."""
     session = async_get_clientsession(hass)
     hass.http.register_view(V0ProxyView(hass, session))
 
+    entry.runtime_data = HASSProxyData(
+        integration=async_get_loaded_integration(hass, entry.domain),
+        dynamic_proxied_urls={},
+    )
+
+    def create_proxied_url(call: ServiceCall) -> None:
+        """Create a proxied URL."""
+        url_id = call.data.get("url_id") or str(uuid.uuid4())
+        ttl = call.data["time_to_live"]
+
+        entry.runtime_data.dynamic_proxied_urls[url_id] = DynamicProxiedURL(
+            url_pattern=call.data["url_pattern"],
+            ssl_verification=call.data["ssl_verification"],
+            ssl_ciphers=call.data["ssl_ciphers"],
+            open_limit=call.data["open_limit"],
+            time_to_live=time.time() + ttl if ttl else 0,
+        )
+
+    def delete_proxied_url(call: ServiceCall) -> None:
+        """Delete a proxied URL."""
+        url_id = call.data["url_id"]
+        dynamic_proxied_urls = entry.runtime_data.dynamic_proxied_urls
+
+        if url_id not in dynamic_proxied_urls:
+            raise ServiceValidationError(
+                translation_domain=DOMAIN,
+                translation_key="url_id_not_found",
+                translation_placeholders={"url_id": url_id},
+            )
+        del entry.runtime_data.dynamic_proxied_urls[url_id]
+
+    if entry.options.get(CONF_DYNAMIC_URLS):
+        hass.services.async_register(
+            DOMAIN,
+            SERVICE_CREATE_PROXIED_URL,
+            create_proxied_url,
+            CREATE_PROXIED_URL_SCHEMA,
+        )
+        hass.services.async_register(
+            DOMAIN,
+            SERVICE_DELETE_PROXIED_URL,
+            delete_proxied_url,
+            DELETE_PROXIED_URL_SCHEMA,
+        )
+
+
+@callback
+async def async_unload_entry(hass: HomeAssistant, _entry: HASSProxyConfigEntry) -> None:
+    """Unload the proxy entry."""
+    hass.services.async_remove(DOMAIN, SERVICE_CREATE_PROXIED_URL)
+    hass.services.async_remove(DOMAIN, SERVICE_DELETE_PROXIED_URL)
+
 
 class HAProxyView(ProxyView):
     """A proxy view for HomeAssistant."""
@@ -44,39 +155,74 @@ def __init__(self, hass: HomeAssistant, websession: aiohttp.ClientSession) -> No
         self._hass = hass
         super().__init__(websession)
 
+    def _get_config_entry(self) -> HASSProxyConfigEntry:
+        """Get the config entry."""
+        return self._hass.config_entries.async_entries(DOMAIN)[0]
+
+    def get_dynamic_proxied_urls(self) -> dict[str, DynamicProxiedURL]:
+        """Get the dynamic proxied URLs."""
+        return self._get_config_entry().runtime_data.dynamic_proxied_urls
+
     def _get_options(self) -> MappingProxyType[str, Any]:
         """Get a ConfigEntry options for a given request."""
-        return self._hass.config_entries.async_entries(DOMAIN)[0].options
+        return self._get_config_entry().options
 
-    def _get_url_to_proxy(self, request: web.Request) -> str:
+    def _get_proxied_url(self, request: web.Request) -> ProxiedURL:
         """Get the URL to proxy."""
         if "url" not in request.query:
-            raise HASSProxyNotFoundRequestError
+            raise HASSProxyLibNotFoundRequestError
 
+        options = self._get_options()
         url_to_proxy = urllib.parse.unquote(request.query["url"])
+
+        for proxied_url in self.get_dynamic_proxied_urls().values():
+            if urlmatch.urlmatch(
+                proxied_url.url_pattern,
+                url_to_proxy,
+                path_required=False,
+            ):
+                return ProxiedURL(
+                    url=url_to_proxy,
+                    ssl_context=self._get_ssl_context(proxied_url.ssl_ciphers)
+                    if proxied_url.ssl_verification
+                    else self._get_ssl_context_no_verify(proxied_url.ssl_ciphers),
+                )
+
         for url_pattern in self._get_options().get("url_patterns", []):
             if urlmatch.urlmatch(url_pattern, url_to_proxy, path_required=False):
-                break
-        else:
-            raise HASSProxyNotFoundRequestError
+                ssl_cipher = options.get(CONF_SSL_CIPHERS)
+                ssl_verification = options.get(CONF_SSL_VERIFICATION, True)
 
-        return url_to_proxy
+                return ProxiedURL(
+                    url=url_to_proxy,
+                    ssl_context=self._get_ssl_context(ssl_cipher)
+                    if ssl_verification
+                    else self._get_ssl_context_no_verify(ssl_cipher),
+                )
 
-    def _get_ssl_context(self) -> ssl.SSLContext:
+        raise HASSProxyLibNotFoundRequestError
+
+    def _get_ssl_context_no_verify(
+        self, ssl_cipher: HASSProxySSLCiphers
+    ) -> ssl.SSLContext:
         """Get an SSL context."""
-        options = self._get_options()
+        return client_context_no_verify(
+            self._proxy_ssl_cipher_to_ha_ssl_cipher(ssl_cipher)
+        )
 
-        if not options.get(CONF_SSL_VERIFICATION, True):
-            return client_context_no_verify(options.get(CONF_SSL_CIPHERS))
-        return client_context(options.get(CONF_SSL_CIPHERS))
+    def _get_ssl_context(self, ssl_ciphers: HASSProxySSLCiphers) -> ssl.SSLContext:
+        """Get an SSL context."""
+        return client_context(self._proxy_ssl_cipher_to_ha_ssl_cipher(ssl_ciphers))
+
+    def _proxy_ssl_cipher_to_ha_ssl_cipher(self, ssl_ciphers: str) -> SSLCipherList:
+        """Convert a proxy SSL cipher to a HA SSL cipher."""
+        if ssl_ciphers == CONF_SSL_CIPHERS_DEFAULT:
+            return SSLCipherList.PYTHON_DEFAULT
+        return ssl_ciphers
 
 
 class V0ProxyView(HAProxyView):
     """A v0 proxy endpoint."""
 
     url = "/api/hass_proxy/v0/"
     name = "api:hass_proxy:v0"
-
-    def _get_proxied_url(self, request: web.Request, **_kwargs: Any) -> str:
-        """Create path."""
-        return ProxiedURL(self._get_url_to_proxy(request), self._get_ssl_context())