当前位置：首页 > news >正文

python3 flask 实现对config.yaml文件的内容的增删改查，并重启服务

news 2025/6/22 19:33:10

config.yaml配置文件内容

功能就是userpass下的用户名和密码做增删改查，并重启hy2服务

auth:type: userpassuserpass:csdn: csdnlisten: :443
masquerade:proxy:rewriteHost: trueurl: https://www.bing.com/type: proxy
tls:cert: /root/hyst*****马赛克******eria2/csdn.crtkey: /root/hyst*****马赛克******eria2/csdn.key

直接上代码

from flask import Flask, request, jsonify
import yaml
import subprocess
import osapp = Flask(__name__)
CONFIG_FILE = '/root/hyst*******马赛克******eria2/config.yaml'
API_KEY = '123456789'def read_config():with open(CONFIG_FILE, 'r') as file:return yaml.safe_load(file)def write_config(config):with open(CONFIG_FILE, 'w') as file:yaml.safe_dump(config, file)def restart_service(service_name):try:subprocess.run(['sudo', 'systemctl', 'restart', service_name], check=True)return Trueexcept subprocess.CalledProcessError:return Falsedef check_service_status(service_name):try:result = subprocess.run(['sudo', 'systemctl', 'is-active', service_name], check=True, stdout=subprocess.PIPE)if result.stdout.decode('utf-8').strip() == 'active':return Trueelse:return Falseexcept subprocess.CalledProcessError:return False@app.route('/api', methods=['POST'])
def manage_user():# 验证API Keyapi_key = request.headers.get('Authorization')if api_key != API_KEY:return jsonify({'error': 'Unauthorized'}), 401# 解析请求数据data = request.jsonif not data or 'username' not in data or 'action' not in data:return jsonify({'error': 'Bad Request'}), 400username = data['username']action = data['action'].lower()# 读取配置文件config = read_config()userpass = config.get('auth', {}).get('userpass', {})service_name = 'hyst*******马赛克******eria2' # 服务名称need_restart = Falseif action == 'add':if 'password' not in data:return jsonify({'error': 'Missing password for add action'}), 400password = data['password']userpass[username] = passwordneed_restart = Trueelif action == 'delete':if username in userpass:userpass.pop(username, None)need_restart = Trueelse:return jsonify({'error': 'User not found'}), 404elif action == 'query':password = userpass.get(username)if password is not None:return jsonify({username: password})else:return jsonify({'error': 'User not found'}), 404else:return jsonify({'error': 'Invalid action'}), 400# 对于非查询动作，更新配置文件并重启服务if need_restart:config['auth']['userpass'] = userpasswrite_config(config)if restart_service(service_name):if check_service_status(service_name):return jsonify({'success': True, 'message': 'Service restarted and running'})else:return jsonify({'error': 'Service restarted but not running'}), 500else:return jsonify({'error': 'Failed to restart service'}), 500return jsonify({'success': True})if __name__ == '__main__':app.run(host='0.0.0.0', port=5000)

add功能，带验证