update acls

Author: LKuemmel

packages/helpermodules/mosquitto_dynsec.py

@@ -1,7 +1,9 @@
 from json import load as json_load
+import json
 import logging
 from pathlib import Path
-from typing import TypedDict, Optional
+import re
+from typing import Tuple, TypedDict, Optional
 
 from helpermodules.subdata import SubData
 from helpermodules.utils.run_command import run_command
@@ -83,6 +85,7 @@ def remove_acl_role(role_template: str, id: int):
 
 
 def check_roles_at_start():
+    update_acls()
     flag_path = Path(Path(__file__).resolve().parents[2]/"ramdisk"/"init_user_management")
     if flag_path.is_file():
         with open(flag_path, "r") as file:
@@ -104,3 +107,140 @@ def check_roles_at_start():
                 if "io" in key:
                     add_acl_role("io-device-<id>-access", value.config.id)
         flag_path.unlink()
+
+
+def _extract_id_from_role_name(role_name: str) -> Optional[int]:
+    numbers = re.findall(r'\d+', role_name)
+    if numbers:
+        return int(numbers[0])
+    return None
+
+
+def _compare_acl(template_acl: MosquittoAcl, configured_acl: MosquittoAcl) -> bool:
+    if template_acl["acltype"] == configured_acl["acltype"]:
+        if re.sub(r'/\d+/', '/<id>/', template_acl["topic"]) == re.sub(r'/\d+/', '/<id>/', configured_acl["topic"]):
+            if template_acl["allow"] == configured_acl["allow"]:
+                if template_acl["priority"] == configured_acl["priority"]:
+                    return True
+    return False
+
+
+def _get_configured_role_data(role_name: str) -> Optional[MosquittoRole]:
+    role_output = run_command([
+        "mosquitto_ctrl", "dynsec", "getRole", role_name])
+    # Parse the text output since it's not JSON
+    role_data = {"rolename": role_name, "acls": []}
+    lines = role_output.strip().split('\n')
+    for line in lines[1:]:  # Skip first line with rolename
+        if "ACLs:" in line:
+            line = line.replace("ACLs:", "")
+        if line.strip() and ':' in line:
+            parts = [p.strip() for p in line.split(':')]
+            if len(parts) >= 4:
+                acl = {
+                    "acltype": parts[0],
+                    "allow": parts[1] == "allow",
+                    "topic": parts[2].split('(')[0].strip(),
+                    "priority": int(parts[3].replace(')', '').strip())
+                }
+                role_data["acls"].append(acl)
+    return role_data
+
+
+def update_acls():
+    def is_version_updated() -> Tuple[str, str]:
+        for role in roles:
+            if "openwb-version" in role:
+                current_version = role.split(":")[1]
+                break
+        else:
+            raise RuntimeError("openwb-version role not found")
+
+        for role in dynsec_config["roles"]:
+            try:
+                if "openwb-version" in role["rolename"]:
+                    template_version = role["rolename"].split(":")[1]
+                    break
+            except Exception:
+                continue
+        else:
+            raise RuntimeError("openwb-version role not found in default-dynamic-security.json")
+
+        if current_version != template_version:
+            log.debug(f"Updating ACLs from version {current_version} to {template_version}")
+        return current_version, template_version
+
+    def get_template_role_data() -> Optional[MosquittoRole]:
+        template_role_data = None
+        for config_role in dynsec_config["roles"]:
+            if (config_role["rolename"] == role_data["rolename"] or
+                    ("openwb-version" in config_role["rolename"] and "openwb-version" in role_data["rolename"])):
+                template_role_data = config_role
+                break
+        else:
+            for config_role in role_templates_config:
+                pattern = config_role["rolename"].replace("<id>", r"\d+")
+                if re.match(pattern, role_data["rolename"]):
+                    template_role_data = config_role
+                    break
+            else:
+                raise RuntimeError(f"Role {role_data['rolename']} not found in default-dynamic-security.json")
+        return template_role_data
+
+    try:
+        roles = _list_acl_roles()
+        with open(_get_packages_path()/"data/config/mosquitto/public/default-dynamic-security.json", "r") as f:
+            dynsec_config = json.load(f)
+        current_version, template_version = is_version_updated()
+        if current_version != template_version:
+            with open(_get_packages_path()/"data/config/mosquitto/public/role-templates.json", "r") as f:
+                role_templates_config = json.load(f)
+            for role_name in roles:
+                try:
+                    role_data = _get_configured_role_data(role_name)
+                    template_role_data = get_template_role_data()
+                    if template_role_data:
+                        # vegleiche die zwei ACL dicts, welche ACLs hinzugefügt, geändert oder entfernt wurden
+                        for acl in role_data["acls"]:
+                            for template_acl in template_role_data["acls"]:
+                                if _compare_acl(template_acl, acl):
+                                    break
+                            else:
+                                log.debug(f"ACL {acl['acltype']}:{'allow' if acl['allow'] else 'deny'}:{acl['topic']}:"
+                                          f"{acl['priority']} in Rolle {role_data['rolename']} wird entfernt.")
+                                run_command([
+                                    "mosquitto_ctrl", "dynsec", "removeRoleAcl", role_data["rolename"],
+                                    acl["acltype"], acl["topic"]
+                                ])
+                        for acl in template_role_data["acls"]:
+                            for role_acl in role_data["acls"]:
+                                if _compare_acl(acl, role_acl):
+                                    break
+                            else:
+                                rolename_id = _extract_id_from_role_name(role_data["rolename"])
+                                if rolename_id is not None:
+                                    acl["topic"] = acl["topic"].replace("<id>", str(rolename_id))
+                                log.debug(f"ACL {acl['acltype']}:{'allow' if acl['allow'] else 'deny'}:{acl['topic']}:"
+                                          f"{acl['priority']} in Rolle {role_data['rolename']} wird hinzugefügt.")
+                                run_command([
+                                    "mosquitto_ctrl", "dynsec", "addRoleAcl", role_data["rolename"],
+                                    acl["acltype"], acl["topic"],
+                                    "allow" if acl["allow"] else "deny",
+                                    str(acl["priority"])
+                                ])
+                    elif template_role_data is None:
+                        log.debug(f"Rolle '{role_data['rolename']}' existiert nicht in den Konfigurationsdateien und"
+                                  " wird gelöscht.")
+                        run_command(["mosquitto_ctrl", "dynsec", "deleteRole", role_data["rolename"]])
+                except Exception:
+                    log.exception(f"Fehler beim Aktualisieren der Rolle '{role_name}'")
+            # bei allen anderen Rollen dürfen nur die ACLs editiert werden,
+            # damit diese in den Benutzergruppen erhalten bleiben
+            run_command(["mosquitto_ctrl", "dynsec", "deleteRole", f"openwb-version:{current_version}"])
+            run_command(["mosquitto_ctrl", "dynsec", "createRole", f"openwb-version:{template_version}"])
+    except Exception:
+        log.exception("Fehler beim Aktualisieren der ACLs")
+
+
+def _get_packages_path() -> Path:
+    return Path(__file__).resolve().parents[2]