Merge pull request #143 from klondi/i2pservice

I2pservice

BOB.cpp

@@ -28,7 +28,7 @@ namespace client
 	void BOBI2PInboundTunnel::Stop ()
 	{
 		m_Acceptor.close();
-		ClearConnections ();
+		ClearHandlers ();
 	}
 
 	void BOBI2PInboundTunnel::Accept ()
@@ -136,7 +136,7 @@ namespace client
 	{
 		LogPrint ("New BOB inbound connection");
 		auto connection = std::make_shared<I2PTunnelConnection>(this, receiver->socket, leaseSet);
-		AddConnection (connection);
+		AddHandler (connection);
 		connection->I2PConnect (receiver->data, receiver->dataLen);
 		delete receiver;
 	}
@@ -154,7 +154,7 @@ namespace client
 
 	void BOBI2POutboundTunnel::Stop ()
 	{
-		ClearConnections ();
+		ClearHandlers ();
 	}	
 
 	void BOBI2POutboundTunnel::Accept ()
@@ -171,7 +171,7 @@ namespace client
 		if (stream)
 		{	
 			auto conn = std::make_shared<I2PTunnelConnection> (this, stream, new boost::asio::ip::tcp::socket (GetService ()), m_Endpoint, m_IsQuiet);
-			AddConnection (conn);
+			AddHandler (conn);
 			conn->Connect ();
 		}	
 	}

BOB.h

@@ -8,6 +8,7 @@
 #include <string>
 #include <boost/asio.hpp>
 #include "I2PTunnel.h"
+#include "I2PService.h"
 #include "Identity.h"
 #include "LeaseSet.h"
 
@@ -41,12 +42,12 @@ namespace client
 	const char BOB_REPLY_ERROR[] = "ERROR %s\n";
 	const char BOB_DATA[] = "NICKNAME %s\n";
 
-	class BOBI2PTunnel: public I2PTunnel
+	class BOBI2PTunnel: public I2PService
 	{
 		public:
 
 			BOBI2PTunnel (ClientDestination * localDestination): 
-				I2PTunnel (localDestination) {};
+				I2PService (localDestination) {};
 
 			virtual void Start () {};
 			virtual void Stop () {};				

Destination.cpp

@@ -387,18 +387,6 @@ namespace client
 		}
 	}	
 
-	void ClientDestination::CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port) {
-		assert(streamRequestComplete);
-		i2p::data::IdentHash identHash;
-		if (i2p::client::context.GetAddressBook ().GetIdentHash (dest, identHash))
-			CreateStream (streamRequestComplete, identHash, port);
-		else
-		{
-			LogPrint (eLogWarning, "Remote destination ", dest, " not found");
-			streamRequestComplete (nullptr);
-		}
-	}
-
 	void ClientDestination::CreateStream (StreamRequestComplete streamRequestComplete, const i2p::data::IdentHash& dest, int port) {
 		assert(streamRequestComplete);
 		const i2p::data::LeaseSet * leaseSet = FindLeaseSet (dest);

Destination.h

@@ -36,7 +36,9 @@ namespace client
 	const char I2CP_PARAM_OUTBOUND_TUNNEL_LENGTH[] = "outbound.length";
 	const int DEFAULT_OUTBOUND_TUNNEL_LENGTH = 3;
 	const int STREAM_REQUEST_TIMEOUT = 60; //in seconds
-	
+
+	typedef std::function<void (std::shared_ptr<i2p::stream::Stream> stream)> StreamRequestComplete;
+
 	class ClientDestination: public i2p::garlic::GarlicDestination
 	{
 		typedef std::function<void (bool success)> RequestComplete;
@@ -49,7 +51,6 @@ namespace client
 			RequestComplete requestComplete;
 		};	
 		
-		typedef std::function<void (std::shared_ptr<i2p::stream::Stream> stream)> StreamRequestComplete;
 		
 		public:
 
@@ -67,7 +68,6 @@ namespace client
 			
 			// streaming
 			i2p::stream::StreamingDestination * GetStreamingDestination () const { return m_StreamingDestination; };
-			void CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port = 0);
 			void CreateStream (StreamRequestComplete streamRequestComplete, const i2p::data::IdentHash& dest, int port = 0);
 			std::shared_ptr<i2p::stream::Stream> CreateStream (const i2p::data::LeaseSet& remote, int port = 0);
 			void AcceptStreams (const i2p::stream::StreamingDestination::Acceptor& acceptor);

HTTPProxy.cpp

@@ -2,16 +2,61 @@
 #include <cassert>
 #include <boost/lexical_cast.hpp>
 #include <boost/regex.hpp>
+#include <string>
+#include <atomic>
 #include "HTTPProxy.h"
 #include "Identity.h"
+#include "Streaming.h"
 #include "Destination.h"
 #include "ClientContext.h"
 #include "I2PEndian.h"
