121 lines
4.4 KiB
Python
121 lines
4.4 KiB
Python
"""
|
|
This plugin provides a command to get random SOCKS5 proxies.
|
|
"""
|
|
|
|
# plugins/proxy.py
|
|
|
|
import os
|
|
import logging
|
|
import random
|
|
import requests
|
|
import socket
|
|
import ssl
|
|
import time
|
|
from datetime import datetime, timedelta
|
|
import simplematrixbotlib as botlib
|
|
|
|
# Constants
|
|
SOCKS5_LIST_URL = 'https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt'
|
|
MAX_TRIES = 5
|
|
PROXY_LIST_FILENAME = 'socks5.txt'
|
|
PROXY_LIST_EXPIRATION = timedelta(hours=8)
|
|
|
|
async def test_socks5_proxy(proxy):
|
|
"""
|
|
Test a SOCKS5 proxy by attempting a connection and sending a request through it.
|
|
|
|
Args:
|
|
proxy (str): The SOCKS5 proxy address in the format 'ip:port'.
|
|
|
|
Returns:
|
|
bool: True if the proxy is working, False otherwise.
|
|
float: The latency in milliseconds if the proxy is working, None otherwise.
|
|
"""
|
|
try:
|
|
ip, port = proxy.split(':')
|
|
start_time = time.time()
|
|
with socket.create_connection((ip, int(port)), timeout=5) as client:
|
|
client.sendall(b'\x05\x01\x00')
|
|
response = client.recv(2)
|
|
if response == b'\x05\x00':
|
|
latency = int(round((time.time() - start_time) * 1000, 0))
|
|
logging.info(f"Successful connection to SOCKS5 proxy {proxy}. Latency: {latency} ms")
|
|
return True, latency
|
|
else:
|
|
logging.info(f"Failed to connect to SOCKS5 proxy {proxy}: Connection refused")
|
|
return False, None
|
|
except Exception as e:
|
|
logging.error(f"Error testing SOCKS5 proxy {proxy}: {e}")
|
|
return False, None
|
|
|
|
async def download_proxy_list():
|
|
"""
|
|
Download the SOCKS5 proxy list file if it doesn't already exist or if it's expired.
|
|
|
|
Returns:
|
|
bool: True if the proxy list is downloaded or up-to-date, False otherwise.
|
|
"""
|
|
try:
|
|
if not os.path.exists(PROXY_LIST_FILENAME) or \
|
|
datetime.now() - datetime.fromtimestamp(os.path.getctime(PROXY_LIST_FILENAME)) > PROXY_LIST_EXPIRATION:
|
|
logging.info("Downloading SOCKS5 proxy list")
|
|
response = requests.get(SOCKS5_LIST_URL, timeout=5)
|
|
with open(PROXY_LIST_FILENAME, 'w') as f:
|
|
f.write(response.text)
|
|
logging.info("Proxy list downloaded successfully")
|
|
return True
|
|
else:
|
|
logging.info("Proxy list already exists and is up-to-date")
|
|
return True
|
|
except Exception as e:
|
|
logging.error(f"Error downloading proxy list: {e}")
|
|
return False
|
|
|
|
|
|
async def handle_command(room, message, bot, prefix, config):
|
|
"""
|
|
Function to handle the !proxy command.
|
|
|
|
Args:
|
|
room (Room): The Matrix room where the command was invoked.
|
|
message (RoomMessage): The message object containing the command.
|
|
|
|
Returns:
|
|
None
|
|
"""
|
|
match = botlib.MessageMatch(room, message, bot, prefix)
|
|
if match.is_not_from_this_bot() and match.prefix() and match.command("proxy"):
|
|
logging.info("Received !proxy command")
|
|
|
|
# Download proxy list if needed
|
|
if not await download_proxy_list():
|
|
await bot.api.send_markdown_message(room.room_id, "Error downloading proxy list")
|
|
logging.error("Error downloading proxy list")
|
|
return
|
|
|
|
try:
|
|
# Read proxies from file
|
|
with open(PROXY_LIST_FILENAME, 'r') as f:
|
|
socks5_proxies = f.read().splitlines()
|
|
random.shuffle(socks5_proxies)
|
|
|
|
# Test proxies
|
|
socks5_proxy = None
|
|
for proxy in socks5_proxies[:MAX_TRIES]:
|
|
success, latency = await test_socks5_proxy(proxy)
|
|
if success:
|
|
socks5_proxy = proxy
|
|
break
|
|
|
|
# Send the first working anonymous proxy of each type to the Matrix room
|
|
if socks5_proxy:
|
|
await bot.api.send_markdown_message(room.room_id, f"✅ Anonymous SOCKS5 Proxy: **{socks5_proxy}** - Latency: **{latency} ms**")
|
|
logging.info(f"Sent SOCKS5 proxy {socks5_proxy} to the room")
|
|
else:
|
|
await bot.api.send_markdown_message(room.room_id, "❌ No working anonymous SOCKS5 proxy found")
|
|
logging.info("No working anonymous SOCKS5 proxy found")
|
|
|
|
except Exception as e:
|
|
logging.error(f"Error handling !proxy command: {e}")
|
|
await bot.api.send_markdown_message(room.room_id, "❌ Error handling !proxy command")
|