From 5405adf485720f02ee5351882ac4a97b1d719466 Mon Sep 17 00:00:00 2001 From: lyuksovannyy Date: Wed, 23 Sep 2026 12:59:15 +0200 Subject: [PATCH] [nobump] Unlocked Restrictions: screenshots, copy-forwarding, protected-chat lock Replaces the pluginBase sample with the plugin: - remove FLAG_SECURE (FlagSecureReason + Window.setFlags) for screenshots - unlock forward/copy/save UI in protected chats and re-send protected messages as copies from the user's account (media re-uploaded, albums regrouped, bot-only parts dropped) - red lock before protected chats' names in the chat list, header and info - settings toggles for all three - verify_bundle.py rewritten for the plugin - bundle.py enables the git hooks on first run; README documents them Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 117 +++++++++++-- plugin.json | 8 +- scripts/post-commit | 3 +- scripts/pre-commit | 3 + src/badge.py | 227 +++++++++++++++++++++++++ src/copier.py | 315 ++++++++++++++++++++++++++++++++++ src/forwarding.py | 122 +++++++++++++ src/greeting/__init__.py | 1 - src/greeting/template.py | 13 -- src/hooking.py | 37 ++++ src/main.py | 106 ++++++++---- src/protection.py | 14 ++ src/screenshots.py | 51 ++++++ src/textutil.py | 2 - tools/bundle.py | 46 +++++ tools/verify_bundle.py | 358 ++++++++++++++++++++++++++++++++------- 16 files changed, 1296 insertions(+), 127 deletions(-) create mode 100644 src/badge.py create mode 100644 src/copier.py create mode 100644 src/forwarding.py delete mode 100644 src/greeting/__init__.py delete mode 100644 src/greeting/template.py create mode 100644 src/hooking.py create mode 100644 src/protection.py create mode 100644 src/screenshots.py delete mode 100644 src/textutil.py diff --git a/README.md b/README.md index 1c35ab1..995ee97 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,56 @@ -# exteraGram Plugin Base +# Unlocked Restrictions -A build toolchain and ADB automation for writing [exteraGram](https://exteragram.app) plugins -as ordinary multi-file Python projects instead of hand-maintaining one giant `.py` file. +An [exteraGram](https://exteragram.app) plugin that lifts the "restrict saving content" +protection on your own device: screenshots work everywhere, messages from protected chats can +be forwarded (as copies), and protected chats are marked with a red lock. + +## Features + +Each feature has its own switch in the plugin settings, all on by default, and toggles take +effect without reloading the plugin. + +- **Allow screenshots.** Removes `FLAG_SECURE`, so screenshots and screen recording work in + protected chats, secret chats and on the passcode screen. +- **Allow forwarding.** The forward, copy and save options appear in protected chats. Forwarding + a protected message, from the long-press menu or the side share button, sends it to the chosen + chat as a new message from your account, with no "Forwarded from" header. Unprotected + messages in the same selection are still forwarded normally. +- **Mark protected chats.** A red lock before the chat's name in the chat list, the chat header + and the chat info screen. + +### How copying works + +Telegram refuses to forward protected content, and also to re-send its media by file id, so a +copy is sent the same way as a message you wrote: + +- text keeps its formatting; photos, videos, files, voice messages and GIFs are downloaded + (or taken from the local cache) and uploaded again; stickers, locations and contacts are sent + directly; +- albums are sent as media groups again (at most 10 items per group); +- bot-only parts are dropped: reply and inline keyboards, games, invoices; content that can't + be reproduced (polls, dice, stories, giveaways, paid media) is skipped, and only its text is + sent if there is any; +- the selection is sent once every item is ready, so one large download delays the batch; +- items that can't be copied are reported in a bulletin. + +Copied videos have no preview thumbnail. The lock covers groups and channels, not private chats +where the other person enabled content protection. + +## Runtime notes + +Things the on-device runtime does differently from the SDK docs, all handled in `src/`: + +- Java lists are not iterable from Python; use `hooking.java_list()`. +- `param.setResult()` in a before-hook does not skip the original method, so the forward hook + empties the message list instead. +- `threading.local` state is not shared between a before-hook and its after-hook. +- Java `String` values arrive as Python `str`, other `CharSequence`s as Java objects. +- Fields named like `java.lang.Object` methods (`SendMessageParams.notify`) must be set with + `hook_utils.set_private_field`. +- Exceptions escaping a Python thread vanish silently, so the copy worker catches everything. +- Shared TL objects (`TLRPC.Chat` and friends) are never modified for display purposes. + +The rest of this README documents the build toolchain inherited from pluginBase. ## How exteraGram plugins actually work @@ -34,15 +83,19 @@ This repo does not use the official Elyx/ElyxBuilder archive format — see plugin.json single source of truth for plugin metadata src/ the plugin, written as an ordinary Python package main.py entry module — exactly one BasePlugin subclass, absolute imports only - greeting/ sample package (proves relative + cross-module imports survive bundling) - textutil.py sample top-level sibling module + screenshots.py FLAG_SECURE removal (FlagSecureReason + Window.setFlags hooks) + forwarding.py unlocks forward UI and intercepts forwards of protected messages + copier.py re-sends protected messages as copies (downloads + re-uploads media, albums) + badge.py red lock before protected chats' names (chat list, header, chat info) + protection.py shared "does this chat restrict saving content" check + hooking.py shared Xposed hook helpers tools/ bundle.py compiles plugin.json + src/ into dist/.plugin devclient.py speaks the on-device dev-server protocol (127.0.0.1:42690) verify_bundle.py offline harness that proves a built bundle behaves correctly, no phone needed release.py publishes a built bundle to a Gitea release (used by CI) .gitea/workflows/ CI that builds, verifies and publishes the .plugin to Releases -scripts/ bash automation for the ADB debug loop (see below) +scripts/ ADB debug-loop automation and the git hooks (see below) docs/sdk-notes.md condensed SDK API surface gathered from the official docs ``` @@ -61,6 +114,8 @@ a private import hook (scoped to a per-module `__builtins__` dict, never touchin files were still on disk. `--dev` (the default) embeds absolute local paths so a debugger resolves breakpoints against your real source files; `--release` embeds repo-relative paths for distribution. Both modes produce byte-reproducible output for a given `src/` + `plugin.json`. +Its first run in a clone also enables the repo's git hooks (see +[Git hooks and versioning](#git-hooks-and-versioning)). ## Device prerequisites @@ -93,6 +148,43 @@ standard `ANDROID_SERIAL` env var for device selection. - `scripts/fetch-sdk-stubs.sh [--beta]` — downloads the latest SDK type stubs into `stubs/` for editor autocomplete. +## Git hooks and versioning + +`scripts/pre-commit` bumps the patch number of `version` in `plugin.json` on every commit +(`1.0.0` → `1.0.1`) and stages it into that same commit. `scripts/post-commit` re-stages +`plugin.json` afterwards, because commits of a file selection (`git commit -- `, which +some IDEs use) are built from a temporary index and would otherwise leave it looking modified. + +**The hooks enable themselves.** Git never activates hooks from repository content, so a +fresh clone, or a repo created from this template, starts without them. The first run of +`tools/bundle.py` fixes that by running `git config core.hooksPath scripts`, and every +workflow runs it: directly, or through `scripts/deploy.sh` and `scripts/watch.sh`. That +includes `--check`, so after cloning run + +```bash +python3 tools/bundle.py --check +``` + +once, before your first commit. A commit made before any build runs without the hooks. +`bundle.py` leaves `core.hooksPath` alone when it is already set (so pointing it elsewhere on +purpose sticks), skips the step in CI (whenever `CI` is set) and outside a git work tree, and +never fails a build over it. To turn the hooks off for a clone, run +`git config core.hooksPath .git/hooks`; `git config --unset core.hooksPath` only lasts +until the next build. + +The bump is skipped when: + +- the commit message contains `[no bump]`, `[skip bump]`, `[no version bump]` or + `[skip version]` (case-insensitive; `nobump`, `no-bump` and `no_bump` also work), +- `NO_BUMP=1` (or `GIT_NO_BUMP=1`) is set, e.g. `NO_BUMP=1 git commit`, which is the only way + for commits written in an editor, since their message doesn't exist yet when the hook runs, +- `version` in `plugin.json` already differs from `HEAD` (you bumped it by hand, e.g. for a + minor or major release), +- a merge, rebase, cherry-pick or revert is in progress. + +Because every version change publishes a release (see below), a normal commit + push to +`main` ships a new patch release. Use a stop word for commits that should not. + ## Publishing a release `.gitea/workflows/release.yml` builds, verifies and publishes the `.plugin` to the repo's @@ -108,7 +200,8 @@ The `version` field in `plugin.json` is the single source of truth: | the same `version` | rebuilds the existing release, replacing its attachment | So every push leaves a downloadable build attached to the current version, and bumping -`version` is what cuts a new release. Pushing a `v*` tag by hand still works and uses that +`version` is what cuts a new release. With the git hooks enabled every commit bumps it, +so every push of a normal commit cuts a new patch release. Pushing a `v*` tag by hand still works and uses that tag verbatim, as does a manual run (Actions → Publish plugin → Run workflow). Every path runs `--check`, builds with `--release`, and gates the upload on @@ -140,9 +233,11 @@ you want when publishing needs to reach beyond this repo. ## Writing your own plugin -1. Edit `plugin.json` (id, name, description, etc.). -2. Write your plugin under `src/`, with `src/main.py` as the entry module (absolute imports +1. Run `python3 tools/bundle.py --check` once. It enables the git hooks that auto-bump the + version on commit (see [Git hooks and versioning](#git-hooks-and-versioning)). +2. Edit `plugin.json` (id, name, description, etc.). +3. Write your plugin under `src/`, with `src/main.py` as the entry module (absolute imports only — it's executed at plugin-module top level, not inside a package). -3. `python3 tools/bundle.py --release && python3 tools/verify_bundle.py dist/.plugin` +4. `python3 tools/bundle.py --release && python3 tools/verify_bundle.py dist/.plugin` (substitute your `id` from `plugin.json`). -4. `bash scripts/deploy.sh --enable` with a device connected. +5. `bash scripts/deploy.sh --enable` with a device connected. diff --git a/plugin.json b/plugin.json index b1fc013..0d635c9 100644 --- a/plugin.json +++ b/plugin.json @@ -1,8 +1,8 @@ { - "id": "hello_bundle", - "name": "Hello Bundle", - "description": "Multi-file exteraGram plugin built by tools/bundle.py", - "author": "@yourhandle", + "id": "unlocked_restrictions", + "name": "Unlocked Restrictions", + "description": "Allows screenshots and forwarding (as a copy) in chats with protected content", + "author": "lyuksovannyy", "version": "1.0.0", "icon": "exteraPlugins/1", "app_version": ">=12.5.1", diff --git a/scripts/post-commit b/scripts/post-commit index bb14819..b44c59a 100755 --- a/scripts/post-commit +++ b/scripts/post-commit @@ -1,5 +1,6 @@ #!/bin/sh -# Companion to scripts/pre-commit. +# Companion to scripts/pre-commit. Both are enabled per clone by tools/bundle.py +# (it sets `core.hooksPath scripts` on its first run). # # `git commit -- ` (also how some IDEs commit a file selection) builds # the commit from a temporary index. The pre-commit bump therefore lands in the diff --git a/scripts/pre-commit b/scripts/pre-commit index b48deac..9d7743a 100755 --- a/scripts/pre-commit +++ b/scripts/pre-commit @@ -1,6 +1,9 @@ #!/usr/bin/env python3 """Increment the patch component of plugin.json's version as part of the commit. +Enabled per clone by tools/bundle.py, which sets `core.hooksPath scripts` on its +first run (git never enables hooks from repository content by itself). + Runs as a pre-commit hook, which is the only commit hook whose `git add` is picked up by the commit currently being created: diff --git a/src/badge.py b/src/badge.py new file mode 100644 index 0000000..8735cc7 --- /dev/null +++ b/src/badge.py @@ -0,0 +1,227 @@ +"""Red lock in front of the names of chats that restrict saving content. + +Shown in the chat list (DialogCell), the chat header (ChatAvatarContainer) +and the chat info screen (ProfileActivity). Chat objects themselves are never +modified: they are shared app-wide and may be written to the database at any +moment, so only the text handed to the views is changed. + +Hook state must not live in threading.local: in this runtime a before-hook +and its after-hook do not see the same thread-local storage. +""" +from base_plugin import MethodHook +from hook_utils import get_private_field +from hooking import hook_all +from protection import chat_is_protected + +# An em space (replaced by the lock) plus a normal space as the gap before the +# name. The lock span is exactly as wide as the em space, so text measured +# with the placeholder keeps its width once the lock is swapped in. +PLACEHOLDER = "  " + +# Title of the protected chat whose DialogCell is being laid out right now, +# or None. DialogCell.buildLayout() runs on the UI thread and does not nest. +_building_title = None + + +def lock_span(paint): + """An ImageSpan drawing a red lock exactly one em space wide for `paint`.""" + from android.graphics import PorterDuff, PorterDuffColorFilter + from android.text.style import ImageSpan + from org.telegram.messenger import ApplicationLoader, R + from org.telegram.ui.ActionBar import Theme + + drawable = ApplicationLoader.applicationContext.getResources().getDrawable(R.drawable.list_secret).mutate() + drawable.setColorFilter( + PorterDuffColorFilter(Theme.getColor(Theme.key_text_RedBold), PorterDuff.Mode.SRC_IN) + ) + box = int(round(paint.measureText(PLACEHOLDER[0]))) + height = int(round(paint.getTextSize() * 0.75)) + ratio = drawable.getIntrinsicWidth() / max(1, drawable.getIntrinsicHeight()) + width = min(box, max(1, round(height * ratio))) + # ImageSpan reports its width as bounds.right, so right-aligning the lock + # in the box keeps the span exactly `box` wide. + drawable.setBounds(box - width, 0, box, height) + return ImageSpan(drawable, ImageSpan.ALIGN_BASELINE) + + +def with_lock(text, paint): + """`text` with the lock in front; a leftover bare placeholder is replaced.""" + from android.text import SpannableStringBuilder, Spanned + + # Java Strings arrive as Python str, other CharSequences as Java objects. + if str(text).startswith(PLACEHOLDER): + if isinstance(text, str): + text = text[len(PLACEHOLDER):] + else: + text = text.subSequence(len(PLACEHOLDER), text.length()) + builder = SpannableStringBuilder(PLACEHOLDER) + builder.append(text) + builder.setSpan(lock_span(paint), 0, 1, Spanned.SPAN_EXCLUSIVE_EXCLUSIVE) + return builder + + +def has_lock(text): + """True when `text` already starts with the lock span.""" + from android.text import Spanned + from android.text.style import ImageSpan + + if text is None or not str(text).startswith(PLACEHOLDER) or not isinstance(text, Spanned): + return False + return len(text.getSpans(0, 1, ImageSpan)) > 0 + + +def _protected_chat(account, dialog_id): + from org.telegram.messenger import MessagesController + + if account is None or dialog_id is None or dialog_id >= 0: + return None + controller = MessagesController.getInstance(account) + chat = controller.getChat(-dialog_id) + return chat if chat_is_protected(controller, chat) else None + + +class DialogCellNameHook(MethodHook): + """DialogCell.buildLayout(): lay the name out with room for the lock. + + Everything positioned after the name (verified badge, mute icon) is + computed inside buildLayout(), so the space has to exist while it runs: + EscapeHook prefixes the chat's title with PLACEHOLDER when buildLayout() + passes it through AndroidUtilities.escape(). Afterwards the placeholder in + the finished single-line layout is swapped for the lock. + """ + + def __init__(self, plugin, enabled): + self.plugin = plugin + self.enabled = enabled + + def before_hooked_method(self, param): + global _building_title + _building_title = None + if not self.enabled(): + return + try: + cell = param.thisObject + chat = _protected_chat( + get_private_field(cell, "currentAccount"), + get_private_field(cell, "currentDialogId"), + ) + if chat is not None and chat.title: + _building_title = chat.title + except Exception as exc: + self.plugin.log(f"lock badge: {exc}") + + def after_hooked_method(self, param): + global _building_title + if _building_title is None: + return + _building_title = None + try: + self._swap_in_lock(param.thisObject) + except Exception as exc: + self.plugin.log(f"lock badge: {exc}") + + @staticmethod + def _swap_in_lock(cell): + from android.text import Layout, StaticLayout + + layout = cell.nameLayout + if layout is None or layout.getLineCount() != 1: + return + text = layout.getText() + if not str(text).startswith(PLACEHOLDER): + return + paint = layout.getPaint() + cell.nameLayout = StaticLayout( + with_lock(text, paint), + paint, + layout.getWidth(), + Layout.Alignment.ALIGN_NORMAL, + 1.0, + 0.0, + False, + ) + + +class EscapeHook(MethodHook): + """AndroidUtilities.escape(String): see DialogCellNameHook.""" + + def after_hooked_method(self, param): + title = _building_title + if title is None or param.args[0] != title: + return + result = param.getResult() + if result is not None and not str(result).startswith(PLACEHOLDER): + param.setResult(PLACEHOLDER + str(result)) + + +class ChatHeaderTitleHook(MethodHook): + """ChatAvatarContainer.setTitle(CharSequence, boolean scam, boolean fake, + boolean verified, boolean premium, EmojiStatus, boolean animated). The + one-argument overload delegates here, so only the long one is handled.""" + + def __init__(self, plugin, enabled): + self.plugin = plugin + self.enabled = enabled + + def before_hooked_method(self, param): + args = param.args + if not self.enabled() or len(args) < 7 or args[0] is None or has_lock(args[0]): + return + try: + container = param.thisObject + fragment = get_private_field(container, "parentFragment") + if fragment is None: + return + if _protected_chat(fragment.getCurrentAccount(), fragment.getDialogId()) is None: + return + paint = get_private_field(container, "titleTextView").getPaint() + args[0] = with_lock(args[0], paint) + except Exception as exc: + self.plugin.log(f"lock badge: {exc}") + + +class ProfileNameHook(MethodHook): + """ProfileActivity.updateProfileData(boolean): prefix the chat info name.""" + + def __init__(self, plugin, enabled): + self.plugin = plugin + self.enabled = enabled + + def after_hooked_method(self, param): + if not self.enabled(): + return + try: + profile = param.thisObject + chat_id = get_private_field(profile, "chatId") + if not chat_id or _protected_chat(profile.getCurrentAccount(), -chat_id) is None: + return + views = get_private_field(profile, "nameTextView") + for i in range(len(views) if views is not None else 0): + view = views[i] + if view is None: + continue + text = view.getText() + if text is not None and len(str(text)) > 0 and not has_lock(text): + view.setText(with_lock(text, view.getPaint())) + except Exception as exc: + self.plugin.log(f"lock badge: {exc}") + + +def install(plugin, enabled): + hooked = hook_all(plugin, "org.telegram.ui.Cells.DialogCell", "buildLayout", DialogCellNameHook(plugin, enabled)) + hooked += hook_all(plugin, "org.telegram.messenger.AndroidUtilities", "escape", EscapeHook()) + hooked += hook_all( + plugin, "org.telegram.ui.Components.ChatAvatarContainer", "setTitle", ChatHeaderTitleHook(plugin, enabled) + ) + hooked += hook_all(plugin, "org.telegram.ui.ProfileActivity", "updateProfileData", ProfileNameHook(plugin, enabled)) + return hooked + + +def refresh_dialogs(): + """Asks the open chat list to rebuild its rows so a toggle shows at once.""" + from client_utils import get_notification_center + from org.telegram.messenger import MessagesController, NotificationCenter + + get_notification_center().postNotificationName( + NotificationCenter.updateInterfaces, MessagesController.UPDATE_MASK_NAME + ) diff --git a/src/copier.py b/src/copier.py new file mode 100644 index 0000000..04fb6c7 --- /dev/null +++ b/src/copier.py @@ -0,0 +1,315 @@ +"""Re-sends messages from protected chats as fresh copies. + +Telegram's server refuses to forward protected content, and also refuses to +re-send its media by file id, so every file is downloaded (or taken from the +local cache) and uploaded again as a new file. +""" +import random +import threading +import time +import traceback + +from android_utils import log, run_on_ui_thread +from hook_utils import set_private_field +from hooking import java_list +from protection import chat_is_protected + +DOWNLOAD_TIMEOUT_S = 600 +DOWNLOAD_POLL_S = 0.5 +MAX_ALBUM_SIZE = 10 + + +class UnsupportedMessage(Exception): + pass + + +def notify_error(text): + from ui.bulletin import BulletinHelper + + BulletinHelper.show_error(text) + + +def notify_info(text): + from ui.bulletin import BulletinHelper + + BulletinHelper.show_info(text) + + +def is_protected(message): + """True when Telegram would refuse to forward `message`. + + Reads the raw `noforwards` fields instead of calling + MessagesController.isChatNoForwards() and friends, since the plugin hooks + those to return false. + """ + from org.telegram.messenger import MessagesController + + owner = message.messageOwner + if owner is None: + return False + if owner.noforwards: + return True + controller = MessagesController.getInstance(message.currentAccount) + dialog_id = message.getDialogId() + if dialog_id < 0: + return chat_is_protected(controller, controller.getChat(-dialog_id)) + user_full = controller.getUserFull(dialog_id) + if user_full is None: + return False + return bool( + getattr(user_full, "noforwards_peer_enabled", False) + or getattr(user_full, "noforwards_my_enabled", False) + ) + + +def copy_messages(plugin, messages, peer, hide_caption, notify, schedule_date, top_msg): + """Starts copying `messages` to `peer` in the background. Call on the UI thread.""" + count = len(messages) + notify_info(f"Copying {count} protected message{'s' if count != 1 else ''}…") + + def copy_all(): + failed = 0 + built = [] # (message, params, has_media), in chat order + for message in sorted(messages, key=lambda m: m.getId()): + try: + params, has_media = _build_params(message, peer, hide_caption) + except UnsupportedMessage as exc: + plugin.log(f"skipping message {message.getId()}: {exc}") + failed += 1 + continue + except Exception as exc: + plugin.log(f"failed to prepare message {message.getId()}: {exc}") + failed += 1 + continue + # `notify` clashes with Object.notify() in Chaquopy, so plain + # attribute assignment fails; set the field through reflection. + set_private_field(params, "notify", notify) + params.scheduleDate = schedule_date + if top_msg is not None: + params.replyToMsg = top_msg + params.replyToTopMsg = top_msg + built.append((message, params, has_media)) + + _group_albums(built) + for message, params, _ in built: + _send_on_ui_thread(message.currentAccount, params) + if failed: + notify_error(f"{failed} of {count} message{'s' if count != 1 else ''} could not be copied") + + def worker(): + # An exception escaping a Python thread is silently lost on device. + try: + copy_all() + except Exception: + plugin.log("copy worker crashed: " + traceback.format_exc()) + notify_error("Copying protected messages failed") + + threading.Thread(target=worker, name="unlocked-restrictions-copy", daemon=True).start() + + +def album_runs(keys): + """Splits a sequence of album keys into runs of consecutive equal non-zero + keys, at most MAX_ALBUM_SIZE long. Returns (start, end) index pairs for + runs of two or more; key 0 means "not part of an album".""" + runs = [] + i = 0 + while i < len(keys): + j = i + 1 + while j < len(keys) and keys[i] != 0 and keys[j] == keys[i] and j - i < MAX_ALBUM_SIZE: + j += 1 + if j - i > 1: + runs.append((i, j)) + i = j + return runs + + +def _group_albums(built): + """Tags copies of album items so they are sent as one media group again. + + SendMessagesHelper groups messages whose params share a "groupId" and + sends the whole group once it sees the one marked "final" — the same + mechanism the attach menu uses. Items whose media had to be dropped are + sent on their own. + """ + from java.util import HashMap + + keys = [ + message.messageOwner.grouped_id if has_media else 0 for message, _, has_media in built + ] + for start, end in album_runs(keys): + group_id = str(random.getrandbits(62) + 1) + for index in range(start, end): + params = built[index][1] + params.params = HashMap() + params.params.put("groupId", group_id) + if index == end - 1: + params.params.put("final", "1") + + +def _send_on_ui_thread(account, params): + from org.telegram.messenger import SendMessagesHelper + + done = threading.Event() + + def send(): + try: + SendMessagesHelper.getInstance(account).sendMessage(params) + except Exception as exc: + log(f"[unlocked_restrictions] send failed: {exc}") + finally: + done.set() + + run_on_ui_thread(send) + # Queue sends one at a time so copies keep the original order. + done.wait(30) + + +def _new_params(text, peer): + from org.telegram.messenger import SendMessagesHelper + + return SendMessagesHelper.SendMessageParams.of(text, peer) + + +def _build_params(message, peer, hide_caption): + """Builds a plain user-account send of `message`'s content. + + Only what a regular client can send is copied. Bot-only parts (reply and + inline keyboards, games, invoices) and content that cannot be reproduced + (polls, dice, stories, giveaways, paid media) are dropped; if the message + has text, that text is still sent on its own. + """ + from org.telegram.messenger import MessageObject + from org.telegram.tgnet import TLRPC + + owner = message.messageOwner + media = MessageObject.getMedia(owner) + text = owner.message or "" + entities = owner.entities + + params = _new_params(None, peer) + has_media = True + if isinstance(media, TLRPC.TL_messageMediaPhoto) and media.photo is not None: + params.photo = _reupload_photo(message, media.photo) + elif isinstance(media, TLRPC.TL_messageMediaDocument) and media.document is not None: + if message.isAnyKindOfSticker(): + # Stickers belong to public sets, so the original file id is fine. + params.document = media.document + else: + params.document, params.path = _reupload_document(message, media.document) + elif isinstance(media, (TLRPC.TL_messageMediaGeo, TLRPC.TL_messageMediaVenue)): + params.location = media + elif isinstance(media, TLRPC.TL_messageMediaContact): + params.user = _contact_user(media) + else: + # No media, a link preview (regenerated from the text), or media a + # user account cannot send. + has_media = False + + if has_media: + if text and not hide_caption: + params.caption = text + params.entities = entities + params.invert_media = bool(getattr(owner, "invert_media", False)) + params.hasMediaSpoilers = bool(getattr(media, "spoiler", False)) + else: + if not text: + kind = media.getClass().getSimpleName() if media is not None else "empty message" + raise UnsupportedMessage(f"nothing a user account can send ({kind})") + params = _new_params(text, peer) + params.entities = entities + + # Never carry over bot-only fields. + params.replyMarkup = None + params.params = None + params.webPage = None + return params, has_media + + +def _contact_user(media): + from org.telegram.tgnet import TLRPC + + user = TLRPC.TL_user() + user.id = media.user_id + user.phone = media.phone_number + user.first_name = media.first_name or "" + user.last_name = media.last_name or "" + return user + + +def _reupload_photo(message, photo): + from org.telegram.messenger import FileLoader, ImageLocation, SendMessagesHelper + + if photo is None: + raise UnsupportedMessage("photo is missing") + best = None + for size in java_list(photo.sizes): + if size.location is None or size.w <= 0 or size.h <= 0: + continue + if best is None or size.w * size.h > best.w * best.h: + best = size + if best is None: + raise UnsupportedMessage("photo has no downloadable size") + + account = message.currentAccount + loader = FileLoader.getInstance(account) + candidates = [loader.getPathToAttach(best, True), loader.getPathToAttach(best, False)] + path = _existing(candidates) + if path is None: + loader.loadFile( + ImageLocation.getForPhoto(best, photo), message, "jpg", FileLoader.PRIORITY_HIGH, 0 + ) + path = _wait_for_file(candidates) + + new_photo = SendMessagesHelper.getInstance(account).generatePhotoSizes(path, None) + if new_photo is None: + raise UnsupportedMessage("could not re-encode photo") + return new_photo + + +def _reupload_document(message, document): + from java.util import ArrayList + from org.telegram.messenger import FileLoader + from org.telegram.tgnet import ConnectionsManager, TLRPC + + account = message.currentAccount + loader = FileLoader.getInstance(account) + candidates = [loader.getPathToAttach(document, True), loader.getPathToAttach(document, False)] + attach_path = message.messageOwner.attachPath + if attach_path: + from java.io import File + + candidates.insert(0, File(attach_path)) + path = _existing(candidates) + if path is None: + loader.loadFile(document, message, FileLoader.PRIORITY_HIGH, 0) + path = _wait_for_file(candidates) + + # id == 0 and access_hash == 0 make SendMessagesHelper upload `path` as a + # new file instead of referencing the protected original. + copy = TLRPC.TL_document() + copy.id = 0 + copy.access_hash = 0 + copy.file_reference = bytes() + copy.date = ConnectionsManager.getInstance(account).getCurrentTime() + copy.mime_type = document.mime_type + copy.size = document.size + copy.dc_id = 0 + copy.attributes = ArrayList(document.attributes) + return copy, path + + +def _existing(files): + for f in files: + if f is not None and f.exists() and f.length() > 0: + return f.getAbsolutePath() + return None + + +def _wait_for_file(files): + deadline = time.monotonic() + DOWNLOAD_TIMEOUT_S + while time.monotonic() < deadline: + path = _existing(files) + if path is not None: + return path + time.sleep(DOWNLOAD_POLL_S) + raise UnsupportedMessage("download timed out") diff --git a/src/forwarding.py b/src/forwarding.py new file mode 100644 index 0000000..256d3a7 --- /dev/null +++ b/src/forwarding.py @@ -0,0 +1,122 @@ +from base_plugin import MethodHook +from hooking import ReturnFalseHook, hook_all, java_list +import copier + + +class CanForwardHook(MethodHook): + """MessageObject.canForwardMessage(): evaluate as if `noforwards` were unset. + + When the original says no and the message is flagged `noforwards`, the + method is called again with the flag cleared and the flag is restored + straight away, because copier.is_protected() still needs to see it. (A + before/after pair can't carry the message between the two callbacks in + this runtime.) The nested call sees `noforwards` unset, so it does not + recurse further. + """ + + def __init__(self, enabled): + self.enabled = enabled + + def after_hooked_method(self, param): + if param.getResult() or not self.enabled(): + return + message = param.thisObject + owner = getattr(message, "messageOwner", None) + if owner is None or not owner.noforwards: + return + owner.noforwards = False + try: + param.setResult(bool(param.method.invoke(message))) + finally: + owner.noforwards = True + + +class ForwardInterceptHook(MethodHook): + """SendMessagesHelper.sendMessage(ArrayList messages, long peer, + boolean forwardFromMyName, boolean hideCaption, boolean notify, int scheduleDate, ...) + + Protected messages are never forwarded: they are pulled out of the list + and re-sent as ordinary messages from the user's account (see copier); + the rest are forwarded as usual. Every overload shares the leading six + parameters, and the shorter ones delegate to the longest, so the outermost + call does the split and the inner calls see only unprotected messages. + """ + + def __init__(self, plugin, enabled): + self.plugin = plugin + self.enabled = enabled + + def before_hooked_method(self, param): + if not self.enabled(): + return + from java.util import ArrayList + + args = param.args + if len(args) < 6 or not isinstance(args[0], ArrayList) or args[0].isEmpty(): + return + flagged = [(m, self._is_protected(m)) for m in java_list(args[0])] + protected = [m for m, is_protected in flagged if is_protected] + if not protected: + return + + # Removing the protected messages must happen even if copying fails, + # otherwise the original sendMessage() would forward them. setResult() + # does not reliably skip the original here, so the list is replaced + # instead: with nothing left, sendMessage() returns early on its own + # isEmpty() check, and the overloads it delegates to get the same list. + remaining = ArrayList() + for m, is_protected in flagged: + if not is_protected: + remaining.add(m) + args[0] = remaining + + try: + copier.copy_messages( + plugin=self.plugin, + messages=protected, + peer=int(args[1]), + hide_caption=bool(args[3]), + notify=bool(args[4]), + schedule_date=int(args[5]), + top_msg=self._topic_root(args), + ) + except Exception as exc: + self.plugin.log(f"copy failed: {exc}") + copier.notify_error(f"Could not copy protected messages: {exc}") + + def _is_protected(self, message): + try: + return copier.is_protected(message) + except Exception as exc: + # When in doubt, copy: a copy always works, a forward of a + # protected message never does. + self.plugin.log(f"protection check failed, copying instead: {exc}") + return True + + @staticmethod + def _topic_root(args): + """Forwarding into a forum topic passes the topic's root message.""" + from org.telegram.messenger import MessageObject + + for i in range(6, len(args)): + if isinstance(args[i], MessageObject): + return args[i] + return None + + +def install(plugin, enabled): + no_forwards = ReturnFalseHook(enabled) + hooked = 0 + for method in ("isChatNoForwards", "isUserNoForwards", "isPeerNoForwards"): + hooked += hook_all(plugin, "org.telegram.messenger.MessagesController", method, no_forwards) + hooked += hook_all(plugin, "org.telegram.ui.ChatActivity", "hasSelectedNoforwardsMessage", no_forwards) + hooked += hook_all( + plugin, "org.telegram.messenger.MessageObject", "canForwardMessage", CanForwardHook(enabled) + ) + hooked += hook_all( + plugin, + "org.telegram.messenger.SendMessagesHelper", + "sendMessage", + ForwardInterceptHook(plugin, enabled), + ) + return hooked diff --git a/src/greeting/__init__.py b/src/greeting/__init__.py deleted file mode 100644 index f0d6578..0000000 --- a/src/greeting/__init__.py +++ /dev/null @@ -1 +0,0 @@ -from .template import render, TemplateError diff --git a/src/greeting/template.py b/src/greeting/template.py deleted file mode 100644 index 5ccc0e9..0000000 --- a/src/greeting/template.py +++ /dev/null @@ -1,13 +0,0 @@ -from textutil import normalize - - -class TemplateError(ValueError): - pass - - -def render(template: str, name: str) -> str: - normalized = normalize(name) - try: - return template.format(name=normalized) - except (KeyError, IndexError, ValueError): - raise TemplateError(f"unsupported placeholder in {template!r}") diff --git a/src/hooking.py b/src/hooking.py new file mode 100644 index 0000000..aeb29c5 --- /dev/null +++ b/src/hooking.py @@ -0,0 +1,37 @@ +from base_plugin import MethodHook +from hook_utils import find_class + + +def java_list(items): + """Copies a java.util.List into a Python list (Java lists aren't iterable here).""" + if items is None: + return [] + return [items.get(i) for i in range(items.size())] + + +class ReturnFalseHook(MethodHook): + """Forces a boolean method to return false while `enabled()` is true.""" + + def __init__(self, enabled): + self.enabled = enabled + + def after_hooked_method(self, param): + if self.enabled(): + param.setResult(False) + + +def hook_all(plugin, class_name, method_name, hook): + """Hooks every overload of `class_name.method_name`; returns how many were hooked.""" + clazz = find_class(class_name) + if clazz is None: + plugin.log(f"class not found: {class_name}") + return 0 + try: + unhooks = plugin.hook_all_methods(clazz, method_name, hook) + except Exception as exc: + plugin.log(f"failed to hook {class_name}.{method_name}: {exc}") + return 0 + count = len(unhooks) if unhooks else 0 + if count == 0: + plugin.log(f"no overloads hooked for {class_name}.{method_name}") + return count diff --git a/src/main.py b/src/main.py index f24e0fa..7ee6229 100644 --- a/src/main.py +++ b/src/main.py @@ -1,50 +1,84 @@ -from base_plugin import BasePlugin, HookResult, HookStrategy -from android_utils import log -from ui.settings import Header, Input, Text -from greeting import render, TemplateError +from base_plugin import BasePlugin +from ui.settings import Header, Switch, Text +import badge +import forwarding +import screenshots -DEFAULT_TEMPLATE = "Hello, {name}!" +SCREENSHOTS_KEY = "allow_screenshots" +FORWARDING_KEY = "allow_forwarding" +LOCK_BADGE_KEY = "protected_lock_badge" -class HelloPlugin(BasePlugin): +class UnlockedRestrictionsPlugin(BasePlugin): def on_plugin_load(self): - log("hello_bundle loaded") - self.add_on_send_message_hook() + # Hooks stay installed for the plugin's lifetime and check the + # settings on every call, so toggles apply without a reload. + screenshots_hooked = screenshots.install(self, self._screenshots_enabled) + forwarding_hooked = forwarding.install(self, self._forwarding_enabled) + badge_hooked = badge.install(self, self._lock_badge_enabled) + self.log( + f"unlocked_restrictions loaded: {screenshots_hooked} screenshot hook(s), " + f"{forwarding_hooked} forwarding hook(s), {badge_hooked} lock badge hook(s)" + ) + if self._screenshots_enabled(): + self._clear_current_window() def on_plugin_unload(self): - log("hello_bundle unloaded") + self.log("unlocked_restrictions unloaded") + + def _screenshots_enabled(self): + return bool(self.get_setting(SCREENSHOTS_KEY, True)) + + def _forwarding_enabled(self): + return bool(self.get_setting(FORWARDING_KEY, True)) + + def _lock_badge_enabled(self): + return bool(self.get_setting(LOCK_BADGE_KEY, True)) + + def _on_lock_badge_change(self, enabled): + try: + badge.refresh_dialogs() + except Exception as exc: + self.log(f"could not refresh the chat list: {exc}") + + def _on_screenshots_change(self, enabled): + if enabled: + self._clear_current_window() + + def _clear_current_window(self): + try: + screenshots.clear_current_window() + except Exception as exc: + self.log(f"could not clear FLAG_SECURE on the current window: {exc}") def create_settings(self): return [ - Header(text="Hello Bundle"), - Input( - key="template", - text="Greeting template", - default=DEFAULT_TEMPLATE, - subtext="Use {name} as placeholder.", - icon="msg_edit", + Header(text="Unlocked Restrictions"), + Switch( + key=SCREENSHOTS_KEY, + text="Allow screenshots", + default=True, + subtext="Screenshots and screen recording in protected chats, secret chats and the passcode screen.", + icon="msg_photos", + on_change=self._on_screenshots_change, + ), + Switch( + key=FORWARDING_KEY, + text="Allow forwarding", + default=True, + subtext="Forwarding from protected chats sends a copy of the message instead.", + icon="msg_forward", + ), + Switch( + key=LOCK_BADGE_KEY, + text="Mark protected chats", + default=True, + subtext="Shows a red lock before the names of chats that restrict saving content.", + icon="msg_secret", + on_change=self._on_lock_badge_change, ), Text( - text="Example", - subtext=".hello alice smith -> Hello, Alice Smith!", + text="Copies have no \"Forwarded from\" header. Media is downloaded and uploaded again, so large files take a while.", icon="msg_info", ), ] - - def on_send_message_hook(self, account, params): - message = getattr(params, "message", None) - if not isinstance(message, str): - return HookResult() - stripped = message.strip() - if not stripped.startswith(".hello"): - return HookResult() - parts = stripped.split(" ", 1) - name = parts[1] if len(parts) > 1 else "" - if not name.strip(): - params.message = "Usage: .hello " - else: - try: - params.message = render(self.get_setting("template", DEFAULT_TEMPLATE), name) - except TemplateError as exc: - params.message = "Bad template: " + str(exc) - return HookResult(strategy=HookStrategy.MODIFY, params=params) diff --git a/src/protection.py b/src/protection.py new file mode 100644 index 0000000..c4b6ff5 --- /dev/null +++ b/src/protection.py @@ -0,0 +1,14 @@ +def chat_is_protected(controller, chat): + """True when `chat` (or the supergroup it migrated to) restricts saving content. + + Reads the raw `noforwards` field instead of calling + MessagesController.isChatNoForwards(), since the plugin hooks that to + return false. + """ + if chat is None: + return False + if chat.migrated_to is not None: + migrated = controller.getChat(chat.migrated_to.channel_id) + if migrated is not None: + return bool(migrated.noforwards) + return bool(chat.noforwards) diff --git a/src/screenshots.py b/src/screenshots.py new file mode 100644 index 0000000..6cb5ca1 --- /dev/null +++ b/src/screenshots.py @@ -0,0 +1,51 @@ +from base_plugin import MethodHook +from hooking import ReturnFalseHook, hook_all + +# android.view.WindowManager.LayoutParams.FLAG_SECURE +FLAG_SECURE = 0x2000 + + +class StripFlagSecureHook(MethodHook): + """Window.setFlags(flags, mask): drop FLAG_SECURE from the flags being set. + + Window.addFlags() delegates to setFlags(), so this catches every path that + tries to secure a window, including ones outside FlagSecureReason. + """ + + def __init__(self, enabled): + self.enabled = enabled + + def before_hooked_method(self, param): + if not self.enabled(): + return + args = param.args + if len(args) != 2: + return + flags = int(args[0]) + if flags & FLAG_SECURE: + args[0] = flags & ~FLAG_SECURE + + +def install(plugin, enabled): + hooked = hook_all( + plugin, + "org.telegram.messenger.FlagSecureReason", + "isSecuredNow", + ReturnFalseHook(enabled), + ) + hooked += hook_all(plugin, "android.view.Window", "setFlags", StripFlagSecureHook(enabled)) + return hooked + + +def clear_current_window(): + """Drops FLAG_SECURE from the window that is already on screen.""" + from android_utils import run_on_ui_thread + from client_utils import get_last_fragment + + def clear(): + fragment = get_last_fragment() + activity = fragment.getParentActivity() if fragment is not None else None + if activity is not None: + activity.getWindow().clearFlags(FLAG_SECURE) + + run_on_ui_thread(clear) diff --git a/src/textutil.py b/src/textutil.py deleted file mode 100644 index 7c3107b..0000000 --- a/src/textutil.py +++ /dev/null @@ -1,2 +0,0 @@ -def normalize(name: str) -> str: - return " ".join(w[:1].upper() + w[1:] for w in name.split()) diff --git a/tools/bundle.py b/tools/bundle.py index 72678be..0060891 100755 --- a/tools/bundle.py +++ b/tools/bundle.py @@ -2,6 +2,9 @@ """Bundle a multi-file exteraGram plugin (plugin.json + src/) into a single .plugin file. Usage: python3 tools/bundle.py [--dev|--release] [--out DIR] [--check] + +Also enables the repo's git hooks (scripts/pre-commit, scripts/post-commit) on +first run; see ensure_git_hooks(). """ from __future__ import annotations @@ -9,13 +12,18 @@ import argparse import ast import base64 import json +import os import re +import stat +import subprocess import sys import zlib from pathlib import Path REPO_ROOT = Path(__file__).resolve().parent.parent SRC_DIR = REPO_ROOT / "src" +HOOKS_DIR = "scripts" +HOOK_NAMES = ("pre-commit", "post-commit") PLUGIN_JSON = REPO_ROOT / "plugin.json" RECOGNIZED_KEYS = { @@ -300,6 +308,43 @@ def assemble_bundle(data: dict, modules: dict, release: bool) -> str: return text +def ensure_git_hooks() -> None: + """Point git at scripts/ for hooks, once per clone. + + Git never enables hooks from repository content, so a fresh clone (or a + repo created from the template) starts without them. Every workflow runs + this bundler first, directly or via scripts/deploy.sh and scripts/watch.sh, + which makes it the one place that can switch them on without an extra + setup step. Skipped in CI, outside a git work tree, and when + core.hooksPath is already set (including to something else on purpose). + Never fails the build. + """ + if os.environ.get("CI"): + return + hooks = [REPO_ROOT / HOOKS_DIR / name for name in HOOK_NAMES] + if not any(h.is_file() for h in hooks): + return + + def git(*args): + return subprocess.run(["git", *args], cwd=REPO_ROOT, capture_output=True, text=True) + + try: + top = git("rev-parse", "--show-toplevel") + if top.returncode != 0 or Path(top.stdout.strip()).resolve() != REPO_ROOT: + return + if git("config", "--get", "core.hooksPath").stdout.strip(): + return + # Git silently skips hooks that are not executable. + for hook in hooks: + if hook.is_file(): + mode = hook.stat().st_mode + hook.chmod(mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) + if git("config", "core.hooksPath", HOOKS_DIR).returncode == 0: + print(f"enabled git hooks from {HOOKS_DIR}/ (git config core.hooksPath {HOOKS_DIR})", file=sys.stderr) + except (OSError, subprocess.SubprocessError): + pass + + def parse_args(argv): parser = argparse.ArgumentParser( description="Bundle a multi-file exteraGram plugin into a single .plugin file." @@ -318,6 +363,7 @@ def parse_args(argv): def main(argv=None) -> int: args = parse_args(sys.argv[1:] if argv is None else argv) + ensure_git_hooks() release = bool(args.release) try: data = load_metadata() diff --git a/tools/verify_bundle.py b/tools/verify_bundle.py index 2dbc171..8819aa7 100755 --- a/tools/verify_bundle.py +++ b/tools/verify_bundle.py @@ -45,6 +45,43 @@ class HookResult: LOG_LINES: list[str] = [] +HOOKED: list[tuple[str, str, object]] = [] + +# Classes the plugin is expected to hook, mapped to the hooked method names. +EXPECTED_HOOKS = { + ("org.telegram.messenger.FlagSecureReason", "isSecuredNow"), + ("android.view.Window", "setFlags"), + ("org.telegram.messenger.MessagesController", "isChatNoForwards"), + ("org.telegram.messenger.MessagesController", "isUserNoForwards"), + ("org.telegram.messenger.MessagesController", "isPeerNoForwards"), + ("org.telegram.ui.ChatActivity", "hasSelectedNoforwardsMessage"), + ("org.telegram.messenger.MessageObject", "canForwardMessage"), + ("org.telegram.messenger.SendMessagesHelper", "sendMessage"), + ("org.telegram.ui.Cells.DialogCell", "buildLayout"), + ("org.telegram.ui.Components.ChatAvatarContainer", "setTitle"), + ("org.telegram.messenger.AndroidUtilities", "escape"), + ("org.telegram.ui.ProfileActivity", "updateProfileData"), +} + + +class MethodHook: + def before_hooked_method(self, param): + pass + + def after_hooked_method(self, param): + pass + + +class MethodReplacement: + pass + + +class FakeJavaClass: + def __init__(self, name): + self.name = name + + def getName(self): + return self.name class BasePlugin: @@ -54,12 +91,19 @@ class BasePlugin: def get_setting(self, key, default=None): return self._settings.get(key, default) - def set_setting(self, key, value): + def set_setting(self, key, value, reload_settings=False): self._settings[key] = value - def add_on_send_message_hook(self): + 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.append((hook_class.getName(), method_name, xposed_hook)) + return [object()] + + def log(self, message): + LOG_LINES.append(str(message)) + def on_plugin_load(self): pass @@ -74,36 +118,87 @@ def fake_log(*args, **kwargs): LOG_LINES.append(" ".join(str(a) for a in args)) +def fake_find_class(name): + return FakeJavaClass(name) + + class _SettingsRow: def __init__(self, **kwargs): self.__dict__.update(kwargs) +class FakeArrayList(list): + def isEmpty(self): + return not self + + def add(self, item): + self.append(item) + + def size(self): + return len(self) + + def get(self, i): + return self[i] + + +class FakeMessageObject: + def __init__(self, protected=False, noforwards=False): + self.protected = protected + self.messageOwner = types.SimpleNamespace(noforwards=noforwards) + + +class FakeParam: + def __init__(self, args=(), this=None, result=None): + self.args = None if args is None else 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: - base_plugin_mod = types.ModuleType("base_plugin") - base_plugin_mod.BasePlugin = BasePlugin - 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 - sys.modules["android_utils"] = android_utils_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 - - -class SimpleParams: - def __init__(self, message): - self.message = message + _module( + "base_plugin", + BasePlugin=BasePlugin, + HookResult=HookResult, + HookStrategy=HookStrategy, + MethodHook=MethodHook, + MethodReplacement=MethodReplacement, + ) + _module("android_utils", log=fake_log, run_on_ui_thread=lambda fn, delay=0: fn()) + _module( + "hook_utils", + find_class=fake_find_class, + set_private_field=lambda obj, name, value: setattr(obj, name, value) or True, + get_private_field=lambda obj, name: getattr(obj, name, None), + ) + _module("client_utils", get_last_fragment=lambda: None) + ui_mod = _module("ui", __path__=[]) + ui_mod.settings = _module( + "ui.settings", + **{n: type(n, (_SettingsRow,), {}) for n in ("Header", "Input", "Text", "Switch", "Divider")}, + ) + ui_mod.bulletin = _module("ui.bulletin", BulletinHelper=types.SimpleNamespace( + show_info=fake_log, show_error=fake_log, show_success=fake_log, + )) + java_mod = _module("java", __path__=[]) + java_mod.util = _module("java.util", ArrayList=FakeArrayList) + org = _module("org", __path__=[]) + org.telegram = _module("org.telegram", __path__=[]) + org.telegram.messenger = _module("org.telegram.messenger", MessageObject=FakeMessageObject) KEY_TO_DUNDER = { @@ -133,6 +228,154 @@ def expected_dunders(meta: dict) -> dict: return result +def verify_hooks(plugin, hooks) -> None: + flag_secure = 0x2000 + strip = hooks[("android.view.Window", "setFlags")] + for enabled in (True, False): + plugin.set_setting("allow_screenshots", enabled) + param = FakeParam(args=[flag_secure | 0x80, flag_secure | 0x80]) + strip.before_hooked_method(param) + want = 0x80 if enabled else flag_secure | 0x80 + check( + f"Window.setFlags strips FLAG_SECURE only when enabled (enabled={enabled})", + param.args[0] == want, + f"flags={param.args[0]:#x}", + ) + + plugin.set_setting("allow_screenshots", True) + param = FakeParam(result=True) + hooks[("org.telegram.messenger.FlagSecureReason", "isSecuredNow")].after_hooked_method(param) + check("FlagSecureReason.isSecuredNow forced false", param.getResult() is False) + + no_forwards = hooks[("org.telegram.messenger.MessagesController", "isChatNoForwards")] + for enabled in (True, False): + plugin.set_setting("allow_forwarding", enabled) + param = FakeParam(result=True) + no_forwards.after_hooked_method(param) + check( + f"isChatNoForwards forced false only when enabled (enabled={enabled})", + param.getResult() is (not enabled), + f"got {param.getResult()!r}", + ) + + plugin.set_setting("allow_forwarding", True) + can_forward = hooks[("org.telegram.messenger.MessageObject", "canForwardMessage")] + message = FakeMessageObject(noforwards=True) + seen = [] + # The re-invoked original answers "forwardable" only if noforwards is unset. + method = types.SimpleNamespace( + invoke=lambda obj: seen.append(obj.messageOwner.noforwards) or not obj.messageOwner.noforwards + ) + param = FakeParam(this=message, result=False) + param.method = method + can_forward.after_hooked_method(param) + check( + "canForwardMessage re-evaluates without noforwards and restores the flag", + param.getResult() is True and seen == [False] and message.messageOwner.noforwards is True, + f"result={param.getResult()!r} seen={seen} flag={message.messageOwner.noforwards!r}", + ) + param = FakeParam(this=FakeMessageObject(noforwards=False), result=False) + param.method = method + can_forward.after_hooked_method(param) + check("canForwardMessage keeps a refusal unrelated to noforwards", param.getResult() is False) + + escape = hooks[("org.telegram.messenger.AndroidUtilities", "escape")] + badge = type(escape).after_hooked_method.__globals__ + badge["_building_title"] = "Chat" + results = [] + for arg in ("Chat", "Other"): + param = FakeParam(args=[arg], result=arg) + escape.after_hooked_method(param) + results.append(param.getResult()) + badge["_building_title"] = None + param = FakeParam(args=["Chat"], result="Chat") + escape.after_hooked_method(param) + results.append(param.getResult()) + check( + "escape() prefixes only the title being laid out", + results == ["\u2003 Chat", "Other", "Chat"], + f"{results}", + ) + + intercept = hooks[("org.telegram.messenger.SendMessagesHelper", "sendMessage")] + copier = type(intercept).before_hooked_method.__globals__["copier"] + + cases = [ + ([], []), + ([0, 0, 0], []), + ([7], []), + ([7, 7, 7], [(0, 3)]), + ([0, 7, 7, 0, 9, 9, 9], [(1, 3), (4, 7)]), + ([7, 7, 9, 9], [(0, 2), (2, 4)]), + ([7, 0, 7], []), + ([5] * 12, [(0, 10), (10, 12)]), + ([5] * 11, [(0, 10)]), + ] + bad = [(k, copier.album_runs(k), want) for k, want in cases if copier.album_runs(k) != want] + check("album_runs groups consecutive album items, max 10 per group", not bad, f"{bad}") + + # Stub out the Java-facing parts of the bundled copier module. + copied = [] + copier.is_protected = lambda m: m.protected + copier.copy_messages = lambda **kwargs: copied.append(kwargs) + + free, locked = FakeMessageObject(), FakeMessageObject(protected=True) + param = FakeParam(args=[FakeArrayList([free, locked]), -100, False, True, True, 0, 0]) + intercept.before_hooked_method(param) + check( + "mixed forward: protected message copied, the rest forwarded", + len(copied) == 1 + and copied[0]["messages"] == [locked] + and copied[0]["peer"] == -100 + and copied[0]["hide_caption"] is True + and list(param.args[0]) == [free] + and not param.result_set, + f"copied={copied} args0={list(param.args[0])} result_set={param.result_set}", + ) + + copied.clear() + param = FakeParam(args=[FakeArrayList([locked]), 5, False, False, True, 0, 0]) + intercept.before_hooked_method(param) + check( + "all-protected forward: nothing left for the original sendMessage to forward", + len(copied) == 1 and list(param.args[0]) == [], + f"copied={copied} args0={list(param.args[0])}", + ) + + copied.clear() + + def broken_check(m): + raise RuntimeError("boom") + + copier.is_protected = broken_check + param = FakeParam(args=[FakeArrayList([free]), 5, False, False, True, 0, 0]) + intercept.before_hooked_method(param) + check( + "failed protection check copies instead of forwarding", + len(copied) == 1 and list(param.args[0]) == [], + f"copied={copied} args0={list(param.args[0])}", + ) + copier.is_protected = lambda m: m.protected + + copied.clear() + param = FakeParam(args=[FakeArrayList([free]), 5, False, False, True, 0, 0]) + intercept.before_hooked_method(param) + check( + "unprotected forward passes through untouched", + not copied and not param.result_set and list(param.args[0]) == [free], + ) + + plugin.set_setting("allow_forwarding", False) + param = FakeParam(args=[FakeArrayList([locked]), 5, False, False, True, 0, 0]) + intercept.before_hooked_method(param) + check("forward interception off when setting disabled", not copied and not param.result_set) + + plugin.set_setting("allow_forwarding", True) + param = FakeParam(args=[object()]) + intercept.before_hooked_method(param) + check("sendMessage(SendMessageParams) overload ignored", not copied and not param.result_set) + + def main(argv=None) -> int: argv = sys.argv[1:] if argv is None else argv if len(argv) != 1: @@ -146,12 +389,12 @@ def main(argv=None) -> int: 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__": "hello_bundle"} + ns = {"__name__": meta["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}" @@ -198,58 +441,55 @@ def main(argv=None) -> int: plugin.on_plugin_load() check( "on_plugin_load emits load log line", - any("hello_bundle loaded" in line for line in LOG_LINES), + any("unlocked_restrictions loaded" in line for line in LOG_LINES), f"log lines: {LOG_LINES}", ) - - params = SimpleParams(".hello alice smith") - result = plugin.on_send_message_hook(account=0, params=params) + hooked = {(cls, method) for cls, method, _ in HOOKED} check( - "'.hello alice smith' renders greeting", - params.message == "Hello, Alice Smith!", - f"got {params.message!r}", - ) - check( - "on_send_message_hook returns MODIFY strategy", - result.strategy == HookStrategy.MODIFY, - f"got {result.strategy}", + "on_plugin_load hooks every target method", + EXPECTED_HOOKS <= hooked, + f"missing: {EXPECTED_HOOKS - hooked}", ) - plugin.set_setting("template", "{nope}") - params2 = SimpleParams(".hello someone") - plugin.on_send_message_hook(account=0, params=params2) + switches = [row for row in plugin.create_settings() if type(row).__name__ == "Switch"] check( - "bad template produces 'Bad template:' message", - params2.message.startswith("Bad template:"), - f"got {params2.message!r}", + "settings expose all toggles, enabled by default", + {(row.key, row.default) for row in switches} + == { + ("allow_screenshots", True), + ("allow_forwarding", True), + ("protected_lock_badge", True), + }, + f"got {[(row.key, row.default) for row in switches]}", ) - # main.py imports `render`/`TemplateError` by name, so they live directly - # in the executed namespace — call render() directly to inspect the raised - # traceback (proves linecache registration for bundled, non-file modules). - render = ns.get("render") - template_error = ns.get("TemplateError") + verify_hooks(plugin, {(cls, method): hook for cls, method, hook in HOOKED}) + + # Raise inside a bundled module to prove linecache registration: the + # traceback must name the original source file and show the failing line. + screenshots = ns.get("screenshots") traceback_text = "" - if render is not None and template_error is not None: + if screenshots is not None: try: - render("{nope}", "x") - except template_error: + screenshots.StripFlagSecureHook(lambda: True).before_hooked_method(FakeParam(args=None)) + except TypeError: traceback_text = traceback.format_exc() check( - "render() traceback names src/greeting/template.py", - "src/greeting/template.py" in traceback_text, + "hook traceback names src/screenshots.py", + "src/screenshots.py" in traceback_text, f"traceback:\n{traceback_text}", ) check( "traceback includes the raising source line text", - "unsupported placeholder" in traceback_text, + "if len(args) != 2" in traceback_text, f"traceback:\n{traceback_text}", ) + bundled = {"main", "hooking", "screenshots", "forwarding", "copier", "badge", "protection"} check( "sys.modules gained no entry for bundled module names", - not ({"greeting", "textutil", "main"} & new_modules), - f"leaked: {({'greeting', 'textutil', 'main'} & new_modules)}", + not (bundled & new_modules), + f"leaked: {bundled & new_modules}", ) if _FAILURES: