order_service.py · 订单与库存管理服务（演示代码）

漏洞描述：全部数据库写/查均使用 Python f-string 将用户可控输入直接嵌入 SQL 语句。攻击者无需任何认证即可构造恶意输入，实现登录绕过、任意数据读取/篡改/删除、整表 DROP。涉及 5 个端点、6 条语句，是本次审计最严重且覆盖面最广的缺陷。

复现步骤：（1）登录绕过：POST /login，参数 name=' OR '1'='1' --，任意 password → 命中第一条用户记录并写入 session；
（2）脱库/破坏：POST /register，参数 name=x', 'x'); DROP TABLE orders;-- → 订单表被删除；
（3）任意退款/批量破坏：POST /refund，参数 oid=1 OR 1=1 → 全部订单置为 refunded。

# 复现（Python requests / curl 均可）
curl -X POST http://target/login \
  -d "name=' OR '1'='1' --&password=whatever"
curl -X POST http://target/refund -d "oid=1 OR 1=1"

修复建议：全部改用参数化查询（? 占位符），杜绝任何字符串拼接 SQL；SQLite 列名/表名不可参数化时应走白名单校验；fetchone() 判空后再取下标。

# 以 login 为例（register/order/refund/stats 同理）
cur = conn.execute(
    "SELECT id FROM users WHERE name=? AND pass=?",
    (name, password))
row = cur.fetchone()

# refund 示例
cur = conn.execute(
    "UPDATE orders SET status='refunded' WHERE id=? AND uid=?",
    (oid, session.get("uid")))

参考链接：OWASP SQL Injection Prevention Cheat Sheet（https://cheatsheetseries.owasp.org/cheatsheets/SQL_Injection_Prevention_Cheat_Sheet.html）；Python sqlite3 参数化文档（https://docs.python.org/3/library/sqlite3.html）。

漏洞描述：/export 端点将请求参数 data 直接 pickle.loads()。pickle 反序列化可构造任意 Python 对象（__reduce__），攻击者可借此执行任意系统命令——这是无前置条件的远程代码执行。同时该端点未做任何认证，公网可达。

复现步骤：

import pickle, os, urllib.parse
class P:
    def __reduce__(self):
        return (os.system, ("calc.exe",))   # 任意命令
payload = urllib.parse.quote(pickle.dumps(P()))
requests.get("http://target/export?data=" + payload)

修复建议：绝不反序列化不可信数据。改为 JSON（json.loads + schema 校验），移除 pickle 导入；若确需传递结构化数据，使用 json 或签名 token。

import json
@app.route("/export")
def export():
    data = request.args.get("data", "[]")
    try:
        payload = json.loads(data)          # 仅限纯 JSON 结构
        # ... 按预期 schema 白名单校验字段后再处理
    except json.JSONDecodeError:
        return "bad data", 400

参考链接：OWASP Deserialization Cheat Sheet（https://cheatsheetseries.owasp.org/cheatsheets/Deserialization_Cheat_Sheet.html）；Python pickle 安全警告（https://docs.python.org/3/library/pickle.html）。

漏洞描述：路由捕获任意 path（含 ../）后直接 os.path.join("avatars", name) 拼接并用 open() 读取，未做路径规范化校验。攻击者可读取服务器任意可读文件：源码、数据库文件 store.db（含明文密码）、系统配置文件等。

复现步骤：

GET /avatar/../../../../etc/passwd
GET /avatar/../store.db      # 直接拉取含明文密码的数据库
GET /avatar/../../order_service.py

修复建议：改用 Flask 的 send_from_directory（内部做路径校验并拒绝越界），并校验解析后绝对路径必须位于白名单目录内；文件名净化（拒绝 ..、空字节、Windows 保留字符）。

import os
from flask import send_from_directory, abort

AVATAR_DIR = os.path.abspath("avatars")

@app.route("/avatar/<path:name>")
def avatar(name):
    # send_from_directory 确保 name 解析后仍在 AVATAR_DIR 内
    return send_from_directory(AVATAR_DIR, name)

参考链接：OWASP Path Traversal（https://owasp.org/www-community/attacks/Path_Traversal）；Flask send_from_directory 文档。

漏洞描述：app.run(host="0.0.0.0", debug=True)：① 绑定全部网卡使服务对局域网/公网暴露；② debug 模式启用 Werkzeug 交互式调试器，在检测到异常时可执行任意 Python 代码（受 PIN 保护，但 PIN 可由文件系统路径/MAC 等环境信息推导，本文件密钥与路径均已泄露/弱化，推导门槛显著降低）；③ 任何未捕获异常会返回完整堆栈（含源码、文件路径、局部变量）。即便关闭 debug 后仍会触发前文已分析的未捕获异常（见 FLASK-AUDIT-014）返回敏感信息。

复现步骤：访问任意触发 500 的端点（如 POST /order 传 pid=abc）→ debug 页加载 → 输入 PIN 后获得 Python shell。

修复建议：生产环境 debug=False（且绝不通过 app.run 直接暴露，应使用 gunicorn/waitress 等生产 WSGI 服务器）；监听地址收敛为 127.0.0.1 并交由反向代理转发；DEBUG 由环境变量强制控制并默认关闭；配全局 @app.errorhandler 返回统一错误页。

# 生产（建议 waitress / gunicorn 承载）
app.run(host="127.0.0.1", port=5000, debug=False)
# 或更优：环境变量门控
app.debug = os.environ.get("FLASK_DEBUG") == "1"   # 默认关闭（fail-closed）

参考链接：Flask 部署文档（https://flask.palletsprojects.com/en/stable/deploying/）；Werkzeug 调试器安全说明。

漏洞描述：用户密码以明文写入 users.pass。一旦数据库泄露（此处已有 SQL 注入 + 路径穿越两条拉库通道），全部用户口令直接暴露；且明文比对（login 中的 pass='{password}'）无法防范撞库/口令复用攻击。违反 OWASP 密码存储备忘单。

修复建议：使用 werkzeug.security.generate_password_hash（PBKDF2）存储，登录时 check_password_hash 比对；查询仅按用户名取记录，密码比对在应用层完成。

from werkzeug.security import generate_password_hash, check_password_hash

# 注册
conn.execute(
    "INSERT INTO users(name, pass) VALUES(?, ?)",
    (name, generate_password_hash(password)))

# 登录（先按用户名取哈希，再在应用层比对）
row = conn.execute("SELECT id, pass FROM users WHERE name=?", (name,)).fetchone()
if row and check_password_hash(row[1], password):
    session["uid"] = row[0]

参考链接：OWASP Password Storage Cheat Sheet（https://cheatsheetseries.owasp.org/cheatsheets/Password_Storage_Cheat_Sheet.html）。

漏洞描述：Flask session 为客户端签名 Cookie，签名密钥 insecure-secret-key-123 硬编码且强度极低。攻击者已知密钥即可用 itsdangerous 直接伪造任意 session["uid"]（包括字符串值，进一步配合 FLASK-AUDIT-001 的 uid 拼接实现注入），冒充任意用户下单/查单。

复现步骤：使用 Python 重放签名 cookie：

from itsdangerous import URLSafeTimedSerializer
s = URLSafeTimedSerializer("insecure-secret-key-123")
cookie = s.dumps({"uid": 999})          # 伪造 uid=999 的会话
# 将 cookie 注入请求即可充当该用户

修复建议：由环境变量注入 ≥32 字节随机密钥，删除硬编码默认值；生产与开发强制分开；SECRET_KEY 缺失时 fail-closed 拒绝启动。

import os, secrets
app.secret_key = os.environ.get("SECRET_KEY")
if not app.secret_key or len(app.secret_key) < 32:
    raise RuntimeError("SECRET_KEY 缺失或过短，禁止启动")

参考链接：Flask Sessions（https://flask.palletsprojects.com/en/stable/quickstart/#sessions）；OWASP Session Management Cheat Sheet。

漏洞描述：target = request.args.get("uid") or uid 允许任意用户通过 ?uid=<他人id> 查询任意用户的订单列表，无任何归属校验。叠加 FLASK-AUDIT-001，target 字符串值同时构成 SQL 注入点。

复现步骤：GET /orders?uid=2（未登录或任意登录态）→ 返回用户 2 的全部订单。

修复建议：完全忽略客户端传入的 uid，仅以当前会话身份查询；如需管理员按 uid 查询，须增加角色校验。

@app.route("/orders")
def my_orders():
    uid = session.get("uid")
    if not uid:
        return "login required", 401
    rows = conn.execute(
        "SELECT * FROM orders WHERE uid=?", (uid,)).fetchall()
    # 仅返回当前登录用户订单，绝不接受 request.args.get("uid")

参考链接：OWASP Broken Object Level Authorization（https://owasp.org/API-Security/editions/2023/en/0xa1-broken-object-level-authorization/）。

