Merge pull request #1489 from FIWARE/forwarding/loop

Implemented support for the Via HTTP header

CHANGES_NEXT_RELEASE

@@ -17,3 +17,4 @@ Fixed issues:
   * #1418 - Performance - Faster startup when there are GeoProperties in DB
   * #1478 - Fixed issue
   * #1479 - Fixed problem with "operations" in registrations
+  * #280  - Using Via header for loop detection (includes a new field 'hostAlias' in the registrations)

src/app/orionld/orionld.cpp

@@ -247,6 +247,7 @@ int             troePoolSize;
 bool            socketService;
 unsigned short  socketServicePort;
 bool            distributed;
+char            brokerId[136];
 bool            noNotifyFalseUpdate;
 bool            idIndex;
 bool            noswap;
@@ -331,6 +332,7 @@ bool            triggerOperation = false;
 #define SOCKET_SERVICE_DESC    "enable the socket service - accept connections via a normal TCP socket"
 #define SOCKET_SERVICE_PORT_DESC  "port to receive new socket service connections"
 #define DISTRIBUTED_DESC       "turn on distributed operation"
+#define BROKER_ID_DESC         "identity of this broker instance for registrations - for the Via header"
 #define FORWARDING_DESC        "turn on distributed operation (deprecated)"
 #define ID_INDEX_DESC          "automatic mongo index on _id.id"
 #define NOSWAP_DESC            "no swapping - for testing only!!!"
@@ -345,7 +347,7 @@ bool            triggerOperation = false;
 #define DBURI_DESC             "complete URI for database connection"
 #define DEBUG_CURL_DESC        "turn on debugging of libcurl - to the broker's logfile"
 #define CSUBCOUNTERS_DESC      "number of subscription counter updates before flush from sub-cache to DB (0: never, 1: always)"
-#define CORE_CONTEXT_DESC      "Core context version (v1.0|v1.3|v1.4|v1.5|v1.6|v1.7) - v1.6 is default"
+#define CORE_CONTEXT_DESC      "core context version (v1.0|v1.3|v1.4|v1.5|v1.6|v1.7) - v1.6 is default"
 
 
 
@@ -432,12 +434,13 @@ PaArgument paArgs[] =
   { "-troeUser",              troeUser,                 "TROE_USER",                 PaString,  PaOpt,  _i "postgres",   PaNL,   PaNL,             TROE_HOST_USER           },
   { "-troePwd",               troePwd,                  "TROE_PWD",                  PaString,  PaOpt,  _i "password",   PaNL,   PaNL,             TROE_HOST_PWD            },
   { "-troePoolSize",          &troePoolSize,            "TROE_POOL_SIZE",            PaInt,     PaOpt,  10,              0,      1000,             TROE_POOL_DESC           },
-  { "-distributed",           &distributed,             "DISTRIBUTED",               PaBool,    PaOpt,  false,           false,  true,             DISTRIBUTED_DESC         },
   { "-noNotifyFalseUpdate",   &noNotifyFalseUpdate,     "NO_NOTIFY_FALSE_UPDATE",    PaBool,    PaOpt,  false,           false,  true,             NO_NOTIFY_FALSE_UPDATE_DESC  },
   { "-triggerOperation",      &triggerOperation,        "TRIGGER_OPERATION",         PaBool,    PaHid,  false,           false,  true,             TRIGGER_OPERATION_DESC   },
   { "-experimental",          &experimental,            "EXPERIMENTAL",              PaBool,    PaOpt,  false,           false,  true,             EXPERIMENTAL_DESC        },
   { "-mongocOnly",            &mongocOnly,              "MONGOCONLY",                PaBool,    PaOpt,  false,           false,  true,             MONGOCONLY_DESC          },
   { "-cSubCounters",          &cSubCounters,            "CSUB_COUNTERS",             PaInt,     PaOpt,  20,              0,      PaNL,             CSUBCOUNTERS_DESC        },
