extendedMediaForwards/tools/verify_bundle.py

#!/usr/bin/env python3
"""Offline verification harness for a built .plugin bundle. Runs without a phone.

Usage: python3 tools/verify_bundle.py <bundle.plugin>
"""
from __future__ import annotations

import ast
import json
import sys
import traceback
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)


class HookStrategy(Enum):
    DEFAULT = auto()
    MODIFY = auto()
    CANCEL = auto()
    MODIFY_FINAL = auto()


@dataclass
class HookResult:
    strategy: HookStrategy = HookStrategy.DEFAULT
    params: object = None


LOG_LINES: list[str] = []


class BasePlugin:
    def __init__(self):
        self._settings = {}
        self.hooked = []

    def get_setting(self, key, default=None):
        return self._settings.get(key, default)

    def set_setting(self, key, value):
        self._settings[key] = value

    def hook_all_methods(self, clazz, name, hook):
        self.hooked.append((clazz, name, hook))
        return [object()]

    def log(self, message):
        fake_log(message)

    def on_plugin_load(self):
        pass

    def on_plugin_unload(self):
        pass

    def create_settings(self):
        return []


class MethodHook:
    pass


def fake_log(*args, **kwargs):
    LOG_LINES.append(" ".join(str(a) for a in args))


class _SettingsRow:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)


class FakeAdapter:
    """Stand-in for PhotoViewer.PageBlocksAdapter."""

    def __init__(self, page, media=None, file=None, video=False):
        self.page = page
        self.media = media
        self.file = file
        self.video = video

    def getParentObject(self):
        return self.page

    def getMedia(self, index):
        return self.media

    def getFile(self, index):
        return self.file

    def isVideo(self, index):
        return self.video


class FakeWebPage:
    def __init__(self, url, display_url=None):
        self.url = url
        self.display_url = display_url


class FakeRichMessage:
    """Stand-in for TL_iv.RichMessage."""


class FakeMessage:
    def __init__(self, rich_message=None):
        self.rich_message = rich_message


class FakeMessageObject:
    def __init__(self, msg_id, dialog_id, rich_message=None):
        self.messageOwner = FakeMessage(rich_message)
        self._id = msg_id
        self._dialog_id = dialog_id

    def getId(self):
        return self._id

    def getDialogId(self):
        return self._dialog_id


class FakeChat:
    def __init__(self, chat_id, channel, username=None):
        self.id = chat_id
        self.channel = channel
        self.username = username


FAKE_CHATS = {
    100: FakeChat(100, channel=True, username="publicchan"),
    200: FakeChat(200, channel=True),
    300: FakeChat(300, channel=False),
}


def install_fakes() -> None:
    base_plugin_mod = types.ModuleType("base_plugin")
    base_plugin_mod.BasePlugin = BasePlugin
    base_plugin_mod.MethodHook = MethodHook
    base_plugin_mod.HookResult = HookResult
    base_plugin_mod.HookStrategy = HookStrategy
    sys.modules["base_plugin"] = base_plugin_mod

    android_utils_mod = types.ModuleType("android_utils")
    android_utils_mod.log = fake_log
    android_utils_mod.run_on_ui_thread = lambda fn, delay=0: fn()

    class OnClickListener:
        def __init__(self, fn):
            self.fn = fn

    android_utils_mod.OnClickListener = OnClickListener
    sys.modules["android_utils"] = android_utils_mod

    hook_utils_mod = types.ModuleType("hook_utils")
    hook_utils_mod.find_class = lambda name: FAKE_CLASSES.get(name)
    hook_utils_mod.get_private_field = lambda obj, field: getattr(obj, field, None)
    sys.modules["hook_utils"] = hook_utils_mod

    client_utils_mod = types.ModuleType("client_utils")
    client_utils_mod.get_selected_account = lambda: 0
    client_utils_mod.get_last_fragment = lambda: None
    client_utils_mod.get_send_messages_helper = lambda account=None: None
    sys.modules["client_utils"] = client_utils_mod

    # `TLRPC.WebPage` is only used for an isinstance() narrowing, so a fake
    # class hierarchy is enough to exercise ivmedia.page_of().
    tgnet_mod = types.ModuleType("org.telegram.tgnet")
    tlrpc = types.SimpleNamespace(
        WebPage=FakeWebPage,
        # `_link_entities` fills these in as plain attributes, so a bare
        # attribute holder reproduces the real TL object closely enough.
        TL_messageEntityTextUrl=lambda: types.SimpleNamespace(
            offset=0, length=0, url=None
        ),
    )
    tgnet_mod.TLRPC = tlrpc
    tgnet_mod.__path__ = []
    tgnet_tl_mod = types.ModuleType("org.telegram.tgnet.tl")
    tgnet_tl_mod.TL_iv = types.SimpleNamespace(RichMessage=FakeRichMessage)
    tgnet_mod.tl = tgnet_tl_mod
    sys.modules["org.telegram.tgnet.tl"] = tgnet_tl_mod

    messenger_mod = types.ModuleType("org.telegram.messenger")
    messenger_mod.MessageObject = FakeMessageObject
    messenger_mod.ChatObject = types.SimpleNamespace(
        isChannel=lambda chat: chat.channel,
        getPublicUsername=lambda chat: chat.username,
    )
    messenger_mod.MessagesController = types.SimpleNamespace(
        getInstance=lambda account: types.SimpleNamespace(getChat=FAKE_CHATS.get)
    )
    sys.modules["org.telegram.messenger"] = messenger_mod

    # `java.util.ArrayList` only ever gets add()ed to and iterated.
    java_mod = types.ModuleType("java")
    java_mod.__path__ = []
    java_util_mod = types.ModuleType("java.util")

    class ArrayList(list):
        def add(self, item):
            self.append(item)
            return True

    java_util_mod.ArrayList = ArrayList
    java_mod.util = java_util_mod
    sys.modules.setdefault("java", java_mod)
    sys.modules["java.util"] = java_util_mod
    sys.modules["org.telegram.tgnet"] = tgnet_mod
    org_mod = types.ModuleType("org")
    org_mod.__path__ = []
    sys.modules.setdefault("org", org_mod)

    ui_mod = types.ModuleType("ui")
    ui_mod.__path__ = []
    sys.modules["ui"] = ui_mod

    ui_settings_mod = types.ModuleType("ui.settings")
    for cls_name in ("Header", "Input", "Text", "Switch"):
        setattr(ui_settings_mod, cls_name, type(cls_name, (_SettingsRow,), {}))
    sys.modules["ui.settings"] = ui_settings_mod
    ui_mod.settings = ui_settings_mod

    ui_bulletin_mod = types.ModuleType("ui.bulletin")

    class BulletinHelper:
        shown = []

        @classmethod
        def show_error(cls, message, fragment=None):
            cls.shown.append(("error", message))

        @classmethod
        def show_success(cls, message, fragment=None):
            cls.shown.append(("success", message))

    ui_bulletin_mod.BulletinHelper = BulletinHelper
    sys.modules["ui.bulletin"] = ui_bulletin_mod
    ui_mod.bulletin = ui_bulletin_mod


