/**
 * Constructor for the TlsStream class.
 *
 * @param role The role of the client.
 * @param sslContext The SSL context for the client.
 */
TlsStream::TlsStream(const Socket::Ptr& socket, TlsRole role, shared_ptr<SSL_CTX> sslContext)
	: m_Socket(socket), m_Role(role)
{
	m_SSL = shared_ptr<SSL>(SSL_new(sslContext.get()), SSL_free);

	if (!m_SSL) {
		BOOST_THROW_EXCEPTION(openssl_error()
			<< boost::errinfo_api_function("SSL_new")
			<< errinfo_openssl_error(ERR_get_error()));
	}

	if (!m_SSLIndexInitialized) {
		m_SSLIndex = SSL_get_ex_new_index(0, const_cast<char *>("TlsStream"), NULL, NULL, NULL);
		m_SSLIndexInitialized = true;
	}

	SSL_set_ex_data(m_SSL.get(), m_SSLIndex, this);

	SSL_set_verify(m_SSL.get(), SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);

	socket->MakeNonBlocking();

	m_BIO = BIO_new_socket(socket->GetFD(), 0);
	BIO_set_nbio(m_BIO, 1);
	SSL_set_bio(m_SSL.get(), m_BIO, m_BIO);

	if (m_Role == TlsRoleServer)
		SSL_set_accept_state(m_SSL.get());
	else
		SSL_set_connect_state(m_SSL.get());
}

/**
 * Constructor for the TlsStream class.
 *
 * @param role The role of the client.
 * @param sslContext The SSL context for the client.
 */
TlsStream::TlsStream(const Socket::Ptr& socket, ConnectionRole role, const shared_ptr<SSL_CTX>& sslContext)
	: m_Eof(false), m_Socket(socket), m_Role(role)
{
	std::ostringstream msgbuf;
	char errbuf[120];

	m_SSL = shared_ptr<SSL>(SSL_new(sslContext.get()), SSL_free);

	if (!m_SSL) {
		msgbuf << "SSL_new() failed with code " << ERR_peek_error() << ", \"" << ERR_error_string(ERR_peek_error(), errbuf) << "\"";
		Log(LogCritical, "TlsStream", msgbuf.str());

		BOOST_THROW_EXCEPTION(openssl_error()
			<< boost::errinfo_api_function("SSL_new")
			<< errinfo_openssl_error(ERR_peek_error()));
	}

	if (!m_SSLIndexInitialized) {
		m_SSLIndex = SSL_get_ex_new_index(0, const_cast<char *>("TlsStream"), NULL, NULL, NULL);
		m_SSLIndexInitialized = true;
	}

	SSL_set_ex_data(m_SSL.get(), m_SSLIndex, this);

	SSL_set_verify(m_SSL.get(), SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);

	socket->MakeNonBlocking();

	SSL_set_fd(m_SSL.get(), socket->GetFD());

	if (m_Role == RoleServer)
		SSL_set_accept_state(m_SSL.get());
	else
		SSL_set_connect_state(m_SSL.get());
}

void LogstashWriter::ReconnectTimerHandler(void)
{
 	if (m_Stream)
		return;

	Socket::Ptr socket;

	if (GetSocketType() == "tcp")
		socket = new TcpSocket();
	else
		socket = new UdpSocket();

	Log(LogNotice, "LogstashWriter")
	    << "Reconnecting to Logstash endpoint '" << GetHost() << "' port '" << GetPort() << "'.";

	try {
		socket->Connect(GetHost(), GetPort());
	} catch (const std::exception&) {
		Log(LogCritical, "LogstashWriter")
		    << "Can't connect to Logstash endpoint '" << GetHost() << "' port '" << GetPort() << "'.";
		return;
	}

	m_Stream = new NetworkStream(socket);
}

void ApiListener::ListenerThreadProc(const Socket::Ptr& server)
{
	Utility::SetThreadName("API Listener");

	server->Listen();

	for (;;) {
		try {
			Socket::Ptr client = server->Accept();
			boost::thread thread(boost::bind(&ApiListener::NewClientHandler, this, client, String(), RoleServer));
			thread.detach();
		} catch (const std::exception&) {
			Log(LogCritical, "ApiListener", "Cannot accept new connection.");
		}
	}
}

/**
 * Constructor for the SocketEvents class.
 */
SocketEvents::SocketEvents(const Socket::Ptr& socket, Object *lifesupportObject)
	: m_ID(m_NextID++), m_FD(socket->GetFD()), m_EnginePrivate(NULL)
{
	boost::call_once(l_SocketIOOnceFlag, &SocketEvents::InitializeEngine);

	Register(lifesupportObject);
}

