diff --git a/libi2pd/SSU2.cpp b/libi2pd/SSU2.cpp
index b7376453..f336ee80 100644
--- a/libi2pd/SSU2.cpp
+++ b/libi2pd/SSU2.cpp
@@ -30,7 +30,8 @@ namespace transport
 	SSU2Session::SSU2Session (SSU2Server& server, std::shared_ptr<const i2p::data::RouterInfo> in_RemoteRouter,
 		std::shared_ptr<const i2p::data::RouterInfo::Address> addr, bool peerTest):
 		TransportSession (in_RemoteRouter, SSU2_CONNECT_TIMEOUT),
-		m_Server (server), m_Address (addr), m_DestConnID (0), m_SourceConnID (0)
+		m_Server (server), m_Address (addr), m_DestConnID (0), m_SourceConnID (0),
+		m_State (eSSU2SessionStateUnknown)
 	{
 		m_NoiseState.reset (new i2p::crypto::NoiseSymmetricState);
 		if (in_RemoteRouter && m_Address)
@@ -242,7 +243,9 @@ namespace transport
 		HandlePayload (decryptedPayload.data (), decryptedPayload.size ());
 		
 		m_Server.AddSession (m_SourceConnID, shared_from_this ());
+		m_State = eSSU2SessionStateEstablished;
 		SendSessionConfirmed (headerX + 16);
+		KDFDataPhase (m_KeyDataSend, m_KeyDataReceive);
 		
 		return true;
 	}	
@@ -377,10 +380,21 @@ namespace transport
 		i2p::data::netdb.PostI2NPMsg (CreateI2NPMessage (eI2NPDummyMsg, ri->GetBuffer (), ri->GetBufferLen ())); // TODO: should insert ri
 		// handle other blocks
 		HandlePayload (decryptedPayload.data () + riSize + 3, decryptedPayload.size () - riSize - 3);
-		
+		m_State = eSSU2SessionStateEstablished;
+		KDFDataPhase (m_KeyDataReceive, m_KeyDataSend);
 		return true;
 	}	
 
+	void SSU2Session::KDFDataPhase (uint8_t * keydata_ab, uint8_t * keydata_ba)
+	{
+		uint8_t keydata[64];
+		i2p::crypto::HKDF (m_NoiseState->m_CK, nullptr, 0, "", keydata); // keydata = HKDF(chainKey, ZEROLEN, "", 64)
+		// ab
+		i2p::crypto::HKDF (keydata, nullptr, 0, "HKDFSSU2DataKeys", keydata_ab); // keydata_ab = HKDF(keydata, ZEROLEN, "HKDFSSU2DataKeys", 64)
+		// ba
+		i2p::crypto::HKDF (keydata + 32, nullptr, 0, "HKDFSSU2DataKeys", keydata_ba); // keydata_ba = HKDF(keydata + 32, ZEROLEN, "HKDFSSU2DataKeys", 64)
+	}	
+		
 	void SSU2Session::SendTokenRequest ()
 	{
 		// we are Alice
@@ -512,6 +526,29 @@ namespace transport
 		SendSessionRequest (headerX[1]);
 		return true;
 	}	
