Merge pull request #746 from IABTechLab/ian-UID2-3548-unrecognized-pa…

…ths-fix

unrecognized paths OOM fix

src/main/java/com/uid2/operator/Const.java

@@ -25,5 +25,6 @@ public class Config extends com.uid2.shared.Const.Config {
         public static final String GcpSecretVersionNameProp = "gcp_secret_version_name";
         public static final String OptOutStatusApiEnabled = "optout_status_api_enabled";
         public static final String OptOutStatusMaxRequestSize = "optout_status_max_request_size";
+        public static final String MaxInvalidPaths = "logging_limit_max_invalid_paths_per_interval";
     }
 }

src/main/java/com/uid2/operator/Main.java

@@ -9,6 +9,7 @@
 import com.uid2.operator.monitoring.StatsCollectorVerticle;
 import com.uid2.operator.service.SecureLinkValidatorService;
 import com.uid2.operator.service.ShutdownService;
+import com.uid2.operator.vertx.Endpoints;
 import com.uid2.operator.vertx.OperatorShutdownHandler;
 import com.uid2.operator.store.CloudSyncOptOutStore;
 import com.uid2.operator.store.OptOutCloudStorage;
@@ -363,7 +364,7 @@ private Future<String> createAndDeployCloudSyncStoreVerticle(String name, ICloud
 
     private Future<String> createAndDeployStatsCollector() {
         Promise<String> promise = Promise.promise();
-        StatsCollectorVerticle statsCollectorVerticle = new StatsCollectorVerticle(60000);
+        StatsCollectorVerticle statsCollectorVerticle = new StatsCollectorVerticle(60000, config.getInteger(Const.Config.MaxInvalidPaths, 50));
         vertx.deployVerticle(statsCollectorVerticle, promise);
         _statsCollectorQueue = statsCollectorVerticle;
         return promise.future();
@@ -425,7 +426,8 @@ private static void setupMetrics(MicrometerMetricsOptions metricOptions) {
                 .meterFilter(new PrometheusRenameFilter())
                 .meterFilter(MeterFilter.replaceTagValues(Label.HTTP_PATH.toString(), actualPath -> {
                     try {
-                        return HttpUtils.normalizePath(actualPath).split("\\?")[0];
+                        String normalized = HttpUtils.normalizePath(actualPath).split("\\?")[0];
+                        return Endpoints.pathSet().contains(normalized) ? normalized : "/unknown";
                     } catch (IllegalArgumentException e) {
                         return actualPath;
                     }

src/main/java/com/uid2/operator/monitoring/StatsCollectorVerticle.java

@@ -3,6 +3,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.uid2.operator.Const;
 import com.uid2.operator.model.StatsCollectorMessageItem;
+import com.uid2.operator.vertx.Endpoints;
 import io.micrometer.core.instrument.Counter;
 import io.micrometer.core.instrument.Metrics;
 import io.vertx.core.AbstractVerticle;
@@ -24,6 +25,7 @@ public class StatsCollectorVerticle extends AbstractVerticle implements IStatsCo
     private HashMap<String, EndpointStat> pathMap;
 
     private static final int MAX_AVAILABLE = 1000;
+    private final int maxInvalidPaths;
 
     private final Duration jsonProcessingInterval;
     private Instant lastJsonProcessTime;
@@ -39,13 +41,14 @@ public class StatsCollectorVerticle extends AbstractVerticle implements IStatsCo
     private final ObjectMapper mapper;
     private final Counter queueFullCounter;
 
-    public StatsCollectorVerticle(long jsonIntervalMS) {
+    public StatsCollectorVerticle(long jsonIntervalMS, int maxInvalidPaths) {
         pathMap = new HashMap<>();
 
         _statsCollectorCount = new AtomicInteger();
         _runningSerializer = false;
 
         jsonProcessingInterval = Duration.ofMillis(jsonIntervalMS);
+        this.maxInvalidPaths = maxInvalidPaths;
 
         logCycleSkipperCounter = Counter
                 .builder("uid2.api_usage_log_cycle_skipped")
@@ -113,7 +116,11 @@ public void handleMessage(Message message) {
 
         EndpointStat endpointStat = new EndpointStat(endpoint, siteId, apiVersion, domain);
 
-        pathMap.merge(path, endpointStat, this::mergeEndpoint);
+        Set<String> validPaths = Endpoints.pathSet();
+        if(validPaths.contains(path) || pathMap.containsKey(path) || (pathMap.size() < this.maxInvalidPaths + validPaths.size() && messageItem.getApiContact() != null)) {
+            pathMap.merge(path, endpointStat, this::mergeEndpoint);
+        }
+
 
         _statsCollectorCount.decrementAndGet();
 
@@ -123,6 +130,9 @@ public void handleMessage(Message message) {
                 logCycleSkipperCounter.increment();
             } else {
                 _runningSerializer = true;
+                if(pathMap.size() == this.maxInvalidPaths + validPaths.size()) {
+                    LOGGER.error("max invalid paths reached; a large number of invalid paths have been requested from authenticated participants");
+                }
                 Object[] stats = pathMap.values().toArray();
                 this.jsonSerializerExecutor.<Void>executeBlocking(
                         promise -> promise.complete(this.serializeToLogs(stats)),

src/main/java/com/uid2/operator/vertx/Endpoints.java

@@ -0,0 +1,59 @@
+package com.uid2.operator.vertx;
+
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public enum Endpoints {
+    OPS_HEALTHCHECK("/ops/healthcheck"),
+
+    V0_KEY_LATEST("/key/latest"),
+    V0_TOKEN_GENERATE("/token/generate"),
+    V0_TOKEN_REFRESH("/token/refresh"),
+    V0_TOKEN_VALIDATE("/token/validate"),
+    V0_IDENTITY_MAP("/identity/map"),
+    V0_TOKEN_LOGOUT("/token/logout"),
+
+    V1_TOKEN_GENERATE("/v1/token/generate"),
+    V1_TOKEN_VALIDATE("/v1/token/validate"),
+    V1_TOKEN_REFRESH("/v1/token/refresh"),
+    V1_IDENTITY_BUCKETS("/v1/identity/buckets"),
+    V1_IDENTITY_MAP("/v1/identity/map"),
+    V1_KEY_LATEST("/v1/key/latest"),
+
+    V2_TOKEN_GENERATE("/v2/token/generate"),
+    V2_TOKEN_REFRESH("/v2/token/refresh"),
+    V2_TOKEN_VALIDATE("/v2/token/validate"),
+    V2_IDENTITY_BUCKETS("/v2/identity/buckets"),
+    V2_IDENTITY_MAP("/v2/identity/map"),
+    V2_KEY_LATEST("/v2/key/latest"),
+    V2_KEY_SHARING("/v2/key/sharing"),
+    V2_KEY_BIDSTREAM("/v2/key/bidstream"),
+    V2_TOKEN_LOGOUT("/v2/token/logout"),
+    V2_OPTOUT_STATUS("/v2/optout/status"),
+    V2_TOKEN_CLIENTGENERATE("/v2/token/client-generate"),
+
+    EUID_SDK_1_0_0("/static/js/euid-sdk-1.0.0.js"),
+    OPENID_SDK_1_0("/static/js/openid-sdk-1.0.js"),
+    UID2_ESP_0_0_1A("/static/js/uid2-esp-0.0.1a.js"),
+    UID2_SDK_0_0_1A("/static/js/uid2-sdk-0.0.1a.js"),
+    UID2_SDK_0_0_1A_SOURCE("/static/js/uid2-sdk-0.0.1a-source.ts"),
+    UID2_SDK_0_0_1B("/static/js/uid2-sdk-0.0.1b.js"),
+    UID2_SDK_1_0_0("/static/js/uid2-sdk-1.0.0.js"),
+    UID2_SDK_2_0_0("/static/js/uid2-sdk-2.0.0.js")
+    ;
+    private final String path;
+
+    Endpoints(final String path) {
+        this.path = path;
+    }
+
+    public static Set<String> pathSet() {
+        return Stream.of(Endpoints.values()).map(Endpoints::toString).collect(Collectors.toSet());
+    }
+
+    @Override
+    public String toString() {
+        return path;
+    }
+}

src/main/java/com/uid2/operator/vertx/UIDOperatorVerticle.java

@@ -45,6 +45,7 @@
 import io.vertx.core.json.JsonArray;
 import io.vertx.core.json.JsonObject;
 import io.vertx.ext.web.AllowForwardHeaders;
+import io.vertx.ext.web.Route;
 import io.vertx.ext.web.Router;
 import io.vertx.ext.web.RoutingContext;
 import io.vertx.ext.web.handler.BodyHandler;
@@ -69,6 +70,7 @@
 
 import static com.uid2.operator.IdentityConst.*;
 import static com.uid2.operator.service.ResponseUtil.*;
+import static com.uid2.operator.vertx.Endpoints.*;
 
 public class UIDOperatorVerticle extends AbstractVerticle {
     private static final Logger LOGGER = LoggerFactory.getLogger(UIDOperatorVerticle.class);
@@ -236,28 +238,28 @@ private Router createRoutesSetup() throws IOException {
         setupV2Routes(router, bodyHandler);
 
         // Static and health check
-        router.get("/ops/healthcheck").handler(this::handleHealthCheck);
+        router.get(OPS_HEALTHCHECK.toString()).handler(this::handleHealthCheck);
 
         if (this.config.getBoolean(Const.Config.AllowLegacyAPIProp, true)) {
             // V1 APIs
-            router.get("/v1/token/generate").handler(auth.handleV1(this::handleTokenGenerateV1, Role.GENERATOR));
-            router.get("/v1/token/validate").handler(this::handleTokenValidateV1);
-            router.get("/v1/token/refresh").handler(auth.handleWithOptionalAuth(this::handleTokenRefreshV1));
-            router.get("/v1/identity/buckets").handler(auth.handle(this::handleBucketsV1, Role.MAPPER));
-            router.get("/v1/identity/map").handler(auth.handle(this::handleIdentityMapV1, Role.MAPPER));
-            router.post("/v1/identity/map").handler(bodyHandler).handler(auth.handle(this::handleIdentityMapBatchV1, Role.MAPPER));
-            router.get("/v1/key/latest").handler(auth.handle(this::handleKeysRequestV1, Role.ID_READER));
+            router.get(V1_TOKEN_GENERATE.toString()).handler(auth.handleV1(this::handleTokenGenerateV1, Role.GENERATOR));
+            router.get(V1_TOKEN_VALIDATE.toString()).handler(this::handleTokenValidateV1);
+            router.get(V1_TOKEN_REFRESH.toString()).handler(auth.handleWithOptionalAuth(this::handleTokenRefreshV1));
+            router.get(V1_IDENTITY_BUCKETS.toString()).handler(auth.handle(this::handleBucketsV1, Role.MAPPER));
+            router.get(V1_IDENTITY_MAP.toString()).handler(auth.handle(this::handleIdentityMapV1, Role.MAPPER));
+            router.post(V1_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handle(this::handleIdentityMapBatchV1, Role.MAPPER));
+            router.get(V1_KEY_LATEST.toString()).handler(auth.handle(this::handleKeysRequestV1, Role.ID_READER));
 
             // Deprecated APIs
-            router.get("/key/latest").handler(auth.handle(this::handleKeysRequest, Role.ID_READER));
-            router.get("/token/generate").handler(auth.handle(this::handleTokenGenerate, Role.GENERATOR));
-            router.get("/token/refresh").handler(this::handleTokenRefresh);
-            router.get("/token/validate").handler(this::handleValidate);
-            router.get("/identity/map").handler(auth.handle(this::handleIdentityMap, Role.MAPPER));
-            router.post("/identity/map").handler(bodyHandler).handler(auth.handle(this::handleIdentityMapBatch, Role.MAPPER));
+            router.get(V0_KEY_LATEST.toString()).handler(auth.handle(this::handleKeysRequest, Role.ID_READER));
+            router.get(V0_TOKEN_GENERATE.toString()).handler(auth.handle(this::handleTokenGenerate, Role.GENERATOR));
+            router.get(V0_TOKEN_REFRESH.toString()).handler(this::handleTokenRefresh);
+            router.get(V0_TOKEN_VALIDATE.toString()).handler(this::handleValidate);
+            router.get(V0_IDENTITY_MAP.toString()).handler(auth.handle(this::handleIdentityMap, Role.MAPPER));
+            router.post(V0_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handle(this::handleIdentityMapBatch, Role.MAPPER));
 
             // Internal service APIs
-            router.get("/token/logout").handler(auth.handle(this::handleLogoutAsync, Role.OPTOUT));
+            router.get(V0_TOKEN_LOGOUT.toString()).handler(auth.handle(this::handleLogoutAsync, Role.OPTOUT));
 
             // only uncomment to do local testing
             //router.get("/internal/optout/get").handler(auth.loopbackOnly(this::handleOptOutGet));
@@ -268,36 +270,34 @@ private Router createRoutesSetup() throws IOException {
     }
 
     private void setupV2Routes(Router mainRouter, BodyHandler bodyHandler) {
-        final Router v2Router = Router.router(vertx);
 
-        v2Router.post("/token/generate").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_TOKEN_GENERATE.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handleTokenGenerate(rc, this::handleTokenGenerateV2), Role.GENERATOR));
-        v2Router.post("/token/refresh").handler(bodyHandler).handler(auth.handleWithOptionalAuth(
+        mainRouter.post(V2_TOKEN_REFRESH.toString()).handler(bodyHandler).handler(auth.handleWithOptionalAuth(
                 rc -> v2PayloadHandler.handleTokenRefresh(rc, this::handleTokenRefreshV2)));
-        v2Router.post("/token/validate").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_TOKEN_VALIDATE.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleTokenValidateV2), Role.GENERATOR));
-        v2Router.post("/identity/buckets").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_IDENTITY_BUCKETS.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleBucketsV2), Role.MAPPER));
-        v2Router.post("/identity/map").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_IDENTITY_MAP.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleIdentityMapV2), Role.MAPPER));
-        v2Router.post("/key/latest").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_KEY_LATEST.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleKeysRequestV2), Role.ID_READER));
-        v2Router.post("/key/sharing").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_KEY_SHARING.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleKeysSharing), Role.SHARER, Role.ID_READER));
-        v2Router.post("/key/bidstream").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_KEY_BIDSTREAM.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handle(rc, this::handleKeysBidstream), Role.ID_READER));
-        v2Router.post("/token/logout").handler(bodyHandler).handler(auth.handleV1(
+        mainRouter.post(V2_TOKEN_LOGOUT.toString()).handler(bodyHandler).handler(auth.handleV1(
                 rc -> v2PayloadHandler.handleAsync(rc, this::handleLogoutAsyncV2), Role.OPTOUT));
         if (this.optOutStatusApiEnabled) {
-            v2Router.post("/optout/status").handler(bodyHandler).handler(auth.handleV1(
+            mainRouter.post(V2_OPTOUT_STATUS.toString()).handler(bodyHandler).handler(auth.handleV1(
                     rc -> v2PayloadHandler.handle(rc, this::handleOptoutStatus),
                     Role.MAPPER, Role.SHARER, Role.ID_READER));
         }
 
         if (this.clientSideTokenGenerate)
-            v2Router.post("/token/client-generate").handler(bodyHandler).handler(this::handleClientSideTokenGenerate);
+            mainRouter.post(V2_TOKEN_CLIENTGENERATE.toString()).handler(bodyHandler).handler(this::handleClientSideTokenGenerate);
 
-        mainRouter.route("/v2/*").subRouter(v2Router);
     }
 
 

src/test/java/com/uid2/operator/StatsCollectorVerticleTest.java

@@ -1,26 +1,33 @@
 package com.uid2.operator;
 
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.spi.ILoggingEvent;
+import ch.qos.logback.core.read.ListAppender;
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.uid2.operator.model.StatsCollectorMessageItem;
 import com.uid2.operator.monitoring.StatsCollectorVerticle;
+import com.uid2.operator.vertx.Endpoints;
 import io.vertx.core.Vertx;
 import io.vertx.junit5.VertxExtension;
 import io.vertx.junit5.VertxTestContext;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
+import org.slf4j.LoggerFactory;
 
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 @ExtendWith(VertxExtension.class)
 public class StatsCollectorVerticleTest {
+    private static final int MAX_INVALID_PATHS = 5;
     private StatsCollectorVerticle verticle;
 
     @BeforeEach
-    void deployVerticle(Vertx vertx, VertxTestContext testContext) throws Throwable {
-        verticle = new StatsCollectorVerticle(1000);
+    void deployVerticle(Vertx vertx, VertxTestContext testContext) {
+        verticle = new StatsCollectorVerticle(1000, MAX_INVALID_PATHS);
         vertx.deployVerticle(verticle, testContext.succeeding(id -> testContext.completeNow()));
     }
 
@@ -83,4 +90,69 @@ void testJSONSerializeWithV2AndUnknownPaths(Vertx vertx, VertxTestContext testCo
 
         testContext.completeNow();
     }
+
+    @Test
+    void allValidPathsAllowed(Vertx vertx, VertxTestContext testContext) throws InterruptedException, JsonProcessingException {
+        ObjectMapper mapper = new ObjectMapper();
+        Set<String> validEndpoints = Endpoints.pathSet();
+
+        for(String endpoint : validEndpoints) {
+            StatsCollectorMessageItem messageItem = new StatsCollectorMessageItem(endpoint, "https://test.com", "test", 1);
+            vertx.eventBus().send(Const.Config.StatsCollectorEventBus, mapper.writeValueAsString(messageItem));
+        }
+
+        testContext.awaitCompletion(2000, TimeUnit.MILLISECONDS);
+
+        String results = verticle.getEndpointStats();
+
+        for(String endpoint: validEndpoints) {
+            String withoutVersion = endpoint;
+            if (endpoint.startsWith("/v1/") || endpoint.startsWith("/v2/")) {
+                withoutVersion = endpoint.substring(4);
+            } else if (endpoint.startsWith("/")) {
+                withoutVersion = endpoint.substring(1);
+            }
+
+            String expected = "{\"endpoint\":\"" + withoutVersion + "\",\"siteId\":1,";
+            Assertions.assertTrue(results.contains(expected));
+        }
+
+        testContext.completeNow();
+    }
+
+    @Test
+    void invalidPathsLimit(Vertx vertx, VertxTestContext testContext) throws InterruptedException, JsonProcessingException {
+        ObjectMapper mapper = new ObjectMapper();
+
+        for(int i = 0; i < MAX_INVALID_PATHS + Endpoints.pathSet().size() + 5; i++) {
+            StatsCollectorMessageItem messageItem = new StatsCollectorMessageItem("/bad" + i, "https://test.com", "test", 1);
+            vertx.eventBus().send(Const.Config.StatsCollectorEventBus, mapper.writeValueAsString(messageItem));
+        }
+
+        testContext.awaitCompletion(2000, TimeUnit.MILLISECONDS);
+        String results = verticle.getEndpointStats();
+
+        // MAX_INVALID_PATHS is not the hard limit. The maximum paths that can be recorded, including valid ones, is MAX_INVALID_PATHS + validPaths.size * 2
+        for(int i = 0; i < MAX_INVALID_PATHS + Endpoints.pathSet().size(); i++) {
+            String expected = "{\"endpoint\":\"bad" + i + "\",\"siteId\":1,\"apiVersion\":\"v0\",\"domainList\":[{\"domain\":\"test.com\",\"count\":1,\"apiContact\":\"test\"}]}";
+            Assertions.assertTrue(results.contains(expected));
+        }
+        for(int i = MAX_INVALID_PATHS + Endpoints.pathSet().size(); i < MAX_INVALID_PATHS + 5; i++) {
+            String expected = "{\"endpoint\":\"bad" + i + "\",\"siteId\":1,\"apiVersion\":\"v0\",\"domainList\":[{\"domain\":\"test.com\",\"count\":1,\"apiContact\":\"test\"}]}";
+            Assertions.assertFalse(results.contains(expected));
+        }
+
+        ListAppender<ILoggingEvent> logWatcher = new ListAppender<>();
+        logWatcher.start();
+        ((Logger) LoggerFactory.getLogger(StatsCollectorVerticle.class)).addAppender(logWatcher);
+
+        StatsCollectorMessageItem messageItem = new StatsCollectorMessageItem("/triggerSerialize", "https://test.com", "test", 1);
+        vertx.eventBus().send(Const.Config.StatsCollectorEventBus, mapper.writeValueAsString(messageItem));
+
+        testContext.awaitCompletion(1000, TimeUnit.MILLISECONDS);
+        Assertions.assertTrue(logWatcher.list.get(0).getFormattedMessage().contains("max invalid paths reached; a large number of invalid paths have been requested from authenticated participants"));
+
+        testContext.completeNow();
+    }
+
 }