如何通过直接修改 IDENTIFY 数据包来获取 discord bot 的移动状态？
How do I get mobile status for discord bot by directly modifying IDENTIFY packet?
显然，discord 机器人可以具有移动状态，而不是默认情况下获得的桌面（在线）状态。
经过一番挖掘，我发现通过修改 discord.gateway.DiscordWebSocket.identify 中的 IDENTIFY packet 将 $browser 的值修改为 Discord Android 或 Discord 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"Discord Android",
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 库并在我的机器上编辑了文件，它最终看起来像这样：
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 和 $device 都编辑为 Discord 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
代码：
import discord
import os
from discord.ext import commands
import mobile_status
import inspect
discord.gateway.DiscordWebSocket.identify = mobile_status.get_mobile()
print(inspect.getsource(discord.gateway.DiscordWebSocket.identify))
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"))
由于每个修补函数（上述函数和 loc["identify"]）都表现出相同的行为，我不能再使用 inspect.getsource(...) 然后依赖 dis.dis 这会导致更多令人失望的结果
拆解后的数据看起来和猴子补丁的工作版本完全一样，所以直接修改的版本尽管功能内容完全相同，但根本无法工作。 （关于拆解数据）
注意：直接做Discord iOS也不行，把$device改成其他值但保持$browser也不行，所有组合我都试过了，none 他们的工作。
TL;DR：如何在运行时不通过猴子修补来获取 discord bot 的移动状态？
以下工作通过子classing 相关的 class，并复制具有相关更改的代码。我们还要subclass the Client class，覆盖掉gateway/websocket class的地方。这会导致大量重复代码，但它确实有效，并且既不需要修改 monkey-patching 也不需要编辑库源代码。
但是，它确实会带来许多与编辑库源代码相同的问题 - 主要是随着库的更新，此代码将变得过时（如果您使用的是存档和过时的版本图书馆，你的问题反而更大了）。
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")
就我个人而言，我认为以下方法确实包含一些运行时 monkey-patching（但没有 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")
至于为什么编辑库源代码对你不起作用，我只能假设你编辑了错误的文件副本，正如人们评论的那样。
DiscordWebSocket.identify 是 non-trivial 并且没有支持的方法来覆盖这些字段。
copy-pasting 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 行。通过覆盖，您通常无需额外的努力即可获得未来的更新。
显然，discord 机器人可以具有移动状态，而不是默认情况下获得的桌面（在线）状态。
经过一番挖掘，我发现通过修改 discord.gateway.DiscordWebSocket.identify 中的 IDENTIFY packet 将 $browser 的值修改为 Discord Android 或 Discord 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"Discord Android",
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 库并在我的机器上编辑了文件，它最终看起来像这样：
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 和 $device 都编辑为 Discord 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
代码：
import discord
import os
from discord.ext import commands
import mobile_status
import inspect
discord.gateway.DiscordWebSocket.identify = mobile_status.get_mobile()
print(inspect.getsource(discord.gateway.DiscordWebSocket.identify))
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"))
由于每个修补函数（上述函数和 loc["identify"]）都表现出相同的行为，我不能再使用 inspect.getsource(...) 然后依赖 dis.dis 这会导致更多令人失望的结果
拆解后的数据看起来和猴子补丁的工作版本完全一样，所以直接修改的版本尽管功能内容完全相同，但根本无法工作。 （关于拆解数据）
注意：直接做Discord iOS也不行，把$device改成其他值但保持$browser也不行，所有组合我都试过了，none 他们的工作。
TL;DR：如何在运行时不通过猴子修补来获取 discord bot 的移动状态？
以下工作通过子classing 相关的 class，并复制具有相关更改的代码。我们还要subclass the Client class，覆盖掉gateway/websocket class的地方。这会导致大量重复代码，但它确实有效，并且既不需要修改 monkey-patching 也不需要编辑库源代码。
但是，它确实会带来许多与编辑库源代码相同的问题 - 主要是随着库的更新，此代码将变得过时（如果您使用的是存档和过时的版本图书馆，你的问题反而更大了）。
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")
就我个人而言，我认为以下方法确实包含一些运行时 monkey-patching（但没有 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")
至于为什么编辑库源代码对你不起作用，我只能假设你编辑了错误的文件副本，正如人们评论的那样。
DiscordWebSocket.identify 是 non-trivial 并且没有支持的方法来覆盖这些字段。
copy-pasting 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 行。通过覆盖，您通常无需额外的努力即可获得未来的更新。