Skip to content

Instantly share code, notes, and snippets.

Created April 7, 2014 22:38
  • Star 7 You must be signed in to star a gist
  • Fork 5 You must be signed in to fork a gist
Star You must be signed in to star a gist
What would you like to do?
PowerDNS PipeBackend python example
#!/usr/bin/python -u
import sys, os, time
import random
class DNSLookup(object):
"""Handle PowerDNS pipe-backend domain name lookups."""
ttl = 30
def __init__(self, query):
"""parse DNS query and produce lookup result.
query: a sequence containing the DNS query as per PowerDNS manual appendix A:
(_type, qname, qclass, qtype, _id, ip) = query
self.has_result = False # has a DNS query response
qname_lower = qname.lower()
"""List of servers to round-robin"""
servers = ['','']
server = random.choice(servers)
self.results = []
if (qtype == 'A' or qtype == 'ANY') and qname_lower == '':
self.results.append('DATA\t%s\t%s\tA\t%d\t-1\t%s' % (qname, qclass, DNSLookup.ttl, server))
self.has_result = True
elif qtype == 'SOA' and qname_lower == '':
% (qname, qclass, qtype))
self.has_result = True
def str_result(self):
"""return string result suitable for pipe-backend output to PowerDNS."""
if self.has_result:
return '\n'.join(self.results)
return ''
class Logger(object):
def __init__(self):
pid = os.getpid()
self.logfile = '/tmp/backend.log'
"""self.logfile = '/tmp/backend-%d.log' % pid"""
def write(self, msg):
logline = '%s|%s\n' % (time.asctime(), msg)
f = file(self.logfile, 'a')
def debug_log(msg):
class PowerDNSbackend(object):
"""The main PowerDNS pipe backend process."""
def __init__(self, filein, fileout):
"""initialise and run PowerDNS pipe backend process."""
self.filein = filein
self.fileout = fileout
self._process_requests() # main program loop
def _process_requests(self):
"""main program loop"""
first_time = True
while 1:
rawline = self.filein.readline()
if rawline == '':
return # EOF detected
line = rawline.rstrip()
debug_log('received from pdns:%s' % line)
if first_time:
if line == 'HELO\t1':
self._fprint('OK\tPython backend firing up')
debug_log('HELO input not received - execution aborted')
rawline = self.filein.readline() # as per docs - read another line before aborting
debug_log('calling sys.exit()')
first_time = False
query = line.split('\t')
if len(query) != 6:
self._fprint('LOG\tPowerDNS sent unparseable line')
debug_log('Performing DNSLookup(%s)' % repr(query))
lookup = DNSLookup(query)
if lookup.has_result:
pdns_result = lookup.str_result()
debug_log('DNSLookup result(%s)' % pdns_result)
def _fprint(self, message):
"""Print the given message with newline and flushing."""
self.fileout.write(message + '\n')
debug_log('sent to pdns:%s' % message)
if __name__ == '__main__':
logger = Logger()
infile = sys.stdin
#outfile = os.fdopen(1, 'w', 1)
outfile = sys.stdout
PowerDNSbackend(infile, outfile)
debug_log('execution failure:' + str(sys.exc_info()[0]))
Copy link

arrfab commented Nov 8, 2017

Hi, simple question about this pdns backend : can you mention which license it has ? I'd like to reuse some part of it so have to ask first the correct license, as often forgotten on github :-)

Copy link

Ashark commented Aug 6, 2021

Could you please port it for python3 or use python2 explicitly in the shebang?
In python 3 there is no file(), I get such error when starting pdns:

File "/etc/powerdns/", line 118, in <module>
  debug_log('execution failure:' + str(sys.exc_info()[0]))
File "/etc/powerdns/", line 55, in debug_log
File "/etc/powerdns/", line 50, in write
  f = file(self.logfile, 'a')
NameError: name 'file' is not defined
[PIPEBackend] Unable to launch, fatal argument error: Child closed pipe

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment