#!/usr/bin/env python3
from gevent import pywsgi
from flask import Flask, request, jsonify, render_template_string
import yaml
import subprocess
import threading
import os
import re
import socket
NETPLAN_FILE = "/etc/netplan/01-netcfg.yaml"
PORT = 6000 # 定义端口常量
app = Flask(__name__)
# 前端 HTML 模板
HTML_TEMPLATE = """
网络配置
网络配置
"""
def is_valid_ip(ip):
"""验证IP地址格式"""
if not ip:
return False
pattern = r'^(\d{1,3}\.){3}\d{1,3}$'
if re.match(pattern, ip):
parts = ip.split('.')
return all(0 <= int(part) <= 255 for part in parts)
return False
def get_current_ip():
"""获取当前服务器的IP地址"""
try:
# 获取本机IP
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
current_ip = s.getsockname()[0]
s.close()
return current_ip
except:
return "127.0.0.1"
def read_netplan():
"""读取当前网络配置"""
try:
with open(NETPLAN_FILE) as f:
data = yaml.safe_load(f)
eth = data["network"]["ethernets"]["eth0"]
addresses = eth.get("addresses", [""])
if addresses and addresses[0]:
ip = addresses[0].split("/")[0]
mask = addresses[0].split("/")[1]
else:
ip = ""
mask = "24"
gateway = eth.get("routes", [{}])[0].get("via", "")
dns = eth.get("nameservers", {}).get("addresses", [""])[0]
return {"ip": ip, "mask": mask, "gateway": gateway, "dns": dns}
except Exception as e:
print(f"读取配置文件出错: {e}")
return {"ip": "", "mask": "24", "gateway": "", "dns": "8.8.8.8"}
def apply_netplan_async(new_ip):
"""异步应用网络配置"""
try:
print(f"开始应用网络配置，新IP: {new_ip}")
# 使用 netplan try 应用配置（20秒超时）
process = subprocess.Popen(
["netplan", "try", "--timeout", "20"],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
)
# 自动确认（按回车键）
try:
stdout, stderr = process.communicate(input='\n', timeout=5)
print(f"网络配置已应用，新IP: {new_ip}")
print(f"输出: {stdout}")
if stderr:
print(f"错误: {stderr}")
# 重启网络服务确保配置生效
subprocess.run(["systemctl", "restart", "systemd-networkd"],
capture_output=True, text=True)
print("网络服务已重启")
except subprocess.TimeoutExpired:
# 如果用户没有手动确认，会自动回滚
print("等待用户确认中...配置将在20秒后自动回滚")
process.kill()
except Exception as e:
print(f"应用网络配置时出错: {e}")
except Exception as e:
print(f"启动netplan进程出错: {e}")
@app.route('/')
def index():
net = read_netplan()
current_ip = get_current_ip()
print(f"当前访问IP: {request.remote_addr}, 服务器IP: {current_ip}")
return render_template_string(HTML_TEMPLATE, **net, port=PORT)
@app.route('/api/network', methods=['POST'])
def set_network():
try:
data = request.json
ip = data.get("ip", "").strip()
mask = data.get("mask", "").strip()
gateway = data.get("gateway", "").strip()
dns = data.get("dns", "").strip()
# 验证IP地址格式
if not is_valid_ip(ip):
return jsonify({
"success": False,
"status": "IP地址格式不正确"
}), 400
if not is_valid_ip(gateway):
return jsonify({
"success": False,
"status": "网关地址格式不正确"
}), 400
if dns and not is_valid_ip(dns):
return jsonify({
"success": False,
"status": "DNS地址格式不正确"
}), 400
# 基础验证
if not all([ip, mask, gateway, dns]):
return jsonify({
"success": False,
"status": "所有字段都必须填写"
}), 400
# 读取并更新配置
with open(NETPLAN_FILE) as f:
config = yaml.safe_load(f)
# 更新网络配置
eth = config["network"]["ethernets"]["eth0"]
eth["dhcp4"] = False
eth["addresses"] = [f"{ip}/{mask}"]
eth["routes"] = [{"to": "default", "via": gateway}]
eth["nameservers"] = {"addresses": [dns]}
# 保存配置文件
with open(NETPLAN_FILE, "w") as f:
yaml.safe_dump(config, f, default_flow_style=False)
print(f"配置文件已保存，新IP: {ip}")
# 在后台异步应用网络配置
threading.Thread(target=apply_netplan_async, args=(ip,), daemon=True).start()
return jsonify({
"success": True,
"status": f"配置已保存并正在应用！\n\n请在20秒内按Enter键确认，否则配置会自动回滚。\n跳转地址: http://{ip}:{PORT}",
"new_ip": ip,
"port": PORT,
"redirect_url": f"http://{ip}:{PORT}"
})
except PermissionError:
return jsonify({
"success": False,
"status": "权限不足，请使用sudo运行此程序"
}), 403
except FileNotFoundError:
return jsonify({
"success": False,
"status": f"配置文件 {NETPLAN_FILE} 不存在"
}), 404
except Exception as e:
return jsonify({
"success": False,
"status": f"保存失败: {str(e)}"
}), 500
@app.route('/check_connection')
def check_connection():
"""检查连接状态"""
return jsonify({
"status": "connected",
"message": "连接成功",
"server_ip": get_current_ip(),
"port": PORT
})
def main():
"""主函数"""
# 检查权限
if os.geteuid() != 0:
print("警告：请使用sudo运行此程序以获取必要的权限")
return
current_ip = get_current_ip()
print("=" * 60)
print("网络配置服务启动")
print("=" * 60)
print(f"当前服务器IP: {current_ip}")
print(f"服务端口: {PORT}")
print(f"访问地址: http://{current_ip}:{PORT}")
print("=" * 60)
print("保存配置后，会自动跳转到新的IP地址")
print("注意：请确保防火墙允许端口访问")
print("=" * 60)
# 检查防火墙设置
try:
result = subprocess.run(["ufw", "status"], capture_output=True, text=True)
if "Status: active" in result.stdout:
print("检测到UFW防火墙，请确保端口已开放:")
print(f" sudo ufw allow {PORT}")
print(f" sudo ufw allow 22 # SSH端口（可选）")
except:
pass
# 使用gevent WSGI服务器
server = pywsgi.WSGIServer(('0.0.0.0', PORT), app)
print(f"服务器已启动，监听端口: {PORT}")
print("按 Ctrl+C 停止服务")
try:
server.serve_forever()
except KeyboardInterrupt:
print("\n服务已停止")
except Exception as e:
print(f"服务器错误: {e}")
if __name__ == "__main__":
main()