[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) <noreply@anthropic.com>

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

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: