#!/usr/bin/env python3 """RDP session monitor — passive RDP connection tracking. Parses RDP Connection Request (Cookie field containing username), CredSSP/NLA TSRequest structures for NTLM domain\\username extraction, and tracks session patterns to identify admin workstations. """ import logging import os import socket import sqlite3 import struct import threading import time from collections import defaultdict from pathlib import Path from typing import Optional from modules.base import BaseModule logger = logging.getLogger("sensor.passive.rdp_monitor") # NTLM signature for CredSSP extraction NTLMSSP_SIGNATURE = b'NTLMSSP\x00' NTLM_AUTHENTICATE = 3 # X.224 Connection Request type X224_CONNECTION_REQUEST = 0xE0 # Admin workstation threshold ADMIN_DEST_THRESHOLD = 3 _DB_INIT = """ CREATE TABLE IF NOT EXISTS rdp_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, timestamp REAL NOT NULL, source_ip TEXT NOT NULL, dest_ip TEXT NOT NULL, username TEXT DEFAULT '', domain TEXT DEFAULT '', client_hostname TEXT DEFAULT '', keyboard_layout TEXT DEFAULT '' ); CREATE INDEX IF NOT EXISTS idx_rdp_src ON rdp_sessions(source_ip); CREATE INDEX IF NOT EXISTS idx_rdp_user ON rdp_sessions(username); CREATE INDEX IF NOT EXISTS idx_rdp_ts ON rdp_sessions(timestamp); """ FLUSH_INTERVAL = 30 class RDPMonitor(BaseModule): """Monitor RDP sessions passively for user/host identification.""" name = "rdp_monitor" module_type = "passive" priority = 150 requires_root = True requires_capture_bus = True def __init__(self, bus, state, config, engine=None): super().__init__(bus, state, config, engine) self._capture_bus = None self._sub_queue = None self._read_thread: Optional[threading.Thread] = None self._flush_thread: Optional[threading.Thread] = None self._db_path = self._resolve_db_path() self._db_conn: Optional[sqlite3.Connection] = None self._write_buffer: list[tuple] = [] self._buffer_lock = threading.Lock() # Track admin patterns: src_ip -> set of dest_ips self._rdp_targets: dict[str, set] = defaultdict(set) self._stats = { "packets_processed": 0, "connection_requests": 0, "credssp_auths": 0, "unique_sessions": 0, "admin_workstations": 0, } # ------------------------------------------------------------------ # BaseModule interface # ------------------------------------------------------------------ def start(self) -> None: if self._running: return self._init_db() self._capture_bus = self.config.get("_capture_bus") if self._capture_bus is None: logger.error("RDPMonitor requires _capture_bus in config") return self._sub_queue = self._capture_bus.subscribe( name=self.name, bpf_filter="port 3389", queue_depth=3000 ) self._running = True self._start_time = time.time() self._pid = os.getpid() self._read_thread = threading.Thread( target=self._read_loop, daemon=True, name="sensor-rdp-read" ) self._read_thread.start() self._flush_thread = threading.Thread( target=self._flush_loop, daemon=True, name="sensor-rdp-flush" ) self._flush_thread.start() self.state.set_module_status(self.name, "running", pid=self._pid) self.bus.emit("MODULE_STARTED", {"module": self.name}, source_module=self.name) logger.info("RDPMonitor started") def stop(self) -> None: if not self._running: return self._running = False if self._capture_bus: self._capture_bus.unsubscribe(self.name) for t in (self._read_thread, self._flush_thread): if t and t.is_alive(): t.join(timeout=5.0) self._flush_buffer() if self._db_conn: self._db_conn.close() self._db_conn = None self.state.set_module_status(self.name, "stopped") self.bus.emit("MODULE_STOPPED", {"module": self.name}, source_module=self.name) logger.info("RDPMonitor stopped — stats: %s", self._stats) def status(self) -> dict: return { "running": self._running, "pid": self._pid, "uptime": time.time() - self._start_time if self._start_time else 0, **self._stats, } def configure(self, config: dict) -> None: self.config.update(config) # ------------------------------------------------------------------ # Packet processing # ------------------------------------------------------------------ def _read_loop(self) -> None: while self._running: result = self._sub_queue.get(timeout=1.0) if result is None: continue ts, pkt = result self._stats["packets_processed"] += 1 try: self._process_packet(pkt, ts) except Exception: logger.debug("Error processing RDP packet", exc_info=True) def _process_packet(self, pkt: bytes, ts: float) -> None: """Extract TCP payload and dispatch to RDP parsers.""" if len(pkt) < 54: return ethertype = struct.unpack("!H", pkt[12:14])[0] eth_offset = 14 if ethertype == 0x8100: if len(pkt) < 58: return ethertype = struct.unpack("!H", pkt[16:18])[0] eth_offset = 18 if ethertype != 0x0800: return ip_header = pkt[eth_offset:] if len(ip_header) < 20: return ihl = (ip_header[0] & 0x0F) * 4 ip_proto = ip_header[9] if ip_proto != 6: return src_ip = socket.inet_ntoa(ip_header[12:16]) dst_ip = socket.inet_ntoa(ip_header[16:20]) if len(ip_header) < ihl + 20: return src_port, dst_port = struct.unpack("!HH", ip_header[ihl:ihl + 4]) tcp_data_off = ((ip_header[ihl + 12] >> 4) & 0xF) * 4 payload = ip_header[ihl + tcp_data_off:] if len(payload) < 6: return # Only process traffic going TO RDP port (client -> server) if dst_port == 3389: self._parse_rdp_client(payload, src_ip, dst_ip, ts) def _parse_rdp_client(self, payload: bytes, src_ip: str, dst_ip: str, ts: float) -> None: """Parse client-to-server RDP traffic.""" # Check for TPKT header (version=3, reserved=0) if payload[0] == 0x03 and payload[1] == 0x00: self._parse_tpkt(payload, src_ip, dst_ip, ts) return # Check for NTLMSSP in CredSSP if NTLMSSP_SIGNATURE in payload: self._parse_credssp_ntlm(payload, src_ip, dst_ip, ts) def _parse_tpkt(self, payload: bytes, src_ip: str, dst_ip: str, ts: float) -> None: """Parse TPKT + X.224 Connection Request for username cookie.""" if len(payload) < 11: return # TPKT: version(1) + reserved(1) + length(2) tpkt_len = struct.unpack("!H", payload[2:4])[0] # X.224: length indicator(1) + type(1) x224_len = payload[4] x224_type = payload[5] & 0xF0 if x224_type != X224_CONNECTION_REQUEST: return self._stats["connection_requests"] += 1 # After X.224 header (variable length), look for Cookie field # Cookie format: "Cookie: mstshash=username\r\n" cookie_data = payload[11:] # Skip past basic X.224 header username = "" client_hostname = "" # Search for mstshash cookie cookie_idx = cookie_data.find(b'Cookie: mstshash=') if cookie_idx >= 0: cookie_start = cookie_idx + 17 # len("Cookie: mstshash=") cookie_end = cookie_data.find(b'\r\n', cookie_start) if cookie_end > cookie_start: username = cookie_data[cookie_start:cookie_end].decode( "ascii", errors="replace" ).strip() if username: self._rdp_targets[src_ip].add(dst_ip) self._stats["unique_sessions"] += 1 self._check_admin_workstation(src_ip) self._record_session(ts, src_ip, dst_ip, username, "", "", "") def _parse_credssp_ntlm(self, payload: bytes, src_ip: str, dst_ip: str, ts: float) -> None: """Extract NTLM credentials from CredSSP TSRequest.""" idx = payload.find(NTLMSSP_SIGNATURE) if idx < 0 or idx + 12 > len(payload): return ntlm_data = payload[idx:] msg_type = struct.unpack(" tuple: """Parse NTLM Type 3 Authenticate for username, domain, workstation.""" if len(data) < 72: return ("", "", "") try: domain_len = struct.unpack(" None: """Check if source IP qualifies as admin workstation.""" targets = self._rdp_targets.get(src_ip, set()) if len(targets) >= ADMIN_DEST_THRESHOLD: current = len([ ip for ip, dests in self._rdp_targets.items() if len(dests) >= ADMIN_DEST_THRESHOLD ]) if current > self._stats["admin_workstations"]: self._stats["admin_workstations"] = current logger.info( "Admin workstation detected: %s (RDP to %d hosts)", src_ip, len(targets) ) # ------------------------------------------------------------------ # Database # ------------------------------------------------------------------ def _resolve_db_path(self) -> str: base = self.config.get("db_dir", os.path.join( os.path.expanduser("~"), ".bigbrother" )) return os.path.join(base, "rdp_monitor.db") def _init_db(self) -> None: Path(self._db_path).parent.mkdir(parents=True, exist_ok=True) self._db_conn = sqlite3.connect(self._db_path, check_same_thread=False) self._db_conn.execute("PRAGMA journal_mode=WAL") self._db_conn.execute("PRAGMA synchronous=NORMAL") self._db_conn.executescript(_DB_INIT) self._db_conn.commit() def _record_session(self, ts: float, src_ip: str, dst_ip: str, username: str, domain: str, client_hostname: str, keyboard_layout: str) -> None: with self._buffer_lock: self._write_buffer.append(( ts, src_ip, dst_ip, username, domain, client_hostname, keyboard_layout )) def _flush_buffer(self) -> None: with self._buffer_lock: batch = list(self._write_buffer) self._write_buffer.clear() if not batch or not self._db_conn: return try: with self._db_conn: self._db_conn.executemany( """INSERT INTO rdp_sessions (timestamp, source_ip, dest_ip, username, domain, client_hostname, keyboard_layout) VALUES (?, ?, ?, ?, ?, ?, ?)""", batch, ) except Exception: logger.exception("Failed to flush RDP sessions") def _flush_loop(self) -> None: while self._running: time.sleep(FLUSH_INTERVAL) try: self._flush_buffer() except Exception: logger.exception("Flush loop error")