import osfrom flask import Flask, request, jsonify, make_response, render_template, redirect, url_forfrom flask_cors import CORSfrom jwt import encode, decode, ExpiredSignatureError, InvalidTokenErrorimport datetimefrom functools import wrapsimport randomimport stringimport timeimport threadingapp = Flask(__name__)CORS(app, supports_credentials=True)app.config['SECRET_KEY'] = os.environ.get('SECRET_KEY', 'your-very-secret-key-123!@#')app.config['TOKEN_EXPIRATION'] = 1200 # Token有效期(分钟)# 模拟用户数据库users = {'admin': {'password': 'adminpass', 'role': 'admin', 'name': 'Administrator'},'operator': {'password': 'operatorpass', 'role': 'operator', 'name': 'System Operator'},'viewer': {'password': 'viewerpass', 'role': 'viewer', 'name': 'View Only User'}}# 模拟主机数据库hosts = [{'id': 'host-001','hostname': 'web-server-01','ip': '192.168.1.100','os': 'Ubuntu 22.04','status': 'online','last_check': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),'cpu': 35,'memory': 45,'storage': 60,'tags': ['web', 'production'],'created_by': 'admin','created_at': '2023-01-15 09:30:00'},{'id': 'host-002','hostname': 'db-server-01','ip': '192.168.1.101','os': 'CentOS 8','status': 'online','last_check': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),'cpu': 15,'memory': 70,'storage': 85,'tags': ['database', 'production'],'created_by': 'admin','created_at': '2023-02-20 14:15:00'},{'id': 'host-003','hostname': 'backup-server-01','ip': '192.168.1.102','os': 'Debian 11','status': 'offline','last_check': '2023-08-14 09:15:42','cpu': 10,'memory': 20,'storage': 30,'tags': ['backup', 'storage'],'created_by': 'operator','created_at': '2023-03-10 11:20:00'}]# 模拟实时数据更新def update_host_metrics():while True:time.sleep(10) # 每10秒更新一次for host in hosts:if host['status'] == 'online':# 随机更新指标host['cpu'] = max(0, min(100, host['cpu'] + random.randint(-5, 5)))host['memory'] = max(0, min(100, host['memory'] + random.randint(-3, 3)))host['storage'] = max(0, min(100, host['storage'] + random.randint(-1, 1)))host['last_check'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')# 随机切换状态if random.random() < 0.05: # 5%的几率切换状态host['status'] = 'offline' if host['status'] == 'online' else 'online'# 启动后台线程更新主机指标if not app.config.get('TESTING'):metric_thread = threading.Thread(target=update_host_metrics, daemon=True)metric_thread.start()# 登录页面@app.route('/')def index():return render_template('login.html')# 主机管理仪表盘@app.route('/dashboard')def dashboard():return render_template('hosts.html')# 登录接口 - 生成Token@app.route('/login', methods=['POST'])def login():data = request.get_json()username = data.get('username')password = data.get('password')if not username or not password:return jsonify({'error': 'Username and password required'}), 400user = users.get(username)if not user or user['password'] != password:return jsonify({'error': 'Invalid credentials'}), 401# 生成JWT Tokentoken = encode({'sub': username,'name': user['name'],'role': user['role'],'iat': datetime.datetime.utcnow(),'exp': datetime.datetime.utcnow() + datetime.timedelta(minutes=app.config['TOKEN_EXPIRATION'])}, app.config['SECRET_KEY'], algorithm='HS256')# 创建响应response = jsonify({'success': True,'message': 'Login successful','redirect': '/dashboard','user': {'username': username,'name': user['name'],'role': user['role']}})# 通过Cookie设置Tokenresponse.set_cookie('access_token',token,httponly=True,secure=False, # 开发环境下为False,生产环境应为Truesamesite='Lax',max_age=app.config['TOKEN_EXPIRATION'] * 60)print(datetime.datetime.utcnow(),"1111")print(datetime.datetime.utcnow() + datetime.timedelta(minutes=app.config['TOKEN_EXPIRATION']), "2222" )print(token, "3333")return response# Token验证装饰器def token_required(roles=None):if roles is None:roles = ['admin', 'operator', 'viewer']def decorator(f):@wraps(f)def decorated(*args, **kwargs):token = request.cookies.get('access_token')if not token:auth_header = request.headers.get('Authorization')if auth_header and auth_header.startswith('Bearer '):token = auth_header.split(' ')[1]if not token:token = request.headers.get('X-Access-Token')if not token:return jsonify({'error': 'Token is missing!'}), 401print(token,"@@@@@@@@@@@@@@@@")print(decode(token, app.config['SECRET_KEY'], algorithms='HS256'))try:data = decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])current_user = {'username': data['sub'],'name': data['name'],'role': data['role']}print(current_user,"###################")# 检查用户角色权限if current_user['role'] not in roles:return jsonify({'error': 'Insufficient permissions!'}), 403except ExpiredSignatureError:return jsonify({'error': 'Token has expired!'}), 401except InvalidTokenError:return jsonify({'error': 'Invalid token!'}), 401# 将用户信息添加到请求上下文中return f(current_user, *args, **kwargs)return decoratedreturn decorator# 主机管理API@app.route('/api/hosts', methods=['GET'])@token_required(roles=['admin', 'operator', 'viewer'])def get_hosts(current_user):# 根据权限过滤主机if current_user['role'] == 'admin':filtered_hosts = hostselse:filtered_hosts = [h for h in hosts if h['created_by'] == current_user['username']]return jsonify({'hosts': filtered_hosts})@app.route('/api/hosts/<host_id>', methods=['GET'])@token_required(roles=['admin', 'operator', 'viewer'])def get_host(current_user, host_id):host = next((h for h in hosts if h['id'] == host_id), None)if not host:return jsonify({'error': 'Host not found'}), 404# 权限检查if current_user['role'] != 'admin' and host['created_by'] != current_user['username']:return jsonify({'error': 'Access denied to this host'}), 403return jsonify(host)@app.route('/api/hosts', methods=['POST'])@token_required(roles=['admin', 'operator'])def add_host(current_user):data = request.get_json()if not data or not data.get('hostname') or not data.get('ip'):return jsonify({'error': 'Hostname and IP are required'}), 400new_host = {'id': f"host-{''.join(random.choices(string.digits, k=3))}",'hostname': data['hostname'],'ip': data['ip'],'os': data.get('os', 'Linux'),'status': 'online','last_check': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S'),'cpu': random.randint(5, 20),'memory': random.randint(20, 50),'storage': random.randint(10, 40),'tags': data.get('tags', []),'created_by': current_user['username'],'created_at': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')}hosts.append(new_host)return jsonify(new_host), 201@app.route('/api/hosts/<host_id>', methods=['PUT'])@token_required(roles=['admin', 'operator'])def update_host(current_user, host_id):data = request.get_json()host = next((h for h in hosts if h['id'] == host_id), None)if not host:return jsonify({'error': 'Host not found'}), 404# 权限检查if current_user['role'] != 'admin' and host['created_by'] != current_user['username']:return jsonify({'error': 'You can only update your own hosts'}), 403if 'hostname' in data:host['hostname'] = data['hostname']if 'ip' in data:host['ip'] = data['ip']if 'os' in data:host['os'] = data['os']if 'tags' in data:host['tags'] = data['tags']if 'status' in data:host['status'] = data['status']host['updated_by'] = current_user['username']host['updated_at'] = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')return jsonify(host)@app.route('/api/hosts/<host_id>', methods=['DELETE'])@token_required(roles=['admin'])def delete_host(current_user, host_id):global hostshost = next((h for h in hosts if h['id'] == host_id), None)if not host:return jsonify({'error': 'Host not found'}), 404# 权限检查if current_user['role'] != 'admin':return jsonify({'error': 'Only admin can delete hosts'}), 403hosts = [h for h in hosts if h['id'] != host_id]return jsonify({'message': 'Host deleted successfully'}), 200@app.route('/api/hosts/<host_id>/execute', methods=['POST'])@token_required(roles=['admin', 'operator'])def execute_command(current_user, host_id):host = next((h for h in hosts if h['id'] == host_id), None)if not host:return jsonify({'error': 'Host not found'}), 404# 权限检查if current_user['role'] != 'admin' and host['created_by'] != current_user['username']:return jsonify({'error': 'You can only execute commands on your own hosts'}), 403if host['status'] != 'online':return jsonify({'error': 'Host is offline'}), 400data = request.get_json()command = data.get('command', '')if not command:return jsonify({'error': 'Command is required'}), 400# 模拟命令执行time.sleep(1.5) # 模拟执行延迟# 生成模拟输出outputs = {'uptime': '14:30:45 up 45 days, 3:15, 1 user, load average: 0.12, 0.08, 0.06','df -h': 'Filesystem Size Used Avail Use% Mounted on\n/dev/sda1 20G 12G 7.2G 62% /','free -m': ' total used free shared buff/cache available\nMem: 3942 1523 874 123 1544 2034\nSwap: 2047 256 1791','top -n 1': 'top - 14:31:01 up 45 days, 3:15, 1 user, load average: 0.12, 0.08, 0.06\nTasks: 215 total, 1 running, 214 sleeping, 0 stopped, 0 zombie\n%Cpu(s): 5.6 us, 1.2 sy, 0.0 ni, 93.0 id, 0.2 wa, 0.0 hi, 0.0 si, 0.0 st\nMiB Mem : 3942.0 total, 874.1 free, 1523.8 used, 1544.1 buff/cache\nMiB Swap: 2047.0 total, 1791.2 free, 255.8 used. 2034.5 avail Mem'}# 如果命令在预设中,返回预设输出,否则返回通用响应if command in outputs:output = outputs[command]else:output = f"Command '{command}' executed successfully on {host['hostname']}"return jsonify({'host': host['hostname'],'command': command,'output': output,'executed_by': current_user['username'],'timestamp': datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')})# 用户信息接口@app.route('/api/user', methods=['GET'])@token_required()def get_user_info(current_user):return jsonify(current_user)# 登出接口@app.route('/logout', methods=['POST'])def logout():response = jsonify({'message': 'Successfully logged out'})response.set_cookie('access_token', '', expires=0)return response# 健康检查接口@app.route('/health')def health_check():return jsonify({'status': 'ok', 'timestamp': datetime.datetime.now().isoformat()})# 404处理@app.errorhandler(404)def page_not_found(e):return render_template('404.html'), 404if __name__ == '__main__':app.run(host='0.0.0.0', port=5002, debug=True)
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。