Fetch DER encoded server public key from a 1.7+ Minecraft server

key_fetcher.py

import sys
import socket
import json
import zlib


def encode_varint(i):
    # standard uleb128 encoding

    if i == 0:
        return b"\x00"

    v = b""
    while i:
        v += bytes([i & 0x7F | (0x80 if (i >> 7) else 0x00)])
        i >>= 7

    return v


def encode_string(s):
    s = s.encode("utf-8")
    return encode_varint(len(s)) + s


def encode_ushort(i):
    return i.to_bytes(2, "big")


def encode_packet(packet_id, packet_data):
    # varint length + varint packet id + packet data

    packet = encode_varint(packet_id) + packet_data
    return encode_varint(len(packet)) + packet


def decode_varint(b):
    i = 0
    s = 0

    while True:
        c = b[0]
        b = b[1:]

        i |= (c & 0x7F) << s
        s += 7

        if c & 0x80 == 0:
            break

    return i, b


def connect(host):
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((socket.gethostbyname(host), 25565))

    # C -> S handshake (status)
    sock.send(
        encode_packet(
            0x00,  # handshake
            encode_varint(774)  # protocol version for 1.21.11
            + encode_string("")  # hostname
            + encode_ushort(25565)  # port
            + encode_varint(1),  # satus state
        )
    )

    # C -> S status request
    sock.send(
        encode_packet(
            0x00,  # status request
            b"",  # no payload
        )
    )

    # S -> C status response
    packet = sock.recv(1 << 24)

    # length prefix
    l, packet = decode_varint(packet)
    assert l > 0
    assert l <= len(packet)
    packet = packet[:l]

    packet_id, packet = decode_varint(packet)
    assert packet_id == 0x00

    # status
    l, packet = decode_varint(packet)
    status = json.loads(packet[:l].decode("utf-8"))
    packet = packet[l:]
    assert len(packet) == 0

    sock.close()
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.connect((socket.gethostbyname(host), 25565))

    # C -> S handshake (login)
    sock.send(
        encode_packet(
            0x00,  # handshake
            # protocol version for 1.21.11
            encode_varint(status["version"]["protocol"])
            + encode_string("")  # hostname
            + encode_ushort(25565)  # port
            + encode_varint(2),  # login state
        )
    )

    # C -> S login start
    sock.send(
        encode_packet(
            0x00,  # login start
            encode_string("Notch")  # username
            # Notch's UUID
            + bytes.fromhex("069a79f444e94726a5befca90e38aaf5"),
        )
    )

    compression = -1
    while True:
        # S -> C
        packet = sock.recv(65536)

        # length prefix
        l, packet = decode_varint(packet)
        assert l > 0
        assert l <= len(packet)
        packet = packet[:l]

        if compression >= 0:
            l, packet = decode_varint(packet)
            if l != 0:
                packet = zlib.decompress(packet)

        packet_id, packet = decode_varint(packet)

        match packet_id:
            case 0x01:
                # encryption request
                # server name
                l, packet = decode_varint(packet)
                packet = packet[l:]

                # key
                l, packet = decode_varint(packet)
                print("Server key:", packet[:l].hex())

                # close connection
                sock.close()
                break
            case 0x02:
                # login success
                print("Cracked server!")
                sock.close()
                break
            case 0x03:
                # set compression
                l, packet = decode_varint(packet)
                compression = l

                assert len(packet) == 0
            case _:
                print("Unknown packet id:", packet_id, packet)
                sock.close()
                break


if __name__ == "__main__":
    if len(sys.argv) > 1:
        connect(sys.argv[1])
    else:
        print(sys.argv[0], "<host>")