admin/ProxyCat
1
0 Fork 0
mirror of https://github.com/honmashironeko/ProxyCat.git synced 2026-10-24 16:44:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

Update

Browse Source
This commit is contained in:
本间白猫 committed 2024-09-30 09:09:22 +08:00
1 parent a42ff1b602
commit 07c03203cf
2 files changed
+187 -654
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ProxyCat-CN/ProxyCat-V1.7.py
-540
View File
@@ -1,540 +0,0 @@
from colorama import init, Fore
from packaging import version
from itertools import cycle
import configparser, threading, logoprint, argparse, logging, asyncio, socket, base64, getip, httpx, time, re, struct, random
init(autoreset=True)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
DEFAULT_CONFIG = {
'port': 1080,
'mode': 'cycle',
'interval': 300,
'username': '',
'password': '',
'use_getip': False,
'proxy_file': 'ip.txt',
}
def load_config(config_file='config.ini'):
config = configparser.ConfigParser()
config.read(config_file, encoding='utf-8')
return {k: v for k, v in config['SETTINGS'].items()}
def load_proxies(file_path='ip.txt'):
with open(file_path, 'r') as file:
return [line.strip() for line in file if '://' in line]
class AsyncProxyServer:
def __init__(self, config):
self.config = {**DEFAULT_CONFIG, **config}
self.username = self.config['username'].strip()
self.password = self.config['password'].strip()
self.auth_required = bool(self.username and self.password)
self.mode = self.config['mode']
self.interval = int(self.config['interval'])
self.use_getip = self.config.get('use_getip', 'False').lower() == 'true'
self.proxy_file = self.config['proxy_file']
self.proxies = self.load_proxies()
self.proxy_cycle = cycle(self.proxies)
self.current_proxy = next(self.proxy_cycle) if self.proxies else "没有可用的代理"
self.last_switch_time = time.time()
self.rate_limiter = asyncio.Queue(maxsize=3000)
def load_proxies(self):
return [getip.newip()] if self.use_getip else load_proxies(self.proxy_file)
async def get_next_proxy(self):
if self.mode == 'load_balance':
return random.choice(self.proxies)
if time.time() - self.last_switch_time >= self.interval:
self.current_proxy = getip.newip() if self.use_getip else next(self.proxy_cycle)
self.last_switch_time = time.time()
logging.info(f"切换到新的代理: {self.current_proxy}")
return self.current_proxy
def time_until_next_switch(self):
return float('inf') if self.mode == 'load_balance' else max(0, self.interval - (time.time() - self.last_switch_time))
async def acquire(self):
await self.rate_limiter.put(None)
await asyncio.sleep(0.001)
self.rate_limiter.get_nowait()
async def handle_client(self, reader, writer):
try:
await self.acquire()
first_byte = await reader.read(1)
if not first_byte:
return
if first_byte == b'\x05':
await self.handle_socks5_connection(reader, writer)
else:
await self._handle_client_impl(reader, writer, first_byte)
except asyncio.CancelledError:
logging.info("客户端处理取消")
except Exception as e:
logging.error(f"客户端处理出错: {e}")
finally:
writer.close()
await writer.wait_closed()
async def handle_socks5_connection(self, reader, writer):
nmethods = ord(await reader.readexactly(1))
methods = await reader.readexactly(nmethods)
if self.auth_required:
writer.write(b'\x05\x02')
else:
writer.write(b'\x05\x00')
await writer.drain()
if self.auth_required:
auth_version = await reader.readexactly(1)
if auth_version != b'\x01':
writer.close()
return
ulen = ord(await reader.readexactly(1))
username = await reader.readexactly(ulen)
plen = ord(await reader.readexactly(1))
password = await reader.readexactly(plen)
if username.decode() != self.username or password.decode() != self.password:
writer.write(b'\x01\x01')
await writer.drain()
writer.close()
return
writer.write(b'\x01\x00')
await writer.drain()
version, cmd, _, atyp = struct.unpack('!BBBB', await reader.readexactly(4))
if cmd != 1:
writer.write(b'\x05\x07\x00\x01\x00\x00\x00\x00\x00\x00')
await writer.drain()
writer.close()
return
if atyp == 1:
dst_addr = socket.inet_ntoa(await reader.readexactly(4))
elif atyp == 3:
addr_len = ord(await reader.readexactly(1))
dst_addr = (await reader.readexactly(addr_len)).decode()
elif atyp == 4:
dst_addr = socket.inet_ntop(socket.AF_INET6, await reader.readexactly(16))
else:
writer.write(b'\x05\x08\x00\x01\x00\x00\x00\x00\x00\x00')
await writer.drain()
writer.close()
return
dst_port = struct.unpack('!H', await reader.readexactly(2))[0]
try:
proxy = await self.get_next_proxy()
proxy_type, proxy_addr = proxy.split('://')
proxy_auth, proxy_host_port = self._split_proxy_auth(proxy_addr)
proxy_host, proxy_port = proxy_host_port.split(':')
proxy_port = int(proxy_port)
remote_reader, remote_writer = await asyncio.open_connection(proxy_host, proxy_port)
if proxy_type == 'socks5':
remote_writer.write(b'\x05\x01\x00')
await remote_writer.drain()
await remote_reader.readexactly(2)
remote_writer.write(b'\x05\x01\x00' +
(b'\x03' + len(dst_addr).to_bytes(1, 'big') + dst_addr.encode() if isinstance(dst_addr, str) else b'\x01' + socket.inet_aton(dst_addr)) +
struct.pack('!H', dst_port))
await remote_writer.drain()
await remote_reader.readexactly(10)
elif proxy_type in ['http', 'https']:
connect_request = f'CONNECT {dst_addr}:{dst_port} HTTP/1.1\r\nHost: {dst_addr}:{dst_port}\r\n'
if proxy_auth:
connect_request += f'Proxy-Authorization: Basic {base64.b64encode(proxy_auth.encode()).decode()}\r\n'
connect_request += '\r\n'
remote_writer.write(connect_request.encode())
await remote_writer.drain()
while True:
line = await remote_reader.readline()
if line == b'\r\n':
break
writer.write(b'\x05\x00\x00\x01\x00\x00\x00\x00\x00\x00')
await writer.drain()
await asyncio.gather(
self._pipe(reader, remote_writer),
self._pipe(remote_reader, writer)
)
except Exception as e:
logging.error(f"SOCKS5 连接错误: {e}")
writer.write(b'\x05\x01\x00\x01\x00\x00\x00\x00\x00\x00')
await writer.drain()
async def _handle_client_impl(self, reader, writer, first_byte):
try:
request_line = first_byte + await reader.readline()
if not request_line:
return
try:
method, path, _ = request_line.decode('utf-8', errors='ignore').split()
except ValueError:
logging.error(f"无效的请求行: {request_line}")
return
headers = {}
while True:
line = await reader.readline()
if line == b'\r\n':
break
if line == b'':
return
try:
name, value = line.decode('utf-8', errors='ignore').strip().split(': ', 1)
headers[name.lower()] = value
except ValueError:
logging.error(f"无效的请求行: {line}")
continue
if self.auth_required and not self._authenticate(headers):
writer.write(b'HTTP/1.1 407 Proxy Authentication Required\r\nProxy-Authenticate: Basic realm="Proxy"\r\n\r\n')
await writer.drain()
return
if method == 'CONNECT':
await self._handle_connect(path, reader, writer)
else:
await self._handle_request(method, path, headers, reader, writer)
except asyncio.CancelledError:
raise
except Exception as e:
logging.error(f"处理客户端请求时出错: {e}")
def _authenticate(self, headers):
if not self.auth_required:
return True
auth = headers.get('proxy-authorization')
if not auth:
return False
try:
scheme, credentials = auth.split()
if scheme.lower() != 'basic':
return False
username, password = base64.b64decode(credentials).decode().split(':')
return username == self.username and password == self.password
except:
return False
async def _handle_connect(self, path, reader, writer):
try:
host, port = path.split(':')
port = int(port)
except ValueError:
logging.error(f"无效的连接路径: {path}")
writer.write(b'HTTP/1.1 400 Bad Request\r\n\r\n')
await writer.drain()
return
proxy = await self.get_next_proxy()
proxy_type, proxy_addr = proxy.split('://')
proxy_auth, proxy_host_port = self._split_proxy_auth(proxy_addr)
proxy_host, proxy_port = proxy_host_port.split(':')
proxy_port = int(proxy_port)
try:
remote_reader, remote_writer = await asyncio.wait_for(asyncio.open_connection(proxy_host, proxy_port),timeout=10)
if proxy_type == 'http':
connect_headers = [f'CONNECT {host}:{port} HTTP/1.1', f'Host: {host}:{port}']
if proxy_auth:
auth_header = f'Proxy-Authorization: Basic {base64.b64encode(proxy_auth.encode()).decode()}'
connect_headers.append(auth_header)
connect_request = '\r\n'.join(connect_headers) + '\r\n\r\n'
remote_writer.write(connect_request.encode())
await remote_writer.drain()
response = await remote_reader.readline()
if not response.startswith(b'HTTP/1.1 200'):
raise Exception("Bad Gateway")
while (await remote_reader.readline()) != b'\r\n':
pass
elif proxy_type in ['socks4', 'socks5']:
if proxy_type == 'socks4':
remote_writer.write(b'\x04\x01' + port.to_bytes(2, 'big') + socket.inet_aton(host) + b'\x00')
else:
remote_writer.write(b'\x05\x01\x00')
await remote_writer.drain()
if (await remote_reader.read(2))[1] == 0:
remote_writer.write(b'\x05\x01\x00\x03' + len(host).to_bytes(1, 'big') + host.encode() + port.to_bytes(2, 'big'))
await remote_writer.drain()
if (await remote_reader.read(10))[1] != 0:
raise Exception("Bad Gateway")
else:
raise Exception("Unsupported proxy type")
writer.write(b'HTTP/1.1 200 Connection Established\r\n\r\n')
await writer.drain()
await asyncio.gather(
self._pipe(reader, remote_writer),
self._pipe(remote_reader, writer)
)
except asyncio.TimeoutError:
logging.error("连接超时")
writer.write(b'HTTP/1.1 504 Gateway Timeout\r\n\r\n')
await writer.drain()
except Exception as e:
logging.error(f"连接错误: {e}")
writer.write(b'HTTP/1.1 502 Bad Gateway\r\n\r\n')
await writer.drain()
def _split_proxy_auth(self, proxy_addr):
match = re.match(r'((?P<username>.+?):(?P<password>.+?)@)?(?P<host>.+)', proxy_addr)
if match:
username = match.group('username')
password = match.group('password')
host = match.group('host')
if username and password:
return f"{username}:{password}", host
return None, proxy_addr
async def _pipe(self, reader, writer):
try:
while True:
try:
data = await reader.read(8192)
if not data:
break
writer.write(data)
await writer.drain()
except asyncio.CancelledError:
break
finally:
writer.close()
await writer.wait_closed()
async def _handle_request(self, method, path, headers, reader, writer):
body = await reader.read()
proxy = await self.get_next_proxy()
proxy_type, proxy_addr = proxy.split('://')
proxy_auth, proxy_host_port = self._split_proxy_auth(proxy_addr)
client_kwargs = {
"limits": httpx.Limits(max_keepalive_connections=500, max_connections=3000),
"timeout": 30,
}
if proxy_type in ['http', 'https']:
client_kwargs["proxies"] = {proxy_type: f"{proxy_type}://{proxy_host_port}"}
elif proxy_type in ['socks4', 'socks5']:
client_kwargs["transport"] = httpx.AsyncHTTPTransport(proxy=f"{proxy_type}://{proxy_host_port}")
if proxy_auth:
headers['Proxy-Authorization'] = f'Basic {base64.b64encode(proxy_auth.encode()).decode()}'
async with httpx.AsyncClient(**client_kwargs) as client:
try:
async with client.stream(method, path, headers=headers, content=body) as response:
await self._write_response(writer, response)
except asyncio.CancelledError:
raise
except Exception as e:
logging.error(f"请求处理出错: {e}")
writer.write(b'HTTP/1.1 502 Bad Gateway\r\n\r\n')
await writer.drain()
async def _write_response(self, writer, response):
try:
writer.write(f'HTTP/1.1 {response.status_code} {response.reason_phrase}\r\n'.encode('utf-8', errors='ignore'))
writer.write(b'Transfer-Encoding: chunked\r\n')
for name, value in response.headers.items():
if name.lower() != 'transfer-encoding':
writer.write(f'{name}: {value}\r\n'.encode('utf-8', errors='ignore'))
writer.write(b'\r\n')
await writer.drain()
async for chunk in response.aiter_bytes(chunk_size=8192):
if asyncio.current_task().cancelled():
raise asyncio.CancelledError()
writer.write(f'{len(chunk):X}\r\n'.encode('utf-8', errors='ignore'))
writer.write(chunk)
writer.write(b'\r\n')
await writer.drain()
writer.write(b'0\r\n\r\n')
await writer.drain()
except asyncio.CancelledError:
logging.info("响应写入被取消")
raise
def print_banner(config):
auth_info = f"{config.get('username')}:{config.get('password')}" if config.get('username') and config.get('password') else "未设置 (无需认证)"
banner_info = [
('公众号', '樱花庄的本间白猫'),
('博客', 'https://y.shironekosan.cn'),
('夸克网盘', 'https://pan.quark.cn/s/39b4b5674570'),
('Github', 'https://github.com/honmashironeko/ProxyCat'),
('百度网盘', 'https://pan.baidu.com/s/1C9LVC9aiaQeYFSj_2mWH1w?pwd=13r5'),
('代理轮换模式', '循环' if config.get('mode') == 'cycle' else '负载均衡' if config.get('mode') == 'load_balance' else '单轮'),
('代理更换时间', f"{config.get('interval')}秒"),
('默认账号密码', auth_info),
('本地监听地址', f"http://{auth_info}@127.0.0.1:{config.get('port')}"),
('本地监听地址', f"socks5://{auth_info}@127.0.0.1:{config.get('port')}"),
]
print(f"{Fore.MAGENTA}{'=' * 55}")
for key, value in banner_info:
print(f"{Fore.YELLOW}{key}: {Fore.GREEN}{value}")
print(f"{Fore.MAGENTA}{'=' * 55}\n")
def update_status(server):
while True:
if server.mode == 'load_balance':
status = f"\r{Fore.YELLOW}当前代理: {Fore.GREEN}{server.current_proxy}"
else:
time_left = server.time_until_next_switch()
status = f"\r{Fore.YELLOW}当前代理: {Fore.GREEN}{server.current_proxy} | {Fore.YELLOW}下次切换: {Fore.GREEN}{time_left:.1f}秒"
print(status, end='', flush=True)
time.sleep(1)
async def handle_client_wrapper(server, reader, writer, clients):
task = asyncio.create_task(server.handle_client(reader, writer))
clients.add(task)
try:
await task
except Exception as e:
logging.error(f"客户端处理出错: {e}")
finally:
clients.remove(task)
async def run_server(server):
clients = set()
server_instance = None
try:
server_instance = await asyncio.start_server(
lambda r, w: handle_client_wrapper(server, r, w, clients),
'0.0.0.0',
int(server.config['port'])
)
async with server_instance:
await server_instance.serve_forever()
except asyncio.CancelledError:
logging.info("服务器正在关闭...")
finally:
if server_instance:
server_instance.close()
await server_instance.wait_closed()
for client in clients:
client.cancel()
await asyncio.gather(*clients, return_exceptions=True)
async def check_proxy(proxy):
proxy_type = proxy.split('://')[0]
check_funcs = {
'http': check_http_proxy,
'https': check_https_proxy,
'socks4': check_socks_proxy,
'socks5': check_socks_proxy
}
if proxy_type not in check_funcs:
return False
try:
return await check_funcs[proxy_type](proxy)
except Exception as e:
logging.error(f"{proxy_type.upper()}代理 {proxy} 检测失败: {e}")
return False
async def check_http_proxy(proxy):
async with httpx.AsyncClient(proxies={'http://': proxy}, timeout=10) as client:
response = await client.get('http://www.baidu.com')
return response.status_code == 200
async def check_https_proxy(proxy):
async with httpx.AsyncClient(proxies={'https://': proxy}, timeout=10) as client:
response = await client.get('https://www.baidu.com')
return response.status_code == 200
async def check_socks_proxy(proxy):
proxy_type, proxy_addr = proxy.split('://')
proxy_host, proxy_port = proxy_addr.split(':')
proxy_port = int(proxy_port)
try:
reader, writer = await asyncio.wait_for(asyncio.open_connection(proxy_host, proxy_port), timeout=5)
writer.write(b'\x05\x01\x00')
await writer.drain()
response = await asyncio.wait_for(reader.readexactly(2), timeout=5)
writer.close()
await writer.wait_closed()
return response == b'\x05\x00'
except Exception:
return False
async def check_proxies(proxies):
valid_proxies = []
for proxy in proxies:
if await check_proxy(proxy):
valid_proxies.append(proxy)
return valid_proxies
async def run_proxy_check(server):
if server.config.get('check_proxies', 'False').lower() == 'true':
logging.info("开始检测代理地址...")
valid_proxies = await check_proxies(server.proxies)
if valid_proxies:
server.proxies = valid_proxies
server.proxy_cycle = cycle(valid_proxies)
server.current_proxy = next(server.proxy_cycle)
logging.info(f"有效代理地址: {valid_proxies}")
else:
logging.error("没有有效的代理地址")
else:
logging.info("代理检测已禁用")
async def check_for_updates():
try:
async with httpx.AsyncClient() as client:
response = await asyncio.wait_for(client.get("https://y.shironekosan.cn/1.html"), timeout=10)
response.raise_for_status()
content = response.text
match = re.search(r'<p>(ProxyCat-V\d+\.\d+)</p>', content)
if match:
latest_version = match.group(1)
CURRENT_VERSION = "ProxyCat-V1.7"
if version.parse(latest_version.split('-V')[1]) > version.parse(CURRENT_VERSION.split('-V')[1]):
print(f"{Fore.YELLOW}发现新版本！当前版本: {CURRENT_VERSION}, 最新版本: {latest_version}")
print(f"{Fore.YELLOW}请访问 https://pan.quark.cn/s/39b4b5674570 获取最新版本。")
print(f"{Fore.YELLOW}请访问 https://github.com/honmashironeko/ProxyCat 获取最新版本。")
print(f"{Fore.YELLOW}请访问 https://pan.baidu.com/s/1C9LVC9aiaQeYFSj_2mWH1w?pwd=13r5 获取最新版本。")
else:
print(f"{Fore.GREEN}当前版本已是最新 ({CURRENT_VERSION})")
else:
print(f"{Fore.RED}无法在响应中找到版本信息")
except Exception as e:
print(f"{Fore.RED}检查更新时发生错误: {e}")
if __name__ == '__main__':
import argparse
parser = argparse.ArgumentParser(description=logoprint.logos())
parser.add_argument('-c', '--config', default='config.ini', help='配置文件路径')
args = parser.parse_args()
config = load_config(args.config)
server = AsyncProxyServer(config)
print_banner(config)
asyncio.run(check_for_updates())
asyncio.run(run_proxy_check(server))
status_thread = threading.Thread(target=update_status, args=(server,), daemon=True)
status_thread.start()
try:
asyncio.run(run_server(server))
except KeyboardInterrupt:
logging.info("程序被用户中断")
ProxyCat-EN/README-EN.md
+187 -114
View File
@@ -1,121 +1,185 @@
![ProxyCat](https://socialify.git.ci/honmashironeko/ProxyCat/image?description=1&descriptionEditable=%E4%B8%80%E6%AC%BE%E8%BD%BB%E9%87%8F%E7%BA%A7%E7%9A%84%E4%BC%98%E7%A7%80%E4%BB%A3%E7%90%86%E6%B1%A0%E4%B8%AD%E9%97%B4%E4%BB%B6%EF%BC%8C%E5%AE%9E%E7%8E%B0%E4%BB%A3%E7%90%86%E7%9A%84%E8%87%AA%E5%8A%A8%E8%BD%AE%E6%8D%A2&font=Bitter&forks=1&issues=1&language=1&logo=https%3A%2F%2Favatars.githubusercontent.com%2Fu%2F139044047%3Fv%3D4&name=1&owner=1&pattern=Circuit%20Board&pulls=1&stargazers=1&theme=Dark)
<p align="center">
<a href="/ProxyCat-EN/README-EN.md">English</a>
·
<a href="/README.md">简体中文</a>
</p>
<p align="center">
<a href="/ProxyCat-EN/README-EN.md">English</a>
·
<a href="/README.md">简体中文</a>
</p>
## Project Overview
## Table of Contents
During penetration testing, one often encounters situations where hiding or changing the IP address is necessary to bypass security devices. However, commercial tunnel proxies are generally very expensive, ranging from 20 to 40 yuan per day, which can be quite prohibitive. The author has observed that short-lived IPs are much more cost-effective, averaging only 0.2 to 3 yuan per day.
- [Development Motivation](#development-motivation)
- [Features](#features)
- [Installation and Usage](#installation-and-usage)
- [Installing Dependencies](#installing-dependencies)
- [Running the Tool](#running-the-tool)
- [Manually Entering Proxy Addresses](#manually-entering-proxy-addresses)
- [Configuration File](#configuration-file)
- [Demonstration](#demonstration)
- [Automatically Obtaining Proxy Addresses via API](#automatically-obtaining-proxy-addresses-via-api)
- [Performance](#performance)
- [Disclaimer](#disclaimer)
- [Changelog](#changelog)
- [Development Plan](#development-plan)
- [Acknowledgements](#acknowledgements)
- [Sponsor Open Source](#sponsor-open-source)
- [Recommended Proxies](#recommended-proxies)
Thus, this tool was developed to transform short-lived IPs (lasting from 1 to 60 minutes) into a stable IP, enabling other tools to utilize them effectively and forming a proxy pool server that can be deployed once for permanent use.
## Development Motivation
During penetration testing, it is often necessary to hide or change IP addresses to bypass security devices. However, tunnel proxies available in the market are expensive, typically costing between 20-40 RMB per day, which is unaffordable for many. I noticed that short-lived IPs offer high cost-effectiveness, with each IP costing only a few cents, averaging 0.2-3 RMB per day.
In summary, **ProxyCat** was born! This tool aims to transform short-lived IPs, lasting from 1 minute to 60 minutes, into fixed IPs for use by other tools, forming a proxy pool server that can be deployed once for permanent use.
![项目原理图](./assets/202408260021207-1725093725174-21.png)
## Supported Features
## Features
```
Local Listening Service: Customizable listening ports, custom authentication for listening ports, Two monitoring protocols.
Proxy Address Support: Four protocol types, timed proxy address changes, automatic address change after each request, API for automatic proxy retrieval, supports authenticated proxies, Support proxy address validity check, Support load balancing mode.
Service Runtime: Changes proxy addresses only upon receiving requests, supports ultra-high concurrency, supports HTTP and HTTPS protocols, checks for version updates at startup.
System Compatibility: Compatible with Windows, Linux, and MacOS; supports VPS remote deployment and local installation.
```
**Multi-Protocol Support**
## Project Usage
- **SOCKS5 Proxy**: Supports the SOCKS5 protocol, suitable for various network environments.
- **HTTP/HTTPS Proxy**: Supports HTTP and HTTPS proxies to meet different application scenarios.
### Install Dependencies
**Proxy Rotation Modes**
This tool is implemented in Python (recommended version 3.8 or higher). Please configure the dependencies using the following commands:
- **Cycle Mode**: Sequentially cycles through each proxy in the list, ensuring balanced usage.
- **Load Balance Mode**: Randomly selects available proxies to distribute traffic load and enhance performance.
- **Custom Mode**: Allows users to customize proxy selection logic to flexibly meet specific needs.
```bash
**Dynamic Proxy Acquisition**
- **Acquire Proxies Using GetIP Function**: Supports dynamically obtaining real-time available proxies through the GetIP function, ensuring the proxies are current and valid.
**Proxy Verification**
- **Automatic Validity Detection**: Automatically checks the availability of proxies at startup, filtering out invalid proxies to ensure the reliability of the proxy list.
- **Supports Multiple Protocols for Verification**: Specifically checks HTTP, HTTPS, and SOCKS5 proxies to improve validation accuracy.
**Authentication Mechanism**
- **Username/Password Authentication**: Supports proxy authentication based on username and password to enhance security and prevent unauthorized access.
**High Concurrency Handling**
- **Asynchronous Architecture**: Implements asynchronous processing based on `asyncio`, supporting large-scale concurrent connections suitable for high-traffic demands.
**Logging and Monitoring**
- **Colored Log Output**: Utilizes `colorama` for colored logs, facilitating real-time monitoring and debugging.
- **Real-Time Status Updates**: Displays the current proxy status and the next switch time to help users understand proxy dynamics.
**Flexible Configuration**
- **Configurable Files**: Easily adjust parameters such as port, mode, and authentication information through the `config.ini` file to adapt to different usage scenarios.
- **Command-Line Arguments**: Supports specifying the configuration file path via command-line arguments for increased convenience.
**Automatic Update Checks**
- **Version Detection**: Built-in version detection feature automatically checks for the latest version and notifies users to ensure continuous software optimization.
## Installation and Usage
### Installing Dependencies
The tool is implemented in Python and is recommended to use **Python 3.8** or higher. Before use, configure the dependencies using the following command:
````bash:c:\Users\hoshi\Documents\GitHub\ProxyCat\requirements.txt
pip install -r requirements.txt
```
# Or recommended to use domestic source:
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple/
````
Run the command `python3 ProxyCat.py -h` in the project directory, and if the following output appears, the configuration is successful:
### Running the Tool
Run the following command in the project directory to view the help information and confirm the configuration is successful:
````bash:c:\Users\hoshi\Documents\GitHub\ProxyCat\ProxyCat.py
python3 ProxyCat.py -h
````
The following output indicates successful configuration:
```
# python3 ProxyCat.py -h
|\ _,,,---,,_ by 本间白猫
|\ _,,,---,,_ by Honma Shironeko
ZZZzz /,`.-'`' -. ;-;;,_
|,4- ) )-,_. ,\ ( `'-'
'---''(_/--' `-'\_) ProxyCat
Usage: ProxyCat.py [-h] [-c]
Arguments:
-h, --help Display this help message and exit
-c C Specify configuration file name (default: config.ini)
Parameters:
-h, --help Show this help message and exit
-c C Specify the configuration file name (default: config.ini)
```
### Manually Enter Proxy Addresses in ip.txt
### Manually Entering Proxy Addresses in ip.txt
Enter proxy addresses in the following format, one per line, into the `ip.txt` file:
Enter proxy addresses in the `ip.txt` file in the following format (`socks5://127.0.0.1:7890` or `http://127.0.0.1:7890`), one per line:
```ip.txt
````plaintext:c:\Users\hoshi\Documents\GitHub\ProxyCat\ip.txt
socks5://127.0.0.1:7890
socks4://127.0.0.1:7890
https://127.0.0.1:7890
http://127.0.0.1:7890
...
```
````
### Configuration File
Configure parameters in `config.ini` (or a custom configuration file):
```
````ini:c:\Users\hoshi\Documents\GitHub\ProxyCat\config.ini
[SETTINGS]
# Local server listening port (default: 1080)
port = 1080
# # Proxy address rotation mode: cycle means circular use, once means stop when used up, and load_balance means load balancing (default: cycle)
# Proxy rotation mode: cycle for sequential use, custom for custom mode, load_balance for load balancing (default: cycle)
mode = cycle
# Proxy address change time (seconds), when set to 0, the IP will be changed for each request (default: 300)
# Proxy switching interval (seconds). Set to 0 to switch IP on every request (default: 300)
interval = 300
# Local server port authentication username (default: neko) No authentication is required when it is empty
# Username for authenticating the local server port (default: neko). Leave empty if no authentication is required
username = neko
# Local server port authentication password (default: 123456) No authentication is required when it is empty
# Password for authenticating the local server port (default: 123456). Leave empty if no authentication is required
password = 123456
# Whether to use the getip module to obtain the proxy address True or False (default: False)
# Whether to use the getip module to obtain proxy addresses True or False (default: False)
use_getip = False
# Proxy address list file (default: ip.txt)
proxy_file = ip.txt
# Whether proxy detection is enabled True or False(default True)
# Whether to enable proxy verification feature True or False (default: True)
check_proxies = True
```
````
Once the corresponding parameters are configured, you may use the following command:
After configuring the corresponding parameters, you can use the tool:
```bash
# python3 ProxyCat.py
```
````bash:c:\Users\hoshi\Documents\GitHub\ProxyCat\ProxyCat.py
python3 ProxyCat.py
````
**Demonstration Result**
### Demonstration
Fixed proxy address (default):
**Fixed Proxy Addresses (Default):**
```
````plaintext:c:\Users\hoshi\Documents\GitHub\ProxyCat\README.md
http://neko:[email protected]:1080
http://127.0.0.1:1080
socks5://neko:[email protected]:1080
socks5://127.0.0.1:1080
```
````
If deployed publicly, replace `127.0.0.1` with your public IP address.
If you are deploying on the public network, replace `127.0.0.1` with your public IP.
![Clip_2024-08-31_16-49-22](./assets/Clip_2024-08-31_16-49-22.png)
![Clip_2024-09-30_09-05-17](C:\Users\hoshi\AppData\Local\Programs\PixPin\Temp\Clip_2024-09-30_09-05-17.png)
### Use API to Automatically Retrieve Proxy Addresses
### Automatically Obtaining Proxy Addresses via API
The tool supports directly calling the API for proxy address retrieval. When you set `use_getip = True`, the tool will no longer read from the local `ip.txt` and will instead execute **getip.py** to fetch new proxy addresses(Please remember to whitelist your IP).
The tool supports directly calling API interfaces to obtain proxy addresses. When you set `use_getip = True`, the tool will no longer read proxy addresses from the local `ip.txt` but will obtain new proxy addresses by executing the **getip.py** script (ensure your IP is whitelisted).
In this case, you need to modify the contents of **getip.py** to use your own API, formatted as `IP:PORT`. The default is `socks5` protocol; change it manually if using `http`.
At this point, you need to modify the content of **getip.py** to your own interface, formatted as `IP:PORT`. The default is the `socks5` protocol. If you need to use `http`, please change it manually.
**Demonstration Result**
@@ -123,102 +187,111 @@ In this case, you need to modify the contents of **getip.py** to use your own AP
![Clip_2024-08-31_20-44-23](https://github.com/user-attachments/assets/42c1f3ef-0e75-4b07-a901-1c8b76f7f9c3)
## Performance
Current testing shows that under adequate server performance for proxy addresses, the tool can support 500 concurrent requests without packet loss, effectively covering most scanning and penetration testing needs.
After actual testing, when the proxy address server has sufficient performance, ProxyCat can handle **1000** concurrent connections without packet loss, covering most scanning and penetration testing needs.
![8e3f79309626ed0e653ba51b6482bff](./assets/8e3f79309626ed0e653ba51b6482bff-1725093725174-23.png)
## Disclaimer
- By downloading, installing, using, or modifying this tool and its related code, you indicate your trust in this tool.
- We take no responsibility for any loss or damage caused to you or others while using this tool.
- If you engage in any illegal activities while using this tool, you assume all corresponding consequences, and we will not bear any legal or associated liabilities.
- Please read and thoroughly understand each clause, especially those that exempt or limit liability, and choose to accept or decline.
- Unless you have read and accepted all terms of this agreement, you are not entitled to download, install, or use this tool.
- Your actions, including downloading, installing, or using, indicate that you have read and agree to these terms.
- If you download, install, use, or modify this tool and related code, you indicate your trust in this tool.
- We do not assume any responsibility for any form of loss or damage to you or others caused by using this tool.
- If you engage in any illegal activities while using this tool, you must bear the corresponding consequences yourself. We will not bear any legal or related responsibilities.
- Please read and fully understand all the terms, especially the clauses that exempt or limit liability, and choose to accept or not accept.
- Unless you have read and accepted all the terms of this agreement, you are not authorized to download, install, or use this tool.
- Your actions of downloading, installing, and using this tool are deemed as your agreement to the above terms.
## Changelog
**2024/09/10**
### 2024/09/29
- Optimize concurrency efficiency, support the next request in advance without receiving the response packet, thereby improving efficiency
- Add a load balancing mode, in which requests will be randomly made to the proxy address, and use concurrent proxies to improve request efficiency
- Agent validity detection modified to be asynchronous, thereby improving efficiency
- Removed the less-used single cycle mode and replaced it with a custom mode, allowing users to customize the proxy switching logic based on needs.
- Modified proxy validity checks to asynchronous for increased speed.
- Removed support for the problematic SOCKS4 protocol.
- Enhanced the logging system aesthetics.
- Improved exception handling logic.
- Added validation for proxy formats to ensure correctness.
**2024/09/09**
### 2024/09/10
- Added function, you can set whether to check the validity of the proxy address in ip.txt when starting for the first time, and only use valid proxy addresses
- function downgrade, support for lower python versions
- Optimized concurrency efficiency, supporting initiating the next request before receiving a response to enhance efficiency.
- Added load balancing mode, randomly sending requests to proxy addresses and using concurrent proxies to improve request efficiency.
- Modified proxy validity checks to asynchronous to improve efficiency.
**2024/09/03**
### 2024/09/09
- Add local socks5 monitoring and adapt to more software
- Replacement of some functions to adapt to lower versions of Python
- Beautify the replayed content
- Added a feature to set whether to perform validity checks on proxy addresses in `ip.txt` during the first startup and use only valid proxies.
- Function downgraded to support lower versions of Python.
**2024/08/31**
### 2024/09/03
- Major structure adjustments for the project.
- Added local SOCKS5 listening to adapt to more software.
- Replaced some functions to support lower versions of Python.
- Enhanced display content aesthetics.
- UI enhancements to indicate the time remaining until the next proxy address change.
- Support for `ctrl+c` to stop operation.
- Significant adjustments to asynchronous requests; tested at 1000 concurrent requests with 5000 packets sent, resulting in about 50 lost packets, achieving ~99% stability; 500 concurrent requests resulted in no packet loss.
- Changed parameter specification from runtime to reading from local ini configuration file for improved usability.
- Support for local no-authentication, adapting to more software proxy modes.
- Added version detection functionality to automatically prompt version information.
- Introduced authentication for proxy server addresses, but this only applies to local reads due to the need for whitelisting in most APIs, hence not provided repeatedly.
- Implemented a feature that updates via `getip` only upon receiving new requests to reduce IP consumption.
- Enhanced automatic recognition of proxy server address protocols to accommodate more proxy providers.
- Added support for HTTPS and SOCKS4 proxy protocols; now covers HTTP, HTTPS, SOCKS5, and SOCKS4 protocols.
### 2024/08/31
**2024/08/25**
- Reorganized the project structure.
- Enhanced display, continuously prompting the next proxy switch time.
- Supported stopping the tool with `Ctrl+C`.
- Significantly shifted to asynchronous requests, improving concurrent efficiency. Tested **1000** concurrent connections with a total of **5000** packets, losing about **50** packets, achieving approximately **99%** stability, and **500** concurrent connections with no packet loss.
- Abandoned the runtime parameter specification approach, modified to read from the local `ini` configuration file for higher usability.
- Supported local unauthenticated access to adapt to more software proxy methods.
- Added version detection feature to automatically prompt version information.
- Added identity verification for proxy server addresses, supporting only local reading as most APIs require whitelisting, thus no duplication was provided.
- Added a feature to update using `getip` only upon receiving new requests to reduce IP consumption.
- Added automatic recognition of proxy server address protocols to adapt to more proxy providers.
- Added support for HTTPS and SOCKS4 proxy protocols, currently covering HTTP, HTTPS, SOCKS5, and SOCKS4 protocols.
- Changed `asyncio.timeout()` to `asyncio.wait_for()` to support lower Python versions.
- Automatically skips empty lines when reading from `ip.txt`.
### 2024/08/25
- Replaced `httpx` with a concurrency pool for performance improvement.
- Added a buffering dictionary to reduce latency for the same site.
- Modified the logic for changing IPs on each request to randomly select a proxy.
- Implemented more efficient structures and algorithms to optimize request handling.
- Automatically skipped empty lines when reading `ip.txt`.
- Replaced `httpx` with a concurrency pool to improve performance.
- Added a buffer dictionary to reduce latency for identical sites.
- Changed the logic of switching IPs on every request to randomly selecting proxies.
- Adopted more efficient structures and algorithms to optimize request handling logic.
**2024/08/24**
### 2024/08/24
- Adopted asynchronous solutions to improve concurrency and reduce timeouts.
- Adopted an asynchronous approach to improve concurrency capabilities and reduce timeouts.
- Encapsulated duplicate code to enhance code reuse.
- Encapsulated redundant code for increased reusability.
### 2024/08/23
**2024/08/23**
- Modified concurrency logic.
- Added identity verification feature.
- Added an IP acquisition interface for permanent IP switching.
- Added a feature to switch IPs on every request.
- Adjusted concurrency logic.
- Added authentication features.
- Added IP retrieval interface for permanent IP changes.
- Introduced functionality to change IP on every request.
## Development Plan
## Development Plans
- [x] Added local server identity verification to prevent unauthorized use during public network deployment.
- [x] Added feature to switch IPs on every request.
- [x] Added a module for automatic acquisition and updating of static proxies for permanent operation.
- [x] Added load balancing mode, using multiple proxy addresses simultaneously to improve concurrency efficiency and reduce single server load.
- [x] Added version detection feature.
- [x] Added support for proxy address identity verification.
- [x] Added a feature to update using `getip` only when receiving new requests to reduce IP consumption.
- [x] Performed batch validity checks on proxy servers in `ip.txt` during the first startup.
- [x] Added local SOCKS protocol listening or fully switched to SOCKS to adapt to more software.
- [ ] Added detailed logging to record the identity of all IPs connecting to ProxyCat and support multiple users.
- [x] Add local server authentication to prevent malicious use in public.
- [x] Introduce functionality to change IP on every request.
- [x] Create a static proxy automatic retrieval module for permanent operation.
- [x] Introduce load balancing mode to use multiple proxy addresses simultaneously, effectively increasing concurrency and reducing load on a single server.
- [x] Add version detection functionality.
- [x] Include support for proxy address authentication.
- [x] Implement functionality for updating via `getip` only upon new requests to minimize IP usage.
- [x] Perform validity checks on all proxy servers in `ip.txt` upon the first startup.
- [x] Add the local listening socks protocol, or completely change it to socks to adapt to more software
If you have good ideas or encounter bugs during use, please contact the author through the following methods to provide feedback!
If you have suggestions or encounter bugs during use, please reach out to the author using the contact information provided!
WeChat Public Account: **Honma Shironeko of Sakura Dormitory**
## Acknowledgments
## Acknowledgements
The following individuals are thanked in no particular order for their contributions to this project.
No particular order is given. Thanks to the mentors who provided help for this project.
[AabyssZG (曾哥)](https://github.com/AabyssZG)
[ProbiusOfficial (探姬)](https://github.com/ProbiusOfficial)
- [AabyssZG (曾哥)](https://github.com/AabyssZG)
- [ProbiusOfficial (探姬)](https://github.com/ProbiusOfficial)
- chars6
![Star History Chart](https://api.star-history.com/svg?repos=honmashironeko/ProxyCat&type=Date)
## Proxy Recommendations
- [Click here to purchase](https://www.ipmart.io?source=Shironeko)
- [Click here to purchase](https://www.ipmart.io?source=Shironeko)