Add URL Display Theme for Raspberry Pi

.github/workflows/check_display_theme_url_display.yml

@@ -0,0 +1,34 @@
+name: Check Display Theme URL Display
+
+on:
+  pull_request:
+    paths:
+      - packages/modules/display_themes/url_display/source/**
+    branches:
+      - master
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    strategy:
+      matrix:
+        node: [ 24 ]
+        # See supported Node.js release schedule at https://nodejs.org/en/about/releases/
+
+    steps:
+      - name: Checkout Repository
+        uses: actions/checkout@v4
+
+      - name: Setup Node.js ${{ matrix.node }}
+        uses: actions/setup-node@v4
+        with:
+          node-version: ${{ matrix.node }}
+          cache: 'npm'
+          cache-dependency-path: packages/modules/display_themes/url_display/source/package-lock.json
+
+      - name: Install Dependencies and Build
+        run: |
+          cd packages/modules/display_themes/url_display/source
+          npm install
+          npm run build --if-present

packages/helpermodules/setdata.py

@@ -13,10 +13,12 @@
 from control import data
 from helpermodules import hardware_configuration, subdata
 from helpermodules.broker import BrokerClient
+from helpermodules.messaging import MessageType, pub_system_message
 from helpermodules.pub import Pub
 from helpermodules.utils.topic_parser import decode_payload, get_index, get_index_position
 from helpermodules.update_config import UpdateConfig
 import dataclass_utils
+from modules.display_themes import deserialize_display_theme
 
 log = logging.getLogger(__name__)
 mqtt_log = logging.getLogger("mqtt")
@@ -905,7 +907,14 @@ def process_optional_topic(self, msg: mqtt.MQTTMessage):
             elif "openWB/set/optional/int_display/standby" in msg.topic:
                 self._validate_value(msg, int, [(0, 600)])
             elif "openWB/set/optional/int_display/theme" in msg.topic:
-                self._validate_value(msg, "json")
+                try:
+                    theme = deserialize_display_theme(decode_payload(msg.payload))
+                    Pub().pub(msg.topic.replace('set/', '', 1), dataclass_utils.asdict(theme))
+                    Pub().pub(msg.topic, "")
+                except ValueError as exc:
+                    log.warning("Ungültige Display-Theme-Konfiguration: %s", exc)
+                    pub_system_message({}, str(exc), MessageType.ERROR)
+                    Pub().pub(msg.topic, "")
             elif "openWB/set/optional/led/active" in msg.topic:
                 self._validate_value(msg, bool)
             else:

packages/helpermodules/subdata.py

@@ -36,6 +36,7 @@
 from modules.common.component_type import ComponentType
 from modules.common.configurable_backup_cloud import ConfigurableBackupCloud
 from modules.common.configurable_tariff import ConfigurableFlexibleTariff, ConfigurableGridFee
+from modules.display_themes import deserialize_display_theme
 from modules.common.simcount.simcounter_state import SimCounterState
 from modules.internal_chargepoint_handler.internal_chargepoint_handler_config import (
     GlobalHandlerData, InternalChargepoint, RfidData)
@@ -758,7 +759,21 @@ def process_optional_topic(self, var: optional.Optional, msg: mqtt.MQTTMessage):
                 elif re.search("/optional/ocpp/", msg.topic) is not None:
                     self.set_json_payload_class(var.data.ocpp, msg)
                 elif re.search("/optional/int_display/", msg.topic) is not None:
-                    self.set_json_payload_class(var.data.int_display, msg)
+                    if msg.topic == "openWB/optional/int_display/theme":
+                        try:
+                            var.data.int_display.theme = deserialize_display_theme(decode_payload(msg.payload))
+                        except ValueError as exc:
+                            log.warning("Ungültige Display-Theme-Konfiguration: %s", exc)
+                            theme = decode_payload(msg.payload)
+                            if isinstance(theme, dict) and theme.get("type") == "url_display":
+                                if not isinstance(theme.get("configuration"), dict):
+                                    theme["configuration"] = {}
+                                theme["configuration"]["url"] = ""
+                                var.data.int_display.theme = deserialize_display_theme(theme)
+                                Pub().pub(msg.topic, asdict(var.data.int_display.theme))
+                            pub_system_message({}, str(exc), MessageType.ERROR)
+                    else:
+                        self.set_json_payload_class(var.data.int_display, msg)
                     if re.search("/(standby|active|rotation)$", msg.topic) is not None:
                         # some topics require an update of the display manager or boot settings
                         run_command([

packages/modules/display_themes/__init__.py

@@ -0,0 +1,29 @@
+import importlib
+from typing import Any
+
+from dataclass_utils import dataclass_from_dict
+
+
+def deserialize_display_theme(config: dict[str, Any]) -> Any:
+    if not isinstance(config, dict):
+        raise ValueError("Die Display-Theme-Konfiguration muss ein JSON-Objekt sein.")
+
+    theme_type = config.get("type")
+    if not isinstance(theme_type, str) or not theme_type.isidentifier():
+        raise ValueError("Der Typ des Display-Themes fehlt.")
+    if "configuration" in config and not isinstance(config["configuration"], dict):
+        raise ValueError("Die Konfiguration des Display-Themes muss ein JSON-Objekt sein.")
+
+    module_name = f"modules.display_themes.{theme_type}.config"
+    try:
+        module = importlib.import_module(f".{theme_type}.config", "modules.display_themes")
+    except ModuleNotFoundError as exc:
+        if exc.name != module_name:
+            raise
+        raise ValueError(f"Unbekanntes Display-Theme: {theme_type}") from exc
+
+    theme = dataclass_from_dict(module.theme_descriptor.configuration_factory, config)
+    default_theme = module.theme_descriptor.configuration_factory()
+    if not isinstance(theme.configuration, type(default_theme.configuration)):
+        raise ValueError("Die Konfiguration des Display-Themes hat einen ungültigen Typ.")
+    return theme

packages/modules/display_themes/url_display/config.py

@@ -0,0 +1,119 @@
+import ipaddress
+import json
+import socket
+import subprocess
+from typing import Optional
+from urllib.parse import urlsplit, urlunsplit
+
+from helpermodules.auto_str import auto_str
+from helpermodules.utils.run_command import run_command
+
+from modules.common.abstract_device import DeviceDescriptor
+
+
+RFC1918_NETWORKS = (
+    ipaddress.ip_network("10.0.0.0/8"),
+    ipaddress.ip_network("172.16.0.0/12"),
+    ipaddress.ip_network("192.168.0.0/16"),
+)
+
+
+def _get_local_ipv4_networks() -> tuple[ipaddress.IPv4Network, ...]:
+    try:
+        output = run_command(["ip", "-j", "-4", "address", "show", "up"])
+        if output is None:
+            raise ValueError("Der Aufruf von 'ip' hat keine Ausgabe geliefert.")
+        interfaces = json.loads(output)
+        return tuple(
+            ipaddress.IPv4Network(f"{address['local']}/{address['prefixlen']}", strict=False)
+            for interface in interfaces
+            for address in interface.get("addr_info", [])
+            if address.get("family") == "inet"
+        )
+    except (KeyError, TypeError, ValueError, subprocess.CalledProcessError) as exc:
+        raise ValueError("Die lokalen IPv4-Netzwerke konnten nicht ermittelt werden.") from exc
+
+
+def validate_url(url: str) -> str:
+    if not isinstance(url, str) or not url.strip():
+        raise ValueError("Bitte eine URL angeben.")
+
+    normalized_url = url.strip()
+    if "://" not in normalized_url:
+        normalized_url = f"http://{normalized_url}"
+
+    parsed_url = urlsplit(normalized_url)
+    if parsed_url.scheme.lower() not in ("http", "https"):
+        raise ValueError("Die URL muss das Protokoll HTTP oder HTTPS verwenden.")
+    if parsed_url.hostname is None:
+        raise ValueError("Die URL enthält keinen gültigen Hostnamen.")
+    if parsed_url.username is not None or parsed_url.password is not None:
+        raise ValueError("Die URL darf keine Zugangsdaten enthalten.")
+    try:
+        parsed_url.port
+    except ValueError as exc:
+        raise ValueError("Die URL enthält keinen gültigen Port.") from exc
+
+    try:
+        resolved_addresses = {
+            ipaddress.IPv4Address(address[4][0])
+            for address in socket.getaddrinfo(
+                parsed_url.hostname,
+                parsed_url.port,
+                family=socket.AF_INET,
+                type=socket.SOCK_STREAM,
+            )
+        }
+    except socket.gaierror as exc:
+        raise ValueError(f"Der Hostname '{parsed_url.hostname}' konnte nicht aufgelöst werden.") from exc
+
+    if not resolved_addresses:
+        raise ValueError(f"Für den Hostnamen '{parsed_url.hostname}' wurde keine IPv4-Adresse gefunden.")
+
+    addresses_outside_rfc1918 = [
+        address
+        for address in resolved_addresses
+        if not any(address in network for network in RFC1918_NETWORKS)
+    ]
+    if not addresses_outside_rfc1918:
+        return urlunsplit(parsed_url._replace(scheme=parsed_url.scheme.lower()))
+
+    local_networks = _get_local_ipv4_networks()
+    invalid_addresses = [
+        str(address)
+        for address in addresses_outside_rfc1918
+        if not any(address in network for network in local_networks)
+    ]
+    if invalid_addresses:
+        raise ValueError(
+            "Die URL muss auf eine IPv4-Adresse im lokalen Netzwerk zeigen. "
+            f"Nicht zulässig: {', '.join(sorted(invalid_addresses))}"
+        )
+
+    return urlunsplit(parsed_url._replace(scheme=parsed_url.scheme.lower()))
+
+
+@auto_str
+class UrlDisplayThemeConfiguration:
+    def __init__(self,
+                 url: str = ""
+                 ) -> None:
+        self.url = validate_url(url) if url else ""
+
+
+@auto_str
+class UrlDisplayTheme:
+    def __init__(self,
+                 name: str = "URL Display",
+                 type: str = "url_display",
+                 official: bool = False,
+                 userManagementSupported: bool = False,
+                 configuration: Optional[UrlDisplayThemeConfiguration] = None) -> None:
+        self.name = name
+        self.type = type
+        self.official = official
+        self.userManagementSupported = userManagementSupported
+        self.configuration = configuration or UrlDisplayThemeConfiguration()
+
+
+theme_descriptor = DeviceDescriptor(configuration_factory=UrlDisplayTheme)

packages/modules/display_themes/url_display/config_test.py

@@ -0,0 +1,135 @@
+import socket
+
+import pytest
+
+from modules.display_themes import deserialize_display_theme
+from modules.display_themes.url_display import config
+
+
+@pytest.fixture(autouse=True)
+def local_networks(monkeypatch):
+    def run_command(command):
+        assert command == ["ip", "-j", "-4", "address", "show", "up"]
+        return '[{"addr_info": [{"family": "inet", "local": "100.64.1.10", "prefixlen": 24}]}]'
+
+    monkeypatch.setattr(
+        config,
+        "run_command",
+        run_command,
+    )
+
+
+def mock_dns(monkeypatch, *addresses):
+    def getaddrinfo(host, port, *, family, type):
+        assert host
+        assert port is None or isinstance(port, int)
+        assert family == socket.AF_INET
+        assert type == socket.SOCK_STREAM
+        return [
+            (socket.AF_INET, socket.SOCK_STREAM, 6, "", (address, 0))
+            for address in addresses
+        ]
+
+    monkeypatch.setattr(
+        config.socket,
+        "getaddrinfo",
+        getaddrinfo,
+    )
+
+
+@pytest.mark.parametrize("address", [
+    "10.0.0.1",
+    "172.16.0.1",
+    "172.31.255.254",
+    "192.168.1.1",
+])
+def test_rfc1918_address_is_allowed(monkeypatch, address):
+    mock_dns(monkeypatch, address)
+
+    assert config.validate_url("https://example.local:8443/path?value=1") == (
+        "https://example.local:8443/path?value=1"
+    )
+
+
+def test_rfc1918_validation_does_not_require_interface_lookup(monkeypatch):
+    mock_dns(monkeypatch, "192.168.1.10")
+    monkeypatch.setattr(
+        config,
+        "run_command",
+        lambda command: pytest.fail(
+            f"Interface lookup should not be called for RFC1918 addresses: {command}"
+        ),
+    )
+
+    assert config.validate_url("openwb.local") == "http://openwb.local"
+
+
+def test_address_in_local_subnet_is_allowed(monkeypatch):
+    mock_dns(monkeypatch, "100.64.1.25")
+
+    assert config.validate_url("evcc.local/status") == "http://evcc.local/status"
+
+
+@pytest.mark.parametrize("address", [
+    "8.8.8.8",
+    "100.64.2.25",
+    "172.15.255.255",
+    "172.32.0.1",
+])
+def test_non_local_address_is_rejected(monkeypatch, address):
+    mock_dns(monkeypatch, address)
+
+    with pytest.raises(ValueError, match="lokalen Netzwerk"):
+        config.validate_url("https://example.com")
+
+
+def test_all_resolved_addresses_must_be_local(monkeypatch):
+    mock_dns(monkeypatch, "192.168.1.10", "8.8.8.8")
+
+    with pytest.raises(ValueError, match="8.8.8.8"):
+        config.validate_url("https://example.local")
+
+
+@pytest.mark.parametrize("url", [
+    "",
+    "ftp://192.168.1.10",
+    "http://user:password@192.168.1.10",
+    "http://192.168.1.10:invalid",
+])
+def test_invalid_url_is_rejected(monkeypatch, url):
+    mock_dns(monkeypatch, "192.168.1.10")
+
+    with pytest.raises(ValueError):
+        config.validate_url(url)
+
+
+def test_unresolvable_hostname_is_rejected(monkeypatch):
+    def raise_gaierror(*args, **kwargs):
+        assert args or kwargs
+        raise socket.gaierror
+
+    monkeypatch.setattr(config.socket, "getaddrinfo", raise_gaierror)
+
+    with pytest.raises(ValueError, match="konnte nicht aufgelöst werden"):
+        config.validate_url("https://missing.local")
+
+
+@pytest.mark.parametrize("configuration", [None, "https://192.168.1.10", 1, []])
+def test_non_object_theme_configuration_is_rejected(configuration):
+    with pytest.raises(ValueError, match="JSON-Objekt"):
+        deserialize_display_theme({
+            "name": "URL Display",
+            "type": "url_display",
+            "configuration": configuration,
+        })
+
+
+@pytest.mark.parametrize("theme_type", ["cards", "colors", "url_display"])
+def test_display_theme_configuration_is_deserialized(theme_type):
+    theme = deserialize_display_theme({
+        "type": theme_type,
+        "configuration": {},
+    })
+
+    assert theme.type == theme_type
+    assert not isinstance(theme.configuration, dict)

packages/modules/display_themes/url_display/source/.gitignore

@@ -0,0 +1 @@
+node_modules/

packages/modules/display_themes/url_display/source/index.html

@@ -0,0 +1,14 @@
+<!doctype html>
+<html lang="de">
+  <head>
+    <meta charset="UTF-8" />
+    <meta name="viewport" content="width=device-width, initial-scale=1.0" />
+    <title>openWB Display - URL</title>
+  </head>
+  <body>
+    <div id="status">
+      <p id="message">Verbinde mit MQTT...</p>
+    </div>
+    <script type="module" src="/src/main.js"></script>
+  </body>
+</html>