Disallow connections to endpoints we're already connected to.

base/utility.cpp

@@ -86,3 +86,23 @@ string Utility::GetCertificateCN(const shared_ptr<X509>& certificate)
 
 	return buffer;
 }
+
+shared_ptr<X509> Utility::GetX509Certificate(string pemfile)
+{
+	X509 *cert;
+	BIO *fpcert = BIO_new(BIO_s_file());
+
+	if (fpcert == NULL)
+		throw OpenSSLException("BIO_new failed", ERR_get_error());
+
+	if (BIO_read_filename(fpcert, pemfile.c_str()) < 0)
+		throw OpenSSLException("BIO_read_filename failed", ERR_get_error());
+
+	cert = PEM_read_bio_X509_AUX(fpcert, NULL, NULL, NULL);
+	if (cert == NULL)
+		throw OpenSSLException("PEM_read_bio_X509_AUX failed", ERR_get_error());
+
+	BIO_free(fpcert);
+
+	return shared_ptr<X509>(cert, X509_free);
+}

base/utility.h

@@ -46,6 +46,7 @@ public:
 
 	static shared_ptr<SSL_CTX> MakeSSLContext(string pubkey, string privkey, string cakey);
 	static string GetCertificateCN(const shared_ptr<X509>& certificate);
+	static shared_ptr<X509> GetX509Certificate(string pemfile);
 };
 
 }

icinga-app/icinga.conf

