python中dns库用法详解（DNS处理模块）

dnspython是python 实现的一个dns场景的工具包。

dnspython（http://www.dnspython.org/）是Python实现的一个DNS工具包，它支持几乎所有的记录类型，可以用于查询、传输并动态更新ZONE信息，同时支持TSIG（事务签名）验证消息和EDNS0（扩展DNS）。在系统管理方面，我们可以利用其查询功能来实现DNS服务监控以及解析结果的校验，可以代替nslookup及dig等工具

更多关于域名知识详解博文：Windows系统DNS部署与安全_IT之一小佬的博客-CSDN博客_windows10搭建dns服务器

安装：

pip install dnspython

dns.resolve源码解析：

def resolve(qname, rdtype=dns.rdatatype.A, rdclass=dns.rdataclass.IN,tcp=False, source=None, raise_on_no_answer=True,source_port=0, lifetime=None, search=None):"""Query nameservers to find the answer to the question.This is a convenience function that uses the default resolverobject to make the query.See ``dns.resolver.Resolver.resolve`` for more information on theparameters."""return get_default_resolver().resolve(qname, rdtype, rdclass, tcp, source,raise_on_no_answer, source_port,lifetime, search)

参数解析：

qname 指定查询的名称
rdtype 指定查询的类型默认是A记录

A记录，将主机名转换为IP地址；

MX记录，邮件交换记录，定义邮件服务器的域名；

CNAME记录，指别名记录，实现域名间的映射；

NS记录，标记区域的域名服务器及授权子域；

PTR记录，反向解析，与A记录相反，将IP转换为主机名；

SOA记录，SOA标记，一个起始授权区的定义。

rdclass 指定查询的网络类型 ,可选的值有IN、CH与HS,默认是 IN
tcp 是否启用tcp查询模式
source 和source_port 指定查询用的源地址和端口
raise_on_no_answer 查询无结果的时候，是否需要抛出异常
lifetime 声明周期配置参数，采用默认值

1、A记录实现A记录查询方法源码

示例代码：

# 导包
import dns.resolverdomain = 'www.baidu.com'# 获取解析对象
query_object = dns.resolver.resolve(qname=domain, rdtype='A')  # 指定查询记录为A
print('查询对象：{}'.format(query_object))print('*' * 100)# 查看response对象
response_object = query_object.response
print('应答对象：{}'.format(response_object))print('*' * 100)# 查询answer对象
answer_object = response_object.answer
print('解析对象：{}'.format(answer_object))print('*' * 100)# 查看解析条目对象
answer_object = response_object.answer
for query_item in answer_object:print('查询条目：{}'.format(query_item))print('*' * 100)# 解析后的记录条目
query_item = response_object.answer[0]
# print(query_item)
for item in query_item:print('解析记录：{}'.format(item))print('*' * 100)

运行结果：

示例代码：

# 导包
import dns.resolverdomain = 'www.baidu.com'query_object = dns.resolver.resolve(qname=domain, rdtype='A')
for query_item in query_object.response.answer:for item in query_item.items:print("{}的A记录解析地址有：{}".format(domain, item))

运行结果：

示例代码：

import dns.resolverdomain = 'www.baidu.com'
# 指定查询类型为A记录
A = dns.resolver.query(domain, 'A')
# 通过response.answer方法获取查询信息
for i in A.response.answer:print('+' * 100)print(i)# 遍历回应信息for j in i.items:print(j)print("*" * 100)

运行结果：

2、MX记录实现MX记录查询方法源码

示例代码：

# 必须这样导入
import dns.resolverret = dns.resolver.query('dnspython.org', 'MX')
print(ret)for data in ret:print(data)print('*' * 100)for data in ret:print(data.exchange, data.preference)print('*' * 100)ret2 = dns.resolver.resolve('dnspython.org', 'MX')
print(ret2)
for data in ret2:print(data.exchange, data.preference)

运行结果：

示例代码：

import dns.resolverdomain = '126.com'# 获取解析对象
query_object = dns.resolver.resolve(domain, 'MX')  # MX记录
print('MX记录的结果：{}'.format(query_object))print('*' * 100)# 从应答的response中获取查询记录
resp_object = query_object.response
print('MX记录的结果：{}'.format(resp_object))print('*' * 100)# 从应答的answer中获取查询目录
answer_object = resp_object.answer
print('MX记录的结果：{}'.format(answer_object))print('*' * 100)# 查看解析条目对象
answer_object = resp_object.answer
for query_item in answer_object:print('查询条目：{}'.format(query_item))print('*' * 100)# 查看解析后的记录条目
query_item = resp_object.answer[0]
for item in query_item.items:print('查询条目：{}'.format(item))print('*' * 100)# 查看解析后的记录条目-详细
query_item = resp_object.answer[0]
for item in query_item.items:print('邮件服务器权重：{}，邮件服务器地址：{}'.format(item.preference, item.exchange))

