Fix return type hints for middleware methods (#6388)

* Fix return type hints for middleware methods

Adjust type hints in SecurityMiddleware to use StreamResponse instead
of Response. This correctly reflects that middleware handlers can return
any StreamResponse subclass, including FileResponse and other streaming
responses.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

* Improve type annotations in SecurityMiddleware

Add proper type parameters to improve type safety:
- Use Callable[[Request], Awaitable[StreamResponse]] for middleware
  handlers instead of bare Callable
- Add type parameter to re.Pattern[str] for ADDONS_ROLE_ACCESS

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

---------

Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
Stefan Agner 2025-12-03 21:51:56 +01:00 committed by GitHub
parent 81b7e54b18
commit 7895bc9007
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194

View File

@ -1,12 +1,12 @@
"""Handle security part of this API."""
from collections.abc import Callable
from collections.abc import Awaitable, Callable
import logging
import re
from typing import Final
from urllib.parse import unquote
from aiohttp.web import Request, Response, middleware
from aiohttp.web import Request, StreamResponse, middleware
from aiohttp.web_exceptions import HTTPBadRequest, HTTPForbidden, HTTPUnauthorized
from awesomeversion import AwesomeVersion
@ -89,7 +89,7 @@ CORE_ONLY_PATHS: Final = re.compile(
)
# Policy role add-on API access
ADDONS_ROLE_ACCESS: dict[str, re.Pattern] = {
ADDONS_ROLE_ACCESS: dict[str, re.Pattern[str]] = {
ROLE_DEFAULT: re.compile(
r"^(?:"
r"|/.+/info"
@ -180,7 +180,9 @@ class SecurityMiddleware(CoreSysAttributes):
return unquoted
@middleware
async def block_bad_requests(self, request: Request, handler: Callable) -> Response:
async def block_bad_requests(
self, request: Request, handler: Callable[[Request], Awaitable[StreamResponse]]
) -> StreamResponse:
"""Process request and tblock commonly known exploit attempts."""
if FILTERS.search(self._recursive_unquote(request.path)):
_LOGGER.warning(
@ -198,7 +200,9 @@ class SecurityMiddleware(CoreSysAttributes):
return await handler(request)
@middleware
async def system_validation(self, request: Request, handler: Callable) -> Response:
async def system_validation(
self, request: Request, handler: Callable[[Request], Awaitable[StreamResponse]]
) -> StreamResponse:
"""Check if core is ready to response."""
if self.sys_core.state not in VALID_API_STATES:
return api_return_error(
@ -208,7 +212,9 @@ class SecurityMiddleware(CoreSysAttributes):
return await handler(request)
@middleware
async def token_validation(self, request: Request, handler: Callable) -> Response:
async def token_validation(
self, request: Request, handler: Callable[[Request], Awaitable[StreamResponse]]
) -> StreamResponse:
"""Check security access of this layer."""
request_from: CoreSysAttributes | None = None
supervisor_token = extract_supervisor_token(request)
@ -279,7 +285,9 @@ class SecurityMiddleware(CoreSysAttributes):
raise HTTPForbidden()
@middleware
async def core_proxy(self, request: Request, handler: Callable) -> Response:
async def core_proxy(
self, request: Request, handler: Callable[[Request], Awaitable[StreamResponse]]
) -> StreamResponse:
"""Validate user from Core API proxy."""
if (
request[REQUEST_FROM] != self.sys_homeassistant