代码审查摘要
该代码包含多项高危安全问题，主要集中在命令执行、数据库访问、密码处理、反序列化及临时文件使用等环节。问题可导致远程命令执行、SQL注入、任意代码执行、凭证泄露、权限提升与信息泄露等严重风险。以下提供分级问题清单、可操作修复方案与一份重构示例代码。
详细问题列表（按严重程度分类）
-
严重
- 命令注入
- 位置：run_backup(cmd) 使用 subprocess.run(cmd, shell=True)
- 问题：用户输入直接进入 shell，允许执行任意命令
- 影响：远程命令执行（RCE）
- SQL 注入
- 位置：load_user 中使用 f-string 拼接 SQL
- 问题：未参数化查询，name/pwd 可注入
- 影响：认证绕过、数据泄露或破坏
- 不安全反序列化
- 位置：handle_payload 使用 pickle.loads
- 问题：反序列化任意不受信数据
- 影响：任意代码执行
-
高
4) 弱密码哈希与敏感信息泄露
- 位置：check_password 使用 MD5；main 打印 token
- 问题：MD5 不适用于口令；将哈希作为 token 输出
- 影响：凭证泄露、离线暴力破解
- 临时文件路径可预测
- 位置：write_temp 使用 /tmp/tmp_{pid}.txt
- 问题：可被抢占/软链接攻击，非原子创建
- 影响：信息泄露、文件覆盖、权限提升
- 外部进程缺少约束
- 位置：run_backup 未设置 check/timeout/绝对路径/白名单
- 影响：僵死进程、执行非预期程序、资源耗尽
-
中
7) 异常被吞掉
- 位置：main 中 except Exception as e: pass
- 问题：错误无法追踪，影藏异常
- 影响：可观测性差，问题难以定位
- 明文回显密码输入
- 位置：main 使用 input 读取 pwd
- 问题：终端回显口令
- 影响：旁观者可见，泄露风险
- 数据库资源管理与风格
- 位置：load_user 手动打开/关闭连接与游标
- 问题：缺少 with 语法上下文管理；错误处理缺失
- 影响：异常时资源释放不及时、代码可读性差
-
低
10) 代码风格（PEP8）
- 位置：import os, sqlite3, ...
- 问题：多库同行导入，不符合 PEP8；部分命名不具语义
- 影响：可读性与一致性降低
11) 信息输出
- 位置：print('user:', user)
- 问题：原样打印数据库记录
- 影响：潜在信息泄露（在日志/控制台）
具体修复建议
-
命令执行
- 禁止 shell=True。使用参数列表调用固定的受信程序，采用白名单策略。
- 使用绝对路径而非依赖 PATH。
- 设置 check=True、timeout，并捕获输出以便审计。
- 限制用户可选动作（如 “full”“inc” 两类备份），而非自由命令行。
-
数据库访问与认证
- 全面改用参数化查询：cur.execute("SELECT ... FROM users WHERE name=? AND pwd=?", (name, pwd))
- 口令存储改造：存储基于 PBKDF2/Argon2 的强哈希（带盐）。推荐：
- 标准库：hashlib.pbkdf2_hmac('sha256', ..., iterations>=200000) + 16字节随机盐（secrets.token_bytes）
- 常量时间比较：hmac.compare_digest
- 认证流程：先按用户名查询取出 salt 与 hash，再校验密码，严禁 WHERE pwd=... 的明文比对。
- 使用 with 管理连接与游标；设置 conn.row_factory=sqlite3.Row 提升可读性。
-
反序列化
- 禁止对不受信数据使用 pickle。若需结构化数据，使用 json.loads 或自定义安全解析器。
- 对输入大小与类型进行约束并校验。
-
临时文件
- 使用 tempfile.NamedTemporaryFile(delete=False) 或 tempfile.mkstemp 安全创建；避免可预测文件名。
- 需要跨进程读取时，保留文件名但保证原子创建与 0600 权限。
-
异常与日志
- 不要吞异常。记录到日志（logging），返回或提示不包含敏感细节的错误信息。
- 为外部进程、数据库操作、解析流程添加具体异常处理与告警。
-
输入与输出
- 使用 getpass.getpass 读取密码，避免回显。
- 不打印 token 或敏感字段到标准输出/日志。
- 对输入设置长度上限与格式校验（如 payload 最大长度）。
-
代码风格与可维护性
- 每行一个 import；函数与变量命名具备语义。
- 为安全关键函数添加文档字符串，解释约束与预期输入。
- 在类型敏感处（如函数返回值）可加上类型注解以提升可维护性。
优化方案
-
安全基线
- 为 SQLite 文件设置严格文件权限（仅运行用户可读写）。
- 数据库 schema 增加唯一索引（users.name UNIQUE）与必要约束。
- 统一日志格式与等级；为安全事件标注审计字段。
-
健壮性
- 限制外部进程资源（timeout 已建议）；必要时在上层加入任务队列与并发限制。
- 为输入数据添加长度阈值（如命令参数、payload 不超过固定大小）。
-
可测试性
- 为密码哈希与验证、输入解析、SQL 查询添加单元测试与集成测试。
- 使用临时数据库与测试替身（mocks）验证异常路径与边界条件。
重构代码示例（安全基线实现示例）
说明：
- 示例将“备份命令”改为固定动作白名单，拒绝任意命令。
- 认证改为“按用户名查 salt/hash -> 本地 PBKDF2 校验”的安全流程。需要数据库包含列：users(name UNIQUE, pwd_hash TEXT, pwd_salt TEXT, role TEXT, ...)。生产中需进行数据迁移。
- 反序列化改为 JSON。
- 临时文件使用 tempfile 安全创建。
- 添加基本日志与异常处理。
import json
import hmac
import hashlib
import logging
import secrets
import shlex
import sqlite3
import subprocess
import tempfile
from getpass import getpass
from typing import Optional, Tuple
# 日志基础配置（生产环境按需要调整 handler/formatter）
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 仅允许声明好的备份动作，对应固定可执行文件与参数
ALLOWED_BACKUP_ACTIONS = {
"full": ["/usr/local/bin/safe_backup", "--full"],
"inc": ["/usr/local/bin/safe_backup", "--incremental"],
}
def run_backup(choice: str) -> None:
"""
仅允许调用白名单动作，禁用 shell。
"""
action = choice.strip().lower()
cmd = ALLOWED_BACKUP_ACTIONS.get(action)
if not cmd:
raise ValueError("invalid backup action")
# 绝对路径 + check + timeout，禁用 shell
result = subprocess.run(cmd, check=True, timeout=600, capture_output=True, text=True)
logger.info("backup finished: %s", result.stdout.strip())
def pbkdf2_hash_password(password: str, *, iterations: int = 200_000) -> Tuple[str, str, int]:
"""
基于 PBKDF2-SHA256 的口令派生；返回 (hex_hash, hex_salt, iterations)
"""
salt = secrets.token_bytes(16)
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
return dk.hex(), salt.hex(), iterations
def pbkdf2_verify_password(password: str, hex_hash: str, hex_salt: str, iterations: int) -> bool:
"""
常量时间比较，验证口令
"""
salt = bytes.fromhex(hex_salt)
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt, iterations)
return hmac.compare_digest(dk.hex(), hex_hash)
def load_user(name: str, password: str) -> Optional[Tuple[int, str]]:
"""
安全认证流程：
1) 参数化查询按用户名取回 hash/salt/迭代
2) 本地验证密码
3) 返回 (id, role)
需数据库列：id, name, role, pwd_hash, pwd_salt, pwd_iter
"""
sql = "SELECT id, role, pwd_hash, pwd_salt, pwd_iter FROM users WHERE name = ?"
try:
with sqlite3.connect("users.db", timeout=5) as conn:
conn.row_factory = sqlite3.Row
row = conn.execute(sql, (name,)).fetchone()
except sqlite3.Error as e:
logger.exception("database error")
raise
if not row:
return None
pwd_hash = row["pwd_hash"]
pwd_salt = row["pwd_salt"]
iterations = int(row["pwd_iter"])
if not pbkdf2_verify_password(password, pwd_hash, pwd_salt, iterations):
return None
return row["id"], row["role"]
def handle_payload_json(s: str):
"""
使用 JSON 解析不受信数据，并限制大小。
"""
if len(s) > 10_000: # 上限防止资源滥用，根据业务调优
raise ValueError("payload too large")
try:
return json.loads(s)
except json.JSONDecodeError as e:
raise ValueError("invalid JSON payload") from e
def write_temp(content: str) -> str:
"""
安全创建临时文件（0600），避免可预测路径与竞态。
"""
with tempfile.NamedTemporaryFile(mode="w", delete=False, encoding="utf-8", prefix="app_", suffix=".txt") as f:
f.write(content)
return f.name
def main():
try:
# 仅允许白名单备份动作（例如：full 或 inc）
backup_choice = input("backup action [full/inc]> ")
run_backup(backup_choice)
name = input("name: ")
pwd = getpass("pwd: ")
user = load_user(name, pwd)
if user is None:
print("authentication failed")
return
print("user:", {"id": user[0], "role": user[1]}) # 输出脱敏/最小化
payload_text = input("payload (json): ")
data = handle_payload_json(payload_text)
logger.info("payload accepted: type=%s", type(data).__name__)
path = write_temp("ok")
logger.info("wrote temp file: %s", path)
# 如果需要会话 token，请使用随机值而非密码派生
session_token = secrets.token_urlsafe(32)
logger.info("session started")
# 不要将 token 打印到终端或日志。这里只示例生成方式，不输出。
except subprocess.TimeoutExpired:
logger.error("backup timeout")
except subprocess.CalledProcessError:
logger.error("backup failed")
except Exception:
# 记录日志，避免向用户暴露敏感细节
logger.exception("unexpected error")
print("internal error")
if __name__ == "__main__":
main()
迁移与落地建议（口令哈希）
- 为 users 表新增列：pwd_hash TEXT, pwd_salt TEXT, pwd_iter INTEGER
- 为现有用户执行一次迁移：读取明文或过渡哈希，使用 PBKDF2 生成新哈希与盐，写回新列；完成后清空旧的明文字段
- 新用户注册/修改密码时：调用 pbkdf2_hash_password 存储三元组
以上修改聚焦安全性与可维护性，避免了命令注入、SQL 注入与不安全反序列化等核心风险，并对口令保护、临时文件与可观测性进行了系统强化。