运行结果：

示例代码：

# 导包
import dns.resolverdomain = '126.com'query_object = dns.resolver.resolve(domain, 'MX')
for query_item in query_object.response.answer:for item in query_item.items:print('邮件服务器权重：{}，邮件服务器地址：{}'.format(item.preference, item.exchange))

运行结果：

示例代码：

import dns.resolver# 输入域名地址
domain = 'baidu.com'# 指定查询类型为MX
MX = dns.resolver.query(domain, 'MX')# 遍历回应结果，输出MX记录的preference及exchanger信息
for i in MX:print('MX preference=', i.preference, 'mail exchange=', i.exchange)

运行结果：

3、NS记录实现NS记录查询方法源码

NS（Name Server）域名服务器记录。用来表明由哪台服务器对该域名进行解析。在注册域名时，总有默认的DNS服务器，每个注册的域名都是由一个DNS域名服务器来进行解析的。但是需要注意的是只能输入一级域名，如：baidu.com；对于二级以及多级域名，如www.baidu.com、wenku.baidu.com则是错误的。

示例代码：

# 导包
import dns.resolverdomain = '126.com'NS = dns.resolver.resolve(domain, 'NS')
for i in NS.response.answer:for j in i.items:print(j.to_text())

运行结果：

4、CNAME记录实现CNAME记录查询方法源码

示例代码：

# 导包
import dns.resolverdomain = 'www.baidu.com'CNAME = dns.resolver.resolve(domain, 'CNAME')
for i in CNAME.response.answer:for j in i.items:print(j.to_text())

运行结果：

5、案例分析：DNS域名轮询业务监控

大部分的DNS解析都是一个域名对应一个IP地址，但是通过DNS轮询技术可以做到一个域名对应多个IP，从而实现最简单高效的负载均衡，不过此方案最大的弊端是目标主机不可用时无法被自动剔除，因此做好业务主机的服务可用监控至关重要。

实现步骤：

实现域名的解析，获取域名所有的A记录解析IP列表
对IP列表进行HTTP级别的探测

具体实现方式：

第一步通过dns.resolver.query()方法获取业务域名A记录信息，查询出所有IP地址列表；

第二步使用httplib模块的request()方法以GET请求监控页面，监控业务所有服务的IP是否正常。

示例代码：

import dns.resolver
import http.client# 定义域名ip列表变量
ip_list = []
# 定义业务域名
app_domain = "www.baidu.com"
# app_domain = "www.blog.csdn.net"# 域名解析函数，解析成功IP将被追加到ip_list
def get_ip_list(domain=""):try:A = dns.resolver.query(domain, 'A')except Exception as e:print("dns resolver error: " + str(e))returnfor i in A.response.answer:for j in i.items:# 追加到ip_list列表ip_list.append(j)return Truedef check_ip(ip):check_url = str(ip) + ": 80"get_content = ""# 定义http连接5秒超时(5)秒http.client.socket.setdefaulttimeout(5)# 创建http连接对象conn = http.client.HTTPConnection(check_url)try:# 发起url请求，添加host主机头conn.request("GET", "/", headers={"Host": app_domain})r = conn.getresponse()# 获取URL页面前15个字符用来可用性效验get_content = r.read(15)finally:# 监控URL页的内容一般是事先定义好的比如http 200等# 判断是否为字节if isinstance(get_content, bytes):get_content = get_content.decode('utf-8')if get_content == "<!DOCTYPE html>":print(str(ip) + "\033[32;1m [ok]\033[0m")else:print(str(ip) + "\033[31;1m [Error]\033[0m")# 获取response的状态码print(r.reason)# 获取response的状态 ok或errorprint(r.status)if __name__ == "__main__":# 域名解析正确且至少返回一个IPif get_ip_list(app_domain) and len(ip_list) > 0:for ip in ip_list:check_ip(ip)else:print("dns resolver error.")

运行结果：

参考博文：

python运维常用模块（三）——DNS处理模块dnspython - jiayou111 - 博客园