This course focuses on setting up a Safe Shopper-Server Protocol utilizing the Python programming language, together with safety for sockets, framing, and key alternate. In case you ever needed to implement your individual safe communication protocol in Python, utilizing solely the usual Python socket library and the cryptography library, that is methods to do it.
To be clear up entrance, this isn’t an entire VPN. No TUN/TAP interfaces or routing of IP visitors is concerned right here. What we’re setting up is a safe client-server protocol — the constructing block upon which an actual VPN, chat utility, or any customized community service utilizing encrypted, authenticated communication sits.
What We’re Constructing
On the finish of this tutorial, you will have a framed message-sending consumer and server that negotiate a brand new encryption key for every session and detect tampering. It isn’t a tunnel — there is no OS-level forwarding or digital community interface concerned.
Design Targets
This differs from a toy socket instance in 4 methods:
- Correct message framing. TCP treats information as a stream, not discrete messages, so we outline our personal message boundaries.
- Actual key alternate. Each session negotiates its personal key reasonably than utilizing a hard and fast key hardcoded into this system.
- Multi-client assist. The server handles a number of shoppers concurrently.
- Visibility into wire visitors. We’ll examine the bytes going over the community and make sure they’re truly encrypted.
Undertaking Setup
python3 -m venv venv
supply venv/bin/activate
pip set up cryptography
Create a server.py and a consumer.py. We’ll construct them piece by piece.
Message Framing in Python Sockets
Receiving an Actual Variety of Bytes
socket.recv() can return fewer bytes than requested. This helper loops till it has precisely what it wants:
def recv_exact(sock, n):
buf = b''
whereas len(buf) < n:
chunk = sock.recv(n - len(buf))
if not chunk:
increase ConnectionError('socket closed earlier than we obtained all the things')
buf += chunk
return buf
Size-Prefixing Messages
Every message is prefixed with its size as 4 bytes (big-endian).
A blind 4-byte size prefix is okay for a demo, however as written it lets a peer declare an arbitrarily giant payload (as much as ~4GB) and pressure the receiver to allocate that a lot reminiscence earlier than a single byte of precise information arrives — a straightforward denial-of-service vector. We cap it right here at 10MB; regulate to no matter your actual message measurement ceiling is.
import struct
**MAX_MESSAGE_SIZE = 10 * 1024 * 1024
def send_framed(sock, payload):
sock.sendall(struct.pack('>I', len(payload)) + payload)
def recv_framed(sock):
header = recv_exact(sock, 4)
size = struct.unpack('>I', header)[0]
**if size > MAX_MESSAGE_SIZE:
increase ValueError(f'peer claims a {size}-byte message, refusing (max {MAX_MESSAGE_SIZE})')**
return recv_exact(sock, size)
Testing With a Giant Payload
A fast echo take a look at confirms the framing logic handles a payload bigger than what a single recv() name would return.
import socket
import threading
def echo_once(conn):
information = recv_framed(conn)
send_framed(conn, information)
srv = socket.socket()
srv.bind(('127.0.0.1', 9000))
srv.hear(1)
threading.Thread(goal=lambda: echo_once(srv.settle for()[0]), daemon=True).begin()
c = socket.create_connection(('127.0.0.1', 9000))
send_framed(c, b'x' * 500000)
end result = recv_framed(c)
assert len(end result) == 500000
print('giant payload survived the spherical journey')
Implementing Key Change With ECDH
Minimal ECDH Handshake
Either side generates an elliptic-curve key pair, and the general public keys are exchanged over the wire.
from cryptography.hazmat.primitives.uneven import ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
def make_keypair():
priv = ec.generate_private_key(ec.SECP384R1())
pub_bytes = priv.public_key().public_bytes(
encoding=serialization.Encoding.X962,
format=serialization.PublicFormat.UncompressedPoint
)
return priv, pub_bytes
Why P-384 reasonably than the extra frequent P-256? Both is an inexpensive selection for this tutorial — P-256 is quicker and simply as sound for many functions. P-384 is used right here for a bigger safety margin; swap in ec.SECP256R1() in the event you’d reasonably match the extra typical default.
Deriving a Per-Session Key
Utilizing the shared secret from the ECDH alternate, HKDF derives a correct symmetric session key.
def get_session_key(priv, other_pub_bytes):
other_pub = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP384R1(), other_pub_bytes)
shared = priv.alternate(ec.ECDH(), other_pub)
return HKDF(algorithm=hashes.SHA256(), size=32, salt=None, data=b'session key').derive(shared)
Why This Beats a Pre-Shared Key
With a single hardcoded key, compromising it as soon as exposes each previous and future session — together with visitors an attacker could have already captured and saved. ECDH generates a recent key for each connection, and no long-term secret ever crosses the wire; if one session’s key leaks, no different session is affected.
Authenticating the Handshake (Stopping MITM)
Uncooked ECDH as proven above negotiates a shared secret with whoever is on the opposite finish of the socket — it doesn’t show that the opposite finish is definitely the consumer or server you meant to speak to. An attacker sitting on the community path can generate their very own ephemeral keypair, full a handshake with the server whereas pretending to be the consumer, and concurrently full a separate handshake with the consumer whereas pretending to be the server.
Either side find yourself with a valid-looking session key, and the attacker sits within the center, transparently decrypting, studying, and re-encrypting each message. ECDH alone provides you confidentiality between two events; it provides you zero assure about who these events are.
The repair is to authenticate the ephemeral public key earlier than trusting it. The only manner to do this right here, with out pulling in a full PKI or TLS stack, is a pre-shared key (PSK) that each side already know out-of-band, used to HMAC-sign both sides’s ephemeral public key. If the signature would not confirm, the connection is dropped earlier than any session secret is derived.
import hmac as hmac_lib
import hashlib
PSK = b'replace-this-with-a-real-provisioned-secret'
def sign_pub(pub_bytes):
return hmac_lib.new(PSK, pub_bytes, hashlib.sha256).digest()
def verify_pub(pub_bytes, tag):
anticipated = sign_pub(pub_bytes)
if not hmac_lib.compare_digest(anticipated, tag):
increase ValueError('handshake authentication failed - doable MITM')
> The literal PSK = b'replace-this-...' above is for illustration solely. In actual deployments, load it from an surroundings variable or a secrets and techniques supervisor (e.g. os.environ['PROTOCOL_PSK'].encode()) — by no means commit an actual PSK to supply management.
Take a look at our hands-on, sensible information to studying Git, with best-practices, industry-accepted requirements, and included cheat sheet. Cease Googling Git instructions and truly study it!
Either side now sends its ephemeral public key plus the HMAC tag over that key, and verifies the tag earlier than deriving the session key. An attacker with out the PSK cannot forge a tag that matches a substituted public key, so a swapped-in secret is caught earlier than it is ever trusted.
Encrypting and Authenticating Information
AES-GCM is a pure match right here: it offers encryption and an authentication tag in a single operation, with no separate HMAC step wanted, and any try to decrypt tampered ciphertext fails loudly with an exception reasonably than silently returning rubbish.
import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
def encrypt(key, information):
nonce = os.urandom(12)
return nonce + AESGCM(key).encrypt(nonce, information, None)
def decrypt(key, blob):
nonce = blob[:12]
ct = blob[12:]
return AESGCM(key).decrypt(nonce, ct, None)
> A random 12-byte nonce per message is okay at this tutorial’s scale, however with a long-lived session key and a excessive message quantity, random 96-bit nonces carry an actual (if small) birthday-bound collision danger — reusing a nonce with the identical key breaks AES-GCM’s safety ensures. For prime-throughput or long-lived periods, want a monotonic counter because the nonce as an alternative of os.urandom(12).
Constructing the Server
Socket Setup, Multi-Shopper Dealing with, and Per-Shopper Keys
A brand new thread is created for each incoming connection; every connection performs its personal handshake and will get its personal session key. The handshake additionally verifies the consumer’s HMAC tag earlier than trusting its public key.
def handle_client(conn, addr):
priv, pub = make_keypair()
send_framed(conn, pub)
send_framed(conn, sign_pub(pub))
peer_pub = recv_framed(conn)
peer_tag = recv_framed(conn)
attempt:
verify_pub(peer_pub, peer_tag)
besides ValueError:
print(f'[!] {addr} failed handshake authentication, dropping connection')
conn.shut()
return
session_key = get_session_key(priv, peer_pub)
print(f'[+] {addr} handshake performed')
whereas True:
attempt:
blob = recv_framed(conn)
**besides (ConnectionError, ValueError) as e:
print(f'[!] {addr} disconnected or despatched a nasty body: {e}')
break**
msg = decrypt(session_key, blob)
print(f'[{addr}] {msg.decode()}')
send_framed(conn, encrypt(session_key, b'ACK: ' + msg))
conn.shut()
def run_server(host='0.0.0.0', port=9000):
srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
srv.bind((host, port))
srv.hear(5)
print(f'listening on {host}:{port}')
whereas True:
conn, addr = srv.settle for()
t = threading.Thread(goal=handle_client, args=(conn, addr), daemon=True)
t.begin()
if __name__ == '__main__':
run_server()
Constructing the Shopper
Regardless of the consumer sorts is encrypted and despatched to the server; no matter arrives on the consumer is learn and decrypted. The consumer additionally verifies the server’s HMAC tag earlier than trusting its public key.
def run_client(host='127.0.0.1', port=9000):
sock = socket.create_connection((host, port))
priv, pub = make_keypair()
peer_pub = recv_framed(sock)
peer_tag = recv_framed(sock)
attempt:
verify_pub(peer_pub, peer_tag)
besides ValueError:
print('[!] server failed handshake authentication, aborting')
sock.shut()
return
send_framed(sock, pub)
send_framed(sock, sign_pub(pub))
session_key = get_session_key(priv, peer_pub)
whereas True:
textual content = enter('> ')
if textual content.decrease() == 'give up':
break
send_framed(sock, encrypt(session_key, textual content.encode()))
reply = decrypt(session_key, recv_framed(sock))
print(reply.decode())
sock.shut()
if __name__ == '__main__':
run_client()
Operating It: A number of Purchasers
Begin the server, then begin two or three terminals in parallel working consumer.py — every will get its personal handshake and session key, dealt with independently by itself thread by the server.
Inspecting Encrypted Site visitors
Set up a packet seize device (tcpdump), seize visitors from a consumer mid-conversation, and examine it in Wireshark. The size prefixes are seen, however the payload is unreadable — affirmation that the AES-GCM layer is doing its job.
tcpdump -i lo -w seize.pcap port 9000
Testing: Unsuitable Key / Tampered Ciphertext
A single modified bit causes decryption to lift an exception as an alternative of returning corrupted information with no indication something went improper — that is the authentication tag at work.
attempt:
decrypt(os.urandom(32), encrypt(session_key, b'good day'))
besides Exception as e:
print('decryption failed as anticipated:', e)
Dealing with Connection Errors
The server’s handle_client loop already wraps recv_framed in a attempt/besides (see the “Constructing the Server” part above), so a consumer that disconnects mid-session — or sends a malformed or outsized body — simply ends that one thread cleanly reasonably than crashing the server or affecting different linked shoppers. On the consumer aspect, a dropped server connection surfaces as a ConnectionError from recv_framed/send_framed; wrap the consumer’s whereas True loop the identical manner in order for you a clear error message as an alternative of a stack hint.
How This Compares to a Actual VPN
To be clear, this is not an actual VPN. Manufacturing VPN protocols, in response to the VPNOverview analysis, are distinguished by the truth that they work on the IP stage with the tabling of complete community visitors circulation in digital interfaces (TUN/TAP) as an alternative of simply securing a selected application-level socket connection. The definition of what we have created right here is to safe one dialog between two endpoints; a real VPN secures all communications that the working system streams to the community.
The place to Go From Right here
There are two pure subsequent steps: make this protocol work with an actual TUN/TAP interface so visitors truly passes by it, and change the hand-rolled crypto layer with a extra vetted framework such because the Noise Protocol Framework or libsodium — each deal with issues like replay safety and key rotation that have been intentionally disregarded right here for readability. The PSK-based HMAC signing proven above is a minimal repair for the handshake authentication hole; a manufacturing system would extra doubtless use mutual TLS or a Noise handshake sample (e.g. Noise_XX or Noise_KK) to get that, together with replay safety, without spending a dime.
Conclusion
We designed a protocol for a client-server utility with correct message framing, elliptic-curve key alternate, authenticated encryption, and multi-client assist — and not using a single fastened secret baked into the code. It isn’t a VPN, nevertheless it’s the muse one is constructed on. The sincere subsequent step from right here is TUN/TAP tunneling paired with a production-quality crypto library, reasonably than persevering with to increase a hand-rolled one.
