vnc none auth support

kvmd/apps/vnc/rfb/__init__.py

@@ -62,6 +62,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         height: int,
         name: str,
         vnc_passwds: List[str],
+        none_auth_only: bool,
     ) -> None:
 
         super().__init__(reader, writer)
@@ -73,6 +74,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         self._height = height
         self.__name = name
         self.__vnc_passwds = vnc_passwds
+        self.__none_auth_only = none_auth_only
 
         self.__rfb_version = 0
         self._encodings = RfbClientEncodings(frozenset())
@@ -124,6 +126,11 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
     async def _on_authorized_vnc_passwd(self, passwd: str) -> str:
         raise NotImplementedError
 
+    async def _on_authorized_none(self) -> bool:
+        raise NotImplementedError
+
+    # =====
+
     async def _on_key_event(self, code: int, state: bool) -> None:
         raise NotImplementedError
 
@@ -201,10 +208,12 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         sec_types: Dict[int, Tuple[str, Callable]] = {}
         if self.__rfb_version > 3:
             sec_types[19] = ("VeNCrypt", self.__handshake_security_vencrypt)
-        if self.__vnc_passwds:
+        if self.__none_auth_only:
+            sec_types[1] = ("None", self.__handshake_security_none)
+        elif self.__vnc_passwds:
             sec_types[2] = ("VNCAuth", self.__handshake_security_vnc_auth)
         if not sec_types:
-            msg = "The client uses a very old protocol 3.3 and VNCAuth is disabled"
+            msg = "The client uses a very old protocol 3.3 and VNCAuth or NoneAuth is disabled"
             await self._write_struct("L", 0, drain=False)  # Refuse old clients using the invalid security type
             await self._write_reason(msg)
             raise RfbError(msg)
@@ -229,6 +238,12 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
 
         await self._write_struct("B", 0)
 
+        if self.__none_auth_only:
+            auth_types = {
+                1: ("VeNCrypt/None", False, self.__handshake_security_none),
+                257: ("VeNCrypt/TLSNone", True, self.__handshake_security_none),
+            }
+        else:
             auth_types = {
                 256: ("VeNCrypt/Plain", False, self.__handshake_security_vencrypt_userpass),
                 259: ("VeNCrypt/TLSPlain", True, self.__handshake_security_vencrypt_userpass),
@@ -266,9 +281,12 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         passwd = await self._read_text(passwd_length)
 
         ok = await self._authorize_userpass(user, passwd)
-
         await self.__handshake_security_send_result(ok, user)
 
+    async def __handshake_security_none(self) -> None:
+        ok = await self._on_authorized_none()
+        await self.__handshake_security_send_result(ok, "")
+
     async def __handshake_security_vnc_auth(self) -> None:
         challenge = rfb_make_challenge()
         await self._write_struct("", challenge)
@@ -287,14 +305,26 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
 
     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)
             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("Invalid username or password" if user else "Invalid password")
+                await self._write_reason(reason)
-            raise RfbError(f"Access denied for user {user!r}" if user else "Access denied")
+            raise RfbError(msg)
 
     # =====
 

kvmd/apps/vnc/server.py

@@ -73,6 +73,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
         streamer: StreamerClient,
 
         vnc_credentials: Dict[str, VncAuthKvmdCredentials],
+        none_auth_only: bool,
         shared_params: _SharedParams,
     ) -> None:
 
@@ -84,6 +85,7 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
             tls_ciphers=tls_ciphers,
             tls_timeout=tls_timeout,
             vnc_passwds=list(vnc_credentials),
+            none_auth_only=none_auth_only,
             **dataclasses.asdict(shared_params),
         )
 
@@ -239,6 +241,11 @@ class _Client(RfbClient):  # pylint: disable=too-many-instance-attributes
             return kc.user
         return ""
 
+    async def _on_authorized_none(self) -> bool:
+        return (await self._authorize_userpass("", ""))
+
+    # =====
+
     async def _on_key_event(self, code: int, state: bool) -> None:
         if (web_name := self.__symmap.get(code)) is not None:  # noqa: E203,E231
             await self.__ws_writer_queue.put({
@@ -321,6 +328,7 @@ class VncServer:  # pylint: disable=too-many-instance-attributes
                 kvmd=kvmd,
                 streamer=streamer,
                 vnc_credentials=(await self.__vnc_auth_manager.read_credentials())[0],
+                none_auth_only=(await kvmd.authorize("", "")),
                 shared_params=shared_params,
             ).run()