# -*- coding: utf-8 -*- """ ☁️ 混沌加密聊天 - 中央服务器 (解决 NAT / 公网穿透) ==================================================== - 服务器只负责: 用户认证 + 群成员管理 + 消息中继 - 所有消息内容由客户端加密, 服务器【不持有群密钥】 - 密钥只在客户端本地, 永不上传到服务器 - 部署: 放到有公网IP的机器, python server.py [端口] """ import os import sys import json import time import base64 import struct import socket import hashlib import threading import sqlite3 SERVER_HOST = '0.0.0.0' SERVER_PORT = 8666 DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'server_data', 'server.db') MAX_FRAME = 256 * 1024 * 1024 # ==================== 帧协议 (与客户端一致) ==================== def send_msg(sock, obj): payload = json.dumps(obj, ensure_ascii=False).encode('utf-8') sock.sendall(struct.pack('>I', len(payload)) + payload) def recv_exact(sock, n): data = b'' while len(data) < n: try: chunk = sock.recv(n - len(data)) except socket.timeout: continue except Exception: return None if not chunk: return None data += chunk return data def recv_msg(sock): header = recv_exact(sock, 4) if header is None: return None length = struct.unpack('>I', header)[0] if length <= 0 or length > MAX_FRAME: return None payload = recv_exact(sock, length) if payload is None: return None try: return json.loads(payload.decode('utf-8')) except Exception: return None # ==================== 服务器数据库 ==================== class ServerDB: def __init__(self, db_path=DB_PATH): os.makedirs(os.path.dirname(db_path), exist_ok=True) self.conn = sqlite3.connect(db_path, check_same_thread=False) self.conn.row_factory = sqlite3.Row self.lock = threading.Lock() self._init() def _init(self): with self.lock: cur = self.conn.cursor() cur.execute('''CREATE TABLE IF NOT EXISTS users( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL, salt TEXT NOT NULL, pass_hash TEXT NOT NULL, created_at REAL)''') cur.execute('''CREATE TABLE IF NOT EXISTS groups( id TEXT PRIMARY KEY, owner TEXT NOT NULL, name TEXT, announcement TEXT, created_at REAL)''') cur.execute('''CREATE TABLE IF NOT EXISTS group_members( group_id TEXT NOT NULL, username TEXT NOT NULL, joined_at REAL, PRIMARY KEY(group_id, username))''') self.conn.commit() def register(self, username, password): salt = os.urandom(16).hex() ph = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), bytes.fromhex(salt), 120000).hex() try: with self.lock: cur = self.conn.cursor() cur.execute('INSERT INTO users(username, salt, pass_hash, created_at) VALUES(?,?,?,?)', (username, salt, ph, time.time())) self.conn.commit() return True, 'ok' except sqlite3.IntegrityError: return False, '该用户名已被注册' def login(self, username, password): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT * FROM users WHERE username=?', (username,)).fetchone() if row is None: return None, '用户不存在' ph = hashlib.pbkdf2_hmac('sha256', password.encode('utf-8'), bytes.fromhex(row['salt']), 120000).hex() if ph != row['pass_hash']: return None, '密码错误' return row, 'ok' def create_group(self, group_id, owner, name): with self.lock: cur = self.conn.cursor() cur.execute('INSERT INTO groups(id, owner, name, created_at) VALUES(?,?,?,?)', (group_id, owner, name, time.time())) cur.execute('INSERT OR IGNORE INTO group_members(group_id, username, joined_at) VALUES(?,?,?)', (group_id, owner, time.time())) self.conn.commit() def add_member(self, group_id, username): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT 1 FROM groups WHERE id=?', (group_id,)).fetchone() if row is None: return False cur.execute('INSERT OR IGNORE INTO group_members(group_id, username, joined_at) VALUES(?,?,?)', (group_id, username, time.time())) self.conn.commit() return True def remove_member(self, group_id, username): with self.lock: cur = self.conn.cursor() cur.execute('DELETE FROM group_members WHERE group_id=? AND username=?', (group_id, username)) self.conn.commit() def get_group(self, group_id): with self.lock: cur = self.conn.cursor() return cur.execute('SELECT id, name, announcement, owner FROM groups WHERE id=?', (group_id,)).fetchone() def list_groups(self, username): with self.lock: cur = self.conn.cursor() rows = cur.execute('''SELECT g.id, g.name, g.owner, g.announcement FROM groups g JOIN group_members m ON m.group_id = g.id WHERE m.username=?''', (username,)).fetchall() return [{'id': r['id'], 'name': r['name'], 'owner': r['owner'], 'announcement': r['announcement'] or ''} for r in rows] def is_owner(self, group_id, username): with self.lock: cur = self.conn.cursor() row = cur.execute('SELECT owner FROM groups WHERE id=?', (group_id,)).fetchone() return bool(row and row['owner'] == username) def set_group_name(self, group_id, name): with self.lock: cur = self.conn.cursor() cur.execute('UPDATE groups SET name=? WHERE id=?', (name, group_id)) self.conn.commit() def set_announcement(self, group_id, text): with self.lock: cur = self.conn.cursor() cur.execute('UPDATE groups SET announcement=? WHERE id=?', (text, group_id)) self.conn.commit() def members(self, group_id): with self.lock: cur = self.conn.cursor() rows = cur.execute('SELECT username FROM group_members WHERE group_id=?', (group_id,)).fetchall() return [r['username'] for r in rows] # ==================== 中央服务器 ==================== class CentralServer: def __init__(self, host=SERVER_HOST, port=SERVER_PORT, db=None): self.host = host self.port = port self.db = db or ServerDB() self.sock = None self.running = False self.clients = {} # cid -> {socket, username, token, addr, lock} self.next_id = 1 self.lock = threading.Lock() def start(self): self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.sock.bind((self.host, self.port)) self.sock.listen(50) self.running = True print(f"☁️ 中央服务器启动: {self.host}:{self.port}") print(f" 数据库: {DB_PATH}") threading.Thread(target=self._accept_loop, daemon=True).start() def stop(self): self.running = False if self.sock: try: self.sock.close() except Exception: pass self.sock = None with self.lock: socks = [c['socket'] for c in self.clients.values()] self.clients.clear() for s in socks: try: s.close() except Exception: pass def _accept_loop(self): while self.running: try: client, addr = self.sock.accept() with self.lock: cid = self.next_id self.next_id += 1 self.clients[cid] = {'socket': client, 'username': None, 'token': None, 'addr': addr, 'lock': threading.Lock()} print(f" 📶 新连接: {addr[0]}:{addr[1]}") threading.Thread(target=self._handler, args=(cid,), daemon=True).start() except Exception: break def _send(self, cid, msg): with self.lock: c = self.clients.get(cid) if c: try: with c['lock']: send_msg(c['socket'], msg) except Exception: pass def _handler(self, cid): client = self.clients[cid]['socket'] # 第一步: 认证 (register 自动登录 或 login) try: msg = recv_msg(client) if msg and msg.get('type') == 'register': username = str(msg.get('username', ''))[:24] password = str(msg.get('password', '')) ok, err = self.db.register(username, password) if not ok: self._send(cid, {'type': 'error', 'msg': err}) self._remove(cid) return row, _ = self.db.login(username, password) self._auth_ok(cid, row) elif msg and msg.get('type') == 'login': username = str(msg.get('username', ''))[:24] password = str(msg.get('password', '')) row, err = self.db.login(username, password) if row is None: self._send(cid, {'type': 'error', 'msg': err}) self._remove(cid) return self._auth_ok(cid, row) else: self._remove(cid) return except Exception: self._remove(cid) return # 第二步: 业务消息循环 while self.running: if cid not in self.clients: break try: msg = recv_msg(client) if msg is None: break self._handle(cid, msg) except Exception: break self._remove(cid) def _auth_ok(self, cid, row): username = row['username'] token = os.urandom(16).hex() with self.lock: if cid in self.clients: self.clients[cid]['username'] = username self.clients[cid]['token'] = token print(f" 👤 {username} 已登录") self._send(cid, {'type': 'login_ok', 'token': token, 'username': username}) def _handle(self, cid, msg): with self.lock: if cid not in self.clients: return username = self.clients[cid]['username'] mtype = msg.get('type') if mtype == 'create_group': group_id = str(msg.get('group_id') or f"g{int(time.time()*1000)}-{os.urandom(3).hex()}") name = str(msg.get('name', ''))[:40] self.db.create_group(group_id, username, name or group_id) print(f" 📡 {username} 创建群 [{name}] ({group_id})") self._send(cid, {'type': 'group_created', 'group_id': group_id, 'name': name}) elif mtype == 'list_groups': groups = self.db.list_groups(username) self._send(cid, {'type': 'groups', 'groups': groups}) elif mtype == 'join_group': group_id = str(msg.get('group_id', '')) if self.db.add_member(group_id, username): self._send(cid, {'type': 'join_ok', 'group_id': group_id}) # 推送当前群名/公告给新成员 ginfo = self.db.get_group(group_id) self._send(cid, {'type': 'group_info', 'group_id': group_id, 'name': ginfo['name'] if ginfo else '', 'announcement': ginfo['announcement'] if ginfo else ''}) self._broadcast_group(group_id, {'type': 'system', 'text': f"👋 {username} 加入了群", 'group_id': group_id}, exclude=cid) else: self._send(cid, {'type': 'error', 'msg': '群不存在'}) elif mtype == 'set_group_name': group_id = str(msg.get('group_id', '')) name = str(msg.get('name', ''))[:40] if self.db.is_owner(group_id, username): self.db.set_group_name(group_id, name) self._broadcast_group(group_id, {'type': 'group_name', 'group_id': group_id, 'name': name}, exclude=None) print(f" 🏷️ {username} 修改群名: {name}") else: self._send(cid, {'type': 'error', 'msg': '仅群主可修改群名'}) elif mtype == 'set_announcement': group_id = str(msg.get('group_id', '')) text = str(msg.get('text', ''))[:500] if self.db.is_owner(group_id, username): self.db.set_announcement(group_id, text) self._broadcast_group(group_id, {'type': 'announcement', 'group_id': group_id, 'text': text}, exclude=None) print(f" 📢 {username} 设置公告: {text[:30]}") else: self._send(cid, {'type': 'error', 'msg': '仅群主可设置公告'}) elif mtype == 'leave_group': group_id = str(msg.get('group_id', '')) self.db.remove_member(group_id, username) self._send(cid, {'type': 'leave_ok', 'group_id': group_id}) self._broadcast_group(group_id, {'type': 'system', 'text': f"👋 {username} 离开了群", 'group_id': group_id}, exclude=cid) elif mtype == 'group_members': group_id = str(msg.get('group_id', '')) self._send(cid, {'type': 'members', 'group_id': group_id, 'members': self.db.members(group_id)}) elif mtype == 'group_message': group_id = str(msg.get('group_id', '')) payload = msg.get('payload') # 转发给同群其他成员 (服务器不解析加密内容) self._broadcast_group(group_id, {'type': 'group_message', 'group_id': group_id, 'from': username, 'payload': payload}, exclude=cid) def _broadcast_group(self, group_id, msg, exclude=None): members = set(self.db.members(group_id)) with self.lock: targets = [(cid2, c['socket'], c['lock']) for cid2, c in self.clients.items() if c.get('username') in members and cid2 != exclude] for cid2, sock, lk in targets: try: with lk: send_msg(sock, msg) except Exception: pass def _remove(self, cid): with self.lock: entry = self.clients.pop(cid, None) if entry: try: entry['socket'].close() except Exception: pass if entry['username']: print(f" 🔌 {entry['username']} 断开") if __name__ == '__main__': port = int(sys.argv[1]) if len(sys.argv) > 1 else SERVER_PORT srv = CentralServer(port=port) srv.start() print("按 Ctrl+C 停止服务器") try: while True: time.sleep(1) except KeyboardInterrupt: srv.stop() print("\n☁️ 服务器已停止")