+  { "-distributed",           &distributed,             "DISTRIBUTED",               PaBool,    PaOpt,  false,           false,  true,             DISTRIBUTED_DESC         },
+  { "-brokerId",              &brokerId,                "BROKER_ID",                 PaStr,     PaOpt,  _i "",           PaNL,   PaNL,             BROKER_ID_DESC           },
   { "-forwarding",            &distributed,             "FORWARDING",                PaBool,    PaHid,  false,           false,  true,             FORWARDING_DESC          },
   { "-socketService",         &socketService,           "SOCKET_SERVICE",            PaBool,    PaHid,  false,           false,  true,             SOCKET_SERVICE_DESC      },
   { "-ssPort",                &socketServicePort,       "SOCKET_SERVICE_PORT",       PaUShort,  PaHid,  1027,            PaNL,   PaNL,             SOCKET_SERVICE_PORT_DESC },
@@ -1189,6 +1192,12 @@ int main(int argC, char* argV[])
   // localIpAndPort - IP:port for X-Forwarded-For
   snprintf(localIpAndPort, sizeof(localIpAndPort), "%s:%d", orionldHostName, port);
 
+  // brokerId - for the Via header
+  if (brokerId[0] == 0)
+    strncpy(brokerId, localIpAndPort, sizeof(brokerId) - 1);
+  else
+    distributed = true;  // Turn on forwarding if the brokerId CLI is used
+
   orionldStateInit(NULL);
 
   // mongocInit calls mongocGeoIndexInit - tenant0 must be ready for that

src/lib/orionld/common/orionldState.h

@@ -240,6 +240,7 @@ typedef struct OrionldStateIn
   char*     host;
   char*     xRealIp;
   char*     xForwardedFor;
+  char*     via;
   char*     connection;
   char*     servicePath;
   char*     xAuthToken;
@@ -600,6 +601,7 @@ extern char              troePwd[256];             // From orionld.cpp
 extern int               troePoolSize;             // From orionld.cpp
 extern char              pgPortString[16];
 extern bool              distributed;              // From orionld.cpp
+extern char              brokerId[136];            // From orionld.cpp
 extern const char*       orionldVersion;
 extern OrionldGeoIndex*  geoIndexList;
 extern OrionldPhase      orionldPhase;

src/lib/orionld/dbModel/dbModelToApiRegistration.cpp

@@ -253,8 +253,11 @@ bool dbModelToApiRegistration(KjNode* dbRegP, bool sysAttrs, bool forCache)
       // Remove "properties" from dbRegP and link the contexts of "properties" to "dbRegP"
       kjChildRemove(dbRegP, propertiesP);
 
-      dbRegP->lastChild->next = propertiesP->value.firstChildP;
-      dbRegP->lastChild       = propertiesP->lastChild;
+      if (propertiesP->value.firstChildP != NULL)
+      {
+        dbRegP->lastChild->next = propertiesP->value.firstChildP;
+        dbRegP->lastChild       = propertiesP->lastChild;
+      }
     }
   }
 

src/lib/orionld/forwarding/CMakeLists.txt

@@ -53,6 +53,8 @@ SET (SOURCES
     regMatchEntityInfoForQuery.cpp
     xForwardedForCompose.cpp
     xForwardedForMatch.cpp
+    viaCompose.cpp
+    viaMatch.cpp
 )
 
 # Include directories

src/lib/orionld/forwarding/distOpRequests.cpp

@@ -39,6 +39,7 @@ extern "C"
 #include "orionld/forwarding/distOpListsMerge.h"                 // distOpListsMerge
 #include "orionld/forwarding/distOpSend.h"                       // distOpSend
 #include "orionld/forwarding/xForwardedForCompose.h"             // xForwardedForCompose
+#include "orionld/forwarding/viaCompose.h"                       // viaCompose
 #include "orionld/forwarding/regMatchForEntityCreation.h"        // regMatchForEntityCreation
 
 