FAKE_CLASSES = {"org.telegram.ui.PhotoViewer": object()}


class SimpleParams:
    def __init__(self, message):
        self.message = message


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


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")

    modules_before = set(sys.modules)
    ns = {"__name__": json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))["id"]}
    exec(compile(source, str(bundle_path), "exec"), ns)
    new_modules = set(sys.modules) - modules_before

    meta = json.loads(PLUGIN_JSON.read_text(encoding="utf-8"))
    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)}",
    )

    plugin_cls = plugin_classes[0] if len(plugin_classes) == 1 else None
    if plugin_cls is not None:
        check(
            "plugin class __module__ matches executed namespace",
            plugin_cls.__module__ == ns["__name__"],
            f"__module__={plugin_cls.__module__!r} != {ns['__name__']!r}",
        )

        plugin = plugin_cls()
        LOG_LINES.clear()
        plugin.on_plugin_load()
        check(
            "on_plugin_load hooks PhotoViewer.onPhotoShow",
            [name for _c, name, _h in plugin.hooked] == ["onPhotoShow"],
            f"hooked: {[n for _c, n, _h in plugin.hooked]}",
        )

        rows = plugin.create_settings()
        check(
            "settings expose the any_article switch",
            any(getattr(row, "key", None) == "any_article" for row in rows),
            f"rows: {[getattr(r, 'key', None) for r in rows]}",
        )

        telegraph = FakeAdapter(FakeWebPage("https://telegra.ph/Some-Article-01-01"))
        graph_org = FakeAdapter(FakeWebPage("http://graph.org/x"))
        other = FakeAdapter(FakeWebPage("https://example.com/post"))
        lookalike = FakeAdapter(FakeWebPage("https://telegra.ph.evil.com/x"))
        no_page = FakeAdapter(None)

        check(
            "telegra.ph and graph.org articles are accepted by default",
            plugin.allows(telegraph) and plugin.allows(graph_org),
        )
        check(
            "non-telegraph article is rejected by default",
            not plugin.allows(other),
        )
        check(
            "host that merely starts with telegra.ph is rejected",
            not plugin.allows(lookalike),
        )
        check(
            "adapter without a WebPage parent is rejected",
            not plugin.allows(no_page),
        )

        plugin.set_setting("any_article", True)
        check(
            "any_article setting accepts other articles",
            plugin.allows(other) and not plugin.allows(no_page),
        )

        rich_msg = FakeAdapter(FakeMessageObject(5, -1000000000200, FakeRichMessage()))
        rich_bare = FakeAdapter(FakeRichMessage())
        plain_msg = FakeAdapter(FakeMessageObject(5, -1000000000200, None))
        check(
            "settings expose the rich_messages switch",
            any(getattr(row, "key", None) == "rich_messages" for row in rows),
        )
        check(
            "rich message media is accepted by default (with or without a message)",
            plugin.allows(rich_msg) and plugin.allows(rich_bare),
        )
        check(
            "message parent without a rich message is rejected",
            not plugin.allows(plain_msg),
        )
        plugin.set_setting("rich_messages", False)
        check(
            "rich_messages setting off rejects rich message media",
            not plugin.allows(rich_msg) and not plugin.allows(rich_bare),
        )

    ivmedia = ns.get("ivmedia")
    sender = ns.get("sender")
    caption = getattr(sender, "CAPTION_TEXT", None)
    link_entities = getattr(sender, "_link_entities", None)
    entities = link_entities(caption, "https://telegra.ph/x") if link_entities and caption else []
    entity = entities[0] if entities else None
    check(
        "caption is a single text_url entity covering the whole caption",
        isinstance(caption, str)
        and len(caption) > 0
        and entity is not None
        and entity.offset == 0
        and entity.length == len(caption)
        and entity.url == "https://telegra.ph/x",
        f"got caption={caption!r} entities={entities!r}",
    )

    message_link = getattr(ivmedia, "message_link", None)
    if message_link is not None:
        # Channel dialog ids are -(chat id) in this fake, like getDialogId().
        links = (
            message_link(0, FakeMessageObject(7, -100)),
            message_link(0, FakeMessageObject(8, -200)),
            message_link(0, FakeMessageObject(9, -300)),
            message_link(0, FakeMessageObject(9, 12345)),
            message_link(0, FakeMessageObject(-1, -100)),
            message_link(0, FakeRichMessage()),
        )
    else:
        links = None
    check(
        "message_link: public/private channels linked; groups, DMs, local and bare rich messages not",
        links
        == (
            "https://t.me/publicchan/7",
            "https://t.me/c/200/8",
            None,
            None,
            None,
            None,
        ),
        f"got {links!r}",
    )

    picker = ns.get("picker")
    selected_targets = getattr(picker, "_selected_targets", None)
    address = getattr(sender, "_address", None)
    targets = None
    addressed = None
    if selected_targets is not None and address is not None:

        class SparseArray:
            def __init__(self, pairs):
                self.pairs = pairs

            def size(self):
                return len(self.pairs)

            def keyAt(self, i):
                return self.pairs[i][0]

            def valueAt(self, i):
                return self.pairs[i][1]

        class TopicMessageObject:
            def __init__(self, account, message, generate_layout, check_media):
                self.message = message
                self.isTopicMainMessage = False

        plain, forum, mono = object(), object(), object()
        topic = types.SimpleNamespace(topicStartMessage="start-msg", from_id="user-peer")
        mono_topic = types.SimpleNamespace(topicStartMessage="unused", from_id="dm-user")
        alert = types.SimpleNamespace(
            selectedDialogs=SparseArray([(11, plain), (-22, forum), (-33, mono)]),
            selectedDialogTopics={forum: topic, mono: mono_topic},
        )
        messenger = sys.modules["org.telegram.messenger"]
        saved = messenger.MessageObject
        messenger.MessageObject = TopicMessageObject
        messenger.DialogObject = types.SimpleNamespace(getPeerDialogId=lambda peer: 4242)
        messenger.MessagesController = types.SimpleNamespace(
            getInstance=lambda account: types.SimpleNamespace(isMonoForum=lambda d: d == -33)
        )
        try:
            targets = selected_targets(alert, 0)
        finally:
            messenger.MessageObject = saved
        params = [
            address(types.SimpleNamespace(replyToMsg=None, replyToTopMsg=None, monoForumPeer=0), t)
            for t in targets
        ]
        addressed = [
            (
                t.peer,
                getattr(p.replyToTopMsg, "message", None),
                p.replyToMsg is p.replyToTopMsg,
                getattr(p.replyToTopMsg, "isTopicMainMessage", None),
                p.monoForumPeer,
            )
            for t, p in zip(targets, params)
        ]
    check(
        "selected forum topic is replied to; mono-forum topic sets monoForumPeer; plain chat untouched",
        addressed
        == [
            (11, None, True, None, 0),
            (-22, "start-msg", True, True, 0),
            (-33, None, True, None, 4242),
        ],
        f"got {addressed!r}",
    )

    media_at = getattr(ivmedia, "media_at", None)
    no_media = getattr(ivmedia, "NoMedia", None)
    raised = False
    traceback_text = ""
    if media_at is not None and no_media is not None:
        try:
            media_at(FakeAdapter(FakeWebPage("https://telegra.ph/x"), media=None), 0)
        except no_media:
            raised = True
            traceback_text = traceback.format_exc()
    check("media_at raises NoMedia when the block has no media", raised)
    check(
        "traceback names src/ivmedia.py with its source line",
        "src/ivmedia.py" in traceback_text and "no downloadable media" in traceback_text,
        f"traceback:\n{traceback_text}",
    )

    check(
        "sys.modules gained no entry for bundled module names",
        not ({"icons", "ivmedia", "sender", "picker", "main"} & new_modules),
        f"leaked: {({'icons', 'ivmedia', 'sender', 'picker', 'main'} & 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())