security: 环境变量密钥、密码加盐哈希、CSRF 防护

- config.py/web_server.py 支持环境变量注入 STF_TOKEN/WEB_SECRET_KEY,新增 .env 加载和 .env.example;.env 加入 gitignore
- 用户密码从裸 SHA-256 改为 werkzeug 加盐哈希(scrypt),旧哈希登录时自动升级
- CSRF:session token + X-CSRF-Token 请求头校验非 GET 请求,前端自动携带
This commit is contained in:
2026-08-08 21:29:39 +08:00
parent afeb0902a1
commit 838886e043
6 changed files with 111 additions and 10 deletions
+36 -2
View File
@@ -17,10 +17,12 @@ import sys
import time
import uuid
import atexit
import secrets
import subprocess
from datetime import datetime
from flask import Flask, jsonify, request, redirect, url_for, render_template, Response, make_response
from flask import (Flask, jsonify, request, redirect, url_for, render_template,
Response, make_response, session)
from flask_login import (LoginManager, login_user, logout_user,
login_required, current_user)
@@ -36,7 +38,9 @@ from tasks import list_task_types, get_task_class
_log = get_logger("web")
app = Flask(__name__)
app.config["SECRET_KEY"] = "dev-secret-key-change-in-production"
# 生产必须通过环境变量 WEB_SECRET_KEY 设置强密钥;缺省用开发密钥(不安全)
app.config["SECRET_KEY"] = os.environ.get(
"WEB_SECRET_KEY", "dev-secret-key-change-in-production")
app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///" + os.path.join(
os.path.dirname(os.path.abspath(__file__)), "data", "users.db")
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
@@ -56,6 +60,35 @@ def load_user(user_id):
return User.query.get(int(user_id))
# ================== CSRF 防护 ==================
def _csrf_token():
"""获取或生成当前会话的 CSRF token(非 GET 请求需在 X-CSRF-Token 头携带)。"""
if "csrf_token" not in session:
session["csrf_token"] = secrets.token_hex(16)
return session["csrf_token"]
@app.before_request
def _csrf_protect():
"""非安全方法(POST/PUT/DELETE/PATCH)校验 X-CSRF-Token 请求头。"""
if request.method not in ("POST", "PUT", "DELETE", "PATCH"):
return
# 登录表单(未登录,尚无 token)和静态文件跳过
if request.endpoint in ("login", "static"):
return
sess = session.get("csrf_token", "")
header = request.headers.get("X-CSRF-Token", "")
if not sess or header != sess:
return jsonify({"ok": False, "error": "CSRF 校验失败"}), 403
@app.route("/api/csrf")
@login_required
def api_csrf():
"""获取 CSRF token(前端非 GET 请求需携带 X-CSRF-Token 头)。"""
return jsonify({"ok": True, "token": _csrf_token()})
# ================== 页面路由 ==================
@app.route("/")
@login_required
@@ -75,6 +108,7 @@ def login():
user = User.query.filter_by(username=username).first()
if user and user.check_password(password):
login_user(user)
_csrf_token() # 建立 CSRF token,前端通过 /api/csrf 获取
_log.info(f"用户 {username} 登录")
return redirect(request.args.get("next") or url_for("index"))
return render_template("admin/login.html", error="用户名或密码错误")