import json import logging import time from pathlib import Path class CrashProtocol: """ BC.Game Crash protocol layer. Handles: WebSocket Engine.IO v3 Socket.IO framing /g/cm namespace Text packets Binary packets Raw packet preservation The binary payload is deliberately preserved rather than inventing field meanings. Once the exact protobuf-style field map is established from the captured traffic, the decoder can be extended without changing the rest of the system. """ ENGINE_OPEN = "0" ENGINE_CLOSE = "1" ENGINE_PING = "2" ENGINE_PONG = "3" ENGINE_MESSAGE = "4" SOCKET_CONNECT = "0" SOCKET_DISCONNECT = "1" SOCKET_EVENT = "2" SOCKET_ACK = "3" SOCKET_ERROR = "4" SOCKET_BINARY_EVENT = "5" SOCKET_BINARY_ACK = "6" def __init__(self, config_path="config.json"): self.logger = logging.getLogger("Protocol") self.config = self._load_config( config_path ) websocket_config = self.config.get( "websocket", {} ) self.namespace = websocket_config.get( "namespace", "/g/cm" ) capture_config = self.config.get( "capture", {} ) self.save_raw_messages = capture_config.get( "save_raw_messages", True ) self.data_dir = Path( capture_config.get( "data_directory", "data" ) ) self.data_dir.mkdir( parents=True, exist_ok=True ) self.raw_file = ( self.data_dir / "protocol_messages.jsonl" ) self.binary_file = ( self.data_dir / "binary_messages.jsonl" ) self.message_count = 0 self.binary_count = 0 self.event_count = 0 # ======================================================== # CONFIGURATION # ======================================================== def _load_config(self, path): with open( path, "r", encoding="utf-8" ) as file: return json.load(file) # ======================================================== # MAIN PARSER # ======================================================== def parse(self, message): self.message_count += 1 if message is None: return { "type": "empty", "raw": None } if isinstance( message, bytes ): return self._parse_binary( message ) if isinstance( message, bytearray ): return self._parse_binary( bytes(message) ) if isinstance( message, str ): return self._parse_text( message ) return { "type": "unknown", "python_type": type(message).__name__, "timestamp": time.time() } # ======================================================== # ENGINE.IO / SOCKET.IO TEXT PARSER # ======================================================== def _parse_text(self, message): timestamp = time.time() if not message: return { "type": "empty", "format": "text", "raw": "", "timestamp": timestamp } packet = { "transport": "websocket", "format": "text", "raw": message, "timestamp": timestamp } # ---------------------------------------------------- # Engine.IO OPEN # ---------------------------------------------------- if message.startswith( self.ENGINE_OPEN ): packet["type"] = "engine_open" packet["engine_io_packet"] = ( self.ENGINE_OPEN ) payload = message[1:] packet["payload_raw"] = payload try: payload_json = json.loads( payload ) packet["payload"] = ( payload_json ) if isinstance( payload_json, dict ): packet["sid"] = ( payload_json.get( "sid" ) ) packet["upgrades"] = ( payload_json.get( "upgrades" ) ) packet["ping_interval"] = ( payload_json.get( "pingInterval" ) ) packet["ping_timeout"] = ( payload_json.get( "pingTimeout" ) ) except json.JSONDecodeError: packet["payload"] = ( payload ) self._store( packet ) return packet # ---------------------------------------------------- # Engine.IO CLOSE # ---------------------------------------------------- if message.startswith( self.ENGINE_CLOSE ): packet["type"] = "engine_close" packet["engine_io_packet"] = ( self.ENGINE_CLOSE ) packet["payload"] = ( message[1:] ) self._store( packet ) return packet # ---------------------------------------------------- # Engine.IO PING # ---------------------------------------------------- if message.startswith( self.ENGINE_PING ): packet["type"] = "engine_ping" packet["engine_io_packet"] = ( self.ENGINE_PING ) packet["payload"] = ( message[1:] ) self._store( packet ) return packet # ---------------------------------------------------- # Engine.IO PONG # ---------------------------------------------------- if message.startswith( self.ENGINE_PONG ): packet["type"] = "engine_pong" packet["engine_io_packet"] = ( self.ENGINE_PONG ) packet["payload"] = ( message[1:] ) self._store( packet ) return packet # ---------------------------------------------------- # Engine.IO MESSAGE # ---------------------------------------------------- if message.startswith( self.ENGINE_MESSAGE ): packet["engine_io_packet"] = ( self.ENGINE_MESSAGE ) socket_payload = message[1:] packet["socket_payload"] = ( socket_payload ) return self._parse_socket_payload( packet, socket_payload ) # ---------------------------------------------------- # Unknown # ---------------------------------------------------- packet["type"] = "unknown" self._store( packet ) return packet # ======================================================== # SOCKET.IO PARSER # ======================================================== def _parse_socket_payload( self, packet, payload ): if not payload: packet["type"] = ( "engine_message_empty" ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO CONNECT # ---------------------------------------------------- if payload.startswith( self.SOCKET_CONNECT ): packet["type"] = ( "socket_namespace" ) packet["socket_io_packet"] = ( self.SOCKET_CONNECT ) namespace = payload[1:] packet["namespace"] = ( namespace ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO DISCONNECT # ---------------------------------------------------- if payload.startswith( self.SOCKET_DISCONNECT ): packet["type"] = ( "socket_disconnect" ) packet["socket_io_packet"] = ( self.SOCKET_DISCONNECT ) packet["namespace"] = ( payload[1:] ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO EVENT # ---------------------------------------------------- if payload.startswith( self.SOCKET_EVENT ): packet["type"] = ( "socket_event" ) packet["socket_io_packet"] = ( self.SOCKET_EVENT ) event_payload = payload[1:] packet.update( self._decode_socket_event( event_payload ) ) self.event_count += 1 self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO ACK # ---------------------------------------------------- if payload.startswith( self.SOCKET_ACK ): packet["type"] = ( "socket_ack" ) packet["socket_io_packet"] = ( self.SOCKET_ACK ) packet["payload"] = ( self._try_json( payload[1:] ) ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO ERROR # ---------------------------------------------------- if payload.startswith( self.SOCKET_ERROR ): packet["type"] = ( "socket_error" ) packet["socket_io_packet"] = ( self.SOCKET_ERROR ) packet["payload"] = ( self._try_json( payload[1:] ) ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO BINARY EVENT # ---------------------------------------------------- if payload.startswith( self.SOCKET_BINARY_EVENT ): packet["type"] = ( "socket_binary_event" ) packet["socket_io_packet"] = ( self.SOCKET_BINARY_EVENT ) binary_payload = payload[1:] packet.update( self._decode_binary_event_header( binary_payload ) ) self._store( packet ) return packet # ---------------------------------------------------- # Socket.IO BINARY ACK # ---------------------------------------------------- if payload.startswith( self.SOCKET_BINARY_ACK ): packet["type"] = ( "socket_binary_ack" ) packet["socket_io_packet"] = ( self.SOCKET_BINARY_ACK ) packet["payload"] = ( payload[1:] ) self._store( packet ) return packet # ---------------------------------------------------- # UNKNOWN SOCKET.IO PACKET # ---------------------------------------------------- packet["type"] = ( "unknown_socket_packet" ) packet["payload"] = ( payload ) self._store( packet ) return packet # ======================================================== # SOCKET.IO EVENT DECODER # ======================================================== def _decode_socket_event( self, payload ): result = { "event_raw": payload } if not payload: return result # Socket.IO normally contains JSON after # packet type 2. BC.Game may use a custom/binary # representation, so JSON decoding is attempted # but never assumed. decoded = self._try_json( payload ) if decoded is not None: result["payload"] = decoded if isinstance( decoded, list ): result["event_name"] = ( decoded[0] if decoded else None ) result["event_data"] = ( decoded[1:] ) elif isinstance( decoded, dict ): result["event_data"] = ( decoded ) return result # ---------------------------------------------------- # Non-JSON event # ---------------------------------------------------- result["binary_payload"] = ( payload.encode( "utf-8", errors="replace" ).hex() ) result["payload_encoding"] = ( "non_json_text" ) return result # ======================================================== # BINARY WEBSOCKET PARSER # ======================================================== def _parse_binary( self, message ): timestamp = time.time() self.binary_count += 1 packet = { "transport": "websocket", "format": "binary", "type": "binary", "timestamp": timestamp, "length": len(message), "hex": message.hex() } # ---------------------------------------------------- # Preserve byte information # ---------------------------------------------------- packet["first_byte"] = ( message[0] if message else None ) packet["first_bytes"] = ( message[:16].hex() ) packet["last_bytes"] = ( message[-16:].hex() if message else "" ) # ---------------------------------------------------- # Basic Engine.IO binary detection # ---------------------------------------------------- if message: first = message[0] packet["possible_engine_packet"] = ( first ) # ---------------------------------------------------- # Socket.IO binary-event marker # ---------------------------------------------------- if message.startswith( b"\x05" ): packet["type"] = ( "socket_binary_event" ) elif message.startswith( b"\x06" ): packet["type"] = ( "socket_binary_ack" ) else: packet["type"] = ( "binary_payload" ) # ---------------------------------------------------- # Binary structure inspection # ---------------------------------------------------- packet["byte_statistics"] = ( self._byte_statistics( message ) ) # ---------------------------------------------------- # Protobuf-style inspection # ---------------------------------------------------- packet["protobuf_candidates"] = ( self.inspect_protobuf( message ) ) self._store_binary( packet ) return packet # ======================================================== # PROTOBUF-STYLE FIELD INSPECTION # ======================================================== def inspect_protobuf( self, data ): """ Performs conservative protobuf-wire-format inspection. This does NOT claim that the payload is protobuf. It identifies byte sequences that are structurally compatible with common protobuf field encodings. Supported wire types: 0 = varint 1 = 64-bit 2 = length-delimited 5 = 32-bit """ if not data: return [] candidates = [] index = 0 maximum_fields = 100 while ( index < len(data) and len(candidates) < maximum_fields ): start = index try: key, key_length = ( self._read_varint( data, index ) ) except ValueError: break if key == 0: break field_number = ( key >> 3 ) wire_type = ( key & 0x07 ) if field_number <= 0: break index += key_length field = { "offset": start, "field_number": field_number, "wire_type": wire_type } try: if wire_type == 0: value, consumed = ( self._read_varint( data, index ) ) field["value"] = value field["encoding"] = ( "varint" ) index += consumed elif wire_type == 1: if index + 8 > len(data): break raw = data[ index:index + 8 ] field["raw"] = ( raw.hex() ) field["encoding"] = ( "fixed64" ) index += 8 elif wire_type == 2: length, consumed = ( self._read_varint( data, index ) ) index += consumed if ( length < 0 or index + length > len(data) ): break raw = data[ index: index + length ] field["length"] = ( length ) field["raw"] = ( raw.hex() ) field["utf8"] = ( self._safe_utf8( raw ) ) field["encoding"] = ( "length_delimited" ) index += length elif wire_type == 5: if index + 4 > len(data): break raw = data[ index:index + 4 ] field["raw"] = ( raw.hex() ) field["encoding"] = ( "fixed32" ) index += 4 else: # Groups / unsupported wire types. break except ValueError: break candidates.append( field ) return candidates # ======================================================== # VARINT DECODER # ======================================================== def _read_varint( self, data, offset ): value = 0 shift = 0 for index in range( offset, min( offset + 10, len(data) ) ): byte = data[index] value |= ( (byte & 0x7F) << shift ) if not ( byte & 0x80 ): return ( value, index - offset + 1 ) shift += 7 raise ValueError( "Invalid varint" ) # ======================================================== # BYTE STATISTICS # ======================================================== def _byte_statistics( self, data ): if not data: return { "length": 0 } unique = len( set(data) ) return { "length": len(data), "unique_bytes": unique, "zero_bytes": data.count(0), "high_bytes": sum( 1 for byte in data if byte >= 128 ), "ascii_bytes": sum( 1 for byte in data if 32 <= byte <= 126 ) } # ======================================================== # BINARY EVENT HEADER # ======================================================== def _decode_binary_event_header( self, payload ): result = { "binary_header_raw": payload } if not payload: return result result["binary_header_hex"] = ( payload.encode( "utf-8", errors="replace" ).hex() ) # Attempt to identify the optional attachment # count used by Socket.IO's binary-event format. digits = "" for character in payload: if character.isdigit(): digits += character else: break if digits: try: result[ "attachment_count" ] = int(digits) result[ "binary_event_payload" ] = payload[ len(digits): ] except ValueError: pass return result # ======================================================== # JSON HELPER # ======================================================== def _try_json( self, value ): if value is None: return None if not isinstance( value, str ): return None try: return json.loads( value ) except ( json.JSONDecodeError, TypeError ): return None # ======================================================== # UTF-8 HELPER # ======================================================== def _safe_utf8( self, data ): if not data: return "" try: text = data.decode( "utf-8" ) if all( char.isprintable() or char in "\r\n\t" for char in text ): return text except UnicodeDecodeError: pass return None # ======================================================== # CRASH EVENT DETECTION # ======================================================== def is_crash_event( self, packet ): if not packet: return False packet_type = packet.get( "type" ) if packet_type in ( "socket_event", "socket_binary_event", "binary_payload" ): return True return False # ======================================================== # CRASH CANDIDATE EXTRACTION # ======================================================== def extract_crash_candidate( self, packet ): """ Returns a normalized candidate structure for crash.py. No multiplier is invented here. A multiplier is only accepted if a future decoder can establish that a particular field genuinely represents the Crash result. """ if not self.is_crash_event( packet ): return None candidate = { "timestamp": packet.get( "timestamp" ), "packet_type": packet.get( "type" ), "namespace": packet.get( "namespace", self.namespace ), "length": packet.get( "length" ), "raw_hex": packet.get( "hex" ) } if "payload" in packet: candidate["payload"] = ( packet["payload"] ) if ( "protobuf_candidates" in packet ): candidate[ "protobuf_candidates" ] = packet[ "protobuf_candidates" ] return candidate # ======================================================== # STORAGE # ======================================================== def _store( self, packet ): if not self.save_raw_messages: return try: safe_packet = dict( packet ) raw = safe_packet.get( "raw" ) if isinstance( raw, bytes ): safe_packet["raw"] = ( raw.hex() ) with open( self.raw_file, "a", encoding="utf-8" ) as file: file.write( json.dumps( safe_packet, ensure_ascii=False, default=str ) + "\n" ) except Exception as error: self.logger.error( "Failed to store protocol packet: %s", error ) def _store_binary( self, packet ): if not self.save_raw_messages: return try: with open( self.binary_file, "a", encoding="utf-8" ) as file: file.write( json.dumps( packet, ensure_ascii=False, default=str ) + "\n" ) except Exception as error: self.logger.error( "Failed to store binary packet: %s", error ) # ======================================================== # DIAGNOSTICS # ======================================================== def get_stats(self): return { "messages": self.message_count, "binary_messages": self.binary_count, "socket_events": self.event_count, "namespace": self.namespace, "raw_storage": self.save_raw_messages }