Bugfixes for the JSON-RPC sub-system.

diff --git a/lib/base/tlsstream.cpp b/lib/base/tlsstream.cpp
index 375f86ca7ec456dbd2976da73b6f5bf897f8b9ce..2b12155e324f787a65c25e16ee02a61bb1e6e548 100644 (file)
--- a/lib/base/tlsstream.cpp
+++ b/lib/base/tlsstream.cpp
@@ -48,42 +48,36 @@ TlsStream::TlsStream(const Stream::Ptr& innerStream, TlsRole role, shared_ptr<SS
 
 void TlsStream::Start(void)
 {
-       m_SSL = shared_ptr<SSL>(SSL_new(m_SSLContext.get()), SSL_free);
-
-       m_SSLContext.reset();
-
-       if (!m_SSL) {
-               BOOST_THROW_EXCEPTION(openssl_error()
-                   << boost::errinfo_api_function("SSL_new")
-                   << errinfo_openssl_error(ERR_get_error()));
-       }
+       {
+               boost::mutex::scoped_lock lock(m_SSLMutex);
 
-       if (!m_SSL)
-               BOOST_THROW_EXCEPTION(std::logic_error("No X509 client certificate was specified."));
+               m_SSL = shared_ptr<SSL>(SSL_new(m_SSLContext.get()), SSL_free);
 
-       if (!m_SSLIndexInitialized) {
-               m_SSLIndex = SSL_get_ex_new_index(0, const_cast<char *>("TlsStream"), NULL, NULL, NULL);
-               m_SSLIndexInitialized = true;
-       }
+               m_SSLContext.reset();
 
-       SSL_set_ex_data(m_SSL.get(), m_SSLIndex, this);
+               if (!m_SSL) {
+                       BOOST_THROW_EXCEPTION(openssl_error()
+                               << boost::errinfo_api_function("SSL_new")
+                               << errinfo_openssl_error(ERR_get_error()));
+               }
 
-       SSL_set_verify(m_SSL.get(), SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);
+               if (!m_SSLIndexInitialized) {
+                       m_SSLIndex = SSL_get_ex_new_index(0, const_cast<char *>("TlsStream"), NULL, NULL, NULL);
+                       m_SSLIndexInitialized = true;
+               }
 
-       m_BIO = BIO_new_I2Stream(m_InnerStream);
-       SSL_set_bio(m_SSL.get(), m_BIO, m_BIO);
+               SSL_set_ex_data(m_SSL.get(), m_SSLIndex, this);
 
-       if (m_Role == TlsRoleServer)
-               SSL_set_accept_state(m_SSL.get());
-       else
-               SSL_set_connect_state(m_SSL.get());
+               SSL_set_verify(m_SSL.get(), SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, NULL);
 
-       /*int rc = SSL_do_handshake(m_SSL.get());
+               m_BIO = BIO_new_I2Stream(m_InnerStream);
+               SSL_set_bio(m_SSL.get(), m_BIO, m_BIO);
 
-       if (rc == 1) {
-               SetConnected(true);
-               OnConnected(GetSelf());
-       }*/
+               if (m_Role == TlsRoleServer)
+                       SSL_set_accept_state(m_SSL.get());
+               else
+                       SSL_set_connect_state(m_SSL.get());
+       }
 
        Stream::Start();
 
@@ -97,7 +91,7 @@ void TlsStream::Start(void)
  */
 shared_ptr<X509> TlsStream::GetClientCertificate(void) const
 {
-       ObjectLock olock(this);
+       boost::mutex::scoped_lock lock(m_SSLMutex);
 
        return shared_ptr<X509>(SSL_get_certificate(m_SSL.get()), &Utility::NullDeleter);
 }
@@ -109,7 +103,7 @@ shared_ptr<X509> TlsStream::GetClientCertificate(void) const
  */
 shared_ptr<X509> TlsStream::GetPeerCertificate(void) const
 {
-       ObjectLock olock(this);
+       boost::mutex::scoped_lock lock(m_SSLMutex);
 
        return shared_ptr<X509>(SSL_get_peer_certificate(m_SSL.get()), X509_free);
 }
@@ -139,15 +133,18 @@ void TlsStream::ClosedHandler(void)
 void TlsStream::HandleIO(void)
 {
        ASSERT(!OwnsLock());
-       ObjectLock olock(this);
 
        char data[16 * 1024];
        int rc;
 
        if (!IsConnected()) {
+               boost::mutex::scoped_lock lock(m_SSLMutex);
+
                rc = SSL_do_handshake(m_SSL.get());
 
                if (rc == 1) {
+                       lock.unlock();
+
                        SetConnected(true);
                } else {
                        switch (SSL_get_error(m_SSL.get(), rc)) {
@@ -170,9 +167,14 @@ void TlsStream::HandleIO(void)
        bool new_data = false, read_ok = true;
 
        while (read_ok) {
+               boost::mutex::scoped_lock lock(m_SSLMutex);
+
                rc = SSL_read(m_SSL.get(), data, sizeof(data));
 
                if (rc > 0) {
+                       lock.unlock();
+
+                       ObjectLock olock(this);
                        m_RecvQueue->Write(data, rc);
                        new_data = true;
                } else {
@@ -194,11 +196,10 @@ void TlsStream::HandleIO(void)
                }
        }
 
-       if (new_data) {
-               olock.Unlock();
+       if (new_data)
                OnDataAvailable(GetSelf());
-               olock.Lock();
-       }
+
+       ObjectLock olock(this);
 
        while (m_SendQueue->GetAvailableBytes() > 0) {
                size_t count = m_SendQueue->GetAvailableBytes();
@@ -211,9 +212,16 @@ void TlsStream::HandleIO(void)
 
                m_SendQueue->Peek(data, count);
 
+               olock.Unlock();
+
+               boost::mutex::scoped_lock lock(m_SSLMutex);
+
                rc = SSL_write(m_SSL.get(), (const char *)data, count);
 
                if (rc > 0) {
+                       lock.unlock();
+
+                       olock.Lock();
                        m_SendQueue->Read(NULL, rc);
                } else {
                        switch (SSL_get_error(m_SSL.get(), rc)) {
@@ -239,13 +247,19 @@ void TlsStream::HandleIO(void)
  */
 void TlsStream::Close(void)
 {
-       ObjectLock olock(this);
+       {
+               boost::mutex::scoped_lock lock(m_SSLMutex);
+       
+               if (m_SSL)
+                       SSL_shutdown(m_SSL.get());
+       }
 
-       if (m_SSL)
-               SSL_shutdown(m_SSL.get());
+       {
+               ObjectLock olock(this);
 
-       m_SendQueue->Close();
-       m_RecvQueue->Close();
+               m_SendQueue->Close();
+               m_RecvQueue->Close();
+       }
 
        Stream::Close();
 }

diff --git a/lib/base/tlsstream.h b/lib/base/tlsstream.h
index 96e3753abf7f65a5e7f5ce111e78a2c7e4e5a3b6..0d4d714e81dc74034cffedad9c19d01e7c7ac23e 100644 (file)
--- a/lib/base/tlsstream.h
+++ b/lib/base/tlsstream.h
@@ -61,6 +61,7 @@ public:
 private:
        shared_ptr<SSL_CTX> m_SSLContext;
        shared_ptr<SSL> m_SSL;
+       mutable boost::mutex m_SSLMutex;
        BIO *m_BIO;
 
        FIFO::Ptr m_SendQueue;

diff --git a/lib/remoting/endpointmanager.cpp b/lib/remoting/endpointmanager.cpp
index 50871ca33899946b5ca001fd1aa490f70e45e438..945633fab5fe51154a7ef8f70b569b15ded04159 100644 (file)
--- a/lib/remoting/endpointmanager.cpp
+++ b/lib/remoting/endpointmanager.cpp
@@ -165,32 +165,30 @@ void EndpointManager::AddConnection(const String& node, const String& service) {
  */
 void EndpointManager::NewClientHandler(const Socket::Ptr& client, TlsRole role)
 {
-       ObjectLock olock(this);
-
-       String peerAddress = client->GetPeerAddress();
        TlsStream::Ptr tlsStream = boost::make_shared<TlsStream>(client, role, m_SSLContext);
-       tlsStream->Start();
 
        m_PendingClients.insert(tlsStream);
-       tlsStream->OnConnected.connect(boost::bind(&EndpointManager::ClientConnectedHandler, this, _1, peerAddress));
+       tlsStream->OnConnected.connect(boost::bind(&EndpointManager::ClientConnectedHandler, this, _1));
        tlsStream->OnClosed.connect(boost::bind(&EndpointManager::ClientClosedHandler, this, _1));
 
        client->Start();
+       tlsStream->Start();
 }
 
-void EndpointManager::ClientConnectedHandler(const Stream::Ptr& client, const String& peerAddress)
+void EndpointManager::ClientConnectedHandler(const Stream::Ptr& client)
 {
-       ObjectLock olock(this);
-
        TlsStream::Ptr tlsStream = static_pointer_cast<TlsStream>(client);
        JsonRpcConnection::Ptr jclient = boost::make_shared<JsonRpcConnection>(tlsStream);
 
-       m_PendingClients.erase(tlsStream);
+       {
+               ObjectLock olock(this);
+               m_PendingClients.erase(tlsStream);
+       }
 
        shared_ptr<X509> cert = tlsStream->GetPeerCertificate();
        String identity = GetCertificateCN(cert);
 
-       Log(LogInformation, "icinga", "New client connection at " + peerAddress + " for identity '" + identity + "'");
+       Log(LogInformation, "icinga", "New client connection for identity '" + identity + "'");
 
        Endpoint::Ptr endpoint = Endpoint::GetByName(identity);
 
@@ -202,10 +200,12 @@ void EndpointManager::ClientConnectedHandler(const Stream::Ptr& client, const St
 
 void EndpointManager::ClientClosedHandler(const Stream::Ptr& client)
 {
-       ObjectLock olock(this);
-
        TlsStream::Ptr tlsStream = static_pointer_cast<TlsStream>(client);
-       m_PendingClients.erase(tlsStream);
+
+       {
+               ObjectLock olock(this);
+               m_PendingClients.erase(tlsStream);
+       }
 }
 
 /**
@@ -370,8 +370,10 @@ void EndpointManager::SubscriptionTimerHandler(void)
 
        subscriptions->Seal();
 
-       if (m_Endpoint)
+       if (m_Endpoint) {
+               ObjectLock olock(m_Endpoint);
                m_Endpoint->SetSubscriptions(subscriptions);
+       }
 }
 
 void EndpointManager::ReconnectTimerHandler(void)

diff --git a/lib/remoting/endpointmanager.h b/lib/remoting/endpointmanager.h
index 5e3a4af7eb499d452bfd0949bbc49997d44e61c7..7acb1c8659495a9161c252c22a24a15e70a88547 100644 (file)
--- a/lib/remoting/endpointmanager.h
+++ b/lib/remoting/endpointmanager.h
@@ -111,8 +111,8 @@ private:
 
        void ReconnectTimerHandler(void);
 
-       void NewClientHandler(const Socket::Ptr& client, TlsRole rol);
-       void ClientConnectedHandler(const Stream::Ptr& client, const String& peerAddress);
+       void NewClientHandler(const Socket::Ptr& client, TlsRole role);
+       void ClientConnectedHandler(const Stream::Ptr& client);
        void ClientClosedHandler(const Stream::Ptr& client);
 };