TCP Socket服务器随机挂起及rsa.pkcs1.DecryptionError异常求助

问题描述 #

• 服务器无规律挂起，随机抛出rsa.pkcs1.DecryptionError解密错误，多数报错集中在密码解密环节
• 曾切换数据接收终止逻辑（从发送END消息改为测试脚本主动关闭连接，服务器通过if socket.fileno() == -1判断连接状态），但问题在逻辑切换前已存在
• 多次校验公私钥、逐行排查代码未解决问题，恢复END消息逻辑也无效；单独打印解密后密码多数正常，偶尔仍出现解密错误

服务器代码 #

import json
import rsa
import socket
import os.path
import netifaces
import json_repair

def get_connection_data():
    if os.path.exists("data/data.json"):
        connection_data = tuple(json_repair.from_file("data/data.json"))
    else:
        interface = netifaces.interfaces()[int(input(f"Enter the index of the interface you want to use. Eg, to use {netifaces.interfaces()[0]} enter 1\n{netifaces.interfaces()}\n")) - 1]
        port = input("What port do you want to user? Press enter for the default \n")
        port = 9000 if port == "" else port
        connection_data = (
            netifaces.ifaddresses(interface)[netifaces.AF_INET][0]["addr"],
            port
        )
        if not os.path.exists("data"):
            os.mkdir("data")
        with open("data/data.json", mode="w") as connection_file:
            json.dump(connection_data, connection_file, indent=4)
    return connection_data

def register_user(user_client, device_address: float):
    print("register_user called")
    if not os.path.exists("data"):
        os.mkdir("data")
    if not os.path.exists(f"data/{device_address}"):
        os.mkdir(f"data/{device_address}")
    all_passwords = {}

    while True:
        service = rsa.decrypt(user_client.recv(1024), priv_key=private_key).decode()
        print("Service is:" + service)

        while service != "END":
            user_client.send("READY".encode())
            username = rsa.decrypt(user_client.recv(1024), priv_key=private_key).decode()
            print("Username is: " + username)
            user_client.send("READY".encode())
            print("Sent ready after receiving username")

            password = rsa.decrypt(user_client.recv(1024), priv_key=private_key).decode()
            print(password)
            user_client.send("READY".encode())
            print("Sent ready after receiving password")

            key = rsa.decrypt(user_client.recv(1024), priv_key=private_key)
            user_client.send("READY".encode())
            print("Sent ready after receiving key")

            if user_client.fileno() == -1:
                break
        if user_client.fileno() == -1:
            break

    all_passwords[service] = {
        username: {
            "password": password,
            "key": key.decode()
        }
    }
    with open(f"data/{device_address}/.passwords.json", mode="w") as passwords_file:
        json.dump(all_passwords, passwords_file, indent=4)

public_key, private_key = rsa.newkeys(1024)
client_public_key = None

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(get_connection_data())
server.listen()

print(f"Listening on {get_connection_data()[0]}:{get_connection_data()[1]}")

while True:
    client, ip_address = server.accept()
    print("Connection accepted")
    client.send(public_key.save_pkcs1("PEM"))
    client_public_key = rsa.PublicKey.load_pkcs1(client.recv(1024))
    print("Received and sent Public keys")

    action = client.recv(1024)
    client.send("READY".encode())
    print("Sent ready message")

    if action.decode().lower() == "register user":
        print("Inside if statement")
        register_user(user_client=client, device_address=ip_address[0])

测试脚本代码 #

import socket
import time
import rsa

IP_ADDRESS = input("Enter server IP: ")
PORT = 9000

pub_key, priv_key = rsa.newkeys(1024)
server_pub_key = None

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
print("Connecting to server")
server.connect((IP_ADDRESS, PORT))

print("Receiving server public key and sending public key")
server_pub_key = rsa.PublicKey.load_pkcs1(server.recv(1024))
server.send(pub_key.save_pkcs1("PEM"))

message = "register user"
server.send(message.encode("utf-8"))
print("Sent action")

time.sleep(1)
print("Waiting for ready message")
server.recv(1024)
print("Received ready message")

all_passwords = {
    "service": {
        "username": {"password": "password", "key": "key"},
        "username2": {"password": "password2", "key": "key2"}
    },
    "service2": {
        "username": {"password": "password", "key": "key"},
        "username2": {"password": "password2", "key": "key2"}
    }
}

for service in all_passwords:
    service_data = all_passwords[service]
    print("Sending service")
    server.send(rsa.encrypt(service.encode(), pub_key=server_pub_key))
    server.recv(1024)

    for username in service_data.keys():
        password = service_data[username]["password"]
        key = service_data[username]["key"]
        print(f"Service: {service} \nUsername: {username} \nPassword: {password} \nKey: {key}")
        
        server.send(rsa.encrypt(username.encode(), pub_key=server_pub_key))
        server.recv(1024)
        print("Received ready after sending username")

        server.send(rsa.encrypt(password.encode(), pub_key=server_pub_key))
        server.recv(1024)
        print("Received ready after sending password")

        server.send(rsa.encrypt(key.encode(), pub_key=server_pub_key))
        server.recv(1024)
        print("received ready after sending key")

server.close()

问题根源分析 #

1. TCP流式传输的粘包/截断问题：socket.recv(1024)无法保证每次接收完整的RSA密文（1024位密钥加密后密文固定为128字节），若单次recv只拿到部分密文或多段密文粘在一起，直接解密会触发DecryptionError。
2. 循环逻辑漏洞：register_user内层while service != "END"未更新service值，会陷入死循环；最终保存all_passwords仅处理最后一组数据，之前的服务数据全部丢失。
3. 连接状态判断失效：socket.fileno() == -1仅在连接被内核清理后才会触发，无法及时检测客户端异常断开，且解密操作在判断前执行，已触发错误。
4. 不可靠的同步方式：测试脚本用time.sleep(1)做同步，依赖网络延迟，容易导致收发时序混乱。

修复与优化方案 #

1. 确保完整接收RSA密文 #

封装固定长度接收函数，保证每次拿到完整的RSA加密数据：

# 服务器端新增完整接收函数
def recv_all(sock, length):
    data = b''
    while len(data) < length:
        chunk = sock.recv(length - len(data))
        if not chunk:
            raise ConnectionError("客户端连接已断开")
        data += chunk
    return data

# 替换所有rsa.decrypt前的recv调用，例如：
service = rsa.decrypt(recv_all(user_client, 128), priv_key=private_key).decode()

2. 修复循环与数据保存逻辑 #

修改register_user函数，完善终止标识逻辑，确保所有数据被正确保存：

def register_user(user_client, device_address):
    print("register_user called")
    if not os.path.exists("data"):
        os.mkdir("data")
    device_dir = f"data/{device_address}"
    if not os.path.exists(device_dir):
        os.mkdir(device_dir)
    all_passwords = {}

    try:
        while True:
            # 接收服务名，用"END"终止所有服务接收
            service = rsa.decrypt(recv_all(user_client, 128), priv_key=private_key).decode()
            print(f"Service is: {service}")
            if service == "END":
                break
            user_client.send("READY".encode())
            
            service_accounts = {}
            while True:
                # 接收用户名，用"END_ACCOUNT"终止当前服务的账号接收
                username = rsa.decrypt(recv_all(user_client, 128), priv_key=private_key).decode()
                if username == "END_ACCOUNT":
                    user_client.send("READY".encode())
                    break
                print(f"Username is: {username}")
                user_client.send("READY".encode())

                password = rsa.decrypt(recv_all(user_client, 128), priv_key=private_key).decode()
                print(password)
                user_client.send("READY".encode())

                key = rsa.decrypt(recv_all(user_client, 128), priv_key=private_key).decode()
                user_client.send("READY".encode())

                service_accounts[username] = {"password": password, "key": key}
            
            all_passwords[service] = service_accounts
    except ConnectionError:
        print("客户端连接异常断开")
    finally:
        # 无论连接是否正常终止，都保存已接收的数据
        with open(f"{device_dir}/.passwords.json", mode="w") as passwords_file:
            json.dump(all_passwords, passwords_file, indent=4)

同时修改测试脚本，添加终止标识：

# 测试脚本循环末尾添加终止逻辑
for service in all_passwords:
    # ... 原有发送逻辑 ...
    # 发送当前服务账号结束标识
    server.send(rsa.encrypt("END_ACCOUNT".encode(), pub_key=server_pub_key))
    server.recv(1024)
# 发送所有服务结束标识
server.send(rsa.encrypt("END".encode(), pub_key=server_pub_key))
server.recv(1024)
server.close()

3. 替换不可靠同步方式 #

删除测试脚本中的time.sleep(1)，完全依赖"READY"消息做同步，确保收发时序正确。

4. 添加异常捕获避免服务器挂起 #

在主循环中捕获异常，单个客户端错误不会影响整个服务器：

# 主循环中调用register_user的部分
if action.decode().lower() == "register user":
    print("Inside if statement")
    try:
        register_user(user_client=client, device_address=ip_address[0])
    except Exception as e:
        print(f"处理注册请求出错: {str(e)}")
    finally:
        client.close()

5. 长期优化建议 #

• 改用对称加密+RSA密钥交换：用RSA仅交换AES对称密钥，实际数据用AES加密，避免RSA的长度限制和性能瓶颈，更高效安全。
• 使用成熟通信框架：替换原生socket，改用websockets、gRPC或HTTP协议，减少底层流式传输的bug。
• 添加数据校验：每个加密数据包附加校验和（如SHA256），确保数据完整性，避免篡改或截断导致的解密错误。
• 完善日志系统：记录请求细节和错误信息，便于快速定位问题。

内容的提问来源于stack exchange，提问作者SmallCoder12