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 提供端到端加密 + 完整性校验 + 身份认证,服务器无法解密内容。
其安全边界符合"服务器不可信"模型,适合对隐私有要求但不涉及合规认证的通信场景。
元数据泄露与密钥分享渠道仍需用户自行权衡与管理。
================================================================================
+272 -160
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 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 = {
'+': '.',
'/': "'",
@@ -23,38 +38,46 @@ 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()
@@ -162,9 +193,11 @@ class EncryptionSystem:
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,
@@ -176,6 +209,7 @@ class EncryptionSystem:
'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)
@@ -190,12 +224,17 @@ class EncryptionSystem:
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"
@@ -208,13 +247,14 @@ 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
@@ -235,17 +275,18 @@ class EncryptionSystem:
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)
def _mix_keys(self, user_password, target_length):
if not self.keys_loaded:
raise Exception("密钥未加载")
# ----------------------------------------------------------------------
# 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("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):
@@ -266,6 +322,7 @@ class EncryptionSystem:
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
@@ -274,12 +331,14 @@ class EncryptionSystem:
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:
@@ -288,6 +347,7 @@ class EncryptionSystem:
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,10 +357,10 @@ 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]
@@ -308,11 +368,18 @@ class EncryptionSystem:
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):
@@ -327,6 +394,7 @@ class EncryptionSystem:
final_key_str = ''.join(final_key)
# Extend if needed
if len(final_key_str) < text_length:
base_key = final_key_str
iteration = 0
@@ -335,19 +403,22 @@ 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 self._ensure_printable(final_key_str)
return 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
@@ -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,10 +508,28 @@ class EncryptionSystem:
except Exception:
return None
def encrypt(self, plaintext, user_password):
if not self.keys_loaded:
return "❌ 错误:密钥未加载"
# ----------------------------------------------------------------------
# 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 "❌ 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('=')
@@ -460,33 +538,44 @@ class EncryptionSystem:
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:
@@ -499,6 +588,7 @@ class EncryptionSystem:
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)
@@ -506,10 +596,13 @@ class EncryptionSystem:
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'))
@@ -517,192 +610,211 @@ class EncryptionSystem:
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. 退出")
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("\n请选择操作 (1-8): ").strip()
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()