Building a SOCKS5 Proxy Server in Python: A Complete Guide
# Building a SOCKS5 Proxy Server in Python: A Complete Guide
Ever wondered how proxy servers work behind the scenes? In this tutorial, we'll build a fully functional **SOCKS5 proxy server** from scratch using Python. By the end, you'll understand the SOCKS5 protocol and have a working proxy server you can use!
---
## 📖 What is a SOCKS5 Proxy?
**SOCKS5** (Socket Secure version 5) is a network protocol that routes your internet traffic through a proxy server. Unlike HTTP proxies, SOCKS5 works at a lower level and can handle any type of traffic — web browsing, emails, file transfers, and more.
### Key Features of SOCKS5:
- ✅ **Protocol Agnostic** — Works with any application protocol
- ✅ **Authentication Support** — Username/password protection
- ✅ **IPv4; IPv6 & Domain Support** — Flexible addressing
- ✅ **Better Performance** — Less overhead than HTTP proxies
---
## 🛠️ Prerequisites
Before we begin, make sure you have:
- Python 3.6 or higher installed
- Basic understanding of Python sockets
- A code editor of your choice
No external libraries required! We'll use only Python's built-in modules:
- `socket` — For network connections
- `threading` — For handling multiple clients
- `select` — For monitoring socket activity
---
## 🏗️ Project Structure
Our proxy server consists of a single `Proxy` class with the following methods:
```
Proxy
├── __init__() # Initialize credentials
├── run() # Start the server
├── handle_client() # Process client connections
├── verify_credentials() # Authenticate users
├── get_available_methods() # Parse auth methods
├── exchange_loop() # Transfer data between client and remote
└── generate_failed_reply() # Create error responses
```
---
## 📝 Step-by-Step Implementation
### Step 1: Setting Up the Foundation
Let's start by importing the required modules and setting up our `Proxy` class:
```python
import socket
import threading
import select
SOCKS_VERSION = 5
class Proxy:
def __init__(self):
self.username = "username"
self.password = "password"
```
**What's happening here?**
- `SOCKS_VERSION = 5` — Defines we're using SOCKS version 5
- The `__init__` method sets up default credentials for authentication
> 💡 **Tip:** In production, load credentials from environment variables or a config file instead of hardcoding them!
---
### Step 2: Starting the Server
The `run()` method creates a TCP server that listens for incoming connections:
```python
def run(self, host, port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((host, port))
s.listen()
print("* Socks5 proxy server is running on {}:{}".format(host, port))
while True:
conn, addr = s.accept()
print("* new connection from {}".format(addr))
t = threading.Thread(target=self.handle_client, args=(conn,))
t.start()
```
**Breaking it down:**
| Line | Purpose |
|------|---------|
| `socket.socket(AF_INET, SOCK_STREAM)` | Creates a TCP socket for IPv4 |
| `s.bind((host, port))` | Binds the socket to the specified address |
| `s.listen()` | Starts listening for connections |
| `s.accept()` | Waits for and accepts a new connection |
| `threading.Thread(...)` | Handles each client in a separate thread |
**Why use threading?**
Threading allows our server to handle multiple clients simultaneously. Without it, only one client could connect at a time!
---
### Step 3: The SOCKS5 Handshake
When a client connects, the SOCKS5 handshake begins. Here's the flow:
```
┌────────┐ ┌────────┐
│ Client │ │ Server │
└───┬────┘ └───┬────┘
│ │
│─── 1. Greeting (version + methods) ─→│
│ │
│←── 2. Method Selection ────────────│
│ │
│─── 3. Username/Password ──────────→│
│ │
│←── 4. Auth Result ─────────────────│
│ │
│─── 5. Connection Request ─────────→│
│ │
│←── 6. Connection Response ─────────│
│ │
│←──── 7. Data Exchange ────────────→│
│ │
```
---
### Step 4: Handling Client Connections
The `handle_client()` method is the heart of our proxy. Let's break it into parts:
#### Part A: Reading the Greeting
```python
def handle_client(self, connection):
# Read version and number of authentication methods
version, nmethods = connection.recv(2)
# Get the list of available authentication methods
methods = self.get_available_methods(nmethods, connection)
# We only accept username/password authentication (method 2)
if 2 not in set(methods):
connection.close()
return
# Tell client we want username/password auth
connection.sendall(bytes([SOCKS_VERSION, 2]))
```
**Authentication Methods:**
| Method Code | Meaning |
|-------------|---------|
| `0` | No authentication required |
| `1` | GSSAPI |
| `2` | Username/Password |
| `255` | No acceptable methods |
#### Part B: Processing Connection Requests
After authentication, the client sends a connection request:
```python
# Read connection request
version, cmd, _, address_type = connection.recv(4)
if address_type == 1: # IPv4
address = socket.inet_ntoa(connection.recv(4))
elif address_type == 3: # Domain name
domain_length = connection.recv(1)[0]
address = connection.recv(domain_length)
address = socket.gethostbyname(address)
# Read port number (2 bytes, big-endian)
port = int.from_bytes(connection.recv(2), 'big', signed=False)
```
**Address Types:**
| Type | Description |
|------|-------------|
| `1` | IPv4 address (4 bytes) |
| `3` | Domain name (variable length) |
| `4` | IPv6 address (16 bytes) |
#### Part C: Establishing the Connection
```python
try:
if cmd == 1: # CONNECT command
remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote.connect((address, port))
bind_address = remote.getsockname()
print("* Connected to {} {}".format(address, port))
else:
connection.close()
# Build success response
addr = int.from_bytes(socket.inet_aton(bind_address[0]), 'big', signed=False)
port = bind_address[1]
reply = b''.join([
SOCKS_VERSION.to_bytes(1, 'big'),
int(0).to_bytes(1, 'big'), # Success status
int(0).to_bytes(1, 'big'), # Reserved
int(1).to_bytes(1, 'big'), # Address type (IPv4)
addr.to_bytes(4, 'big'), # Bound address
port.to_bytes(2, 'big') # Bound port
])
except Exception as e:
reply = self.generate_failed_reply(address_type, 5)
connection.sendall(reply)
```
---
### Step 5: Authentication Verification
The `verify_credentials()` method checks the username and password:
```python
def verify_credentials(self, connection):
version = ord(connection.recv(1)) # Should be 1
# Read username
username_len = ord(connection.recv(1))
username = connection.recv(username_len).decode('utf-8')
# Read password
password_len = ord(connection.recv(1))
password = connection.recv(password_len).decode('utf-8')
if username == self.username and password == self.password:
# Success! Status = 0
response = bytes([version, 0])
connection.sendall(response)
return True
# Failure! Status != 0
response = bytes([version, 0xFF])
connection.sendall(response)
connection.close()
return False
```
**Response Status Codes:**
- `0` — Success
- Any other value — Failure
---
### Step 6: Data Exchange Loop
Once connected, data flows between the client and remote server:
```python
def exchange_loop(self, client, remote):
while True:
# Wait for data from either socket
r, w, e = select.select([client, remote], [], [])
if client in r:
data = client.recv(4096)
if remote.send(data) <= 0:
break
if remote in r:
data = remote.recv(4096)
if client.send(data) <= 0:
break
```
**How `select.select()` works:**
- Monitors multiple sockets for activity
- Returns when any socket has data ready to read
- More efficient than constantly checking each socket
---
### Step 7: Helper Methods
#### Getting Available Authentication Methods
```python
def get_available_methods(self, nmethods, connection):
methods = []
for i in range(nmethods):
methods.append(ord(connection.recv(1)))
return methods
```
#### Generating Error Responses
```python
def generate_failed_reply(self, address_type, error_number):
return b''.join([
SOCKS_VERSION.to_bytes(1, 'big'),
error_number.to_bytes(1, 'big'),
int(0).to_bytes(1, 'big'),
address_type.to_bytes(1, 'big'),
int(0).to_bytes(4, 'big'),
int(0).to_bytes(4, 'big')
])
```
**Common Error Codes:**
| Code | Meaning |
|------|---------|
| `1` | General SOCKS server failure |
| `2` | Connection not allowed |
| `3` | Network unreachable |
| `4` | Host unreachable |
| `5` | Connection refused |
| `7` | Command not supported |
| `8` | Address type not supported |
---
## 🚀 Running the Proxy Server
### Starting the Server
```python
if __name__ == "__main__":
proxy = Proxy()
proxy.run("127.0.0.1", 3000)
```
Run the script:
```bash
python socks5_proxy.py
```
You should see:
```
* Socks5 proxy server is running on 127.0.0.1:3000
```
---
## 🔧 Configuring Your Applications
### Browser Configuration (Firefox)
1. Open **Settings** → **Network Settings**
2. Select **Manual proxy configuration**
3. Set **SOCKS Host**: `127.0.0.1`
4. Set **Port**: `3000`
5. Select **SOCKS v5**
6. Enter your username and password if prompted
### Using with cURL
```bash
curl --socks5 127.0.0.1:3000 --proxy-user username:password https://example.com
```
### Using with Python Requests
```python
import requests
proxies = {
'http': 'socks5://username:[email protected]:3000',
'https': 'socks5://username:[email protected]:3000'
}
response = requests.get('https://example.com', proxies=proxies)
print(response.text)
```
> **Note:** You'll need to install `requests[socks]` for SOCKS support:
> ```bash
> pip install requests[socks]
> ```
---
## 📊 Complete Code
Here's the full implementation:
```python
import socket
import threading
import select
SOCKS_VERSION = 5
class Proxy:
def __init__(self):
self.username = "username"
self.password = "password"
def handle_client(self, connection):
# greeting header
# read and unpack 2 bytes from a client
version, nmethods = connection.recv(2)
# get available methods [0, 1, 2]
methods = self.get_available_methods(nmethods, connection)
# accept only USERNAME/PASSWORD auth
if 2 not in set(methods):
# close connection
connection.close()
return
# send welcome message
connection.sendall(bytes([SOCKS_VERSION, 2]))
if not self.verify_credentials(connection):
return
# request (version=5)
version, cmd, _, address_type = connection.recv(4)
if address_type == 1: # IPv4
address = socket.inet_ntoa(connection.recv(4))
elif address_type == 3: # Domain name
domain_length = connection.recv(1)[0]
address = connection.recv(domain_length)
address = socket.gethostbyname(address)
# convert bytes to unsigned short array
port = int.from_bytes(connection.recv(2), 'big', signed=False)
try:
if cmd == 1: # CONNECT
remote = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote.connect((address, port))
bind_address = remote.getsockname()
print("* Connected to {} {}".format(address, port))
else:
connection.close()
addr = int.from_bytes(socket.inet_aton(bind_address[0]), 'big', signed=False)
port = bind_address[1]
reply = b''.join([
SOCKS_VERSION.to_bytes(1, 'big'),
int(0).to_bytes(1, 'big'),
int(0).to_bytes(1, 'big'),
int(1).to_bytes(1, 'big'),
addr.to_bytes(4, 'big'),
port.to_bytes(2, 'big')
])
except Exception as e:
# return connection refused error
reply = self.generate_failed_reply(address_type, 5)
connection.sendall(reply)
# establish data exchange
if reply[1] == 0 and cmd == 1:
self.exchange_loop(connection, remote)
connection.close()
def exchange_loop(self, client, remote):
while True:
# wait until client or remote is available for read
r, w, e = select.select([client, remote], [], [])
if client in r:
data = client.recv(4096)
if remote.send(data) <= 0:
break
if remote in r:
data = remote.recv(4096)
if client.send(data) <= 0:
break
def generate_failed_reply(self, address_type, error_number):
return b''.join([
SOCKS_VERSION.to_bytes(1, 'big'),
error_number.to_bytes(1, 'big'),
int(0).to_bytes(1, 'big'),
address_type.to_bytes(1, 'big'),
int(0).to_bytes(4, 'big'),
int(0).to_bytes(4, 'big')
])
def verify_credentials(self, connection):
version = ord(connection.recv(1)) # should be 1
username_len = ord(connection.recv(1))
username = connection.recv(username_len).decode('utf-8')
password_len = ord(connection.recv(1))
password = connection.recv(password_len).decode('utf-8')
if username == self.username and password == self.password:
# success, status = 0
response = bytes([version, 0])
connection.sendall(response)
return True
# failure, status != 0
response = bytes([version, 0xFF])
connection.sendall(response)
connection.close()
return False
def get_available_methods(self, nmethods, connection):
methods = []
for i in range(nmethods):
methods.append(ord(connection.recv(1)))
return methods
def run(self, host, port):
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.bind((host, port))
s.listen()
print("* Socks5 proxy server is running on {}:{}".format(host, port))
while True:
conn, addr = s.accept()
print("* new connection from {}".format(addr))
t = threading.Thread(target=self.handle_client, args=(conn,))
t.start()
if __name__ == "__main__":
proxy = Proxy()
proxy.run("127.0.0.1", 3000)
```
---
## ⚠️ Security Considerations
Before deploying this proxy server, keep these points in mind:
1. **Change Default Credentials** — Never use the default `username`/`password` in production
2. **Use TLS/SSL** — Consider wrapping connections in SSL for encrypted communication
3. **Implement Rate Limiting** — Prevent abuse by limiting connections per IP
4. **Add Logging** — Track usage and detect suspicious activity
5. **Firewall Rules** — Only allow connections from trusted IP addresses
---
## 🎯 Possible Enhancements
Want to take this further? Here are some ideas:
- [ ] Add IPv6 support
- [ ] Implement UDP ASSOCIATE command
- [ ] Add connection timeout handling
- [ ] Create a configuration file for settings
- [ ] Add logging with different verbosity levels
- [ ] Implement access control lists (ACL)
- [ ] Add support for no-authentication mode
---
## 📚 Conclusion
Congratulations! You've built a fully functional SOCKS5 proxy server in Python. You now understand:
- ✅ How the SOCKS5 protocol works
- ✅ The handshake and authentication process
- ✅ How to handle multiple clients with threading
- ✅ How to relay data between client and remote servers
This proxy server is great for learning and development purposes. For production use, consider adding robust error handling, logging, and security features.
---
## 📎 References
- [RFC 1928 - SOCKS Protocol Version 5](https://tools.ietf.org/html/rfc1928)
- [RFC 1929 - Username/Password Authentication for SOCKS V5](https://tools.ietf.org/html/rfc1929)
- [Python Socket Documentation](https://docs.python.org/3/library/socket.html)
---
*Happy Coding! 🐍*
Comments (0)