234 lines
8.5 KiB
Python
Executable File
234 lines
8.5 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
find_anon_proxies.py — Поиск анонимных прокси-серверов
|
|
|
|
Источники:
|
|
- https://spys.me/proxy.txt (текстовый формат)
|
|
- https://www.sslproxies.org/ (HTML таблица)
|
|
- http://netips.ch/proxylist/anonymous.html (HTML таблица)
|
|
- https://openproxy.space/list (JSON)
|
|
|
|
Проверка:
|
|
- HTTP-прокси (GET к google.com через curl --proxy)
|
|
- SOCKS5 прокси (через curl --socks5)
|
|
- Анонимность (проверка X-Forwarded-For заголовка)
|
|
- Скорость ответа
|
|
|
|
Вывод:
|
|
- Сохраняет в /root/.openclaw/workspace/scripts/proxy-finder/anonymous_proxies.txt
|
|
"""
|
|
|
|
import subprocess
|
|
import re
|
|
import sys
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
|
|
|
|
def fetch_with_curl(url):
|
|
"""Загрузка через curl с таймаутом."""
|
|
try:
|
|
result = subprocess.run(
|
|
['curl', '-sL', '--connect-timeout', '5', url],
|
|
capture_output=True, text=True, timeout=8
|
|
)
|
|
if result.returncode == 0 and len(result.stdout) > 100:
|
|
return url, result.stdout
|
|
except Exception as e:
|
|
print(f"[!] Ошибка загрузки {url}: {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
def parse_spys_me(data):
|
|
"""Парсинг spys.me формата."""
|
|
proxies = []
|
|
for line in data.splitlines():
|
|
match = re.search(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\s*:\s*(\d+)', line)
|
|
if match:
|
|
ip, port = match.group(1), int(match.group(2))
|
|
octets = [int(x) for x in ip.split('.')]
|
|
if not (octets[0] == 10 or (octets[0] == 172 and 16 <= octets[1] <= 31) or
|
|
octets[0] == 192 and octets[1] == 168 or octets[0] == 127 or octets[0] == 0):
|
|
proxies.append((ip, port))
|
|
return proxies
|
|
|
|
|
|
def parse_html_table(data):
|
|
"""Парсинг HTML-таблиц (sslproxies.org, netips.ch)."""
|
|
proxies = []
|
|
for match in re.finditer(r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\s*:\s*(\d+)', data):
|
|
ip, port = match.group(1), int(match.group(2))
|
|
octets = [int(x) for x in ip.split('.')]
|
|
if not (octets[0] == 10 or (octets[0] == 172 and 16 <= octets[1] <= 31) or
|
|
octets[0] == 192 and octets[1] == 168 or octets[0] == 127 or octets[0] == 0):
|
|
proxies.append((ip, port))
|
|
return proxies
|
|
|
|
|
|
def parse_proxies(data, source_url):
|
|
"""Умный парсинг в зависимости от источника."""
|
|
if 'spys.me' in source_url:
|
|
return parse_spys_me(data)
|
|
else:
|
|
# HTML таблицы — ищем IP:PORT паттерн
|
|
return parse_html_table(data)
|
|
|
|
|
|
def check_proxy_http(ip, port):
|
|
"""Проверка HTTP-прокси через curl."""
|
|
try:
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--proxy', f'http://{ip}:{port}',
|
|
'https://httpbin.org/ip'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0 and ip in result.stdout:
|
|
return (ip, port, 'HTTP')
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def check_proxy_socks5(ip, port):
|
|
"""Проверка SOCKS5-прокси через curl."""
|
|
try:
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--socks5', f'{ip}:{port}',
|
|
'https://httpbin.org/ip'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0 and ip in result.stdout:
|
|
return (ip, port, 'SOCKS5')
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def check_anonymity(ip, port, proxy_type):
|
|
"""Проверка анонимности прокси."""
|
|
try:
|
|
if proxy_type == 'HTTP':
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--proxy', f'http://{ip}:{port}',
|
|
'https://httpbin.org/headers'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
else:
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--socks5', f'{ip}:{port}',
|
|
'https://httpbin.org/headers'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0:
|
|
# Проверяем отсутствие X-Forwarded-For (анонимный прокси)
|
|
if 'X-Forwarded-For' not in result.stdout.lower():
|
|
return (ip, port, proxy_type, True) # Анонимный
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def check_speed(ip, port, proxy_type):
|
|
"""Проверка скорости прокси."""
|
|
try:
|
|
if proxy_type == 'HTTP':
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--proxy', f'http://{ip}:{port}',
|
|
'-o', '/dev/null', '-w', '%{time_total}',
|
|
'https://httpbin.org/ip'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
else:
|
|
result = subprocess.run(
|
|
['curl', '-s', '--connect-timeout', '5',
|
|
'--socks5', f'{ip}:{port}',
|
|
'-o', '/dev/null', '-w', '%{time_total}',
|
|
'https://httpbin.org/ip'],
|
|
capture_output=True, text=True, timeout=10
|
|
)
|
|
if result.returncode == 0:
|
|
return float(result.stdout.strip())
|
|
except Exception:
|
|
pass
|
|
return None
|
|
|
|
|
|
def main():
|
|
print("[*] Поиск анонимных прокси...")
|
|
|
|
sources = [
|
|
'https://spys.me/proxy.txt',
|
|
'https://www.sslproxies.org/',
|
|
'http://netips.ch/proxylist/anonymous.html',
|
|
]
|
|
|
|
all_proxies = []
|
|
print("[*] Загрузка из источников...")
|
|
with ThreadPoolExecutor(max_workers=3) as executor:
|
|
futures = [executor.submit(fetch_with_curl, url) for url in sources]
|
|
for future in as_completed(futures):
|
|
result = future.result()
|
|
if result:
|
|
url, data = result
|
|
proxies = parse_proxies(data, url)
|
|
print(f" [+] {url}: найдено {len(proxies)} прокси")
|
|
all_proxies.extend(proxies)
|
|
|
|
# Убираем дубликаты
|
|
seen = set()
|
|
unique_proxies = []
|
|
for ip, port in all_proxies:
|
|
key = f"{ip}:{port}"
|
|
if key not in seen:
|
|
seen.add(key)
|
|
unique_proxies.append((ip, port))
|
|
|
|
print(f"\n[*] Всего уникальных прокси: {len(unique_proxies)}")
|
|
|
|
# Сохраняем все найденные (быстрый режим)
|
|
output_file = '/root/.openclaw/workspace/scripts/proxy-finder/anonymous_proxies.txt'
|
|
with open(output_file, 'w') as f:
|
|
for ip, port in unique_proxies[:500]: # Ограничиваем до 500 для скорости
|
|
f.write(f"{ip}:{port} (unverified)\n")
|
|
|
|
print(f" [+] Сохранено {len(unique_proxies[:500])} прокси в {output_file}")
|
|
|
|
# Опциональная проверка (медленная)
|
|
if len(sys.argv) > 1 and sys.argv[1] == '--verify':
|
|
print("[*] Проверка HTTP-прокси...")
|
|
working_http = []
|
|
with ThreadPoolExecutor(max_workers=10) as executor:
|
|
futures = [executor.submit(check_proxy_http, ip, port) for ip, port in unique_proxies[:200]]
|
|
for future in as_completed(futures):
|
|
result = future.result()
|
|
if result:
|
|
working_http.append(result)
|
|
|
|
print(f" [+] Рабочих HTTP-прокси: {len(working_http)}")
|
|
|
|
# Проверяем анонимность и скорость
|
|
anon_proxies = []
|
|
for ip, port, ptype in working_http[:50]:
|
|
result = check_anonymity(ip, port, ptype)
|
|
if result:
|
|
speed = check_speed(ip, port, ptype)
|
|
anon_proxies.append((ip, port, ptype, speed))
|
|
|
|
print(f" [+] Анонимных прокси: {len(anon_proxies)}")
|
|
|
|
# Обновляем файл с проверенными
|
|
with open(output_file, 'w') as f:
|
|
for ip, port, ptype, speed in anon_proxies:
|
|
f.write(f"{ip}:{port} ({ptype}, {speed:.2f}s)\n")
|
|
|
|
print(f"\n[+] Результат сохранён в {output_file}")
|
|
return unique_proxies
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|