fix: allow EOSdash through an external reverse proxy (#1355)

Support externally proxied EOSdash (#1320)

* test: verify optional dashboard port stays unpublished

* docs: describe optional proxy access and public URL

src/akkudoktoreos/server/eos.py

@@ -10,6 +10,7 @@ import traceback
 from contextlib import asynccontextmanager
 from enum import Enum
 from typing import Annotated, Any, AsyncGenerator, Dict, List, Optional, Union
+from urllib.parse import urlsplit
 
 import psutil
 import uvicorn
@@ -2376,38 +2377,96 @@ def _sanitize_redirect_path(path: str) -> Optional[str]:
     return "/".join(parts)
 
 
+def _trusted_request_hosts() -> set[str]:
+    """Host names that may be taken from a request to address EOSdash.
+
+    The `Host` header is sent by the client and is not trusted. Reflecting it into an
+    absolute redirect would turn every EOS server into an open redirect. Only hosts that
+    the configuration already knows are accepted. Deployments behind a reverse proxy
+    announce their public address by `server.eosdash_public_url` instead.
+
+    Returns:
+        set[str]: Lower case host names, without brackets around IPv6 addresses.
+    """
+    settings = get_config().server
+    hosts = {"localhost", "127.0.0.1", "::1"}
+    for host in (settings.host, settings.eosdash_host):
+        if host and str(host) not in ("0.0.0.0", "::"):  # noqa: S104
+            hosts.add(str(host).lower())
+    hosts.add(get_host_ip())
+    return hosts
+
+
+def _eosdash_base_url(request: Request) -> Optional[str]:
+    """Return the configured public dashboard URL or a direct-access URL.
+
+    A proxy's public dashboard route cannot be inferred from its EOS API route.
+    Configure eosdash_public_url for TLS termination, port mappings or prefixes.
+
+    Args:
+        request: The request to take the host from for direct access.
+
+    Returns:
+        Optional[str]: Base URL of EOSdash without trailing slash. `None` if the request
+        host is not a trusted host and no public URL is configured.
+    """
+    settings = get_config().server
+    if settings.eosdash_public_url:
+        return settings.eosdash_public_url.rstrip("/")
+    port = settings.eosdash_port or 8504
+    scheme = request.url.scheme
+    if scheme not in ("http", "https"):
+        scheme = "http"
+    # Request.url honours the Host header and parses bracketed IPv6 correctly.
+    # Proxy scheme handling belongs to the ASGI server's trusted proxy middleware;
+    # do not trust arbitrary raw X-Forwarded-* headers here.
+    host = urlsplit(str(request.url)).hostname or ""
+    if not host or host in ("0.0.0.0", "::"):  # noqa: S104
+        host = str(settings.eosdash_host or settings.host)
+        if host in ("0.0.0.0", "::"):  # noqa: S104
+            host = get_host_ip()
+    if host.lower() not in _trusted_request_hosts():
+        return None
+    if ":" in host:
+        host = f"[{host}]"
+    return f"{scheme}://{host}:{port}"
+
+
+def _eosdash_unknown_page(request: Request, status_code: int) -> HTMLResponse:
+    """Error page for a request that can not be answered with an EOSdash address."""
+    error_page = create_error_page(
+        status_code=str(status_code),
+        error_title="EOSdash Address Unknown",
+        error_message=(
+            f"URL is unknown: '{request.url}'. EOSdash can not be addressed for host "
+            f"'{request.url.netloc}'. Set 'server.eosdash_public_url' to the public "
+            "address of EOSdash."
+        ),
+        error_details="Untrusted request host",
+    )
+    return HTMLResponse(content=error_page, status_code=status_code)
+
+
 def redirect(request: Request, path: str) -> Union[HTMLResponse, RedirectResponse]:
+    base_url = _eosdash_base_url(request)
+
     # Path is not for EOSdash
     if not (path.startswith("eosdash") or path == ""):
-        host = get_config().server.eosdash_host
-        if host is None:
-            host = get_config().server.host
-        host = str(host)
-        port = get_config().server.eosdash_port
-        if port is None:
-            port = 8504
-        if host == "0.0.0.0":  # noqa: S104
-            # Use IP of EOS host
-            host = get_host_ip()
-        url = f"http://{host}:{port}/"
+        if base_url is None:
+            return _eosdash_unknown_page(request, 404)
         error_page = create_error_page(
             status_code="404",
             error_title="Page Not Found",
-            error_message=f"""<pre>
-URL is unknown: '{request.url}'
-Did you want to connect to <a href="{url}" class="back-button">EOSdash</a>?
-</pre>
-""",
+            error_message=f"URL is unknown: '{request.url}'. Did you want to connect to EOSdash?",
             error_details="Unknown URL",
+            link_url=f"{base_url}/",
+            link_label="Open EOSdash",
         )
         return HTMLResponse(content=error_page, status_code=404)
 
-    host = str(get_config().server.eosdash_host)
-    if host == "0.0.0.0":  # noqa: S104
-        # Use IP of EOS host
-        host = get_host_ip()
-    if host and get_config().server.eosdash_port:
-        base_url = f"http://{host}:{get_config().server.eosdash_port}"
+    if get_config().server.eosdash_port:
+        if base_url is None:
+            return _eosdash_unknown_page(request, 404)
         safe_path = _sanitize_redirect_path(path) or ""
         url = f"{base_url}/{safe_path}"
         return RedirectResponse(url=url, status_code=303)

src/akkudoktoreos/server/rest/error.py

@@ -1,7 +1,7 @@
 import html
 import traceback
 from dataclasses import dataclass
-from typing import cast
+from typing import Optional, cast
 
 from fastapi import FastAPI, Request
 from fastapi.exceptions import HTTPException, RequestValidationError
@@ -186,6 +186,7 @@ ERROR_PAGE_TEMPLATE = """
         <h2 class="error-title">ERROR_TITLE</h2>
         <p class="error-message">ERROR_MESSAGE</p>
         <div class="error-details">ERROR_DETAILS</div>
+        ERROR_LINK
         <a href="/docs" class="back-button">Back to Home</a>
     </div>
 </body>
@@ -194,11 +195,39 @@ ERROR_PAGE_TEMPLATE = """
 
 
 def create_error_page(
-    status_code: str, error_title: str, error_message: str, error_details: str
+    status_code: str,
+    error_title: str,
+    error_message: str,
+    error_details: str,
+    link_url: Optional[str] = None,
+    link_label: str = "Open link",
 ) -> str:
-    """Create an error page by replacing placeholders in the template."""
+    """Create an error page by replacing placeholders in the template.
+
+    The message and the details are escaped, so they are always shown as text. Markup in
+    them is not rendered. Use `link_url` to offer a link on the page.
+
+    Args:
+        status_code: The HTTP status code to display.
+        error_title: The title of the error.
+        error_message: The error message, shown as text.
+        error_details: The error details, shown as text.
+        link_url: Target of an extra link on the page. No link is added for `None`.
+        link_label: The label of the extra link.
+
+    Returns:
+        str: The error page as HTML.
+    """
+    link = ""
+    if link_url:
+        link = (
+            f'<a href="{html.escape(link_url, quote=True)}" class="back-button">'
+            f"{html.escape(link_label)}</a>"
+        )
+    # Insert the link first, so escaped text of the other placeholders is never replaced.
     return (
-        ERROR_PAGE_TEMPLATE.replace("STATUS_CODE", status_code)
+        ERROR_PAGE_TEMPLATE.replace("ERROR_LINK", link)
+        .replace("STATUS_CODE", status_code)
         .replace("ERROR_TITLE", error_title)
         .replace("ERROR_MESSAGE", html.escape(error_message))
         .replace("ERROR_DETAILS", html.escape(error_details))

src/akkudoktoreos/server/server.py

@@ -8,6 +8,7 @@ import socket
 import sys
 import time
 from typing import Any, Optional
+from urllib.parse import urlsplit
 
 try:
     # Only available on Linux/Unix type systems
@@ -429,6 +430,18 @@ class ServerCommonSettings(SettingsBaseModel):
             ],
         },
     )
