Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def write_packet_names():
latest_protocol = packets.default_protocol_version
latest_minecraft = packets.minecraft_versions[latest_protocol]
text = textwrap.dedent("""
Packet Names
============
See the `Minecraft Coalition Wiki`_ for a details on every packet.
.. _Minecraft Coalition Wiki: http://wiki.vg/Protocol
""")
title = "Minecraft %s" % latest_minecraft
text += "\n%s\n%s\n\n" % (title, "-"*len(title))
names = defaultdict(set)
for key, name in packets.packet_names.items():
protocol, _, direction, _ = key
def get_packet_name(self, ident):
key = (self.protocol_version, self.protocol_mode, self.recv_direction,
ident)
try:
return packets.packet_names[key]
except KeyError:
raise ProtocolError("No name known for packet: %s" % (key,))
# Encrypt
data = self.cipher.encrypt(data)
# Send
self.transport.write(data)
class Factory(protocol.Factory, object):
protocol = Protocol
ticker_type = Ticker
log_level = logging.INFO
connection_timeout = 30
force_protocol_version = None
minecraft_versions = packets.minecraft_versions
def buildProtocol(self, addr):
return self.protocol(self, addr)
def get_buff_type(self, protocol_version):
"""
Gets a buffer type for the given protocol version.
"""
for ver, cls in reversed(buff_types):
if protocol_version >= ver:
return cls
def get_packet_ident(self, name):
key = (self.protocol_version, self.protocol_mode, self.send_direction,
name)
try:
return packets.packet_idents[key]
except KeyError:
raise ProtocolError("No ID known for packet: %s" % (key,))
def write_packet_names():
latest_protocol = packets.default_protocol_version
latest_minecraft = packets.minecraft_versions[latest_protocol]
text = textwrap.dedent("""
Packet Names
============
See the `Minecraft Coalition Wiki`_ for a details on every packet.
.. _Minecraft Coalition Wiki: http://wiki.vg/Protocol
""")
title = "Minecraft %s" % latest_minecraft
text += "\n%s\n%s\n\n" % (title, "-"*len(title))
names = defaultdict(set)
for key, name in packets.packet_names.items():
#: The logger for this protocol.
logger = None
#: A reference to a :class:`Ticker` instance.
ticker = None
#: A reference to the factory
factory = None
#: The IP address of the remote.
remote_addr = None
recv_direction = None
send_direction = None
protocol_version = packets.default_protocol_version
protocol_mode = "init"
compression_threshold = -1
in_game = False
closed = False
def __init__(self, factory, remote_addr):
self.factory = factory
self.remote_addr = remote_addr
self.buff_type = self.factory.get_buff_type(self.protocol_version)
self.recv_buff = self.buff_type()
self.cipher = Cipher()
self.logger = logging.getLogger("%s{%s}" % (
self.__class__.__name__,
self.remote_addr.host))