WIP: protocol abstraction

Work on abstracting the CollabVMServer so it now calls into a interface for sending/recieving protocol messages. This will allow cleaner bringup of a fully binary protocol, and generally is just cleaner code.

Mostly everything is parsd/running through this new layer, although there are some TODO items:

- NetworkClient/... should just spit out a Buffer or something that eventually turns into or has one
- TCP protocol will need to be revamped so we can support an actual binary protocol on top of it. The current thing is line based
- More admin op stuff needs to be handled
- The handlers are a bit jumbled around atm
- There is still a good amount of code which assumes guacamole which needs to be rewritten

dont use this branch fuckers

cvmts/src/BinRectsProtocol.ts

@@ -0,0 +1,6 @@
+import * as msgpack from 'msgpackr';
+import { CollabVMProtocolMessage, CollabVMProtocolMessageType } from '@cvmts/collab-vm-1.2-binary-protocol';
+
+// TODO: reimplement binrects protocol
+// we can just create/proxy a GuacamoleProtocol manually,
+// and for the rects do our own thing

cvmts/src/GuacamoleProtocol.ts

@@ -0,0 +1,282 @@
+import pino from 'pino';
+import { IProtocol, IProtocolHandlers, ListEntry, ProtocolAddUser, ProtocolChatHistory, ScreenRect } from './Protocol';
+import { User } from './User';
+
+import * as cvm from '@cvmts/cvm-rs';
+
+// CollabVM protocol implementation for Guacamole.
+export class GuacamoleProtocol implements IProtocol {
+	private handlers: IProtocolHandlers | null = null;
+	private logger = pino({
+		name: 'CVMTS.GuacamoleProtocol'
+	});
+
+	private user: User | null = null;
+
+	init(u: User): void {
+		this.user = u;
+	}
+
+	setHandler(handlers: IProtocolHandlers): void {
+		this.handlers = handlers;
+	}
+
+	private __processMessage_admin(decodedElements: string[]): boolean {
+		switch (decodedElements[1]) {
+			case '2':
+				// Login
+				if (decodedElements.length !== 3) return false;
+				this.handlers?.onAdminLogin(this.user!, decodedElements[2]);
+				break;
+			case '5':
+				// QEMU Monitor
+				if (decodedElements.length !== 4) return false;
+				// [2] node
+				// [3] cmd
+				break;
+			case '8':
+				// Restore
+				break;
+			case '10':
+				// Reboot
+				if (decodedElements.length !== 3) return false;
+				// [2] - node
+				break;
+			case '12':
+			// Ban
+
+			case '13':
+				// Force Vote
+				if (decodedElements.length !== 3) return false;
+
+				break;
+			case '14':
+				// Mute
+				if (decodedElements.length !== 4) return false;
+				break;
+			case '15':
+			// Kick
+			case '16':
+				// End turn
+				if (decodedElements.length !== 3) return false;
+				break;
+			case '17':
+				// Clear turn queue
+				if (decodedElements.length !== 3) return false;
+				// [2] - node
+				break;
+			case '18':
+				// Rename user
+				if (decodedElements.length !== 4) return false;
+
+				// [2] - username
+				// [3] - new username
+				break;
+			case '19':
+				// Get IP
+				if (decodedElements.length !== 3) return false;
+				break;
+			case '20':
+				// Steal turn
+				break;
+			case '21':
+				// XSS
+				if (decodedElements.length !== 3) return false;
+				// [2] message
+				break;
+			case '22':
+				// Toggle turns
+				if (decodedElements.length !== 3) return false;
+				// [2] 0 == disable 1 == enable
+				break;
+			case '23':
+				// Indefinite turn
+				break;
+			case '24':
+				// Hide screen
+
+				if (decodedElements.length !== 3) return false;
+				// 0 - hide
+				// 1 - unhide
+
+				break;
+			case '25':
+				if (decodedElements.length !== 3) return false;
+				// [2]
+				break;
+		}
+		return true;
+	}
+
+	processMessage(buffer: Buffer): boolean {
+		let decodedElements = cvm.guacDecode(buffer.toString('utf-8'));
+		if (decodedElements.length < 1) return false;
+
+		// The first element is the "opcode".
+		switch (decodedElements[0]) {
+			case 'nop':
+				this.handlers?.onNop(this.user!);
+				break;
+			case 'cap':
+				if (decodedElements.length < 2) return false;
+				this.handlers?.onCapabilityUpgrade(this.user!, decodedElements.slice(1));
+				break;
+			case 'login':
+				if (decodedElements.length !== 2) return false;
+				this.handlers?.onLogin(this.user!, decodedElements[1]);
+				break;
+			case 'noflag':
+				this.handlers?.onNoFlag(this.user!);
+				break;
+			case 'list':
+				this.handlers?.onList(this.user!);
+				break;
+			case 'connect':
+				if (decodedElements.length !== 2) return false;
+				this.handlers?.onConnect(this.user!, decodedElements[1]);
+				break;
+			case 'view':
+				{
+					if (decodedElements.length !== 3) return false;
+					let viewMode = parseInt(decodedElements[2]);
+					if (viewMode == undefined) return false;
+
+					this.handlers?.onView(this.user!, decodedElements[1], viewMode);
+				}
+				break;
+			case 'rename':
+				this.handlers?.onRename(this.user!, decodedElements[1]);
+				break;
+			case 'chat':
+				if (decodedElements.length !== 2) return false;
+				this.handlers?.onChat(this.user!, decodedElements[1]);
+				break;
+			case 'turn':
+				let forfeit = false;
+				if (decodedElements.length > 2) return false;
+				if (decodedElements.length == 1) {
+					 forfeit = false;
+				} else {
+					if (decodedElements[1] == '0') forfeit = true;
+					else if (decodedElements[1] == '1') forfeit = false;
+				}
+
+				this.handlers?.onTurnRequest(this.user!, forfeit);
+				break;
+			case 'mouse':
+				if (decodedElements.length !== 4) return false;
+
+				let x = parseInt(decodedElements[1]);
+				let y = parseInt(decodedElements[2]);
+				let mask = parseInt(decodedElements[3]);
+				if (x === undefined || y === undefined || mask === undefined) return false;
+
+				this.handlers?.onMouse(this.user!, x, y, mask);
+				break;
+			case 'key':
+				if (decodedElements.length !== 3) return false;
+				var keysym = parseInt(decodedElements[1]);
+				var down = parseInt(decodedElements[2]);
+				if (keysym === undefined || (down !== 0 && down !== 1)) return false;
+				this.handlers?.onKey(this.user!, keysym, down === 1);
+				break;
+			case 'vote':
+				if (decodedElements.length !== 2) return false;
+				let choice = parseInt(decodedElements[1]);
+				if (choice == undefined) return false;
+				this.handlers?.onVote(this.user!, choice);
+				break;
+
+			case 'admin':
+				if (decodedElements.length < 2) return false;
+				return this.__processMessage_admin(decodedElements);
+		}
+
+		return true;
+	}
+
+	// Senders
+
+	sendAuth(authServer: string): void {
+		this.user?.sendMsg(cvm.guacEncode('auth', authServer));
+	}
+
+	sendNop(): void {
+		this.user?.sendMsg(cvm.guacEncode('nop'));
+	}
+
+	sendSync(now: number): void {
+		this.user?.sendMsg(cvm.guacEncode('sync', now.toString()));
+	}
+
+	sendConnectFailResponse(): void {
+		this.user?.sendMsg(cvm.guacEncode('connect', '0'));
+	}
+
+	sendConnectOKResponse(votes: boolean): void {
+		this.user?.sendMsg(cvm.guacEncode('connect', '1', '1', votes ? '1' : '0', '0'));
+	}
+
+	sendLoginResponse(ok: boolean, message: string | undefined): void {
+		if (ok) {
+			this.user?.sendMsg(cvm.guacEncode('login', '1'));
+			return;
+		} else {
+			this.user?.sendMsg(cvm.guacEncode('login', '0', message!));
+		}
+	}
+
+	sendChatMessage(username: string, message: string): void {
+		this.user?.sendMsg(cvm.guacEncode('chat', username, message));
+	}
+
+	sendChatHistoryMessage(history: ProtocolChatHistory[]): void {
+		let arr = ['chat'];
+		for (let a of history) {
+			arr.push(a.user);
+			arr.push(a.msg);
+		}
+
+		this.user?.sendMsg(cvm.guacEncode(...arr));
+	}
+
+	sendAddUser(users: ProtocolAddUser[]): void {
+		let arr = ['adduser', users.length.toString()];
+		for (let user of users) {
+			arr.push(user.username);
+			arr.push(user.rank.toString());
+		}
+
+		this.user?.sendMsg(cvm.guacEncode(...arr));
+	}
+
+	sendRemUser(users: string[]): void {
+		let arr = ['remuser', users.length.toString()];
+
+		for (let user of users) {
+			arr.push(user);
+		}
+
+		this.user?.sendMsg(cvm.guacEncode(...arr));
+	}
+
+	sendListResponse(list: ListEntry[]): void {
+		let arr = ['list'];
+		for (let node of list) {
+			arr.push(node.id);
+			arr.push(node.name);
+			arr.push(node.thumbnail.toString('base64'));
+		}
+
+		this.user?.sendMsg(cvm.guacEncode(...arr));
+	}
+
+	sendScreenResize(width: number, height: number): void {
+		this.user?.sendMsg(cvm.guacEncode('size', '0', width.toString(), height.toString()));
+	}
+
+	sendScreenUpdate(rect: ScreenRect): void {
+		this.user?.sendMsg(cvm.guacEncode('png', '0', '0', rect.x.toString(), rect.y.toString(), rect.data.toString('base64')));
+		this.sendSync(Date.now());
+	}
+}

