Browse Source

make it work

pull/1122/head
Jeff Becker 7 years ago
parent
commit
f2e6fad104
No known key found for this signature in database
GPG Key ID: F357B3B42F6F9B05
  1. 8
      libi2pd/CryptoWorker.h
  2. 13
      libi2pd/NTCPSession.cpp
  3. 11
      libi2pd/NTCPSession.h

8
libi2pd/CryptoWorker.h

@ -6,7 +6,6 @@
#include <deque> #include <deque>
#include <thread> #include <thread>
#include <vector> #include <vector>
#include <pair>
#include <memory> #include <memory>
namespace i2p namespace i2p
@ -17,7 +16,7 @@ namespace worker
struct ThreadPool struct ThreadPool
{ {
typedef std::function<void(void)> ResultFunc; typedef std::function<void(void)> ResultFunc;
typedef std::function<Result(void)> WorkFunc; typedef std::function<ResultFunc(void)> WorkFunc;
typedef std::pair<std::shared_ptr<Caller>, WorkFunc> Job; typedef std::pair<std::shared_ptr<Caller>, WorkFunc> Job;
typedef std::mutex mtx_t; typedef std::mutex mtx_t;
typedef std::unique_lock<mtx_t> lock_t; typedef std::unique_lock<mtx_t> lock_t;
@ -30,9 +29,9 @@ namespace worker
while(workers--) while(workers--)
{ {
threads.emplace_back([this] { threads.emplace_back([this] {
Job job;
for (;;) for (;;)
{ {
Job job;
{ {
lock_t lock(this->queue_mutex); lock_t lock(this->queue_mutex);
this->condition.wait( this->condition.wait(
@ -41,8 +40,9 @@ namespace worker
job = std::move(this->jobs.front()); job = std::move(this->jobs.front());
this->jobs.pop_front(); this->jobs.pop_front();
} }
ResultFunc result = job.second();
job.first->GetService().post(result);
} }
job.first->GetService().post(job.second());
}); });
} }
} }

13
libi2pd/NTCPSession.cpp

@ -105,6 +105,11 @@ namespace transport
transports.PeerConnected (shared_from_this ()); transports.PeerConnected (shared_from_this ());
} }
boost::asio::io_service & NTCPSession::GetService()
{
return m_Server.GetService();
}
void NTCPSession::ClientLogin () void NTCPSession::ClientLogin ()
{ {
if (!m_DHKeysPair) if (!m_DHKeysPair)
@ -196,11 +201,8 @@ namespace transport
m_Encryption.SetIV (y + 240); m_Encryption.SetIV (y + 240);
m_Decryption.SetIV (m_Establisher->phase1.HXxorHI + 16); m_Decryption.SetIV (m_Establisher->phase1.HXxorHI + 16);
m_Encryption.Encrypt ((uint8_t *)&m_Establisher->phase2.encrypted, sizeof(m_Establisher->phase2.encrypted), (uint8_t *)&m_Establisher->phase2.encrypted); m_Encryption.Encrypt ((uint8_t *)&m_Establisher->phase2.encrypted, sizeof(m_Establisher->phase2.encrypted), (uint8_t *)&m_Establisher->phase2.encrypted);
boost::asio::async_write (m_Socket, boost::asio::buffer (&m_Establisher->phase2, sizeof (NTCPPhase2)), boost::asio::transfer_all (), boost::asio::async_write(m_Socket, boost::asio::buffer (&m_Establisher->phase2, sizeof (NTCPPhase2)), boost::asio::transfer_all(), std::bind(&NTCPSession::HandlePhase2Sent, shared_from_this(), std::placeholders::_1, std::placeholders::_2, tsB));
std::bind(&NTCPSession::HandlePhase2Sent, shared_from_this (), std::placeholders::_1, std::placeholders::_2, tsB));
} }
void NTCPSession::HandlePhase2Sent (const boost::system::error_code& ecode, std::size_t bytes_transferred, uint32_t tsB) void NTCPSession::HandlePhase2Sent (const boost::system::error_code& ecode, std::size_t bytes_transferred, uint32_t tsB)
@ -249,7 +251,6 @@ namespace transport
{ {
m_Decryption.SetIV (m_Establisher->phase2.pubKey + 240); m_Decryption.SetIV (m_Establisher->phase2.pubKey + 240);
m_Encryption.SetIV (m_Establisher->phase1.HXxorHI + 16); m_Encryption.SetIV (m_Establisher->phase1.HXxorHI + 16);
m_Decryption.Decrypt((uint8_t *)&m_Establisher->phase2.encrypted, sizeof(m_Establisher->phase2.encrypted), (uint8_t *)&m_Establisher->phase2.encrypted); m_Decryption.Decrypt((uint8_t *)&m_Establisher->phase2.encrypted, sizeof(m_Establisher->phase2.encrypted), (uint8_t *)&m_Establisher->phase2.encrypted);
// verify // verify
uint8_t xy[512]; uint8_t xy[512];
@ -289,7 +290,6 @@ namespace transport
buf += paddingSize; buf += paddingSize;
len += paddingSize; len += paddingSize;
} }
SignedData s; SignedData s;
s.Insert (m_Establisher->phase1.pubKey, 256); // x s.Insert (m_Establisher->phase1.pubKey, 256); // x
s.Insert (m_Establisher->phase2.pubKey, 256); // y s.Insert (m_Establisher->phase2.pubKey, 256); // y
@ -297,7 +297,6 @@ namespace transport
s.Insert (tsA); // tsA s.Insert (tsA); // tsA
s.Insert (m_Establisher->phase2.encrypted.timestamp, 4); // tsB s.Insert (m_Establisher->phase2.encrypted.timestamp, 4); // tsB
s.Sign (keys, buf); s.Sign (keys, buf);
m_Encryption.Encrypt(m_ReceiveBuffer, len, m_ReceiveBuffer); m_Encryption.Encrypt(m_ReceiveBuffer, len, m_ReceiveBuffer);
boost::asio::async_write (m_Socket, boost::asio::buffer (m_ReceiveBuffer, len), boost::asio::transfer_all (), boost::asio::async_write (m_Socket, boost::asio::buffer (m_ReceiveBuffer, len), boost::asio::transfer_all (),
std::bind(&NTCPSession::HandlePhase3Sent, shared_from_this (), std::placeholders::_1, std::placeholders::_2, tsA)); std::bind(&NTCPSession::HandlePhase3Sent, shared_from_this (), std::placeholders::_1, std::placeholders::_2, tsA));

