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 support@ripple.moe 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 support@ripple.moe 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