diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a4ed7f4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,10 @@ +.idea/ +venv/ +dist/ +*23*/ +*egg-info/ + +*_ff.* +*.session +.env +.env.bk \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..ae2ba25 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,13 @@ +beautifulsoup4==4.12.2 +certifi==2023.7.22 +charset-normalizer==3.2.0 +colorama==0.4.6 +idna==3.4 +pyaes>=1.6.1 +pyasn1>=0.5.0 +python-dotenv>=1.0.0 +requests==2.31.0 +rsa>=4.9 +soupsieve>=2.5 +Telethon==1.30.3 +urllib3>=2.0.5 diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..fd80b38 --- /dev/null +++ b/setup.py @@ -0,0 +1,36 @@ +from setuptools import ( + setup, + find_packages, +) + +setup( + name="atop", + version="0.1.8-1", + author="Aaarghhh", + author_email="giacomo@udontneed.it", + packages=["atop", "atop.modules"], + package_dir={'':'src'}, + include_package_data=True, + entry_points={"console_scripts": ["a-ton-of-privacy = atop.atop:run"]}, + url="https://github.com/aaarghhh/a_TON_of_privacy", + license="MIT", + description='"A TON of Privacy" formally called ATOP ... is a tool for conducting OSINT investigations on TON NFTs.', + long_description=open("README.md").read(), + long_description_content_type="text/markdown", + install_requires=[ + "beautifulsoup4==4.12.2", + "certifi==2023.7.22", + "charset-normalizer==3.2.0", + "colorama==0.4.6", + "idna==3.4", + "pyaes>=1.6.1", + "pyasn1>=0.5.0", + "python-dotenv>=1.0.0", + "requests==2.31.0", + "rsa>=4.9", + "soupsieve>=2.5", + "Telethon==1.30.3", + "urllib3>=2.0.5" + ], + zip_safe=False, +) diff --git a/src/__init__.py b/src/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/atop/__init__.py b/src/atop/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/atop/atop.py b/src/atop/atop.py new file mode 100644 index 0000000..544efb7 --- /dev/null +++ b/src/atop/atop.py @@ -0,0 +1,635 @@ +from colorama import Fore, Style +import requests + +import re +import json +import argparse +import random +import time +from datetime import datetime +from dotenv import load_dotenv +from atop.modules.telegramhelper import TelegramHelper +from atop.modules.const import user_agent +import os + +delays = [0.2, 0.5, 0.6, 0.5, 0.1, 0.4, 1] + + +def gdelay(): + return random.choice(delays) + + +def print_banner(): + print( + """ +Welcome in the realm of.....""" + + Fore.RED + + """ + + ▄▄▄ ▄▄▄█████▓ ▒█████ ███▄ █ ▒█████ █████▒ +▒████▄ ▓ ██▒ ▓▒▒██▒ ██▒ ██ ▀█ █ ▒██▒ ██▒▓██ ▒ +▒██ ▀█▄ ▒ ▓██░ ▒░▒██░ ██▒▓██ ▀█ ██▒ ▒██░ ██▒▒████ ░ +░██▄▄▄▄██ ░ ▓██▓ ░ ▒██ ██░▓██▒ ▐▌██▒ ▒██ ██░░▓█▒ ░ + ▓█ ▓██▒ ▒██▒ ░ ░ ████▓▒░▒██░ ▓██░ ░ ████▓▒░░▒█░ + ▒▒ ▓▒█░ ▒ ░░ ░ ▒░▒░▒░ ░ ▒░ ▒ ▒ ░ ▒░▒░▒░ ▒ ░ + ▒ ▒▒ ░ ░ ░ ▒ ▒░ ░ ░░ ░ ▒░ ░ ▒ ▒░ ░ + ░ ▒ ░ ░ ░ ░ ▒ ░ ░ ░ ░ ░ ░ ▒ ░ ░ + ░ ░ ░ ░ ░ ░ ░ + + ██▓███ ██▀███ ██▓ ██▒ █▓ ▄▄▄ ▄████▄▓██ ██▓ +▓██░ ██▒▓██ ▒ ██▒▓██▒▓██░ █▒▒████▄ ▒██▀ ▀█ ▒██ ██▒ +▓██░ ██▓▒▓██ ░▄█ ▒▒██▒ ▓██ █▒░▒██ ▀█▄ ▒▓█ ▄ ▒██ ██░ +▒██▄█▓▒ ▒▒██▀▀█▄ ░██░ ▒██ █░░░██▄▄▄▄██ ▒▓▓▄ ▄██▒░ ▐██▓░ +▒██▒ ░ ░░██▓ ▒██▒░██░ ▒▀█░ ▓█ ▓██▒▒ ▓███▀ ░░ ██▒▓░ +▒▓▒░ ░ ░░ ▒▓ ░▒▓░░▓ ░ ▐░ ▒▒ ▓▒█░░ ░▒ ▒ ░ ██▒▒▒ +░▒ ░ ░▒ ░ ▒░ ▒ ░ ░ ░░ ▒ ▒▒ ░ ░ ▒ ▓██ ░▒░ +░░ ░░ ░ ▒ ░ ░░ ░ ▒ ░ ▒ ▒ ░░ + ░ ░ ░ ░ ░░ ░ ░ ░ + ░ ░ ░ ░ +v 0.1.8 """ + + Style.RESET_ALL + ) + + +class Ton_retriever: + telegram_pivot = False + silent = False + step = 10000 + offset = 0 + target = "" + tor_proxy = False + + comprehensive = False + currentua = "" + address = "" + nft_name = "" + is_scam = "" + owner_name = "" + info = None + transactions = None + nfts = None + type = "" + kind = "" + ens_detail = None + session = None + + asset_in_sale = False + + # pivoting Telegram account + api_hash = None + api_id = None + api_telephone = None + sessionstring = None + tgchecker = None + outputdir = None + + proxy = False + s_proto = "socks5" + s_proxy = "127.0.0.1" + s_port = "9050" + + @staticmethod + def sleeping_time(): + time.sleep(gdelay()) + + @staticmethod + def ipf_ens(domain, session=None): + ipfs_url = "" + request_api = f"https://{domain}.limo" + try: + if not session: + res = requests.get(request_api, timeout=5) + else: + res = session.get(request_api, timeout=5) + if res.status_code == 200: + ipfs_url = res.headers["X-Ipfs-Path"] + except Exception as exx: + pass + time.sleep(0.3) + return ipfs_url + + def get_session(self): + ## session stora i cookie + self.session = requests.session() + if self.proxy: + self.session.proxies = { + "http": f"{self.s_proto}://{self.s_proxy}:{self.s_port}", + "https": f"{self.s_proto}://{self.s_proxy}:{self.s_port}", + } + self.user_agent_retrieve(self) + self.session.headers = {"User-Agent": self.ua()} + + def __init__( + self, + _telephone_num, + _comprehensive, + _tor, + _silent, + _telegram_pivot, + outputdir=None, + telegram_api_hash=None, + telegram_api_id=None, + telegram_api_telephone=None, + sessionstring=None, + ): + if _telegram_pivot: + load_dotenv() + try: + self.api_id = os.getenv("API_ID") + self.api_hash = os.getenv("API_HASH") + self.api_telephone = os.getenv("PHONE_NUMBER") + self.sessionstring = os.getenv("SESSION_STRING") + if not self.api_id and telegram_api_id: + self.api_id = telegram_api_id + if not self.api_hash and telegram_api_hash: + self.api_hash = telegram_api_hash + if not self.api_telephone and telegram_api_telephone: + self.api_telephone = telegram_api_telephone + if not self.sessionstring and sessionstring: + self.sessionstring = sessionstring + self.outputdir = outputdir + if not ( + (self.api_telephone != "" or self.sessionstring) + and (self.api_hash != "" and self.api_telephone != "") + ): + print( + "You have to setup your sockpuppet.. you haven't compiled your env data.. " + ) + exit(0) + else: + self.telegram_pivot = True + self.tgchecker = TelegramHelper( + self.api_hash, + self.api_id, + self.api_telephone, + self.outputdir, + self.sessionstring, + ) + except Exception as exx: + pass + + self.comprehensive = _comprehensive + if not self.check_format(_telephone_num): + if not self.silent: + print("\n [!] WRONG INPUT FORMAT") + return 1 + self.stop_cycle = False + self.proxy = _tor + self.silent = _silent + self.get_session() + if not self.silent: + print(f"\n [!] START CRAWLING.... {self.kind}: {self.target} \n") + + """ + TON DNS 0:b774d95eb20543f186c06b371ab88ad704f7e256130caf96189368a7d0cb6ccf + TON NICKNAME 0:80d78a35f955a14b679faa887ff4cd5bfc0f43b4a4eea2a7e6927f3701b273c2 + TON NUMBERS 0:0e41dc1dc3c9067ed24248580e12b3359818d83dee0304fabcf80845eafafdb2 + """ + + def check_format(self, _string): + status = False + if re.match(r"\+?888[0-9\s]{0,12}", _string.strip()): + status = True + if _string[0] != "+": + _string = "+" + _string + self.target = _string.replace(" ", "") + self.kind = "NUMBER" + self.type = ( + "0e41dc1dc3c9067ed24248580e12b3359818d83dee0304fabcf80845eafafdb2" + ) + if re.match(r"[a-z0-9-_]+\.ton", _string.strip()): + status = True + self.target = _string.strip() + self.kind = "DOMAIN" + self.type = ( + "b774d95eb20543f186c06b371ab88ad704f7e256130caf96189368a7d0cb6ccf" + ) + if re.match(r"@[a-z0-9]", _string.strip()): + status = True + self.target = _string.strip() + self.kind = "NICKNAME" + self.type = ( + "80d78a35f955a14b679faa887ff4cd5bfc0f43b4a4eea2a7e6927f3701b273c2" + ) + return status + + def user_agent_retrieve(self): + try: + for browser in ["chrome", "edge", "firefox", "safari", "opera"]: + with self.session.get( + f"http://useragentstring.com/pages/useragentstring.php?name={browser}" + ) as response: + tt = response.content.decode("utf-8") + count = 10 + for ua in re.findall(r"
  • ]*>([^<]*)<\/a><\/li>", tt): + if ua not in self.user_agents: + self.user_agents.append(ua) + count -= 1 + if count == 0: + break + except Exception as exx: + self.user_agents = user_agent + + def ua(self): + return random.choice(self.user_agents) + + def print_info(self): + if not self.address: + print(f" [-] {self.kind} NOT FOUND, {self.offset} {self.kind} PROCESSED...") + return + else: + balance = "N/A" + last_date = datetime.min + print( + """ + ░▒████████████████████ TON ██████████████████████▒░ + """ + ) + + header = f" [+] Details for {self.kind.lower()}: " + self.target + if self.asset_in_sale: + header = header + " ( asset in sale! )" + print(header) + print(" ├ Owner address: ", str(self.address)) + print(" ├ Is scam: ", str(self.is_scam)) + if self.owner_name != "": + print(" ├ Owner name: ", str(self.owner_name)) + + if self.info: + if "result" in self.info.keys(): + balance = str(int(self.info["result"]["balance"]) / 1000000000) + + if self.transactions and len(self.transactions): + last_date = datetime.fromtimestamp(self.transactions[0]["utime"]) + + print(" ├ Last activity: ", last_date.strftime("%Y-%m-%d %H:%M:%S")) + print(" ├ Balance: ", str(balance)) + print(" └ ------------------------------------\n") + + processnft = False + if self.nfts: + if "data" in self.nfts.keys(): + if "nftItemsByOwner" in self.nfts["data"].keys(): + print( + " [+] ", + "NFTs found: %s" + % len(self.nfts["data"]["nftItemsByOwner"]["items"]), + ) + processnft = True + if processnft: + first = True + for nftff in self.nfts["data"]["nftItemsByOwner"]["items"]: + if not first: + print(" |") + print(" ├ Address: %s" % (nftff["address"])) + print(" | Name: %s, Kind: %s" % (nftff["name"], nftff["kind"])) + is_collection = "" + if "collection" in nftff.keys() and nftff["collection"]: + is_collection = nftff["collection"]["name"] + print(" | Collection: %s" % (is_collection)) + ## pivoting Telegram account + if ( + "tg-data" in nftff.keys() + and self.telegram_pivot + and nftff["tg-data"] + ): + if nftff["tg-data"][1] and nftff["tg-data"][1] > 0: + TelegramHelper.print_entity(nftff["tg-data"]) + + if "image" in nftff.keys() and nftff["image"]: + if ( + "originalUrl" in nftff["image"].keys() + and nftff["image"]["originalUrl"] + ): + print(" | Url: %s" % (nftff["image"]["originalUrl"])) + first = False + print(" └ ------------------------------------") + + if self.comprehensive and self.ens_detail: + print( + """ + ░▒████████████████████ ETH ██████████████████████▒░ + """ + ) + if "data" in self.ens_detail.keys(): + if "domains" in self.ens_detail["data"].keys(): + if len(self.ens_detail["data"]["domains"]) == 1: + print( + " [+] ", + f"Details for related {self.kind.lower()} ENS domain: " + + self.ens_detail["data"]["domains"][0]["name"], + ) + print( + " ├ Owner address: ", + self.ens_detail["data"]["domains"][0]["owner"]["id"], + ) + date = datetime.fromtimestamp( + int( + self.ens_detail["data"]["domains"][0][ + "registration" + ]["registrationDate"] + ) + ) + print( + " ├ Registration: ", + date.strftime("%Y-%m-%d %H:%M:%S"), + ) + date = datetime.fromtimestamp( + int( + self.ens_detail["data"]["domains"][0][ + "registration" + ]["expiryDate"] + ) + ) + print(" ├ Expiry: ", date.strftime("%Y-%m-%d %H:%M:%S")) + print(" └ ------------------------------------") + + first = True + for domain in self.ens_detail["data"]["domains"][0]["owner"]["domains"]: + if not first: + print(" |") + else: + addr = self.ens_detail["data"]["domains"][0]["owner"]["id"] + print("\n [+] ", f"Domains related to the ETH address: {addr}") + print(" ├ Address: %s" % (domain["id"])) + date = datetime.fromtimestamp(int(domain["createdAt"])) + print( + " | Name: %s, created at: %s" + % (domain["name"], date.strftime("%Y-%m-%d %H:%M:%S")) + ) + if domain["resolver"]: + print(" | Resolver: %s" % (domain["resolver"]["address"])) + current_ipfs = Ton_retriever.ipf_ens( + domain["name"], session=self.session + ) + if current_ipfs != "": + print(" | IPFS root: %s" % current_ipfs) + first = False + print(" └ ------------------------------------") + + def enrich_telegram_asset(self): + try: + for currentnft in self.nfts["data"]["nftItemsByOwner"]["items"]: + currentnft["tg-data"] = None + if "collection" in currentnft.keys() and currentnft["collection"]: + if ( + currentnft["collection"]["name"].strip() + == "Anonymous Telegram Numbers" + and self.telegram_pivot + ): + if self.tgchecker: + currentnft[ + "tg-data" + ] = self.tgchecker.check_telegram_number(currentnft["name"]) + if ( + currentnft["collection"]["name"].strip() == "Telegram Usernames" + and self.telegram_pivot + ): + if self.tgchecker: + nickname_data = self.tgchecker.check_telegram_nickname( + currentnft["name"] + ) + if nickname_data: + if nickname_data["apidetail"]: + currentnft["tg-data"] = [ + currentnft["name"], + nickname_data["apidetail"].id, + nickname_data, + ] + else: + currentnft["tg-data"] = [ + currentnft["name"], + -1, + None, + ] + else: + currentnft["tg-data"] = [currentnft["name"], -1, None] + except Exception as exx: + if not self.silent: + print(f"[-] AN ISSUE OCCURRED DURING RETRIEVING TELEGRAM INFO... {exx}") + exit(1) + + def pivot_ens(self): + ens_domain = self.target.split(".")[0] + ".eth" + request_api = "https://api.thegraph.com/subgraphs/name/ensdomains/ens" + req_body = { + "query": '{ domains(where: {name: "%s"}) { name owner { id domains { id name createdAt parent { labelName } resolver { texts address } } } registration { registrationDate expiryDate } }}' + % ens_domain, + "variables": {}, + } + try: + res = self.session.post(request_api, json=req_body).text + self.ens_detail = json.loads(res) + except Exception as exx: + if not self.silent: + print(f"[-] AN ISSUE OCCURRED DURING RETRIEVING ENS INFO... {exx}") + exit(1) + + def request_address_nft(self, addr): + request_api = "https://api.getgems.io/graphql" + req_body = { + "query": "\nquery NftItemConnection($ownerAddress: String!, $first: Int!, $after: String) {\n nftItemsByOwner(ownerAddress: $ownerAddress, first: $first, after: $after) {\n cursor\n items {\n id\n name\n address\n index\n kind\n image: content {\n type: __typename\n ... on NftContentImage {\n originalUrl\n thumb: image {\n sized(width: 480, height: 480)\n }\n }\n ... on NftContentLottie {\n preview: image {\n sized(width: 480, height: 480)\n }\n }\n ... on NftContentVideo {\n cover: preview(width: 480, height: 480)\n }\n }\n collection {\n address\n name\n isVerified\n }\n sale {\n ... on NftSaleFixPrice {\n fullPrice\n }\n }\n }\n }\n}", + "variables": {"first": 100, "ownerAddress": addr}, + } + try: + res = self.session.post(request_api, json=req_body).text + self.nfts = json.loads(res) + if self.tgchecker: + self.enrich_telegram_asset() + except Exception as exx: + if not self.silent: + print("[-] AN ISSUE OCCURRED DURING RETRIEVING NFT INFO...") + exit(1) + + def request_address_info(self, addr): + c_addr = addr.split(":")[1] + request_api = ( + "https://api.ton.cat/v2/explorer/getWalletInformation?address=0%3A" + c_addr + ) + try: + res = self.session.get(request_api).text + self.info = json.loads(res) + except Exception as exx: + if not self.silent: + print(" [-] AN ISSUE OCCURRED DURING RETRIEVING ADDRESS INFO...") + exit(1) + + def request_address_transctions(self, addr): + c_addr = addr.split(":")[1] + request_api = f"https://toncenter.com/api/index/getTransactionsByAddress?address=0%3A{c_addr}&limit=20&offset=0&include_msg_body=true" + try: + res = self.session.get(request_api).text + self.transactions = json.loads(res) + except Exception as exx: + if not self.silent: + print(" [-] AN ISSUE OCCURRED DURING RETRIEVING TRANSACTIONS INFO...") + exit(1) + + def request_info(self): + count = 0 + request_api = f"https://tonapi.io/v1/nft/searchItems?collection=0%3A{self.type}&include_on_sale=false&limit={self.step}&offset={self.offset}" + try: + res = self.session.get(request_api).text + obj = json.loads(res) + if "message" in obj.keys(): + if obj["message"] == "rate limit exceeded": + print( + f" [-] RATE LIMIT EXCEEDED... {self.offset} NUMBERS CHECKED.." + ) + exit(1) + + for element in obj["nft_items"]: + count += 1 + search_field = "" + if "name" in element["metadata"].keys(): + search_field = element["metadata"]["name"].replace(" ", "") + + if self.kind == "NICKNAME": + search_field = "@" + element["metadata"]["name"] + + elif "dns" in element.keys(): + if element["dns"]: + search_field = element["dns"] + + if search_field == self.target: + self.nft_name = search_field + _owner = element["owner"]["address"] + _is_scam = element["owner"]["is_scam"] + + # handling auction smartcontract + if ( + "sale" in element.keys() + and element["owner"]["address"] + != element["sale"]["owner"]["address"] + ): + self.asset_in_sale = True + _owner = element["sale"]["owner"]["address"] + _is_scam = element["sale"]["owner"]["is_scam"] + if "name" in element["sale"]["owner"]: + self.owner_name = element["sale"]["owner"]["name"] + elif "name" in element["owner"].keys(): + self.owner_name = element["owner"]["name"] + + self.address = _owner + self.is_scam = _is_scam + self.request_address_info(_owner) + self.request_address_transctions(_owner) + self.request_address_nft(_owner) + self.stop_cycle = True + break + + except Exception as exx: + if not self.silent: + print( + f" [-] THERE WAS SOME ISSUE DURING REQUESTING INFO ABOUT TON {self.kind} ..." + ) + exit(1) + return count + + def start_searching(self): + self.offset = 0 + current_finding = self.step + while not self.stop_cycle and current_finding == self.step: + current_finding = self.request_info() + self.offset += current_finding + time.sleep(gdelay()) + if self.comprehensive: + if self.kind == "DOMAIN": + self.pivot_ens() + + @staticmethod + def telegram_generate_session(): + TelegramHelper.generate_string_session() + + +def run(): + parser = argparse.ArgumentParser( + description="Launch me, and you'll receive a TON of privacy..." + ) + parser.add_argument( + "--target", + required=False, + help=" [?] TON number, nickname or domain to analyze ...", + ) + parser.add_argument( + "-l", + "--login", + default=False, + help=" [?] Create session string for Telegram login ...", + action="store_true", + ) + """ + if a flag comprehensive is True + a deep inspection will be done: + - domain -> pivoting on ENS domain + - nickname -> Telepathy check nickname details + - telephone -> TBA? + """ + parser.add_argument( + "-c", + "--comprehensive", + required=False, + default=False, + help=" [?] Comprehensive research which includes pivoting on ENS domains ...", + action="store_true", + ) + parser.add_argument( + "-t", + "--tor", + required=False, + default=False, + help=" [?] Use TOR as SOCK5 proxy ...", + action="store_true", + ) + parser.add_argument( + "-p", + "--phonepivot", + required=False, + default=False, + help=" [?] Pivot Telegram account ...", + action="store_true", + ) + parser.add_argument( + "-s", + "--silent", + required=False, + default=False, + help=" [?] Disable any print, useful if you don't want to print on stdout ...", + action="store_true", + ) + parser.add_argument( + "--picpath", + required=False, + help="[?] where to store profile pics ...", + ) + try: + print_banner() + args = parser.parse_args() + + if not args.target and not args.login: + parser.print_help() + exit(0) + + if args.login: + Ton_retriever.telegram_generate_session() + exit(0) + else: + ton_ret = Ton_retriever( + args.target, + args.comprehensive, + args.tor, + args.silent, + args.phonepivot, + args.picpath, + ) + ton_ret.start_searching() + if not args.silent: + ton_ret.print_info() + + except KeyboardInterrupt: + print("[-] ATOP was killed ...") + exit(1) + + +if __name__ == "__main__": + run() diff --git a/src/atop/modules/__init__.py b/src/atop/modules/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/atop/modules/const.py b/src/atop/modules/const.py new file mode 100644 index 0000000..9f76b40 --- /dev/null +++ b/src/atop/modules/const.py @@ -0,0 +1,56 @@ +user_agent = [ + # Chrome + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.113 Safari/537.36", + "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36", + "Mozilla/5.0 (Windows NT 5.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36", + "Mozilla/5.0 (Windows NT 6.2; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.90 Safari/537.36", + "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/44.0.2403.157 Safari/537.36", + "Mozilla/5.0 (Windows NT 6.3; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/60.0.3112.113 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36", + "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/57.0.2987.133 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36", + "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/55.0.2883.87 Safari/537.36", + # Firefox + "Mozilla/4.0 (compatible; MSIE 9.0; Windows NT 6.1)", + "Mozilla/5.0 (Windows NT 6.1; WOW64; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; WOW64; Trident/5.0)", + "Mozilla/5.0 (Windows NT 6.1; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (Windows NT 6.2; WOW64; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.0; Trident/5.0)", + "Mozilla/5.0 (Windows NT 6.3; WOW64; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (compatible; MSIE 9.0; Windows NT 6.1; Trident/5.0)", + "Mozilla/5.0 (Windows NT 6.1; Win64; x64; Trident/7.0; rv:11.0) like Gecko", + "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.1; WOW64; Trident/6.0)", + "Mozilla/5.0 (compatible; MSIE 10.0; Windows NT 6.1; Trident/6.0)", + "Mozilla/4.0 (compatible; MSIE 8.0; Windows NT 5.1; Trident/4.0; .NET CLR 2.0.50727; .NET CLR 3.0.4506.2152; .NET CLR 3.5.30729)" + # Safari + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.16 (KHTML, like Gecko) RockMelt/0.9.50.549 Chrome/10.0.648.205 Safari/534.16" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.18 (KHTML, like Gecko) Chrome/11.0.661.0 Safari/534.18" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.19 (KHTML, like Gecko) Chrome/11.0.661.0 Safari/534.19" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.21 (KHTML, like Gecko) Chrome/11.0.678.0 Safari/534.21" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.21 (KHTML, like Gecko) Chrome/11.0.682.0 Safari/534.21" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.3 (KHTML, like Gecko) Chrome/6.0.458.1 Safari/534.3" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.3 (KHTML, like Gecko) Chrome/6.0.461.0 Safari/534.3" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.3 (KHTML, like Gecko) Chrome/6.0.472.53 Safari/534.3" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.3 (KHTML, like Gecko) Iron/6.0.475 Safari/534" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.30 (KHTML, like Gecko) Chrome/12.0.724.100 Safari/534.30" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.6 (KHTML, like Gecko) Chrome/7.0.500.0 Safari/534.6" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Chrome/7.0.514.0 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Chrome/7.0.517.41 Safari/534.7 ChromePlus/1.5.0.0 ChromePlus/1.5.0.0" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Chrome/7.0.517.41 Safari/534.7 ChromePlus/1.5.0.0alpha1" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Flock/3.5.2.4599 Chrome/7.0.517.442 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Iron/7.0.520.0 Chrome/7.0.520.0 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Iron/7.0.520.1 Chrome/7.0.520.1 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) Iron/7.0.520.1 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) RockMelt/0.8.36.116 Chrome/7.0.517.44 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.7 (KHTML, like Gecko) RockMelt/0.8.36.128 Chrome/7.0.517.44 Safari/534.7" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/534.9 (KHTML, like Gecko) Chrome/7.0.531.0 Safari/534.9" + "Mozilla/5.0 (Windows; U; Windows NT 5.1; en-US) AppleWebKit/525.19 (KHTML, like Gecko) Iron/0.2.152.0 Safari/13657880.525", + # OTHER + "Mozilla/5.0 (Windows NT 5.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2224.3 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:53.0) Gecko/20100101 Firefox/53.0", + "Mozilla/5.0 (Windows; U; Windows NT 5.1; zh-CN; rv:1.9) Gecko/20080705 Firefox/3.0 Kapiko/3.0", + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/51.0.2704.79", +] diff --git a/src/atop/modules/telegramhelper.py b/src/atop/modules/telegramhelper.py new file mode 100644 index 0000000..59f3687 --- /dev/null +++ b/src/atop/modules/telegramhelper.py @@ -0,0 +1,339 @@ +import re +import time + +from telethon import TelegramClient, errors, events, sync +from telethon.tl.types import InputPhoneContact +from telethon import functions, types +from telethon.errors import FloodWaitError +from getpass import getpass +from bs4 import BeautifulSoup + +from telethon.sessions import StringSession +from atop.modules.const import user_agent + +import requests +import random +import os + +from telethon.tl.types import PeerChat, PeerChannel, Channel, User, Chat + + +class TelegramHelper: + _api_hash = "" + _api_id = "" + _telephone_sock = "" + _client = None + _outdir = "" + _sessionstring = None + + @staticmethod + def generate_string_session(): + api_id = input(" [!] Please enter your API ID:\n") + api_hash = input(" [!] Please enter your API Hash:\n") + phone_number = input(" [!] Please enter your phone number:\n") + # validate with regex phone number + _api_id = 0 + if not phone_number.startswith("+"): + phone_number = "+" + phone_number.replace(" ", "") + if not re.compile(r"^(\+)[0-9]{11,12}$").match(phone_number): + print(" [-] Phone number is valid.") + return -1 + if not re.compile(r"^\d+$").match(api_id): + print(" [-] App id is valid.") + return -1 + else: + _api_id = int(api_id) + with TelegramClient(StringSession(), _api_id, api_hash) as client: + print(client.session.save()) + return 0 + + @staticmethod + def parse_html_page(url, session=None): + data_web = { + "nickname": "@" + url.replace("https://t.me/", ""), + "participants": "N/A", + "image": "N/A", + "kind": "N/A", + "description": "N/A", + "name": "N/A", + } + if not session: + s = requests.Session() + else: + s = session + s.max_redirects = 10 + s.headers["User-Agent"] = random.choice(user_agent) + URL = s.get(url) + URL.encoding = "utf-8" + html_content = URL.text + soup = BeautifulSoup(html_content, "html.parser") + + try: + action = soup.find("div", {"class": ["tgme_page_additional"]}).text + if not "you can view and join" in action: + data_web["kind"] = "user" + except: + pass + try: + data_web["name"] = soup.find("div", {"class": ["tgme_page_title"]}).text + except: + data_web["name"] = "N/A" + try: + data_web["image"] = soup.find("div", {"class": ["tgme_page_photo"]}).find( + "img" + )["src"] + except: + data_web["description"] = "N/A" + try: + data_web["description"] = ( + soup.find("div", {"class": ["tgme_page_description"]}) + .getText(separator="\n") + .replace("\n", " ") + ) + except: + data_web["description"] = "N/A" + try: + if data_web["kind"] != "user": + group_participants = soup.find( + "div", {"class": ["tgme_page_extra"]} + ).text + if "member" in group_participants: + sep = "member" + stripped = group_participants.split(sep, 1)[0] + data_web["participants"] = stripped.replace(" ", "") + data_web["kind"] = "group" + else: + sep = "subscriber" + stripped = group_participants.split(sep, 1)[0] + data_web["participants"] = stripped.replace(" ", "") + data_web["kind"] = "channel" + except: + data_web["participants"] = "N/A" + return data_web + + def __init__(self, api_hash, api_id, telephone_sock, outdir, sessionstring=None): + if not api_hash or not api_id or (not telephone_sock and not sessionstring): + print( + "You have to setup your sockpuppet.. you haven't compiled your .env data.. " + ) + if not sessionstring: + self._api_hash = api_hash + self._api_id = api_id + self._outdir = outdir + self._telephone_sock = telephone_sock + else: + self._api_hash = api_hash + self._api_id = api_id + self._sessionstring = sessionstring + self.create_client() + + def create_client(self): + if self._sessionstring: + self._client = TelegramClient( + StringSession(self._sessionstring), self._api_id, self._api_hash + ) + self._client.connect() + if not self._client.is_user_authorized(): + print(f"[-] TELEGRAM ISN'T AUTHENTICATE PLS RECREATE A SESSIONSTRING...") + exit(1) + else: + self._client = TelegramClient( + self._telephone_sock, self._api_id, self._api_hash + ) + self._client.connect() + if not self._client.is_user_authorized(): + self._client.send_code_request(self._telephone_sock) + try: + self._client.sign_in( + self._telephone_sock, + input("Enter the code (sent on telegram): "), + ) + except errors.SessionPasswordNeededError: + pw = getpass( + "Two-Step Verification enabled. Please enter your account password: " + ) + self._client.sign_in(password=pw) + + def retrieve_entity(self, _target): + current_entity = None + try: + current_entity = self._client.get_entity(_target) + except Exception as exx: + try: + current_entity = self._client.get_entity(int(_target)) + except: + pass + pass + """if not current_entity: + try: + current_entity = self._client.get_entity(PeerChannel(_target)) + except Exception as exx: + pass + if not current_entity: + try: + current_entity = self._client.get_entity(PeerChat(_target)) + except Exception as exx: + pass""" + return current_entity + + @staticmethod + def create_tg_url(handle): + return "https://t.me/{}".format(handle.split("@")[1]) + + @staticmethod + def print_entity(entity_data): + if entity_data[1] > 0: + if type(entity_data[2]["apidetail"]) == Channel: + print( + " | Found Channel id: {}, title: {}, forum: {}, creation date: {}".format( + entity_data[2]["apidetail"].id, + entity_data[2]["apidetail"].title, + str(entity_data[2]["apidetail"].forum), + entity_data[2]["apidetail"].date.strftime("%d/%m/%Y, %H:%M:%S"), + ) + ) + if type(entity_data[2]["apidetail"]) == Chat: + print( + " | Found Group id: {}, title:{}, forum{}, creation date {}".format( + entity_data[2]["apidetail"].id, + entity_data[2]["apidetail"].title, + str(entity_data[2]["apidetail"].forum), + entity_data[2]["apidetail"].date.strftime("%m/%d/%Y, %H:%M:%S"), + ) + ) + if type(entity_data[2]["apidetail"]) == User: + print( + " | Found User id: {}, first name:{}, last name:{}, lang code: {}".format( + entity_data[2]["apidetail"].id, + entity_data[2]["apidetail"].first_name, + entity_data[2]["apidetail"].last_name, + entity_data[2]["apidetail"].lang_code, + ) + ) + TelegramHelper.print_web(entity_data) + else: + print(" | No Telegram account found..") + + @staticmethod + def print_web(entity_data): + if ( + entity_data[2] + and "apidetail" in entity_data[2].keys() + and "webdetail" in entity_data[2].keys() + ): + if type(entity_data[2]["apidetail"]) == Channel: + print( + " | Name: {}, Description: {}, Subscribers: {}".format( + entity_data[2]["webdetail"]["name"], + entity_data[2]["webdetail"]["description"], + entity_data[2]["webdetail"]["participants"], + ) + ) + print( + " | Profilepic: {}".format(entity_data[2]["webdetail"]["image"]) + ) + if type(entity_data[2]["apidetail"]) == Chat: + print( + " | Name: {}, Description: {}, Members: {}".format( + entity_data[2]["webdetail"]["name"], + entity_data[2]["webdetail"]["description"], + entity_data[2]["webdetail"]["participants"], + ) + ) + print( + " | Profilepic: {}".format(entity_data[2]["webdetail"]["image"]) + ) + if type(entity_data[2]["apidetail"]) == User: + print( + " | Name: {}, Description: {}".format( + entity_data[2]["webdetail"]["name"], + entity_data[2]["webdetail"]["description"], + ) + ) + print( + " | Profilepic: {}".format(entity_data[2]["webdetail"]["image"]) + ) + + def check_telegram_nickname(self, _telegra_to_check): + ent = None + web = None + try: + ent = self.retrieve_entity(_telegra_to_check) + except: + pass + if ent: + try: + web = TelegramHelper.parse_html_page( + TelegramHelper.create_tg_url(_telegra_to_check) + ) + except: + pass + return {"apidetail": ent, "webdetail": web} + + def check_telegram_number(self, _number_to_check): + try: + contact = InputPhoneContact( + client_id=0, phone=_number_to_check, first_name="", last_name="" + ) + try: + contacts = self._client( + functions.contacts.ImportContactsRequest([contact]) + ) + except FloodWaitError as e: + time.sleep(e.seconds + 0.2) + print("[!] Too many requests Waiting {}".format(e.seconds + 1)) + contacts = self._client( + functions.contacts.ImportContactsRequest([contact]) + ) + if len(contacts.to_dict()["users"]) > 0: + username = contacts.to_dict()["users"][0]["username"] + id = contacts.to_dict()["users"][0]["id"] + telegram_details = None + if id: + if self._outdir and self._outdir != "": + filename = self._client.download_profile_photo( + id, download_big=True + ) + if filename: + os.rename( + os.path.join(self._outdir, str(filename)), + str(id) + ".jpg", + ) + + ###tru retrieving user details in contcat list + if not username: + username = "N/A" + if username != "N/A": + telegram_details = self.check_telegram_nickname( + "@" + username + ) + try: + try: + self._client( + functions.contacts.DeleteContactsRequest(id=[id]) + ) + except FloodWaitError as e: + time.sleep(e.seconds + 0.2) + self._client( + functions.contacts.DeleteContactsRequest(id=[id]) + ) + except Exception as exx: + return None, -2, None + return username, id, telegram_details + else: + return None, -1, None + else: + return None, -1, None + + except IndexError as e: + print( + " | Error happened during retrieving information about this %s Ton Number " + % (_number_to_check) + ) + + except TypeError as e: + print( + f" | TypeError: {e}. --> The error might have occured due to the inability to delete the {_number_to_check} from the contact list." + ) + except: + raise diff --git a/test-requirements.txt b/test-requirements.txt new file mode 100644 index 0000000..a254f77 --- /dev/null +++ b/test-requirements.txt @@ -0,0 +1,17 @@ +attrs==22.2.0 +certifi==2022.12.7 +charset-normalizer==3.0.1 +colorama==0.4.6 +exceptiongroup==1.1.0 +idna==3.4 +iniconfig==2.0.0 +packaging==23.0 +pluggy==1.0.0 +pyaes==1.6.1 +pyasn1==0.4.8 +pytest==7.2.1 +requests==2.28.2 +rsa==4.9 +Telethon==1.26.1 +tomli==2.0.1 +urllib3==1.26.14 diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_ens.py b/tests/test_ens.py new file mode 100644 index 0000000..06ed2ee --- /dev/null +++ b/tests/test_ens.py @@ -0,0 +1,38 @@ +import pytest +from src.atop.atop import Ton_retriever +from src.atop.modules.telegramhelper import TelegramHelper + +@pytest.fixture +def domains_ens_test(): + return ["vitalik.eth", "ahahahahahjdhassjkgsdajkhsdga.eth"] + + +def test_ipf2ens_ok(domains_ens_test): + ipfs = Ton_retriever.ipf_ens(domains_ens_test[0]) + assert ipfs != "" + +def test_ipf2ens_no(domains_ens_test): + ipfs = Ton_retriever.ipf_ens(domains_ens_test[1]) + assert ipfs == "" + +def test_web_client(): + result = TelegramHelper.parse_html_page("https://t.me/aaarghhh") + assert "user" == result["kind"] + result = TelegramHelper.parse_html_page("https://t.me/testcanalebla") + assert "channel" == result["kind"] + result = TelegramHelper.parse_html_page("https://t.me/gruppotest01") + assert "group" == result["kind"] + +def test_telegram_api(): + current_parser = Ton_retriever("+888...", True, False, True, True, None, "0000", + "000000000", None, "000=") + current_parser.start_searching() + found = False + if current_parser.nfts: + if "data" in current_parser.nfts.keys(): + print(current_parser.nfts["data"]) + if "nftItemsByOwner" in current_parser.nfts["data"].keys(): + found = True + assert found == True + +