漏洞描述：退款端点：① 无任何登录校验；② 不校验订单归属（任意用户可退款他人订单）；③ 不校验订单状态机（可对任意状态订单执行退款，包括从未支付/已发货订单）；④ 配合 SQL 注入（FLASK-AUDIT-001）可批量退款全表。直接破坏资金与订单数据完整性。

复现步骤：POST /refund 传 oid=1 → 订单 1 被置为 refunded，无论其归属与状态。

修复建议：认证 + 归属校验 + 状态机原子条件更新三合一：

@app.route("/refund", methods=["POST"])
def refund():
    uid = session.get("uid")
    if not uid:
        return "login required", 401
    oid = request.form["oid"]
    # 仅允许退款本人处于可退状态('new')的订单；rowcount 验证影响行数
    cur = conn.execute(
        "UPDATE orders SET status='refunded' "
        "WHERE id=? AND uid=? AND status='new'",
        (oid, uid))
    if cur.rowcount == 0:
        return "not refundable", 400
    conn.commit()
    return "refunded"

漏洞描述：下单采用"读库存 → 判断 → sleep → 扣减"的 TOCTOU 模式，非原子操作。两个并发请求同时读到剩余库存 N，均通过 stock >= qty 判断后各自扣减，导致库存变负、实际销量超过库存。第 41 行 time.sleep(0.05) 将竞态窗口放大约 5× 以上，使超卖可稳定复现。

复现步骤：库存为 10，同时并发 20 个 POST /order（qty=1）→ 大部分全部成功，库存最终为 -10，实际下单数 20。

修复建议：改为原子条件 UPDATE：仅当 stock >= qty 时才扣减，并校验 rowcount；扣库存与建订单放入同一显式事务（BEGIN IMMEDIATE）。

@app.route("/order", methods=["POST"])
def place_order():
    uid = session.get("uid")
    if not uid:
        return "login required", 401
    pid = request.form["pid"]; qty = int(request.form["qty"])
    conn.execute("BEGIN IMMEDIATE")     # 立即排他锁，阻止并发写
    cur = conn.execute(
        "UPDATE products SET stock=stock-? WHERE id=? AND stock>=?",
        (qty, pid, qty))
    if cur.rowcount == 0:
        conn.execute("ROLLBACK")
        return "out of stock"
    conn.execute(
        "INSERT INTO orders(uid, pid, qty, status) VALUES(?,?,?,'new')",
        (uid, pid, qty))
    conn.commit()

参考链接：OWASP Transaction Integrity；Python sqlite3 事务与隔离级别文档。

漏洞描述：两处结构性问题：① sqlite3.connect 在模块级创建并被所有请求线程共享——sqlite3 连接默认非线程安全，Flask 开发服务器（threaded=True）下并发请求会触发 "SQLite objects created in a thread can only be used in that same thread" 错误或数据竞争；② 扣库存与插入订单是两条独立 execute，默认 autocommit 下彼此无事务边界——若第二步失败，库存已被扣减却无订单，资金/库存账目错乱。

修复建议：连接改为 per-request（flask.g）并在请求结束时关闭；写操作包入显式事务（BEGIN IMMEDIATE ... COMMIT/ROLLBACK），见 FLASK-AUDIT-009 示例；启用 WAL 模式提升并发读。

from flask import g

def get_db():
    if "db" not in g:
        g.db = sqlite3.connect(DB, check_same_thread=False)
        g.db.row_factory = sqlite3.Row
        g.db.execute("PRAGMA journal_mode=WAL")
    return g.db

@app.teardown_appcontext
def close_db(exc=None):
    db = g.pop("db", None)
    if db is not None:
        db.close()

漏洞描述：① place_order 中 uid = session.get("uid") 未校验登录，未登录时 uid=None 仍会继续执行 → 向 orders 插入 uid 为 NULL 的订单（配合 SQL 注入可改为任意值）；② /export（RCE 入口）与 /stats（全量订单数据 + 产品数据）完全无需认证即可访问。所有敏感端点缺少统一登录门禁。

修复建议：为 /order、/export、/stats 统一增加登录/授权校验（可封装装饰器）；未登录返回 401 并在前端跳转登录。

from functools import wraps
from flask import abort

def login_required(f):
    @wraps(f)
    def wrapper(*a, **kw):
        if not session.get("uid"):
            abort(401)
        return f(*a, **kw)
    return wrapper

@app.route("/order", methods=["POST"])
@login_required
def place_order(): ...

漏洞描述：所有状态变更端点均为裸 POST 表单，无 CSRF token 校验，且会话 Cookie 未设置 SameSite（见 FLASK-AUDIT-018）。攻击者可构造恶意网页自动提交表单，在受害者浏览器中以受害者会话触发下单/退款/注册，实现跨站请求伪造。

