70 lines
2.4 KiB
Python
70 lines
2.4 KiB
Python
#!/usr/bin/env python
|
|
# CleanSocks - Developed by acidvegas in Python (https://git.acid.vegas/proxytools)
|
|
|
|
'''
|
|
Requirements:
|
|
PySocks (https://pypi.python.org/pypi/pysocks)
|
|
|
|
This script will clean a list of proxies by removing duplicates, checking for valid formats (IP:PORT), and testing if the proxies are working
|
|
'''
|
|
|
|
import argparse
|
|
import concurrent.futures
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
sys.dont_write_bytecode = True
|
|
|
|
def is_proxy(proxy):
|
|
return re.match('^(([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5])\.){3}([0-9]|[1-9][0-9]|1[0-9]{2}|2[0-4][0-9]|25[0-5]):(?:6553[0-5]|655[0-2][0-9]|65[0-4][0-9]{2}|6[0-4][0-9]{3}|[1-5][0-9]{4}|[1-9][0-9]{1,3}|[0-9])$', proxy)
|
|
|
|
def test_proxy(proxy):
|
|
global good
|
|
ip, port = proxy.split(':')
|
|
try:
|
|
sock = socks.socksocket()
|
|
sock.set_proxy(socks.SOCKS5, ip, int(port))
|
|
sock.settimeout(args.timeout)
|
|
sock.connect(('www.google.com', 80))
|
|
except:
|
|
print('BAD | ' + proxy)
|
|
else:
|
|
print('GOOD | ' + proxy)
|
|
good.append(proxy)
|
|
finally:
|
|
sock.close()
|
|
|
|
parser = argparse.ArgumentParser(usage='%(prog)s <input> <output> [options]')
|
|
parser.add_argument('input', help='file to scan')
|
|
parser.add_argument('output', help='file to output')
|
|
parser.add_argument('-t', '--threads', help='number of threads (default: 100)', default=100, type=int)
|
|
parser.add_argument('-x', '--timeout', help='socket timeout seconds (default: 15)', default=15, type=int)
|
|
args = parser.parse_args()
|
|
try:
|
|
import socks
|
|
except ImportError:
|
|
raise SystemExit('missing pysocks module (https://pypi.python.org/pypi/pysocks)')
|
|
if not os.path.isfile(args.input):
|
|
raise SystemExit('no such input file')
|
|
proxies = set([line.strip() for line in open(args.input).readlines() if is_proxy(line)])
|
|
if not proxies:
|
|
raise SystemExit('no proxies found from input file')
|
|
deduped, ips = list(), list()
|
|
for proxy in proxies:
|
|
ip = proxy.split(':')[0]
|
|
if ip not in ips:
|
|
ips.append(ip)
|
|
deduped.append(proxy)
|
|
deduped.sort()
|
|
good = list()
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
|
|
checks = {executor.submit(test_proxy, proxy): proxy for proxy in deduped}
|
|
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))
|
|
print('Total : ' + format(len(proxies), ',d'))
|
|
print('Good : ' + format(len(good), ',d'))
|
|
print('Bad : ' + format(len(proxies)-len(good), ',d')) |