+++ /dev/null
-# Domain Name Server (DNS) interface
-#
-# See RFC 1035:
-# ------------------------------------------------------------------------
-# Network Working Group P. Mockapetris
-# Request for Comments: 1035 ISI
-# November 1987
-# Obsoletes: RFCs 882, 883, 973
-#
-# DOMAIN NAMES - IMPLEMENTATION AND SPECIFICATION
-# ------------------------------------------------------------------------
-
-
-import string
-
-import dnstype
-import dnsclass
-import dnsopcode
-
-
-# Low-level 16 and 32 bit integer packing and unpacking
-
-def pack16bit(n):
- return chr((n>>8)&0xFF) + chr(n&0xFF)
-
-def pack32bit(n):
- return chr((n>>24)&0xFF) + chr((n>>16)&0xFF) \
- + chr((n>>8)&0xFF) + chr(n&0xFF)
-
-def unpack16bit(s):
- return (ord(s[0])<<8) | ord(s[1])
-
-def unpack32bit(s):
- return (ord(s[0])<<24) | (ord(s[1])<<16) \
- | (ord(s[2])<<8) | ord(s[3])
-
-def addr2bin(addr):
- if type(addr) == type(0):
- return addr
- bytes = string.splitfields(addr, '.')
- if len(bytes) != 4: raise ValueError, 'bad IP address'
- n = 0
- for byte in bytes: n = n<<8 | string.atoi(byte)
- return n
-
-def bin2addr(n):
- return '%d.%d.%d.%d' % ((n>>24)&0xFF, (n>>16)&0xFF,
- (n>>8)&0xFF, n&0xFF)
-
-
-# Packing class
-
-class Packer:
- def __init__(self):
- self.buf = ''
- self.index = {}
- def getbuf(self):
- return self.buf
- def addbyte(self, c):
- if len(c) != 1: raise TypeError, 'one character expected'
- self.buf = self.buf + c
- def addbytes(self, bytes):
- self.buf = self.buf + bytes
- def add16bit(self, n):
- self.buf = self.buf + pack16bit(n)
- def add32bit(self, n):
- self.buf = self.buf + pack32bit(n)
- def addaddr(self, addr):
- n = addr2bin(addr)
- self.buf = self.buf + pack32bit(n)
- def addstring(self, s):
- self.addbyte(chr(len(s)))
- self.addbytes(s)
- def addname(self, name):
- # Domain name packing (section 4.1.4)
- # Add a domain name to the buffer, possibly using pointers.
- # The case of the first occurrence of a name is preserved.
- # Redundant dots are ignored.
- list = []
- for label in string.splitfields(name, '.'):
- if label:
- if len(label) > 63:
- raise PackError, 'label too long'
- list.append(label)
- keys = []
- for i in range(len(list)):
- key = string.upper(string.joinfields(list[i:], '.'))
- keys.append(key)
- if self.index.has_key(key):
- pointer = self.index[key]
- break
- else:
- i = len(list)
- pointer = None
- # Do it into temporaries first so exceptions don't
- # mess up self.index and self.buf
- buf = ''
- offset = len(self.buf)
- index = []
- for j in range(i):
- label = list[j]
- n = len(label)
- if offset + len(buf) < 0x3FFF:
- index.append( (keys[j], offset + len(buf)) )
- else:
- print 'dnslib.Packer.addname:',
- print 'warning: pointer too big'
- buf = buf + (chr(n) + label)
- if pointer:
- buf = buf + pack16bit(pointer | 0xC000)
- else:
- buf = buf + '\0'
- self.buf = self.buf + buf
- for key, value in index:
- self.index[key] = value
- def dump(self):
- keys = self.index.keys()
- keys.sort()
- print '-'*40
- for key in keys:
- print '%20s %3d' % (key, self.index[key])
- print '-'*40
- space = 1
- for i in range(0, len(self.buf)+1, 2):
- if self.buf[i:i+2] == '**':
- if not space: print
- space = 1
- continue
- space = 0
- print '%4d' % i,
- for c in self.buf[i:i+2]:
- if ' ' < c < '\177':
- print ' %c' % c,
- else:
- print '%2d' % ord(c),
- print
- print '-'*40
-
-
-# Unpacking class
-
-UnpackError = 'dnslib.UnpackError' # Exception
-
-class Unpacker:
- def __init__(self, buf):
- self.buf = buf
- self.offset = 0
- def getbyte(self):
- c = self.buf[self.offset]
- self.offset = self.offset + 1
- return c
- def getbytes(self, n):
- s = self.buf[self.offset : self.offset + n]
- if len(s) != n: raise UnpackError, 'not enough data left'
- self.offset = self.offset + n
- return s
- def get16bit(self):
- return unpack16bit(self.getbytes(2))
- def get32bit(self):
- return unpack32bit(self.getbytes(4))
- def getaddr(self):
- return bin2addr(self.get32bit())
- def getstring(self):
- return self.getbytes(ord(self.getbyte()))
- def getname(self):
- # Domain name unpacking (section 4.1.4)
- c = self.getbyte()
- i = ord(c)
- if i & 0xC0 == 0xC0:
- d = self.getbyte()
- j = ord(d)
- pointer = ((i<<8) | j) & ~0xC000
- save_offset = self.offset
- try:
- self.offset = pointer
- domain = self.getname()
- finally:
- self.offset = save_offset
- return domain
- if i == 0:
- return ''
- domain = self.getbytes(i)
- remains = self.getname()
- if not remains:
- return domain
- else:
- return domain + '.' + remains
-
-
-# Test program for packin/unpacking (section 4.1.4)
-
-def testpacker():
- N = 25
- R = range(N)
- import timing
- # See section 4.1.4 of RFC 1035
- timing.start()
- for i in R:
- p = Packer()
- p.addbytes('*' * 20)
- p.addname('f.ISI.ARPA')
- p.addbytes('*' * 8)
- p.addname('Foo.F.isi.arpa')
- p.addbytes('*' * 18)
- p.addname('arpa')
- p.addbytes('*' * 26)
- p.addname('')
- timing.finish()
- print round(timing.milli() * 0.001 / N, 3), 'seconds per packing'
- p.dump()
- u = Unpacker(p.buf)
- u.getbytes(20)
- u.getname()
- u.getbytes(8)
- u.getname()
- u.getbytes(18)
- u.getname()
- u.getbytes(26)
- u.getname()
- timing.start()
- for i in R:
- u = Unpacker(p.buf)
- res = (u.getbytes(20),
- u.getname(),
- u.getbytes(8),
- u.getname(),
- u.getbytes(18),
- u.getname(),
- u.getbytes(26),
- u.getname())
- timing.finish()
- print round(timing.milli() * 0.001 / N, 3), 'seconds per unpacking'
- for item in res: print item
-
-
-# Pack/unpack RR toplevel format (section 3.2.1)
-
-class RRpacker(Packer):
- def __init__(self):
- Packer.__init__(self)
- self.rdstart = None
- def addRRheader(self, name, type, klass, ttl, *rest):
- self.addname(name)
- self.add16bit(type)
- self.add16bit(klass)
- self.add32bit(ttl)
- if rest:
- if res[1:]: raise TypeError, 'too many args'
- rdlength = rest[0]
- else:
- rdlength = 0
- self.add16bit(rdlength)
- self.rdstart = len(self.buf)
- def patchrdlength(self):
- rdlength = unpack16bit(self.buf[self.rdstart-2:self.rdstart])
- if rdlength == len(self.buf) - self.rdstart:
- return
- rdata = self.buf[self.rdstart:]
- save_buf = self.buf
- ok = 0
- try:
- self.buf = self.buf[:self.rdstart-2]
- self.add16bit(len(rdata))
- self.buf = self.buf + rdata
- ok = 1
- finally:
- if not ok: self.buf = save_buf
- def endRR(self):
- if self.rdstart is not None:
- self.patchrdlength()
- self.rdstart = None
- def getbuf(self):
- if self.rdstart is not None: self.patchrdlenth()
- return Packer.getbuf(self)
- # Standard RRs (section 3.3)
- def addCNAME(self, name, klass, ttl, cname):
- self.addRRheader(name, dnstype.CNAME, klass, ttl)
- self.addname(cname)
- self.endRR()
- def addHINFO(self, name, klass, ttl, cpu, os):
- self.addRRheader(name, dnstype.HINFO, klass, ttl)
- self.addstring(cpu)
- self.addstring(os)
- self.endRR()
- def addMX(self, name, klass, ttl, preference, exchange):
- self.addRRheader(name, dnstype.MX, klass, ttl)
- self.add16bit(preference)
- self.addname(exchange)
- self.endRR()
- def addNS(self, name, klass, ttl, nsdname):
- self.addRRheader(name, dnstype.NS, klass, ttl)
- self.addname(nsdname)
- self.endRR()
- def addPTR(self, name, klass, ttl, ptrdname):
- self.addRRheader(name, dnstype.PTR, klass, ttl)
- self.addname(ptrdname)
- self.endRR()
- def addSOA(self, name, klass, ttl,
- mname, rname, serial, refresh, retry, expire, minimum):
- self.addRRheader(name, dnstype.SOA, klass, ttl)
- self.addname(mname)
- self.addname(rname)
- self.add32bit(serial)
- self.add32bit(refresh)
- self.add32bit(retry)
- self.add32bit(expire)
- self.add32bit(minimum)
- self.endRR()
- def addTXT(self, name, klass, ttl, list):
- self.addRRheader(name, dnstype.TXT, klass, ttl)
- for txtdata in list:
- self.addstring(txtdata)
- self.endRR()
- # Internet specific RRs (section 3.4) -- class = IN
- def addA(self, name, ttl, address):
- self.addRRheader(name, dnstype.A, dnsclass.IN, ttl)
- self.addaddr(address)
- self.endRR()
- def addWKS(self, name, ttl, address, protocol, bitmap):
- self.addRRheader(name, dnstype.WKS, dnsclass.IN, ttl)
- self.addaddr(address)
- self.addbyte(chr(protocol))
- self.addbytes(bitmap)
- self.endRR()
-
-
-class RRunpacker(Unpacker):
- def __init__(self, buf):
- Unpacker.__init__(self, buf)
- self.rdend = None
- def getRRheader(self):
- name = self.getname()
- type = self.get16bit()
- klass = self.get16bit()
- ttl = self.get32bit()
- rdlength = self.get16bit()
- self.rdend = self.offset + rdlength
- return (name, type, klass, ttl, rdlength)
- def endRR(self):
- if self.offset != self.rdend:
- raise UnpackError, 'end of RR not reached'
- def getCNAMEdata(self):
- return self.getname()
- def getHINFOdata(self):
- return self.getstring(), self.getstring()
- def getMXdata(self):
- return self.get16bit(), self.getname()
- def getNSdata(self):
- return self.getname()
- def getPTRdata(self):
- return self.getname()
- def getSOAdata(self):
- return self.getname(), \
- self.getname(), \
- self.get32bit(), \
- self.get32bit(), \
- self.get32bit(), \
- self.get32bit(), \
- self.get32bit()
- def getTXTdata(self):
- list = []
- while self.offset != self.rdend:
- list.append(self.getstring())
- return list
- def getAdata(self):
- return self.getaddr()
- def getWKSdata(self):
- address = self.getaddr()
- protocol = ord(self.getbyte())
- bitmap = self.getbytes(self.rdend - self.offset)
- return address, protocol, bitmap
-
-
-# Pack/unpack Message Header (section 4.1)
-
-class Hpacker(Packer):
- def addHeader(self, id, qr, opcode, aa, tc, rd, ra, z, rcode,
- qdcount, ancount, nscount, arcount):
- self.add16bit(id)
- self.add16bit((qr&1)<<15 | (opcode*0xF)<<11 | (aa&1)<<10
- | (tc&1)<<9 | (rd&1)<<8 | (ra&1)<<7
- | (z&7)<<4 | (rcode&0xF))
- self.add16bit(qdcount)
- self.add16bit(ancount)
- self.add16bit(nscount)
- self.add16bit(arcount)
-
-class Hunpacker(Unpacker):
- def getHeader(self):
- id = self.get16bit()
- flags = self.get16bit()
- qr, opcode, aa, tc, rd, ra, z, rcode = (
- (flags>>15)&1,
- (flags>>11)&0xF,
- (flags>>10)&1,
- (flags>>9)&1,
- (flags>>8)&1,
- (flags>>7)&1,
- (flags>>4)&7,
- (flags>>0)&0xF)
- qdcount = self.get16bit()
- ancount = self.get16bit()
- nscount = self.get16bit()
- arcount = self.get16bit()
- return (id, qr, opcode, aa, tc, rd, ra, z, rcode,
- qdcount, ancount, nscount, arcount)
-
-
-# Pack/unpack Question (section 4.1.2)
-
-class Qpacker(Packer):
- def addQuestion(self, qname, qtype, qclass):
- self.addname(qname)
- self.add16bit(qtype)
- self.add16bit(qclass)
-
-class Qunpacker(Unpacker):
- def getQuestion(self):
- return self.getname(), self.get16bit(), self.get16bit()
-
-
-# Pack/unpack Message(section 4)
-# NB the order of the base classes is important for __init__()!
-
-class Mpacker(RRpacker, Qpacker, Hpacker):
- pass
-
-class Munpacker(RRunpacker, Qunpacker, Hunpacker):
- pass
-
-
-# Routines to print an unpacker to stdout, for debugging.
-# These affect the unpacker's current position!
-
-def dumpM(u):
- print 'HEADER:',
- (id, qr, opcode, aa, tc, rd, ra, z, rcode,
- qdcount, ancount, nscount, arcount) = u.getHeader()
- print 'id=%d,' % id,
- print 'qr=%d, opcode=%d, aa=%d, tc=%d, rd=%d, ra=%d, z=%d, rcode=%d,' \
- % (qr, opcode, aa, tc, rd, ra, z, rcode)
- if tc: print '*** response truncated! ***'
- if rcode: print '*** nonzero error code! (%d) ***' % rcode
- print ' qdcount=%d, ancount=%d, nscount=%d, arcount=%d' \
- % (qdcount, ancount, nscount, arcount)
- for i in range(qdcount):
- print 'QUESTION %d:' % i,
- dumpQ(u)
- for i in range(ancount):
- print 'ANSWER %d:' % i,
- dumpRR(u)
- for i in range(nscount):
- print 'AUTHORITY RECORD %d:' % i,
- dumpRR(u)
- for i in range(arcount):
- print 'ADDITIONAL RECORD %d:' % i,
- dumpRR(u)
-
-def dumpQ(u):
- qname, qtype, qclass = u.getQuestion()
- print 'qname=%s, qtype=%d(%s), qclass=%d(%s)' \
- % (qname,
- qtype, dnstype.typestr(qtype),
- qclass, dnsclass.classstr(qclass))
-
-def dumpRR(u):
- name, type, klass, ttl, rdlength = u.getRRheader()
- typename = dnstype.typestr(type)
- print 'name=%s, type=%d(%s), class=%d(%s), ttl=%d' \
- % (name,
- type, typename,
- klass, dnsclass.classstr(klass),
- ttl)
- mname = 'get%sdata' % typename
- if hasattr(u, mname):
- print ' formatted rdata:', getattr(u, mname)()
- else:
- print ' binary rdata:', u.getbytes(rdlength)
-
-
-# Test program
-
-def test():
- import sys
- import getopt
- import socket
- protocol = 'udp'
- server = 'cnri.reston.va.us' # XXX adapt this to your local
- port = 53
- opcode = dnsopcode.QUERY
- rd = 0
- qtype = dnstype.MX
- qname = 'cwi.nl'
- try:
- opts, args = getopt.getopt(sys.argv[1:], 'Trs:tu')
- if len(args) > 2: raise getopt.error, 'too many arguments'
- except getopt.error, msg:
- print msg
- print 'Usage: python dnslib.py',
- print '[-T] [-r] [-s server] [-t] [-u]',
- print '[qtype [qname]]'
- print '-T: run testpacker() and exit'
- print '-r: recursion desired (default not)'
- print '-s server: use server (default %s)' % server
- print '-t: use TCP protocol'
- print '-u: use UDP protocol (default)'
- print 'qtype: query type (default %s)' % \
- dnstype.typestr(qtype)
- print 'qname: query name (default %s)' % qname
- print 'Recognized qtype values:'
- qtypes = dnstype.typemap.keys()
- qtypes.sort()
- n = 0
- for qtype in qtypes:
- n = n+1
- if n >= 8: n = 1; print
- print '%s = %d' % (dnstype.typemap[qtype], qtype),
- print
- sys.exit(2)
- for o, a in opts:
- if o == '-T': testpacker(); return
- if o == '-t': protocol = 'tcp'
- if o == '-u': protocol = 'udp'
- if o == '-s': server = a
- if o == '-r': rd = 1
- if args[0:]:
- try:
- qtype = eval(string.upper(args[0]), dnstype.__dict__)
- except (NameError, SyntaxError):
- print 'bad query type:', `args[0]`
- sys.exit(2)
- if args[1:]:
- qname = args[1]
- if qtype == dnstype.AXFR:
- print 'Query type AXFR, protocol forced to TCP'
- protocol = 'tcp'
- print 'QTYPE %d(%s)' % (qtype, dnstype.typestr(qtype))
- m = Mpacker()
- m.addHeader(0,
- 0, opcode, 0, 0, rd, 0, 0, 0,
- 1, 0, 0, 0)
- m.addQuestion(qname, qtype, dnsclass.IN)
- request = m.getbuf()
- if protocol == 'udp':
- s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
- s.connect((server, port))
- s.send(request)
- reply = s.recv(1024)
- else:
- s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
- s.connect((server, port))
- s.send(pack16bit(len(request)) + request)
- s.shutdown(1)
- f = s.makefile('r')
- header = f.read(2)
- if len(header) < 2:
- print '*** EOF ***'
- return
- count = unpack16bit(header)
- reply = f.read(count)
- if len(reply) != count:
- print '*** Incomplete reply ***'
- return
- u = Munpacker(reply)
- dumpM(u)
- if protocol == 'tcp' and qtype == dnstype.AXFR:
- while 1:
- header = f.read(2)
- if len(header) < 2:
- print '========== EOF =========='
- break
- count = unpack16bit(header)
- if not count:
- print '========== ZERO COUNT =========='
- break
- print '========== NEXT =========='
- reply = f.read(count)
- if len(reply) != count:
- print '*** Incomplete reply ***'
- break
- u = Munpacker(reply)
- dumpM(u)
-
-
-# Run test program when called as a script
-
-if __name__ == '__main__':
- test()
projects / python / commitdiff