@@ -4,7 +4,7 @@
 		"demo": { "replicate": "0" }
 	},
 	"rpcconnection": {
-		"kekslistener": { "replicate": "0", "hostname": "127.0.0.1", "port": "7777" }
+		"kekslistener": { "replicate": "0", "hostname": "::1", "port": "7777" }
 	},
 	"rpclistener": {
 		"kekslistener": { "replicate": "0", "port": "7777" }

icinga/discoverycomponent.cpp

@@ -11,7 +11,7 @@ void DiscoveryComponent::Start(void)
 {
 	m_DiscoveryEndpoint = make_shared<VirtualEndpoint>();
 	m_DiscoveryEndpoint->RegisterMethodHandler("message::Welcome",
-		bind_weak(&DiscoveryComponent::GetPeersMessageHandler, shared_from_this()));
+		bind_weak(&DiscoveryComponent::WelcomeMessageHandler, shared_from_this()));
 
 	m_DiscoveryEndpoint->RegisterMethodSource("discovery::PeerAvailable");
 	m_DiscoveryEndpoint->RegisterMethodHandler("discovery::GetPeers",
@@ -28,8 +28,34 @@ void DiscoveryComponent::Stop(void)
 		mgr->UnregisterEndpoint(m_DiscoveryEndpoint);
 }
 
+int DiscoveryComponent::CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea)
+{
+	if (endpoint == neea.Endpoint)
+		return 0;
+
+	if (endpoint->GetIdentity() == neea.Endpoint->GetIdentity()) {
+		Application::Log("Detected duplicate identity (" + endpoint->GetIdentity() + " - Disconnecting endpoint.");
+
+		endpoint->Stop();
+		GetEndpointManager()->UnregisterEndpoint(endpoint);
+	}
+
+	return 0;
+}
+
 int DiscoveryComponent::WelcomeMessageHandler(const NewRequestEventArgs& neea)
 {
+	if (neea.Sender->GetIdentity() == GetEndpointManager()->GetIdentity()) {
+		Application::Log("Detected loop-back connection - Disconnecting endpoint.");
+
+		neea.Sender->Stop();
+		GetEndpointManager()->UnregisterEndpoint(neea.Sender);
+
+		return 0;
+	}
+
+	GetEndpointManager()->ForeachEndpoint(bind(&DiscoveryComponent::CheckExistingEndpoint, this, neea.Sender, _1));
+
 	JsonRpcRequest request;
 	request.SetMethod("discovery::GetPeers");
 	GetEndpointManager()->SendUnicastRequest(m_DiscoveryEndpoint, neea.Sender, request);

icinga/discoverycomponent.h

@@ -14,6 +14,8 @@ private:
 	int WelcomeMessageHandler(const NewRequestEventArgs& neea);
 	int GetPeersMessageHandler(const NewRequestEventArgs& nrea);
 
+	int CheckExistingEndpoint(Endpoint::Ptr endpoint, const NewEndpointEventArgs& neea);
+
 public:
 	virtual string GetName(void) const;
 	virtual void Start(void);

icinga/endpoint.h

@@ -53,6 +53,8 @@ public:
 	virtual void ProcessRequest(Endpoint::Ptr sender, const JsonRpcRequest& message) = 0;
 	virtual void ProcessResponse(Endpoint::Ptr sender, const JsonRpcResponse& message) = 0;
 
+	virtual void Stop(void) = 0;
+
 	Event<NewMethodEventArgs> OnNewMethodSink;
 	Event<NewMethodEventArgs> OnNewMethodSource;
 

icinga/endpointmanager.cpp

@@ -2,11 +2,17 @@
 
 using namespace icinga;
 
-EndpointManager::EndpointManager(shared_ptr<SSL_CTX> sslContext)
+EndpointManager::EndpointManager(string identity, shared_ptr<SSL_CTX> sslContext)
 {
+	m_Identity = identity;
 	m_SSLContext = sslContext;
 }
 
+string EndpointManager::GetIdentity(void) const
+{
+	return m_Identity;
+}
+
 void EndpointManager::AddListener(unsigned short port)
 {
 	stringstream s;
@@ -160,8 +166,13 @@ void EndpointManager::ForeachEndpoint(function<int (const NewEndpointEventArgs&)
 {
 	NewEndpointEventArgs neea;
 	neea.Source = shared_from_this();
-	for (list<Endpoint::Ptr>::iterator i = m_Endpoints.begin(); i != m_Endpoints.end(); i++) {
-		neea.Endpoint = *i;
+
+	list<Endpoint::Ptr>::iterator prev, i;
+	for (i = m_Endpoints.begin(); i != m_Endpoints.end(); ) {
+		prev = i;
+		i++;
+
+		neea.Endpoint = *prev;
 		callback(neea);
 	}
 }

icinga/endpointmanager.h

@@ -11,7 +11,9 @@ struct I2_ICINGA_API NewEndpointEventArgs : public EventArgs
 
 class I2_ICINGA_API EndpointManager : public Object
 {
+	string m_Identity;
 	shared_ptr<SSL_CTX> m_SSLContext;
+
 	list<JsonRpcServer::Ptr> m_Servers;
 	list<Endpoint::Ptr> m_Endpoints;
 
@@ -27,7 +29,9 @@ public:
 	typedef shared_ptr<EndpointManager> Ptr;
 	typedef weak_ptr<EndpointManager> WeakPtr;
 
-	EndpointManager(shared_ptr<SSL_CTX> sslContext);
+	EndpointManager(string identity, shared_ptr<SSL_CTX> sslContext);
+
+	string GetIdentity(void) const;
 
 	void AddListener(unsigned short port);
 	void AddConnection(string host, unsigned short port);

icinga/icingaapplication.cpp

@@ -22,9 +22,14 @@ int IcingaApplication::Main(const vector<string>& args)
 		return EXIT_FAILURE;
 	}
 
+	shared_ptr<X509> cert = Utility::GetX509Certificate("icinga-c1.crt");
+	string identity = Utility::GetCertificateCN(cert);
+
+	Application::Log("My identity: " + identity);
+
 	shared_ptr<SSL_CTX> sslContext = Utility::MakeSSLContext("icinga-c1.crt", "icinga-c1.key", "ca.crt");
 
-	m_EndpointManager = make_shared<EndpointManager>(sslContext);
+	m_EndpointManager = make_shared<EndpointManager>(identity, sslContext);
 
 	string componentDirectory = GetExeDirectory() + "/../lib/icinga";
 	AddComponentSearchDir(componentDirectory);

icinga/jsonrpcendpoint.cpp

@@ -199,3 +199,11 @@ int JsonRpcEndpoint::VerifyCertificateHandler(const VerifyCertificateEventArgs&
 
 	return 0;
 }
+
+void JsonRpcEndpoint::Stop(void)
+{
+	if (m_Client) {
+		m_Client->Close();
+		m_Client = JsonRpcClient::Ptr();
+	}
+}

icinga/jsonrpcendpoint.h

@@ -50,6 +50,8 @@ public:
 
 	virtual void ProcessRequest(Endpoint::Ptr sender, const JsonRpcRequest& message);
 	virtual void ProcessResponse(Endpoint::Ptr sender, const JsonRpcResponse& message);
+
+	virtual void Stop(void);
 };
 
 }

icinga/virtualendpoint.cpp

@@ -83,3 +83,8 @@ bool VirtualEndpoint::IsAllowedMethodSource(string method) const
 {
 	return true;
 }
+
+void VirtualEndpoint::Stop(void)
+{
+	/* Nothing to do here. */
+}

icinga/virtualendpoint.h

@@ -38,6 +38,8 @@ public:
 
 	virtual void ProcessRequest(Endpoint::Ptr sender, const JsonRpcRequest& message);
 	virtual void ProcessResponse(Endpoint::Ptr sender, const JsonRpcResponse& message);
+
+	virtual void Stop(void);
 };
 
 }