はじめに
とある申請の条件として記事を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)