kea/src/bin/xfrin/xfrin.py.in

#!@PYTHON@

# Copyright (C) 2009-2013  Internet Systems Consortium.
#
# Permission to use, copy, modify, and distribute this software for any
# purpose with or without fee is hereby granted, provided that the above
# copyright notice and this permission notice appear in all copies.
#
# THE SOFTWARE IS PROVIDED "AS IS" AND INTERNET SYSTEMS CONSORTIUM
# DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL
# INTERNET SYSTEMS CONSORTIUM BE LIABLE FOR ANY SPECIAL, DIRECT,
# INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
# FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
# NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
# WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

import sys; sys.path.append ('@@PYTHONPATH@@')
import os
import signal
import isc
import asyncore
import struct
import threading
import socket
import random
import time
from functools import reduce
from optparse import OptionParser, OptionValueError
from isc.config.ccsession import *
from isc.statistics import Counters
from isc.notify import notify_out
import isc.util.process
from isc.util.address_formatter import AddressFormatter
from isc.datasrc import DataSourceClient, ZoneFinder
import isc.net.parse
from isc.xfrin.diff import Diff
from isc.server_common.auth_command import auth_loadzone_command
from isc.server_common.tsig_keyring import init_keyring, get_keyring
from isc.server_common.datasrc_clients_mgr import DataSrcClientsMgr, ConfigError
from isc.log_messages.xfrin_messages import *
from isc.dns import *

isc.log.init("b10-xfrin", buffer=True)
logger = isc.log.Logger("xfrin")

# Pending system-wide debug level definitions, the ones we
# use here are hardcoded for now
DBG_PROCESS = logger.DBGLVL_TRACE_BASIC
DBG_COMMANDS = logger.DBGLVL_TRACE_DETAIL

isc.util.process.rename()

# If B10_FROM_BUILD or B10_FROM_SOURCE is set in the environment, we
# use data files from a directory relative to that, otherwise we use
# the ones installed on the system
SPECFILE_PATH = "@datadir@/@PACKAGE@"\
    .replace("${datarootdir}", "@datarootdir@")\
    .replace("${prefix}", "@prefix@")
if "B10_FROM_SOURCE" in os.environ:
    SPECFILE_PATH = os.environ["B10_FROM_SOURCE"] + "/src/bin/xfrin"
SPECFILE_LOCATION = SPECFILE_PATH + "/xfrin.spec"

AUTH_MODULE_NAME = 'Auth'
XFROUT_MODULE_NAME = 'Xfrout'

# Remote module and identifiers (according to their spec files)
ZONE_MANAGER_MODULE_NAME = 'Zonemgr'

REFRESH_FROM_ZONEMGR = 'refresh_from_zonemgr'

# Constants for debug levels.
DBG_XFRIN_TRACE = logger.DBGLVL_TRACE_BASIC

# These two default are currently hard-coded. For config this isn't
# necessary, but we need these defaults for optional command arguments
# (TODO: have similar support to get default values for command
# arguments as we do for config options)
DEFAULT_MASTER_PORT = 53
DEFAULT_ZONE_CLASS = RRClass.IN

__version__ = 'BIND10'

# Internal result codes of an xfr session
XFRIN_OK = 0                    # normal success
XFRIN_FAIL = 1                  # general failure (internal/external)

class XfrinException(Exception):
    pass

class XfrinProtocolError(Exception):
    '''An exception raised for errors encountered in xfrin protocol handling.
    '''
    pass

class XfrinZoneError(Exception):
    '''
    An exception raised when the received zone is broken enough to be unusable.
    '''
    pass

class XfrinZoneUptodate(Exception):
    '''
    Thrown when the zone is already up to date, so there's no need to download
    the zone. This is not really an error case (but it's still an exceptional
    condition and the control flow is different than usual).
    '''
    pass

class XfrinZoneInfoException(Exception):
    """This exception is raised if there is an error in the given
       configuration (part), or when a command does not have a required
       argument or has bad arguments, for instance when the zone's master
       address is not a valid IP address, when the zone does not
       have a name, or when multiple settings are given for the same
       zone."""
    pass

def _check_zone_name(zone_name_str):
    """Checks if the given zone name is a valid domain name, and returns
    it as a Name object. Raises an XfrinException if it is not."""
    try:
        # In the _zones dict, part of the key is the zone name,
        # but due to a limitation in the Name class, we
        # cannot directly use it as a dict key, and we use to_text()
        #
        # Downcase the name here for that reason.
        return Name(zone_name_str, True)
    except (EmptyLabel, TooLongLabel, BadLabelType, BadEscape,
            TooLongName, IncompleteName) as ne:
        raise XfrinZoneInfoException("bad zone name: " + zone_name_str + " (" + str(ne) + ")")

def _check_zone_class(zone_class_str):
    """If the given argument is a string: checks if the given class is
       a valid one, and returns an RRClass object if so.
       Raises XfrinZoneInfoException if not.
       If it is None, this function returns the default RRClass.IN"""
    if zone_class_str is None:
        return DEFAULT_ZONE_CLASS
    try:
        return RRClass(zone_class_str)
    except InvalidRRClass as irce:
        raise XfrinZoneInfoException("bad zone class: " + zone_class_str + " (" + str(irce) + ")")

def format_zone_str(zone_name, zone_class):
    """Helper function to format a zone name and class as a string of
       the form '<name>/<class>'.
       Parameters:
       zone_name (isc.dns.Name) name to format
       zone_class (isc.dns.RRClass) class to format
    """
    return zone_name.to_text(True) + '/' + str(zone_class)

def format_addrinfo(addrinfo):
    """Helper function to format the addrinfo as a string of the form
       <addr>:<port> (for IPv4) or [<addr>]:port (for IPv6). For unix domain
       sockets, and unknown address families, it returns a basic string
       conversion of the third element of the passed tuple.
       Parameters:
       addrinfo: a 3-tuple consisting of address family, socket type, and,
                 depending on the family, either a 2-tuple with the address
                 and port, or a filename
    """
    try:
        if addrinfo[0] == socket.AF_INET:
            return str(addrinfo[2][0]) + ":" + str(addrinfo[2][1])
        elif addrinfo[0] == socket.AF_INET6:
            return "[" + str(addrinfo[2][0]) + "]:" + str(addrinfo[2][1])
        else:
            return str(addrinfo[2])
    except IndexError:
        raise TypeError("addrinfo argument to format_addrinfo() does not "
                        "appear to be consisting of (family, socktype, (addr, port))")

def get_soa_serial(soa_rdata):
    '''Extract the serial field of SOA RDATA and return it as a Serial object.

    We don't have to be very efficient here, so we first dump the entire RDATA
    as a string and convert the first corresponding field.  This should be
    sufficient in practice, but may not always work when the MNAME or RNAME
    contains an (escaped) space character in their labels.  Ideally there
    should be a more direct and convenient way to get access to the SOA
    fields.
    '''
    return Serial(int(soa_rdata.to_text().split()[2]))

