new_rrset.add_rdata(isc.dns.Rdata(rrset.get_type(), rrclass, wire))
return new_rrset
+ def collect_rrsets(collection, rrset):
+ '''
+ Helper function to collect similar rrsets.
+ Collect all rrsets with the same name, class, and type
+ collection is the currently collected list of RRsets,
+ rrset is the RRset to add;
+ if an RRset with the same name, class and type as the
+ given rrset exists in the collection, its rdata fields
+ are added to that RRset. Otherwise, the rrset is added
+ to the given collection.
+ TTL is ignored.
+ This method does not check rdata contents for duplicate
+ values.
+
+ The collection and its rrsets are modified in-place,
+ this method does not return anything.
+ '''
+ found = False
+ for existing_rrset in collection:
+ if existing_rrset.get_name() == rrset.get_name() and\
+ existing_rrset.get_class() == rrset.get_class() and\
+ existing_rrset.get_type() == rrset.get_type():
+ for rdata in rrset.get_rdata():
+ existing_rrset.add_rdata(rdata)
+ found = True
+ if not found:
+ collection.append(rrset)
+
+class DDNS_SOA:
+ '''Class to handle the SOA in the DNS update '''
+
+ def __get_serial_internal(self, origin_soa):
+ '''Get serial number from soa'''
+ return Serial(int(origin_soa.get_rdata()[0].to_text().split()[2]))
+
+ def __write_soa_internal(self, origin_soa, soa_num):
+ '''Write back serial number to soa'''
+ new_soa = RRset(origin_soa.get_name(), origin_soa.get_class(),
+ RRType.SOA(), origin_soa.get_ttl())
+ soa_rdata_parts = origin_soa.get_rdata()[0].to_text().split()
+ soa_rdata_parts[2] = str(soa_num.get_value())
+ new_soa.add_rdata(Rdata(origin_soa.get_type(), origin_soa.get_class(),
+ " ".join(soa_rdata_parts)))
+ return new_soa
+
+ def soa_update_check(self, origin_soa, new_soa):
+ '''Check whether the new soa is valid. If the serial number is bigger
+ than the old one, it is valid, then return True, otherwise, return
+ False. Make sure the origin_soa and new_soa parameters are not none
+ before invoke soa_update_check.
+ Parameters:
+ origin_soa, old SOA resource record.
+ new_soa, new SOA resource record.
+ Output:
+ if the serial number of new soa is bigger than the old one, return
+ True, otherwise return False.
+ '''
+ old_serial = self.__get_serial_internal(origin_soa)
+ new_serial = self.__get_serial_internal(new_soa)
+ if(new_serial > old_serial):
+ return True
+ else:
+ return False
+
+ def update_soa(self, origin_soa, inc_number = 1):
+ ''' Update the soa number incrementally as RFC 2136. Please make sure
+ that the origin_soa exists and not none before invoke this function.
+ Parameters:
+ origin_soa, the soa resource record which will be updated.
+ inc_number, the number which will be added into the serial number of
+ origin_soa, the default value is one.
+ Output:
+ The new origin soa whoes serial number has been updated.
+ '''
+ soa_num = self.__get_serial_internal(origin_soa)
+ soa_num = soa_num + inc_number
+ if soa_num.get_value() == 0:
+ soa_num = soa_num + 1
+ return self.__write_soa_internal(origin_soa, soa_num)
+
class UpdateSession:
'''Protocol handling for a single dynamic update request.
# serial magic and add the newly created one
# get it from DS and to increment and stuff
- result, old_soa, _ = self.__diff.find(self.__zname, RRType.SOA())
-
- if self.__added_soa is not None:
- new_soa = self.__added_soa
- # serial check goes here
+ result, old_soa, _ = self.__finder.find(self.__zname, RRType.SOA(),
+ ZoneFinder.NO_WILDCARD |
+ ZoneFinder.FIND_GLUE_OK)
+ # We may implement recovering from missing SOA data at some point, but
+ # for now servfail on such a broken state
+ if result != ZoneFinder.SUCCESS:
+ raise UpdateError("Error finding SOA record in datasource.",
+ self.__zname, self.__zclass, Rcode.SERVFAIL())
+ serial_operation = DDNS_SOA()
+ if self.__added_soa is not None and\
+ serial_operation.soa_update_check(old_soa, self.__added_soa):
+ new_soa = self.__added_soa
else:
- new_soa = old_soa
# increment goes here
+ new_soa = serial_operation.update_soa(old_soa)
- diff.delete_data(old_soa)
- diff.add_data(new_soa)
+ self.__diff.delete_data(old_soa)
+ self.__diff.add_data(new_soa)
def __do_update(self):
'''Scan, check, and execute the Update section in the
add_rdata(rrset, rdata)
return rrset
- def add_rdata(rrset, rdata):
- '''
- Helper function for easily adding Rdata fields to RRsets.
- This function assumes the given rdata is of type string or bytes,
- and corresponds to the given rrset
+ class SessionModuleTests(unittest.TestCase):
+ '''Tests for module-level functions in the session.py module'''
+
+ def test_foreach_rr_in_rrset(self):
+ rrset = create_rrset("www.example.org", TEST_RRCLASS,
+ RRType.A(), 3600, [ "192.0.2.1" ])
+
+ l = []
+ for rr in foreach_rr(rrset):
+ l.append(str(rr))
+ self.assertEqual(["www.example.org. 3600 IN A 192.0.2.1\n"], l)
+
+ add_rdata(rrset, "192.0.2.2")
+ add_rdata(rrset, "192.0.2.3")
+
+ # but through the generator, there should be several 1-line entries
+ l = []
+ for rr in foreach_rr(rrset):
+ l.append(str(rr))
+ self.assertEqual(["www.example.org. 3600 IN A 192.0.2.1\n",
+ "www.example.org. 3600 IN A 192.0.2.2\n",
+ "www.example.org. 3600 IN A 192.0.2.3\n",
+ ], l)
+
+ def test_convert_rrset_class(self):
+ # Converting an RRSET to a different class should work
+ # if the rdata types can be converted
+ rrset = create_rrset("www.example.org", RRClass.NONE(), RRType.A(),
+ 3600, [ b'\xc0\x00\x02\x01', b'\xc0\x00\x02\x02'])
+
+ rrset2 = convert_rrset_class(rrset, RRClass.IN())
+ self.assertEqual("www.example.org. 3600 IN A 192.0.2.1\n" +
+ "www.example.org. 3600 IN A 192.0.2.2\n",
+ str(rrset2))
+
+ rrset3 = convert_rrset_class(rrset2, RRClass.NONE())
+ self.assertEqual("www.example.org. 3600 CLASS254 A \\# 4 " +
+ "c0000201\nwww.example.org. 3600 CLASS254 " +
+ "A \\# 4 c0000202\n",
+ str(rrset3))
+
+ # depending on what type of bad data is given, a number
+ # of different exceptions could be raised (TODO: i recall
+ # there was a ticket about making a better hierarchy for
+ # dns/parsing related exceptions)
+ self.assertRaises(InvalidRdataLength, convert_rrset_class,
+ rrset, RRClass.CH())
+ add_rdata(rrset, b'\xc0\x00')
+ self.assertRaises(DNSMessageFORMERR, convert_rrset_class,
+ rrset, RRClass.IN())
+
+ def test_collect_rrsets(self):
+ '''
+ Tests the 'rrset collector' method, which collects rrsets
+ with the same name and type
+ '''
+ collected = []
+
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
+ RRType.A(), 0, [ "192.0.2.1" ]))
+ # Same name and class, different type
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
+ RRType.TXT(), 0, [ "one" ]))
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
+ RRType.A(), 0, [ "192.0.2.2" ]))
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.IN(),
+ RRType.TXT(), 0, [ "two" ]))
+ # Same class and type as an existing one, different name
+ collect_rrsets(collected, create_rrset("b.example.org", RRClass.IN(),
+ RRType.A(), 0, [ "192.0.2.3" ]))
+ # Same name and type as an existing one, different class
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.CH(),
+ RRType.TXT(), 0, [ "one" ]))
+ collect_rrsets(collected, create_rrset("b.example.org", RRClass.IN(),
+ RRType.A(), 0, [ "192.0.2.4" ]))
+ collect_rrsets(collected, create_rrset("a.example.org", RRClass.CH(),
+ RRType.TXT(), 0, [ "two" ]))
+
+ strings = [ rrset.to_text() for rrset in collected ]
+ # note + vs , in this list
+ expected = ['a.example.org. 0 IN A 192.0.2.1\n' +
+ 'a.example.org. 0 IN A 192.0.2.2\n',
+ 'a.example.org. 0 IN TXT "one"\n' +
+ 'a.example.org. 0 IN TXT "two"\n',
+ 'b.example.org. 0 IN A 192.0.2.3\n' +
+ 'b.example.org. 0 IN A 192.0.2.4\n',
+ 'a.example.org. 0 CH TXT "one"\n' +
+ 'a.example.org. 0 CH TXT "two"\n']
+
+ self.assertEqual(expected, strings)
+
+ class SessionTestBase(unittest.TestCase):
+ '''Base class for all sesion related tests.
+
+ It just initializes common test parameters in its setUp() and defines
+ some common utility method(s).
+
'''
- rrset.add_rdata(isc.dns.Rdata(rrset.get_type(),
- rrset.get_class(),
- rdata))
+ def setUp(self):
+ shutil.copyfile(READ_ZONE_DB_FILE, WRITE_ZONE_DB_FILE)
+ self._datasrc_client = DataSourceClient("sqlite3",
+ WRITE_ZONE_DB_CONFIG)
+ self._update_msg = create_update_msg()
+ self._acl_map = {(TEST_ZONE_NAME, TEST_RRCLASS):
+ REQUEST_LOADER.load([{"action": "ACCEPT"}])}
+ self._session = UpdateSession(self._update_msg, TEST_CLIENT4,
+ ZoneConfig([], TEST_RRCLASS,
+ self._datasrc_client,
+ self._acl_map))
+ self._session._get_update_zone()
+ self._session._create_diff()
+
+ def tearDown(self):
+ # With the Updater created in _get_update_zone, and tests
+ # doing all kinds of crazy stuff, one might get database locked
+ # errors if it doesn't clean up explicitely after each test
+ self._session = None
+
+ def check_response(self, msg, expected_rcode):
+ '''Perform common checks on update resposne message.'''
+ self.assertTrue(msg.get_header_flag(Message.HEADERFLAG_QR))
+ # note: we convert opcode to text it'd be more helpful on failure.
+ self.assertEqual(Opcode.UPDATE().to_text(), msg.get_opcode().to_text())
+ self.assertEqual(expected_rcode.to_text(), msg.get_rcode().to_text())
+ # All sections should be cleared
+ self.assertEqual(0, msg.get_rr_count(SECTION_ZONE))
+ self.assertEqual(0, msg.get_rr_count(SECTION_PREREQUISITE))
+ self.assertEqual(0, msg.get_rr_count(SECTION_UPDATE))
+ self.assertEqual(0, msg.get_rr_count(Message.SECTION_ADDITIONAL))
-
- class SessionTest(unittest.TestCase):
- '''Session tests'''
- def setUp(self):
- shutil.copyfile(READ_ZONE_DB_FILE, WRITE_ZONE_DB_FILE)
- self.__datasrc_client = DataSourceClient("sqlite3",
- WRITE_ZONE_DB_CONFIG)
- self.__update_msgdata, self.__update_msg = create_update_msg()
- self.__session = UpdateSession(self.__update_msg,
- self.__update_msgdata, TEST_CLIENT4,
- ZoneConfig([], TEST_RRCLASS,
- self.__datasrc_client))
- self.__session._UpdateSession__get_update_zone()
-
- def check_response(self, msg, expected_rcode):
- '''Perform common checks on update resposne message.'''
- self.assertTrue(msg.get_header_flag(Message.HEADERFLAG_QR))
- # note: we convert opcode to text it'd be more helpful on failure.
- self.assertEqual(Opcode.UPDATE().to_text(), msg.get_opcode().to_text())
- self.assertEqual(expected_rcode.to_text(), msg.get_rcode().to_text())
- # All sections should be cleared
- self.assertEqual(0, msg.get_rr_count(SECTION_ZONE))
- self.assertEqual(0, msg.get_rr_count(SECTION_PREREQUISITE))
- self.assertEqual(0, msg.get_rr_count(SECTION_UPDATE))
- self.assertEqual(0, msg.get_rr_count(Message.SECTION_ADDITIONAL))
+class TestDDNSSOA(unittest.TestCase):
+ '''unittest for the DDNS_SOA'''
+ def test_update_soa(self):
+ '''unittest for update_soa function'''
+ soa_update = DDNS_SOA()
+ soa_rr = create_rrset("example.org", TEST_RRCLASS,
+ RRType.SOA(), 3600, ["ns1.example.org. " +
+ "admin.example.org. " +
+ "1233 3600 1800 2419200 7200"])
+ expected_soa_rr = create_rrset("example.org", TEST_RRCLASS,
+ RRType.SOA(), 3600, ["ns1.example.org. "
+ + "admin.example.org. " +
+ "1234 3600 1800 2419200 7200"])
+ self.assertEqual(soa_update.update_soa(soa_rr).get_rdata()[0].to_text(),
+ expected_soa_rr.get_rdata()[0].to_text())
+ max_serial = 2 ** 32 - 1
+ soa_rdata = "%d %s"%(max_serial,"3600 1800 2419200 7200")
+ soa_rr = create_rrset("example.org", TEST_RRCLASS, RRType.SOA(), 3600,
+ ["ns1.example.org. " + "admin.example.org. " +
+ soa_rdata])
+ expected_soa_rr = create_rrset("example.org", TEST_RRCLASS,
+ RRType.SOA(), 3600, ["ns1.example.org. "
+ + "admin.example.org. " +
+ "1 3600 1800 2419200 7200"])
+ self.assertEqual(soa_update.update_soa(soa_rr).get_rdata()[0].to_text(),
+ expected_soa_rr.get_rdata()[0].to_text())
+
+ def test_soa_update_check(self):
+ '''unittest for soa_update_check function'''
+ small_soa_rr = create_rrset("example.org", TEST_RRCLASS, RRType.SOA(),
+ 3600, ["ns1.example.org. " +
+ "admin.example.org. " +
+ "1233 3600 1800 2419200 7200"])
+ large_soa_rr = create_rrset("example.org", TEST_RRCLASS, RRType.SOA(),
+ 3600, ["ns1.example.org. " +
+ "admin.example.org. " +
+ "1234 3600 1800 2419200 7200"])
+ soa_update = DDNS_SOA()
+ # The case of (i1 < i2 and i2 - i1 < 2^(SERIAL_BITS - 1)) in rfc 1982
+ self.assertTrue(soa_update.soa_update_check(small_soa_rr,
+ large_soa_rr))
+ self.assertFalse(soa_update.soa_update_check(large_soa_rr,
+ small_soa_rr))
+ small_serial = 1235 + 2 ** 31
+ soa_rdata = "%d %s"%(small_serial,"3600 1800 2419200 7200")
+ small_soa_rr = create_rrset("example.org", TEST_RRCLASS, RRType.SOA(),
+ 3600, ["ns1.example.org. " +
+ "admin.example.org. " +
+ soa_rdata])
+ large_soa_rr = create_rrset("example.org", TEST_RRCLASS, RRType.SOA(),
+ 3600, ["ns1.example.org. " +
+ "admin.example.org. " +
+ "1234 3600 1800 2419200 7200"])
+ # The case of (i1 > i2 and i1 - i2 > 2^(SERIAL_BITS - 1)) in rfc 1982
+ self.assertTrue(soa_update.soa_update_check(small_soa_rr,
+ large_soa_rr))
+ self.assertFalse(soa_update.soa_update_check(large_soa_rr,
+ small_soa_rr))
+
+ class SessionTest(SessionTestBase):
+ '''Basic session tests'''
def test_handle(self):
'''Basic update case'''
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_soa_apex,
self.rrset_update_soa_del ])
- self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
- isc.dns.Name("example.org"),
- RRType.SOA(),
- orig_soa_rrset)
+ self.check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("example.org"),
+ RRType.SOA(),
+ incremented_soa_rrset_01)
# If we delete everything at the apex, the SOA and NS rrsets should be
- # untouched
+ # untouched (but serial will be incremented)
self.check_full_handle_result(Rcode.NOERROR(),
[ self.rrset_update_del_name_apex ])
- self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
- isc.dns.Name("example.org"),
- RRType.SOA(),
- orig_soa_rrset)
- self.__check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
- isc.dns.Name("example.org"),
- RRType.NS(),
- orig_ns_rrset)
+ self.check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("example.org"),
+ RRType.SOA(),
+ incremented_soa_rrset_02)
+ self.check_inzone_data(isc.datasrc.ZoneFinder.SUCCESS,
+ isc.dns.Name("example.org"),
+ RRType.NS(),
+ orig_ns_rrset)
# but the MX should be gone
- self.check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
- isc.dns.Name("example.org"),
- RRType.MX())
+ self.__check_inzone_data(isc.datasrc.ZoneFinder.NXRRSET,
+ isc.dns.Name("example.org"),
+ RRType.MX())
# Deleting the NS rrset by name and type only, it should also be left
# untouched
projects / thirdparty / kea.git / commitdiff
