Pythonで認証コードを返すだけのマイクラサーバーを作ってみる

# はじめに
接続したときに「あなたの認証コードは〇〇です」みたいなの返すやつ作ってみたい...
[前の記事](https://qiita.com/n_mache/items/b76cff45e3b99a65a3a0)書いてて「これサーバー側も作れそう」ってなったからやってみます。

:::note info
**追記**
GitHubにレポジトリを作成しました。
Pull RequestやIssueなどお待ちしています。
https://github.com/n-mache/pymcauthserver
:::
:::note warn
動作確認はなるべく行いましたが、脆弱性やバグが残っている可能性があります。
実用化する際はドキュメントとかを全て読んで、コードが問題ないことを確認して下さい。
:::
# Quarry
[Quarry](https://pypi.org/project/quarry/)ってモジュールを見つけたけど、どうやら更新が止まってて1.20.4では機能しないらしい...
1から実装するしか無さそう？
# 実際にやってみる
認証コード返すだけのサーバーだし、ステータスとログイン処理さえあればいいはずなので、
## モジュール
`cryptography` と `requests` だけ標準で入ってないから `pip install` で入れないと動かない
```python
import socket #TCP通信とかをするやつ
import json #JSONをパースしてくれるやつ
import struct #バイナリを簡単にするやつ(?)
import threading #同時接続とかをするために使うやつ
from cryptography.hazmat.backends import default_backend #暗号化関係
from cryptography.hazmat.primitives.asymmetric import rsa, padding #暗号化関係
from cryptography.hazmat.primitives.ciphers import algorithms, modes #暗号化関係
from cryptography.hazmat.primitives import ciphers, serialization #暗号化関係
import os #OSの機能を使うためのやつ
import random #ランダム生成に使うやつ
import string #文字とかを使うやつ
import requests #HTTP通信をするためのやつ
import hashlib #ハッシュを生成するためのやつ
import uuid #UUIDをいろいろ扱えるやつ
import time #時間を扱うためのやつ
```
## VarInt
(前の記事のやつをそのまま使う)
```python
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
```
## TCPサーバー
0.0.0.0:25565で建ててみる
```python
svhost = "0.0.0.0"
svport = 25565

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((svhost, svport))
server_socket.listen()
print(f"{svhost}:{svport}で接続を受け入れています。")
```
## 接続を処理する
仕様上、処理中は新規接続がブロックされてしまうらしい...
同時接続とかもしたいからhandle_client関数で処理するようにして、threadingからいいかんじに叩く
```python
def handle_client(client, address):
```
### handshake
handshakeはStateが1で「ステータス」、2で「ログイン」って[この記事](https://wiki.vg/Protocol#Handshake)に書かれてるからそれで分岐させてみる
```python
    res = b""
    while True:
        r = client.recv(1)
        res += r
        if not r[0] & 128:break
    size = decode_varint(res)
    res = client.recv(size)
    recv = b""
    pos = 0
    for r in res:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    packet_id = decode_varint(recv)
    if packet_id != 0:
        client.close()
        return
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    protocol_version = decode_varint(recv)
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    size = decode_varint(recv)
    addr = res[pos:pos+size].decode("utf-8")
    pos += size
    port = struct.unpack(">H", res[pos:pos+2])[0]
    pos += 2
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    next_state = decode_varint(recv)
    if next_state != 1 and next_state != 2:
        client.close()
        return
```
## ステータス応答
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654072/85a3f2f6-d8c6-86f4-cf1f-dfcf6b1633b0.png)
このコードをそのまま実行すればこの画像みたいな表示になります。
`data = json.dumps...`の部分を書き換えれば表示も変更できます。
```python
    if next_state == 1:
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res != b"\x00":
            client.close()
            return
        data = json.dumps({"version": {"name": "認証サーバー", "protocol": protocol_version}, "players": {"max": 1, "online": 0}, "description": {"text": "認証サーバー"}}).encode()
        data = b"\x00"+encode_varint(len(data))+data
        client.send(encode_varint(len(data))+data)
        pos = 0
        res = b""
        while True:
            r = client.recv(1)
            res += r
            pos += 1
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res[0] != 1:
            client.close()
            return
        data = res
        data = encode_varint(len(data))+data
        client.send(data)
        client.close()
```
## ログイン
正直ここでコード発行して切っちゃっても良さそうだけど...
それだと他人のMCIDとUUIDで発行できちゃう可能性があるので認証までやります。
```python
    if next_state == 2:
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        pos = 0
        if res[0] != 0:
            client.close()
            return
        pos += 1
        recv = b""
        for r in res[pos:]:
            recv += bytes([r])
            pos += 1
            if not r & 128:break
        size = decode_varint(recv)
        username = res[pos:pos+size].decode("utf-8")
```
### 暗号化リクエスト
キーとかサーバーIDとかを生成してクライアントに送るらしい
```python
        key = rsa.generate_private_key(public_exponent=65537, key_size=1024, backend=default_backend())
        public_key = key.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
        verify_token = os.urandom(4)
        server_id = ("".join(random.choices(string.ascii_lowercase+string.digits,k=10))).encode('ascii')
        data = b"\x01"+encode_varint(len(server_id))+server_id+encode_varint(len(public_key))+public_key+encode_varint(len(verify_token))+verify_token
        data = encode_varint(len(data))+data
        client.send(data)
```
### 暗号化応答
暗号化応答ではクライアント側から共有暗号鍵っていうのが送られてくるらしい
```python
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res[0] != 1:
            client.close()
            return
        pos = 1
        recv = b""
        for r in res[pos:]:
            recv += bytes([r])
            pos += 1
            if not r & 128:break
        size = decode_varint(recv)
        shared_secret = res[pos:pos+size]
        shared_secret = key.decrypt(shared_secret, padding.PKCS1v15())
        pos += size
        recv = b""
        for r in res[pos:]:
            pos += 1
            if len(recv) == 0 and r == 0:continue
            recv += bytes([r])
            if not r & 128:break
        size = decode_varint(recv)
        client_verify_token = res[pos:pos+size]
        client_verify_token = key.decrypt(client_verify_token, padding.PKCS1v15())
        if verify_token != verify_token:
            client.close()
            return
        cipher = ciphers.Cipher(algorithms.AES(shared_secret), modes.CFB8(shared_secret), backend=default_backend())
        encryptor = cipher.encryptor()
        decryptor = cipher.decryptor()
```
### 認証
ここだけうまくいかなくてめちゃくちゃ苦戦しました。
(具体的にはうまくいくときとうまくいかないときがあった)
wiki.vgを見てみたら「Minecraftで使用されるSha1.hexdigest()メゾットは非標準であることに注意してください」ってちゃんと書いてあった
```python
        hash = hashlib.sha1()
        hash.update(server_id)
        hash.update(shared_secret)
        hash.update(public_key)
        hash = int(hash.hexdigest(), 16)
        if hash >> 156 & 8:
            hash = "-"+format(hash*-1 & 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF,"x")
        else:
            hash = format(hash,"x")
        has_joined = requests.get("https://sessionserver.mojang.com/session/minecraft/hasJoined",params={"username":username,"serverId":hash})
        has_joined_response = {}
        try:has_joined_response = has_joined.json()
        except:pass
        if has_joined.status_code != 200 or not "id" in has_joined_response:
            data = json.dumps({"text": "認証できませんでした。", "color": "red"}).encode()
            data = b"\x00"+encode_varint(len(data))+data
            data = encode_varint(len(data))+data
            data = encryptor.update(data)
            client.send(data)
            client.close()
            return
```
### 認証コードを発行する
ランダムな数字6桁を適当に発行してみてます。
(この部分変えれば任意のものにできる)
```python
        verify_code = "".join(random.choices(string.digits,k=6))
        print(has_joined_response["name"]+" ("+has_joined_response["id"]+") に認証コードを発行しました。")
        data = json.dumps({"text": "§a認証コードを発行しました§r\n§bユーザー名:§r §f"+has_joined_response["name"]+"§r\n§b認証コード:§r §f"+verify_code+"§r"}).encode()
        data = b"\x00"+encode_varint(len(data))+data
        data = encode_varint(len(data))+data
        data = encryptor.update(data)
        client.send(data)
        client.close()
```
### 認証コードを保存する
ファイルに保存してもいいしDBに保存してもいいし...
今回はファイルに保存するようにしてみます。
(`mcauths` という名前でディレクトリ作らないと動かないので注意)
```python
        with open("mcauths/"+has_joined_response["id"], "w") as f:
            f.write(json.dumps({"code": verify_code, "expire": time.time()+600}))
```
## 接続を受け入れる
サーバーのsocketに対してaccept関数を使うと受け入れができるらしい
処理中に接続をブロックさせたくないのでthreadingを使用
```python
while True:
    client, address = server_socket.accept()
    threading.Thread(target=handle_client,args=(client, address,)).start()
```
# 全体のコード
```python
import socket #TCP通信とかをするやつ
import json #JSONをパースしてくれるやつ
import struct #バイナリを簡単にするやつ(?)
import threading #同時接続とかをするために使うやつ
from cryptography.hazmat.backends import default_backend #暗号化関係
from cryptography.hazmat.primitives.asymmetric import rsa, padding #暗号化関係
from cryptography.hazmat.primitives.ciphers import algorithms, modes #暗号化関係
from cryptography.hazmat.primitives import ciphers, serialization #暗号化関係
import os #OSの機能を使うためのやつ
import random #ランダム生成に使うやつ
import string #文字とかを使うやつ
import requests #HTTP通信をするためのやつ
import hashlib #ハッシュを生成するためのやつ
import uuid #UUIDをいろいろ扱えるやつ
import time #時間を扱うためのやつ

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

svhost = "0.0.0.0"
svport = 25565

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind((svhost, svport))
server_socket.listen()
print(f"{svhost}:{svport}で接続を受け入れています。")

def handle_client(client, address):
    res = b""
    while True:
        r = client.recv(1)
        res += r
        if not r[0] & 128:break
    size = decode_varint(res)
    res = client.recv(size)
    recv = b""
    pos = 0
    for r in res:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    packet_id = decode_varint(recv)
    if packet_id != 0:
        client.close()
        return
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    protocol_version = decode_varint(recv)
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    size = decode_varint(recv)
    addr = res[pos:pos+size].decode("utf-8")
    pos += size
    port = struct.unpack(">H", res[pos:pos+2])[0]
    pos += 2
    recv = b""
    for r in res[pos:]:
        recv += bytes([r])
        pos += 1
        if not r & 128:break
    next_state = decode_varint(recv)
    if next_state != 1 and next_state != 2:
        client.close()
        return
    if next_state == 1:
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res != b"\x00":
            client.close()
            return
        data = json.dumps({"version": {"name": "認証サーバー", "protocol": protocol_version}, "players": {"max": 1, "online": 0}, "description": {"text": "認証サーバー"}}).encode()
        data = b"\x00"+encode_varint(len(data))+data
        client.send(encode_varint(len(data))+data)
        pos = 0
        res = b""
        while True:
            r = client.recv(1)
            res += r
            pos += 1
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res[0] != 1:
            client.close()
            return
        data = res
        data = encode_varint(len(data))+data
        client.send(data)
        client.close()
    if next_state == 2:
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        pos = 0
        if res[0] != 0:
            client.close()
            return
        pos += 1
        recv = b""
        for r in res[pos:]:
            recv += bytes([r])
            pos += 1
            if not r & 128:break
        size = decode_varint(recv)
        username = res[pos:pos+size].decode("utf-8")
        key = rsa.generate_private_key(public_exponent=65537, key_size=1024, backend=default_backend())
        public_key = key.public_key().public_bytes(serialization.Encoding.DER, serialization.PublicFormat.SubjectPublicKeyInfo)
        verify_token = os.urandom(4)
        server_id = ("".join(random.choices(string.ascii_lowercase+string.digits,k=10))).encode('ascii')
        data = b"\x01"+encode_varint(len(server_id))+server_id+encode_varint(len(public_key))+public_key+encode_varint(len(verify_token))+verify_token
        data = encode_varint(len(data))+data
        client.send(data)
        res = b""
        while True:
            r = client.recv(1)
            res += r
            if not r[0] & 128:break
        size = decode_varint(res)
        res = client.recv(size)
        if res[0] != 1:
            client.close()
            return
        pos = 1
        recv = b""
        for r in res[pos:]:
            recv += bytes([r])
            pos += 1
            if not r & 128:break
        size = decode_varint(recv)
        shared_secret = res[pos:pos+size]
        shared_secret = key.decrypt(shared_secret, padding.PKCS1v15())
        pos += size
        recv = b""
        for r in res[pos:]:
            pos += 1
            if len(recv) == 0 and r == 0:continue
            recv += bytes([r])
            if not r & 128:break
        size = decode_varint(recv)
        client_verify_token = res[pos:pos+size]
        client_verify_token = key.decrypt(client_verify_token, padding.PKCS1v15())
        if verify_token != client_verify_token:
            client.close()
            return
        cipher = ciphers.Cipher(algorithms.AES(shared_secret), modes.CFB8(shared_secret), backend=default_backend())
        encryptor = cipher.encryptor()
        decryptor = cipher.decryptor()
        hash = hashlib.sha1()
        hash.update(server_id)
        hash.update(shared_secret)
        hash.update(public_key)
        hash = int(hash.hexdigest(), 16)
        if hash >> 156 & 8:
            hash = "-"+format(hash*-1 & 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF,"x")
        else:
            hash = format(hash,"x")
        has_joined = requests.get("https://sessionserver.mojang.com/session/minecraft/hasJoined",params={"username":username,"serverId":hash})
        has_joined_response = {}
        try:has_joined_response = has_joined.json()
        except:pass
        if has_joined.status_code != 200 or not "id" in has_joined_response:
            data = json.dumps({"text": "認証できませんでした。", "color": "red"}).encode()
            data = b"\x00"+encode_varint(len(data))+data
            data = encode_varint(len(data))+data
            data = encryptor.update(data)
            client.send(data)
            client.close()
            return
        verify_code = "".join(random.choices(string.digits,k=6))
        print(has_joined_response["name"]+" ("+has_joined_response["id"]+") に認証コードを発行しました。")
        data = json.dumps({"text": "§a認証コードを発行しました§r\n§bユーザー名:§r §f"+has_joined_response["name"]+"§r\n§b認証コード:§r §f"+verify_code+"§r"}).encode()
        data = b"\x00"+encode_varint(len(data))+data
        data = encode_varint(len(data))+data
        data = encryptor.update(data)
        client.send(data)
        client.close()
        # 認証コードを保存する処理をここへ
        # ファイルに保存する例
        with open("mcauths/"+has_joined_response["id"], "w") as f:
            f.write(json.dumps({"code": verify_code, "expire": time.time()+600}))

while True:
    client, address = server_socket.accept()
    threading.Thread(target=handle_client,args=(client, address,)).start()
```
# ここまでの進捗
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654072/b0601f6f-53eb-5585-818e-7af73b003443.png)
最終的にこうなった
# サイト側の実装例
これはPHPで実装する例です。
`$mcauths_dir` の部分だけさっきのディレクトリのパスに変更すればそのまま使えるはず...

:::note warn
__注意__
サイトで公開される場所にmcauthsディレクトリを置くと認証コードが盗み見られる可能性があります。
:::
```php
<?php
if (isset($_POST["mcid"]) && isset($_POST["code"])){
    $mcauths_dir = "/path/to/mcauths/"; // ディレクトリのパスには最後にスラッシュ("/")をつける
    $mcid = $_POST["mcid"];
    $code = $_POST["code"];
    $uuid = @file_get_contents("https://api.mojang.com/users/profiles/minecraft/".$mcid);
    if ($uuid === false){
        http_response_code(400);
        exit("UUIDの取得に失敗しました。");
    }
    $res = json_decode($uuid, true);
    $uuid = $res["id"];
    $mcid = $res["name"];
    if (!is_file($mcauths_dir.$uuid)){
        http_response_code(400);
        exit("認証情報が存在しません。");
    }
    $auth = json_decode(file_get_contents($mcauths_dir.$uuid), true);
    if ($auth["code"] !== $code){
        http_response_code(400);
        exit("認証に失敗しました。");
    }
    if ($auth["expire"] < time()){
        http_response_code(400);
        unlink($mcauths_dir.$uuid);
        exit("認証コードの有効期限が切れています。");
    }
    // 認証成功
    unlink($mcauths_dir.$uuid);
    exit("認証に成功しました。<br>UUID: ".$uuid."<br>NAME: ".$mcid);
}
?>
<form method="POST">
    MCID: <input name="mcid"><br>
    CODE: <input name="code"><br>
    <input type="submit">
</form>
```
## 動作
![mcauth_demo.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3654072/1f24a579-5559-c8bf-b6bd-eb0508c6c275.gif)
# 参考
- [Protocol Encryption - wiki.vg](https://wiki.vg/Protocol_Encryption)
- [Protocol - wiki.vg](https://wiki.vg/Protocol)
- [uuidモジュール - Pythonドキュメント](https://docs.python.org/ja/3/library/uuid.html)
- [PythonでTCP通信(サーバ編) - Qiita](https://qiita.com/keiusukematsuda/items/aacded313fdb6c08f410)