Moved mojang.api.net.slp to mojang.minecraft.proto.slp

mojang/minecraft/proto/slp/__init__.py

@@ -0,0 +1,23 @@
+import socket
+from typing import Tuple
+
+from .current import slp
+from .old import slp_1_6, slp_prior_1_4, slp_prior_1_6
+
+
+def ping(addr: Tuple[str, int], timeout: int = 3):
+    response = None
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.connect(addr)
+        sock.settimeout(timeout)
+
+        _methods = [slp, slp_1_6, slp_prior_1_6, slp_prior_1_6]
+        while len(_methods) > 0 and response is None:
+            method = _methods.pop(0)
+            try:
+                response = method(sock, addr[0], addr[1])
+            except socket.error:
+                pass
+
+    return response
+

mojang/minecraft/proto/slp/_structures.py

@@ -0,0 +1,15 @@
+from dataclasses import dataclass, field
+from typing import Tuple, List
+
+@dataclass(frozen=True)
+class Players:
+    count: Tuple[int, int] = field()
+    list: List[str] = field()
+
+@dataclass(frozen=True)
+class SLPResponse:
+    protocol_version: int = field()
+    version: str = field()
+    motd: str = field()
+    players: Players = field()
+    ping: int = field()

mojang/minecraft/proto/slp/current.py

@@ -0,0 +1,104 @@
+import socket
+import struct
+import io
+import time
+import json
+from typing import IO
+
+from ._structures import SLPResponse, Players
+
+# Types helper
+def _write_varint(buffer: IO, value: int):
+    while True:
+        byte = value & 0x7f
+        value >>= 7
+
+        if value > 0:
+            byte |= 0x80
+
+        buffer.write(struct.pack('B', byte))
+
+        if value == 0:
+            break
+
+def _read_varint(buffer: IO):
+    val = 0
+
+    for i in range(5):
+        byte = buffer.read(1)
+        if len(byte) == 0:
+            break
+
+        val |= (ord(byte) & 0x7f) << (7*i)
+
+        if ord(byte) & 0x80 == 0:
+            break
+
+    return val
+
+def _write_string(buffer: IO, value: str, encoding: str = 'utf-8'):
+    value = value.encode(encoding)
+    _write_varint(buffer, len(value))
+    buffer.write(value)
+
+def _read_string(buffer: IO, encoding: str = 'utf-8'):
+    length = _read_varint(buffer)
+    return buffer.read(length).decode(encoding)
+
+# Packet helper
+def _write_packet(sock: socket.socket, packet: bytes):
+    with sock.makefile('wb') as buffer:
+        _write_varint(buffer, len(packet))
+        buffer.write(packet)
+
+def _read_packet(sock: socket.socket):
+    with sock.makefile('rb') as buffer:
+        length = _read_varint(buffer)
+        packet = buffer.read(length)
+
+    return packet
+
+
+def slp(sock: socket.socket, hostname: str = 'locahost', port: int = 25565):
+    # Send handshake packet
+    with io.BytesIO() as buffer:
+        _write_varint(buffer, 0x00)
+        _write_varint(buffer, 0)
+        _write_string(buffer, hostname)
+        buffer.write(struct.pack('>h', port))
+        _write_varint(buffer, 0x01)
+
+        _write_packet(sock, buffer.getvalue())
+
+    # Send request packet
+    with io.BytesIO() as buffer:
+        _write_varint(buffer, 0x00)
+
+        _write_packet(sock, buffer.getvalue())
+
+    # Receive response
+    packet = _read_packet(sock)    
+    with io.BytesIO(packet) as buffer:
+        packet_id = _read_varint(buffer)
+        response = json.loads(_read_string(buffer))
+
+    # Ping
+    with io.BytesIO() as buffer:
+        _write_varint(buffer, 0x01)
+        buffer.write(struct.pack('>q', int(time.time() * 1000)))
+
+        _write_packet(sock, buffer.getvalue())
+
+    # Pong
+    packet = _read_packet(sock)
+    with io.BytesIO(packet) as buffer:
+        packet_id = _read_varint(buffer)
+        ping_start = struct.unpack('>q', buffer.read(8))[0]
+
+    return SLPResponse(
+        protocol_version=response['version']['protocol'], 
+        version=response['version']['name'], 
+        motd=response['description']['text'],
+        players=Players((response['players']['online'], response['players']['max']), response['players'].get('sample', [])),
+        ping=int(time.time() * 1000) - ping_start
+    )

mojang/minecraft/proto/slp/old.py

@@ -0,0 +1,57 @@
+import socket
+import struct
+
+from ._structures import Players, SLPResponse
+
+
+def slp_1_6(sock: socket.socket, hostname: str = 'localhost', port: int = 25565):
+    # Send request
+    encoded_hostname = hostname.encode('utf-16be')
+    header = struct.pack('>BBBh22s', 0xFE, 0x01, 0xFA, 11,'MC|PingHost'.encode('utf-16be'))
+    body = struct.pack('>hBh{}si'.format(len(encoded_hostname)), 7 + len(encoded_hostname), 80, len(hostname), encoded_hostname, port)
+    packet = header + body
+
+    sock.send(packet)
+
+    # Receive response
+    with sock.makefile('rb') as buffer:
+        buffer.read(9) # Skip 9 first bytes
+        protocol_ver, version, motd, nplayers, maxplayers = buffer.read().decode('utf-16-be').split('\x00')
+
+    return SLPResponse(
+        protocol_version=protocol_ver, 
+        version=version, 
+        motd=motd,
+        players=Players((int(nplayers), int(maxplayers)), []),
+        ping=None
+    )
+
+def slp_prior_1_6(sock: socket.socket, *args):
+    sock.send(struct.pack('>BB', 0xFE, 0x01))
+
+    with sock.makefile('rb') as buffer:
+        buffer.read(9) # Skip 9 first bytes
+        protocol_ver, version, motd, nplayers, maxplayers = buffer.read().decode('utf-16-be').split('\x00')
+
+    return SLPResponse(
+        protocol_version=protocol_ver, 
+        version=version, 
+        motd=motd,
+        players=Players((int(nplayers), int(maxplayers)), []),
+        ping=None
+    )
+
+def slp_prior_1_4(sock: socket.socket, *args):
+    sock.send(struct.pack('>B', 0xFE))
+
+    with sock.makefile('rb') as buffer:
+        buffer.read(3) # Skip 3 first bytes
+        motd, nplayers, maxplayers = buffer.read().decode('utf-16-be').split('\xa7')
+
+    return SLPResponse(
+        protocol_version=None, 
+        version=None, 
+        motd=motd,
+        players=Players((int(nplayers), int(maxplayers)), []),
+        ping=None
+    )