PurpleI2P
/
i2pd
mirror of https://github.com/PurpleI2P/i2pd.git
1
0
Fork 0
Code Issues Releases Activity
Browse Source

Merge branch 'fix_http_proxy_500'

pull/557/head
Jeff Becker 9 years ago
parent
a1c81a63dd e795de5562
commit
543a372435
No known key found for this signature in database
GPG Key ID: AB950234D6EA286B
6 changed files with 403 additions and 265 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 1
      AddressBook.cpp
  2. 439
      HTTPProxy.cpp
  3. 25
      HTTPProxy.h
  4. 184
      I2CP.cpp
  5. 8
      I2CP.h
  6. 11
      Streaming.cpp

1
AddressBook.cpp
Unescape View File

@ -744,7 +744,6 @@ namespace client
LogPrint (eLogInfo, "Addressbook: received ", m_Link, " ETag: ", m_Etag, " Last-Modified: ", m_LastModified); LogPrint (eLogInfo, "Addressbook: received ", m_Link, " ETag: ", m_Etag, " Last-Modified: ", m_LastModified);
if (!response.eof () && !response.fail ()) if (!response.eof () && !response.fail ())
{ {
success = true;
if (!isChunked) if (!isChunked)
success = ProcessResponse (response, isGzip); success = ProcessResponse (response, isGzip);
else else

439
HTTPProxy.cpp
Unescape View File

@ -1,14 +1,8 @@
#include <cstring> #include <cstring>
#include <cassert> #include <cassert>
#include <boost/lexical_cast.hpp>
#include <string> #include <string>
#include <atomic> #include <atomic>
#include <memory>
#include <set>
#include <boost/asio.hpp>
#include <mutex>
#include "I2PService.h"
#include "Destination.h"
#include "HTTPProxy.h" #include "HTTPProxy.h"
#include "util.h" #include "util.h"
#include "Identity.h" #include "Identity.h"
@ -20,57 +14,71 @@
#include "Config.h" #include "Config.h"
#include "HTTP.h" #include "HTTP.h"
namespace i2p { namespace i2p
namespace proxy { {
bool str_rmatch(std::string & str, const char *suffix) { namespace proxy
auto pos = str.rfind (suffix); {
if (pos == std::string::npos) static const size_t http_buffer_size = 8192;
return false; /* not found */ class HTTPProxyHandler: public i2p::client::I2PServiceHandler, public std::enable_shared_from_this<HTTPProxyHandler>
if (str.length() == (pos + std::strlen(suffix)))
return true; /* match */
return false;
}
class HTTPReqHandler: public i2p::client::I2PServiceHandler, public std::enable_shared_from_this<HTTPReqHandler>
{ {
private: private:
enum state
bool HandleRequest(std::size_t len); {
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 HandleSockRecv(const boost::system::error_code & ecode, std::size_t bytes_transfered);
void Terminate(); void Terminate();
void AsyncSockRead(); void AsyncSockRead();
void HTTPRequestFailed(const char *message); void HTTPRequestFailed(/*std::string message*/);
void RedirectToJumpService(std::string & host); void RedirectToJumpService();
bool ExtractAddressHelper(i2p::http::URL & url, std::string & b64); void ExtractRequest();
void SanitizeHTTPRequest(i2p::http::HTTPReq & req); bool IsI2PAddress();
bool ValidateHTTPRequest();
void HandleJumpServices();
bool CreateHTTPRequest(uint8_t *http_buff, std::size_t len);
void SentHTTPFailed(const boost::system::error_code & ecode); void SentHTTPFailed(const boost::system::error_code & ecode);
void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream); void HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream);
uint8_t m_http_buff[http_buffer_size];
std::shared_ptr<boost::asio::ip::tcp::socket> m_sock; std::shared_ptr<boost::asio::ip::tcp::socket> m_sock;
std::vector<unsigned char> m_recv_buf; /* as "downstream recieve buffer", from client to me */ std::string m_request; //Data left to be sent
std::vector<unsigned char> m_send_buf; /* as "upstream send buffer", from me to remote host */ 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: public:
HTTPReqHandler(HTTPProxy * parent, std::shared_ptr<boost::asio::ip::tcp::socket> sock) : HTTPProxyHandler(HTTPProxyServer * parent, std::shared_ptr<boost::asio::ip::tcp::socket> sock) :
I2PServiceHandler(parent), m_sock(sock), m_recv_buf(8192), m_send_buf(0) {}; I2PServiceHandler(parent), m_sock(sock)
~HTTPReqHandler() { Terminate(); } { EnterState(GET_METHOD); }
~HTTPProxyHandler() { Terminate(); }
void Handle () { AsyncSockRead(); } void Handle () { AsyncSockRead(); }
}; };
void HTTPReqHandler::AsyncSockRead() void HTTPProxyHandler::AsyncSockRead()
{ {
LogPrint(eLogDebug, "HTTPProxy: async sock read"); LogPrint(eLogDebug, "HTTPProxy: async sock read");
if (!m_sock) { if(m_sock) {
m_sock->async_receive(boost::asio::buffer(m_http_buff, http_buffer_size),
std::bind(&HTTPProxyHandler::HandleSockRecv, shared_from_this(),
std::placeholders::_1, std::placeholders::_2));
} else {
LogPrint(eLogError, "HTTPProxy: no socket for read"); LogPrint(eLogError, "HTTPProxy: no socket for read");
return;
} }
m_sock->async_receive(boost::asio::buffer(m_recv_buf),
std::bind(&HTTPReqHandler::HandleSockRecv, shared_from_this(),
std::placeholders::_1, std::placeholders::_2));
} }
void HTTPReqHandler::Terminate() { void HTTPProxyHandler::Terminate() {
if (Kill()) return; if (Kill()) return;
if (m_sock) if (m_sock)
{ {
@ -81,175 +89,204 @@ namespace proxy {
Done(shared_from_this()); Done(shared_from_this());
} }
void HTTPReqHandler::HTTPRequestFailed(const char *message) /* All hope is lost beyond this point */
//TODO: handle this apropriately
void HTTPProxyHandler::HTTPRequestFailed(/*HTTPProxyHandler::errTypes error*/)
{ {
i2p::http::HTTPRes res; static std::string response = "HTTP/1.0 500 Internal Server Error\r\nContent-type: text/html\r\nContent-length: 0\r\n\r\n";
res.code = 500; boost::asio::async_write(*m_sock, boost::asio::buffer(response,response.size()),
res.add_header("Content-Type", "text/plain"); std::bind(&HTTPProxyHandler::SentHTTPFailed, shared_from_this(), std::placeholders::_1));
res.add_header("Connection", "close");
res.body = message;
res.body += "\r\n";
std::string response = res.to_string();
boost::asio::async_write(*m_sock, boost::asio::buffer(response, response.size()),
std::bind(&HTTPReqHandler::SentHTTPFailed, shared_from_this(), std::placeholders::_1));
} }
void HTTPReqHandler::RedirectToJumpService(std::string & host) void HTTPProxyHandler::RedirectToJumpService(/*HTTPProxyHandler::errTypes error*/)
{ {
i2p::http::HTTPRes res; std::stringstream response;
i2p::http::URL url; std::string httpAddr; i2p::config::GetOption("http.address", httpAddr);
uint16_t httpPort; i2p::config::GetOption("http.port", httpPort);
i2p::config::GetOption("http.address", url.host); response << "HTTP/1.1 302 Found\r\nLocation: http://" << httpAddr << ":" << httpPort << "/?page=jumpservices&address=" << m_address << "\r\n\r\n";
i2p::config::GetOption("http.port", url.port); boost::asio::async_write(*m_sock, boost::asio::buffer(response.str (),response.str ().length ()),
url.schema = "http"; std::bind(&HTTPProxyHandler::SentHTTPFailed, shared_from_this(), std::placeholders::_1));
url.path = "/"; }
url.query = "page=jumpservices&address=";
url.query += host;
res.code = 302; /* redirect */
res.add_header("Location", url.to_string().c_str());
res.add_header("Connection", "close");
std::string response = res.to_string(); void HTTPProxyHandler::EnterState(HTTPProxyHandler::state nstate)
boost::asio::async_write(*m_sock, boost::asio::buffer(response, response.length()), {
std::bind(&HTTPReqHandler::SentHTTPFailed, shared_from_this(), std::placeholders::_1)); m_state = nstate;
} }
bool HTTPReqHandler::ExtractAddressHelper(i2p::http::URL & url, std::string & b64) void HTTPProxyHandler::ExtractRequest()
{ {
const char *param = "i2paddresshelper="; LogPrint(eLogDebug, "HTTPProxy: request: ", m_method, " ", m_url);
std::size_t pos = url.query.find(param); i2p::http::URL url;
std::size_t len = std::strlen(param); url.parse (m_url);
std::map<std::string, std::string> params; m_address = url.host;
m_port = url.port;
m_path = url.path;
if (url.query.length () > 0) m_path += "?" + url.query;
if (!m_port) m_port = 80;
LogPrint(eLogDebug, "HTTPProxy: server: ", m_address, ", port: ", m_port, ", path: ", m_path);
}
if (pos == std::string::npos) bool HTTPProxyHandler::ValidateHTTPRequest()
return false; /* not found */ {
if (!url.parse_query(params)) if ( m_version != "HTTP/1.0" && m_version != "HTTP/1.1" )
{
LogPrint(eLogError, "HTTPProxy: unsupported version: ", m_version);
HTTPRequestFailed(); //TODO: send right stuff
return false; return false;
}
std::string value = params["i2paddresshelper"];
len += value.length();
b64 = i2p::http::UrlDecode(value);
url.query.replace(pos, len, "");
return true; return true;
} }
void HTTPReqHandler::SanitizeHTTPRequest(i2p::http::HTTPReq & req) void HTTPProxyHandler::HandleJumpServices()
{ {
req.del_header("Referer"); static const char * helpermark1 = "?i2paddresshelper=";
req.del_header("Via"); static const char * helpermark2 = "&i2paddresshelper=";
req.del_header("Forwarded"); size_t addressHelperPos1 = m_path.rfind (helpermark1);
std::vector<std::string> toErase; size_t addressHelperPos2 = m_path.rfind (helpermark2);
for (auto it : req.headers) { size_t addressHelperPos;
if (it.first.compare(0, 12, "X-Forwarded-")) { if (addressHelperPos1 == std::string::npos)
toErase.push_back(it.first); {
} else if (it.first.compare(0, 6, "Proxy-")) { if (addressHelperPos2 == std::string::npos)
toErase.push_back(it.first); return; //Not a jump service
} else { else
/* allow this header */ addressHelperPos = addressHelperPos2;
}
} }
for (auto header : toErase) { else
req.headers.erase(header); {
if (addressHelperPos2 == std::string::npos)
addressHelperPos = addressHelperPos1;
else if ( addressHelperPos1 > addressHelperPos2 )
addressHelperPos = addressHelperPos1;
else
addressHelperPos = addressHelperPos2;
} }
/* replace headers */ auto base64 = m_path.substr (addressHelperPos + strlen(helpermark1));
req.add_header("Connection", "close", true); /* keep-alive conns not supported yet */ base64 = i2p::util::http::urlDecode(base64); //Some of the symbols may be urlencoded
req.add_header("User-Agent", "MYOB/6.66 (AN/ON)", true); /* privacy */ LogPrint (eLogInfo, "HTTPProxy: jump service for ", m_address, ", inserting to address book");
//TODO: this is very dangerous and broken. We should ask the user before doing anything see http://pastethis.i2p/raw/pn5fL4YNJL7OSWj3Sc6N/
//TODO: we could redirect the user again to avoid dirtiness in the browser
i2p::client::context.GetAddressBook ().InsertAddress (m_address, base64);
m_path.erase(addressHelperPos);
} }
/** bool HTTPProxyHandler::IsI2PAddress()
* @param len length of data in m_recv_buf
* @return true on processed request or false if more data needed
*/
bool HTTPReqHandler::HandleRequest(std::size_t len)
{ {
i2p::http::HTTPReq req; auto pos = m_address.rfind (".i2p");
i2p::http::URL url; if (pos != std::string::npos && (pos+4) == m_address.length ())
std::string b64; {
return true;
int req_len = 0;
req_len = req.parse((const char *) m_recv_buf.data(), len);
if (req_len == 0)
return false; /* need more data */
if (req_len < 0) {
LogPrint(eLogError, "HTTPProxy: unable to parse request");
HTTPRequestFailed("invalid request");
return true; /* parse error */
} }
return false;
}
/* parsing success, now let's look inside request */ bool HTTPProxyHandler::CreateHTTPRequest(uint8_t *http_buff, std::size_t len)
LogPrint(eLogDebug, "HTTPProxy: requested: ", req.uri); {
url.parse(req.uri); ExtractRequest(); //TODO: parse earlier
if (!ValidateHTTPRequest()) return false;
if (ExtractAddressHelper(url, b64)) { HandleJumpServices();
i2p::client::context.GetAddressBook ().InsertAddress (url.host, b64);
std::string message = "added b64 from addresshelper for " + url.host + " to address book";
LogPrint (eLogInfo, "HTTPProxy: ", message);
message += ", please reload page";
HTTPRequestFailed(message.c_str());
return true; /* request processed */
}
i2p::data::IdentHash identHash; i2p::data::IdentHash identHash;
if (str_rmatch(url.host, ".i2p")) { if (IsI2PAddress ())
if (!i2p::client::context.GetAddressBook ().GetIdentHash (url.host, identHash)) { {
RedirectToJumpService(url.host); if (!i2p::client::context.GetAddressBook ().GetIdentHash (m_address, identHash)){
return true; /* request processed */ RedirectToJumpService();
return false;
} }
/* TODO: outproxy handler here */
} else {
std::string message = "Host " + url.host + " not inside i2p network, but outproxy support still missing";
HTTPRequestFailed(message.c_str());
LogPrint (eLogWarning, "HTTPProxy: ", message);
return true;
} }
SanitizeHTTPRequest(req);
m_request = m_method;
m_request.push_back(' ');
m_request += m_path;
m_request.push_back(' ');
m_request += m_version;
m_request.push_back('\r');
m_request.push_back('\n');
m_request.append("Connection: close\r\n");
// TODO: temporary shortcut. Must be implemented properly
uint8_t * eol = nullptr;
bool isEndOfHeader = false;
while (!isEndOfHeader && len && (eol = (uint8_t *)memchr (http_buff, '\r', len)))
{
if (eol)
{
*eol = 0; eol++;
if (strncmp ((const char *)http_buff, "Referer", 7) && strncmp ((const char *)http_buff, "Connection", 10)) // strip out referer and connection
{
if (!strncmp ((const char *)http_buff, "User-Agent", 10)) // replace UserAgent
m_request.append("User-Agent: MYOB/6.66 (AN/ON)");
else
m_request.append ((const char *)http_buff);
m_request.append ("\r\n");
}
isEndOfHeader = !http_buff[0];
auto l = eol - http_buff;
http_buff = eol;
len -= l;
if (len > 0) // \r
{
http_buff++;
len--;
}
}
}
m_request.append(reinterpret_cast<const char *>(http_buff),len);
return true;
}
std::string dest_host = url.host; bool HTTPProxyHandler::HandleData(uint8_t *http_buff, std::size_t len)
uint16_t dest_port = url.port; {
/* set proper 'Host' header in upstream request */ while (len > 0)
auto h = req.headers.find("Host"); {
if (dest_host != "") { //TODO: fallback to finding HOst: header if needed
/* absolute url, replace 'Host' header */ switch (m_state)
std::string h = dest_host; {
if (dest_port != 0 && dest_port != 80) case GET_METHOD:
h += ":" + std::to_string(dest_port); switch (*http_buff)
req.add_header("Host", h, true); {
} else if (h != req.headers.end()) { case ' ': EnterState(GET_HOSTNAME); break;
/* relative url and 'Host' header provided. transparent proxy mode? */ default: m_method.push_back(*http_buff); break;
i2p::http::URL u; }
std::string t = "http://" + h->second; break;
u.parse(t); case GET_HOSTNAME:
dest_host = u.host; switch (*http_buff)
dest_port = u.port; {
} else { case ' ': EnterState(GET_HTTPV); break;
/* relative url and missing 'Host' header */ default: m_url.push_back(*http_buff); break;
std::string message = "Can't detect destination host from request"; }
HTTPRequestFailed(message.c_str()); break;
return true; case GET_HTTPV:
switch (*http_buff)
{
case '\r': EnterState(GET_HTTPVNL); break;
default: m_version.push_back(*http_buff); break;
}
break;
case GET_HTTPVNL:
switch (*http_buff)
{
case '\n': EnterState(DONE); break;
default:
LogPrint(eLogError, "HTTPProxy: rejected invalid request ending with: ", ((int)*http_buff));
HTTPRequestFailed(); //TODO: add correct code
return false;
}
break;
default:
LogPrint(eLogError, "HTTPProxy: invalid state: ", m_state);
HTTPRequestFailed(); //TODO: add correct code 500
return false;
}
http_buff++;
len--;
if (m_state == DONE)
return CreateHTTPRequest(http_buff,len);
} }
/* make relative url */
url.schema = "";
url.host = "";
req.uri = url.to_string();
/* drop original request from input buffer */
m_recv_buf.erase(m_recv_buf.begin(), m_recv_buf.begin() + req_len);
/* build new buffer from modified request and data from original request */
std::string request = req.to_string();
m_send_buf.assign(request.begin(), request.end());
m_send_buf.insert(m_send_buf.end(), m_recv_buf.begin(), m_recv_buf.end());
/* connect to destination */
GetOwner()->CreateStream (std::bind (&HTTPReqHandler::HandleStreamRequestComplete,
shared_from_this(), std::placeholders::_1), dest_host, dest_port);
return true; return true;
} }
void HTTPReqHandler::HandleSockRecv(const boost::system::error_code & ecode, std::size_t len) void HTTPProxyHandler::HandleSockRecv(const boost::system::error_code & ecode, std::size_t len)
{ {
LogPrint(eLogDebug, "HTTPProxy: sock recv: ", len, " bytes"); LogPrint(eLogDebug, "HTTPProxy: sock recv: ", len, " bytes");
if(ecode) if(ecode)
@ -259,44 +296,54 @@ namespace proxy {
return; return;
} }
if (HandleRequest(len)) { if (HandleData(m_http_buff, len))
m_recv_buf.clear(); {
return; /* request processed */ if (m_state == DONE)
{
LogPrint(eLogDebug, "HTTPProxy: requested: ", m_url);
GetOwner()->CreateStream (std::bind (&HTTPProxyHandler::HandleStreamRequestComplete,
shared_from_this(), std::placeholders::_1), m_address, m_port);
}
else
AsyncSockRead();
} }
AsyncSockRead();
} }
void HTTPReqHandler::SentHTTPFailed(const boost::system::error_code & ecode) void HTTPProxyHandler::SentHTTPFailed(const boost::system::error_code & ecode)
{ {
if (ecode) if (ecode)
LogPrint (eLogError, "HTTPProxy: Closing socket after sending failure because: ", ecode.message ()); LogPrint (eLogError, "HTTPProxy: Closing socket after sending failure because: ", ecode.message ());
Terminate(); Terminate();
} }
void HTTPReqHandler::HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream) void HTTPProxyHandler::HandleStreamRequestComplete (std::shared_ptr<i2p::stream::Stream> stream)
{ {
if (!stream) { if (stream)
{
if (Kill()) return;
LogPrint (eLogInfo, "HTTPProxy: New I2PTunnel connection");
auto connection = std::make_shared<i2p::client::I2PTunnelConnection>(GetOwner(), m_sock, stream);
GetOwner()->AddHandler (connection);
connection->I2PConnect (reinterpret_cast<const uint8_t*>(m_request.data()), m_request.size());
Done(shared_from_this());
}
else
{
LogPrint (eLogError, "HTTPProxy: error when creating the stream, check the previous warnings for more info"); LogPrint (eLogError, "HTTPProxy: error when creating the stream, check the previous warnings for more info");
HTTPRequestFailed("error when creating the stream, check logs"); HTTPRequestFailed(); // TODO: Send correct error message host unreachable
return;
} }
if (Kill())
return;
LogPrint (eLogDebug, "HTTPProxy: New I2PTunnel connection");
auto connection = std::make_shared<i2p::client::I2PTunnelConnection>(GetOwner(), m_sock, stream);
GetOwner()->AddHandler (connection);
connection->I2PConnect (m_send_buf.data(), m_send_buf.size());
Done (shared_from_this());
} }
HTTPProxy::HTTPProxy(const std::string& address, int port, std::shared_ptr<i2p::client::ClientDestination> localDestination): HTTPProxyServer::HTTPProxyServer(const std::string& address, int port, std::shared_ptr<i2p::client::ClientDestination> localDestination):
TCPIPAcceptor(address, port, localDestination ? localDestination : i2p::client::context.GetSharedLocalDestination ()) TCPIPAcceptor(address, port, localDestination ? localDestination : i2p::client::context.GetSharedLocalDestination ())
{ {
} }
std::shared_ptr<i2p::client::I2PServiceHandler> HTTPProxy::CreateHandler(std::shared_ptr<boost::asio::ip::tcp::socket> socket) std::shared_ptr<i2p::client::I2PServiceHandler> HTTPProxyServer::CreateHandler(std::shared_ptr<boost::asio::ip::tcp::socket> socket)
{ {
return std::make_shared<HTTPReqHandler> (this, socket); return std::make_shared<HTTPProxyHandler> (this, socket);
} }
} // http
} // i2p }
}

25
HTTPProxy.h
Unescape View File

@ -1,21 +1,32 @@
#ifndef HTTP_PROXY_H__ #ifndef HTTP_PROXY_H__
#define HTTP_PROXY_H__ #define HTTP_PROXY_H__
namespace i2p { #include <memory>
namespace proxy { #include <set>
class HTTPProxy: public i2p::client::TCPIPAcceptor #include <boost/asio.hpp>
#include <mutex>
#include "I2PService.h"
#include "Destination.h"
namespace i2p
{
namespace proxy
{
class HTTPProxyServer: public i2p::client::TCPIPAcceptor
{ {
public: public:
HTTPProxy(const std::string& address, int port, std::shared_ptr<i2p::client::ClientDestination> localDestination = nullptr); HTTPProxyServer(const std::string& address, int port, std::shared_ptr<i2p::client::ClientDestination> localDestination = nullptr);
~HTTPProxy() {}; ~HTTPProxyServer() {};
protected: protected:
// Implements TCPIPAcceptor // Implements TCPIPAcceptor
std::shared_ptr<i2p::client::I2PServiceHandler> CreateHandler(std::shared_ptr<boost::asio::ip::tcp::socket> socket); std::shared_ptr<i2p::client::I2PServiceHandler> CreateHandler(std::shared_ptr<boost::asio::ip::tcp::socket> socket);
const char* GetName() { return "HTTP Proxy"; } const char* GetName() { return "HTTP Proxy"; }
}; };
} // http
} // i2p typedef HTTPProxyServer HTTPProxy;
}
}
#endif #endif

184
I2CP.cpp
Unescape View File

@ -23,7 +23,7 @@ namespace i2p
namespace client namespace client
{ {
I2CPDestination::I2CPDestination (I2CPSession& owner, std::shared_ptr<const i2p::data::IdentityEx> identity, bool isPublic, const std::map<std::string, std::string>& params): I2CPDestination::I2CPDestination (std::shared_ptr<I2CPSession> owner, std::shared_ptr<const i2p::data::IdentityEx> identity, bool isPublic, const std::map<std::string, std::string>& params):
LeaseSetDestination (isPublic, &params), m_Owner (owner), m_Identity (identity) LeaseSetDestination (isPublic, &params), m_Owner (owner), m_Identity (identity)
{ {
} }
@ -37,7 +37,7 @@ namespace client
{ {
uint32_t length = bufbe32toh (buf); uint32_t length = bufbe32toh (buf);
if (length > len - 4) length = len - 4; if (length > len - 4) length = len - 4;
m_Owner.SendMessagePayloadMessage (buf + 4, length); m_Owner->SendMessagePayloadMessage (buf + 4, length);
} }
void I2CPDestination::CreateNewLeaseSet (std::vector<std::shared_ptr<i2p::tunnel::InboundTunnel> > tunnels) void I2CPDestination::CreateNewLeaseSet (std::vector<std::shared_ptr<i2p::tunnel::InboundTunnel> > tunnels)
@ -46,9 +46,9 @@ namespace client
m_LeaseSetExpirationTime = ls.GetExpirationTime (); m_LeaseSetExpirationTime = ls.GetExpirationTime ();
uint8_t * leases = ls.GetLeases (); uint8_t * leases = ls.GetLeases ();
leases[-1] = tunnels.size (); leases[-1] = tunnels.size ();
htobe16buf (leases - 3, m_Owner.GetSessionID ()); htobe16buf (leases - 3, m_Owner->GetSessionID ());
size_t l = 2/*sessionID*/ + 1/*num leases*/ + i2p::data::LEASE_SIZE*tunnels.size (); size_t l = 2/*sessionID*/ + 1/*num leases*/ + i2p::data::LEASE_SIZE*tunnels.size ();
m_Owner.SendI2CPMessage (I2CP_REQUEST_VARIABLE_LEASESET_MESSAGE, leases - 3, l); m_Owner->SendI2CPMessage (I2CP_REQUEST_VARIABLE_LEASESET_MESSAGE, leases - 3, l);
} }
void I2CPDestination::LeaseSetCreated (const uint8_t * buf, size_t len) void I2CPDestination::LeaseSetCreated (const uint8_t * buf, size_t len)
@ -78,10 +78,10 @@ namespace client
if (ls) if (ls)
{ {
bool sent = s->SendMsg (msg, ls); bool sent = s->SendMsg (msg, ls);
s->m_Owner.SendMessageStatusMessage (nonce, sent ? eI2CPMessageStatusGuaranteedSuccess : eI2CPMessageStatusGuaranteedFailure); s->m_Owner->SendMessageStatusMessage (nonce, sent ? eI2CPMessageStatusGuaranteedSuccess : eI2CPMessageStatusGuaranteedFailure);
} }
else else
s->m_Owner.SendMessageStatusMessage (nonce, eI2CPMessageStatusNoLeaseSet); s->m_Owner->SendMessageStatusMessage (nonce, eI2CPMessageStatusNoLeaseSet);
}); });
} }
} }
@ -117,9 +117,8 @@ namespace client
I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<boost::asio::ip::tcp::socket> socket): I2CPSession::I2CPSession (I2CPServer& owner, std::shared_ptr<boost::asio::ip::tcp::socket> socket):
m_Owner (owner), m_Socket (socket), m_Owner (owner), m_Socket (socket),
m_NextMessage (nullptr), m_NextMessageLen (0), m_NextMessageOffset (0), m_NextMessage (nullptr), m_NextMessageLen (0), m_NextMessageOffset (0),
m_MessageID (0) m_SessionID (0xFFFF), m_MessageID (0), m_IsSendAccepted (true)
{ {
RAND_bytes ((uint8_t *)&m_SessionID, 2);
} }
I2CPSession::~I2CPSession () I2CPSession::~I2CPSession ()
@ -134,6 +133,7 @@ namespace client
void I2CPSession::Stop () void I2CPSession::Stop ()
{ {
Terminate ();
} }
void I2CPSession::ReadProtocolByte () void I2CPSession::ReadProtocolByte ()
@ -167,24 +167,59 @@ namespace client
size_t offset = 0; // from m_Buffer size_t offset = 0; // from m_Buffer
if (m_NextMessage) if (m_NextMessage)
{ {
if (m_NextMessageOffset + bytes_transferred <= m_NextMessageLen) if (!m_NextMessageLen) // we didn't receive header yet
{ {
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, bytes_transferred); if (m_NextMessageOffset + bytes_transferred < I2CP_HEADER_SIZE)
m_NextMessageOffset += bytes_transferred; {
offset = bytes_transferred; // still no complete header
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, bytes_transferred);
m_NextMessageOffset += bytes_transferred;
offset = bytes_transferred;
}
else
{
// we know message length now
offset = I2CP_HEADER_SIZE - m_NextMessageOffset;
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, offset);
m_NextMessageLen = bufbe32toh (m_NextMessage + I2CP_HEADER_LENGTH_OFFSET) + I2CP_HEADER_SIZE;
m_NextMessageOffset = I2CP_HEADER_SIZE;
}
}
if (offset < bytes_transferred)
{
auto msgRemainingLen = m_NextMessageLen - m_NextMessageOffset;
auto bufRemainingLen = bytes_transferred - offset;
if (bufRemainingLen < msgRemainingLen)
{
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer + offset, bufRemainingLen);
m_NextMessageOffset += bufRemainingLen;
offset += bufRemainingLen;
}
else
{
// m_NextMessage complete
offset += msgRemainingLen;
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer + offset, msgRemainingLen);
HandleNextMessage (m_NextMessage);
delete[] m_NextMessage;
m_NextMessage = nullptr;
}
} }
else
{
// m_NextMessage complete
offset = m_NextMessageLen - m_NextMessageOffset;
memcpy (m_NextMessage + m_NextMessageOffset, m_Buffer, offset);
HandleNextMessage (m_NextMessage);
delete[] m_NextMessage;
m_NextMessage = nullptr;
}
} }
// process the rest
while (offset < bytes_transferred) while (offset < bytes_transferred)
{ {
if (bytes_transferred - offset < I2CP_HEADER_SIZE)
{
// we don't have message header yet
m_NextMessage = new uint8_t[0xFFFF]; // allocate 64K
m_NextMessageLen = 0; // we must set message length later
m_NextMessageOffset = bytes_transferred - offset;
memcpy (m_NextMessage, m_Buffer + offset, m_NextMessageOffset); // just copy it
break;
}
auto msgLen = bufbe32toh (m_Buffer + offset + I2CP_HEADER_LENGTH_OFFSET) + I2CP_HEADER_SIZE; auto msgLen = bufbe32toh (m_Buffer + offset + I2CP_HEADER_LENGTH_OFFSET) + I2CP_HEADER_SIZE;
if (msgLen > 0xFFFF) // 64K if (msgLen > 0xFFFF) // 64K
{ {
@ -236,14 +271,20 @@ namespace client
void I2CPSession::SendI2CPMessage (uint8_t type, const uint8_t * payload, size_t len) void I2CPSession::SendI2CPMessage (uint8_t type, const uint8_t * payload, size_t len)
{ {
auto l = len + I2CP_HEADER_SIZE; auto socket = m_Socket;
uint8_t * buf = new uint8_t[l]; if (socket)
htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len); {
buf[I2CP_HEADER_TYPE_OFFSET] = type; auto l = len + I2CP_HEADER_SIZE;
memcpy (buf + I2CP_HEADER_SIZE, payload, len); uint8_t * buf = new uint8_t[l];
boost::asio::async_write (*m_Socket, boost::asio::buffer (buf, l), boost::asio::transfer_all (), htobe32buf (buf + I2CP_HEADER_LENGTH_OFFSET, len);
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (), buf[I2CP_HEADER_TYPE_OFFSET] = type;
std::placeholders::_1, std::placeholders::_2, buf)); memcpy (buf + I2CP_HEADER_SIZE, payload, len);
boost::asio::async_write (*socket, boost::asio::buffer (buf, l), boost::asio::transfer_all (),
std::bind(&I2CPSession::HandleI2CPMessageSent, shared_from_this (),
std::placeholders::_1, std::placeholders::_2, buf));
}
else
LogPrint (eLogError, "I2CP: Can't write to the socket");
} }
void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, const uint8_t * buf) void I2CPSession::HandleI2CPMessageSent (const boost::system::error_code& ecode, std::size_t bytes_transferred, const uint8_t * buf)
@ -276,21 +317,24 @@ namespace client
size_t offset = 0; size_t offset = 0;
while (offset < len) while (offset < len)
{ {
auto semicolon = (const uint8_t *)memchr (buf + offset, ';', len - offset); std::string param = ExtractString (buf + offset, len - offset);
if (semicolon) offset += param.length () + 1;
if (buf[offset] != '=')
{ {
auto l = semicolon - buf - offset + 1; LogPrint (eLogWarning, "I2CP: Unexpected character ", buf[offset], " instead '=' after ", param);
auto equal = (const uint8_t *)memchr (buf + offset, '=', l); break;
if (equal) }
{ offset++;
auto l1 = equal - buf - offset + 1;
mapping.insert (std::make_pair (std::string ((const char *)(buf + offset), l1 -1), std::string value = ExtractString (buf + offset, len - offset);
std::string ((const char *)(buf + offset + l1), l - l1 - 2))); offset += value.length () + 1;
} if (buf[offset] != ';')
offset += l; {
} LogPrint (eLogWarning, "I2CP: Unexpected character ", buf[offset], " instead ';' after ", value);
else break;
break; }
offset++;
mapping.insert (std::make_pair (param, value));
} }
} }
@ -311,23 +355,45 @@ namespace client
void I2CPSession::CreateSessionMessageHandler (const uint8_t * buf, size_t len) void I2CPSession::CreateSessionMessageHandler (const uint8_t * buf, size_t len)
{ {
RAND_bytes ((uint8_t *)&m_SessionID, 2);
auto identity = std::make_shared<i2p::data::IdentityEx>(); auto identity = std::make_shared<i2p::data::IdentityEx>();
size_t offset = identity->FromBuffer (buf, len); size_t offset = identity->FromBuffer (buf, len);
if (!offset)
{
LogPrint (eLogError, "I2CP: create session maformed identity");
SendSessionStatusMessage (3); // invalid
return;
}
uint16_t optionsSize = bufbe16toh (buf + offset); uint16_t optionsSize = bufbe16toh (buf + offset);
offset += 2; offset += 2;
if (optionsSize > len - offset)
{
LogPrint (eLogError, "I2CP: options size ", optionsSize, "exceeds message size");
SendSessionStatusMessage (3); // invalid
return;
}
std::map<std::string, std::string> params; std::map<std::string, std::string> params;
ExtractMapping (buf + offset, optionsSize, params); ExtractMapping (buf + offset, optionsSize, params);
offset += optionsSize; offset += optionsSize; // options
if (params[I2CP_PARAM_MESSAGE_RELIABILITY] == "none") m_IsSendAccepted = false;
offset += 8; // date offset += 8; // date
if (identity->Verify (buf, offset, buf + offset)) // signature if (identity->Verify (buf, offset, buf + offset)) // signature
{ {
bool isPublic = true; bool isPublic = true;
if (params[I2CP_PARAM_DONT_PUBLISH_LEASESET] == "true") isPublic = false; if (params[I2CP_PARAM_DONT_PUBLISH_LEASESET] == "true") isPublic = false;
m_Destination = std::make_shared<I2CPDestination>(*this, identity, isPublic, params); if (!m_Destination)
m_Destination->Start (); {
SendSessionStatusMessage (1); // created m_Destination = std::make_shared<I2CPDestination>(shared_from_this (), identity, isPublic, params);
LogPrint (eLogDebug, "I2CP: session ", m_SessionID, " created"); SendSessionStatusMessage (1); // created
LogPrint (eLogDebug, "I2CP: session ", m_SessionID, " created");
m_Destination->Start ();
}
else
{
LogPrint (eLogError, "I2CP: session already exists");
SendSessionStatusMessage (4); // refused
}
} }
else else
{ {
@ -340,7 +406,11 @@ namespace client
{ {
SendSessionStatusMessage (0); // destroy SendSessionStatusMessage (0); // destroy
LogPrint (eLogDebug, "I2CP: session ", m_SessionID, " destroyed"); LogPrint (eLogDebug, "I2CP: session ", m_SessionID, " destroyed");
Terminate (); if (m_Destination)
{
m_Destination->Stop ();
m_Destination = 0;
}
} }
void I2CPSession::ReconfigureSessionMessageHandler (const uint8_t * buf, size_t len) void I2CPSession::ReconfigureSessionMessageHandler (const uint8_t * buf, size_t len)
@ -359,6 +429,7 @@ namespace client
void I2CPSession::SendMessageStatusMessage (uint32_t nonce, I2CPMessageStatus status) void I2CPSession::SendMessageStatusMessage (uint32_t nonce, I2CPMessageStatus status)
{ {
if (!nonce) return; // don't send status with zero nonce
uint8_t buf[15]; uint8_t buf[15];
htobe16buf (buf, m_SessionID); htobe16buf (buf, m_SessionID);
htobe32buf (buf + 2, m_MessageID++); htobe32buf (buf + 2, m_MessageID++);
@ -402,7 +473,8 @@ namespace client
uint32_t payloadLen = bufbe32toh (buf + offset); uint32_t payloadLen = bufbe32toh (buf + offset);
offset += 4; offset += 4;
uint32_t nonce = bufbe32toh (buf + offset + payloadLen); uint32_t nonce = bufbe32toh (buf + offset + payloadLen);
SendMessageStatusMessage (nonce, eI2CPMessageStatusAccepted); // accepted if (m_IsSendAccepted)
SendMessageStatusMessage (nonce, eI2CPMessageStatusAccepted); // accepted
m_Destination->SendMsgTo (buf + offset, payloadLen, identity.GetIdentHash (), nonce); m_Destination->SendMsgTo (buf + offset, payloadLen, identity.GetIdentHash (), nonce);
} }
} }
@ -418,7 +490,7 @@ namespace client
void I2CPSession::HostLookupMessageHandler (const uint8_t * buf, size_t len) void I2CPSession::HostLookupMessageHandler (const uint8_t * buf, size_t len)
{ {
uint16_t sessionID = bufbe16toh (buf); uint16_t sessionID = bufbe16toh (buf);
if (sessionID == m_SessionID) if (sessionID == m_SessionID || sessionID == 0xFFFF) // -1 means without session
{ {
uint32_t requestID = bufbe32toh (buf + 2); uint32_t requestID = bufbe32toh (buf + 2);
//uint32_t timeout = bufbe32toh (buf + 6); //uint32_t timeout = bufbe32toh (buf + 6);
@ -445,15 +517,17 @@ namespace client
return; return;
} }
if (m_Destination) std::shared_ptr<LeaseSetDestination> destination = m_Destination;
if(!destination) destination = i2p::client::context.GetSharedLocalDestination ();
if (destination)
{ {
auto ls = m_Destination->FindLeaseSet (ident); auto ls = destination->FindLeaseSet (ident);
if (ls) if (ls)
SendHostReplyMessage (requestID, ls->GetIdentity ()); SendHostReplyMessage (requestID, ls->GetIdentity ());
else else
{ {
auto s = shared_from_this (); auto s = shared_from_this ();
m_Destination->RequestDestination (ident, destination->RequestDestination (ident,
[s, requestID](std::shared_ptr<i2p::data::LeaseSet> leaseSet) [s, requestID](std::shared_ptr<i2p::data::LeaseSet> leaseSet)
{ {
s->SendHostReplyMessage (requestID, leaseSet ? leaseSet->GetIdentity () : nullptr); s->SendHostReplyMessage (requestID, leaseSet ? leaseSet->GetIdentity () : nullptr);

8
I2CP.h
Unescape View File

@ -56,14 +56,15 @@ namespace client
}; };
// params // params
const char I2CP_PARAM_DONT_PUBLISH_LEASESET[] = "i2cp.dontPublishLeaseSet "; const char I2CP_PARAM_DONT_PUBLISH_LEASESET[] = "i2cp.dontPublishLeaseSet";
const char I2CP_PARAM_MESSAGE_RELIABILITY[] = "i2cp.messageReliability";
class I2CPSession; class I2CPSession;
class I2CPDestination: public LeaseSetDestination class I2CPDestination: public LeaseSetDestination
{ {
public: public:
I2CPDestination (I2CPSession& owner, std::shared_ptr<const i2p::data::IdentityEx> identity, bool isPublic, const std::map<std::string, std::string>& params); I2CPDestination (std::shared_ptr<I2CPSession> owner, std::shared_ptr<const i2p::data::IdentityEx> identity, bool isPublic, const std::map<std::string, std::string>& params);
void SetEncryptionPrivateKey (const uint8_t * key); void SetEncryptionPrivateKey (const uint8_t * key);
void LeaseSetCreated (const uint8_t * buf, size_t len); // called from I2CPSession void LeaseSetCreated (const uint8_t * buf, size_t len); // called from I2CPSession
@ -87,7 +88,7 @@ namespace client
private: private:
I2CPSession& m_Owner; std::shared_ptr<I2CPSession> m_Owner;
std::shared_ptr<const i2p::data::IdentityEx> m_Identity; std::shared_ptr<const i2p::data::IdentityEx> m_Identity;
uint8_t m_EncryptionPrivateKey[256]; uint8_t m_EncryptionPrivateKey[256];
uint64_t m_LeaseSetExpirationTime; uint64_t m_LeaseSetExpirationTime;
@ -148,6 +149,7 @@ namespace client
std::shared_ptr<I2CPDestination> m_Destination; std::shared_ptr<I2CPDestination> m_Destination;
uint16_t m_SessionID; uint16_t m_SessionID;
uint32_t m_MessageID; uint32_t m_MessageID;
bool m_IsSendAccepted;
}; };
typedef void (I2CPSession::*I2CPMessageHandler)(const uint8_t * buf, size_t len); typedef void (I2CPSession::*I2CPMessageHandler)(const uint8_t * buf, size_t len);

11
Streaming.cpp
Unescape View File

@ -232,6 +232,11 @@ namespace stream
bool acknowledged = false; bool acknowledged = false;
auto ts = i2p::util::GetMillisecondsSinceEpoch (); auto ts = i2p::util::GetMillisecondsSinceEpoch ();
uint32_t ackThrough = packet->GetAckThrough (); uint32_t ackThrough = packet->GetAckThrough ();
if (ackThrough > m_SequenceNumber)
{
LogPrint (eLogError, "Streaming: Unexpected ackThrough=", ackThrough, " > seqn=", m_SequenceNumber);
return;
}
int nackCount = packet->GetNACKCount (); int nackCount = packet->GetNACKCount ();
for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();) for (auto it = m_SentPackets.begin (); it != m_SentPackets.end ();)
{ {
@ -336,9 +341,9 @@ namespace stream
htobe32buf (packet + size, m_SequenceNumber++); htobe32buf (packet + size, m_SequenceNumber++);
size += 4; // sequenceNum size += 4; // sequenceNum
if (isNoAck) if (isNoAck)
htobe32buf (packet + size, m_LastReceivedSequenceNumber);
else
htobuf32 (packet + size, 0); htobuf32 (packet + size, 0);
else
htobe32buf (packet + size, m_LastReceivedSequenceNumber);
size += 4; // ack Through size += 4; // ack Through
packet[size] = 0; packet[size] = 0;
size++; // NACK count size++; // NACK count
@ -521,7 +526,7 @@ namespace stream
size += 4; // receiveStreamID size += 4; // receiveStreamID
htobe32buf (packet + size, m_SequenceNumber++); htobe32buf (packet + size, m_SequenceNumber++);
size += 4; // sequenceNum size += 4; // sequenceNum
htobe32buf (packet + size, m_LastReceivedSequenceNumber); htobe32buf (packet + size, m_LastReceivedSequenceNumber >= 0 ? m_LastReceivedSequenceNumber : 0);
size += 4; // ack Through size += 4; // ack Through
packet[size] = 0; packet[size] = 0;
size++; // NACK count size++; // NACK count

Write Preview
Loading…
Cancel
Save