From 12f0c7406f1a78cc4572b0eeae4f9a0d3c45cafb Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Thu, 8 Oct 2026 08:52:03 +0200 Subject: [PATCH 1/3] fis display user name generation --- .../mosquitto_dynsec/mosquitto_dynsec.py | 6 +++--- .../helpermodules/mosquitto_dynsec/user_handler.py | 12 +++++++----- 2 files changed, 10 insertions(+), 8 deletions(-) diff --git a/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py b/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py index f131d2e485..6fce4d85fb 100644 --- a/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py +++ b/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py @@ -98,15 +98,15 @@ def verify_password_reset_token(username: str, token: str) -> bool: def check_required_users(): - def create_user(ip_address: str): - success, user_name = create_display_user(ip_address, user_name="Display-Intern") + def create_user(ip_address: str = "127.0.0.1"): + success, user_name = create_display_user(ip_address) if success: log.info(f"Created user '{user_name}' for cp display at {ip_address}") else: log.error(f"Failed to create user for cp display at {ip_address}") # Always create user for localhost to ensure access to local displays, even if no chargepoints are configured - create_user("127.0.0.1") + create_user() # Create users for chargepoints of type 'external_openwb' for cp in SubData.cp_data.values(): cp_type = cp.chargepoint.data.config.type diff --git a/packages/helpermodules/mosquitto_dynsec/user_handler.py b/packages/helpermodules/mosquitto_dynsec/user_handler.py index be7305fd34..981ae6219f 100644 --- a/packages/helpermodules/mosquitto_dynsec/user_handler.py +++ b/packages/helpermodules/mosquitto_dynsec/user_handler.py @@ -1,6 +1,6 @@ import logging from secrets import token_hex -from typing import Optional, Tuple +from typing import Tuple from json import dump as json_dump from helpermodules.mosquitto_dynsec.role_handler import _get_base_path @@ -27,9 +27,8 @@ def add_user_to_group(username: str, groupname: str) -> None: run_command(["mosquitto_ctrl", "dynsec", "addGroupClient", groupname, username]) -def create_display_user(ip_address: str, user_name: Optional[str] = None) -> Tuple[bool, str]: - if user_name is None: - user_name = f"Display-{ip_address}" +def create_display_user(ip_address: str) -> Tuple[bool, str]: + user_name = "Display-Intern" if ip_address == "127.0.0.1" else f"Display-{ip_address}" file_name = f"display-{ip_address}".replace('.', '_') if user_exists(user_name): log.info(f"User '{user_name}' already exists") @@ -49,6 +48,9 @@ def create_display_user(ip_address: str, user_name: Optional[str] = None) -> Tup def remove_display_user(ip_address: str) -> bool: + if ip_address == "127.0.0.1": + log.error("Cannot remove the internal display user") + return False user_name = f"Display-{ip_address}" if remove_user(user_name): remove_user_credentials(user_name.lower().replace('.', '_')) @@ -60,7 +62,7 @@ def remove_display_user(ip_address: str) -> bool: def user_exists(username: str) -> bool: result = run_command(["mosquitto_ctrl", "dynsec", "getClient", username], process_exception=True) - if username in result: + if result is not None and username in result: return True return False From c33178d94acf316dff7248c238ccfc704eb036b3 Mon Sep 17 00:00:00 2001 From: benderl Date: Thu, 8 Oct 2026 11:18:44 +0200 Subject: [PATCH 2/3] Apply batched suggestions from code review Co-authored-by: Thomas Papendieck <14850347+tpd-opitz@users.noreply.github.com> --- packages/helpermodules/mosquitto_dynsec/user_handler.py | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/packages/helpermodules/mosquitto_dynsec/user_handler.py b/packages/helpermodules/mosquitto_dynsec/user_handler.py index 981ae6219f..3c9025f81e 100644 --- a/packages/helpermodules/mosquitto_dynsec/user_handler.py +++ b/packages/helpermodules/mosquitto_dynsec/user_handler.py @@ -3,6 +3,7 @@ from typing import Tuple from json import dump as json_dump +INTERNAL_DISPLAY_IP = "127.0.0.1" from helpermodules.mosquitto_dynsec.role_handler import _get_base_path from helpermodules.utils.run_command import run_command @@ -28,7 +29,7 @@ def add_user_to_group(username: str, groupname: str) -> None: def create_display_user(ip_address: str) -> Tuple[bool, str]: - user_name = "Display-Intern" if ip_address == "127.0.0.1" else f"Display-{ip_address}" + user_name = "Display-Intern" if ip_address == INTERNAL_DISPLAY_IP else f"Display-{ip_address}" file_name = f"display-{ip_address}".replace('.', '_') if user_exists(user_name): log.info(f"User '{user_name}' already exists") @@ -48,7 +49,7 @@ def create_display_user(ip_address: str) -> Tuple[bool, str]: def remove_display_user(ip_address: str) -> bool: - if ip_address == "127.0.0.1": + if ip_address == INTERNAL_DISPLAY_IP: log.error("Cannot remove the internal display user") return False user_name = f"Display-{ip_address}" @@ -62,9 +63,7 @@ def remove_display_user(ip_address: str) -> bool: def user_exists(username: str) -> bool: result = run_command(["mosquitto_ctrl", "dynsec", "getClient", username], process_exception=True) - if result is not None and username in result: - return True - return False + return result is not None and username in result def add_user(username: str, password: str) -> bool: From b3236d20bbdabe9e4b6d7c2920a1919038a17f1d Mon Sep 17 00:00:00 2001 From: Lutz Bender Date: Thu, 8 Oct 2026 11:34:07 +0200 Subject: [PATCH 3/3] fixes --- .../mosquitto_dynsec/mosquitto_dynsec.py | 8 ++++---- .../mosquitto_dynsec/user_handler.py | 17 ++++++++++------- 2 files changed, 14 insertions(+), 11 deletions(-) diff --git a/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py b/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py index 6fce4d85fb..08db4d89e9 100644 --- a/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py +++ b/packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py @@ -98,12 +98,12 @@ def verify_password_reset_token(username: str, token: str) -> bool: def check_required_users(): - def create_user(ip_address: str = "127.0.0.1"): - success, user_name = create_display_user(ip_address) + def create_user(ip_address: Optional[str] = None): + success, user_name, assigned_ip = create_display_user(ip_address) if success: - log.info(f"Created user '{user_name}' for cp display at {ip_address}") + log.info(f"Created user '{user_name}' for cp display at {assigned_ip}") else: - log.error(f"Failed to create user for cp display at {ip_address}") + log.error(f"Failed to create user for cp display at {assigned_ip}") # Always create user for localhost to ensure access to local displays, even if no chargepoints are configured create_user() diff --git a/packages/helpermodules/mosquitto_dynsec/user_handler.py b/packages/helpermodules/mosquitto_dynsec/user_handler.py index 3c9025f81e..4d3b5d9bf8 100644 --- a/packages/helpermodules/mosquitto_dynsec/user_handler.py +++ b/packages/helpermodules/mosquitto_dynsec/user_handler.py @@ -1,12 +1,13 @@ import logging from secrets import token_hex -from typing import Tuple +from typing import Tuple, Optional from json import dump as json_dump -INTERNAL_DISPLAY_IP = "127.0.0.1" from helpermodules.mosquitto_dynsec.role_handler import _get_base_path from helpermodules.utils.run_command import run_command +INTERNAL_DISPLAY_IP = "127.0.0.1" +INTERNAL_DISPLAY_USER = "Display-Intern" USER_CREDENTIALS_PATH = _get_base_path() / "data" / "clients" log = logging.getLogger(__name__) @@ -28,16 +29,18 @@ def add_user_to_group(username: str, groupname: str) -> None: run_command(["mosquitto_ctrl", "dynsec", "addGroupClient", groupname, username]) -def create_display_user(ip_address: str) -> Tuple[bool, str]: - user_name = "Display-Intern" if ip_address == INTERNAL_DISPLAY_IP else f"Display-{ip_address}" +def create_display_user(ip_address: Optional[str] = None) -> Tuple[bool, str, str]: + if ip_address is None: + ip_address = INTERNAL_DISPLAY_IP + user_name = INTERNAL_DISPLAY_USER if ip_address == INTERNAL_DISPLAY_IP else f"Display-{ip_address}" file_name = f"display-{ip_address}".replace('.', '_') if user_exists(user_name): log.info(f"User '{user_name}' already exists") - return True, user_name + return True, user_name, ip_address password = token_hex(16) if not add_user(user_name, password): log.error(f"Failed to create user '{user_name}' for cp display at {ip_address}") - return False, user_name + return False, user_name, ip_address add_user_to_group(user_name, "display") store_user_credentials( file_name, @@ -45,7 +48,7 @@ def create_display_user(ip_address: str) -> Tuple[bool, str]: password ) log.info(f"Created user '{user_name}' for cp display at {ip_address}") - return True, user_name + return True, user_name, ip_address def remove_display_user(ip_address: str) -> bool: