How to Mock mTLS Between Services in Python
Simulate mutual TLS authentication between two services using Python's ssl module with self-signed certificates.
Python code
66 linesimport ssl
import socket
import threading
import tempfile
from pathlib import Path
import subprocess
def create_test_cert(cert_path: Path, key_path: Path, common_name: str = "localhost"):
"""Generate a self-signed certificate using openssl."""
subprocess.run([
"openssl", "req", "-x509", "-newkey", "rsa:2048",
"-keyout", str(key_path), "-out", str(cert_path),
"-days", "365", "-nodes", "-subj", f"/CN={common_name}"
], check=True, capture_output=True)
def start_mtls_server(certfile: str, keyfile: str, port: int = 8443):
"""Start a simple mTLS server that requires client certificates."""
context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
context.load_cert_chain(certfile=certfile, keyfile=keyfile)
# Require and verify client certificates
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(cafile=certfile) # Use CA cert for verification
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(("127.0.0.1", port))
server_socket.listen(1)
def handle_client(conn):
try:
with conn:
data = conn.recv(1024)
print(f"Server received: {data.decode()}")
conn.sendall(b"Hello from mTLS server!")
except ssl.SSLError as e:
print(f"TLS handshake failed: {e}")
print(f"Server listening on port {port} with mTLS required")
client_conn, addr = server_socket.accept()
handle_client(client_conn)
server_socket.close()
def connect_mtls_client(certfile: str, keyfile: str, port: int = 8443):
"""Connect as a client with certificate authentication."""
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
context.load_cert_chain(certfile=certfile, keyfile=keyfile)
context.check_hostname = False
context.verify_mode = ssl.CERT_NONE # Skip server verification for demo
with socket.create_connection(("127.0.0.1", port)) as raw_sock:
with context.wrap_socket(raw_sock, server_hostname="localhost") as tls_sock:
tls_sock.sendall(b"Authenticated request")
response = tls_sock.recv(1024)
print(f"Client received: {response.decode()}")
if __name__ == "__main__":
with tempfile.TemporaryDirectory() as tmpdir:
tmp = Path(tmpdir)
cert = tmp / "cert.pem"
key = tmp / "key.pem"
create_test_cert(cert, key)
server_thread = threading.Thread(target=start_mtls_server, args=(str(cert), str(key)))
server_thread.start()
connect_mtls_client(str(cert), str(key))
server_thread.join()
Output
Server listening on port 8443 with mTLS required
Client received: Hello from mTLS server!
Server received: Authenticated request
How it works
The server creates an SSL context with PROTOCOL_TLS_SERVER and requires client certificates by setting verify_mode to CERT_REQUIRED. The load_verify_locations call uses the CA certificate to validate incoming client certificates. The client creates a context with create_default_context, loads its certificate chain, and disables hostname checking to simplify local testing. Both sides exchange encrypted data after the TLS handshake completes, demonstrating how mTLS works between services. The server accepts one connection and handles it in a thread, while the client connects synchronously.
Common mistakes
- Using the same certificate for both client and server without separating CA and leaf certs
- Forgetting to set `check_hostname = False` when using self-signed certs locally
- Not handling SSL handshake errors gracefully on the server side
- Sharing private keys between services instead of using distinct certs per service
Variations
- Use `ssl.create_default_context` with a custom CA bundle for production validation
- Replace socket-based communication with an HTTP server like Flask or FastAPI wrapped in TLS
Real-world use cases
- Testing service-to-service authentication in a local development environment without a real CA.
- Validating mTLS configuration before deploying a new microservice behind an API gateway.
- Creating integration tests that verify mutual TLS handshakes in CI pipelines.
Sponsored
More from Microservices patterns
- BFF aggregation pattern: combine multiple service responses in Python easy
- Backward Compatible Schema Evolution in Python medium
- Bulkhead Thread Pool per Service Mock in Python medium
- CQRS with Separate Read and Write Repositories in Python medium
- Cache-Aside Pattern in Python: Per-Service Mock easy
- Consumer Driven Contract Pact Mock in Python medium
Keep learning
Related tutorials and quizzes for this topic.