Merge pull request #1669 from majestrate/inbound-convo-wrong-path-fix-2021-06-14

[bugfix] make inbound convotags reply on the correct path

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)
 

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:

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);
+          if (SendToOrQueue(*maybe, pkt.ConstBuffer(), type))
-          return;
+          {
+            MarkIPActive(dst);
+            return;
+          }
         }
         // try establishing a path to this guy
         // will fail if it's an inbound convo

llarp/link/server.cpp

@@ -481,10 +481,9 @@ namespace llarp
   bool
   ILinkLayer::PutSession(const std::shared_ptr<ILinkSession>& 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;

llarp/link/server.hpp

@@ -245,7 +245,7 @@ namespace llarp
     SecretKey m_SecretKey;
 
     using AuthedLinks = std::unordered_multimap<RouterID, std::shared_ptr<ILinkSession>>;
-    using Pending = std::unordered_multimap<SockAddr, std::shared_ptr<ILinkSession>>;
+    using Pending = std::unordered_map<SockAddr, std::shared_ptr<ILinkSession>>;
     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));

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<path::Path_ptr> 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<ProtocolMessage> msg)
+        path::Path_ptr, const PathID_t from, std::shared_ptr<ProtocolMessage> msg)
     {
-      msg->sender.UpdateAddr();
       PutSenderFor(msg->tag, msg->sender, true);
-      PutReplyIntroFor(msg->tag, path->intro);
+      Introduction intro = msg->introReply;
-      Introduction intro;
+      if (HasInboundConvo(msg->sender.Addr()))
-      intro.pathID = from;
+      {
-      intro.router = PubKey{path->Endpoint()};
+        intro.pathID = from;
-      intro.expiresAt = std::min(path->ExpireTime(), msg->introReply.expiresAt);
+      }
-      intro.latency = path->intro.latency;
+      PutReplyIntroFor(msg->tag, intro);
-      PutIntroFor(msg->tag, intro);
-      PutReplyIntroFor(msg->tag, path->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<routing::PathTransferMessage>();
         ProtocolFrame& f = transfer->T;
@@ -1805,87 +1824,86 @@ namespace llarp
         if (const auto maybe = GetBestConvoTagFor(remote))
         {
           // the remote guy's intro
-          Introduction remoteIntro;
+          Introduction replyIntro;
-          Introduction replyPath;
           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
+            LogWarn(
-            ForEachPath([&](path::Path_ptr path) {
+                Name(),
-              if (path->intro == replyPath)
+                " has no path for intro router ",
-              {
+                RouterID{replyIntro.router},
-                p = path;
+                " for inbound convo T=",
-                return;
+                tag);
-              }
+            return false;
-              if (p && p->ExpiresSoon(now) && path->IsReady()
-                  && path->intro.router == replyPath.router)
-              {
-                p = path;
-              }
-            });
           }
-          else
-            p = GetPathByRouter(remoteIntro.router);
 
-          if (p)
+          f.T = tag;
+          // TODO: check expiration of our end
+          auto m = std::make_shared<ProtocolMessage>(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;
+            m->seqno = *maybe;
-            // TODO: check expiration of our end
-            auto m = std::make_shared<ProtocolMessage>(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;
           }
           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
+      LogTrace("Making an outbound session and queuing the data");
-      if (WantsOutboundSession(remote))
+      // add pending traffic
-      {
+      auto& traffic = m_state->m_PendingTraffic;
-        LogTrace("Making an outbound session and queuing the data");
+      traffic[remote].emplace_back(data, t);
-        // add pending traffic
+      EnsurePathToService(
-        auto& traffic = m_state->m_PendingTraffic;
+          remote,
-        traffic[remote].emplace_back(data, t);
+          [self = this](Address addr, OutboundContext* ctx) {
-        EnsurePathToService(
+            if (ctx)
-            remote,
+            {
-            [self = this](Address addr, OutboundContext* ctx) {
+              for (auto& pending : self->m_state->m_PendingTraffic[addr])
-              if (ctx)
-              {
-                for (auto& pending : self->m_state->m_PendingTraffic[addr])
-                {
-                  ctx->AsyncEncryptAndSendTo(pending.Buffer(), pending.protocol);
-                }
-              }
-              else
               {
-                LogWarn("no path made to ", addr);
+                ctx->AsyncEncryptAndSendTo(pending.Buffer(), pending.protocol);
               }
-              self->m_state->m_PendingTraffic.erase(addr);
+            }
-            },
+            else
-            PathAlignmentTimeout());
+            {
-        return true;
+              LogWarn("no path made to ", addr);
-      }
+            }
-      LogDebug("SendOrQueue failed: no inbound/outbound sessions");
+            self->m_state->m_PendingTraffic.erase(addr);
-      return false;
+          },
+          PathAlignmentTimeout());
+      return true;
     }
 
     bool

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
     {

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 (ReadyToSend())
-      if ((remoteIntro.router.IsZero() or m_BadIntros.count(remoteIntro))
+      {
-          and GetPathByRouter(m_NextIntro.router))
+        // if we dont have a cached session key after sending intro we are in a fugged state so
-        SwapIntros();
+        // 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<Introduction> 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)
+    OutboundContext::MarkIntroBad(const Introduction&, llarp_time_t)
-    {
+    {}
-      // insert bad intro
-      m_BadIntros[intro] = now;
-    }
 
     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)
             {

llarp/service/outbound_context.hpp

@@ -161,7 +161,6 @@ namespace llarp
       uint64_t m_UpdateIntrosetTX = 0;
       IntroSet currentIntroSet;
       Introduction m_NextIntro;
-      std::unordered_map<Introduction, llarp_time_t> m_BadIntros;
       llarp_time_t lastShift = 0s;
       uint16_t m_LookupFails = 0;
       uint16_t m_BuildFails = 0;

llarp/service/protocol.cpp

@@ -367,9 +367,15 @@ namespace llarp
                 AuthResult result) {
               if (result.code == AuthResultCode::eAuthAccepted)
               {
-                handler->PutSenderFor(msg->tag, msg->sender, true);
+                if (handler->WantsOutboundSession(msg->sender.Addr()))
-                handler->PutIntroFor(msg->tag, msg->introReply);
+                {
-                handler->PutReplyIntroFor(msg->tag, fromIntro);
+                  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());