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

#!@PYTHON@

# Copyright (C) 2009-2011  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
from optparse import OptionParser, OptionValueError
from isc.config.ccsession import *
from isc.notify import notify_out
import isc.util.process
from isc.datasrc import DataSourceClient, ZoneFinder
import isc.net.parse
from isc.xfrin.diff import Diff
from isc.log_messages.xfrin_messages import *

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

try:
    from pydnspp import *
except ImportError as e:
    # C++ loadable module may not be installed; even so the xfrin process
    # must keep running, so we warn about it and move forward.
    logger.error(XFRIN_IMPORT_DNS, str(e))

isc.util.process.rename()

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

XFROUT_MODULE_NAME = 'Xfrout'
ZONE_MANAGER_MODULE_NAME = 'Zonemgr'
REFRESH_FROM_ZONEMGR = 'refresh_from_zonemgr'
ZONE_XFRIN_FAILED = 'zone_xfrin_failed'

# Constants for debug levels, to be removed when we have #1074.
DBG_XFRIN_TRACE = 3

# 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'
# define xfrin rcode
XFRIN_OK = 0
XFRIN_FAIL = 1

class XfrinException(Exception):
    pass

class XfrinProtocolError(Exception):
    '''An exception raised for errors encountered in xfrin protocol handling.
    '''
    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 get_soa_serial(soa_rdata):
    '''Extract the serial field of an SOA RDATA and returns it as an intger.

    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 int(soa_rdata.to_text().split()[2])

class XfrinState:
    '''
    The states of the incomding *XFR state machine.
    '''
    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)

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() + ' given)')
        conn._end_serial = get_soa_serial(rr.get_rdata()[0])

        # FIXME: we need to check the serial is actually greater than ours.
        # To do so, however, we need a way to find records from datasource.
        # Complete that part later as a separate task.  (Always performing
        # xfr could be inefficient, but shouldn't do any harm otherwise)

        self.set_xfrstate(conn, XfrinFirstData())
        return True

class XfrinFirstData(XfrinState):
    def handle_rr(self, conn, rr):
        # If the transfer begins with one SOA record, it is an AXFR,
        # if it begins with two SOAs (the serial of the second one being
        # equal to our serial), it is an IXFR.
        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())
            self.set_xfrstate(conn, XfrinIXFRDeleteSOA())
        else:
            logger.debug(DBG_XFRIN_TRACE, XFRIN_GOT_NONINCREMENTAL_RESP,
                 conn.zone_str())
            self.set_xfrstate(conn, XfrinAXFR())
        return False    # need to revisit this RR in an update context

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')
        conn._diff.remove_data(rr)
        self.set_xfrstate(conn, XfrinIXFRDelete())
        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.remove_data(rr)
        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())
        return True

class XfrinIXFRAdd(XfrinState):
    def handle_rr(self, conn, rr):
        if rr.get_type() == RRType.SOA():
            soa_serial = get_soa_serial(rr.get_rdata()[0])
            if soa_serial == conn._end_serial:
                conn._diff.commit()
                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:
                conn._diff.commit()
                self.set_xfrstate(conn, XfrinIXFRDeleteSOA())
                return False
        conn._diff.add_data(rr)
        return True

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

class XfrinAXFR(XfrinState):
    def handle_rr(self, conn, rr):
        raise XfrinException('Falling back from IXFR to AXFR not ' + \
                                 'supported yet')

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

    def __init__(self,
                 sock_map, zone_name, rrclass, db_file, shutdown_event,
                 master_addrinfo, tsig_key = None, verbose = False,
                 idle_timeout = 60):
        ''' idle_timeout: max idle time for read data from socket.
            db_file: specify the data source file.
            check_soa: when it's true, check soa first before sending xfr query
        '''

        asyncore.dispatcher.__init__(self, map=sock_map)
        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
        self._end_serial = None # essentially private

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

        # Data source handlers
        self._db_file = db_file # temporary for sqlite3 specific code
        self._datasrc_client = self._get_datasrc_client(rrclass)

        self.create_socket(master_addrinfo[0], master_addrinfo[1])
        self._sock_map = sock_map
        self._soa_rr_count = 0
        self._idle_timeout = idle_timeout
        self.setblocking(1)
        self._shutdown_event = shutdown_event
        self._verbose = verbose
        self._master_address = master_addrinfo[2]
        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 = self.__create_tsig_ctx

    def __create_tsig_ctx(self, key):
        return TSIGContext(key)

    def _get_datasrc_client(self, rrclass):
        # Create a client here once #1206 is done
        return None

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

    def get_xfrstate(self):
        return self.__state

    def zone_str(self):
        '''A convenient function for logging to include zone name and class'''
        return self._zone_name.to_text() + '/' + str(self._rrclass)

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

        try:
            self.connect(self._master_address)
            return True
        except socket.error as e:
            logger.error(XFRIN_CONNECT_MASTER, self._master_address, str(e))
            return False

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

        query_type is either SOA, AXFR or IXFR.  For type IXFR, it searches
        the associated data source for the current SOA record to include
        it in the query.  If the corresponding zone or the SOA record
        cannot be found, 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))
        if query_type == RRType.IXFR():
            # get the zone finder.  this must be SUCCESS (not even
            # PARTIALMATCH) because we are specifying the zone origin name.
            result, finder = self._datasrc_client.find_zone(self._zone_name)
            if result != DataSourceClient.SUCCESS:
                raise XfrinException('Zone not found in the given data ' +
                                     'source: ' + self.zone_str())
            result, soa_rrset = finder.find(self._zone_name, RRType.SOA(),
                                            None, ZoneFinder.FIND_DEFAULT)
            if result != ZoneFinder.SUCCESS:
                raise XfrinException('SOA RR not found in zone: ' +
                                     self.zone_str())
            # Especially for database-based zones, working zones may be in
            # a broken state where it has more than one SOA RR.  We proactively
            # check the condition and abort the xfr attempt if we identify it.
            if soa_rrset.get_rdata_count() != 1:
                raise XfrinException('Invalid number of SOA RRs for ' +
                                     self.zone_str() + ': ' +
                                     str(soa_rrset.get_rdata_count()))
            msg.add_rrset(Message.SECTION_AUTHORITY, soa_rrset)
            self._request_serial = get_soa_serial(soa_rrset.get_rdata()[0])
        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 XfrinException('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 XfrinException('Unexpected TSIG in response')

    def _check_soa_serial(self):
        ''' Compare the soa serial, if soa serial in master is less than
        the soa serial in local, Finish xfrin.
        False: soa serial in master is less or equal to the local one.
        True: soa serial in master is bigger
        '''

        self._send_query(RRType.SOA())
        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)

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

        # perform some minimal level validation.  It's an open issue how
        # strict we should be (see the comment in _check_response_header())
        self._check_response_header(msg)

        # TODO, need select soa record from data source then compare the two
        # serial, current just return OK, since this function hasn't been used
        # now.
        return XFRIN_OK

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

        try:
            ret = XFRIN_OK
            if check_soa:
                logstr = 'SOA check for \'%s\' ' % self.zone_str()
                ret =  self._check_soa_serial()

            if ret == XFRIN_OK:
                if not ixfr_first:
                    logger.info(XFRIN_AXFR_TRANSFER_STARTED, self.zone_str())
                    self._send_query(RRType.AXFR())
                    isc.datasrc.sqlite3_ds.load(self._db_file,
                                                self._zone_name.to_text(),
                                                self._handle_axfrin_response)
                    logger.info(XFRIN_AXFR_TRANSFER_SUCCESS, self.zone_str())

        except XfrinException as e:
            logger.error(XFRIN_AXFR_TRANSFER_FAILURE, self.zone_str(), str(e))
            ret = XFRIN_FAIL
            #TODO, recover data source.
        except isc.datasrc.sqlite3_ds.Sqlite3DSError as e:
            logger.error(XFRIN_AXFR_DATABASE_FAILURE, self.zone_str(), str(e))
            ret = XFRIN_FAIL
        except UserWarning as e:
            # XXX: this is an exception from our C++ library via the
            # Boost.Python binding.  It would be better to have more more
            # specific exceptions, but at this moment this is the finest
            # granularity.
            logger.error(XFRIN_AXFR_INTERNAL_FAILURE, self.zone_str(), str(e))
            ret = XFRIN_FAIL
        finally:
           self.close()

        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 XfrinException('error response: %s' % msg_rcode.to_text())

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

        if msg.get_qid() != self._query_id:
            raise XfrinException('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_ANSWER) == 0:
            raise XfrinException('answer section is empty')

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

    def _handle_answer_section(self, answer_section):
        '''Return a generator for the reponse in one tcp package to a zone transfer.'''

        for rrset in answer_section:
            rrset_name = rrset.get_name().to_text()
            rrset_ttl = int(rrset.get_ttl().to_text())
            rrset_class = rrset.get_class().to_text()
            rrset_type = rrset.get_type().to_text()

            for rdata in rrset.get_rdata():
                # Count the soa record count
                if rrset.get_type() == RRType.SOA():
                    self._soa_rr_count += 1

                    # XXX: the current DNS message parser can't preserve the
                    # RR order or separete the beginning and ending SOA RRs.
                    # As a short term workaround, we simply ignore the second
                    # SOA, and ignore the erroneous case where the transfer
                    # session doesn't end with an SOA.
                    if (self._soa_rr_count == 2):
                        # Avoid inserting soa record twice
                        break

                rdata_text = rdata.to_text()
                yield (rrset_name, rrset_ttl, rrset_class, rrset_type,
                       rdata_text)

    def _handle_axfrin_response(self):
        '''Return a generator for the response to a zone transfer. '''
        while True:
            data_len = self._get_request_response(2)
            msg_len = socket.htons(struct.unpack('H', data_len)[0])
            recvdata = self._get_request_response(msg_len)
            msg = Message(Message.PARSE)
            msg.from_wire(recvdata)

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

            # Perform response status validation
            self._check_response_status(msg)

            answer_section = msg.get_section(Message.SECTION_ANSWER)
            for rr in self._handle_answer_section(answer_section):
                yield rr

            if self._soa_rr_count == 2:
                break

            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 log_info(self, msg, type='info'):
        # Overwrite the log function, log nothing
        pass

def process_xfrin(server, xfrin_recorder, zone_name, rrclass, db_file,
                  shutdown_event, master_addrinfo, check_soa, verbose,
                  tsig_key):
    xfrin_recorder.increment(zone_name)
    sock_map = {}
    conn = XfrinConnection(sock_map, zone_name, rrclass, db_file,
                           shutdown_event, master_addrinfo,
                           tsig_key, verbose)
    ret = XFRIN_FAIL
    if conn.connect_to_master():
        ret = conn.do_xfrin(check_soa)

    # 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)
    xfrin_recorder.decrement(zone_name)


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:
    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"""
        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(config_data.get('tsig_key'))
        self.set_ixfr_disabled(config_data.get('ixfr_disabled'))

    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(self, tsig_key_str):
        """Set the tsig_key for this zone, given a TSIG key string
           representation. If tsig_key_str is None, no TSIG key will
           be set. Raises XfrinZoneInfoException if tsig_key_str cannot
           be parsed."""
        if tsig_key_str is None:
            self.tsig_key = None
        else:
            try:
                self.tsig_key = TSIGKey(tsig_key_str)
            except InvalidParameter as ipe:
                logger.error(XFRIN_BAD_TSIG_KEY_STRING, tsig_key_str)
                errmsg = "bad TSIG key string: " + tsig_key_str
                raise XfrinZoneInfoException(errmsg)

    def set_ixfr_disabled(self, ixfr_disabled):
        """Set ixfr_disabled. If set to False (the default), it will use
           IXFR for incoming transfers. If set to True, it will use AXFR.
           At this moment there is no automatic fallback"""
        # don't care what type it is; if evaluates to true, set to True
        if ixfr_disabled:
            self.ixfr_disabled = True
        else:
            self.ixfr_disabled = False

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

class Xfrin:
    def __init__(self, verbose = False):
        self._max_transfers_in = 10
        self._zones = {}
        self._cc_setup()
        self.recorder = XfrinRecorder()
        self._shutdown_event = threading.Event()
        self._verbose = verbose

    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()
        config_data = self._module_cc.get_full_config()
        self.config_handler(config_data)

    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 shutdown(self):
        ''' shutdown the xfrin process. the thread which is doing xfrin should be
        terminated.
        '''
        self._shutdown_event.set()
        main_thread = threading.currentThread()
        for th in threading.enumerate():
            if th is main_thread:
                continue
            th.join()

    def command_handler(self, command, args):
        answer = create_answer(0)
        try:
            if command == 'shutdown':
                self._shutdown_event.set()
            elif command == 'notify' or command == REFRESH_FROM_ZONEMGR:
                # Xfrin receives the refresh/notify command from zone manager.
                # notify command maybe has the parameters which
                # specify the notifyfrom address and port, according the RFC1996, zone
                # transfer should starts first from the notifyfrom, but now, let 'TODO' it.
                # (using the value now, while we can only set one master address, would be
                # a security hole. Once we add the ability to have multiple master addresses,
                # we should check if it matches one of them, and then use it.)
                (zone_name, rrclass) = self._parse_zone_name_and_class(args)
                zone_info = self._get_zone_info(zone_name, rrclass)
                if zone_info is None:
                    # TODO what to do? no info known about zone. defaults?
                    errmsg = "Got notification to retransfer unknown zone " + zone_name.to_text()
                    logger.error(XFRIN_RETRANSFER_UNKNOWN_ZONE, zone_name.to_text())
                    answer = create_answer(1, errmsg)
                else:
                    master_addr = zone_info.get_master_addr_info()
                    ret = self.xfrin_start(zone_name,
                                           rrclass,
                                           self._get_db_file(),
                                           master_addr,
                                           zone_info.tsig_key,
                                           True)
                    answer = create_answer(ret[0], ret[1])

            elif command == 'retransfer' or command == 'refresh':
                # Xfrin receives the retransfer/refresh from cmdctl(sent by bindctl).
                # If the command has specified master address, do transfer from the
                # master address, or else do transfer from the configured masters.
                (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:
                    tsig_key = zone_info.tsig_key
                db_file = args.get('db_file') or self._get_db_file()
                ret = self.xfrin_start(zone_name,
                                       rrclass,
                                       db_file,
                                       master_addr,
                                       tsig_key,
                                       (False if command == 'retransfer' else True))
                answer = create_answer(ret[0], ret[1])

            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 _get_db_file(self):
        #TODO, the db file path should be got in auth server's configuration
        # if we need access to this configuration more often, we
        # should add it on start, and not remove it here
        # (or, if we have writable ds, we might not need this in
        # the first place)
        self._module_cc.add_remote_config(AUTH_SPECFILE_LOCATION)
        db_file, is_default = self._module_cc.get_remote_config_value("Auth", "database_file")
        if is_default and "B10_FROM_BUILD" in os.environ:
            # this too should be unnecessary, but currently the
            # 'from build' override isn't stored in the config
            # (and we don't have writable datasources yet)
            db_file = os.environ["B10_FROM_BUILD"] + os.sep + "bind10_zones.sqlite3"
        self._module_cc.remove_remote_config(AUTH_SPECFILE_LOCATION)
        return db_file

    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, 'zone_class': zone_class.to_text()}
        if xfr_result == XFRIN_OK:
            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)
                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)
                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(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)
                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):
        while not self._shutdown_event.is_set():
            self._cc_check_command()

    def xfrin_start(self, zone_name, rrclass, db_file, master_addrinfo, tsig_key,
                    check_soa = True):
        if "pydnspp" not in sys.modules:
            return (1, "xfrin failed, can't load dns message python library: 'pydnspp'")

        # 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')

        xfrin_thread = threading.Thread(target = process_xfrin,
                                        args = (self,
                                                self.recorder,
                                                zone_name,
                                                rrclass,
                                                db_file,
                                                self._shutdown_event,
                                                master_addrinfo, check_soa,
                                                self._verbose,
                                                tsig_key))

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


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="display more about what is going on")

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(verbose = options.verbose)
        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()

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