Skip to content
GitLab
Commit 1177bfe3 authored by Jelte Jansen's avatar Jelte Jansen
Browse files

[master] Merge branch 'trac1298' 

Conflicts:
	src/bin/xfrin/tests/xfrin_test.py
	src/bin/xfrin/xfrin.py.in
parents e6a596fe c03e6df1
Hide whitespace changes
Inline Side-by-side
src/bin/xfrin/tests/xfrin_test.py
View file @ 1177bfe3
......@@ -2022,6 +2022,19 @@ class TestXfrin(unittest.TestCase):
self.assertEqual(self.xfr.command_handler("notify",
self.args)['result'][0], 1)
# also try a different port in the actual command
zones = { 'zones': [
{ 'name': TEST_ZONE_NAME_STR,
'master_addr': TEST_MASTER_IPV6_ADDRESS,
'master_port': str(int(TEST_MASTER_PORT) + 1)
}
]}
self.xfr.config_handler(zones)
# the command should now fail
self.assertEqual(self.xfr.command_handler("notify",
self.args)['result'][0], 1)
def test_command_handler_notify_known_zone(self):
# try it with a known zone
self.args['master'] = TEST_MASTER_IPV6_ADDRESS
......@@ -2037,21 +2050,6 @@ class TestXfrin(unittest.TestCase):
self.assertEqual(self.xfr.command_handler("notify",
self.args)['result'][0], 0)
# Note: The rest of the tests won't pass due to the change in #1298
# We should probably simply remove the test cases, but for now we
# just comment them out. (Note also that the comment about 'not
# from the config' is now wrong, because we used the matching address.)
#
# and see if we used the address from the command, and not from
# the config
# This is actually NOT the address given in the command, which
# would at this point not make sense, see the TODO in
# xfrin.py.in Xfrin.command_handler())
# self.assertEqual(TEST_MASTER_IPV4_ADDRESS,
# self.xfr.xfrin_started_master_addr)
# self.assertEqual(int(TEST_MASTER_PORT),
# self.xfr.xfrin_started_master_port)
def test_command_handler_unknown(self):
self.assertEqual(self.xfr.command_handler("xxx", None)['result'][0], 1)
......@@ -2413,6 +2411,58 @@ class TestXfrinProcess(unittest.TestCase):
"""
self.__do_test([XFRIN_FAIL, XFRIN_FAIL],
[RRType.IXFR(), RRType.AXFR()], RRType.IXFR())
class TestFormatting(unittest.TestCase):
# If the formatting functions are moved to a more general library
# (ticket #1379), these tests should be moved with them.
def test_format_zone_str(self):
self.assertEqual("example.com/IN",
format_zone_str(isc.dns.Name("example.com"),
isc.dns.RRClass("IN")))
self.assertEqual("example.com/CH",
format_zone_str(isc.dns.Name("example.com"),
isc.dns.RRClass("CH")))
self.assertEqual("example.org/IN",
format_zone_str(isc.dns.Name("example.org"),
isc.dns.RRClass("IN")))
def test_format_addrinfo(self):
# This test may need to be updated if the input type is changed,
# right now it is a nested tuple:
# (family, sockettype, (address, port))
# of which sockettype is ignored
self.assertEqual("192.0.2.1:53",
format_addrinfo((socket.AF_INET, socket.SOCK_STREAM,
("192.0.2.1", 53))))
self.assertEqual("192.0.2.2:53",
format_addrinfo((socket.AF_INET, socket.SOCK_STREAM,
("192.0.2.2", 53))))
self.assertEqual("192.0.2.1:54",
format_addrinfo((socket.AF_INET, socket.SOCK_STREAM,
("192.0.2.1", 54))))
self.assertEqual("[2001:db8::1]:53",
format_addrinfo((socket.AF_INET6, socket.SOCK_STREAM,
("2001:db8::1", 53))))
self.assertEqual("[2001:db8::2]:53",
format_addrinfo((socket.AF_INET6, socket.SOCK_STREAM,
("2001:db8::2", 53))))
self.assertEqual("[2001:db8::1]:54",
format_addrinfo((socket.AF_INET6, socket.SOCK_STREAM,
("2001:db8::1", 54))))
self.assertEqual("/some/file",
format_addrinfo((socket.AF_UNIX, socket.SOCK_STREAM,
"/some/file")))
# second element of passed tuple should be ignored
self.assertEqual("192.0.2.1:53",
format_addrinfo((socket.AF_INET, None,
("192.0.2.1", 53))))
self.assertEqual("192.0.2.1:53",
format_addrinfo((socket.AF_INET, "Just some string",
("192.0.2.1", 53))))
self.assertRaises(TypeError, format_addrinfo, 1)
self.assertRaises(TypeError, format_addrinfo,
(socket.AF_INET, "asdf"))
self.assertRaises(TypeError, format_addrinfo,
(socket.AF_INET, "asdf", ()))
if __name__== "__main__":
try:
......
src/bin/xfrin/xfrin.py.in
View file @ 1177bfe3
......@@ -122,6 +122,36 @@ def _check_zone_class(zone_class_str):
except InvalidRRClass as irce:
raise XfrinZoneInfoException("bad zone class: " + zone_class_str + " (" + str(irce) + ")")
def format_zone_str(zone_name, zone_class):
"""Helper function to format a zone name and class as a string of
the form '<name>/<class>'.
Parameters:
zone_name (isc.dns.Name) name to format
zone_class (isc.dns.RRClass) class to format
"""
return zone_name.to_text(True) + '/' + str(zone_class)
def format_addrinfo(addrinfo):
"""Helper function to format the addrinfo as a string of the form
<addr>:<port> (for IPv4) or [<addr>]:port (for IPv6). For unix domain
sockets, and unknown address families, it returns a basic string
conversion of the third element of the passed tuple.
Parameters:
addrinfo: a 3-tuple consisting of address family, socket type, and,
depending on the family, either a 2-tuple with the address
and port, or a filename
"""
try:
if addrinfo[0] == socket.AF_INET:
return str(addrinfo[2][0]) + ":" + str(addrinfo[2][1])
elif addrinfo[0] == socket.AF_INET6:
return "[" + str(addrinfo[2][0]) + "]:" + str(addrinfo[2][1])
else:
return str(addrinfo[2])
except IndexError:
raise TypeError("addrinfo argument to format_addrinfo() does not "
"appear to be consisting of (family, socktype, (addr, port))")
def get_soa_serial(soa_rdata):
'''Extract the serial field of an SOA RDATA and returns it as an intger.
......@@ -498,8 +528,8 @@ class XfrinConnection(asyncore.dispatcher):
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)
'''A convenience function for logging to include zone name and class'''
return format_zone_str(self._zone_name, self._rrclass)
def connect_to_master(self):
'''Connect to master in TCP.'''
......@@ -1094,20 +1124,22 @@ class Xfrin:
# 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_str = format_zone_str(zone_name, rrclass)
zone_info = self._get_zone_info(zone_name, rrclass)
notify_addr = self._parse_master_and_port(args, 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())
errmsg = "Got notification to retransfer unknown zone " + zone_str
logger.info(XFRIN_RETRANSFER_UNKNOWN_ZONE, zone_str)
answer = create_answer(1, errmsg)
else:
request_type = RRType.AXFR()
if zone_info.use_ixfr:
request_type = RRType.IXFR()
master_addr = zone_info.get_master_addr_info()
if notify_addr == master_addr:
if notify_addr[0] == master_addr[0] and\
notify_addr[2] == master_addr[2]:
ret = self.xfrin_start(zone_name,
rrclass,
self._get_db_file(),
......@@ -1116,11 +1148,12 @@ class Xfrin:
True)
answer = create_answer(ret[0], ret[1])
else:
errmsg = "Got notification for " + zone_name.to_text()\
+ "from unknown address: " + notify_addr[2][0];
logger.error(XFRIN_NOTIFY_UNKNOWN_MASTER,
zone_name.to_text(), notify_addr[2][0],
master_addr[2][0])
notify_addr_str = format_addrinfo(notify_addr)
master_addr_str = format_addrinfo(master_addr)
errmsg = "Got notification for " + zone_str\
+ "from unknown address: " + notify_addr_str;
logger.info(XFRIN_NOTIFY_UNKNOWN_MASTER, zone_str,
notify_addr_str, master_addr_str)
answer = create_answer(1, errmsg)
elif command == 'retransfer' or command == 'refresh':
......
src/lib/dns/python/name_python.cc
View file @ 1177bfe3
......@@ -25,6 +25,8 @@
#include "messagerenderer_python.h"
#include "name_python.h"
#include <iostream>
using namespace isc::dns;
using namespace isc::dns::python;
using namespace isc::util;
......@@ -97,7 +99,7 @@ int Name_init(s_Name* self, PyObject* args);
void Name_destroy(s_Name* self);
PyObject* Name_toWire(s_Name* self, PyObject* args);
PyObject* Name_toText(s_Name* self);
PyObject* Name_toText(s_Name* self, PyObject* args);
PyObject* Name_str(PyObject* self);
PyObject* Name_getLabelCount(s_Name* self);
PyObject* Name_at(s_Name* self, PyObject* args);
......@@ -120,8 +122,9 @@ PyMethodDef Name_methods[] = {
"Returns the length" },
{ "get_labelcount", reinterpret_cast<PyCFunction>(Name_getLabelCount), METH_NOARGS,
"Returns the number of labels" },
{ "to_text", reinterpret_cast<PyCFunction>(Name_toText), METH_NOARGS,
"Returns the string representation" },
{ "to_text", reinterpret_cast<PyCFunction>(Name_toText), METH_VARARGS,
"Returns the string representation. The optional argument must be either"
"True of False. If True, the final dot will be omitted." },
{ "to_wire", reinterpret_cast<PyCFunction>(Name_toWire), METH_VARARGS,
"Converts the Name object to wire format.\n"
"The argument can be either a MessageRenderer or an object that "
......@@ -278,8 +281,24 @@ Name_getLabelCount(s_Name* self) {
}
PyObject*
Name_toText(s_Name* self) {
return (Py_BuildValue("s", self->cppobj->toText().c_str()));
Name_toText(s_Name* self, PyObject* args) {
PyObject* omit_final_dot_obj = NULL;
if (PyArg_ParseTuple(args, "|O", &omit_final_dot_obj)) {
bool omit_final_dot = false;
if (omit_final_dot_obj != NULL) {
if (PyBool_Check(omit_final_dot_obj) != 0) {
omit_final_dot = (omit_final_dot_obj == Py_True);
} else {
PyErr_SetString(PyExc_TypeError,
"Optional argument 1 of to_text() should be True of False");
return (NULL);
}
}
return (Py_BuildValue("s",
self->cppobj->toText(omit_final_dot).c_str()));
} else {
return (NULL);
}
}
PyObject*
......
src/lib/dns/python/tests/name_python_test.py
View file @ 1177bfe3
......@@ -121,6 +121,15 @@ class NameTest(unittest.TestCase):
self.assertEqual(".", str(self.name2))
self.assertEqual("something.completely.different.", self.name3.to_text())
self.assertEqual("example.com.", self.name1.to_text(False))
self.assertEqual("example.com", self.name1.to_text(True))
# make sure it does not behave unexpectedly on wrong arguments
self.assertRaises(TypeError, self.name1.to_text, True, 1)
self.assertRaises(TypeError, self.name1.to_text, 1)
self.assertRaises(TypeError, self.name1.to_text, [])
self.assertRaises(TypeError, self.name1.to_text, "foo")
def test_to_wire(self):
b1 = bytearray()
self.name1.to_wire(b1)
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment