Filter AAAA record replies with unbound

response-filter.py

AAA_BLOCKLIST = [
    # ADD domains here
]


def init_standard(id, env):
    log_info(
        f"response-filter: AAAA response filter initialized; blocked_urls={AAAA_BLOCKLIST!r}"
    )
    return True


def deinit(id):
    return True


def inform_super(id, qstate, superqstate, qdata):
    return True


def operate(id, event, qstate, qdata):
    log_query_info(
        VERB_QUERY, f"response-filter: operate called; id={id}", qstate.qinfo
    )

    if (event == MODULE_EVENT_NEW) or (event == MODULE_EVENT_PASS):
        if qstate.qinfo.qtype == RR_TYPE_AAAA and any(
            qstate.qinfo.qname_str.endswith(blocked) for blocked in AAAA_BLOCKLIST
        ):
            # RFC 4074 - 3. Expected Behavior
            #
            # Suppose that an authoritative server has an A RR but has no AAAA RR
            # for a host name.  Then, the server should return a response to a
            # query for an AAAA RR of the name with the response code (RCODE) being
            # 0 (indicating no error) and with an empty answer section (see
            # Sections 4.3.2 and 6.2.4 of [1]).  Such a response indicates that
            # there is at least one RR of a different type than AAAA for the
            # queried name, and the stub resolver can then look for A RRs.
            msg = DNSMessage(
                qstate.qinfo.qname_str, RR_TYPE_A, RR_CLASS_IN, PKT_QR | PKT_RA | PKT_AA
            )

            success = msg.set_return_msg(qstate)
            if not success:
                qstate.ext_state[id] = MODULE_ERROR
                log_err(
                    f"response-filter: cannot block AAAA request, failed to set response"
                )
                return False

            # We don't need validation, result is valid
            msg.security = 2
            qstate.return_rcode = RCODE_NOERROR
            qstate.ext_state[id] = MODULE_FINISHED
            log_info(
                f"response-filter: blocking AAAA request; qname={qstate.qinfo.qname_str!r}"
            )
            return True
        else:
            # Pass the query to validator
            qstate.ext_state[id] = MODULE_WAIT_MODULE
            return True

    if event == MODULE_EVENT_MODDONE:
        qstate.ext_state[id] = MODULE_FINISHED
        return True

    qstate.ext_state[id] = MODULE_ERROR
    return True

unbound.conf

server:
    directory: ""
    username: ""
    # make sure unbound can access entropy from inside the chroot.
    # e.g. on linux the use these commands (on BSD, devfs(8) is used):
    #      mount --bind -n /dev/random /etc/unbound/dev/random
    # and  mount --bind -n /dev/log /etc/unbound/dev/log
    chroot: ""
    pidfile: ""
    # verbosity: 1        # uncomment and increase to get more logging.
    # listen on all interfaces, answer queries from the local subnet.
    interface: 127.0.0.50
    interface: ::1
    access-control: 127.0.0.0/8 allow
    access-control: ::1/128 allow
    logfile: ""   # Empty string means STDOUT
    verbosity: 2  # Uncomment and increase to get more logging.
    hide-identity: yes
    hide-version: yes

    module-config: "python iterator"

python:
    python-script: "./response-filter.py"

forward-zone:
    name: "."
    forward-addr: xxx.xxx.xxx.xxx