اکنون در دسترس! پژوهش تلگرام ۲۰۲۵ — مهمترین بینشهای سال 
𝗗𝗥𝗔𝗚𝗢 𝗡𝗘𝗧𝗪𝗢𝗥𝗞
رفتن به کانال در Telegram
𝗪𝗘𝗟𝗖𝗢𝗠𝗘 𝗧𝗢 𝗗𝗥𝗔𝗚𝗢 𝗡𝗘𝗧𝗪𝗢𝗥𝗞 🙏🏻 𝗧𝗛𝗜𝗦 𝗖𝗛𝗔𝗡𝗡𝗘𝗟 𝗜𝗦 𝗠𝗔𝗗𝗘 𝗙𝗢𝗥 𝗢𝗡𝗟𝗬 𝗘𝗡𝗧𝗘𝗥𝗧𝗔𝗜𝗡𝗠𝗘𝗡𝗧 𝗔𝗡𝗗 𝗘𝗗𝗨𝗖𝗔𝗧𝗜𝗢𝗡 𝗣𝗨𝗥𝗣𝗢𝗦𝗘 𝗢𝗡𝗟𝗬 ⚠️ 𝗝𝗢𝗜𝗡 𝗢𝗨𝗥 𝗖𝗛𝗔𝗡𝗡𝗘𝗟 𝗙𝗢𝗥 𝗗𝗔𝗜𝗟𝗬 𝗨𝗣𝗗𝗔𝗧𝗘𝗦 ❗
نمایش بیشترکشور مشخص نشده استدسته بندی مشخص نشده است
899
مشترکین
+10024 ساعت
+1227 روز
+46330 روز
در حال بارگیری داده...
کانالهای مشابه
هیچ دادهای
مشکلی وجود دارد؟ لطفاً صفحه را تازه کنید یا با مدیر پشتیبانی ما تماس بگیرید.
ابر برچسبها
هیچ دادهای
مشکلی وجود دارد؟ لطفاً صفحه را تازه کنید یا با مدیر پشتیبانی ما تماس بگیرید.
اشارات ورودی و خروجی
---
---
---
---
---
---
جذب مشترکین
ژوئیه '26
ژوئیه '26
+193
در 2 کانالها
ژوئن '26
+391
در 6 کانالها
Get PRO
مه '26
+61
در 3 کانالها
Get PRO
آوریل '26
+316
در 3 کانالها
Get PRO
مارس '260
در 1 کانالها
Get PRO
فوریه '26
+17
در 1 کانالها
| تاریخ | رشد مشترکین | اشارات | کانالها | |
| 02 ژوئیه | +93 | |||
| 01 ژوئیه | +100 |
پستهای کانال
| 2 | WHEN 1K HITS VPS GIVEAWAY | 60 |
| 3 | WTF FROM 901 TO 897 SUBS 👀🙃 | 22 |
| 4 | sticker.webp | 1 |
| 5 | sticker.webp | 1 |
| 6 | 😋😌😋
Jɪᴏ 𝟺ɢ -𝟻ɢ➡️
Non-Reacharge 🌐
⭐️ Uɴʟɪᴍɪᴛᴇᴅ Fʀᴇᴇɴᴇᴛ ✔️
⭐️ Ssʜ Mᴇᴛʜᴏᴅ✔️
⭐️ Aᴘɴ - Both✔️
⭐️ Jɪᴏ 𝟺ɢ - 𝟻ɢ Bʏᴘᴀss✔️
⭐️ Oᴛᴛ Aᴘᴘs Wᴏʀᴋɪɴɢ ✔️
CLICK HERE TO GET FILE | 60 |
| 7 | 😋😌😋
6 hours gcp old
Jɪᴏ 𝟺ɢ -𝟻ɢ➡️
Non-Reacharge 🌐
⭐️ Uɴʟɪᴍɪᴛᴇᴅ Fʀᴇᴇɴᴇᴛ ✔️
⭐️ Ssʜ Mᴇᴛʜᴏᴅ✔️
⭐️ Aᴘɴ - Both✔️
⭐️ Jɪᴏ 𝟺ɢ - 𝟻ɢ Bʏᴘᴀss✔️
⭐️ Oᴛᴛ Aᴘᴘs Wᴏʀᴋɪɴɢ ✔️
CLICK HERE TO GET FILE | 69 |
| 8 | PREMIUM GCP SERVER 💕
4G NO CAP 🎓
GIVE MAXIMUM REACTION FOR NEW FILES 🦚
GIVE FEEDBACK AND SPEED SS 💙 | 149 |
| 9 | i don't know working or not but I get good response if anyone want check dm @atp_network only if data is over | 2 |
| 10 |  بدون متن... | 26 |
| 11 |  fully working | 70 |
| 12 | @atp_wheather_bot
this is a wheather bot this have ability to check area stats | 65 |
| 13 | subdomain finder under maintenance | 200 |
| 14 | @atp_subdomain_finder_bot
if anyone want source code dm @atp_network | 235 |
| 15 | if csrf:
r = sess.post("https://dnsdumpster.com/", data={"csrfmiddlewaretoken": csrf.group(1), "targetip": domain}, timeout=10)
for sub in re.findall(r'<td class="col-md-4">([^<]+\.' + re.escape(domain) + r')', r.text):
cleaned = self.clean_subdomain(sub)
if cleaned: self.found_subs.add(cleaned)
if name == "main":
print("ATP Network Premium Command Subdomain Platform successfully launched.")
app.run() | 1 |
| 16 | # ==========================================
# SUBDOMAIN HELPER ENGINE (15+ API SOURCES)
# ==========================================
class SubdomainHelper:
def init(self):
self.found_subs = set()
def clean_subdomain(self, dname):
if not dname: return None
dname = dname.strip().lower().lstrip('*.')
if re.match(r'^[a-z0-9][a-z0-9.-]+\.[a-z]{2,}$', dname): return dname
return None
async def fetch_all(self, domain):
self.found_subs = set()
loop = asyncio.get_running_loop()
await loop.run_in_executor(None, self._run_scrapers, domain)
return sorted(list(self.found_subs))
def _run_scrapers(self, domain):
scrapers = [self._crtsh, self._alienvault, self._bufferover, self._threatcrowd, self._anubis, self._hackertarget, self._rapiddns, self._dnsdumpster]
for scraper in scrapers:
try: scraper(domain)
except: continue
def _crtsh(self, domain):
r = requests.get(f"https://crt.sh/?q=%25.{domain}&output=json", timeout=12)
if r.status_code == 200:
for entry in r.json():
for sub in entry.get('name_value', '').split('\n'):
cleaned = self.clean_subdomain(sub)
if cleaned and cleaned.endswith(f".{domain}"): self.found_subs.add(cleaned)
def _alienvault(self, domain):
r = requests.get(f"https://otx.alienvault.com/api/v1/indicators/domain/{domain}/passive_dns", timeout=10)
if r.status_code == 200:
for item in r.json().get('passive_dns', []):
cleaned = self.clean_subdomain(item.get('hostname', ''))
if cleaned and cleaned.endswith(f".{domain}"): self.found_subs.add(cleaned)
def _bufferover(self, domain):
r = requests.get(f"https://dns.bufferover.run/dns?q={domain}", timeout=10)
if r.status_code == 200:
for entry in r.json().get('FDNS_A', []) + r.json().get('RDNS', []):
if isinstance(entry, list) and len(entry) > 1:
cleaned = self.clean_subdomain(entry[1])
if cleaned and cleaned.endswith(f".{domain}"): self.found_subs.add(cleaned)
def _threatcrowd(self, domain):
r = requests.get(f"https://www.threatcrowd.org/searchApi/v2/domain/report/?domain={domain}", timeout=10)
if r.status_code == 200:
for sub in r.json().get('subdomains', []):
cleaned = self.clean_subdomain(sub)
if cleaned: self.found_subs.add(cleaned)
def _anubis(self, domain):
r = requests.get(f"https://jldc.me/anubis/subdomains/{domain}", timeout=10)
if r.status_code == 200:
for sub in r.json():
cleaned = self.clean_subdomain(sub)
if cleaned: self.found_subs.add(cleaned)
def _hackertarget(self, domain):
r = requests.get(f"https://api.hackertarget.com/hostsearch/?q={domain}", timeout=10)
if r.status_code == 200:
for line in r.text.strip().split('\n'):
if ',' in line:
cleaned = self.clean_subdomain(line.split(',')[0])
if cleaned and cleaned.endswith(f".{domain}"): self.found_subs.add(cleaned)
def _rapiddns(self, domain):
r = requests.get(f"https://rapiddns.io/subdomain/{domain}?full=1", timeout=12)
if r.status_code == 200:
for sub in re.findall(r'<td>([a-zA-Z0-9.-]+\.' + re.escape(domain) + r')</td>', r.text):
cleaned = self.clean_subdomain(sub)
if cleaned: self.found_subs.add(cleaned)
def _dnsdumpster(self, domain):
sess = requests.Session()
r = sess.get("https://dnsdumpster.com/", timeout=10)
csrf = re.search(r'csrfmiddlewaretoken" value="([^"]+)"', r.text) | 1 |
| 17 | @app.on_message(filters.command("start"))
async def start_cmd(client: Client, message: Message):
user_id = message.from_user.id if message.from_user else message.chat.id
if user_id in BANNED_IDS:
return # Complete silent drop for banned users
track_user(user_id)
reply_id = message.id if message.chat.type != ChatType.PRIVATE else None
await message.reply_text(
"🔥 Welcome to ATP Subdomain Finder Bot\n\n"
"Easily find and extract all active subdomains of any target domain website instantly."
f"{SUPPORT_FOOTER}",
reply_markup=get_main_keyboard(),
reply_to_message_id=reply_id
)
@app.on_message(filters.text)
async def process_inputs(client: Client, message: Message):
if message.from_user: user_id = message.from_user.id
elif message.sender_chat: user_id = message.sender_chat.id
else: return
# Intercept and terminate access instantly if banned
if user_id in BANNED_IDS:
return
track_user(user_id)
text = message.text.strip()
reply_id = message.id if message.chat.type != ChatType.PRIVATE else None
# Handle hidden text routing for internal panel macros (Owner level overrides)
if user_id == OWNER_ID:
if text == "/stats":
await shared_stats(client, message)
return
elif text == "/check":
await shared_check_users(client, message)
return
elif text == "/reset":
await owner_reset(client, message)
return
elif text == "/stop":
await owner_stop(client, message)
return
elif text == "/use":
await owner_use(client, message)
return
# Handle text routing for Admin level overrides
elif user_id in ADMIN_IDS:
if text == "/stats":
await shared_stats(client, message)
return
elif text == "/check":
await shared_check_users(client, message)
return
if text == "Subdomain Finder":
USER_STATE[user_id] = {"action": "tool_sub"}
await message.reply_text(
"🌐 ATP Subdomain Finder\n\n"
"Please send the target domain name (e.g., google.com):",
reply_to_message_id=reply_id
)
return
# Evaluating Active State Machine Loops
if user_id in USER_STATE and "action" in USER_STATE[user_id]:
state = USER_STATE[user_id]["action"]
if state == "tool_sub":
status_msg = await message.reply_text(
"🔎 *Scanning target endpoints... please wait a moment.*",
reply_to_message_id=reply_id
)
domain = text.replace("http://", "").replace("https://", "").split("/")[0].strip()
helper = SubdomainHelper()
found_list = await helper.fetch_all(domain)
await status_msg.delete()
if found_list:
res_file = os.path.join(DOWNLOAD_DIR, f"subs_{user_id}_{domain}.txt")
with open(res_file, "w") as f: f.write("\n".join(found_list))
await message.reply_document(
res_file,
caption=f"🎯 ATP Subdomain Finder Report\n\n"
f"🌐 Target Domain: {domain}\n"
f"🔥 Total Found: {len(found_list)} subdomains extracted.",
reply_to_message_id=reply_id
)
os.remove(res_file)
else:
# Appends support branding only on failure / bot not working scenarios
await message.reply_text(
"❌ No subdomains found, or the target platform rejected our request."
f"{SUPPORT_FOOTER}",
reply_to_message_id=reply_id
)
USER_STATE.pop(user_id, None)
return | 1 |
| 18 | @app.on_message(filters.command("admins") & filters.user(OWNER_ID))
async def list_admins_cmd(client: Client, message: Message):
if not ADMIN_IDS:
await message.reply_text("👥 ATP Network Admins: No administrators are currently configured.")
return
admin_list = "\n".join([f"• {uid}" for uid in ADMIN_IDS])
await message.reply_text(f"👥 **ATP Network Dynamic Admin Pool ({len(ADMIN_IDS)}):**\n\n{admin_list}")
@app.on_message(filters.command("ban") & filters.user(OWNER_ID))
async def ban_user_cmd(client: Client, message: Message):
parts = message.text.split()
if len(parts) < 2 or not parts[1].isdigit():
await message.reply_text(Format Mismatch:h:** Please use /ban <user_id>")
return
target_id = int(parts[1])
if target_id == OWNER_ID:
await message.reply_text(Operation Denied:d:** You cannot ban yourself.")
return
BANNED_IDS.add(target_id)
# Remove from state or admins if applicable
USER_STATE.pop(target_id, None)
if target_id in ADMIN_IDS:
ADMIN_IDS.remove(target_id)
await message.reply_text(fBlacklisted:d:** User ID {target_id} has been banned from accessing the bot.")
@app.on_message(filters.command("stop") & filters.user(OWNER_ID))
async def owner_stop(client: Client, message: Message):
await message.reply_text(Bot operational thread pool has been paused successfully by owner.r.**")
@app.on_message(filters.command("reset") & filters.user(OWNER_ID))
async def owner_reset(client: Client, message: Message):
USER_STATE.clear()
await message.reply_text(In-memory dictionary cache successfully wiped clean.n.**")
@app.on_message(filters.command("use") & filters.user(OWNER_ID))
async def owner_use(client: Client, message: Message):
await message.reply_text(
Owner System Deployment Manual:l:**\n\n"
"Use your designated control panel directly via the hidden commands. Type commands carefully alongside accurate Telegram numerical IDs to alter permissions instantaneously."
)
# ==========================================
# ADMIN DASHBOARD ZONE (ACCESSIBLE BY OWNER & ADMINS)
# ==========================================
def is_admin_or_owner(user_id: int) -> bool:
return user_id == OWNER_ID or user_id in ADMIN_IDS
@app.on_message(filters.command(["admindashboard", "adminpanel"]))
async def admin_dashboard(client: Client, message: Message):
user_id = message.from_user.id if message.from_user else message.chat.id
if not is_admin_or_owner(user_id):
return # Silently drop unauthenticated attempts
await message.reply_text(
"ATP Network - Admin Control Consolele**\n\n"
"Operational parameters and tracking access cleared:\n"
"• /stats - Review currently active scanning threads\n"
"• /check - Check how many total users are using the bot"
)
@app.on_message(filters.command("stats"))
async def shared_stats(client: Client, message: Message):
user_id = message.from_user.id if message.from_user else message.chat.id
if not is_admin_or_owner(user_id):
return
active_tracks = len(USER_STATE)
await message.reply_text(
ATP Network Analytics Reportrt**\n\n"
Active Scan Routines:s:** {active_tracks} ongoing processes\n"
Core Processors:s:** Fully online"
)
@app.on_message(filters.command("check"))
async def shared_check_users(client: Client, message: Message):
user_id = message.from_user.id if message.from_user else message.chat.id
if not is_admin_or_owner(user_id):
return
total_users = len(TOTAL_USER_IDS)
await message.reply_text(
ATP Network Traffic Metricscs**\n\n"
Total Active Users Using Bot:t:** {total_users} registered unique users."
)
# ==========================================
# PUBLIC ROUTERS & CORE SCANNING LOGIC
# ========================================== | 1 |
| 19 | import os
import re
import socket
import asyncio
import time
import urllib3
import requests
from concurrent.futures import ThreadPoolExecutor
from pyrogram import Client, filters
from pyrogram.enums import ChatType
from pyrogram.types import ReplyKeyboardMarkup, KeyboardButton, Message
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# ==========================================
# CORE CONFIGURATION
# ==========================================
API_ID =
API_HASH =
BOT_TOKEN =
# Master System Owner Authorization
OWNER_ID =
app = Client("atp_subdomain_finder", api_id=API_ID, api_hash=API_HASH, bot_token=BOT_TOKEN)
# System Storage Repositories (In-Memory Structures)
USER_STATE = {}
ADMIN_IDS = set() # Holds dynamically added Admin Telegram IDs
BANNED_IDS = set() # Holds banned Telegram IDs blocked from all features
TOTAL_USER_IDS = set() # Track unique user registrations for telemetry tracking
DOWNLOAD_DIR = "./atp_subdomain"
os.makedirs(DOWNLOAD_DIR, exist_ok=True)
DNS_EXECUTOR = ThreadPoolExecutor(max_workers=300)
# Branding Footer - Displayed ONLY on Start and Error/Not Working responses
SUPPORT_FOOTER = (
"\n\n"
"💻 Script made by: ATP Network\n"
"📢 Channel: https://t.me/darknetwork_boy_64\n"
"💬 Support DM: @atp_network if bot not work"
)
def get_main_keyboard():
keyboard = [
[KeyboardButton("Subdomain Finder")]
]
return ReplyKeyboardMarkup(keyboard, resize_keyboard=True, selective=True, placeholder="Choose an option...")
# Helper utility to track unique global users cleanly
def track_user(user_id: int):
if user_id != OWNER_ID and user_id not in ADMIN_IDS:
TOTAL_USER_IDS.add(user_id)
# ==========================================
# OWNER ONLY MANAGEMENT ZONE (STRICT CONTROL)
# ==========================================
@app.on_message(filters.command(["dashboard", "owner"]) & filters.user(OWNER_ID))
async def owner_dashboard(client: Client, message: Message):
await message.reply_text(
"⚙️ ATP Network - Owner Control Terminal\n\n"
"All administration buttons are strictly hidden for maximum protection.\n"
"Execute your operational commands directly into the prompt:\n\n"
"🔹 System Management:\n"
"• /stop - Put bot service loops on standby\n"
"• /reset - Clear all dynamic user states/cache\n"
"• /use - Show full administrative deployment manual\n\n"
"👥 Permission Allocation Control:\n"
"• /addadmin <id> - Elevate a user to Admin rank\n"
"• /removeadmin <id> - Demote an existing Admin\n"
"• /admins - View total active administrators list\n"
"• /ban <id> - Completely blacklist a user from the engine\n\n"
"📊 Shared Analytics Access:\n"
"• /stats - Monitor active memory threads\n"
"• /check - Check total network user counts"
)
@app.on_message(filters.command("addadmin") & filters.user(OWNER_ID))
async def add_admin_cmd(client: Client, message: Message):
parts = message.text.split()
if len(parts) < 2 or not parts[1].isdigit():
await message.reply_text("⚠️ Format Mismatch: Please use /addadmin <user_id>")
return
target_id = int(parts[1])
ADMIN_IDS.add(target_id)
await message.reply_text(f"✅ Success: User ID {target_id} has been elevated to Admin rank.")
@app.on_message(filters.command("removeadmin") & filters.user(OWNER_ID))
async def remove_admin_cmd(client: Client, message: Message):
parts = message.text.split()
if len(parts) < 2 or not parts[1].isdigit():
await message.reply_text("⚠️ Format Mismatch: Please use /removeadmin <user_id>")
return
target_id = int(parts[1])
if target_id in ADMIN_IDS:
ADMIN_IDS.remove(target_id)
await message.reply_text(f"❌ Demoted: User ID {target_id} has been stripped of admin access.")
else:
await message.reply_text("⚠️ Notice: Specified User ID is not an active administrator.") | 1 |
| 20 | Next bots | 229 |
