Unable to connect to vncwebsocket endpoint via API-TOKEN

[pandada8]

I used to have some code interactive with vncwebsocket via the following procedure. It's used to connect VM serial directly to the user terminal.

1. generate a PVEVNC ticket via /api2/json/nodes/{node}/qemu/{vmid}/termproxy
2. Submit Ticket and user to wss://<host>:8006/api2/json/nodes/{node}/qemu/{vmid}/vncwebsocket

When switching to API Token. I found that you can still generate PVEVNC Ticket from step1, but Unable to connect the websocket endpoint.
During further diagnosing, I find the following log from pvedaemon:

authentication failure; rhost=127.0.0.1 user=root@pam!<redacted> msg=value 'root@pam!<redacted>' does not look like a valid user name
command '/usr/bin/termproxy 5900 --path /vms/10020 --perm VM.Console -- /usr/sbin/qm terminal 10020 -escape 0 -iface serial0' failed: exit code 1

what's the correct way using vncwebsocket endpoint via api token ?

A small reproduce script is attached below.

"""
Install dependency by pip install proxmoxer websocket-client
"""
import proxmoxer
import websocket
import ssl
import sys
from urllib.parse import urlencode

endpoint = '<replace>'
vmid = 10020
serial='<replace>'
node = '<replace>'
opts = {"cert_reqs": ssl.CERT_NONE, "check_hostname": False}

p = proxmoxer.ProxmoxAPI(endpoint, user='<replace>', token_name="<replace>", token_value="<replace>", verify_ssl=False)
# p = proxmoxer.ProxmoxAPI(endpoint, user='<replace>', password='<replace>', verify_ssl=False)
data = p.nodes(node).qemu(vmid).termproxy.post(serial=serial)
query = {'vncticket': data['ticket'], 'port': data['port']}
extra_opts = {}
handshake = ''
if hasattr(p._backend.auth, 'token_name'):
    extra_opts['header'] = {'Authorization': f'PVEAPIToken={p._backend.auth.username}!{p._backend.auth.token_name}={p._backend.auth.token_value}'}
    # handshake = f'{p._backend.auth.username}:{data["ticket"]}\n'
    handshake = f'{p._backend.auth.username}!{p._backend.auth.token_name}:{data["ticket"]}\n'
else:
    extra_opts['cookie'] = '; '.join(['='.join(i) for i in p._backend.auth.get_cookies().items()])
    handshake = f'{p._backend.auth.username}:{data["ticket"]}\n'
ws = websocket.create_connection(f"wss://{endpoint}/api2/json/nodes/{node}/qemu/{vmid}/vncwebsocket?{urlencode(query)}", sslopt=opts, **extra_opts)
ws.send(handshake)
print(ws.recv())
# should be 'OK'
ws.close()

 

[navshimo]

hi! i manages to create a termproxy websocket session successfully. i hope it can help you and the others in the future (to be honest it took me one entire hellish week to finally get it to work).
disclaimer 1: i tried to use tokens and it didn't work. i tried whatever i could think of.
disclaimer 2: i used aoihttp instead of websocket and websockets modules, so that i can monitor and debug the traffic using burp suit. you should remove the lines related to proxy. lines: 11, 12, 22, 23, 56 and 64.

import os
import ssl
import warnings
import asyncio
from dotenv import load_dotenv
from proxmoxer import ProxmoxAPI
import aiohttp
from time import sleep


os.environ['HTTP_PROXY'] = 'http://localhost:8080'
os.environ['HTTPS_PROXY'] = 'http://localhost:8080'

load_dotenv(dotenv_path='.env')
vmid = '310'

warnings.filterwarnings(action="ignore")
node_name: str = os.environ['PROXMOX_NODE']
proxmox_host: str = os.environ['PROXMOX_SERVER']
proxmox_user: str = os.environ['PROXMOX_USERNAME']
proxmox_password = os.environ['PROXMOX_PASSWORD']
proxy_host: str = os.environ['PROXY_HOST']  # Replace with your proxy host
proxy_port: int = int(os.environ['PROXY_PORT'])  # Replace with your proxy port

proxmox = ProxmoxAPI(
    host=proxmox_host,
    user=proxmox_user,
    password=proxmox_password,
    verify_ssl=False,
)

termproxy = proxmox.nodes(node_name).lxc(vmid).termproxy.post()

ticket: str = termproxy.get('ticket')
port: str = termproxy.get('port')

proxmox.nodes.get()
cookie, _ = proxmox.get_tokens()


async def connect_websocket():
    headers = {
        'Authorization': f'PVEAPIToken={proxmox_user}',
        'Sec-WebSocket-Protocol': 'bianry',
        'Pragma': 'no-cache',
        'Cache-Control': 'no-cache',
        'Cookie': f'PVEAuthCookie={cookie}'
    }

    ssl_defaults = ssl.get_default_verify_paths()
    ssl_context = ssl.create_default_context(cafile=ssl_defaults.cafile)
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE

    websocket_url = f'wss://{proxmox_host}:8006/api2/json/nodes/{node_name}/lxc/{vmid}/vncwebsocket'
    proxy_server = f"http://{proxy_host}:{proxy_port}"
    query_params = {'port': port, 'vncticket': ticket}

    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(
            f'{websocket_url}',
            headers=headers,
            ssl=ssl_context,
            proxy=proxy_server,
            params=query_params,
        ) as websocket:
            try:
                # Send initial message after the connection is established
                await websocket.send_str(f'{proxmox_user}:{ticket}\n')
                # Read and print incoming messages
                response = await websocket.receive()
                print(f"Received response: {response.data}")

                # based on the monitored traffic (burp suite) the following should be sent to server:
                await websocket.send_str('1:86:24:')
                   
                # now do whatever you want!!
            except aiohttp.ClientError as e:
                print(f"WebSocket connection error: {e}")
            except asyncio.CancelledError:
                print("WebSocket connection cancelled.")

loop = asyncio.get_event_loop()
loop.run_until_complete(connect_websocket())