+ self.assert_(self._checkRsaPrivateKey(cert1))
+ self.assert_(self._checkCertificate(cert1))
+
+
+class TestPathJoin(unittest.TestCase):
+ """Testing case for PathJoin"""
+
+ def testBasicItems(self):
+ mlist = ["/a", "b", "c"]
+ self.failUnlessEqual(PathJoin(*mlist), "/".join(mlist))
+
+ def testNonAbsPrefix(self):
+ self.failUnlessRaises(ValueError, PathJoin, "a", "b")
+
+ def testBackTrack(self):
+ self.failUnlessRaises(ValueError, PathJoin, "/a", "b/../c")
+
+ def testMultiAbs(self):
+ self.failUnlessRaises(ValueError, PathJoin, "/a", "/b")
+
+
+class TestHostInfo(unittest.TestCase):
+ """Testing case for HostInfo"""
+
+ def testUppercase(self):
+ data = "AbC.example.com"
+ self.failUnlessEqual(HostInfo.NormalizeName(data), data.lower())
+
+ def testTooLongName(self):
+ data = "a.b." + "c" * 255
+ self.failUnlessRaises(OpPrereqError, HostInfo.NormalizeName, data)
+
+ def testTrailingDot(self):
+ data = "a.b.c"
+ self.failUnlessEqual(HostInfo.NormalizeName(data + "."), data)
+
+ def testInvalidName(self):
+ data = [
+ "a b",
+ "a/b",
+ ".a.b",
+ "a..b",
+ ]
+ for value in data:
+ self.failUnlessRaises(OpPrereqError, HostInfo.NormalizeName, value)
+
+ def testValidName(self):
+ data = [
+ "a.b",
+ "a-b",
+ "a_b",
+ "a.b.c",
+ ]
+ for value in data:
+ HostInfo.NormalizeName(value)
+
+
+class TestParseAsn1Generalizedtime(unittest.TestCase):
+ def test(self):
+ # UTC
+ self.assertEqual(utils._ParseAsn1Generalizedtime("19700101000000Z"), 0)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20100222174152Z"),
+ 1266860512)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20380119031407Z"),
+ (2**31) - 1)
+
+ # With offset
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20100222174152+0000"),
+ 1266860512)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20100223131652+0000"),
+ 1266931012)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20100223051808-0800"),
+ 1266931088)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("20100224002135+1100"),
+ 1266931295)
+ self.assertEqual(utils._ParseAsn1Generalizedtime("19700101000000-0100"),
+ 3600)
+
+ # Leap seconds are not supported by datetime.datetime
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime,
+ "19841231235960+0000")
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime,
+ "19920630235960+0000")
+
+ # Errors
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime, "")
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime, "invalid")
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime,
+ "20100222174152")
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime,
+ "Mon Feb 22 17:47:02 UTC 2010")
+ self.assertRaises(ValueError, utils._ParseAsn1Generalizedtime,
+ "2010-02-22 17:42:02")
+
+
+class TestGetX509CertValidity(testutils.GanetiTestCase):
+ def setUp(self):
+ testutils.GanetiTestCase.setUp(self)
+
+ pyopenssl_version = distutils.version.LooseVersion(OpenSSL.__version__)
+
+ # Test whether we have pyOpenSSL 0.7 or above
+ self.pyopenssl0_7 = (pyopenssl_version >= "0.7")
+
+ if not self.pyopenssl0_7:
+ warnings.warn("This test requires pyOpenSSL 0.7 or above to"
+ " function correctly")
+
+ def _LoadCert(self, name):
+ return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
+ self._ReadTestData(name))
+
+ def test(self):
+ validity = utils.GetX509CertValidity(self._LoadCert("cert1.pem"))
+ if self.pyopenssl0_7:
+ self.assertEqual(validity, (1266919967, 1267524767))
+ else:
+ self.assertEqual(validity, (None, None))
+
+
+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)
projects / ganeti-local / blobdiff