lib/cpp/test/TNonblockingSSLServerTest.cpp - packaging/sources/thrift - Gitiles

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements. See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership. The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License. You may obtain a copy of the License at
  *
  *   http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  * KIND, either express or implied. See the License for the
  * specific language governing permissions and limitations
  * under the License.
  */

 #define BOOST_TEST_MODULE TNonblockingSSLServerTest
 #include <boost/test/unit_test.hpp>
 #include <boost/smart_ptr.hpp>
 #include <boost/shared_ptr.hpp>
 #include <boost/filesystem.hpp>
 #include <boost/format.hpp>

 #include "thrift/server/TNonblockingServer.h"
 #include "thrift/transport/TSSLSocket.h"
 #include "thrift/transport/TNonblockingSSLServerSocket.h"

 #include "gen-cpp/ParentService.h"

 #include <event.h>

 using namespace apache::thrift;
 using apache::thrift::concurrency::Guard;
 using apache::thrift::concurrency::Monitor;
 using apache::thrift::concurrency::Mutex;
 using apache::thrift::server::TServerEventHandler;
 using apache::thrift::transport::TSSLSocketFactory;
 using apache::thrift::transport::TSSLSocket;

 struct Handler : public test::ParentServiceIf {
   void addString(const std::string& s) { strings_.push_back(s); }
   void getStrings(std::vector<std::string>& _return) { _return = strings_; }
   std::vector<std::string> strings_;

   // dummy overrides not used in this test
   int32_t incrementGeneration() { return 0; }
   int32_t getGeneration() { return 0; }
   void getDataWait(std::string&, const int32_t) {}
   void onewayWait() {}
   void exceptionWait(const std::string&) {}
   void unexpectedExceptionWait(const std::string&) {}
 };

 boost::filesystem::path keyDir;
 boost::filesystem::path certFile(const std::string& filename)
 {
   return keyDir / filename;
 }

 struct GlobalFixtureSSL
 {
     GlobalFixtureSSL()
     {
       using namespace boost::unit_test::framework;
       for (int i = 0; i < master_test_suite().argc; ++i)
       {
         BOOST_TEST_MESSAGE(boost::format("argv[%1%] = \"%2%\"") % i % master_test_suite().argv[i]);
       }

 #ifdef __linux__
       // OpenSSL calls send() without MSG_NOSIGPIPE so writing to a socket that has
       // disconnected can cause a SIGPIPE signal...
       signal(SIGPIPE, SIG_IGN);
 #endif

       TSSLSocketFactory::setManualOpenSSLInitialization(true);
       apache::thrift::transport::initializeOpenSSL();

       keyDir = boost::filesystem::current_path().parent_path().parent_path().parent_path() / "test" / "keys";
       if (!boost::filesystem::exists(certFile("server.crt")))
       {
         keyDir = boost::filesystem::path(master_test_suite().argv[master_test_suite().argc - 1]);
         if (!boost::filesystem::exists(certFile("server.crt")))
         {
           throw std::invalid_argument("The last argument to this test must be the directory containing the test certificate(s).");
         }
       }
     }

     virtual ~GlobalFixtureSSL()
     {
       apache::thrift::transport::cleanupOpenSSL();
 #ifdef __linux__
       signal(SIGPIPE, SIG_DFL);
 #endif
     }
 };

 #if (BOOST_VERSION >= 105900)
 BOOST_GLOBAL_FIXTURE(GlobalFixtureSSL);
 #else
 BOOST_GLOBAL_FIXTURE(GlobalFixtureSSL)
 #endif

 stdcxx::shared_ptr<TSSLSocketFactory> createServerSocketFactory() {
   stdcxx::shared_ptr<TSSLSocketFactory> pServerSocketFactory;

   pServerSocketFactory.reset(new TSSLSocketFactory());
   pServerSocketFactory->ciphers("ALL:!ADH:!LOW:!EXP:!MD5:@STRENGTH");
   pServerSocketFactory->loadCertificate(certFile("server.crt").string().c_str());
   pServerSocketFactory->loadPrivateKey(certFile("server.key").string().c_str());
   pServerSocketFactory->server(true);
   return pServerSocketFactory;
 }

 stdcxx::shared_ptr<TSSLSocketFactory> createClientSocketFactory() {
   stdcxx::shared_ptr<TSSLSocketFactory> pClientSocketFactory;

   pClientSocketFactory.reset(new TSSLSocketFactory());
   pClientSocketFactory->authenticate(true);
   pClientSocketFactory->loadCertificate(certFile("client.crt").string().c_str());
   pClientSocketFactory->loadPrivateKey(certFile("client.key").string().c_str());
   pClientSocketFactory->loadTrustedCertificates(certFile("CA.pem").string().c_str());
   return pClientSocketFactory;
 }

 class Fixture {
 private:
   struct ListenEventHandler : public TServerEventHandler {
     public:
       ListenEventHandler(Mutex* mutex) : listenMonitor_(mutex), ready_(false) {}

       void preServe() /* override */ {
         Guard g(listenMonitor_.mutex());
         ready_ = true;
         listenMonitor_.notify();
       }

       Monitor listenMonitor_;
       bool ready_;
   };

   struct Runner : public apache::thrift::concurrency::Runnable {
     int port;
     stdcxx::shared_ptr<event_base> userEventBase;
     stdcxx::shared_ptr<TProcessor> processor;
     stdcxx::shared_ptr<server::TNonblockingServer> server;
     stdcxx::shared_ptr<ListenEventHandler> listenHandler;
     stdcxx::shared_ptr<TSSLSocketFactory> pServerSocketFactory;
     stdcxx::shared_ptr<transport::TNonblockingSSLServerSocket> socket;
     Mutex mutex_;

     Runner() {
       listenHandler.reset(new ListenEventHandler(&mutex_));
     }

     virtual void run() {
       // When binding to explicit port, allow retrying to workaround bind failures on ports in use
       int retryCount = port ? 10 : 0;
       pServerSocketFactory = createServerSocketFactory();
       startServer(retryCount);
     }

     void readyBarrier() {
       // block until server is listening and ready to accept connections
       Guard g(mutex_);
       while (!listenHandler->ready_) {
         listenHandler->listenMonitor_.wait();
       }
     }
   private:
     void startServer(int retry_count) {
       try {
         socket.reset(new transport::TNonblockingSSLServerSocket(port, pServerSocketFactory));
         server.reset(new server::TNonblockingServer(processor, socket));
 	      server->setServerEventHandler(listenHandler);
         server->setNumIOThreads(1);
         if (userEventBase) {
           server->registerEvents(userEventBase.get());
         }
         server->serve();
       } catch (const transport::TTransportException&) {
         if (retry_count > 0) {
           ++port;
           startServer(retry_count - 1);
         } else {
           throw;
         }
       }
     }
   };

   struct EventDeleter {
     void operator()(event_base* p) { event_base_free(p); }
   };

 protected:
   Fixture() : processor(new test::ParentServiceProcessor(stdcxx::make_shared<Handler>())) {}

   ~Fixture() {
     if (server) {
       server->stop();
     }
     if (thread) {
       thread->join();
     }
   }

   void setEventBase(event_base* user_event_base) {
     userEventBase_.reset(user_event_base, EventDeleter());
   }

   int startServer(int port) {
     stdcxx::shared_ptr<Runner> runner(new Runner);
     runner->port = port;
     runner->processor = processor;
     runner->userEventBase = userEventBase_;

     apache::thrift::stdcxx::scoped_ptr<apache::thrift::concurrency::ThreadFactory> threadFactory(
         new apache::thrift::concurrency::PlatformThreadFactory(
 #if !USE_BOOST_THREAD && !USE_STD_THREAD
             concurrency::PlatformThreadFactory::OTHER, concurrency::PlatformThreadFactory::NORMAL,
             1,
 #endif
             false));
     thread = threadFactory->newThread(runner);
     thread->start();
     runner->readyBarrier();

     server = runner->server;
     return runner->port;
   }

   bool canCommunicate(int serverPort) {
     stdcxx::shared_ptr<TSSLSocketFactory> pClientSocketFactory = createClientSocketFactory();
     stdcxx::shared_ptr<TSSLSocket> socket = pClientSocketFactory->createSocket("localhost", serverPort);
     socket->open();
     test::ParentServiceClient client(stdcxx::make_shared<protocol::TBinaryProtocol>(
         stdcxx::make_shared<transport::TFramedTransport>(socket)));
     client.addString("foo");
     std::vector<std::string> strings;
     client.getStrings(strings);
     return strings.size() == 1 && !(strings[0].compare("foo"));
   }

 private:
   stdcxx::shared_ptr<event_base> userEventBase_;
   stdcxx::shared_ptr<test::ParentServiceProcessor> processor;
 protected:
   stdcxx::shared_ptr<server::TNonblockingServer> server;
 private:
   stdcxx::shared_ptr<apache::thrift::concurrency::Thread> thread;

 };

 BOOST_AUTO_TEST_SUITE(TNonblockingSSLServerTest)

 BOOST_FIXTURE_TEST_CASE(get_specified_port, Fixture) {
   int specified_port = startServer(12345);
   BOOST_REQUIRE_GE(specified_port, 12345);
   BOOST_REQUIRE_EQUAL(server->getListenPort(), specified_port);
   BOOST_CHECK(canCommunicate(specified_port));

   server->stop();
 }

 BOOST_FIXTURE_TEST_CASE(get_assigned_port, Fixture) {
   int specified_port = startServer(0);
   BOOST_REQUIRE_EQUAL(specified_port, 0);
   int assigned_port = server->getListenPort();
   BOOST_REQUIRE_NE(assigned_port, 0);
   BOOST_CHECK(canCommunicate(assigned_port));

   server->stop();
 }

 BOOST_FIXTURE_TEST_CASE(provide_event_base, Fixture) {
   event_base* eb = event_base_new();
   setEventBase(eb);
   startServer(0);

   // assert that the server works
   BOOST_CHECK(canCommunicate(server->getListenPort()));
 #if LIBEVENT_VERSION_NUMBER > 0x02010400
   // also assert that the event_base is actually used when it's easy
   BOOST_CHECK_GT(event_base_get_num_events(eb, EVENT_BASE_COUNT_ADDED), 0);
 #endif
 }

 BOOST_AUTO_TEST_SUITE_END()
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing,
	* software distributed under the License is distributed on an
	* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
	* KIND, either express or implied. See the License for the
	* specific language governing permissions and limitations
	* under the License.
	*/

	#define BOOST_TEST_MODULE TNonblockingSSLServerTest
	#include <boost/test/unit_test.hpp>
	#include <boost/smart_ptr.hpp>
	#include <boost/shared_ptr.hpp>
	#include <boost/filesystem.hpp>
	#include <boost/format.hpp>

	#include "thrift/server/TNonblockingServer.h"
	#include "thrift/transport/TSSLSocket.h"
	#include "thrift/transport/TNonblockingSSLServerSocket.h"

	#include "gen-cpp/ParentService.h"

	#include <event.h>

	using namespace apache::thrift;
	using apache::thrift::concurrency::Guard;
	using apache::thrift::concurrency::Monitor;
	using apache::thrift::concurrency::Mutex;
	using apache::thrift::server::TServerEventHandler;
	using apache::thrift::transport::TSSLSocketFactory;
	using apache::thrift::transport::TSSLSocket;

	struct Handler : public test::ParentServiceIf {
	void addString(const std::string& s) { strings_.push_back(s); }
	void getStrings(std::vector<std::string>& _return) { _return = strings_; }
	std::vector<std::string> strings_;

	// dummy overrides not used in this test
	int32_t incrementGeneration() { return 0; }
	int32_t getGeneration() { return 0; }
	void getDataWait(std::string&, const int32_t) {}
	void onewayWait() {}
	void exceptionWait(const std::string&) {}
	void unexpectedExceptionWait(const std::string&) {}
	};

	boost::filesystem::path keyDir;
	boost::filesystem::path certFile(const std::string& filename)
	{
	return keyDir / filename;
	}

	struct GlobalFixtureSSL
	{
	GlobalFixtureSSL()
	{
	using namespace boost::unit_test::framework;
	for (int i = 0; i < master_test_suite().argc; ++i)
	{
	BOOST_TEST_MESSAGE(boost::format("argv[%1%] = \"%2%\"") % i % master_test_suite().argv[i]);
	}

	#ifdef __linux__
	// OpenSSL calls send() without MSG_NOSIGPIPE so writing to a socket that has
	// disconnected can cause a SIGPIPE signal...
	signal(SIGPIPE, SIG_IGN);
	#endif

	TSSLSocketFactory::setManualOpenSSLInitialization(true);
	apache::thrift::transport::initializeOpenSSL();

	keyDir = boost::filesystem::current_path().parent_path().parent_path().parent_path() / "test" / "keys";
	if (!boost::filesystem::exists(certFile("server.crt")))
	{
	keyDir = boost::filesystem::path(master_test_suite().argv[master_test_suite().argc - 1]);
	if (!boost::filesystem::exists(certFile("server.crt")))
	{
	throw std::invalid_argument("The last argument to this test must be the directory containing the test certificate(s).");
	}
	}
	}

	virtual ~GlobalFixtureSSL()
	{
	apache::thrift::transport::cleanupOpenSSL();
	#ifdef __linux__
	signal(SIGPIPE, SIG_DFL);
	#endif
	}
	};

	#if (BOOST_VERSION >= 105900)
	BOOST_GLOBAL_FIXTURE(GlobalFixtureSSL);
	#else
	BOOST_GLOBAL_FIXTURE(GlobalFixtureSSL)
	#endif

	stdcxx::shared_ptr<TSSLSocketFactory> createServerSocketFactory() {
	stdcxx::shared_ptr<TSSLSocketFactory> pServerSocketFactory;

	pServerSocketFactory.reset(new TSSLSocketFactory());
	pServerSocketFactory->ciphers("ALL:!ADH:!LOW:!EXP:!MD5:@STRENGTH");
	pServerSocketFactory->loadCertificate(certFile("server.crt").string().c_str());
	pServerSocketFactory->loadPrivateKey(certFile("server.key").string().c_str());
	pServerSocketFactory->server(true);
	return pServerSocketFactory;
	}

	stdcxx::shared_ptr<TSSLSocketFactory> createClientSocketFactory() {
	stdcxx::shared_ptr<TSSLSocketFactory> pClientSocketFactory;

	pClientSocketFactory.reset(new TSSLSocketFactory());
	pClientSocketFactory->authenticate(true);
	pClientSocketFactory->loadCertificate(certFile("client.crt").string().c_str());
	pClientSocketFactory->loadPrivateKey(certFile("client.key").string().c_str());
	pClientSocketFactory->loadTrustedCertificates(certFile("CA.pem").string().c_str());
	return pClientSocketFactory;
	}

	class Fixture {
	private:
	struct ListenEventHandler : public TServerEventHandler {
	public:
	ListenEventHandler(Mutex* mutex) : listenMonitor_(mutex), ready_(false) {}

	void preServe() /* override */ {
	Guard g(listenMonitor_.mutex());
	ready_ = true;
	listenMonitor_.notify();
	}

	Monitor listenMonitor_;
	bool ready_;
	};

	struct Runner : public apache::thrift::concurrency::Runnable {
	int port;
	stdcxx::shared_ptr<event_base> userEventBase;
	stdcxx::shared_ptr<TProcessor> processor;
	stdcxx::shared_ptr<server::TNonblockingServer> server;
	stdcxx::shared_ptr<ListenEventHandler> listenHandler;
	stdcxx::shared_ptr<TSSLSocketFactory> pServerSocketFactory;
	stdcxx::shared_ptr<transport::TNonblockingSSLServerSocket> socket;
	Mutex mutex_;

	Runner() {
	listenHandler.reset(new ListenEventHandler(&mutex_));
	}

	virtual void run() {
	// When binding to explicit port, allow retrying to workaround bind failures on ports in use
	int retryCount = port ? 10 : 0;
	pServerSocketFactory = createServerSocketFactory();
	startServer(retryCount);
	}

	void readyBarrier() {
	// block until server is listening and ready to accept connections
	Guard g(mutex_);
	while (!listenHandler->ready_) {
	listenHandler->listenMonitor_.wait();
	}
	}
	private:
	void startServer(int retry_count) {
	try {
	socket.reset(new transport::TNonblockingSSLServerSocket(port, pServerSocketFactory));
	server.reset(new server::TNonblockingServer(processor, socket));
	server->setServerEventHandler(listenHandler);
	server->setNumIOThreads(1);
	if (userEventBase) {
	server->registerEvents(userEventBase.get());
	}
	server->serve();
	} catch (const transport::TTransportException&) {
	if (retry_count > 0) {
	++port;
	startServer(retry_count - 1);
	} else {
	throw;
	}
	}
	}
	};

	struct EventDeleter {
	void operator()(event_base* p) { event_base_free(p); }
	};

	protected:
	Fixture() : processor(new test::ParentServiceProcessor(stdcxx::make_shared<Handler>())) {}

	~Fixture() {
	if (server) {
	server->stop();
	}
	if (thread) {
	thread->join();
	}
	}

	void setEventBase(event_base* user_event_base) {
	userEventBase_.reset(user_event_base, EventDeleter());
	}

	int startServer(int port) {
	stdcxx::shared_ptr<Runner> runner(new Runner);
	runner->port = port;
	runner->processor = processor;
	runner->userEventBase = userEventBase_;

	apache::thrift::stdcxx::scoped_ptr<apache::thrift::concurrency::ThreadFactory> threadFactory(
	new apache::thrift::concurrency::PlatformThreadFactory(
	#if !USE_BOOST_THREAD && !USE_STD_THREAD
	concurrency::PlatformThreadFactory::OTHER, concurrency::PlatformThreadFactory::NORMAL,
	1,
	#endif
	false));
	thread = threadFactory->newThread(runner);
	thread->start();
	runner->readyBarrier();

	server = runner->server;
	return runner->port;
	}

	bool canCommunicate(int serverPort) {
	stdcxx::shared_ptr<TSSLSocketFactory> pClientSocketFactory = createClientSocketFactory();
	stdcxx::shared_ptr<TSSLSocket> socket = pClientSocketFactory->createSocket("localhost", serverPort);
	socket->open();
	test::ParentServiceClient client(stdcxx::make_shared<protocol::TBinaryProtocol>(
	stdcxx::make_shared<transport::TFramedTransport>(socket)));
	client.addString("foo");
	std::vector<std::string> strings;
	client.getStrings(strings);
	return strings.size() == 1 && !(strings[0].compare("foo"));
	}

	private:
	stdcxx::shared_ptr<event_base> userEventBase_;
	stdcxx::shared_ptr<test::ParentServiceProcessor> processor;
	protected:
	stdcxx::shared_ptr<server::TNonblockingServer> server;
	private:
	stdcxx::shared_ptr<apache::thrift::concurrency::Thread> thread;

	};

	BOOST_AUTO_TEST_SUITE(TNonblockingSSLServerTest)

	BOOST_FIXTURE_TEST_CASE(get_specified_port, Fixture) {
	int specified_port = startServer(12345);
	BOOST_REQUIRE_GE(specified_port, 12345);
	BOOST_REQUIRE_EQUAL(server->getListenPort(), specified_port);
	BOOST_CHECK(canCommunicate(specified_port));

	server->stop();
	}

	BOOST_FIXTURE_TEST_CASE(get_assigned_port, Fixture) {
	int specified_port = startServer(0);
	BOOST_REQUIRE_EQUAL(specified_port, 0);
	int assigned_port = server->getListenPort();
	BOOST_REQUIRE_NE(assigned_port, 0);
	BOOST_CHECK(canCommunicate(assigned_port));

	server->stop();
	}

	BOOST_FIXTURE_TEST_CASE(provide_event_base, Fixture) {
	event_base* eb = event_base_new();
	setEventBase(eb);
	startServer(0);

	// assert that the server works
	BOOST_CHECK(canCommunicate(server->getListenPort()));
	#if LIBEVENT_VERSION_NUMBER > 0x02010400
	// also assert that the event_base is actually used when it's easy
	BOOST_CHECK_GT(event_base_get_num_events(eb, EVENT_BASE_COUNT_ADDED), 0);
	#endif
	}

	BOOST_AUTO_TEST_SUITE_END()