ewized · October 5, 2025 05:50 · Oct 3, 2021 · Oct 3, 2021 · Sep 28, 2015 · Sep 28, 2015
diff --git a/statusping.py b/statusping.py
@@ -23,6 +23,7 @@
 # OTHER DEALINGS IN THE SOFTWARE.
 #
 # For more information, please refer to <http://unlicense.org/>
+
 import socket
 import struct
 import json

diff --git a/statusping.py b/statusping.py
@@ -1,4 +1,28 @@
 #!/usr/bin/python3
+# This is free and unencumbered software released into the public domain.
+#
+# Anyone is free to copy, modify, publish, use, compile, sell, or
+# distribute this software, either in source code form or as a compiled
+# binary, for any purpose, commercial or non-commercial, and by any
+# means.
+#
+# In jurisdictions that recognize copyright laws, the author or authors
+# of this software dedicate any and all copyright interest in the
+# software to the public domain. We make this dedication for the benefit
+# of the public at large and to the detriment of our heirs and
+# successors. We intend this dedication to be an overt act of
+# relinquishment in perpetuity of all present and future rights to this
+# software under copyright law.
+#
+# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
+# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
+# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
+# IN NO EVENT SHALL THE AUTHORS BE LIABLE FOR ANY CLAIM, DAMAGES OR
+# OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE,
+# ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+# OTHER DEALINGS IN THE SOFTWARE.
+#
+# For more information, please refer to <http://unlicense.org/>
 import socket
 import struct
 import json

diff --git a/statusping.py b/statusping.py
@@ -7,10 +7,11 @@
 class StatusPing:
     """ Get the ping status for the Minecraft server """
 
-    def __init__(self, host='localhost', port=25565):
+    def __init__(self, host='localhost', port=25565, timeout=5):
         """ Init the hostname and the port """
         self._host = host
         self._port = port
+        self._timeout = timeout
 
     def _unpack_varint(self, sock):
         """ Unpack the varint """
@@ -88,6 +89,7 @@ def _read_fully(self, connection, extra_varint=False):
     def get_status(self):
         """ Get the status response """
         with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as connection:
+            connection.settimeout(self._timeout)
             connection.connect((self._host, self._port))
 
             # Send handshake + status request

diff --git a/statusping.py b/statusping.py
@@ -71,6 +71,10 @@ def _read_fully(self, connection, extra_varint=False):
         byte = b''
 
         if extra_varint:
+            # Packet contained netty header offset for this
+            if packet_id > packet_length:
+                self._unpack_varint(connection)
+
             extra_length = self._unpack_varint(connection)
 
             while len(byte) < extra_length:

diff --git a/statusping.py b/statusping.py
@@ -2,6 +2,7 @@
 import socket
 import struct
 import json
+import time
 
 class StatusPing:
     """ Get the ping status for the Minecraft server """

diff --git a/mcping.py → statusping.py b/mcping.py → statusping.py
diff --git a/mcping.py b/mcping.py
@@ -0,0 +1,103 @@
+#!/usr/bin/python3
+import socket
+import struct
+import json
+
+class StatusPing:
+    """ Get the ping status for the Minecraft server """
+
+    def __init__(self, host='localhost', port=25565):
+        """ Init the hostname and the port """
+        self._host = host
+        self._port = port
+
+    def _unpack_varint(self, sock):
+        """ Unpack the varint """
+        data = 0
+        for i in range(5):
+            ordinal = sock.recv(1)
+
+            if len(ordinal) == 0:
+                break
+
+            byte = ord(ordinal)
+            data |= (byte & 0x7F) << 7*i
+
+            if not byte & 0x80:
+                break
+
+        return data
+
+    def _pack_varint(self, data):
+        """ Pack the var int """
+        ordinal = b''
+
+        while True:
+            byte = data & 0x7F
+            data >>= 7
+            ordinal += struct.pack('B', byte | (0x80 if data > 0 else 0))
+
+            if data == 0:
+                break
+
+        return ordinal
+
+    def _pack_data(self, data):
+        """ Page the data """
+        if type(data) is str:
+            data = data.encode('utf8')
+            return self._pack_varint(len(data)) + data
+        elif type(data) is int:
+            return struct.pack('H', data)
+        elif type(data) is float:
+            return struct.pack('L', int(data))
+        else:
+            return data
+
+    def _send_data(self, connection, *args):
+        """ Send the data on the connection """
+        data = b''
+
+        for arg in args:
+            data += self._pack_data(arg)
+
+        connection.send(self._pack_varint(len(data)) + data)
+
+    def _read_fully(self, connection, extra_varint=False):
+        """ Read the connection and return the bytes """
+        packet_length = self._unpack_varint(connection)
+        packet_id = self._unpack_varint(connection)
+        byte = b''
+
+        if extra_varint:
+            extra_length = self._unpack_varint(connection)
+
+            while len(byte) < extra_length:
+                byte += connection.recv(extra_length)
+
+        else:
+            byte = connection.recv(packet_length)
+
+        return byte
+
+    def get_status(self):
+        """ Get the status response """
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as connection:
+            connection.connect((self._host, self._port))
+
+            # Send handshake + status request
+            self._send_data(connection, b'\x00\x00', self._host, self._port, b'\x01')
+            self._send_data(connection, b'\x00')
+
+            # Read response, offset for string length
+            data = self._read_fully(connection, extra_varint=True)
+
+            # Send and read unix time
+            self._send_data(connection, b'\x01', time.time() * 1000)
+            unix = self._read_fully(connection)
+
+        # Load json and return
+        response = json.loads(data.decode('utf8'))
+        response['ping'] = int(time.time() * 1000) - struct.unpack('L', unix)[0]
+
+        return response