Building a Secure Client-Server Protocol in Python: Sockets, Framing & Key Exchange

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.

To be clear up front, this is not a complete VPN. No TUN/TAP interfaces or routing of IP traffic is involved here. What we're constructing is a secure client-server protocol — the building block upon which a real VPN, chat application, or any custom network service using encrypted, authenticated communication sits.

What We're Building

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 — there's no OS-level forwarding or virtual network interface involved.

Design Goals

This differs from a toy socket example in four ways:

• Proper message framing. TCP treats data as a stream, not discrete messages, so we define our own message boundaries.
• Real key exchange. Every session negotiates its own key rather than using a fixed key hardcoded into the program.
• Multi-client support. The server handles multiple clients simultaneously.
• Visibility into wire traffic. We'll inspect the bytes going over the network and confirm they're actually encrypted.

Project Setup

python3 -m venv venv
source venv/bin/activate
pip install cryptography

Create a server.py and a client.py. We'll build them piece by piece.

Message Framing in Python Sockets

Receiving an Exact Number of Bytes

socket.recv() can return fewer bytes than requested. This helper loops until it has exactly what it needs:

def recv_exact(sock, n):
    # sockets don't guarantee a full read in one shot, so loop it
    buf = b''
    while len(buf) < n:
        chunk = sock.recv(n - len(buf))  # ask only for what's left
        if not chunk:
            # peer closed the connection mid-message, nothing more to do
            raise ConnectionError('socket closed before we got everything')
        buf += chunk
    return buf

Length-Prefixing Messages

Each message is prefixed with its length as 4 bytes (big-endian).

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 — an easy denial-of-service vector. We cap it here at 10MB; adjust to whatever your real message size ceiling is.

import struct

**MAX_MESSAGE_SIZE = 10 * 1024 * 1024  # 10MB — adjust to your actual needs**

def send_framed(sock, payload):
    # write the size first so the other end knows exactly how much to read
    sock.sendall(struct.pack('>I', len(payload)) + payload)

def recv_framed(sock):
    # header tells us the payload length, then we grab exactly that many bytes
    header = recv_exact(sock, 4)
    length = struct.unpack('>I', header)[0]
    **if length > MAX_MESSAGE_SIZE:
        raise ValueError(f'peer claims a {length}-byte message, refusing (max {MAX_MESSAGE_SIZE})')**
    return recv_exact(sock, length)

Testing With a Large Payload

A quick echo test confirms the framing logic handles a payload larger than what a single recv() call would return.

import socket
import threading

def echo_once(conn):
    # just bounce whatever we receive straight back
    data = recv_framed(conn)
    send_framed(conn, data)

srv = socket.socket()
srv.bind(('127.0.0.1', 9000))
srv.listen(1)
# spin up a throwaway thread so accept() doesn't block this test
threading.Thread(target=lambda: echo_once(srv.accept()[0]), daemon=True).start()

c = socket.create_connection(('127.0.0.1', 9000))
send_framed(c, b'x' * 500000)  # deliberately bigger than one recv() would return
result = recv_framed(c)
assert len(result) == 500000
print('large payload survived the round trip')

Implementing Key Exchange With ECDH

Minimal ECDH Handshake

Each side generates an elliptic-curve key pair, and the public keys are exchanged over the wire.

from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.kdf.hkdf import HKDF

def make_keypair():
    # SECP384R1 - solid security margin and the library supports it well
    priv = ec.generate_private_key(ec.SECP384R1())
    # send this over the wire so the other side can do its half of ECDH
    pub_bytes = priv.public_key().public_bytes(
        encoding=serialization.Encoding.X962,
        format=serialization.PublicFormat.UncompressedPoint
    )
    return priv, pub_bytes

Why P-384 rather than the more common P-256? Either is a reasonable choice for this tutorial — 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.

Deriving a Per-Session Key

Using the shared secret from the ECDH exchange, HKDF derives a proper symmetric session key.

def get_session_key(priv, other_pub_bytes):
    # rebuild the peer's public key object from the bytes they sent us
    other_pub = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP384R1(), other_pub_bytes)
    shared = priv.exchange(ec.ECDH(), other_pub)
    # don't use the raw ECDH secret as-is - HKDF turns it into a clean 32-byte key
    return HKDF(algorithm=hashes.SHA256(), length=32, salt=None, info=b'session key').derive(shared)

Why This Beats a Pre-Shared Key

With a single hardcoded key, compromising it once exposes every past and future session — 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.

Authenticating the Handshake (Preventing MITM)

Raw ECDH as shown above negotiates a shared secret with whoever is on the other end of the socket — 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.

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.

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.

import hmac as hmac_lib
import hashlib

# PSK must be provisioned to both client and server ahead of time,
# out-of-band - e.g. baked into a config file at deploy time.
# This is NOT the session key; it only authenticates the handshake.
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):
    expected = sign_pub(pub_bytes)
    # compare_digest avoids leaking timing info about how many bytes matched
    if not hmac_lib.compare_digest(expected, tag):
        raise ValueError('handshake authentication failed - possible MITM')

> 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()) — never commit a real PSK to source control.

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.

Encrypting and Authenticating Data

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.

import os
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

def encrypt(key, data):
    nonce = os.urandom(12)  # must be unique per message, never reused with the same key
    return nonce + AESGCM(key).encrypt(nonce, data, None)

def decrypt(key, blob):
    # nonce was tacked on the front during encrypt(), so split it back off
    nonce = blob[:12]
    ct = blob[12:]
    return AESGCM(key).decrypt(nonce, ct, None)

> 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 — 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).

Building the Server

Socket Setup, Multi-Client Handling, and Per-Client Keys

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.

def handle_client(conn, addr):
    # every client gets its own keypair and, therefore, its own session key
    priv, pub = make_keypair()
    send_framed(conn, pub)
    send_framed(conn, sign_pub(pub))
    peer_pub = recv_framed(conn)
    peer_tag = recv_framed(conn)
    try:
        verify_pub(peer_pub, peer_tag)
    except ValueError:
        print(f'[!] {addr} failed handshake authentication, dropping connection')
        conn.close()
        return
    session_key = get_session_key(priv, peer_pub)
    print(f'[+] {addr} handshake done')

    while True:
        try:
            blob = recv_framed(conn)
        **except (ConnectionError, ValueError) as e:
            print(f'[!] {addr} disconnected or sent a bad frame: {e}')
            break**
        msg = decrypt(session_key, blob)
        print(f'[{addr}] {msg.decode()}')
        send_framed(conn, encrypt(session_key, b'ACK: ' + msg))
    conn.close()

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)  # avoid "address in use" on restart
    srv.bind((host, port))
    srv.listen(5)
    print(f'listening on {host}:{port}')
    while True:
        conn, addr = srv.accept()
        # hand each connection its own thread so clients don't block one another
        t = threading.Thread(target=handle_client, args=(conn, addr), daemon=True)
        t.start()

if __name__ == '__main__':
    run_server()

Building the Client

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.

def run_client(host='127.0.0.1', port=9000):
    sock = socket.create_connection((host, port))
    priv, pub = make_keypair()
    # server sends its public key first in this version, then we send ours
    peer_pub = recv_framed(sock)
    peer_tag = recv_framed(sock)
    try:
        verify_pub(peer_pub, peer_tag)
    except ValueError:
        print('[!] server failed handshake authentication, aborting')
        sock.close()
        return
    send_framed(sock, pub)
    send_framed(sock, sign_pub(pub))
    session_key = get_session_key(priv, peer_pub)

    while True:
        text = input('> ')
        if text.lower() == 'quit':
            break
        send_framed(sock, encrypt(session_key, text.encode()))
        reply = decrypt(session_key, recv_framed(sock))
        print(reply.decode())
    sock.close()

if __name__ == '__main__':
    run_client()

Running It: Multiple Clients

Start the server, then start two or three terminals in parallel running client.py — each gets its own handshake and session key, handled independently on its own thread by the server.

Inspecting Encrypted Traffic

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 — confirmation that the AES-GCM layer is doing its job.

tcpdump -i lo -w capture.pcap port 9000

Testing: Wrong Key / Tampered Ciphertext

A single changed bit causes decryption to raise an exception instead of returning corrupted data with no indication anything went wrong — that's the authentication tag at work.

# swap the real key for a random one and watch it blow up
try:
    decrypt(os.urandom(32), encrypt(session_key, b'hello'))
except Exception as e:
    print('decryption failed as expected:', e)

Handling Connection Errors

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 — or sends a malformed or oversized frame — 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.

How This Compares to a Real VPN

To be clear, this isn't a real VPN. Production VPN protocols, according to the VPNOverview research, are distinguished by the fact that they work at the IP level with the tabling of entire network traffic flow in virtual interfaces (TUN/TAP) instead of just securing a particular application-level socket connection. The definition of what we've created here is to secure one conversation between two endpoints; a true VPN secures all communications that the operating system streams to the network.

Where to Go From Here

There are two natural next steps: make this protocol work with a real TUN/TAP interface so traffic actually passes through it, and replace the hand-rolled crypto layer with a more vetted framework such as the Noise Protocol Framework or libsodium — both handle things like replay protection and key rotation that were deliberately left out here for clarity. The PSK-based HMAC signing shown above is a minimal fix for the handshake authentication gap; a production system would more likely use mutual TLS or a Noise handshake pattern (e.g. Noise_XX or Noise_KK) to get that, along with replay protection, for free.

Conclusion

We designed a protocol for a client-server application with proper message framing, elliptic-curve key exchange, authenticated encryption, and multi-client support — without a single fixed secret baked into the code. It's not a VPN, but it's the foundation one is built on. The honest next step from here is TUN/TAP tunneling paired with a production-quality crypto library, rather than continuing to extend a hand-rolled one.