/**
 * Constructor for the TlsStream class.
 *
 * @param role The role of the client.
 * @param sslContext The SSL context for the client.
 */
TlsStream::TlsStream(const Socket::Ptr& socket, const String& hostname, ConnectionRole role, const boost::shared_ptr<SSL_CTX>& sslContext)
	: SocketEvents(socket, this), m_Eof(false), m_HandshakeOK(false), m_VerifyOK(true), m_ErrorCode(0),
	  m_ErrorOccurred(false),  m_Socket(socket), m_Role(role), m_SendQ(new FIFO()), m_RecvQ(new FIFO()),
	  m_CurrentAction(TlsActionNone), m_Retry(false), m_Shutdown(false)
{
	std::ostringstream msgbuf;
	char errbuf[120];

	m_SSL = boost::shared_ptr<SSL>(SSL_new(sslContext.get()), SSL_free);

	if (!m_SSL) {
		msgbuf << "SSL_new() failed with code " << ERR_peek_error() << ", \"" << ERR_error_string(ERR_peek_error(), errbuf) << "\"";
		Log(LogCritical, "TlsStream", msgbuf.str());

		BOOST_THROW_EXCEPTION(openssl_error()
			<< boost::errinfo_api_function("SSL_new")
			<< errinfo_openssl_error(ERR_peek_error()));
	}

	if (!m_SSLIndexInitialized) {
		m_SSLIndex = SSL_get_ex_new_index(0, const_cast<char *>("TlsStream"), NULL, NULL, NULL);
		m_SSLIndexInitialized = true;
	}

	SSL_set_ex_data(m_SSL.get(), m_SSLIndex, this);

	SSL_set_verify(m_SSL.get(), SSL_VERIFY_PEER | SSL_VERIFY_CLIENT_ONCE, &TlsStream::ValidateCertificate);

	socket->MakeNonBlocking();

	SSL_set_fd(m_SSL.get(), socket->GetFD());

	if (m_Role == RoleServer)
		SSL_set_accept_state(m_SSL.get());
	else {
#ifdef SSL_CTRL_SET_TLSEXT_HOSTNAME
		if (!hostname.IsEmpty())
			SSL_set_tlsext_host_name(m_SSL.get(), hostname.CStr());
#endif /* SSL_CTRL_SET_TLSEXT_HOSTNAME */

		SSL_set_connect_state(m_SSL.get());
	}
}

void Epoll::add(Socket::Ptr p_socket, const EventTypes& p_ets) const
{
  int fd = p_socket->getDescriptor();
  int option = EPOLL_CTL_ADD;

  epoll_event e;
  ::memset(&e, 0, sizeof(e));
  e.data.fd = fd;
  e.events = nativeEvents(p_ets);

  if (::epoll_ctl(m_epollFd, option, fd, &e) < 0)
  {
    throw std::runtime_error("Add handler to epoll failed");
  }
}

void Epoll::modify(Socket::Ptr p_socket, const EventTypes& p_ets) const
{
  int option = EPOLL_CTL_MOD;
  epollCtl(option, p_socket->getDescriptor(), p_ets);
}

/**
 * Processes a new client connection.
 *
 * @param client The new client.
 */
