4. 设备端密码管理器搭建（二）：主密码哈希验证与密码增删改查

4.1 主密码哈希验证（PBKDF2）

核心流程示意图

PBKDF2 验证代码示例（Python）

import hashlib
import base64
import os

# ---------- 生成哈希（注册/设置主密码时） ----------
def hash_master_password(password: str, salt: bytes = None) -> dict:
    """使用 PBKDF2-SHA256 生成哈希，返回盐值和哈希值"""
    if salt is None:
        salt = os.urandom(16)  # 生成 16 字节随机盐值
    iterations = 100000  # 迭代次数
    dk = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode('utf-8'),
        salt,
        iterations
    )
    return {
        'salt': base64.b64encode(salt).decode('utf-8'),
        'hash': base64.b64encode(dk).decode('utf-8'),
        'iterations': iterations
    }

# ---------- 验证主密码（登录时） ----------
def verify_master_password(password: str, stored_salt: str, stored_hash: str, iterations: int) -> bool:
    """验证输入的主密码是否与存储的哈希匹配"""
    salt = base64.b64decode(stored_salt)
    dk = hashlib.pbkdf2_hmac(
        'sha256',
        password.encode('utf-8'),
        salt,
        iterations
    )
    return base64.b64encode(dk).decode('utf-8') == stored_hash

# ---------- 使用示例 ----------
# 设置主密码
result = hash_master_password("MySecureMasterP@ss123")
print(f"盐值: {result['salt']}")
print(f"哈希: {result['hash']}")

# 验证主密码
is_valid = verify_master_password(
    "MySecureMasterP@ss123",
    result['salt'],
    result['hash'],
    result['iterations']
)
print(f"验证结果: {is_valid}")  # 输出 True

4.2 密码记录的增删改查（CRUD）

在通过主密码验证后，用户可以对存储的密码记录进行管理。每条记录通常包含：网站/应用名称、用户名、密码、备注。我们使用加密的本地文件（如 SQLite + AES 加密）来存储这些记录。

数据模型与操作表

操作	功能描述	核心代码片段
创建 (Create)	添加一条新的密码记录	INSERT INTO vault (site, username, password, note) VALUES (?, ?, ?, ?)
读取 (Read)	查询所有记录或按关键词搜索	SELECT * FROM vault WHERE site LIKE ? OR username LIKE ?
更新 (Update)	修改已有记录的密码或备注	UPDATE vault SET password=?, note=? WHERE id=?
删除 (Delete)	移除一条密码记录	DELETE FROM vault WHERE id=?

CRUD 操作流程图

完整 CRUD 实现示例（Python + SQLite）

import sqlite3
from cryptography.fernet import Fernet

# 初始化加密密钥（实际应用中应从主密码派生）
cipher_key = Fernet.generate_key()
cipher = Fernet(cipher_key)

def encrypt_password(plaintext: str) -> bytes:
    return cipher.encrypt(plaintext.encode('utf-8'))

def decrypt_password(ciphertext: bytes) -> str:
    return cipher.decrypt(ciphertext).decode('utf-8')

# 连接数据库并创建表
conn = sqlite3.connect('vault.db')
cursor = conn.cursor()
cursor.execute('''
    CREATE TABLE IF NOT EXISTS vault (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        site TEXT NOT NULL,
        username TEXT NOT NULL,
        password_encrypted BLOB NOT NULL,
        note TEXT DEFAULT ''
    )
''')
conn.commit()

# ---------- 创建 (Create) ----------
def add_entry(site: str, username: str, password: str, note: str = ""):
    enc_pw = encrypt_password(password)
    cursor.execute(
        "INSERT INTO vault (site, username, password_encrypted, note) VALUES (?, ?, ?, ?)",
        (site, username, enc_pw, note)
    )
    conn.commit()
    print(f"✅ 已添加: {site} / {username}")

# ---------- 读取 (Read) ----------
def search_entries(keyword: str = ""):
    if keyword:
        cursor.execute(
            "SELECT id, site, username, password_encrypted, note FROM vault WHERE site LIKE ? OR username LIKE ?",
            (f"%{keyword}%", f"%{keyword}%")
        )
    else:
        cursor.execute("SELECT id, site, username, password_encrypted, note FROM vault")
    rows = cursor.fetchall()
    for row in rows:
        decrypted_pw = decrypt_password(row[3])
        print(f"ID: {row[0]} | 站点: {row[1]} | 用户: {row[2]} | 密码: {decrypted_pw} | 备注: {row[4]}")
    return rows

# ---------- 更新 (Update) ----------
def update_entry(entry_id: int, new_password: str, new_note: str = None):
    enc_pw = encrypt_password(new_password)
    if new_note is not None:
        cursor.execute("UPDATE vault SET password_encrypted=?, note=? WHERE id=?", (enc_pw, new_note, entry_id))
    else:
        cursor.execute("UPDATE vault SET password_encrypted=? WHERE id=?", (enc_pw, entry_id))
    conn.commit()
    print(f"✅ 已更新 ID {entry_id}")

# ---------- 删除 (Delete) ----------
def delete_entry(entry_id: int):
    cursor.execute("DELETE FROM vault WHERE id=?", (entry_id,))
    conn.commit()
    print(f"🗑️ 已删除 ID {entry_id}")

# ---------- 使用示例 ----------
add_entry("example.com", "alice", "P@ssw0rd123", "工作邮箱")
add_entry("github.com", "alice_dev", "G!tHub_2025")
search_entries("example")
update_entry(1, "NewP@ss456", "更新备注")
delete_entry(2)
conn.close()