hass-web-proxy-lib 0.0.5

A library to proxy web traffic through Home Assistant integrations.

hass-web-proxy-lib

A small Home Assistant library to proxy web traffic through Home Assistant. Used by the Home Assistant Web Proxy Integration and any other integration that needs to proxy traffic through Home Assistant.

This library is not itself an integration but rather can be used by integration developers to offer proxying capabilities within their own integration.

Usage

Use this library as part of your custom integration by declaring a new HomeAssistantView that inherits from the ProxyView or WebsocketProxyView class in this library. Callers must implement the _get_proxied_url method to return a ProxiedURL object containing a destination URL for a given proxy request, or raising an exception to indicate an error condition.

Example Usage

Proxies a GET request from https://$HA_INSTANCE/api/my_integration/proxy/ through to /dir/file relative to the URL stored in the config entry data.

@callback
async def async_setup_entry(hass: HomeAssistant) -> None:
    """Set up the HASS web proxy entry."""
    session = async_get_clientsession(hass)
    hass.http.register_view(MyProxyView(hass, session))


class MyProxyView(ProxyView):
    """A proxy view for My Integration."""

    url = "/api/my_integration/proxy/"
    name = "api:my_integration:proxy"

    def _get_proxied_url(self, request: web.Request) -> ProxiedURL:
        """Get the URL to proxy."""
        # Retrieve host to connect to from config entry data.
        # Specifics depend on your application, this is an example only.
        config_entries = hass.config_entries.async_entries(DOMAIN):
        if not config_entries:
            raise HASSWebProxyLibNotFoundRequestError
        url = config_entries[0].data.get(CONF_URL)
        if not url:
            raise HASSWebProxyLibNotFoundRequestError
        return ProxiedURL(url=f"${url}/dir/file")

See the hass-web-proxy-integration for a more complete example of usage of this library.

Key Classes

ProxyView

The main class to inherit from for simple GET request proxying. Inheritors must implement _get_proxied_url(...) to return a ProxiedURL object.

WebsocketProxyView

The class to inherit from for websocket proxying. Inheritors must implement _get_proxied_url(...) to return a ProxiedURL object.

ProxiedURL

A small dataclass returned by overridden _get_proxied_url(...) methods that describes how the library should proxy a given request.

Field name	Default	Description
url		The destination URL a given request should be made to, e.g. https://my-backend.my-domain.io.
allow_unauthenticated	False	When False or unset, unauthenticated HA traffic will be rejected with a 401 Unauthorized status. When True, unauthenticated traffic will be allowed.
headers		An optional dictionary of headers to set on the outbound request.
query_params		An optional dictionary of query parameters to set in the target URL. This is a convenience alternative to the caller simply adding the query string parameters onto the url parameter.
ssl_context		An optional SSLContext object that should be used for secure onward requests.

Errors

HASSWebProxyLibBadRequestError

Can be raised by _get_proxied_url(...) to indicate a bad request (400 Bad Request).

HASSWebProxyLibUnauthorizedRequestError

Can be raised by _get_proxied_url(...) to indicate an unauthorized request (401 Unauthorized).

HASSWebProxyLibForbiddenRequestError

Can be raised by _get_proxied_url(...) to indicate a forbidden request (403 Forbidden).

HASSWebProxyLibNotFoundRequestError

Can be raised by _get_proxied_url(...) to indicate a request is not found request (404 Not Found).

HASSWebProxyLibExpiredError

Can be raised by _get_proxied_url(...) to indicate an expired / permanently removed resource is not available (410 Gone).

Testing

This library also contains a small test utility and fixture file that can be used to test proxying.

Test Handlers

response_handler(request: web.Request) -> web.Response

A small response handler that will return a json object containing:

Field name	Description
headers	A dictionary of the request headers received.
url	The full URL/querystring requested.

ws_response_handler(request: web.Request) -> web.WebSocketResponse

A small websocket response handler that will initially return a json object containing headers and url (as in response_handler above), and then will simply echo back any future text or binary data.

Test Fixtures

The local_server fixture will start a small aiohttp server that can be "proxied to". The server listens to / and /ws for simple GET requests and websockets respectively using the above handlers. The resultant server address will be available within the local_server variable as a URL object.

Example Test Usage

import pytest
from hass_web_proxy_lib.tests.utils import response_handler, ws_response_handler

# Add the HA and local_server fixtures.
pytest_plugins = [
    "pytest_homeassistant_custom_component",
]

# Start a server that listens to requests for /dir/file and /dir/websocket .
@pytest.fixture
async def local_backend(hass: HomeAssistant, aiohttp_server: Any) -> Any:
    """Start a local backend."""

    # Start a local backend.
    app = web.Application()
    app.add_routes([
        web.get("/dir/file", response_handler),
        web.get("/dir/websocket", ws_response_handler),

    ])

    # Create a config entry for my integration, passing in the local backend
    # address as config entry data.
    config_entry: MockConfigEntry = MockConfigEntry(
        entry_id="74565ad414754616000674c87bdc876c",
        domain=DOMAIN,
        data={CONF_URL: str(app.make_url("/"))},
        title="My entry"
    )
    config_entry.add_to_hass(hass)

    # Setup the integration.
    await hass.config_entries.async_setup(config_entry.entry_id)
    await hass.async_block_till_done()

    return await aiohttp_server(app)


async def test_proxy_view_success(
    hass: HomeAssistant,
    local_backend: Any,
    hass_client: Any,
) -> None:
    """Test that a valid URL proxies successfully."""
    authenticated_hass_client = await hass_client()
    # The proxy code above in `_get_proxied_url` should result in a request to
    # $URL/dir/file which should return an OK HTTP status.
    resp = await authenticated_hass_client.get(f"/api/my_integration/proxy/")
    assert resp.status == HTTPStatus.OK