如何通过直接修改IDENTIFY数据包来获取discord bot的移动状态？

Question

显然，discord 机器人可以具有移动状态，而不是默认获得的桌面（在线）状态。

经过一番挖掘后，我发现这种状态是通过修改 的值来实现的，IDENTIFY packet或者discord.gateway.DiscordWebSocket.identify理论上应该$browser让我们获得移动状态。Discord AndroidDiscord iOS

修改我在网上找到的执行此操作的代码片段后，我最终得到以下结果：

def get_mobile():
    """
    The Gateway's IDENTIFY packet contains a properties field, containing $os, $browser and $device fields.
    Discord uses that information to know when your phone client and only your phone client has connected to Discord,
    from there they send the extended presence object.
    The exact field that is checked is the $browser field. If it's set to Discord Android on desktop,
    the mobile indicator is is triggered by the desktop client. If it's set to Discord Client on mobile,
    the mobile indicator is not triggered by the mobile client.
    The specific values for the $os, $browser, and $device fields are can change from time to time.
    """
    import ast
    import inspect
    import re
    import discord

    def source(o):
        s = inspect.getsource(o).split("\n")
        indent = len(s[0]) - len(s[0].lstrip())

        return "\n".join(i[indent:] for i in s)

    source_ = source(discord.gateway.DiscordWebSocket.identify)
    patched = re.sub(
        r'([\'"]\$browser[\'"]:\s?[\'"]).+([\'"])',
        r"\1Discord Android\2",
        source_,
    )

    loc = {}
    exec(compile(ast.parse(patched), "<string>", "exec"), discord.gateway.__dict__, loc)
    return loc["identify"]

现在剩下要做的就是覆盖discord.gateway.DiscordWebSocket.identify主文件中的运行时期间，如下所示：

import discord
import os
from discord.ext import commands
import mobile_status

discord.gateway.DiscordWebSocket.identify = mobile_status.get_mobile()
bot = commands.Bot(command_prefix="?")

@bot.event
async def on_ready():
    print(f"Sucessfully logged in as {bot.user}")

bot.run(os.getenv("DISCORD_TOKEN"))

我们确实成功获取了手机状态

但问题是，我想直接修改文件（包含该函数），而不是在运行时对其进行猴子修补。所以我在本地克隆了 dpy lib 并在我的机器上编辑了该文件，它最终看起来像这样：

    async def identify(self):
        """Sends the IDENTIFY packet."""
        payload = {
            'op': self.IDENTIFY,
            'd': {
                'token': self.token,
                'properties': {
                    '$os': sys.platform,
                    '$browser': 'Discord Android',
                    '$device': 'Discord Android',
                    '$referrer': '',
                    '$referring_domain': ''
                },
                'compress': True,
                'large_threshold': 250,
                'v': 3
            }
        }
     # ...

（为了安全起见，$browser对两者进行了编辑）$deviceDiscord Android

但这不起作用，只是给了我常规的桌面在线图标。
因此，我做的下一件事是在对函数进行猴子修补后identify检查该函数，这样我就可以查看源代码，看看之前出了什么问题，但由于运气不好，我得到了这个错误：

Traceback (most recent call last):
  File "c:\Users\Achxy\Desktop\fresh\file.py", line 8, in <module>
    print(inspect.getsource(discord.gateway.DiscordWebSocket.identify))
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 1024, in getsource
    lines, lnum = getsourcelines(object)
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 1006, in getsourcelines
    lines, lnum = findsource(object)
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 835, in findsource
    raise OSError('could not get source code')
OSError: could not get source code

代码 ：

Traceback (most recent call last):
  File "c:\Users\Achxy\Desktop\fresh\file.py", line 8, in <module>
    print(inspect.getsource(discord.gateway.DiscordWebSocket.identify))
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 1024, in getsource
    lines, lnum = getsourcelines(object)
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 1006, in getsourcelines
    lines, lnum = findsource(object)
  File "C:\Users\Achxy\AppData\Local\Programs\Python\Python39\lib\inspect.py", line 835, in findsource
    raise OSError('could not get source code')
OSError: could not get source code

