Initial Release of IGNCore version 2.5

core/aochat/bot.py

@@ -0,0 +1,117 @@
+import select
+import socket
+import struct
+
+from core.aochat.client_packets import LoginRequest, LoginSelect
+from core.aochat.crypt import generate_login_key
+from core.aochat.server_packets import ServerPacket, LoginOK, LoginError, LoginCharacterList
+from core.logger import Logger
+
+
+class Bot:
+    def __init__(self):
+        self.socket = None
+        self.char_id = None
+        self.char_name = None
+        self.logger = Logger(__name__)
+
+    def connect(self, host, port):
+        self.logger.info("Connecting to '%s:%d'" % (host, port))
+        self.socket = socket.create_connection((host, port), 10)
+
+    def disconnect(self):
+        if self.socket:
+            self.socket.shutdown(socket.SHUT_RDWR)
+            self.socket.close()
+            self.socket = None
+
+    def login(self, username, password, character):
+        character = character.capitalize()
+
+        # read seed packet
+        self.logger.info("Logging in as '%s'" % character)
+        seed_packet = self.read_packet(10)
+        seed = seed_packet.seed
+
+        # send back challenge
+        key = generate_login_key(seed, username, password)
+        login_request_packet = LoginRequest(0, username, key)
+        self.send_packet(login_request_packet)
+
+        # read character list
+        character_list_packet: LoginCharacterList = self.read_packet()
+        if isinstance(character_list_packet, LoginError):
+            self.logger.error("Error logging in: %s" % character_list_packet.message)
+            return False
+        if character not in character_list_packet.names:
+            self.logger.error("Character '%s' does not exist on this account" % character)
+            return False
+        index = character_list_packet.names.index(character)
+
+        # select character
+        self.char_id = character_list_packet.char_ids[index]
+        self.char_name = character_list_packet.names[index]
+        login_select_packet = LoginSelect(self.char_id)
+        self.send_packet(login_select_packet)
+
+        # wait for OK
+        packet = self.read_packet()
+        if packet.id == LoginOK.id:
+            self.logger.info("Connected!")
+            return packet
+        else:
+            self.logger.error("Error logging in: %s" % packet.message)
+            return False
+
+    def read_packet(self, max_delay_time=1):
+        """
+        Wait for packet from server.
+        """
+
+        read, write, error = select.select([self.socket], [], [], max_delay_time)
+        if not read:
+            return None
+        else:
+            # Read data from server
+            head = self.read_bytes(4)
+            packet_type, packet_length = struct.unpack(">2H", head)
+            data = self.read_bytes(packet_length)
+
+            try:
+                return ServerPacket.get_instance(packet_type, data)
+            except Exception as e:
+                self.logger.error(f"Error parsing packet parameters for packet_type {packet_type} and payload: {data}",
+                                  e)
+                return None
+
+    def send_packet(self, packet):
+        data = packet.to_bytes()
+        data = struct.pack(">2H", packet.id, len(data)) + data
+
+        self.write_bytes(data)
+
+    def read_bytes(self, num_bytes):
+        data = bytes()
+
+        while num_bytes > 0:
+            chunk = self.socket.recv(num_bytes)
+
+            if len(chunk) == 0:
+                raise EOFError
+
+            num_bytes -= len(chunk)
+            data = data + chunk
+
+        return data
+
+    def write_bytes(self, data):
+        num_bytes = len(data)
+
+        while num_bytes > 0:
+            sent = self.socket.send(data)
+
+            if sent == 0:
+                raise EOFError
+
+            data = data[sent:]
+            num_bytes -= sent