AI-Agent/dns-serrvice
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity
Files
0a18b94d84d6b0242ebbbbe54edaf0f14cb6c4f2
dns-serrvice/app.py
T
AI Agent 0a18b94d84 feat: BIND DNS Web Manager - Web interface for managing BIND9 DNS server 
- Flask + SQLite web app with user authentication
- Zone management: create/delete zones, edit raw zone files
- Record management: add/delete A/AAAA/CNAME/MX/TXT/NS/PTR/SRV/CAA records
- Service control: start/stop/restart/reload BIND via systemctl/rndc
- Configuration editor: edit named.conf.options/local with syntax validation
- DNS query testing: online dig tool
- Audit log: all operations logged with user/timestamp
- BIND9 backend, listening on port 53
2026-07-09 17:24:44 +08:00

1076 lines
35 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
BIND DNS Web Manager - Web interface for managing BIND9 DNS server
Features: zone management, record CRUD, service control, config editing, query testing
"""
import os
import re
import json
import subprocess
import shlex
import ipaddress
from datetime import datetime, timezone
from functools import wraps
from flask import (
Flask, render_template, redirect, url_for, request, session,
flash, jsonify, abort, Response
)
from flask_sqlalchemy import SQLAlchemy
from werkzeug.security import generate_password_hash, check_password_hash
app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ.get("DNS_WEB_SECRET", "dns-web-mgmt-secret-key-2026")
app.config["SQLALCHEMY_DATABASE_URI"] = f"sqlite:///{os.path.join(app.instance_path, 'dns_web.db')}"
app.config["SQLALCHEMY_TRACK_MODIFICATIONS"] = False
db = SQLAlchemy(app)
# ─── BIND paths ───
BIND_CONF_DIR = "/etc/bind"
BIND_ZONES_DIR = "/etc/bind/zones"
BIND_CONF_LOCAL = "/etc/bind/named.conf.local"
BIND_CONF_OPTIONS = "/etc/bind/named.conf.options"
BIND_RNDC_KEY = "/etc/bind/rndc.key"
BIND_SERVICE = "named"
# ─── Record types ───
RECORD_TYPES = ["A", "AAAA", "CNAME", "MX", "TXT", "NS", "PTR", "SRV", "CAA", "SOA"]
RECORD_FIELDS = {
"A": ["ttl", "ip"],
"AAAA": ["ttl", "ip"],
"CNAME": ["ttl", "target"],
"MX": ["ttl", "priority", "target"],
"TXT": ["ttl", "text"],
"NS": ["ttl", "target"],
"PTR": ["ttl", "target"],
"SRV": ["ttl", "priority", "weight", "port", "target"],
"CAA": ["ttl", "flags", "tag", "value"],
}
# ════════════════════════════════════════════════════════════════════
# Models
# ════════════════════════════════════════════════════════════════════
class User(db.Model):
__tablename__ = "users"
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
password_hash = db.Column(db.String(200), nullable=False)
role = db.Column(db.String(20), default="admin")
created_at = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
def set_password(self, pw):
self.password_hash = generate_password_hash(pw)
def check_password(self, pw):
return check_password_hash(self.password_hash, pw)
class AuditLog(db.Model):
__tablename__ = "audit_logs"
id = db.Column(db.Integer, primary_key=True)
user = db.Column(db.String(80))
action = db.Column(db.String(200), nullable=False)
detail = db.Column(db.Text)
timestamp = db.Column(db.DateTime, default=lambda: datetime.now(timezone.utc))
@staticmethod
def log(user, action, detail=""):
db.session.add(AuditLog(user=user, action=action, detail=detail))
db.session.commit()
# ════════════════════════════════════════════════════════════════════
# Auth
# ════════════════════════════════════════════════════════════════════
def current_user():
uid = session.get("user_id")
if uid is None:
return None
return db.session.get(User, uid)
def login_required(f):
@wraps(f)
def wrapper(*a, **kw):
if session.get("user_id") is None:
return redirect(url_for("login", next=request.path))
return f(*a, **kw)
return wrapper
@app.context_processor
def inject_globals():
return dict(current_user=current_user())
# ════════════════════════════════════════════════════════════════════
# BIND helpers
# ════════════════════════════════════════════════════════════════════
def run_cmd(cmd, timeout=10):
"""Run shell command, return (returncode, stdout, stderr)."""
try:
r = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
return r.returncode, r.stdout.strip(), r.stderr.strip()
except subprocess.TimeoutExpired:
return -1, "", "Command timed out"
except Exception as e:
return -1, "", str(e)
def bind_service_status():
rc, out, _ = run_cmd("systemctl is-active named 2>&1")
enabled_rc, _, _ = run_cmd("systemctl is-enabled named 2>&1")
return {
"active": out.strip() == "active",
"status": out.strip(),
"enabled": enabled_rc == 0 and "enabled" in run_cmd("systemctl is-enabled named 2>&1")[1],
}
def bind_reload():
"""Reload BIND zones via rndc."""
rc, out, err = run_cmd("rndc reload 2>&1")
return rc == 0, out + err
def bind_version():
rc, out, _ = run_cmd("named -v 2>&1")
return out.strip()
def parse_named_conf_local():
"""Parse named.conf.local, return list of zone dicts."""
zones = []
try:
with open(BIND_CONF_LOCAL) as f:
content = f.read()
except FileNotFoundError:
return zones
# Match: zone "example.com" { type master; file "/etc/bind/zones/db.example.com"; };
pattern = re.compile(
r'zone\s+"([^"]+)"\s*\{\s*'
r'type\s+(\w+)\s*;\s*'
r'file\s+"([^"]+)"\s*;\s*'
r'(?:allow-update\s*\{[^}]*\}\s*;\s*)?'
r'\}\s*;',
re.MULTILINE
)
for m in pattern.finditer(content):
zone_name = m.group(1)
zone_type = m.group(2)
zone_file = m.group(3)
zones.append({
"name": zone_name,
"type": zone_type,
"file": zone_file,
"filename": os.path.basename(zone_file),
})
return zones
def write_named_conf_local(zones):
"""Rewrite named.conf.local from zone list."""
lines = ['// Local zone configurations - managed by DNS Web UI\n']
for z in zones:
lines.append(f'\nzone "{z["name"]}" {{\n')
lines.append(f' type {z["type"]};\n')
lines.append(f' file "{z["file"]}";\n')
lines.append(' allow-update { none; };\n')
lines.append('};\n')
content = ''.join(lines)
with open(BIND_CONF_LOCAL, 'w') as f:
f.write(content)
# Ensure permissions
run_cmd(f"chown root:bind {BIND_CONF_LOCAL}")
run_cmd(f"chmod 644 {BIND_CONF_LOCAL}")
def parse_zone_file(filepath):
"""Parse a BIND zone file, return SOA + records list.
Properly handles multi-line SOA with parentheses, inline comments,
$TTL/$ORIGIN directives, and blank-name continuation records.
"""
records = []
soa = {}
try:
with open(filepath) as f:
raw_lines = f.readlines()
except FileNotFoundError:
return soa, records
# ── Step 1: Normalise: strip comments, join multi-line SOA ──
# Remove inline comments (everything after unquoted ;) and track lines
clean_lines = []
in_parens = False
merged = ""
for line in raw_lines:
# Strip inline comments (simple approach - no quoted ; in our zone files)
if ';' in line:
# Check if ; is inside quotes
in_quote = False
cut_pos = len(line)
for i, ch in enumerate(line):
if ch == '"':
in_quote = not in_quote
elif ch == ';' and not in_quote:
cut_pos = i
break
line = line[:cut_pos]
stripped = line.strip()
if in_parens:
merged += " " + stripped
if ')' in stripped:
in_parens = False
clean_lines.append(merged.replace('(', ' ').replace(')', ' ').strip())
merged = ""
continue
if '(' in stripped and ')' not in stripped:
in_parens = True
merged = stripped.replace('(', ' ')
continue
if stripped:
clean_lines.append(stripped)
# ── Step 2: Parse normalised lines ──
origin = ""
default_ttl = "86400"
prev_name = ""
for line in clean_lines:
parts = line.split()
if parts[0] == '$TTL':
default_ttl = parts[1] if len(parts) > 1 else default_ttl
continue
if parts[0] == '$ORIGIN':
origin = parts[1] if len(parts) > 1 else origin
continue
# SOA record
if 'SOA' in parts:
soa_idx = parts.index('SOA')
soa_name = parts[0] if parts[0] != '@' else (origin.rstrip('.') if origin else '@')
soa["name"] = soa_name
# SOA data: ns email serial refresh retry expire minimum
soa_data = parts[soa_idx + 1:]
soa["raw"] = ' '.join(soa_data)
if len(soa_data) >= 7:
soa["ns"] = soa_data[0]
soa["email"] = soa_data[1]
soa["serial"] = soa_data[2]
soa["refresh"] = soa_data[3]
soa["retry"] = soa_data[4]
soa["expire"] = soa_data[5]
soa["minimum"] = soa_data[6]
continue
# Skip blank-only lines (shouldn't happen after normalisation)
if not parts:
continue
# Regular record
name = parts[0]
if name == '@':
name = origin.rstrip('.') if origin else '@'
elif name == '':
name = prev_name
else:
prev_name = name
idx = 1
ttl = ""
if idx < len(parts) and parts[idx].isdigit():
ttl = parts[idx]
idx += 1
# Class
if idx < len(parts) and parts[idx] in ('IN', 'CH', 'HS'):
idx += 1
if idx >= len(parts):
continue
rtype = parts[idx].upper()
idx += 1
rdata = ' '.join(parts[idx:])
records.append({
"name": name,
"ttl": ttl,
"type": rtype,
"data": rdata.strip(),
})
return soa, records
def generate_zone_file(zone_name, records, soa=None):
"""Generate zone file content from records list."""
now = datetime.now().strftime("%Y%m%d01")
if not soa:
soa = {
"ns": f"ns1.{zone_name}.",
"email": f"admin.{zone_name}.",
"serial": now,
"refresh": "3600",
"retry": "900",
"expire": "604800",
"minimum": "86400",
}
lines = [
f"$TTL {soa.get('minimum', '86400')}\n",
f'@ IN SOA {soa["ns"]} {soa["email"]} (\n',
f' {soa["serial"]} ; Serial\n',
f' {soa["refresh"]} ; Refresh\n',
f' {soa["retry"]} ; Retry\n',
f' {soa["expire"]} ; Expire\n',
f' {soa["minimum"]} ; Minimum TTL\n',
f' )\n',
]
# Add NS records for the zone itself
has_ns = any(r["type"] == "NS" for r in records)
if not has_ns:
lines.append(f'@ IN NS ns1.{zone_name}.\n')
lines.append(f'ns1 IN A 127.0.0.1\n')
for r in records:
name = r["name"]
ttl = f' {r.get("ttl", "")}' if r.get("ttl") else ''
rtype = r["type"]
data = r["data"]
lines.append(f'{name}{ttl} IN {rtype:<6} {data}\n')
return ''.join(lines)
def append_record_to_file(filepath, name, ttl, rtype, data):
"""Append a new record to the zone file and bump the serial.
Returns (success, message).
"""
try:
with open(filepath) as f:
content = f.read()
except FileNotFoundError:
return False, "Zone file not found"
# Build the new record line
ttl_str = f'{ttl} ' if ttl else ''
new_line = f'{name} {ttl_str}IN {rtype:<6} {data}\n'
# Append record at end of file
content = content.rstrip('\n') + '\n' + new_line
# Update serial
content = bump_serial_in_content(content)
with open(filepath, 'w') as f:
f.write(content)
run_cmd(f"chown bind:bind {filepath}")
return True, "OK"
def delete_record_from_file(filepath, zone_name, idx):
"""Delete the Nth record (0-indexed, excluding SOA) from the zone file.
Returns (success, deleted_record_str, message).
"""
soa, records = parse_zone_file(filepath)
if idx < 0 or idx >= len(records):
return False, "", "Record index out of range"
deleted = records[idx]
# Read raw lines and find the matching record line to remove
with open(filepath) as f:
raw_lines = f.readlines()
# We need to find the Nth non-SOA, non-directive, non-comment, non-empty line
# and remove it. This is tricky because the record might span lines.
# Strategy: find the line that contains the record's name and type.
# Build a search pattern for the record
target_name = deleted["name"]
target_type = deleted["type"]
target_data = deleted["data"]
# Find and remove the matching line
new_lines = []
record_idx = 0
removed = False
in_soa_block = False # Track multi-line SOA inside parentheses
for line in raw_lines:
stripped = line.strip()
if not stripped or stripped.startswith(';') or stripped.startswith('$'):
new_lines.append(line)
continue
# Track SOA multi-line block (lines between ( and ))
if 'SOA' in stripped.split():
new_lines.append(line)
if '(' in stripped and ')' not in stripped:
in_soa_block = True
continue
# Skip lines inside SOA block
if in_soa_block:
new_lines.append(line)
if ')' in stripped:
in_soa_block = False
continue
# This is a record line
if not removed and record_idx == idx:
# Skip this line (don't append)
removed = True
continue
record_idx += 1
new_lines.append(line)
if not removed:
return False, "", "Could not find record to delete"
content = ''.join(new_lines)
content = bump_serial_in_content(content)
with open(filepath, 'w') as f:
f.write(content)
run_cmd(f"chown bind:bind {filepath}")
return True, f'{deleted["name"]} {deleted["type"]} {deleted["data"]}', "OK"
def bump_serial_in_content(content):
"""Find and increment the SOA serial in zone file content."""
# Match serial number in SOA - it's the number after the email in SOA
# Pattern: SOA ns email serial ...
lines = content.split('\n')
in_soa = False
for i, line in enumerate(lines):
if 'SOA' in line:
in_soa = True
continue
if in_soa:
# Find the serial number (first number in SOA block after ns/email)
stripped = line.strip()
# Remove comments
if ';' in stripped:
stripped = stripped[:stripped.index(';')].strip()
if stripped:
parts = stripped.split()
if parts and parts[0].isdigit():
new_serial = serial_update(parts[0])
lines[i] = line.replace(parts[0], new_serial, 1)
break
return '\n'.join(lines)
def serial_update(serial_str):
"""Increment serial number."""
try:
s = int(serial_str)
today = int(datetime.now().strftime("%Y%m%d"))
counter = s % 100
base = s // 100
if base == today:
s = today * 100 + counter + 1
else:
s = today * 100 + 1
return str(s)
except ValueError:
return datetime.now().strftime("%Y%m%d01")
def validate_record(rtype, data):
"""Validate record data, return (bool, error_msg)."""
parts = data.split()
try:
if rtype in ("A", "AAAA"):
ip = parts[0]
ipaddress.ip_address(ip)
if rtype == "A" and ":" in ip:
return False, "A record must use IPv4"
if rtype == "AAAA" and ":" not in ip:
return False, "AAAA record must use IPv6"
elif rtype == "CNAME":
if not parts[0].endswith('.'):
return False, "CNAME target must be FQDN (end with .)"
elif rtype == "MX":
if len(parts) < 2:
return False, "MX needs priority and target"
int(parts[0])
if not parts[1].endswith('.'):
return False, "MX target must be FQDN (end with .)"
elif rtype == "NS":
if not parts[0].endswith('.'):
return False, "NS target must be FQDN (end with .)"
elif rtype == "PTR":
if not parts[0].endswith('.'):
return False, "PTR target must be FQDN (end with .)"
elif rtype == "SRV":
if len(parts) < 4:
return False, "SRV needs priority weight port target"
elif rtype == "TXT":
if not data:
return False, "TXT data cannot be empty"
except ValueError as e:
return False, f"Invalid data: {e}"
return True, ""
# ════════════════════════════════════════════════════════════════════
# Routes
# ════════════════════════════════════════════════════════════════════
@app.route("/")
@login_required
def dashboard():
status = bind_service_status()
zones = parse_named_conf_local()
zone_count = len(zones)
total_records = 0
for z in zones:
try:
_, recs = parse_zone_file(z["file"])
z["record_count"] = len(recs)
total_records += len(recs)
except Exception:
z["record_count"] = 0
recent_logs = AuditLog.query.order_by(AuditLog.timestamp.desc()).limit(10).all()
version = bind_version()
return render_template("dashboard.html",
status=status, zones=zones, zone_count=zone_count,
total_records=total_records, recent_logs=recent_logs,
version=version)
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
username = request.form.get("username", "")
password = request.form.get("password", "")
user = User.query.filter_by(username=username).first()
if user and user.check_password(password):
session["user_id"] = user.id
AuditLog.log(username, "登录")
nxt = request.args.get("next", "/")
return redirect(nxt)
flash("用户名或密码错误", "error")
return render_template("login.html")
@app.route("/logout")
def logout():
u = current_user()
if u:
AuditLog.log(u.username, "登出")
session.pop("user_id", None)
return redirect(url_for("login"))
@app.route("/change-password", methods=["GET", "POST"])
@login_required
def change_password():
u = current_user()
if request.method == "POST":
old = request.form.get("old_password", "")
new = request.form.get("new_password", "")
confirm = request.form.get("confirm_password", "")
if not u.check_password(old):
flash("旧密码错误", "error")
elif len(new) < 6:
flash("新密码至少6位", "error")
elif new != confirm:
flash("两次密码不一致", "error")
else:
u.set_password(new)
db.session.commit()
AuditLog.log(u.username, "修改密码")
flash("密码修改成功", "success")
return redirect(url_for("dashboard"))
return render_template("change_password.html")
# ─── Zone management ───
@app.route("/zones")
@login_required
def zone_list():
zones = parse_named_conf_local()
for z in zones:
try:
_, recs = parse_zone_file(z["file"])
z["record_count"] = len(recs)
except Exception:
z["record_count"] = 0
return render_template("zones.html", zones=zones)
@app.route("/zones/create", methods=["GET", "POST"])
@login_required
def zone_create():
if request.method == "POST":
zone_name = request.form.get("zone_name", "").strip().rstrip('.')
zone_type = request.form.get("zone_type", "master")
ns_server = request.form.get("ns_server", "").strip()
admin_email = request.form.get("admin_email", "").strip()
if not zone_name:
flash("请输入域名", "error")
return render_template("zone_form.html")
# Validate zone name
if not re.match(r'^[a-zA-Z0-9][a-zA-Z0-9.-]*$', zone_name):
flash("域名格式无效", "error")
return render_template("zone_form.html")
zone_file = os.path.join(BIND_ZONES_DIR, f"db.{zone_name}")
# Check duplicate
existing = parse_named_conf_local()
if any(z["name"] == zone_name for z in existing):
flash(f"域名 {zone_name} 已存在", "error")
return render_template("zone_form.html")
# Generate initial zone file
now = datetime.now().strftime("%Y%m%d01")
ns = ns_server if ns_server else f"ns1.{zone_name}."
email = admin_email.replace('@', '.') if admin_email else f"admin.{zone_name}."
if not ns.endswith('.'):
ns += '.'
if not email.endswith('.'):
email += '.'
zone_content = f"""$TTL 86400
@ IN SOA {ns} {email} (
{now} ; Serial
3600 ; Refresh
900 ; Retry
604800 ; Expire
86400 ; Minimum TTL
)
@ IN NS {ns}
{ns.split('.')[0]} IN A 127.0.0.1
"""
# Write zone file
with open(zone_file, 'w') as f:
f.write(zone_content)
run_cmd(f"chown bind:bind {zone_file}")
run_cmd(f"chmod 644 {zone_file}")
# Validate zone
rc, out, err = run_cmd(f"named-checkzone {zone_name} {zone_file}")
if rc != 0:
os.remove(zone_file)
flash(f"Zone 文件验证失败: {out} {err}", "error")
return render_template("zone_form.html")
# Add to named.conf.local
existing.append({"name": zone_name, "type": zone_type, "file": zone_file})
write_named_conf_local(existing)
# Reload BIND
bind_reload()
u = current_user()
AuditLog.log(u.username, "创建Zone", f"域名: {zone_name}, 类型: {zone_type}")
flash(f"域名 {zone_name} 创建成功", "success")
return redirect(url_for("zone_detail", zone_name=zone_name))
return render_template("zone_form.html")
@app.route("/zones/<zone_name>")
@login_required
def zone_detail(zone_name):
zone_name = zone_name.rstrip('.')
zones = parse_named_conf_local()
zone = next((z for z in zones if z["name"] == zone_name), None)
if not zone:
abort(404)
soa, records = parse_zone_file(zone["file"])
# Read raw file
try:
with open(zone["file"]) as f:
raw_content = f.read()
except Exception:
raw_content = ""
return render_template("zone_detail.html", zone=zone, records=records,
soa=soa, raw_content=raw_content,
record_types=RECORD_TYPES, record_fields=RECORD_FIELDS)
@app.route("/zones/<zone_name>/delete", methods=["POST"])
@login_required
def zone_delete(zone_name):
zone_name = zone_name.rstrip('.')
zones = parse_named_conf_local()
zone = next((z for z in zones if z["name"] == zone_name), None)
if not zone:
flash("域名不存在", "error")
return redirect(url_for("zone_list"))
# Delete zone file
if os.path.exists(zone["file"]):
os.remove(zone["file"])
# Remove from named.conf.local
zones = [z for z in zones if z["name"] != zone_name]
write_named_conf_local(zones)
bind_reload()
u = current_user()
AuditLog.log(u.username, "删除Zone", f"域名: {zone_name}")
flash(f"域名 {zone_name} 已删除", "success")
return redirect(url_for("zone_list"))
@app.route("/zones/<zone_name>/raw", methods=["GET", "POST"])
@login_required
def zone_raw(zone_name):
zone_name = zone_name.rstrip('.')
zones = parse_named_conf_local()
zone = next((z for z in zones if z["name"] == zone_name), None)
if not zone:
abort(404)
if request.method == "POST":
content = request.form.get("content", "")
# Backup
backup_path = zone["file"] + ".bak"
if os.path.exists(zone["file"]):
run_cmd(f"cp {zone['file']} {backup_path}")
# Validate
with open("/tmp/zone_check.tmp", 'w') as f:
f.write(content)
rc, out, err = run_cmd(f"named-checkzone {zone_name} /tmp/zone_check.tmp")
if rc != 0:
flash(f"Zone 文件验证失败: {err}", "error")
return render_template("zone_raw.html", zone=zone, content=content)
# Write
with open(zone["file"], 'w') as f:
f.write(content)
run_cmd(f"chown bind:bind {zone['file']}")
run_cmd(f"rm -f /tmp/zone_check.tmp")
bind_reload()
u = current_user()
AuditLog.log(u.username, "编辑Zone原始文件", f"域名: {zone_name}")
flash("Zone 文件已保存并重新加载", "success")
return redirect(url_for("zone_detail", zone_name=zone_name))
try:
with open(zone["file"]) as f:
content = f.read()
except Exception:
content = ""
return render_template("zone_raw.html", zone=zone, content=content)
# ─── Record management ───
@app.route("/zones/<zone_name>/records/add", methods=["POST"])
@login_required
def record_add(zone_name):
zone_name = zone_name.rstrip('.')
zones = parse_named_conf_local()
zone = next((z for z in zones if z["name"] == zone_name), None)
if not zone:
abort(404)
name = request.form.get("name", "").strip()
rtype = request.form.get("type", "").strip().upper()
ttl = request.form.get("ttl", "").strip()
data = request.form.get("data", "").strip()
if not rtype or not data:
flash("记录类型和数据不能为空", "error")
return redirect(url_for("zone_detail", zone_name=zone_name))
# Validate
ok, err = validate_record(rtype, data)
if not ok:
flash(f"记录验证失败: {err}", "error")
return redirect(url_for("zone_detail", zone_name=zone_name))
if not name:
name = "@"
# Append record directly to zone file and bump serial
ok, msg = append_record_to_file(zone["file"], name, ttl, rtype, data)
if not ok:
flash(f"添加记录失败: {msg}", "error")
return redirect(url_for("zone_detail", zone_name=zone_name))
# Validate zone
rc, out, err = run_cmd(f"named-checkzone {zone_name} {zone['file']}")
if rc != 0:
flash(f"Zone 验证失败: {out} {err}", "error")
return redirect(url_for("zone_detail", zone_name=zone_name))
bind_reload()
u = current_user()
AuditLog.log(u.username, "添加记录", f"域名: {zone_name}, {name} {rtype} {data}")
flash(f"记录 {name} {rtype} 添加成功", "success")
return redirect(url_for("zone_detail", zone_name=zone_name))
@app.route("/zones/<zone_name>/records/<int:idx>/delete", methods=["POST"])
@login_required
def record_delete(zone_name, idx):
zone_name = zone_name.rstrip('.')
zones = parse_named_conf_local()
zone = next((z for z in zones if z["name"] == zone_name), None)
if not zone:
abort(404)
# Delete record from file and bump serial
ok, deleted_str, msg = delete_record_from_file(zone["file"], zone_name, idx)
if not ok:
flash(f"删除记录失败: {msg}", "error")
return redirect(url_for("zone_detail", zone_name=zone_name))
bind_reload()
u = current_user()
AuditLog.log(u.username, "删除记录", f"域名: {zone_name}, {deleted_str}")
flash(f"记录 {deleted_str} 已删除", "success")
return redirect(url_for("zone_detail", zone_name=zone_name))
# ─── Service control ───
@app.route("/service/<action>")
@login_required
def service_control(action):
u = current_user()
if action == "start":
rc, out, err = run_cmd("systemctl start named 2>&1")
AuditLog.log(u.username, "启动BIND服务")
msg = "BIND 服务已启动" if rc == 0 else f"启动失败: {err}"
elif action == "stop":
rc, out, err = run_cmd("systemctl stop named 2>&1")
AuditLog.log(u.username, "停止BIND服务")
msg = "BIND 服务已停止" if rc == 0 else f"停止失败: {err}"
elif action == "restart":
rc, out, err = run_cmd("systemctl restart named 2>&1")
AuditLog.log(u.username, "重启BIND服务")
msg = "BIND 服务已重启" if rc == 0 else f"重启失败: {err}"
elif action == "reload":
ok, msg = bind_reload()
AuditLog.log(u.username, "重载BIND配置")
msg = "BIND 配置已重载" if ok else f"重载失败: {msg}"
else:
msg = "未知操作"
flash(msg, "success" if "失败" not in msg else "error")
return redirect(url_for("dashboard"))
# ─── Configuration ───
@app.route("/config")
@login_required
def config_view():
try:
with open(BIND_CONF_OPTIONS) as f:
options_content = f.read()
except Exception:
options_content = ""
try:
with open(BIND_CONF_LOCAL) as f:
local_content = f.read()
except Exception:
local_content = ""
try:
with open("/etc/bind/named.conf") as f:
main_content = f.read()
except Exception:
main_content = ""
# Check config
rc, out, err = run_cmd("named-checkconf 2>&1")
config_ok = rc == 0
config_msg = out + err
return render_template("config.html",
options_content=options_content,
local_content=local_content,
main_content=main_content,
config_ok=config_ok,
config_msg=config_msg)
@app.route("/config/options", methods=["POST"])
@login_required
def config_options_save():
content = request.form.get("content", "")
backup = BIND_CONF_OPTIONS + ".bak"
run_cmd(f"cp {BIND_CONF_OPTIONS} {backup}")
# Write to temp and validate
with open("/tmp/named_check.tmp", 'w') as f:
f.write(content)
rc, out, err = run_cmd("named-checkconf /tmp/named_check.tmp 2>&1")
run_cmd("rm -f /tmp/named_check.tmp")
if rc != 0:
flash(f"配置验证失败: {err}", "error")
return redirect(url_for("config_view"))
with open(BIND_CONF_OPTIONS, 'w') as f:
f.write(content)
run_cmd(f"chown root:bind {BIND_CONF_OPTIONS}")
bind_reload()
u = current_user()
AuditLog.log(u.username, "修改BIND options配置")
flash("options 配置已保存", "success")
return redirect(url_for("config_view"))
@app.route("/config/local", methods=["POST"])
@login_required
def config_local_save():
content = request.form.get("content", "")
backup = BIND_CONF_LOCAL + ".bak"
run_cmd(f"cp {BIND_CONF_LOCAL} {backup}")
with open(BIND_CONF_LOCAL, 'w') as f:
f.write(content)
run_cmd(f"chown root:bind {BIND_CONF_LOCAL}")
rc, out, err = run_cmd("named-checkconf 2>&1")
if rc != 0:
run_cmd(f"cp {backup} {BIND_CONF_LOCAL}")
flash(f"配置验证失败，已恢复: {err}", "error")
return redirect(url_for("config_view"))
bind_reload()
u = current_user()
AuditLog.log(u.username, "修改BIND local配置")
flash("local 配置已保存", "success")
return redirect(url_for("config_view"))
# ─── Query test ───
@app.route("/query", methods=["GET", "POST"])
@login_required
def query_test():
result = None
if request.method == "POST":
domain = request.form.get("domain", "").strip()
qtype = request.form.get("qtype", "A").strip().upper()
server = request.form.get("server", "").strip()
if not domain:
flash("请输入域名", "error")
return render_template("query.html", result=None)
# Build dig command
port = ""
cmd = f"dig {shlex.quote(domain)} {shlex.quote(qtype)} @127.0.0.1"
if server:
cmd = f"dig {shlex.quote(domain)} {shlex.quote(qtype)} {shlex.quote(server)}"
rc, out, err = run_cmd(cmd, timeout=15)
result = {
"domain": domain,
"qtype": qtype,
"server": server or "127.0.0.1:53",
"output": out if out else err,
"success": rc == 0,
}
return render_template("query.html", result=result, qtypes=RECORD_TYPES)
# ─── Logs ───
@app.route("/logs")
@login_required
def logs():
page = request.args.get("page", 1, type=int)
per_page = 50
pagination = AuditLog.query.order_by(AuditLog.timestamp.desc()).paginate(
page=page, per_page=per_page, error_out=False
)
return render_template("logs.html", logs=pagination.items, pagination=pagination)
# ─── API ───
@app.route("/api/status")
@login_required
def api_status():
return jsonify(bind_service_status())
# ─── Error handlers ───
@app.errorhandler(404)
def not_found(e):
return render_template("error.html", code=404, message="页面不存在"), 404
@app.errorhandler(500)
def server_error(e):
return render_template("error.html", code=500, message="服务器错误"), 500
# ════════════════════════════════════════════════════════════════════
# Init
# ════════════════════════════════════════════════════════════════════
def seed_admin():
from sqlalchemy.exc import IntegrityError
try:
if not User.query.filter_by(username="admin").first():
a = User(username="admin", role="admin")
a.set_password("admin")
db.session.add(a)
db.session.commit()
print("[init] Default admin user created (admin/admin)")
except IntegrityError:
db.session.rollback()
print("[init] Admin user already exists (race condition handled)")
if __name__ == "__main__":
os.makedirs(app.instance_path, exist_ok=True)
with app.app_context():
db.create_all()
seed_admin()
app.run(host="0.0.0.0", port=5300, debug=True)
Reference in New Issue View Git Blame Copy Permalink