minecraft/networking/connection.py

   1 from collections import deque
   2 from threading import RLock
   3 import zlib
   4 import threading
   5 import socket
   6 import timeit
   7 import select
   8 import sys
   9 import json
  10 import re
  11
  12 from .types import VarInt
  13 from .packets import clientbound, serverbound
  14 from . import packets, encryption
  15 from .. import (
  16     utility, KNOWN_MINECRAFT_VERSIONS, SUPPORTED_MINECRAFT_VERSIONS,
  17     SUPPORTED_PROTOCOL_VERSIONS, PROTOCOL_VERSION_INDICES
  18 )
  19 from ..exceptions import (
  20     VersionMismatch, LoginDisconnect, IgnorePacket, InvalidState
  21 )
  22
  23
  24 STATE_STATUS = 1
  25 STATE_PLAYING = 2
  26
  27
  28 class ConnectionContext(object):
  29     """A ConnectionContext encapsulates the static configuration parameters
  30     shared by the Connection class with other classes, such as Packet.
  31     Importantly, it can be used without knowing the interface of Connection.
  32     """
  33     def __init__(self, **kwds):
  34         self.protocol_version = kwds.get('protocol_version')
  35
  36     def protocol_earlier(self, other_pv):
  37         """Returns True if the protocol version of this context was published
  38            earlier than 'other_pv', or else False."""
  39         return utility.protocol_earlier(self.protocol_version, other_pv)
  40
  41     def protocol_earlier_eq(self, other_pv):
  42         """Returns True if the protocol version of this context was published
  43            earlier than, or is equal to, 'other_pv', or else False."""
  44         return utility.protocol_earlier_eq(self.protocol_version, other_pv)
  45
  46     def protocol_later(self, other_pv):
  47         """Returns True if the protocol version of this context was published
  48            later than 'other_pv', or else False."""
  49         return utility.protocol_earlier(other_pv, self.protocol_version)
  50
  51     def protocol_later_eq(self, other_pv):
  52         """Returns True if the protocol version of this context was published
  53            later than, or is equal to, 'other_pv', or else False."""
  54         return utility.protocol_earlier_eq(other_pv, self.protocol_version)
  55
  56     def protocol_in_range(self, start_pv, end_pv):
  57         """Returns True if the protocol version of this context was published
  58            later than, or is equal to, 'start_pv' and was published earlier
  59            than 'end_pv' (analogously to Python's 'range' function)."""
  60         return (utility.protocol_earlier(self.protocol_version, end_pv) and
  61                 utility.protocol_earlier_eq(start_pv, self.protocol_version))
  62
  63
  64 class _ConnectionOptions(object):
  65     def __init__(self, address=None, port=None, compression_threshold=-1,
  66                  compression_enabled=False):
  67         self.address = address
  68         self.port = port
  69         self.compression_threshold = compression_threshold
  70         self.compression_enabled = compression_enabled
  71
  72
  73 class Connection(object):
  74     """This class represents a connection to a minecraft
  75     server, it handles everything from connecting, sending packets to
  76     handling default network behaviour
  77     """
  78     def __init__(
  79         self,
  80         address,
  81         port=25565,
  82         auth_token=None,
  83         username=None,
  84         initial_version=None,
  85         allowed_versions=None,
  86         handle_exception=None,
  87         handle_exit=None,
  88     ):
  89         """Sets up an instance of this object to be able to connect to a
  90         minecraft server.
  91
  92         The connect method needs to be called in order to actually begin
  93         the connection
  94
  95         :param address: address of the server to connect to
  96         :param port(int): port of the server to connect to
  97         :param auth_token: :class:`minecraft.authentication.AuthenticationToken`
  98                            object. If None, no authentication is attempted and
  99                            the server is assumed to be running in offline mode.
 100         :param username: Username string; only applicable in offline mode.
 101         :param initial_version: A Minecraft version ID string or protocol
 102                                 version number to use if the server's protocol
 103                                 version cannot be determined. (Although it is
 104                                 now somewhat inaccurate, this name is retained
 105                                 for backward compatibility.)
 106         :param allowed_versions: A set of versions, each being a Minecraft
 107                                  version ID string or protocol version number,
 108                                  restricting the versions that the client may
 109                                  use in connecting to the server.
 110         :param handle_exception: The final exception handler. This is triggered
 111                                  when an exception occurs in the networking
 112                                  thread that is not caught normally. After
 113                                  any other user-registered exception handlers
 114                                  are run, the final exception (which may be the
 115                                  original exception or one raised by another
 116                                  handler) is passed, regardless of whether or
 117                                  not it was caught by another handler, to the
 118                                  final handler, which may be a function obeying
 119                                  the protocol of 'register_exception_handler';
 120                                  the value 'None', meaning that if the
 121                                  exception was otherwise uncaught, it is
 122                                  re-raised from the networking thread after
 123                                  closing the connection; or the value 'False',
 124                                  meaning that the exception is never re-raised.
 125         :param handle_exit: A function to be called when a connection to a
 126                             server terminates, not caused by an exception,
 127                             and not with the intention to automatically
 128                             reconnect. Exceptions raised from this function
 129                             will be handled by any matching exception handlers.
 130         """  # NOQA
 131
 132         # This lock is re-entrant because it may be acquired in a re-entrant
 133         # manner from within an outgoing packet
 134         self._write_lock = RLock()
 135
 136         self.networking_thread = None
 137         self.new_networking_thread = None
 138         self.packet_listeners = []
 139         self.early_packet_listeners = []
 140         self.outgoing_packet_listeners = []
 141         self.early_outgoing_packet_listeners = []
 142         self._exception_handlers = []
 143
 144         def proto_version(version):
 145             if isinstance(version, str):
 146                 proto_version = SUPPORTED_MINECRAFT_VERSIONS.get(version)
 147             elif isinstance(version, int):
 148                 proto_version = version
 149             else:
 150                 proto_version = None
 151             if proto_version not in SUPPORTED_PROTOCOL_VERSIONS:
 152                 raise ValueError('Unsupported version number: %r.' % version)
 153             return proto_version
 154
 155         if allowed_versions is None:
 156             self.allowed_proto_versions = set(SUPPORTED_PROTOCOL_VERSIONS)
 157         else:
 158             allowed_versions = set(map(proto_version, allowed_versions))
 159             self.allowed_proto_versions = allowed_versions
 160
 161         latest_allowed_proto = max(self.allowed_proto_versions,
 162                                    key=PROTOCOL_VERSION_INDICES.get)
 163
 164         if initial_version is None:
 165             self.default_proto_version = latest_allowed_proto
 166         else:
 167             self.default_proto_version = proto_version(initial_version)
 168
 169         self.context = ConnectionContext(protocol_version=latest_allowed_proto)
 170
 171         self.options = _ConnectionOptions()
 172         self.options.address = address
 173         self.options.port = port
 174         self.auth_token = auth_token
 175         self.username = username
 176         self.connected = False
 177
 178         self.handle_exception = handle_exception
 179         self.exception, self.exc_info = None, None
 180         self.handle_exit = handle_exit
 181
 182         # The reactor handles all the default responses to packets,
 183         # it should be changed per networking state
 184         self.reactor = PacketReactor(self)
 185
 186     def _start_network_thread(self):
 187         with self._write_lock:
 188             if self.networking_thread is not None and \
 189                not self.networking_thread.interrupt or \
 190                self.new_networking_thread is not None:
 191                 raise InvalidState('A networking thread is already running.')
 192             elif self.networking_thread is None:
 193                 self.networking_thread = NetworkingThread(self)
 194                 self.networking_thread.start()
 195             else:
 196                 # This thread will wait until the existing thread exits, and
 197                 # then set 'networking_thread' to itself and
 198                 # 'new_networking_thread' to None.
 199                 self.new_networking_thread \
 200                     = NetworkingThread(self, previous=self.networking_thread)
 201                 self.new_networking_thread.start()
 202
 203     def write_packet(self, packet, force=False):
 204         """Writes a packet to the server.
 205
 206         If force is set to true, the method attempts to acquire the write lock
 207         and write the packet out immediately, and as such may block.
 208
 209         If force is false then the packet will be added to the end of the
 210         packet writing queue to be sent 'as soon as possible'
 211
 212         :param packet: The :class:`network.packets.Packet` to write
 213         :param force(bool): Specifies if the packet write should be immediate
 214         """
 215         packet.context = self.context
 216         if force:
 217             with self._write_lock:
 218                 self._write_packet(packet)
 219         else:
 220             self._outgoing_packet_queue.append(packet)
 221
 222     def listener(self, *packet_types, **kwds):
 223         """
 224         Shorthand decorator to register a function as a packet listener.
 225
 226         Wraps :meth:`minecraft.networking.connection.register_packet_listener`
 227         :param packet_types: Packet types to listen for.
 228         :param kwds: Keyword arguments for `register_packet_listener`
 229         """
 230         def listener_decorator(handler_func):
 231             self.register_packet_listener(handler_func, *packet_types, **kwds)
 232             return handler_func
 233
 234         return listener_decorator
 235
 236     def exception_handler(self, *exc_types, **kwds):
 237         """
 238         Shorthand decorator to register a function as an exception handler.
 239         """
 240         def exception_handler_decorator(handler_func):
 241             self.register_exception_handler(handler_func, *exc_types, **kwds)
 242             return handler_func
 243
 244         return exception_handler_decorator
 245
 246     def register_packet_listener(self, method, *packet_types, **kwds):
 247         """
 248         Registers a listener method which will be notified when a packet of
 249         a selected type is received.
 250
 251         If :class:`minecraft.networking.connection.IgnorePacket` is raised from
 252         within this method, no subsequent handlers will be called. If
 253         'early=True', this has the additional effect of preventing the default
 254         in-built action; this could break the internal state of the
 255         'Connection', so should be done with care. If, in addition,
 256         'outgoing=True', this will prevent the packet from being written to the
 257         network.
 258
 259         :param method: The method which will be called back with the packet
 260         :param packet_types: The packets to listen for
 261         :param outgoing: If 'True', this listener will be called on outgoing
 262                          packets just after they are sent to the server, rather
 263                          than on incoming packets.
 264         :param early: If 'True', this listener will be called before any
 265                       built-in default action is carried out, and before any
 266                       listeners with 'early=False' are called. If
 267                       'outgoing=True', the listener will be called before the
 268                       packet is written to the network, rather than afterwards.
 269         """
 270         outgoing = kwds.pop('outgoing', False)
 271         early = kwds.pop('early', False)
 272         target = self.packet_listeners if not early and not outgoing \
 273             else self.early_packet_listeners if early and not outgoing \
 274             else self.outgoing_packet_listeners if not early \
 275             else self.early_outgoing_packet_listeners
 276         target.append(packets.PacketListener(method, *packet_types, **kwds))
 277
 278     def register_exception_handler(self, handler_func, *exc_types, **kwds):
 279         """
 280         Register a function to be called when an unhandled exception occurs
 281         in the networking thread.
 282
 283         When multiple exception handlers are registered, they act like 'except'
 284         clauses in a Python 'try' clause, with the earliest matching handler
 285         catching the exception, and any later handlers catching any uncaught
 286         exception raised from within an earlier handler.
 287
 288         Regardless of the presence or absence of matching handlers, any such
 289         exception will cause the connection and the networking thread to
 290         terminate, the final exception handler will be called (see the
 291         'handle_exception' argument of the 'Connection' contructor), and the
 292         original exception - or the last exception raised by a handler - will
 293         be set as the 'exception' and 'exc_info' attributes of the
 294         'Connection'.
 295
 296         :param handler_func: A function taking two arguments: the exception
 297         object 'e' as in 'except Exception as e:', and the corresponding
 298         3-tuple given by 'sys.exc_info()'. The return value of the function is
 299         ignored, but any exception raised in it replaces the original
 300         exception, and may be passed to later exception handlers.
 301
 302         :param exc_types: The types of exceptions that this handler shall
 303         catch, as in 'except (exc_type_1, exc_type_2, ...) as e:'. If this is
 304         empty, the handler will catch all exceptions.
 305
 306         :param early: If 'True', the exception handler is registered before
 307         any existing exception handlers in the handling order.
 308         """
 309         early = kwds.pop('early', False)
 310         assert not kwds, 'Unexpected keyword arguments: %r' % (kwds,)
 311         if early:
 312             self._exception_handlers.insert(0, (handler_func, exc_types))
 313         else:
 314             self._exception_handlers.append((handler_func, exc_types))
 315
 316     def _pop_packet(self):
 317         # Pops the topmost packet off the outgoing queue and writes it out
 318         # through the socket
 319         #
 320         # Mostly an internal convenience function, caller should make sure
 321         # they have the write lock acquired to avoid issues caused by
 322         # asynchronous access to the socket.
 323         # This should be the only method that removes elements from the
 324         # outbound queue
 325         if len(self._outgoing_packet_queue) == 0:
 326             return False
 327         else:
 328             self._write_packet(self._outgoing_packet_queue.popleft())
 329             return True
 330
 331     def _write_packet(self, packet):
 332         # Immediately writes the given packet to the network. The caller must
 333         # have the write lock acquired before calling this method.
 334         try:
 335             for listener in self.early_outgoing_packet_listeners:
 336                 listener.call_packet(packet)
 337
 338             if self.options.compression_enabled:
 339                 packet.write(self.socket, self.options.compression_threshold)
 340             else:
 341                 packet.write(self.socket)
 342
 343             for listener in self.outgoing_packet_listeners:
 344                 listener.call_packet(packet)
 345         except IgnorePacket:
 346             pass
 347
 348     def status(self, handle_status=None, handle_ping=False):
 349         """Issue a status request to the server and then disconnect.
 350
 351         :param handle_status: a function to be called with the status
 352                               dictionary None for the default behaviour of
 353                               printing the dictionary to standard output, or
 354                               False to ignore the result.
 355         :param handle_ping: a function to be called with the measured latency
 356                             in milliseconds, None for the default handler,
 357                             which prints the latency to standard outout, or
 358                             False, to prevent measurement of the latency.
 359         """
 360         with self._write_lock:  # pylint: disable=not-context-manager
 361             self._check_connection()
 362
 363             self._connect()
 364             self._handshake(next_state=STATE_STATUS)
 365             self._start_network_thread()
 366
 367             do_ping = handle_ping is not False
 368             self.reactor = StatusReactor(self, do_ping=do_ping)
 369
 370             if handle_status is False:
 371                 self.reactor.handle_status = lambda *args, **kwds: None
 372             elif handle_status is not None:
 373                 self.reactor.handle_status = handle_status
 374
 375             if handle_ping is False:
 376                 self.reactor.handle_ping = lambda *args, **kwds: None
 377             elif handle_ping is not None:
 378                 self.reactor.handle_ping = handle_ping
 379
 380             request_packet = serverbound.status.RequestPacket()
 381             self.write_packet(request_packet)
 382
 383     def connect(self):
 384         """
 385         Attempt to begin connecting to the server.
 386         May safely be called multiple times after the first, i.e. to reconnect.
 387         """
 388         # Hold the lock throughout, in case connect() is called from the
 389         # networking thread while another connection is in progress.
 390         with self._write_lock:  # pylint: disable=not-context-manager
 391             self._check_connection()
 392
 393             # It is important that this is set correctly even when connecting
 394             # in status mode, as some servers, e.g. SpigotMC with the
 395             # ProtocolSupport plugin, use it to determine the correct response.
 396             self.context.protocol_version \
 397                 = max(self.allowed_proto_versions,
 398                       key=PROTOCOL_VERSION_INDICES.get)
 399
 400             self.spawned = False
 401             self._connect()
 402             if len(self.allowed_proto_versions) == 1:
 403                 # There is exactly one allowed protocol version, so skip the
 404                 # process of determining the server's version, and immediately
 405                 # connect.
 406                 self._handshake(next_state=STATE_PLAYING)
 407                 login_start_packet = serverbound.login.LoginStartPacket()
 408                 if self.auth_token:
 409                     login_start_packet.name = self.auth_token.profile.name
 410                 else:
 411                     login_start_packet.name = self.username
 412                 self.write_packet(login_start_packet)
 413                 self.reactor = LoginReactor(self)
 414             else:
 415                 # Determine the server's protocol version by first performing a
 416                 # status query.
 417                 self._handshake(next_state=STATE_STATUS)
 418                 self.write_packet(serverbound.status.RequestPacket())
 419                 self.reactor = PlayingStatusReactor(self)
 420             self._start_network_thread()
 421
 422     def _check_connection(self):
 423         if self.networking_thread is not None and \
 424            not self.networking_thread.interrupt or \
 425            self.new_networking_thread is not None:
 426             raise InvalidState('There is an existing connection.')
 427
 428     def _connect(self):
 429         # Connect a socket to the server and create a file object from the
 430         # socket.
 431         # The file object is used to read any and all data from the socket
 432         # since it's "guaranteed" to read the number of bytes specified,
 433         # the socket itself will mostly be used to write data upstream to
 434         # the server.
 435         self._outgoing_packet_queue = deque()
 436
 437         info = socket.getaddrinfo(self.options.address, self.options.port,
 438                                   0, socket.SOCK_STREAM)
 439
 440         # Prefer to use IPv4 (for backward compatibility with previous
 441         # versions that always resolved hostnames to IPv4 addresses),
 442         # then IPv6, then other address families.
 443         def key(ai):
 444             return 0 if ai[0] == socket.AF_INET else \
 445                    1 if ai[0] == socket.AF_INET6 else 2
 446         ai_faml, ai_type, ai_prot, _ai_cnam, ai_addr = min(info, key=key)
 447
 448         self.socket = socket.socket(ai_faml, ai_type, ai_prot)
 449         self.socket.connect(ai_addr)
 450         self.file_object = self.socket.makefile("rb", 0)
 451         self.options.compression_enabled = False
 452         self.options.compression_threshold = -1
 453         self.connected = True
 454
 455     def disconnect(self, immediate=False):
 456         """Terminate the existing server connection, if there is one.
 457            If 'immediate' is True, do not attempt to write any packets.
 458         """
 459         with self._write_lock:  # pylint: disable=not-context-manager
 460             self.connected = False
 461
 462             if not immediate and self.socket is not None:
 463                 # Flush any packets remaining in the queue.
 464                 while self._pop_packet():
 465                     pass
 466
 467             if self.new_networking_thread is not None:
 468                 self.new_networking_thread.interrupt = True
 469             elif self.networking_thread is not None:
 470                 self.networking_thread.interrupt = True
 471
 472             if self.socket is not None:
 473                 try:
 474                     self.socket.shutdown(socket.SHUT_RDWR)
 475                 except socket.error:
 476                     pass
 477                 finally:
 478                     self.file_object.close()
 479                     self.socket.close()
 480                     self.socket = None
 481
 482     def _handshake(self, next_state=STATE_PLAYING):
 483         handshake = serverbound.handshake.HandShakePacket()
 484         handshake.protocol_version = self.context.protocol_version
 485         handshake.server_address = self.options.address
 486         handshake.server_port = self.options.port
 487         handshake.next_state = next_state
 488
 489         self.write_packet(handshake)
 490
 491     def _handle_exception(self, exc, exc_info):
 492         final_handler = self.handle_exception
 493
 494         # Call the current PacketReactor's exception handler.
 495         try:
 496             if self.reactor.handle_exception(exc, exc_info):
 497                 return
 498         except Exception as new_exc:
 499             exc, exc_info = new_exc, sys.exc_info()
 500
 501         # Call the user-registered exception handlers in order.
 502         for handler, exc_types in self._exception_handlers:
 503             if not exc_types or isinstance(exc, exc_types):
 504                 try:
 505                     handler(exc, exc_info)
 506                     caught = True
 507                     break
 508                 except Exception as new_exc:
 509                     exc, exc_info = new_exc, sys.exc_info()
 510         else:
 511             caught = False
 512
 513         # Call the user-specified final exception handler.
 514         if final_handler not in (None, False):
 515             try:
 516                 final_handler(exc, exc_info)
 517             except Exception as new_exc:
 518                 exc, exc_info = new_exc, sys.exc_info()
 519
 520         # For backward compatibility, try to set the 'exc_info' attribute.
 521         try:
 522             exc.exc_info = exc_info
 523         except (TypeError, AttributeError):
 524             pass
 525
 526         # Record the exception.
 527         self.exception, self.exc_info = exc, exc_info
 528
 529         # The following condition being false indicates that an exception
 530         # handler has initiated a new connection, meaning that we should not
 531         # interfere with the connection state. Otherwise, make sure that any
 532         # current connection is completely terminated.
 533         if (self.new_networking_thread or self.networking_thread).interrupt:
 534             self.disconnect(immediate=True)
 535
 536         # If allowed by the final exception handler, re-raise the exception.
 537         if final_handler is None and not caught:
 538             exc_value, exc_tb = exc_info[1:]
 539             raise exc_value.with_traceback(exc_tb)
 540
 541     def _version_mismatch(self, server_protocol=None, server_version=None):
 542         if server_protocol is None:
 543             server_protocol = KNOWN_MINECRAFT_VERSIONS.get(server_version)
 544
 545         if server_protocol is None:
 546             vs = 'version' if server_version is None else \
 547                  ('version of %s' % server_version)
 548         else:
 549             vs = ('protocol version of %d' % server_protocol) + \
 550                  ('' if server_version is None else ' (%s)' % server_version)
 551         ss = 'supported, but not allowed for this connection' \
 552              if server_protocol in SUPPORTED_PROTOCOL_VERSIONS \
 553              else 'not supported'
 554         err = VersionMismatch("Server's %s is %s." % (vs, ss))
 555         err.server_protocol = server_protocol
 556         err.server_version = server_version
 557         raise err
 558
 559     def _handle_exit(self):
 560         if not self.connected and self.handle_exit is not None:
 561             self.handle_exit()
 562
 563     def _react(self, packet):
 564         try:
 565             for listener in self.early_packet_listeners:
 566                 listener.call_packet(packet)
 567             self.reactor.react(packet)
 568             for listener in self.packet_listeners:
 569                 listener.call_packet(packet)
 570         except IgnorePacket:
 571             pass
 572
 573
 574 class NetworkingThread(threading.Thread):
 575     def __init__(self, connection, previous=None):
 576         threading.Thread.__init__(self)
 577         self.interrupt = False
 578         self.connection = connection
 579         self.name = "Networking Thread"
 580         self.daemon = True
 581
 582         self.previous_thread = previous
 583
 584     def run(self):
 585         try:
 586             if self.previous_thread is not None:
 587                 if self.previous_thread.is_alive():
 588                     self.previous_thread.join()
 589                 with self.connection._write_lock:
 590                     self.connection.networking_thread = self
 591                     self.connection.new_networking_thread = None
 592             self._run()
 593             self.connection._handle_exit()
 594         except Exception as e:
 595             self.interrupt = True
 596             self.connection._handle_exception(e, sys.exc_info())
 597         finally:
 598             with self.connection._write_lock:
 599                 self.connection.networking_thread = None
 600
 601     def _run(self):
 602         while not self.interrupt:
 603             # Attempt to write out as many as 300 packets.
 604             num_packets = 0
 605             with self.connection._write_lock:
 606                 try:
 607                     while not self.interrupt and self.connection._pop_packet():
 608                         num_packets += 1
 609                         if num_packets >= 300:
 610                             break
 611                     exc_info = None
 612                 except IOError:
 613                     exc_info = sys.exc_info()
 614
 615                 # If any packets remain to be written, resume writing as soon
 616                 # as possible after reading any available packets; otherwise,
 617                 # wait for up to 50ms (1 tick) for new packets to arrive.
 618                 if self.connection._outgoing_packet_queue:
 619                     read_timeout = 0
 620                 else:
 621                     read_timeout = 0.05
 622
 623             # Read and react to as many as 50 packets.
 624             while num_packets < 50 and not self.interrupt:
 625                 packet = self.connection.reactor.read_packet(
 626                     self.connection.file_object, timeout=read_timeout)
 627                 if not packet:
 628                     break
 629                 num_packets += 1
 630                 self.connection._react(packet)
 631                 read_timeout = 0
 632
 633                 # Ignore the earlier exception if a disconnect packet is
 634                 # received, as it may have been caused by trying to write to
 635                 # the closed socket, which does not represent a program error.
 636                 if exc_info is not None and packet.packet_name == "disconnect":
 637                     exc_info = None
 638
 639             if exc_info is not None:
 640                 exc_value, exc_tb = exc_info[1:]
 641                 raise exc_value.with_traceback(exc_tb)
 642
 643
 644 class PacketReactor(object):
 645     """
 646     Reads and reacts to packets
 647     """
 648     state_name = None
 649
 650     # Handshaking is considered the "default" state
 651     get_clientbound_packets = staticmethod(clientbound.handshake.get_packets)
 652
 653     def __init__(self, connection):
 654         self.connection = connection
 655         context = self.connection.context
 656         self.clientbound_packets = {
 657             packet.get_id(context): packet
 658             for packet in self.__class__.get_clientbound_packets(context)}
 659
 660     def read_packet(self, stream, timeout=0):
 661         # Block for up to `timeout' seconds waiting for `stream' to become
 662         # readable, returning `None' if the timeout elapses.
 663         ready_to_read = select.select([stream], [], [], timeout)[0]
 664
 665         if ready_to_read:
 666             length = VarInt.read(stream)
 667
 668             packet_data = packets.PacketBuffer()
 669             packet_data.send(stream.read(length))
 670             # Ensure we read all the packet
 671             while len(packet_data.get_writable()) < length:
 672                 packet_data.send(
 673                     stream.read(length - len(packet_data.get_writable())))
 674             packet_data.reset_cursor()
 675
 676             if self.connection.options.compression_enabled:
 677                 decompressed_size = VarInt.read(packet_data)
 678                 if decompressed_size > 0:
 679                     decompressor = zlib.decompressobj()
 680                     decompressed_packet = decompressor.decompress(
 681                                                        packet_data.read())
 682                     assert len(decompressed_packet) == decompressed_size, \
 683                         'decompressed length %d, but expected %d' % \
 684                         (len(decompressed_packet), decompressed_size)
 685                     packet_data.reset()
 686                     packet_data.send(decompressed_packet)
 687                     packet_data.reset_cursor()
 688
 689             packet_id = VarInt.read(packet_data)
 690
 691             # If we know the structure of the packet, attempt to parse it
 692             # otherwise, just return an instance of the base Packet class.
 693             if packet_id in self.clientbound_packets:
 694                 packet = self.clientbound_packets[packet_id]()
 695                 packet.context = self.connection.context
 696                 packet.read(packet_data)
 697             else:
 698                 packet = packets.Packet()
 699                 packet.context = self.connection.context
 700                 packet.id = packet_id
 701             return packet
 702         else:
 703             return None
 704
 705     def react(self, packet):
 706         """Called with each incoming packet after early packet listeners are
 707            run (if none of them raise 'IgnorePacket'), but before regular
 708            packet listeners are run. If this method raises 'IgnorePacket', no
 709            subsequent packet listeners will be called for this packet.
 710         """
 711         raise NotImplementedError("Call to base reactor")
 712
 713     def handle_exception(self, exc, exc_info):
 714         """Called when an exception is raised in the networking thread. If this
 715            method returns True, the default action will be prevented and the
 716            exception ignored (but the networking thread will still terminate).
 717         """
 718         return False
 719
 720
 721 class LoginReactor(PacketReactor):
 722     get_clientbound_packets = staticmethod(clientbound.login.get_packets)
 723
 724     def react(self, packet):
 725         if packet.packet_name == "encryption request":
 726
 727             secret = encryption.generate_shared_secret()
 728             token, encrypted_secret = encryption.encrypt_token_and_secret(
 729                 packet.public_key, packet.verify_token, secret)
 730
 731             # A server id of '-' means the server is in offline mode
 732             if packet.server_id != '-':
 733                 server_id = encryption.generate_verification_hash(
 734                     packet.server_id, secret, packet.public_key)
 735                 if self.connection.auth_token is not None:
 736                     self.connection.auth_token.join(server_id)
 737
 738             encryption_response = serverbound.login.EncryptionResponsePacket()
 739             encryption_response.shared_secret = encrypted_secret
 740             encryption_response.verify_token = token
 741
 742             # Forced because we'll have encrypted the connection by the time
 743             # it reaches the outgoing queue
 744             self.connection.write_packet(encryption_response, force=True)
 745
 746             # Enable the encryption
 747             cipher = encryption.create_AES_cipher(secret)
 748             encryptor = cipher.encryptor()
 749             decryptor = cipher.decryptor()
 750             self.connection.socket = encryption.EncryptedSocketWrapper(
 751                 self.connection.socket, encryptor, decryptor)
 752             self.connection.file_object = \
 753                 encryption.EncryptedFileObjectWrapper(
 754                     self.connection.file_object, decryptor)
 755
 756         elif packet.packet_name == "disconnect":
 757             # Receiving a disconnect packet in the login state indicates an
 758             # abnormal condition. Raise an exception explaining the situation.
 759             try:
 760                 msg = json.loads(packet.json_data)['text']
 761             except (ValueError, TypeError, KeyError):
 762                 msg = packet.json_data
 763             match = re.match(r"Outdated (client! Please use|server!"
 764                              r" I'm still on) (?P<ver>\S+)$", msg)
 765             if match:
 766                 ver = match.group('ver')
 767                 self.connection._version_mismatch(server_version=ver)
 768             raise LoginDisconnect('The server rejected our login attempt '
 769                                   'with: "%s".' % msg)
 770
 771         elif packet.packet_name == "login success":
 772             self.connection.reactor = PlayingReactor(self.connection)
 773
 774         elif packet.packet_name == "set compression":
 775             self.connection.options.compression_threshold = packet.threshold
 776             self.connection.options.compression_enabled = True
 777
 778         elif packet.packet_name == "login plugin request":
 779             self.connection.write_packet(
 780                 serverbound.login.PluginResponsePacket(
 781                     message_id=packet.message_id, successful=False))
 782
 783
 784 class PlayingReactor(PacketReactor):
 785     get_clientbound_packets = staticmethod(clientbound.play.get_packets)
 786
 787     def react(self, packet):
 788         if packet.packet_name == "set compression":
 789             self.connection.options.compression_threshold = packet.threshold
 790             self.connection.options.compression_enabled = True
 791
 792         elif packet.packet_name == "keep alive":
 793             keep_alive_packet = serverbound.play.KeepAlivePacket()
 794             keep_alive_packet.keep_alive_id = packet.keep_alive_id
 795             self.connection.write_packet(keep_alive_packet)
 796
 797         elif packet.packet_name == "player position and look":
 798             if self.connection.context.protocol_later_eq(107):
 799                 teleport_confirm = serverbound.play.TeleportConfirmPacket()
 800                 teleport_confirm.teleport_id = packet.teleport_id
 801                 self.connection.write_packet(teleport_confirm)
 802             else:
 803                 position_response = serverbound.play.PositionAndLookPacket()
 804                 position_response.x = packet.x
 805                 position_response.feet_y = packet.y
 806                 position_response.z = packet.z
 807                 position_response.yaw = packet.yaw
 808                 position_response.pitch = packet.pitch
 809                 position_response.on_ground = True
 810                 self.connection.write_packet(position_response)
 811             self.connection.spawned = True
 812
 813         elif packet.packet_name == "disconnect":
 814             self.connection.disconnect()
 815
 816
 817 class StatusReactor(PacketReactor):
 818     get_clientbound_packets = staticmethod(clientbound.status.get_packets)
 819
 820     def __init__(self, connection, do_ping=False):
 821         super(StatusReactor, self).__init__(connection)
 822         self.do_ping = do_ping
 823
 824     def react(self, packet):
 825         if packet.packet_name == "response":
 826             status_dict = json.loads(packet.json_response)
 827             if self.do_ping:
 828                 ping_packet = serverbound.status.PingPacket()
 829                 # NOTE: it may be better to depend on the `monotonic' package
 830                 # or something similar for more accurate time measurement.
 831                 ping_packet.time = int(1000 * timeit.default_timer())
 832                 self.connection.write_packet(ping_packet)
 833             else:
 834                 self.connection.disconnect()
 835             self.handle_status(status_dict)
 836
 837         elif packet.packet_name == "ping":
 838             if self.do_ping:
 839                 now = int(1000 * timeit.default_timer())
 840                 self.connection.disconnect()
 841                 self.handle_ping(now - packet.time)
 842
 843     def handle_status(self, status_dict):
 844         print(status_dict)
 845
 846     def handle_ping(self, latency_ms):
 847         print('Ping: %d ms' % latency_ms)
 848
 849
 850 class PlayingStatusReactor(StatusReactor):
 851     def __init__(self, connection):
 852         super(PlayingStatusReactor, self).__init__(connection, do_ping=False)
 853
 854     def handle_status(self, status):
 855         if status == {}:
 856             # This can occur when we connect to a Mojang server while it is
 857             # still initialising, so it must not cause the client to connect
 858             # with the default version.
 859             raise IOError('Invalid server status.')
 860         elif 'version' not in status or 'protocol' not in status['version']:
 861             return self.handle_failure()
 862
 863         proto = status['version']['protocol']
 864         if proto not in self.connection.allowed_proto_versions:
 865             self.connection._version_mismatch(
 866                 server_protocol=proto,
 867                 server_version=status['version'].get('name'))
 868
 869         self.handle_proto_version(proto)
 870
 871     def handle_proto_version(self, proto_version):
 872         self.connection.allowed_proto_versions = {proto_version}
 873         self.connection.connect()
 874
 875     def handle_failure(self):
 876         self.handle_proto_version(self.connection.default_proto_version)
 877
 878     def handle_exception(self, exc, exc_info):
 879         if isinstance(exc, EOFError):
 880             # An exception of this type may indicate that the server does not
 881             # properly support status queries, so we treat it as non-fatal.
 882             self.connection.disconnect(immediate=True)
 883             self.handle_failure()
 884             return True