TLS/SSL 证书生成、自签名证书、请求 CA 签发证书以及使用 Python TCP 服务器与客户端进行加密通讯
问题： TCP 通讯能否加密
最近遇到了通讯加密的问题，我们都知道 TLS/SSL 证书一般都是用来给HTTP进行加密使用的，使其成为HTTPS。
但HTTP也是TCP通讯，那么 TLS/SSL 证书应该也可以使用到TCP的服务器与客户端通讯加密中。
当我们不使用加密证书时，通讯的内容是可以通过抓包工具（Wireshark）获得，并且很轻易的拿到其中发送的内容（明文）
直接就拿到了“aaaa”数据
使用了加密证书再次发送“aaaa”数据,可以发现，无法直接看出内容
那么我们就来实现一下TCP通讯加密的过程。
一、自签名证书步骤
IP="192.168.1.2"
openssl req -x509 -newkey rsa:1024 -keyout ca.key -out ca.crt -days 365 -nodes -subj "/CN=$IP" -addext "subjectAltName=IP:$IP"
Windows10安装ca证书：
- 双击ca.crt
- 安装证书
- 安装到受信任的根证书下
- 检测是否安装成功，再次双击ca.crt
- 选择证书路径，证书状态显示“该证书没有问题”就OK了
二、CA服务器签发证书
# 1. 生成服务器私钥
openssl genrsa -out server.key 2048
# 2. 使用 cert.cnf 生成 CSR
openssl req -new -key server.key -out server.csr -config cert.cnf
使用域名的话，就使用DNS.x 可以配置多个，可以使用通配符，CN与DNS要一致
cert.cnf内容：
[req]
distinguished_name = req_distinguished_name
x509_extensions = v3_req
prompt = no
[req_distinguished_name]
C = CN
ST = Beijing
L = Beijing
O = MyOrg
OU = MyUnit
CN = my.com
[v3_req]
subjectAltName = @alt_names
[alt_names]
DNS.1 = *.my.com
DNS.2 = *.ai.my.com
# IP.1 = 192.168.1.2
三、自搭建CA服务器（扩展）
# 1. 生成 CA 私钥（2048位，建议生产环境使用4096位）
openssl genrsa -out ca.key 4096
# 2. 生成自签名的 CA 根证书（有效期10年，主题中的 CN 是 CA 的名称）
#openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 -out ca.crt -subj "/CN=MyCA"
openssl req -x509 -new -nodes -key ca.key -sha256 -days 3650 -out ca.crt -config ca.cnf -extensions v3_ca
ca.cnf内容：
[ req ]
default_bits = 4096
distinguished_name = req_distinguished_name
string_mask = utf8only
default_md = sha256
x509_extensions = v3_ca
[ req_distinguished_name ]
countryName = Country Name (2 letter code)
countryName_default = CN
countryName_min = 2
countryName_max = 2
stateOrProvinceName = State or Province Name (full name)
stateOrProvinceName_default = Province
localityName = Locality Name (eg, city)
localityName_default = City
organizationName = Organization Name (eg, company)
organizationName_default = My Company Ltd
organizationalUnitName = Organizational Unit Name (eg, section)
organizationalUnitName_default = IT Department
commonName = Common Name (e.g. server FQDN or YOUR name)
commonName_default = My Root CA
emailAddress = Email Address
emailAddress_default = admin@example.com
[ v3_ca ]
subjectKeyIdentifier = hash
authorityKeyIdentifier = keyid:always,issuer
basicConstraints = critical, CA:true
keyUsage = critical, digitalSignature, cRLSign, keyCertSign
nsCertType = sslCA, emailCA
# 使用 CA 私钥和 CSR 签署服务器证书（有效期1年）
openssl x509 -req -in server.csr -CA ca.crt -CAkey ca.key -CAcreateserial -out server.crt -days 365 -sha256
四、TCP服务器 Python代码
核心代码：
class KeepaliveSSLServer:
def __init__(self, host='0.0.0.0', port=8443, certfile='server.crt', keyfile='server.key'):
self.host = host
self.port = port
self.certfile = certfile
self.keyfile = keyfile
self.running = False
def handle_client(self, conn, addr):
"""Handle client connection (keep alive)"""
......
data = conn.recv(4096)
message = data.decode('utf-8')
print(f"{client_name} 📥 Received: {message}")
......
# Send response
response = f"Server received: '{message}' [Time: {time.strftime('%H:%M:%S')}]"
conn.sendall(response.encode('utf-8'))
......
def start(self):
"""Start server"""
try:
# Create SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)
# Create TCP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.host, self.port))
sock.listen(5)
# Wrap SSL
ssock = context.wrap_socket(sock, server_side=True)
self.running = True
......
# Accept connections
while self.running:
try:
conn, addr = ssock.accept()
# Create separate thread for each client
client_thread = threading.Thread(
target=self.handle_client,
args=(conn, addr),
daemon=True
)
client_thread.start()
time.sleep(0.1) # Avoid CPU spinning
except Exception as e:
print(f"Accept connection error: {e}")
continue
except Exception as e:
print(f"❌ Server failed to start: {e}")
import traceback
traceback.print_exc()
finally:
try:
ssock.close()
sock.close()
except:
pass
print("✅ Server stopped")
if __name__ == '__main__':
# Configuration
CERT_FILE = 'server.crt'
KEY_FILE = 'server.key'
# Start server
server = KeepaliveSSLServer(
host='0.0.0.0',
port=8443,
certfile=CERT_FILE,
keyfile=KEY_FILE
)
server.start()
完整代码：
# server_keepalive.py
import socket
import ssl
import threading
import time
class KeepaliveSSLServer:
def __init__(self, host='0.0.0.0', port=8443, certfile='server.crt', keyfile='server.key'):
self.host = host
self.port = port
self.certfile = certfile
self.keyfile = keyfile
self.running = False
def handle_client(self, conn, addr):
"""Handle client connection (keep alive)"""
client_name = f"[{addr[0]}:{addr[1]}]"
print(f"{client_name} 🟢 Connected")
try:
while True:
# Receive data (with timeout to avoid permanent block)
conn.settimeout(30.0) # 30 seconds timeout
try:
data = conn.recv(4096)
if not data:
print(f"{client_name} 🔴 Client closed connection")
break
message = data.decode('utf-8')
print(f"{client_name} 📥 Received: {message}")
# Handle special commands
if message.lower() in ['exit', 'quit']:
response = f"{client_name} Exit command received, closing connection"
conn.sendall(response.encode('utf-8'))
print(f"{client_name} 📤 Sent: {response}")
break
# Send response
response = f"Server received: '{message}' [Time: {time.strftime('%H:%M:%S')}]"
conn.sendall(response.encode('utf-8'))
print(f"{client_name} 📤 Sent: {response}")
except socket.timeout:
# Continue waiting after timeout (keep connection alive)
print(f"{client_name} ⏰ Waiting for message…")
continue
except Exception as e:
print(f"{client_name} ❌ Receive error: {e}")
break
except Exception as e:
print(f"{client_name} ❌ Processing error: {e}")
finally:
conn.close()
print(f"{client_name} 🔴 Connection closed")
def start(self):
"""Start server"""
try:
# Create SSL context
context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
context.load_cert_chain(certfile=self.certfile, keyfile=self.keyfile)
# Create TCP socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind((self.host, self.port))
sock.listen(5)
# Wrap SSL
ssock = context.wrap_socket(sock, server_side=True)
self.running = True
print("=" * 60)
print(f"🔐 SSL Keep-Alive Server Started")
print(f" Listening: {self.host}:{self.port}")
print(f" Cert: {self.certfile}")
print(f" Key: {self.keyfile}")
print("=" * 60)
print("Waiting for connections… (Ctrl+C to stop)\\n")
# Accept connections
while self.running:
try:
conn, addr = ssock.accept()
# Create separate thread for each client
client_thread = threading.Thread(
target=self.handle_client,
args=(conn, addr),
daemon=True
)
client_thread.start()
time.sleep(0.1) # Avoid CPU spinning
except KeyboardInterrupt:
print("\\n🛑 Shutting down server…")
break
except Exception as e:
print(f"Accept connection error: {e}")
continue
except Exception as e:
print(f"❌ Server failed to start: {e}")
import traceback
traceback.print_exc()
finally:
try:
ssock.close()
sock.close()
except:
pass
print("✅ Server stopped")
if __name__ == '__main__':
# Configuration
CERT_FILE = 'server.crt'
KEY_FILE = 'server.key'
# Start server
server = KeepaliveSSLServer(
host='0.0.0.0',
port=8443,
certfile=CERT_FILE,
keyfile=KEY_FILE
)
server.start()
五、TCP 客户端 Python代码
核心代码：
class InteractiveSSLClient:
def __init__(self, server_ip='192.168.1.2', server_port=8443, certfile='ca.crt'):
self.server_ip = server_ip
self.server_port = server_port
self.certfile = certfile
def start(self):
"""Start interactive client"""
try:
# Create SSL context
# 1.使用系统的根证书目录
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
# 2.如果没安装证书到系统，就在代码中加载(使用自签名的根证书)
# context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
# context.load_verify_locations(cafile=self.certfile)
context.check_hostname = False
context.verify_mode = ssl.CERT_REQUIRED
# Create and connect
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssock = context.wrap_socket(sock, server_hostname=self.server_ip)
ssock.connect((self.server_ip, self.server_port))
......
# Loop reading keyboard input
while True:
try:
# Read keyboard input
user_input = input("\\n📤 Enter message: ").strip()
......
# Send message
ssock.sendall(user_input.encode('utf-8'))
print(f"✅ Sent: {user_input}")
# Receive server response
response = ssock.recv(4096).decode('utf-8')
print(f"📥 Server response: {response}")
except Exception as e:
print(f"\\n❌ Send/Receive error: {e}")
break
# Close connection
ssock.close()
......
if __name__ == '__main__':
# Configuration
# SERVER_IP = '192.168.1.2' # Change to your server IP
SERVER_IP = 'my.com' # Change to your server IP
SERVER_PORT = 8443
CERT_FILE = 'ca.crt' # Certificate file path
# Start client
client = InteractiveSSLClient(
server_ip=SERVER_IP,
server_port=SERVER_PORT,
certfile=CERT_FILE
)
client.start()
完整代码：
import socket
import ssl
import hashlib
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
class InteractiveSSLClient:
def __init__(self, server_ip='192.168.1.2', server_port=8443, certfile='ca.crt'):
self.server_ip = server_ip
self.server_port = server_port
self.certfile = certfile
def start(self):
"""Start interactive client"""
try:
# Create SSL context
# 1.使用系统的根证书目录
context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
# 2.如果没安装证书到系统，就在代码中加载(使用自签名的根证书)
# context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
# context.load_verify_locations(cafile=self.certfile)
context.check_hostname = False
context.verify_mode = ssl.CERT_REQUIRED
# Create and connect
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
ssock = context.wrap_socket(sock, server_hostname=self.server_ip)
ssock.connect((self.server_ip, self.server_port))
# 获取本地端口号
local_address = ssock.getsockname()
print("本地地址:", local_address)
print("本地端口:", local_address[1])
# 获取证书字典（基本信息）
cert_dict = ssock.getpeercert()
# 获取二进制证书（用于计算指纹）
cert_bin = ssock.getpeercert(binary_form=True)
# 计算指纹
sha256_fingerprint = hashlib.sha256(cert_bin).hexdigest()
sha1_fingerprint = hashlib.sha1(cert_bin).hexdigest()
# 2. 提取公钥
cert = x509.load_der_x509_certificate(cert_bin, default_backend())
public_key = cert.public_key()
public_key_pem = public_key.public_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PublicFormat.SubjectPublicKeyInfo
)
# 3. 计算公钥指纹（只对公钥部分）
pubkey_fingerprint = hashlib.sha256(public_key_pem).hexdigest()
# 提取信息
info = {
"issuer": tuple_to_dict(cert_dict.get('issuer', [])),
"subject": tuple_to_dict(cert_dict.get('subject', [])),
"validity": {
"not_before": cert_dict.get('notBefore', 'N/A'),
"not_after": cert_dict.get('notAfter', 'N/A'),
},
"fingerprints": {
# "SHA256": ':'.join(sha256_fingerprint[i:i+2] for i in range(0, len(sha256_fingerprint), 2)),
"SHA256": sha256_fingerprint,
"SHA1": ':'.join(sha1_fingerprint[i:i + 2] for i in range(0, len(sha1_fingerprint), 2)),
},
"san": cert_dict.get('subjectAltName', 'N/A'),
"version": cert_dict.get('version', 'N/A'),
"serial_number": cert_dict.get('serialNumber', 'N/A'),
"pubkey_fingerprint": pubkey_fingerprint,
}
try:
print("=" * 70)
print("📊 CERTIFICATE DETAILS (No cryptography)")
print("=" * 70)
# 显示颁发者
print("\\n📌 颁发者 (Issuer):")
for key, value in info['issuer'].items():
print(f" {key}: {value}")
# 显示主体
print("\\n👤 颁发给 (Subject):")
for key, value in info['subject'].items():
print(f" {key}: {value}")
# 有效期
print("\\n⏰ 有效期:")
print(f" 开始: {info['validity']['not_before']}")
print(f" 到期: {info['validity']['not_after']}")
# 指纹
print("\\n🔐 SHA-256 指纹:")
print(f" 证书: {info['fingerprints']['SHA256']}")
# print(f" SHA-1: {info['fingerprints']['SHA1']}")
print(f" 公钥: {info['pubkey_fingerprint']}")
# 其他
print("\\n📋 其他:")
print(f" 版本: {info['version']}")
print(f" 序列号: {info['serial_number']}")
print(f" SAN: {info['san']}")
print("=" * 70)
except Exception as e:
print(f"❌ 错误: {e}")
import traceback
traceback.print_exc()
print("=" * 60)
print(f"✅ Connected to Server")
print(f" Server: {self.server_ip}:{self.server_port}")
print(f" Protocol: {ssock.version()}")
print(f" Cipher: {ssock.cipher()[0]}")
print("=" * 60)
print("\\nUsage:")
print("1. Type message and press Enter to send to server")
print("2. Type 'exit' or 'quit' to disconnect")
print("3. Type 'help' to view commands")
print("-" * 60)
# Loop reading keyboard input
while True:
try:
# Read keyboard input
user_input = input("\\n📤 Enter message: ").strip()
# Check exit commands
if user_input.lower() in ['exit', 'quit', 'q']:
print("\\n👋 Disconnecting…")
break
# Check help command
if user_input.lower() == 'help':
print("\\nAvailable commands:")
print(" exit/quit/q – Disconnect from server")
print(" help – Show this help")
print(" Other text – Send to server")
continue
# Check empty input
if not user_input:
print("⚠️ Please enter non-empty message")
continue
# Send message
ssock.sendall(user_input.encode('utf-8'))
print(f"✅ Sent: {user_input}")
# Receive server response
response = ssock.recv(4096).decode('utf-8')
print(f"📥 Server response: {response}")
except KeyboardInterrupt:
print("\\n\\n👋 Ctrl+C detected, disconnecting…")
break
except Exception as e:
print(f"\\n❌ Send/Receive error: {e}")
break
# Close connection
ssock.close()
print("✅ Connection closed")
except ssl.SSLError as e:
print(f"\\n❌ SSL Error: {e}")
print("Tip: Check if certificate matches server IP")
except ConnectionRefusedError:
print(f"\\n❌ Connection refused: {self.server_ip}:{self.server_port}")
print("Tip: Ensure server is running")
except Exception as e:
print(f"\\n❌ Error: {e}")
import traceback
traceback.print_exc()
def tuple_to_dict(data):
"""
将 (('key1', 'val1'),) 格式转换为 {'key1': 'val1'}
"""
return {item[0][0]: item[0][1] for item in data}
if __name__ == '__main__':
# Configuration
# SERVER_IP = '192.168.1.2' # Change to your server IP
SERVER_IP = 'my.com' # Change to your server IP
SERVER_PORT = 8443
CERT_FILE = 'ca.crt' # Certificate file path
# Start client
client = InteractiveSSLClient(
server_ip=SERVER_IP,
server_port=SERVER_PORT,
certfile=CERT_FILE
)
client.start()