由于每个修补功能（上述一个和）都表现出相同的行为，loc["identify"]我无法再使用inspect.getsource(...)然后依赖dis.dis这导致更令人失望的结果

反汇编的数据看起来与猴子补丁的工作版本完全相同，因此尽管功能内容完全相同，但直接修改的版本根本无法工作。（关于拆解数据）

注意：Discord iOS直接执行也不起作用，将 更改$device为其他值但保留$browser不起作用，我已经尝试了所有组合，但没有一个起作用。

TL;DR：如何在运行时不对其进行猴子修补的情况下获取不和谐机器人的移动状态？

Answer 1

DiscordWebSocket.identify非常重要，并且没有受支持的方法来覆盖这些字段。

复制粘贴 35* 行代码来修改 2 行代码的更可维护的替代方法是子类化，然后覆盖DiscordWebSocket.send_as_json（4 行自定义代码），并修补classmethod DiscordWebSocket.from_client以实例化子类：

import os

from discord.ext import commands
from discord.gateway import DiscordWebSocket


class MyDiscordWebSocket(DiscordWebSocket):

    async def send_as_json(self, data):
        if data.get('op') == self.IDENTIFY:
            if data.get('d', {}).get('properties', {}).get('$browser') is not None:
                data['d']['properties']['$browser'] = 'Discord Android'
                data['d']['properties']['$device'] = 'Discord Android'
        await super().send_as_json(data)


DiscordWebSocket.from_client = MyDiscordWebSocket.from_client
bot = commands.Bot(command_prefix="?")


@bot.event
async def on_ready():
    print(f"Sucessfully logged in as {bot.user}")

bot.run(os.getenv("DISCORD_TOKEN"))

*Pycord 1.7.3 中的 39 行。通过覆盖，您通常无需额外的努力即可获得未来的更新。

Answer 2

以下工作是对相关类进行子类化，并通过相关更改复制代码。我们还必须对该Client类进行子类化，以覆盖使用 gateway/websocket 类的位置。这会导致大量重复的代码，但它确实有效，并且不需要肮脏的猴子修补或编辑库源代码。

但是，它确实存在许多与编辑库源代码相同的问题 - 主要是随着库的更新，此代码将变得过时（如果您使用的是库的存档和过时版本，则您有相反更大的问题）。

import asyncio
import sys

import aiohttp

import discord
from discord.gateway import DiscordWebSocket, _log
from discord.ext.commands import Bot


class MyGateway(DiscordWebSocket):

    async def identify(self):
        payload = {
            'op': self.IDENTIFY,
            'd': {
                'token': self.token,
                'properties': {
                    '$os': sys.platform,
                    '$browser': 'Discord Android',
                    '$device': 'Discord Android',
                    '$referrer': '',
                    '$referring_domain': ''
                },
                'compress': True,
                'large_threshold': 250,
                'v': 3
            }
        }

        if self.shard_id is not None and self.shard_count is not None:
            payload['d']['shard'] = [self.shard_id, self.shard_count]

        state = self._connection
        if state._activity is not None or state._status is not None:
            payload['d']['presence'] = {
                'status': state._status,
                'game': state._activity,
                'since': 0,
                'afk': False
            }

        if state._intents is not None:
            payload['d']['intents'] = state._intents.value

        await self.call_hooks('before_identify', self.shard_id, initial=self._initial_identify)
        await self.send_as_json(payload)
        _log.info('Shard ID %s has sent the IDENTIFY payload.', self.shard_id)


