#!@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. 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 | | | checks, then | +--+ commit) | (non SOA, add) | | (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 instantiotion 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]) # FIXME: we need to check the serial is actually greater than ours. # To do so, however, we need to implement serial number arithmetic. # Although it wouldn't be a big task, we'll leave it for a separate # task for now. (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): '''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()) 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 need to create a new Diff object now. conn._diff = Diff(conn._datasrc_client, conn._zone_name) conn._diff.delete_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.delete_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()) 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 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()) # 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. In this process all the AXFR changes are committed to the data source. There might be more actions here, but for now we simply return False, indicating there will be no more message to receive. """ conn._diff.commit() return False class XfrinConnection(asyncore.dispatcher): '''Do xfrin in this class. ''' def __init__(self, sock_map, zone_name, rrclass, datasrc_client, shutdown_event, master_addrinfo, tsig_key=None, idle_timeout=60): '''Constructor of the XfirnConnection class. idle_timeout: max idle time for read data from socket. datasrc_client: the data source client object used for the XFR session. This will eventually replace db_file completely. ''' 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.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._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 __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, a working zone 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, request_type=RRType.AXFR()): '''Do an xfr session by sending xfr request and parsing responses.''' try: ret = XFRIN_OK self._request_type = request_type # Right now RRType.[IA]XFR().to_text() is 'TYPExxx', so we need # to hardcode here. request_str = 'IXFR' if request_type == RRType.IXFR() else 'AXFR' if check_soa: ret = self._check_soa_serial() if ret == XFRIN_OK: logger.info(XFRIN_XFR_TRANSFER_STARTED, request_str, self.zone_str()) self._send_query(self._request_type) self.__state = XfrinInitialSOA() self._handle_xfrin_responses() logger.info(XFRIN_XFR_TRANSFER_SUCCESS, request_str, self.zone_str()) except (XfrinException, XfrinProtocolError) as e: logger.error(XFRIN_XFR_TRANSFER_FAILURE, request_str, self.zone_str(), 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, request_str, self.zone_str(), str(e)) ret = XFRIN_FAIL finally: # 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 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_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_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]) recvdata = self._get_request_response(msg_len) msg = Message(Message.PARSE) msg.from_wire(recvdata, Message.PRESERVE_ORDER) # 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 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, tsig_key, request_type): xfrin_recorder.increment(zone_name) # Create a data source client used in this XFR session. Right now we # still assume an sqlite3-based data source, and use both the old and new # data source APIs. We also need to use a mock client for tests. # For a temporary workaround to deal with these situations, we skip the # creation when the given file is none (the test case). Eventually # this code will be much cleaner. datasrc_client = None if db_file is not None: # temporary hardcoded sqlite initialization. Once we decide on # the config specification, we need to update this (TODO) # this may depend on #1207, or any followup ticket created for #1207 datasrc_type = "sqlite3" datasrc_config = "{ \"database_file\": \"" + db_file + "\"}" datasrc_client = DataSourceClient(datasrc_type, datasrc_config) # Create a TCP connection for the XFR session and perform the operation. sock_map = {} conn = XfrinConnection(sock_map, zone_name, rrclass, datasrc_client, shutdown_event, master_addrinfo, tsig_key) ret = XFRIN_FAIL if conn.connect_to_master(): ret = conn.do_xfrin(check_soa, request_type) # 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): self._max_transfers_in = 10 self._zones = {} self._cc_setup() self.recorder = XfrinRecorder() self._shutdown_event = threading.Event() 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, RRType.AXFR(), 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 request_type = RRType.AXFR() if zone_info: tsig_key = zone_info.tsig_key if not zone_info.ixfr_disabled: request_type = RRType.IXFR() db_file = args.get('db_file') or self._get_db_file() ret = self.xfrin_start(zone_name, rrclass, db_file, master_addr, tsig_key, request_type, (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.to_text(), '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, request_type, 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, tsig_key, request_type)) 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="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() if __name__ == '__main__': main(Xfrin)