This course focuses on constructing a Secure Client-Server Protocol using the Python programming language, including security for sockets, framing, and key exchange. If you ever wanted to implement your own secure communication protocol in Python, using only the standard Python socket library and the cryptography library, this is how to do it.<\/p>\n
To be clear up front, this is not a complete VPN<\/strong>. No TUN\/TAP interfaces or routing of IP traffic is involved here. What we’re constructing is a secure client-server protocol \u2014 the building block upon which a real VPN, chat application, or any custom network service using encrypted, authenticated communication sits.<\/p>\n At the end of this tutorial, you’ll have a framed message-sending client and server that negotiate a new encryption key for each session and detect tampering. It’s not a tunnel \u2014 there’s no OS-level forwarding or virtual network interface involved.<\/p>\n This differs from a toy socket example in four ways:<\/p>\n Proper message framing.<\/strong> TCP treats data as a stream, not discrete messages, so we define our own message boundaries. python3 -m venv venv Create a server.py and a client.py. We’ll build them piece by piece.<\/p>\n socket.recv() can return fewer bytes than requested. This helper loops until it has exactly what it needs:<\/p>\n def<\/span> recv_exact<\/span>(sock, n<\/span>):<\/span> Each message is prefixed with its length as 4 bytes (big-endian).<\/p>\n A blind 4-byte length prefix is fine for a demo, but as written it lets a peer claim an arbitrarily large payload (up to ~4GB) and force the receiver to allocate that much memory before a single byte of actual data arrives \u2014 an easy denial-of-service vector. We cap it here at 10MB; adjust to whatever your real message size ceiling is.<\/p>\n import<\/span> struct<\/p>\n **MAX_MESSAGE_SIZE = 10<\/span> * 1024<\/span> * 1024<\/span> # 10MB \u2014 adjust to your actual needs**<\/span><\/p>\n def<\/span> send_framed<\/span>(sock, payload<\/span>):<\/span> def<\/span> recv_framed<\/span>(sock<\/span>):<\/span> A quick echo test confirms the framing logic handles a payload larger than what a single recv() call would return.<\/p>\n import<\/span> socket def<\/span> echo_once<\/span>(conn<\/span>):<\/span> srv = socket.socket() c = socket.create_connection((‘127.0.0.1’<\/span>, 9000<\/span>)) Each side generates an elliptic-curve key pair, and the public keys are exchanged over the wire.<\/p>\n from<\/span> cryptography.hazmat.primitives.asymmetric import<\/span> ec def<\/span> make_keypair<\/span>():<\/span> Why P-384 rather than the more common P-256? Either is a reasonable choice for this tutorial \u2014 P-256 is faster and just as sound for most applications. P-384 is used here for a larger security margin; swap in ec.SECP256R1() if you’d rather match the more typical default.<\/p>\n Using the shared secret from the ECDH exchange, HKDF derives a proper symmetric session key.<\/p>\n def<\/span> get_session_key<\/span>(priv, other_pub_bytes<\/span>):<\/span> With a single hardcoded key, compromising it once exposes every past and future session \u2014 including traffic an attacker may have already captured and stored. ECDH generates a fresh key for every connection, and no long-term secret ever crosses the wire; if one session’s key leaks, no other session is affected.<\/p>\n Raw ECDH as shown above negotiates a shared secret with whoever is on the other end of the socket \u2014 it does not prove that the other end is actually the client or server you intended to talk to. An attacker sitting on the network path can generate their own ephemeral keypair, complete a handshake with the server while pretending to be the client, and simultaneously complete a separate handshake with the client while pretending to be the server.<\/p>\n Both sides end up with a valid-looking session key, and the attacker sits in the middle, transparently decrypting, reading, and re-encrypting every message. ECDH alone gives you confidentiality between two parties; it gives you zero guarantee about who those parties are.<\/p>\n The fix is to authenticate the ephemeral public key before trusting it. The simplest way to do that here, without pulling in a full PKI or TLS stack, is a pre-shared key (PSK) that both sides already know out-of-band, used to HMAC-sign each side’s ephemeral public key. If the signature doesn’t verify, the connection is dropped before any session key is derived.<\/p>\n import<\/span> hmac as<\/span> hmac_lib # PSK must be provisioned to both client and server ahead of time,<\/span> def<\/span> sign_pub<\/span>(pub_bytes<\/span>):<\/span> def<\/span> verify_pub<\/span>(pub_bytes, tag<\/span>):<\/span> > The literal PSK = b’replace-this-…’ above is for illustration only. In real deployments, load it from an environment variable or a secrets manager (e.g. os.environ[‘PROTOCOL_PSK’].encode()) \u2014 never commit a real PSK to source control.<\/strong><\/p>\n Each side now sends its ephemeral public key plus the HMAC tag over that key, and verifies the tag before deriving the session key. An attacker without the PSK can’t forge a tag that matches a substituted public key, so a swapped-in key is caught before it’s ever trusted.<\/p>\n AES-GCM is a natural fit here: it provides encryption and an authentication tag in a single operation, with no separate HMAC step needed, and any attempt to decrypt tampered ciphertext fails loudly with an exception rather than silently returning garbage.<\/p>\n import<\/span> os def<\/span> encrypt<\/span>(key, data<\/span>):<\/span> def<\/span> decrypt<\/span>(key, blob<\/span>):<\/span> > A random 12-byte nonce per message is fine at this tutorial’s scale, but with a long-lived session key and a high message volume, random 96-bit nonces carry a real (if small) birthday-bound collision risk \u2014 reusing a nonce with the same key breaks AES-GCM’s security guarantees. For high-throughput or long-lived sessions, prefer a monotonic counter as the nonce instead of os.urandom(12).<\/strong><\/p>\n A new thread is created for every incoming connection; each connection performs its own handshake and gets its own session key. The handshake also verifies the client’s HMAC tag before trusting its public key.<\/p>\n def<\/span> handle_client<\/span>(conn, addr<\/span>):<\/span> while<\/span> True<\/span>: def<\/span> run_server<\/span>(host=‘0.0.0.0’<\/span>, port=9000<\/span><\/span>):<\/span> if<\/span> __name__ == ‘__main__’<\/span>: Whatever the user types is encrypted and sent to the server; whatever arrives at the client is read and decrypted. The client also verifies the server’s HMAC tag before trusting its public key.<\/p>\n def<\/span> run_client<\/span>(host=‘127.0.0.1’<\/span>, port=9000<\/span><\/span>):<\/span> while<\/span> True<\/span>: if<\/span> __name__ == ‘__main__’<\/span>: Start the server, then start two or three terminals in parallel running client.py \u2014 each gets its own handshake and session key, handled independently on its own thread by the server.<\/p>\n Install a packet capture tool (tcpdump), capture traffic from a client mid-conversation, and view it in Wireshark. The length prefixes are visible, but the payload is unreadable \u2014 confirmation that the AES-GCM layer is doing its job.<\/p>\n tcpdump -i lo -w capture.pcap port 9000<\/p>\n A single changed bit causes decryption to raise an exception instead of returning corrupted data with no indication anything went wrong \u2014 that’s the authentication tag at work.<\/p>\n # swap the real key for a random one and watch it blow up<\/span> The server’s handle_client loop already wraps recv_framed in a try\/except (see the «Building the Server» section above), so a client that disconnects mid-session \u2014 or sends a malformed or oversized frame \u2014 just ends that one thread cleanly rather than crashing the server or affecting other connected clients. On the client side, a dropped server connection surfaces as a ConnectionError from recv_framed\/send_framed; wrap the client’s while True loop the same way if you want a clean error message instead of a stack trace.<\/p>\nWhat We’re Building<\/h2>\n
Design Goals<\/h2>\n
\nReal key exchange.<\/strong> Every session negotiates its own key rather than using a fixed key hardcoded into the program.
\nMulti-client support.<\/strong> The server handles multiple clients simultaneously.
\nVisibility into wire traffic.<\/strong> We’ll inspect the bytes going over the network and confirm they’re actually encrypted.<\/p>\nProject Setup<\/h2>\n
\nsource<\/span> venv\/bin\/activate
\npip install cryptography<\/p>\nMessage Framing in Python Sockets<\/h2>\n
Receiving an Exact Number of Bytes<\/h3>\n
\n # sockets don’t guarantee a full read in one shot, so loop it<\/span>
\n buf = b»<\/span>
\n while<\/span> len<\/span>(buf) < n:
\n chunk = sock.recv(n — len<\/span>(buf)) # ask only for what’s left<\/span>
\n if<\/span> not<\/span> chunk:
\n # peer closed the connection mid-message, nothing more to do<\/span>
\n raise<\/span> ConnectionError(‘socket closed before we got everything’<\/span>)
\n buf += chunk
\n return<\/span> buf<\/p>\nLength-Prefixing Messages<\/h3>\n
\n # write the size first so the other end knows exactly how much to read<\/span>
\n sock.sendall(struct.pack(‘>I’<\/span>, len<\/span>(payload)) + payload)<\/p>\n
\n # header tells us the payload length, then we grab exactly that many bytes<\/span>
\n header = recv_exact(sock, 4<\/span>)
\n length = struct.unpack(‘>I’<\/span>, header)[0<\/span>]
\n **if<\/span> length > MAX_MESSAGE_SIZE:
\n raise<\/span> ValueError(f’peer claims a {length}<\/span>-byte message, refusing (max {MAX_MESSAGE_SIZE}<\/span>)’<\/span>)**
\n return<\/span> recv_exact(sock, length)<\/p>\nTesting With a Large Payload<\/h3>\n
\nimport<\/span> threading<\/p>\n
\n # just bounce whatever we receive straight back<\/span>
\n data = recv_framed(conn)
\n send_framed(conn, data)<\/p>\n
\nsrv.bind((‘127.0.0.1’<\/span>, 9000<\/span>))
\nsrv.listen(1<\/span>)
\n# spin up a throwaway thread so accept() doesn’t block this test<\/span>
\nthreading.Thread(target=lambda<\/span>: echo_once(srv.accept()[0<\/span>]), daemon=True<\/span>).start()<\/p>\n
\nsend_framed(c, b’x’<\/span> * 500000<\/span>) # deliberately bigger than one recv() would return<\/span>
\nresult = recv_framed(c)
\nassert<\/span> len<\/span>(result) == 500000<\/span>
\nprint<\/span>(‘large payload survived the round trip’<\/span>)<\/p>\nImplementing Key Exchange With ECDH<\/h2>\n
Minimal ECDH Handshake<\/h3>\n
\nfrom<\/span> cryptography.hazmat.primitives import<\/span> hashes, serialization
\nfrom<\/span> cryptography.hazmat.primitives.kdf.hkdf import<\/span> HKDF<\/p>\n
\n # SECP384R1 — solid security margin and the library supports it well<\/span>
\n priv = ec.generate_private_key(ec.SECP384R1())
\n # send this over the wire so the other side can do its half of ECDH<\/span>
\n pub_bytes = priv.public_key().public_bytes(
\n encoding=serialization.Encoding.X962,
\n format<\/span>=serialization.PublicFormat.UncompressedPoint
\n )
\n return<\/span> priv, pub_bytes<\/p>\nDeriving a Per-Session Key<\/h3>\n
\n # rebuild the peer’s public key object from the bytes they sent us<\/span>
\n other_pub = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP384R1(), other_pub_bytes)
\n shared = priv.exchange(ec.ECDH(), other_pub)
\n # don’t use the raw ECDH secret as-is — HKDF turns it into a clean 32-byte key<\/span>
\n return<\/span> HKDF(algorithm=hashes.SHA256(), length=32<\/span>, salt=None<\/span>, info=b’session key’<\/span>).derive(shared)<\/p>\nWhy This Beats a Pre-Shared Key<\/h3>\n
Authenticating the Handshake (Preventing MITM)<\/h2>\n
\nimport<\/span> hashlib<\/p>\n
\n# out-of-band — e.g. baked into a config file at deploy time.<\/span>
\n# This is NOT the session key; it only authenticates the handshake.<\/span>
\nPSK = b’replace-this-with-a-real-provisioned-secret’<\/span><\/p>\n
\n return<\/span> hmac_lib.new(PSK, pub_bytes, hashlib.sha256).digest()<\/p>\n
\n expected = sign_pub(pub_bytes)
\n # compare_digest avoids leaking timing info about how many bytes matched<\/span>
\n if<\/span> not<\/span> hmac_lib.compare_digest(expected, tag):
\n raise<\/span> ValueError(‘handshake authentication failed — possible MITM’<\/span>)<\/p>\nEncrypting and Authenticating Data<\/h2>\n
\nfrom<\/span> cryptography.hazmat.primitives.ciphers.aead import<\/span> AESGCM<\/p>\n
\n nonce = os.urandom(12<\/span>) # must be unique per message, never reused with the same key<\/span>
\n return<\/span> nonce + AESGCM(key).encrypt(nonce, data, None<\/span>)<\/p>\n
\n # nonce was tacked on the front during encrypt(), so split it back off<\/span>
\n nonce = blob[:12<\/span>]
\n ct = blob[12<\/span>:]
\n return<\/span> AESGCM(key).decrypt(nonce, ct, None<\/span>)<\/p>\nBuilding the Server<\/h2>\n
Socket Setup, Multi-Client Handling, and Per-Client Keys<\/h3>\n
\n # every client gets its own keypair and, therefore, its own session key<\/span>
\n priv, pub = make_keypair()
\n send_framed(conn, pub)
\n send_framed(conn, sign_pub(pub))
\n peer_pub = recv_framed(conn)
\n peer_tag = recv_framed(conn)
\n try<\/span>:
\n verify_pub(peer_pub, peer_tag)
\n except<\/span> ValueError:
\n print<\/span>(f'[!] {addr}<\/span> failed handshake authentication, dropping connection’<\/span>)
\n conn.close()
\n return<\/span>
\n session_key = get_session_key(priv, peer_pub)
\n print<\/span>(f'[+] {addr}<\/span> handshake done’<\/span>)<\/p>\n
\n try<\/span>:
\n blob = recv_framed(conn)
\n **except<\/span> (ConnectionError, ValueError) as<\/span> e:
\n print<\/span>(f'[!] {addr}<\/span> disconnected or sent a bad frame: {e}<\/span>‘<\/span>)
\n break<\/span>**
\n msg = decrypt(session_key, blob)
\n print<\/span>(f'[{addr}<\/span>] {msg.decode()}<\/span>‘<\/span>)
\n send_framed(conn, encrypt(session_key, b’ACK: ‘<\/span> + msg))
\n conn.close()<\/p>\n
\n srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
\n srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1<\/span>) # avoid «address in use» on restart<\/span>
\n srv.bind((host, port))
\n srv.listen(5<\/span>)
\n print<\/span>(f’listening on {host}<\/span>:{port}<\/span>‘<\/span>)
\n while<\/span> True<\/span>:
\n conn, addr = srv.accept()
\n # hand each connection its own thread so clients don’t block one another<\/span>
\n t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True<\/span>)
\n t.start()<\/p>\n
\n run_server()<\/p>\nBuilding the Client<\/h2>\n
\n sock = socket.create_connection((host, port))
\n priv, pub = make_keypair()
\n # server sends its public key first in this version, then we send ours<\/span>
\n peer_pub = recv_framed(sock)
\n peer_tag = recv_framed(sock)
\n try<\/span>:
\n verify_pub(peer_pub, peer_tag)
\n except<\/span> ValueError:
\n print<\/span>(‘[!] server failed handshake authentication, aborting’<\/span>)
\n sock.close()
\n return<\/span>
\n send_framed(sock, pub)
\n send_framed(sock, sign_pub(pub))
\n session_key = get_session_key(priv, peer_pub)<\/p>\n
\n text = input<\/span>(‘> ‘<\/span>)
\n if<\/span> text.lower() == ‘quit’<\/span>:
\n break<\/span>
\n send_framed(sock, encrypt(session_key, text.encode()))
\n reply = decrypt(session_key, recv_framed(sock))
\n print<\/span>(reply.decode())
\n sock.close()<\/p>\n
\n run_client()<\/p>\nRunning It: Multiple Clients<\/h2>\n
Inspecting Encrypted Traffic<\/h2>\n
Testing: Wrong Key \/ Tampered Ciphertext<\/h2>\n
\ntry<\/span>:
\n decrypt(os.urandom(32<\/span>), encrypt(session_key, b’hello’<\/span>))
\nexcept<\/span> Exception as<\/span> e:
\n print<\/span>(‘decryption failed as expected:’<\/span>, e)<\/p>\nHandling Connection Errors<\/h2>\n
How This Compares to a Real VPN<\/h2>\n