class XfrinState:
    '''
    The states of the incoming *XFR state machine.

    We (will) handle both IXFR and AXFR with a single integrated state
    machine because they cannot be distinguished immediately - an AXFR
    response to an IXFR request can only be detected when the first two (2)
    response RRs have already been received.

    The following diagram summarizes the state transition.  After sending
    the query, xfrin starts the process with the InitialSOA state (all
    IXFR/AXFR response begins with an SOA).  When it reaches IXFREnd
    or AXFREnd, the process successfully completes.

                             (AXFR or
            (recv SOA)        AXFR-style IXFR)  (SOA, add)
    InitialSOA------->FirstData------------->AXFR--------->AXFREnd
         |                |                  |  ^         (post xfr
         |(IXFR &&        |                  |  |        checks, then
         | recv SOA       |                  +--+        commit)
         | not new)       |            (non SOA, add)
         V                |
    IXFRUptodate          |                     (non SOA, delete)
               (pure IXFR,|                           +-------+
            keep handling)|             (Delete SOA)  V       |
                          + ->IXFRDeleteSOA------>IXFRDelete--+
                                   ^                   |
                (see SOA, not end, |          (see SOA)|
            commit, keep handling) |                   |
                                   |                   V
                      +---------IXFRAdd<----------+IXFRAddSOA
        (non SOA, add)|         ^  |    (Add SOA)
                      ----------+  |
                                   |(see SOA w/ end serial, commit changes)
                                   V
                                IXFREnd

    Note that changes are committed for every "difference sequence"
    (i.e. changes for one SOA update).  This means when an IXFR response
    contains multiple difference sequences and something goes wrong
    after several commits, these changes have been published and visible
    to clients even if the IXFR session is subsequently aborted.
    It is not clear if this is valid in terms of the protocol specification.
    Section 4 of RFC 1995 states:

       An IXFR client, should only replace an older version with a newer
       version after all the differences have been successfully processed.

    If this "replacement" is for the changes of one difference sequence
    and "all the differences" mean the changes for that sequence, this
    implementation strictly follows what RFC states.  If this is for
    the entire IXFR response (that may contain multiple sequences),
    we should implement it with one big transaction and one final commit
    at the very end.

    For now, we implement it with multiple smaller commits for two
    reasons.  First, this is what BIND 9 does, and we generally port
    the implementation logic here.  BIND 9 has been supporting IXFR
    for many years, so the fact that it still behaves this way
    probably means it at least doesn't cause a severe operational
    problem in practice.  Second, especially because BIND 10 would
    often uses a database backend, a larger transaction could cause an
    undesirable effects, e.g. suspending normal lookups for a longer
    period depending on the characteristics of the database.  Even if
    we find something wrong in a later sequeunce and abort the
    session, we can start another incremental update from what has
    been validated, or we can switch to AXFR to replace the zone
    completely.

    This implementation uses the state design pattern, where each state
    is represented as a subclass of the base XfrinState class.  Each concrete
    subclass of XfrinState is assumed to define two methods: handle_rr() and
    finish_message().  These methods handle specific part of XFR protocols
    and (if necessary) perform the state transition.

    Conceptually, XfrinState and its subclasses are a "friend" of
    XfrinConnection and are assumed to be allowed to access its internal
    information (even though Python does not have a strict access control
    between different classes).

    The XfrinState and its subclasses are designed to be stateless, and
    can be used as singleton objects.  For now, however, we always instantiate
    a new object for every state transition, partly because the introduction
    of singleton will make a code bit complicated, and partly because
    the overhead of object instantiation wouldn't be significant for xfrin.

    '''
    def set_xfrstate(self, conn, new_state):
        '''Set the XfrConnection to a given new state.

        As a "friend" class, this method intentionally gets access to the
        connection's "private" method.

        '''
        conn._XfrinConnection__set_xfrstate(new_state)

    def handle_rr(self, conn):
        '''Handle one RR of an XFR response message.

        Depending on the state, the RR is generally added or deleted in the
        corresponding data source, or in some special cases indicates
        a specifi transition, such as starting a new IXFR difference
        sequence or completing the session.

        All subclass has their specific behaviors for this method, so
        there is no default definition.  If the base class version
        is called, it's a bug of the caller, and it's notified via
        an XfrinException exception.

        This method returns a boolean value: True if the given RR was
        fully handled and the caller should go to the next RR; False
        if the caller needs to call this method with the (possibly) new
        state for the same RR again.

        '''
        raise XfrinException("Internal bug: " +
                             "XfrinState.handle_rr() called directly")

    def finish_message(self, conn):
        '''Perform any final processing after handling all RRs of a response.

        This method then returns a boolean indicating whether to continue
        receiving the message.  Unless it's in the end of the entire XFR
        session, we should continue, so this default method simply returns
        True.

        '''
        return True

