diff --git a/CMakeLists.txt b/CMakeLists.txt index f98a5671e..ea2651591 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -16,7 +16,7 @@ if(CCACHE_PROGRAM) endif() project(lokinet - VERSION 0.9.3 + VERSION 0.9.4 DESCRIPTION "lokinet - IP packet onion router" LANGUAGES C CXX) diff --git a/contrib/py/admin/lokinetmon b/contrib/py/admin/lokinetmon index 4c36b34e6..2c45b2435 100755 --- a/contrib/py/admin/lokinetmon +++ b/contrib/py/admin/lokinetmon @@ -155,7 +155,7 @@ class Monitor: def time_to(timestamp): """ return time until timestamp in seconds formatted""" if timestamp: - val = int((timestamp - now()) / 1000) + val = (timestamp - now()) / 1000.0 if val < 0: return "{} seconds ago".format(0-val) else: @@ -449,7 +449,7 @@ class Monitor: for intro in context['introset']['intros'] or []: y_pos = self._display_intro(y_pos, intro, "introset intro", context['paths']) for path in context['paths']: - y_pos = self._display_intro(y_pos, path['intro'], "inbound path intro", context['paths']) + y_pos = self._display_intro(y_pos, path['intro'], "inbound path [created {}]".format(self.time_to(path['buildStarted'])), context['paths']) return y_pos @@ -489,12 +489,8 @@ class Monitor: for intro in context['currentRemoteIntroset']['intros'] or []: y_pos = self._display_intro(y_pos, intro, "introset intro", paths) y_pos += 1 - for intro in context['badIntros'] or []: - y_pos = self._display_intro(y_pos, intro, "bad intro", paths) - y_pos += 1 return y_pos - def display_data(self): """ draw main window """ if self.data is not None: diff --git a/llarp/handlers/tun.cpp b/llarp/handlers/tun.cpp index 7ab8f1dcb..ccbb88221 100644 --- a/llarp/handlers/tun.cpp +++ b/llarp/handlers/tun.cpp @@ -397,6 +397,14 @@ namespace llarp service::Address addr, auto msg, bool isV6) -> bool { using service::Address; using service::OutboundContext; + if (HasInboundConvo(addr)) + { + // if we have an inbound convo to this address don't mark as outbound so we don't have a + // state race this codepath is hit when an application verifies that reverse and forward + // dns records match for an inbound session + SendDNSReply(addr, this, msg, reply, isV6); + return true; + } MarkAddressOutbound(addr); return EnsurePathToService( addr, @@ -424,6 +432,7 @@ namespace llarp service::Address addr, auto msg) -> bool { using service::Address; using service::OutboundContext; + // TODO: how do we handle SRV record lookups for inbound sessions? MarkAddressOutbound(addr); return EnsurePathToService( addr, @@ -1100,10 +1109,13 @@ namespace llarp } // try sending it on an existing convotag // this succeds for inbound convos, probably. - if (SendToOrQueue(to, pkt.ConstBuffer(), type)) + if (auto maybe = GetBestConvoTagFor(to)) { - MarkIPActive(dst); - return; + if (SendToOrQueue(*maybe, pkt.ConstBuffer(), type)) + { + MarkIPActive(dst); + return; + } } // try establishing a path to this guy // will fail if it's an inbound convo diff --git a/llarp/link/server.cpp b/llarp/link/server.cpp index 6e40b0df4..b8c9f748f 100644 --- a/llarp/link/server.cpp +++ b/llarp/link/server.cpp @@ -481,10 +481,9 @@ namespace llarp bool ILinkLayer::PutSession(const std::shared_ptr& s) { - static constexpr size_t MaxSessionsPerEndpoint = 5; Lock_t lock(m_PendingMutex); const auto address = s->GetRemoteEndpoint(); - if (m_Pending.count(address) >= MaxSessionsPerEndpoint) + if (m_Pending.count(address)) return false; m_Pending.emplace(address, s); return true; diff --git a/llarp/link/server.hpp b/llarp/link/server.hpp index cc3b5e84e..e58e4da56 100644 --- a/llarp/link/server.hpp +++ b/llarp/link/server.hpp @@ -245,7 +245,7 @@ namespace llarp SecretKey m_SecretKey; using AuthedLinks = std::unordered_multimap>; - using Pending = std::unordered_multimap>; + using Pending = std::unordered_map>; mutable DECLARE_LOCK(Mutex_t, m_AuthedLinksMutex, ACQUIRED_BEFORE(m_PendingMutex)); AuthedLinks m_AuthedLinks GUARDED_BY(m_AuthedLinksMutex); mutable DECLARE_LOCK(Mutex_t, m_PendingMutex, ACQUIRED_AFTER(m_AuthedLinksMutex)); diff --git a/llarp/service/endpoint.cpp b/llarp/service/endpoint.cpp index f995a3b84..796d9e536 100644 --- a/llarp/service/endpoint.cpp +++ b/llarp/service/endpoint.cpp @@ -400,7 +400,7 @@ namespace llarp { for (const auto& item : Sessions()) { - if (item.second.remote.Addr() == addr && item.second.inbound) + if (item.second.remote.Addr() == addr and item.second.inbound) return true; } return false; @@ -420,9 +420,24 @@ namespace llarp void Endpoint::PutSenderFor(const ConvoTag& tag, const ServiceInfo& info, bool inbound) { + if (info.Addr().IsZero()) + { + LogError(Name(), " cannot put invalid service info ", info, " T=", tag); + return; + } auto itr = Sessions().find(tag); if (itr == Sessions().end()) { + if (WantsOutboundSession(info.Addr()) and inbound) + { + LogWarn( + Name(), + " not adding sender for ", + info.Addr(), + " session is inbound and we want outbound T=", + tag); + return; + } itr = Sessions().emplace(tag, Session{}).first; itr->second.inbound = inbound; itr->second.remote = info; @@ -587,7 +602,10 @@ namespace llarp bool Endpoint::PublishIntroSet(const EncryptedIntroSet& introset, AbstractRouter* r) { - const auto paths = GetManyPathsWithUniqueEndpoints(this, llarp::dht::IntroSetRelayRedundancy); + const auto paths = GetManyPathsWithUniqueEndpoints( + this, + llarp::dht::IntroSetRelayRedundancy, + dht::Key_t{introset.derivedSigningKey.as_array()}); if (paths.size() != llarp::dht::IntroSetRelayRedundancy) { @@ -917,6 +935,7 @@ namespace llarp }); constexpr size_t min_unique_lns_endpoints = 2; + constexpr size_t max_unique_lns_endpoints = 7; // not enough paths if (paths.size() < min_unique_lns_endpoints) @@ -951,10 +970,16 @@ namespace llarp handler(result); }; + // pick up to max_unique_lns_endpoints random paths to do lookups from + std::vector chosenpaths; + chosenpaths.insert(chosenpaths.begin(), paths.begin(), paths.end()); + std::shuffle(chosenpaths.begin(), chosenpaths.end(), CSRNG{}); + chosenpaths.resize(std::min(paths.size(), max_unique_lns_endpoints)); + auto resultHandler = - m_state->lnsTracker.MakeResultHandler(name, paths.size(), maybeInvalidateCache); + m_state->lnsTracker.MakeResultHandler(name, chosenpaths.size(), maybeInvalidateCache); - for (const auto& path : paths) + for (const auto& path : chosenpaths) { LogInfo(Name(), " lookup ", name, " from ", path->Endpoint()); auto job = new LookupNameJob{this, GenTXID(), name, resultHandler}; @@ -1075,18 +1100,15 @@ namespace llarp bool Endpoint::HandleDataMessage( - path::Path_ptr path, const PathID_t from, std::shared_ptr msg) + path::Path_ptr, const PathID_t from, std::shared_ptr msg) { - msg->sender.UpdateAddr(); PutSenderFor(msg->tag, msg->sender, true); - PutReplyIntroFor(msg->tag, path->intro); - Introduction intro; - intro.pathID = from; - intro.router = PubKey{path->Endpoint()}; - intro.expiresAt = std::min(path->ExpireTime(), msg->introReply.expiresAt); - intro.latency = path->intro.latency; - PutIntroFor(msg->tag, intro); - PutReplyIntroFor(msg->tag, path->intro); + Introduction intro = msg->introReply; + if (HasInboundConvo(msg->sender.Addr())) + { + intro.pathID = from; + } + PutReplyIntroFor(msg->tag, intro); ConvoTagRX(msg->tag); return ProcessDataMessage(msg); } @@ -1429,7 +1451,7 @@ namespace llarp if (not m_IntrosetLookupFilter.Insert(remote)) return true; - const auto paths = GetManyPathsWithUniqueEndpoints(this, NumParallelLookups, remote.ToKey()); + const auto paths = GetManyPathsWithUniqueEndpoints(this, NumParallelLookups); using namespace std::placeholders; const dht::Key_t location = remote.ToKey(); @@ -1453,7 +1475,7 @@ namespace llarp path->Endpoint(), order, GenTXID(), - timeout + (2 * path->intro.latency)); + timeout + (2 * path->intro.latency) + IntrosetLookupGraceInterval); LogInfo( "doing lookup for ", remote, @@ -1791,12 +1813,9 @@ namespace llarp LogTrace("SendToOrQueue: dropping because data.sz == 0"); return false; } - - // inbound conversation - const auto now = Now(); - if (HasInboundConvo(remote)) { + // inbound conversation LogTrace("Have inbound convo"); auto transfer = std::make_shared(); ProtocolFrame& f = transfer->T; @@ -1805,87 +1824,86 @@ namespace llarp if (const auto maybe = GetBestConvoTagFor(remote)) { // the remote guy's intro - Introduction remoteIntro; - Introduction replyPath; + Introduction replyIntro; SharedSecret K; const auto tag = *maybe; - if (!GetCachedSessionKeyFor(tag, K)) + if (not GetCachedSessionKeyFor(tag, K)) { - LogError("no cached key for T=", tag); + LogError(Name(), " no cached key for inbound session from ", remote, " T=", tag); return false; } - if (!GetIntroFor(tag, remoteIntro)) + if (not GetReplyIntroFor(tag, replyIntro)) { - LogError("no intro for T=", tag); + LogError(Name(), "no reply intro for inbound session from ", remote, " T=", tag); return false; } - if (GetReplyIntroFor(tag, replyPath)) + // get path for intro + auto p = GetPathByRouter(replyIntro.router); + + if (not p) { - // get path for intro - ForEachPath([&](path::Path_ptr path) { - if (path->intro == replyPath) - { - p = path; - return; - } - if (p && p->ExpiresSoon(now) && path->IsReady() - && path->intro.router == replyPath.router) - { - p = path; - } - }); + LogWarn( + Name(), + " has no path for intro router ", + RouterID{replyIntro.router}, + " for inbound convo T=", + tag); + return false; } - else - p = GetPathByRouter(remoteIntro.router); - if (p) + f.T = tag; + // TODO: check expiration of our end + auto m = std::make_shared(f.T); + m->PutBuffer(data); + f.N.Randomize(); + f.C.Zero(); + f.R = 0; + transfer->Y.Randomize(); + m->proto = t; + m->introReply = p->intro; + m->sender = m_Identity.pub; + if (auto maybe = GetSeqNoForConvo(f.T)) { - f.T = tag; - // TODO: check expiration of our end - auto m = std::make_shared(f.T); - m->PutBuffer(data); - f.N.Randomize(); - f.C.Zero(); - f.R = 0; - transfer->Y.Randomize(); - m->proto = t; - m->introReply = p->intro; - PutReplyIntroFor(f.T, m->introReply); - m->sender = m_Identity.pub; - if (auto maybe = GetSeqNoForConvo(f.T)) - { - m->seqno = *maybe; - } - else - { - LogWarn(Name(), " no session T=", f.T); - return false; - } - f.S = m->seqno; - f.F = m->introReply.pathID; - transfer->P = remoteIntro.pathID; - auto self = this; - Router()->QueueWork([transfer, p, m, K, self]() { - if (not transfer->T.EncryptAndSign(*m, K, self->m_Identity)) - { - LogError("failed to encrypt and sign"); - return; - } - self->m_SendQueue.tryPushBack(SendEvent_t{transfer, p}); - }); - return true; + m->seqno = *maybe; } else { - LogTrace("SendToOrQueue failed to return via inbound: no path"); + LogWarn(Name(), " could not set sequence number, no session T=", f.T); + return false; } + f.S = m->seqno; + f.F = p->intro.pathID; + transfer->P = replyIntro.pathID; + auto self = this; + Router()->QueueWork([transfer, p, m, K, self]() { + if (not transfer->T.EncryptAndSign(*m, K, self->m_Identity)) + { + LogError("failed to encrypt and sign for sessionn T=", transfer->T.T); + return; + } + self->m_SendQueue.tryPushBack(SendEvent_t{transfer, p}); + }); + return true; } else { - LogWarn("Have inbound convo from ", remote, " but get-best returned none; bug?"); + LogWarn( + Name(), + " SendToOrQueue on inbound convo from ", + remote, + " but get-best returned none; bug?"); } } + if (not WantsOutboundSession(remote)) + { + LogWarn( + Name(), + " SendToOrQueue on outbound session we did not mark as outbound (remote=", + remote, + ")"); + return false; + } // Failed to find a suitable inbound convo, look for outbound LogTrace("Not an inbound convo"); @@ -1900,34 +1918,28 @@ namespace llarp return true; } } - // if we want to make an outbound session - if (WantsOutboundSession(remote)) - { - LogTrace("Making an outbound session and queuing the data"); - // add pending traffic - auto& traffic = m_state->m_PendingTraffic; - traffic[remote].emplace_back(data, t); - EnsurePathToService( - remote, - [self = this](Address addr, OutboundContext* ctx) { - if (ctx) - { - for (auto& pending : self->m_state->m_PendingTraffic[addr]) - { - ctx->AsyncEncryptAndSendTo(pending.Buffer(), pending.protocol); - } - } - else + LogTrace("Making an outbound session and queuing the data"); + // add pending traffic + auto& traffic = m_state->m_PendingTraffic; + traffic[remote].emplace_back(data, t); + EnsurePathToService( + remote, + [self = this](Address addr, OutboundContext* ctx) { + if (ctx) + { + for (auto& pending : self->m_state->m_PendingTraffic[addr]) { - LogWarn("no path made to ", addr); + ctx->AsyncEncryptAndSendTo(pending.Buffer(), pending.protocol); } - self->m_state->m_PendingTraffic.erase(addr); - }, - PathAlignmentTimeout()); - return true; - } - LogDebug("SendOrQueue failed: no inbound/outbound sessions"); - return false; + } + else + { + LogWarn("no path made to ", addr); + } + self->m_state->m_PendingTraffic.erase(addr); + }, + PathAlignmentTimeout()); + return true; } bool diff --git a/llarp/service/hidden_service_address_lookup.hpp b/llarp/service/hidden_service_address_lookup.hpp index 2e656a885..164b365d2 100644 --- a/llarp/service/hidden_service_address_lookup.hpp +++ b/llarp/service/hidden_service_address_lookup.hpp @@ -8,6 +8,9 @@ namespace llarp { namespace service { + /// interval for which we will add to lookup timeout interval + constexpr auto IntrosetLookupGraceInterval = 20s; + struct Endpoint; struct HiddenServiceAddressLookup : public IServiceLookup { diff --git a/llarp/service/outbound_context.cpp b/llarp/service/outbound_context.cpp index 85ebbb869..54ad74602 100644 --- a/llarp/service/outbound_context.cpp +++ b/llarp/service/outbound_context.cpp @@ -49,6 +49,7 @@ namespace llarp ShiftIntroduction(false); UpdateIntroSet(); SwapIntros(); + markedBad = remoteIntro.IsExpired(Now()); } return true; } @@ -151,17 +152,17 @@ namespace llarp return false; if (remoteIntro.router.IsZero()) return false; - return IntroSent(); + return IntroSent() and GetPathByRouter(remoteIntro.router); } void OutboundContext::ShiftIntroRouter(const RouterID r) { const auto now = Now(); - Introduction selectedIntro; + Introduction selectedIntro{}; for (const auto& intro : currentIntroSet.intros) { - if (intro.expiresAt > selectedIntro.expiresAt && intro.router != r) + if (intro.expiresAt > selectedIntro.expiresAt and intro.router != r) { selectedIntro = intro; } @@ -289,7 +290,7 @@ namespace llarp path->Endpoint(), relayOrder, m_Endpoint->GenTXID(), - (IntrosetUpdateInterval / 2) + (2 * path->intro.latency)); + (IntrosetUpdateInterval / 2) + (2 * path->intro.latency) + IntrosetLookupGraceInterval); relayOrder++; if (job->SendRequestViaPath(path, m_Endpoint->Router())) updatingIntroSet = true; @@ -314,11 +315,6 @@ namespace llarp obj["currentRemoteIntroset"] = currentIntroSet.ExtractStatus(); obj["nextIntro"] = m_NextIntro.ExtractStatus(); obj["readyToSend"] = ReadyToSend(); - std::transform( - m_BadIntros.begin(), - m_BadIntros.end(), - std::back_inserter(obj["badIntros"]), - [](const auto& item) -> util::StatusObject { return item.first.ExtractStatus(); }); return obj; } @@ -337,10 +333,14 @@ namespace llarp { SwapIntros(); } - - if ((remoteIntro.router.IsZero() or m_BadIntros.count(remoteIntro)) - and GetPathByRouter(m_NextIntro.router)) - SwapIntros(); + if (ReadyToSend()) + { + // if we dont have a cached session key after sending intro we are in a fugged state so + // expunge + SharedSecret discardme; + if (not m_DataHandler->GetCachedSessionKeyFor(currentConvoTag, discardme)) + return true; + } if (m_GotInboundTraffic and m_LastInboundTraffic + sendTimeout <= now) { @@ -351,22 +351,36 @@ namespace llarp } // check for stale intros // update the introset if we think we need to - if (currentIntroSet.HasStaleIntros(now, path::intro_path_spread)) + if (currentIntroSet.HasStaleIntros(now, path::intro_path_spread) + or remoteIntro.ExpiresSoon(now, path::intro_path_spread)) { UpdateIntroSet(); + ShiftIntroduction(false); + } + + if (ReadyToSend()) + { + if (not remoteIntro.router.IsZero() and not GetPathByRouter(remoteIntro.router)) + { + // pick another good intro if we have no path on our current intro + std::vector otherIntros; + ForEachPath([now, router = remoteIntro.router, &otherIntros](auto path) { + if (path and path->IsReady() and path->Endpoint() != router + and not path->ExpiresSoon(now, path::intro_path_spread)) + { + otherIntros.emplace_back(path->intro); + } + }); + if (not otherIntros.empty()) + { + std::shuffle(otherIntros.begin(), otherIntros.end(), CSRNG{}); + remoteIntro = otherIntros[0]; + } + } } // lookup router in intro if set and unknown if (not m_NextIntro.router.IsZero()) m_Endpoint->EnsureRouterIsKnown(m_NextIntro.router); - // expire bad intros - auto itr = m_BadIntros.begin(); - while (itr != m_BadIntros.end()) - { - if (now > itr->second && now - itr->second > path::default_lifetime) - itr = m_BadIntros.erase(itr); - else - ++itr; - } if (ReadyToSend() and not m_ReadyHooks.empty()) { @@ -386,6 +400,8 @@ namespace llarp { // send a keep alive to keep this session alive KeepAlive(); + if (markedBad) + return true; } // if we are dead return true so we are removed return timeout > 0s ? (now >= timeout && now - timeout > sendTimeout) @@ -433,13 +449,18 @@ namespace llarp return false; size_t numValidPaths = 0; - ForEachPath([now, &numValidPaths](path::Path_ptr path) { + bool havePathToNextIntro = false; + ForEachPath([now, this, &havePathToNextIntro, &numValidPaths](path::Path_ptr path) { if (not path->IsReady()) return; if (not path->intro.ExpiresSoon(now, path::default_lifetime - path::intro_path_spread)) + { numValidPaths++; + if (path->intro.router == m_NextIntro.router) + havePathToNextIntro = true; + } }); - return numValidPaths < numDesiredPaths; + return numValidPaths < numDesiredPaths or not havePathToNextIntro; } void @@ -449,11 +470,8 @@ namespace llarp } void - OutboundContext::MarkIntroBad(const Introduction& intro, llarp_time_t now) - { - // insert bad intro - m_BadIntros[intro] = now; - } + OutboundContext::MarkIntroBad(const Introduction&, llarp_time_t) + {} bool OutboundContext::IntroSent() const @@ -488,13 +506,12 @@ namespace llarp continue; if (m_Endpoint->SnodeBlacklist().count(intro.router)) continue; - if (m_BadIntros.find(intro) == m_BadIntros.end() && remoteIntro.router == intro.router) + if (remoteIntro.router == intro.router) { if (intro.expiresAt > m_NextIntro.expiresAt) { success = true; m_NextIntro = intro; - return true; } } } @@ -508,7 +525,7 @@ namespace llarp m_Endpoint->EnsureRouterIsKnown(intro.router); if (intro.ExpiresSoon(now)) continue; - if (m_BadIntros.find(intro) == m_BadIntros.end() && m_NextIntro != intro) + if (m_NextIntro != intro) { if (intro.expiresAt > m_NextIntro.expiresAt) { diff --git a/llarp/service/outbound_context.hpp b/llarp/service/outbound_context.hpp index 4242e85bc..8e27500e6 100644 --- a/llarp/service/outbound_context.hpp +++ b/llarp/service/outbound_context.hpp @@ -161,7 +161,6 @@ namespace llarp uint64_t m_UpdateIntrosetTX = 0; IntroSet currentIntroSet; Introduction m_NextIntro; - std::unordered_map m_BadIntros; llarp_time_t lastShift = 0s; uint16_t m_LookupFails = 0; uint16_t m_BuildFails = 0; diff --git a/llarp/service/protocol.cpp b/llarp/service/protocol.cpp index 89c6a09aa..8ecf32ac8 100644 --- a/llarp/service/protocol.cpp +++ b/llarp/service/protocol.cpp @@ -367,9 +367,15 @@ namespace llarp AuthResult result) { if (result.code == AuthResultCode::eAuthAccepted) { - handler->PutSenderFor(msg->tag, msg->sender, true); - handler->PutIntroFor(msg->tag, msg->introReply); - handler->PutReplyIntroFor(msg->tag, fromIntro); + if (handler->WantsOutboundSession(msg->sender.Addr())) + { + handler->PutSenderFor(msg->tag, msg->sender, false); + } + else + { + handler->PutSenderFor(msg->tag, msg->sender, true); + } + handler->PutReplyIntroFor(msg->tag, msg->introReply); handler->PutCachedSessionKeyFor(msg->tag, sharedKey); handler->SendAuthResult(path, from, msg->tag, result); LogInfo("auth okay for T=", msg->tag, " from ", msg->sender.Addr());