cvmts/src/Protocol.ts

@@ -0,0 +1,119 @@
+import { Rank, User } from './User';
+
+// We should probably put this in the binproto repository or something
+enum UpgradeCapability {
+	Binary = 'bin'
+}
+
+export interface ScreenRect {
+	x: number;
+	y: number;
+	data: Buffer;
+}
+
+export interface ListEntry {
+	id: string;
+	name: string;
+	thumbnail: Buffer;
+}
+
+export interface ProtocolChatHistory {
+	user: string;
+	msg: string;
+}
+
+export interface ProtocolAddUser {
+	username: string;
+	rank: Rank;
+}
+
+// Protocol handlers. This is implemented by a layer that wants to listen to CollabVM protocol messages.
+export interface IProtocolHandlers {
+	onNop(user: User): void;
+
+	onNoFlag(user: User): void;
+
+	// Called when the client requests a capability upgrade
+	onCapabilityUpgrade(user: User, capability: Array<String>): boolean;
+
+	onLogin(user: User, token: string): void;
+
+	// Called on turn request
+	onTurnRequest(user: User, forfeit: boolean): void;
+
+	onVote(user: User, choice: number): void;
+
+	onList(user: User): void;
+	onConnect(user: User, node: string): void;
+	onView(user: User, node: string, viewMode: number): void;
+
+	onAdminLogin(user: User, password: string): void;
+	onAdminMonitor(user: User, node: string, command: string): void;
+
+	onRename(user: User, newName: string|undefined): void;
+	onChat(user: User, message: string): void;
+
+	onKey(user: User, keysym: number, pressed: boolean): void;
+	onMouse(user: User, x: number, y: number, buttonMask: number): void;
+}
+
+// Abstracts away all of the CollabVM protocol details
+export interface IProtocol {
+	init(u: User): void;
+
+	// Sets handler object.
+	setHandler(handlers: IProtocolHandlers): void;
+
+	// Parses a single CollabVM protocol message and fires the given handler.
+	// This function does not catch any thrown errors; it is the caller's responsibility
+	// to handle errors. It should, however, catch invalid parameters without failing.
+	processMessage(buffer: Buffer): boolean;
+
+	// Senders
+
+	sendNop(): void;
+	sendSync(now: number): void;
+
+	sendAuth(authServer: string): void;
+
+	sendConnectFailResponse(): void;
+	sendConnectOKResponse(votes: boolean): void;
+
+	sendLoginResponse(ok: boolean, message: string | undefined): void;
+
+	sendChatMessage(username: '' | string, message: string): void;
+	sendChatHistoryMessage(history: ProtocolChatHistory[]): void;
+
+	sendAddUser(users: ProtocolAddUser[]): void;
+	sendRemUser(users: string[]): void;
+
+	sendListResponse(list: ListEntry[]): void;
+
+	sendScreenResize(width: number, height: number): void;
+
+	// Sends a rectangle update to the user.
+	sendScreenUpdate(rect: ScreenRect): void;
+}
+
+// Holds protocol factories.
+export class ProtocolManager {
+	private protocols = new Map<String, () => IProtocol>();
+
+	// Registers a protocol with the given name.
+	registerProtocol(name: string, protocolFactory: () => IProtocol) {
+		if (!this.protocols.has(name)) this.protocols.set(name, protocolFactory);
+	}
+
+	// Creates an instance of a given protocol for a user.
+	createProtocol(name: string, user: User): IProtocol {
+		if (!this.protocols.has(name)) throw new Error(`ProtocolManager does not have protocol \"${name}\"`);
+
+		let factory = this.protocols.get(name)!;
+		let proto = factory();
+		proto.init(user);
+		return proto;
+	}
+}
+
+/// Global protocol manager
+export let TheProtocolManager = new ProtocolManager();