@@ -130,6 +131,7 @@ DistOp* distOpRequests(char* entityId, char* entityType, DistOpType operation, K
 
   // Now that we've found all matching registrations we can add ourselves to the X-forwarded-For header
   char* xff = xForwardedForCompose(orionldState.in.xForwardedFor, localIpAndPort);
+  char* via = viaCompose(orionldState.in.via, brokerId);
 
   // Enqueue all forwarded requests
   int forwards = 0;  // Debugging purposees
@@ -138,7 +140,7 @@ DistOp* distOpRequests(char* entityId, char* entityType, DistOpType operation, K
     // Send the forwarded request and await all responses
     if ((distOpP->regP != NULL) && (distOpP->error == false))
     {
-      if (distOpSend(distOpP, dateHeader, xff, false, NULL) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, via, false, NULL) == 0)
       {
         distOpP->error = false;
         orionldState.distOp.requests += 1;

src/lib/orionld/forwarding/distOpSend.cpp

@@ -329,7 +329,7 @@ void bodyCompact(DistOpType operation, KjNode* requestBody, OrionldContext* fwdC
 //
 // distOpSend -
 //
-bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, bool local, const char* entityIds)
+bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, const char* viaHeader, bool local, const char* entityIds)
 {
   //
   // Figure out the @context to use for the forwarded request
@@ -484,6 +484,9 @@ bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedF
   // X-Forwarded-For
   headers = curl_slist_append(headers, xForwardedForHeader);
 
+  // Via
+  headers = curl_slist_append(headers, viaHeader);
+
   // Custom headers from Registration::contextSourceInfo
   char* infoTenant    = NULL;
   char* accept        = NULL;

src/lib/orionld/forwarding/distOpSend.h

@@ -33,6 +33,6 @@
 //
 // distOpSend -
 //
-extern bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, bool local, const char* entityIds);
+extern bool distOpSend(DistOp* distOpP, const char* dateHeader, const char* xForwardedForHeader, const char* viaHeader, bool local, const char* entityIds);
 
 #endif  // SRC_LIB_ORIONLD_FORWARDING_DISTOPSEND_H_

src/lib/orionld/forwarding/distOpsSend.cpp

@@ -29,6 +29,8 @@
 #include "orionld/forwarding/DistOp.h"                              // DistOp
 #include "orionld/forwarding/distOpSend.h"                          // distOpSend
 #include "orionld/forwarding/xForwardedForCompose.h"                // xForwardedForCompose
+#include "orionld/forwarding/viaCompose.h"                          // viaCompose
+#include "orionld/forwarding/distOpsSend.h"                         // Own interface
 
 
 
@@ -39,6 +41,7 @@
 int distOpsSend(DistOp* distOpList, bool local)
 {
   char* xff = xForwardedForCompose(orionldState.in.xForwardedFor, localIpAndPort);
+  char* via = viaCompose(orionldState.in.via, brokerId);
 
   char dateHeader[70];
   snprintf(dateHeader, sizeof(dateHeader), "Date: %s", orionldState.requestTimeString);  // MOVE to orionldStateInit, for example
@@ -51,7 +54,7 @@ int distOpsSend(DistOp* distOpList, bool local)
     {
       distOpP->onlyIds = true;
 
-      if (distOpSend(distOpP, dateHeader, xff, local, NULL) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, via, local, NULL) == 0)
         distOpP->error = false;
       else
         distOpP->error = true;
@@ -104,6 +107,7 @@ int distOpsSend(DistOp* distOpList, bool local)
 int distOpsSend2(DistOpListItem* distOpList)
 {
   char* xff = xForwardedForCompose(orionldState.in.xForwardedFor, localIpAndPort);
+  char* via = viaCompose(orionldState.in.via, brokerId);
 
   char dateHeader[70];
   snprintf(dateHeader, sizeof(dateHeader), "Date: %s", orionldState.requestTimeString);  // MOVE to orionldStateInit, for example
@@ -118,7 +122,7 @@ int distOpsSend2(DistOpListItem* distOpList)
     {
       distOpP->onlyIds = false;
 
-      if (distOpSend(distOpP, dateHeader, xff, false, doItemP->entityIds) == 0)
+      if (distOpSend(distOpP, dateHeader, xff, via, false, doItemP->entityIds) == 0)
         distOpP->error = false;
       else
         distOpP->error = true;

src/lib/orionld/forwarding/regMatchForBatchDelete.cpp

@@ -36,6 +36,7 @@ extern "C"
 #include "orionld/forwarding/regMatchOperation.h"                // regMatchOperation
 #include "orionld/forwarding/regMatchInformationArrayForGet.h"   // regMatchInformationArrayForGet
 #include "orionld/forwarding/xForwardedForMatch.h"               // xForwardedForMatch
+#include "orionld/forwarding/viaMatch.h"                         // viaMatch
 #include "orionld/forwarding/regMatchForBatchDelete.h"           // Own interface
 
 
@@ -56,17 +57,23 @@ DistOp* regMatchForBatchDelete
 
   for (RegCacheItem* regP = orionldState.tenantP->regCache->regList; regP != NULL; regP = regP->next)
   {
+    if ((regP->mode & regMode) == 0)
+    {
+       LM_T(LmtRegMatch, ("%s: No match due to regMode", regP->regId));
+       continue;
+    }
+
     // Loop detection
-    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
+    if (viaMatch(orionldState.in.via, regP->hostAlias) == true)
     {
-      LM_T(LmtRegMatch, ("%s: No match due to loop detection", regP->regId));
+      LM_T(LmtRegMatch, ("%s: No Reg Match due to Loop (Via)", regP->regId));
       continue;
     }
 
-    if ((regP->mode & regMode) == 0)
+    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
     {
-       LM_T(LmtRegMatch, ("%s: No match due to regMode", regP->regId));
-       continue;
+      LM_T(LmtRegMatch, ("%s: No match due to loop detection", regP->regId));
+      continue;
     }
 
     if (regMatchOperation(regP, operation) == false)

src/lib/orionld/forwarding/regMatchForEntityCreation.cpp

@@ -36,6 +36,7 @@ extern "C"
 #include "orionld/forwarding/regMatchOperation.h"                // regMatchOperation
 #include "orionld/forwarding/regMatchInformationArray.h"         // regMatchInformationArray
 #include "orionld/forwarding/xForwardedForMatch.h"               // xForwardedForMatch
+#include "orionld/forwarding/viaMatch.h"                         // viaMatch
 #include "orionld/forwarding/regMatchForEntityCreation.h"        // Own interface
 
 
@@ -118,16 +119,22 @@ DistOp* regMatchForEntityCreation
 
   for (RegCacheItem* regP = orionldState.tenantP->regCache->regList; regP != NULL; regP = regP->next)
   {
+    if ((regP->mode & regMode) == 0)
+    {
+      // LM_T(LmtRegMatch, ("%s: No Reg Match due to regMode (0x%x vs 0x%x)", regP->regId, regP->mode, regMode));
+      continue;
+    }
+
     // Loop detection
-    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
+    if (viaMatch(orionldState.in.via, regP->hostAlias) == true)
     {
-      LM_T(LmtRegMatch, ("%s: No Reg Match due to loop detection", regP->regId));
+      LM_T(LmtRegMatch, ("%s: No Reg Match due to Loop (Via)", regP->regId));
       continue;
     }
 
-    if ((regP->mode & regMode) == 0)
+    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
     {
-      // LM_T(LmtRegMatch, ("%s: No Reg Match due to regMode (0x%x vs 0x%x)", regP->regId, regP->mode, regMode));
+      LM_T(LmtRegMatch, ("%s: No Reg Match due to loop detection", regP->regId));
       continue;
     }
 

src/lib/orionld/forwarding/regMatchForEntityGet.cpp

@@ -39,6 +39,7 @@ extern "C"
 #include "orionld/forwarding/regMatchOperation.h"                // regMatchOperation
 #include "orionld/forwarding/regMatchInformationArrayForGet.h"   // regMatchInformationArrayForGet
 #include "orionld/forwarding/xForwardedForMatch.h"               // xForwardedForMatch
+#include "orionld/forwarding/viaMatch.h"                         // viaMatch
 #include "orionld/forwarding/regMatchForEntityGet.h"             // Own interface
 
 
@@ -74,18 +75,22 @@ DistOp* regMatchForEntityGet  // FIXME: +entity-type
       regP->regId = (regIdP != NULL)? regIdP->value.s : (char*) "unknown registration";
     }
 
-    LM_T(LmtRegMatch, ("Treating registration '%s' for registrations of mode '%s'", regP->regId, registrationModeToString(regMode)));
+    if ((regP->mode & regMode) == 0)
+    {
+       LM_T(LmtRegMatch, ("%s: No match due to regMode", regP->regId));
+       continue;
+    }
 
     // Loop detection
-    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
+    if (viaMatch(orionldState.in.via, regP->hostAlias) == true)
     {
-      LM_T(LmtRegMatch, ("No Reg Match due to loop detection"));
+      LM_T(LmtRegMatch, ("%s: No Reg Match due to Loop (Via)", regP->regId));
       continue;
     }
 
-    if ((regP->mode & regMode) == 0)
+    if (xForwardedForMatch(orionldState.in.xForwardedFor, regP->ipAndPort) == true)
     {
-      LM_T(LmtRegMatch, ("%s: No Reg Match due to regMode", regP->regId));
+      LM_T(LmtRegMatch, ("No Reg Match due to loop detection"));
       continue;
     }
 

src/lib/orionld/forwarding/viaCompose.cpp

@@ -0,0 +1,64 @@
+/*
+*
+* Copyright 2023 FIWARE Foundation e.V.
+*
+* This file is part of Orion-LD Context Broker.
+*
+* Orion-LD Context Broker is free software: you can redistribute it and/or
+* modify it under the terms of the GNU Affero General Public License as
+* published by the Free Software Foundation, either version 3 of the
+* License, or (at your option) any later version.
+*
+* Orion-LD Context Broker is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Affero General Public License
+* along with Orion-LD Context Broker. If not, see http://www.gnu.org/licenses/.
+*
+* For those usages not covered by this license please contact with
+* orionld at fiware dot org
+*
+* Author: Ken Zangelin
+*/
+#include <stdio.h>                                               // snprintf
+#include <string.h>                                              // strlen
+
+extern "C"
+{
+#include "kalloc/kaAlloc.h"                                      // kaAlloc
+}
+
+#include "logMsg/logMsg.h"                                       // LM_*
+
+#include "orionld/common/orionldState.h"                         // orionldState
+#include "orionld/forwarding/viaCompose.h"                       // Own interface
+
+
+
+// -----------------------------------------------------------------------------
+//
+// viaCompose -
+//
+char* viaCompose(char* via, char* self)
+{
+  int viaLen = 6;  // strlen("Via: ") + '\0'
+
+  if (via != NULL)
+    viaLen += strlen(via);
+  viaLen += 6 + strlen(self);  // 4: ", 1.1 "
+
+  char* viaHeader = kaAlloc(&orionldState.kalloc, viaLen);
+
+  if (viaHeader == NULL)
+    LM_X(1, ("Out of memory (kaAlloc failed to allocate %d bytes", viaLen));
+
+  if (via != NULL)
+    snprintf(viaHeader, viaLen, "Via: %s, 1.1 %s", via, self);  // Hardcoding HTTP version 1.1 ...
+  else
+    snprintf(viaHeader, viaLen, "Via: 1.1 %s", self);
+
+  LM_T(LmtHeaders, ("via: '%s'", viaHeader));
+  return viaHeader;
+}