diff --git a/supervisor/api/middleware/security.py b/supervisor/api/middleware/security.py index 6cbb2eecd..9c44022af 100644 --- a/supervisor/api/middleware/security.py +++ b/supervisor/api/middleware/security.py @@ -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