#!/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 # ════════════════════════════════════════════════════════════════════ # python3.7 #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) # python3.6 def run_cmd(cmd, timeout=10): """Run shell command, return (returncode, stdout, stderr).""" try: r = subprocess.run( cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=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/") @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//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//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//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//records//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/") @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)