221 lines
8.5 KiB
Python
221 lines
8.5 KiB
Python
"""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)
|