changes from copilot review

Author: benderl

packages/helpermodules/command.py

@@ -365,20 +365,27 @@ def removeChargepoint(self, connection_id: str, payload: dict) -> None:
                 payload, connection_id,
                 f'Die ID \'{cp_id}\' ist größer als die maximal vergebene '
                 f'ID \'{self.max_id_hierarchy}\'.', MessageType.ERROR)
-        cp_type = SubData.cp_data.get(f"cp{cp_id}", None).chargepoint.data.config.type
-        cp_ip = (SubData.cp_data.get(f"cp{cp_id}", None).chargepoint.data.config.configuration.get("ip_address")
-                 if cp_type == "external_openwb" else "127.0.0.1" if cp_type == "internal_openwb" else None)
-        if cp_type == "external_openwb" and cp_ip is not None:
-            # check if ip is used in other chargepoints, if not remove display user
-            for cp in SubData.cp_data.values():
-                if (
-                    cp.chargepoint.data.config.configuration.get("ip_address", None) == cp_ip
-                    and cp.chargepoint.num != cp_id
-                ):
-                    log.info(f"IP {cp_ip} is still used by cp{cp.chargepoint.num}, not removing display user")
-                    break
-            else:
-                remove_display_user(cp_ip)
+            return
+        cp = SubData.cp_data.get(f"cp{cp_id}", None)
+        if cp is None:
+            pub_user_message(
+                payload, connection_id,
+                f'Ladepunkt mit ID \'{cp_id}\' existiert nicht.', MessageType.ERROR)
+            return
+        cp_type = cp.chargepoint.data.config.type
+        if cp_type == "external_openwb":
+            cp_ip = cp.chargepoint.data.config.configuration.get("ip_address")
+            if cp_ip is not None:
+                # check if ip is used in other charge points, if not remove display user
+                for cp in SubData.cp_data.values():
+                    if (
+                        cp.chargepoint.data.config.configuration.get("ip_address", None) == cp_ip
+                        and cp.chargepoint.num != cp_id
+                    ):
+                        log.info(f"IP {cp_ip} is still used by cp{cp.chargepoint.num}, not removing display user")
+                        break
+                else:
+                    remove_display_user(cp_ip)
         remove_acl_role("chargepoint-<id>-access", cp_id)
         remove_acl_role("chargepoint-<id>-write-access", cp_id)
         ProcessBrokerBranch(f'chargepoint/{cp_id}/').remove_topics()

packages/helpermodules/mosquitto_dynsec/mosquitto_dynsec.py

@@ -3,32 +3,32 @@
 import requests
 from json import load as json_load, dump as json_dump
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Tuple
 from datetime import datetime, timedelta
 from passlib.hash import bcrypt
 from secrets import token_hex
 
 from helpermodules.subdata import SubData
 from helpermodules.utils.run_command import run_command
-from helpermodules.mosquitto_dynsec.role_handler import add_acl_role, update_acls
+from helpermodules.mosquitto_dynsec.role_handler import _get_base_path, add_acl_role, update_acls
 from helpermodules.mosquitto_dynsec.user_handler import create_display_user, user_exists
 from modules.common.component_type import special_to_general_type_mapping
 
-TOKEN_DATA_PATH = Path(Path(__file__).resolve().parents[2]/"ramdisk"/"password_reset_tokens")
+TOKEN_DATA_PATH = Path(_get_base_path() / "ramdisk" / "password_reset_tokens")
 
 log = logging.getLogger(__name__)
 
 
 def get_user_email(username: str) -> Optional[str]:
     email_regexp = r'(^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z0-9]{2,7}$)'
     if user_exists(username):
-        result = run_command(["/var/www/html/openWB/runs/dynsec_helper/get_user_mail.sh", username])
+        result = run_command([f"{_get_base_path()}/runs/dynsec_helper/get_user_mail.sh", username])
         email = result.strip()
         return email if re.fullmatch(email_regexp, email) else None
     return None
 
 
