This course focuses on establishing a Safe Consumer-Server Protocol utilizing the Python programming language, together with safety for sockets, framing, and key trade. If you happen to ever wished to implement your individual safe communication protocol in Python, utilizing solely the usual Python socket library and the cryptography library, that is easy methods to do it.
To be clear up entrance, this isn’t a whole VPN. No TUN/TAP interfaces or routing of IP site visitors is concerned right here. What we’re establishing 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 may have a framed message-sending consumer and server that negotiate a brand new encryption key for every session and detect tampering. It is not a tunnel — there is not any 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 trade. Each session negotiates its personal key somewhat than utilizing a hard and fast key hardcoded into this system.
- Multi-client help. The server handles a number of purchasers concurrently.
- Visibility into wire site visitors. We’ll examine the bytes going over the community and make sure they’re really encrypted.
Mission 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 Precise 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 bought all the pieces')
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 ok for a demo, however as written it lets a peer declare an arbitrarily massive payload (as much as ~4GB) and power the receiver to allocate that a lot reminiscence earlier than a single byte of precise information arrives — a simple denial-of-service vector. We cap it right here at 10MB; alter to no matter your actual message dimension 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 Massive 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.pay attention(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)
consequence = recv_framed(c)
assert len(consequence) == 500000
print('massive payload survived the spherical journey')
Implementing Key Alternate With ECDH
Minimal ECDH Handshake
All sides 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 somewhat than the extra frequent P-256? Both is an affordable 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 case you’d somewhat match the extra typical default.
Deriving a Per-Session Key
Utilizing the shared secret from the ECDH trade, 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.trade(ec.ECDH(), other_pub)
return HKDF(algorithm=hashes.SHA256(), size=32, salt=None, information=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 site visitors an attacker might 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 supposed 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.
Each 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 offers you confidentiality between two events; it offers you zero assure about who these events are.
The repair is to authenticate the ephemeral public key earlier than trusting it. The only means to try this right here, with out pulling in a full PKI or TLS stack, is a pre-shared key (PSK) that either side already know out-of-band, used to HMAC-sign either side’s ephemeral public key. If the signature would not confirm, the connection is dropped earlier than any session secret’s 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 - potential 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 really be taught it!
All sides 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’s caught earlier than it is ever trusted.
Encrypting and Authenticating Information
AES-GCM is a pure match right here: it gives 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 somewhat 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 ok 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 top-throughput or long-lived periods, favor a monotonic counter because the nonce as a substitute of os.urandom(12).
Constructing the Server
Socket Setup, Multi-Consumer Dealing with, and Per-Consumer 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 accomplished')
whereas True:
attempt:
blob = recv_framed(conn)
**besides (ConnectionError, ValueError) as e:
print(f'[!] {addr} disconnected or despatched a foul 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.pay attention(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 Consumer
Regardless of the consumer varieties 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() == 'stop':
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 Visitors
Set up a packet seize instrument (tcpdump), seize site visitors from a consumer mid-conversation, and look at 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: Flawed Key / Tampered Ciphertext
A single modified bit causes decryption to lift an exception as a substitute of returning corrupted information with no indication something went unsuitable — that is the authentication tag at work.
attempt:
decrypt(os.urandom(32), encrypt(session_key, b'hey'))
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 somewhat than crashing the server or affecting different related purchasers. On the consumer facet, a dropped server connection surfaces as a ConnectionError from recv_framed/send_framed; wrap the consumer’s whereas True loop the identical means if you need a clear error message as a substitute of a stack hint.
How This Compares to a Actual VPN
To be clear, this is not an actual VPN. Manufacturing VPN protocols, in accordance with the VPNOverview analysis, are distinguished by the truth that they work on the IP stage with the tabling of whole community site visitors stream in digital interfaces (TUN/TAP) as a substitute of simply securing a specific 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 site visitors really passes by it, and substitute 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 ignored 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 probably 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 trade, authenticated encryption, and multi-client help — with no single fastened secret baked into the code. It is not a VPN, nevertheless it’s the inspiration one is constructed on. The sincere subsequent step from right here is TUN/TAP tunneling paired with a production-quality crypto library, somewhat than persevering with to increase a hand-rolled one.
