套接字使用特定的 DNS 服务器解析 DNS

Question

我想用特定的 DNS 服务器解析 DNS，例如 Google 的 8.8.8.8。我的实际 Python 代码是：

import socket

def getIP(d):
    try:
        data = socket.gethostbyname(d)
        ip = repr(data)
        return True
    except Exception:
        # fail gracefully!
        return False

可以使用 Python 吗？

Answer 1

您可以使用 dnspython：http://www.dnspython.org/ 在 ubuntu/debian 上，您可以使用：

sudo apt-get install python-dnspython

否则通过：

sudo pip install dnspython

或者通过以下方式下载源安装：

sudo python setup.py install

你的代码应该是这样的：

from dns import resolver

res = resolver.Resolver()
res.nameservers = ['8.8.8.8']

answers = res.query('stackexchange.com')

for rdata in answers:
    print (rdata.address)

编辑： 由于 OP 似乎在 Mac OS X 上使用它时遇到问题，因此我为安装它所做的工作（仅限本地用户）：

git clone git://github.com/rthalley/dnspython.git
cd dnspython
python setup.py install --user

Answer 2

dnspython 是一个成熟的库，但对于到达这里的人来说可能有点矫枉过正。所以，我提供了一个简单的单文件implementation。

import socket
import ipaddress


def parse_dns_string(reader, data):
    res = ''
    to_resue = None
    bytes_left = 0

    for ch in data:
        if not ch:
            break

        if to_resue is not None:
            resue_pos = chr(to_resue) + chr(ch)
            res += reader.reuse(resue_pos)
            break

        if bytes_left:
            res += chr(ch)
            bytes_left -= 1
            continue

        if (ch >> 6) == 0b11 and reader is not None:
            to_resue = ch - 0b11000000
        else:
            bytes_left = ch

        if res:
            res += '.'

    return res


class StreamReader:
    def __init__(self, data):
        self.data = data
        self.pos = 0

    def read(self, len_):
        pos = self.pos
        if pos >= len(self.data):
            raise

        res = self.data[pos: pos+len_]
        self.pos += len_
        return res

    def reuse(self, pos):
        pos = int.from_bytes(pos.encode(), 'big')
        return parse_dns_string(None, self.data[pos:])


def make_dns_query_domain(domain):
    def f(s):
        return chr(len(s)) + s

    parts = domain.split('.')
    parts = list(map(f, parts))
    return ''.join(parts).encode()


def make_dns_request_data(dns_query):
    req = b'\xaa\xbb\x01\x00\x00\x01\x00\x00\x00\x00\x00\x00'
    req += dns_query
    req += b'\x00\x00\x01\x00\x01'
    return req


def add_record_to_result(result, type_, data, reader):
    if type_ == 'A':
        item = str(ipaddress.IPv4Address(data))
    elif type_ == 'CNAME':
        item = parse_dns_string(reader, data)
    else:
        return

    result.setdefault(type_, []).append(item)


def parse_dns_response(res, dq_len, req):
    reader = StreamReader(res)

    def get_query(s):
        return s[12:12+dq_len]

    data = reader.read(len(req))
    assert(get_query(data) == get_query(req))

    def to_int(bytes_):
        return int.from_bytes(bytes_, 'big')

    result = {}
    res_num = to_int(data[6:8])
    for i in range(res_num):
        reader.read(2)
        type_num = to_int(reader.read(2))

        type_ = None
        if type_num == 1:
            type_ = 'A'
        elif type_num == 5:
            type_ = 'CNAME'

        reader.read(6)
        data = reader.read(2)
        data = reader.read(to_int(data))
        add_record_to_result(result, type_, data, reader)

    return result


def dns_lookup(domain, address):
    dns_query = make_dns_query_domain(domain)
    dq_len = len(dns_query)

    req = make_dns_request_data(dns_query)
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(2)

    try:
        sock.sendto(req, (address, 53))
        res, _ = sock.recvfrom(1024 * 4)
        result = parse_dns_response(res, dq_len, req)
    except Exception:
        return
    finally:
        sock.close()

    return result


if __name__ == '__main__':
    print(dns_lookup('www.stackoverflow.com', "8.8.8.8"))

输出：

{'CNAME': ['stackoverflow.com'], 'A': ['151.101.1.69', '151.101.65.69', '151.101.129.69', '151.101.193.69']}