import struct import binascii import logging from gzip import GzipFile from time import time from collections import defaultdict from io import BytesIO import gevent from gevent import event, queue from random import shuffle from steam.steamid import SteamID from steam.enums import EResult, EUniverse from steam.enums.emsg import EMsg from steam.core import crypto from steam.core.connection import TCPConnection from steam.core.msg import Msg, MsgProto from eventemitter import EventEmitter from steam.util import ip_from_int, is_proto, clear_proto_bit class CMClient(EventEmitter): """ CMClient provides a secure message channel to Steam CM servers Can be used as mixing or on it's own. Incoming messages are parsed and emitted as events using their :class:`steam.enums.emsg.EMsg` as event identifier """ EVENT_CONNECTED = 'connected' """Connection establed to CM server """ EVENT_DISCONNECTED = 'disconnected' """Connection closed """ EVENT_RECONNECT = 'reconnect' """Delayed connect :param delay: delay in seconds :type delay: int """ EVENT_CHANNEL_SECURED = 'channel_secured' """After successful completion of encryption handshake """ EVENT_ERROR = 'error' """When login is denied :param eresult: reason :type eresult: :class:`.EResult` """ EVENT_EMSG = 0 """All incoming messages are emitted with their :class:`.EMsg` number. """ PROTOCOL_TCP = 0 #: TCP protocol enum PROTOCOL_UDP = 1 #: UDP protocol enum verbose_debug = False #: print message connects in debug cm_servers = None #: a instance of :class:`.CMServerList` current_server_addr = None #: (ip, port) tuple _seen_logon = False _connecting = False connected = False #: :class:`True` if connected to CM channel_secured = False #: :class:`True` once secure channel handshake is complete channel_key = None #: channel encryption key channel_hmac = None #: HMAC secret steam_id = SteamID() #: :class:`.SteamID` of the current user session_id = None #: session id when logged in _recv_loop = None _heartbeat_loop = None _LOG = None def __init__(self, protocol=PROTOCOL_TCP): self._LOG = logging.getLogger("CMClient") self.cm_servers = CMServerList() if protocol == CMClient.PROTOCOL_TCP: self.connection = TCPConnection() else: raise ValueError("Only TCP is supported") self.on(EMsg.ChannelEncryptRequest, self.__handle_encrypt_request), self.on(EMsg.Multi, self.__handle_multi), self.on(EMsg.ClientLogOnResponse, self._handle_logon), self.on(EMsg.ClientCMList, self._handle_cm_list), def emit(self, event, *args): if event is not None: self._LOG.debug("Emit event: %s" % repr(event)) super(CMClient, self).emit(event, *args) def connect(self, retry=0, delay=0): """Initiate connection to CM. Blocks until connected unless ``retry`` is specified. :param retry: number of retries before returning. Unlimited when set to ``None`` :type retry: :class:`int` :param delay: delay in secnds before connection attempt :type delay: :class:`int` :return: successful connection :rtype: :class:`bool` """ if self.connected: self._LOG.debug("Connect called, but we are connected?") return if self._connecting: self._LOG.debug("Connect called, but we are already connecting.") return self._connecting = True if delay: self._LOG.debug("Delayed connect: %d seconds" % delay) self.emit(self.EVENT_RECONNECT, delay) self.sleep(delay) self._LOG.debug("Connect initiated.") for i, server_addr in enumerate(self.cm_servers): if retry and i > retry: return False start = time() if self.connection.connect(server_addr): break diff = time() - start self._LOG.debug("Failed to connect. Retrying...") if diff < 5: self.sleep(5 - diff) self.current_server_addr = server_addr self.connected = True self.emit(self.EVENT_CONNECTED) self._recv_loop = gevent.spawn(self._recv_messages) self._connecting = False return True def disconnect(self): """Close connection""" if not self.connected: return self.connected = False self.connection.disconnect() if self._heartbeat_loop: self._heartbeat_loop.kill() self._recv_loop.kill() self._reset_attributes() self.emit(self.EVENT_DISCONNECTED) def _reset_attributes(self): for name in ['connected', 'channel_secured', 'channel_key', 'channel_hmac', 'steam_id', 'session_id', '_seen_logon', '_recv_loop', '_heartbeat_loop', ]: self.__dict__.pop(name, None) def send(self, message): """ Send a message :param message: a message instance :type message: :class:`steam.core.msg.Msg`, :class:`steam.core.msg.MsgProto` """ if not isinstance(message, (Msg, MsgProto)): raise ValueError("Expected Msg or MsgProto, got %s" % message) if self.steam_id: message.steamID = self.steam_id if self.session_id: message.sessionID = self.session_id if self.verbose_debug: self._LOG.debug("Outgoing: %s\n%s" % (repr(message), str(message))) else: self._LOG.debug("Outgoing: %s", repr(message)) data = message.serialize() if self.channel_key: if self.channel_hmac: data = crypto.symmetric_encrypt_HMAC(data, self.channel_key, self.channel_hmac) else: data = crypto.symmetric_encrypt(data, self.channel_key) self.connection.put_message(data) def _recv_messages(self): for message in self.connection: if not self.connected: break if self.channel_key: if self.channel_hmac: try: message = crypto.symmetric_decrypt_HMAC(message, self.channel_key, self.channel_hmac) except RuntimeError as e: self._LOG.exception(e) break else: message = crypto.symmetric_decrypt(message, self.channel_key) gevent.spawn(self._parse_message, message) self.idle() if not self._seen_logon and self.channel_secured: if self.wait_event('disconnected', timeout=5) is not None: return gevent.spawn(self.disconnect) def _parse_message(self, message): emsg_id, = struct.unpack_from(" %s)" % ( len(msg.body.message_body), msg.body.size_unzipped, )) with GzipFile(fileobj=BytesIO(msg.body.message_body)) as f: data = f.read() if len(data) != msg.body.size_unzipped: self._LOG.fatal("Unzipped size mismatch") gevent.spawn(self.disconnect) return else: data = msg.body.message_body while len(data) > 0: size, = struct.unpack_from(" total: self._LOG.debug("Added %d new CM addresses." % (len(self.list) - total))