11
libi2pd/NTCPSession.h

@ -56,6 +56,7 @@ namespace transport
void Done (); void Done ();
boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; }; boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; };
boost::asio::io_service & GetService();
bool IsEstablished () const { return m_IsEstablished; }; bool IsEstablished () const { return m_IsEstablished; };
bool IsTerminated () const { return m_IsTerminated; }; bool IsTerminated () const { return m_IsTerminated; };
@ -172,6 +173,11 @@ namespace transport
void SetSessionLimits(uint16_t softLimit, uint16_t hardLimit) { m_SoftLimit = softLimit; m_HardLimit = hardLimit; } void SetSessionLimits(uint16_t softLimit, uint16_t hardLimit) { m_SoftLimit = softLimit; m_HardLimit = hardLimit; }
bool ShouldLimit() const { return ShouldHardLimit() || ShouldSoftLimit(); } bool ShouldLimit() const { return ShouldHardLimit() || ShouldSoftLimit(); }
void Work(std::shared_ptr<NTCPSession> conn, Pool::WorkFunc work)
{
m_CryptoPool->Offer({conn, work});
}
private: private:
/** @brief return true for hard limit */ /** @brief return true for hard limit */
@ -196,11 +202,6 @@ namespace transport
void ScheduleTermination (); void ScheduleTermination ();
void HandleTerminationTimer (const boost::system::error_code& ecode); void HandleTerminationTimer (const boost::system::error_code& ecode);
void Work(std::shared_ptr<NTCPSession> conn, Pool::WorkFunc work)
{
m_CryptoPool->Offer({conn, work});
}
private: private:
bool m_IsRunning; bool m_IsRunning;

Loading…
Cancel
Save