/
/
/
1"""Base Webserver logic for an HTTPServer that can handle dynamic routes."""
2
3from __future__ import annotations
4
5from collections.abc import Callable, Coroutine, Mapping
6from typing import TYPE_CHECKING, Any, Final
7
8from aiohttp import web
9
10from music_assistant.constants import WILDCARD_BIND_IPS
11
12if TYPE_CHECKING:
13 import logging
14
15 from aiohttp.typedefs import Handler
16
17
18MAX_CLIENT_SIZE: Final = 1024**2 * 16
19MAX_LINE_SIZE: Final = 24570
20
21# grace period handlers get to finish before they are cancelled on shutdown
22DEFAULT_SHUTDOWN_TIMEOUT: Final = 10
23
24# Type alias for dynamic route handlers
25DynamicRouteHandler = Callable[
26 [web.Request], Coroutine[Any, Any, web.Response | web.StreamResponse]
27]
28
29
30REDACTED_HEADER_VALUE: Final = "<redacted>"
31
32
33def redact_sensitive_headers(headers: Mapping[str, str]) -> dict[str, str]:
34 """
35 Return request headers with credential-bearing values redacted.
36
37 :param headers: Headers to prepare for logging.
38 """
39 return {
40 key: REDACTED_HEADER_VALUE
41 if key.lower().startswith(("authorization", "proxy-authorization"))
42 else value
43 for key, value in headers.items()
44 }
45
46
47class Webserver:
48 """Base Webserver logic for an HTTPServer that can handle dynamic routes."""
49
50 def __init__(
51 self,
52 logger: logging.Logger,
53 enable_dynamic_routes: bool = False,
54 ) -> None:
55 """Initialize instance."""
56 self.logger = logger
57 # the below gets initialized in async setup
58 self._apprunner: web.AppRunner | None = None
59 self._webapp: web.Application | None = None
60 self._tcp_site: web.TCPSite | None = None
61 self._static_routes: list[tuple[str, str, Handler]] | None = None
62 self._dynamic_routes: dict[str, DynamicRouteHandler] | None = (
63 {} if enable_dynamic_routes else None
64 )
65 self._bind_port: int | None = None
66 self._bind_ip: str | None = None
67 self._ingress_tcp_site: web.TCPSite | None = None
68
69 async def setup(
70 self,
71 bind_ip: str | None,
72 bind_port: int,
73 static_routes: list[tuple[str, str, Handler]] | None = None,
74 static_content: tuple[str, str, str] | None = None,
75 ingress_tcp_site_params: tuple[str, int] | None = None,
76 app_state: dict[str, Any] | None = None,
77 ssl_context: Any | None = None,
78 ) -> None:
79 """
80 Async initialize of module.
81
82 :param bind_ip: IP address to bind to. An unavailable address falls back to all
83 interfaces. The effective address is available as the ``bind_ip`` property.
84 :param bind_port: Port to bind to, or 0 to let the OS assign a free one, which
85 requires a specific ``bind_ip``. The assigned port is available as the
86 ``port`` property.
87 :param static_routes: List of static routes to register.
88 :param static_content: Tuple of (path, directory, name) for static content.
89 :param ingress_tcp_site_params: Tuple of (host, port) for ingress TCP site.
90 :param app_state: Optional dict of key-value pairs to set on app before starting.
91 :param ssl_context: Optional SSL context for HTTPS support.
92 """
93 self._bind_port = bind_port
94 self._static_routes = static_routes
95 self._webapp = web.Application(
96 logger=self.logger,
97 client_max_size=MAX_CLIENT_SIZE,
98 handler_args={
99 "max_line_size": MAX_LINE_SIZE,
100 "max_field_size": MAX_LINE_SIZE,
101 },
102 )
103 # Set app state before starting
104 if app_state:
105 for key, value in app_state.items():
106 self._webapp[key] = value
107 self._apprunner = web.AppRunner(
108 self._webapp, access_log=None, shutdown_timeout=DEFAULT_SHUTDOWN_TIMEOUT
109 )
110 # add static routes
111 if self._static_routes:
112 for method, path, handler in self._static_routes:
113 self._webapp.router.add_route(method, path, handler)
114 if static_content:
115 self._webapp.router.add_static(
116 static_content[0], static_content[1], name=static_content[2]
117 )
118 # register catch-all route to handle dynamic routes (if enabled)
119 if self._dynamic_routes is not None:
120 self._webapp.router.add_route("*", "/{tail:.*}", self._handle_catch_all)
121 await self._apprunner.setup()
122 # set host to None to bind to all addresses on both IPv4 and IPv6
123 host = None if bind_ip in WILDCARD_BIND_IPS else bind_ip
124 if bind_port == 0 and host is None:
125 # a wildcard bind gets one socket per address family, each with its own
126 # OS-assigned port, so there is no single port to publish
127 msg = "An OS-assigned port requires a specific bind address"
128 raise ValueError(msg)
129 try:
130 self._tcp_site = web.TCPSite(
131 self._apprunner, host=host, port=bind_port, ssl_context=ssl_context
132 )
133 await self._tcp_site.start()
134 except OSError:
135 if host is None:
136 raise
137 if bind_port == 0:
138 # binding all interfaces is no fallback for an OS-assigned port
139 raise
140 # the configured interface is not available, retry on all interfaces
141 self.logger.error(
142 "Could not bind to %s, will start on all interfaces as fallback!", host
143 )
144 host = None
145 self._tcp_site = web.TCPSite(
146 self._apprunner, host=host, port=bind_port, ssl_context=ssl_context
147 )
148 await self._tcp_site.start()
149 self._bind_ip = host
150 # port 0 asks the OS for a free port, which it only picks at bind time
151 if bind_port == 0:
152 self._bind_port = self._apprunner.addresses[0][1]
153 # start additional ingress TCP site if configured
154 # this is only used if we're running in the context of an HA add-on
155 # which proxies our frontend and api through ingress
156 if ingress_tcp_site_params:
157 # Store ingress site reference in app for security checks
158 self._webapp["ingress_site"] = ingress_tcp_site_params
159 self._ingress_tcp_site = web.TCPSite(
160 self._apprunner,
161 host=ingress_tcp_site_params[0],
162 port=ingress_tcp_site_params[1],
163 )
164 await self._ingress_tcp_site.start()
165
166 async def close(self) -> None:
167 """Cleanup on exit."""
168 # stop/clean webserver
169 if self._tcp_site:
170 await self._tcp_site.stop()
171 if self._ingress_tcp_site:
172 await self._ingress_tcp_site.stop()
173 if self._apprunner:
174 await self._apprunner.cleanup()
175 if self._webapp:
176 await self._webapp.shutdown()
177 await self._webapp.cleanup()
178
179 @property
180 def port(self) -> int | None:
181 """Return the port of this webserver."""
182 return self._bind_port
183
184 @property
185 def bind_ip(self) -> str | None:
186 """Return the IP address this webserver is bound to (None for all interfaces)."""
187 return self._bind_ip
188
189 def register_dynamic_route(
190 self,
191 path: str,
192 handler: Callable[[web.Request], Coroutine[Any, Any, web.Response | web.StreamResponse]],
193 method: str = "*",
194 ) -> Callable[[], None]:
195 """Register a dynamic route on the webserver, returns handler to unregister."""
196 if self._dynamic_routes is None:
197 msg = "Dynamic routes are not enabled"
198 raise RuntimeError(msg)
199 key = f"{method}.{path}"
200 if key in self._dynamic_routes:
201 msg = f"Route {path} already registered."
202 raise RuntimeError(msg)
203 self._dynamic_routes[key] = handler
204
205 def _remove() -> None:
206 assert self._dynamic_routes is not None # for type checking
207 self._dynamic_routes.pop(key, None)
208
209 return _remove
210
211 def unregister_dynamic_route(self, path: str, method: str = "*") -> None:
212 """Unregister a dynamic route from the webserver."""
213 if self._dynamic_routes is None:
214 msg = "Dynamic routes are not enabled"
215 raise RuntimeError(msg)
216 key = f"{method}.{path}"
217 self._dynamic_routes.pop(key, None)
218
219 async def serve_static(self, file_path: str, request: web.Request) -> web.FileResponse:
220 """Serve file response."""
221 headers = {"Cache-Control": "no-cache"}
222 return web.FileResponse(file_path, headers=headers)
223
224 async def _handle_catch_all(self, request: web.Request) -> web.Response | web.StreamResponse:
225 """Redirect request to correct destination."""
226 # find handler for the request
227 # Try exact match first
228 for key in (f"{request.method}.{request.path}", f"*.{request.path}"):
229 assert self._dynamic_routes is not None # for type checking
230 if handler := self._dynamic_routes.get(key):
231 return await handler(request)
232 # Try prefix match (for routes registered with /*)
233 if self._dynamic_routes is not None:
234 for route_key, handler in list(self._dynamic_routes.items()):
235 method, path = route_key.split(".", 1)
236 if method in (request.method, "*") and path.endswith("/*"):
237 prefix = path[:-2]
238 if request.path.startswith(prefix):
239 return await handler(request)
240 # deny all other requests
241 self.logger.warning(
242 "Received unhandled %s request to %s from %s\nheaders: %s\n",
243 request.method,
244 request.path,
245 request.remote,
246 redact_sensitive_headers(request.headers),
247 )
248 return web.Response(status=404)
249