面试题：Python socket模块在复杂网络环境下的安全与可靠性设计

1. 数据加密

使用cryptography库实现数据加密。

from cryptography.fernet import Fernet

# 生成密钥
key = Fernet.generate_key()
cipher_suite = Fernet(key)

# 加密数据
def encrypt_data(data):
    return cipher_suite.encrypt(data.encode())

# 解密数据
def decrypt_data(encrypted_data):
    return cipher_suite.decrypt(encrypted_data).decode()

2. 防止中间人攻击

使用数字签名和证书验证。借助cryptography库生成私钥、公钥对，并使用私钥对数据签名，接收方使用公钥验证签名。

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes

# 生成私钥
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)

# 生成公钥
public_key = private_key.public_key()

# 数据签名
def sign_data(data, private_key):
    signature = private_key.sign(
        data.encode(),
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH
        ),
        hashes.SHA256()
    )
    return signature

# 验证签名
def verify_signature(data, signature, public_key):
    try:
        public_key.verify(
            signature,
            data.encode(),
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        return True
    except:
        return False

3. 处理网络异常情况

处理网络延迟

使用超时机制，在socket操作中设置超时时间。

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(5)  # 设置5秒超时
try:
    sock.connect(('127.0.0.1', 12345))
    # 发送和接收数据操作
except socket.timeout:
    print("连接超时")

处理丢包

采用可靠的传输协议（如TCP），并且实现重传机制。如果在一定时间内没有收到确认信息，则重发数据。

# 发送数据并等待确认
def send_data_with_ack(sock, data):
    while True:
        sock.sendall(data)
        try:
            sock.settimeout(2)  # 设置等待确认的超时时间
            ack = sock.recv(1024)
            if ack.decode() == 'ACK':
                break
        except socket.timeout:
            continue

处理恶意攻击

过滤非法IP地址和端口，对收到的数据进行合法性检查，防止恶意构造的数据导致程序崩溃。

import re

# 简单的IP地址合法性检查
def is_valid_ip(ip):
    pattern = re.compile(r'^((25[0 - 5]|2[0 - 4][0 - 9]|[01]?[0 - 9][0 - 9]?)\.){3}(25[0 - 5]|2[0 - 4][0 - 9]|[01]?[0 - 9][0 - 9]?)$')
    return bool(pattern.match(ip))

# 简单的端口合法性检查
def is_valid_port(port):
    return 1 <= port <= 65535

4. 整体通信系统示例

服务器端

import socket
from cryptography.fernet import Fernet
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes

# 生成密钥
key = Fernet.generate_key()
cipher_suite = Fernet(key)

# 生成私钥
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)

# 生成公钥
public_key = private_key.public_key()

# 数据签名
def sign_data(data, private_key):
    signature = private_key.sign(
        data.encode(),
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH
        ),
        hashes.SHA256()
    )
    return signature

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.bind(('127.0.0.1', 12345))
sock.listen(1)

while True:
    conn, addr = sock.accept()
    # 发送公钥给客户端
    public_pem = public_key.public_bytes(
        encoding=serialization.Encoding.PEM,
        format=serialization.PublicFormat.SubjectPublicKeyInfo
    )
    conn.sendall(public_pem)

    # 接收并验证数据
    encrypted_data = conn.recv(1024)
    decrypted_data = cipher_suite.decrypt(encrypted_data).decode()
    signature = conn.recv(1024)
    if verify_signature(decrypted_data, signature, public_key):
        print(f"接收到数据: {decrypted_data}")
        conn.sendall(b'ACK')
    else:
        print("签名验证失败")
    conn.close()

客户端

import socket
from cryptography.fernet import Fernet
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa, padding
from cryptography.hazmat.primitives import serialization, hashes

# 生成密钥
key = Fernet.generate_key()
cipher_suite = Fernet(key)

# 生成私钥
private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)

# 生成公钥
public_key = private_key.public_key()

# 数据签名
def sign_data(data, private_key):
    signature = private_key.sign(
        data.encode(),
        padding.PSS(
            mgf=padding.MGF1(hashes.SHA256()),
            salt_length=padding.PSS.MAX_LENGTH
        ),
        hashes.SHA256()
    )
    return signature

# 验证签名
def verify_signature(data, signature, public_key):
    try:
        public_key.verify(
            signature,
            data.encode(),
            padding.PSS(
                mgf=padding.MGF1(hashes.SHA256()),
                salt_length=padding.PSS.MAX_LENGTH
            ),
            hashes.SHA256()
        )
        return True
    except:
        return False

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('127.0.0.1', 12345))

# 接收服务器公钥
public_pem = sock.recv(1024)
server_public_key = serialization.load_pem_public_key(
    public_pem,
    backend=default_backend()
)

data = "Hello, Server!"
encrypted_data = cipher_suite.encrypt(data.encode())
signature = sign_data(data, private_key)

sock.sendall(encrypted_data)
sock.sendall(signature)

ack = sock.recv(1024)
if ack.decode() == 'ACK':
    print("数据发送成功")
sock.close()

通过以上设计方案，能在一定程度上保证在复杂网络环境下通信系统的高度安全且可靠，实现数据加密、防止中间人攻击以及处理各类网络异常情况以确保数据完整性和通信连续性。

星途面试题库