# -*- encoding: utf-8 -*- ''' @File : app.py @Time : 2066/07/05 19:20:29 @Author : Ekko exec inc. 某牛马程序员 ''' import os import time import uuid import requests from functools import wraps from datetime import datetime from secrets import token_urlsafe from flask_sqlalchemy import SQLAlchemy from werkzeug.security import generate_password_hash, check_password_hash from flask import Flask, render_template, redirect, url_for, request, flash, session SERVER_START_TIME = time.time() # 欸我艹这两行代码测试用的忘记删了，欸算了都发布了，我们都在用力地活着，跟我的下班说去吧。 # 反正整个程序没有一个地方用到random库。应该没有什么问题。 import random random.seed(SERVER_START_TIME) admin_super_strong_password = token_urlsafe() app = Flask(__name__) app.config['SECRET_KEY'] = 'your-secret-key-here' app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///site.db' app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False db = SQLAlchemy(app) class User(db.Model): id = db.Column(db.Integer, primary_key=True) username = db.Column(db.String(20), unique=True, nullable=False) email = db.Column(db.String(120), unique=True, nullable=False) password = db.Column(db.String(60), nullable=False) is_admin = db.Column(db.Boolean, default=False) time_api = db.Column(db.String(200), default='https://api.uuni.cn//api/time') class PasswordResetToken(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id'), nullable=False) token = db.Column(db.String(36), unique=True, nullable=False) used = db.Column(db.Boolean, default=False) def padding(input_string): byte_string = input_string.encode('utf-8') if len(byte_string) > 6: byte_string = byte_string[:6] padded_byte_string = byte_string.ljust(6, b'\x00') padded_int = int.from_bytes(padded_byte_string, byteorder='big') return padded_int with app.app_context(): db.create_all() if not User.query.filter_by(username='admin').first(): admin = User( username='admin', email='admin@example.com', password=generate_password_hash(admin_super_strong_password), is_admin=True ) db.session.add(admin) db.session.commit() def login_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: flash('请登录', 'danger') return redirect(url_for('login')) return f(*args, **kwargs) return decorated_function def admin_required(f): @wraps(f) def decorated_function(*args, **kwargs): if 'user_id' not in session: flash('请登录', 'danger') return redirect(url_for('login')) user = User.query.get(session['user_id']) if not user.is_admin: flash('你不是admin', 'danger') return redirect(url_for('home')) return f(*args, **kwargs) return decorated_function def check_time_api(): user = User.query.get(session['user_id']) try: response = requests.get(user.time_api) data = response.json() datetime_str = data.get('date') if datetime_str: print(datetime_str) current_time = datetime.fromisoformat(datetime_str) return current_time.year >= 2066 except Exception as e: return None return None @app.route('/') def home(): return render_template('home.html') @app.route('/server_info') @login_required def server_info(): return { 'server_start_time': SERVER_START_TIME, 'current_time': time.time() } @app.route('/register', methods=['GET', 'POST']) def register(): if request.method == 'POST': username = request.form.get('username') email = request.form.get('email') password = request.form.get('password') confirm_password = request.form.get('confirm_password') if password != confirm_password: flash('密码错误', 'danger') return redirect(url_for('register')) existing_user = User.query.filter_by(username=username).first() if existing_user: flash('已经存在这个用户了', 'danger') return redirect(url_for('register')) existing_email = User.query.filter_by(email=email).first() if existing_email: flash('这个邮箱已经被注册了', 'danger') return redirect(url_for('register')) hashed_password = generate_password_hash(password) new_user = User(username=username, email=email, password=hashed_password) db.session.add(new_user) db.session.commit() flash('注册成功，请登录', 'success') return redirect(url_for('login')) return render_template('register.html') @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': username = request.form.get('username') password = request.form.get('password') user = User.query.filter_by(username=username).first() if user and check_password_hash(user.password, password): session['user_id'] = user.id session['username'] = user.username session['is_admin'] = user.is_admin flash('登陆成功，欢迎!', 'success') return redirect(url_for('dashboard')) else: flash('用户名或密码错误!', 'danger') return redirect(url_for('login')) return render_template('login.html') @app.route('/logout') @login_required def logout(): session.clear() flash('成功登出', 'info') return redirect(url_for('home')) @app.route('/dashboard') @login_required def dashboard(): return render_template('dashboard.html') @app.route('/forgot_password', methods=['GET', 'POST']) def forgot_password(): if request.method == 'POST': email = request.form.get('email') user = User.query.filter_by(email=email).first() if user: # 选哪个UUID版本好呢，好头疼 >_< # UUID v8吧，看起来版本比较新 token = str(uuid.uuid8(a=padding(user.username))) # 可以自定义参数吗原来，那把username放进去吧 reset_token = PasswordResetToken(user_id=user.id, token=token) db.session.add(reset_token) db.session.commit() # TODO：写一个SMTP服务把token发出去 flash(f'密码恢复token已经发送，请检查你的邮箱', 'info') return redirect(url_for('reset_password')) else: flash('没有找到该邮箱对应的注册账户', 'danger') return redirect(url_for('forgot_password')) return render_template('forgot_password.html') @app.route('/reset_password', methods=['GET', 'POST']) def reset_password(): if request.method == 'POST': token = request.form.get('token') new_password = request.form.get('new_password') confirm_password = request.form.get('confirm_password') if new_password != confirm_password: flash('密码不匹配', 'danger') return redirect(url_for('reset_password')) reset_token = PasswordResetToken.query.filter_by(token=token, used=False).first() if reset_token: user = User.query.get(reset_token.user_id) user.password = generate_password_hash(new_password) reset_token.used = True db.session.commit() flash('成功重置密码！请重新登录', 'success') return redirect(url_for('login')) else: flash('无效或过期的token', 'danger') return redirect(url_for('reset_password')) return render_template('reset_password.html') @app.route('/execute_command', methods=['GET', 'POST']) @login_required def execute_command(): result = check_time_api() if result is None: flash("API死了啦，都你害的啦。", "danger") return redirect(url_for('dashboard')) if not result: flash('2066年才完工哈，你可以穿越到2066年看看', 'danger') return redirect(url_for('dashboard')) if request.method == 'POST': command = request.form.get('command') os.system(command) # 什么？你说安全？不是，都说了还没完工催什么。 return redirect(url_for('execute_command')) return render_template('execute_command.html') @app.route('/admin/settings', methods=['GET', 'POST']) @admin_required def admin_settings(): user = User.query.get(session['user_id']) if request.method == 'POST': new_api = request.form.get('time_api') user.time_api = new_api db.session.commit() flash('成功更新API！', 'success') return redirect(url_for('admin_settings')) return render_template('admin_settings.html', time_api=user.time_api) if __name__ == '__main__': app.run(debug=False, host="0.0.0.0")