PDNS Pipe

dns.py

#!/usr/bin/python3 -u

# Implementation of a Pipe Backend for PowerDNS
# https://doc.powerdns.com/md/authoritative/backend-pipe/

# We need Python unbuffered so we use -u
# Another ways to achieve it in
# http://stackoverflow.com/questions/107705/disable-output-buffering

import sys


def build_answer(qname, qtype, answer, ttl=60):
    return ("%s\tIN\t%s\t%d\t1\t%s" % (qname, qtype, ttl, answer))


def send_log(msg):
    sys.stdout.write("LOG\t%s\n" % msg)


def send_data(msg):
    sys.stdout.write("DATA\t%s\n" % msg)
    send_log("Sent %s" % msg)


# Check first line to ensure that we are properly initialized
line = sys.stdin.readline().rstrip()
if line != "HELO\t1":
    sys.stdout.write("FAIL\n")
    sys.exit(1)
sys.stdout.write("OK Python backend is alive!\n")

# Process each line!
while True:
    raw_line = sys.stdin.readline()
    line = raw_line.rstrip()
    args = line.split("\t")
    if len(args) < 6:
        send_log("PowerDNS sent me unparseable line")
        sys.stdout.write("FAIL\n")
        continue
    rtype, qname, qclass, qtype, id, ip = args
    send_log("Received [[ %s ]]" % line)
    # PDNS will use the SOA to decide which backend to use. We have to answer only
    # when the query is as close as possible:
    # i.e.: answer host.example.com not to example.com.
    if qtype == "SOA":
        send_data(build_answer(qname, 'SOA', 'rdns.as210934.me  2021081104 28800 7200 604800 86400'))
    if qtype in ("PTR", "ANY"):
        if 'ip6.arpa' in qname:
            answer = qname.replace('.ip6.arpa', '').replace('.', ' ').split()
            return_answer = ''
            for i in range(len(answer)):
                return_answer += answer[len(answer) - i - 1]
                if i % 4 == 3:
                    return_answer += "-"
            send_data(build_answer(qname, 'PTR', return_answer[:-1] + ".16clouds.com"))

    sys.stdout.write("END\n")