Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- ### SERVER ###
- from twisted.internet import defer, reactor
- from twisted.names import dns, error, server, common
- class MockDNSResolver(common.ResolverBase):
- def _lookup(self, name, cls, type, timeout):
- print("Incoming query for:", name)
- record = dns.Record_A(address=b"127.0.0.5")
- answer = dns.RRHeader(name=name, payload=record)
- return defer.succeed(([answer], [], []))
- if __name__ == "__main__":
- clients = [MockDNSResolver()]
- factory = server.DNSServerFactory(clients=clients)
- protocol = dns.DNSDatagramProtocol(controller=factory)
- reactor.listenUDP(10053, protocol)
- reactor.listenTCP(10053, factory)
- reactor.run()
- ### CLIENT ###
- from twisted.internet import reactor
- from twisted.names import client
- resolver = client.createResolver(servers=[('127.0.0.1', 10053)])
- d = resolver.getHostByName('does.not.exist')
- def callback(result):
- print('Result:', result)
- d.addBoth(callback)
- reactor.callLater(4, reactor.stop)
- reactor.run()
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement