constants.py

from enum import Enum
from . import packet as packet

class data_type(Enum):
    BYTE     = 0
    UINT16   = 1
    SINT16   = 2
    UINT32   = 3
    SINT32   = 4
    UINT64   = 5
    SINT64   = 6
    STRING   = 7
    FLOAT    = 8
    BBYTES   = 9
    INT_LIST = 10

class server_packets:
    # misc packets
    notification = lambda message: packet.build(packet.id.server_notification, [[message, data_type.STRING]])

    # login error packets
    @staticmethod
    def login_banned():
        packets = packet.build(packet.id.server_user_id, [[-1, data_type.SINT32]])
        packets += server_packets.notification("You are banned. You can appeal after one month since your ban by sending an email to [email protected] from the email address you've used to sign up.")
        return packets

    @staticmethod
    def login_locked():
        packets = packet.build(packet.id.server_user_id, [[-1, data_type.SINT32]])
        packets += server_packets.notification("Your account is locked. You can't log in, but your profile and scores are still visible from the website. If you want to unlock your account, send an email to [email protected] from the email address you've used to sign up.")
        return packets

    login_failed      = lambda: packet.build(packet.id.server_user_id, [[-1, data_type.SINT32]])
    force_update      = lambda: packet.build(packet.id.server_user_id, [[-2, data_type.SINT32]])
    login_error       = lambda: packet.build(packet.id.server_user_id, [[-5, data_type.SINT32]])
    need_supporter    = lambda: packet.build(packet.id.server_user_id, [[-6, data_type.SINT32]])
    need_verification = lambda: packet.build(packet.id.server_user_id, [[-8, data_type.SINT32]])

    # login packets
    user_id          = lambda uid:        packet.build(packet.id.server_user_id, [[uid, data_type.SINT32]])
    silence_end_time = lambda seconds:    packet.build(packet.id.server_silence_end, [[seconds, data_type.UINT32]])
    protocol_version = lambda version=19: packet.build(packet.id.server_protocol_version, [[version, data_type.UINT32]])
    main_menu_icon   = lambda icon:       packet.build(packet.id.server_main_menu_icon, [[icon, data_type.STRING]])

packet.py

import struct
from .constants import data_type as d_type
from types import SimpleNamespace

class uleb128:
    @staticmethod
    def encode(num):
        if num==0:
            return bytearray(b"\x00")

        arr=bytearray()
        while num>0:
            arr.append(num&127)
            num>>=7

            if num!=0:
                arr[-1]|=128

        return arr

    @staticmethod
    def decode(num):
        shift = 0
        arr = [0, 0]

        while True:
            b = num[arr[1]]
            arr[1] += 1
            arr[0] |= int(b & 127) << shift
            if b & 128 == 0:
                break
            shift += 7

        return arr

class data:
    @staticmethod
    def unpack(data, data_type: d_type):
        pack_type = "<%s"%{
            d_type.UINT16: "H",
            d_type.SINT16: "h",
            d_type.UINT32: "L",
            d_type.SINT32: "l",
            d_type.UINT64: "Q",
            d_type.SINT64: "q",
            d_type.STRING: "s",
            d_type.FLOAT: "f"
        }.get(data_type, "B")
        return struct.unpack(pack_type, bytes(data))[0]

    @staticmethod
    def pack(packet_data, data_type: d_type):
        packed_data = bytes()
        pack = True
        pack_type = ""

        if data_type == d_type.BBYTES:
            pack = False
            packed_data = packet_data
        elif data_type == d_type.INT_LIST:
            pack = False
            packed_data = data.pack(len(packet_data), d_type.UINT16)
            for i in packet_data:
                packed_data += data.pack(i, d_type.SINT32)
        elif data_type == d_type.STRING:
            pack = False
            if len(packet_data) == 0:
                packed_data += b"\x00"
            else:
                packed_data += b"\x0B"
                packed_data += uleb128.encode(len(packet_data))
                packed_data += str.encode(packet_data, "latin_1", "ignore")
        else:
            pack_type = "<%s"%{
                d_type.UINT16:"H",
                d_type.SINT16:"h",
                d_type.UINT32:"L",
                d_type.SINT32:"l",
                d_type.UINT64:"Q",
                d_type.SINT64:"q",
                d_type.STRING:"s",
                d_type.FLOAT:"f"
            }.get(data_type, "B")

        if pack:
            packed_data += struct.pack(pack_type, packet_data)

        return packed_data

def get_id(stream):
    return data.unpack(stream[0:2], d_type.UINT16)

def length(stream):
    return data.unpack(stream[3:7], d_type.UINT32)

