username/s5p
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial SOCKS5 proxy with chain support

Browse Source 
Asyncio-based SOCKS5 server that tunnels connections through
configurable chains of SOCKS5, SOCKS4/4a, and HTTP CONNECT proxies.
Tor integration via standard SOCKS5 hop.
This commit is contained in:
user
2026-02-15 03:10:25 +01:00
commit 0710dda8da
21 changed files with 1117 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.gitignore vendored Normal file
View File

@@ -0,0 +1,8 @@
__pycache__/
*.py[cod]
*.egg-info/
*.egg
.eggs/
dist/
build/
.venv/

14
Makefile Normal file
View File

@@ -0,0 +1,14 @@
.PHONY: install test lint clean
install:
pip install -e .
test:
pytest tests/ -v
lint:
ruff check src/ tests/
clean:
rm -rf build/ dist/ src/*.egg-info
find . -type d -name __pycache__ -exec rm -rf {} +

40
PROJECT.md Normal file
View File

@@ -0,0 +1,40 @@
# s5p -- Project
## Purpose
A lightweight SOCKS5 proxy server that chains connections through Tor and/or
arbitrary proxy hops (SOCKS4, SOCKS5, HTTP CONNECT).
## Motivation
Existing solutions (`proxychains-ng`) rely on `LD_PRELOAD` hacks, only work
on Linux, and intercept at the library level. s5p is a proper SOCKS5 server
that any application can use natively -- no injection required.
## Architecture
```
TCP tunnel tunnel
Client -------> s5p -------> Hop 1 -------> Hop 2 -------> Target
SOCKS5 proto1 proto2 protoN
```
- **server.py** -- asyncio SOCKS5 server, chain builder, bidirectional relay
- **proto.py** -- protocol handshake implementations (SOCKS5, SOCKS4/4a, HTTP CONNECT)
- **config.py** -- YAML config loading, proxy URL parsing
- **cli.py** -- argparse CLI, logging setup
## Dependencies
| Package | Purpose |
|---------|---------|
| pyyaml | Config file parsing |
All other functionality uses Python stdlib (`asyncio`, `socket`, `struct`).
## Design Decisions
- **No LD_PRELOAD** -- clean SOCKS5 server, works with any client
- **asyncio** -- single-threaded event loop, efficient for I/O-bound proxying
- **Domain passthrough** -- never resolve DNS locally to prevent leaks
- **Tor as a hop** -- no special Tor handling; it's just `socks5://127.0.0.1:9050`

72
README.md Normal file
View File

@@ -0,0 +1,72 @@
# s5p
SOCKS5 proxy server with Tor and proxy-chain support. Routes connections
through configurable chains of SOCKS4, SOCKS5, and HTTP CONNECT proxies.
## Features
- SOCKS5 server (RFC 1928)
- Proxy chaining: tunnel through multiple hops in sequence
- Supported hop protocols: SOCKS5, SOCKS4/4a, HTTP CONNECT
- Per-hop authentication (username/password)
- DNS leak prevention (domain names forwarded to proxies, never resolved locally)
- Tor integration (Tor is just another SOCKS5 hop)
- Pure Python, asyncio-based, minimal dependencies
## Quick Start
```bash
# Install
cd ~/git/s5p
python -m venv .venv && source .venv/bin/activate
pip install -e .
# Run with Tor
s5p -C socks5://127.0.0.1:9050
# Run with a chain: Tor -> external proxy
s5p -C socks5://127.0.0.1:9050,socks5://proxy:1080
# Run with config file
s5p -c config/example.yaml
# Test it
curl --proxy socks5h://127.0.0.1:1080 https://check.torproject.org/api/ip
```
## Configuration
```yaml
listen: 127.0.0.1:1080
timeout: 10
chain:
- socks5://127.0.0.1:9050 # Tor
- socks5://user:pass@proxy:1080 # exit-side proxy
- http://proxy2:8080 # HTTP CONNECT proxy
```
## CLI Reference
```
s5p [-c FILE] [-l [HOST:]PORT] [-C URL[,URL,...]] [-t SEC] [-v|-q]
Options:
-c, --config FILE YAML config file
-l, --listen [HOST:]PORT Listen address (default: 127.0.0.1:1080)
-C, --chain URL[,URL] Comma-separated proxy chain
-t, --timeout SEC Per-hop timeout (default: 10)
-v, --verbose Debug logging
-q, --quiet Errors only
-V, --version Show version
```
## How Chaining Works
```
Client -> s5p -> Hop1 -> Hop2 -> ... -> HopN -> Destination
```
s5p connects to Hop1 via TCP, negotiates the hop protocol (SOCKS5/4/HTTP),
then over that tunnel negotiates with Hop2, and so on. The final hop connects
to the actual destination. Each hop only sees its immediate neighbors.

30
ROADMAP.md Normal file
View File

@@ -0,0 +1,30 @@
# s5p -- Roadmap
## v0.1.0 (current)
- [x] SOCKS5 server (CONNECT command)
- [x] Proxy chaining (SOCKS5, SOCKS4/4a, HTTP CONNECT)
- [x] Per-hop authentication
- [x] YAML config + CLI flags
- [x] DNS leak prevention
## v0.2.0
- [ ] SOCKS5 server authentication (username/password)
- [ ] Tor control port integration (circuit renewal via NEWNYM)
- [ ] Connection retry with configurable backoff
- [ ] Metrics (connections/sec, bytes relayed, hop latency)
## v0.3.0
- [ ] UDP ASSOCIATE support (SOCKS5 UDP relay)
- [ ] BIND support
- [ ] Chain randomization (random order, random subset)
- [ ] Hot-reload config on SIGHUP
## v1.0.0
- [ ] Stable API and config format
- [ ] Comprehensive test suite with mock proxies
- [ ] Systemd service unit
- [ ] Performance benchmarks

18
TASKS.md Normal file
View File

@@ -0,0 +1,18 @@
# s5p -- Tasks
## Current
- [x] Scaffold project structure
- [x] Implement SOCKS5 server
- [x] Implement protocol handshakes (SOCKS5, SOCKS4/4a, HTTP CONNECT)
- [x] Implement chain builder
- [x] CLI and config loading
- [x] Unit tests (config, proto)
- [x] Documentation
- [ ] Smoke test with Tor
## Next
- [ ] Integration tests with mock proxy server
- [ ] SOCKS5 server-side authentication
- [ ] Tor control port integration

26
TODO.md Normal file
View File

@@ -0,0 +1,26 @@
# s5p -- Backlog
## Features
- SOCKS5 BIND and UDP ASSOCIATE commands
- Chain randomization modes (random, round-robin)
- Per-destination chain rules (bypass chain for local addresses)
- Hot config reload on SIGHUP
- Systemd socket activation
## Performance
- Benchmark relay throughput vs direct connection
- Tune buffer sizes for different workloads
- Connection pooling for frequently-used chains
## Security
- Optional SOCKS5 server authentication
- Rate limiting per source IP
- Access control lists
## Docs
- Man page
- Architecture diagram

15
config/example.yaml Normal file
View File

@@ -0,0 +1,15 @@
# s5p configuration
listen: 127.0.0.1:1080
timeout: 10
log_level: info
# Proxy chain -- connections tunnel through each hop in order.
# Supported protocols: socks5://, socks4://, http://
#
# Example: route through Tor, then an external SOCKS5 proxy
chain:
- socks5://127.0.0.1:9050 # Tor
# - socks5://user:pass@proxy:1080 # authenticated SOCKS5
# - socks4://proxy:1080 # SOCKS4/4a
# - http://user:pass@proxy:8080 # HTTP CONNECT

46
docs/CHEATSHEET.md Normal file
View File

@@ -0,0 +1,46 @@
# s5p -- Cheatsheet
## CLI
```
s5p # direct, listen :1080
s5p -C socks5://127.0.0.1:9050 # through Tor
s5p -C socks5://tor:9050,http://px:8080 # Tor + HTTP proxy
s5p -c config/example.yaml # from config file
s5p -l 0.0.0.0:9999 # custom listen address
s5p -t 30 # 30s per-hop timeout
s5p -v # debug logging
s5p -q # errors only
```
## Proxy URLs
```
socks5://host:port
socks5://user:pass@host:port
socks4://host:port
http://host:port
http://user:pass@host:port
```
## Testing
```bash
# Check exit IP
curl -x socks5h://127.0.0.1:1080 https://httpbin.org/ip
# Verbose curl
curl -v -x socks5h://127.0.0.1:1080 https://example.com
# With timeout
curl --max-time 30 -x socks5h://127.0.0.1:1080 https://example.com
```
## Troubleshooting
| Symptom | Check |
|---------|-------|
| Connection refused | Is Tor running? `ss -tlnp \| grep 9050` |
| Timeout | Increase `-t`, check proxy reachability |
| DNS leak | Use `socks5h://` (not `socks5://`) in client |
| Auth failed | Verify credentials in proxy URL |

39
docs/INSTALL.md Normal file
View File

@@ -0,0 +1,39 @@
# s5p -- Installation
## Prerequisites
- Python >= 3.11
- pip
- Tor (optional, for Tor-based chains)
## Install
```bash
cd ~/git/s5p
python -m venv .venv
source .venv/bin/activate
pip install -e .
```
## Verify
```bash
s5p --version
which s5p
```
## Install Tor (optional)
```bash
sudo apt install tor
sudo systemctl enable --now tor
# Verify Tor SOCKS5 port
ss -tlnp | grep 9050
```
## Symlink (alternative)
```bash
ln -sf ~/git/s5p/.venv/bin/s5p ~/.local/bin/s5p
```

73
docs/USAGE.md Normal file
View File

@@ -0,0 +1,73 @@
# s5p -- Usage
## Basic Usage
```bash
# Direct proxy (no chain, just a SOCKS5 server)
s5p
# Through Tor
s5p -C socks5://127.0.0.1:9050
# Through Tor + another proxy
s5p -C socks5://127.0.0.1:9050,socks5://proxy:1080
# Custom listen address
s5p -l 0.0.0.0:9999 -C socks5://127.0.0.1:9050
# From config file
s5p -c config/example.yaml
# Debug mode
s5p -v -C socks5://127.0.0.1:9050
```
## Config File
```yaml
listen: 127.0.0.1:1080
timeout: 10
log_level: info
chain:
- socks5://127.0.0.1:9050
- http://user:pass@proxy:8080
```
## Proxy URL Format
```
protocol://[username:password@]host[:port]
```
| Protocol | Default Port | Auth Support |
|----------|-------------|-------------|
| socks5 | 1080 | username/password |
| socks4 | 1080 | none |
| http | 8080 | Basic |
## Testing the Proxy
```bash
# Check exit IP via Tor
curl --proxy socks5h://127.0.0.1:1080 https://check.torproject.org/api/ip
# Fetch a page
curl --proxy socks5h://127.0.0.1:1080 https://example.com
# Use with Firefox: set SOCKS5 proxy to 127.0.0.1:1080, enable remote DNS
```
Note: use `socks5h://` (not `socks5://`) with curl to send DNS through the proxy.
## Chain Order
Hops are traversed left-to-right:
```
-C hop1,hop2,hop3
Client -> s5p -> hop1 -> hop2 -> hop3 -> Destination
```
Each hop only sees its immediate neighbors.

26
pyproject.toml Normal file
View File

@@ -0,0 +1,26 @@
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"
[project]
name = "s5p"
version = "0.1.0"
description = "SOCKS5 proxy with Tor and proxy-chain support"
requires-python = ">=3.11"
dependencies = ["pyyaml>=6.0"]
[project.scripts]
s5p = "s5p.cli:main"
[tool.setuptools.packages.find]
where = ["src"]
[tool.ruff]
target-version = "py311"
line-length = 100
[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP"]
[tool.pytest.ini_options]
testpaths = ["tests"]

3
src/s5p/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
"""s5p -- SOCKS5 proxy with chain support."""
__version__ = "0.1.0"

7
src/s5p/__main__.py Normal file
View File

@@ -0,0 +1,7 @@
"""Allow running as: python -m s5p."""
import sys
from .cli import main
sys.exit(main())

83
src/s5p/cli.py Normal file
View File

@@ -0,0 +1,83 @@
"""Command-line interface for s5p."""
from __future__ import annotations
import argparse
import asyncio
import logging
from . import __version__
from .config import Config, load_config, parse_proxy_url
from .server import serve
def _setup_logging(level: str) -> None:
"""Configure logging with a compact muted format."""
numeric = getattr(logging, level.upper(), logging.INFO)
fmt = "\033[2m%(asctime)s\033[0m %(levelname).1s %(message)s"
logging.basicConfig(level=numeric, format=fmt, datefmt="%H:%M:%S")
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
"""Parse command-line arguments."""
p = argparse.ArgumentParser(
prog="s5p",
description="SOCKS5 proxy with Tor and proxy-chain support",
)
p.add_argument(
"-V", "--version", action="version", version=f"s5p {__version__}"
)
p.add_argument("-c", "--config", metavar="FILE", help="YAML config file")
p.add_argument(
"-l", "--listen", metavar="[HOST:]PORT",
help="listen address (default: 127.0.0.1:1080)",
)
p.add_argument(
"-C", "--chain", metavar="URL[,URL,...]",
help="comma-separated proxy chain URLs",
)
p.add_argument(
"-t", "--timeout", type=float, metavar="SEC",
help="per-hop timeout in seconds (default: 10)",
)
p.add_argument("-v", "--verbose", action="store_true", help="debug logging")
p.add_argument("-q", "--quiet", action="store_true", help="errors only")
return p.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
"""Entry point."""
args = _parse_args(argv)
config = load_config(args.config) if args.config else Config()
if args.listen:
if ":" in args.listen:
host, port_str = args.listen.rsplit(":", 1)
config.listen_host = host
config.listen_port = int(port_str)
else:
config.listen_port = int(args.listen)
if args.chain:
config.chain = [parse_proxy_url(u.strip()) for u in args.chain.split(",")]
if args.timeout is not None:
config.timeout = args.timeout
if args.verbose:
config.log_level = "debug"
elif args.quiet:
config.log_level = "error"
_setup_logging(config.log_level)
try:
asyncio.run(serve(config))
except KeyboardInterrupt:
return 0
except Exception as e:
logging.getLogger("s5p").error("%s", e)
return 1
return 0

101
src/s5p/config.py Normal file
View File

@@ -0,0 +1,101 @@
"""Configuration loading and proxy URL parsing."""
from __future__ import annotations
from dataclasses import dataclass, field
from pathlib import Path
from urllib.parse import urlparse
import yaml
DEFAULT_PORTS = {"socks5": 1080, "socks4": 1080, "http": 8080}
@dataclass
class ChainHop:
"""A single proxy hop in the chain."""
proto: str
host: str
port: int
username: str | None = None
password: str | None = None
def __str__(self) -> str:
auth = f"{self.username}@" if self.username else ""
return f"{self.proto}://{auth}{self.host}:{self.port}"
@dataclass
class Config:
"""Server configuration."""
listen_host: str = "127.0.0.1"
listen_port: int = 1080
chain: list[ChainHop] = field(default_factory=list)
timeout: float = 10.0
log_level: str = "info"
def parse_proxy_url(url: str) -> ChainHop:
"""Parse a proxy URL like socks5://user:pass@host:port."""
parsed = urlparse(url)
proto = parsed.scheme.lower()
if proto not in ("socks5", "socks4", "http"):
raise ValueError(f"unsupported protocol: {proto}")
host = parsed.hostname
if not host:
raise ValueError(f"missing host in proxy URL: {url}")
port = parsed.port or DEFAULT_PORTS.get(proto, 1080)
return ChainHop(
proto=proto,
host=host,
port=port,
username=parsed.username,
password=parsed.password,
)
def load_config(path: str | Path) -> Config:
"""Load configuration from a YAML file."""
path = Path(path)
with path.open() as f:
raw = yaml.safe_load(f) or {}
config = Config()
if "listen" in raw:
listen = raw["listen"]
if isinstance(listen, str) and ":" in listen:
host, port_str = listen.rsplit(":", 1)
config.listen_host = host
config.listen_port = int(port_str)
elif isinstance(listen, (str, int)):
config.listen_port = int(listen)
if "timeout" in raw:
config.timeout = float(raw["timeout"])
if "log_level" in raw:
config.log_level = raw["log_level"]
if "chain" in raw:
for entry in raw["chain"]:
if isinstance(entry, str):
config.chain.append(parse_proxy_url(entry))
elif isinstance(entry, dict):
config.chain.append(
ChainHop(
proto=entry.get("proto", "socks5"),
host=entry["host"],
port=int(entry["port"]),
username=entry.get("username"),
password=entry.get("password"),
)
)
return config

183
src/s5p/proto.py Normal file
View File

@@ -0,0 +1,183 @@
"""Protocol handshake implementations for SOCKS4/4a, SOCKS5, and HTTP CONNECT."""
from __future__ import annotations
import asyncio
import base64
import socket
import struct
from enum import IntEnum
class Socks5Reply(IntEnum):
"""SOCKS5 reply codes (RFC 1928)."""
SUCCEEDED = 0x00
GENERAL_FAILURE = 0x01
NOT_ALLOWED = 0x02
NETWORK_UNREACHABLE = 0x03
HOST_UNREACHABLE = 0x04
CONNECTION_REFUSED = 0x05
TTL_EXPIRED = 0x06
COMMAND_NOT_SUPPORTED = 0x07
ADDRESS_TYPE_NOT_SUPPORTED = 0x08
class Socks5AddrType(IntEnum):
"""SOCKS5 address types."""
IPV4 = 0x01
DOMAIN = 0x03
IPV6 = 0x04
class ProtoError(Exception):
"""Protocol negotiation error."""
def __init__(self, message: str, reply: int = Socks5Reply.GENERAL_FAILURE):
super().__init__(message)
self.reply = reply
def encode_address(host: str) -> tuple[int, bytes]:
"""Encode host as SOCKS5 address. Returns (atyp, encoded_bytes)."""
try:
return Socks5AddrType.IPV4, socket.inet_aton(host)
except OSError:
pass
try:
return Socks5AddrType.IPV6, socket.inet_pton(socket.AF_INET6, host)
except OSError:
pass
encoded = host.encode("ascii")
return Socks5AddrType.DOMAIN, bytes([len(encoded)]) + encoded
async def read_socks5_address(
reader: asyncio.StreamReader,
) -> tuple[str, int]:
"""Read a SOCKS5 address (atyp + addr + port) from the stream."""
atyp = (await reader.readexactly(1))[0]
if atyp == Socks5AddrType.IPV4:
addr = socket.inet_ntoa(await reader.readexactly(4))
elif atyp == Socks5AddrType.DOMAIN:
length = (await reader.readexactly(1))[0]
addr = (await reader.readexactly(length)).decode("ascii")
elif atyp == Socks5AddrType.IPV6:
addr = socket.inet_ntop(socket.AF_INET6, await reader.readexactly(16))
else:
raise ProtoError(
f"unsupported address type: {atyp:#x}",
Socks5Reply.ADDRESS_TYPE_NOT_SUPPORTED,
)
port = struct.unpack("!H", await reader.readexactly(2))[0]
return addr, port
async def socks5_connect(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
host: str,
port: int,
username: str | None = None,
password: str | None = None,
) -> None:
"""Negotiate a SOCKS5 CONNECT through the given stream."""
methods = b"\x00\x02" if username and password else b"\x00"
writer.write(bytes([0x05, len(methods)]) + methods)
await writer.drain()
data = await reader.readexactly(2)
if data[0] != 0x05:
raise ProtoError(f"socks5: unexpected version {data[0]:#x}")
method = data[1]
if method == 0x02:
if not (username and password):
raise ProtoError("socks5: server requires auth but no credentials")
uname = username.encode("utf-8")
passwd = password.encode("utf-8")
writer.write(
bytes([0x01, len(uname)]) + uname + bytes([len(passwd)]) + passwd
)
await writer.drain()
auth_resp = await reader.readexactly(2)
if auth_resp[1] != 0x00:
raise ProtoError("socks5: authentication failed")
elif method == 0xFF:
raise ProtoError("socks5: no acceptable authentication methods")
elif method != 0x00:
raise ProtoError(f"socks5: unsupported method {method:#x}")
atyp, addr_bytes = encode_address(host)
writer.write(
struct.pack("!BBB", 0x05, 0x01, 0x00)
+ bytes([atyp])
+ addr_bytes
+ struct.pack("!H", port)
)
await writer.drain()
resp = await reader.readexactly(3) # VER, REP, RSV
if resp[1] != 0x00:
raise ProtoError(f"socks5: connect failed (reply={resp[1]:#x})", resp[1])
await read_socks5_address(reader)
async def socks4_connect(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
host: str,
port: int,
) -> None:
"""Negotiate a SOCKS4/4a CONNECT through the given stream."""
port_bytes = struct.pack("!H", port)
try:
ip_bytes = socket.inet_aton(host)
writer.write(b"\x04\x01" + port_bytes + ip_bytes + b"\x00")
except OSError:
# SOCKS4a: sentinel IP 0.0.0.1, then domain after userid null
writer.write(
b"\x04\x01"
+ port_bytes
+ b"\x00\x00\x00\x01\x00"
+ host.encode("ascii")
+ b"\x00"
)
await writer.drain()
resp = await reader.readexactly(8)
if resp[1] != 0x5A:
raise ProtoError(f"socks4: request rejected (status={resp[1]:#x})")
async def http_connect(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
host: str,
port: int,
username: str | None = None,
password: str | None = None,
) -> None:
"""Negotiate an HTTP CONNECT tunnel through the given stream."""
request = f"CONNECT {host}:{port} HTTP/1.1\r\nHost: {host}:{port}\r\n"
if username and password:
creds = base64.b64encode(f"{username}:{password}".encode()).decode()
request += f"Proxy-Authorization: Basic {creds}\r\n"
request += "\r\n"
writer.write(request.encode())
await writer.drain()
line = await reader.readline()
if not line:
raise ProtoError("http: empty response")
parts = line.decode("utf-8", errors="replace").split(None, 2)
if len(parts) < 2 or not parts[1].startswith("2"):
raise ProtoError(
f"http: connect failed: {line.decode('utf-8', errors='replace').strip()}"
)
while True:
header_line = await reader.readline()
if header_line in (b"\r\n", b"\n", b""):
break

235
src/s5p/server.py Normal file
View File

@@ -0,0 +1,235 @@
"""SOCKS5 proxy server with proxy-chain support."""
from __future__ import annotations
import asyncio
import logging
import struct
import time
from .config import ChainHop, Config
from .proto import (
ProtoError,
Socks5Reply,
http_connect,
read_socks5_address,
socks4_connect,
socks5_connect,
)
logger = logging.getLogger("s5p")
BUFFER_SIZE = 65536
# -- relay -------------------------------------------------------------------
async def _relay(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
) -> None:
"""Unidirectional data relay."""
try:
while True:
data = await reader.read(BUFFER_SIZE)
if not data:
break
writer.write(data)
await writer.drain()
except (ConnectionError, asyncio.CancelledError, OSError):
pass
finally:
try:
writer.close()
await writer.wait_closed()
except (OSError, ConnectionError):
pass
# -- chain building ----------------------------------------------------------
async def _negotiate_hop(
reader: asyncio.StreamReader,
writer: asyncio.StreamWriter,
hop: ChainHop,
dest_host: str,
dest_port: int,
) -> None:
"""Negotiate a single hop in the chain."""
if hop.proto == "socks5":
await socks5_connect(reader, writer, dest_host, dest_port, hop.username, hop.password)
elif hop.proto == "socks4":
await socks4_connect(reader, writer, dest_host, dest_port)
elif hop.proto == "http":
await http_connect(reader, writer, dest_host, dest_port, hop.username, hop.password)
else:
raise ProtoError(f"unsupported protocol: {hop.proto}")
async def build_chain(
chain: list[ChainHop],
target_host: str,
target_port: int,
timeout: float = 10.0,
) -> tuple[asyncio.StreamReader, asyncio.StreamWriter]:
"""Build a tunnel through the proxy chain to the target.
Connects to the first hop via TCP, then negotiates each subsequent
hop over the tunnel established by the previous one.
"""
if not chain:
return await asyncio.wait_for(
asyncio.open_connection(target_host, target_port),
timeout=timeout,
)
reader, writer = await asyncio.wait_for(
asyncio.open_connection(chain[0].host, chain[0].port),
timeout=timeout,
)
try:
for i, hop in enumerate(chain):
if i + 1 < len(chain):
dest_host = chain[i + 1].host
dest_port = chain[i + 1].port
else:
dest_host = target_host
dest_port = target_port
await asyncio.wait_for(
_negotiate_hop(reader, writer, hop, dest_host, dest_port),
timeout=timeout,
)
logger.debug(
"hop %d/%d ok %s -> %s:%d",
i + 1,
len(chain),
hop.proto,
dest_host,
dest_port,
)
except Exception:
writer.close()
raise
return reader, writer
# -- SOCKS5 server -----------------------------------------------------------
def _socks5_reply(rep: int) -> bytes:
"""Build a minimal SOCKS5 reply packet."""
return struct.pack("!BBB", 0x05, rep, 0x00) + b"\x01\x00\x00\x00\x00\x00\x00"
async def _handle_client(
client_reader: asyncio.StreamReader,
client_writer: asyncio.StreamWriter,
config: Config,
) -> None:
"""Handle a single SOCKS5 client connection."""
peer = client_writer.get_extra_info("peername")
tag = f"{peer[0]}:{peer[1]}" if peer else "?"
try:
# -- greeting --
header = await asyncio.wait_for(client_reader.readexactly(2), timeout=10.0)
if header[0] != 0x05:
logger.warning("[%s] bad socks version: %d", tag, header[0])
return
methods = await client_reader.readexactly(header[1])
if 0x00 not in methods:
client_writer.write(b"\x05\xff")
await client_writer.drain()
return
client_writer.write(b"\x05\x00")
await client_writer.drain()
# -- connect request --
req = await asyncio.wait_for(client_reader.readexactly(3), timeout=10.0)
if req[0] != 0x05:
return
if req[1] != 0x01:
client_writer.write(_socks5_reply(Socks5Reply.COMMAND_NOT_SUPPORTED))
await client_writer.drain()
return
target_host, target_port = await read_socks5_address(client_reader)
logger.info("[%s] connect %s:%d", tag, target_host, target_port)
# -- build chain --
t0 = time.monotonic()
remote_reader, remote_writer = await build_chain(
config.chain, target_host, target_port, timeout=config.timeout
)
dt = time.monotonic() - t0
logger.debug("[%s] chain up in %.0fms", tag, dt * 1000)
# -- success --
client_writer.write(_socks5_reply(Socks5Reply.SUCCEEDED))
await client_writer.drain()
# -- relay --
await asyncio.gather(
_relay(client_reader, remote_writer),
_relay(remote_reader, client_writer),
)
except ProtoError as e:
logger.warning("[%s] %s", tag, e)
try:
client_writer.write(_socks5_reply(e.reply))
await client_writer.drain()
except OSError:
pass
except asyncio.TimeoutError:
logger.warning("[%s] timeout", tag)
try:
client_writer.write(_socks5_reply(Socks5Reply.TTL_EXPIRED))
await client_writer.drain()
except OSError:
pass
except (ConnectionError, OSError) as e:
logger.debug("[%s] %s", tag, e)
try:
client_writer.write(_socks5_reply(Socks5Reply.CONNECTION_REFUSED))
await client_writer.drain()
except OSError:
pass
except Exception:
logger.exception("[%s] unexpected error", tag)
finally:
try:
client_writer.close()
await client_writer.wait_closed()
except OSError:
pass
# -- entry point -------------------------------------------------------------
async def serve(config: Config) -> None:
"""Start the SOCKS5 proxy server."""
async def on_client(r: asyncio.StreamReader, w: asyncio.StreamWriter) -> None:
await _handle_client(r, w, config)
srv = await asyncio.start_server(on_client, config.listen_host, config.listen_port)
addrs = ", ".join(str(s.getsockname()) for s in srv.sockets)
logger.info("listening on %s", addrs)
if config.chain:
for i, hop in enumerate(config.chain):
logger.info(" chain[%d] %s", i, hop)
else:
logger.info(" mode: direct (no chain)")
async with srv:
await srv.serve_forever()

0
tests/__init__.py Normal file
View File

76
tests/test_config.py Normal file
View File

@@ -0,0 +1,76 @@
"""Tests for configuration loading and proxy URL parsing."""
import pytest
from s5p.config import ChainHop, Config, parse_proxy_url
class TestParseProxyUrl:
"""Test proxy URL parsing."""
def test_socks5_basic(self):
hop = parse_proxy_url("socks5://127.0.0.1:9050")
assert hop.proto == "socks5"
assert hop.host == "127.0.0.1"
assert hop.port == 9050
assert hop.username is None
assert hop.password is None
def test_socks5_with_auth(self):
hop = parse_proxy_url("socks5://user:pass@proxy.example.com:1080")
assert hop.proto == "socks5"
assert hop.host == "proxy.example.com"
assert hop.port == 1080
assert hop.username == "user"
assert hop.password == "pass"
def test_socks4(self):
hop = parse_proxy_url("socks4://10.0.0.1:1080")
assert hop.proto == "socks4"
assert hop.host == "10.0.0.1"
assert hop.port == 1080
def test_http_connect(self):
hop = parse_proxy_url("http://proxy:8080")
assert hop.proto == "http"
assert hop.host == "proxy"
assert hop.port == 8080
def test_default_port_socks5(self):
hop = parse_proxy_url("socks5://host")
assert hop.port == 1080
def test_default_port_http(self):
hop = parse_proxy_url("http://host")
assert hop.port == 8080
def test_unsupported_protocol(self):
with pytest.raises(ValueError, match="unsupported protocol"):
parse_proxy_url("ftp://host:21")
def test_missing_host(self):
with pytest.raises(ValueError, match="missing host"):
parse_proxy_url("socks5://")
class TestChainHop:
"""Test ChainHop string representation."""
def test_str_without_auth(self):
hop = ChainHop(proto="socks5", host="localhost", port=9050)
assert str(hop) == "socks5://localhost:9050"
def test_str_with_auth(self):
hop = ChainHop(proto="http", host="proxy", port=8080, username="u", password="p")
assert str(hop) == "http://u@proxy:8080"
class TestConfig:
"""Test Config defaults."""
def test_defaults(self):
c = Config()
assert c.listen_host == "127.0.0.1"
assert c.listen_port == 1080
assert c.chain == []
assert c.timeout == 10.0

22
tests/test_proto.py Normal file
View File

@@ -0,0 +1,22 @@
"""Tests for protocol helpers."""
from s5p.proto import Socks5AddrType, encode_address
class TestEncodeAddress:
"""Test SOCKS5 address encoding."""
def test_ipv4(self):
atyp, data = encode_address("127.0.0.1")
assert atyp == Socks5AddrType.IPV4
assert data == b"\x7f\x00\x00\x01"
def test_ipv6(self):
atyp, data = encode_address("::1")
assert atyp == Socks5AddrType.IPV6
assert len(data) == 16
def test_domain(self):
atyp, data = encode_address("example.com")
assert atyp == Socks5AddrType.DOMAIN
assert data == bytes([11]) + b"example.com"