+class TestSignX509Certificate(unittest.TestCase):
+ KEY = "My private key!"
+ KEY_OTHER = "Another key"
+
+ def test(self):
+ # Generate certificate valid for 5 minutes
+ (_, cert_pem) = utils.GenerateSelfSignedX509Cert(None, 300)
+
+ cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
+ cert_pem)
+
+ # No signature at all
+ self.assertRaises(errors.GenericError,
+ utils.LoadSignedX509Certificate, cert_pem, self.KEY)
+
+ # Invalid input
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ "", self.KEY)
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ "X-Ganeti-Signature: \n", self.KEY)
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ "X-Ganeti-Sign: $1234$abcdef\n", self.KEY)
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ "X-Ganeti-Signature: $1234567890$abcdef\n", self.KEY)
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ "X-Ganeti-Signature: $1234$abc\n\n" + cert_pem, self.KEY)
+
+ # Invalid salt
+ for salt in list("-_@$,:;/\\ \t\n"):
+ self.assertRaises(errors.GenericError, utils.SignX509Certificate,
+ cert_pem, self.KEY, "foo%sbar" % salt)
+
+ for salt in ["HelloWorld", "salt", string.letters, string.digits,
+ utils.GenerateSecret(numbytes=4),
+ utils.GenerateSecret(numbytes=16),
+ "{123:456}".encode("hex")]:
+ signed_pem = utils.SignX509Certificate(cert, self.KEY, salt)
+
+ self._Check(cert, salt, signed_pem)
+
+ self._Check(cert, salt, "X-Another-Header: with a value\n" + signed_pem)
+ self._Check(cert, salt, (10 * "Hello World!\n") + signed_pem)
+ self._Check(cert, salt, (signed_pem + "\n\na few more\n"
+ "lines----\n------ at\nthe end!"))
+
+ def _Check(self, cert, salt, pem):
+ (cert2, salt2) = utils.LoadSignedX509Certificate(pem, self.KEY)
+ self.assertEqual(salt, salt2)
+ self.assertEqual(cert.digest("sha1"), cert2.digest("sha1"))
+
+ # Other key
+ self.assertRaises(errors.GenericError, utils.LoadSignedX509Certificate,
+ pem, self.KEY_OTHER)
+
+
+class TestMakedirs(unittest.TestCase):
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def testNonExisting(self):
+ path = utils.PathJoin(self.tmpdir, "foo")
+ utils.Makedirs(path)
+ self.assert_(os.path.isdir(path))
+
+ def testExisting(self):
+ path = utils.PathJoin(self.tmpdir, "foo")
+ os.mkdir(path)
+ utils.Makedirs(path)
+ self.assert_(os.path.isdir(path))
+
+ def testRecursiveNonExisting(self):
+ path = utils.PathJoin(self.tmpdir, "foo/bar/baz")
+ utils.Makedirs(path)
+ self.assert_(os.path.isdir(path))
+
+ def testRecursiveExisting(self):
+ path = utils.PathJoin(self.tmpdir, "B/moo/xyz")
+ self.assert_(not os.path.exists(path))
+ os.mkdir(utils.PathJoin(self.tmpdir, "B"))
+ utils.Makedirs(path)
+ self.assert_(os.path.isdir(path))
+
+
+class TestRetry(testutils.GanetiTestCase):
+ def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
+ self.retries = 0
+
+ @staticmethod
+ def _RaiseRetryAgain():
+ raise utils.RetryAgain()
+
+ @staticmethod
+ def _RaiseRetryAgainWithArg(args):
+ raise utils.RetryAgain(*args)
+
+ def _WrongNestedLoop(self):
+ return utils.Retry(self._RaiseRetryAgain, 0.01, 0.02)
+
+ def _RetryAndSucceed(self, retries):
+ if self.retries < retries:
+ self.retries += 1
+ raise utils.RetryAgain()
+ else:
+ return True
+
+ def testRaiseTimeout(self):
+ self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
+ self._RaiseRetryAgain, 0.01, 0.02)
+ self.failUnlessRaises(utils.RetryTimeout, utils.Retry,
+ self._RetryAndSucceed, 0.01, 0, args=[1])
+ self.failUnlessEqual(self.retries, 1)
+
+ def testComplete(self):
+ self.failUnlessEqual(utils.Retry(lambda: True, 0, 1), True)
+ self.failUnlessEqual(utils.Retry(self._RetryAndSucceed, 0, 1, args=[2]),
+ True)
+ self.failUnlessEqual(self.retries, 2)
+
+ def testNestedLoop(self):
+ try:
+ self.failUnlessRaises(errors.ProgrammerError, utils.Retry,
+ self._WrongNestedLoop, 0, 1)
+ except utils.RetryTimeout:
+ self.fail("Didn't detect inner loop's exception")
+
+ def testTimeoutArgument(self):
+ retry_arg="my_important_debugging_message"
+ try:
+ utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02, args=[[retry_arg]])
+ except utils.RetryTimeout, err:
+ self.failUnlessEqual(err.args, (retry_arg, ))
+ else:
+ self.fail("Expected timeout didn't happen")
+
+ def testRaiseInnerWithExc(self):
+ retry_arg="my_important_debugging_message"
+ try:
+ try:
+ utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
+ args=[[errors.GenericError(retry_arg, retry_arg)]])
+ except utils.RetryTimeout, err:
+ err.RaiseInner()
+ else:
+ self.fail("Expected timeout didn't happen")
+ except errors.GenericError, err:
+ self.failUnlessEqual(err.args, (retry_arg, retry_arg))
+ else:
+ self.fail("Expected GenericError didn't happen")
+
+ def testRaiseInnerWithMsg(self):
+ retry_arg="my_important_debugging_message"
+ try:
+ try:
+ utils.Retry(self._RaiseRetryAgainWithArg, 0.01, 0.02,
+ args=[[retry_arg, retry_arg]])
+ except utils.RetryTimeout, err:
+ err.RaiseInner()
+ else:
+ self.fail("Expected timeout didn't happen")
+ except utils.RetryTimeout, err:
+ self.failUnlessEqual(err.args, (retry_arg, retry_arg))
+ else:
+ self.fail("Expected RetryTimeout didn't happen")
+
+
+class TestLineSplitter(unittest.TestCase):
+ def test(self):
+ lines = []
+ ls = utils.LineSplitter(lines.append)
+ ls.write("Hello World\n")
+ self.assertEqual(lines, [])
+ ls.write("Foo\n Bar\r\n ")
+ ls.write("Baz")
+ ls.write("Moo")
+ self.assertEqual(lines, [])
+ ls.flush()
+ self.assertEqual(lines, ["Hello World", "Foo", " Bar"])
+ ls.close()
+ self.assertEqual(lines, ["Hello World", "Foo", " Bar", " BazMoo"])
+
+ def _testExtra(self, line, all_lines, p1, p2):
+ self.assertEqual(p1, 999)
+ self.assertEqual(p2, "extra")
+ all_lines.append(line)
+
+ def testExtraArgsNoFlush(self):
+ lines = []
+ ls = utils.LineSplitter(self._testExtra, lines, 999, "extra")
+ ls.write("\n\nHello World\n")
+ ls.write("Foo\n Bar\r\n ")
+ ls.write("")
+ ls.write("Baz")
+ ls.write("Moo\n\nx\n")
+ self.assertEqual(lines, [])
+ ls.close()
+ self.assertEqual(lines, ["", "", "Hello World", "Foo", " Bar", " BazMoo",
+ "", "x"])
+
+
+class TestReadLockedPidFile(unittest.TestCase):
+ def setUp(self):
+ self.tmpdir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def testNonExistent(self):
+ path = utils.PathJoin(self.tmpdir, "nonexist")
+ self.assert_(utils.ReadLockedPidFile(path) is None)
+
+ def testUnlocked(self):
+ path = utils.PathJoin(self.tmpdir, "pid")
+ utils.WriteFile(path, data="123")
+ self.assert_(utils.ReadLockedPidFile(path) is None)
+
+ def testLocked(self):
+ path = utils.PathJoin(self.tmpdir, "pid")
+ utils.WriteFile(path, data="123")
+
+ fl = utils.FileLock.Open(path)
+ try:
+ fl.Exclusive(blocking=True)
+
+ self.assertEqual(utils.ReadLockedPidFile(path), 123)
+ finally:
+ fl.Close()
+
+ self.assert_(utils.ReadLockedPidFile(path) is None)
+
+ def testError(self):
+ path = utils.PathJoin(self.tmpdir, "foobar", "pid")
+ utils.WriteFile(utils.PathJoin(self.tmpdir, "foobar"), data="")
+ # open(2) should return ENOTDIR
+ self.assertRaises(EnvironmentError, utils.ReadLockedPidFile, path)
+
+
+class TestCertVerification(testutils.GanetiTestCase):
+ def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
+
+ self.tmpdir = tempfile.mkdtemp()
+
+ def tearDown(self):
+ shutil.rmtree(self.tmpdir)
+
+ def testVerifyCertificate(self):
+ cert_pem = utils.ReadFile(self._TestDataFilename("cert1.pem"))
+ cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
+ cert_pem)
+
+ # Not checking return value as this certificate is expired
+ utils.VerifyX509Certificate(cert, 30, 7)
+
+
+class TestVerifyCertificateInner(unittest.TestCase):
+ def test(self):
+ vci = utils._VerifyCertificateInner
+
+ # Valid
+ self.assertEqual(vci(False, 1263916313, 1298476313, 1266940313, 30, 7),
+ (None, None))
+
+ # Not yet valid
+ (errcode, msg) = vci(False, 1266507600, 1267544400, 1266075600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_WARNING)
+
+ # Expiring soon
+ (errcode, msg) = vci(False, 1266507600, 1267544400, 1266939600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_ERROR)
+
+ (errcode, msg) = vci(False, 1266507600, 1267544400, 1266939600, 30, 1)
+ self.assertEqual(errcode, utils.CERT_WARNING)
+
+ (errcode, msg) = vci(False, 1266507600, None, 1266939600, 30, 7)
+ self.assertEqual(errcode, None)
+
+ # Expired
+ (errcode, msg) = vci(True, 1266507600, 1267544400, 1266939600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_ERROR)
+
+ (errcode, msg) = vci(True, None, 1267544400, 1266939600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_ERROR)
+
+ (errcode, msg) = vci(True, 1266507600, None, 1266939600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_ERROR)
+
+ (errcode, msg) = vci(True, None, None, 1266939600, 30, 7)
+ self.assertEqual(errcode, utils.CERT_ERROR)
+
+
+class TestHmacFunctions(unittest.TestCase):
+ # Digests can be checked with "openssl sha1 -hmac $key"
+ def testSha1Hmac(self):
+ self.assertEqual(utils.Sha1Hmac("", ""),
+ "fbdb1d1b18aa6c08324b7d64b71fb76370690e1d")
+ self.assertEqual(utils.Sha1Hmac("3YzMxZWE", "Hello World"),
+ "ef4f3bda82212ecb2f7ce868888a19092481f1fd")
+ self.assertEqual(utils.Sha1Hmac("TguMTA2K", ""),
+ "f904c2476527c6d3e6609ab683c66fa0652cb1dc")
+
+ longtext = 1500 * "The quick brown fox jumps over the lazy dog\n"
+ self.assertEqual(utils.Sha1Hmac("3YzMxZWE", longtext),
+ "35901b9a3001a7cdcf8e0e9d7c2e79df2223af54")
+
+ def testSha1HmacSalt(self):
+ self.assertEqual(utils.Sha1Hmac("TguMTA2K", "", salt="abc0"),
+ "4999bf342470eadb11dfcd24ca5680cf9fd7cdce")
+ self.assertEqual(utils.Sha1Hmac("TguMTA2K", "", salt="abc9"),
+ "17a4adc34d69c0d367d4ffbef96fd41d4df7a6e8")
+ self.assertEqual(utils.Sha1Hmac("3YzMxZWE", "Hello World", salt="xyz0"),
+ "7f264f8114c9066afc9bb7636e1786d996d3cc0d")
+
+ def testVerifySha1Hmac(self):
+ self.assert_(utils.VerifySha1Hmac("", "", ("fbdb1d1b18aa6c08324b"
+ "7d64b71fb76370690e1d")))
+ self.assert_(utils.VerifySha1Hmac("TguMTA2K", "",
+ ("f904c2476527c6d3e660"
+ "9ab683c66fa0652cb1dc")))
+
+ digest = "ef4f3bda82212ecb2f7ce868888a19092481f1fd"
+ self.assert_(utils.VerifySha1Hmac("3YzMxZWE", "Hello World", digest))
+ self.assert_(utils.VerifySha1Hmac("3YzMxZWE", "Hello World",
+ digest.lower()))
+ self.assert_(utils.VerifySha1Hmac("3YzMxZWE", "Hello World",
+ digest.upper()))
+ self.assert_(utils.VerifySha1Hmac("3YzMxZWE", "Hello World",
+ digest.title()))
+
+ def testVerifySha1HmacSalt(self):
+ self.assert_(utils.VerifySha1Hmac("TguMTA2K", "",
+ ("17a4adc34d69c0d367d4"
+ "ffbef96fd41d4df7a6e8"),
+ salt="abc9"))
+ self.assert_(utils.VerifySha1Hmac("3YzMxZWE", "Hello World",
+ ("7f264f8114c9066afc9b"
+ "b7636e1786d996d3cc0d"),
+ salt="xyz0"))
+
+
+class TestIgnoreSignals(unittest.TestCase):
+ """Test the IgnoreSignals decorator"""
+
+ @staticmethod
+ def _Raise(exception):
+ raise exception
+
+ @staticmethod
+ def _Return(rval):
+ return rval
+
+ def testIgnoreSignals(self):
+ sock_err_intr = socket.error(errno.EINTR, "Message")
+ sock_err_intr.errno = errno.EINTR
+ sock_err_inval = socket.error(errno.EINVAL, "Message")
+ sock_err_inval.errno = errno.EINVAL
+
+ env_err_intr = EnvironmentError(errno.EINTR, "Message")
+ env_err_inval = EnvironmentError(errno.EINVAL, "Message")
+
+ self.assertRaises(socket.error, self._Raise, sock_err_intr)
+ self.assertRaises(socket.error, self._Raise, sock_err_inval)
+ self.assertRaises(EnvironmentError, self._Raise, env_err_intr)
+ self.assertRaises(EnvironmentError, self._Raise, env_err_inval)
+
+ self.assertEquals(utils.IgnoreSignals(self._Raise, sock_err_intr), None)
+ self.assertEquals(utils.IgnoreSignals(self._Raise, env_err_intr), None)
+ self.assertRaises(socket.error, utils.IgnoreSignals, self._Raise,
+ sock_err_inval)
+ self.assertRaises(EnvironmentError, utils.IgnoreSignals, self._Raise,
+ env_err_inval)
+
+ self.assertEquals(utils.IgnoreSignals(self._Return, True), True)
+ self.assertEquals(utils.IgnoreSignals(self._Return, 33), 33)
+
+