Files
lyuksovannyyandClaude Opus 5.5 00553970b7
Publish plugin / publish (push) Successful in 5s
[no bump] Ghost Mode: AyuGram's ghost mode with per-account and per-chat settings
Read receipts, story views, typing and online status are held back per
the options; read on interact, scheduled and silent sending as in AyuGram.
Every option resolves chat -> account -> global; forum topics follow
their group. Settings live in the plugin page, a chat/profile menu popup,
and a drawer row for the current account.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-09 15:54:41 +02:00

1310 lines
46 KiB
Python
Executable File

#!/usr/bin/env python3
"""Offline verification harness for a built .plugin bundle. Runs without a phone.
Executes the bundle against faked SDK modules and faked Telegram classes, then
drives the plugin's hooks the way the app would and checks what reaches the
network.
Usage: python3 tools/verify_bundle.py <bundle.plugin>
"""
from __future__ import annotations
import ast
import json
import sys
import types
from dataclasses import dataclass
from enum import Enum, auto
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parent.parent
PLUGIN_JSON = REPO_ROOT / "plugin.json"
_FAILURES = []
def check(name: str, condition: bool, detail: str = "") -> None:
if condition:
print(f"PASS {name}")
else:
message = f"FAIL {name}"
if detail:
message += f": {detail}"
print(message, file=sys.stderr)
_FAILURES.append(name)
# --- SDK fakes ---------------------------------------------------------------
class HookStrategy(Enum):
DEFAULT = auto()
MODIFY = auto()
CANCEL = auto()
MODIFY_FINAL = auto()
@dataclass
class HookResult:
strategy: HookStrategy = HookStrategy.DEFAULT
request: object = None
response: object = None
update: object = None
updates: object = None
error: object = None
params: object = None
class MenuItemType(Enum):
MESSAGE_CONTEXT_MENU = auto()
DRAWER_MENU = auto()
MAIN_MENU = auto()
CHAT_ACTION_MENU = auto()
PROFILE_ACTION_MENU = auto()
@dataclass
class MenuItemData:
menu_type: MenuItemType
text: str
on_click: object
item_id: str = None
icon: str = None
subtext: str = None
condition: str = None
priority: int = 0
class MethodHook:
def before_hooked_method(self, param):
pass
def after_hooked_method(self, param):
pass
LOG_LINES: list[str] = []
HOOKED: dict[tuple[str, str], object] = {}
ADDED_HOOKS: list[str] = []
MENU_ITEMS: list[MenuItemData] = []
SETTINGS_RELOADS: list[str] = []
class BasePlugin:
def __init__(self):
self._settings = {}
def get_setting(self, key, default=None):
return self._settings.get(key, default)
def set_setting(self, key, value, reload_settings=False):
self._settings[key] = value
if reload_settings:
SETTINGS_RELOADS.append(key)
def add_hook(self, name, match_substring=False, priority=0):
ADDED_HOOKS.append(name)
def remove_hook(self, name):
ADDED_HOOKS.remove(name)
def add_on_send_message_hook(self, priority=0):
pass
def hook_all_methods(self, hook_class, method_name, xposed_hook=None, priority=None, **kwargs):
HOOKED[(hook_class.getName(), method_name)] = xposed_hook
return [object()]
def unhook_method(self, unhook):
pass
def add_menu_item(self, item):
MENU_ITEMS.append(item)
return f"item{len(MENU_ITEMS)}"
def remove_menu_item(self, item_id):
return True
def log(self, message):
LOG_LINES.append(str(message))
class FakeJavaClass:
def __init__(self, name):
self.name = name
def getName(self):
return self.name
class _SettingsRow:
def __init__(self, **kwargs):
self.__dict__.update(kwargs)
BULLETINS: list[str] = []
DIALOGS: list["FakeAlertDialogBuilder"] = []
class FakeAlertDialogBuilder:
def __init__(self, context, *args, **kwargs):
self.view = None
self.dismiss_listener = None
self.title = None
self.dismissed = False
DIALOGS.append(self)
def set_title(self, title):
self.title = title
return self
def set_view(self, view, height=-2):
self.view = view
return self
def set_negative_button(self, text, listener=None):
return self
def set_on_dismiss_listener(self, listener=None):
self.dismiss_listener = listener
return self
def show(self):
return self
def dismiss(self):
self.dismissed = True
if self.dismiss_listener is not None:
self.dismiss_listener(self)
@property
def rows(self):
return self.view.children[0].children
@property
def items(self):
return [str(r.text) for r in self.rows]
def find(self, title):
return next(i for i, text in enumerate(self.items) if text.endswith(title))
def click(self, index):
self.rows[index].listener.onClick(None)
# --- Android view fakes ------------------------------------------------------
class FakeSpannable:
def __init__(self, text):
self.text = text
self.spans = [] # (span, start, end)
def setSpan(self, span, start, end, flags):
self.spans.append((span, start, end))
def __str__(self):
return self.text
def colored(self):
"""[(utf-16 slice text, color)] of the color spans."""
units = self.text.encode("utf-16-le")
return [
(units[2 * a : 2 * b].decode("utf-16-le"), span.color)
for span, a, b in self.spans
if hasattr(span, "color")
]
class FakeView:
def __init__(self, context=None):
self.children = []
self.text = None
self.listener = None
def addView(self, view, *args):
self.children.append(view)
def removeAllViews(self):
self.children = []
def setText(self, text):
self.text = text
def setOnClickListener(self, listener):
self.listener = listener
def __getattr__(self, name): # setPadding, setTextColor, setGravity, ...
if name.startswith("set"):
return lambda *args: None
raise AttributeError(name)
GREEN, RED, GRAY = 0x11, 0x22, 0x33
class FakeTheme:
key_color_green, key_color_red, key_dialogTextGray2 = "green", "red", "gray"
key_dialogTextBlack, key_listSelector = "black", "selector"
COLORS = {"green": GREEN, "red": RED, "gray": GRAY, "black": 0, "selector": 0}
@classmethod
def getColor(cls, key):
return cls.COLORS[key]
@staticmethod
def createSelectorDrawable(color, mask):
return object()
class FakeDrawable:
def __init__(self, size):
self.size = size
def getIntrinsicWidth(self):
return self.size
def getIntrinsicHeight(self):
return self.size
class FakeLayerDrawable(FakeDrawable):
def __init__(self, layers):
self.layers = list(layers)
self.size = None
def setLayerSize(self, index, width, height):
self.size = width
class FakeImageView:
def __init__(self, drawable):
self.drawable = drawable
def getDrawable(self):
return self.drawable
def setImageDrawable(self, drawable):
self.drawable = drawable
# --- Telegram fakes ----------------------------------------------------------
class JavaObject:
def getClass(self):
name = type(self).__name__
return types.SimpleNamespace(getSimpleName=lambda: name)
def _tl(name, **defaults):
def __init__(self, **kwargs):
self.__dict__.update(defaults)
self.__dict__.update(kwargs)
return type(name, (JavaObject,), {"__init__": __init__})
TL_inputPeerUser = _tl("TL_inputPeerUser", user_id=0, chat_id=0, channel_id=0)
TL_inputPeerChat = _tl("TL_inputPeerChat", user_id=0, chat_id=0, channel_id=0)
TL_inputPeerChannel = _tl("TL_inputPeerChannel", user_id=0, chat_id=0, channel_id=0)
TL_inputPeerSelf = _tl("TL_inputPeerSelf", user_id=0, chat_id=0, channel_id=0)
TL_inputChannel = _tl("TL_inputChannel", channel_id=0)
TL_inputEncryptedChat = _tl("TL_inputEncryptedChat", chat_id=0)
TLRPC = types.SimpleNamespace(
TL_messages_readHistory=_tl("TL_messages_readHistory", peer=None, max_id=0),
TL_channels_readHistory=_tl("TL_channels_readHistory", channel=None, max_id=0),
TL_account_updateStatus=_tl("TL_account_updateStatus", offline=False),
TL_messages_setTyping=_tl("TL_messages_setTyping", peer=None),
TL_messages_setEncryptedTyping=_tl("TL_messages_setEncryptedTyping", peer=None),
TL_messages_sendMessage=_tl("TL_messages_sendMessage", peer=None),
TL_messages_forwardMessages=_tl("TL_messages_forwardMessages", to_peer=None),
TL_stories_readStories=_tl("TL_stories_readStories", peer=None),
TL_messages_readMessageContents=_tl("TL_messages_readMessageContents"),
TL_messages_readEncryptedHistory=_tl("TL_messages_readEncryptedHistory", peer=None),
TL_messages_readDiscussion=_tl("TL_messages_readDiscussion", peer=None, msg_id=0, read_max_id=0),
)
ACCOUNTS = {0: 111, 1: 222} # slot -> Telegram user id
BOT_ID = 999
CHANNEL_ID = 777
FORUM_ID = 888
FORUM_TOPICS = {1: 500, 10: 610, 20: 720} # topic id -> newest message id
NOW = 1_700_000_000
NEW_MESSAGES = 42 # NotificationCenter.didReceiveNewMessages
SENT: list[tuple[int, object]] = [] # (account, request) sent through ConnectionsManager
class FakeUserConfig:
MAX_ACCOUNT_COUNT = 4
selectedAccount = 0
def __init__(self, slot):
self.slot = slot
@classmethod
def getInstance(cls, slot):
return cls(slot)
def isClientActivated(self):
return self.slot in ACCOUNTS
def getClientUserId(self):
return ACCOUNTS.get(self.slot, 0)
def getCurrentUser(self):
uid = ACCOUNTS.get(self.slot)
return types.SimpleNamespace(id=uid, first_name=f"User{uid}", username=None) if uid else None
class FakeUserObject:
@staticmethod
def getUserName(user):
return getattr(user, "first_name", "") or "user"
@staticmethod
def isUserSelf(user):
return user.id in ACCOUNTS.values()
class FakeDialogObject:
@staticmethod
def makeEncryptedDialogId(chat_id):
return 0x4000000000000000 | (chat_id & 0xFFFFFFFF)
@staticmethod
def isEncryptedDialog(dialog_id):
return bool(dialog_id & 0x4000000000000000) and dialog_id > 0
@staticmethod
def getEncryptedChatId(dialog_id):
return dialog_id & 0xFFFFFFFF
class FakeSparse(dict):
def get(self, key, default=None):
return super().get(key, default)
class FakeMessagesController:
instances: dict[int, "FakeMessagesController"] = {}
def __init__(self, account):
self.account = account
self.dialogs_dict = FakeSparse()
self.differences = []
@classmethod
def getInstance(cls, account):
return cls.instances.setdefault(account, cls(account))
def getUser(self, user_id):
return types.SimpleNamespace(id=user_id, bot=user_id == BOT_ID, first_name=f"U{user_id}", username=None)
def getChat(self, chat_id):
return types.SimpleNamespace(
id=chat_id,
title=f"Chat {chat_id}",
channel=chat_id in (CHANNEL_ID, FORUM_ID),
forum=chat_id == FORUM_ID,
)
def getTopicsController(self):
topics = {
i: types.SimpleNamespace(id=i, top_message=top) for i, top in FORUM_TOPICS.items()
}
return types.SimpleNamespace(
findTopic=lambda chat_id, topic_id: topics.get(topic_id) if chat_id == FORUM_ID else None,
getTopics=lambda chat_id: FakeArrayList(topics.values()) if chat_id == FORUM_ID else None,
)
def getInputPeer(self, dialog_id):
if dialog_id > 0:
return TL_inputPeerUser(user_id=dialog_id)
return TL_inputPeerChat(chat_id=-dialog_id)
def getInputChannel(self, chat_id):
return TL_inputChannel(channel_id=chat_id)
def processNewDifferenceParams(self, *args):
self.differences.append(args)
class FakeChatObject:
@staticmethod
def isChannel(chat):
return bool(getattr(chat, "channel", False))
class FakeConnectionsManager:
def __init__(self, account):
self.account = account
@classmethod
def getInstance(cls, account):
return cls(account)
def getCurrentTime(self):
return NOW
def sendRequest(self, request, callback):
SENT.append((self.account, request))
if callback is not None:
callback.fn(types.SimpleNamespace(pts=5, pts_count=1), None)
class FakeRequestCallback:
def __init__(self, fn, account=None):
self.fn = fn
class FakeArrayList(list):
def isEmpty(self):
return not self
def size(self):
return len(self)
def get(self, i):
return self[i]
class FakeChatActivity(JavaObject):
def __init__(self, account, dialog):
self.account = account
self.dialog = dialog
self.panel_hidden = False
self.waitingForSendingMessageLoad = True
def getCurrentAccount(self):
return self.account
def getDialogId(self):
return self.dialog
def getParentActivity(self):
return object()
def hideFieldPanel(self, animated):
self.panel_hidden = True
ChatActivity = type("ChatActivity", (FakeChatActivity,), {})
LAST_FRAGMENT = [None]
PLUGINS_ENTRY, ARCHIVE_ENTRY, FEED_ENTRY = 7, 1, 9
MAIN_MENU_LAYOUT = FakeArrayList([ARCHIVE_ENTRY, FEED_ENTRY])
GHOST_DRAWABLE = 0x7F0812AB
class FakeDrawerMenuItemView:
def __init__(self, context):
self.menu_item = None
self.listener = None
def setMenuItem(self, layout_id, account, icon, text):
self.menu_item = (layout_id, account, icon, text)
def setOnClickListener(self, listener):
self.listener = listener
class FakeLinearLayout:
def __init__(self):
self.children = []
def getContext(self):
return object()
def addView(self, view):
self.children.append(view)
class FakeDrawerMenuView:
def __init__(self):
self.container = FakeLinearLayout()
self.closed = 0
self.onItemClick = types.SimpleNamespace(run=self._close)
def _close(self):
self.closed += 1
class FakeResources:
def getIdentifier(self, name, kind, package):
return GHOST_DRAWABLE if (name, kind) == ("ghost", "drawable") else 0
class FakeParam:
def __init__(self, args=(), this=None, result=None):
self.args = list(args)
self.thisObject = this
self._result = result
self.result_set = False
def getResult(self):
return self._result
def setResult(self, value):
self._result = value
self.result_set = True
def _module(name, **attrs):
mod = types.ModuleType(name)
mod.__dict__.update(attrs)
sys.modules[name] = mod
return mod
def install_fakes() -> None:
_module(
"base_plugin",
BasePlugin=BasePlugin,
HookResult=HookResult,
HookStrategy=HookStrategy,
MethodHook=MethodHook,
MenuItemData=MenuItemData,
MenuItemType=MenuItemType,
)
_module(
"android_utils",
log=lambda *a, **k: LOG_LINES.append(" ".join(map(str, a))),
run_on_ui_thread=lambda fn, delay=0: fn(),
OnClickListener=lambda fn: types.SimpleNamespace(onClick=fn),
)
_module(
"hook_utils",
find_class=FakeJavaClass,
get_private_field=lambda obj, name: getattr(obj, name, None),
set_private_field=lambda obj, name, value: setattr(obj, name, value) or True,
)
_module(
"client_utils",
get_last_fragment=lambda: LAST_FRAGMENT[0],
RequestCallback=FakeRequestCallback,
)
ui = _module("ui", __path__=[])
ui.settings = _module(
"ui.settings",
**{n: type(n, (_SettingsRow,), {}) for n in ("Header", "Input", "Text", "Switch", "Selector", "Divider")},
)
ui.bulletin = _module(
"ui.bulletin",
BulletinHelper=types.SimpleNamespace(
show_info=lambda text, fragment=None: BULLETINS.append(text),
show_error=lambda text, fragment=None: BULLETINS.append("ERROR " + text),
),
)
ui.alert = _module("ui.alert", AlertDialogBuilder=FakeAlertDialogBuilder)
android = _module("android", __path__=[])
android.widget = _module("android.widget", LinearLayout=type("LinearLayout", (FakeView,), {"VERTICAL": 1}),
ScrollView=FakeView, TextView=FakeView)
android.text = _module("android.text", __path__=[], SpannableStringBuilder=FakeSpannable,
Spanned=types.SimpleNamespace(SPAN_EXCLUSIVE_EXCLUSIVE=33),
TextUtils=types.SimpleNamespace(TruncateAt=types.SimpleNamespace(END=1)))
android.text.style = _module("android.text.style",
ForegroundColorSpan=lambda color: types.SimpleNamespace(color=color),
StyleSpan=lambda style: types.SimpleNamespace(style=style))
android.graphics = _module("android.graphics", __path__=[], Typeface=types.SimpleNamespace(BOLD=1))
android.graphics.drawable = _module("android.graphics.drawable", LayerDrawable=FakeLayerDrawable)
android.util = _module("android.util", TypedValue=types.SimpleNamespace(COMPLEX_UNIT_DIP=1))
android.view = _module("android.view", Gravity=types.SimpleNamespace(CENTER_VERTICAL=16, START=0x800003))
java = _module("java", __path__=[])
java.util = _module("java.util", ArrayList=FakeArrayList)
org = _module("org", __path__=[])
org.telegram = _module("org.telegram", __path__=[])
org.telegram.messenger = _module(
"org.telegram.messenger",
UserConfig=FakeUserConfig,
UserObject=FakeUserObject,
DialogObject=FakeDialogObject,
MessagesController=FakeMessagesController,
ChatObject=FakeChatObject,
NotificationCenter=types.SimpleNamespace(didReceiveNewMessages=NEW_MESSAGES),
AndroidUtilities=types.SimpleNamespace(dp=lambda v: int(v * 3)),
ApplicationLoader=types.SimpleNamespace(
applicationContext=types.SimpleNamespace(
getResources=lambda: FakeResources(), getPackageName=lambda: "com.exteragram.messenger"
)
),
)
com = _module("com", __path__=[])
com.exteragram = _module("com.exteragram", __path__=[])
com.exteragram.messenger = _module(
"com.exteragram.messenger",
__path__=[],
ExteraConfig=types.SimpleNamespace(getMainMenuLayout=lambda: MAIN_MENU_LAYOUT),
MainMenuItem=types.SimpleNamespace(PLUGINS=types.SimpleNamespace(getId=lambda: PLUGINS_ENTRY)),
)
com.exteragram.messenger.drawer = _module(
"com.exteragram.messenger.drawer", DrawerMenuItemView=FakeDrawerMenuItemView
)
org.telegram.ui = _module("org.telegram.ui", __path__=[])
org.telegram.ui.ActionBar = _module("org.telegram.ui.ActionBar", Theme=FakeTheme)
org.telegram.tgnet = _module(
"org.telegram.tgnet", TLRPC=TLRPC, ConnectionsManager=FakeConnectionsManager
)
KEY_TO_DUNDER = {
"id": "__id__",
"name": "__name__",
"description": "__description__",
"author": "__author__",
"version": "__version__",
"icon": "__icon__",
"app_version": "__app_version__",
"sdk_version": "__sdk_version__",
"requirements": "__requirements__",
}
def expected_dunders(meta: dict) -> dict:
result = {}
for key, dunder in KEY_TO_DUNDER.items():
if key not in meta:
continue
value = meta[key]
if key in ("description", "icon") and value == "":
continue
if key == "requirements" and value == []:
continue
result[dunder] = value
return result
# --- behaviour ---------------------------------------------------------------
def reset(plugin):
plugin._settings.clear()
SENT.clear()
BULLETINS.clear()
DIALOGS.clear()
LAST_FRAGMENT[0] = None
def pre(plugin, request, account=0):
return plugin.pre_request_hook(type(request).__name__, account, request).strategy
def read(dialog):
if dialog > 0:
return TLRPC.TL_messages_readHistory(peer=TL_inputPeerUser(user_id=dialog), max_id=10)
return TLRPC.TL_messages_readHistory(peer=TL_inputPeerChat(chat_id=-dialog), max_id=10)
def verify_requests(plugin):
D, C = HookStrategy.DEFAULT, HookStrategy.CANCEL
reset(plugin)
check("reads pass with ghost mode off", pre(plugin, read(555)) == D)
plugin.set_setting("g_no_read", True)
check(
"global Don't read messages drops reads on every account",
pre(plugin, read(555), 0) == C and pre(plugin, read(555), 1) == C,
)
plugin.set_setting("a_222_no_read", 2) # account 1 (uid 222): Off
check(
"account override beats the global switch",
pre(plugin, read(555), 0) == C and pre(plugin, read(555), 1) == D,
)
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).NO_READ, 2)
check(
"chat override beats the account",
pre(plugin, read(555), 0) == D and pre(plugin, read(556), 0) == C,
)
plugin.prefs.set_chat_choice(222, -42, plugin_features(plugin).NO_READ, 1)
check(
"chat override can turn ghost on in an account that has it off",
pre(plugin, read(-42), 1) == C and pre(plugin, read(555), 1) == D,
)
reset(plugin)
plugin.prefs.set_chat_choice(111, -CHANNEL_ID, plugin_features(plugin).NO_READ, 1)
channel_read = types.new_class("TL_channels_readHistory", (JavaObject,))()
channel_read.channel = TL_inputChannel(channel_id=CHANNEL_ID)
check("channel reads resolve to -channel_id", pre(plugin, channel_read) == C)
self_read = TLRPC.TL_messages_readHistory(peer=TL_inputPeerSelf())
plugin.prefs.set_chat_choice(111, 111, plugin_features(plugin).NO_READ, 1)
check("inputPeerSelf resolves to the account's own id", pre(plugin, self_read) == C)
reset(plugin)
plugin.set_setting("g_no_read", True)
LAST_FRAGMENT[0] = ChatActivity(0, 555)
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).NO_READ, 2)
check(
"readMessageContents follows the chat on screen",
pre(plugin, TLRPC.TL_messages_readMessageContents()) == D,
)
LAST_FRAGMENT[0] = None
check(
"readMessageContents without a chat on screen follows the account",
pre(plugin, TLRPC.TL_messages_readMessageContents()) == C,
)
reset(plugin)
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).NO_TYPING, 1)
typing = lambda d: TLRPC.TL_messages_setTyping(peer=TL_inputPeerUser(user_id=d))
check("per-chat Don't send typing", pre(plugin, typing(555)) == C and pre(plugin, typing(556)) == D)
secret = FakeDialogObject.makeEncryptedDialogId(7)
plugin.prefs.set_chat_choice(111, secret, plugin_features(plugin).NO_TYPING, 1)
encrypted = TLRPC.TL_messages_setEncryptedTyping(peer=TL_inputEncryptedChat(chat_id=7))
check("secret chat typing resolves to its encrypted dialog id", pre(plugin, encrypted) == C)
stories = lambda d: TLRPC.TL_stories_readStories(peer=TL_inputPeerUser(user_id=d))
plugin.set_setting("g_no_read_stories", True)
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).NO_READ_STORIES, 2)
check("per-chat Don't read stories", pre(plugin, stories(555)) == D and pre(plugin, stories(556)) == C)
# online status
reset(plugin)
plugin.set_setting("g_no_online", True)
status = TLRPC.TL_account_updateStatus(offline=False)
result = plugin.pre_request_hook("TL_account_updateStatus", 0, status)
check(
"Don't send online sends offline instead",
result.strategy == HookStrategy.MODIFY and status.offline is True,
f"{result.strategy} offline={status.offline}",
)
plugin.set_setting("a_222_no_online", 2)
status = TLRPC.TL_account_updateStatus(offline=False)
plugin.pre_request_hook("TL_account_updateStatus", 1, status)
check("Don't send online is per account", status.offline is False)
reset(plugin)
plugin.set_setting("g_auto_offline", True)
status = TLRPC.TL_account_updateStatus(offline=False)
plugin.pre_request_hook("TL_account_updateStatus", 1, status)
plugin.post_request_hook("TL_account_updateStatus", 1, object(), None)
check(
"Go offline automatically follows online with offline on the same account",
status.offline is False
and len(SENT) == 1
and SENT[0][0] == 1
and type(SENT[0][1]).__name__ == "TL_account_updateStatus"
and SENT[0][1].offline is True,
f"{SENT}",
)
plugin.pre_request_hook("TL_account_updateStatus", 1, SENT[0][1])
plugin.post_request_hook("TL_account_updateStatus", 1, object(), None)
check("the offline status doesn't trigger another", len(SENT) == 1)
# read on interact
reset(plugin)
plugin.set_setting("g_no_read", True)
FakeMessagesController.getInstance(0).dialogs_dict[555] = types.SimpleNamespace(top_message=4321)
send = TLRPC.TL_messages_sendMessage(peer=TL_inputPeerUser(user_id=555))
pre(plugin, send)
plugin.post_request_hook("TL_messages_sendMessage", 0, object(), None)
readers = [r for a, r in SENT if type(r).__name__ == "TL_messages_readHistory"]
check(
"Read on interact reads the chat up to its newest message after sending",
len(readers) == 1 and readers[0].max_id == 4321 and readers[0].peer.user_id == 555,
f"{SENT}",
)
check("the plugin's own read receipt is let through once", pre(plugin, readers[0]) == D)
check("…and only once", pre(plugin, readers[0]) == C)
SENT.clear()
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).READ_ON_INTERACT, 2)
pre(plugin, send)
plugin.post_request_hook("TL_messages_sendMessage", 0, object(), None)
check("Read on interact can be turned off per chat", not SENT, f"{SENT}")
SENT.clear()
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).READ_ON_INTERACT, 0)
pre(plugin, send)
plugin.post_request_hook("TL_messages_sendMessage", 0, None, object())
check("a failed send doesn't read the chat", not SENT)
reset(plugin)
FakeMessagesController.getInstance(0).dialogs_dict[-CHANNEL_ID] = types.SimpleNamespace(top_message=99)
plugin.set_setting("g_no_read", True)
forward = TLRPC.TL_messages_forwardMessages(to_peer=TL_inputPeerChannel(channel_id=CHANNEL_ID))
pre(plugin, forward)
plugin.post_request_hook("TL_messages_forwardMessages", 0, object(), None)
check(
"Read on interact on a channel uses channels.readHistory",
len(SENT) == 1
and type(SENT[0][1]).__name__ == "TL_channels_readHistory"
and SENT[0][1].channel.channel_id == CHANNEL_ID
and SENT[0][1].max_id == 99,
f"{SENT}",
)
def plugin_features(plugin):
return type(plugin.prefs).resolve.__globals__["features"]
def params(**kwargs):
base = dict(peer=555, scheduleDate=0, notify=True, photo=None, document=None)
base.update(kwargs)
return types.SimpleNamespace(**base)
def verify_sending(plugin):
reset(plugin)
p = params()
result = plugin.on_send_message_hook(0, p)
check("sending untouched with options off", result.strategy == HookStrategy.DEFAULT and p.scheduleDate == 0)
plugin.set_setting("g_scheduled", True)
p = params()
result = plugin.on_send_message_hook(0, p)
check(
"Use scheduled messages schedules 12 s ahead",
result.strategy == HookStrategy.MODIFY and p.scheduleDate == NOW + 12,
f"{p.scheduleDate - NOW}",
)
p = params(peer=556, photo=object())
plugin.on_send_message_hook(0, p)
check("photos get extra upload time", p.scheduleDate == NOW + 22, f"{p.scheduleDate - NOW}")
p2 = params(peer=556, photo=object())
plugin.on_send_message_hook(0, p2)
check("album parts share one date", p2.scheduleDate == p.scheduleDate)
skipped = {
"bots": params(peer=BOT_ID),
"Saved Messages": params(peer=111),
"secret chats": params(peer=FakeDialogObject.makeEncryptedDialogId(3)),
"already scheduled": params(scheduleDate=NOW + 3600),
}
for what, p in skipped.items():
before = p.scheduleDate
plugin.on_send_message_hook(0, p)
check(f"no auto-schedule for {what}", p.scheduleDate == before, f"{p.scheduleDate}")
plugin.prefs.set_chat_choice(111, 557, plugin_features(plugin).SCHEDULED, 2)
p = params(peer=557)
plugin.on_send_message_hook(0, p)
check("Use scheduled messages off per chat", p.scheduleDate == 0)
reset(plugin)
plugin.prefs.set_chat_choice(111, 555, plugin_features(plugin).SILENT, 1)
a, b = params(), params(peer=556)
plugin.on_send_message_hook(0, a)
plugin.on_send_message_hook(0, b)
check("Send without sound per chat", a.notify is False and b.notify is True)
# forwards
reset(plugin)
plugin.set_setting("g_scheduled", True)
plugin.set_setting("g_silent", True)
forward = HOOKED[("org.telegram.messenger.SendMessagesHelper", "sendMessage")]
helper = types.SimpleNamespace(currentAccount=0)
param = FakeParam(args=[FakeArrayList([object()]), 555, False, False, True, 0, None], this=helper)
forward.before_hooked_method(param)
check(
"forwards are scheduled and silent too",
param.args[5] == NOW + 12 and param.args[4] is False,
f"{param.args}",
)
param = FakeParam(args=[object()], this=helper)
forward.before_hooked_method(param)
check("sendMessage(SendMessageParams) overload left alone", param.args == param.args[:1])
notification = HOOKED[("org.telegram.ui.ChatActivity", "didReceivedNotification")]
seen = []
for dialog, scheduled in ((555, True), (558, True), (555, False)):
chat = ChatActivity(0, dialog)
chat.isPaused = False
param = FakeParam(args=[NEW_MESSAGES, 0, [dialog, FakeArrayList(), scheduled]], this=chat)
notification.before_hooked_method(param)
seen.append(chat.isPaused)
notification.after_hooked_method(param)
seen.append(chat.isPaused)
check(
"the chat doesn't jump to scheduled messages after auto-scheduling (paused for that call only)",
seen == [True, False, False, False, False, False],
f"{seen}",
)
chat = ChatActivity(0, 555)
HOOKED[("org.telegram.ui.ChatActivity", "forwardMessages")].after_hooked_method(FakeParam(this=chat))
check(
"the forward panel closes after a scheduled forward",
chat.panel_hidden and chat.waitingForSendingMessageLoad is False,
)
def verify_ui(plugin):
reset(plugin)
rows = plugin.create_settings()
keys = {getattr(r, "key", None) for r in rows}
want = {f"g_{k}" for k in plugin_features(plugin).BY_KEY}
check("settings page has a global switch per option", want <= keys, f"{want - keys}")
accounts = [r for r in rows if getattr(r, "create_sub_fragment", None)]
check("settings page lists every signed-in account", len(accounts) == len(ACCOUNTS))
master = next(r for r in rows if getattr(r, "key", None) == "g_ghost")
master.on_change(True)
essentials = plugin_features(plugin).ESSENTIALS
check(
"master switch turns every essential on",
all(plugin.get_setting(f"g_{f.key}") is True for f in essentials),
)
page = accounts[1].create_sub_fragment()
selectors = [r for r in page if getattr(r, "items", None)]
check(
"account page has a selector per option plus Ghost mode",
len(selectors) == len(plugin_features(plugin).ALL) + 1,
f"{len(selectors)}",
)
selectors[0].on_change(2) # Ghost mode: Off for account 1
check(
"account Ghost mode selector sets that account's essentials only",
all(plugin.get_setting(f"a_222_{f.key}") == 2 for f in essentials)
and all(plugin.get_setting(f"a_111_{f.key}") is None for f in essentials),
)
check(
"…and the account is out of ghost mode while the global one is on",
not plugin.prefs.ghost_active(222) and plugin.prefs.ghost_active(111),
)
# the chat dialog
reset(plugin)
LAST_FRAGMENT[0] = ChatActivity(0, 555)
chat_item = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.CHAT_ACTION_MENU)
chat_item.on_click({"account": 0, "dialog_id": 555, "fragment": LAST_FRAGMENT[0]})
check("chat menu opens the chat dialog", len(DIALOGS) == 1, f"{len(DIALOGS)}")
dialog = DIALOGS[-1]
check(
"chat rows read '<emoji> <state> <title>', untouched options Inherit",
dialog.items[dialog.find("Don't read messages")] == "📨 Inherit Don't read messages"
and dialog.items[0] == "👻 Off Ghost mode",
f"{dialog.items}",
)
row = dialog.find("Don't read messages")
check("Inherit is gray", dialog.rows[row].text.colored() == [("Inherit", GRAY)], f"{dialog.rows[row].text.colored()}")
dialog.click(row)
check(
"tapping an option sets it for this chat and redraws in place",
plugin.prefs.chat_choice(111, 555, plugin_features(plugin).NO_READ) == 1
and len(DIALOGS) == 1
and dialog.items[row] == "📨 On Don't read messages",
f"{plugin.prefs.chat_overrides(111)} dialogs={len(DIALOGS)} {dialog.items}",
)
check("On is green", dialog.rows[row].text.colored() == [("On", GREEN)], f"{dialog.rows[row].text.colored()}")
dialog.click(row)
check(
"…then Off, in red",
dialog.items[row] == "📨 Off Don't read messages" and dialog.rows[row].text.colored() == [("Off", RED)],
f"{dialog.items[row]!r}",
)
dialog.click(row)
check("…then back to Inherit", dialog.items[row] == "📨 Inherit Don't read messages", f"{dialog.items[row]!r}")
dialog.click(row)
check(
"the chat now drops reads, other chats don't",
pre(plugin, read(555)) == HookStrategy.CANCEL and pre(plugin, read(556)) == HookStrategy.DEFAULT,
)
dialog.click(0) # Ghost mode in this chat (currently off): turn on
chat_essentials = [f for f in essentials if f.chat]
check(
"chat ghost toggle sets the chat's essentials",
all(plugin.prefs.chat_choice(111, 555, f) == 1 for f in chat_essentials)
and dialog.items[0] == "👻 On Ghost mode",
f"{plugin.prefs.chat_overrides(111)} {dialog.items[0]!r}",
)
check(
"every row has an emoji prefix",
all(not text[0].isascii() for text in dialog.items),
f"{dialog.items}",
)
dialog.click(dialog.find("Use account settings"))
check(
"reset removes the chat's exceptions",
not plugin.prefs.chat_overrides(111) and not any(t.endswith("Use account settings") for t in dialog.items),
)
dialog.click(0) # ghost on again, so reading is blocked and "Read all" shows
dialog.click(dialog.find("Read all messages now"))
check("Read all messages now closes the popup", dialog.dismissed)
# message menu
reset(plugin)
message = types.SimpleNamespace(getDialogId=lambda: 555, getId=lambda: 77, currentAccount=1)
item = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.MESSAGE_CONTEXT_MENU)
item.on_click({"message": message})
check(
"Read until here sends a read receipt up to that message from its account",
len(SENT) == 1 and SENT[0][0] == 1 and SENT[0][1].max_id == 77,
f"{SENT}",
)
# drawer: the current account's options
reset(plugin)
drawer = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.DRAWER_MENU)
check("drawer item has the ghost icon", drawer.icon == "ghost", f"{drawer.icon!r}")
chat_item = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.CHAT_ACTION_MENU)
check("chat menu item has the ghost icon", chat_item.icon == "ghost", f"{chat_item.icon!r}")
FakeUserConfig.selectedAccount = 1
LAST_FRAGMENT[0] = ChatActivity(1, 0) # whatever screen is under the drawer
drawer.on_click({})
check(
"drawer opens the current account's options",
len(DIALOGS) == 1 and "User222" in DIALOGS[0].title,
f"{[d.title for d in DIALOGS]}",
)
check(
"account dialog lists every option",
len(DIALOGS[0].items) == 1 + len(plugin_features(plugin).ALL),
f"{DIALOGS[0].items}",
)
dialog = DIALOGS[0]
check(
"account rows show Global for options the account doesn't set",
dialog.items[dialog.find("Send without sound")] == "🔕 Global Send without sound",
f"{dialog.items}",
)
dialog.click(0)
check(
"account dialog ghost toggle turns ghost mode on for that account only",
plugin.prefs.ghost_active(222) and not plugin.prefs.ghost_active(111),
)
row = dialog.find("Send without sound")
dialog.click(row)
check(
"tapping an option cycles the account's choice",
plugin.get_setting("a_222_silent") == 1 and dialog.items[row] == "🔕 On Send without sound",
f"{plugin.get_setting('a_222_silent')} {dialog.items[row]!r}",
)
dialog.click(dialog.find("Use global settings"))
check("reset returns the account to the global settings", not plugin.prefs.account_overrides(222))
FakeUserConfig.selectedAccount = 0
def verify_drawer(plugin):
reset(plugin)
rebuild = HOOKED[("com.exteragram.messenger.drawer.DrawerMenuView", "rebuildMenu")]
menu = FakeDrawerMenuView()
rebuild.after_hooked_method(FakeParam(args=[1, None], this=menu))
rows = menu.container.children
check(
"without a Plugins entry the drawer gets a Ghost mode row with the ghost icon",
len(rows) == 1 and rows[0].menu_item == (PLUGINS_ENTRY, 1, GHOST_DRAWABLE, "Ghost mode"),
f"{[r.menu_item for r in rows]}",
)
LAST_FRAGMENT[0] = ChatActivity(1, 0)
rows[0].listener.onClick(None)
check(
"the row closes the drawer and opens that account's options",
menu.closed == 1 and len(DIALOGS) == 1 and "User222" in DIALOGS[0].title,
f"closed={menu.closed} {[d.title for d in DIALOGS]}",
)
MAIN_MENU_LAYOUT.append(PLUGINS_ENTRY)
menu = FakeDrawerMenuView()
rebuild.after_hooked_method(FakeParam(args=[0, None], this=menu))
check("with a Plugins entry the app's own item is used, no second row", not menu.container.children)
MAIN_MENU_LAYOUT.remove(PLUGINS_ENTRY)
def verify_icons(plugin):
sub_item = HOOKED[("org.telegram.ui.ActionBar.ActionBarMenuSubItem", "setTextAndIcon")]
item = types.SimpleNamespace(imageView=FakeImageView(FakeDrawable(96)))
sub_item.after_hooked_method(FakeParam(args=["Ghost mode", GHOST_DRAWABLE], this=item))
check(
"the ghost icon in chat/profile menus is drawn at 24dp",
isinstance(item.imageView.drawable, FakeLayerDrawable) and item.imageView.drawable.size == 72,
f"{vars(item.imageView.drawable)}",
)
other = types.SimpleNamespace(imageView=FakeImageView(FakeDrawable(96)))
sub_item.after_hooked_method(FakeParam(args=["Forward", 12345], this=other))
check("other menu icons are left alone", not isinstance(other.imageView.drawable, FakeLayerDrawable))
drawer_item = HOOKED[("com.exteragram.messenger.drawer.DrawerMenuItemView", "setMenuItem")]
row = types.SimpleNamespace(iconView=FakeImageView(FakeDrawable(96)))
drawer_item.after_hooked_method(FakeParam(args=[7, 0, GHOST_DRAWABLE, "Ghost mode"], this=row))
check(
"the ghost icon in the drawer is drawn at 24dp",
isinstance(row.iconView.drawable, FakeLayerDrawable) and row.iconView.drawable.size == 72,
)
small = types.SimpleNamespace(iconView=FakeImageView(FakeDrawable(72)))
drawer_item.after_hooked_method(FakeParam(args=[7, 0, GHOST_DRAWABLE, "Ghost mode"], this=small))
check("an icon already 24dp isn't wrapped", not isinstance(small.iconView.drawable, FakeLayerDrawable))
def verify_forums(plugin):
"""A forum is one chat: its settings cover all topics."""
D, C = HookStrategy.DEFAULT, HookStrategy.CANCEL
reset(plugin)
nr = plugin_features(plugin).NO_READ
plugin.prefs.set_chat_choice(111, -FORUM_ID, nr, 1)
discussion = lambda topic: TLRPC.TL_messages_readDiscussion(
peer=TL_inputPeerChannel(channel_id=FORUM_ID), msg_id=topic, read_max_id=600
)
check(
"a forum's setting applies to every topic's read receipts",
all(pre(plugin, discussion(t)) == C for t in FORUM_TOPICS),
)
typing = TLRPC.TL_messages_setTyping(peer=TL_inputPeerChannel(channel_id=FORUM_ID))
typing.top_msg_id = 10
plugin.prefs.set_chat_choice(111, -FORUM_ID, plugin_features(plugin).NO_TYPING, 1)
check("…and to typing in any topic", pre(plugin, typing) == C)
check("the forum's settings are stored once, for the group", list(plugin.prefs.chat_overrides(111)) == [-FORUM_ID])
# read on interact in a topic reads that topic
send = TLRPC.TL_messages_sendMessage(peer=TL_inputPeerChannel(channel_id=FORUM_ID))
send.reply_to = types.SimpleNamespace(top_msg_id=10, reply_to_msg_id=10)
pre(plugin, send)
plugin.post_request_hook("TL_messages_sendMessage", 0, object(), None)
check(
"Read on interact in a topic reads that topic (readDiscussion)",
len(SENT) == 1
and type(SENT[0][1]).__name__ == "TL_messages_readDiscussion"
and SENT[0][1].msg_id == 10
and SENT[0][1].read_max_id == FORUM_TOPICS[10],
f"{[(type(r).__name__, vars(r)) for _, r in SENT]}",
)
check("…and it isn't blocked by the forum's own setting", pre(plugin, SENT[0][1]) == D)
SENT.clear()
general = TLRPC.TL_messages_sendMessage(peer=TL_inputPeerChannel(channel_id=FORUM_ID))
pre(plugin, general)
plugin.post_request_hook("TL_messages_sendMessage", 0, object(), None)
check(
"a message to General reads the General topic",
len(SENT) == 1 and SENT[0][1].msg_id == 1 and SENT[0][1].read_max_id == FORUM_TOPICS[1],
f"{[vars(r) for _, r in SENT]}",
)
pre(plugin, SENT[0][1]) # as the app would, which uses up its allowance
# Read all messages now
SENT.clear()
LAST_FRAGMENT[0] = ChatActivity(0, -FORUM_ID)
chat_item = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.CHAT_ACTION_MENU)
chat_item.on_click({"account": 0, "dialog_id": -FORUM_ID})
DIALOGS[-1].click(DIALOGS[-1].find("Read all messages now"))
check(
"Read all messages now reads every topic up to its newest message",
sorted((r.msg_id, r.read_max_id) for _, r in SENT) == sorted(FORUM_TOPICS.items()),
f"{[vars(r) for _, r in SENT]}",
)
check("…and all of them get through", all(pre(plugin, r) == D for _, r in SENT))
check("…once", pre(plugin, SENT[0][1]) == C)
# Read until here on a topic message
SENT.clear()
owner = types.SimpleNamespace(
reply_to=types.SimpleNamespace(forum_topic=True, reply_to_top_id=20, reply_to_msg_id=715),
action=None,
)
message = types.SimpleNamespace(
getDialogId=lambda: -FORUM_ID, getId=lambda: 716, currentAccount=0, messageOwner=owner
)
item = next(i for i in MENU_ITEMS if i.menu_type == MenuItemType.MESSAGE_CONTEXT_MENU)
item.on_click({"message": message})
check(
"Read until here in a topic reads that topic up to the message",
len(SENT) == 1 and SENT[0][1].msg_id == 20 and SENT[0][1].read_max_id == 716,
f"{[vars(r) for _, r in SENT]}",
)
def main(argv=None) -> int:
argv = sys.argv[1:] if argv is None else argv
if len(argv) != 1:
print("usage: tools/verify_bundle.py <bundle.plugin>", file=sys.stderr)
return 1
bundle_path = Path(argv[0])
if not bundle_path.is_file():
print(f"error: bundle not found at {bundle_path}", file=sys.stderr)
return 1
install_fakes()
source = bundle_path.read_text(encoding="utf-8")
meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
modules_before = set(sys.modules)
ns = {"__name__": meta["id"]}
exec(compile(source, str(bundle_path), "exec"), ns)
new_modules = set(sys.modules) - modules_before
expected = expected_dunders(meta)
mismatches = [
f"{dunder}: expected {value!r}, got {ns.get(dunder)!r}"
for dunder, value in expected.items()
if ns.get(dunder) != value
]
check("metadata constants match plugin.json", not mismatches, "; ".join(mismatches))
tree = ast.parse(source, filename=str(bundle_path))
top_level_assigned = {
target.id
for node in tree.body
if isinstance(node, ast.Assign)
for target in node.targets
if isinstance(target, ast.Name)
}
check(
"metadata exposed as top-level literal assignments (AST)",
set(expected) <= top_level_assigned,
f"missing: {set(expected) - top_level_assigned}",
)
plugin_classes = [
value
for value in ns.values()
if isinstance(value, type) and issubclass(value, BasePlugin) and value is not BasePlugin
]
check("exactly one BasePlugin subclass defined", len(plugin_classes) == 1, f"found {len(plugin_classes)}")
if len(plugin_classes) == 1:
plugin = plugin_classes[0]()
plugin.on_plugin_load()
check("on_plugin_load emits load log line", any("ghost_mode loaded" in l for l in LOG_LINES), f"{LOG_LINES}")
check("no errors while loading", not [l for l in LOG_LINES if "loaded" not in l], f"{LOG_LINES}")
network = type(plugin.network).pre.__globals__
check("every request ghost mode handles is hooked", set(network["NAMES"]) <= set(ADDED_HOOKS))
expected_hooks = {
("org.telegram.messenger.SendMessagesHelper", "sendMessage"),
("org.telegram.ui.ChatActivity", "didReceivedNotification"),
("org.telegram.ui.ChatActivity", "forwardMessages"),
("com.exteragram.messenger.drawer.DrawerMenuView", "rebuildMenu"),
("org.telegram.ui.ActionBar.ActionBarMenuSubItem", "setTextAndIcon"),
("com.exteragram.messenger.drawer.DrawerMenuItemView", "setMenuItem"),
}
check("Xposed hooks installed", expected_hooks <= set(HOOKED), f"{set(HOOKED)}")
check(
"menu items: chat, profile, message, drawer",
{i.menu_type for i in MENU_ITEMS}
== {
MenuItemType.CHAT_ACTION_MENU,
MenuItemType.PROFILE_ACTION_MENU,
MenuItemType.MESSAGE_CONTEXT_MENU,
MenuItemType.DRAWER_MENU,
},
)
for part in (verify_requests, verify_sending, verify_ui, verify_drawer, verify_icons, verify_forums):
try:
part(plugin)
except Exception:
import traceback
check(part.__name__, False, traceback.format_exc())
plugin.on_plugin_unload()
check("unload removes the request hooks", not ADDED_HOOKS, f"{ADDED_HOOKS}")
errors = [l for l in LOG_LINES if "failed" in l or "hook:" in l]
check("no hook errors logged", not errors, f"{errors}")
bundled = {"main", "features", "prefs", "peers", "network", "sending", "menus", "settings_page", "hooking", "drawer", "icons", "options_dialog"}
check("sys.modules gained no entry for bundled module names", not (bundled & new_modules), f"{bundled & new_modules}")
if _FAILURES:
print(f"\n{len(_FAILURES)} check(s) failed: {', '.join(_FAILURES)}", file=sys.stderr)
return 1
print("\nall checks passed")
return 0
if __name__ == "__main__":
sys.exit(main())