复现步骤：在攻击者站点植入自动提交脚本，受害者已登录状态下访问 → 该页面向 http://target/refund POST oid=1 → 受害者订单被退款。

<form action="http://target/refund" method="POST">
  <input name="oid" value="1">
</form>
<script>document.forms[0].submit()</script>

修复建议：接入 Flask-WTF CSRFProtect（全站自动校验），模板表单内嵌 csrf_token；同时设置 SESSION_COOKIE_SAMESITE="Lax" 作为纵深。

from flask_wtf.csrf import CSRFProtect
csrf = CSRFProtect()
csrf.init_app(app)          # 对所有 POST 自动校验 __token__ 字段/头
# 模板：<input type="hidden" name="csrf_token" value="{{ csrf_token() }}">

参考链接：OWASP CSRF Prevention Cheat Sheet；Flask-WTF CSRFProtect 文档。

漏洞描述：登录端点无限流、无账号锁定、无失败计数。配合明文密码（FLASK-AUDIT-005）与 SQL 注入风险，暴力破解可直接命中弱口令用户。

修复建议：按 IP + 用户名维度记录失败次数，超阈值（如 5 次/10 分钟）锁定；建议接入成熟的限流中间件（如 flask-limiter）。

from flask_limiter import Limiter
limiter = Limiter(key_func=lambda: request.remote_addr)
limiter.init_app(app)

@app.route("/login", methods=["POST"])
@limiter.limit("5 per minute")     # 失败/成功合并限流，按需细化
def login(): ...

漏洞描述：① int(request.form["pid"]) 遇非数字/缺失字段直接抛 ValueError/BadRequestKeyError → 500；② cur.fetchone()[0] 在商品 id 不存在时 fetchone 返回 None → AttributeError → 500。在 debug 模式（FLASK-AUDIT-004）下 500 会泄露完整堆栈与源码。任一字段异常输入即可低成本触发，构成稳定性/信息泄露攻击面。

修复建议：表单字段做存在性与类型校验，DB 结果判空；返回 400 并记录日志。

pid_s = request.form.get("pid")
if not pid_s or not pid_s.isdigit():
    return "invalid pid", 400
pid, qty = int(pid_s), int(request.form.get("qty", 0))
row = conn.execute("SELECT stock FROM products WHERE id=?", (pid,)).fetchone()
if row is None:
    return "product not found", 404

漏洞描述：except Exception: pass 静默吞掉所有异常：无日志、无回滚、无错误返回。若 conn.execute 失败而事务处于中间态，后续请求复用同一全局连接将携带脏状态继续执行，造成数据一致性隐性损坏；同时运维无法发现失败退款。

修复建议：捕获具体异常类型，记录完整 traceback（logging.exception），失败时 ROLLBACK，并返回可诊断的错误响应。

@app.route("/refund", methods=["POST"])
def refund():
    oid = request.form["oid"]
    try:
        conn.execute("BEGIN IMMEDIATE")
        conn.execute(
            "UPDATE orders SET status='refunded' WHERE id=? AND uid=? AND status='new'",
            (oid, session.get("uid")))
        conn.commit()
    except sqlite3.Error:
        conn.execute("ROLLBACK")
        app.logger.exception("refund failed: oid=%s uid=%s", oid, session.get("uid"))
        return "refund error", 500
    return "done"

漏洞描述：未注册 @app.errorhandler(404/500)，未配置任何日志（access log、业务审计、异常 log 均缺失）。404/500 时 debug 模式泄露堆栈、生产模式返回裸错误；下单/退款/登录等关键操作无可审计痕迹，无法追溯异常与攻击行为。

修复建议：注册统一错误处理器；配置 logging（文件/RotatingFileHandler），对登录、下单、退款记录结构化审计日志。

import logging
from logging.handlers import RotatingFileHandler

handler = RotatingFileHandler("app.log", maxBytes=1_000_000, backupCount=5)
handler.setFormatter(logging.Formatter(
    "%(asctime)s %(levelname)s [%(thread)s] %(message)s"))
app.logger.addHandler(handler); app.logger.setLevel(logging.INFO)

@app.errorhandler(404)
def not_found(e):
    app.logger.warning("404 %s", request.path)
    return "not found", 404

@app.errorhandler(500)
def server_error(e):
    app.logger.exception("500 on %s", request.path)
    return "internal error", 500

