AI-Agent/dns-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity
Files
1614e37386f10e931f212203c546b0764ff0d714
dns-service/app.py
T
Hermes 1614e37386 feat: structured upstream DNS / allow-recursion editor in Web UI 
新增配置管理页面 '上游 DNS 转发' 卡片，让管理员通过两个 textarea
分别维护 forwarders 和 allow-recursion，不需要手写 BIND 语法：

  - 每行一个 IP / CIDR
  - 支持 # 和 // 注释
  - 留空 = 清空该字段（forwarders 留空表示走根提示；
    allow-recursion 留空表示只有 localhost/localnets 能递归）

实现要点：

1. _parse_options_lists() 从 options 文件文本里抽出 forwarders /
   allow-recursion 列表（用大括号平衡匹配，不依赖单行格式）。
2. _replace_or_append_option() 在 options block 里精准替换已有 directive，
   自动跟随原缩进；找不到则按 4 空格缩进追加到 'options {' 之后；
   options block 本身不存在则新建一个。
3. _ensure_options_file() 自动创建缺失的 options 文件（带合理的最小
   默认值），避免 'No such file' 错误。
4. config_upstream_save() 写文件前用 named-checkconf 验证语法，
   失败则不写入；成功则备份 .bak + 写文件 + rndc reload。

附带的修复：config_view 进入页面时也调用 _ensure_options_file()，
解决了 'Web 端第一次配置 options 文件不存在' 的问题。

模板 templates/config.html 新增 '上游 DNS 转发' 卡片，放在 named.conf
主配置卡片和 named.conf.options 卡片之间，placeholder 给出常用示例
（223.5.5.5 / 114.114.114.114 / RFC1918 三段）。

回归测试（dev 环境）：
- _parse_options_lists: 正确从样例 options 抽出 3 forwarders + 2 recursion
- _replace_or_append_option: 替换后缩进正确，保留其它 directives
- round-trip：表单输入 -> 写文件 -> 再解析 -> 一致
2026-07-24 17:58:56 +08:00

1407 lines
48 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 ───
# 优先级：环境变量 > 自动探测 > 内置默认值。
# 自动探测按"哪个目录/文件存在就用谁"在常见路径里回退，覆盖 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/<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']} 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/<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")
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
def _replace_or_append_option(content, directive, new_block):
"""Replace an existing '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 nested braces in directives by counting brace depth (so multi-line
directives like `forwarders { 1.1.1.1; }` work; nested ones like `key ...`
are not expected in options block).
"""
# Try to find an existing directive with balanced-brace body
pattern = re.compile(
r'(' + re.escape(directive) + r')\s*\{',
re.MULTILINE,
)
m = pattern.search(content)
if m:
start = m.start()
# Detect indentation by looking at whitespace before the directive
line_start = content.rfind('\n', 0, start) + 1
indent = ''
for ch in content[line_start:start]:
if ch in ' \t':
indent += ch
else:
break
# Walk forward from `{` to find the matching `};`
depth = 0
i = m.end() - 1 # points at `{`
for j in range(i, len(content)):
c = content[j]
if c == '{':
depth += 1
elif c == '}':
depth -= 1
if depth == 0:
# Expect `;` shortly after; consume any whitespace
end = j + 1
while end < len(content) and content[end] in ' \t\n':
end += 1
if end < len(content) and content[end] == ';':
end += 1
# Indent new_block to match the directive's indentation
indented = '\n'.join(
(indent + line) if line else line
for line in new_block.split('\n')
)
return content[:start] + indented + content[end:]
return content # malformed, give up
# Not found — append after the opening "options {" line
opt_match = re.search(r'options\s*\{', content)
if opt_match:
insert_at = opt_match.end()
# Insert after the newline that follows
nl = content.find('\n', insert_at)
if nl >= 0:
insert_at = nl + 1
# Indent new_block 4 spaces (typical options-body indent)
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:]
# No options block — create one
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)
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)
@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", "")
new_fwd = _parse_list(raw_fwd)
new_rc = _parse_list(raw_rc)
_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),
)
# 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}")
flash(f"已保存：forwarders {len(new_fwd)} 条，allow-recursion {len(new_rc)} 条", "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