pymetaserver/fetcher.py

from authentication import authenticate
import connectpacket
import datapacket
import identitypacket
import serverpacket
import encoder
import socket

nestData = {"level": 0}

def convert(data, type):
    match type:
        case 1:
            return int.from_bytes(data, byteorder="little", signed=True)
        case 3:
            return ord(data)
        case 4:
            return data.decode()
        case 5:
            return bool(data)
        case 6:
            return
    return "Invalid" + str(type)

def parse(packet):
    directorySize = 0
    bytesToParse = 0
    isParsingKey = True
    parsedKey = ""
    playerListNullSkipped = False
    parsedValueType = None
    nestLevel = nestData["level"]
    nextNestLevel = nestLevel + 1
    data = {}

    #if nestLevel > 0:
    #    print("NESTED level", nestLevel, packet)

    for idx, byte in enumerate(packet):

        if nestLevel:
            if not nestLevel in nestData:
                nestData[nestLevel] = 0
            else:
                nestData[nestLevel] += 1

        if nextNestLevel in nestData:
            if nestData[nextNestLevel]:
                nestData[nextNestLevel] -= 1
                #print("SKIPPING")
                #print("IDX:", idx)
                #print("PACKET SIZE:", len(packet))
                continue

        if idx == 0:
            if byte != ord("<"):
                return "Malformed"
            else:
                continue

        if not bytesToParse and not ("header" in data and data["header"] == "pn" and directorySize):
            if not directorySize and not isParsingKey:
                if not playerListNullSkipped:
                    data["header"] = parsedKey
                    del data[parsedKey]
                    parsedKey = ""
                if data["header"] == "pn" and not playerListNullSkipped:
                    playerListNullSkipped = True
                    continue
                else:
                    directorySize = byte
                    isParsingKey = True
            elif not isParsingKey and not parsedValueType:
                parsedValueType = byte
                if parsedValueType == 0:
                    nestData[nestLevel] += 1
                    return data
                elif parsedValueType == 1:
                    bytesToParse = 4
                elif parsedValueType == 3 or parsedValueType == 5:
                    bytesToParse = 1
            else:
                if chr(byte) == "\x00" or chr(byte) == ">":
                    #if nestData["level"] > 1 and chr(byte) == "\x00":
                    #    nestData[nestLevel] += 1
                    return data
                bytesToParse = byte
        else:
            bytesToParse -= 1
            if isParsingKey:
                if chr(byte) == "<":
                    bytesToParse = 1
                    #print("NESTDATA INITIAL:", nestData)
                    nestData["level"] += 1
                    nestedPacket = parse(packet[idx:])
                    nestData["level"] -= 1
                    #print("PARSING:", packet[idx:])
                    #print("PARSED:", nestedPacket)
                    #print("NESTDATA RESULT:", nestData)
                    #print("NESTDATA HEADER:", nestedPacket["header"])
                    #print("NESTLEVEL:", nestLevel)
                    name = nestedPacket["header"]
                    data[name] = dict(nestedPacket)
                    #print("RESULTING PACKET:", data)
                    continue
                elif chr(byte) == "\x00" or chr(byte) == ">":
                    return data
                else:
                    parsedKey += chr(byte)
            else:
                data[parsedKey] += bytes([byte])

            if bytesToParse == 0:
                if isParsingKey:
                    #print(parsedKey)
                    data[parsedKey] = b"";
                else:
                    data[parsedKey] = convert(data[parsedKey], int(parsedValueType))
                    parsedKey = "";
                    parsedValueType = None
                isParsingKey = not isParsingKey

    return data

def packetType(data):
    if not "header" in data or not "c" in data:
        return
    if data["header"] == "m":
        match data["c"]:
            case "ma":
                return "Register"
            case "mb":
                return "List"
            case "md":
                return "Authentication"
            case "mf":
                return "Data"
            case "mh":
                return "ServerDetails"
    elif data["header"] == "match":
        match data["c"]:
            case "aa":
                return "Identity"
            case "ac":
                return "Connect"

def respond(packet, address, config, cur):
    data = parse(packet)
    print("Packet received:", data)
    print("Its raw data:", packet)
    type = packetType(data)
    match type:
        case "Data":
            return [encoder.encode(datapacket.generate(address, data["ec"], data["DataType"], config))]
        case "Register":
            cur.execute(
                "REPLACE INTO servers (name, game, version, localIp, localPort, playersOnline, demoPlayers, maxPlayers, spectatorsOnline, maxSpectators, gameType, scoreMode, timePlaying, password, modPath, state, teamCount, globalIp, globalPort, players, authkey) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", 
                (data["dg"], data["dh"], data["di"], data["de"], data["df"], data["dj"], data["dk"], data["dl"], data["dm"], data["dn"], data["dw"], data["dx"], data.get("dy"), data.get("dt"), data.get("eb"), data["dv"], data["ea"], address[0], data["db"], str(data["pn"]), data["dq"]))
        case "List":
            packets = []
            authData = authenticate(data, cur, bool(int(config.get("Authentication", "KeyAuthentication"))), bool(int(config.get("Authentication", "VersionAuthentication"))))
            authPacket = encoder.encode(authData)
            packets = [authPacket]
            if authData["me"] == 1:
                packets += [encoder.encode(serverpacket.getList(data, cur))]
            return packets
        case "Authentication":
            return [encoder.encode(authenticate(data, cur, bool(int(config.get("Authentication", "KeyAuthentication"))), bool(int(config.get("Authentication", "VersionAuthentication")))))]
        case "ServerDetails":
            packets = []
            authData = authenticate(data, cur, bool(int(config.get("Authentication", "KeyAuthentication"))), bool(int(config.get("Authentication", "VersionAuthentication"))))
            authPacket = encoder.encode(authData)
            packets = [authPacket]
            if authData["me"] == 1:
                packets += [encoder.encode(serverpacket.getServer(data["da"], data["db"], data, cur, True))]
            return packets
        case "Identity":
            return [encoder.encode(identitypacket.generate(address, data))]
        case "Connect":
            message = encoder.encode(connectpacket.generate(address, data))
            send_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
            addr = (data["ah"], data["ai"])
            send_socket.sendto(message, addr)
        case _:
            print('Unknown packet' + str(type));

    return []