import asyncio
import socket
import subprocess
import re
from datetime import datetime
from app import BOT, Message
@BOT.add_cmd(cmd="ping")
async def enhanced_ping(bot: BOT, message: Message):
"""
CMD: PING
INFO: Enhanced ping command with detailed statistics.
FLAGS: -c [count] for number of pings (default: 4)
USAGE: .ping google.com
.ping -c 10 8.8.8.8
"""
target = message.filtered_input
if not target:
await message.reply("β No target provided!\n"
"Usage: .ping [hostname/ip]\n"
"Example: .ping google.com")
return
# Parse count flag
count = 4 # Default count
if "-c" in message.flags:
try:
count_index = message.flags.index("-c") + 1
if count_index < len(message.flags):
count = min(int(message.flags[count_index]), 20) # Max 20 pings
except (ValueError, IndexError):
pass
response = await message.reply(f"π Pinging {target}...\nSending {count} packets...")
try:
# Run ping command
if count == 1:
cmd = ['ping', '-c', '1', target]
else:
cmd = ['ping', '-c', str(count), target]
process = await asyncio.create_subprocess_exec(
*cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
error_msg = stderr.decode().strip()
await response.edit(f"β Ping failed!\n"
f"Target: {target}\n"
f"Error: {error_msg}")
return
output = stdout.decode().strip()
# Parse ping results
lines = output.split('\n')
# Extract statistics
stats_line = next((line for line in lines if 'transmitted' in line), '')
time_line = next((line for line in lines if 'min/avg/max' in line), '')
# Parse individual ping times
ping_times = []
for line in lines:
if 'time=' in line:
time_match = re.search(r'time=([0-9.]+)', line)
if time_match:
ping_times.append(float(time_match.group(1)))
# Build result
ping_text = f"π Ping Results\n\n"
ping_text += f"Target: {target}\n"
ping_text += f"Packets: {count}\n"
if stats_line:
ping_text += f"Statistics: {stats_line}\n"
if time_line:
ping_text += f"Timing: {time_line}\n"
if ping_times:
avg_time = sum(ping_times) / len(ping_times)
min_time = min(ping_times)
max_time = max(ping_times)
ping_text += f"\nπ Detailed Analysis:\n"
ping_text += f"Average: {avg_time:.2f} ms\n"
ping_text += f"Minimum: {min_time:.2f} ms\n"
ping_text += f"Maximum: {max_time:.2f} ms\n"
# Simple quality assessment
if avg_time < 50:
quality = "π’ Excellent"
elif avg_time < 100:
quality = "π‘ Good"
elif avg_time < 200:
quality = "π Fair"
else:
quality = "π΄ Poor"
ping_text += f"Quality: {quality}"
ping_text += f"\n\nβ
Ping completed!"
await response.edit(ping_text)
except Exception as e:
await response.edit(f"β Ping error:\n{str(e)}")
@BOT.add_cmd(cmd="whois")
async def whois_lookup(bot: BOT, message: Message):
"""
CMD: WHOIS
INFO: Perform WHOIS lookup for domains and IP addresses.
USAGE: .whois google.com
.whois 8.8.8.8
"""
target = message.filtered_input
if not target:
await message.reply("β No domain/IP provided!\n"
"Usage: .whois [domain/ip]\n"
"Example: .whois google.com")
return
response = await message.reply(f"π Looking up WHOIS for {target}...")
try:
# Run whois command
process = await asyncio.create_subprocess_exec(
'whois', target,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
if process.returncode != 0:
error_msg = stderr.decode().strip()
await response.edit(f"β WHOIS lookup failed!\n"
f"Target: {target}\n"
f"Error: {error_msg}")
return
output = stdout.decode().strip()
# Parse important information
lines = output.split('\n')
# Extract key information
registrar = ""
creation_date = ""
expiration_date = ""
name_servers = []
for line in lines:
line = line.strip()
if 'Registrar:' in line:
registrar = line.split(':', 1)[1].strip()
elif 'Creation Date:' in line or 'Created:' in line:
creation_date = line.split(':', 1)[1].strip()
elif 'Expiration Date:' in line or 'Expires:' in line:
expiration_date = line.split(':', 1)[1].strip()
elif 'Name Server:' in line:
ns = line.split(':', 1)[1].strip().lower()
if ns not in name_servers:
name_servers.append(ns)
# Truncate output if too long
if len(output) > 3000:
output = output[:3000] + "\n... (output truncated)"
whois_text = f"π WHOIS Lookup Results\n\n"
whois_text += f"Domain/IP: {target}\n"
if registrar:
whois_text += f"Registrar: {registrar}\n"
if creation_date:
whois_text += f"Created: {creation_date}\n"
if expiration_date:
whois_text += f"Expires: {expiration_date}\n"
if name_servers:
whois_text += f"Name Servers:\n"
for ns in name_servers[:5]: # Show max 5 name servers
whois_text += f" β’ {ns}\n"
whois_text += f"\nπ Full WHOIS Data:\n{output}"
await response.edit(whois_text)
except Exception as e:
await response.edit(f"β WHOIS error:\n{str(e)}")
@BOT.add_cmd(cmd="nslookup")
async def dns_lookup(bot: BOT, message: Message):
"""
CMD: NSLOOKUP
INFO: Perform DNS lookup for domains.
FLAGS: -type [A/AAAA/MX/NS/TXT] for specific record types
USAGE: .nslookup google.com
.nslookup -type MX gmail.com
"""
target = message.filtered_input
if not target:
await message.reply("β No domain provided!\n"
"Usage: .nslookup [domain]\n"
"Example: .nslookup google.com")
return
# Parse record type
record_type = "A" # Default
if "-type" in message.flags:
try:
type_index = message.flags.index("-type") + 1
if type_index < len(message.flags):
record_type = message.flags[type_index].upper()
except (ValueError, IndexError):
pass
response = await message.reply(f"π DNS lookup for {target} ({record_type})...")
try:
# Run nslookup command
process = await asyncio.create_subprocess_exec(
'nslookup', '-type=' + record_type, target,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
stdout, stderr = await process.communicate()
output = stdout.decode().strip()
if process.returncode != 0 or "can't find" in output.lower():
await response.edit(f"β DNS lookup failed!\n"
f"Domain: {target}\n"
f"Type: {record_type}\n"
f"Output: {output}")
return
# Parse results for cleaner display
lines = output.split('\n')
# Extract relevant information
results = []
in_answer_section = False
for line in lines:
line = line.strip()
if 'Non-authoritative answer:' in line:
in_answer_section = True
continue
elif in_answer_section and line:
if not line.startswith('Name:') and not line.startswith('Address:'):
results.append(line)
dns_text = f"π DNS Lookup Results\n\n"
dns_text += f"Domain: {target}\n"
dns_text += f"Record Type: {record_type}\n\n"
if results:
dns_text += f"π Results:\n"
for result in results[:10]: # Limit to 10 results
dns_text += f"{result}\n"
dns_text += f"\nπ Full Output:\n{output}"
await response.edit(dns_text)
except Exception as e:
await response.edit(f"β DNS lookup error:\n{str(e)}")
@BOT.add_cmd(cmd="ipinfo")
async def ip_info(bot: BOT, message: Message):
"""
CMD: IPINFO
INFO: Get information about an IP address or domain.
USAGE: .ipinfo 8.8.8.8
.ipinfo google.com
"""
target = message.filtered_input
if not target:
await message.reply("β No IP/domain provided!\n"
"Usage: .ipinfo [ip/domain]\n"
"Example: .ipinfo 8.8.8.8")
return
response = await message.reply(f"π Getting IP information for {target}...")
try:
# First, resolve domain to IP if needed
try:
ip_address = socket.gethostbyname(target)
except socket.gaierror:
# Assume it's already an IP
ip_address = target
# Validate IP address
try:
socket.inet_aton(ip_address)
except socket.error:
await response.edit(f"β Invalid IP address or domain!\n"
f"Target: {target}")
return
# Get basic IP information
info_text = f"π IP Address Information\n\n"
info_text += f"Target: {target}\n"
info_text += f"IP Address: {ip_address}\n"
# Check if it's a private IP
ip_parts = ip_address.split('.')
if len(ip_parts) == 4:
first_octet = int(ip_parts[0])
second_octet = int(ip_parts[1])
if (first_octet == 10 or
(first_octet == 172 and 16 <= second_octet <= 31) or
(first_octet == 192 and second_octet == 168) or
first_octet == 127):
info_text += f"Type: π Private/Local IP\n"
else:
info_text += f"Type: π Public IP\n"
# Try to get hostname if different from input
if target != ip_address:
try:
hostname = socket.gethostbyaddr(ip_address)[0]
info_text += f"Hostname: {hostname}\n"
except socket.herror:
pass
# Try reverse DNS lookup
try:
reverse_dns = socket.gethostbyaddr(ip_address)[0]
if reverse_dns != target:
info_text += f"Reverse DNS: {reverse_dns}\n"
except socket.herror:
info_text += f"Reverse DNS: Not available\n"
info_text += f"\nβ
IP information retrieved!"
await response.edit(info_text)
except Exception as e:
await response.edit(f"β IP info error:\n{str(e)}")
@BOT.add_cmd(cmd="traceroute")
async def trace_route(bot: BOT, message: Message):
"""
CMD: TRACEROUTE
INFO: Trace the route to a destination.
USAGE: .traceroute google.com
"""
target = message.filtered_input
if not target:
await message.reply("β No target provided!\n"
"Usage: .traceroute [hostname/ip]\n"
"Example: .traceroute google.com")
return
response = await message.reply(f"π Tracing route to {target}...\nThis may take a moment...")
try:
# Run traceroute command (use traceroute on Linux/Mac, tracert on Windows)
try:
# Try traceroute first (Linux/Mac)
process = await asyncio.create_subprocess_exec(
'traceroute', '-m', '15', target, # Max 15 hops
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
except FileNotFoundError:
# Try tracert (Windows)
process = await asyncio.create_subprocess_exec(
'tracert', '-h', '15', target, # Max 15 hops
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=60.0)
except asyncio.TimeoutError:
process.kill()
await response.edit("β° Traceroute timed out after 60 seconds!")
return
if process.returncode != 0:
error_msg = stderr.decode().strip()
await response.edit(f"β Traceroute failed!\n"
f"Target: {target}\n"
f"Error: {error_msg}")
return
output = stdout.decode().strip()
# Truncate if too long
if len(output) > 3500:
output = output[:3500] + "\n... (output truncated)"
trace_text = f"πΊοΈ Traceroute Results\n\n"
trace_text += f"Target: {target}\n"
trace_text += f"Max Hops: 15\n\n"
trace_text += f"π Route Trace:\n{output}"
trace_text += f"\nβ
Traceroute completed!"
await response.edit(trace_text)
except Exception as e:
await response.edit(f"β Traceroute error:\n{str(e)}")