# echo-server.py import socket import time import codecs import sys import enum from enum import IntFlag import struct import bottle import threading from sys import exit import vmixoutput HOST = "127.0.0.1" # Standard loopback interface address (localhost) PORT = 27569 # Port to listen on (non-privileged ports are > 1023) HTTPPORT = 8098 parsedData = {} class InfoType(enum.Enum): u8 = 0 s8 = 1 s16 = 2 s32 = 3 f32 = 4 string = 5 class PlayerFlags(IntFlag): PlayerInfo = 0x1 Health = 0x2 Inventory = 0x4 Ammo = 0x8 Score = 0x10 GMStat = 0x20 class TypedData: value = 0 endIndex = 0 class MGPacket: def findNullTerm(self,startindex, data=None): if data == None: data = self.rawPacket return data.find(b'\x00', startindex) def getByteInt(self, index, length, data=None, isSigned=False): if data == None: data = self.rawPacket return int.from_bytes(data[index:index+length], "little", signed=isSigned) def getStringAtIndex(self, index, data=None): if data == None: data = self.rawPacket return data[index:self.findNullTerm(index,data)].decode("latin-1","ignore") def parseRemainingPacket(self, data): if len(data) > 1: print("[[CONTINUING PARSING DATA WITH LENGTH {}]]".format(len(data))) subparsedData = MGPacket(data) def scrubPlayerName(self, rawName): replacementDict={ "^^": "", "^a": "", "^1": "", "^2": "", "^3": "", "^4": "", "^5": "", "^6": "", "^7": "", "^8": "", "^9": "", "^b": "", "^d": "", "^m": "", "^h": "", "^s": "", "^r": "" } parsedName = rawName for i, char in enumerate(parsedName): if char == "^" and parsedName[i+1] == "x": replacementDict[parsedName[i:i+5]] = "" elif char == "^" and parsedName[i+1] == "&": replacementDict[parsedName[i:i+4]] = "" elif char == "^" and parsedName[i+1] == "[": endindex = parsedName[i:].index("^]") replacementDict[parsedName[i:endindex+2+i]] = "" elif char == "^" and parsedName[i+1] == "{": endindex = parsedName[i:].index("}") replacementDict[parsedName[i:endindex+1+i]] = "" elif char == "^" and parsedName[i+1] == "'": endindex = parsedName[i:].index("=") replacementDict[parsedName[i:endindex+1+i]] = "" elif char == "^" and parsedName[i+1] == "U": replacementDict[parsedName[i:i+6]] = "" for chars, matches in replacementDict.items(): parsedName = parsedName.replace(chars, matches) return parsedName def getDataAfterType(self, type, index, data=None): if data == None: data = self.rawPacket result = TypedData() if type == InfoType['u8']: result.value = int.from_bytes(data[index + 1 : index + 2], "little") result.endIndex = index + 2 return result elif type == InfoType['s8']: result.value = -1 return result elif type == InfoType['s16']: result.value = -1 return result elif type == InfoType['s32']: result.value = -1 return result elif type == InfoType['f32']: result.value = -1 return result elif type == InfoType['string']: result.value = data[ index + 1 : self.findNullTerm(index + 1, data) ].decode("latin-1","ignore") result.endIndex = self.findNullTerm(index + 1, data) #result.value = self.scrubPlayerName(result.value) return result def __init__(self, rawPacket): rawArray = [] for byte in rawPacket: rawArray.append(hex(byte)) print("\n [[ReceivedData]]", rawArray) self.rawPacket = rawPacket parsedData = {} header = int.from_bytes(rawPacket[:1], "little") # get first byte if header == 0: parsedData['packetType'] = 'GAMEMODEINFO' parsedData['protocol'] = int.from_bytes(self.rawPacket[1:5], "little") if parsedData['protocol'] != 1: return choppedBlock = self.rawPacket[6:] #gonna need to test these with >1 global stat numGlobalStats = self.getByteInt(5,1) parsedData['numGlobalStats'] = numGlobalStats while numGlobalStats > 0: parsedData['globalStat' + str(self.getByteInt(0,1,choppedBlock)) ] = self.getStringAtIndex(1, choppedBlock) choppedBlock = choppedBlock[self.findNullTerm(1,choppedBlock)+1:] numGlobalStats -= 1 numPlayerStats = self.getByteInt(0,1,choppedBlock) parsedData['numPlayerStats'] = numPlayerStats while numPlayerStats > 0: parsedData['playerStat' + str(self.getByteInt(1,1,choppedBlock)) ] = self.getStringAtIndex(2, choppedBlock) choppedBlock = choppedBlock[self.findNullTerm(1,choppedBlock)+1:] numPlayerStats -= 1 choppedBlock = choppedBlock[self.findNullTerm(0,choppedBlock)+1:] #fix for arbitrary lengths print(parsedData) elif header == 1: parsedData['packetType'] = 'PLAYERINFO' parsedData['playerNum'] = self.getByteInt(1,1) updateFlags = (self.rawPacket[2] << 0) + (self.rawPacket[3] << 8) parsedData['updateFlags'] = updateFlags choppedBlock = self.rawPacket[4:] #player data updates per flag, eating the data as we go if bool(PlayerFlags.PlayerInfo & updateFlags): parsedData['name'] = self.getStringAtIndex(0, choppedBlock) print("HI") print(parsedData['name']) teamPosition = self.findNullTerm(0, choppedBlock)+1 parsedData['team'] = self.getByteInt(teamPosition,1, choppedBlock) if teamPosition+1 < len(choppedBlock): choppedBlock = choppedBlock[teamPosition+1:] if bool(PlayerFlags.Health & updateFlags): parsedData['health'] = self.getByteInt(0,2,choppedBlock,True) parsedData['bleedingAmount'] = self.getByteInt(2,1,choppedBlock) parsedData['deadFlags'] = self.getByteInt(3,1,choppedBlock) if len(choppedBlock) > 3: choppedBlock = choppedBlock[4:] if bool(PlayerFlags.Inventory & updateFlags): parsedData['heldWeapon'] = self.getByteInt(0,1,choppedBlock) parsedData['weapon1'] = self.getByteInt(1,1,choppedBlock) parsedData['weapon2'] = self.getByteInt(2,1,choppedBlock) parsedData['throwable'] = self.getByteInt(3,1,choppedBlock) parsedData['itemA'] = self.getByteInt(4,1,choppedBlock) parsedData['itemB'] = self.getByteInt(5,1,choppedBlock) if len(choppedBlock) > 5: choppedBlock = choppedBlock[6:] if bool(PlayerFlags.Ammo & updateFlags): #print("\n ammo block: {}".format(choppedBlock)) parsedData['ammo1'] = self.getByteInt(0,1,choppedBlock) parsedData['ammo2'] = self.getByteInt(1,1,choppedBlock) parsedData['mags1'] = self.getByteInt(2,1,choppedBlock) parsedData['mags2'] = self.getByteInt(3,1,choppedBlock) if len(choppedBlock) > 3: choppedBlock = choppedBlock[4:] if bool(PlayerFlags.Score & updateFlags): #print("\n score block: {}".format(choppedBlock)) parsedData['score'] = self.getByteInt(0,2,choppedBlock,True) parsedData['kills'] = self.getByteInt(2,2,choppedBlock,True) parsedData['deaths'] = self.getByteInt(4,2,choppedBlock,True) if len(choppedBlock) > 5: choppedBlock = choppedBlock[6:] if bool(PlayerFlags.GMStat & updateFlags): #print("\n stat block: {}".format(choppedBlock)) numStats = self.getByteInt(0,1,choppedBlock) parsedData['numStats'] = numStats choppedBlock = choppedBlock[1:] while numStats > 0: statType = InfoType(self.getByteInt(1,1,choppedBlock)) statData = self.getDataAfterType(statType,1,choppedBlock) parsedData['stat' + str(self.getByteInt(0,1,choppedBlock)) ] = statData.value choppedBlock = choppedBlock[statData.endIndex:] numStats -= 1 #print("\n data after stat block: {}".format(choppedBlock)) print(parsedData) vmixoutput.processPlayerInputStream(parsedData) elif header == 2: parsedData['packetType'] = 'GLOBALSTATINFO' parsedData['statNum'] = self.getByteInt(1,1) statType = InfoType(self.getByteInt(2,1)) parsedData['statType'] = statType statValue = self.getDataAfterType(statType,2) parsedData['statValue'] = statValue.value choppedBlock = self.rawPacket[statValue.endIndex + 1:] #print(parsedData) elif header == 3: parsedData['packetType'] = 'CHAT' chatType = self.getByteInt(1,1) #print(chatType) if chatType == 1: parsedData['chatType'] = 'globalchat' playerNum = self.getByteInt(2,1) parsedData['playerNum'] = playerNum if playerNum == 255: parsedData['playerName'] = self.getStringAtIndex(3) parsedData['message'] = self.rawPacket[self.findNullTerm(4)+1:self.findNullTerm(9)].decode("latin-1","ignore") else: parsedData['message'] = self.rawPacket[3:self.findNullTerm(8)].decode("latin-1","ignore") if chatType == 2: parsedData['chatType'] = 'killtracker' parsedData['killerNum'] = self.getByteInt(2,1) parsedData['victimNum'] = self.getByteInt(3,1) parsedData['killerWeapon'] = self.getStringAtIndex(4) if chatType == 3: parsedData['chatType'] = 'announcement' parsedData['message'] = self.getStringAtIndex(2) choppedBlock = self.rawPacket[self.findNullTerm(4)+1:] elif header == 4: parsedData['packetType'] = 'SPECTATORINFO' print(parsedData) else: parsedData['packetType'] = "UNKNOWN" try: choppedBlock except NameError: print("") else: if len(choppedBlock) == len(rawPacket): print("length parsing error, passing") return else: self.parseRemainingPacket(choppedBlock) class GameListenThread (threading.Thread): def __init__(self): threading.Thread.__init__(self) self.running = False def exit(self): self.running = False print("Exiting listen thread") sys.exit() def run(self): with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.bind((HOST, PORT)) s.listen() conn, addr = s.accept() print("Hi! Listening on {} {}".format(HOST,PORT)) with conn: try: print(f"Connected by {addr}") self.running = True while True: if self.running: data = conn.recv(1024) parsedData = MGPacket(data) time.sleep(0.1) else: break except (KeyboardInterrupt, SystemExit): self.exit() class HttpThread (threading.Thread): def __init__(self): threading.Thread.__init__(self) def exit(self): print("Exiting http thread") sys.exit() def run(self): app = application = bottle.default_app() bottle.run(host = HOST, port = HTTPPORT) def main(): #httpthread = HttpThread() try: gamelistenthread = GameListenThread() #testData = b'\x01\x020\x00\x00\x00\x18\x00\r\x00\x01\x00\x00\x02\x01\x08\x1e\x00\x00\x00\x07\x01\x00\x00\x00\x00\x00\x14\x00\x00\x00\x00\x00\x00\t\x00\x11\x00' #parsedData = MGPacket(testData) gamelistenthread.start() while gamelistenthread.is_alive(): gamelistenthread.join(1) #httpthread.start() except (KeyboardInterrupt, SystemExit): print("Exiting on keyboard interrupt") gamelistenthread.exit() if __name__ == "__main__": exit(main())