Added a python proxy usage guide and started on a mass dnsbl checker
This commit is contained in:
parent
ff4b3d9fea
commit
d25ddff0c5
97
dnsbl.py
Normal file
97
dnsbl.py
Normal file
@ -0,0 +1,97 @@
|
|||||||
|
#!/usr/bin/env python
|
||||||
|
# DNSBL - Developed by acidvegas in Python (https://git.acid.vegas/proxytools)
|
||||||
|
|
||||||
|
# NOT FINISHED - WORK IN PROGRESS
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import ipaddress
|
||||||
|
import socket
|
||||||
|
|
||||||
|
try:
|
||||||
|
import aiodns
|
||||||
|
except ImportError:
|
||||||
|
raise SystemExit('missing required library \'aiodns\' (https://pypi.org/project/aiodns/)')
|
||||||
|
|
||||||
|
DNSBL_LIST = [
|
||||||
|
'b.barracudacentral.org',
|
||||||
|
'cbl.abuseat.org',
|
||||||
|
'http.dnsbl.sorbs.net',
|
||||||
|
'misc.dnsbl.sorbs.net',
|
||||||
|
'socks.dnsbl.sorbs.net',
|
||||||
|
'web.dnsbl.sorbs.net',
|
||||||
|
'dnsbl-1.uceprotect.net',
|
||||||
|
'dnsbl-2.uceprotect.net',
|
||||||
|
'dnsbl-3.uceprotect.net',
|
||||||
|
'db.wpbl.info',
|
||||||
|
'zen.spamhaus.org',
|
||||||
|
'spam.dnsbl.sorbs.net',
|
||||||
|
'noptr.spamrats.com',
|
||||||
|
'cbl.anti-spam.org.cn',
|
||||||
|
'dnsbl.dronebl.org',
|
||||||
|
'dnsbl.inps.de',
|
||||||
|
'dnsbl.sorbs.net',
|
||||||
|
'drone.abuse.ch',
|
||||||
|
'duinv.aupads.org',
|
||||||
|
'dul.dnsbl.sorbs.net',
|
||||||
|
'dyna.spamrats.com',
|
||||||
|
'dynip.rothen.com',
|
||||||
|
'ips.backscatterer.org',
|
||||||
|
'ix.dnsbl.manitu.net',
|
||||||
|
'korea.services.net',
|
||||||
|
'orvedb.aupads.org',
|
||||||
|
'osps.dnsbl.net.au',
|
||||||
|
'osrs.dnsbl.net.au',
|
||||||
|
'owfs.dnsbl.net.au',
|
||||||
|
'pbl.spamhaus.org',
|
||||||
|
'phishing.rbl.msrbl.net',
|
||||||
|
'probes.dnsbl.net.au',
|
||||||
|
'proxy.bl.gweep.ca',
|
||||||
|
'rbl.interserver.net',
|
||||||
|
'rdts.dnsbl.net.au',
|
||||||
|
'relays.bl.gweep.ca',
|
||||||
|
'relays.nether.net',
|
||||||
|
'residential.block.transip.nl',
|
||||||
|
'ricn.dnsbl.net.au',
|
||||||
|
'rmst.dnsbl.net.au',
|
||||||
|
'smtp.dnsbl.sorbs.net',
|
||||||
|
'spam.abuse.ch',
|
||||||
|
'spam.dnsbl.anonmails.de',
|
||||||
|
'spam.rbl.msrbl.net',
|
||||||
|
'spam.spamrats.com',
|
||||||
|
'spamrbl.imp.ch',
|
||||||
|
't3direct.dnsbl.net.au',
|
||||||
|
'tor.dnsbl.sectoor.de',
|
||||||
|
'torserver.tor.dnsbl.sectoor.de',
|
||||||
|
'ubl.lashback.com',
|
||||||
|
'ubl.unsubscore.com',
|
||||||
|
'virus.rbl.jp',
|
||||||
|
'virus.rbl.msrbl.net',
|
||||||
|
'wormrbl.imp.ch',
|
||||||
|
'xbl.spamhaus.org',
|
||||||
|
'z.mailspike.net',
|
||||||
|
'zombie.dnsbl.sorbs.net',
|
||||||
|
]
|
||||||
|
|
||||||
|
async def check_dnsbl(ip, dnsbl):
|
||||||
|
reversed_ip = ipaddress.ip_address(ip).reverse_pointer
|
||||||
|
try:
|
||||||
|
resolver = aiodns.DNSResolver()
|
||||||
|
lookup = f'{reversed_ip}.{dnsbl}'
|
||||||
|
await resolver.query(lookup, 'A')
|
||||||
|
except:
|
||||||
|
return None
|
||||||
|
|
||||||
|
async def main(ip):
|
||||||
|
tasks = [check_dnsbl(ip, dnsbl) for dnsbl in DNSBL_LIST]
|
||||||
|
blacklisted_on = [res for res in await asyncio.gather(*tasks) if res]
|
||||||
|
if blacklisted_on:
|
||||||
|
print(f"{ip} is blacklisted on the following DNSBLs:")
|
||||||
|
for bl in blacklisted_on:
|
||||||
|
print(f"- {bl}")
|
||||||
|
else:
|
||||||
|
print(f"{ip} is not blacklisted on any known DNSBLs.")
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
ip_address = input("Enter the IP address to check: ")
|
||||||
|
loop = asyncio.get_event_loop()
|
||||||
|
loop.run_until_complete(main(ip_address))
|
126
pythonproxy.md
Normal file
126
pythonproxy.md
Normal file
@ -0,0 +1,126 @@
|
|||||||
|
# Proxy usage with Python
|
||||||
|
|
||||||
|
## [aiosocks](https://pypi.org/project/aiosocks/)
|
||||||
|
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
import aiosocks
|
||||||
|
|
||||||
|
async def proxy_example(proxy: str, use_ssl: bool = False):
|
||||||
|
'''Proxy can be in IP:PORT format or USER:PASS@IP:PORT format'''
|
||||||
|
|
||||||
|
auth = proxy.split('@')[0].split(':') if '@' in proxy else None
|
||||||
|
proxy_ip, proxy_port = proxy.split('@')[1].split(':') if '@' in proxy else proxy.split(':')
|
||||||
|
|
||||||
|
options = {
|
||||||
|
'proxy' : aiosocks.Socks5Addr(proxy_ip, proxy_port),
|
||||||
|
'proxy_auth' : aiosocks.Socks5Auth(*auth) if auth else None,
|
||||||
|
'dst' : (host, port),
|
||||||
|
'limit' : 1024,
|
||||||
|
'ssl' : ssl._create_unverified_context() if use_ssl else None,
|
||||||
|
'family' : 2 # 2 = IPv4 | 10 = IPv6
|
||||||
|
}
|
||||||
|
|
||||||
|
reader, writer = await asyncio.wait_for(aiosocks.open_connection(**options), 15) # 15 second timeout
|
||||||
|
|
||||||
|
while True:
|
||||||
|
data = await asyncio.wait_for(reader.readuntil(b'\r\n'), 300) # 5 minute timeout on no data received
|
||||||
|
print(data.decode().strip()) # Print the response from the server
|
||||||
|
```
|
||||||
|
|
||||||
|
## [aiohttp](https://pypi.org/project/aiohttp)
|
||||||
|
```python
|
||||||
|
import asyncio
|
||||||
|
import aiohttp
|
||||||
|
|
||||||
|
async def proxy_example(proxy: str, url: str):
|
||||||
|
'''Proxy can be in IP:PORT format or USER:PASS@IP:PORT format'''
|
||||||
|
|
||||||
|
async with aiohttp.ClientSession() as session:
|
||||||
|
async with session.get('https://google.com', proxy=f'http://{proxy}', timeout=15) as response:
|
||||||
|
if response.status == 200: # 200 = success
|
||||||
|
print(response.text()) # Print the response from the server
|
||||||
|
```
|
||||||
|
|
||||||
|
## [http.client](https://docs.python.org/3/library/http.client.html)
|
||||||
|
I really don't use this library much at all, so this is some LM generated function...
|
||||||
|
|
||||||
|
```python
|
||||||
|
import base64
|
||||||
|
import http.client
|
||||||
|
|
||||||
|
def proxy_example(proxy: str, url):
|
||||||
|
'''Proxy can be in IP:PORT format or USER:PASS@IP:PORT format'''
|
||||||
|
|
||||||
|
auth = proxy.split('@')[0].split(':') if '@' in proxy else None
|
||||||
|
proxy_host, proxy_port = proxy.split('@')[1].split(':') if '@' in proxy else proxy.split(':')
|
||||||
|
|
||||||
|
scheme, rest = url.split('://', 1)
|
||||||
|
host, path = rest.split('/', 1)
|
||||||
|
path = '/' + path
|
||||||
|
|
||||||
|
if scheme == 'https':
|
||||||
|
conn = http.client.HTTPConnection(proxy_host, proxy_port)
|
||||||
|
conn.request('CONNECT', host)
|
||||||
|
response = conn.getresponse()
|
||||||
|
if response.status != 200:
|
||||||
|
print("Failed to establish a tunnel via proxy.")
|
||||||
|
print(response.status, response.reason)
|
||||||
|
return
|
||||||
|
conn = http.client.HTTPSConnection(proxy_host, proxy_port, context=None)
|
||||||
|
conn.set_tunnel(host)
|
||||||
|
else:
|
||||||
|
conn = http.client.HTTPConnection(proxy_host, proxy_port)
|
||||||
|
path = url
|
||||||
|
|
||||||
|
headers = {}
|
||||||
|
if auth:
|
||||||
|
auth = base64.b64encode(f'{auth[0]}:{auth[1]}'.encode()).decode()
|
||||||
|
headers['Proxy-Authorization'] = f'Basic {auth}'
|
||||||
|
|
||||||
|
|
||||||
|
conn.request('GET', path, headers=headers)
|
||||||
|
response = conn.getresponse()
|
||||||
|
print(response.status, response.reason)
|
||||||
|
if response.status == 200:
|
||||||
|
data = response.read()
|
||||||
|
print(data.decode())
|
||||||
|
|
||||||
|
conn.close()
|
||||||
|
```
|
||||||
|
|
||||||
|
## [requests](https://pypi.org/project/requests/)
|
||||||
|
```python
|
||||||
|
import requests
|
||||||
|
|
||||||
|
def proxy_example(proxy: str, url: str):
|
||||||
|
'''Proxy can be in IP:PORT format or USER:PASS@IP:PORT format'''
|
||||||
|
|
||||||
|
proxy_handler = {'http': 'http://'+proxy, 'https': 'https://'+proxy}
|
||||||
|
response = requests.get(url, proxies=proxies)
|
||||||
|
print(response.text)
|
||||||
|
```
|
||||||
|
|
||||||
|
## [urllib.request](https://docs.python.org/3/library/urllib.html)
|
||||||
|
```python
|
||||||
|
import urllib.request
|
||||||
|
|
||||||
|
def proxy_example(proxy: str, url: str):
|
||||||
|
'''Proxy can be in IP:PORT format or USER:PASS@IP:PORT format'''
|
||||||
|
|
||||||
|
proxy_handler = urllib.request.ProxyHandler({'http': proxy, 'https': proxy})
|
||||||
|
opener = urllib.request.build_opener(proxy_handler)
|
||||||
|
|
||||||
|
if '@' in proxy: # Handle authentication
|
||||||
|
creds, address = proxy.split('@')
|
||||||
|
username, password = creds.split(':')
|
||||||
|
auth_header = urllib.request.HTTPBasicAuthHandler()
|
||||||
|
auth_header.add_password(realm=None, uri=proxy, user=username, passwd=password)
|
||||||
|
opener.add_handler(auth_header)
|
||||||
|
|
||||||
|
urllib.request.install_opener(opener)
|
||||||
|
|
||||||
|
response = urllib.request.urlopen(url, timeout=15)
|
||||||
|
if response.code == 200:
|
||||||
|
print(response.read().decode())
|
||||||
|
```
|
@ -1,58 +0,0 @@
|
|||||||
#!/usr/bin/env python
|
|
||||||
# Residential Proxy Usage Example - Developed by acidvegas in Python (https://git.acid.vegas/proxytools)
|
|
||||||
|
|
||||||
'''
|
|
||||||
Residential proxies are typically in a user:pass@host:port format, rotating on every request.
|
|
||||||
|
|
||||||
These example below show how to use these proxies with the aiosocks library and the requests library.
|
|
||||||
'''
|
|
||||||
|
|
||||||
import asyncio
|
|
||||||
import ssl
|
|
||||||
|
|
||||||
try:
|
|
||||||
import aiosocks
|
|
||||||
except ImportError:
|
|
||||||
raise SystemExit('missing required library \'aiosocks\' (https://pypi.org/project/aiosocks/)')
|
|
||||||
|
|
||||||
try:
|
|
||||||
import requests
|
|
||||||
except ImportError:
|
|
||||||
raise SystemExit('missing required library \'requestss\' (https://pypi.org/project/requests/)')
|
|
||||||
|
|
||||||
async def tcp_example(proxy: str, host: str, port: int, use_ssl: bool = False):
|
|
||||||
'''
|
|
||||||
Make a connection to a TCP server through a proxy.
|
|
||||||
|
|
||||||
:param proxy: the proxy to use in the format of ip:port
|
|
||||||
:param host: the host to connect to
|
|
||||||
:param port: the port to connect to
|
|
||||||
:param use_ssl: whether or not to use SSL
|
|
||||||
'''
|
|
||||||
auth = proxy.split('@')[0].split(':') if '@' in proxy else None
|
|
||||||
proxy_ip, proxy_port = proxy.split('@')[1].split(':') if '@' in proxy else proxy.split(':')
|
|
||||||
options = {
|
|
||||||
'proxy' : aiosocks.Socks5Addr(proxy_ip, proxy_port),
|
|
||||||
'proxy_auth' : aiosocks.Socks5Auth(*auth) if auth else None,
|
|
||||||
'dst' : (host, port),
|
|
||||||
'limit' : 1024,
|
|
||||||
'ssl' : ssl._create_unverified_context() if use_ssl else None,
|
|
||||||
'family' : 2
|
|
||||||
}
|
|
||||||
reader, writer = await asyncio.wait_for(aiosocks.open_connection(**options), 15) # 15 second timeout
|
|
||||||
while True:
|
|
||||||
if reader.at_eof(): # Check if the connection has been closed
|
|
||||||
break
|
|
||||||
data = await asyncio.wait_for(reader.readuntil(b'\r\n'), 300) # 5 minute timeout on no data received
|
|
||||||
line = data.decode('utf-8').strip()
|
|
||||||
print(line) # Print the data received from the server
|
|
||||||
|
|
||||||
async def http_example(proxy: str, url: str):
|
|
||||||
'''
|
|
||||||
Make a HTTP request through a proxy.
|
|
||||||
|
|
||||||
:param proxy: the proxy to use in the format of ip:port
|
|
||||||
:param url: the url to request
|
|
||||||
'''
|
|
||||||
response = requests.get(url, proxies={'http': proxy, 'https':proxy}, timeout=15) # 15 second timeout
|
|
||||||
return response.text
|
|
Loading…
Reference in New Issue
Block a user