From: Jelte Jansen Date: Thu, 22 Dec 2011 14:55:21 +0000 (+0100) Subject: [1522] ccsession socketrequestor implementation X-Git-Tag: trac2351_base~304^2~15^2~22 X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=294566c2b9a3dff9b9d3f929f09ad8f7a9d760c2;p=thirdparty%2Fkea.git [1522] ccsession socketrequestor implementation releaseSocket not implemented yet --- diff --git a/src/lib/config/ccsession.h b/src/lib/config/ccsession.h index 50bb65c83c..0f8e34a49b 100644 --- a/src/lib/config/ccsession.h +++ b/src/lib/config/ccsession.h @@ -315,7 +315,41 @@ public: isc::data::ConstElementPtr getRemoteConfigValue( const std::string& module_name, const std::string& identifier) const; - + + /** + * Send a message to the underlying CC session. + * This has the same interface as isc::cc::Session::group_sendmsg() + * + * \param msg see isc::cc::Session::group_sendmsg() + * \param group see isc::cc::Session::group_sendmsg() + * \param instance see isc::cc::Session::group_sendmsg() + * \param to see isc::cc::Session::group_sendmsg() + * \return see isc::cc::Session::group_sendmsg() + */ + int groupSendMsg(isc::data::ConstElementPtr msg, + std::string group, + std::string instance = "*", + std::string to = "*") { + return session_.group_sendmsg(msg, group, instance, to); + }; + + /** + * Receive a message from the underlying CC session. + * This has the same interface as isc::cc::Session::group_recvmsg() + * + * \param envelope see isc::cc::Session::group_recvmsg() + * \param msg see isc::cc::Session::group_recvmsg() + * \param nonblock see isc::cc::Session::group_recvmsg() + * \param seq see isc::cc::Session::group_recvmsg() + * \return see isc::cc::Session::group_recvmsg() + */ + bool groupRecvMsg(isc::data::ConstElementPtr& envelope, + isc::data::ConstElementPtr& msg, + bool nonblock = true, + int seq = -1) { + return session_.group_recvmsg(envelope, msg, nonblock, seq); + }; + private: ModuleSpec readModuleSpecification(const std::string& filename); void startCheck(); diff --git a/src/lib/server_common/socket_request.cc b/src/lib/server_common/socket_request.cc index 328816f328..e7cba64e02 100644 --- a/src/lib/server_common/socket_request.cc +++ b/src/lib/server_common/socket_request.cc @@ -14,11 +14,259 @@ #include "socket_request.h" +#include +#include +#include +#include + +#include +#include +#include + +// XXX TODO remove +#include + namespace isc { namespace server_common { namespace { SocketRequestor* requestor(NULL); + +// Before the boss process calls send_fd, it first sends this +// string to indicate success +static const std::string CREATOR_SOCKET_OK("1"); + +// Before the boss process calls send_fd, it first sends this +// string to indicate failure +static const std::string CREATOR_SOCKET_UNAVAILABLE("0"); + +// The name of the ccsession command to request a socket from boss +// (the actual format of command and response are hardcoded in their +// respective methods) +static const std::string REQUEST_SOCKET_COMMAND("get_socket"); + +// This implementation class for SocketRequestor uses +// a ModuleCCSession for communication with the boss process, +// and fd_share to read out the socket(s). +// Since we only use a reference to the session, it must never +// be closed during the lifetime of this class +class SocketRequestorCCSession : public SocketRequestor { +public: + SocketRequestorCCSession(config::ModuleCCSession& session) : + session_(session) + {} + + ~SocketRequestorCCSession() { + closeFdShareSockets(); + } + + virtual SocketID requestSocket(Protocol protocol, + const std::string& address, + uint16_t port, ShareMode share_mode, + const std::string& share_name) { + + isc::data::ConstElementPtr request_msg = + createRequestSocketMessage(protocol, address, port, + share_mode, share_name); + + // Send it to boss + int seq = session_.groupSendMsg(request_msg, "Boss"); + + // Get the answer from the boss. + // Just do a blocking read, we can't really do much anyway + isc::data::ConstElementPtr env, recv_msg; + if (!session_.groupRecvMsg(env, recv_msg, false, seq)) { + isc_throw(isc::config::CCSessionError, + "Incomplete response when requesting socket"); + } + + // Read the socket file from the answer + std::string token, path; + readRequestSocketAnswer(recv_msg, token, path); + // get the domain socket over which we will receive the + // real socket + int sock_pass_fd = getFdShareSocket(path); + + // and finally get the socket itself + int passed_sock_fd = getSocketFd(sock_pass_fd); + return (SocketID(passed_sock_fd, token)); + }; + + virtual void releaseSocket(const std::string& token) { + (void)token; + }; + +private: + // Creates the cc session message to request a socket. + // The actual command format is hardcoded, and should match + // the format as read in bind10_src.py.in + isc::data::ConstElementPtr + createRequestSocketMessage(Protocol protocol, + const std::string& address, + uint16_t port, ShareMode share_mode, + const std::string& share_name) + { + isc::data::ElementPtr request = isc::data::Element::createMap(); + request->set("address", isc::data::Element::create(address)); + request->set("port", isc::data::Element::create(port)); + switch (protocol) { + case SocketRequestor::UDP: + request->set("protocol", isc::data::Element::create("UDP")); + break; + case SocketRequestor::TCP: + request->set("protocol", isc::data::Element::create("TCP")); + break; + } + switch (share_mode) { + case DONT_SHARE: + request->set("share_mode", + isc::data::Element::create("NO")); + break; + case SHARE_SAME: + request->set("share_mode", + isc::data::Element::create("SAMEAPP")); + break; + case SHARE_ANY: + request->set("share_mode", + isc::data::Element::create("ANY")); + break; + } + request->set("share_name", isc::data::Element::create(share_name)); + + return (isc::config::createCommand(REQUEST_SOCKET_COMMAND, request)); + } + + // Checks and parses the response receive from Boss + // If successful, token and path will be set to the values found in the + // answer. + // If the response was an error response, or does not contain the + // expected elements, a CCSessionError is raised. + void readRequestSocketAnswer(isc::data::ConstElementPtr recv_msg, + std::string& token, + std::string& path) { + int rcode; + isc::data::ConstElementPtr answer = isc::config::parseAnswer(rcode, recv_msg); + if (rcode != 0) { + isc_throw(isc::config::CCSessionError, + "Error response when requesting socket: " << + answer->str()); + } + + if (!answer || + !answer->contains("token") || + !answer->contains("path")) { + isc_throw(isc::config::CCSessionError, + "Malformed answer when requesting socket"); + } + token = answer->get("token")->stringValue(); + path = answer->get("path")->stringValue(); + } + + // Returns the domain socket file descriptor + // If we had not opened it yet, opens it now + int + getFdShareSocket(const std::string& path) { + if (fd_share_sockets_.find(path) == fd_share_sockets_.end()) { + int new_fd = createFdShareSocket(path); + fd_share_sockets_[path] = new_fd; + return (new_fd); + } else { + return (fd_share_sockets_[path]); + } + } + + // Connect to the domain socket that has been received from Boss. + // (i.e. the one that is used to pass created sockets over). + // + // This should only be called if the socket had not been connected to + // already. To get the socket and reuse existing ones, use + // getFdShareSocket() + // + // \param path The domain socket to connect to + // \exception SocketError if the socket cannot be connected to + // \return the socket file descriptor + int + createFdShareSocket(const std::string& path) { + int sock_pass_fd = socket(AF_UNIX, SOCK_STREAM, 0); + if (sock_pass_fd == -1) { + isc_throw(SocketError, "Unable to open domain socket " << path << + ": " << strerror(errno)); + } + struct sockaddr_un sock_pass_addr; + sock_pass_addr.sun_family = AF_UNIX; + if (path.size() > sizeof(sock_pass_addr.sun_path)) { + isc_throw(SocketError, "Unable to open domain socket " << path << + ": path too long"); + } + + strcpy(sock_pass_addr.sun_path, path.c_str()); + size_t len = strlen(sock_pass_addr.sun_path) + + sizeof(sock_pass_addr.sun_family); + if (connect(sock_pass_fd, + (struct sockaddr *)&sock_pass_addr, + len) == -1) { + isc_throw(SocketError, "Unable to open domain socket " << path << + ": " << strerror(errno)); + } + return (sock_pass_fd); + } + + // Reads a socket fd over the given socket (using recv_fd()). + // + // \exception SocketError if the socket cannot be read + // \return the socket fd that has been read + int getSocketFd(int sock_pass_fd) { + // Boss first sends some data to signal that getting the socket + // from its cache succeeded + char status[2]; + memset(status, 0, 2); + if (isc::util::io::read_data(sock_pass_fd, &status, 1) < 1) { + isc_throw(SocketError, + "Error reading status code while requesting socket"); + } + // Actual status value hardcoded by boss atm. + if (CREATOR_SOCKET_UNAVAILABLE == status) { + isc_throw(SocketError, + "CREATOR_SOCKET_UNAVAILABLE returned"); + } else if (CREATOR_SOCKET_OK != status) { + isc_throw(SocketError, + "Unknown status code returned before recv_fd " << status); + } + + int passed_sock_fd = isc::util::io::recv_fd(sock_pass_fd); + + // check for error values of passed_sock_fd (see fd_share.h) + if (passed_sock_fd <= 0) { + switch (passed_sock_fd) { + case isc::util::io::FD_COMM_ERROR: + isc_throw(SocketError, + "FD_COMM_ERROR while requesting socket"); + break; + case isc::util::io::FD_OTHER_ERROR: + isc_throw(SocketError, + "FD_OTHER_ERROR while requesting socket"); + break; + default: + isc_throw(SocketError, + "Unknown error while requesting socket"); + } + } + return (passed_sock_fd); + } + + // Closes the sockets that has been used for fd_share + void + closeFdShareSockets() { + std::map::iterator it; + for (it = fd_share_sockets_.begin(); it != fd_share_sockets_.end(); ++it ) { + close((*it).second); + } + } + + config::ModuleCCSession& session_; + std::map fd_share_sockets_; +}; + } SocketRequestor& @@ -35,5 +283,25 @@ SocketRequestor::initTest(SocketRequestor* new_requestor) { requestor = new_requestor; } + +void +SocketRequestor::init(config::ModuleCCSession& session) { + if (requestor != NULL) { + isc_throw(InvalidOperation, "The socket requestor was already initialized"); + } else { + requestor = new SocketRequestorCCSession(session); + } +} + +void +SocketRequestor::cleanup() { + if (requestor != NULL) { + delete requestor; + requestor = NULL; + } else { + isc_throw(InvalidOperation, "The socket requestor is not initialized"); + } +} + } } diff --git a/src/lib/server_common/socket_request.h b/src/lib/server_common/socket_request.h index f5e71ca629..e208753957 100644 --- a/src/lib/server_common/socket_request.h +++ b/src/lib/server_common/socket_request.h @@ -151,8 +151,11 @@ public: /// It can be called only once per the life of application. /// /// \param session the CC session that'll be used to talk to the - /// socket creator. - /// \throw InvalidOperation when it is called more than once. + /// socket creator. + /// \param socket_path the path of the domain socket that is used to + /// the pass the actual sockets around. + /// \throw InvalidOperation when it is called more than once, + /// when socket_path is empty static void init(config::ModuleCCSession& session); /// \brief Initialization for tests @@ -171,6 +174,16 @@ public: /// an "virgin" state (which acts as if initTest or init was never /// called before). static void initTest(SocketRequestor* requestor); + + /// \brief Destroy the singleton instance + /// + /// Calling this function is not strictly necessary; the socket + /// requestor is a singleton anyway. However, for some tests it + /// is useful to destroy and recreate it, as well as for programs + /// that want to be completely clean on exit. + /// After this method has been called, all operations except init + /// will fail. + static void cleanup(); }; /// \brief Access the requestor object. diff --git a/src/lib/server_common/tests/socket_requestor_test.cc b/src/lib/server_common/tests/socket_requestor_test.cc index 6d49aaa178..ed805e0f6a 100644 --- a/src/lib/server_common/tests/socket_requestor_test.cc +++ b/src/lib/server_common/tests/socket_requestor_test.cc @@ -16,6 +16,24 @@ #include +#include +#include +#include + +#include + +#include +#include +#include +#include + +#include + +#include +#include + +using namespace isc::data; +using namespace isc::config; using namespace isc::server_common; using namespace isc; @@ -54,4 +72,331 @@ TEST(SocketRequestorAccess, initialized) { SocketRequestor::initTest(NULL); } +// This class contains a fake (module)ccsession to emulate answers from Boss +// and creates a local domain socket to emulate fd_sharing +class SocketRequestorTest : public ::testing::Test { +public: + SocketRequestorTest() : session(ElementPtr(new ListElement), + ElementPtr(new ListElement), + ElementPtr(new ListElement)), + specfile(std::string(TEST_DATA_PATH) + "/spec.spec") + { + session.getMessages()->add(createAnswer()); + cc_session.reset(new ModuleCCSession(specfile, session, NULL, NULL, + false, false)); + SocketRequestor::init(*cc_session); + }; + + ~SocketRequestorTest() { + SocketRequestor::cleanup(); + } + + // Do a standard request with some default values + SocketRequestor::SocketID + doRequest() { + return (socketRequestor().requestSocket(SocketRequestor::UDP, + "192.0.2.1", 12345, + SocketRequestor::DONT_SHARE, + "test")); + } + + // Creates a valid socket request answer, as it would be sent by + // Boss. 'valid' in terms of format, not values + void + addAnswer(const std::string& token, const std::string& path) { + ElementPtr answer_part = Element::createMap(); + answer_part->set("token", Element::create(token)); + answer_part->set("path", Element::create(path)); + session.getMessages()->add(createAnswer(0, answer_part)); + } + + // Clears the messages the client sent to far on the fake msgq + // (for easier access to new messages later) + void + clearMsgQueue() { + while(session.getMsgQueue()->size() > 0) { + session.getMsgQueue()->remove(0); + } + } + + isc::cc::FakeSession session; + std::auto_ptr cc_session; + std::string specfile; +}; + +// helper function to create the request packet as we expect the +// socket requestor to send +ConstElementPtr createExpectedRequest(const std::string& address, + int port, + const std::string& protocol, + const std::string& share_mode, + const std::string& share_name) { + // create command arguments + ElementPtr command_args = Element::createMap(); + command_args->set("address", Element::create(address)); + command_args->set("port", Element::create(port)); + command_args->set("protocol", Element::create(protocol)); + command_args->set("share_mode", Element::create(share_mode)); + command_args->set("share_name", Element::create(share_name)); + + // create the envelope + ElementPtr packet = Element::createList(); + packet->add(Element::create("Boss")); + packet->add(Element::create("*")); + packet->add(createCommand("get_socket", command_args)); + + return (packet); +} + +TEST_F(SocketRequestorTest, testSocketRequestMessages) { + // Should raise CCSessionError if there is no answer + // We are only testing the request messages that are sent, + // so for this test that is no problem + clearMsgQueue(); + ConstElementPtr expected_request; + + expected_request = createExpectedRequest("192.0.2.1", 12345, "UDP", + "NO", "test"); + ASSERT_THROW(socketRequestor().requestSocket(SocketRequestor::UDP, + "192.0.2.1", 12345, + SocketRequestor::DONT_SHARE, + "test"), + CCSessionError); + ASSERT_EQ(1, session.getMsgQueue()->size()); + ASSERT_EQ(*expected_request, *(session.getMsgQueue()->get(0))); + + clearMsgQueue(); + expected_request = createExpectedRequest("192.0.2.2", 1, "TCP", + "ANY", "test2"); + ASSERT_THROW(socketRequestor().requestSocket(SocketRequestor::TCP, + "192.0.2.2", 1, + SocketRequestor::SHARE_ANY, + "test2"), + CCSessionError); + ASSERT_EQ(1, session.getMsgQueue()->size()); + ASSERT_EQ(*expected_request, *(session.getMsgQueue()->get(0))); + + clearMsgQueue(); + expected_request = createExpectedRequest("::1", 2, "UDP", + "SAMEAPP", "test3"); + ASSERT_THROW(socketRequestor().requestSocket(SocketRequestor::UDP, + "::1", 2, + SocketRequestor::SHARE_SAME, + "test3"), + CCSessionError); + ASSERT_EQ(1, session.getMsgQueue()->size()); + ASSERT_EQ(*expected_request, *(session.getMsgQueue()->get(0))); +} + +TEST_F(SocketRequestorTest, testBadRequestAnswers) { + // Should raise CCSessionError if there is no answer + ASSERT_THROW(doRequest(), CCSessionError); + + // Also if the answer does not match the format + session.getMessages()->add(createAnswer()); + ASSERT_THROW(doRequest(), CCSessionError); + + // Now a 'real' answer, should fail on socket connect (no such file) + addAnswer("foo", "/does/not/exist"); + ASSERT_THROW(doRequest(), SocketRequestor::SocketError); + + // Another failure (domain socket path too long) + std::string long_str(1000, 'x'); + addAnswer("foo", long_str); + ASSERT_THROW(doRequest(), SocketRequestor::SocketError); + + // Send back an error response + session.getMessages()->add(createAnswer(1, "error")); + ASSERT_THROW(doRequest(), CCSessionError); +} + +// Helper test class that creates a random domain socket +// When run() is called, it creates the socket, forks, and the child will +// listen for a connection, then send all the data passed to run to that +// connection, and then close the socket +class TestSocket { +public: + TestSocket() : fd_(-1) { + path_ = strdup("test_socket.XXXXXX"); + // Misuse mkstemp to generate a file name. + int f = mkstemp(path_); + // Just need the name, so immediately close + close(f); + } + + ~TestSocket() { + cleanup(); + } + + void + cleanup() { + unlink(path_); + free(path_); + if (fd_ != -1) { + close(fd_); + } + } + + // Returns the path used for the socket + const char* getPath() const { + return (path_); + } + + // create socket, fork, and serve if child + void run(std::vector data) { + try { + create(); + int child_pid = fork(); + if (child_pid == 0) { + serve(data); + exit(0); + } else { + // parent does not need fd anymore + close(fd_); + fd_ = -1; + } + } catch (const std::exception&) { + cleanup(); + throw; + } + } +private: + // Actually create the socket and listen on it + void + create() { + fd_ = socket(AF_UNIX, SOCK_STREAM, 0); + if (fd_ == -1) { + isc_throw(Exception, "Unable to create socket"); + } + struct sockaddr_un socket_address; + socket_address.sun_family = AF_UNIX; + int len = strlen(path_); + if (len > sizeof(socket_address.sun_path)) { + isc_throw(Exception, + "mkstemp() created a filename too long for sun_path"); + } + strncpy(socket_address.sun_path, path_, len); + + len += sizeof(socket_address.sun_family); + // Remove the random file we created so we can reuse it for + // a domain socket connection. This contains a minor race condition + // but for the purposes of this test it should be small enough + unlink(path_); + if (bind(fd_, (struct sockaddr *)&socket_address, len) == -1) { + isc_throw(Exception, + "unable to bind to test domain socket " << path_ << + ": " << strerror(errno)); + } + + listen(fd_, 1); + } + + // Accept one connection, then send all values from the vector using + // send_fd() (prepended by a status code 'ok'). + // There are a few specific exceptions; + // when the value is -1, it will send back an error value (signaling + // CREATOR_SOCKET_UNAVAILABLE) + // when the value is -2, it will send a byte signaling CREATOR_SOCKET_OK + // first, and then one byte from some string (i.e. bad data, not using + // send_fd()) + // When it runs out of data, the socket is closed and the fork exists + // (it will exit too if there is any error on this side) + void + serve(std::vector data) { + struct sockaddr_un client_address; + socklen_t ca_len = sizeof(client_address); + int client_fd = accept(fd_, + (struct sockaddr*) &client_address, + &ca_len); + if (client_fd == -1) { + isc_throw(Exception, "Error in accept(): " << strerror(errno)); + } + BOOST_FOREACH(int cur_data, data) { + int result; + if (cur_data == -1) { + // send 'CREATOR_SOCKET_UNAVAILABLE' + result = isc::util::io::write_data(client_fd, "0", 1); + } else if (cur_data == -2) { + // send 'CREATOR_SOCKET_OK' first + result = isc::util::io::write_data(client_fd, "1", 1); + if (result == 1) { + result = send(client_fd, "a", 1, 0); + } + } else { + // send 'CREATOR_SOCKET_OK' first + std::cout << "[XX] SENDING ON " << path_ << std::endl; + result = isc::util::io::write_data(client_fd, "1", 1); + if (result == 1) { + result = isc::util::io::send_fd(client_fd, cur_data); + } + } + if (result < 0) { + isc_throw(Exception, "Error in send_fd(): " << strerror(errno)); + } + } + close(client_fd); + } + + int fd_; + char *path_; +}; + +TEST_F(SocketRequestorTest, testSocketPassing) { + TestSocket ts; + std::vector data; + data.push_back(1); + data.push_back(2); + data.push_back(3); + data.push_back(-1); + data.push_back(-2); + data.push_back(1); + ts.run(data); + + // 1 should be ok + addAnswer("foo", ts.getPath()); + SocketRequestor::SocketID socket_id = doRequest(); + ASSERT_EQ("foo", socket_id.second); + + // 2 should be ok too + addAnswer("bar", ts.getPath()); + socket_id = doRequest(); + ASSERT_EQ("bar", socket_id.second); + + // 3 should be ok too (reuse earlier token) + addAnswer("foo", ts.getPath()); + socket_id = doRequest(); + ASSERT_EQ("foo", socket_id.second); + + // -1 should not + addAnswer("foo", ts.getPath()); + ASSERT_THROW(doRequest(), SocketRequestor::SocketError); + + // -2 should not + addAnswer("foo", ts.getPath()); + ASSERT_THROW(doRequest(), SocketRequestor::SocketError); + + // Create a second socket server, to test that multiple different + // domains sockets would work as well (even though we don't actually + // use that feature) + TestSocket ts2; + std::vector data2; + data2.push_back(1); + ts2.run(data2); + // 1 should be ok + addAnswer("foo", ts2.getPath()); + socket_id = doRequest(); + ASSERT_EQ("foo", socket_id.second); + + // Now use first one again + addAnswer("foo", ts.getPath()); + socket_id = doRequest(); + ASSERT_EQ("foo", socket_id.second); + + // Vector is now empty, so the socket should be gone + addAnswer("foo", ts.getPath()); + ASSERT_THROW(doRequest(), SocketRequestor::SocketError); + +} + + }