mirror of
https://github.com/home-assistant/core.git
synced 2026-08-24 02:24:51 -05:00
Unify query token auth in http views (#173082)
This commit is contained in:
@@ -1,18 +1,19 @@
|
||||
"""The Brands integration."""
|
||||
|
||||
from collections import deque
|
||||
from collections.abc import Container, Mapping
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from random import SystemRandom
|
||||
import time
|
||||
from typing import Any, Final
|
||||
from typing import Any, Final, override
|
||||
|
||||
from aiohttp import ClientError, hdrs, web
|
||||
from aiohttp import ClientError, web
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.core import HomeAssistant, callback, valid_domain
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
@@ -108,23 +109,18 @@ def _read_brand_file(brand_dir: Path, image: str) -> bytes | None:
|
||||
class _BrandsBaseView(HomeAssistantView):
|
||||
"""Base view for serving brand images."""
|
||||
|
||||
requires_auth = False
|
||||
use_query_token_for_auth = True
|
||||
|
||||
def __init__(self, hass: HomeAssistant) -> None:
|
||||
"""Initialize the view."""
|
||||
self._hass = hass
|
||||
self._cache_dir = Path(hass.config.cache_path(DOMAIN))
|
||||
|
||||
def _authenticate(self, request: web.Request) -> None:
|
||||
"""Authenticate the request using Bearer token or query token."""
|
||||
access_tokens: deque[str] = self._hass.data[DOMAIN]
|
||||
authenticated = (
|
||||
request[KEY_AUTHENTICATED] or request.query.get("token") in access_tokens
|
||||
)
|
||||
if not authenticated:
|
||||
if hdrs.AUTHORIZATION in request.headers:
|
||||
raise web.HTTPUnauthorized
|
||||
raise web.HTTPForbidden
|
||||
@callback
|
||||
@override
|
||||
def get_valid_auth_tokens(self, match_info: Mapping[str, str]) -> Container[str]:
|
||||
"""Return valid auth tokens, which can be used for query token authentication."""
|
||||
return self._hass.data[DOMAIN]
|
||||
|
||||
async def _serve_from_custom_integration(
|
||||
self,
|
||||
@@ -240,8 +236,6 @@ class BrandsIntegrationView(_BrandsBaseView):
|
||||
image: str,
|
||||
) -> web.Response:
|
||||
"""Handle GET request for an integration brand image."""
|
||||
self._authenticate(request)
|
||||
|
||||
if not valid_domain(domain) or image not in ALLOWED_IMAGES:
|
||||
return web.Response(status=HTTPStatus.NOT_FOUND)
|
||||
|
||||
@@ -274,8 +268,6 @@ class BrandsHardwareView(_BrandsBaseView):
|
||||
image: str,
|
||||
) -> web.Response:
|
||||
"""Handle GET request for a hardware brand image."""
|
||||
self._authenticate(request)
|
||||
|
||||
if not CATEGORY_RE.match(category):
|
||||
return web.Response(status=HTTPStatus.NOT_FOUND)
|
||||
# Hardware images have dynamic names like "manufacturer_model.png"
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
import asyncio
|
||||
import collections
|
||||
from collections.abc import Awaitable, Callable, Coroutine
|
||||
from collections.abc import Awaitable, Callable, Container, Coroutine, Mapping
|
||||
from contextlib import suppress
|
||||
from dataclasses import asdict, dataclass
|
||||
from datetime import datetime, timedelta
|
||||
@@ -12,16 +12,16 @@ import logging
|
||||
import os
|
||||
from random import SystemRandom
|
||||
import time
|
||||
from typing import Any, Final, final
|
||||
from typing import Any, Final, final, override
|
||||
|
||||
from aiohttp import hdrs, web
|
||||
from aiohttp import web
|
||||
import attr
|
||||
from propcache.api import cached_property, under_cached_property
|
||||
import voluptuous as vol
|
||||
from webrtc_models import RTCIceCandidateInit
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.components.media_player import (
|
||||
ATTR_MEDIA_CONTENT_ID,
|
||||
ATTR_MEDIA_CONTENT_TYPE,
|
||||
@@ -776,30 +776,26 @@ class Camera(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
|
||||
class CameraView(HomeAssistantView):
|
||||
"""Base CameraView."""
|
||||
|
||||
requires_auth = False
|
||||
use_query_token_for_auth = True
|
||||
|
||||
def __init__(self, component: EntityComponent[Camera]) -> None:
|
||||
"""Initialize a basic camera view."""
|
||||
self.component = component
|
||||
|
||||
@callback
|
||||
@override
|
||||
def get_valid_auth_tokens(self, match_info: Mapping[str, str]) -> Container[str]:
|
||||
"""Return valid auth tokens, which can be used for query token authentication."""
|
||||
if (camera := self.component.get_entity(match_info["entity_id"])) is None:
|
||||
return ()
|
||||
|
||||
return camera.access_tokens
|
||||
|
||||
async def get(self, request: web.Request, entity_id: str) -> web.StreamResponse:
|
||||
"""Start a GET request."""
|
||||
if (camera := self.component.get_entity(entity_id)) is None:
|
||||
raise web.HTTPNotFound
|
||||
|
||||
authenticated = (
|
||||
request[KEY_AUTHENTICATED]
|
||||
or request.query.get("token") in camera.access_tokens
|
||||
)
|
||||
|
||||
if not authenticated:
|
||||
# Attempt with invalid bearer token, raise unauthorized
|
||||
# so ban middleware can handle it.
|
||||
if hdrs.AUTHORIZATION in request.headers:
|
||||
raise web.HTTPUnauthorized
|
||||
# Invalid sigAuth or camera access token
|
||||
raise web.HTTPForbidden
|
||||
|
||||
if not camera.is_on:
|
||||
_LOGGER.debug("Camera is off")
|
||||
raise web.HTTPServiceUnavailable
|
||||
|
||||
@@ -2,20 +2,21 @@
|
||||
|
||||
import asyncio
|
||||
import collections
|
||||
from collections.abc import Container, Mapping
|
||||
from contextlib import suppress
|
||||
from dataclasses import dataclass
|
||||
from datetime import datetime, timedelta
|
||||
import logging
|
||||
import os
|
||||
from random import SystemRandom
|
||||
from typing import Final, final
|
||||
from typing import Final, final, override
|
||||
|
||||
from aiohttp import hdrs, web
|
||||
from aiohttp import web
|
||||
import httpx
|
||||
from propcache.api import cached_property
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.http import KEY_AUTHENTICATED, KEY_HASS, HomeAssistantView
|
||||
from homeassistant.components.http import KEY_HASS, HomeAssistantView
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import CONTENT_TYPE_MULTIPART, EVENT_HOMEASSISTANT_STOP
|
||||
from homeassistant.core import (
|
||||
@@ -314,33 +315,28 @@ class ImageView(HomeAssistantView):
|
||||
"""View to serve an image."""
|
||||
|
||||
name = "api:image:image"
|
||||
requires_auth = False
|
||||
use_query_token_for_auth = True
|
||||
url = "/api/image_proxy/{entity_id}"
|
||||
|
||||
def __init__(self, component: EntityComponent[ImageEntity]) -> None:
|
||||
"""Initialize an image view."""
|
||||
self.component = component
|
||||
|
||||
async def _authenticate_request(
|
||||
self, request: web.Request, entity_id: str
|
||||
) -> ImageEntity:
|
||||
"""Authenticate request and return image entity."""
|
||||
@callback
|
||||
@override
|
||||
def get_valid_auth_tokens(self, match_info: Mapping[str, str]) -> Container[str]:
|
||||
"""Return valid auth tokens, which can be used for query token authentication."""
|
||||
if (image_entity := self.component.get_entity(match_info["entity_id"])) is None:
|
||||
return ()
|
||||
|
||||
return image_entity.access_tokens
|
||||
|
||||
@callback
|
||||
def _get_image_entity(self, entity_id: str) -> ImageEntity:
|
||||
"""Get image entity from request."""
|
||||
if (image_entity := self.component.get_entity(entity_id)) is None:
|
||||
raise web.HTTPNotFound
|
||||
|
||||
authenticated = (
|
||||
request[KEY_AUTHENTICATED]
|
||||
or request.query.get("token") in image_entity.access_tokens
|
||||
)
|
||||
|
||||
if not authenticated:
|
||||
# Attempt with invalid bearer token, raise unauthorized
|
||||
# so ban middleware can handle it.
|
||||
if hdrs.AUTHORIZATION in request.headers:
|
||||
raise web.HTTPUnauthorized
|
||||
# Invalid sigAuth or image entity access token
|
||||
raise web.HTTPForbidden
|
||||
|
||||
return image_entity
|
||||
|
||||
async def head(self, request: web.Request, entity_id: str) -> web.Response:
|
||||
@@ -349,7 +345,7 @@ class ImageView(HomeAssistantView):
|
||||
This is sent by some DLNA renderers, like Samsung ones, prior to sending
|
||||
the GET request.
|
||||
"""
|
||||
image_entity = await self._authenticate_request(request, entity_id)
|
||||
image_entity = self._get_image_entity(entity_id)
|
||||
|
||||
# Don't use `handle` as we don't care about the stream case, we only want
|
||||
# to verify that the image exists.
|
||||
@@ -365,7 +361,7 @@ class ImageView(HomeAssistantView):
|
||||
|
||||
async def get(self, request: web.Request, entity_id: str) -> web.StreamResponse:
|
||||
"""Start a GET request."""
|
||||
image_entity = await self._authenticate_request(request, entity_id)
|
||||
image_entity = self._get_image_entity(entity_id)
|
||||
return await self.handle(request, image_entity)
|
||||
|
||||
async def handle(
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
import asyncio
|
||||
import collections
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Container, Mapping
|
||||
from contextlib import suppress
|
||||
import datetime as dt
|
||||
from enum import StrEnum
|
||||
@@ -12,7 +12,7 @@ import hashlib
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
import secrets
|
||||
from typing import Any, Final, Required, TypedDict, final
|
||||
from typing import Any, Final, Required, TypedDict, final, override
|
||||
from urllib.parse import quote, urlparse
|
||||
|
||||
import aiohttp
|
||||
@@ -24,7 +24,7 @@ import voluptuous as vol
|
||||
from yarl import URL
|
||||
|
||||
from homeassistant.components import websocket_api
|
||||
from homeassistant.components.http import KEY_AUTHENTICATED, HomeAssistantView
|
||||
from homeassistant.components.http import HomeAssistantView
|
||||
from homeassistant.components.websocket_api import ERR_NOT_SUPPORTED, ERR_UNKNOWN_ERROR
|
||||
from homeassistant.config_entries import ConfigEntry
|
||||
from homeassistant.const import ( # noqa: F401
|
||||
@@ -50,7 +50,7 @@ from homeassistant.const import ( # noqa: F401
|
||||
STATE_PLAYING,
|
||||
STATE_STANDBY,
|
||||
)
|
||||
from homeassistant.core import HomeAssistant, SupportsResponse
|
||||
from homeassistant.core import HomeAssistant, SupportsResponse, callback
|
||||
from homeassistant.helpers import config_validation as cv
|
||||
from homeassistant.helpers.aiohttp_client import async_get_clientsession
|
||||
from homeassistant.helpers.entity import Entity, EntityDescription
|
||||
@@ -1249,7 +1249,7 @@ class MediaPlayerEntity(Entity, cached_properties=CACHED_PROPERTIES_WITH_ATTR_):
|
||||
class MediaPlayerImageView(HomeAssistantView):
|
||||
"""Media player view to serve an image."""
|
||||
|
||||
requires_auth = False
|
||||
use_query_token_for_auth = True
|
||||
url = "/api/media_player_proxy/{entity_id}"
|
||||
name = "api:media_player:image"
|
||||
extra_urls = [
|
||||
@@ -1262,6 +1262,15 @@ class MediaPlayerImageView(HomeAssistantView):
|
||||
"""Initialize a media player view."""
|
||||
self.component = component
|
||||
|
||||
@callback
|
||||
@override
|
||||
def get_valid_auth_tokens(self, match_info: Mapping[str, str]) -> Container[str]:
|
||||
"""Return valid auth tokens, which can be used for query token authentication."""
|
||||
if (player := self.component.get_entity(match_info["entity_id"])) is None:
|
||||
return ()
|
||||
|
||||
return (player.access_token,)
|
||||
|
||||
async def get(
|
||||
self,
|
||||
request: web.Request,
|
||||
@@ -1271,21 +1280,9 @@ class MediaPlayerImageView(HomeAssistantView):
|
||||
) -> web.Response:
|
||||
"""Start a get request."""
|
||||
if (player := self.component.get_entity(entity_id)) is None:
|
||||
status = (
|
||||
HTTPStatus.NOT_FOUND
|
||||
if request[KEY_AUTHENTICATED]
|
||||
else HTTPStatus.UNAUTHORIZED
|
||||
)
|
||||
return web.Response(status=status)
|
||||
return web.Response(status=HTTPStatus.NOT_FOUND)
|
||||
|
||||
assert isinstance(player, MediaPlayerEntity)
|
||||
authenticated = (
|
||||
request[KEY_AUTHENTICATED]
|
||||
or request.query.get("token") == player.access_token
|
||||
)
|
||||
|
||||
if not authenticated:
|
||||
return web.Response(status=HTTPStatus.UNAUTHORIZED)
|
||||
|
||||
if media_content_type and media_content_id:
|
||||
media_image_id = request.query.get("media_image_id")
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Helper to track the current http request."""
|
||||
|
||||
from collections.abc import Awaitable, Callable
|
||||
from collections.abc import Awaitable, Callable, Container, Mapping
|
||||
from contextvars import ContextVar
|
||||
from http import HTTPStatus
|
||||
import inspect
|
||||
@@ -20,7 +20,7 @@ import voluptuous as vol
|
||||
|
||||
from homeassistant import exceptions
|
||||
from homeassistant.const import CONTENT_TYPE_JSON
|
||||
from homeassistant.core import Context, HomeAssistant, is_callback
|
||||
from homeassistant.core import Context, HomeAssistant, callback, is_callback
|
||||
from homeassistant.util.json import JSON_ENCODE_EXCEPTIONS, format_unserializable_data
|
||||
|
||||
from .json import find_paths_unserializable_data, json_bytes, json_dumps
|
||||
@@ -55,7 +55,13 @@ def request_handler_factory(
|
||||
|
||||
authenticated = request.get(KEY_AUTHENTICATED, False)
|
||||
|
||||
if view.requires_auth and not authenticated:
|
||||
if view.use_query_token_for_auth and not authenticated:
|
||||
token = request.query.get("token")
|
||||
if token and token in view.get_valid_auth_tokens(request.match_info):
|
||||
_LOGGER.debug("Authenticated request with query token")
|
||||
authenticated = True
|
||||
|
||||
if (view.requires_auth or view.use_query_token_for_auth) and not authenticated:
|
||||
# Import here to avoid circular dependency with network.py
|
||||
from .network import NoURLAvailableError, get_url # noqa: PLC0415
|
||||
|
||||
@@ -129,6 +135,7 @@ class HomeAssistantView:
|
||||
extra_urls: list[str] = []
|
||||
# Views inheriting from this class can override this
|
||||
requires_auth = True
|
||||
use_query_token_for_auth = False
|
||||
cors_allowed = False
|
||||
|
||||
@staticmethod
|
||||
@@ -204,3 +211,8 @@ class HomeAssistantView:
|
||||
if allow_cors:
|
||||
for route in routes:
|
||||
allow_cors(route)
|
||||
|
||||
@callback
|
||||
def get_valid_auth_tokens(self, match_info: Mapping[str, str]) -> Container[str]:
|
||||
"""Return valid auth tokens, which can be used for query token authentication."""
|
||||
return ()
|
||||
|
||||
@@ -809,30 +809,30 @@ async def test_token_query_param_authentication(
|
||||
assert await resp.read() == FAKE_PNG
|
||||
|
||||
|
||||
async def test_unauthenticated_request_forbidden(
|
||||
async def test_unauthenticated_request_unauthorized(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
aioclient_mock: AiohttpClientMocker,
|
||||
) -> None:
|
||||
"""Test that unauthenticated requests are forbidden."""
|
||||
"""Test that unauthenticated requests are unauthorized."""
|
||||
client = await hass_client_no_auth()
|
||||
|
||||
resp = await client.get("/api/brands/integration/hue/icon.png")
|
||||
assert resp.status == HTTPStatus.FORBIDDEN
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
resp = await client.get("/api/brands/hardware/boards/green.png")
|
||||
assert resp.status == HTTPStatus.FORBIDDEN
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
|
||||
async def test_invalid_token_forbidden(
|
||||
async def test_invalid_token_unauthorized(
|
||||
hass: HomeAssistant,
|
||||
hass_client_no_auth: ClientSessionGenerator,
|
||||
) -> None:
|
||||
"""Test that an invalid access token in query param is forbidden."""
|
||||
"""Test that an invalid access token in query param is unauthorized."""
|
||||
client = await hass_client_no_auth()
|
||||
resp = await client.get("/api/brands/integration/hue/icon.png?token=invalid_token")
|
||||
|
||||
assert resp.status == HTTPStatus.FORBIDDEN
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
|
||||
async def test_invalid_bearer_token_unauthorized(
|
||||
|
||||
@@ -693,6 +693,30 @@ async def test_camera_proxy_stream(hass_client: ClientSessionGenerator) -> None:
|
||||
assert response.status == HTTPStatus.BAD_GATEWAY
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_camera")
|
||||
async def test_camera_proxy_query_token_auth(
|
||||
hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
|
||||
) -> None:
|
||||
"""Test the camera proxy authenticates via the access token query param."""
|
||||
client = await hass_client_no_auth()
|
||||
|
||||
state = hass.states.get("camera.demo_camera")
|
||||
assert state is not None
|
||||
|
||||
# A valid access token in the query param authenticates the request
|
||||
resp = await client.get(state.attributes["entity_picture"])
|
||||
assert resp.status == HTTPStatus.OK
|
||||
assert await resp.read() == b"Test"
|
||||
|
||||
# Without a token the request is unauthorized
|
||||
resp = await client.get("/api/camera_proxy/camera.demo_camera")
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
# An invalid token is also unauthorized
|
||||
resp = await client.get("/api/camera_proxy/camera.demo_camera?token=invalid")
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
|
||||
@pytest.mark.usefixtures("mock_camera")
|
||||
async def test_state_streaming(hass: HomeAssistant) -> None:
|
||||
"""Camera state."""
|
||||
|
||||
@@ -404,14 +404,27 @@ async def test_failed_login_attempts_counter(
|
||||
|
||||
app.router.add_get(
|
||||
"/auth_true",
|
||||
request_handler_factory(hass, Mock(requires_auth=True), auth_true_handler),
|
||||
request_handler_factory(
|
||||
hass,
|
||||
Mock(requires_auth=True, use_query_token_for_auth=False),
|
||||
auth_true_handler,
|
||||
),
|
||||
)
|
||||
app.router.add_get(
|
||||
"/auth_false",
|
||||
request_handler_factory(hass, Mock(requires_auth=True), auth_handler),
|
||||
request_handler_factory(
|
||||
hass,
|
||||
Mock(requires_auth=True, use_query_token_for_auth=False),
|
||||
auth_handler,
|
||||
),
|
||||
)
|
||||
app.router.add_get(
|
||||
"/", request_handler_factory(hass, Mock(requires_auth=False), auth_handler)
|
||||
"/",
|
||||
request_handler_factory(
|
||||
hass,
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
auth_handler,
|
||||
),
|
||||
)
|
||||
|
||||
setup_bans(hass, app, 5)
|
||||
|
||||
@@ -61,7 +61,7 @@ async def test_handling_unauthorized(mock_request: Mock) -> None:
|
||||
with pytest.raises(HTTPUnauthorized):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=False),
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
AsyncMock(side_effect=Unauthorized),
|
||||
)(mock_request)
|
||||
|
||||
@@ -71,7 +71,7 @@ async def test_handling_invalid_data(mock_request: Mock) -> None:
|
||||
with pytest.raises(HTTPBadRequest):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=False),
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
AsyncMock(side_effect=vol.Invalid("yo")),
|
||||
)(mock_request)
|
||||
|
||||
@@ -81,7 +81,7 @@ async def test_handling_service_not_found(mock_request: Mock) -> None:
|
||||
with pytest.raises(HTTPInternalServerError):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=False),
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
AsyncMock(side_effect=ServiceNotFound("test", "test")),
|
||||
)(mock_request)
|
||||
|
||||
@@ -90,7 +90,7 @@ async def test_not_running(mock_request_with_stopping: Mock) -> None:
|
||||
"""Test we get a 503 when not running."""
|
||||
response = await request_handler_factory(
|
||||
mock_request_with_stopping.app[KEY_HASS],
|
||||
Mock(requires_auth=False),
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
AsyncMock(side_effect=Unauthorized),
|
||||
)(mock_request_with_stopping)
|
||||
assert response.status == HTTPStatus.SERVICE_UNAVAILABLE
|
||||
@@ -101,11 +101,64 @@ async def test_invalid_handler(mock_request: Mock) -> None:
|
||||
with pytest.raises(TypeError):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=False),
|
||||
Mock(requires_auth=False, use_query_token_for_auth=False),
|
||||
AsyncMock(return_value=["not valid"]),
|
||||
)(mock_request)
|
||||
|
||||
|
||||
async def test_query_token_auth_valid(mock_request: Mock) -> None:
|
||||
"""Test authentication with a valid query token."""
|
||||
mock_request.get = Mock(return_value=False)
|
||||
mock_request.query = {"token": "valid-token"}
|
||||
handler = AsyncMock(return_value=None)
|
||||
|
||||
response = await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(
|
||||
requires_auth=False,
|
||||
use_query_token_for_auth=True,
|
||||
get_valid_auth_tokens=Mock(return_value={"valid-token"}),
|
||||
),
|
||||
handler,
|
||||
)(mock_request)
|
||||
|
||||
assert response.status == HTTPStatus.OK
|
||||
handler.assert_awaited_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"query",
|
||||
[{"token": "wrong-token"}, {}],
|
||||
ids=["invalid_token", "missing_token"],
|
||||
)
|
||||
async def test_query_token_auth_unauthorized(
|
||||
mock_request: Mock, query: dict[str, str]
|
||||
) -> None:
|
||||
"""Test an invalid or missing query token is rejected."""
|
||||
mock_request.get = Mock(return_value=False)
|
||||
mock_request.query = query
|
||||
handler = AsyncMock()
|
||||
|
||||
with (
|
||||
patch(
|
||||
"homeassistant.helpers.network.get_url",
|
||||
return_value="https://example.com",
|
||||
),
|
||||
pytest.raises(HTTPUnauthorized),
|
||||
):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(
|
||||
requires_auth=False,
|
||||
use_query_token_for_auth=True,
|
||||
get_valid_auth_tokens=Mock(return_value={"valid-token"}),
|
||||
),
|
||||
handler,
|
||||
)(mock_request)
|
||||
|
||||
handler.assert_not_awaited()
|
||||
|
||||
|
||||
async def test_requires_auth_includes_www_authenticate(
|
||||
mock_request: Mock,
|
||||
) -> None:
|
||||
@@ -120,7 +173,7 @@ async def test_requires_auth_includes_www_authenticate(
|
||||
):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=True),
|
||||
Mock(requires_auth=True, use_query_token_for_auth=False),
|
||||
AsyncMock(),
|
||||
)(mock_request)
|
||||
assert exc_info.value.headers["WWW-Authenticate"] == (
|
||||
@@ -143,7 +196,7 @@ async def test_requires_auth_omits_www_authenticate_without_url(
|
||||
):
|
||||
await request_handler_factory(
|
||||
mock_request.app[KEY_HASS],
|
||||
Mock(requires_auth=True),
|
||||
Mock(requires_auth=True, use_query_token_for_auth=False),
|
||||
AsyncMock(),
|
||||
)(mock_request)
|
||||
assert "WWW-Authenticate" not in exc_info.value.headers
|
||||
@@ -212,7 +265,7 @@ async def test_requires_auth_www_authenticate_prefer_external(
|
||||
with pytest.raises(HTTPUnauthorized) as exc_info:
|
||||
await request_handler_factory(
|
||||
hass,
|
||||
Mock(requires_auth=True),
|
||||
Mock(requires_auth=True, use_query_token_for_auth=False),
|
||||
AsyncMock(),
|
||||
)(mock_current_request)
|
||||
|
||||
|
||||
@@ -234,24 +234,30 @@ async def test_fetch_image_unauthenticated(
|
||||
client = await hass_client_no_auth()
|
||||
|
||||
resp = await client.get("/api/image_proxy/image.test")
|
||||
assert resp.status == HTTPStatus.FORBIDDEN
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
resp = await client.get("/api/image_proxy/image.test")
|
||||
assert resp.status == HTTPStatus.FORBIDDEN
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
resp = await client.get(
|
||||
"/api/image_proxy/image.test", headers={hdrs.AUTHORIZATION: "blabla"}
|
||||
)
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
# An invalid token is also unauthorized
|
||||
resp = await client.get("/api/image_proxy/image.test?token=invalid")
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
state = hass.states.get("image.test")
|
||||
resp = await client.get(state.attributes["entity_picture"])
|
||||
assert resp.status == HTTPStatus.OK
|
||||
body = await resp.read()
|
||||
assert body == b"Test"
|
||||
|
||||
# Unknown entities are also unauthorized for an unauthenticated client, so
|
||||
# their existence is not leaked
|
||||
resp = await client.get("/api/image_proxy/image.unknown")
|
||||
assert resp.status == HTTPStatus.NOT_FOUND
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
|
||||
@respx.mock
|
||||
|
||||
@@ -113,6 +113,28 @@ async def test_get_image_http(
|
||||
assert content == b"image"
|
||||
|
||||
|
||||
async def test_get_image_http_unauthenticated(
|
||||
hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
|
||||
) -> None:
|
||||
"""Test get image via http command without a valid token is unauthorized."""
|
||||
await async_setup_component(
|
||||
hass, "media_player", {"media_player": {"platform": "demo"}}
|
||||
)
|
||||
await hass.async_block_till_done()
|
||||
|
||||
client = await hass_client_no_auth()
|
||||
|
||||
# Without a token the request is unauthorized
|
||||
resp = await client.get("/api/media_player_proxy/media_player.bedroom")
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
# An invalid token is also unauthorized
|
||||
resp = await client.get(
|
||||
"/api/media_player_proxy/media_player.bedroom?token=invalid"
|
||||
)
|
||||
assert resp.status == HTTPStatus.UNAUTHORIZED
|
||||
|
||||
|
||||
async def test_get_image_http_remote(
|
||||
hass: HomeAssistant, hass_client_no_auth: ClientSessionGenerator
|
||||
) -> None:
|
||||
|
||||
Reference in New Issue
Block a user