improved security checks

kvmd/apps/ipmi/auth.py

@@ -70,9 +70,13 @@ class IpmiAuthManager:
 
             (ipmi_user, ipmi_passwd) = left.split(":")
             ipmi_user = ipmi_user.strip()
+            if len(ipmi_user) == 0:
+                raise IpmiPasswdError(f"Empty IPMI user (left) at line #{number}")
 
             (kvmd_user, kvmd_passwd) = right.split(":")
             kvmd_user = kvmd_user.strip()
+            if len(kvmd_user) == 0:
+                raise IpmiPasswdError(f"Empty KVMD user (left) at line #{number}")
 
             if ipmi_user in credentials:
                 raise IpmiPasswdError(f"Found duplicating user {ipmi_user!r} (left) at line #{number}")

kvmd/apps/kvmd/auth.py

@@ -71,6 +71,8 @@ class AuthManager:
         return self.__enabled
 
     async def authorize(self, user: str, passwd: str) -> bool:
+        assert user == user.strip()
+        assert user
         assert self.__enabled
         assert self.__internal_service
 
@@ -87,6 +89,8 @@ class AuthManager:
         return ok
 
     async def login(self, user: str, passwd: str) -> Optional[str]:
+        assert user == user.strip()
+        assert user
         assert self.__enabled
         if (await self.authorize(user, passwd)):
             for (token, token_user) in self.__tokens.items():

kvmd/apps/vnc/rfb/__init__.py

@@ -277,54 +277,58 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
 
     async def __handshake_security_vencrypt_userpass(self) -> None:
         (user_length, passwd_length) = await self._read_struct("LL")
-        user = await self._read_text(user_length)
+        user = (await self._read_text(user_length)).strip()
         passwd = await self._read_text(passwd_length)
 
-        ok = await self._authorize_userpass(user, passwd)
-        await self.__handshake_security_send_result(ok, user)
+        allow = await self._authorize_userpass(user, passwd)
+        if allow:
+            assert user
+        await self.__handshake_security_send_result(
+            allow=allow,
+            allow_msg=f"Access granted for user {user!r}",
+            deny_msg=f"Access denied for user {user!r}",
+            deny_reason="Invalid username or password",
+        )
 
     async def __handshake_security_none(self) -> None:
-        ok = await self._on_authorized_none()
-        await self.__handshake_security_send_result(ok, "")
+        allow = await self._on_authorized_none()
+        await self.__handshake_security_send_result(
+            allow=allow,
+            allow_msg="NoneAuth access granted",
+            deny_msg="NoneAuth access denied",
+            deny_reason="Access denied",
+        )
 
     async def __handshake_security_vnc_auth(self) -> None:
         challenge = rfb_make_challenge()
         await self._write_struct("", challenge)
 
-        (ok, user) = (False, "")
+        user = ""
         response = (await self._read_struct("16s"))[0]
         for passwd in self.__vnc_passwds:
             passwd_bytes = passwd.encode("utf-8", errors="ignore")
             if rfb_encrypt_challenge(challenge, passwd_bytes) == response:
                 user = await self._on_authorized_vnc_passwd(passwd)
                 if user:
-                    ok = True
+                    assert user == user.strip()
                 break
 
-        await self.__handshake_security_send_result(ok, user)
+        await self.__handshake_security_send_result(
+            allow=bool(user),
+            allow_msg="VNCAuth access granted for user {user!r}",
+            deny_msg="VNCAuth access denied (user not found)",
+            deny_reason="Invalid password",
+        )
 
-    async def __handshake_security_send_result(self, ok: bool, user: str) -> None:
-        if ok:
-            if self.__none_auth_only:
-                assert len(user) == 0
-                get_logger(0).info("[main] Client %s: Anonymous access granted", self._remote)
-            else:
-                assert user
-                get_logger(0).info("[main] Client %s: Access granted for user %r", self._remote, user)
+    async def __handshake_security_send_result(self, allow: bool, allow_msg: str, deny_msg: str, deny_reason: str) -> None:
+        if allow:
+            get_logger(0).info("[main] Client %s: %s", self._remote, allow_msg)
             await self._write_struct("L", 0)
         else:
             await self._write_struct("L", 1, drain=(self.__rfb_version < 8))
-            if self.__none_auth_only:
-                reason = msg = "Anonymous access denied"
-            elif user:
-                reason = "Invalid username or password"
-                msg = f"Access denied for user {user!r}"
-            else:
-                reason = "Invalid password"
-                msg = "Access denied"
             if self.__rfb_version >= 8:
-                await self._write_reason(reason)
-            raise RfbError(msg)
+                await self._write_reason(deny_reason)
+            raise RfbError(deny_msg)
 
     # =====
 

kvmd/apps/vnc/vncauth.py

@@ -81,6 +81,8 @@ class VncAuthManager:
 
             (kvmd_user, kvmd_passwd) = kvmd_userpass.split(":")
             kvmd_user = kvmd_user.strip()
+            if len(kvmd_user) == 0:
+                raise VncAuthError(f"Empty KVMD user (right part) at line #{number}")
 
             if vnc_passwd in credentials:
                 raise VncAuthError(f"Found duplicating VNC password (left part) at line #{number}")

kvmd/plugins/auth/htpasswd.py

@@ -43,5 +43,7 @@ class Plugin(BaseAuthService):
         }
 
     async def authorize(self, user: str, passwd: str) -> bool:
+        assert user == user.strip()
+        assert user
         htpasswd = passlib.apache.HtpasswdFile(self.__path)
         return htpasswd.check_password(user, passwd)

kvmd/plugins/auth/http.py

@@ -71,6 +71,8 @@ class Plugin(BaseAuthService):
         }
 
     async def authorize(self, user: str, passwd: str) -> bool:
+        assert user == user.strip()
+        assert user
         session = self.__ensure_session()
         try:
             async with session.request(

kvmd/plugins/auth/pam.py

@@ -67,6 +67,8 @@ class Plugin(BaseAuthService):
         }
 
     async def authorize(self, user: str, passwd: str) -> bool:
+        assert user == user.strip()
+        assert user
         async with self.__lock:
             return (await aiotools.run_async(self.__inner_authorize, user, passwd))