class XfrinInitialSOA(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() != RRType.SOA:
            raise XfrinProtocolError('First RR in zone transfer must be SOA ('
                                     + rr.get_type().to_text() + ' received)')
        conn._end_serial = get_soa_serial(rr.get_rdata()[0])

        if conn._request_type == RRType.IXFR and \
                conn._end_serial <= conn._request_serial:
            logger.info(XFRIN_IXFR_UPTODATE, conn.zone_str(),
                        conn._request_serial, conn._end_serial)
            self.set_xfrstate(conn, XfrinIXFRUptodate())
        else:
            self.set_xfrstate(conn, XfrinFirstData())

        return True

class XfrinFirstData(XfrinState):
    def handle_rr(self, conn, rr):
        '''Handle the first RR after initial SOA in an XFR session.

        This state happens exactly once in an XFR session, where
        we decide whether it's incremental update ("real" IXFR) or
        non incremental update (AXFR or AXFR-style IXFR).
        If we initiated IXFR and the transfer begins with two SOAs
        (the serial of the second one being equal to our serial),
        it's incremental; otherwise it's non incremental.

        This method always return False (unlike many other handle_rr()
        methods) because this first RR must be examined again in the
        determined update context.

        Note that in the non incremental case the RR should normally be
        something other SOA, but it's still possible it's an SOA with a
        different serial than ours.  The only possible interpretation at
        this point is that it's non incremental update that only consists
        of the SOA RR.  It will result in broken zone (for example, it
        wouldn't even contain an apex NS) and should be rejected at post
        XFR processing, but in terms of the XFR session processing we
        accept it and move forward.

        Note further that, in the half-broken SOA-only transfer case,
        these two SOAs are supposed to be the same as stated in Section 2.2
        of RFC 5936.  We don't check that condition here, either; we'll
        leave whether and how to deal with that situation to the end of
        the processing of non incremental update.  See also a related
        discussion at the IETF dnsext wg:
        http://www.ietf.org/mail-archive/web/dnsext/current/msg07908.html

        '''
        if conn._request_type == RRType.IXFR and \
                rr.get_type() == RRType.SOA and \
                conn._request_serial == get_soa_serial(rr.get_rdata()[0]):
            logger.debug(DBG_XFRIN_TRACE, XFRIN_GOT_INCREMENTAL_RESP,
                         conn.zone_str())
            conn._diff = None # Will be created on-demand
            self.set_xfrstate(conn, XfrinIXFRDeleteSOA())
        else:
            logger.debug(DBG_XFRIN_TRACE, XFRIN_GOT_NONINCREMENTAL_RESP,
                 conn.zone_str())
            # We are now going to add RRs to the new zone.  We need create
            # a Diff object.  It will be used throughtout the XFR session.
            conn._diff = Diff(conn._datasrc_client, conn._zone_name, True)
            self.set_xfrstate(conn, XfrinAXFR())
        return False

class XfrinIXFRDeleteSOA(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() != RRType.SOA:
            # this shouldn't happen; should this occur it means an internal
            # bug.
            raise XfrinException(rr.get_type().to_text() +
                                 ' RR is given in IXFRDeleteSOA state')
        # This is the beginning state of one difference sequence (changes
        # for one SOA update).  We may need to create a new Diff object now.
        # Note also that we (unconditionally) enable journaling here.  The
        # Diff constructor may internally disable it, however, if the
        # underlying data source doesn't support journaling.
        if conn._diff is None:
            conn._diff = Diff(conn._datasrc_client, conn._zone_name, False,
                              True)
        conn._diff.delete_data(rr)
        self.set_xfrstate(conn, XfrinIXFRDelete())
        conn.get_transfer_stats().ixfr_deletion_count += 1
        return True

class XfrinIXFRDelete(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() == RRType.SOA:
            # This is the only place where current_serial is set
            conn._current_serial = get_soa_serial(rr.get_rdata()[0])
            self.set_xfrstate(conn, XfrinIXFRAddSOA())
            return False
        conn._diff.delete_data(rr)
        conn.get_transfer_stats().ixfr_deletion_count += 1
        return True

class XfrinIXFRAddSOA(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() != RRType.SOA:
            # this shouldn't happen; should this occur it means an internal
            # bug.
            raise XfrinException(rr.get_type().to_text() +
                                 ' RR is given in IXFRAddSOA state')
        conn._diff.add_data(rr)
        self.set_xfrstate(conn, XfrinIXFRAdd())
        conn.get_transfer_stats().ixfr_addition_count += 1
        return True

class XfrinIXFRAdd(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() == RRType.SOA:
            # This SOA marks the end of a difference sequence
            conn.get_transfer_stats().ixfr_changeset_count += 1
            soa_serial = get_soa_serial(rr.get_rdata()[0])
            if soa_serial == conn._end_serial:
                # The final part is there. Finish the transfer by
                # checking the last TSIG (if required), the zone data and
                # committing.
                conn.finish_transfer()
                self.set_xfrstate(conn, XfrinIXFREnd())
                return True
            elif soa_serial != conn._current_serial:
                raise XfrinProtocolError('IXFR out of sync: expected ' +
                                         'serial ' +
                                         str(conn._current_serial) +
                                         ', got ' + str(soa_serial))
            else:
                # Apply a change to the database. But don't commit it yet,
                # we can't know if the message is/will be properly signed.
                # A complete commit will happen after the last bit.
                conn._diff.apply()
                self.set_xfrstate(conn, XfrinIXFRDeleteSOA())
                return False
        conn._diff.add_data(rr)
        conn.get_transfer_stats().ixfr_addition_count += 1
        return True

class XfrinIXFREnd(XfrinState):
    def handle_rr(self, conn, rr):
        raise XfrinProtocolError('Extra data after the end of IXFR diffs: ' +
                                 rr.to_text())

    def finish_message(self, conn):
        '''Final processing after processing an entire IXFR session.

        There will be more actions here, but for now we simply return False,
        indicating there will be no more message to receive.

        '''
        return False

class XfrinIXFRUptodate(XfrinState):
    def handle_rr(self, conn, rr):
        raise XfrinProtocolError('Extra data after single IXFR response ' +
                                 rr.to_text())

    def finish_message(self, conn):
        raise XfrinZoneUptodate

class XfrinAXFR(XfrinState):
    def handle_rr(self, conn, rr):
        """
        Handle the RR by putting it into the zone.
        """
        conn._diff.add_data(rr)
        if rr.get_type() == RRType.SOA:
            # SOA means end.  Don't commit it yet - we need to perform
            # post-transfer checks

            soa_serial = get_soa_serial(rr.get_rdata()[0])
            if conn._end_serial != soa_serial:
                logger.warn(XFRIN_AXFR_INCONSISTENT_SOA, conn.zone_str(),
                            conn._end_serial, soa_serial)

            self.set_xfrstate(conn, XfrinAXFREnd())
        conn.get_transfer_stats().axfr_rr_count += 1
        # Yes, we've eaten this RR.
        return True

class XfrinAXFREnd(XfrinState):
    def handle_rr(self, conn, rr):
        raise XfrinProtocolError('Extra data after the end of AXFR: ' +
                                 rr.to_text())

    def finish_message(self, conn):
        """
        Final processing after processing an entire AXFR session.

        This simply calls the finish_transfer method of the connection
        that ensures it is signed by TSIG (if required), the zone data
        is valid and commits it.
        """
        conn.finish_transfer()
        return False

class XfrinTransferStats:
    """
    This class keeps a record of transfer data for logging purposes.
    It records number of messages, rrs, and bytes transfered, as well
    as the start and end time. The start time is set upon instantiation of
    this class. The end time is set the first time finalize(),
    get_running_time(), or get_bytes_per_second() is called. The end time is
    set only once; subsequent calls to any of these methods does not modify
    it further.
    All _count instance variables can be directly set as needed by the
    class collecting these results.
    """
    def __init__(self):
        self.message_count = 0
        self.axfr_rr_count = 0
        self.byte_count = 0
        self.ixfr_changeset_count = 0;
        self.ixfr_deletion_count = 0;
        self.ixfr_addition_count = 0;
        self._start_time = time.time()
        self._end_time = None

    def finalize(self):
        """Sets the end time to time.time() if not done already."""
        if self._end_time is None:
            self._end_time = time.time()

    def get_running_time(self):
        """Calls finalize(), then returns the difference between creation
           and finalization time"""
        self.finalize()
        return self._end_time - self._start_time

    def get_bytes_per_second(self):
        """Returns the number of bytes per second, based on the result of
           get_running_time() and the value of bytes_count."""
        runtime = self.get_running_time()
        if runtime > 0.0:
            return float(self.byte_count) / runtime
        else:
            # This should never happen, but if some clock is so
            # off or reset in the meantime, we do need to return
            # *something* (and not raise an error)
            if self.byte_count == 0:
                return 0.0
            else:
                return float("inf")


class XfrinConnection(asyncore.dispatcher):
    '''Do xfrin in this class. '''

    def __init__(self,
                 sock_map, zone_name, rrclass, datasrc_client,
                 shutdown_event, master_addrinfo, zone_soa, tsig_key=None,
                 idle_timeout=60):
        """Constructor of the XfirnConnection class.

        Parameters:
          sock_map: empty dict, used with asyncore.
          zone_name (dns.Name): Zone name.
          rrclass (dns.RRClass): Zone RR class.
          datasrc_client (DataSourceClient): the data source client object
            used for the XFR session.
          shutdown_event (threading.Event): used for synchronization with
            parent thread.
          master_addrinfo (tuple: (sock family, sock type, sockaddr)):
            address and port of the master server.
          zone_soa (RRset or None): SOA RRset of zone's current SOA or None
            if it's not available.
          idle_timeout (int): max idle time for read data from socket.

        """

        asyncore.dispatcher.__init__(self, map=sock_map)

        # The XFR state.  Conceptually this is purely private, so we emphasize
        # the fact by the double underscore.  Other classes are assumed to
        # get access to this via get_xfrstate(), and only XfrinState classes
        # are assumed to be allowed to modify it via __set_xfrstate().
        self.__state = None

        # Requested transfer type (RRType.AXFR or RRType.IXFR).  The actual
        # transfer type may differ due to IXFR->AXFR fallback:
        self._request_type = None

        # Zone parameters
        self._zone_name = zone_name
        self._rrclass = rrclass

        # Data source handler
        self._datasrc_client = datasrc_client
        self._zone_soa = zone_soa

        self._sock_map = sock_map
        self._soa_rr_count = 0
        self._idle_timeout = idle_timeout
        self._shutdown_event = shutdown_event
        self._master_addrinfo = master_addrinfo
        self._tsig_key = tsig_key
        self._tsig_ctx = None
        # tsig_ctx_creator is introduced to allow tests to use a mock class for
        # easier tests (in normal case we always use the default)
        self._tsig_ctx_creator = lambda key : TSIGContext(key)

        # keep a record of this specific transfer to log on success
        # (time, rr/s, etc)
        self._transfer_stats = XfrinTransferStats()
        self._counters = Counters(SPECFILE_LOCATION)

    def init_socket(self):
        '''Initialize the underlyig socket.

        This is essentially a part of __init__() and is expected to be
        called immediately after the constructor.  It's separated from
        the constructor because otherwise we might not be able to close
        it if the constructor raises an exception after opening the socket.
        '''
        self.create_socket(self._master_addrinfo[0], self._master_addrinfo[1])
        self.socket.setblocking(1)

    def __set_xfrstate(self, new_state):
        self.__state = new_state

    def get_xfrstate(self):
        return self.__state

    def get_transfer_stats(self):
        """Returns the transfer stats object, used to measure transfer time,
           and number of messages/records/bytes transfered."""
        return self._transfer_stats

    def zone_str(self):
        '''A convenience function for logging to include zone name and class'''
        return format_zone_str(self._zone_name, self._rrclass)

    def connect_to_master(self):
        '''Connect to master in TCP.'''

        try:
            self.connect(self._master_addrinfo[2])
            return True
        except socket.error as e:
            logger.error(XFRIN_CONNECT_MASTER, self._master_addrinfo[2],
                         str(e))
            return False

    def _create_query(self, query_type):
        '''Create an XFR-related query message.

        query_type is either SOA, AXFR or IXFR.  An IXFR query needs the
        zone's current SOA record.  If it's not known, it raises an
        XfrinException exception.  Note that this may not necessarily a
        broken configuration; for the first attempt of transfer the secondary
        may not have any boot-strap zone information, in which case IXFR
        simply won't work.  The xfrin should then fall back to AXFR.
        _request_serial is recorded for later use.

        '''
        msg = Message(Message.RENDER)
        query_id = random.randint(0, 0xFFFF)
        self._query_id = query_id
        msg.set_qid(query_id)
        msg.set_opcode(Opcode.QUERY)
        msg.set_rcode(Rcode.NOERROR)
        msg.add_question(Question(self._zone_name, self._rrclass, query_type))

        # Remember our serial, if known
        self._request_serial = get_soa_serial(self._zone_soa.get_rdata()[0]) \
            if self._zone_soa is not None else None

        # Set the authority section with our SOA for IXFR
        if query_type == RRType.IXFR:
            if self._zone_soa is None:
                # (incremental) IXFR doesn't work without known SOA
                raise XfrinException('Failed to create IXFR query due to no ' +
                                     'SOA for ' + self.zone_str())
            msg.add_rrset(Message.SECTION_AUTHORITY, self._zone_soa)

        return msg

    def _send_data(self, data):
        size = len(data)
        total_count = 0
        while total_count < size:
            count = self.send(data[total_count:])
            total_count += count

    def _send_query(self, query_type):
        '''Send query message over TCP. '''

        msg = self._create_query(query_type)
        render = MessageRenderer()
        # XXX Currently, python wrapper doesn't accept 'None' parameter in this
        # case, we should remove the if statement and use a universal
        # interface later.
        if self._tsig_key is not None:
            self._tsig_ctx = self._tsig_ctx_creator(self._tsig_key)
            msg.to_wire(render, self._tsig_ctx)
        else:
            msg.to_wire(render)

        header_len = struct.pack('H', socket.htons(render.get_length()))
        self._send_data(header_len)
        self._send_data(render.get_data())

    def _asyncore_loop(self):
        '''
        This method is a trivial wrapper for asyncore.loop().  It's extracted from
        _get_request_response so that we can test the rest of the code without
        involving actual communication with a remote server.'''
        asyncore.loop(self._idle_timeout, map=self._sock_map, count=1)

    def _get_request_response(self, size):
        recv_size = 0
        data = b''
        while recv_size < size:
            self._recv_time_out = True
            self._need_recv_size = size - recv_size
            self._asyncore_loop()
            if self._recv_time_out:
                raise XfrinException('receive data from socket time out.')

            recv_size += self._recvd_size
            data += self._recvd_data

        return data

    def _check_response_tsig(self, msg, response_data):
        tsig_record = msg.get_tsig_record()
        if self._tsig_ctx is not None:
            tsig_error = self._tsig_ctx.verify(tsig_record, response_data)
            if tsig_error != TSIGError.NOERROR:
                raise XfrinProtocolError('TSIG verify fail: %s' %
                                         str(tsig_error))
        elif tsig_record is not None:
            # If the response includes a TSIG while we didn't sign the query,
            # we treat it as an error.  RFC doesn't say anything about this
            # case, but it clearly states the server must not sign a response
            # to an unsigned request.  Although we could be flexible, no sane
            # implementation would return such a response, and since this is
            # part of security mechanism, it's probably better to be more
            # strict.
            raise XfrinProtocolError('Unexpected TSIG in response')

    def _check_response_tsig_last(self):
        """
        Check there's a signature at the last message.
        """
        if self._tsig_ctx is not None:
            if not self._tsig_ctx.last_had_signature():
                raise XfrinProtocolError('TSIG verify fail: no TSIG on last '+
                                         'message')

    def __validate_error(self, reason):
        '''
        Used as error callback below.
        '''
        logger.error(XFRIN_ZONE_INVALID, self._zone_name, self._rrclass,
                     reason)

    def __validate_warning(self, reason):
        '''
        Used as warning callback below.
        '''
        logger.warn(XFRIN_ZONE_WARN, self._zone_name, self._rrclass, reason)

    def finish_transfer(self):
        """
        Perform any necessary checks after a transfer. Then complete the
        transfer by committing the transaction into the data source.
        """
        self._check_response_tsig_last()
        if not check_zone(self._zone_name, self._rrclass,
                          self._diff.get_rrset_collection(),
                          (self.__validate_error, self.__validate_warning)):
            raise XfrinZoneError('Validation of the new zone failed')
        self._diff.commit()

    def __parse_soa_response(self, msg, response_data):
        '''Parse a response to SOA query and extract the SOA from answer.

        This is a subroutine of _check_soa_serial().  This method also
        validates message, and rejects bogus responses with XfrinProtocolError.

        If everything is okay, it returns the SOA RR from the answer section
        of the response.

        '''
        # Check TSIG integrity and validate the header.  Unlike AXFR/IXFR,
        # we should be more strict for SOA queries and check the AA flag, too.
        self._check_response_tsig(msg, response_data)
        self._check_response_header(msg)
        if not msg.get_header_flag(Message.HEADERFLAG_AA):
            raise XfrinProtocolError('non-authoritative answer to SOA query')

        # Validate the question section
        n_question = msg.get_rr_count(Message.SECTION_QUESTION)
        if n_question != 1:
            raise XfrinProtocolError('Invalid response to SOA query: ' +
                                     '(' + str(n_question) + ' questions, 1 ' +
                                     'expected)')
        resp_question = msg.get_question()[0]
        if resp_question.get_name() != self._zone_name or \
                resp_question.get_class() != self._rrclass or \
                resp_question.get_type() != RRType.SOA:
            raise XfrinProtocolError('Invalid response to SOA query: '
                                     'question mismatch: ' +
                                     str(resp_question))

        # Look into the answer section for SOA
        soa = None
        for rr in msg.get_section(Message.SECTION_ANSWER):
            if rr.get_type() == RRType.SOA:
                if soa is not None:
                    raise XfrinProtocolError('SOA response had multiple SOAs')
                soa = rr
            # There should not be a CNAME record at top of zone.
            if rr.get_type() == RRType.CNAME:
                raise XfrinProtocolError('SOA query resulted in CNAME')

        # If SOA is not found, try to figure out the reason then report it.
        if soa is None:
            # See if we have any SOA records in the authority section.
            for rr in msg.get_section(Message.SECTION_AUTHORITY):
                if rr.get_type() == RRType.NS:
                    raise XfrinProtocolError('SOA query resulted in referral')
                if rr.get_type() == RRType.SOA:
                    raise XfrinProtocolError('SOA query resulted in NODATA')
            raise XfrinProtocolError('No SOA record found in response to ' +
                                     'SOA query')

        # Check if the SOA is really what we asked for
        if soa.get_name() != self._zone_name or \
                soa.get_class() != self._rrclass:
            raise XfrinProtocolError("SOA response doesn't match query: " +
                                     str(soa))

        # All okay, return it
        return soa

    def _get_ipver_str(self):
        """Returns a 'v4' or 'v6' string representing a IP version
        depending on the socket family. This is for an internal use
        only (except for tests). This is supported only for IP sockets.
        It raises a ValueError exception on other address families.

        """
        if self.socket.family == socket.AF_INET:
            return 'v4'
        elif self.socket.family == socket.AF_INET6:
            return 'v6'
        raise ValueError("Invalid address family. "
                         "This is supported only for IP sockets")

    def _check_soa_serial(self):
        '''Send SOA query and compare the local and remote serials.

        If we know our local serial and the remote serial isn't newer
        than ours, we abort the session with XfrinZoneUptodate.
        On success it returns XFRIN_OK for testing.  The caller won't use it.

        '''

        self._send_query(RRType.SOA)
        # count soaoutv4 or soaoutv6 requests
        self._counters.inc('zones', self._rrclass.to_text(),
                           self._zone_name.to_text(), 'soaout' +
                           self._get_ipver_str())
        data_len = self._get_request_response(2)
        msg_len = socket.htons(struct.unpack('H', data_len)[0])
        soa_response = self._get_request_response(msg_len)
        msg = Message(Message.PARSE)
        msg.from_wire(soa_response, Message.PRESERVE_ORDER)

        # Validate/parse the rest of the response, and extract the SOA
        # from the answer section
        soa = self.__parse_soa_response(msg, soa_response)

        # Compare the two serials.  If ours is 'new', abort with ZoneUptodate.
        primary_serial = get_soa_serial(soa.get_rdata()[0])
        if self._request_serial is not None and \
                self._request_serial >= primary_serial:
            if self._request_serial != primary_serial:
                logger.info(XFRIN_ZONE_SERIAL_AHEAD, primary_serial,
                            self.zone_str(),
                            format_addrinfo(self._master_addrinfo),
                            self._request_serial)
            raise XfrinZoneUptodate

        return XFRIN_OK

    def do_xfrin(self, check_soa, request_type=RRType.AXFR):
        '''Do an xfr session by sending xfr request and parsing responses.'''

        try:
            ret = XFRIN_OK
            self._request_type = request_type
            req_str = request_type.to_text()
            if check_soa:
                self._check_soa_serial()
                self.close()
                self.init_socket()
                if not self.connect_to_master():
                    raise XfrinException('Unable to reconnect to master')

            # start statistics timer
            # Note: If the timer for the zone is already started but
            # not yet stopped due to some error, the last start time
            # is overwritten at this point.
            self._counters.start_timer('zones',
                                       self._rrclass.to_text(),
                                       self._zone_name.to_text(),
                                       'last_' + req_str.lower() +
                                       '_duration')
            logger.info(XFRIN_XFR_TRANSFER_STARTED, req_str, self.zone_str())
            # An AXFR or IXFR is being requested.
            self._counters.inc('zones', self._rrclass.to_text(),
                               self._zone_name.to_text(),
                               req_str.lower() + 'req' +
                               self._get_ipver_str())
            self._send_query(self._request_type)
            self.__state = XfrinInitialSOA()
            self._handle_xfrin_responses()
            # Depending what data was found, we log different status reports
            # (In case of an AXFR-style IXFR, print the 'AXFR' message)
            if self._transfer_stats.axfr_rr_count == 0:
                logger.info(XFRIN_IXFR_TRANSFER_SUCCESS,
                            self.zone_str(),
                            self._transfer_stats.message_count,
                            self._transfer_stats.ixfr_changeset_count,
                            self._transfer_stats.ixfr_deletion_count,
                            self._transfer_stats.ixfr_addition_count,
                            self._transfer_stats.byte_count,
                            "%.3f" % self._transfer_stats.get_running_time(),
                            "%.f" % self._transfer_stats.get_bytes_per_second()
                           )
            else:
                logger.info(XFRIN_TRANSFER_SUCCESS,
                            req_str,
                            self.zone_str(),
                            self._transfer_stats.message_count,
                            self._transfer_stats.axfr_rr_count,
                            self._transfer_stats.byte_count,
                            "%.3f" % self._transfer_stats.get_running_time(),
                            "%.f" % self._transfer_stats.get_bytes_per_second()
                           )
        except XfrinZoneUptodate:
            # Eventually we'll probably have to treat this case as a trigger
            # of trying another primary server, etc, but for now we treat it
            # as "success".
            pass
        except XfrinZoneError:
            # The log message doesn't contain the exception text, since there's
            # only one place where the exception is thrown now and it'd be the
            # same generic message every time.
            logger.error(XFRIN_INVALID_ZONE_DATA, self.zone_str(),
                         format_addrinfo(self._master_addrinfo))
            ret = XFRIN_FAIL
        except XfrinProtocolError as e:
            logger.info(XFRIN_XFR_TRANSFER_PROTOCOL_VIOLATION, req_str,
                        self.zone_str(),
                        format_addrinfo(self._master_addrinfo), str(e))
            ret = XFRIN_FAIL
        except XfrinException as e:
            logger.error(XFRIN_XFR_TRANSFER_FAILURE, req_str,
                         self.zone_str(),
                         format_addrinfo(self._master_addrinfo), str(e))
            ret = XFRIN_FAIL
        except Exception as e:
            # Catching all possible exceptions like this is generally not a
            # good practice, but handling an xfr session could result in
            # so many types of exceptions, including ones from the DNS library
            # or from the data source library.  Eventually we'd introduce a
            # hierarchy for exception classes from a base "ISC exception" and
            # catch it here, but until then we need broadest coverage so that
            # we won't miss anything.

            logger.error(XFRIN_XFR_OTHER_FAILURE, req_str,
                         self.zone_str(), str(e))
            ret = XFRIN_FAIL
        finally:
            # A xfrsuccess or xfrfail counter is incremented depending on
            # the result.
            result = {XFRIN_OK: 'xfrsuccess', XFRIN_FAIL: 'xfrfail'}[ret]
            self._counters.inc('zones', self._rrclass.to_text(),
                               self._zone_name.to_text(), result)
            # The started statistics timer is finally stopped only in
            # a successful case.
            if ret == XFRIN_OK:
                self._counters.stop_timer('zones',
                                          self._rrclass.to_text(),
                                          self._zone_name.to_text(),
                                          'last_' + req_str.lower() +
                                          '_duration')
            # Make sure any remaining transaction in the diff is closed
            # (if not yet - possible in case of xfr-level exception) as soon
            # as possible
            self._diff = None
        return ret

    def _check_response_header(self, msg):
        '''Perform minimal validation on responses'''

        # It's not clear how strict we should be about response validation.
        # BIND 9 ignores some cases where it would normally be considered a
        # bogus response.  For example, it accepts a response even if its
        # opcode doesn't match that of the corresponding request.
        # According to an original developer of BIND 9 some of the missing
        # checks are deliberate to be kind to old implementations that would
        # cause interoperability trouble with stricter checks.

        msg_rcode = msg.get_rcode()
        if msg_rcode != Rcode.NOERROR:
            raise XfrinProtocolError('error response: %s' %
                                     msg_rcode.to_text())

        if not msg.get_header_flag(Message.HEADERFLAG_QR):
            raise XfrinProtocolError('response is not a response')

        if msg.get_qid() != self._query_id:
            raise XfrinProtocolError('bad query id')

    def _check_response_status(self, msg):
        '''Check validation of xfr response. '''

        self._check_response_header(msg)

        if msg.get_rr_count(Message.SECTION_QUESTION) > 1:
            raise XfrinProtocolError('query section count greater than 1')

    def _handle_xfrin_responses(self):
        read_next_msg = True
        while read_next_msg:
            data_len = self._get_request_response(2)
            msg_len = socket.htons(struct.unpack('H', data_len)[0])
            self._transfer_stats.byte_count += msg_len + 2
            recvdata = self._get_request_response(msg_len)
            msg = Message(Message.PARSE)
            msg.from_wire(recvdata, Message.PRESERVE_ORDER)
            self._transfer_stats.message_count += 1

            # TSIG related checks, including an unexpected signed response
            self._check_response_tsig(msg, recvdata)

            # Perform response status validation
            self._check_response_status(msg)

            for rr in msg.get_section(Message.SECTION_ANSWER):
                rr_handled = False
                while not rr_handled:
                    rr_handled = self.__state.handle_rr(self, rr)

            read_next_msg = self.__state.finish_message(self)

            if self._shutdown_event.is_set():
                raise XfrinException('xfrin is forced to stop')

    def handle_read(self):
        '''Read query's response from socket. '''

        self._recvd_data = self.recv(self._need_recv_size)
        self._recvd_size = len(self._recvd_data)
        self._recv_time_out = False

    def writable(self):
        '''Ignore the writable socket. '''

        return False

def __get_initial_xfr_type(zone_soa, request_ixfr, zname, zclass, master_addr):
    """Determine the initial xfr request type.

    This is a dedicated subroutine of __process_xfrin.
    """
    if zone_soa is None:
        # This is a kind of special case, so we log it at info level.
        logger.info(XFRIN_INITIAL_AXFR, format_zone_str(zname, zclass),
                    AddressFormatter(master_addr))
        return RRType.AXFR
    if request_ixfr == ZoneInfo.REQUEST_IXFR_DISABLED:
        logger.debug(DBG_XFRIN_TRACE, XFRIN_INITIAL_IXFR_DISABLED,
                     format_zone_str(zname, zclass),
                     AddressFormatter(master_addr))
        return RRType.AXFR

    assert(request_ixfr == ZoneInfo.REQUEST_IXFR_FIRST or
           request_ixfr == ZoneInfo.REQUEST_IXFR_ONLY)
    logger.debug(DBG_XFRIN_TRACE, XFRIN_INITIAL_IXFR,
                     format_zone_str(zname, zclass),
                     AddressFormatter(master_addr))
    return RRType.IXFR

def __process_xfrin(server, zone_name, rrclass, datasrc_client, zone_soa,
                    shutdown_event, master_addrinfo, check_soa, tsig_key,
                    request_ixfr, conn_class):
    conn = None
    exception = None
    ret = XFRIN_FAIL
    try:
        # Determine the initialreuqest type: AXFR or IXFR.
        request_type = __get_initial_xfr_type(zone_soa, request_ixfr,
                                              zone_name, rrclass,
                                              master_addrinfo[2])

        # Create a TCP connection for the XFR session and perform the
        # operation.
        sock_map = {}
        # In case we were asked to do IXFR and that one fails, we try again
        # with AXFR. But only if we could actually connect to the server.
        #
        # So we start with retry as True, which is set to false on each
        # attempt. In the case of connected but failed IXFR, we set it to true
        # once again.
        retry = True
        while retry:
            retry = False
            conn = conn_class(sock_map, zone_name, rrclass, datasrc_client,
                              shutdown_event, master_addrinfo, zone_soa,
                              tsig_key)
            conn.init_socket()
            ret = XFRIN_FAIL
            if conn.connect_to_master():
                ret = conn.do_xfrin(check_soa, request_type)
                if ret == XFRIN_FAIL and request_type == RRType.IXFR:
                    # IXFR failed for some reason. It might mean the server
                    # can't handle it, or we don't have the zone or we are out
                    # of sync or whatever else. So we retry with with AXFR, as
                    # it may succeed in many such cases; if "IXFR only" is
                    # specified in request_ixfr, however, we suppress the
                    # fallback.
                    if request_ixfr == ZoneInfo.REQUEST_IXFR_ONLY:
                        logger.warn(XFRIN_XFR_TRANSFER_FALLBACK_DISABLED,
                                    conn.zone_str())
                    else:
                        retry = True
                        request_type = RRType.AXFR
                        logger.warn(XFRIN_XFR_TRANSFER_FALLBACK,
                                    conn.zone_str())
                        conn.close()
                        conn = None

    except Exception as ex:
        # If exception happens, just remember it here so that we can re-raise
        # after cleaning up things.  We don't log it here because we want
        # eliminate smallest possibility of having an exception in logging
        # itself.
        exception = ex

    # asyncore.dispatcher requires explicit close() unless its lifetime
    # from born to destruction is closed within asyncore.loop, which is not
    # the case for us.  We always close() here, whether or not do_xfrin
    # succeeds, and even when we see an unexpected exception.
    if conn is not None:
        conn.close()

    # Publish the zone transfer result news, so zonemgr can reset the
    # zone timer, and xfrout can notify the zone's slaves if the result
    # is success.
    server.publish_xfrin_news(zone_name, rrclass, ret)

    if exception is not None:
        raise exception

def process_xfrin(server, xfrin_recorder, zone_name, rrclass, datasrc_client,
                  zone_soa, shutdown_event, master_addrinfo, check_soa,
                  tsig_key, request_ixfr, conn_class=XfrinConnection):
    # Even if it should be rare, the main process of xfrin session can
    # raise an exception.  In order to make sure the lock in xfrin_recorder
    # is released in any cases, we delegate the main part to the helper
    # function in the try block, catch any exceptions, then release the lock.
    xfrin_recorder.increment(zone_name)
    exception = None
    try:
        __process_xfrin(server, zone_name, rrclass, datasrc_client, zone_soa,
                        shutdown_event, master_addrinfo, check_soa, tsig_key,
                        request_ixfr, conn_class)
    except Exception as ex:
        # don't log it until we complete decrement().
        exception = ex
    xfrin_recorder.decrement(zone_name)

    if exception is not None:
        if request_ixfr == ZoneInfo.REQUEST_IXFR_DISABLED:
            typestr = "AXFR"
        else:
            typestr = "IXFR"
        logger.error(XFRIN_XFR_PROCESS_FAILURE, typestr, zone_name.to_text(),
                     str(rrclass), str(exception))

class XfrinRecorder:
    def __init__(self):
        self._lock = threading.Lock()
        self._zones = []

    def increment(self, zone_name):
        self._lock.acquire()
        self._zones.append(zone_name)
        self._lock.release()

    def decrement(self, zone_name):
        self._lock.acquire()
        if zone_name in self._zones:
            self._zones.remove(zone_name)
        self._lock.release()

    def xfrin_in_progress(self, zone_name):
        self._lock.acquire()
        ret = zone_name in self._zones
        self._lock.release()
        return ret

    def count(self):
        self._lock.acquire()
        ret = len(self._zones)
        self._lock.release()
        return ret

class ZoneInfo:
    # Internal values corresponding to request_ixfr
    REQUEST_IXFR_FIRST = 0      # request_ixfr=yes, use IXFR 1st then AXFR
    REQUEST_IXFR_ONLY = 1       # request_ixfr=only, use IXFR only
    REQUEST_IXFR_DISABLED = 2   # request_ixfr=no, AXFR-only

    # Map from configuration values for request_ixfr to internal values
    # This is a constant; don't modify.
    REQUEST_IXFR_CFG_TO_VAL = { 'yes': REQUEST_IXFR_FIRST,
                                'only': REQUEST_IXFR_ONLY,
                                'no': REQUEST_IXFR_DISABLED }

    def __init__(self, config_data, module_cc):
        """Creates a zone_info with the config data element as
           specified by the 'zones' list in xfrin.spec. Module_cc is
           needed to get the defaults from the specification"""
        # Handle deprecated config parameter explicitly for the moment.
        if config_data.get('use_ixfr') is not None:
            raise XfrinZoneInfoException('use_ixfr was deprecated.' +
                                         'use rquest_ixfr')

        self._module_cc = module_cc
        self.set_name(config_data.get('name'))
        self.set_master_addr(config_data.get('master_addr'))

        self.set_master_port(config_data.get('master_port'))
        self.set_zone_class(config_data.get('class'))
        self.set_tsig_key_name(config_data.get('tsig_key'))
        self.set_request_ixfr(config_data.get('request_ixfr'))

    @property
    def request_ixfr(self):
        """Policy on the use of IXFR.

        Possible values are REQUEST_IXFR_xxx, internally stored in
        __request_ixfr, read-only outside of the class.

        """
        return self.__request_ixfr

    def set_name(self, name_str):
        """Set the name for this zone given a name string.
           Raises XfrinZoneInfoException if name_str is None or if it
           cannot be parsed."""
        if name_str is None:
            raise XfrinZoneInfoException("Configuration zones list "
                                         "element does not contain "
                                         "'name' attribute")
        else:
            self.name = _check_zone_name(name_str)

    def set_master_addr(self, master_addr_str):
        """Set the master address for this zone given an IP address
           string. Raises XfrinZoneInfoException if master_addr_str is
           None or if it cannot be parsed."""
        if master_addr_str is None:
            raise XfrinZoneInfoException("master address missing from config data")
        else:
            try:
                self.master_addr = isc.net.parse.addr_parse(master_addr_str)
            except ValueError:
                logger.error(XFRIN_BAD_MASTER_ADDR_FORMAT, master_addr_str)
                errmsg = "bad format for zone's master: " + master_addr_str
                raise XfrinZoneInfoException(errmsg)

    def set_master_port(self, master_port_str):
        """Set the master port given a port number string. If
           master_port_str is None, the default from the specification
           for this module will be used. Raises XfrinZoneInfoException if
           the string contains an invalid port number"""
        if master_port_str is None:
            self.master_port = self._module_cc.get_default_value("zones/master_port")
        else:
            try:
                self.master_port = isc.net.parse.port_parse(master_port_str)
            except ValueError:
                logger.error(XFRIN_BAD_MASTER_PORT_FORMAT, master_port_str)
                errmsg = "bad format for zone's master port: " + master_port_str
                raise XfrinZoneInfoException(errmsg)

    def set_zone_class(self, zone_class_str):
        """Set the zone class given an RR class str (e.g. "IN"). If
           zone_class_str is None, it will default to what is specified
           in the specification file for this module. Raises
           XfrinZoneInfoException if the string cannot be parsed."""
        # TODO: remove _str
        self.class_str = zone_class_str or self._module_cc.get_default_value("zones/class")
        if zone_class_str == None:
            #TODO rrclass->zone_class
            self.rrclass = RRClass(self._module_cc.get_default_value("zones/class"))
        else:
            try:
                self.rrclass = RRClass(zone_class_str)
            except InvalidRRClass:
                logger.error(XFRIN_BAD_ZONE_CLASS, zone_class_str)
                errmsg = "invalid zone class: " + zone_class_str
                raise XfrinZoneInfoException(errmsg)

    def set_tsig_key_name(self, tsig_key_str):
        """Set the name of the tsig_key for this zone. If tsig_key_str
           is None, no TSIG key will be used. This name is used to
           find the TSIG key to use for transfers in the global TSIG
           key ring.
           Raises XfrinZoneInfoException if tsig_key_str is not a valid
           (dns) name."""
        if tsig_key_str is None:
            self.tsig_key_name = None
        else:
            # can throw a number of exceptions but it is just one
            # call, so Exception should be OK here
            try:
                self.tsig_key_name = Name(tsig_key_str)
            except Exception as exc:
                raise XfrinZoneInfoException("Bad TSIG key name: " + str(exc))

    def get_tsig_key(self):
        if self.tsig_key_name is None:
            return None
        result, key = get_keyring().find(self.tsig_key_name)
        if result != isc.dns.TSIGKeyRing.SUCCESS:
            raise XfrinZoneInfoException("TSIG key not found in keyring: " +
                                         self.tsig_key_name.to_text())
        else:
            return key

    def set_request_ixfr(self, request_ixfr):
        if request_ixfr is None:
            request_ixfr = \
                self._module_cc.get_default_value("zones/request_ixfr")
        try:
            self.__request_ixfr = self.REQUEST_IXFR_CFG_TO_VAL[request_ixfr]
        except KeyError:
            raise XfrinZoneInfoException('invalid value for request_ixfr: ' +
                                         request_ixfr)

    def get_master_addr_info(self):
        return (self.master_addr.family, socket.SOCK_STREAM,
                (str(self.master_addr), self.master_port))

def _do_auth_loadzone(server, zone_name, zone_class):
    msg = auth_loadzone_command(server._module_cc, zone_name, zone_class)
    if msg is not None:
        param = msg['command'][1]
        logger.debug(DBG_XFRIN_TRACE, XFRIN_AUTH_LOADZONE, param["origin"],
                     param["class"])
        seq = server._send_cc_session.group_sendmsg(msg, AUTH_MODULE_NAME,
                                                    want_answer=True)
        answer, env = server._send_cc_session.group_recvmsg(False, seq)

class Xfrin:
    def __init__(self):
        self._max_transfers_in = 10
        self._zones = {}
        self.recorder = XfrinRecorder()
        self._shutdown_event = threading.Event()
        self._counters = Counters(SPECFILE_LOCATION)
        # This is essentially private, but we allow tests to customize it.
        self._datasrc_clients_mgr = DataSrcClientsMgr()

        # Initial configuration
        self._cc_setup()
        config_data = self._module_cc.get_full_config()
        self.config_handler(config_data)
        # data_sources configuration should be ready with cfgmgr, so this
        # shouldn't fail; if it ever does we simply propagate the exception
        # to terminate the program.
        self._module_cc.add_remote_config_by_name('data_sources',
                                                  self._datasrc_config_handler)
        init_keyring(self._module_cc)

    def _cc_setup(self):
        '''This method is used only as part of initialization, but is
        implemented separately for convenience of unit tests; by letting
        the test code override this method we can test most of this class
        without requiring a command channel.'''
        # Create one session for sending command to other modules, because the
        # listening session will block the send operation.
        self._send_cc_session = isc.cc.Session()
        self._module_cc = isc.config.ModuleCCSession(SPECFILE_LOCATION,
                                                     self.config_handler,
                                                     self.command_handler)
        self._module_cc.start()

    def _cc_check_command(self):
        '''This is a straightforward wrapper for cc.check_command,
        but provided as a separate method for the convenience
        of unit tests.'''
        self._module_cc.check_command(False)

    def _get_zone_info(self, name, rrclass):
        """Returns the ZoneInfo object containing the configured data
           for the given zone name. If the zone name did not have any
           data, returns None"""
        return self._zones.get((name.to_text(), rrclass.to_text()))

    def _add_zone_info(self, zone_info):
        """Add the zone info. Raises a XfrinZoneInfoException if a zone
           with the same name and class is already configured"""
        key = (zone_info.name.to_text(), zone_info.class_str)
        if key in self._zones:
            raise XfrinZoneInfoException("zone " + str(key) +
                                       " configured multiple times")
        self._zones[key] = zone_info

    def _clear_zone_info(self):
        self._zones = {}

    def config_handler(self, new_config):
        # backup all config data (should there be a problem in the new
        # data)
        old_max_transfers_in = self._max_transfers_in
        old_zones = self._zones

        self._max_transfers_in = \
            new_config.get("transfers_in") or self._max_transfers_in

        if 'zones' in new_config:
            self._clear_zone_info()
            for zone_config in new_config.get('zones'):
                try:
                    zone_info = ZoneInfo(zone_config, self._module_cc)
                    self._add_zone_info(zone_info)
                except XfrinZoneInfoException as xce:
                    self._zones = old_zones
                    self._max_transfers_in = old_max_transfers_in
                    return create_answer(1, str(xce))

        return create_answer(0)

    def _datasrc_config_handler(self, new_config, config_data):
        """Configuration handler of the 'data_sources' module.

        The actual handling is deletegated to the DataSrcClientsMgr class;
        this method is a simple wrapper.

        This is essentially private, but implemented as 'protected' so tests
        can refer to it; other external use is prohibited.

        """
        try:
            self._datasrc_clients_mgr.reconfigure(new_config)
        except isc.server_common.datasrc_clients_mgr.ConfigError as ex:
            logger.error(XFRIN_DATASRC_CONFIG_ERROR, ex)

    def shutdown(self):
        ''' shutdown the xfrin process. the thread which is doing xfrin should be
        terminated.
        '''
        self._module_cc.send_stopping()
        self._shutdown_event.set()
        main_thread = threading.currentThread()
        for th in threading.enumerate():
            if th is main_thread:
                continue
            th.join()

    def __validate_notify_addr(self, notify_addr, zone_str, zone_info):
        """Validate notify source as a destination for xfr source.

        This is called from __handle_xfr_command in case xfr is triggered
        by ZoneMgr either due to incoming Notify or periodic refresh event.

        """
        if zone_info is None:
            # TODO what to do? no info known about zone. defaults?
            errmsg = "Got notification to retransfer unknown zone " + zone_str
            logger.info(XFRIN_RETRANSFER_UNKNOWN_ZONE, zone_str)
            return create_answer(1, errmsg)
        else:
            master_addr = zone_info.get_master_addr_info()
            if (notify_addr[0] != master_addr[0] or
                notify_addr[2] != master_addr[2]):
                notify_addr_str = format_addrinfo(notify_addr)
                master_addr_str = format_addrinfo(master_addr)
                errmsg = "Got notification for " + zone_str\
                    + "from unknown address: " + notify_addr_str;
                logger.info(XFRIN_NOTIFY_UNKNOWN_MASTER, zone_str,
                            notify_addr_str, master_addr_str)
                return create_answer(1, errmsg)

        # Notified address is okay
        return None

    def __get_running_request_ixfr(self, arg_request_ixfr, zone_info):
        """Determine the request_ixfr policy for a specific transfer.

        This is a dedicated subroutine of __handle_xfr_command.

        """
        # If explicitly specified, use it.
        if arg_request_ixfr is not None:
            return arg_request_ixfr
        # Otherwise, if zone info is known, use its value.
        if zone_info is not None:
            return zone_info.request_ixfr
        # Otherwise, use the default value for ZoneInfo
        request_ixfr_def = \
            self._module_cc.get_default_value("zones/request_ixfr")
        return ZoneInfo.REQUEST_IXFR_CFG_TO_VAL[request_ixfr_def]

    def __handle_xfr_command(self, args, check_soa, addr_validator,
                             request_ixfr):
        """Common subroutine for handling transfer commands.

        This helper method unifies both cases of transfer command from
        ZoneMgr or from a user.  Depending on who invokes the transfer,
        details of validation and parameter selection slightly vary.
        These conditions are passed through parameters and handled in the
        unified code of this method accordingly.

        If this is from the ZoneMgr due to incoming notify, zone transfer
        should start from the notify's source address as long as it's
        configured as a master address, according to RFC1996.  The current
        implementation conforms to it in a limited way: we can only set one
        master address. Once we add the ability to have multiple master
        addresses, we should check if it matches one of them, and then use it.

        In case of transfer command from the user, if the command specifies
        the master address, use that one; otherwise try to use a configured
        master address for the zone.

        """
        (zone_name, rrclass) = self._parse_zone_name_and_class(args)
        master_addr = self._parse_master_and_port(args, zone_name, rrclass)
        zone_info = self._get_zone_info(zone_name, rrclass)
        tsig_key = None if zone_info is None else zone_info.get_tsig_key()
        zone_str = format_zone_str(zone_name, rrclass) # for logging
        answer = addr_validator(master_addr, zone_str, zone_info)
        if answer is not None:
            return answer
        request_ixfr = self.__get_running_request_ixfr(request_ixfr, zone_info)
        ret = self.xfrin_start(zone_name, rrclass, master_addr, tsig_key,
                               request_ixfr, check_soa)
        return create_answer(ret[0], ret[1])

    def command_handler(self, command, args):
        logger.debug(DBG_XFRIN_TRACE, XFRIN_RECEIVED_COMMAND, command)
        answer = create_answer(0)
        try:
            if command == 'shutdown':
                self._shutdown_event.set()
            elif command == 'notify' or command == REFRESH_FROM_ZONEMGR:
                # refresh/notify command from zone manager.
                # The address has to be validated and always perform SOA check.
                addr_validator = \
                    lambda x, y, z: self.__validate_notify_addr(x, y, z)
                answer = self.__handle_xfr_command(args, True, addr_validator,
                                                   None)
            elif command == 'retransfer':
                # retransfer from cmdctl (sent by bindctl).
                # No need for address validation, skip SOA check, and always
                # use AXFR.
                answer = self.__handle_xfr_command(
                    args, False, lambda x, y, z: None,
                    ZoneInfo.REQUEST_IXFR_DISABLED)
            elif command == 'refresh':
                # retransfer from cmdctl (sent by bindctl).  similar to
                # retransfer, but do SOA check, and honor request_ixfr config.
                answer = self.__handle_xfr_command(
                    args, True, lambda x, y, z: None, None)
            # return statistics data to the stats daemon
            elif command == "getstats":
                # The log level is here set to debug in order to avoid
                # that a log becomes too verbose. Because the
                # b10-stats daemon is periodically asking to the
                # b10-xfrin daemon.
                answer = create_answer(0, self._counters.get_statistics())

            else:
                answer = create_answer(1, 'unknown command: ' + command)
        except XfrinException as err:
            logger.error(XFRIN_COMMAND_ERROR, command, str(err))
            answer = create_answer(1, str(err))
        return answer

    def _parse_zone_name_and_class(self, args):
        zone_name_str = args.get('zone_name')
        if zone_name_str is None:
            raise XfrinException('zone name should be provided')

        return (_check_zone_name(zone_name_str),
                _check_zone_class(args.get('zone_class')))

    def _parse_master_and_port(self, args, zone_name, zone_class):
        """
        Return tuple (family, socktype, sockaddr) for address and port in given
        args dict.
        IPv4 and IPv6 are the only supported addresses now, so sockaddr will be
        (address, port). The socktype is socket.SOCK_STREAM for now.
        """
        # check if we have configured info about this zone, in case
        # port or master are not specified
        zone_info = self._get_zone_info(zone_name, zone_class)

        addr_str = args.get('master')
        if addr_str is None:
            if zone_info is not None:
                addr = zone_info.master_addr
            else:
                raise XfrinException("Master address not given or "
                                     "configured for " + zone_name.to_text())
        else:
            try:
                addr = isc.net.parse.addr_parse(addr_str)
            except ValueError as err:
                raise XfrinException("failed to resolve master address %s: %s" %
                                     (addr_str, str(err)))

        port_str = args.get('port')
        if port_str is None:
            if zone_info is not None:
                port = zone_info.master_port
            else:
                port = DEFAULT_MASTER_PORT
        else:
            try:
                port = isc.net.parse.port_parse(port_str)
            except ValueError as err:
                raise XfrinException("failed to parse port=%s: %s" %
                                     (port_str, str(err)))

        return (addr.family, socket.SOCK_STREAM, (str(addr), port))

    def publish_xfrin_news(self, zone_name, zone_class, xfr_result):
        '''Send command to xfrout/zone manager module.
        If xfrin has finished successfully for one zone, tell the good
        news(command: zone_new_data_ready) to zone manager and xfrout.
        if xfrin failed, just tell the bad news to zone manager, so that
        it can reset the refresh timer for that zone. '''
        param = {'zone_name': zone_name.to_text(),
                 'zone_class': zone_class.to_text()}
        if xfr_result == XFRIN_OK:
            # FIXME: Due to the hack with two different CC sessions
            # (see the _cc_setup comment) and the fact the rpc_call
            # is a high-level call present only at ModuleCCSession,
            # we are forced to use the primitive way of manually
            # calling group_sendmsg and the group_recvmsg. Also, why
            # do we do group_recvmsg when we don't need the answer?
            # And why is this direct RPC call if a notification would
            # be more appropriate?
            _do_auth_loadzone(self, zone_name, zone_class)
            msg = create_command(notify_out.ZONE_NEW_DATA_READY_CMD, param)
            # catch the exception, in case msgq has been killed.
            try:
                seq = self._send_cc_session.group_sendmsg(msg,
                                                          XFROUT_MODULE_NAME,
                                                          want_answer=True)
                try:
                    answer, env = self._send_cc_session.group_recvmsg(False,
                                                                      seq)
                except isc.cc.session.SessionTimeout:
                    pass        # for now we just ignore the failure
                seq = self._send_cc_session.group_sendmsg(msg,
                                                          ZONE_MANAGER_MODULE_NAME,
                                                          want_answer=True)
                try:
                    answer, env = self._send_cc_session.group_recvmsg(False,
                                                                      seq)
                except isc.cc.session.SessionTimeout:
                    pass        # for now we just ignore the failure
            except socket.error as err:
                logger.error(XFRIN_MSGQ_SEND_ERROR, XFROUT_MODULE_NAME, ZONE_MANAGER_MODULE_NAME)

        else:
            msg = create_command(notify_out.ZONE_XFRIN_FAILED, param)
            # catch the exception, in case msgq has been killed.
            try:
                seq = self._send_cc_session.group_sendmsg(msg, ZONE_MANAGER_MODULE_NAME,
                                                          want_answer=True)
                try:
                    answer, env = self._send_cc_session.group_recvmsg(False,
                                                                      seq)
                except isc.cc.session.SessionTimeout:
                    pass        # for now we just ignore the failure
            except socket.error as err:
                logger.error(XFRIN_MSGQ_SEND_ERROR_ZONE_MANAGER, ZONE_MANAGER_MODULE_NAME)

    def startup(self):
        logger.debug(DBG_PROCESS, XFRIN_STARTED)
        while not self._shutdown_event.is_set():
            self._cc_check_command()

    def xfrin_start(self, zone_name, rrclass, master_addrinfo, tsig_key,
                    request_ixfr, check_soa=True):
        # check max_transfer_in, else return quota error
        if self.recorder.count() >= self._max_transfers_in:
            return (1, 'xfrin quota error')

        if self.recorder.xfrin_in_progress(zone_name):
            return (1, 'zone xfrin is in progress')

        # Identify the data source to which the zone content is transferred,
        # and get the current zone SOA from the data source (if available).
        # Note that we do this before spawning the xfrin session thread.
        # find() on the client list and use of ZoneFinder (in _get_zone_soa())
        # should be completed within the same single thread.
        datasrc_client = None
        clist = self._datasrc_clients_mgr.get_client_list(rrclass)
        if clist is None:
            return (1, 'no data source is configured for class %s' % rrclass)

        try:
            datasrc_client = clist.find(zone_name, True, False)[0]
            if datasrc_client is None: # can happen, so log it separately.
                logger.error(XFRIN_DATASRC_UNKNOWN,
                             format_zone_str(zone_name, rrclass))
                return (1, 'data source to transfer %s to is unknown' %
                        format_zone_str(zone_name, rrclass))
            zone_soa = _get_zone_soa(datasrc_client, zone_name, rrclass)
        except isc.datasrc.Error as ex:
            # rare case error. re-raise as XfrinException so it'll be logged
            # in command_handler().
            raise XfrinException('unexpected failure in datasrc module: ' +
                                 str(ex))

        xfrin_thread = threading.Thread(target=process_xfrin,
                                        args=(self, self.recorder,
                                              zone_name, rrclass,
                                              datasrc_client, zone_soa,
                                              self._shutdown_event,
                                              master_addrinfo, check_soa,
                                              tsig_key, request_ixfr))

        xfrin_thread.start()
        return (0, 'zone xfrin is started')

def _get_zone_soa(datasrc_client, zone_name, zone_class):
    """Retrieve the current SOA RR of the zone to be transferred.

    This function is essentially private to the module, but will also
    be called (or tweaked) from tests; no one else should use this
    function directly.

    The specified zone is expected to exist in the data source referenced
    by the given datasrc_client at the point of the call to this function.
    If this is not met XfrinException exception will be raised.

    It will be used for various purposes in subsequent xfr protocol
    processing.   It is validly possible that the zone is currently
    empty and therefore doesn't have an SOA, so this method doesn't
    consider it an error and returns None in such a case.  It may or
    may not result in failure in the actual processing depending on
    how the SOA is used.

    When the zone has an SOA RR, this method makes sure that it's
    valid, i.e., it has exactly one RDATA; if it is not the case
    this method returns None.

    """
    # get the zone finder.  this must be SUCCESS (not even
    # PARTIALMATCH) because we are specifying the zone origin name.
    result, finder = datasrc_client.find_zone(zone_name)
    if result != DataSourceClient.SUCCESS:
        # The data source doesn't know the zone.  In the context of this
        # function is called, this shouldn't happen.
        raise XfrinException("unexpected result: zone %s doesn't exist" %
                             format_zone_str(zone_name, zone_class))
    result, soa_rrset, _ = finder.find(zone_name, RRType.SOA)
    if result != ZoneFinder.SUCCESS:
        logger.info(XFRIN_ZONE_NO_SOA, format_zone_str(zone_name, zone_class))
        return None
    if soa_rrset.get_rdata_count() != 1:
        logger.warn(XFRIN_ZONE_MULTIPLE_SOA,
                    format_zone_str(zone_name, zone_class),
                    soa_rrset.get_rdata_count())
        return None
    return soa_rrset

xfrind = None

def signal_handler(signal, frame):
    if xfrind:
        xfrind.shutdown()
    sys.exit(0)

def set_signal_handler():
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)

def set_cmd_options(parser):
    parser.add_option("-v", "--verbose", dest="verbose", action="store_true",
            help="This option is obsolete and has no effect.")

def main(xfrin_class, use_signal=True):
    """The main loop of the Xfrin daemon.

    @param xfrin_class: A class of the Xfrin object.  This is normally Xfrin,
    but can be a subclass of it for customization.
    @param use_signal: True if this process should catch signals.  This is
    normally True, but may be disabled when this function is called in a
    testing context."""
    global xfrind

    try:
        parser = OptionParser(version = __version__)
        set_cmd_options(parser)
        (options, args) = parser.parse_args()

        if use_signal:
            set_signal_handler()
        xfrind = xfrin_class()
        xfrind.startup()
    except KeyboardInterrupt:
        logger.info(XFRIN_STOPPED_BY_KEYBOARD)
    except isc.cc.session.SessionError as e:
        logger.error(XFRIN_CC_SESSION_ERROR, str(e))
    except Exception as e:
        logger.error(XFRIN_UNKNOWN_ERROR, str(e))

    if xfrind:
        xfrind.shutdown()

    logger.info(XFRIN_EXITING)

if __name__ == '__main__':
    main(Xfrin)