默认适配 Ubuntu/Debian(sites-available/sites-enabled),同时兼容 CentOS/RHEL(conf.d,文件后缀 .disabled 控制启停)。
功能:新增/修改站点、启用/暂停、nginx -t 测试、reload、查看健康状态
技术:Flask + Jinja2 模板;元数据 JSON 记录站点表单字段,便于编辑回显
安全:支持管理口令(ADMIN_TOKEN),默认仅监听 127.0.0.1,建议内网或 SSH 隧道访问
权限:需 root 运行(或 sudo 精准授权)
一、快速开始
1) 安装依赖
Ubuntu/Debian
sudo apt update && sudo apt install -y python3-venv nginx
CentOS/RHEL
sudo dnf install -y python3 nginx
2) 放置项目(示例目录 /opt/nginx-admin)
mkdir -p /opt/nginx-admin && cd /opt/nginx-admin
把下面三个文件按路径保存:
app.py
templates/site.conf.j2
static/index.html
3) 创建虚拟环境并安装
python3 -m venv .venv
. .venv/bin/activate
pip install flask jinja2
4) 运行
可选安全口令:export ADMIN_TOKEN='请自己设置一串复杂口令'
启动:sudo .venv/bin/python app.py
若设置了 ADMIN_TOKEN,首次访问可在 URL 带 ?token=你的口令,或在页面顶部输入口令
二、后端 app.py说明
默认路径:Ubuntu/Debian 使用 /etc/nginx/sites-available 与 /etc/nginx/sites-enabled;CentOS/RHEL 使用 /etc/nginx/conf.d(本代码自动兼容)
想明确指定路径,可用环境变量覆写:
SITES_AVAILABLE, SITES_ENABLED
口令:环境变量 ADMIN_TOKEN;如果不设置,则不启用鉴权(仅限内网/隧道环境)
保存为 app.py
import os, shutil, subprocess, tempfile, json
from pathlib import Path
from functools import wraps
from flask import Flask, request, jsonify, send_from_directory
from jinja2 import Environment, FileSystemLoader
app = Flask(__name__, static_folder="static")
# 可用环境变量覆写
NGINX_BIN = shutil.which("nginx") or "/usr/sbin/nginx"
SYSTEMCTL = shutil.which("systemctl") or "/usr/bin/systemctl"
SITES_AVAILABLE = Path(os.getenv("SITES_AVAILABLE", "/etc/nginx/sites-available"))
SITES_ENABLED = Path(os.getenv("SITES_ENABLED", "/etc/nginx/sites-enabled"))
COMBINED_DIR = SITES_AVAILABLE.resolve() == SITES_ENABLED.resolve() # CentOS/RHEL 常为 True
CONF_EXT = ".conf"
DISABLED_EXT = ".disabled"
META_DIR = SITES_AVAILABLE / ".meta"
TEMPLATE_DIR = Path("templates")
env = Environment(loader=FileSystemLoader(str(TEMPLATE_DIR)))
tpl = env.get_template("site.conf.j2")
ADMIN_TOKEN = os.getenv("ADMIN_TOKEN") # 建议设置
def require_auth(fn):
@wraps(fn)
def wrapper(*args, **kwargs):
if not ADMIN_TOKEN:
return fn(*args, **kwargs)
token = (
(request.headers.get("Authorization") or "").replace("Bearer ", "", 1)
or request.headers.get("X-Admin-Token")
or request.args.get("token")
)
if token != ADMIN_TOKEN:
return jsonify({"ok": False, "msg": "Unauthorized"}), 401
return fn(*args, **kwargs)
return wrapper
def run(cmd):
p = subprocess.run(cmd, text=True, capture_output=True)
return p.returncode == 0, (p.stdout + "\n" + p.stderr).strip()
def atomic_write(path: Path, content: str):
path.parent.mkdir(parents=True, exist_ok=True)
with tempfile.NamedTemporaryFile("w", delete=False, dir=str(path.parent)) as tf:
tf.write(content)
tmp = tf.name
os.replace(tmp, path)
def nginx_test():
return run([NGINX_BIN, "-t"])
def nginx_reload():
ok, out = run([SYSTEMCTL, "reload", "nginx"])
if not ok:
ok2, out2 = run([NGINX_BIN, "-s", "reload"])
out = out + "\n" + out2
return ok2, out
return ok, out
def conf_available_path(name: str) -> Path:
return SITES_AVAILABLE / f"{name}{CONF_EXT}"
def conf_enabled_symlink(name: str) -> Path:
return SITES_ENABLED / f"{name}{CONF_EXT}"
def conf_disabled_path(name: str) -> Path:
return SITES_AVAILABLE / f"{name}{CONF_EXT}{DISABLED_EXT}"
def meta_path(name: str) -> Path:
return META_DIR / f"{name}.json"
def is_enabled(name: str) -> bool:
if COMBINED_DIR:
return conf_available_path(name).exists()
else:
return conf_enabled_symlink(name).exists()
def enable_site(name: str):
if COMBINED_DIR:
# CentOS/RHEL: 用 .disabled 后缀控制启停
disabled = conf_disabled_path(name)
active = conf_available_path(name)
if disabled.exists():
os.replace(disabled, active)
# 如果两者都不存在,不在此创建文件;写入时创建
else:
src = conf_available_path(name)
dst = conf_enabled_symlink(name)
if not dst.exists():
dst.symlink_to(src.resolve())
def disable_site(name: str):
if COMBINED_DIR:
active = conf_available_path(name)
disabled = conf_disabled_path(name)
if active.exists():
os.replace(active, disabled)
else:
dst = conf_enabled_symlink(name)
if dst.exists():
dst.unlink()
def render_conf(payload: dict) -> str:
allowed = {"server_name", "root", "proxy_pass", "enable_ssl", "ssl_cert", "ssl_key", "https_redirect"}
data = {k: v for k, v in (payload or {}).items() if k in allowed}
if not data.get("server_name"):
raise ValueError("server_name is required")
# root 与 proxy_pass 二选一,优先 proxy_pass
if data.get("proxy_pass") and data.get("root"):
data.pop("root", None)
return tpl.render(**data)
def save_meta(name: str, payload: dict):
META_DIR.mkdir(parents=True, exist_ok=True)
atomic_write(meta_path(name), json.dumps(payload, ensure_ascii=False, indent=2))
def load_meta(name: str) -> dict:
p = meta_path(name)
if p.exists():
return json.loads(p.read_text())
return {}
def create_or_update_site(payload: dict, enable=True):
server_name = payload.get("server_name")
if not server_name:
return False, "server_name is required"
# 记录当前启用状态
was_enabled = is_enabled(server_name)
# 渲染配置 & 选择写入路径
conf_text = render_conf(payload)
if COMBINED_DIR:
target_path = conf_available_path(server_name) if (enable or was_enabled) else conf_disabled_path(server_name)
else:
target_path = conf_available_path(server_name)
# 备份
backup_path = None
if target_path.exists():
backup_path = target_path.with_suffix(target_path.suffix + ".bak")
shutil.copy2(target_path, backup_path)
# 写入配置与元数据
atomic_write(target_path, conf_text)
save_meta(server_name, payload)
# 根据状态启用/保持状态
if enable or was_enabled:
try:
enable_site(server_name)
ok, out = nginx_test()
if not ok:
# 回滚
if backup_path and backup_path.exists():
shutil.copy2(backup_path, target_path)
disable_site(server_name)
return False, f"nginx -t failed:\n{out}"
ok2, out2 = nginx_reload()
if not ok2:
return False, f"nginx reload failed:\n{out2}"
return True, "applied and reloaded"
except Exception as e:
disable_site(server_name)
if backup_path and backup_path.exists():
shutil.copy2(backup_path, target_path)
return False, str(e)
else:
return True, "saved (not enabled)"
def list_all_sites():
items = {}
if COMBINED_DIR:
# 识别 .conf 和 .conf.disabled
for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}"):
items[p.stem] = True
for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}{DISABLED_EXT}"):
name = p.name.replace(CONF_EXT + DISABLED_EXT, "")
items[name] = items.get(name, False) and True or False
else:
for p in SITES_AVAILABLE.glob(f"*{CONF_EXT}"):
name = p.stem
items[name] = conf_enabled_symlink(name).exists()
return [{"server_name": k, "enabled": v} for k, v in sorted(items.items(), key=lambda x: x[0])]
# ---------- Routes ----------
@app.get("/")
def index():
return send_from_directory(app.static_folder, "index.html")
@app.get("/health")
@require_auth
def health():
ok, out = nginx_test()
return jsonify({"nginx_ok": ok, "output": out})
@app.post("/nginx/reload")
@require_auth
def reload_api():
ok, out = nginx_reload()
return (jsonify({"ok": ok, "msg": out}), 200 if ok else 400)
@app.get("/sites")
@require_auth
def sites_list():
return jsonify(list_all_sites())
@app.get("/sites/<server_name>")
@require_auth
def site_get(server_name):
enabled = is_enabled(server_name)
meta = load_meta(server_name)
return jsonify({"server_name": server_name, "enabled": enabled, "meta": meta})
@app.post("/sites")
@require_auth
def site_create():
ok, msg = create_or_update_site(request.json or {}, enable=True)
return (jsonify({"ok": ok, "msg": msg}), 200 if ok else 400)
@app.put("/sites/<server_name>")
@require_auth
def site_update(server_name):
data = request.json or {}
data["server_name"] = server_name
ok, msg = create_or_update_site(data, enable=False)
return (jsonify({"ok": ok, "msg": msg}), 200 if ok else 400)
@app.post("/sites/<server_name>/enable")
@require_auth
def site_enable(server_name):
enable_site(server_name)
ok, out = nginx_test()
if not ok:
disable_site(server_name)
return jsonify({"ok": False, "msg": out}), 400
ok2, out2 = nginx_reload()
return (jsonify({"ok": ok2, "msg": out2}), 200 if ok2 else 400)
@app.post("/sites/<server_name>/disable")
@require_auth
def site_disable(server_name):
disable_site(server_name)
ok, out = nginx_test()
if not ok:
return jsonify({"ok": False, "msg": out}), 400
ok2, out2 = nginx_reload()
return (jsonify({"ok": ok2, "msg": out2}), 200 if ok2 else 400)
if __name__ == "__main__":
# 默认仅监听本机,避免暴露在公网;需要外网可改为 0.0.0.0 并加防护
app.run(host="127.0.0.1", port=8080) 三、Jinja2 配置模板保存为 templates/site.conf.j2
{% set use_proxy = proxy_pass is defined and proxy_pass %}
{% set use_ssl = enable_ssl|default(false) %}
# HTTP
server {
listen 80;
server_name {{ server_name }};
{% if https_redirect and use_ssl %}
return 301 https://$host$request_uri;
{% else %}
{% if use_proxy %}
location / {
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_pass {{ proxy_pass }};
}
{% else %}
root {{ root }};
index index.html index.htm;
location / {
try_files $uri $uri/ =404;
}
{% endif %}
{% endif %}
}
{% if use_ssl %}
server {
listen 443 ssl http2;
server_name {{ server_name }};
ssl_certificate {{ ssl_cert }};
ssl_certificate_key {{ ssl_key }};
{% if use_proxy %}
location / {
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_pass {{ proxy_pass }};
}
{% else %}
root {{ root }};
index index.html index.htm;
location / {
try_files $uri $uri/ =404;
}
{% endif %}
}
{% endif %} 四、前端页面保存为 static/index.html
<!doctype html>
<html lang="zh-cn">
<head>
<meta charset="utf-8" />
<title>Nginx 可视化管理</title>
<meta name="viewport" content="width=device-width, initial-scale=1" />
<style>
body { font-family: system-ui, -apple-system, Segoe UI, Roboto, Helvetica, Arial, "Apple Color Emoji", "Segoe UI Emoji"; margin: 20px; }
.row { display: flex; gap: 24px; flex-wrap: wrap; }
.col { flex: 1 1 380px; min-width: 320px; }
table { width: 100%; border-collapse: collapse; margin-top: 8px; }
th, td { border: 1px solid #eee; padding: 8px; text-align: left; }
th { background: #fafafa; }
.muted { color: #666; }
.ok { color: #1677ff; }
.bad { color: #d4380d; }
input[type=text] , input[type=url] { width: 100%; padding: 8px; box-sizing: border-box; }
label { display: block; margin: 8px 0 4px; }
.btn { padding: 6px 12px; margin-right: 8px; border: 1px solid #ddd; background: #fff; cursor: pointer; border-radius: 4px; }
.btn.primary { background: #1677ff; color: #fff; border-color: #1677ff; }
.btn.danger { background: #ff4d4f; color: #fff; border-color: #ff4d4f; }
.btn:disabled { opacity: 0.6; cursor: not-allowed; }
.tag { display: inline-block; padding: 2px 8px; background: #f0f5ff; color: #1d39c4; border-radius: 999px; font-size: 12px; }
.card { border: 1px solid #eee; border-radius: 8px; padding: 12px 16px; }
.help { font-size: 12px; color: #999; }
.hr { height: 1px; background: #eee; margin: 12px 0; }
</style>
</head>
<body>
<h2>Nginx 可视化管理</h2>
<div class="help">提示:默认仅在本机监听;生产环境请务必设置 ADMIN_TOKEN,并通过内网或隧道访问。</div>
<div style="margin: 8px 0;">
<label>管理口令(可留空,若后端设置了 ADMIN_TOKEN 则必填):</label>
<input id="token" type="text" placeholder="Bearer Token" />
</div>
<div class="row">
<div class="col card">
<div style="display:flex; align-items:center; justify-content:space-between;">
<h3>站点列表</h3>
<div>
<button class="btn" onclick="testNginx()">测试 nginx</button>
<button class="btn" onclick="reloadNginx()">重载 nginx</button>
<button class="btn" onclick="loadSites()">刷新列表</button>
</div>
</div>
<div id="health" class="help">状态未知</div>
<table>
<thead>
<tr><th>域名</th><th>状态</th><th>操作</th></tr>
</thead>
<tbody id="siteRows"></tbody>
</table>
</div>
<div class="col card">
<h3 id="formTitle">新增 / 编辑站点</h3>
<div class="help">修改现有站点时请先在左侧点击站点加载其配置</div>
<label>域名(server_name)</label>
<input id="server_name" type="text" placeholder="example.com 或 foo.example.com" />
<div class="hr"></div>
<div>
<label>站点类型</label>
<label><input type="radio" name="mode" value="static" checked> 静态目录</label>
<label><input type="radio" name="mode" value="proxy"> 反向代理</label>
</div>
<div id="staticFields">
<label>根目录(root)</label>
<input id="root" type="text" placeholder="/var/www/example" />
</div>
<div id="proxyFields" style="display:none;">
<label>代理地址(proxy_pass)</label>
<input id="proxy_pass" type="url" placeholder="http://127.0.0.1:3000" />
</div>
<div class="hr"></div>
<div>
<label><input id="enable_ssl" type="checkbox"> 启用 SSL(已准备好证书时勾选)</label>
<div id="sslFields" style="display:none;">
<label>证书路径(ssl_certificate)</label>
<input id="ssl_cert" type="text" placeholder="/etc/letsencrypt/live/example.com/fullchain.pem" />
<label>私钥路径(ssl_certificate_key)</label>
<input id="ssl_key" type="text" placeholder="/etc/letsencrypt/live/example.com/privkey.pem" />
<label><input id="https_redirect" type="checkbox" checked> 80 强制跳转到 443</label>
</div>
</div>
<div class="hr"></div>
<div>
<button class="btn primary" onclick="submitForm()">保存/应用</button>
<button class="btn" onclick="resetForm()">清空表单</button>
<span id="formMsg" class="help"></span>
</div>
</div>
</div>
<script>
let current = null;
function getToken() {
const urlToken = new URLSearchParams(location.search).get("token");
if (urlToken) {
localStorage.setItem("token", urlToken);
history.replaceState({}, "", location.pathname);
}
const saved = localStorage.getItem("token") || "";
const input = document.getElementById("token");
if (!input.value && saved) input.value = saved;
return input.value.trim();
}
function api(path, options={}) {
const token = getToken();
const headers = options.headers || {};
if (token) headers["Authorization"] = "Bearer " + token;
headers["Content-Type"] = "application/json";
return fetch(path, {...options, headers});
}
async function loadSites() {
document.getElementById("siteRows").innerHTML = "<tr><td colspan='3' class='muted'>加载中...</td></tr>";
try {
const res = await api("/sites");
if (!res.ok) throw new Error(await res.text());
const data = await res.json();
const rows = data.map(it => {
const status = it.enabled ? "<span class='tag'>已启用</span>" : "<span class='muted'>已暂停</span>";
const btn = it.enabled
? `<button class='btn' onclick="toggleSite('${it.server_name}', false)">暂停</button>`
: `<button class='btn' onclick="toggleSite('${it.server_name}', true)">启用</button>`;
const edit = `<button class='btn' onclick="editSite('${it.server_name}')">编辑</button>`;
return `<tr><td>${it.server_name}</td><td>${status}</td><td>${edit} ${btn}</td></tr>`;
}).join("") || "<tr><td colspan='3' class='muted'>暂无站点</td></tr>";
document.getElementById("siteRows").innerHTML = rows;
} catch (e) {
document.getElementById("siteRows").innerHTML = "<tr><td colspan='3' class='bad'>加载失败</td></tr>";
}
}
async function editSite(name) {
try {
const res = await api(`/sites/${encodeURIComponent(name)}`);
if (!res.ok) throw new Error(await res.text());
const data = await res.json();
current = data.server_name;
document.getElementById("formTitle").innerText = `编辑:${current}`;
fillForm(data.meta || {});
} catch (e) {
alert("加载失败:" + e.message);
}
}
function fillForm(meta) {
document.getElementById("server_name").value = meta.server_name || current || "";
const mode = meta.proxy_pass ? "proxy" : "static";
document.querySelector(`input[name="mode"][value="${mode}"]`).checked = true;
document.getElementById("root").value = meta.root || "";
document.getElementById("proxy_pass").value = meta.proxy_pass || "";
document.getElementById("enable_ssl").checked = !!meta.enable_ssl;
document.getElementById("ssl_cert").value = meta.ssl_cert || "";
document.getElementById("ssl_key").value = meta.ssl_key || "";
document.getElementById("https_redirect").checked = !!meta.https_redirect;
toggleFields();
}
function collectForm() {
const server_name = document.getElementById("server_name").value.trim();
const mode = document.querySelector('input[name="mode"]:checked').value;
const root = document.getElementById("root").value.trim();
const proxy_pass = document.getElementById("proxy_pass").value.trim();
const enable_ssl = document.getElementById("enable_ssl").checked;
const ssl_cert = document.getElementById("ssl_cert").value.trim();
const ssl_key = document.getElementById("ssl_key").value.trim();
const https_redirect = document.getElementById("https_redirect").checked;
const meta = { server_name, enable_ssl, https_redirect };
if (mode === "proxy") meta.proxy_pass = proxy_pass;
else meta.root = root;
if (enable_ssl) { meta.ssl_cert = ssl_cert; meta.ssl_key = ssl_key; }
return meta;
}
async function submitForm() {
const meta = collectForm();
const msg = document.getElementById("formMsg");
msg.innerText = "提交中...";
try {
if (!meta.server_name) throw new Error("server_name 不能为空");
let res;
// 如果当前在编辑同名站点,则更新;否则创建
if (current && current === meta.server_name) {
res = await api(`/sites/${encodeURIComponent(current)}`, { method: "PUT", body: JSON.stringify(meta) });
} else {
res = await api("/sites", { method: "POST", body: JSON.stringify(meta) });
}
const data = await res.json();
if (!res.ok || !data.ok) throw new Error(data.msg || "未知错误");
msg.innerText = "成功:" + data.msg;
loadSites();
current = meta.server_name;
document.getElementById("formTitle").innerText = `编辑:${current}`;
} catch (e) {
msg.innerText = "失败:" + e.message;
}
}
async function toggleSite(name, enable) {
const path = enable ? `/sites/${encodeURIComponent(name)}/enable` : `/sites/${encodeURIComponent(name)}/disable`;
const res = await api(path, { method: "POST" });
const data = await res.json().catch(()=>({}));
if (!res.ok || !data.ok) {
alert("操作失败:" + (data.msg || res.statusText));
}
loadSites();
}
async function testNginx() {
const el = document.getElementById("health");
el.innerText = "测试中...";
const res = await api("/health");
const data = await res.json().catch(()=>({}));
if (res.ok) {
el.innerHTML = data.nginx_ok ? `<span class='ok'>nginx -t 通过</span>` : `<span class='bad'>nginx -t 失败</span>`;
el.innerHTML += "<div class='help'><pre style='white-space:pre-wrap'>" + (data.output || "") + "</pre></div>";
} else {
el.innerHTML = `<span class='bad'>健康检查失败</span>`;
}
}
async function reloadNginx() {
const res = await api("/nginx/reload", { method: "POST" });
const data = await res.json().catch(()=>({}));
if (!res.ok || !data.ok) alert("重载失败:" + (data.msg || res.statusText));
else alert("已重载");
}
function toggleFields() {
const mode = document.querySelector('input[name="mode"]:checked').value;
document.getElementById("staticFields").style.display = mode === "static" ? "" : "none";
document.getElementById("proxyFields").style.display = mode === "proxy" ? "" : "none";
const ssl = document.getElementById("enable_ssl").checked;
document.getElementById("sslFields").style.display = ssl ? "" : "none";
}
function resetForm() {
current = null;
document.getElementById("formTitle").innerText = "新增 / 编辑站点";
document.getElementById("server_name").value = "";
document.querySelector('input[name="mode"][value="static"]').checked = true;
document.getElementById("root").value = "";
document.getElementById("proxy_pass").value = "";
document.getElementById("enable_ssl").checked = false;
document.getElementById("ssl_cert").value = "";
document.getElementById("ssl_key").value = "";
document.getElementById("https_redirect").checked = true;
toggleFields();
document.getElementById("formMsg").innerText = "";
}
document.querySelectorAll('input[name="mode"]').forEach(el => el.addEventListener("change", toggleFields));
document.getElementById("enable_ssl").addEventListener("change", toggleFields);
window.addEventListener("DOMContentLoaded", () => {
getToken();
toggleFields();
loadSites();
testNginx();
});
</script>
</body>
</html> 五、systemd 托管(可选)1) sudo vim /etc/systemd/system/nginx-admin.service
[Unit] Description=Simple Nginx Admin Panel (Flask) After=network.target [Service] WorkingDirectory=/opt/nginx-admin Environment=ADMIN_TOKEN=请设置复杂口令 ExecStart=/opt/nginx-admin/.venv/bin/python app.py User=root Restart=on-failure RestartSec=3 # 如需非 root 运行,请配合 sudoers 授权 nginx -t 与 reload [Install] WantedBy=multi-user.target
2) 启动
sudo systemctl daemon-reload
sudo systemctl enable --now nginx-admin
查看日志:journalctl -u nginx-admin -f
六、CentOS/RHEL 适配说明
常见路径为 /etc/nginx/conf.d,并不区分 sites-available/sites-enabled
本代码自动识别“同一目录”模式:用 conf 文件 + .disabled 后缀控制启停
启用:example.conf
暂停:example.conf.disabled
如需手动指定:
export SITES_AVAILABLE=/etc/nginx/conf.d
export SITES_ENABLED=/etc/nginx/conf.d
七、常见问题
权限问题:必须能写入配置目录,并具备 nginx -t 与 reload 权限。最简单是以 root 运行;更安全是用 sudoers 精准授权。
安全建议:
设置 ADMIN_TOKEN,禁止公网直接访问该管理端口
默认仅监听 127.0.0.1,建议通过 SSH 隧道访问:ssh -L 8080:127.0.0.1:8080 your-host
证书:
建站后可用 certbot 或 acme.sh 签发证书,再勾选“启用 SSL”并填入证书路径
示例(webroot 模式):certbot certonly --webroot -w /var/www/example -d example.com
需要我帮你把这套脚本按你的环境定制吗?回我这些信息:
你的系统与版本(Ubuntu/CentOS/AlmaLinux 等)
Nginx 配置惯例(sites-available/sites-enabled 还是 conf.d)
主要站点类型(静态/反代/混合),是否要内置 Let’s Encrypt 一键签证
网友回复