-def generate_password_reset_token(username: str) -> list[str, int]:
+def generate_password_reset_token(username: str) -> Tuple[str, int]:
     token = token_hex(16)
     token_hash = bcrypt.hash(token)
     expires_at = datetime.now() + timedelta(hours=1)
@@ -68,15 +68,15 @@ def send_password_reset_to_server(email: str, token: str, expires_at: int) -> No
             timeout=timeout,
             verify=True
         )
-        print(f"{response.status_code}: {response.text}")
+        log.debug(f"Password reset token request returned {response.status_code}: {response.text}")
     except requests.exceptions.Timeout:
         error = f"Error: request timed out after {timeout}s"
     except requests.exceptions.ConnectionError:
         error = f"Error: connection to {TOKEN_SERVER_URL} failed"
     except requests.RequestException as e:
         error = f"HTTP-Error: {e}"
     if error is not None or response.status_code != 200:
-        print(f"Failed to send password reset email: {error or f'status {response.status_code}'}")
+        log.error(f"Failed to send password reset email: {error or f'status {response.status_code}'}")
 
 
 def verify_password_reset_token(username: str, token: str) -> bool:

packages/helpermodules/mosquitto_dynsec/role_handler.py

@@ -3,6 +3,7 @@
 from typing import Tuple, TypedDict, Optional
 from pathlib import Path
 from json import load as json_load
+from copy import deepcopy
 
 from helpermodules.utils.run_command import run_command
 
@@ -24,12 +25,12 @@ class MosquittoRole(TypedDict):
     acls: list[MosquittoAcl]
 
 
-def _get_packages_path() -> Path:
+def _get_base_path() -> Path:
     return Path(__file__).resolve().parents[3]
 
 
 def _get_default_roles() -> list[MosquittoRole]:
