"""Request-scoped transactions, authentication and account administration helpers.""" import os import secrets import re from functools import wraps from pathlib import Path from flask import (abort, current_app, flash, g, redirect, render_template, request, session, url_for) from sqlalchemy import text from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.orm.exc import StaleDataError from werkzeug.local import LocalProxy from werkzeug.security import check_password_hash, generate_password_hash from models import User def next_page(value): """Only preserve known local destinations from an approval notification.""" if value in ('/', '/orders'): return value if isinstance(value, str) and re.fullmatch(r'/orders/\d+/(?:view|versions/\d+)', value): return value return '/' def login_redirect(): return redirect(url_for('login', next=next_page(request.path))) def get_db(): if 'db' not in g: g.db = current_app.extensions['po_session_factory']() return g.db db = LocalProxy(get_db) def current_user(): uid = session.get('user_id') if not isinstance(uid, int): return None user = db.get(User, uid) if not user or not user.active or session.get('session_version') != user.session_version: session.clear() return None return user def login_required(f): @wraps(f) def wrapper(*args, **kwargs): if not current_user(): return login_redirect() return f(*args, **kwargs) return wrapper def role_required(*roles): def decorator(f): @wraps(f) def wrapper(*args, **kwargs): user = current_user() if not user: return login_redirect() if user.role != 'admin' and user.role not in roles: abort(403) return f(*args, **kwargs) return wrapper return decorator def csrf_token(): if 'csrf_token' not in session: session['csrf_token'] = secrets.token_hex(32) return session['csrf_token'] def validate_password(password): if len(password) < 10 or len(password) > 128: raise ValueError('密码须为 10 至 128 个字符') if password in ('1234567890', 'admin123456', 'password123'): raise ValueError('请勿使用常见密码') def persistent_secret(path): path = Path(path) try: with path.open('x', encoding='ascii') as f: value = secrets.token_hex(32) f.write(value) except FileExistsError: value = path.read_text(encoding='ascii').strip() if len(value) < 32: raise RuntimeError('本地密钥文件不完整，请检查 %s' % path) return value def sign_in(user): session.clear() session.update(user_id=user.id, user_name=user.name, role=user.role, session_version=user.session_version) def init_security(app, factory, instance): app.extensions['po_session_factory'] = factory app.secret_key = os.environ.get('PO_SECRET_KEY') or persistent_secret(Path(instance) / 'session.key') if len(app.secret_key) < 32: raise RuntimeError('PO_SECRET_KEY 至少需要 32 个字符') app.config.update(SESSION_COOKIE_HTTPONLY=True, SESSION_COOKIE_SAMESITE='Lax', SESSION_COOKIE_SECURE=os.environ.get('PO_HTTPS') == '1') with factory() as bootstrap_db: if not bootstrap_db.query(User).count(): persistent_secret(Path(instance) / 'setup.token') @app.before_request def protect_request(): if request.method in ('POST', 'PUT', 'PATCH', 'DELETE'): expected = session.get('csrf_token', '') supplied = request.form.get('csrf_token') or request.headers.get('X-CSRF-Token', '') if not expected or not secrets.compare_digest(expected.encode('utf-8'), supplied.encode('utf-8')): abort(400, description='页面已过期，请刷新后重新操作') # Serialize SQLite writes before reading state or allocating order numbers. # Every request owns its Session; GET requests remain independent readers. db.execute(text('BEGIN IMMEDIATE')) user = current_user() if user and user.must_change_password and request.endpoint not in ('change_password', 'logout', 'static'): return redirect(url_for('change_password', next=next_page(request.path))) @app.teardown_appcontext def close_db(error=None): connection = g.pop('db', None) if connection is not None: connection.rollback() connection.close() @app.errorhandler(ValueError) def invalid_input(error): db.rollback() return render_template('error.html', message=str(error)), 400 @app.errorhandler(SQLAlchemyError) def database_error(error): db.rollback() if isinstance(error, StaleDataError): message, status = '订单已被其他人修改，请刷新后核对最新内容再操作。', 409 elif isinstance(error, IntegrityError): message, status = '数据已存在或被其他记录引用，请检查登录名、单号等内容后重试。', 409 else: app.logger.exception('Database request failed') message, status = '暂时无法保存，请稍后重试；本次操作未生效。', 503 return render_template('error.html', message=message), status @app.context_processor def security_context(): return {'csrf_token': csrf_token, 'request_key': lambda: secrets.token_hex(16), 'current_user': current_user} @app.route('/setup', methods=['GET', 'POST']) def setup(): if db.query(User).count(): return redirect(url_for('login')) if request.method == 'POST': expected = (Path(instance) / 'setup.token').read_text(encoding='ascii').strip() if not secrets.compare_digest(expected.encode('utf-8'), request.form.get('setup_token', '').encode('utf-8')): abort(403) username = request.form.get('username', '').strip() if not username or len(username) > 50: raise ValueError('请填写不超过 50 字的管理员登录名') password = request.form.get('password', '') validate_password(password) user = User(username=username, name=request.form.get('name', '').strip() or '系统管理员', password_hash=generate_password_hash(password), role='admin', active=True, session_version=1, must_change_password=False) db.add(user) db.commit() sign_in(user) return redirect(url_for('index')) return render_template('setup.html') @app.route('/login', methods=['GET', 'POST']) def login(): if not db.query(User).count(): return redirect(url_for('setup')) if request.method == 'POST': user = db.query(User).filter_by(username=request.form.get('username', '').strip()).first() if user and user.active and check_password_hash(user.password_hash, request.form.get('password', '')): sign_in(user) destination = next_page(request.form.get('next')) return redirect(url_for('change_password', next=destination) if user.must_change_password else destination) flash('用户名或密码错误', 'error') return render_template('login.html') @app.route('/logout', methods=['POST']) def logout(): session.clear() return redirect(url_for('login')) @app.route('/account/password', methods=['GET', 'POST']) @login_required def change_password(): user = current_user() if request.method == 'POST': old = request.form.get('old_password', '') password = request.form.get('password', '') if not check_password_hash(user.password_hash, old): raise ValueError('原密码不正确') validate_password(password) if old == password or password != request.form.get('confirm_password'): raise ValueError('新密码须与原密码不同，且两次输入一致') user.password_hash = generate_password_hash(password) user.must_change_password = False user.session_version += 1 db.commit() sign_in(user) flash('密码已更新，其他登录已失效', 'ok') return redirect(next_page(request.form.get('next'))) return render_template('password.html', user=user)