Made cleansocks.py asyncronous to test proxies faster and added notes to floodbl.py

cleansocks.py

@@ -8,28 +8,44 @@ This script will clean a list of proxies by removing duplicates, checking for va
 '''
 
 import argparse
-import concurrent.futures
+import asyncio
 import os
 import re
 
 # Globals
+all       = list()
 good      = list()
 print_bad = True
 
-def check(proxy):
+async def check(semaphore, proxy):
+	async with semaphore:
 		ip, port = proxy.split(':')
+		options = {
+			'proxy'      : aiosocks.Socks5Addr(proxy.split(':')[0], int(proxy.split(':')[1])),
+			'proxy_auth' : None,
+			'dst'        : ('www.google.com',80),
+			'limit'      : 1024,
+			'ssl'        : None,
+			'family'     : 2
+		}
 		try:
-		sock = socks.socksocket()
+		    await asyncio.wait_for(aiosocks.open_connection(**options), 15)
-		sock.set_proxy(socks.SOCKS5, ip, int(port))
-		sock.settimeout(args.timeout)
-		sock.connect('www.google.com', 80)) # NOTE: use your own host instead in-case connections are blocked
 		except:
 			if print_bad:
 				print('\033[1;31mBAD\033[0m  \033[30m|\033[0m ' + proxy)
 		else:
 			print('\033[1;32mGOOD\033[0m \033[30m|\033[0m ' + proxy)
+			if ip not in all:
+				all.append(ip)
 				good.append(proxy)
 
+async def main(targets):
+	sema = asyncio.BoundedSemaphore(500)
+	jobs = list()
+	for target in targets:
+		jobs.append(asyncio.ensure_future(check(sema, target)))
+	await asyncio.gather(*jobs)
+
 # Main
 print('#'*56)
 print('#{0}#'.format(''.center(54)))
@@ -45,19 +61,16 @@ parser.add_argument('-t', '--threads', help='number of threads      (default: 10
 parser.add_argument('-x', '--timeout', help='socket timeout seconds (default: 15)',  default=15,  type=int)
 args = parser.parse_args()
 try:
-	import socks
+	import aiosocks
 except ImportError:
-	raise SystemExit('missing pysocks module (https://pypi.python.org/pypi/pysocks)')
+	raise SystemExit('missing pysocks module (https://pypi.org/project/aiosocks/)')
 if not os.path.isfile(args.input):
 	raise SystemExit('no such input file')
 initial = len(open(args.input).readlines())
 proxies = set([proxy for proxy in re.findall('[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+', open(args.input).read(), re.MULTILINE)])
 if not proxies:
 	raise SystemExit('no proxies found from input file')
-with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
+asyncio.run(main(proxies))
-	checks = {executor.submit(check, proxy): proxy for proxy in proxies}
-	for future in concurrent.futures.as_completed(checks):
-		checks[future]
 good.sort()
 with open(args.output, 'w') as output_file:
 	output_file.write('\n'.join(good))

floodbl.py

@@ -7,6 +7,39 @@ Test proxies against a set of Domain Name System-based Blackhole Lists (DNSBL) o
 
 '''
 
+'''
+Notes for future improvement:
+
+To query an IPv6 address, you must expand it, then reverse it into "nibble" format.
+    e.g. if the IP was 2001:db8::1, you expand it to 2001:0db8:0000:0000:0000:0000:0000:0001 and reverse it.
+    In nibble format it is 1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2 and add on the dns blacklist you require.
+
+        e.g.   1.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.0.8.b.d.0.1.0.0.2.tor.dan.me.uk
+
+    If the IP has a match, the DNS server will respond with an "A" record of 127.0.0.100.
+    It will also respond with a "TXT" record with extra information as per below:
+
+        N:<nodename>/P:<port1[,port2]>/F:<flags>
+
+    port1 is the OR (onion router) port, port2 (if specified) is the DR (directory) port.
+    Flags are defined as follows:
+
+        E     Exit
+        X     Hidden Exit
+        A     Authority
+        B     BadExit
+        C     NoEdConsensus
+        D     V2Dir
+        F     Fast
+        G     Guard
+        H     HSDir
+        N     Named
+        R     Running
+        S     Stable
+        U     Unnamed
+        V     Valid
+'''
+
 import argparse
 import concurrent.futures
 import os