Skip to content

Instantly share code, notes, and snippets.

@d-Rickyy-b
Last active May 10, 2023 22:33
Show Gist options
  • Save d-Rickyy-b/405c2341d762fa87e73edd8f584830e6 to your computer and use it in GitHub Desktop.
Save d-Rickyy-b/405c2341d762fa87e73edd8f584830e6 to your computer and use it in GitHub Desktop.
Small python script to extract (forward) messages from a bot using telegram as phishing backend / https://blog.rico-j.de/telegram-phishing-backend
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# Script to forward the content of a chat between a Telegram Bot and a user/chat to another user/chat
# Make sure to send /start to the bot on Telegram beforehand, otherwise Telegram won't allow the bot to forward messages to you!
import argparse
import json
import logging
import requests
import signal
import time
logging.basicConfig(format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO)
file = None
proxy_dict = {}
def get_bot_info(token):
"""Log information about the bot's username, first_name and id"""
get_me_url = f"https://api.telegram.org/bot{token}/getMe"
resp = requests.get(get_me_url, proxies=proxy_dict)
if resp.status_code == 200:
res = resp.json().get("result")
first_name, username, bot_id = res.get("first_name"), res.get("username"), res.get('id')
logging.info(f"Bot Info: Name: '{first_name}', Username: '@{username}', ID: '{bot_id}'")
def get_chat_info(token, from_chat):
"""Log details for a specific chat"""
get_chat_url = f"https://api.telegram.org/bot{token}/getChat?chat_id={from_chat}"
resp = requests.get(get_chat_url, proxies=proxy_dict)
if resp.status_code == 200:
res = resp.json().get("result")
first_name, username, bot_id, title = res.get("first_name"), res.get("username"), res.get('id'), res.get('title')
logging.info(f"Chat Info: Name: '{first_name}', Username: '@{username}', ID: '{bot_id}', Title: '{title}'")
else:
logging.error(f"Error while retrieving chat info - {resp.status_code}!")
logging.error(resp.content)
# Negative chat IDs = groups / channels
if from_chat.startswith("-"):
get_chat_membercount_url = f"https://api.telegram.org/bot{token}/getChatMemberCount?chat_id={from_chat}"
resp = requests.get(get_chat_membercount_url, proxies=proxy_dict)
if resp.status_code == 200:
res = resp.json().get("result")
logging.info(f"Chat Member Count: {res}")
else:
logging.error("Error while retrieving chat member count!")
get_chat_admins_url = f"https://api.telegram.org/bot{token}/getChatAdministrators?chat_id={from_chat}"
resp = requests.get(get_chat_admins_url, proxies=proxy_dict)
if resp.status_code == 200:
res = resp.json().get("result")
logging.info(f"Chat Admin Info: {res}")
else:
logging.error("Error while retrieving chat admin info!")
def check_latest_message_id(token, from_chat):
"""
Checks the latest message_id of 'from_chat' by sending a message to the chat.
This is dangerous, because it could be detected by the chat owners.
"""
test_message = "This bot is currently experiencing issues. Please check @BotFather for further details!"
send_message_url = f"https://api.telegram.org/bot{token}/sendMessage?chat_id={from_chat}&text={test_message}"
delete_message_url = f"https://api.telegram.org/bot{token}/deleteMessage?chat_id={from_chat}&message_id="
# TODO if group chat: check if bot got permissions to delete the message
send_message_resp = requests.get(send_message_url, proxies=proxy_dict)
if send_message_resp.status_code == 200:
latest_message_id = send_message_resp.json().get("result", {}).get("message_id")
logging.info(f"Latest message ID: {latest_message_id}")
del_message_resp = requests.get(f"{delete_message_url}{latest_message_id}", proxies=proxy_dict)
if del_message_resp.status_code != 200:
logging.warning("Couldn't delete sent message!")
logging.warning(f"{del_message_resp.status_code}, {del_message_resp.content}")
def forward_messages(token, from_chat, to_chat, offset=1, max_consecutive_errors=10):
"""Forwards all the messages from 'from_chat' starting at 'offset' to 'to_chat'"""
url = f"https://api.telegram.org/bot{token}/forwardMessage?chat_id={to_chat}&from_chat_id={from_chat}&message_id="
bot_id = token.split(":")[0]
error_counter = 0
while True:
# Query Telegram to forward the message with the given ID to the receiver
resp = requests.get(f"{url}{offset}", proxies=proxy_dict)
if resp.status_code == 200:
error_counter = 0
store_to_file(json.loads(resp.content), bot_id, offset)
elif resp.status_code == 429:
# Wait and try again with the same message
retry_after = resp.json().get("parameters", {}).get("retry_after", 1)
logging.warning(f"Flood wait at offset {offset}! Sleeping {retry_after} seconds!")
time.sleep(retry_after)
offset -= 1
elif resp.status_code == 400 and resp.json().get("description") == "Bad Request: chat not found":
logging.error("Please check your destination chat_id! The bot can't find a chat with that ID.")
logging.error("Did you send '/start' to the bot already?")
exit(1)
elif resp.status_code == 400 and resp.json().get("description") == "Bad Request: the message can\'t be forwarded":
logging.warning(f"Received status code: {resp.status_code} at offset {offset}! Something weird happened - the message can't be forwarded")
else:
desc = resp.json().get("description")
logging.warning(f"Received status code: {resp.status_code} at offset {offset}! {desc}")
# If the Telegram API responds with a status code != 200, the message doesn't exist
# That could mean that it was deleted, or it's the last message in that chat!
# If there are 10 consecutive errors we stop the data collection!
error_counter += 1
if error_counter >= max_consecutive_errors:
logging.error(f"{max_consecutive_errors} consecutive errors! Current offset: {offset}! Exiting!")
exit(1)
# Telegram allows for 1 msg/s per chat - so we need this sleep in order to not get http 429 errors
time.sleep(1)
offset += 1
# Regularly display the current offset
if offset % 100 == 0:
logging.info(f"Current offset: {offset}")
def get_updates(token):
"""Get updates from the bot to identify users"""
get_updates_url = f"https://api.telegram.org/bot{token}/getUpdates"
resp = requests.get(get_updates_url, proxies=proxy_dict)
if resp.status_code == 200:
res = resp.json().get("result")
logging.info(f"Updates: {res}")
else:
logging.error(f"Error while retrieving updates - {resp.status_code}!")
logging.error(resp.content)
def store_to_file(content, bot_id, offset):
"""Store some data to a file"""
global file
if not file:
file = open(f"{bot_id}.json", "a", encoding="utf-8", buffering=1)
message = {"offset": offset, "message": content}
file.write(json.dumps(message) + "\n")
file.flush()
def signal_handler(sig, frame):
"""Signal handler for processing CTRL+C"""
logging.warning("Received SIGINT. Stopping!")
exit(0)
if __name__ == "__main__":
# Argument setup
parser = argparse.ArgumentParser(description='Tool for extracting phishing content from Telegram bots.')
parser.add_argument("-t", "--token", type=str, required=True, help="The bot token of the phishing bot")
parser.add_argument("-s", "--source-chat", type=str, required=True, help="The chat to which the bot sent the phishing from the website")
parser.add_argument("-d", "--dest-chat", type=str, required=True, help="The chat where to forward the found messages to")
parser.add_argument("-o", "--offset", type=int, required=False, default=1, help="Offset for message_id, skips <offset> number of messages")
parser.add_argument("-i", "--info", required=False, action=argparse.BooleanOptionalAction, help="Only print bot and chat info, then exit")
parser.add_argument("-e", "--max-errors", type=int, required=False, default=10, help="Maximum consecutive errors")
parser.add_argument("-l", "--latest-message", required=False, action=argparse.BooleanOptionalAction, help="Tries to find the latest message_id of a chat by sending a new message. WARNING! If the message couldn't get deleted, the chat members can see the message sent by the bot!")
parser.add_argument("-p", "--proxy", type=str, required=False, help="Proxy address in the format of http://hostname:port")
# parser.add_argument("-m", "--monitor", type=bool, required=False, action=argparse.BooleanOptionalAction, help="If set to monitor mode, the script will not break, but wait for new messages to arrive")
args = parser.parse_args()
if args.proxy:
proxy_dict = {"http": args.proxy, "https": args.proxy}
if args.info:
# Receive metadata for chat
get_bot_info(args.token)
get_chat_info(args.token, args.source_chat)
# If the user specified to only get bot info, exit here
exit()
if args.latest_message:
logging.warning("Sending new message to chat to identify the latest message_id")
check_latest_message_id(args.token, args.source_chat)
signal.signal(signal.SIGINT, signal_handler)
forward_messages(args.token, args.source_chat, args.dest_chat, args.offset, args.max_errors)
@d-Rickyy-b
Copy link
Author