+    eosdash_public_url: Optional[str] = Field(
+        default=None,
+        json_schema_extra={
+            "description": (
+                "Public EOSdash base URL for redirects and error-page links, including an "
+                "optional proxy path prefix. Set this for reverse proxies or mapped ports; "
+                "it does not change the bind address. Without it, direct access uses the "
+                "request host and EOSdash port. Raw forwarded headers are not used."
+            ),
+            "examples": ["https://energy.example.com/dashboard"],
+        },
+    )
     eosdash_supervise_interval_sec: int = Field(
         default=10,
         json_schema_extra={
@@ -464,6 +477,31 @@ class ServerCommonSettings(SettingsBaseModel):
         },
     )
 
+    @field_validator("eosdash_public_url")
+    @classmethod
+    def validate_eosdash_public_url(cls, value: Optional[str]) -> Optional[str]:
+        """Require an absolute HTTP(S) base URL without credentials or query data."""
+        if value is None:
+            return None
+        parsed = urlsplit(value)
+        if (
+            parsed.scheme not in ("http", "https")
+            or not parsed.hostname
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.query
+            or parsed.fragment
+            or "?" in value
+            or "#" in value
+            or "\\" in value
+            or any(character.isspace() or ord(character) < 32 for character in value)
+        ):
+            raise ValueError("EOSdash public URL must be an absolute HTTP(S) base URL.")
+        # Accessing port also validates its numeric range.
+        if parsed.port == 0:
+            raise ValueError("EOSdash public URL port must be positive.")
+        return value.rstrip("/")
+
     @field_validator("host", "eosdash_host", mode="before")
     def validate_server_host(cls, value: Optional[str]) -> Optional[str]:
         if isinstance(value, str):