-    with open(_get_packages_path() /
+    with open(_get_base_path() /
               "data" / "config" / "mosquitto" / "public" / "default-dynamic-security.json", 'r',
               encoding='utf-8') as file:
         dynsec_config = json_load(file)
@@ -38,7 +39,7 @@ def _get_default_roles() -> list[MosquittoRole]:
 
 
 def _get_role_templates() -> list[MosquittoRole]:
-    with open(_get_packages_path() /
+    with open(_get_base_path() /
               "data" / "config" / "mosquitto" / "public" / "role-templates.json", 'r', encoding='utf-8') as file:
         roles: list[MosquittoRole] = json_load(file)
     return roles if roles else []
@@ -51,21 +52,22 @@ def extract_id_from_role_name(role_name: str) -> Optional[int]:
     return None
 
 
-def get_acl_role_data(role_template: str, id: int) -> MosquittoRole:
-    roles = _get_role_templates()
-    role_data = None
-    for role in roles:
-        if role["rolename"] == role_template:
-            role_data = role
-            break
-    if role_data is None:
-        raise ValueError(f"Kein passendes Rollen-Template für '{role_template}' gefunden.")
-    role_data["rolename"] = role_data["rolename"].replace("<id>", str(id))
-    role_data["textname"] = role_data["textname"].replace("<id>", str(id))
-    role_data["textdescription"] = role_data["textdescription"].replace("<id>", str(id))
-    for acl in role_data["acls"]:
-        acl["topic"] = acl["topic"].replace("<id>", str(id))
-    return role_data
+def get_acl_role_data(role_template_name: str, id: int) -> MosquittoRole:
+    template_roles = _get_role_templates()
+    for template_role in template_roles:
+        if template_role["rolename"] == role_template_name:
+            role_data: MosquittoRole = {
+                "rolename": template_role["rolename"].replace("<id>", str(id)),
+                "textname": template_role["textname"].replace("<id>", str(id)),
+                "textdescription": template_role["textdescription"].replace("<id>", str(id)),
+                "acls": []
+            }
+            for acl in template_role["acls"]:
+                new_acl = deepcopy(acl)
+                new_acl["topic"] = new_acl["topic"].replace("<id>", str(id))
+                role_data["acls"].append(new_acl)
+            return role_data
+    raise ValueError(f"Kein passendes Rollen-Template für '{role_template_name}' gefunden.")
 
 
 def get_configured_role_data(role_name: str) -> Optional[MosquittoRole]:

packages/helpermodules/mosquitto_dynsec/user_handler.py

@@ -3,10 +3,10 @@
 from typing import Optional, Tuple
 from json import dump as json_dump
 
-from helpermodules.mosquitto_dynsec.role_handler import _get_packages_path
+from helpermodules.mosquitto_dynsec.role_handler import _get_base_path
 from helpermodules.utils.run_command import run_command
 
-USER_CREDENTIALS_PATH = _get_packages_path() / "data" / "clients"
+USER_CREDENTIALS_PATH = _get_base_path() / "data" / "clients"
 log = logging.getLogger(__name__)
 
 
@@ -58,7 +58,7 @@ def remove_display_user(ip_address: str) -> bool:
     return False
 
 
-def user_exists(username: str) -> Optional[dict]:
+def user_exists(username: str) -> bool:
     result = run_command(["mosquitto_ctrl", "dynsec", "getClient", username], process_exception=True)
     if username in result:
         return True

packages/modules/display_themes/cards/source/src/App.vue

@@ -159,12 +159,15 @@ export default {
     createConnection() {
       const { protocol, host, port, path, ...options } = this.connection;
       const connectUrl = `${protocol}://${host}:${port}${path}`;
-      const [user, pass] = this.$cookies.get("mqtt")?.split(":") || [null, null];
+      const [user, pass] = this.$cookies
+        .get("mqtt")
+        ?.match(/^([^:]+):(.+)$/)
+        ?.slice(1) || [null, null];
       if (!(user && pass)) {
         console.debug("Anonymous mqtt connection (no cookie set)");
       }
       if ((this.nodeEnv !== "production" || protocol == "wss") && user && pass) {
-        console.debug("Using mqtt credentials from cookie:", user, "/", pass);
+        console.debug(`Using mqtt credentials from cookie: "${user}" / "${pass.charAt(0)}..."`);
         options.username = user;
         options.password = pass;
         if (user === "admin" && pass === "openwb") {

packages/modules/display_themes/cards/source/src/components/UserIndicator.vue

@@ -4,13 +4,14 @@ import { library } from "@fortawesome/fontawesome-svg-core";
 import { FontAwesomeIcon } from "@fortawesome/vue-fontawesome";
 import {
   faCircleUser as fasCircleUser,
+  faLock as fasLock,
   faArrowRightToBracket as fasArrowRightToBracket,
   faArrowRightFromBracket as fasArrowRightFromBracket,
   faEye as fasEye,
   faEyeSlash as fasEyeSlash,
 } from "@fortawesome/free-solid-svg-icons";
 /* add icons to the library */
-library.add(fasCircleUser, fasArrowRightToBracket, fasArrowRightFromBracket, fasEye, fasEyeSlash);
+library.add(fasCircleUser, fasLock, fasArrowRightToBracket, fasArrowRightFromBracket, fasEye, fasEyeSlash);
 import { useMqttStore } from "@/stores/mqtt.js";
 
 export default {

packages/modules/web_themes/koala/source/src/stores/index.ts

@@ -11,7 +11,7 @@ declare module 'pinia' {
   // eslint-disable-next-line @typescript-eslint/no-empty-object-type
   export interface PiniaCustomProperties {
     // add your custom properties here, if any
-    readonly router: Router;
+    readonly router: Router | null;
   }
 }
 

packages/modules/web_themes/koala/source/src/stores/mqtt-store.ts

@@ -34,9 +34,9 @@ export const useMqttStore = defineStore('mqtt', () => {
   let mqttUser = null;
   let mqttPass = null;
   if ($q.cookies.has('mqtt')) {
-    [mqttUser, mqttPass] = decodeURIComponent($q.cookies.get('mqtt')).split(
-      ':',
-    ) || [null, null];
+    [mqttUser, mqttPass] = decodeURIComponent($q.cookies.get('mqtt'))
+      ?.match(/^([^:]+):(.+)$/)
+      ?.slice(1) || [null, null];
   }
 
   // local variables