cvmts/src/TCP/TCPServer.ts

@@ -1,3 +1,6 @@
+// TODO: replace tcp protocol with smth like
+// struct msg { beu32 len; char data[len] }
+// (along with a length cap obviously)
 import EventEmitter from 'events';
 import NetworkServer from '../NetworkServer.js';
 import { Server, Socket } from 'net';

cvmts/src/User.ts

@@ -8,6 +8,7 @@ import NetworkClient from './NetworkClient.js';
 import { CollabVMCapabilities } from '@cvmts/collab-vm-1.2-binary-protocol';
 import pino from 'pino';
 import { BanManager } from './BanManager.js';
+import { IProtocol, TheProtocolManager } from './Protocol.js';
 
 export class User {
 	socket: NetworkClient;
@@ -22,6 +23,7 @@ export class User {
 	Config: IConfig;
 	IP: IPData;
 	Capabilities: CollabVMCapabilities;
+	protocol: IProtocol;
 	turnWhitelist: boolean = false;
 	// Hide flag. Only takes effect if the user is logged in.
 	noFlag: boolean = false;
@@ -44,6 +46,9 @@ export class User {
 		this.msgsSent = 0;
 		this.Capabilities = new CollabVMCapabilities();
 
+		// All clients default to the Guacamole protocol.
+		this.protocol = TheProtocolManager.createProtocol('guacamole', this);
+
 		this.socket.on('disconnect', () => {
 			// Unref the ip data for this connection
 			this.IP.Unref();
@@ -52,11 +57,6 @@ export class User {
 			clearInterval(this.msgRecieveInterval);
 		});
 
-		this.socket.on('msg', (e) => {
-			clearTimeout(this.nopRecieveTimeout);
-			clearInterval(this.msgRecieveInterval);
-			this.msgRecieveInterval = setInterval(() => this.onNoMsg(), 10000);
-		});
 
 		this.nopSendInterval = setInterval(() => this.sendNop(), 5000);
 		this.msgRecieveInterval = setInterval(() => this.onNoMsg(), 10000);
@@ -84,8 +84,14 @@ export class User {
 		return username;
 	}
 
+	onNop() {
+		clearTimeout(this.nopRecieveTimeout);
+		clearInterval(this.msgRecieveInterval);
+		this.msgRecieveInterval = setInterval(() => this.onNoMsg(), 10000);
+	}
+
 	sendNop() {
-		this.socket.send('3.nop;');
+		this.protocol.sendNop();
 	}
 
 	sendMsg(msg: string) {
@@ -107,7 +113,7 @@ export class User {
 		this.socket.close();
 	}
 
-	onMsgSent() {
+	onChatMsgSent() {
 		if (!this.Config.collabvm.automute.enabled) return;
 		// rate limit guest and unregistered chat messages, but not staff ones
 		switch (this.rank) {
@@ -153,5 +159,5 @@ export enum Rank {
 	// After all these years
 	Registered = 1,
 	Admin = 2,
-	Moderator = 3,
+	Moderator = 3
 }

cvmts/src/index.ts

@@ -16,6 +16,8 @@ import pino from 'pino';
 import { Database } from './Database.js';
 import { BanManager } from './BanManager.js';
 import { QemuVMShim } from './vm/qemu.js';
+import { TheProtocolManager } from './Protocol.js';
+import { GuacamoleProtocol } from './GuacamoleProtocol.js';
 
 let logger = pino();
 
@@ -97,17 +99,20 @@ async function start() {
 	process.on('SIGINT', async () => await stop());
 	process.on('SIGTERM', async () => await stop());
 
+	// Register protocol(s)
+	TheProtocolManager.registerProtocol("guacamole", () => new GuacamoleProtocol);
+
 	await VM.Start();
 	// Start up the server
 	var CVM = new CollabVMServer(Config, VM, banmgr, auth, geoipReader);
 
 	var WS = new WSServer(Config, banmgr);
-	WS.on('connect', (client: User) => CVM.addUser(client));
+	WS.on('connect', (client: User) => CVM.connectionOpened(client));
 	WS.start();
 
 	if (Config.tcp.enabled) {
 		var TCP = new TCPServer(Config, banmgr);
-		TCP.on('connect', (client: User) => CVM.addUser(client));
+		TCP.on('connect', (client: User) => CVM.connectionOpened(client));
 		TCP.start();
 	}
 }