class MyBot(Bot):

    async def connect(self, *, reconnect: bool = True) -> None:
        """|coro|

        Creates a websocket connection and lets the websocket listen
        to messages from Discord. This is a loop that runs the entire
        event system and miscellaneous aspects of the library. Control
        is not resumed until the WebSocket connection is terminated.

        Parameters
        -----------
        reconnect: :class:`bool`
            If we should attempt reconnecting, either due to internet
            failure or a specific failure on Discord's part. Certain
            disconnects that lead to bad state will not be handled (such as
            invalid sharding payloads or bad tokens).

        Raises
        -------
        :exc:`.GatewayNotFound`
            If the gateway to connect to Discord is not found. Usually if this
            is thrown then there is a Discord API outage.
        :exc:`.ConnectionClosed`
            The websocket connection has been terminated.
        """

        backoff = discord.client.ExponentialBackoff()
        ws_params = {
            'initial': True,
            'shard_id': self.shard_id,
        }
        while not self.is_closed():
            try:
                coro = MyGateway.from_client(self, **ws_params)
                self.ws = await asyncio.wait_for(coro, timeout=60.0)
                ws_params['initial'] = False
                while True:
                    await self.ws.poll_event()
            except discord.client.ReconnectWebSocket as e:
                _log.info('Got a request to %s the websocket.', e.op)
                self.dispatch('disconnect')
                ws_params.update(sequence=self.ws.sequence, resume=e.resume, session=self.ws.session_id)
                continue
            except (OSError,
                    discord.HTTPException,
                    discord.GatewayNotFound,
                    discord.ConnectionClosed,
                    aiohttp.ClientError,
                    asyncio.TimeoutError) as exc:

                self.dispatch('disconnect')
                if not reconnect:
                    await self.close()
                    if isinstance(exc, discord.ConnectionClosed) and exc.code == 1000:
                        # clean close, don't re-raise this
                        return
                    raise

                if self.is_closed():
                    return

                # If we get connection reset by peer then try to RESUME
                if isinstance(exc, OSError) and exc.errno in (54, 10054):
                    ws_params.update(sequence=self.ws.sequence, initial=False, resume=True, session=self.ws.session_id)
                    continue

                # We should only get this when an unhandled close code happens,
                # such as a clean disconnect (1000) or a bad state (bad token, no sharding, etc)
                # sometimes, discord sends us 1000 for unknown reasons so we should reconnect
                # regardless and rely on is_closed instead
                if isinstance(exc, discord.ConnectionClosed):
                    if exc.code == 4014:
                        raise discord.PrivilegedIntentsRequired(exc.shard_id) from None
                    if exc.code != 1000:
                        await self.close()
                        raise

                retry = backoff.delay()
                _log.exception("Attempting a reconnect in %.2fs", retry)
                await asyncio.sleep(retry)
                # Always try to RESUME the connection
                # If the connection is not RESUME-able then the gateway will invalidate the session.
                # This is apparently what the official Discord client does.
                ws_params.update(sequence=self.ws.sequence, resume=True, session=self.ws.session_id)


bot = MyBot(command_prefix="?")


@bot.event
async def on_ready():
    print(f"Sucessfully logged in as {bot.user}")

bot.run("YOUR_BOT_TOKEN")

就我个人而言，我认为以下方法（确实包含一些运行时猴子修补（但没有 AST 操作））对于此目的来说更干净：

import sys
from discord.gateway import DiscordWebSocket, _log
from discord.ext.commands import Bot


async def identify(self):
    payload = {
        'op': self.IDENTIFY,
        'd': {
            'token': self.token,
            'properties': {
                '$os': sys.platform,
                '$browser': 'Discord Android',
                '$device': 'Discord Android',
                '$referrer': '',
                '$referring_domain': ''
            },
            'compress': True,
            'large_threshold': 250,
            'v': 3
        }
    }

    if self.shard_id is not None and self.shard_count is not None:
        payload['d']['shard'] = [self.shard_id, self.shard_count]

    state = self._connection
    if state._activity is not None or state._status is not None:
        payload['d']['presence'] = {
            'status': state._status,
            'game': state._activity,
            'since': 0,
            'afk': False
        }

    if state._intents is not None:
        payload['d']['intents'] = state._intents.value

    await self.call_hooks('before_identify', self.shard_id, initial=self._initial_identify)
    await self.send_as_json(payload)
    _log.info('Shard ID %s has sent the IDENTIFY payload.', self.shard_id)


DiscordWebSocket.identify = identify
bot = Bot(command_prefix="?")


@bot.event
async def on_ready():
    print(f"Sucessfully logged in as {bot.user}")

bot.run("YOUR_DISCORD_TOKEN")

至于为什么编辑库源代码对你不起作用，我只能假设你编辑了错误的文件副本，正如人们评论的那样。