7 Commits
Author SHA1 Message Date
dvs bad851f741 更新 chaoscrypt.py 2026-08-28 00:55:02 +00:00
dvs b311483963 更新 README.md 2026-08-28 00:53:38 +00:00
dvs ef90efd199 revert 587ef2140b
revert 删除 .gitignore
2026-08-28 00:51:40 +00:00
dvs 587ef2140b 删除 .gitignore 2026-08-27 10:46:24 +00:00
dvs c352eef2ba 更新 README.md 2026-08-27 10:44:26 +00:00
dvs 124b511d5f v2.0: 新增 ChaosCryptChat 端到端加密聊天系统 + 更新文档 2026-08-26 16:02:23 +08:00
dvs 79f4ad8b78 docs: README 新增 GitHub 克隆地址 2026-08-25 18:09:21 +08:00
6 changed files with 5337 additions and 937 deletions
+20
View File
@@ -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
+418
View File
@@ -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☁️ 服务器已停止")
+342 -706
View File
File diff suppressed because it is too large Load Diff
+67
View File
@@ -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
View File
@@ -1,3 +1,7 @@
"""
A multi-layer encryption system with key management, substitution ciphers,
XOR operations, and dynamic key derivation.
"""
import base64 import base64
import json import json
@@ -5,12 +9,23 @@ import os
import time import time
from datetime import datetime from datetime import datetime
class EncryptionSystem: class EncryptionSystem:
"""Main encryption engine with key-based transformation layers."""
def __init__(self, key_file=None, key_password=None): 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_file = key_file
self.key_password = key_password self.key_password = key_password
self.keys_loaded = False self.keys_loaded = False
# Base64 special character mappings for safe transport
self.special_encrypt = { self.special_encrypt = {
'+': '.', '+': '.',
'/': "'", '/': "'",
@@ -20,41 +35,49 @@ class EncryptionSystem:
'.': '+', '.': '+',
"'": '/' "'": '/'
} }
self.flip_pattern = None self.flip_pattern = None
# Try loading the key file if provided or find default
if key_file: if key_file:
if os.path.exists(key_file): if os.path.exists(key_file):
if key_password is None: 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._load_keys(key_file, key_password)
self.keys_loaded = True self.keys_loaded = True
else: else:
print(f"⚠️ 密钥文件 {key_file} 不存在") print(f"⚠️ Key file {key_file} not found")
self.keys_loaded = False self.keys_loaded = False
else: else:
default_key = "encryption.key" default_key = "encryption.key"
if os.path.exists(default_key): if os.path.exists(default_key):
self.key_file = default_key self.key_file = default_key
if key_password is None: 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._load_keys(default_key, key_password)
self.keys_loaded = True self.keys_loaded = True
else: else:
print("=" * 60) print("=" * 60)
print("首次启动,请先生成密钥文件") print("First launch detected. Please generate a key file first.")
print("=" * 60) print("=" * 60)
self.keys_loaded = False self.keys_loaded = False
# ----------------------------------------------------------------------
# Key generation utilities
# ----------------------------------------------------------------------
def _generate_random_alphabet(self, lowercase=False): def _generate_random_alphabet(self, lowercase=False):
"""Generate a shuffled alphabet string."""
chars = list("abcdefghijklmnopqrstuvwxyz" if lowercase else "ABCDEFGHIJKLMNOPQRSTUVWXYZ") chars = list("abcdefghijklmnopqrstuvwxyz" if lowercase else "ABCDEFGHIJKLMNOPQRSTUVWXYZ")
n = len(chars) n = len(chars)
# Fisher-Yates shuffle using secure random bytes
for i in range(n - 1, 0, -1): for i in range(n - 1, 0, -1):
j = int.from_bytes(os.urandom(1), 'big') % (i + 1) j = int.from_bytes(os.urandom(1), 'big') % (i + 1)
chars[i], chars[j] = chars[j], chars[i] chars[i], chars[j] = chars[j], chars[i]
return ''.join(chars) return ''.join(chars)
def _generate_random_digit_mapping(self): def _generate_random_digit_mapping(self):
"""Create a random mapping for digits 0-9 to alphabet characters."""
digits = list("0123456789") digits = list("0123456789")
mapping_chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") mapping_chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")
n = len(mapping_chars) n = len(mapping_chars)
@@ -67,72 +90,69 @@ class EncryptionSystem:
return mapping return mapping
def _generate_random_equal_mapping(self): def _generate_random_equal_mapping(self):
"""Generate mapping for Base64 padding count (0-3)."""
chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz") chars = list("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz")
n = len(chars) n = len(chars)
for i in range(n - 1, 0, -1): for i in range(n - 1, 0, -1):
j = int.from_bytes(os.urandom(1), 'big') % (i + 1) j = int.from_bytes(os.urandom(1), 'big') % (i + 1)
chars[i], chars[j] = chars[j], chars[i] chars[i], chars[j] = chars[j], chars[i]
mapping = { return {
'0': chars[0], '0': chars[0],
'1': chars[1], '1': chars[1],
'2': chars[2], '2': chars[2],
'3': chars[3] '3': chars[3]
} }
return mapping
def _generate_random_flip_pattern(self): def _generate_random_flip_pattern(self):
pattern = [] """Generate a 10-bit pattern for case-flipping."""
for _ in range(10): return [int.from_bytes(os.urandom(1), 'big') % 2 for _ in range(10)]
bit = int.from_bytes(os.urandom(1), 'big') % 2
pattern.append(bit)
return pattern
def _generate_long_key(self, length=4096): def _generate_long_key(self, length=4096):
"""Generate a long hex key (4096 chars by default)."""
chars = "0123456789abcdef" chars = "0123456789abcdef"
result = [] return ''.join(chars[int.from_bytes(os.urandom(1), 'big') % 16] for _ in range(length))
for _ in range(length):
idx = int.from_bytes(os.urandom(1), 'big') % 16
result.append(chars[idx])
return ''.join(result)
def _generate_short_key(self, length=512): def _generate_short_key(self, length=512):
"""Generate a short hex key (512 chars by default)."""
chars = "0123456789abcdef" chars = "0123456789abcdef"
result = [] return ''.join(chars[int.from_bytes(os.urandom(1), 'big') % 16] for _ in range(length))
for _ in range(length):
idx = int.from_bytes(os.urandom(1), 'big') % 16 # ----------------------------------------------------------------------
result.append(chars[idx]) # Key obfuscation and persistence
return ''.join(result) # ----------------------------------------------------------------------
def _obfuscate_keys(self, keys_data): def _obfuscate_keys(self, keys_data):
"""Obfuscate key data using base64 + reversal + Caesar shift."""
json_str = json.dumps(keys_data) json_str = json.dumps(keys_data)
b64 = base64.b64encode(json_str.encode()).decode() b64 = base64.b64encode(json_str.encode()).decode()
reversed_b64 = b64[::-1] reversed_b64 = b64[::-1]
shifted = ''.join([chr((ord(c) + 1) % 128) for c in reversed_b64]) shifted = ''.join(chr((ord(c) + 1) % 128) for c in reversed_b64)
final = base64.b64encode(shifted.encode()).decode() return base64.b64encode(shifted.encode()).decode()
return final
def _deobfuscate_keys(self, obfuscated_data): def _deobfuscate_keys(self, obfuscated_data):
"""Reverse the obfuscation to recover original key data."""
try: try:
shifted = base64.b64decode(obfuscated_data.encode()).decode() 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] b64 = reversed_b64[::-1]
json_str = base64.b64decode(b64.encode()).decode() json_str = base64.b64decode(b64.encode()).decode()
return json.loads(json_str) return json.loads(json_str)
except Exception: except Exception as exc:
raise Exception("密钥文件损坏") raise Exception("Key file corrupted") from exc
def _xor_encrypt_data(self, data, password): def _xor_encrypt_data(self, data, password):
result = [] """XOR encrypt data with a password, returning hex string."""
key_len = len(password) key_len = len(password)
for i, char in enumerate(data): return ''.join(
xor_result = ord(char) ^ ord(password[i % key_len]) f"{ord(char) ^ ord(password[i % key_len]):02x}"
result.append(f"{xor_result:02x}") for i, char in enumerate(data)
return ''.join(result) )
def _xor_decrypt_data(self, hex_data, password): def _xor_decrypt_data(self, hex_data, password):
"""XOR decrypt hex data with a password."""
try: try:
result = []
key_len = len(password) key_len = len(password)
result = []
for i in range(0, len(hex_data), 2): for i in range(0, len(hex_data), 2):
if i + 1 < len(hex_data): if i + 1 < len(hex_data):
hex_byte = hex_data[i:i+2] hex_byte = hex_data[i:i+2]
@@ -143,17 +163,28 @@ class EncryptionSystem:
return None return None
def generate_keys(self, save_path=None, key_password=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: if key_password is None:
key_password = input("请设置密钥文件密码: ") key_password = input("Set key file password: ")
confirm = input("请再次输入密码确认: ") confirm = input("Confirm password: ")
if key_password != confirm: if key_password != confirm:
print("❌ 密码不匹配") print("❌ Passwords do not match")
return None return None
print("=" * 60) print("=" * 60)
print("正在生成随机密钥...") print("Generating random keys...")
print("=" * 60) print("=" * 60)
# Generate all key components
self.upper_mapping = self._generate_random_alphabet(lowercase=False) self.upper_mapping = self._generate_random_alphabet(lowercase=False)
self.lower_mapping = self._generate_random_alphabet(lowercase=True) self.lower_mapping = self._generate_random_alphabet(lowercase=True)
self.digit_mapping = self._generate_random_digit_mapping() self.digit_mapping = self._generate_random_digit_mapping()
@@ -161,10 +192,12 @@ class EncryptionSystem:
self.flip_pattern = self._generate_random_flip_pattern() self.flip_pattern = self._generate_random_flip_pattern()
self.long_key = self._generate_long_key(4096) self.long_key = self._generate_long_key(4096)
self.short_key = self._generate_short_key(512) 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.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.equal_reverse = {v: k for k, v in self.equal_mapping.items()}
# Bundle keys into a dictionary
keys_data = { keys_data = {
'upper_mapping': self.upper_mapping, 'upper_mapping': self.upper_mapping,
'lower_mapping': self.lower_mapping, 'lower_mapping': self.lower_mapping,
@@ -175,30 +208,36 @@ class EncryptionSystem:
'short_key': self.short_key, 'short_key': self.short_key,
'generated_at': datetime.now().isoformat() 'generated_at': datetime.now().isoformat()
} }
# Obfuscate and encrypt the key data
obfuscated = self._obfuscate_keys(keys_data) obfuscated = self._obfuscate_keys(keys_data)
encrypted = self._xor_encrypt_data(obfuscated, key_password) encrypted = self._xor_encrypt_data(obfuscated, key_password)
if save_path is None: if save_path is None:
save_path = "encryption.key" save_path = "encryption.key"
with open(save_path, 'w') as f: with open(save_path, 'w') as f:
f.write(encrypted) f.write(encrypted)
self.key_file = save_path self.key_file = save_path
self.key_password = key_password self.key_password = key_password
self.keys_loaded = True self.keys_loaded = True
self._build_maps() self._build_maps()
print(f"✅ 密钥已生成并保存到: {save_path}") print(f"✅ Keys generated and saved to: {save_path}")
print("=" * 60) print("=" * 60)
return save_path return save_path
# ----------------------------------------------------------------------
# Internal helpers
# ----------------------------------------------------------------------
def _build_maps(self): def _build_maps(self):
"""Build encryption and decryption maps from shuffled alphabets."""
self.upper_original = "ABCDEFGHIJKLMNOPQRSTUVWXYZ" self.upper_original = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
self.lower_original = "abcdefghijklmnopqrstuvwxyz" self.lower_original = "abcdefghijklmnopqrstuvwxyz"
self.encrypt_map = {} self.encrypt_map = {}
self.decrypt_map = {} self.decrypt_map = {}
for i in range(26): for i in range(26):
@@ -208,18 +247,19 @@ class EncryptionSystem:
self.decrypt_map[self.lower_mapping[i]] = self.lower_original[i] self.decrypt_map[self.lower_mapping[i]] = self.lower_original[i]
def _load_keys(self, key_file, key_password): def _load_keys(self, key_file, key_password):
"""Load and decrypt keys from a key file."""
try: try:
with open(key_file, 'r') as f: with open(key_file, 'r') as f:
encrypted_data = f.read() encrypted_data = f.read()
decrypted = self._xor_decrypt_data(encrypted_data, key_password) decrypted = self._xor_decrypt_data(encrypted_data, key_password)
if decrypted is None: if decrypted is None:
print("❌ 密码错误") print("❌ Incorrect password")
self.keys_loaded = False self.keys_loaded = False
return return
keys_data = self._deobfuscate_keys(decrypted) keys_data = self._deobfuscate_keys(decrypted)
self.upper_mapping = keys_data['upper_mapping'] self.upper_mapping = keys_data['upper_mapping']
self.lower_mapping = keys_data['lower_mapping'] self.lower_mapping = keys_data['lower_mapping']
self.digit_mapping = keys_data['digit_mapping'] self.digit_mapping = keys_data['digit_mapping']
@@ -227,25 +267,26 @@ class EncryptionSystem:
self.flip_pattern = keys_data['flip_pattern'] self.flip_pattern = keys_data['flip_pattern']
self.long_key = keys_data['long_key'] self.long_key = keys_data['long_key']
self.short_key = keys_data['short_key'] self.short_key = keys_data['short_key']
self.digit_reverse = {v: k for k, v in self.digit_mapping.items()} 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.equal_reverse = {v: k for k, v in self.equal_mapping.items()}
self.key_password = key_password self.key_password = key_password
self._build_maps() self._build_maps()
self.keys_loaded = True self.keys_loaded = True
print(f"✅ 密钥已从 {key_file} 加载") print(f"✅ Keys loaded from {key_file}")
except Exception: except Exception:
print("❌ 加载失败") print("❌ Loading failed")
self.keys_loaded = False self.keys_loaded = False
def _ensure_printable(self, text): def _ensure_printable(self, text):
"""Force all characters into printable ASCII range (32-126)."""
result = [] result = []
for c in text: for c in text:
val = ord(c) val = ord(c)
if val == 124: if val == 124: # '|' is used as a separator, keep it
result.append('|') result.append('|')
elif val < 32 or val > 126: elif val < 32 or val > 126:
val = val % 95 + 32 val = val % 95 + 32
@@ -254,10 +295,25 @@ class EncryptionSystem:
result.append(c) result.append(c)
return ''.join(result) return ''.join(result)
# ----------------------------------------------------------------------
# Key derivation
# ----------------------------------------------------------------------
def _mix_keys(self, user_password, target_length): 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: if not self.keys_loaded:
raise Exception("密钥未加载") raise Exception("Keys not loaded")
# Interleave user password and short key
mixed = [] mixed = []
max_len = max(len(user_password), len(self.short_key)) max_len = max(len(user_password), len(self.short_key))
for i in range(max_len): for i in range(max_len):
@@ -265,7 +321,8 @@ class EncryptionSystem:
mixed.append(ord(user_password[i])) mixed.append(ord(user_password[i]))
if i < len(self.short_key): if i < len(self.short_key):
mixed.append(ord(self.short_key[i])) mixed.append(ord(self.short_key[i]))
# Apply bit transformations
for i in range(len(mixed)): for i in range(len(mixed)):
if i % 3 == 0: if i % 3 == 0:
mixed[i] = (mixed[i] << 1) & 0xFF mixed[i] = (mixed[i] << 1) & 0xFF
@@ -273,21 +330,24 @@ class EncryptionSystem:
mixed[i] = (mixed[i] >> 1) & 0xFF mixed[i] = (mixed[i] >> 1) & 0xFF
else: else:
mixed[i] = mixed[i] ^ 0x5A mixed[i] = mixed[i] ^ 0x5A
# Swap pairs
for i in range(0, len(mixed) - 3, 4): for i in range(0, len(mixed) - 3, 4):
mixed[i], mixed[i+3] = mixed[i+3], mixed[i] mixed[i], mixed[i+3] = mixed[i+3], mixed[i]
mixed[i+1], mixed[i+2] = mixed[i+2], mixed[i+1] mixed[i+1], mixed[i+2] = mixed[i+2], mixed[i+1]
mixed.reverse() mixed.reverse()
# Convert to printable characters
result = [] result = []
for x in mixed: for x in mixed:
if x < 32 or x > 126: if x < 32 or x > 126:
x = x % 95 + 32 x = x % 95 + 32
result.append(chr(x)) result.append(chr(x))
result_str = ''.join(result) result_str = ''.join(result)
# Extend if needed by repeating with variations
if len(result_str) < target_length: if len(result_str) < target_length:
base_key = result_str base_key = result_str
final_key = base_key final_key = base_key
@@ -297,22 +357,29 @@ class EncryptionSystem:
next_chunk = base_key[::-1] next_chunk = base_key[::-1]
elif iteration % 3 == 1: elif iteration % 3 == 1:
next_chunk = base_key[::-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: else:
next_chunk = base_key[::-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)
final_key += next_chunk final_key += next_chunk
iteration += 1 iteration += 1
result_str = final_key[:target_length] result_str = final_key[:target_length]
return self._ensure_printable(result_str) return self._ensure_printable(result_str)
def _derive_final_key(self, user_password, text_length): 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) 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] reversed_hex = hex_key[::-1]
# Convert back to printable characters
final_key = [] final_key = []
for i in range(0, len(reversed_hex), 2): for i in range(0, len(reversed_hex), 2):
if i + 1 < len(reversed_hex): if i + 1 < len(reversed_hex):
@@ -324,9 +391,10 @@ class EncryptionSystem:
final_key.append(chr(val)) final_key.append(chr(val))
except ValueError: except ValueError:
final_key.append('x') final_key.append('x')
final_key_str = ''.join(final_key) final_key_str = ''.join(final_key)
# Extend if needed
if len(final_key_str) < text_length: if len(final_key_str) < text_length:
base_key = final_key_str base_key = final_key_str
iteration = 0 iteration = 0
@@ -335,22 +403,25 @@ class EncryptionSystem:
next_chunk = base_key[::-1] next_chunk = base_key[::-1]
elif iteration % 3 == 1: elif iteration % 3 == 1:
next_chunk = base_key[::-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: else:
next_chunk = base_key[::-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)
final_key_str += next_chunk final_key_str += next_chunk
iteration += 1 iteration += 1
final_key_str = final_key_str[:text_length] final_key_str = final_key_str[:text_length]
final_key_str = self._ensure_printable(final_key_str) return self._ensure_printable(final_key_str)
return final_key_str # ----------------------------------------------------------------------
# Core encryption operations
# ----------------------------------------------------------------------
def _apply_flip(self, text): def _apply_flip(self, text):
"""Apply case-flipping based on the flip pattern."""
if self.flip_pattern is None: if self.flip_pattern is None:
return text return text
result = [] result = []
for i, ch in enumerate(text): for i, ch in enumerate(text):
idx = i % len(self.flip_pattern) idx = i % len(self.flip_pattern)
@@ -361,15 +432,16 @@ class EncryptionSystem:
return ''.join(result) return ''.join(result)
def _xor_encrypt_with_final_key(self, text, user_password): 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)) final_key = self._derive_final_key(user_password, len(text))
key_len = len(final_key) key_len = len(final_key)
result = [] return ''.join(
for i, char in enumerate(text): f"{ord(char) ^ ord(final_key[i % key_len]):02x}"
xor_result = ord(char) ^ ord(final_key[i % key_len]) for i, char in enumerate(text)
result.append(f"{xor_result:02x}") )
return ''.join(result)
def _xor_decrypt_with_final_key(self, hex_text, user_password): def _xor_decrypt_with_final_key(self, hex_text, user_password):
"""XOR decrypt with the derived final key."""
try: try:
text_length = len(hex_text) // 2 text_length = len(hex_text) // 2
final_key = self._derive_final_key(user_password, text_length) final_key = self._derive_final_key(user_password, text_length)
@@ -390,47 +462,35 @@ class EncryptionSystem:
return None return None
def _substitute_letters(self, text, mapping): def _substitute_letters(self, text, mapping):
result = [] """Apply a substitution map to letters."""
for char in text: return ''.join(mapping.get(char, char) for char in text)
if char in mapping:
result.append(mapping[char])
else:
result.append(char)
return ''.join(result)
def _encode_digit(self, num_str): def _encode_digit(self, num_str):
result = [] """Encode a digit string using the digit mapping."""
for char in num_str: return ''.join(self.digit_mapping.get(char, char) for char in num_str)
if char in self.digit_mapping:
result.append(self.digit_mapping[char])
else:
result.append(char)
return ''.join(result)
def _decode_digit(self, encoded_str): def _decode_digit(self, encoded_str):
result = [] """Decode a digit string using the reverse digit mapping."""
for char in encoded_str: return ''.join(self.digit_reverse.get(char, char) for char in encoded_str)
if char in self.digit_reverse:
result.append(self.digit_reverse[char])
else:
result.append(char)
return ''.join(result)
def _get_dynamic_key(self, text_length): def _get_dynamic_key(self, text_length):
"""Generate a dynamic key from the long key, repeated as needed."""
key = self.long_key key = self.long_key
while len(key) < text_length: if len(key) < text_length:
key += self.long_key repeats = (text_length // len(key)) + 1
key = (key * repeats)[:text_length]
return key[:text_length] return key[:text_length]
def _xor_encrypt_with_key(self, text, key): def _xor_encrypt_with_key(self, text, key):
"""XOR encrypt with a fixed key, returning hex."""
key_len = len(key) key_len = len(key)
result = [] return ''.join(
for i, char in enumerate(text): f"{ord(char) ^ ord(key[i % key_len]):02x}"
xor_result = ord(char) ^ ord(key[i % key_len]) for i, char in enumerate(text)
result.append(f"{xor_result:02x}") )
return ''.join(result)
def _xor_decrypt_with_key(self, hex_text, key): def _xor_decrypt_with_key(self, hex_text, key):
"""XOR decrypt with a fixed key."""
try: try:
key_len = len(key) key_len = len(key)
result = [] result = []
@@ -448,45 +508,74 @@ class EncryptionSystem:
except Exception: except Exception:
return None return None
# ----------------------------------------------------------------------
# Public API
# ----------------------------------------------------------------------
def encrypt(self, plaintext, user_password): 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: 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') b64 = base64.b64encode(plaintext.encode('utf-8')).decode('utf-8')
equal_count = b64.count('=') equal_count = b64.count('=')
processed = b64 processed = b64
for old, new in self.special_encrypt.items(): for old, new in self.special_encrypt.items():
processed = processed.replace(old, new) processed = processed.replace(old, new)
processed = processed.rstrip('=') processed = processed.rstrip('=')
# Step 2: Letter substitution + flip + reverse
sub = self._substitute_letters(processed, self.encrypt_map) sub = self._substitute_letters(processed, self.encrypt_map)
flipped = self._apply_flip(sub) flipped = self._apply_flip(sub)
reversed_text = flipped[::-1] reversed_text = flipped[::-1]
# Step 3: Append padding count
equal_char = self.equal_mapping[str(equal_count)] equal_char = self.equal_mapping[str(equal_count)]
with_equal = f"{reversed_text}|{equal_char}" with_equal = f"{reversed_text}|{equal_char}"
# Step 4: Dynamic key (long key) encryption
key_length_str = str(len(with_equal)) key_length_str = str(len(with_equal))
key_length_encoded = self._encode_digit(key_length_str) key_length_encoded = self._encode_digit(key_length_str)
dynamic_key = self._get_dynamic_key(len(with_equal)) dynamic_key = self._get_dynamic_key(len(with_equal))
encrypted_by_dynamic = self._xor_encrypt_with_key(with_equal, dynamic_key) 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}" combined = f"{encrypted_by_dynamic}|{key_length_encoded}"
final_encrypted = self._xor_encrypt_with_final_key(combined, user_password) final_encrypted = self._xor_encrypt_with_final_key(combined, user_password)
return final_encrypted return final_encrypted
def decrypt(self, ciphertext, user_password): 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: if not self.keys_loaded:
return "❌ 解密失败" return "❌ Decryption failed"
try: try:
# Step 1: Decrypt with derived final key
combined = self._xor_decrypt_with_final_key(ciphertext, user_password) combined = self._xor_decrypt_with_final_key(ciphertext, user_password)
if combined is None: if combined is None:
return "❌ 解密失败" return "❌ Decryption failed"
# Step 2: Extract hex data and length indicator
if '|' in combined: if '|' in combined:
parts = combined.split('|') parts = combined.split('|')
if len(parts) >= 2: if len(parts) >= 2:
@@ -498,211 +587,234 @@ class EncryptionSystem:
else: else:
hex_data = combined hex_data = combined
key_length_encoded = 'g' key_length_encoded = 'g'
# Step 3: Decode length and get dynamic key
key_length_str = self._decode_digit(key_length_encoded) key_length_str = self._decode_digit(key_length_encoded)
try: try:
key_length = int(key_length_str) key_length = int(key_length_str)
except ValueError: except ValueError:
key_length = 16 key_length = 16
dynamic_key = self._get_dynamic_key(key_length) 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) xor_decrypted = self._xor_decrypt_with_key(hex_data, dynamic_key)
if xor_decrypted is None: if xor_decrypted is None:
return "❌ 解密失败" return "❌ Decryption failed"
# Step 5: Extract main data and padding count
if '|' in xor_decrypted: if '|' in xor_decrypted:
main_part, equal_char = xor_decrypted.split('|') main_part, equal_char = xor_decrypted.split('|')
equal_count = int(self.equal_reverse.get(equal_char, '0')) equal_count = int(self.equal_reverse.get(equal_char, '0'))
else: else:
main_part = xor_decrypted main_part = xor_decrypted
equal_count = 0 equal_count = 0
# Step 6: Reverse, flip, substitute
reversed_text = main_part[::-1] reversed_text = main_part[::-1]
flipped = self._apply_flip(reversed_text) flipped = self._apply_flip(reversed_text)
sub = self._substitute_letters(flipped, self.decrypt_map) sub = self._substitute_letters(flipped, self.decrypt_map)
# Step 7: Restore Base64 special chars and padding
for old, new in self.special_decrypt.items(): for old, new in self.special_decrypt.items():
sub = sub.replace(old, new) sub = sub.replace(old, new)
b64_with_equal = sub + '=' * equal_count b64_with_equal = sub + '=' * equal_count
if len(b64_with_equal) % 4 != 0: 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') decoded = base64.b64decode(b64_with_equal.encode('utf-8')).decode('utf-8')
return decoded return decoded
except Exception: except Exception:
return "❌ 解密失败" return "❌ Decryption failed"
def print_keys(self): def print_keys(self):
"""Display information about the currently loaded keys."""
if not self.keys_loaded: if not self.keys_loaded:
print("❌ 未加载密钥") print("❌ Keys not loaded")
return return
print("=" * 60) print("=" * 60)
print("密钥信息") print("Key Information")
print("=" * 60) print("=" * 60)
print(f"密钥文件: {self.key_file}") print(f"Key file: {self.key_file}")
print(f"大写映射表: {self.upper_mapping}") print(f"Uppercase mapping: {self.upper_mapping}")
print(f"小写映射表: {self.lower_mapping}") print(f"Lowercase mapping: {self.lower_mapping}")
print(f"翻转模式: {self.flip_pattern}") print(f"Flip pattern: {self.flip_pattern}")
print(f"长密钥长度: {len(self.long_key)} 位") print(f"Long key length: {len(self.long_key)} bits")
print(f"短密钥长度: {len(self.short_key)} 位") print(f"Short key length: {len(self.short_key)} bits")
print("=" * 60) print("=" * 60)
# ----------------------------------------------------------------------
# CLI Entry Point
# ----------------------------------------------------------------------
def main(): def main():
"""Command-line interface for the encryption system."""
print("=" * 60) print("=" * 60)
print("欢迎使用加密系统") print("Welcome to the Encryption System")
print("=" * 60) print("=" * 60)
crypto = None crypto = None
key_password = None key_password = None
# Try loading default key file
if os.path.exists("encryption.key"): 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) crypto = EncryptionSystem("encryption.key", key_password)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
crypto = None crypto = None
else: else:
print("\n未找到默认密钥文件,请先生成") print("\nDefault key file not found. Please generate one first.")
choice = input("是否生成默认密钥文件?(y/n): ").strip().lower() choice = input("Generate default key file? (y/n): ").strip().lower()
if choice == 'y': if choice == 'y':
crypto = EncryptionSystem() crypto = EncryptionSystem()
key_password = crypto.generate_keys("encryption.key") result = crypto.generate_keys("encryption.key")
if key_password is None: if result is None:
print("❌ 生成失败") print("❌ Generation failed")
crypto = None crypto = None
else: else:
print("⚠️ 请使用模式4或5指定密钥文件,或模式6生成新密钥") print("⚠️ Use options 4, 5, or 6 to manage key files manually.")
# Main interaction loop
while True: while True:
print("\n请选择操作:") print("\nSelect an option:")
print("1. 使用默认密钥加密") print("1. Encrypt with default key")
print("2. 使用默认密钥解密") print("2. Decrypt with default key")
print("3. 生成新密钥(覆盖默认)") print("3. Generate new key (overwrite default)")
print("4. 使用指定密钥文件加密") print("4. Encrypt with custom key file")
print("5. 使用指定密钥文件解密") print("5. Decrypt with custom key file")
print("6. 生成密钥并保存到当前文件夹") print("6. Generate key and save to current folder")
print("7. 查看当前密钥信息") print("7. View current key info")
print("8. 退出") print("8. Exit")
choice = input("\n请选择操作 (1-8): ").strip() choice = input("\nEnter choice (1-8): ").strip()
if choice == '1': if choice == '1':
# Encrypt with default key
if crypto is None or not crypto.keys_loaded: if crypto is None or not crypto.keys_loaded:
if os.path.exists("encryption.key"): if os.path.exists("encryption.key"):
if key_password is None: if key_password is None:
key_password = input("请输入默认密钥文件密码: ") key_password = input("Enter default key file password: ")
crypto = EncryptionSystem("encryption.key", key_password) crypto = EncryptionSystem("encryption.key", key_password)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
continue continue
else: else:
print("❌ 密钥文件不存在") print("❌ Key file not found")
continue continue
password = input("请输入加密密码: ") password = input("Enter encryption password: ")
text = input("请输入要加密的文本: ") text = input("Enter text to encrypt: ")
if text and password: if text and password:
encrypted = crypto.encrypt(text, password) encrypted = crypto.encrypt(text, password)
print(f"\n✅ 加密结果: {encrypted}") print(f"\n✅ Encrypted result: {encrypted}")
elif choice == '2': elif choice == '2':
# Decrypt with default key
if crypto is None or not crypto.keys_loaded: if crypto is None or not crypto.keys_loaded:
if os.path.exists("encryption.key"): if os.path.exists("encryption.key"):
if key_password is None: if key_password is None:
key_password = input("请输入默认密钥文件密码: ") key_password = input("Enter default key file password: ")
crypto = EncryptionSystem("encryption.key", key_password) crypto = EncryptionSystem("encryption.key", key_password)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
continue continue
else: else:
print("❌ 密钥文件不存在") print("❌ Key file not found")
continue continue
password = input("请输入加密密码: ") password = input("Enter encryption password: ")
text = input("请输入要解密的密文: ") text = input("Enter ciphertext to decrypt: ")
if text and password: if text and password:
decrypted = crypto.decrypt(text, password) decrypted = crypto.decrypt(text, password)
print(f"\n✅ 解密结果: {decrypted}") print(f"\n✅ Decrypted result: {decrypted}")
elif choice == '3': elif choice == '3':
# Generate and overwrite default key
crypto = EncryptionSystem() crypto = EncryptionSystem()
key_password = crypto.generate_keys("encryption.key") result = crypto.generate_keys("encryption.key")
if key_password is None: if result is None:
print("❌ 生成失败") print("❌ Generation failed")
else: else:
print("✅ 默认密钥已更新") print("✅ Default key updated")
elif choice == '4': 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): if not os.path.exists(key_file):
print(f"❌ 文件不存在") print(f"❌ File not found: {key_file}")
continue continue
kp = input(f"请输入密钥文件密码: ") kp = input("Enter key file password: ")
crypto = EncryptionSystem(key_file, kp) crypto = EncryptionSystem(key_file, kp)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
continue continue
password = input("请输入加密密码: ") password = input("Enter encryption password: ")
text = input("请输入要加密的文本: ") text = input("Enter text to encrypt: ")
if text and password: if text and password:
encrypted = crypto.encrypt(text, password) encrypted = crypto.encrypt(text, password)
print(f"\n✅ 加密结果: {encrypted}") print(f"\n✅ Encrypted result: {encrypted}")
elif choice == '5': 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): if not os.path.exists(key_file):
print(f"❌ 文件不存在") print(f"❌ File not found: {key_file}")
continue continue
kp = input(f"请输入密钥文件密码: ") kp = input("Enter key file password: ")
crypto = EncryptionSystem(key_file, kp) crypto = EncryptionSystem(key_file, kp)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
continue continue
password = input("请输入加密密码: ") password = input("Enter encryption password: ")
text = input("请输入要解密的密文: ") text = input("Enter ciphertext to decrypt: ")
if text and password: if text and password:
decrypted = crypto.decrypt(text, password) decrypted = crypto.decrypt(text, password)
print(f"\n✅ 解密结果: {decrypted}") print(f"\n✅ Decrypted result: {decrypted}")
elif choice == '6': elif choice == '6':
# Generate new key with timestamp
timestamp = int(time.time()) timestamp = int(time.time())
filename = f"key_{timestamp}.key" filename = f"key_{timestamp}.key"
crypto = EncryptionSystem() crypto = EncryptionSystem()
kp = crypto.generate_keys(filename) result = crypto.generate_keys(filename)
if kp is None: if result is None:
print("❌ 生成失败") print("❌ Generation failed")
else: else:
print(f"✅ 密钥已保存到: {filename}") print(f"✅ Key saved to: {filename}")
elif choice == '7': elif choice == '7':
# Show key info
if crypto is None or not crypto.keys_loaded: if crypto is None or not crypto.keys_loaded:
if os.path.exists("encryption.key"): if os.path.exists("encryption.key"):
if key_password is None: if key_password is None:
key_password = input("请输入默认密钥文件密码: ") key_password = input("Enter default key file password: ")
crypto = EncryptionSystem("encryption.key", key_password) crypto = EncryptionSystem("encryption.key", key_password)
if not crypto.keys_loaded: if not crypto.keys_loaded:
print("❌ 加载失败") print("❌ Loading failed")
continue continue
else: else:
print("❌ 未加载密钥") print("❌ Keys not loaded")
continue continue
crypto.print_keys() crypto.print_keys()
elif choice == '8': elif choice == '8':
print("感谢使用,再见!") print("Goodbye!")
break break
else: else:
print("❌ 无效选择") print("❌ Invalid choice")
if __name__ == "__main__": if __name__ == "__main__":
main() main()