void ApiListener::NewClientHandlerInternal(const Socket::Ptr& client, const String& hostname, ConnectionRole role)
{
	CONTEXT("Handling new API client connection");

	String conninfo;

	if (role == RoleClient)
		conninfo = "to";
	else
		conninfo = "from";

	conninfo += " " + client->GetPeerAddress();

	TlsStream::Ptr tlsStream;

	{
		ObjectLock olock(this);
		try {
			tlsStream = new TlsStream(client, hostname, role, m_SSLContext);
		} catch (const std::exception&) {
			Log(LogCritical, "ApiListener")
			    << "Cannot create TLS stream from client connection (" << conninfo << ")";
			return;
		}
	}

	try {
		tlsStream->Handshake();
	} catch (const std::exception& ex) {
		Log(LogCritical, "ApiListener")
		    << "Client TLS handshake failed (" << conninfo << ")";
		return;
	}

	boost::shared_ptr<X509> cert = tlsStream->GetPeerCertificate();
	String identity;
	Endpoint::Ptr endpoint;
	bool verify_ok = false;

	if (cert) {
		try {
			identity = GetCertificateCN(cert);
		} catch (const std::exception&) {
			Log(LogCritical, "ApiListener")
			    << "Cannot get certificate common name from cert path: '" << GetCertPath() << "'.";
			return;
		}

		verify_ok = tlsStream->IsVerifyOK();
		if (!hostname.IsEmpty()) {
			if (identity != hostname) {
				Log(LogWarning, "ApiListener")
					<< "Unexpected certificate common name while connecting to endpoint '"
				    << hostname << "': got '" << identity << "'";
				return;
			} else if (!verify_ok) {
				Log(LogWarning, "ApiListener")
					<< "Certificate validation failed for endpoint '" << hostname
					<< "': " << tlsStream->GetVerifyError();
				return;
			}
		}

		if (verify_ok)
			endpoint = Endpoint::GetByName(identity);

		{
			Log log(LogInformation, "ApiListener");

			log << "New client connection for identity '" << identity << "' " << conninfo;

			if (!verify_ok)
				log << " (certificate validation failed: " << tlsStream->GetVerifyError() << ")";
			else if (!endpoint)
				log << " (no Endpoint object found for identity)";
		}
	} else {
		Log(LogInformation, "ApiListener")
		    << "New client connection " << conninfo << " (no client certificate)";
	}

	ClientType ctype;

	if (role == RoleClient) {
		Dictionary::Ptr message = new Dictionary();
		message->Set("jsonrpc", "2.0");
		message->Set("method", "icinga::Hello");
		message->Set("params", new Dictionary());
		JsonRpc::SendMessage(tlsStream, message);
		ctype = ClientJsonRpc;
	} else {
		tlsStream->WaitForData(5);

		if (!tlsStream->IsDataAvailable()) {
			Log(LogWarning, "ApiListener")
//.........这里部分代码省略.........

void ExternalCommandListener::CommandPipeThread(const String& commandPath)
{
	Utility::SetThreadName("Command Pipe");

	struct stat statbuf;
	bool fifo_ok = false;

	if (lstat(commandPath.CStr(), &statbuf) >= 0) {
		if (S_ISFIFO(statbuf.st_mode) && access(commandPath.CStr(), R_OK) >= 0) {
			fifo_ok = true;
		} else {
			if (unlink(commandPath.CStr()) < 0) {
				BOOST_THROW_EXCEPTION(posix_error()
				    << boost::errinfo_api_function("unlink")
				    << boost::errinfo_errno(errno)
				    << boost::errinfo_file_name(commandPath));
			}
		}
	}

	mode_t mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP;

	if (!fifo_ok && mkfifo(commandPath.CStr(), S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP) < 0) {
		Log(LogCritical, "ExternalCommandListener")
		    << "mkfifo() for fifo path '" << commandPath << "' failed with error code " << errno << ", \"" << Utility::FormatErrorNumber(errno) << "\"";
		return;
	}

	/* mkfifo() uses umask to mask off some bits, which means we need to chmod() the
	 * fifo to get the right mask. */
	if (chmod(commandPath.CStr(), mode) < 0) {
		Log(LogCritical, "ExternalCommandListener")
		    << "chmod() on fifo '" << commandPath << "' failed with error code " << errno << ", \"" << Utility::FormatErrorNumber(errno) << "\"";
		return;
	}

	for (;;) {
		int fd = open(commandPath.CStr(), O_RDWR | O_NONBLOCK);

		if (fd < 0) {
			Log(LogCritical, "ExternalCommandListener")
			    << "open() for fifo path '" << commandPath << "' failed with error code " << errno << ", \"" << Utility::FormatErrorNumber(errno) << "\"";
			return;
		}

		FIFO::Ptr fifo = new FIFO();
		Socket::Ptr sock = new Socket(fd);
		StreamReadContext src;

		for (;;) {
			sock->Poll(true, false);

			char buffer[8192];
			size_t rc;

			try {
				rc = sock->Read(buffer, sizeof(buffer));
			} catch (const std::exception& ex) {
				/* We have read all data. */
				if (errno == EAGAIN)
					continue;

				Log(LogWarning, "ExternalCommandListener")
				    << "Cannot read from command pipe." << DiagnosticInformation(ex);
				break;
			}

			/* Empty pipe (EOF) */
			if (rc == 0)
				continue;

			fifo->Write(buffer, rc);

			for (;;) {
				String command;
				StreamReadStatus srs = fifo->ReadLine(&command, src);

				if (srs != StatusNewItem)
					break;

				try {
					Log(LogInformation, "ExternalCommandListener")
					    << "Executing external command: " << command;

					ExternalCommandProcessor::Execute(command);
				} catch (const std::exception& ex) {
					Log(LogWarning, "ExternalCommandListener")
					    << "External command failed: " << DiagnosticInformation(ex, false);
					Log(LogNotice, "ExternalCommandListener")
					    << "External command failed: " << DiagnosticInformation(ex, true);
				}
			}
		}
	}
}