+#include "I2PTunnel.h"
 
 namespace i2p
 {
 namespace proxy
 {
+	static const size_t http_buffer_size = 8192;
+	class HTTPProxyHandler: public i2p::client::I2PServiceHandler, public std::enable_shared_from_this<HTTPProxyHandler> {
+		private:
+			enum state {
+				GET_METHOD,
+				GET_HOSTNAME,
+				GET_HTTPV,
+				GET_HTTPVNL, //TODO: fallback to finding HOst: header if needed
+				DONE
+			};
+
+			void EnterState(state nstate);
+			bool HandleData(uint8_t *http_buff, std::size_t len);
+			void HandleSockRecv(const boost::system::error_code & ecode, std::size_t bytes_transfered);
+			void Terminate();
+			void AsyncSockRead();
+			void HTTPRequestFailed(/*std::string message*/);
+			void ExtractRequest();
+			bool ValidateHTTPRequest();
+			bool CreateHTTPRequest(uint8_t *http_buff, std::size_t len);
+			void SentHTTPFailed(const boost::system::error_code & ecode);
+			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
+
+			uint8_t m_http_buff[http_buffer_size];
+			boost::asio::ip::tcp::socket * m_sock;
+			std::string m_request; //Data left to be sent
+			std::string m_url; //URL
+			std::string m_method; //Method
+			std::string m_version; //HTTP version
+			std::string m_address; //Address
+			std::string m_path; //Path
+			int m_port; //Port
+			state m_state;//Parsing state
+
+		public:
+			HTTPProxyHandler(HTTPProxyServer * parent, boost::asio::ip::tcp::socket * sock) : 
+				I2PServiceHandler(parent), m_sock(sock)
+				{ AsyncSockRead(); EnterState(GET_METHOD); }
+			~HTTPProxyHandler() { Terminate(); }
+	};
+
 	void HTTPProxyHandler::AsyncSockRead()
 	{
 		LogPrint(eLogDebug,"--- HTTP Proxy async sock read");
@@ -24,19 +69,15 @@ namespace proxy
 		}
 	}
 
-	void HTTPProxyHandler::Done() {
-		if (m_parent) m_parent->RemoveHandler (shared_from_this ());
-	}
-
 	void HTTPProxyHandler::Terminate() {
-		if (dead.exchange(true)) return;
+		if (Kill()) return;
 		if (m_sock) {
 			LogPrint(eLogDebug,"--- HTTP Proxy close sock");
 			m_sock->close();
 			delete m_sock;
 			m_sock = nullptr;
 		}
-		Done();
+		Done(shared_from_this());
 	}
 
 	/* All hope is lost beyond this point */
@@ -155,8 +196,7 @@ namespace proxy
 		if (HandleData(m_http_buff, len)) {
 			if (m_state == DONE) {
 				LogPrint(eLogInfo,"--- HTTP Proxy requested: ", m_url);
-				m_parent->GetLocalDestination ()->CreateStream (
+				GetOwner()->CreateStream (std::bind (&HTTPProxyHandler::HandleStreamRequestComplete,
-						std::bind (&HTTPProxyHandler::HandleStreamRequestComplete,
 						this, std::placeholders::_1), m_address, m_port);
 			} else {
 				AsyncSockRead();
@@ -178,12 +218,12 @@ namespace proxy
 	void HTTPProxyHandler::HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream)
 	{
 		if (stream) {
-			if (dead.exchange(true)) return;
+			if (Kill()) return;
 			LogPrint (eLogInfo,"--- HTTP Proxy New I2PTunnel connection");
-			auto connection = std::make_shared<i2p::client::I2PTunnelConnection>((i2p::client::I2PTunnel *)m_parent, m_sock, stream);
+			auto connection = std::make_shared<i2p::client::I2PTunnelConnection>(GetOwner(), m_sock, stream);
-			m_parent->AddConnection (connection);
+			GetOwner()->AddHandler (connection);
 			connection->I2PConnect (reinterpret_cast<const uint8_t*>(m_request.data()), m_request.size());
-			Done();
+			Done(shared_from_this());
 		} else {
 			LogPrint (eLogError,"--- HTTP Proxy Issue when creating the stream, check the previous warnings for more info.");
 			HTTPRequestFailed(); // TODO: Send correct error message host unreachable
@@ -200,7 +240,6 @@ namespace proxy
 	{
 		m_Acceptor.close();
 		m_Timer.cancel ();
-		ClearConnections ();
 		ClearHandlers();
 	}
 
@@ -211,23 +250,6 @@ namespace proxy
 			std::placeholders::_1, newSocket));
 	}
 
-	void  HTTPProxyServer::AddHandler (std::shared_ptr<HTTPProxyHandler> handler) {
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.insert (handler);
-	}
-
-	void  HTTPProxyServer::RemoveHandler (std::shared_ptr<HTTPProxyHandler> handler)
-	{
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.erase (handler);
-	}
-
-	void  HTTPProxyServer::ClearHandlers ()
-	{
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.clear ();
-	}
-
 	void HTTPProxyServer::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
 	{
 		if (!ecode)

HTTPProxy.h

@@ -2,67 +2,17 @@
 #define HTTP_PROXY_H__
 
 #include <memory>
-#include <string>
 #include <set>
 #include <boost/asio.hpp>
 #include <mutex>
-#include <atomic>
+#include "I2PService.h"
-#include "Identity.h"
-#include "Streaming.h"
-#include "I2PTunnel.h"
 
 namespace i2p
 {
 namespace proxy
 {
-
+	class HTTPProxyHandler;
-	const size_t http_buffer_size = 8192;
+	class HTTPProxyServer: public i2p::client::I2PService
-
-	class HTTPProxyServer;
-	class HTTPProxyHandler: public std::enable_shared_from_this<HTTPProxyHandler> {
-		private:
-			enum state {
-				GET_METHOD,
-				GET_HOSTNAME,
-				GET_HTTPV,
-				GET_HTTPVNL, //TODO: fallback to finding HOst: header if needed
-				DONE
-			};
-
-			void EnterState(state nstate);
-			bool HandleData(uint8_t *http_buff, std::size_t len);
-			void HandleSockRecv(const boost::system::error_code & ecode, std::size_t bytes_transfered);
-			void Done();
-			void Terminate();
-			void AsyncSockRead();
-			void HTTPRequestFailed(/*std::string message*/);
-			void ExtractRequest();
-			bool ValidateHTTPRequest();
-			bool CreateHTTPRequest(uint8_t *http_buff, std::size_t len);
-			void SentHTTPFailed(const boost::system::error_code & ecode);
-			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
-
-			uint8_t m_http_buff[http_buffer_size];
-			HTTPProxyServer * m_parent;
-			boost::asio::ip::tcp::socket * m_sock;
-			std::string m_request; //Data left to be sent
-			std::string m_url; //URL
-			std::string m_method; //Method
-			std::string m_version; //HTTP version
-			std::string m_address; //Address
-			std::string m_path; //Path
-			int m_port; //Port
-			std::atomic<bool> dead; //To avoid cleaning up multiple times
-			state m_state;//Parsing state
-
-		public:
-			HTTPProxyHandler(HTTPProxyServer * parent, boost::asio::ip::tcp::socket * sock) : 
-				m_parent(parent), m_sock(sock), dead(false)
-				{ AsyncSockRead(); EnterState(GET_METHOD); }
-			~HTTPProxyHandler() { Terminate(); }
-	};
-
-	class HTTPProxyServer: public i2p::client::I2PTunnel
 	{
 		private:
 			std::set<std::shared_ptr<HTTPProxyHandler> > m_Handlers;
@@ -76,16 +26,13 @@ namespace proxy
 			void HandleAccept(const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
 
 		public:
-			HTTPProxyServer(int port) : I2PTunnel(nullptr),
+			HTTPProxyServer(int port) : I2PService(i2p::data::SIGNING_KEY_TYPE_DSA_SHA1),
 				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
 				m_Timer (GetService ()) {};
 			~HTTPProxyServer() { Stop(); }
 
 			void Start ();
 			void Stop ();
-			void AddHandler (std::shared_ptr<HTTPProxyHandler> handler);
-			void RemoveHandler (std::shared_ptr<HTTPProxyHandler> handler);
-			void ClearHandlers ();
 	};
 
 	typedef HTTPProxyServer HTTPProxy;

I2PService.cpp

@@ -0,0 +1,36 @@
+#include "Destination.h"
+#include "Identity.h"
+#include "ClientContext.h"
+#include "I2PService.h"
+
+
+namespace i2p
+{
+namespace client
+{
+	static const i2p::data::SigningKeyType I2P_SERVICE_DEFAULT_KEY_TYPE = i2p::data::SIGNING_KEY_TYPE_ECDSA_SHA256_P256;
+
+	I2PService::I2PService (ClientDestination * localDestination):
+		m_LocalDestination (localDestination ? localDestination :
+					i2p::client::context.CreateNewLocalDestination (false, I2P_SERVICE_DEFAULT_KEY_TYPE))
+	{
+	}
+	
+	I2PService::I2PService (i2p::data::SigningKeyType kt):
+		m_LocalDestination (i2p::client::context.CreateNewLocalDestination (false, kt))
+	{
+	}
+	
+	void I2PService::CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port) {
+		assert(streamRequestComplete);
+		i2p::data::IdentHash identHash;
+		if (i2p::client::context.GetAddressBook ().GetIdentHash (dest, identHash))
+			m_LocalDestination->CreateStream (streamRequestComplete, identHash, port);
+		else
+		{
+			LogPrint (eLogWarning, "Remote destination ", dest, " not found");
+			streamRequestComplete (nullptr);
+		}
+	}
+}
+}

I2PService.h

@@ -0,0 +1,75 @@
+#ifndef I2PSERVICE_H__
+#define I2PSERVICE_H__
+
+#include <atomic>
+#include <mutex>
+#include <unordered_set>
+#include <memory>
+#include <boost/asio.hpp>
+#include "Destination.h"
+#include "Identity.h"
+
+namespace i2p
+{
+namespace client
+{
+	class I2PServiceHandler;
+	class I2PService
+	{
+		public:
+			I2PService (ClientDestination * localDestination  = nullptr);
+			I2PService (i2p::data::SigningKeyType kt);
+			virtual ~I2PService () { ClearHandlers (); }
+
+			inline void AddHandler (std::shared_ptr<I2PServiceHandler> conn) {
+				std::unique_lock<std::mutex> l(m_HandlersMutex);
+				m_Handlers.insert(conn);
+			}
+			inline void RemoveHandler (std::shared_ptr<I2PServiceHandler> conn) {
+				std::unique_lock<std::mutex> l(m_HandlersMutex);
+				m_Handlers.erase(conn);
+			}
+			inline void ClearHandlers () {
+				std::unique_lock<std::mutex> l(m_HandlersMutex);
+				m_Handlers.clear();
+			}
+
+			inline ClientDestination * GetLocalDestination () { return m_LocalDestination; };
+			inline void SetLocalDestination (ClientDestination * dest) { m_LocalDestination = dest; };
+			void CreateStream (StreamRequestComplete streamRequestComplete, const std::string& dest, int port = 0);
+
+			inline boost::asio::io_service& GetService () { return m_LocalDestination->GetService (); };
+
+			virtual void Start () = 0;
+			virtual void Stop () = 0;
+
+		private:
+
+			ClientDestination * m_LocalDestination;
+			std::unordered_set<std::shared_ptr<I2PServiceHandler> > m_Handlers;
+			std::mutex m_HandlersMutex;
+	};
+
+	/*Simple interface for I2PHandlers, allows detection of finalization amongst other things */
+	class I2PServiceHandler
+	{
+		public:
+			I2PServiceHandler(I2PService * parent) : m_Service(parent), m_Dead(false) { }
+			virtual ~I2PServiceHandler() { }
+		protected:
+			// Call when terminating or handing over to avoid race conditions
+			inline bool Kill() { return m_Dead.exchange(true); }
+			// Call to know if the handler is dead
+			inline bool Dead() { return m_Dead; }
+			// Call when done to clean up (make sure Kill is called first)
+			inline void Done(std::shared_ptr<I2PServiceHandler> me) { if(m_Service) m_Service->RemoveHandler(me); }
+			// Call to talk with the owner
+			inline I2PService * GetOwner() { return m_Service; }
+		private:
+			I2PService *m_Service;
+			std::atomic<bool> m_Dead; //To avoid cleaning up multiple times
+	};
+}
+}
+
+#endif

I2PTunnel.cpp

@@ -9,24 +9,25 @@ namespace i2p
 {
 namespace client
 {
-	I2PTunnelConnection::I2PTunnelConnection (I2PTunnel * owner, 
+	I2PTunnelConnection::I2PTunnelConnection (I2PService * owner, 
 	    boost::asio::ip::tcp::socket * socket, const i2p::data::LeaseSet * leaseSet): 
-		m_Socket (socket), m_Owner (owner), m_RemoteEndpoint (socket->remote_endpoint ()),
+		I2PServiceHandler(owner), m_Socket (socket), m_RemoteEndpoint (socket->remote_endpoint ()),
 		m_IsQuiet (true)
 	{
-		m_Stream = m_Owner->GetLocalDestination ()->CreateStream (*leaseSet);
+		m_Stream = GetOwner()->GetLocalDestination ()->CreateStream (*leaseSet);
 	}	
 
-	I2PTunnelConnection::I2PTunnelConnection (I2PTunnel * owner,
+	I2PTunnelConnection::I2PTunnelConnection (I2PService * owner,
 	    boost::asio::ip::tcp::socket * socket, std::shared_ptr<i2p::stream::Stream> stream):
-		m_Socket (socket), m_Stream (stream), m_Owner (owner),
+		I2PServiceHandler(owner), m_Socket (socket), m_Stream (stream),
 		m_RemoteEndpoint (socket->remote_endpoint ()), m_IsQuiet (true)
 	{
 	}
 
-	I2PTunnelConnection::I2PTunnelConnection (I2PTunnel * owner, std::shared_ptr<i2p::stream::Stream> stream,  
+	I2PTunnelConnection::I2PTunnelConnection (I2PService * owner, std::shared_ptr<i2p::stream::Stream> stream,
 	    boost::asio::ip::tcp::socket * socket, const boost::asio::ip::tcp::endpoint& target, bool quiet):
-		m_Socket (socket), m_Stream (stream), m_Owner (owner), m_RemoteEndpoint (target), m_IsQuiet (quiet)
+		I2PServiceHandler(owner), m_Socket (socket), m_Stream (stream),
+		m_RemoteEndpoint (target), m_IsQuiet (quiet)
 	{
 	}
 
@@ -56,15 +57,15 @@ namespace client
 	}	
 		
 	void I2PTunnelConnection::Terminate ()
-	{	
+	{
+		if (Kill()) return;
 		if (m_Stream)
 		{
 			m_Stream->Close ();
 			m_Stream.reset ();
 		}	
 		m_Socket->close ();
-		if (m_Owner)
+		Done(shared_from_this ());
-			m_Owner->RemoveConnection (shared_from_this ());
 	}			
 
 	void I2PTunnelConnection::Receive ()
@@ -150,28 +151,8 @@ namespace client
 		}
 	}
 
-	I2PTunnel::I2PTunnel (ClientDestination * localDestination) :
-		m_LocalDestination (localDestination ? localDestination :
-			i2p::client::context.CreateNewLocalDestination (false, I2P_TUNNEL_DEFAULT_KEY_TYPE))
-	{
-	}
-	void I2PTunnel::AddConnection (std::shared_ptr<I2PTunnelConnection> conn)
-	{
-		m_Connections.insert (conn);
-	}
-		
-	void I2PTunnel::RemoveConnection (std::shared_ptr<I2PTunnelConnection> conn)
-	{
-		m_Connections.erase (conn);
-	}	
-	
-	void I2PTunnel::ClearConnections ()
-	{
-		m_Connections.clear ();
-	}	
-		
 	I2PClientTunnel::I2PClientTunnel (const std::string& destination, int port, ClientDestination * localDestination): 
-		I2PTunnel (localDestination),
+		I2PService (localDestination),
 		m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
 		m_Timer (GetService ()), m_Destination (destination), m_DestinationIdentHash (nullptr)
 	{
@@ -193,7 +174,7 @@ namespace client
 	{
 		m_Acceptor.close();
 		m_Timer.cancel ();
-		ClearConnections ();
+		ClearHandlers ();
 		auto *originalIdentHash = m_DestinationIdentHash;
 		m_DestinationIdentHash = nullptr;
 		delete originalIdentHash;
@@ -250,7 +231,7 @@ namespace client
 		{
 			LogPrint (eLogInfo,"New I2PTunnel connection");
 			auto connection = std::make_shared<I2PTunnelConnection>(this, socket, stream);
-			AddConnection (connection);
+			AddHandler (connection);
 			connection->I2PConnect ();
 		}
 		else
@@ -261,7 +242,7 @@ namespace client
 	}
 
 	I2PServerTunnel::I2PServerTunnel (const std::string& address, int port, ClientDestination * localDestination): 
-		I2PTunnel (localDestination), m_Endpoint (boost::asio::ip::address::from_string (address), port)
+		I2PService (localDestination), m_Endpoint (boost::asio::ip::address::from_string (address), port)
 	{
 	}
 	
@@ -272,7 +253,7 @@ namespace client
 
 	void I2PServerTunnel::Stop ()
 	{
-		ClearConnections ();
+		ClearHandlers ();
 	}	
 
 	void I2PServerTunnel::Accept ()
@@ -289,7 +270,7 @@ namespace client
 		if (stream)
 		{	
 			auto conn = std::make_shared<I2PTunnelConnection> (this, stream, new boost::asio::ip::tcp::socket (GetService ()), m_Endpoint);
-			AddConnection (conn);
+			AddHandler (conn);
 			conn->Connect ();
 		}	
 	}

I2PTunnel.h

@@ -9,6 +9,7 @@
 #include "Identity.h"
 #include "Destination.h"
 #include "Streaming.h"
+#include "I2PService.h"
 
 namespace i2p
 {
@@ -17,21 +18,19 @@ namespace client
 	const size_t I2P_TUNNEL_CONNECTION_BUFFER_SIZE = 8192;
 	const int I2P_TUNNEL_CONNECTION_MAX_IDLE = 3600; // in seconds	
 	const int I2P_TUNNEL_DESTINATION_REQUEST_TIMEOUT = 10; // in seconds
-	const i2p::data::SigningKeyType I2P_TUNNEL_DEFAULT_KEY_TYPE = i2p::data::SIGNING_KEY_TYPE_ECDSA_SHA256_P256;
 
-	class I2PTunnel;
+
-	class I2PTunnelConnection: public std::enable_shared_from_this<I2PTunnelConnection>
+	class I2PTunnelConnection: public I2PServiceHandler, public std::enable_shared_from_this<I2PTunnelConnection>
 	{
 		public:
 
-			I2PTunnelConnection (I2PTunnel * owner, boost::asio::ip::tcp::socket * socket,
+			I2PTunnelConnection (I2PService * owner, boost::asio::ip::tcp::socket * socket,
 				const i2p::data::LeaseSet * leaseSet); // to I2P
-			I2PTunnelConnection (I2PTunnel * owner, boost::asio::ip::tcp::socket * socket,
+			I2PTunnelConnection (I2PService * owner, boost::asio::ip::tcp::socket * socket,
 				std::shared_ptr<i2p::stream::Stream> stream); // to I2P using simplified API :)
-			I2PTunnelConnection (I2PTunnel * owner, std::shared_ptr<i2p::stream::Stream> stream,  boost::asio::ip::tcp::socket * socket, 
+			I2PTunnelConnection (I2PService * owner, std::shared_ptr<i2p::stream::Stream> stream,  boost::asio::ip::tcp::socket * socket, 
 				const boost::asio::ip::tcp::endpoint& target, bool quiet = true); // from I2P
 			~I2PTunnelConnection ();
-
 			void I2PConnect (const uint8_t * msg = nullptr, size_t len = 0);
 			void Connect ();
 			
@@ -52,33 +51,11 @@ namespace client
 			uint8_t m_Buffer[I2P_TUNNEL_CONNECTION_BUFFER_SIZE], m_StreamBuffer[I2P_TUNNEL_CONNECTION_BUFFER_SIZE];
 			boost::asio::ip::tcp::socket * m_Socket;
 			std::shared_ptr<i2p::stream::Stream> m_Stream;
-			I2PTunnel * m_Owner;
 			boost::asio::ip::tcp::endpoint m_RemoteEndpoint;
 			bool m_IsQuiet; // don't send destination
-	};	
+	};
-
-	class I2PTunnel
-	{
-		public:
-
-			I2PTunnel (ClientDestination * localDestination  = nullptr);
-			virtual ~I2PTunnel () { ClearConnections (); }; 
-
-			void AddConnection (std::shared_ptr<I2PTunnelConnection> conn);
-			void RemoveConnection (std::shared_ptr<I2PTunnelConnection> conn);	
-			void ClearConnections ();
-			ClientDestination * GetLocalDestination () { return m_LocalDestination; };
-			void SetLocalDestination (ClientDestination * dest) { m_LocalDestination = dest; }; 			
-
-			boost::asio::io_service& GetService () { return m_LocalDestination->GetService (); };
-			
-		private:
 
-			ClientDestination * m_LocalDestination;
+	class I2PClientTunnel: public I2PService
-			std::set<std::shared_ptr<I2PTunnelConnection> > m_Connections;
-	};	
-	
-	class I2PClientTunnel: public I2PTunnel
 	{
 		public:
 
@@ -103,7 +80,7 @@ namespace client
 			const i2p::data::IdentHash * m_DestinationIdentHash;
 	};	
 
-	class I2PServerTunnel: public I2PTunnel
+	class I2PServerTunnel: public I2PService
 	{
 		public:
 
@@ -121,7 +98,7 @@ namespace client
 
 			boost::asio::ip::tcp::endpoint m_Endpoint;		
 	};
-}		
+}
 }	
 
 #endif

SOCKS.cpp

@@ -1,15 +1,137 @@
 #include <cstring>
 #include <cassert>
+#include <string>
+#include <atomic>
 #include "SOCKS.h"
 #include "Identity.h"
+#include "Streaming.h"
 #include "Destination.h"
 #include "ClientContext.h"
 #include "I2PEndian.h"
+#include "I2PTunnel.h"
 
 namespace i2p
 {
 namespace proxy
 {
+	static const size_t socks_buffer_size = 8192;
+	static const size_t max_socks_hostname_size = 255; // Limit for socks5 and bad idea to traverse
+
+	struct SOCKSDnsAddress {
+		uint8_t size;
+		char value[max_socks_hostname_size];
+		void FromString (std::string str) {
+			size = str.length();
+			if (str.length() > max_socks_hostname_size) size = max_socks_hostname_size;
+			memcpy(value,str.c_str(),size);
+		}
+		std::string ToString() { return std::string(value, size); }
+		void push_back (char c) { value[size++] = c; }
+	};
+
+	class SOCKSServer;
+	class SOCKSHandler: public i2p::client::I2PServiceHandler, public std::enable_shared_from_this<SOCKSHandler> {
+		private:
+			enum state {
+				GET_SOCKSV,
+				GET_COMMAND,
+				GET_PORT,
+				GET_IPV4,
+				GET4_IDENT,
+				GET4A_HOST,
+				GET5_AUTHNUM,
+				GET5_AUTH,
+				GET5_REQUESTV,
+				GET5_GETRSV,
+				GET5_GETADDRTYPE,
+				GET5_IPV6,
+				GET5_HOST_SIZE,
+				GET5_HOST,
+				DONE
+			};
+			enum authMethods {
+				AUTH_NONE = 0, //No authentication, skip to next step
+				AUTH_GSSAPI = 1, //GSSAPI authentication
+				AUTH_USERPASSWD = 2, //Username and password
+				AUTH_UNACCEPTABLE = 0xff //No acceptable method found
+			};
+			enum addrTypes {
+				ADDR_IPV4 = 1, //IPv4 address (4 octets)
+				ADDR_DNS = 3, // DNS name (up to 255 octets)
+				ADDR_IPV6 = 4 //IPV6 address (16 octets)
+			};
+			enum errTypes {
+				SOCKS5_OK = 0, // No error for SOCKS5
+				SOCKS5_GEN_FAIL = 1, // General server failure
+				SOCKS5_RULE_DENIED = 2, // Connection disallowed by ruleset
+				SOCKS5_NET_UNREACH = 3, // Network unreachable
+				SOCKS5_HOST_UNREACH = 4, // Host unreachable
+				SOCKS5_CONN_REFUSED = 5, // Connection refused by the peer
+				SOCKS5_TTL_EXPIRED = 6, // TTL Expired
+				SOCKS5_CMD_UNSUP = 7, // Command unsuported
+				SOCKS5_ADDR_UNSUP = 8, // Address type unsuported
+				SOCKS4_OK = 90, // No error for SOCKS4
+				SOCKS4_FAIL = 91, // Failed establishing connecting or not allowed
+				SOCKS4_IDENTD_MISSING = 92, // Couldn't connect to the identd server
+				SOCKS4_IDENTD_DIFFER = 93 // The ID reported by the application and by identd differ
+			};
+			enum cmdTypes {
+				CMD_CONNECT = 1, // TCP Connect
+				CMD_BIND = 2, // TCP Bind
+				CMD_UDP = 3 // UDP associate
+			};
+			enum socksVersions {
+				SOCKS4 = 4, // SOCKS4
+				SOCKS5 = 5 // SOCKS5
+			};
+			union address {
+				uint32_t ip;
+				SOCKSDnsAddress dns;
+				uint8_t ipv6[16];
+			};
+
+			void EnterState(state nstate, uint8_t parseleft = 1);
+			bool HandleData(uint8_t *sock_buff, std::size_t len);
+			bool ValidateSOCKSRequest();
+			void HandleSockRecv(const boost::system::error_code & ecode, std::size_t bytes_transfered);
+			void Terminate();
+			void AsyncSockRead();
+			boost::asio::const_buffers_1 GenerateSOCKS5SelectAuth(authMethods method);
+			boost::asio::const_buffers_1 GenerateSOCKS4Response(errTypes error, uint32_t ip, uint16_t port);
+			boost::asio::const_buffers_1 GenerateSOCKS5Response(errTypes error, addrTypes type, const address &addr, uint16_t port);
+			bool Socks5ChooseAuth();
+			void SocksRequestFailed(errTypes error);
+			void SocksRequestSuccess();
+			void SentSocksFailed(const boost::system::error_code & ecode);
+			void SentSocksDone(const boost::system::error_code & ecode);
+			void SentSocksResponse(const boost::system::error_code & ecode);
+			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
+
+			uint8_t m_sock_buff[socks_buffer_size];
+			boost::asio::ip::tcp::socket * m_sock;
+			std::shared_ptr<i2p::stream::Stream> m_stream;
+			uint8_t *m_remaining_data; //Data left to be sent
+			uint8_t m_response[7+max_socks_hostname_size];
+			address m_address; //Address
+			std::size_t m_remaining_data_len; //Size of the data left to be sent
+			uint32_t m_4aip; //Used in 4a requests
+			uint16_t m_port;
+			uint8_t m_command;
+			uint8_t m_parseleft; //Octets left to parse
+			authMethods m_authchosen; //Authentication chosen
+			addrTypes m_addrtype; //Address type chosen
+			socksVersions m_socksv; //Socks version
+			cmdTypes m_cmd; // Command requested
+			state m_state;
+
+		public:
+			SOCKSHandler(SOCKSServer * parent, boost::asio::ip::tcp::socket * sock) :
+				I2PServiceHandler(parent), m_sock(sock), m_stream(nullptr),
+				m_authchosen(AUTH_UNACCEPTABLE), m_addrtype(ADDR_IPV4)
+				{ m_address.ip = 0; EnterState(GET_SOCKSV); AsyncSockRead(); }
+			~SOCKSHandler() { Terminate(); }
+	};
+
 	void SOCKSHandler::AsyncSockRead()
 	{
 		LogPrint(eLogDebug,"--- SOCKS async sock read");
@@ -22,12 +144,8 @@ namespace proxy
 		}
 	}
 
-	void SOCKSHandler::Done() {
-		if (m_parent) m_parent->RemoveHandler (shared_from_this ());
-	}
-
 	void SOCKSHandler::Terminate() {
-		if (dead.exchange(true)) return;
+		if (Kill()) return;
 		if (m_sock) {
 			LogPrint(eLogDebug,"--- SOCKS close sock");
 			m_sock->close();
@@ -38,7 +156,7 @@ namespace proxy
 			LogPrint(eLogDebug,"--- SOCKS close stream");
 			m_stream.reset ();
 		}
-		Done();
+		Done(shared_from_this());
 	}
 
 	boost::asio::const_buffers_1 SOCKSHandler::GenerateSOCKS4Response(SOCKSHandler::errTypes error, uint32_t ip, uint16_t port)
@@ -54,7 +172,7 @@ namespace proxy
 	boost::asio::const_buffers_1 SOCKSHandler::GenerateSOCKS5Response(SOCKSHandler::errTypes error, SOCKSHandler::addrTypes type,
 								   const SOCKSHandler::address &addr, uint16_t port)
 	{
-		size_t size;
+		size_t size = 6;
 		assert(error <= SOCKS5_ADDR_UNSUP);
 		m_response[0] = '\x05'; //Version
 		m_response[1] = error; //Response code
@@ -125,7 +243,7 @@ namespace proxy
 				break;
 			case SOCKS5:
 				LogPrint(eLogInfo,"--- SOCKS5 connection success");
-				auto s = i2p::client::context.GetAddressBook().ToAddress(m_parent->GetLocalDestination()->GetIdentHash());
+				auto s = i2p::client::context.GetAddressBook().ToAddress(GetOwner()->GetLocalDestination()->GetIdentHash());
 				address ad; ad.dns.FromString(s);
 				//HACK only 16 bits passed in port as SOCKS5 doesn't allow for more
 				response = GenerateSOCKS5Response(SOCKS5_OK, ADDR_DNS, ad, m_stream->GetRecvStreamID());
@@ -336,8 +454,7 @@ namespace proxy
 		if (HandleData(m_sock_buff, len)) {
 			if (m_state == DONE) {
 				LogPrint(eLogInfo,"--- SOCKS requested ", m_address.dns.ToString(), ":" , m_port);
-				m_parent->GetLocalDestination ()->CreateStream (
+				GetOwner()->CreateStream ( std::bind (&SOCKSHandler::HandleStreamRequestComplete,
-						std::bind (&SOCKSHandler::HandleStreamRequestComplete,
 						this, std::placeholders::_1), m_address.dns.ToString(), m_port);
 			} else {
 				AsyncSockRead();
@@ -359,12 +476,12 @@ namespace proxy
 	void SOCKSHandler::SentSocksDone(const boost::system::error_code & ecode)
 	{
 		if (!ecode) {
-			if (dead.exchange(true)) return;
+		if (Kill()) return;
 			LogPrint (eLogInfo,"--- SOCKS New I2PTunnel connection");
-			auto connection = std::make_shared<i2p::client::I2PTunnelConnection>((i2p::client::I2PTunnel *)m_parent, m_sock, m_stream);
+			auto connection = std::make_shared<i2p::client::I2PTunnelConnection>(GetOwner(), m_sock, m_stream);
-			m_parent->AddConnection (connection);
+			GetOwner()->AddHandler (connection);
 			connection->I2PConnect (m_remaining_data,m_remaining_data_len);
-			Done();
+			Done(shared_from_this());
 		}
 		else
 		{
@@ -402,7 +519,6 @@ namespace proxy
 	{
 		m_Acceptor.close();
 		m_Timer.cancel ();
-		ClearConnections ();
 		ClearHandlers();
 	}
 
@@ -413,23 +529,6 @@ namespace proxy
 			std::placeholders::_1, newSocket));
 	}
 
-	void  SOCKSServer::AddHandler (std::shared_ptr<SOCKSHandler> handler) {
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.insert (handler);
-	}
-
-	void  SOCKSServer::RemoveHandler (std::shared_ptr<SOCKSHandler> handler)
-	{
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.erase (handler);
-	}
-
-	void  SOCKSServer::ClearHandlers ()
-	{
-		std::unique_lock<std::mutex> l(m_HandlersMutex);
-		m_Handlers.clear ();
-	}
-
 	void SOCKSServer::HandleAccept (const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket)
 	{
 		if (!ecode)

SOCKS.h

@@ -2,142 +2,17 @@
 #define SOCKS_H__
 
 #include <memory>
-#include <string>
 #include <set>
 #include <boost/asio.hpp>
 #include <mutex>
-#include <atomic>
+#include "I2PService.h"
-#include "Identity.h"
-#include "Streaming.h"
-#include "I2PTunnel.h"
 
 namespace i2p
 {
 namespace proxy
 {
-
+	class SOCKSHandler;
-	const size_t socks_buffer_size = 8192;
+	class SOCKSServer: public i2p::client::I2PService
-	const size_t max_socks_hostname_size = 255; // Limit for socks5 and bad idea to traverse
-
-	struct SOCKSDnsAddress {
-		uint8_t size;
-		char value[max_socks_hostname_size];
-		void FromString (std::string str) {
-			size = str.length();
-			if (str.length() > max_socks_hostname_size) size = max_socks_hostname_size;
-			memcpy(value,str.c_str(),size);
-		}
-		std::string ToString() { return std::string(value, size); }
-		void push_back (char c) { value[size++] = c; }
-	};
-
-	class SOCKSServer;
-	class SOCKSHandler: public std::enable_shared_from_this<SOCKSHandler> {
-		private:
-			enum state {
-				GET_SOCKSV,
-				GET_COMMAND,
-				GET_PORT,
-				GET_IPV4,
-				GET4_IDENT,
-				GET4A_HOST,
-				GET5_AUTHNUM,
-				GET5_AUTH,
-				GET5_REQUESTV,
-				GET5_GETRSV,
-				GET5_GETADDRTYPE,
-				GET5_IPV6,
-				GET5_HOST_SIZE,
-				GET5_HOST,
-				DONE
-			};
-			enum authMethods {
-				AUTH_NONE = 0, //No authentication, skip to next step
-				AUTH_GSSAPI = 1, //GSSAPI authentication
-				AUTH_USERPASSWD = 2, //Username and password
-				AUTH_UNACCEPTABLE = 0xff //No acceptable method found
-			};
-			enum addrTypes {
-				ADDR_IPV4 = 1, //IPv4 address (4 octets)
-				ADDR_DNS = 3, // DNS name (up to 255 octets)
-				ADDR_IPV6 = 4 //IPV6 address (16 octets)
-			};
-			enum errTypes {
-				SOCKS5_OK = 0, // No error for SOCKS5
-				SOCKS5_GEN_FAIL = 1, // General server failure
-				SOCKS5_RULE_DENIED = 2, // Connection disallowed by ruleset
-				SOCKS5_NET_UNREACH = 3, // Network unreachable
-				SOCKS5_HOST_UNREACH = 4, // Host unreachable
-				SOCKS5_CONN_REFUSED = 5, // Connection refused by the peer
-				SOCKS5_TTL_EXPIRED = 6, // TTL Expired
-				SOCKS5_CMD_UNSUP = 7, // Command unsuported
-				SOCKS5_ADDR_UNSUP = 8, // Address type unsuported
-				SOCKS4_OK = 90, // No error for SOCKS4
-				SOCKS4_FAIL = 91, // Failed establishing connecting or not allowed
-				SOCKS4_IDENTD_MISSING = 92, // Couldn't connect to the identd server
-				SOCKS4_IDENTD_DIFFER = 93 // The ID reported by the application and by identd differ
-			};
-			enum cmdTypes {
-				CMD_CONNECT = 1, // TCP Connect
-				CMD_BIND = 2, // TCP Bind
-				CMD_UDP = 3 // UDP associate
-			};
-			enum socksVersions {
-				SOCKS4 = 4, // SOCKS4
-				SOCKS5 = 5 // SOCKS5
-			};
-			union address {
-				uint32_t ip;
-				SOCKSDnsAddress dns;
-				uint8_t ipv6[16];
-			};
-
-			void EnterState(state nstate, uint8_t parseleft = 1);
-			bool HandleData(uint8_t *sock_buff, std::size_t len);
-			bool ValidateSOCKSRequest();
-			void HandleSockRecv(const boost::system::error_code & ecode, std::size_t bytes_transfered);
-			void Done();
-			void Terminate();
-			void AsyncSockRead();
-			boost::asio::const_buffers_1 GenerateSOCKS5SelectAuth(authMethods method);
-			boost::asio::const_buffers_1 GenerateSOCKS4Response(errTypes error, uint32_t ip, uint16_t port);
-			boost::asio::const_buffers_1 GenerateSOCKS5Response(errTypes error, addrTypes type, const address &addr, uint16_t port);
-			bool Socks5ChooseAuth();
-			void SocksRequestFailed(errTypes error);
-			void SocksRequestSuccess();
-			void SentSocksFailed(const boost::system::error_code & ecode);
-			void SentSocksDone(const boost::system::error_code & ecode);
-			void SentSocksResponse(const boost::system::error_code & ecode);
-			void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
-
-			uint8_t m_sock_buff[socks_buffer_size];
-			SOCKSServer * m_parent;
-			boost::asio::ip::tcp::socket * m_sock;
-			std::shared_ptr<i2p::stream::Stream> m_stream;
-			uint8_t *m_remaining_data; //Data left to be sent
-			uint8_t m_response[7+max_socks_hostname_size];
-			address m_address; //Address
-			std::size_t m_remaining_data_len; //Size of the data left to be sent
-			uint32_t m_4aip; //Used in 4a requests
-			uint16_t m_port;
-			uint8_t m_command;
-			uint8_t m_parseleft; //Octets left to parse
-			authMethods m_authchosen; //Authentication chosen
-			addrTypes m_addrtype; //Address type chosen
-			socksVersions m_socksv; //Socks version
-			cmdTypes m_cmd; // Command requested
-			state m_state;
-			std::atomic<bool> dead; //To avoid cleaning up multiple times
-
-		public:
-			SOCKSHandler(SOCKSServer * parent, boost::asio::ip::tcp::socket * sock) : 
-				m_parent(parent), m_sock(sock), m_stream(nullptr),
-				m_authchosen(AUTH_UNACCEPTABLE), m_addrtype(ADDR_IPV4), dead(false)
-				{ m_address.ip = 0; EnterState(GET_SOCKSV); AsyncSockRead(); }
-			~SOCKSHandler() { Terminate(); }
-	};
-
-	class SOCKSServer: public i2p::client::I2PTunnel
 	{
 		private:
 			std::set<std::shared_ptr<SOCKSHandler> > m_Handlers;
@@ -151,16 +26,13 @@ namespace proxy
 			void HandleAccept(const boost::system::error_code& ecode, boost::asio::ip::tcp::socket * socket);
 
 		public:
-			SOCKSServer(int port) : I2PTunnel(nullptr),
+			SOCKSServer(int port) : I2PService(nullptr),
 				m_Acceptor (GetService (), boost::asio::ip::tcp::endpoint (boost::asio::ip::tcp::v4(), port)),
 				m_Timer (GetService ()) {};
 			~SOCKSServer() { Stop(); }
 
 			void Start ();
 			void Stop ();
-			void AddHandler (std::shared_ptr<SOCKSHandler> handler);
-			void RemoveHandler (std::shared_ptr<SOCKSHandler> handler);
-			void ClearHandlers ();
 	};
 
 	typedef SOCKSServer SOCKSProxy;

build/CMakeLists.txt

@@ -48,6 +48,7 @@ set (DAEMON_SRC
   "${CMAKE_SOURCE_DIR}/Daemon.cpp"
   "${CMAKE_SOURCE_DIR}/HTTPProxy.cpp"
   "${CMAKE_SOURCE_DIR}/HTTPServer.cpp"
+  "${CMAKE_SOURCE_DIR}/I2PService.cpp"
   "${CMAKE_SOURCE_DIR}/I2PControl.cpp"	
   "${CMAKE_SOURCE_DIR}/I2PTunnel.cpp"
   "${CMAKE_SOURCE_DIR}/SAM.cpp"

build/autotools/Makefile.am

@@ -8,7 +8,7 @@ i2p_SOURCES	= AddressBook.cpp CryptoConst.cpp Daemon.cpp		\
 		  SSUData.cpp Streaming.cpp TransitTunnel.cpp		\
 		  Transports.cpp Tunnel.cpp TunnelEndpoint.cpp		\
 		  TunnelGateway.cpp TunnelPool.cpp UPnP.cpp aes.cpp	\
-		  base64.cpp i2p.cpp util.cpp 		  		\
+		  base64.cpp i2p.cpp util.cpp I2PService.cpp  		\
 		  							\
 		  AddressBook.h CryptoConst.h Daemon.h ElGamal.h	\
 		  Garlic.h HTTPProxy.h HTTPServer.h I2NPProtocol.h	\
@@ -19,7 +19,7 @@ i2p_SOURCES	= AddressBook.cpp CryptoConst.cpp Daemon.cpp		\
 		  TransitTunnel.h Transports.h Tunnel.h TunnelBase.h	\
 		  TunnelConfig.h TunnelEndpoint.h TunnelGateway.h	\
 		  TunnelPool.h UPnP.h aes.h base64.h config.h hmac.h	\
-		  util.h version.h
+		  util.h version.h I2PService.h
 
 AM_LDFLAGS	= @BOOST_DATE_TIME_LIB@ @BOOST_FILESYSTEM_LIB@		\
 		  @BOOST_PROGRAM_OPTIONS_LIB@ @BOOST_REGEX_LIB@		\

filelist.mk

@@ -10,14 +10,14 @@ ifeq ($(UNAME),Darwin)
 # This is needed on OS X for some reason I don't understand (yet).
 # Else will get linker error about unknown symbols. - torkel
 	COMMON_SRC += \
-	  BOB.cpp ClientContext.cpp Daemon.cpp I2PTunnel.cpp SAM.cpp SOCKS.cpp \
+	  BOB.cpp ClientContext.cpp Daemon.cpp I2PTunnel.cpp I2PService.cpp SAM.cpp SOCKS.cpp \
 	  UPnP.cpp HTTPServer.cpp HTTPProxy.cpp i2p.cpp DaemonLinux.cpp
 endif
 
 
 # also: Daemon{Linux,Win32}.cpp will be added later
 DAEMON_SRC = $(COMMON_SRC) \
-  BOB.cpp ClientContext.cpp Daemon.cpp I2PTunnel.cpp SAM.cpp SOCKS.cpp UPnP.cpp \
+  BOB.cpp ClientContext.cpp Daemon.cpp I2PTunnel.cpp I2PService.cpp SAM.cpp SOCKS.cpp UPnP.cpp \
   HTTPServer.cpp HTTPProxy.cpp I2PControl.cpp i2p.cpp
 
 LIB_SRC := $(COMMON_SRC) \