v2.0 streaming redesign: concurrent scan+spray, RDP connect button, live stats fix, terminal output

ip_utils.py

@@ -0,0 +1,87 @@
+"""
+FastRDP-NG IP Utilities
+Parse IP ranges, CIDR notation, stick to basics.
+"""
+
+import ipaddress
+import re
+from typing import Generator
+
+IP_RANGE_RE = re.compile(
+    r'^(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\s*-\s*(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})$'
+)
+SINGLE_IP_RE = re.compile(r'^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$')
+
+
+def parse_line(line: str) -> Generator[str, None, None]:
+    """Parse a single line: single IP, dash range, CIDR."""
+    line = line.strip()
+    if not line or line.startswith('#') or line.startswith('//'):
+        return
+
+    # CIDR: 192.168.1.0/24
+    if '/' in line:
+        try:
+            network = ipaddress.IPv4Network(line, strict=False)
+            for host in network.hosts():
+                yield str(host)
+        except ValueError:
+            pass
+        return
+
+    # Dash range: 192.168.1.1-192.168.1.255
+    match = IP_RANGE_RE.match(line)
+    if match:
+        try:
+            start = int(ipaddress.IPv4Address(match.group(1)))
+            end = int(ipaddress.IPv4Address(match.group(2)))
+            if start > end:
+                start, end = end, start
+            # Cap at /16 to prevent memory bombs
+            if end - start > 65536:
+                end = start + 65536
+            for ip_int in range(start, end + 1):
+                yield str(ipaddress.IPv4Address(ip_int))
+        except ValueError:
+            pass
+        return
+
+    # Single IP
+    if SINGLE_IP_RE.match(line):
+        yield line
+
+
+def parse_ranges_file(filepath: str) -> Generator[str, None, None]:
+    """Parse ranges.txt, yield individual IPs lazily."""
+    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
+        for line in f:
+            yield from parse_line(line)
+
+
+def count_ips(filepath: str) -> int:
+    """Quick count of total IPs for progress tracking."""
+    total = 0
+    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith('#') or line.startswith('//'):
+                continue
+            if '/' in line:
+                try:
+                    n = ipaddress.IPv4Network(line, strict=False)
+                    total += max(0, n.num_addresses - 2)
+                except ValueError:
+                    total += 1
+            elif IP_RANGE_RE.match(line):
+                m = IP_RANGE_RE.match(line)
+                if m is None:
+                    total += 1
+                    continue
+                s = int(ipaddress.IPv4Address(m.group(1)))
+                e = int(ipaddress.IPv4Address(m.group(2)))
+                if s > e:
+                    s, e = e, s
+                total += min(e - s + 1, 65536)
+            elif SINGLE_IP_RE.match(line):
+                total += 1
+    return total