はじめに

とある申請の条件として記事を1つ以上出す必要があったため、ステータスを取得するプログラムを書くことにしました。

そもそもステータスって？

調べてみるとプロトコルを解説する記事がありました。

各部位

アイコン

base64で返ってくるみたい
(今回のコードだと取得してない)

プレイヤー数

0/20みたいなやつです。
オンライン数と最大数を取得できればよさそう

MOTD

"Message of the day" の略らしいです。
まあサーバーの説明みたいな...?

バージョン

画像には書かれてませんが、バージョンの名前とプロトコルバージョン(数字)が返ってくるみたいです。

実際にやってみる

必要なモジュール

import socket #TCP通信とかをするやつ
import json #JSONをパースしてくれるやつ
import struct #バイナリを簡単にするやつ(?)

VarInt

なにそれ聞いたこと無いんだけど()
ということで調べました。
検索でヒットしたProtocol Buffers: バイナリフォーマット(Wire Format)の中身という記事に書かれていました。
例えば25565という数字は1101 1101 1100 0111 0000 0001になるそうです。

def encode_varint(num):
    res = b""   
    while num:
        b = num & 127
        num = num >> 7
        if num != 0:
            b |= 128
        res += bytes([b])
    return res

def decode_varint(data):
    val = 0
    shift = 0
    for d in data:
        val |= (d & 127) << shift
        if not (d & 128):break
        shift += 7
    return val

サーバーに接続する

今回は例として、n-mache.work:25565を対象としています。
接続先を変更する場合はhost変数の値を書き換えればいいです。

host = "n-mache.work" #接続先のIPアドレスとか
port = 25565 #接続先のポートとか(基本的に25565)

s = socket.socket()
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
s.connect((host, port))

handshake

今回は例としてプロトコルバージョンは1.20.4 (765)と指定しています。

0x00 (Packet ID)
プロトコルバージョンのVarInt
ホスト名の長さのVarInt
ホスト名
ポートのVarInt
StateのVarInt (1を送ると「ステータス取得」、2を送ると「サーバー接続(ログイン)」になるっぽい)

の全てを1つにしたもの、送る前にバイト数も指定(例えば"test"なら"4test"みたいな感じ)

protocol_version = 765

data = b"\x00"+encode_varint(protocol_version)+encode_varint(len(host.encode()))+host.encode()+struct.pack(">H", port)+encode_varint(1)
s.send(bytes([len(data)])+data)

ステータスを取得する

s.send(b"\x01\x00")

res = b""
while True:
    r = s.recv(1)
    if r == b"\x00":break
    res += r
size = decode_varint(res)
res = b""
while True:
    r = s.recv(1)
    res += r
    if not r[0] & 128:break
size = decode_varint(res)
status = json.loads(s.recv(size).decode())

これで取得は完了です。

ステータスを表示する

これは例ですが、n-mache.work:25565の場合では

{'version': {'name': 'Waterfall 1.8.x, 1.9.x, 1.10.x, 1.11.x, 1.12.x, 1.13.x, 1.14.x, 1.15.x, 1.16.x, 1.17.x, 1.18.x, 1.19.x, 1.20.x', 'protocol': 764}, 'players': {'max': 20, 'online': 0}, 'description': {'extra': [{'text': 'mache server'}], 'text': ''}, 'modinfo': {'type': 'FML', 'modList': []}}

というJSONが返却されます。
ここからversion, players, descriptionを取り出します。

if "version" in status and "name" in status["version"]:
    print("バージョン: "+status["version"]["name"])
if "players" in status:
    print("プレイヤー数: "+str(status["players"].get("max","?"))+"人中"+str(status["players"].get("online","?"))+"人")
if "description" in status:
    motd = None
    if "text" in status["description"] and len(status["description"]["text"]) > 0:
        motd = status["description"]["text"]
    if motd == None and "extra" in status["description"]:
        motd = ""
        for extra in status["description"]["extra"]:
            if not "text" in extra:continue
            motd += extra["text"]
    if motd != None:
        print("MOTD: "+motd)

実行すると、こんな感じのが出力されます。

バージョン: Waterfall 1.8.x, 1.9.x, 1.10.x, 1.11.x, 1.12.x, 1.13.x, 1.14.x, 1.15.x, 1.16.x, 1.17.x, 1.18.x, 1.19.x, 1.20.x
プレイヤー数: 20人中0人
MOTD: mache server

全体のコード

import socket
import json
import struct

def encode_varint(num):
    res = b""
    while num:
        b = num & 127
        num = num >> 7
        if num != 0:
            b |= 128
        res += bytes([b])
    return res

def decode_varint(data):
    val = 0
    shift = 0
    for d in data:
        val |= (d & 127) << shift
        if not (d & 128):break
        shift += 7
    return val

host = "n-mache.work"
port = 25565

s = socket.socket()
s.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
s.connect((host, port))

protocol_version = 765

data = b"\x00"+encode_varint(protocol_version)+encode_varint(len(host.encode()))+host.encode()+struct.pack(">H", port)+encode_varint(1)
s.send(bytes([len(data)])+data)

s.send(b"\x01\x00")

res = b""
while True:
    r = s.recv(1)
    if r == b"\x00":break
    res += r
size = decode_varint(res)
res = b""
while True:
    r = s.recv(1)
    res += r
    if not r[0] & 128:break
size = decode_varint(res)
status = json.loads(s.recv(size).decode())

if "version" in status and "name" in status["version"]:
    print("バージョン: "+status["version"]["name"])
if "players" in status:
    print("プレイヤー数: "+str(status["players"].get("max","?"))+"人中"+str(status["players"].get("online","?"))+"人")
if "description" in status:
    motd = None
    if "text" in status["description"] and len(status["description"]["text"]) > 0:
        motd = status["description"]["text"]
    if motd == None and "extra" in status["description"]:
        motd = ""
        for extra in status["description"]["extra"]:
            if not "text" in extra:continue
            motd += extra["text"]
    if motd != None:
        print("MOTD: "+motd)

Pythonでマインクラフトサーバーからステータスを取得して表示する