Implementing X509Plain VeNCrypt security type 262 in place of TLSPlain security type 259.

This is a fix for both bVNC on iOS and for devices that do not support Anonymous Diffie
Hellman ciphers (e.g. Android past API 23).

It provides a mechanism for verifying and approving the identity of the server and ensuring
subsequent connections to the same server are not under MITM attack.

kvmd/apps/vnc/rfb/__init__.py

@@ -42,6 +42,7 @@ from .encodings import RfbClientEncodings
 
 from .crypto import rfb_make_challenge
 from .crypto import rfb_encrypt_challenge
+from .crypto import create_self_signed_cert_if_nonexistent, key_file_name, cert_file_name
 
 from .stream import RfbClientStream
 
@@ -261,7 +262,7 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
         else:
             auth_types = {256: ("VeNCrypt/Plain", False, self.__handshake_security_vencrypt_userpass)}
             if self.__tls_ciphers:
-                auth_types[259] = ("VeNCrypt/TLSPlain", True, self.__handshake_security_vencrypt_userpass)
+                auth_types[262] = ("VeNCrypt/X509Plain", True, self.__handshake_security_vencrypt_userpass)
             if self.__vnc_passwds:
                 # Vinagre не умеет работать с VNC Auth через VeNCrypt, но это его проблемы,
                 # так как он своеобразно трактует рекомендации VeNCrypt.
@@ -284,6 +285,8 @@ class RfbClient(RfbClientStream):  # pylint: disable=too-many-instance-attribute
             assert self.__tls_ciphers, (self.__tls_ciphers, auth_name, tls, handler)
             await self._write_struct("B", 1)  # Ack
             ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+            create_self_signed_cert_if_nonexistent(key_file_name, cert_file_name)
+            ssl_context.load_cert_chain(keyfile=key_file_name, certfile=cert_file_name)
             ssl_context.set_ciphers(self.__tls_ciphers)
             await self._start_tls(ssl_context, self.__tls_timeout)
 

kvmd/apps/vnc/rfb/crypto.py

@@ -26,6 +26,14 @@ from typing import List
 
 import passlib.crypto.des
 
+from OpenSSL import crypto, SSL
+from socket import gethostname
+from pprint import pprint
+from time import gmtime, mktime
+import os.path
+
+key_file_name = "private_vnc.key"
+cert_file_name = "self_signed_cert.crt"
 
 # =====
 def rfb_make_challenge() -> bytes:
@@ -51,3 +59,30 @@ def _make_key(passwd: bytes) -> bytes:
                 btgt = btgt | (1 << 7 - index)
         key.append(btgt)
     return bytes(key)
+
+
+def create_self_signed_cert_if_nonexistent(key_file, cert_file):
+    if os.path.isfile(key_file) and os.path.isfile(cert_file):
+        return
+
+    key = crypto.PKey()
+    key.generate_key(crypto.TYPE_RSA, 2048)
+
+    cert = crypto.X509()
+    cert.get_subject().C = "CA"
+    cert.get_subject().ST = "Toronto"
+    cert.get_subject().L = "Toronto"
+    cert.get_subject().O = "Company Ltd"
+    cert.get_subject().OU = "Company Ltd"
+    cert.get_subject().CN = gethostname()
+    cert.set_serial_number(1000)
+    cert.gmtime_adj_notBefore(0)
+    cert.gmtime_adj_notAfter(100*365*24*60*60)
+    cert.set_issuer(cert.get_subject())
+    cert.set_pubkey(key)
+    cert.sign(key, 'sha256')
+
+    open(key_file, "wt").write(
+        crypto.dump_privatekey(crypto.FILETYPE_PEM, key).decode('utf-8'))
+    open(cert_file, "wt").write(
+        crypto.dump_certificate(crypto.FILETYPE_PEM, cert).decode('utf-8'))