lokinet/llarp/service/endpoint.cpp

1271 lines
35 KiB
C++
Raw Normal View History

#include <service/endpoint.hpp>
2018-12-12 00:48:54 +00:00
#include <dht/messages/findintro.hpp>
2019-01-16 00:24:16 +00:00
#include <dht/messages/findrouter.hpp>
#include <dht/messages/gotintro.hpp>
#include <dht/messages/gotrouter.hpp>
#include <dht/messages/pubintro.hpp>
#include <nodedb.hpp>
#include <profiling.hpp>
#include <router/abstractrouter.hpp>
#include <routing/dht_message.hpp>
2019-06-19 22:30:07 +00:00
#include <routing/path_transfer_message.hpp>
#include <service/endpoint_state.hpp>
#include <service/endpoint_util.hpp>
#include <service/hidden_service_address_lookup.hpp>
#include <service/outbound_context.hpp>
2018-12-12 02:15:08 +00:00
#include <service/protocol.hpp>
2019-01-11 01:19:36 +00:00
#include <util/logic.hpp>
#include <util/str.hpp>
#include <util/buffer.hpp>
2019-06-02 21:19:10 +00:00
#include <util/memfn.hpp>
#include <hook/shell.hpp>
namespace llarp
{
namespace service
{
Endpoint::Endpoint(const std::string& name, AbstractRouter* r,
Context* parent)
: path::Builder(r, 3, path::default_len), context(parent)
{
m_state = std::make_unique< EndpointState >();
m_state->m_Router = r;
m_state->m_Name = name;
m_state->m_Tag.Zero();
}
bool
Endpoint::SetOption(const std::string& k, const std::string& v)
{
2019-07-18 16:28:17 +00:00
return m_state->SetOption(k, v, *this);
2018-08-09 19:02:17 +00:00
}
2019-04-08 12:01:52 +00:00
llarp_ev_loop_ptr
Endpoint::EndpointNetLoop()
{
if(m_state->m_IsolatedNetLoop)
return m_state->m_IsolatedNetLoop;
2019-07-06 17:03:40 +00:00
return Router()->netloop();
}
2018-08-16 14:34:15 +00:00
bool
Endpoint::NetworkIsIsolated() const
{
return m_state->m_IsolatedLogic.get() != nullptr
&& m_state->m_IsolatedNetLoop != nullptr;
2018-08-09 19:02:17 +00:00
}
2018-08-10 03:51:38 +00:00
bool
Endpoint::HasPendingPathToService(const Address& addr) const
{
return m_state->m_PendingServiceLookups.find(addr)
!= m_state->m_PendingServiceLookups.end();
2018-08-10 03:51:38 +00:00
}
void
Endpoint::RegenAndPublishIntroSet(llarp_time_t now, bool forceRebuild)
{
std::set< Introduction > I;
if(!GetCurrentIntroductionsWithFilter(
I, [now](const service::Introduction& intro) -> bool {
return now < intro.expiresAt
&& intro.expiresAt - now > (2 * 60 * 1000);
}))
{
2019-04-21 15:40:32 +00:00
LogWarn("could not publish descriptors for endpoint ", Name(),
" because we couldn't get enough valid introductions");
2018-10-29 16:48:36 +00:00
if(ShouldBuildMore(now) || forceRebuild)
2018-11-22 15:52:04 +00:00
ManualRebuild(1);
return;
}
introSet().I.clear();
for(auto& intro : I)
{
introSet().I.emplace_back(std::move(intro));
}
if(introSet().I.size() == 0)
{
2019-04-21 15:40:32 +00:00
LogWarn("not enough intros to publish introset for ", Name());
if(ShouldBuildMore(now) || forceRebuild)
ManualRebuild(1);
return;
}
introSet().topic = m_state->m_Tag;
if(!m_Identity.SignIntroSet(introSet(), now))
{
2019-04-21 15:40:32 +00:00
LogWarn("failed to sign introset for endpoint ", Name());
return;
}
if(PublishIntroSet(Router()))
{
2019-04-21 15:40:32 +00:00
LogInfo("(re)publishing introset for endpoint ", Name());
}
else
{
2019-04-21 15:40:32 +00:00
LogWarn("failed to publish intro set for endpoint ", Name());
}
}
bool
Endpoint::IsReady() const
{
const auto now = Now();
if(introSet().I.size() == 0)
return false;
if(introSet().IsExpired(now))
return false;
return true;
}
bool
Endpoint::HasPendingRouterLookup(const RouterID remote) const
{
const auto& routers = m_state->m_PendingRouters;
return routers.find(remote) != routers.end();
}
bool
Endpoint::GetEndpointWithConvoTag(const ConvoTag tag,
llarp::AlignedBuffer< 32 >& addr,
bool& snode) const
{
auto itr = Sessions().find(tag);
if(itr != Sessions().end())
{
snode = false;
addr = itr->second.remote.Addr();
return true;
}
else
{
for(const auto& item : m_state->m_SNodeSessions)
{
if(item.second.second == tag)
{
snode = true;
addr = item.first;
return true;
}
}
}
return false;
}
bool
Endpoint::IntrosetIsStale() const
{
return introSet().HasExpiredIntros(Now());
}
2019-02-11 17:14:43 +00:00
util::StatusObject
Endpoint::ExtractStatus() const
2019-02-08 19:43:25 +00:00
{
2019-02-11 17:14:43 +00:00
auto obj = path::Builder::ExtractStatus();
obj.Put("identity", m_Identity.pub.Addr().ToString());
return m_state->ExtractStatus(obj);
2019-02-08 19:43:25 +00:00
}
void
2018-07-18 22:50:05 +00:00
Endpoint::Tick(llarp_time_t now)
{
2019-04-23 16:13:22 +00:00
path::Builder::Tick(now);
2018-07-19 04:58:39 +00:00
// publish descriptors
2018-07-18 22:50:05 +00:00
if(ShouldPublishDescriptors(now))
{
RegenAndPublishIntroSet(now);
}
2019-02-21 19:26:59 +00:00
2018-12-13 12:27:14 +00:00
// expire snode sessions
EndpointUtil::ExpireSNodeSessions(now, m_state->m_SNodeSessions);
// expire pending tx
EndpointUtil::ExpirePendingTx(now, m_state->m_PendingLookups);
2018-08-14 21:17:18 +00:00
// expire pending router lookups
EndpointUtil::ExpirePendingRouterLookups(now, m_state->m_PendingRouters);
2018-08-14 21:17:18 +00:00
// prefetch addrs
for(const auto& addr : m_state->m_PrefetchAddrs)
{
if(!EndpointUtil::HasPathToService(addr, m_state->m_RemoteSessions))
{
2018-08-22 15:52:10 +00:00
if(!EnsurePathToService(
addr,
[](ABSL_ATTRIBUTE_UNUSED Address addr,
ABSL_ATTRIBUTE_UNUSED OutboundContext* ctx) {},
10000))
{
2019-04-21 15:40:32 +00:00
LogWarn("failed to ensure path to ", addr);
}
}
}
#ifdef TESTNET
2018-07-19 04:58:39 +00:00
// prefetch tags
for(const auto& tag : m_state->m_PrefetchTags)
2018-07-18 03:10:21 +00:00
{
auto itr = m_state->m_PrefetchedTags.find(tag);
if(itr == m_state->m_PrefetchedTags.end())
2018-07-18 03:10:21 +00:00
{
itr =
m_state->m_PrefetchedTags.emplace(tag, CachedTagResult(tag, this))
.first;
2018-07-19 04:58:39 +00:00
}
for(const auto& introset : itr->second.result)
{
2018-08-10 03:51:38 +00:00
if(HasPendingPathToService(introset.A.Addr()))
continue;
std::array< byte_t, 128 > tmp = {0};
2019-02-02 23:12:42 +00:00
llarp_buffer_t buf(tmp);
2019-07-08 14:22:47 +00:00
if(SendToServiceOrQueue(introset.A.Addr(), buf, eProtocolControl))
2019-04-21 15:40:32 +00:00
LogInfo(Name(), " send message to ", introset.A.Addr(), " for tag ",
tag.ToString());
2019-02-08 19:43:25 +00:00
else
2019-04-21 15:40:32 +00:00
LogWarn(Name(), " failed to send/queue data to ", introset.A.Addr(),
" for tag ", tag.ToString());
2018-07-18 03:10:21 +00:00
}
2018-07-18 22:50:05 +00:00
itr->second.Expire(now);
if(itr->second.ShouldRefresh(now))
2018-07-18 03:10:21 +00:00
{
auto path = PickRandomEstablishedPath();
if(path)
{
2018-08-14 21:17:18 +00:00
auto job = new TagLookupJob(this, &itr->second);
if(!job->SendRequestViaPath(path, Router()))
2019-04-21 15:40:32 +00:00
LogError(Name(), " failed to send tag lookup");
}
else
{
2019-04-21 15:40:32 +00:00
LogError(Name(), " has no paths for tag lookup");
2018-07-18 03:10:21 +00:00
}
}
}
#endif
2019-02-05 14:50:33 +00:00
// deregister dead sessions
EndpointUtil::DeregisterDeadSessions(now, m_state->m_DeadSessions);
// tick remote sessions
EndpointUtil::TickRemoteSessions(now, m_state->m_RemoteSessions,
m_state->m_DeadSessions);
2019-02-09 14:37:24 +00:00
// expire convotags
EndpointUtil::ExpireConvoSessions(now, Sessions());
2018-09-24 15:52:25 +00:00
}
bool
Endpoint::Stop()
{
// stop remote sessions
EndpointUtil::StopRemoteSessions(m_state->m_RemoteSessions);
// stop snode sessions
EndpointUtil::StopSnodeSessions(m_state->m_SNodeSessions);
if(m_OnDown)
m_OnDown->NotifyAsync(NotifyParams());
2019-04-21 15:40:32 +00:00
return path::Builder::Stop();
}
2018-07-18 03:10:21 +00:00
uint64_t
Endpoint::GenTXID()
{
uint64_t txid = randint();
const auto& lookups = m_state->m_PendingLookups;
while(lookups.find(txid) != lookups.end())
2018-07-18 03:10:21 +00:00
++txid;
return txid;
}
2018-07-16 03:32:13 +00:00
std::string
Endpoint::Name() const
{
return m_state->m_Name + ":" + m_Identity.pub.Name();
2018-07-16 03:32:13 +00:00
}
2018-08-04 02:59:32 +00:00
void
Endpoint::PutLookup(IServiceLookup* lookup, uint64_t txid)
{
m_state->m_PendingLookups.emplace(
txid, std::unique_ptr< IServiceLookup >(lookup));
2018-08-04 02:59:32 +00:00
}
bool
2019-05-03 13:15:03 +00:00
Endpoint::HandleGotIntroMessage(dht::GotIntroMessage_constptr msg)
{
2018-07-18 03:10:21 +00:00
std::set< IntroSet > remote;
auto currentPub = m_state->m_CurrentPublishTX;
for(const auto& introset : msg->I)
{
if(!introset.Verify(Now()))
{
if(m_Identity.pub == introset.A && currentPub == msg->T)
IntroSetPublishFail();
return true;
2018-07-18 22:50:05 +00:00
}
if(m_Identity.pub == introset.A && currentPub == msg->T)
2018-07-18 22:50:05 +00:00
{
2019-04-21 15:40:32 +00:00
LogInfo(
"got introset publish confirmation for hidden service endpoint ",
2018-07-16 03:32:13 +00:00
Name());
2018-07-17 06:17:13 +00:00
IntroSetPublished();
2018-07-18 03:10:21 +00:00
return true;
}
2019-07-06 17:03:40 +00:00
remote.insert(introset);
}
auto& lookups = m_state->m_PendingLookups;
auto itr = lookups.find(msg->T);
if(itr == lookups.end())
2018-07-18 03:10:21 +00:00
{
2019-04-21 15:40:32 +00:00
LogWarn("invalid lookup response for hidden service endpoint ", Name(),
" txid=", msg->T);
2018-07-20 04:50:28 +00:00
return true;
2018-07-18 03:10:21 +00:00
}
2018-08-14 21:17:18 +00:00
std::unique_ptr< IServiceLookup > lookup = std::move(itr->second);
lookups.erase(itr);
2018-08-14 21:17:18 +00:00
lookup->HandleResponse(remote);
return true;
}
bool
Endpoint::HasInboundConvo(const Address& addr) const
{
for(const auto& item : Sessions())
{
if(item.second.remote.Addr() == addr && item.second.inbound)
return true;
}
return false;
}
2018-08-09 19:02:17 +00:00
void
Endpoint::PutSenderFor(const ConvoTag& tag, const ServiceInfo& info,
bool inbound)
2018-08-09 19:02:17 +00:00
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
{
itr = Sessions().emplace(tag, Session{}).first;
2019-07-01 13:44:25 +00:00
itr->second.inbound = inbound;
itr->second.remote = info;
2018-08-09 19:02:17 +00:00
}
2018-10-29 16:48:36 +00:00
itr->second.lastUsed = Now();
2018-08-09 19:02:17 +00:00
}
bool
Endpoint::GetSenderFor(const ConvoTag& tag, ServiceInfo& si) const
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
return false;
si = itr->second.remote;
return true;
}
void
Endpoint::PutIntroFor(const ConvoTag& tag, const Introduction& intro)
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
{
return;
2018-08-09 19:02:17 +00:00
}
itr->second.intro = intro;
2018-10-29 16:48:36 +00:00
itr->second.lastUsed = Now();
2018-08-09 19:02:17 +00:00
}
bool
Endpoint::GetIntroFor(const ConvoTag& tag, Introduction& intro) const
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
return false;
intro = itr->second.intro;
return true;
}
2019-02-21 16:45:33 +00:00
void
Endpoint::PutReplyIntroFor(const ConvoTag& tag, const Introduction& intro)
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2019-02-21 16:45:33 +00:00
{
return;
2019-02-21 16:45:33 +00:00
}
itr->second.replyIntro = intro;
itr->second.lastUsed = Now();
}
bool
Endpoint::GetReplyIntroFor(const ConvoTag& tag, Introduction& intro) const
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2019-02-21 16:45:33 +00:00
return false;
intro = itr->second.replyIntro;
return true;
}
2018-08-09 19:02:17 +00:00
bool
Endpoint::GetConvoTagsForService(const Address& addr,
2018-08-09 19:02:17 +00:00
std::set< ConvoTag >& tags) const
{
return EndpointUtil::GetConvoTagsForService(Sessions(), addr, tags);
2018-08-09 19:02:17 +00:00
}
bool
Endpoint::GetCachedSessionKeyFor(const ConvoTag& tag,
SharedSecret& secret) const
2018-08-09 19:02:17 +00:00
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
return false;
secret = itr->second.sharedKey;
2018-08-09 19:02:17 +00:00
return true;
}
void
Endpoint::PutCachedSessionKeyFor(const ConvoTag& tag, const SharedSecret& k)
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
{
itr = Sessions().emplace(tag, Session{}).first;
2018-08-09 19:02:17 +00:00
}
itr->second.sharedKey = k;
2018-10-29 16:48:36 +00:00
itr->second.lastUsed = Now();
2018-08-09 19:02:17 +00:00
}
bool
Endpoint::LoadKeyFile()
{
const auto& keyfile = m_state->m_Keyfile;
if(!keyfile.empty())
{
if(!m_Identity.EnsureKeys(keyfile))
{
LogError("Can't ensure keyfile [", keyfile, "]");
return false;
}
}
else
{
m_Identity.RegenerateKeys();
}
return true;
}
bool
Endpoint::Start()
{
// how can I tell if a m_Identity isn't loaded?
2018-08-09 19:02:17 +00:00
if(!m_DataHandler)
{
m_DataHandler = this;
}
2018-08-16 14:34:15 +00:00
// this does network isolation
while(m_state->m_OnInit.size())
2018-08-09 19:02:17 +00:00
{
if(m_state->m_OnInit.front()())
m_state->m_OnInit.pop_front();
2018-08-09 19:02:17 +00:00
else
{
2019-04-21 15:40:32 +00:00
LogWarn("Can't call init of network isolation");
2018-08-09 19:02:17 +00:00
return false;
}
2018-08-09 19:02:17 +00:00
}
return true;
}
Endpoint::~Endpoint()
{
if(m_OnUp)
m_OnUp->Stop();
if(m_OnDown)
m_OnDown->Stop();
if(m_OnReady)
m_OnReady->Stop();
}
2018-07-18 03:10:21 +00:00
bool
Endpoint::PublishIntroSet(AbstractRouter* r)
2018-07-18 03:10:21 +00:00
{
// publish via near router
RouterID location = m_Identity.pub.Addr().as_array();
2018-09-24 14:31:58 +00:00
auto path = GetEstablishedPathClosestTo(location);
2018-10-23 17:15:22 +00:00
return path && PublishIntroSetVia(r, path);
2018-07-18 03:10:21 +00:00
}
2018-09-18 14:48:06 +00:00
struct PublishIntroSetJob : public IServiceLookup
{
IntroSet m_IntroSet;
Endpoint* m_Endpoint;
PublishIntroSetJob(Endpoint* parent, uint64_t id,
const IntroSet& introset)
: IServiceLookup(parent, id, "PublishIntroSet")
, m_IntroSet(introset)
, m_Endpoint(parent)
{
}
std::shared_ptr< routing::IMessage >
2018-09-18 14:48:06 +00:00
BuildRequestMessage()
{
auto msg = std::make_shared< routing::DHTMessage >();
msg->M.emplace_back(
std::make_unique< dht::PublishIntroMessage >(m_IntroSet, txid, 1));
2018-09-18 14:48:06 +00:00
return msg;
}
bool
HandleResponse(const std::set< IntroSet >& response)
{
if(response.size())
m_Endpoint->IntroSetPublished();
else
m_Endpoint->IntroSetPublishFail();
return true;
}
};
2018-07-18 03:10:21 +00:00
void
Endpoint::IntroSetPublishFail()
{
auto now = Now();
if(ShouldPublishDescriptors(now))
{
RegenAndPublishIntroSet(now);
}
2019-04-21 15:40:32 +00:00
else if(NumInStatus(path::ePathEstablished) < 3)
{
if(introSet().HasExpiredIntros(now))
ManualRebuild(1);
}
2018-09-18 14:48:06 +00:00
}
bool
Endpoint::PublishIntroSetVia(AbstractRouter* r, path::Path_ptr path)
2018-09-18 14:48:06 +00:00
{
auto job = new PublishIntroSetJob(this, GenTXID(), introSet());
2018-09-18 14:48:06 +00:00
if(job->SendRequestViaPath(path, r))
{
m_state->m_LastPublishAttempt = Now();
2018-09-18 14:48:06 +00:00
return true;
}
return false;
2018-07-18 03:10:21 +00:00
}
2019-05-07 17:46:38 +00:00
void
Endpoint::ResetInternalState()
{
path::Builder::ResetInternalState();
static auto resetState = [](auto& container, auto getter) {
std::for_each(container.begin(), container.end(), [getter](auto& item) {
getter(item)->ResetInternalState();
});
2019-05-07 17:46:38 +00:00
};
resetState(m_state->m_RemoteSessions,
[](const auto& item) { return item.second; });
resetState(m_state->m_SNodeSessions,
[](const auto& item) { return item.second.first; });
2019-05-07 17:46:38 +00:00
}
2018-07-18 03:10:21 +00:00
bool
2018-07-18 22:50:05 +00:00
Endpoint::ShouldPublishDescriptors(llarp_time_t now) const
2018-07-18 03:10:21 +00:00
{
2019-04-21 15:40:32 +00:00
if(NumInStatus(path::ePathEstablished) < 3)
return false;
2019-04-01 19:56:11 +00:00
// make sure we have all paths that are established
// in our introset
2019-06-17 13:05:37 +00:00
size_t numNotInIntroset = 0;
2019-04-23 16:13:22 +00:00
ForEachPath([&](const path::Path_ptr& p) {
2019-04-05 14:58:22 +00:00
if(!p->IsReady())
return;
for(const auto& i : introSet().I)
2019-04-05 14:58:22 +00:00
{
if(i == p->intro)
return;
}
2019-06-17 12:43:16 +00:00
++numNotInIntroset;
2019-04-05 14:58:22 +00:00
});
auto lastpub = m_state->m_LastPublishAttempt;
if(m_state->m_IntroSet.HasExpiredIntros(now) || numNotInIntroset > 1)
{
return now - lastpub >= INTROSET_PUBLISH_RETRY_INTERVAL;
}
return now - lastpub >= INTROSET_PUBLISH_INTERVAL;
2018-07-18 03:10:21 +00:00
}
void
Endpoint::IntroSetPublished()
{
m_state->m_LastPublish = Now();
2019-04-21 15:40:32 +00:00
LogInfo(Name(), " IntroSet publish confirmed");
if(m_OnReady)
m_OnReady->NotifyAsync(NotifyParams());
m_OnReady = nullptr;
2018-07-18 03:10:21 +00:00
}
void
Endpoint::IsolatedNetworkMainLoop()
{
m_state->m_IsolatedNetLoop = llarp_make_ev_loop();
m_state->m_IsolatedLogic = std::make_shared< llarp::Logic >();
if(SetupNetworking())
llarp_ev_loop_run_single_process(m_state->m_IsolatedNetLoop,
m_state->m_IsolatedLogic);
else
{
m_state->m_IsolatedNetLoop.reset();
m_state->m_IsolatedLogic.reset();
}
2018-08-09 19:02:17 +00:00
}
2019-05-10 16:19:33 +00:00
bool
Endpoint::SelectHop(llarp_nodedb* db, const std::set< RouterID >& prev,
RouterContact& cur, size_t hop, path::PathRole roles)
{
std::set< RouterID > exclude = prev;
for(const auto& snode : SnodeBlacklist())
2019-05-10 16:19:33 +00:00
exclude.insert(snode);
return path::Builder::SelectHop(db, exclude, cur, hop, roles);
}
bool
Endpoint::ShouldBundleRC() const
{
return m_state->m_BundleRC;
}
2018-07-22 23:14:29 +00:00
void
2019-04-21 15:40:32 +00:00
Endpoint::PutNewOutboundContext(const service::IntroSet& introset)
2018-07-22 23:14:29 +00:00
{
Address addr;
introset.A.CalculateAddress(addr.as_array());
2018-07-22 23:14:29 +00:00
auto& remoteSessions = m_state->m_RemoteSessions;
auto& serviceLookups = m_state->m_PendingServiceLookups;
if(remoteSessions.count(addr) >= MAX_OUTBOUND_CONTEXT_COUNT)
2018-07-22 23:14:29 +00:00
{
auto itr = remoteSessions.find(addr);
auto range = serviceLookups.equal_range(addr);
auto i = range.first;
if(i != range.second)
{
i->second(addr, itr->second.get());
++i;
}
serviceLookups.erase(addr);
return;
2018-07-22 23:14:29 +00:00
}
auto it = remoteSessions.emplace(
2019-04-23 16:13:22 +00:00
addr, std::make_shared< OutboundContext >(introset, this));
2019-04-21 15:40:32 +00:00
LogInfo("Created New outbound context for ", addr.ToString());
2018-07-22 23:14:29 +00:00
// inform pending
auto range = serviceLookups.equal_range(addr);
auto itr = range.first;
if(itr != range.second)
2018-07-22 23:14:29 +00:00
{
itr->second(addr, it->second.get());
++itr;
2018-07-22 23:14:29 +00:00
}
serviceLookups.erase(addr);
2018-07-22 23:14:29 +00:00
}
void
Endpoint::HandleVerifyGotRouter(dht::GotRouterMessage_constptr msg,
llarp_async_verify_rc* j)
{
auto& pendingRouters = m_state->m_PendingRouters;
auto itr = pendingRouters.find(msg->R[0].pubkey);
if(itr != pendingRouters.end())
{
if(j->valid)
itr->second.InformResult(msg->R);
else
itr->second.InformResult({});
pendingRouters.erase(itr);
}
delete j;
}
2018-08-10 21:34:11 +00:00
bool
2019-05-03 13:15:03 +00:00
Endpoint::HandleGotRouterMessage(dht::GotRouterMessage_constptr msg)
2018-08-10 21:34:11 +00:00
{
if(msg->R.size())
2018-08-10 21:34:11 +00:00
{
llarp_async_verify_rc* job = new llarp_async_verify_rc;
job->nodedb = Router()->nodedb();
job->cryptoworker = Router()->threadpool();
job->diskworker = Router()->diskworker();
job->logic = Router()->logic();
job->hook = std::bind(&Endpoint::HandleVerifyGotRouter, this, msg,
std::placeholders::_1);
job->rc = msg->R[0];
2018-08-10 21:34:11 +00:00
llarp_nodedb_async_verify(job);
2019-05-03 13:15:03 +00:00
}
else
{
auto& routers = m_state->m_PendingRouters;
auto itr = routers.begin();
while(itr != routers.end())
{
if(itr->second.txid == msg->txid)
{
itr->second.InformResult({});
itr = routers.erase(itr);
}
else
++itr;
}
2018-08-10 21:34:11 +00:00
}
return true;
2018-08-10 21:34:11 +00:00
}
void
Endpoint::EnsureRouterIsKnown(const RouterID& router)
{
2018-08-14 22:07:58 +00:00
if(router.IsZero())
return;
if(!Router()->nodedb()->Has(router))
2018-08-10 21:34:11 +00:00
{
2019-05-03 13:15:03 +00:00
LookupRouterAnon(router, nullptr);
2018-12-19 17:48:29 +00:00
}
}
2018-08-10 21:34:11 +00:00
2018-12-19 17:48:29 +00:00
bool
2019-05-03 13:15:03 +00:00
Endpoint::LookupRouterAnon(RouterID router, RouterLookupHandler handler)
2018-12-19 17:48:29 +00:00
{
auto& routers = m_state->m_PendingRouters;
if(routers.find(router) == routers.end())
2018-12-19 17:48:29 +00:00
{
auto path = GetEstablishedPathClosestTo(router);
routing::DHTMessage msg;
auto txid = GenTXID();
msg.M.emplace_back(
std::make_unique< dht::FindRouterMessage >(txid, router));
2018-12-19 17:48:29 +00:00
if(path && path->SendRoutingMessage(msg, Router()))
2018-12-19 17:48:29 +00:00
{
routers.emplace(router, RouterLookupJob(this, handler));
2018-12-19 17:48:29 +00:00
return true;
2018-08-10 21:34:11 +00:00
}
}
2018-12-19 17:48:29 +00:00
return false;
2018-08-10 21:34:11 +00:00
}
void
Endpoint::HandlePathBuilt(path::Path_ptr p)
{
2019-06-02 21:19:10 +00:00
p->SetDataHandler(util::memFn(&Endpoint::HandleHiddenServiceFrame, this));
p->SetDropHandler(util::memFn(&Endpoint::HandleDataDrop, this));
p->SetDeadChecker(util::memFn(&Endpoint::CheckPathIsDead, this));
path::Builder::HandlePathBuilt(p);
}
bool
2019-04-23 16:13:22 +00:00
Endpoint::HandleDataDrop(path::Path_ptr p, const PathID_t& dst,
uint64_t seq)
{
2019-04-21 15:40:32 +00:00
LogWarn(Name(), " message ", seq, " dropped by endpoint ", p->Endpoint(),
" via ", dst);
return true;
}
std::unordered_map< std::string, std::string >
Endpoint::NotifyParams() const
{
return {{"LOKINET_ADDR", m_Identity.pub.Addr().ToString()}};
}
bool
2019-06-28 14:12:20 +00:00
Endpoint::HandleDataMessage(path::Path_ptr path, const PathID_t from,
2019-05-03 13:15:03 +00:00
std::shared_ptr< ProtocolMessage > msg)
{
2019-06-06 10:52:27 +00:00
msg->sender.UpdateAddr();
PutSenderFor(msg->tag, msg->sender, true);
2019-06-28 14:12:20 +00:00
PutReplyIntroFor(msg->tag, path->intro);
Introduction intro;
2019-07-01 13:44:25 +00:00
intro.pathID = from;
intro.router = PubKey(path->Endpoint());
2019-06-28 14:48:00 +00:00
intro.expiresAt = std::min(path->ExpireTime(), msg->introReply.expiresAt);
2019-06-28 14:12:20 +00:00
PutIntroFor(msg->tag, intro);
2018-09-18 17:48:26 +00:00
return ProcessDataMessage(msg);
}
2018-11-29 14:01:13 +00:00
bool
2019-07-01 13:44:25 +00:00
Endpoint::HasPathToSNode(const RouterID ident) const
2018-11-29 14:01:13 +00:00
{
auto range = m_state->m_SNodeSessions.equal_range(ident);
auto itr = range.first;
2018-11-29 14:01:13 +00:00
while(itr != range.second)
{
if(itr->second.first->IsReady())
2018-11-29 14:01:13 +00:00
{
return true;
}
++itr;
}
return false;
}
2018-11-29 13:12:35 +00:00
bool
2019-05-03 13:15:03 +00:00
Endpoint::ProcessDataMessage(std::shared_ptr< ProtocolMessage > msg)
2018-11-29 13:12:35 +00:00
{
2019-06-11 16:44:05 +00:00
if(msg->proto == eProtocolTrafficV4 || msg->proto == eProtocolTrafficV6)
2018-11-29 13:12:35 +00:00
{
util::Lock l(&m_state->m_InboundTrafficQueueMutex);
m_state->m_InboundTrafficQueue.emplace(msg);
2019-05-22 17:47:33 +00:00
return true;
2018-11-29 13:12:35 +00:00
}
2019-07-06 17:03:40 +00:00
if(msg->proto == eProtocolControl)
2018-11-29 13:12:35 +00:00
{
// TODO: implement me (?)
// right now it's just random noise
2018-11-29 13:12:35 +00:00
return true;
}
return false;
}
2019-03-08 16:00:45 +00:00
void
Endpoint::RemoveConvoTag(const ConvoTag& t)
{
Sessions().erase(t);
2019-03-08 16:00:45 +00:00
}
bool
Endpoint::HandleHiddenServiceFrame(path::Path_ptr p,
const ProtocolFrame& frame)
{
if(frame.R)
2019-03-08 16:00:45 +00:00
{
// handle discard
ServiceInfo si;
if(!GetSenderFor(frame.T, si))
2019-03-08 16:00:45 +00:00
return false;
// verify source
if(!frame.Verify(si))
2019-03-08 16:00:45 +00:00
return false;
// remove convotag it doesn't exist
LogWarn("remove convotag T=", frame.T);
RemoveConvoTag(frame.T);
2019-03-08 16:00:45 +00:00
return true;
}
if(!frame.AsyncDecryptAndVerify(EndpointLogic(), p, CryptoWorker(),
m_Identity, m_DataHandler))
2019-03-08 16:00:45 +00:00
{
// send discard
ProtocolFrame f;
f.R = 1;
f.T = frame.T;
2019-03-08 16:00:45 +00:00
f.F = p->intro.pathID;
if(!f.Sign(m_Identity))
2019-03-08 16:00:45 +00:00
return false;
2019-05-02 16:23:31 +00:00
{
util::Lock lock(&m_state->m_SendQueueMutex);
m_state->m_SendQueue.emplace_back(
2019-05-02 16:23:31 +00:00
std::make_shared< const routing::PathTransferMessage >(f,
frame.F),
p);
}
return true;
2019-03-08 16:00:45 +00:00
}
return true;
}
2019-04-23 16:13:22 +00:00
void Endpoint::HandlePathDied(path::Path_ptr)
2018-09-17 15:32:37 +00:00
{
2019-03-30 13:02:10 +00:00
RegenAndPublishIntroSet(Now(), true);
}
bool
Endpoint::CheckPathIsDead(path::Path_ptr, llarp_time_t dlt)
{
2019-04-05 14:58:22 +00:00
return dlt > path::alive_timeout;
}
2018-08-10 21:34:11 +00:00
bool
2018-10-15 15:43:41 +00:00
Endpoint::OnLookup(const Address& addr, const IntroSet* introset,
const RouterID& endpoint)
2018-08-10 21:34:11 +00:00
{
auto now = Now();
auto& fails = m_state->m_ServiceLookupFails;
auto& lookups = m_state->m_PendingServiceLookups;
2018-10-10 21:31:03 +00:00
if(introset == nullptr || introset->IsExpired(now))
{
2019-04-21 15:40:32 +00:00
LogError(Name(), " failed to lookup ", addr.ToString(), " from ",
endpoint);
fails[endpoint] = fails[endpoint] + 1;
// inform one
auto itr = lookups.find(addr);
if(itr != lookups.end())
{
itr->second(addr, nullptr);
lookups.erase(itr);
}
2018-08-10 21:34:11 +00:00
return false;
}
2019-07-06 17:03:40 +00:00
PutNewOutboundContext(*introset);
2018-08-10 21:34:11 +00:00
return true;
}
2018-07-19 04:58:39 +00:00
bool
2019-07-01 13:44:25 +00:00
Endpoint::EnsurePathToService(const Address remote, PathEnsureHook hook,
ABSL_ATTRIBUTE_UNUSED llarp_time_t timeoutMS,
bool randomPath)
2018-07-19 04:58:39 +00:00
{
path::Path_ptr path = nullptr;
if(randomPath)
path = PickRandomEstablishedPath();
else
path = GetEstablishedPathClosestTo(remote.ToRouter());
if(!path)
{
2019-04-21 15:40:32 +00:00
LogWarn("No outbound path for lookup yet");
2018-12-02 15:26:26 +00:00
BuildOne();
return false;
}
2019-04-21 15:40:32 +00:00
LogInfo(Name(), " Ensure Path to ", remote.ToString());
auto& sessions = m_state->m_RemoteSessions;
2018-07-22 23:14:29 +00:00
{
auto itr = sessions.find(remote);
if(itr != sessions.end())
2018-07-22 23:14:29 +00:00
{
2018-08-22 15:52:10 +00:00
hook(itr->first, itr->second.get());
2018-07-22 23:14:29 +00:00
return true;
}
}
auto& lookups = m_state->m_PendingServiceLookups;
if(lookups.count(remote) >= MaxConcurrentLookups)
{
2019-04-21 15:40:32 +00:00
LogWarn(Name(), " has too many pending service lookups for ",
remote.ToString());
return false;
}
using namespace std::placeholders;
2018-08-10 21:34:11 +00:00
HiddenServiceAddressLookup* job = new HiddenServiceAddressLookup(
2019-06-02 21:19:10 +00:00
this, util::memFn(&Endpoint::OnLookup, this), remote, GenTXID());
2019-04-21 15:40:32 +00:00
LogInfo("doing lookup for ", remote, " via ", path->Endpoint());
2018-08-10 21:34:11 +00:00
if(job->SendRequestViaPath(path, Router()))
{
lookups.emplace(remote, hook);
2018-08-10 21:34:11 +00:00
return true;
}
2019-04-21 15:40:32 +00:00
LogError("send via path failed");
2018-08-10 21:34:11 +00:00
return false;
2018-07-19 04:58:39 +00:00
}
2018-11-29 13:12:35 +00:00
void
2019-07-01 13:44:25 +00:00
Endpoint::EnsurePathToSNode(const RouterID snode, SNodeEnsureHook h)
2018-11-29 13:12:35 +00:00
{
auto& nodeSessions = m_state->m_SNodeSessions;
using namespace std::placeholders;
if(nodeSessions.count(snode) == 0)
2018-11-29 13:12:35 +00:00
{
ConvoTag tag;
// TODO: check for collision lol no we don't but maybe we will...
// some day :DDDDD
tag.Randomize();
2019-04-23 16:13:22 +00:00
auto session = std::make_shared< exit::SNodeSession >(
snode,
2019-07-01 13:44:25 +00:00
[=](const llarp_buffer_t& pkt) -> bool {
/// TODO: V6
return HandleInboundPacket(tag, pkt, eProtocolTrafficV4);
2019-07-01 13:44:25 +00:00
},
2019-07-18 16:28:17 +00:00
Router(), numPaths, numHops, false, ShouldBundleRC());
m_state->m_SNodeSessions.emplace(snode, std::make_pair(session, tag));
2018-11-29 13:12:35 +00:00
}
2019-04-30 21:36:27 +00:00
EnsureRouterIsKnown(snode);
auto range = nodeSessions.equal_range(snode);
auto itr = range.first;
while(itr != range.second)
{
if(itr->second.first->IsReady())
h(snode, itr->second.first);
2019-03-07 15:17:29 +00:00
else
2019-04-30 21:36:27 +00:00
{
itr->second.first->AddReadyHook(std::bind(h, snode, _1));
itr->second.first->BuildOne();
2019-04-30 21:36:27 +00:00
}
++itr;
}
2018-11-29 13:12:35 +00:00
}
bool
2019-02-02 23:12:42 +00:00
Endpoint::SendToSNodeOrQueue(const RouterID& addr,
const llarp_buffer_t& buf)
2018-11-29 13:12:35 +00:00
{
2019-06-11 16:44:05 +00:00
auto pkt = std::make_shared< net::IPPacket >();
2019-04-30 21:36:27 +00:00
if(!pkt->Load(buf))
2018-11-29 13:12:35 +00:00
return false;
2019-04-30 21:36:27 +00:00
EnsurePathToSNode(addr, [pkt](RouterID, exit::BaseSession_ptr s) {
if(s)
s->QueueUpstreamTraffic(*pkt, routing::ExitPadSize);
});
return true;
2018-11-29 13:12:35 +00:00
}
2019-04-30 16:07:17 +00:00
void Endpoint::Pump(llarp_time_t)
2019-04-25 17:15:56 +00:00
{
const auto& sessions = m_state->m_SNodeSessions;
auto& queue = m_state->m_InboundTrafficQueue;
2019-04-30 13:56:39 +00:00
EndpointLogic()->queue_func([&]() {
2019-05-22 16:20:50 +00:00
// send downstream packets to user for snode
for(const auto& item : sessions)
item.second.first->FlushDownstream();
// send downstream traffic to user for hidden service
util::Lock lock(&m_state->m_InboundTrafficQueueMutex);
while(queue.size())
{
const auto& msg = queue.top();
2019-07-18 16:28:17 +00:00
const llarp_buffer_t buf(msg->payload);
HandleInboundPacket(msg->tag, buf, msg->proto);
queue.pop();
}
2019-04-30 13:56:39 +00:00
});
2019-04-30 16:07:17 +00:00
auto router = Router();
// TODO: locking on this container
for(const auto& item : m_state->m_RemoteSessions)
2019-04-30 16:07:17 +00:00
item.second->FlushUpstream();
// TODO: locking on this container
for(const auto& item : sessions)
item.second.first->FlushUpstream();
util::Lock lock(&m_state->m_SendQueueMutex);
// send outbound traffic
for(const auto& item : m_state->m_SendQueue)
2019-04-30 16:07:17 +00:00
item.second->SendRoutingMessage(*item.first, router);
m_state->m_SendQueue.clear();
2019-04-25 17:15:56 +00:00
}
bool
Endpoint::EnsureConvo(ABSL_ATTRIBUTE_UNUSED const AlignedBuffer< 32 > addr,
bool snode,
ABSL_ATTRIBUTE_UNUSED ConvoEventListener_ptr ev)
{
if(snode)
{
}
// TODO: something meaningful
return false;
}
2018-08-22 15:52:10 +00:00
bool
2019-06-06 10:52:27 +00:00
Endpoint::SendToServiceOrQueue(const service::Address& remote,
const llarp_buffer_t& data, ProtocolType t)
2018-08-22 15:52:10 +00:00
{
// inbound converstation
2019-07-18 16:28:17 +00:00
const auto now = Now();
2018-11-14 12:23:08 +00:00
if(HasInboundConvo(remote))
{
auto transfer = std::make_shared< routing::PathTransferMessage >();
ProtocolFrame& f = transfer->T;
std::shared_ptr< path::Path > p;
std::set< ConvoTag > tags;
if(GetConvoTagsForService(remote, tags))
{
2019-06-28 14:12:20 +00:00
// the remote guy's intro
Introduction remoteIntro;
2019-06-28 14:12:20 +00:00
Introduction replyPath;
SharedSecret K;
// pick tag
for(const auto& tag : tags)
{
if(tag.IsZero())
continue;
if(!GetCachedSessionKeyFor(tag, K))
continue;
2019-06-28 14:12:20 +00:00
if(!GetReplyIntroFor(tag, replyPath))
continue;
if(!GetIntroFor(tag, remoteIntro))
continue;
// get path for intro
ForEachPath([&](path::Path_ptr path) {
if(path->intro == replyPath)
{
p = path;
2019-07-01 20:45:00 +00:00
return;
}
if(p && p->ExpiresSoon(now) && path->IsReady()
&& path->intro.router == replyPath.router)
{
2019-06-28 14:12:20 +00:00
p = path;
}
2019-06-28 14:12:20 +00:00
});
if(p)
{
2019-06-28 14:12:20 +00:00
f.T = tag;
}
}
if(p)
{
// TODO: check expiration of our end
ProtocolMessage m(f.T);
m.PutBuffer(data);
f.N.Randomize();
f.C.Zero();
transfer->Y.Randomize();
m.proto = t;
m.introReply = p->intro;
PutReplyIntroFor(f.T, m.introReply);
m.sender = m_Identity.pub;
m.seqno = GetSeqNoForConvo(f.T);
f.S = 1;
f.F = m.introReply.pathID;
transfer->P = remoteIntro.pathID;
if(!f.EncryptAndSign(m, K, m_Identity))
{
LogError("failed to encrypt and sign");
return false;
}
LogDebug(Name(), " send ", data.sz, " via ", remoteIntro.router);
{
util::Lock lock(&m_state->m_SendQueueMutex);
m_state->m_SendQueue.emplace_back(transfer, p);
}
return true;
}
}
}
// outbound converstation
auto& sessions = m_state->m_RemoteSessions;
if(EndpointUtil::HasPathToService(remote, sessions))
2018-08-22 15:52:10 +00:00
{
auto range = sessions.equal_range(remote);
auto itr = range.first;
while(itr != range.second)
{
if(itr->second->ReadyToSend())
{
itr->second->AsyncEncryptAndSendTo(data, t);
return true;
}
++itr;
}
2018-08-22 15:52:10 +00:00
}
auto& traffic = m_state->m_PendingTraffic;
traffic[remote].emplace_back(data, t);
// no converstation
2019-06-19 22:30:07 +00:00
return EnsurePathToService(
remote,
[&](Address r, OutboundContext* c) {
if(c)
{
c->UpdateIntroSet(true);
for(auto& pending : traffic[r])
{
2019-06-19 22:30:07 +00:00
c->AsyncEncryptAndSendTo(pending.Buffer(), pending.protocol);
}
2019-06-19 22:30:07 +00:00
}
traffic.erase(r);
2019-06-19 22:30:07 +00:00
},
5000, true);
}
2018-08-22 15:52:10 +00:00
2019-03-08 17:00:13 +00:00
bool
Endpoint::HasConvoTag(const ConvoTag& t) const
{
return Sessions().find(t) != Sessions().end();
2019-03-08 17:00:13 +00:00
}
2018-08-09 19:02:17 +00:00
uint64_t
Endpoint::GetSeqNoForConvo(const ConvoTag& tag)
{
auto itr = Sessions().find(tag);
if(itr == Sessions().end())
2018-08-09 19:02:17 +00:00
return 0;
return ++(itr->second.seqno);
}
2019-03-08 14:36:24 +00:00
bool
Endpoint::ShouldBuildMore(llarp_time_t now) const
{
2019-05-02 16:23:31 +00:00
const bool should = path::Builder::ShouldBuildMore(now);
2019-01-16 00:24:16 +00:00
// determine newest intro
Introduction intro;
if(!GetNewestIntro(intro))
return should;
// time from now that the newest intro expires at
2019-05-02 16:23:31 +00:00
if(intro.ExpiresSoon(now))
2018-09-27 11:07:20 +00:00
return should;
2019-05-02 16:31:08 +00:00
const auto dlt = intro.expiresAt - now;
2018-09-27 11:09:00 +00:00
return should
|| ( // try spacing tunnel builds out evenly in time
(dlt <= (path::default_lifetime / 4))
2019-07-18 16:28:17 +00:00
&& (NumInStatus(path::ePathBuilding) < numPaths));
}
2019-05-22 16:20:50 +00:00
std::shared_ptr< Logic >
2018-08-09 19:02:17 +00:00
Endpoint::RouterLogic()
2018-07-19 04:58:39 +00:00
{
return Router()->logic();
2018-07-19 04:58:39 +00:00
}
2019-05-22 16:20:50 +00:00
std::shared_ptr< Logic >
2018-08-09 19:02:17 +00:00
Endpoint::EndpointLogic()
{
return m_state->m_IsolatedLogic ? m_state->m_IsolatedLogic
: Router()->logic();
2018-08-09 19:02:17 +00:00
}
2019-07-09 13:47:24 +00:00
std::shared_ptr< llarp::thread::ThreadPool >
Endpoint::CryptoWorker()
2018-07-19 04:58:39 +00:00
{
return Router()->threadpool();
2018-07-19 04:58:39 +00:00
}
AbstractRouter*
Endpoint::Router()
{
return m_state->m_Router;
}
const std::set< RouterID >&
Endpoint::SnodeBlacklist() const
{
return m_state->m_SnodeBlacklist;
}
const IntroSet&
Endpoint::introSet() const
{
return m_state->m_IntroSet;
}
IntroSet&
Endpoint::introSet()
{
return m_state->m_IntroSet;
}
const ConvoMap&
Endpoint::Sessions() const
{
return m_state->m_Sessions;
}
ConvoMap&
Endpoint::Sessions()
{
return m_state->m_Sessions;
}
2018-07-12 18:21:44 +00:00
} // namespace service
2018-07-16 03:32:13 +00:00
} // namespace llarp