Working copy

plugins/proxy.py

@@ -5,53 +5,69 @@ import logging
 import random
 import requests
 import socket
-import asyncio
+import ssl
 import time
+from datetime import datetime, timedelta
 import simplematrixbotlib as botlib
 
-# Function to test SOCKS4 proxies
-async def test_socks4_proxy(proxy):
-    try:
-        start_time = time.time()
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.settimeout(5)
-        s.connect((proxy.split(':')[0], int(proxy.split(':')[1])))
-        latency = round((time.time() - start_time) * 1000, 2)
-        logging.info(f"Tested SOCKS4 proxy {proxy}. Latency: {latency} ms")
-        return True, latency
-    except Exception as e:
-        logging.error(f"Error testing SOCKS4 proxy {proxy}: {e}")
-        return False, None
+# 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)
 
-# Function to test SOCKS5 proxies
 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()
-        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        s.settimeout(5)
-        s.connect((proxy.split(':')[0], int(proxy.split(':')[1])))
-        latency = round((time.time() - start_time) * 1000, 2)
-        logging.info(f"Tested SOCKS5 proxy {proxy}. Latency: {latency} ms")
-        return True, latency
+        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
 
-# Function to test HTTP proxies
-async def test_http_proxy(proxy):
-    local_ip = requests.get("https://api.ipify.org").text
+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:
-        response = requests.get("https://api.ipify.org", proxies={"http": proxy}, timeout=5).text
-        if response.strip() != local_ip.strip():
-            logging.info(f"Tested anonymous HTTP proxy {proxy}")
+        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(f"HTTP proxy {proxy} is not anonymous")
-            return False
+            logging.info("Proxy list already exists and is up-to-date")
+            return True
     except Exception as e:
-        logging.error(f"Error testing HTTP proxy {proxy}: {e}")
+        logging.error(f"Error downloading proxy list: {e}")
         return False
 
+
 async def handle_command(room, message, bot, PREFIX):
     """
     Function to handle the !proxy command.
@@ -66,38 +82,35 @@ async def handle_command(room, message, bot, PREFIX):
     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:
-            # Fetch SOCKS4 proxy
-            socks4_proxies = requests.get('https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks4.txt', timeout=5).text.splitlines()
-            random.shuffle(socks4_proxies)
-            for proxy in socks4_proxies:
-                is_working, latency = await test_socks4_proxy(proxy)
-                if is_working:
-                    await bot.api.send_text_message(room.room_id, f"SOCKS4 Proxy: {proxy} - Latency: {latency} ms")
-                    break
-
-            # Fetch SOCKS5 proxy
-            socks5_proxies = requests.get('https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/socks5.txt', timeout=5).text.splitlines()
+            # Read proxies from file
+            with open(PROXY_LIST_FILENAME, 'r') as f:
+                socks5_proxies = f.read().splitlines()
             random.shuffle(socks5_proxies)
-            for proxy in socks5_proxies:
-                is_working, latency = await test_socks5_proxy(proxy)
-                if is_working:
-                    await bot.api.send_text_message(room.room_id, f"SOCKS5 Proxy: {proxy} - Latency: {latency} ms")
+
+            # 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
 
-            # Fetch HTTP proxy
-            http_proxies = requests.get('https://raw.githubusercontent.com/TheSpeedX/SOCKS-List/master/http.txt', timeout=5).text.splitlines()
-            random.shuffle(http_proxies)
-            for proxy in http_proxies:
-                is_working = await test_http_proxy(proxy)
-                if is_working:
-                    await bot.api.send_text_message(room.room_id, f"HTTP 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")
 
-            logging.info("Sent proxies to the room")
-        except asyncio.TimeoutError:
-            await bot.api.send_text_message(room.room_id, "Failed to fetch or test proxies. The operation timed out.")
-            logging.error("Proxy fetch and test operation timed out")
         except Exception as e:
-            await bot.api.send_text_message(room.room_id, f"An error occurred: {e}")
-            logging.error(f"Error fetching or testing proxies: {e}")
+            logging.error(f"Error handling !proxy command: {e}")
+            await bot.api.send_markdown_message(room.room_id, "❌ Error handling !proxy command")