qwebirc/dns.py

   1 from twisted.names import client
   2 from twisted.internet import reactor, defer
   3
   4 class LookupException(Exception): pass
   5 class VerificationException(Exception): pass
   6 TimeoutException = defer.TimeoutError
   7
   8 def lookupPTR(ip, *args, **kwargs):
   9   def callback(result):
  10     answer, auth, add = result
  11
  12     if len(answer) == 0:
  13       raise LookupException, "No ANSWERS in PTR response for %s." % repr(ip)
  14     return str(answer[0].payload.name)
  15
  16   ptr = ".".join(ip.split(".")[::-1]) + ".in-addr.arpa."
  17   return client.lookupPointer(ptr, **kwargs).addCallback(callback)
  18
  19 def lookupAs(hostname, *args, **kwargs):
  20   def callback(result):
  21     answer, auth, add = result
  22     if len(answer) == 0:
  23       raise LookupException, "No ANSWERS in A response for %s." % repr(hostname)
  24     return [x.payload.dottedQuad() for x in answer]
  25
  26   return client.lookupAddress(hostname, *args, **kwargs).addCallback(callback)
  27
  28 def lookupAndVerifyPTR(ip, *args, **kwargs):
  29   d = defer.Deferred()
  30
  31   def gotPTRResult(ptr):
  32     def gotAResult(a_records):
  33       if ip in a_records:
  34         d.callback(ptr)
  35       else:
  36         raise VerificationException("IP mismatch: %s != %s%s" % (repr(ip), repr(ptr), repr(a_records)))
  37     lookupAs(ptr, *args, **kwargs).addCallback(gotAResult).addErrback(d.errback)
  38
  39   lookupPTR(ip, *args, **kwargs).addCallback(gotPTRResult).addErrback(d.errback)
  40   return d
  41
  42 if __name__ == "__main__":
  43   import sys
  44
  45   def callback(x):
  46     print x
  47     reactor.stop()
  48
  49   def errback(x):
  50     x.printTraceback()
  51     reactor.stop()
  52
  53   d = lookupAndVerifyPTR(sys.argv[1], timeout=[.001])
  54   d.addCallbacks(callback, errback)
  55
  56   reactor.run()