+
+	void SSU2Session::ProcessData (uint8_t * buf, size_t len)
+	{
+		Header header;
+		memcpy (header.buf, buf, 16);
+		header.ll[1] ^= CreateHeaderMask (m_KeyDataReceive + 32, buf + (len - 12));
+		if (header.h.type != eSSU2Data) 
+		{
+			LogPrint (eLogWarning, "SSU2: Unexpected message type  ", (int)header.h.type);
+			return;
+		}	
+		uint8_t payload[SSU2_MTU];
+		size_t payloadSize = len - 32;
+		uint8_t nonce[12];
+		CreateNonce (be32toh (header.h.packetNum), nonce);
+		if (!i2p::crypto::AEADChaCha20Poly1305 (buf + 16, payloadSize, header.buf, 16, 
+			m_KeyDataReceive, nonce, payload, payloadSize, false))
+		{
+			LogPrint (eLogWarning, "SSU2: Data AEAD verification failed ");
+			return;
+		}	
+		HandlePayload (payload, payloadSize);
+	}	
 		
 	void SSU2Session::HandlePayload (const uint8_t * buf, size_t len)
 	{
@@ -799,8 +836,10 @@ namespace transport
 		auto it = m_Sessions.find (connID);
 		if (it != m_Sessions.end ())
 		{
-			// TODO: check state
-			it->second->ProcessSessionConfirmed (buf, len);
+			if (it->second->IsEstablished ())
+				it->second->ProcessData (buf, len);
+			else	
+				it->second->ProcessSessionConfirmed (buf, len);
 		}	
 		else 
 		{
diff --git a/libi2pd/SSU2.h b/libi2pd/SSU2.h
index ca6a50a2..92e8912d 100644
--- a/libi2pd/SSU2.h
+++ b/libi2pd/SSU2.h
@@ -34,6 +34,7 @@ namespace transport
 		eSSU2SessionRequest = 0,
 		eSSU2SessionCreated = 1,
 		eSSU2SessionConfirmed = 2,
+		eSSU2Data = 6,
 		eSSU2Retry = 9,
 		eSSU2TokenRequest = 10
 	};
@@ -64,6 +65,15 @@ namespace transport
 		eSSU2BlkPadding = 254
 	};
 
+	enum SSU2SessionState
+	{
+		eSSU2SessionStateUnknown,
+		eSSU2SessionStateEstablished,
+		eSSU2SessionStateClosed,
+		eSSU2SessionStateFailed
+	};
+
+	
 	// RouterInfo flags
 	const uint8_t SSU2_ROUTER_INFO_FLAG_REQUEST_FLOOD = 0x01;
 	const uint8_t SSU2_ROUTER_INFO_FLAG_GZIP = 0x02;	
@@ -95,11 +105,13 @@ namespace transport
 			void Connect ();
 			void Done () override {};
 			void SendI2NPMessages (const std::vector<std::shared_ptr<I2NPMessage> >& msgs) override {};
-
+			bool IsEstablished () const { return m_State == eSSU2SessionStateEstablished; };
+			
 			void ProcessFirstIncomingMessage (uint64_t connID, uint8_t * buf, size_t len);
 			bool ProcessSessionCreated (uint8_t * buf, size_t len);
 			bool ProcessSessionConfirmed (uint8_t * buf, size_t len);
 			bool ProcessRetry (uint8_t * buf, size_t len);
+			void ProcessData (uint8_t * buf, size_t len);
 			
 		private:
 
@@ -109,9 +121,10 @@ namespace transport
 			void SendSessionRequest (uint64_t token = 0);
 			void SendSessionCreated (const uint8_t * X);
 			void SendSessionConfirmed (const uint8_t * Y);
+			void KDFDataPhase (uint8_t * keydata_ab, uint8_t * keydata_ba);
 			void SendTokenRequest ();
 			void SendRetry ();
-
+			
 			void HandlePayload (const uint8_t * buf, size_t len);
 			bool ExtractEndpoint (const uint8_t * buf, size_t size, boost::asio::ip::udp::endpoint& ep);
 			size_t CreateAddressBlock (const boost::asio::ip::udp::endpoint& ep, uint8_t * buf, size_t len);
@@ -126,6 +139,8 @@ namespace transport
 			std::shared_ptr<const i2p::data::RouterInfo::Address> m_Address;
 			boost::asio::ip::udp::endpoint m_RemoteEndpoint;
 			uint64_t m_DestConnID, m_SourceConnID;
+			SSU2SessionState m_State;
+			uint8_t m_KeyDataSend[64], m_KeyDataReceive[64]; 
 	};
 
 	class SSU2Server:  private i2p::util::RunnableServiceWithWork