419 lines
16 KiB
Python
419 lines
16 KiB
Python
# -*- 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☁️ 服务器已停止")
|