def asyncrat_magic()

in asyncrat/asyncrat.py [0:0]

• 33 lines of code
• 6 McCabe index (conditional complexity)

    def asyncrat_magic(self, p: ProcessMemory, addr: int) -> Config | None:
        try:
            strings_offset = cast(UInt32, p.uint32v(addr + 0x40))
            strings_size = cast(UInt32, p.uint32v(addr + 0x44))
            raw: bytes = p.readv(addr + strings_offset, strings_size)
            data = raw.split(b"\x00\x00")
            key = base64.b64decode(self.get_string(data, 7))
            log.debug("extracted key: %s", str(key))

            config = {
                self.family: {
                    "hosts": self.decrypt_config_item_list(key, data, 2),
                    "ports": self.decrypt_config_item_list(key, data, 1),
                    "version": self.decrypt_config_item_printable(key, data, 3),
                    "install_folder": self.get_wide_string(data, 5),
                    "install_file": self.get_wide_string(data, 6),
                    "install": self.decrypt_config_item_printable(key, data, 4),
                    "mutex": self.decrypt_config_item_printable(key, data, 8),
                    "pastebin": self.decrypt(key, base64.b64decode(data[12][1:])).encode("ascii").replace(b"\x0f", b""),
                },
            }
            if config[self.family].get("pastebin", None) and config[self.family]["pastebin"] != "null":
                try:
                    req = requests.get(url=config[self.family]["pastebin"])
                    if req.status_code == 200:
                        data = req.content.split(b"\x3a")
                        config[self.family]["host"] = data[0].decode("ascii", "ignore")
                        config[self.family]["ports"] = [data[1].decode("ascii", "ignore")]
                except requests.exceptions.RequestException as error:
                    log.warning(error)
            return config
        except requests.exceptions.RequestException as error:
            log.warning(error)
            return None