#!/usr/bin/env python3 """Empirical probe: what exactly is visible via Bot API getUpdates vs MTProto history. Answers the question raised during the P0 pilot (docs/log-source-discovery.md): Bitsight collected 5M stealer logs via bot tokens — but getUpdates returns only INCOMING updates, while stealers exfiltrate BY SENDING AS the bot. What gives? Test matrix (all VERIFIED = executed + observed): A. bot -> user private chat (sendMessage): in getUpdates? B. user -> bot private message: in getUpdates? C. bot -> group post: in getUpdates? D. user -> group post (privacy default): in getUpdates? E. group history via MTProto user client: are the bot's own posts visible? F. exportChatInviteLink with bot token (bot promoted to admin in basic group) G. getChat(user) — operator profiling surface Creates a throwaway bot via @BotFather + a throwaway basic group, using the project MTProto session (.env TG_* via tg_session_load.build_session_file). Usage: .venv/bin/python scripts/tg_botapi_probe.py """ import asyncio import json import os import random import re import string import sys import time import urllib.request from pathlib import Path sys.path.insert(0, str(Path(__file__).parent)) from tg_session_load import build_session_file, load_dotenv, ENV_FILE, SESSION_DIR BOTFATHER = "BotFather" TOKEN_RE = re.compile(r"(? dict: import urllib.parse url = f"https://api.telegram.org/bot{token}/{method}" if params: url += "?" + urllib.parse.urlencode(params) with urllib.request.urlopen(url, timeout=30) as r: return json.loads(r.read()) def poll_updates(token: str, prefix: str, offset: int = 0, tries: int = 8, pause: int = 3) -> tuple[bool, list, int]: """Poll getUpdates until a message starting with prefix appears.""" ups: list = [] for _ in range(tries): ups = bot_api(token, "getUpdates", offset=offset).get("result", []) if any((u.get("message") or u.get("channel_post") or {}) .get("text", "").startswith(prefix) for u in ups): return True, ups, offset if ups: offset = max(u["update_id"] for u in ups) + 1 time.sleep(pause) return False, ups, offset async def wait_botfather_reply(app, since_id: int, timeout: int = 30) -> str: for _ in range(timeout): async for msg in app.get_chat_history(BOTFATHER, limit=3): if msg.id > since_id and msg.text and not msg.outgoing: return msg.text await asyncio.sleep(1) raise TimeoutError("BotFather did not reply") async def last_msg_id(app) -> int: async for msg in app.get_chat_history(BOTFATHER, limit=1): return msg.id return 0 async def delete_bot(app, username: str): mid = await last_msg_id(app) await app.send_message(BOTFATHER, "/deletebot") await wait_botfather_reply(app, mid) mid = await last_msg_id(app) await app.send_message(BOTFATHER, f"@{username}") # BotFather confirmation prompt: "Send 'Yes, I am totally sure.' to confirm" conf = None for _ in range(20): async for msg in app.get_chat_history(BOTFATHER, limit=5): if msg.id > mid and msg.text and not msg.outgoing: m = re.search(r"Send '([^']+)' to confirm", msg.text) if m: conf = m.group(1) elif "exactly like this:" in msg.text: conf = (msg.text.split("exactly like this:")[-1] .strip().splitlines()[0].strip()) if conf: break await asyncio.sleep(1) if not conf: return "no confirmation prompt (already deleted?)" mid = await last_msg_id(app) await app.send_message(BOTFATHER, conf) return await wait_botfather_reply(app, mid) async def main(): load_dotenv(str(ENV_FILE)) api_id = int(os.environ["TG_API_ID"]) api_hash = os.environ["TG_API_HASH"] auth_key_hex = os.environ["TG_AUTH_KEY"].strip() dc_id = int(os.environ["TG_DC_ID"]) user_id = int(os.environ["TG_USER_ID"]) build_session_file(api_id, auth_key_hex, dc_id, user_id) from pyrogram import Client from pyrogram.raw import functions, types as raw_types app = Client(name="breach_session", api_id=api_id, api_hash=api_hash, workdir=str(SESSION_DIR)) await app.start() me = await app.get_me() print(f"[*] user: uid={me.id}") # cleanup orphans from previous crashed runs for orphan in ("probe_ta6nk1_bot", "probe_sdzh2n_bot", "probe_d2ywi6_bot"): try: r = await delete_bot(app, orphan) print(f" orphan @{orphan}: {r.splitlines()[0][:60]}") except Exception as e: print(f" orphan @{orphan}: cleanup skipped ({type(e).__name__})") suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=6)) bot_name = f"Probe {suffix}" bot_username = f"probe_{suffix}_bot" # --- create bot via BotFather --- print(f"[*] creating bot @{bot_username} via BotFather...") mid = await last_msg_id(app) await app.send_message(BOTFATHER, "/newbot") await wait_botfather_reply(app, mid) mid = await last_msg_id(app) await app.send_message(BOTFATHER, bot_name) await wait_botfather_reply(app, mid) mid = await last_msg_id(app) await app.send_message(BOTFATHER, bot_username) reply = await wait_botfather_reply(app, mid) m = TOKEN_RE.search(reply) if not m: print(f" [FAIL] no token in reply: {reply[:200]}") await app.stop() return token = m.group() print(f" [+] token: {token}") results: dict = {} # --- B (first): user -> bot private message (also unlocks bot->user DM) --- await app.send_message(bot_username, "probe-B user->bot") ok, ups, offset = poll_updates(token, "probe-B") results["B user->bot private in getUpdates"] = ok if not ok: print(f" [debug] B: updates seen: {json.dumps(ups)[:400]}") # --- A: bot -> user private chat (allowed only after user /start) --- try: bot_api(token, "sendMessage", chat_id=me.id, text="probe-A bot->user") ok, ups, offset = poll_updates(token, "probe-A", offset) results["A bot->user private (sendMessage) in getUpdates"] = ok if not ok: print(f" [debug] A: updates seen: {json.dumps(ups)[:400]}") except Exception as e: results["A bot->user private (sendMessage)"] = f"send failed: {e}" # --- C: create basic group with bot, bot posts --- group = await app.create_group(f"probe-{suffix}", bot_username) gid = group.id time.sleep(2) bot_api(token, "sendMessage", chat_id=gid, text="probe-C bot->group") ok, ups, offset = poll_updates(token, "probe-C", offset) results["C bot->group post in getUpdates"] = ok if not ok: print(f" [debug] C: updates seen: {json.dumps(ups)[:400]}") # --- D: user posts in group (bot privacy default ON) --- await app.send_message(gid, "probe-D user->group") ok, ups, offset = poll_updates(token, "probe-D", offset) results["D user->group post in getUpdates (privacy default ON)"] = ok if not ok: print(f" [debug] D: updates seen: {json.dumps(ups)[:400]}") # --- E: full group history via MTProto user client --- texts = [] async for msg in app.get_chat_history(gid, limit=20): if msg.text: texts.append(msg.text) results["E bot's own group post visible in MTProto history"] = any( t.startswith("probe-C") for t in texts) # --- F: promote bot to admin (raw, basic group) + exportChatInviteLink --- try: r = await app.invoke(functions.contacts.ResolveUsername(username=bot_username)) bot_user = r.users[0] await app.invoke(functions.messages.EditChatAdmin( chat_id=-gid, # basic group id is negative in raw API user_id=raw_types.InputUser(user_id=bot_user.id, access_hash=bot_user.access_hash), is_admin=True)) time.sleep(2) link = bot_api(token, "exportChatInviteLink", chat_id=gid) results["F exportChatInviteLink by bot token (bot=admin)"] = ( link.get("result") if link.get("ok") else f"api error: {link}") except Exception as e: results["F exportChatInviteLink by bot token (bot=admin)"] = f"error: {e}" # --- G: getChat on private-chat counterpart (operator profiling) --- try: ch = bot_api(token, "getChat", chat_id=me.id) r = ch.get("result", {}) results["G getChat(user) reveals"] = ( f"first_name={r.get('first_name')!r} username={r.get('username')!r}") except Exception as e: results["G getChat(user) reveals"] = f"error: {e}" print("\n=== PROBE RESULTS (executed + observed) ===") for k, v in results.items(): print(f" {k}: {v}") # cleanup: delete bot try: reply = await delete_bot(app, bot_username) print(f"[*] bot deleted: {reply.splitlines()[0][:60]}") except Exception as e: print(f"[*] bot deletion failed: {e}") await app.stop() if __name__ == "__main__": asyncio.run(main())