Initial commit

Author: Steven

byte_utils.py

@@ -0,0 +1,52 @@
+import struct
+
+
+def read_varint(byte, i):
+    result = 0
+    bytes = 0
+    while True:
+        try:
+            byte_in = byte[i]
+            i += 1
+            result |= (byte_in & 0x7F) << (bytes * 7)
+            if bytes > 32:
+                raise IOError("Packet is too long!")
+            if (byte_in & 0x80) != 0x80:
+                return result, i
+        except IndexError:
+            print("A invalid byte array was requested!")
+
+
+def read_utf(byte, i):
+    (length, i) = read_varint(byte, i)
+    ip = byte[i:(i + length)].decode('utf-8')
+    i += length
+    return ip, i
+
+
+def read_ushort(byte, i):
+    new_i = i + 2
+    return struct.unpack(">H", byte[i:new_i]), new_i
+
+
+def read_long(byte, i):
+    new_i = i + 8
+    return struct.unpack(">q", byte[i:new_i]), new_i
+
+
+def write_varint(byte, value):
+    part = None
+    while True:
+        part = value & 0x7F
+        value >>= 7
+        if value != 0:
+            part |= 0x80
+        byte.append(part)
+        if value == 0:
+            break
+
+
+def write_utf(byte, value):
+    write_varint(byte, len(value))
+    for b in value.encode():
+        byte.append(b)

main.py

@@ -0,0 +1,63 @@
+#!/usr/bin/env python3
+
+import base64
+import json
+import os.path
+
+from socket_server import SocketServer
+
+server = None
+
+
+def main():
+    if os.path.exists("config.json"):
+        with open("config.json", 'r') as file:
+            configuration = json.load(file)
+
+        ip = configuration["ip"]
+        port = configuration["port"]
+        motd = configuration["motd"]["1"] + "\n" + configuration["motd"]["2"]
+        version_text = configuration["version_text"]
+        kick_message = ""
+        server_icon = None
+
+        for message in configuration["kick_message"]:
+            kick_message += message + "\n"
+        kick_message = kick_message[:-2]
+
+        if not os.path.exists(configuration["server_icon"]):
+            print("Server icon doesn't exists - submitting none...")
+        else:
+            with open(configuration["server_icon"], 'rb') as image:
+                server_icon = "data:image/png;base64," + base64.b64encode(image.read()).decode()
+        try:
+            global server
+            server = SocketServer(ip, port, motd, version_text, kick_message, server_icon)
+            server.start()
+        except KeyboardInterrupt:
+            server.close()
+            exit(1)
+        except Exception as e:
+            print(e)
+    else:
+        configuration = {}
+        configuration["ip"] = "0.0.0.0"
+        configuration["port"] = 25565
+        configuration["motd"] = {}
+        configuration["motd"]["1"] = "§4Maintenance!"
+        configuration["motd"]["2"] = "§aCheck example.com for more information!"
+        configuration["version_text"] = "§4Maintenance"
+        configuration["kick_message"] = ["§bSorry", "", "§aThis server is offline!"]
+        configuration["server_icon"] = "server_icon.png"
+        configuration["samples"] = ["§bexample.com", "", "§4Maintenance"]
+
+        with open("config.json", 'w') as file:
+            json.dump(configuration, file, sort_keys=True, indent=4, ensure_ascii=False)
+
+        print("[!] A new configuration was created!")
+        print("Please check the settings in the config.json!")
+        exit(1)
+
+
+if __name__ == '__main__':
+    main()

socket_server.py

@@ -0,0 +1,97 @@
+import _thread
+import json
+import socket
+import uuid
+
+import byte_utils
+
+
+class SocketServer:
+    ip = None
+    port = None
+    motd = None
+    version_text = None
+    kick_message = None
+    samples = None
+    server_icon = None
+    s = None
+
+    def __init__(self, ip, port, motd, version_text, kick_message, samples, server_icon):
+        self.ip = ip
+        self.port = port
+        self.motd = motd
+        self.version_text = version_text
+        self.kick_message = kick_message
+        self.samples = samples
+        self.server_icon = server_icon
+
+    def on_new_client(self, client_socket, addr):
+        data = client_socket.recv(1024)
+        (length, i) = byte_utils.read_varint(data, 0)
+        (packetID, i) = byte_utils.read_varint(data, i)
+
+        if packetID == 0:
+            (version, i) = byte_utils.read_varint(data, i)
+            (ip, i) = byte_utils.read_utf(data, i)
+            (port_tuple, i) = byte_utils.read_ushort(data, i)
+            (state, i) = byte_utils.read_varint(data, i)
+            if state == 1:
+                print("%s:%s has pinged the server (%s:%s)" % (addr[0], addr[1], ip, port_tuple[0]))
+
+                motd = {}
+                motd["version"] = {}
+                motd["version"]["name"] = self.version_text
+                motd["version"]["protocol"] = 2
+                motd["players"] = {}
+                motd["players"]["max"] = 0
+                motd["players"]["online"] = 0
+                motd["players"]["sample"] = []
+
+                for sample in self.samples:
+                    motd["players"]["sample"].append({"name": sample, "id": str(uuid.uuid4())})
+
+                motd["description"] = {"text": self.motd}
+
+                if len(self.server_icon) > 0:
+                    motd["favicon"] = self.server_icon
+
+                print(motd)
+
+                self.write_response(client_socket, json.dumps(motd))
+            elif state == 2:
+                print("%s:%s has tried to connect to the server (%s:%s)" % (addr[0], addr[1], ip, port_tuple[0]))
+                self.write_response(client_socket, json.dumps({"text": self.kick_message}))
+            else:
+                print("%s:%s tried to request a login/ping with a unknown state: %s" % (addr[0], addr[1], state))
+        elif packetID == 1:
+            (long, i) = byte_utils.read_long(data, i)
+            response = bytearray()
+            byte_utils.write_varint(response, 9)
+            byte_utils.write_varint(response, 1)
+            bytearray.append(long)
+            client_socket.sendall(bytearray)
+            print("Send pong to %s" % addr)
+        else:
+            print("%s tried to request a unknown packet: %s" % (addr, packetID))
+
+    def write_response(self, client_socket, response):
+        response_array = bytearray()
+        byte_utils.write_varint(response_array, 0)
+        byte_utils.write_utf(response_array, response)
+        length = bytearray()
+        byte_utils.write_varint(length, len(response_array))
+        client_socket.sendall(length)
+        client_socket.sendall(response_array)
+
+    def start(self):
+        self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.s.bind((self.ip, self.port))
+        self.s.settimeout(5000)
+        self.s.listen(30)
+        print("Server started!")
+        while 1:
+            (client, address) = self.s.accept()
+            _thread.start_new_thread(self.on_new_client, (client, address,))
+
+    def close(self):
+        self.s.close()