def build(packet, packet_data:list=[]):
    _pdata = packet_data
    packet_data = bytes()
    for i in _pdata:
        packet_data += data.pack(i[0], i[1])

    packet_length = len(packet_data)
    packet_bytes = bytes()

    # Return packet as bytes
    packet_bytes += struct.pack("<h", packet)  # packet id (int16)
    packet_bytes += bytes(b"\x00")  # unused byte
    packet_bytes += struct.pack("<l", packet_length)  # packet lenght (iint32)
    packet_bytes += packet_data  # packet data
    return packet_bytes

def read(stream, structure:list=[], has_first_bytes=None):
    packet_data = {}

    pos = SimpleNamespace(end=7, start=7) if has_first_bytes else SimpleNamespace(end=0, start=0)

    for i in structure:
        pos.start = pos.end
        unpack = True
        if i[1] == d_type.INT_LIST:
            unpack = False

            length = data.unpack(stream[pos.start:pos.start + 2], d_type.UINT16)

            packet_data[i[0]] = []
            for j in range(0, length):
                packed_data = stream[pos.start+2+(4*j) : pos.start+2+(4*(j+1))]
                packet_data[i[0]].append(data.unpack(packed_data, d_type.SINT32))

            pos.end = pos.start + 2 + (4 * length)
        elif i[1] == d_type.STRING:
            unpack = False

            if stream[pos.start] == 0:
                packet_data[i[0]] = ""
                pos.end = pos.start + 1
            else:
                length = uleb128.decode(stream[pos.start + 1:])
                pos.end = pos.start + length[0] + length[1] + 1

                packet_data[i[0]] = ""
                for j in stream[pos.start + 1 + length[1]:pos.end]:
                    packet_data[i[0]] += chr(j)
        else:
            pos.end = pos.start + {
                d_type.BYTE: 1,
                d_type.UINT16: 2, d_type.SINT16: 2,
                d_type.UINT32: 4, d_type.SINT32: 4,
                d_type.UINT64: 8, d_type.SINT64: 8
            }.get(i[1])

        if unpack:
            packet_data[i[0]] = data.unpack(stream[pos.start:pos.end], i[1])

class id:
    client_change_action = 0
    client_send_public_message = 1
    client_logout = 2
    client_request_status_update = 3
    server_user_id = 5
    server_send_message = 7
    server_user_stats = 11
    server_user_logout = 12
    server_spectator_joined = 13
    server_spectator_left = 14
    server_spectate_frames = 15
    client_start_spectating = 16
    client_stop_spectating = 17
    client_spectate_frames = 18
    client_cant_spectate = 21
    server_spectator_cant_spectate = 22
    server_notification = 24
    client_send_private_message = 25
    server_update_match = 26
    server_new_match = 27
    server_dispose_match = 28
    client_part_lobby = 29
    client_join_lobby = 30
    client_create_match = 31
    client_join_match = 32
    client_part_match = 33
    server_match_join_success = 36
    server_match_join_fail = 37
    client_match_change_slot = 38
    client_match_ready = 39
    client_match_lock = 40
    client_match_change_settings = 41
    server_fellow_spectator_joined = 42
    server_fellow_spectator_left = 43
    client_match_start = 44
    server_match_start = 46
    client_match_score_update = 47
    server_match_score_update = 48
    client_match_complete = 49
    server_match_transfer_host = 50
    client_match_change_mods = 51
    client_match_load_complete = 52
    server_match_all_players_loaded = 53
    client_match_no_beatmap = 54
    client_match_not_ready = 55
    client_match_failed = 56
    server_match_player_failed = 57
    server_match_complete = 58
    client_match_has_beatmap = 59
    client_match_skip_request = 60
    server_match_skip = 61
    client_channel_join = 63
    server_channel_join_success = 64
    server_channel_info = 65
    server_channel_kicked = 66
    client_match_transfer_host = 70
    server_supporter_gmt = 71
    server_friends_list = 72
    client_friend_add = 73
    client_friend_remove = 74
    server_protocol_version = 75
    server_main_menu_icon = 76
    client_match_change_team = 77
    client_channel_part = 78
    server_match_player_skipped = 81
    client_set_away_message = 82
    server_user_panel = 83
    client_user_stats_request = 85
    server_restart = 86
    client_invite = 87
    server_invite = 88
    server_channel_info_end = 89
    client_match_change_password = 90
    server_match_change_password = 91
    server_silence_end = 92
    server_user_silenced = 94
    server_user_presence_bundle = 96
    client_user_panel_request = 97
    client_tournament_match_info_request = 93
    server_match_abort = 106
    server_switch_server = 107
    client_tournament_join_match_channel = 108
    client_tournament_leave_match_channel = 109