142 lines
3.7 KiB
Python
142 lines
3.7 KiB
Python
from flask import Flask, render_template, request, jsonify, abort, redirect, url_for
|
|
from functools import wraps
|
|
import os
|
|
import subprocess
|
|
import base64
|
|
|
|
app = Flask(__name__)
|
|
|
|
# 配置
|
|
SQUID_PASSWD_FILE = 'config/squid_passwd'
|
|
ADMIN_PASSWORD = os.getenv('SQUID_PASSWORD', 'admin123')
|
|
|
|
|
|
class User:
|
|
def __init__(self, name, password, is_active=True):
|
|
self.name = name
|
|
self.password = password
|
|
self.is_active = is_active
|
|
|
|
|
|
def basic_auth_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
auth = request.authorization
|
|
if not auth or not (auth.username == 'admin' and auth.password == ADMIN_PASSWORD):
|
|
return ('Unauthorized', 401,
|
|
{'WWW-Authenticate': 'Basic realm="Authorization Required"'})
|
|
return f(*args, **kwargs)
|
|
|
|
return decorated
|
|
|
|
|
|
def read_squid_file():
|
|
users = []
|
|
try:
|
|
with open(SQUID_PASSWD_FILE, 'r') as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
|
|
is_active = not line.startswith('#')
|
|
if not is_active:
|
|
line = line[1:]
|
|
|
|
parts = line.split(':', 1)
|
|
if len(parts) == 2:
|
|
users.append(User(parts[0], parts[1], is_active))
|
|
except FileNotFoundError:
|
|
pass
|
|
return users
|
|
|
|
|
|
def write_squid_file(users):
|
|
with open(SQUID_PASSWD_FILE, 'w') as f:
|
|
for user in users:
|
|
line = f"{'#' if not user.is_active else ''}{user.name}:{user.password}\n"
|
|
f.write(line)
|
|
|
|
|
|
def create_user(username, password):
|
|
users = read_squid_file()
|
|
if any(u.name == username for u in users):
|
|
return False
|
|
|
|
try:
|
|
subprocess.run(['htpasswd', '-b', SQUID_PASSWD_FILE, username, password], check=True)
|
|
return True
|
|
except subprocess.CalledProcessError:
|
|
return False
|
|
|
|
|
|
@app.route('/')
|
|
@basic_auth_required
|
|
def index():
|
|
users = read_squid_file()
|
|
return render_template('clients.html', users=users)
|
|
|
|
|
|
@app.route('/toggle/<username>', methods=['POST'])
|
|
@basic_auth_required
|
|
def toggle_user(username):
|
|
users = read_squid_file()
|
|
for user in users:
|
|
if user.name == username:
|
|
user.is_active = not user.is_active
|
|
break
|
|
|
|
write_squid_file(users)
|
|
return '', 200
|
|
|
|
|
|
@app.route('/delete/<username>', methods=['POST'])
|
|
@basic_auth_required
|
|
def delete_user(username):
|
|
users = [u for u in read_squid_file() if u.name != username]
|
|
write_squid_file(users)
|
|
return '', 200
|
|
|
|
|
|
@app.route('/save_user', methods=['POST'])
|
|
@basic_auth_required
|
|
def save_user():
|
|
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'}), 400
|
|
|
|
try:
|
|
subprocess.run(['htpasswd', '-b', SQUID_PASSWD_FILE, username, password], check=True)
|
|
return '', 200
|
|
except subprocess.CalledProcessError:
|
|
return jsonify({'error': 'Failed to update password'}), 500
|
|
|
|
|
|
@app.route('/create_user', methods=['POST'])
|
|
@basic_auth_required
|
|
def handle_create_user():
|
|
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'}), 400
|
|
|
|
if create_user(username, password):
|
|
return '', 200
|
|
else:
|
|
return jsonify({'error': 'User already exists'}), 409
|
|
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
return ('', 401, {'WWW-Authenticate': 'Basic realm="Authorization Required"'})
|
|
|
|
|
|
if __name__ == '__main__':
|
|
if not os.path.exists('config'):
|
|
os.makedirs('config')
|
|
app.run(host='0.0.0.0', port=8080, debug=True) |