███████╗ ██████╗ ██████╗██╗ ██╗███████╗██████╗

██╔════╝██╔═══██╗██╔════╝██║ ██╔╝██╔════╝██╔══██╗

███████╗██║ ██║██║ █████╔╝ █████╗ ██████╔╝

╚════██║██║ ██║██║ ██╔═██╗ ██╔══╝ ██╔══██╗

███████║╚██████╔╝╚██████╗██║ ██╗███████╗██║ ██║

╚══════╝ ╚═════╝ ╚═════╝╚═╝ ╚═╝╚══════╝╚═╝ ╚═╝

Check Valid Working SOCKS Proxy

def __init__(self, queue, timeout):

self.timeout = timeout

self.q = queue

threading.Thread.__init__(self)

def isSocks4(self, host, port, soc):

ipaddr = socket.inet_aton(host)

port_pack = struct.pack(">H", port)

packet4 = b"\x04\x01" + port_pack + ipaddr + b"\x00"

soc.sendall(packet4)

data = soc.recv(8)

if len(data) < 2:

# Null response

return False

if data[0] != int("0x00", 16):

# Bad data

return False

if data[1] != int("0x5A", 16):

# Server returned an error

return False

return True

def isSocks5(self, host, port, soc):

soc.sendall(b"\x05\x01\x00")

data = soc.recv(2)

if len(data) < 2:

# Null response

return False

if data[0] != int("0x05", 16):

# Not socks5

return False

if data[1] != int("0x00", 16):

# Requires authentication

return False

return True

def getSocksVersion(self, proxy):

host, port = proxy.split(":")

try:

port = int(port)

if port < 0 or port > 65536:

print(f"Invalid: {proxy}")

return 0

except Exception:

print(f"Invalid: {proxy}")

s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

s.settimeout(self.timeout)

try:

s.connect((host, port))

if self.isSocks4(host, port, s):

s.close()

return 4

elif self.isSocks5(host, port, s):

s.close()

return 5

else:

print(f"Not a SOCKS: {proxy}")

s.close()

return 0

except socket.timeout:

print(f"Timeout: {proxy}")

s.close()

return 0

except socket.error:

print(f"Connection refused: {proxy}")

s.close()

return 0

def run(self):

while True:

proxy = self.q.get()

version = self.getSocksVersion(proxy)

if version in [5, 4]:

print(f"Working: {proxy}")

socksProxies.put(proxy)

def __init__(self, queue, outputPath):

self.q = queue

self.outputPath = outputPath

threading.Thread.__init__(self)

def run(self):

while True:

toWrite = self.q.qsize()

with open(self.outputPath, "a+") as outputFile:

for _ in range(toWrite):

proxy = self.q.get()

outputFile.write(proxy + "\n")

self.q.task_done()

pattern = r"\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{2,5}"

data = ""

for source in sources:

try:

if is_url:

req = Request(

source,

headers={

"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_9_3) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/35.0.1916.47 Safari/537.36"

},

)

response = urlopen(req)

data += response.read().decode("utf-8") + "\n"

else:

with open(source) as file:

data += file.read() + "\n"

print(f"Processed: {source}")

except Exception:

print(f"Skipping, Error Occured: {source}")

print(f"Loaded {len(proxies)} proxies !!!")

for proxy in proxies:

checkQueue.put(proxy)

for _ in range(threads):

thread = ThreadChecker(checkQueue, timeout)

thread.setDaemon(True)

thread.start()

time.sleep(0.25)

wT = ThreadWriter(socksProxies, outputPath)

wT.setDaemon(True)

wT.start()

checkQueue.join()

socksProxies.join()

socker can utilize multiple sources like file, inbuilt apis, custom urls.

File Mode:

python3 socker.py -i proxys_to_be_checked.txt

URL Mode:

python3 socker.py -u proxlylist1.site -u proxlylist2.site

Auto Mode:

python3 socker.py -auto

These modes can also be used simultaneously to gather more proxies.

python3 socker.py -i proxys_to_be_checked.txt -auto

socker is not intented for malicious uses.

description=description,

epilog="Coded by SpeedX !!!",

formatter_class=argparse.RawTextHelpFormatter,