#!/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 ─── # 优先级：环境变量 > 自动探测 > 内置默认值。 # 自动探测按"哪个目录/文件存在就用谁"在常见路径里回退，覆盖 Debian/Ubuntu、 # RHEL/CentOS/Rocky、AlmaLinux、FreeBSD 等发行版默认布局；如有特殊安装路径， # 可用环境变量强制指定。 # # 注意 BIND_CONF_LOCAL 的探测顺序：/etc/named.conf.local 优先于 # /etc/bind/named.conf.local —— 因为 RHEL/CentOS 的 /etc/named.conf 默认 # include 的是 /etc/named.conf.local，而 Debian/Ubuntu 默认 include 的是 # /etc/bind/named.conf.local。优先探测 RHEL 路径可以避免"两边文件都存在但 # named 只读其中一份"的歧义场景。 _BIND_PATH_CANDIDATES = { "BIND_CONF_DIR": ["/etc/bind", "/etc", "/etc/named"], "BIND_ZONES_DIR": ["/etc/bind/zones", "/var/named", "/var/named/data"], "BIND_CONF_LOCAL": ["/etc/named.conf.local", "/etc/bind/named.conf.local"], "BIND_CONF_OPTIONS": ["/etc/named.conf", "/etc/bind/named.conf.options"], "BIND_RNDC_KEY": ["/etc/bind/rndc.key", "/etc/rndc.key", "/var/named/key"], } def _detect_bind_path(candidates): """Return first existing path in candidates; fall back to first (built-in default).""" for p in candidates: if os.path.exists(p): return p return candidates[0] BIND_CONF_DIR = os.environ.get("BIND_CONF_DIR", _detect_bind_path(_BIND_PATH_CANDIDATES["BIND_CONF_DIR"])) BIND_ZONES_DIR = os.environ.get("BIND_ZONES_DIR", _detect_bind_path(_BIND_PATH_CANDIDATES["BIND_ZONES_DIR"])) BIND_CONF_LOCAL = os.environ.get("BIND_CONF_LOCAL", _detect_bind_path(_BIND_PATH_CANDIDATES["BIND_CONF_LOCAL"])) BIND_CONF_OPTIONS = os.environ.get("BIND_CONF_OPTIONS", _detect_bind_path(_BIND_PATH_CANDIDATES["BIND_CONF_OPTIONS"])) BIND_RNDC_KEY = os.environ.get("BIND_RNDC_KEY", _detect_bind_path(_BIND_PATH_CANDIDATES["BIND_RNDC_KEY"])) BIND_SERVICE = os.environ.get("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 parse_all_managed_zones(): """Return all zone names that BIND currently has loaded (master/slave/in-view), including zones declared in the main named.conf outside BIND_CONF_LOCAL. Uses `named-checkconf -p` which prints the fully-resolved configuration with all included files expanded, so we can detect duplicates the user might have added manually (e.g. a zone declared in /etc/named.conf that web UI doesn't know about). Returns a dict: zone_name -> {"type": str, "file": str, "source": "local"|"main"}. """ rc, out, err = run_cmd("named-checkconf -p 2>&1", timeout=15) if rc != 0: # If named-checkconf fails (e.g. broken config), fall back to local-only # so the UI doesn't break entirely. return {z["name"]: {"type": z["type"], "file": z["file"], "source": "local"} for z in parse_named_conf_local()} # Parse the dumped config: zone statements with a non-builtin type (i.e. not # "hint" and not under automatic empty-zones). We deliberately include builtin # zones too and let callers filter. result = {} pattern = re.compile( r'zone\s+"([^"]+)"\s+(?:IN\s+)?\{\s*' r'type\s+(\w+)\s*;\s*' r'(?:file\s+"([^"]+)"\s*;\s*)?', re.MULTILINE, ) for m in pattern.finditer(out): name, ztype, zfile = m.group(1), m.group(2), m.group(3) or "" # Skip builtin/hint zones and the "empty" automatic zones (no file). if ztype == "hint" or not zfile: continue result[name] = {"type": ztype, "file": zfile, "source": "unknown"} # Mark which ones came from BIND_CONF_LOCAL for the UI to know whether to # let the user edit/delete them. local_names = {z["name"] for z in parse_named_conf_local()} for name in result: if name in local_names: result[name]["source"] = "local" else: result[name]["source"] = "main" return result 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} 2>/dev/null || chown named:named {filepath} 2>/dev/null") run_cmd(f"chmod 644 {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} 2>/dev/null || chown named:named {filepath} 2>/dev/null") run_cmd(f"chmod 644 {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 — both against zones we manage in BIND_CONF.local AND # against zones declared elsewhere in named.conf (e.g. user manually # added a zone block in /etc/named.conf). Without the second check, # creating a zone with a name already declared in the main config would # produce a duplicate zone error on BIND reload. existing = parse_named_conf_local() all_managed = parse_all_managed_zones() if any(z["name"] == zone_name for z in existing): flash(f"域名 {zone_name} 已在 Web UI 中存在", "error") return render_template("zone_form.html") if zone_name in all_managed: flash( f"域名 {zone_name} 已在 named.conf 中声明（不在 Web UI 管理范围内），" f"请先到 named.conf 中注释/删除后再创建", "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 try: os.makedirs(BIND_ZONES_DIR, exist_ok=True) except PermissionError as e: flash(f"无法创建 zone 目录 {BIND_ZONES_DIR}（权限不足）：{e}", "error") return render_template("zone_form.html") with open(zone_file, 'w') as f: f.write(zone_content) # BIND runs as user 'bind' (Debian/Ubuntu) or 'named' (RHEL/CentOS). # Try Debian first; fall back to RHEL. run_cmd(f"chown bind:bind {zone_file} 2>/dev/null || chown named:named {zone_file} 2>/dev/null") 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']} 2>/dev/null || chown named:named {zone['file']} 2>/dev/null") run_cmd(f"chmod 644 {zone['file']}") 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") AuditLog.log(u.username, "启动BIND服务") msg = "BIND 服务已启动" if rc == 0 else f"启动失败: {out} {err}" elif action == "stop": rc, out, err = run_cmd("systemctl stop named") AuditLog.log(u.username, "停止BIND服务") msg = "BIND 服务已停止" if rc == 0 else f"停止失败: {out} {err}" elif action == "restart": rc, out, err = run_cmd("systemctl restart named") AuditLog.log(u.username, "重启BIND服务") msg = "BIND 服务已重启" if rc == 0 else f"重启失败: {out} {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 ─── def _parse_options_lists(content): """Extract forwarders and allow-recursion IP/CIDR lists from a named.conf.options-style text. Returns: forwarders: list[str] — IPs in order; [] if not present recursion: list[str] — CIDRs/acls in order; [] if not present """ forwarders = [] recursion = [] fw_match = re.search( r'forwarders\s*\{([^}]*)\}', content, re.DOTALL, ) if fw_match: body = fw_match.group(1) for token in re.findall(r'\d+\.\d+\.\d+\.\d+|[0-9a-fA-F:]+', body): # Skip plain numbers (matches like "port 53") — only accept IPv4 dotted quad or IPv6 hex if '.' in token and token.count('.') == 3: forwarders.append(token) elif ':' in token and re.match(r'^[0-9a-fA-F:]+$', token): forwarders.append(token) rc_match = re.search( r'allow-recursion\s*\{([^}]*)\}', content, re.DOTALL, ) if rc_match: body = rc_match.group(1) # Pull out quoted strings and unquoted tokens for m in re.finditer(r'"([^"]+)"|([\w./-]+)', body): token = m.group(1) or m.group(2) if token and token not in ('any', 'none', 'localhost', 'localnets'): recursion.append(token) return forwarders, recursion _VALIDATION_VALUES = ("auto", "yes", "no") def _parse_options_dnssec(content): """Return current dnssec-validation value as one of 'auto'/'yes'/'no'. Returns 'auto' if the directive is absent (BIND default). """ m = re.search(r'dnssec-validation\s+(\w+)\s*;', content) if not m: return "auto" val = m.group(1).lower() if val in _VALIDATION_VALUES: return val return "auto" def _listen_on_v4_set(content): """True if the options block contains any IPv4 'listen-on' directive.""" # Match listen-on (port N)? { ... } but NOT listen-on-v6 return bool(re.search(r'(?;'.""" m = re.search(r'recursion\s+(\w+)\s*;', content) if not m: return "yes" val = m.group(1).lower() return val if val in ("yes", "no") else "yes" def _replace_or_append_option(content, directive, new_block): """Replace an existing 'directive { ... };' or 'directive ;' in an options block with new_block, or append it after the opening 'options {' if not present. `new_block` is the full directive text (without leading/trailing newline). Handles: - brace-delimited: 'forwarders { ... };' / 'allow-recursion { ... };' - value-terminated: 'dnssec-validation auto;' / 'recursion yes;' Nested braces are matched by counting depth. Comments ('//', '#', '/*...*/') are skipped so the directive name doesn't match inside comment text. """ # Build a mask of which character ranges are inside comments. We treat the # whole file linearly: # - '//' to end-of-line: line comment (BIND and shell style) # - '#' to end-of-line: line comment (BIND 9.18+ accepts # too) # - '/* ... */': block comment n = len(content) in_comment = [False] * n # True at position i = i is inside a comment i = 0 while i < n: c = content[i] # End of block comment if in_comment[i] is False and i + 1 < n and c == '/' and content[i + 1] == '*': j = i + 2 depth = 1 while j < n and depth > 0: if j + 1 < n and content[j] == '*' and content[j + 1] == '/': depth -= 1 j += 2 else: j += 1 for k in range(i, j): in_comment[k] = True i = j continue # Line comment // ... \n if in_comment[i] is False and i + 1 < n and c == '/' and content[i + 1] == '/': j = i while j < n and content[j] != '\n': in_comment[j] = True j += 1 i = j continue # Line comment # ... \n (skip only if at start of token; be conservative # and treat any '#' preceded by whitespace or start-of-line as a comment) if in_comment[i] is False and c == '#': # Only treat as comment if preceded by whitespace or start-of-line prev_ok = (i == 0) or content[i - 1] in ' \t' if prev_ok: j = i while j < n and content[j] != '\n': in_comment[j] = True j += 1 i = j continue i += 1 pattern = re.compile( r'(?= len(content): continue if content[i] == '{': depth = 0 for j in range(i, len(content)): # Skip over comments inside the brace body if in_comment[j]: continue c = content[j] if c == '{': depth += 1 elif c == '}': depth -= 1 if depth == 0: end = j + 1 while end < len(content) and content[end] in ' \t\n': end += 1 if end < len(content) and content[end] == ';': end += 1 indented = '\n'.join( (indent + line) if line else line for line in new_block.split('\n') ) return content[:start] + indented + content[end:] continue else: end = i while end < len(content) and content[end] != ';': end += 1 if end < len(content): end += 1 indented = '\n'.join( (indent + line) if line else line for line in new_block.split('\n') ) return content[:start] + indented + content[end:] # Not found — append after the opening "options {" line opt_match = re.search(r'options\s*\{', content) if opt_match: insert_at = opt_match.end() nl = content.find('\n', insert_at) if nl >= 0: insert_at = nl + 1 indented = '\n'.join(' ' + line if line else line for line in new_block.split('\n')) indented = '\n' + indented + '\n' return content[:insert_at] + indented + content[insert_at:] return 'options {\n' + new_block + '\n};\n' def _format_list_block(directive, items, indent=4): """Format a 'directive { item1; item2; ... };' block.""" if not items: return f"{directive} {{ }};" pad = ' ' * indent lines = [f"{directive} {{"] for item in items: lines.append(f"{pad}{item};") lines.append("};") return "\n".join(lines) def _ensure_options_file(): """Create BIND_CONF_OPTIONS with a minimal valid options block if missing. Ensures the parent directory exists and the file is owned by root:bind (Debian) / root:named (RHEL), mode 644. """ if os.path.exists(BIND_CONF_OPTIONS): return parent = os.path.dirname(BIND_CONF_OPTIONS) if parent and not os.path.exists(parent): os.makedirs(parent, exist_ok=True) with open(BIND_CONF_OPTIONS, 'w') as f: f.write("options {\n" " directory \"/var/cache/bind\";\n" " listen-on port 53 { any; };\n" " listen-on-v6 { none; };\n" " allow-query { any; };\n" " dnssec-validation auto;\n" " auth-nxdomain no;\n" "};\n") # Ownership: Debian uses bind:named fallback (run_cmd) run_cmd(f"chown root:bind {BIND_CONF_OPTIONS} 2>/dev/null || chown root:named {BIND_CONF_OPTIONS} 2>/dev/null") run_cmd(f"chmod 644 {BIND_CONF_OPTIONS}") @app.route("/config") @login_required def config_view(): _ensure_options_file() 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") config_ok = rc == 0 config_msg = out + err # Parse structured fields for the upstream-DNS form forwarders, recursion = _parse_options_lists(options_content) dnssec_value = _parse_options_dnssec(options_content) listen_on_v4 = _parse_listen_on_v4(options_content) recursion_value = _parse_recursion(options_content) 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, forwarders=forwarders, recursion=recursion, dnssec_value=dnssec_value, dnssec_values=_VALIDATION_VALUES, listen_on_v4=listen_on_v4, recursion_value=recursion_value) @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") run_cmd("rm -f /tmp/named_check.tmp") if rc != 0: flash(f"配置验证失败: {out} {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") if rc != 0: run_cmd(f"cp {backup} {BIND_CONF_LOCAL}") flash(f"配置验证失败，已恢复: {out} {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")) @app.route("/config/upstream", methods=["POST"]) @login_required def config_upstream_save(): """Structured edit of forwarders + allow-recursion in named.conf.options. The textarea content for each field is parsed line-by-line; comments (#, //) and blank lines are stripped. Whitespace and trailing dots are trimmed. This avoids the user accidentally breaking BIND syntax inside a free-form text editor. """ def _parse_list(raw): items = [] for line in (raw or "").splitlines(): s = line.strip() if not s or s.startswith("#") or s.startswith("//"): continue s = s.rstrip(";").strip() if s.startswith("acl ") or s.startswith("key "): # Skip acl/key definitions; only accept inline IPs/CIDRs continue if s: items.append(s) return items raw_fwd = request.form.get("forwarders", "") raw_rc = request.form.get("recursion", "") raw_listen = request.form.get("listen_on_v4", "") dnssec_in = (request.form.get("dnssec") or "auto").strip().lower() if dnssec_in not in _VALIDATION_VALUES: dnssec_in = "auto" recursion_in = (request.form.get("recursion_toggle") or "yes").strip().lower() if recursion_in not in ("yes", "no"): recursion_in = "yes" new_fwd = _parse_list(raw_fwd) new_rc = _parse_list(raw_rc) new_listen = _parse_list(raw_listen) _ensure_options_file() try: with open(BIND_CONF_OPTIONS) as f: content = f.read() except Exception as e: flash(f"读取 {BIND_CONF_OPTIONS} 失败: {e}", "error") return redirect(url_for("config_view")) # Build new directives new_content = _replace_or_append_option( content, "forwarders", _format_list_block("forwarders", new_fwd), ) new_content = _replace_or_append_option( new_content, "allow-recursion", _format_list_block("allow-recursion", new_rc), ) new_content = _replace_or_append_option( new_content, "dnssec-validation", f"dnssec-validation {dnssec_in};", ) # listen-on: empty list -> listen-on port 53 { any; }; (default safe) if not new_listen: new_listen = ["any"] new_content = _replace_or_append_option( new_content, "listen-on", _format_list_block("listen-on port 53", new_listen), ) new_content = _replace_or_append_option( new_content, "recursion", f"recursion {recursion_in};", ) # Validate BEFORE writing to the live file tmp = "/tmp/named_check.tmp" with open(tmp, 'w') as f: f.write(new_content) rc, out, err = run_cmd(f"named-checkconf {tmp}") run_cmd(f"rm -f {tmp}") if rc != 0: flash(f"配置验证失败: {out} {err}", "error") return redirect(url_for("config_view")) # Backup + write backup = BIND_CONF_OPTIONS + ".bak" run_cmd(f"cp {BIND_CONF_OPTIONS} {backup}") with open(BIND_CONF_OPTIONS, 'w') as f: f.write(new_content) run_cmd(f"chown root:bind {BIND_CONF_OPTIONS} 2>/dev/null || chown root:named {BIND_CONF_OPTIONS} 2>/dev/null") bind_reload() u = current_user() AuditLog.log(u.username, "修改上游DNS转发配置", f"forwarders={new_fwd}; allow-recursion={new_rc}; " f"dnssec={dnssec_in}; listen-on={new_listen}; recursion={recursion_in}") flash( f"已保存：forwarders {len(new_fwd)} 条，allow-recursion {len(new_rc)} 条，" f"dnssec-validation {dnssec_in}，listen-on {len(new_listen)} 项，" f"recursion {recursion_in}", "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)