Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bad851f741 | ||
|
|
b311483963 | ||
|
|
ef90efd199 | ||
|
|
587ef2140b | ||
|
|
c352eef2ba | ||
|
|
124b511d5f | ||
|
|
79f4ad8b78 |
+20
@@ -0,0 +1,20 @@
|
||||
# 运行时数据(聊天记录/账户/服务器状态)
|
||||
data/
|
||||
server_data/
|
||||
ChaosCryptChat/data/
|
||||
ChaosCryptChat/server_data/
|
||||
|
||||
# 密钥文件(严禁上传!)
|
||||
*.key
|
||||
|
||||
# Python 缓存
|
||||
__pycache__/
|
||||
*.pyc
|
||||
|
||||
# 媒体缓存
|
||||
media/
|
||||
|
||||
# 系统文件
|
||||
Thumbs.db
|
||||
.DS_Store
|
||||
desktop.ini
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,418 @@
|
||||
# -*- 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☁️ 服务器已停止")
|
||||
+67
@@ -765,4 +765,71 @@ DeriveKey(K\_user, K\_short, n) 包含:
|
||||
================================================================================
|
||||
|
||||
|
||||
================================================================================
|
||||
十、ChaosCryptChat 聊天模块安全分析(v2.0)
|
||||
================================================================================
|
||||
|
||||
本部分分析 ChaosCryptChat(端到端加密聊天系统)的安全模型。
|
||||
|
||||
10.1 威胁模型
|
||||
|
||||
假设:
|
||||
- 中央服务器可被攻击者完全控制(最坏情况)
|
||||
- 网络传输可被监听、篡改、重放
|
||||
- 客户端本地文件可信(不讨论端侧木马)
|
||||
|
||||
10.2 端到端加密
|
||||
|
||||
消息路径:
|
||||
明文 → 混沌加密(群密钥 + nonce 派生消息密钥)→ 密文 → 网络 → 密文 → 混沌解密 → 明文
|
||||
|
||||
关键属性:
|
||||
- 群密钥只存在于客户端本地,永不上传服务器
|
||||
- 服务器仅中继密文,无法解密任何消息
|
||||
- 每条消息使用独立 nonce 派生消息密钥,避免重放
|
||||
- HMAC-SHA256 完整性校验:密文被篡改将导致解密失败
|
||||
|
||||
10.3 中央服务器模型
|
||||
|
||||
服务器职责:用户认证 + 群成员管理 + 消息中继
|
||||
|
||||
服务器不持有:
|
||||
- 群密钥(session key)
|
||||
- 消息明文
|
||||
- 每用户密钥(user key,经服务器持有密钥加密后下发)
|
||||
|
||||
服务器被攻陷的影响:
|
||||
- 无法解密历史或实时聊天内容
|
||||
- 可能泄露:用户名、登录时间、中继日志、群成员关系
|
||||
- 可实施:拒绝服务、成员关系观察(元数据泄露)
|
||||
|
||||
10.4 身份认证
|
||||
|
||||
用户密码:PBKDF2-HMAC-SHA256,120,000 次迭代 + 随机盐
|
||||
每用户密钥:加入群时签发,用于 HMAC 签名,防止成员冒充他人
|
||||
群密钥认证:加入群需提供群密钥派生的认证凭据,证明是群成员
|
||||
|
||||
10.5 消息完整性
|
||||
|
||||
文本/语音/文件消息均携带 HMAC:
|
||||
- 文本: HMAC(密文 + 群密钥)
|
||||
- 二进制: HMAC('BIN|' + 群密钥 + nonce + 密文)
|
||||
|
||||
篡改检测:接收方重算 HMAC 不匹配 → 拒绝消息
|
||||
|
||||
10.6 已知限制
|
||||
|
||||
1. 端到端(P2P)模式下,群主主机既是聊天者又是服务器,若群主被攻陷则群聊失守
|
||||
2. 服务器虽无法解密内容,但能观察到元数据(谁和谁在何时通信)
|
||||
3. 群密钥通过群主手动分享,存在被截获的风险(需可信渠道传递)
|
||||
4. 语音播放/录音依赖第三方库(pygame/sounddevice),其安全性不在本系统保证范围
|
||||
5. 消息撤回仅做本地标记 + 服务器广播,已离线成员仍可能看到撤回前的消息
|
||||
|
||||
10.7 结论
|
||||
|
||||
ChaosCryptChat 提供端到端加密 + 完整性校验 + 身份认证,服务器无法解密内容。
|
||||
其安全边界符合"服务器不可信"模型,适合对隐私有要求但不涉及合规认证的通信场景。
|
||||
元数据泄露与密钥分享渠道仍需用户自行权衡与管理。
|
||||
|
||||
================================================================================
|
||||
|
||||
|
||||
+343
-231
@@ -1,3 +1,7 @@
|
||||
"""
|
||||
A multi-layer encryption system with key management, substitution ciphers,
|
||||
XOR operations, and dynamic key derivation.
|
||||
"""
|
||||
|
||||
import base64
|
||||
import json
|
||||
@@ -5,12 +9,23 @@ import os
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
|
||||
class EncryptionSystem:
|
||||
"""Main encryption engine with key-based transformation layers."""
|
||||
|
||||
def __init__(self, key_file=None, key_password=None):
|
||||
"""
|
||||
Initialize the encryption system with an optional key file.
|
||||
|
||||
Args:
|
||||
key_file: Path to the encryption key file.
|
||||
key_password: Password for decrypting the key file.
|
||||
"""
|
||||
self.key_file = key_file
|
||||
self.key_password = key_password
|
||||
self.keys_loaded = False
|
||||
|
||||
|
||||
# Base64 special character mappings for safe transport
|
||||
self.special_encrypt = {
|
||||
'+': '.',
|
||||
'/': "'",
|
||||
@@ -20,41 +35,49 @@ class EncryptionSystem:
|
||||
'.': '+',
|
||||
"'": '/'
|
||||
}
|
||||
|
||||
|
||||
self.flip_pattern = None
|
||||
|
||||
|
||||
# Try loading the key file if provided or find default
|
||||
if key_file:
|
||||
if os.path.exists(key_file):
|
||||
if key_password is None:
|
||||
key_password = input(f"请输入密钥文件 {key_file} 的密码: ")
|
||||
key_password = input(f"Enter password for key file {key_file}: ")
|
||||
self._load_keys(key_file, key_password)
|
||||
self.keys_loaded = True
|
||||
else:
|
||||
print(f"⚠️ 密钥文件 {key_file} 不存在")
|
||||
print(f"⚠️ Key file {key_file} not found")
|
||||
self.keys_loaded = False
|
||||
else:
|
||||
default_key = "encryption.key"
|
||||
if os.path.exists(default_key):
|
||||
self.key_file = default_key
|
||||
if key_password is None:
|
||||
key_password = input(f"请输入密钥文件 {default_key} 的密码: ")
|
||||
key_password = input(f"Enter password for key file {default_key}: ")
|
||||
self._load_keys(default_key, key_password)
|
||||
self.keys_loaded = True
|
||||
else:
|
||||
print("=" * 60)
|
||||
print("首次启动,请先生成密钥文件")
|
||||
print("First launch detected. Please generate a key file first.")
|
||||
print("=" * 60)
|
||||
self.keys_loaded = False
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Key generation utilities
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def _generate_random_alphabet(self, lowercase=False):
|
||||
"""Generate a shuffled alphabet string."""
|
||||
chars = list("abcdefghijklmnopqrstuvwxyz" if lowercase else "ABCDEFGHIJKLMNOPQRSTUVWXYZ")
|
||||
n = len(chars)
|
||||
# Fisher-Yates shuffle using secure random bytes
|
||||
for i in range(n - 1, 0, -1):
|
||||
j = int.from_bytes(os.urandom(1), 'big') % (i + 1)
|
||||
chars[i], chars[j] = chars[j], chars[i]
|
||||
return ''.join(chars)
|
||||
|
||||
def _generate_random_digit_mapping(self):
|
||||
"""Create a random mapping for digits 0-9 to alphabet characters."""
|
||||
digits = list("0123456789")
|
||||
mapping_chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")
|
||||
n = len(mapping_chars)
|
||||
@@ -67,72 +90,69 @@ class EncryptionSystem:
|
||||
return mapping
|
||||
|
||||
def _generate_random_equal_mapping(self):
|
||||
"""Generate mapping for Base64 padding count (0-3)."""
|
||||
chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")
|
||||
n = len(chars)
|
||||
for i in range(n - 1, 0, -1):
|
||||
j = int.from_bytes(os.urandom(1), 'big') % (i + 1)
|
||||
chars[i], chars[j] = chars[j], chars[i]
|
||||
mapping = {
|
||||
return {
|
||||
'0': chars[0],
|
||||
'1': chars[1],
|
||||
'2': chars[2],
|
||||
'3': chars[3]
|
||||
}
|
||||
return mapping
|
||||
|
||||
def _generate_random_flip_pattern(self):
|
||||
pattern = []
|
||||
for _ in range(10):
|
||||
bit = int.from_bytes(os.urandom(1), 'big') % 2
|
||||
pattern.append(bit)
|
||||
return pattern
|
||||
"""Generate a 10-bit pattern for case-flipping."""
|
||||
return [int.from_bytes(os.urandom(1), 'big') % 2 for _ in range(10)]
|
||||
|
||||
def _generate_long_key(self, length=4096):
|
||||
"""Generate a long hex key (4096 chars by default)."""
|
||||
chars = "0123456789abcdef"
|
||||
result = []
|
||||
for _ in range(length):
|
||||
idx = int.from_bytes(os.urandom(1), 'big') % 16
|
||||
result.append(chars[idx])
|
||||
return ''.join(result)
|
||||
return ''.join(chars[int.from_bytes(os.urandom(1), 'big') % 16] for _ in range(length))
|
||||
|
||||
def _generate_short_key(self, length=512):
|
||||
"""Generate a short hex key (512 chars by default)."""
|
||||
chars = "0123456789abcdef"
|
||||
result = []
|
||||
for _ in range(length):
|
||||
idx = int.from_bytes(os.urandom(1), 'big') % 16
|
||||
result.append(chars[idx])
|
||||
return ''.join(result)
|
||||
return ''.join(chars[int.from_bytes(os.urandom(1), 'big') % 16] for _ in range(length))
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Key obfuscation and persistence
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def _obfuscate_keys(self, keys_data):
|
||||
"""Obfuscate key data using base64 + reversal + Caesar shift."""
|
||||
json_str = json.dumps(keys_data)
|
||||
b64 = base64.b64encode(json_str.encode()).decode()
|
||||
reversed_b64 = b64[::-1]
|
||||
shifted = ''.join([chr((ord(c) + 1) % 128) for c in reversed_b64])
|
||||
final = base64.b64encode(shifted.encode()).decode()
|
||||
return final
|
||||
shifted = ''.join(chr((ord(c) + 1) % 128) for c in reversed_b64)
|
||||
return base64.b64encode(shifted.encode()).decode()
|
||||
|
||||
def _deobfuscate_keys(self, obfuscated_data):
|
||||
"""Reverse the obfuscation to recover original key data."""
|
||||
try:
|
||||
shifted = base64.b64decode(obfuscated_data.encode()).decode()
|
||||
reversed_b64 = ''.join([chr((ord(c) - 1) % 128) for c in shifted])
|
||||
reversed_b64 = ''.join(chr((ord(c) - 1) % 128) for c in shifted)
|
||||
b64 = reversed_b64[::-1]
|
||||
json_str = base64.b64decode(b64.encode()).decode()
|
||||
return json.loads(json_str)
|
||||
except Exception:
|
||||
raise Exception("密钥文件损坏")
|
||||
except Exception as exc:
|
||||
raise Exception("Key file corrupted") from exc
|
||||
|
||||
def _xor_encrypt_data(self, data, password):
|
||||
result = []
|
||||
"""XOR encrypt data with a password, returning hex string."""
|
||||
key_len = len(password)
|
||||
for i, char in enumerate(data):
|
||||
xor_result = ord(char) ^ ord(password[i % key_len])
|
||||
result.append(f"{xor_result:02x}")
|
||||
return ''.join(result)
|
||||
return ''.join(
|
||||
f"{ord(char) ^ ord(password[i % key_len]):02x}"
|
||||
for i, char in enumerate(data)
|
||||
)
|
||||
|
||||
def _xor_decrypt_data(self, hex_data, password):
|
||||
"""XOR decrypt hex data with a password."""
|
||||
try:
|
||||
result = []
|
||||
key_len = len(password)
|
||||
result = []
|
||||
for i in range(0, len(hex_data), 2):
|
||||
if i + 1 < len(hex_data):
|
||||
hex_byte = hex_data[i:i+2]
|
||||
@@ -143,17 +163,28 @@ class EncryptionSystem:
|
||||
return None
|
||||
|
||||
def generate_keys(self, save_path=None, key_password=None):
|
||||
"""
|
||||
Generate a fresh set of encryption keys and save to a key file.
|
||||
|
||||
Args:
|
||||
save_path: Path to save the key file (default: encryption.key).
|
||||
key_password: Password to protect the key file.
|
||||
|
||||
Returns:
|
||||
The password used, or None if generation failed.
|
||||
"""
|
||||
if key_password is None:
|
||||
key_password = input("请设置密钥文件密码: ")
|
||||
confirm = input("请再次输入密码确认: ")
|
||||
key_password = input("Set key file password: ")
|
||||
confirm = input("Confirm password: ")
|
||||
if key_password != confirm:
|
||||
print("❌ 密码不匹配")
|
||||
print("❌ Passwords do not match")
|
||||
return None
|
||||
|
||||
|
||||
print("=" * 60)
|
||||
print("正在生成随机密钥...")
|
||||
print("Generating random keys...")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
# Generate all key components
|
||||
self.upper_mapping = self._generate_random_alphabet(lowercase=False)
|
||||
self.lower_mapping = self._generate_random_alphabet(lowercase=True)
|
||||
self.digit_mapping = self._generate_random_digit_mapping()
|
||||
@@ -161,10 +192,12 @@ class EncryptionSystem:
|
||||
self.flip_pattern = self._generate_random_flip_pattern()
|
||||
self.long_key = self._generate_long_key(4096)
|
||||
self.short_key = self._generate_short_key(512)
|
||||
|
||||
|
||||
# Build reverse mappings
|
||||
self.digit_reverse = {v: k for k, v in self.digit_mapping.items()}
|
||||
self.equal_reverse = {v: k for k, v in self.equal_mapping.items()}
|
||||
|
||||
|
||||
# Bundle keys into a dictionary
|
||||
keys_data = {
|
||||
'upper_mapping': self.upper_mapping,
|
||||
'lower_mapping': self.lower_mapping,
|
||||
@@ -175,30 +208,36 @@ class EncryptionSystem:
|
||||
'short_key': self.short_key,
|
||||
'generated_at': datetime.now().isoformat()
|
||||
}
|
||||
|
||||
|
||||
# Obfuscate and encrypt the key data
|
||||
obfuscated = self._obfuscate_keys(keys_data)
|
||||
encrypted = self._xor_encrypt_data(obfuscated, key_password)
|
||||
|
||||
|
||||
if save_path is None:
|
||||
save_path = "encryption.key"
|
||||
|
||||
|
||||
with open(save_path, 'w') as f:
|
||||
f.write(encrypted)
|
||||
|
||||
|
||||
self.key_file = save_path
|
||||
self.key_password = key_password
|
||||
self.keys_loaded = True
|
||||
self._build_maps()
|
||||
|
||||
print(f"✅ 密钥已生成并保存到: {save_path}")
|
||||
|
||||
print(f"✅ Keys generated and saved to: {save_path}")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
return save_path
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Internal helpers
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def _build_maps(self):
|
||||
"""Build encryption and decryption maps from shuffled alphabets."""
|
||||
self.upper_original = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
||||
self.lower_original = "abcdefghijklmnopqrstuvwxyz"
|
||||
|
||||
|
||||
self.encrypt_map = {}
|
||||
self.decrypt_map = {}
|
||||
for i in range(26):
|
||||
@@ -208,18 +247,19 @@ class EncryptionSystem:
|
||||
self.decrypt_map[self.lower_mapping[i]] = self.lower_original[i]
|
||||
|
||||
def _load_keys(self, key_file, key_password):
|
||||
"""Load and decrypt keys from a key file."""
|
||||
try:
|
||||
with open(key_file, 'r') as f:
|
||||
encrypted_data = f.read()
|
||||
|
||||
|
||||
decrypted = self._xor_decrypt_data(encrypted_data, key_password)
|
||||
if decrypted is None:
|
||||
print("❌ 密码错误")
|
||||
print("❌ Incorrect password")
|
||||
self.keys_loaded = False
|
||||
return
|
||||
|
||||
|
||||
keys_data = self._deobfuscate_keys(decrypted)
|
||||
|
||||
|
||||
self.upper_mapping = keys_data['upper_mapping']
|
||||
self.lower_mapping = keys_data['lower_mapping']
|
||||
self.digit_mapping = keys_data['digit_mapping']
|
||||
@@ -227,25 +267,26 @@ class EncryptionSystem:
|
||||
self.flip_pattern = keys_data['flip_pattern']
|
||||
self.long_key = keys_data['long_key']
|
||||
self.short_key = keys_data['short_key']
|
||||
|
||||
|
||||
self.digit_reverse = {v: k for k, v in self.digit_mapping.items()}
|
||||
self.equal_reverse = {v: k for k, v in self.equal_mapping.items()}
|
||||
|
||||
|
||||
self.key_password = key_password
|
||||
self._build_maps()
|
||||
|
||||
|
||||
self.keys_loaded = True
|
||||
print(f"✅ 密钥已从 {key_file} 加载")
|
||||
|
||||
print(f"✅ Keys loaded from {key_file}")
|
||||
|
||||
except Exception:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
self.keys_loaded = False
|
||||
|
||||
def _ensure_printable(self, text):
|
||||
"""Force all characters into printable ASCII range (32-126)."""
|
||||
result = []
|
||||
for c in text:
|
||||
val = ord(c)
|
||||
if val == 124:
|
||||
if val == 124: # '|' is used as a separator, keep it
|
||||
result.append('|')
|
||||
elif val < 32 or val > 126:
|
||||
val = val % 95 + 32
|
||||
@@ -254,10 +295,25 @@ class EncryptionSystem:
|
||||
result.append(c)
|
||||
return ''.join(result)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Key derivation
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def _mix_keys(self, user_password, target_length):
|
||||
"""
|
||||
Mix the user password with the short key using bit operations and swaps.
|
||||
|
||||
Args:
|
||||
user_password: User-provided password.
|
||||
target_length: Desired output length.
|
||||
|
||||
Returns:
|
||||
A mixed printable string.
|
||||
"""
|
||||
if not self.keys_loaded:
|
||||
raise Exception("密钥未加载")
|
||||
|
||||
raise Exception("Keys not loaded")
|
||||
|
||||
# Interleave user password and short key
|
||||
mixed = []
|
||||
max_len = max(len(user_password), len(self.short_key))
|
||||
for i in range(max_len):
|
||||
@@ -265,7 +321,8 @@ class EncryptionSystem:
|
||||
mixed.append(ord(user_password[i]))
|
||||
if i < len(self.short_key):
|
||||
mixed.append(ord(self.short_key[i]))
|
||||
|
||||
|
||||
# Apply bit transformations
|
||||
for i in range(len(mixed)):
|
||||
if i % 3 == 0:
|
||||
mixed[i] = (mixed[i] << 1) & 0xFF
|
||||
@@ -273,21 +330,24 @@ class EncryptionSystem:
|
||||
mixed[i] = (mixed[i] >> 1) & 0xFF
|
||||
else:
|
||||
mixed[i] = mixed[i] ^ 0x5A
|
||||
|
||||
|
||||
# Swap pairs
|
||||
for i in range(0, len(mixed) - 3, 4):
|
||||
mixed[i], mixed[i+3] = mixed[i+3], mixed[i]
|
||||
mixed[i+1], mixed[i+2] = mixed[i+2], mixed[i+1]
|
||||
|
||||
|
||||
mixed.reverse()
|
||||
|
||||
|
||||
# Convert to printable characters
|
||||
result = []
|
||||
for x in mixed:
|
||||
if x < 32 or x > 126:
|
||||
x = x % 95 + 32
|
||||
result.append(chr(x))
|
||||
|
||||
|
||||
result_str = ''.join(result)
|
||||
|
||||
|
||||
# Extend if needed by repeating with variations
|
||||
if len(result_str) < target_length:
|
||||
base_key = result_str
|
||||
final_key = base_key
|
||||
@@ -297,22 +357,29 @@ class EncryptionSystem:
|
||||
next_chunk = base_key[::-1]
|
||||
elif iteration % 3 == 1:
|
||||
next_chunk = base_key[::-1]
|
||||
next_chunk = ''.join([chr((ord(c) + iteration) % 95 + 32) for c in next_chunk])
|
||||
next_chunk = ''.join(chr((ord(c) + iteration) % 95 + 32) for c in next_chunk)
|
||||
else:
|
||||
next_chunk = base_key[::-1]
|
||||
next_chunk = ''.join([chr((ord(c) ^ iteration) % 95 + 32) for c in next_chunk])
|
||||
next_chunk = ''.join(chr((ord(c) ^ iteration) % 95 + 32) for c in next_chunk)
|
||||
final_key += next_chunk
|
||||
iteration += 1
|
||||
result_str = final_key[:target_length]
|
||||
|
||||
|
||||
return self._ensure_printable(result_str)
|
||||
|
||||
def _derive_final_key(self, user_password, text_length):
|
||||
"""
|
||||
Derive a final encryption key from the user password and short key.
|
||||
|
||||
The process includes mixing, hex conversion, reversal, and extension.
|
||||
"""
|
||||
mixed_key = self._mix_keys(user_password, text_length * 2)
|
||||
|
||||
hex_key = ''.join([f"{ord(c):02x}" for c in mixed_key])
|
||||
|
||||
# Convert to hex and reverse
|
||||
hex_key = ''.join(f"{ord(c):02x}" for c in mixed_key)
|
||||
reversed_hex = hex_key[::-1]
|
||||
|
||||
|
||||
# Convert back to printable characters
|
||||
final_key = []
|
||||
for i in range(0, len(reversed_hex), 2):
|
||||
if i + 1 < len(reversed_hex):
|
||||
@@ -324,9 +391,10 @@ class EncryptionSystem:
|
||||
final_key.append(chr(val))
|
||||
except ValueError:
|
||||
final_key.append('x')
|
||||
|
||||
|
||||
final_key_str = ''.join(final_key)
|
||||
|
||||
|
||||
# Extend if needed
|
||||
if len(final_key_str) < text_length:
|
||||
base_key = final_key_str
|
||||
iteration = 0
|
||||
@@ -335,22 +403,25 @@ class EncryptionSystem:
|
||||
next_chunk = base_key[::-1]
|
||||
elif iteration % 3 == 1:
|
||||
next_chunk = base_key[::-1]
|
||||
next_chunk = ''.join([chr((ord(c) + iteration) % 95 + 32) for c in next_chunk])
|
||||
next_chunk = ''.join(chr((ord(c) + iteration) % 95 + 32) for c in next_chunk)
|
||||
else:
|
||||
next_chunk = base_key[::-1]
|
||||
next_chunk = ''.join([chr((ord(c) ^ iteration) % 95 + 32) for c in next_chunk])
|
||||
next_chunk = ''.join(chr((ord(c) ^ iteration) % 95 + 32) for c in next_chunk)
|
||||
final_key_str += next_chunk
|
||||
iteration += 1
|
||||
|
||||
|
||||
final_key_str = final_key_str[:text_length]
|
||||
final_key_str = self._ensure_printable(final_key_str)
|
||||
|
||||
return final_key_str
|
||||
return self._ensure_printable(final_key_str)
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Core encryption operations
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def _apply_flip(self, text):
|
||||
"""Apply case-flipping based on the flip pattern."""
|
||||
if self.flip_pattern is None:
|
||||
return text
|
||||
|
||||
|
||||
result = []
|
||||
for i, ch in enumerate(text):
|
||||
idx = i % len(self.flip_pattern)
|
||||
@@ -361,15 +432,16 @@ class EncryptionSystem:
|
||||
return ''.join(result)
|
||||
|
||||
def _xor_encrypt_with_final_key(self, text, user_password):
|
||||
"""XOR encrypt with the derived final key, returning hex."""
|
||||
final_key = self._derive_final_key(user_password, len(text))
|
||||
key_len = len(final_key)
|
||||
result = []
|
||||
for i, char in enumerate(text):
|
||||
xor_result = ord(char) ^ ord(final_key[i % key_len])
|
||||
result.append(f"{xor_result:02x}")
|
||||
return ''.join(result)
|
||||
return ''.join(
|
||||
f"{ord(char) ^ ord(final_key[i % key_len]):02x}"
|
||||
for i, char in enumerate(text)
|
||||
)
|
||||
|
||||
def _xor_decrypt_with_final_key(self, hex_text, user_password):
|
||||
"""XOR decrypt with the derived final key."""
|
||||
try:
|
||||
text_length = len(hex_text) // 2
|
||||
final_key = self._derive_final_key(user_password, text_length)
|
||||
@@ -390,47 +462,35 @@ class EncryptionSystem:
|
||||
return None
|
||||
|
||||
def _substitute_letters(self, text, mapping):
|
||||
result = []
|
||||
for char in text:
|
||||
if char in mapping:
|
||||
result.append(mapping[char])
|
||||
else:
|
||||
result.append(char)
|
||||
return ''.join(result)
|
||||
"""Apply a substitution map to letters."""
|
||||
return ''.join(mapping.get(char, char) for char in text)
|
||||
|
||||
def _encode_digit(self, num_str):
|
||||
result = []
|
||||
for char in num_str:
|
||||
if char in self.digit_mapping:
|
||||
result.append(self.digit_mapping[char])
|
||||
else:
|
||||
result.append(char)
|
||||
return ''.join(result)
|
||||
"""Encode a digit string using the digit mapping."""
|
||||
return ''.join(self.digit_mapping.get(char, char) for char in num_str)
|
||||
|
||||
def _decode_digit(self, encoded_str):
|
||||
result = []
|
||||
for char in encoded_str:
|
||||
if char in self.digit_reverse:
|
||||
result.append(self.digit_reverse[char])
|
||||
else:
|
||||
result.append(char)
|
||||
return ''.join(result)
|
||||
"""Decode a digit string using the reverse digit mapping."""
|
||||
return ''.join(self.digit_reverse.get(char, char) for char in encoded_str)
|
||||
|
||||
def _get_dynamic_key(self, text_length):
|
||||
"""Generate a dynamic key from the long key, repeated as needed."""
|
||||
key = self.long_key
|
||||
while len(key) < text_length:
|
||||
key += self.long_key
|
||||
if len(key) < text_length:
|
||||
repeats = (text_length // len(key)) + 1
|
||||
key = (key * repeats)[:text_length]
|
||||
return key[:text_length]
|
||||
|
||||
def _xor_encrypt_with_key(self, text, key):
|
||||
"""XOR encrypt with a fixed key, returning hex."""
|
||||
key_len = len(key)
|
||||
result = []
|
||||
for i, char in enumerate(text):
|
||||
xor_result = ord(char) ^ ord(key[i % key_len])
|
||||
result.append(f"{xor_result:02x}")
|
||||
return ''.join(result)
|
||||
return ''.join(
|
||||
f"{ord(char) ^ ord(key[i % key_len]):02x}"
|
||||
for i, char in enumerate(text)
|
||||
)
|
||||
|
||||
def _xor_decrypt_with_key(self, hex_text, key):
|
||||
"""XOR decrypt with a fixed key."""
|
||||
try:
|
||||
key_len = len(key)
|
||||
result = []
|
||||
@@ -448,45 +508,74 @@ class EncryptionSystem:
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# Public API
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def encrypt(self, plaintext, user_password):
|
||||
"""
|
||||
Encrypt plaintext with the given password.
|
||||
|
||||
Workflow:
|
||||
1. Base64 encode the plaintext.
|
||||
2. Apply letter substitution and case-flipping.
|
||||
3. Reverse the string and append padding count.
|
||||
4. XOR encrypt with dynamic key (long key).
|
||||
5. XOR encrypt with derived final key (user password + short key).
|
||||
|
||||
Returns:
|
||||
Encrypted ciphertext as a hex string.
|
||||
"""
|
||||
if not self.keys_loaded:
|
||||
return "❌ 错误:密钥未加载"
|
||||
|
||||
return "❌ Error: Keys not loaded"
|
||||
|
||||
# Step 1: Base64 encode and handle special chars
|
||||
b64 = base64.b64encode(plaintext.encode('utf-8')).decode('utf-8')
|
||||
equal_count = b64.count('=')
|
||||
|
||||
|
||||
processed = b64
|
||||
for old, new in self.special_encrypt.items():
|
||||
processed = processed.replace(old, new)
|
||||
processed = processed.rstrip('=')
|
||||
|
||||
|
||||
# Step 2: Letter substitution + flip + reverse
|
||||
sub = self._substitute_letters(processed, self.encrypt_map)
|
||||
flipped = self._apply_flip(sub)
|
||||
reversed_text = flipped[::-1]
|
||||
|
||||
|
||||
# Step 3: Append padding count
|
||||
equal_char = self.equal_mapping[str(equal_count)]
|
||||
with_equal = f"{reversed_text}|{equal_char}"
|
||||
|
||||
|
||||
# Step 4: Dynamic key (long key) encryption
|
||||
key_length_str = str(len(with_equal))
|
||||
key_length_encoded = self._encode_digit(key_length_str)
|
||||
|
||||
dynamic_key = self._get_dynamic_key(len(with_equal))
|
||||
encrypted_by_dynamic = self._xor_encrypt_with_key(with_equal, dynamic_key)
|
||||
|
||||
|
||||
# Step 5: Final encryption with derived key
|
||||
combined = f"{encrypted_by_dynamic}|{key_length_encoded}"
|
||||
final_encrypted = self._xor_encrypt_with_final_key(combined, user_password)
|
||||
|
||||
|
||||
return final_encrypted
|
||||
|
||||
def decrypt(self, ciphertext, user_password):
|
||||
"""
|
||||
Decrypt ciphertext with the given password.
|
||||
|
||||
Returns:
|
||||
The original plaintext, or an error message on failure.
|
||||
"""
|
||||
if not self.keys_loaded:
|
||||
return "❌ 解密失败"
|
||||
|
||||
return "❌ Decryption failed"
|
||||
|
||||
try:
|
||||
# Step 1: Decrypt with derived final key
|
||||
combined = self._xor_decrypt_with_final_key(ciphertext, user_password)
|
||||
if combined is None:
|
||||
return "❌ 解密失败"
|
||||
|
||||
return "❌ Decryption failed"
|
||||
|
||||
# Step 2: Extract hex data and length indicator
|
||||
if '|' in combined:
|
||||
parts = combined.split('|')
|
||||
if len(parts) >= 2:
|
||||
@@ -498,211 +587,234 @@ class EncryptionSystem:
|
||||
else:
|
||||
hex_data = combined
|
||||
key_length_encoded = 'g'
|
||||
|
||||
|
||||
# Step 3: Decode length and get dynamic key
|
||||
key_length_str = self._decode_digit(key_length_encoded)
|
||||
try:
|
||||
key_length = int(key_length_str)
|
||||
except ValueError:
|
||||
key_length = 16
|
||||
|
||||
|
||||
dynamic_key = self._get_dynamic_key(key_length)
|
||||
|
||||
# Step 4: Decrypt with dynamic key
|
||||
xor_decrypted = self._xor_decrypt_with_key(hex_data, dynamic_key)
|
||||
if xor_decrypted is None:
|
||||
return "❌ 解密失败"
|
||||
|
||||
return "❌ Decryption failed"
|
||||
|
||||
# Step 5: Extract main data and padding count
|
||||
if '|' in xor_decrypted:
|
||||
main_part, equal_char = xor_decrypted.split('|')
|
||||
equal_count = int(self.equal_reverse.get(equal_char, '0'))
|
||||
else:
|
||||
main_part = xor_decrypted
|
||||
equal_count = 0
|
||||
|
||||
|
||||
# Step 6: Reverse, flip, substitute
|
||||
reversed_text = main_part[::-1]
|
||||
flipped = self._apply_flip(reversed_text)
|
||||
sub = self._substitute_letters(flipped, self.decrypt_map)
|
||||
|
||||
|
||||
# Step 7: Restore Base64 special chars and padding
|
||||
for old, new in self.special_decrypt.items():
|
||||
sub = sub.replace(old, new)
|
||||
|
||||
|
||||
b64_with_equal = sub + '=' * equal_count
|
||||
|
||||
|
||||
if len(b64_with_equal) % 4 != 0:
|
||||
return "❌ 解密失败"
|
||||
|
||||
return "❌ Decryption failed"
|
||||
|
||||
# Step 8: Base64 decode
|
||||
decoded = base64.b64decode(b64_with_equal.encode('utf-8')).decode('utf-8')
|
||||
return decoded
|
||||
|
||||
|
||||
except Exception:
|
||||
return "❌ 解密失败"
|
||||
return "❌ Decryption failed"
|
||||
|
||||
def print_keys(self):
|
||||
"""Display information about the currently loaded keys."""
|
||||
if not self.keys_loaded:
|
||||
print("❌ 未加载密钥")
|
||||
print("❌ Keys not loaded")
|
||||
return
|
||||
|
||||
|
||||
print("=" * 60)
|
||||
print("密钥信息")
|
||||
print("Key Information")
|
||||
print("=" * 60)
|
||||
print(f"密钥文件: {self.key_file}")
|
||||
print(f"大写映射表: {self.upper_mapping}")
|
||||
print(f"小写映射表: {self.lower_mapping}")
|
||||
print(f"翻转模式: {self.flip_pattern}")
|
||||
print(f"长密钥长度: {len(self.long_key)} 位")
|
||||
print(f"短密钥长度: {len(self.short_key)} 位")
|
||||
print(f"Key file: {self.key_file}")
|
||||
print(f"Uppercase mapping: {self.upper_mapping}")
|
||||
print(f"Lowercase mapping: {self.lower_mapping}")
|
||||
print(f"Flip pattern: {self.flip_pattern}")
|
||||
print(f"Long key length: {len(self.long_key)} bits")
|
||||
print(f"Short key length: {len(self.short_key)} bits")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
# ----------------------------------------------------------------------
|
||||
# CLI Entry Point
|
||||
# ----------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
"""Command-line interface for the encryption system."""
|
||||
print("=" * 60)
|
||||
print("欢迎使用加密系统")
|
||||
print("Welcome to the Encryption System")
|
||||
print("=" * 60)
|
||||
|
||||
|
||||
crypto = None
|
||||
key_password = None
|
||||
|
||||
|
||||
# Try loading default key file
|
||||
if os.path.exists("encryption.key"):
|
||||
key_password = input("请输入默认密钥文件 (encryption.key) 的密码: ")
|
||||
key_password = input("Enter password for default key file (encryption.key): ")
|
||||
crypto = EncryptionSystem("encryption.key", key_password)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
crypto = None
|
||||
else:
|
||||
print("\n未找到默认密钥文件,请先生成")
|
||||
choice = input("是否生成默认密钥文件?(y/n): ").strip().lower()
|
||||
print("\nDefault key file not found. Please generate one first.")
|
||||
choice = input("Generate default key file? (y/n): ").strip().lower()
|
||||
if choice == 'y':
|
||||
crypto = EncryptionSystem()
|
||||
key_password = crypto.generate_keys("encryption.key")
|
||||
if key_password is None:
|
||||
print("❌ 生成失败")
|
||||
result = crypto.generate_keys("encryption.key")
|
||||
if result is None:
|
||||
print("❌ Generation failed")
|
||||
crypto = None
|
||||
else:
|
||||
print("⚠️ 请使用模式4或5指定密钥文件,或模式6生成新密钥")
|
||||
|
||||
print("⚠️ Use options 4, 5, or 6 to manage key files manually.")
|
||||
|
||||
# Main interaction loop
|
||||
while True:
|
||||
print("\n请选择操作:")
|
||||
print("1. 使用默认密钥加密")
|
||||
print("2. 使用默认密钥解密")
|
||||
print("3. 生成新密钥(覆盖默认)")
|
||||
print("4. 使用指定密钥文件加密")
|
||||
print("5. 使用指定密钥文件解密")
|
||||
print("6. 生成密钥并保存到当前文件夹")
|
||||
print("7. 查看当前密钥信息")
|
||||
print("8. 退出")
|
||||
|
||||
choice = input("\n请选择操作 (1-8): ").strip()
|
||||
|
||||
print("\nSelect an option:")
|
||||
print("1. Encrypt with default key")
|
||||
print("2. Decrypt with default key")
|
||||
print("3. Generate new key (overwrite default)")
|
||||
print("4. Encrypt with custom key file")
|
||||
print("5. Decrypt with custom key file")
|
||||
print("6. Generate key and save to current folder")
|
||||
print("7. View current key info")
|
||||
print("8. Exit")
|
||||
|
||||
choice = input("\nEnter choice (1-8): ").strip()
|
||||
|
||||
if choice == '1':
|
||||
# Encrypt with default key
|
||||
if crypto is None or not crypto.keys_loaded:
|
||||
if os.path.exists("encryption.key"):
|
||||
if key_password is None:
|
||||
key_password = input("请输入默认密钥文件密码: ")
|
||||
key_password = input("Enter default key file password: ")
|
||||
crypto = EncryptionSystem("encryption.key", key_password)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
continue
|
||||
else:
|
||||
print("❌ 密钥文件不存在")
|
||||
print("❌ Key file not found")
|
||||
continue
|
||||
|
||||
password = input("请输入加密密码: ")
|
||||
text = input("请输入要加密的文本: ")
|
||||
|
||||
password = input("Enter encryption password: ")
|
||||
text = input("Enter text to encrypt: ")
|
||||
if text and password:
|
||||
encrypted = crypto.encrypt(text, password)
|
||||
print(f"\n✅ 加密结果: {encrypted}")
|
||||
|
||||
print(f"\n✅ Encrypted result: {encrypted}")
|
||||
|
||||
elif choice == '2':
|
||||
# Decrypt with default key
|
||||
if crypto is None or not crypto.keys_loaded:
|
||||
if os.path.exists("encryption.key"):
|
||||
if key_password is None:
|
||||
key_password = input("请输入默认密钥文件密码: ")
|
||||
key_password = input("Enter default key file password: ")
|
||||
crypto = EncryptionSystem("encryption.key", key_password)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
continue
|
||||
else:
|
||||
print("❌ 密钥文件不存在")
|
||||
print("❌ Key file not found")
|
||||
continue
|
||||
|
||||
password = input("请输入加密密码: ")
|
||||
text = input("请输入要解密的密文: ")
|
||||
|
||||
password = input("Enter encryption password: ")
|
||||
text = input("Enter ciphertext to decrypt: ")
|
||||
if text and password:
|
||||
decrypted = crypto.decrypt(text, password)
|
||||
print(f"\n✅ 解密结果: {decrypted}")
|
||||
|
||||
print(f"\n✅ Decrypted result: {decrypted}")
|
||||
|
||||
elif choice == '3':
|
||||
# Generate and overwrite default key
|
||||
crypto = EncryptionSystem()
|
||||
key_password = crypto.generate_keys("encryption.key")
|
||||
if key_password is None:
|
||||
print("❌ 生成失败")
|
||||
result = crypto.generate_keys("encryption.key")
|
||||
if result is None:
|
||||
print("❌ Generation failed")
|
||||
else:
|
||||
print("✅ 默认密钥已更新")
|
||||
|
||||
print("✅ Default key updated")
|
||||
|
||||
elif choice == '4':
|
||||
key_file = input("请输入密钥文件路径: ")
|
||||
# Encrypt with custom key file
|
||||
key_file = input("Enter key file path: ")
|
||||
if not os.path.exists(key_file):
|
||||
print(f"❌ 文件不存在")
|
||||
print(f"❌ File not found: {key_file}")
|
||||
continue
|
||||
|
||||
kp = input(f"请输入密钥文件密码: ")
|
||||
|
||||
kp = input("Enter key file password: ")
|
||||
crypto = EncryptionSystem(key_file, kp)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
continue
|
||||
|
||||
password = input("请输入加密密码: ")
|
||||
text = input("请输入要加密的文本: ")
|
||||
|
||||
password = input("Enter encryption password: ")
|
||||
text = input("Enter text to encrypt: ")
|
||||
if text and password:
|
||||
encrypted = crypto.encrypt(text, password)
|
||||
print(f"\n✅ 加密结果: {encrypted}")
|
||||
|
||||
print(f"\n✅ Encrypted result: {encrypted}")
|
||||
|
||||
elif choice == '5':
|
||||
key_file = input("请输入密钥文件路径: ")
|
||||
# Decrypt with custom key file
|
||||
key_file = input("Enter key file path: ")
|
||||
if not os.path.exists(key_file):
|
||||
print(f"❌ 文件不存在")
|
||||
print(f"❌ File not found: {key_file}")
|
||||
continue
|
||||
|
||||
kp = input(f"请输入密钥文件密码: ")
|
||||
|
||||
kp = input("Enter key file password: ")
|
||||
crypto = EncryptionSystem(key_file, kp)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
continue
|
||||
|
||||
password = input("请输入加密密码: ")
|
||||
text = input("请输入要解密的密文: ")
|
||||
|
||||
password = input("Enter encryption password: ")
|
||||
text = input("Enter ciphertext to decrypt: ")
|
||||
if text and password:
|
||||
decrypted = crypto.decrypt(text, password)
|
||||
print(f"\n✅ 解密结果: {decrypted}")
|
||||
|
||||
print(f"\n✅ Decrypted result: {decrypted}")
|
||||
|
||||
elif choice == '6':
|
||||
# Generate new key with timestamp
|
||||
timestamp = int(time.time())
|
||||
filename = f"key_{timestamp}.key"
|
||||
crypto = EncryptionSystem()
|
||||
kp = crypto.generate_keys(filename)
|
||||
if kp is None:
|
||||
print("❌ 生成失败")
|
||||
result = crypto.generate_keys(filename)
|
||||
if result is None:
|
||||
print("❌ Generation failed")
|
||||
else:
|
||||
print(f"✅ 密钥已保存到: {filename}")
|
||||
|
||||
print(f"✅ Key saved to: {filename}")
|
||||
|
||||
elif choice == '7':
|
||||
# Show key info
|
||||
if crypto is None or not crypto.keys_loaded:
|
||||
if os.path.exists("encryption.key"):
|
||||
if key_password is None:
|
||||
key_password = input("请输入默认密钥文件密码: ")
|
||||
key_password = input("Enter default key file password: ")
|
||||
crypto = EncryptionSystem("encryption.key", key_password)
|
||||
if not crypto.keys_loaded:
|
||||
print("❌ 加载失败")
|
||||
print("❌ Loading failed")
|
||||
continue
|
||||
else:
|
||||
print("❌ 未加载密钥")
|
||||
print("❌ Keys not loaded")
|
||||
continue
|
||||
crypto.print_keys()
|
||||
|
||||
|
||||
elif choice == '8':
|
||||
print("感谢使用,再见!")
|
||||
print("Goodbye!")
|
||||
break
|
||||
|
||||
|
||||
else:
|
||||
print("❌ 无效选择")
|
||||
print("❌ Invalid choice")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
main()
|
||||
Reference in New Issue
Block a user