dnsPython 中的 TXT 记录
我已经实现了一个简单的 DNS 服务器。 它只是用 TXT 记录进行响应。我正在主持 脚本作为 example.com 的 NS 服务器。 NS服务器是xyzk它 当我发出类似以下内容时工作正常:
$ dig demo.example.com @xyzk
; <<>> DiG 9.3.6-P1-RedHat-9.3.6-4.P1.el5_4.1 <<>> demo.example.com @x.y.z.k
;; global options: printcmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 10028
;; flags: qr rd ra; QUERY: 1, ANSWER: 1, AUTHORITY: 2, ADDITIONAL: 2
;; QUESTION SECTION:
;demo.example.com. IN A
;; ANSWER SECTION:
demo.example.com. 1000 IN TXT "test message"
但当我向其中一个提出完全相同的问题时它不起作用 公共名称服务器(例如 Sun 的 DNS,位于 4.2.2.1):
$ dig demo.example.com @4.2.2.1
(当我发出 $ dig demo.example 时,我得到同样的结果。 com @4.2.2.1 TXT
)
; <<>> DiG 9.3.6-P1-RedHat-9.3.6-4.P1.el5_4.1 <<>> demo.example.com
;; global options: printcmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: SERVFAIL, id: 10905
;; flags: qr rd ra; QUERY: 1, ANSWER: 0, AUTHORITY: 0, ADDITIONAL: 0
;; QUESTION SECTION:
;demo.example.com. IN A
;; Query time: 5837 msec
;; SERVER: 172.16.1.1#53(172.16.1.1)
;; WHEN: Tue Jul 20 04:01:27 2010
;; MSG SIZE rcvd: 75
你们知道出了什么问题吗?有趣的是如果我改变 响应类型为 CNAME,而不是 TXT, 效果很好。
def dns_respond(resp, q, data_type):
rrset = dns.rrset.from_text(q.name, 1000,
dns.rdataclass.IN, dns.rdatatype.TXT, 'test message')
resp.answer.append(rrset)
return resp
def dns_ok(resp, q, data = None, msg = ''):
return dns_respond(resp = resp, q = q, data_type = 'OK')
def dns_error(resp, q):
return dns_respond(resp = resp, q = q, data_type = 'Error')
def requestHandler(address, message):
resp = None
message_id = ord(message[0]) * 256 + ord(message[1])
logging.debug('msg id = ' + str(message_id))
if message_id in serving_ids:
# the request is already taken, drop this message
logging.debug('I am already serving this request.')
return
serving_ids.append(message_id)
msg = dns.message.from_wire(message)
op = msg.opcode()
if op == 0:
# standard and inverse query
qs = msg.question
if len(qs) > 0:
q = qs[0]
logging.debug('request is ' + str(q))
if q.rdtype == dns.rdatatype.A:
resp = std_qry(msg)
else:
# not implemented
#resp = std_qry(msg)
resp = make_response(qry=msg, RCODE=4)
else:
# not implemented
resp = make_response(qry=msg, RCODE=4)
if resp:
s.sendto(resp.to_wire(), address)
def std_qry(msg):
qs = msg.question
logging.debug(str(len(qs)) + ' questions.')
answers = []
nxdomain = False
for q in qs:
resp = make_response(qry=msg)
logging.critical('Sending...')
return dns_ok(resp, q)
def make_response(qry=None, id=None, RCODE=0):
if qry is None and id is None:
raise Exception, 'bad use of make_response'
if qry is None:
resp = dns.message.Message(id)
# QR = 1
resp.flags |= dns.flags.QR
if RCODE != 1:
raise Exception, 'bad use of make_response'
else:
resp = dns.message.make_response(qry)
#resp.flags |= dns.flags.AA
resp.flags |= dns.flags.RA
resp.set_rcode(RCODE)
return resp
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', 53))
logging.debug('binded to UDP port 53.')
serving_ids = []
while True:
logging.debug('waiting requests.')
message, address = s.recvfrom(1024)
logging.debug('serving a request.')
requestHandler(address, message)
I have implemented a simple DNS server.
It simply responds with a TXT record. I am hosting the
script as the NS server for example.com. The NS server is x.y.z.k. It
works fine when I issue something like:
$ dig demo.example.com @x.y.z.k
; <<>> DiG 9.3.6-P1-RedHat-9.3.6-4.P1.el5_4.1 <<>> demo.example.com @x.y.z.k
;; global options: printcmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: NOERROR, id: 10028
;; flags: qr rd ra; QUERY: 1, ANSWER: 1, AUTHORITY: 2, ADDITIONAL: 2
;; QUESTION SECTION:
;demo.example.com. IN A
;; ANSWER SECTION:
demo.example.com. 1000 IN TXT "test message"
but it does not work when I ask the very same question from one of the
public name servers (For example Sun's DNS at 4.2.2.1):
$ dig demo.example.com @4.2.2.1
(I get the same thing when I issue $ dig demo.example.com @4.2.2.1 TXT
)
; <<>> DiG 9.3.6-P1-RedHat-9.3.6-4.P1.el5_4.1 <<>> demo.example.com
;; global options: printcmd
;; Got answer:
;; ->>HEADER<<- opcode: QUERY, status: SERVFAIL, id: 10905
;; flags: qr rd ra; QUERY: 1, ANSWER: 0, AUTHORITY: 0, ADDITIONAL: 0
;; QUESTION SECTION:
;demo.example.com. IN A
;; Query time: 5837 msec
;; SERVER: 172.16.1.1#53(172.16.1.1)
;; WHEN: Tue Jul 20 04:01:27 2010
;; MSG SIZE rcvd: 75
Do you guys have any idea what is wrong? Interestingly if I change
the response type to something like CNAME, instead of TXT,
it works fine.
def dns_respond(resp, q, data_type):
rrset = dns.rrset.from_text(q.name, 1000,
dns.rdataclass.IN, dns.rdatatype.TXT, 'test message')
resp.answer.append(rrset)
return resp
def dns_ok(resp, q, data = None, msg = ''):
return dns_respond(resp = resp, q = q, data_type = 'OK')
def dns_error(resp, q):
return dns_respond(resp = resp, q = q, data_type = 'Error')
def requestHandler(address, message):
resp = None
message_id = ord(message[0]) * 256 + ord(message[1])
logging.debug('msg id = ' + str(message_id))
if message_id in serving_ids:
# the request is already taken, drop this message
logging.debug('I am already serving this request.')
return
serving_ids.append(message_id)
msg = dns.message.from_wire(message)
op = msg.opcode()
if op == 0:
# standard and inverse query
qs = msg.question
if len(qs) > 0:
q = qs[0]
logging.debug('request is ' + str(q))
if q.rdtype == dns.rdatatype.A:
resp = std_qry(msg)
else:
# not implemented
#resp = std_qry(msg)
resp = make_response(qry=msg, RCODE=4)
else:
# not implemented
resp = make_response(qry=msg, RCODE=4)
if resp:
s.sendto(resp.to_wire(), address)
def std_qry(msg):
qs = msg.question
logging.debug(str(len(qs)) + ' questions.')
answers = []
nxdomain = False
for q in qs:
resp = make_response(qry=msg)
logging.critical('Sending...')
return dns_ok(resp, q)
def make_response(qry=None, id=None, RCODE=0):
if qry is None and id is None:
raise Exception, 'bad use of make_response'
if qry is None:
resp = dns.message.Message(id)
# QR = 1
resp.flags |= dns.flags.QR
if RCODE != 1:
raise Exception, 'bad use of make_response'
else:
resp = dns.message.make_response(qry)
#resp.flags |= dns.flags.AA
resp.flags |= dns.flags.RA
resp.set_rcode(RCODE)
return resp
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', 53))
logging.debug('binded to UDP port 53.')
serving_ids = []
while True:
logging.debug('waiting requests.')
message, address = s.recvfrom(1024)
logging.debug('serving a request.')
requestHandler(address, message)
如果你对这篇内容有疑问,欢迎到本站社区发帖提问 参与讨论,获取更多帮助,或者扫码二维码加入 Web 技术交流群。
绑定邮箱获取回复消息
由于您还没有绑定你的真实邮箱,如果其他用户或者作者回复了您的评论,将不能在第一时间通知您!
发布评论
评论(1)
我可以看到一些真正的错误会阻止此工作:
TXT
记录返回到A
记录查询 - 这可能是表演的阻碍。AA
标志(权威答案)而不是RA
此外,还有一些与协议相关的问题需要修复:
NS
记录和SOA< /code> 询问时记录,否则您的代表团可能无法可靠地工作
NXDOMAIN
(rcode = 3)NOERROR
(rcode = 0),而不是NOTIMPL
REFUSED
(rcode = 5)I can see a few real bugs that'll stop this working:
TXT
record to anA
record query - this is probably the show stopper.AA
flag (authoritative answer) instead ofRA
In addition there are a few protocol-related issues that need fixing:
NS
records andSOA
records when asked, or your delegation may not work reliablyNXDOMAIN
(rcode = 3)NOERROR
(rcode = 0) instead ofNOTIMPL
REFUSED
(rcode = 5)