Examples¶
The quarry source tree includes a few example uses of the quarry
module:
# List examples
$ python -m examples
# Run an example
$ python -m examples.client_messenger
If you have quarry
in your python search path, you can run the example
files directly.
Clients¶
Pinger¶
"""
Pinger example client
This example client connects to a server in "status" mode to retrieve some
information about the server. The information returned is what you'd normally
see in the "Multiplayer" menu of the official client.
"""
from twisted.internet import reactor
from quarry.net.client import ClientFactory, ClientProtocol
class PingProtocol(ClientProtocol):
def status_response(self, data):
for k, v in sorted(data.items()):
if k != "favicon":
self.logger.info("%s --> %s" % (k, v))
reactor.stop()
class PingFactory(ClientFactory):
protocol = PingProtocol
protocol_mode_next = "status"
def main(argv):
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("host")
parser.add_argument("-p", "--port", default=25565, type=int)
args = parser.parse_args(argv)
factory = PingFactory()
factory.connect(args.host, args.port)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Player Lister¶
"""
Player lister example client
This client requires a Mojang account for online-mode servers. It logs in to
the server and prints the players listed in the tab menu.
"""
from twisted.internet import reactor, defer
from quarry.net.client import ClientFactory, ClientProtocol
from quarry.net.auth import ProfileCLI
class PlayerListProtocol(ClientProtocol):
def setup(self):
self.players = {}
def packet_player_list_item(self, buff):
# 1.7.x
if self.protocol_version <= 5:
p_player_name = buff.unpack_string()
p_online = buff.unpack('?')
p_ping = buff.unpack('h')
if p_online:
self.players[p_player_name] = {
'name': p_player_name,
'ping': p_ping
}
elif p_player_name in self.players:
del self.players[p_player_name]
# 1.8.x
else:
p_action = buff.unpack_varint()
p_count = buff.unpack_varint()
for i in range(p_count):
p_uuid = buff.unpack_uuid()
if p_action == 0: # ADD_PLAYER
p_player_name = buff.unpack_string()
p_properties_count = buff.unpack_varint()
p_properties = {}
for j in range(p_properties_count):
p_property_name = buff.unpack_string()
p_property_value = buff.unpack_string()
p_property_is_signed = buff.unpack('?')
if p_property_is_signed:
p_property_signature = buff.unpack_string()
p_properties[p_property_name] = p_property_value
p_gamemode = buff.unpack_varint()
p_ping = buff.unpack_varint()
p_has_display_name = buff.unpack('?')
if p_has_display_name:
p_display_name = buff.unpack_chat()
else:
p_display_name = None
self.players[p_uuid] = {
'name': p_player_name,
'properties': p_properties,
'gamemode': p_gamemode,
'ping': p_ping,
'display_name': p_display_name
}
elif p_action == 1: # UPDATE_GAMEMODE
p_gamemode = buff.unpack_varint()
if p_uuid in self.players:
self.players[p_uuid]['gamemode'] = p_gamemode
elif p_action == 2: # UPDATE_LATENCY
p_ping = buff.unpack_varint()
if p_uuid in self.players:
self.players[p_uuid]['ping'] = p_ping
elif p_action == 3: # UPDATE_DISPLAY_NAME
p_has_display_name = buff.unpack('?')
if p_has_display_name:
p_display_name = buff.unpack_chat()
else:
p_display_name = None
if p_uuid in self.players:
self.players[p_uuid]['display_name'] = p_display_name
elif p_action == 4: # REMOVE_PLAYER
if p_uuid in self.players:
del self.players[p_uuid]
def packet_chunk_data(self, buff):
buff.discard()
# convert self.players into a more readable format
printable_players = []
for data in self.players.values():
printable_players.append((data['name'], data['ping']))
for display_name, ping in sorted(printable_players):
self.logger.info("%4sms %s" % (ping, display_name))
reactor.stop()
class PlayerListFactory(ClientFactory):
protocol = PlayerListProtocol
@defer.inlineCallbacks
def run(args):
# Log in
profile = yield ProfileCLI.make_profile(args)
# Create factory
factory = PlayerListFactory(profile)
# Connect!
factory.connect(args.host, args.port)
def main(argv):
parser = ProfileCLI.make_parser()
parser.add_argument("host")
parser.add_argument("-p", "--port", default=25565, type=int)
args = parser.parse_args(argv)
run(args)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Chat Logger¶
"""
Chat logger example client
This client stays in-game after joining. It prints chat messages received from
the server and slowly rotates (thanks c45y for the idea).
"""
from twisted.internet import reactor, defer
from quarry.net.client import ClientFactory, SpawningClientProtocol
from quarry.net.auth import ProfileCLI
class ChatLoggerProtocol(SpawningClientProtocol):
def packet_chat_message(self, buff):
p_text = buff.unpack_chat()
# 1.7.x
if self.protocol_version <= 5:
pass
# 1.8.x
else:
p_position = buff.unpack('B')
self.logger.info(":: %s" % p_text)
class ChatLoggerFactory(ClientFactory):
protocol = ChatLoggerProtocol
@defer.inlineCallbacks
def run(args):
# Log in
profile = yield ProfileCLI.make_profile(args)
# Create factory
factory = ChatLoggerFactory(profile)
# Connect!
factory.connect(args.host, args.port)
def main(argv):
parser = ProfileCLI.make_parser()
parser.add_argument("host")
parser.add_argument("-p", "--port", default=25565, type=int)
args = parser.parse_args(argv)
run(args)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Messenger¶
"""
Messenger example client
Bridges minecraft chat (in/out) with stdout and stdin.
"""
import os
import sys
from twisted.internet import defer, reactor, stdio
from twisted.protocols import basic
from quarry.net.auth import ProfileCLI
from quarry.net.client import ClientFactory, SpawningClientProtocol
class StdioProtocol(basic.LineReceiver):
delimiter = os.linesep.encode('ascii')
in_encoding = getattr(sys.stdin, "encoding", 'utf8')
out_encoding = getattr(sys.stdout, "encoding", 'utf8')
def lineReceived(self, line):
self.minecraft_protocol.send_chat(line.decode(self.in_encoding))
def send_line(self, text):
self.sendLine(text.encode(self.out_encoding))
class MinecraftProtocol(SpawningClientProtocol):
spawned = False
def packet_chat_message(self, buff):
p_text = buff.unpack_chat().to_string()
# 1.7.x
if self.protocol_version <= 5:
p_position = 0
# 1.8.x
else:
p_position = buff.unpack('B')
if p_position in (0, 1) and p_text.strip():
self.stdio_protocol.send_line(p_text)
def send_chat(self, text):
self.send_packet("chat_message", self.buff_type.pack_string(text))
class MinecraftFactory(ClientFactory):
protocol = MinecraftProtocol
log_level = "WARN"
def buildProtocol(self, addr):
minecraft_protocol = super(MinecraftFactory, self).buildProtocol(addr)
stdio_protocol = StdioProtocol()
minecraft_protocol.stdio_protocol = stdio_protocol
stdio_protocol.minecraft_protocol = minecraft_protocol
stdio.StandardIO(stdio_protocol)
return minecraft_protocol
@defer.inlineCallbacks
def run(args):
# Log in
profile = yield ProfileCLI.make_profile(args)
# Create factory
factory = MinecraftFactory(profile)
# Connect!
factory.connect(args.host, args.port)
def main(argv):
parser = ProfileCLI.make_parser()
parser.add_argument("host")
parser.add_argument("port", nargs='?', default=25565, type=int)
args = parser.parse_args(argv)
run(args)
reactor.run()
if __name__ == "__main__":
main(sys.argv[1:])
Servers¶
Downtime Server¶
"""
Example "downtime" server
This server kicks players with the MOTD when they try to connect. It can be
useful for when you want players to know that your usual server is down for
maintenance.
"""
from twisted.internet import reactor
from quarry.net.server import ServerFactory, ServerProtocol
class DowntimeProtocol(ServerProtocol):
def packet_login_start(self, buff):
buff.discard()
self.close(self.factory.motd)
class DowntimeFactory(ServerFactory):
protocol = DowntimeProtocol
def main(argv):
# Parse options
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-a", "--host", default="", help="address to listen on")
parser.add_argument("-p", "--port", default=25565, type=int, help="port to listen on")
parser.add_argument("-m", "--message", default="We're down for maintenance",
help="message to kick users with")
args = parser.parse_args(argv)
# Create factory
factory = DowntimeFactory()
factory.motd = args.message
# Listen
factory.listen(args.host, args.port)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Auth Server¶
"""
Example "auth" server
This server authenticates players with the mojang session server, then kicks
them. Useful for server websites that ask users for a valid Minecraft account.
"""
from twisted.internet import reactor
from quarry.net.server import ServerFactory, ServerProtocol
class AuthProtocol(ServerProtocol):
def player_joined(self):
# This method gets called when a player successfully joins the server.
# If we're in online mode (the default), this means auth with the
# session server was successful and the user definitely owns the
# display name they claim to.
# Call super. This switches us to "play" mode, marks the player as
# in-game, and does some logging.
ServerProtocol.player_joined(self)
# Define your own logic here. It could be an HTTP request to an API,
# or perhaps an update to a database table.
display_name = self.display_name
ip_addr = self.remote_addr.host
self.logger.info("[%s authed with IP %s]" % (display_name, ip_addr))
# Kick the player.
self.close("Thanks, you are now registered!")
class AuthFactory(ServerFactory):
protocol = AuthProtocol
motd = "Auth Server"
def main(argv):
# Parse options
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-a", "--host", default="", help="address to listen on")
parser.add_argument("-p", "--port", default=25565, type=int, help="port to listen on")
args = parser.parse_args(argv)
# Create factory
factory = AuthFactory()
# Listen
factory.listen(args.host, args.port)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Chat Room Server¶
"""
Example "chat room" server
This server authenticates players, then spawns them in an empty world and does
the bare minimum to keep them in-game. Players can speak to eachother using
chat.
Supports Minecraft 1.15. Earlier versions will not work as the packet formats
differ.
"""
from twisted.internet import reactor
from quarry.net.server import ServerFactory, ServerProtocol
class ChatRoomProtocol(ServerProtocol):
def player_joined(self):
# Call super. This switches us to "play" mode, marks the player as
# in-game, and does some logging.
ServerProtocol.player_joined(self)
# Send "Join Game" packet
self.send_packet("join_game",
self.buff_type.pack("iBqiB",
0, # entity id
3, # game mode
0, # dimension
0, # hashed seed
0), # max players
self.buff_type.pack_string("flat"), # level type
self.buff_type.pack_varint(1), # view distance
self.buff_type.pack("??",
False, # reduced debug info
True)) # show respawn screen
# Send "Player Position and Look" packet
self.send_packet("player_position_and_look",
self.buff_type.pack("dddff?",
0, # x
255, # y
0, # z
0, # yaw
0, # pitch
0b00000), # flags
self.buff_type.pack_varint(0)) # teleport id
# Start sending "Keep Alive" packets
self.ticker.add_loop(20, self.update_keep_alive)
# Announce player joined
self.factory.send_chat(u"\u00a7e%s has joined." % self.display_name)
def player_left(self):
ServerProtocol.player_left(self)
# Announce player left
self.factory.send_chat(u"\u00a7e%s has left." % self.display_name)
def update_keep_alive(self):
# Send a "Keep Alive" packet
# 1.7.x
if self.protocol_version <= 338:
payload = self.buff_type.pack_varint(0)
# 1.12.2
else:
payload = self.buff_type.pack('Q', 0)
self.send_packet("keep_alive", payload)
def packet_chat_message(self, buff):
# When we receive a chat message from the player, ask the factory
# to relay it to all connected players
p_text = buff.unpack_string()
self.factory.send_chat("<%s> %s" % (self.display_name, p_text))
class ChatRoomFactory(ServerFactory):
protocol = ChatRoomProtocol
motd = "Chat Room Server"
def send_chat(self, message):
for player in self.players:
player.send_packet("chat_message",player.buff_type.pack_chat(message) + player.buff_type.pack('B', 0) )
def main(argv):
# Parse options
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-a", "--host", default="", help="address to listen on")
parser.add_argument("-p", "--port", default=25565, type=int, help="port to listen on")
args = parser.parse_args(argv)
# Create factory
factory = ChatRoomFactory()
# Listen
factory.listen(args.host, args.port)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])
Proxies¶
Chat Hider Proxy¶
"""
"Quiet mode" example proxy
Allows a client to turn on "quiet mode" which hides chat messages
"""
from twisted.internet import reactor
from quarry.net.proxy import DownstreamFactory, Bridge
class QuietBridge(Bridge):
quiet_mode = False
def packet_upstream_chat_message(self, buff):
buff.save()
chat_message = self.read_chat(buff, "upstream")
self.logger.info(" >> %s" % chat_message)
if chat_message.startswith("/quiet"):
# Switch mode
self.quiet_mode = not self.quiet_mode
action = self.quiet_mode and "enabled" or "disabled"
msg = "Quiet mode %s" % action
self.downstream.send_packet("chat_message",
self.write_chat(msg, "downstream"))
elif self.quiet_mode and not chat_message.startswith("/"):
# Don't let the player send chat messages in quiet mode
msg = "Can't send messages while in quiet mode"
self.downstream.send_packet("chat_message",
self.write_chat(msg, "downstream"))
else:
# Pass to upstream
buff.restore()
self.upstream.send_packet("chat_message", buff.read())
def packet_downstream_chat_message(self, buff):
chat_message = self.read_chat(buff, "downstream")
self.logger.info(" :: %s" % chat_message)
if self.quiet_mode and chat_message.startswith("<"):
# Ignore message we're in quiet mode and it looks like chat
pass
else:
# Pass to downstream
buff.restore()
self.downstream.send_packet("chat_message", buff.read())
def read_chat(self, buff, direction):
buff.save()
if direction == "upstream":
p_text = buff.unpack_string()
return p_text
elif direction == "downstream":
p_text = str(buff.unpack_chat())
# 1.7.x
if self.upstream.protocol_version <= 5:
p_position = 0
# 1.8.x
else:
p_position = buff.unpack('B')
if p_position in (0, 1):
return p_text
def write_chat(self, text, direction):
if direction == "upstream":
return self.buff_type.pack_string(text)
elif direction == "downstream":
data = self.buff_type.pack_chat(text)
# 1.7.x
if self.downstream.protocol_version <= 5:
pass
# 1.8.x
else:
data += self.buff_type.pack('B', 0)
return data
class QuietDownstreamFactory(DownstreamFactory):
bridge_class = QuietBridge
motd = "Proxy Server"
def main(argv):
# Parse options
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("-a", "--listen-host", default="", help="address to listen on")
parser.add_argument("-p", "--listen-port", default=25565, type=int, help="port to listen on")
parser.add_argument("-b", "--connect-host", default="127.0.0.1", help="address to connect to")
parser.add_argument("-q", "--connect-port", default=25565, type=int, help="port to connect to")
args = parser.parse_args(argv)
# Create factory
factory = QuietDownstreamFactory()
factory.connect_host = args.connect_host
factory.connect_port = args.connect_port
# Listen
factory.listen(args.listen_host, args.listen_port)
reactor.run()
if __name__ == "__main__":
import sys
main(sys.argv[1:])