漏洞描述：对每行订单再执行一次产品查询（N+1），订单量为 N 时共 N+1 次查询；该端点未认证（FLASK-AUDIT-011），攻击者可高并发请求放大 DB 负载，将 O(N) 查询叠加并发请求转化为资源耗尽型 DoS；同时 :82 的 f-string 拼接（值来自 DB，常规不可注入，但被注入列污染后成注入链）属于纵深隐患。

复现步骤：订单表 10 万行时单次请求即触发 10 万次子查询；高并发下拖垮 SQLite。

修复建议：改 JOIN 单查询 + 分页；并为 /stats 加认证与限流。

rows = conn.execute(
    "SELECT o.*, p.name AS product_name "
    "FROM orders o JOIN products p ON p.id = o.pid "
    "ORDER BY o.id DESC LIMIT 100").fetchall()   # 一次查询 + 分页

漏洞描述：未配置 SESSION_COOKIE_SECURE（会话可在 HTTP 明文传输中被嗅探劫持）、SESSION_COOKIE_SAMESITE（未设 Lax/Strict 则跨站 POST 携带 Cookie，放大 CSRF 面，见 FLASK-AUDIT-012）。SESSION_COOKIE_HTTPONLY 依赖 Flask 默认值 True（未显式声明）。

修复建议：显式声明全部 Cookie 安全属性；生产强制 Secure + SameSite=Lax/Strict。

app.config.update(
    SESSION_COOKIE_HTTPONLY=True,
    SESSION_COOKIE_SECURE=not app.debug,   # 生产 HTTPS 下必开
    SESSION_COOKIE_SAMESITE="Lax",
)

漏洞描述：全站未设置 Content-Security-Policy、X-Content-Type-Options、X-Frame-Options。当前唯一渲染点 /orders（render_template_string，第 54 行）依赖 Jinja 默认 autoescape 输出 {{ r }}（元组字符串化后整体转义），直接注入 XSS 的路径受限，但该渲染模式脆弱（任一字段改为 |safe 或引入富文本即失守），缺 CSP 使纵深防线为零。

修复建议：全局 after_request 注入安全响应头，生产环境配 CSP。

@app.after_request
def set_headers(resp):
    resp.headers["Content-Security-Policy"] = "default-src 'self'"
    resp.headers["X-Content-Type-Options"] = "nosniff"
    resp.headers["X-Frame-Options"] = "DENY"
    return resp

漏洞描述：常量 STOCK_WARN_THRESHOLD = 100 声明后从未被引用，为死代码/魔法数字残留；命名暗示的"库存预警"功能并未实现。

修复建议：实现预警逻辑（库存低于阈值时告警/限购）或直接删除该常量。

漏洞描述：time.sleep(0.05) 在请求处理路径上同步阻塞：① 将竞态窗口放大数倍（见 FLASK-AUDIT-009）；② 在开发服务器单 worker 多线程模型下，使单个请求占用线程 50ms，高并发请求可迅速占满线程池 → 可用性 DoS。

修复建议：删除该 sleep；竞态由原子 UPDATE + 事务解决（见 FLASK-AUDIT-009），不需要"模拟延迟"。

漏洞描述：① 第 86 行 init_db() 在模块顶层执行（而非仅在 __main__ 内），任何 import order_service（测试、gunicorn preload、其他模块）都会触发建表并连接数据库——隐藏副作用、难以测试；② 第 2 行导入 threading 但全程未使用，属冗余依赖面。

修复建议：把 init_db() 移入应用工厂/启动入口（if __name__ == "__main__": 或 create_app 中）；删除未使用的 threading 导入。

def create_app():
    app = Flask(__name__)
    init_db()                     # 仅在显式创建应用时初始化
    return app

if __name__ == "__main__":
    create_app().run(host="127.0.0.1", debug=False)

漏洞描述：① /orders 将 SQLite 行对象（元组）直接 {{ r }} 输出，向客户端暴露完整内部列结构（字段顺序、内部类型），方便攻击者测绘 schema；② orders 表未对 uid/pid 建索引，WHERE uid=?（查单）与 JOIN（统计）均全表扫描，数据增长后线性劣化。

修复建议：模板渲染命名字段（使用 sqlite3.Row + 明确字段），并建立索引。

conn.row_factory = sqlite3.Row   # 行对象支持 r["name"] 按字段访问
# init_db 中：
conn.execute("CREATE INDEX IF NOT EXISTS idx_orders_uid ON orders(uid)")
conn.execute("CREATE INDEX IF NOT EXISTS idx_orders_pid ON orders(pid)")