ipmi: usinc usc auth

kvmd/apps/ipmi/auth.py

@@ -20,7 +20,13 @@
 # ========================================================================== #
 
 
-import dataclasses
+import threading
+import functools
+import time
+
+from ...logging import get_logger
+
+from ... import tools
 
 
 # =====
@@ -29,60 +35,42 @@ class IpmiPasswdError(Exception):
         super().__init__(f"Syntax error at {path}:{lineno}: {msg}")
 
 
-@dataclasses.dataclass(frozen=True)
-class IpmiUserCredentials:
-    ipmi_user: str
-    ipmi_passwd: str
-    kvmd_user: str
-    kvmd_passwd: str
-
-
 class IpmiAuthManager:
     def __init__(self, path: str) -> None:
         self.__path = path
-        with open(path) as file:
-            self.__credentials = self.__parse_passwd_file(file.read().split("\n"))
+        self.__lock = threading.Lock()
 
-    def __contains__(self, ipmi_user: str) -> bool:
-        return (ipmi_user in self.__credentials)
+    def get(self, user: str) -> (str | None):
+        creds = self.__get_credentials(int(time.time()))
+        return creds.get(user)
 
-    def __getitem__(self, ipmi_user: str) -> str:
-        return self.__credentials[ipmi_user].ipmi_passwd
+    @functools.lru_cache(maxsize=1)
+    def __get_credentials(self, ts: int) -> dict[str, str]:
+        _ = ts
+        with self.__lock:
+            try:
+                return self.__read_credentials()
+            except Exception as ex:
+                get_logger().error("%s", tools.efmt(ex))
+            return {}
 
-    def get_credentials(self, ipmi_user: str) -> IpmiUserCredentials:
-        return self.__credentials[ipmi_user]
+    def __read_credentials(self) -> dict[str, str]:
+        with open(self.__path) as file:
+            creds: dict[str, str] = {}
+            for (lineno, line) in tools.passwds_splitted(file.read()):
+                if " -> " in line:  # Compatibility with old ipmipasswd file format
+                    line = line.split(" -> ", 1)[0]
 
-    def __parse_passwd_file(self, lines: list[str]) -> dict[str, IpmiUserCredentials]:
-        credentials: dict[str, IpmiUserCredentials] = {}
-        for (lineno, line) in enumerate(lines):
-            if len(line.strip()) == 0 or line.lstrip().startswith("#"):
-                continue
+                if ":" not in line:
+                    raise IpmiPasswdError(self.__path, lineno, "Missing ':' operator")
 
-            if " -> " not in line:
-                raise IpmiPasswdError(self.__path, lineno, "Missing ' -> ' operator")
+                (user, passwd) = line.split(":", 1)
+                user = user.strip()
+                if len(user) == 0:
+                    raise IpmiPasswdError(self.__path, lineno, "Empty IPMI user")
 
-            (left, right) = map(str.lstrip, line.split(" -> ", 1))
-            for (name, pair) in [("left", left), ("right", right)]:
-                if ":" not in pair:
-                    raise IpmiPasswdError(self.__path, lineno, f"Missing ':' operator in {name} credentials")
+                if user in creds:
+                    raise IpmiPasswdError(self.__path, lineno, f"Found duplicating user {user!r}")
 
-            (ipmi_user, ipmi_passwd) = left.split(":")
-            ipmi_user = ipmi_user.strip()
-            if len(ipmi_user) == 0:
-                raise IpmiPasswdError(self.__path, lineno, "Empty IPMI user (left)")
-
-            (kvmd_user, kvmd_passwd) = right.split(":")
-            kvmd_user = kvmd_user.strip()
-            if len(kvmd_user) == 0:
-                raise IpmiPasswdError(self.__path, lineno, "Empty KVMD user (left)")
-
-            if ipmi_user in credentials:
-                raise IpmiPasswdError(self.__path, lineno, f"Found duplicating user {ipmi_user!r} (left)")
-
-            credentials[ipmi_user] = IpmiUserCredentials(
-                ipmi_user=ipmi_user,
-                ipmi_passwd=ipmi_passwd,
-                kvmd_user=kvmd_user,
-                kvmd_passwd=kvmd_passwd,
-            )
-        return credentials
+                creds[user] = passwd
+            return creds