This code is licensed MIT!

@d-Rickyy-b
Copy link
Author

Today I added CLI argument parsing so that you don't need to manipulate the code itself in order to run the script.

Example usage: python phish_extractor.py -t "1234567890:ABCDEFGHI-1QWERTYUIOPASDF" -s "1234118078" -d "813451231"

usage: phish_extractor.py [-h] -t TOKEN -s SOURCE_CHAT -d DEST_CHAT [-o OFFSET] [-i | --info | --no-info]

Tool for extracting phishing content from Telegram bots.

optional arguments:
  -h, --help            show this help message and exit
  -t TOKEN, --token TOKEN
                        The bot token of the phishing bot
  -s SOURCE_CHAT, --source-chat SOURCE_CHAT
                        The chat to which the bot sent the phishing from the website
  -d DEST_CHAT, --dest-chat DEST_CHAT
                        The chat where to forward the found messages to
  -o OFFSET, --offset OFFSET
                        Offset for message_id, skips <offset> number of messages
  -i, --info, --no-info
                        Only print bot and chat info, then exit

@d-Rickyy-b
Copy link
Author

d-Rickyy-b commented Oct 19, 2021

Today I added:

  • -e, --max-errors to specify the amount of consecutive errors that can happen one after another before stopping the script
  • -l, --latest-message to find the latest message in a chat. ⚠️ WARNING this is an experimental feature that sends a message to the specified chat in order to find the latest message_id.
  • A Signal handler that catches CTRL+C to stop the script gracefully and not with an exception.
  • Store forwarded messages into own file named after <bot_id>.json
  • Store offset in output json
  • Proxy support via -p, --proxy - for now it's the same proxy for http/https. Might change in another update when I got more time.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment