Add text console support for Proxmox guests

confluent_server/confluent/plugins/hardwaremanagement/proxmox.py

@@ -11,6 +11,14 @@ import eventlet
 import confluent.interface.console as conapi
 import io
 import urllib.parse as urlparse
+import eventlet.green.ssl as ssl
+
+
+try:
+    websocket = eventlet.import_patched('websocket')
+    wso = websocket.WebSocket
+except Exception:
+    wso = object
 
 class RetainedIO(io.BytesIO):
     # Need to retain buffer after close
@@ -20,14 +28,138 @@ class RetainedIO(io.BytesIO):
         self.resultbuffer = self.getbuffer()
         super().close()
 
+class WrappedWebSocket(wso):
+
+    def set_verify_callback(self, callback):
+        self._certverify = callback
+
+    def connect(self, url, **options):
+
+        add_tls = url.startswith('wss://')
+        if add_tls:
+            hostname, port, resource, _ = websocket._url.parse_url(url)
+            if hostname[0] != '[' and ':' in hostname:
+                hostname = '[{0}]'.format(hostname)
+            if resource[0] != '/':
+                resource = '/{0}'.format(resource)
+            url = 'ws://{0}:8006{1}'.format(hostname,resource)
+        else:
+            return super(WrappedWebSocket, self).connect(url, **options)
+        self.sock_opt.timeout = options.get('timeout', self.sock_opt.timeout)
+        self.sock, addrs = websocket._http.connect(url, self.sock_opt, websocket._http.proxy_info(**options),
+                                           options.pop('socket', None))
+        self.sock = ssl.wrap_socket(self.sock, cert_reqs=ssl.CERT_NONE)
+        # The above is supersedeed by the _certverify, which provides
+        # known-hosts style cert validaiton
+        bincert = self.sock.getpeercert(binary_form=True)
+        if not self._certverify(bincert):
+            raise pygexc.UnrecognizedCertificate('Unknown certificate', bincert)
+        try:
+            try:
+                self.handshake_response = websocket._handshake.handshake(self.sock, *addrs, **options)
+            except TypeError:
+                self.handshake_response = websocket._handshake.handshake(self.sock, url, *addrs, **options)
+            if self.handshake_response.status in websocket._handshake.SUPPORTED_REDIRECT_STATUSES:
+                options['redirect_limit'] = options.pop('redirect_limit', 3) - 1
+                if options['redirect_limit'] < 0:
+                     raise Exception('Redirect limit hit')
+                url = self.handshake_response.headers['location']
+                self.sock.close()
+                return self.connect(url, **options)
+            self.connected = True
+        except:
+            if self.sock:
+                self.sock.close()
+                self.sock = None
+            raise
+
+
 class PmxConsole(conapi.Console):
-    pass
-    # this more closely resembles OpenBMC.., websocket based and all
+    def __init__(self, consdata, node, configmanager, apiclient):
+        self.ws = None
+        self.consdata = consdata
+        self.nodeconfig = configmanager
+        self.connected = False
+        self.bmc = consdata['server']
+        self.node = node
+        self.recvr = None
+        self.apiclient = apiclient
+
+    def recvdata(self):
+        while self.connected:
+            try:
+                pendingdata = self.ws.recv()
+            except websocket.WebSocketConnectionClosedException:
+                pendingdata = ''
+            if pendingdata == '':
+                self.datacallback(conapi.ConsoleEvent.Disconnect)
+                return
+            self.datacallback(pendingdata)
+
+    def connect(self, callback):
+        if self.apiclient.get_vm_power(self.node) != 'on':
+            callback(conapi.ConsoleEvent.Disconnect)
+            return
+        # socket = new WebSocket(socketURL, 'binary'); - subprotocol binary
+        # client handshake is:
+        #     socket.send(PVE.UserName + ':' + ticket + "\n");
+
+        # Peer sends 'OK' on handshake, other than that it's direct pass through
+        # send '2' every 30 seconds for keepalive
+        # data is xmitted with 0:<len>:data
+        # resize is sent with 1:columns:rows:""
+        self.datacallback = callback
+        kv = util.TLSCertVerifier(
+            self.nodeconfig, self.node, 'pubkeys.tls_hardwaremanager').verify_cert
+        bmc = self.bmc
+        if '%' in self.bmc:
+            prefix = self.bmc.split('%')[0]
+            bmc = prefix + ']'
+        self.ws = WrappedWebSocket(host=bmc)
+        self.ws.set_verify_callback(kv)
+        ticket = self.consdata['ticket']
+        user = self.consdata['user']
+        port = self.consdata['port']
+        urlticket = urlparse.quote(ticket)
+        host = self.consdata['host']
+        guest = self.consdata['guest']
+        pac = self.consdata['pac']  # fortunately, we terminate this on our end, but it does kind of reduce the value of the
+        # 'ticket' approach, as the general cookie must be provided as cookie along with the VNC ticket
+        self.ws.connect(f'wss://{self.bmc}:8006/api2/json/nodes/{host}/{guest}/vncwebsocket?port={port}&vncticket={urlticket}',
+                        host=bmc, cookie=f'PVEAuthCookie={pac}', # cookie='XSRF-TOKEN={0}; SESSION={1}'.format(wc.cookies['XSRF-TOKEN'], wc.cookies['SESSION']),
+                        subprotocols=['binary'])
+        self.ws.send(f'{user}:{ticket}\n')
+        data = self.ws.recv()
+        if data == b'OK':
+            self.ws.recv()  # swallow the 'starting serial terminal' message
+            self.connected = True
+            self.recvr = eventlet.spawn(self.recvdata)
+        else:
+            print(repr(data))
+        return
+
+    def write(self, data):
+        try:
+            dlen = str(len(data))
+            data = data.decode()
+            self.ws.send('0:' + dlen + ':' + data)
+        except websocket.WebSocketConnectionClosedException:
+            self.datacallback(conapi.ConsoleEvent.Disconnect)
+
+    def close(self):
+        if self.recvr:
+            self.recvr.kill()
+            self.recvr = None
+        if self.ws:
+            self.ws.close()
+        self.connected = False
+        self.datacallback = None
 
 class PmxApiClient:
     def __init__(self, server, user, password, configmanager):
         self.user = user
         self.password = password
+        self.pac = None
         if configmanager:
             cv = util.TLSCertVerifier(
                 configmanager, server, 'pubkeys.tls'
@@ -40,6 +172,7 @@ class PmxApiClient:
             self.password = self.password.decode()
         except Exception:
             pass
+        self.server = server
         self.wc = webclient.SecureHTTPConnection(server, port=8006, verifycallback=cv)
         self.vmmap = {}
         self.login()
@@ -54,6 +187,7 @@ class PmxApiClient:
         loginbody = urlparse.urlencode(loginform)
         rsp = self.wc.grab_json_response_with_status('/api2/json/access/ticket', loginbody)
         self.wc.cookies['PVEAuthCookie'] = rsp[0]['data']['ticket']
+        self.pac = rsp[0]['data']['ticket']
         self.wc.set_header('CSRFPreventionToken', rsp[0]['data']['CSRFPreventionToken'])
 
 
@@ -111,8 +245,14 @@ class PmxApiClient:
     def get_vm_serial(self, vm):
         # This would be termproxy
         # Example url
-        #wss:///api2/json/nodes/{host}/{guest}/vncwebsocket?port=5900&vncticket=URLENCODEDTICKET
-        raise Exception('TODO')
+        host, guest = self.get_vm(vm)
+        rsp = self.wc.grab_json_response_with_status(f'/api2/json/nodes/{host}/{guest}/termproxy', method='POST')
+        consdata = rsp[0]['data']
+        consdata['server'] = self.server
+        consdata['host'] = host
+        consdata['guest'] = guest
+        consdata['pac'] = self.pac
+        return consdata
 
     def get_vm_bootdev(self, vm):
         host, guest = self.get_vm(vm)
@@ -238,10 +378,10 @@ def update(nodes, element, configmanager, inputdata):
 
 # assume this is only console for now
 def create(nodes, element, configmanager, inputdata):
-    clientsbynode = prep_vcsa_clients(nodes, configmanager)
+    clientsbynode = prep_proxmox_clients(nodes, configmanager)
     for node in nodes:
         serialdata = clientsbynode[node].get_vm_serial(node)
-        return VmConsole(serialdata['server'], serialdata['port'], serialdata['tls'])
+        return PmxConsole(serialdata, node, configmanager, clientsbynode[node])