acidvegas/proxytools
acidvegas/proxytools
3
Fork 0
Code Issues Pull Requests Activity

Asyncronous IP checking

Browse Source
This commit is contained in:
Dionysus 2023-06-09 23:53:53 -04:00
parent 9dcd2cee03
commit 6aa2d5c48d
Signed by: acidvegas
GPG Key ID: EF4B922DB85DC9DE
1 changed files with 28 additions and 29 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

57
proxytools/floodbl.py
View File

@ -2,35 +2,38 @@
# FloodBL - Developed by acidvegas in Python (https://git.acid.vegas/proxytools) # FloodBL - Developed by acidvegas in Python (https://git.acid.vegas/proxytools)
''' '''
This script will test proxies against a set of Domain Name System-based Blackhole Lists (DNSBL) or Real-time Blackhole Lists (RBL)
Test proxies against a set of Domain Name System-based Blackhole Lists (DNSBL) or Real-time Blackhole Lists (RBL)
''' '''
import asyncio
import argparse import argparse
import concurrent.futures
import os import os
import re import re
import socket import socket
dnsbls = ('dnsbl.dronebl.org','rbl.efnetrbl.org','torexit.dan.me.uk') blackholes = ('dnsbl.dronebl.org','rbl.efnetrbl.org','torexit.dan.me.uk')
def dnsbl_check(proxy): async def check(sema, proxy):
global good async with sema:
bad = False ip = proxy.split(':')[0]
ip = proxy.split(':')[0] formatted_ip = '.'.join(ip.split('.')[::-1])
formatted_ip = '.'.join(ip.split('.')[::-1]) for dnsbl in blackholes:
for dnsbl in dnsbls: try:
try: socket.gethostbyname(f'{formatted_ip}.{dnsbl}')
socket.gethostbyname(f'{formatted_ip}.{dnsbl}') print('\033[1;32mGOOD\033[0m \033[30m|\033[0m ' + ip)
except socket.gaierror: good.append(proxy)
pass except socket.gaierror:
else: print('\033[1;31mBAD\033[0m \033[30m|\033[0m ' + ip.ljust(22) + f'\033[30m({dnsbl})\033[0m')
bad = True break
break
if bad: async def main(proxies, threads):
print('BAD | ' + ip) sema = asyncio.BoundedSemaphore(threads) # B O U N D E D S E M A P H O R E G A N G
else: jobs = list()
good.append(proxy) for proxy in proxies:
print('GOOD | ' + ip) jobs.append(asyncio.ensure_future(check(sema, proxy)))
await asyncio.gather(*jobs)
# Main # Main
parser = argparse.ArgumentParser(usage='%(prog)s <input> <output> [options]') parser = argparse.ArgumentParser(usage='%(prog)s <input> <output> [options]')
@ -43,14 +46,10 @@ if not os.path.isfile(args.input):
proxies = re.findall('[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+', open(args.input).read(), re.MULTILINE) proxies = re.findall('[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+:[0-9]+', open(args.input).read(), re.MULTILINE)
if not proxies: if not proxies:
raise SystemExit('no proxies found from input file') raise SystemExit('no proxies found from input file')
good = list() asyncio.run(main(proxies, args.threads))
with concurrent.futures.ThreadPoolExecutor(max_workers=args.threads) as executor:
checks = {executor.submit(dnsbl_check, proxy): proxy for proxy in proxies}
for future in concurrent.futures.as_completed(checks):
checks[future]
good.sort() good.sort()
with open(args.output, 'w') as output_file: with open(args.output, 'w') as output_file:
output_file.write('\n'.join(good)) output_file.write('\n'.join(good))
print('Total : ' + format(len(proxies), ',d')) print('\n\033[34mTotal\033[0m : ' + format(len(proxies), ',d'))
print('Good : ' + format(len(good), ',d')) print('\033[34mGood\033[0m : ' + format(len(good), ',d'))
print('Bad : ' + format(len(proxies)-len(good), ',d')) print('\033[34mBad\033[0m : ' + format(len(proxies)-len(good), ',d'))