基于Django Channels的Shamir秘密共享电子投票原型可行性及本地WSS部署咨询

作为做过实时协作加密项目的开发者，我可以明确告诉你：你的方案非常合理且完全可行。Shamir's Secret Sharing（SSS）天生适配“多方共同解密”的场景，而Django Channels刚好能帮你实现投票者状态的实时同步和结果的即时推送，两者结合的逻辑闭环非常顺畅。

一、方案可行性详解 #

你的核心需求是“所有投票者完成密钥输入后才展示解密后的投票结果”，这个流程用SSS + Django Channels完全能实现：

• 投票阶段：收集完所有投票后，用SSS把解密密钥拆分成N个分片（N等于投票者数量），每个投票者对应一个分片
• 等待阶段：通过WebSocket实时追踪每个投票者的登录状态和密钥提交情况
• 触发阶段：当最后一个投票者提交分片后，后端自动合并所有分片还原密钥，解密投票结果，再通过WebSocket实时推送给所有在线用户

我之前用类似架构做过多方密钥协作的系统，逻辑上没有任何问题，只要把SSS的解密逻辑和Channels的实时事件触发做好整合就行。

二、本地运行WSS（WebSocket Secure）完全可行 #

本地开发时完全可以搭建WSS，只需要生成自签名SSL证书就行。虽然浏览器会弹出“证书不受信任”的警告，但这只是因为证书不是正规CA签发的，完全不影响开发测试。等上线生产环境时，换成Let's Encrypt之类的免费可信证书就可以了。

三、具体操作步骤 #

1. 安装依赖 #

先把需要的包装好：

pip install django channels channels-redis

• channels：Django生态的WebSocket核心框架
• channels-redis：作为Channels的通道层后端（需要先在本地装Redis服务，嫌麻烦的话开发初期也可以用内存后端，但Redis更稳定）

2. 修改Django项目配置 #

• 在settings.py里把Channels加入INSTALLED_APPS：

INSTALLED_APPS = [
    # ... 你原来的app
    'channels',
]

• 替换ASGI应用（替代原来的WSGI）：

ASGI_APPLICATION = '你的项目名.asgi.application'

• 配置通道层（以Redis为例）：

CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

3. 编写WebSocket消费者（核心逻辑） #

在你的app里新建consumers.py，处理投票者状态同步和结果推送：

from channels.generic.websocket import AsyncWebsocketConsumer
from channels.db import database_sync_to_async
import json
from .models import VoteSession, Voter

class VoteResultConsumer(AsyncWebsocketConsumer):
    async def connect(self):
        # 加入对应投票会话的WebSocket组
        self.session_id = self.scope['url_route']['kwargs']['session_id']
        self.group_name = f'vote_session_{self.session_id}'

        await self.channel_layer.group_add(
            self.group_name,
            self.channel_name
        )
        await self.accept()

        # 连接成功后先检查是否所有投票者已就绪
        await self.check_all_voters_ready()

    async def disconnect(self, close_code):
        await self.channel_layer.group_discard(
            self.group_name,
            self.channel_name
        )

    # 接收前端发来的密钥分片
    async def receive(self, text_data):
        text_data_json = json.loads(text_data)
        voter_id = text_data_json['voter_id']
        secret_shard = text_data_json['secret_shard']

        # 保存分片到数据库（同步操作转异步）
        await database_sync_to_async(self.save_secret_shard)(voter_id, secret_shard)

        # 保存后再次检查是否全员就绪
        await self.check_all_voters_ready()

    def save_secret_shard(self, voter_id, secret_shard):
        voter = Voter.objects.get(id=voter_id)
        voter.secret_shard = secret_shard
        voter.is_ready = True
        voter.save()

    async def check_all_voters_ready(self):
        # 查询当前会话的所有投票者是否都提交了分片
        session = await database_sync_to_async(VoteSession.objects.get)(id=self.session_id)
        all_ready = await database_sync_to_async(
            lambda: session.voters.filter(is_ready=True).count() == session.voters.count()
        )()

        if all_ready:
            # 合并Shamir分片，解密投票结果
            result = await database_sync_to_async(self.decrypt_vote_result)(session)
            # 推送给组内所有在线用户
            await self.channel_layer.group_send(
                self.group_name,
                {
                    'type': 'vote_result_message',
                    'result': result
                }
            )

    def decrypt_vote_result(self, session):
        # 这里替换成你的Shamir密钥合并和解密逻辑
        # 示例：收集所有分片，用SSS库合并出密钥，再解密结果
        shards = [v.secret_shard for v in session.voters.all()]
        from shamir import merge_shards  # 假设你用的Shamir库有这个函数
        secret_key = merge_shards(shards)
        # 解密投票结果的逻辑...
        return {"result": "投票结果：选项A 5票，选项B 3票"}

    # 把结果发送给前端
    async def vote_result_message(self, event):
        result = event['result']
        await self.send(text_data=json.dumps({
            'result': result
        }))

4. 配置WebSocket路由 #

在项目根目录新建routing.py：

from django.urls import re_path
from . import consumers

websocket_urlpatterns = [
    re_path(r'ws/vote/(?P<session_id>\d+)/$', consumers.VoteResultConsumer.as_asgi()),
]

然后修改asgi.py整合路由：

import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.auth import AuthMiddlewareStack
import 你的项目名.routing

os.environ.setdefault('DJANGO_SETTINGS_MODULE', '你的项目名.settings')

application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            你的项目名.routing.websocket_urlpatterns
        )
    ),
})

5. 本地WSS配置与前端连接 #

首先生成自签名SSL证书：

# 生成私钥
openssl genrsa -out localhost.key 2048
# 生成证书签名请求
openssl req -new -key localhost.key -out localhost.csr
# 生成自签名证书
openssl x509 -req -days 365 -in localhost.csr -signkey localhost.key -out localhost.crt

然后用带SSL的命令启动Django：

python manage.py runserver_plus --cert localhost.crt --key localhost.key

前端连接WSS的示例代码（JS）：

const sessionId = 1; // 替换为实际的投票会话ID
const socket = new WebSocket(`wss://localhost:8000/ws/vote/${sessionId}/`);

socket.onopen = function(e) {
    console.log("WebSocket连接成功");
};

socket.onmessage = function(event) {
    const data = JSON.parse(event.data);
    if (data.result) {
        // 把结果展示到页面上
        document.getElementById("vote-result").innerText = data.result;
    }
};

// 提交密钥分片的函数
function submitSecretShard(voterId, secretShard) {
    socket.send(JSON.stringify({
        'voter_id': voterId,
        'secret_shard': secretShard
    }));
}

四、小提示 #

• 本地WSS的浏览器警告：可以在浏览器里手动信任自签名证书，或者开发初期先用WS（非加密）测试逻辑，最后再切换到WSS
• Shamir密钥的安全性：虽然WSS已经加密了传输，但存储分片时最好在数据库里做个加密，更稳妥
• 通道层替代：如果不想装Redis，开发时可以用channels.layers.InMemoryChannelLayer，但生产环境必须用Redis或其他分布式通道层

内容的提问来源于stack exchange，提问作者B.Ghost