如何写DNS解析查询脚本

wen 实用脚本 24

如何写DNS解析查询脚本(从入门到实战)

目录导读

  1. 什么是DNS解析查询脚本?
  2. 脚本的核心功能与适用场景
  3. 常用编程语言与工具选择(Python/Bash/Go)
  4. 手写第一个DNS查询脚本(Python示例)
  5. 进阶功能:多域名批量查询、超时处理、结果格式化
  6. 常见问题与避坑指南(问答环节)
  7. 性能优化与安全考量
  8. 脚本部署与自动化集成建议

什么是DNS解析查询脚本?

DNS(域名系统)解析查询脚本,本质上是一段自动化程序,用于向DNS服务器发送查询请求,获取域名对应的IP地址、MX记录、CNAME等资源记录,这类脚本广泛应用于网络诊断、网站运维、安全审计以及自动化部署流程中。

如何写DNS解析查询脚本

核心价值

  • 快速检测域名可用性与解析状态
  • 批量验证上千个域名的配置正确性
  • 监控DNS变更并触发告警

脚本的核心功能与适用场景

一个成熟的DNS查询脚本至少应包含:

  • 基础查询:A记录、AAAA记录、CNAME、MX、NS、TXT
  • 错误处理:超时、NXDOMAIN、SERVFAIL等状态码
  • 输出格式化:JSON/CSV/表格(便于后续解析)

典型场景举例

  • CDN部署前验证域名指向是否正确
  • 安全团队检测DNS劫持或恶意域名解析
  • DevOps自动化中检查服务发现域名状态

常用编程语言与工具选择

语言/工具 优点 适用场景
Python + dnspython 语法简洁、第三方库强大 复杂逻辑、批量查询
Bash + dig/nslookup 零依赖、快速执行 Linux环境下的快速诊断
Go + net.LookupHost 高性能、并发友好 高吞吐量生产环境
Node.js + dns模块 事件驱动 实时监控系统

典型的选择逻辑

  • 如果只需要几个命令行的查询,Bash脚本最快
  • 如果需要复杂的错误处理和多种记录类型,Python最通用
  • 如果需要极高性能(例如每秒查询10000次),Go是首选

手写第一个DNS查询脚本(Python示例)

环境准备

pip install dnspython

基础查询脚本

import dns.resolver
def query_a_record(domain):
    try:
        result = dns.resolver.resolve(domain, 'A')
        for ip in result:
            print(f'{domain} -> {ip}')
    except dns.resolver.NXDOMAIN:
        print(f'{domain} 不存在')
    except dns.resolver.Timeout:
        print(f'{domain} 查询超时')
    except Exception as e:
        print(f'错误: {e}')
if __name__ == '__main__':
    query_a_record('example.com')

运行效果

example.com -> 93.184.216.34

问答环节

Q:为什么选择dns.resolver而不是socket.gethostbyname
A:socket.gethostbyname不支持指定DNS服务器、超时控制,且无法查询MX、TXT等记录。dns.resolver提供完全控制,包括EDNS、DNSSEC验证等。


进阶功能:多域名批量查询、超时处理、结果格式化

批量查询 + 输出JSON

import dns.resolver
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
def query_all(domains):
    results = {}
    with ThreadPoolExecutor(max_workers=20) as executor:
        futures = {executor.submit(query_single, d): d for d in domains}
        for future in as_completed(futures):
            domain = futures[future]
            results[domain] = future.result()
    return results
def query_single(domain):
    try:
        answer = dns.resolver.resolve(domain, 'A', lifetime=5)
        return [str(r) for r in answer]
    except:
        return None
# 使用示例
domains = ['example.com', 'google.com', 'github.com']
res = query_all(domains)
print(json.dumps(res, indent=2))

超时与重试机制

import time
import dns.resolver
def robust_query(domain, retries=3, timeout=2):
    for attempt in range(retries):
        try:
            resolver = dns.resolver.Resolver()
            resolver.timeout = timeout
            resolver.lifetime = timeout
            answer = resolver.resolve(domain, 'A')
            return answer
        except dns.resolver.Timeout:
            if attempt == retries - 1:
                raise
            time.sleep(1)

要点

  • 使用lifetime参数限制总查询时间
  • 指数退避重试避免雪崩

常见问题与避坑指南(问答环节)

Q1:脚本查询结果与nslookup不一致?
A:可能原因:

  • 脚本使用了系统默认DNS(如8.8.8.8),而nslookup用了本机缓存
  • 某些域名配置了分地域解析(GSLB),不同DNS服务器返回不同IP
  • 建议显式指定resolver.nameservers = ['8.8.8.8']

Q2:如何处理CNAME循环解析?
A:设置最大跟随次数,防止无限递归:

def resolve_with_follow(domain, max_follow=10):
    visited = []
    for _ in range(max_follow):
        if domain in visited:
            raise Exception('CNAME循环')
        visited.append(domain)
        answer = dns.resolver.resolve(domain, 'CNAME')
        if answer:
            domain = str(answer[0].target)
        else:
            return dns.resolver.resolve(domain, 'A')

Q3:脚本在高并发下返回OSError: [Errno 24] Too many open files
A:原因:并发socket耗尽系统文件描述符。
解决方案:

  • ThreadPoolExecutor前设置ulimit -n 65535
  • 使用asyncio事件循环而非多线程

Q4:如何验证DNSSEC签名?
A:使用dns.dnssec模块:

import dns.dnssec
answer = dns.resolver.resolve('example.com', 'A', raise_on_no_answer=False)
if dns.dnssec.validate(answer):
    print('DNSSEC验证通过')

性能优化与安全考量

性能优化策略

技术 效果 实现方式
连接池复用 减少TCP握手 使用dns.resolver.Resolver()全局实例
异步I/O 提高并发 asyncio + dns.asyncquery
缓存DNS结果 降低DNS服务器负载 dns.resolver.Resolver().cache = dns.resolver.Cache()
就近DNS服务器 减少延迟 配置离业务服务器最近的DNS IP

安全注意事项

  • 避免未经授权的扫描:大量DNS查询可能被ISP或目标DNS视为攻击
  • 隐蔽性:设置合理的查询间隔(如0.1秒/次),添加随机化
  • 数据保密:如查询结果包含内部域名,输出时需脱敏处理

脚本部署与自动化集成建议

部署方式

# 作为系统服务(Linux)
[Unit]
Description=DNS health checker
After=network.target
[Service]
ExecStart=/usr/bin/python3 /opt/dns_checker.py
Restart=always
User=nobody
[Install]
WantedBy=multi-user.target

与监控系统集成(Prometheus)

from prometheus_client import start_http_server, Gauge
gauge = Gauge('dns_resolution_ms', 'DNS resolution time', ['domain'])
gauge.labels(domain='example.com').set(resolution_time_ms)

与CI/CD流水线结合

# GitLab CI 示例
validate-dns:
  script:
    - python3 dns_check.py --domains deploy.example.com
  only:
    - main

写一个高效的DNS解析查询脚本,关键在于理解DNS协议特性、选择合适的技术栈,并针对生产环境做好错误处理与性能优化,从简单的单域名查询,到支持递归、批量、并发、监控的完整工具链,每一步都建立在实际需求之上,希望本文能帮助你快速构建属于自己的DNS查询利器,让域名运维工作更加自动化和可靠。

延伸参考:RFC 1035(DNS协议规范)、dns.resolver官方文档、DNSSEC实践指南

抱歉,评论功能暂时关闭!