Cherry-pick proxy-16.0 release commits

proxy/pom.xml

@@ -4,7 +4,7 @@
 
   <groupId>com.wavefront</groupId>
   <artifactId>proxy</artifactId>
-  <version>15.1</version>
+  <version>16.0</version>
 
   <name>Wavefront Proxy</name>
   <description>Service for batching and relaying metric traffic to Wavefront</description>

proxy/src/main/java/com/wavefront/agent/ProxyConfigDef.java

@@ -1577,7 +1577,8 @@ public abstract class ProxyConfigDef extends Configuration {
 
   @Parameter(
           names = {"--metricQuerySamplingDryRun"},
-          description = "Enable query-aware sampling dry run mode. Bloom filters and sampling decisions are evaluated but points are not dropped.")
+          arity = 1,
+          description = "Enable query-aware sampling dry run mode. Bloom filters and sampling decisions are evaluated but points are not dropped. (Default: false)")
   @ProxyConfigOption(category = Categories.INPUT, subCategory = SubCategories.METRICS)
   boolean metricQuerySamplingDryRun = false;
 }

proxy/src/main/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticator.java

@@ -64,7 +64,8 @@ class HttpGetTokenIntrospectionAuthenticator extends TokenIntrospectionAuthentic
 
   @Override
   boolean callAuthService(@Nonnull String token) throws Exception {
-    HttpGet request = new HttpGet(tokenIntrospectionServiceUrl.replace("{{token}}", token));
+    HttpGet request =
+        new HttpGet(tokenIntrospectionServiceUrl.replace("{{token}}", urlEncodeToken(token)));
     if (tokenIntrospectionServiceAuthorizationHeader != null) {
       request.setHeader("Authorization", tokenIntrospectionServiceAuthorizationHeader);
     }

proxy/src/main/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticator.java

@@ -72,7 +72,8 @@ class Oauth2TokenIntrospectionAuthenticator extends TokenIntrospectionAuthentica
   @Override
   boolean callAuthService(@Nonnull String token) throws Exception {
     boolean result;
-    HttpPost request = new HttpPost(tokenIntrospectionServiceUrl.replace("{{token}}", token));
+    HttpPost request =
+        new HttpPost(tokenIntrospectionServiceUrl.replace("{{token}}", urlEncodeToken(token)));
     request.setHeader("Content-Type", "application/x-www-form-urlencoded");
     request.setHeader("Accept", "application/json");
     if (tokenIntrospectionAuthorizationHeader != null) {

proxy/src/main/java/com/wavefront/agent/auth/TokenIntrospectionAuthenticator.java

@@ -6,6 +6,8 @@
 import com.yammer.metrics.Metrics;
 import com.yammer.metrics.core.Counter;
 import com.yammer.metrics.core.MetricName;
+import java.io.UnsupportedEncodingException;
+import java.net.URLEncoder;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import java.util.logging.Level;
@@ -86,6 +88,24 @@ public Boolean reload(@Nonnull String key, @Nonnull Boolean oldValue) {
 
   abstract boolean callAuthService(@Nonnull String token) throws Exception;
 
+  /**
+   * Percent-encodes a token for safe substitution into a URL template (e.g. replacing a {@code
+   * {{token}}} placeholder). Tokens are taken verbatim from inbound requests and may contain
+   * characters that are meaningful in a URL (such as {@code / ? # & = %}); encoding them as an
+   * opaque value prevents a malicious token from altering the target host, path, or query string
+   * of the introspection request.
+   */
+  static String urlEncodeToken(@Nonnull String token) {
+    try {
+      // URLEncoder escapes everything except [A-Za-z0-9.\-_*], encoding space as '+'; normalize
+      // '+' to '%20' since we're encoding a URL path/query component, not a form field.
+      return URLEncoder.encode(token, "UTF-8").replace("+", "%20");
+    } catch (UnsupportedEncodingException e) {
+      // UTF-8 is guaranteed to be supported by every JVM.
+      throw new AssertionError(e);
+    }
+  }
+
   @Override
   public boolean authorize(@Nullable String token) {
     if (token == null) {

proxy/src/main/java/com/wavefront/agent/listeners/DataDogPortUnificationHandler.java

@@ -225,7 +225,12 @@ protected void handleHttpMessage(final ChannelHandlerContext ctx, final FullHttp
               new TaggedMetricName("listeners", "http-relay.duration-nanos", "port", handle));
       long startNanos = System.nanoTime();
       try {
-        String outgoingUrl = requestRelayTarget.replaceFirst("/*$", "") + request.uri();
+        String path = uri.getRawPath();
+        if (path == null || !path.startsWith("/")) {
+          throw new URISyntaxException(request.uri(), "Relative path must start with /");
+        }
+        String outgoingUrl = requestRelayTarget.replaceFirst("/*$", "") + path +
+            (uri.getRawQuery() == null ? "" : "?" + uri.getRawQuery());
         HttpPost outgoingRequest = new HttpPost(outgoingUrl);
 
         request

proxy/src/test/java/com/wavefront/agent/HttpEndToEndTest.java

@@ -10,7 +10,6 @@
 import static com.wavefront.api.agent.Constants.PUSH_FORMAT_LOGS_JSON_ARR;
 import static com.wavefront.api.agent.Constants.PUSH_FORMAT_LOGS_JSON_LINES;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
@@ -34,10 +33,14 @@
 import io.netty.util.CharsetUtil;
 import java.io.File;
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
@@ -145,8 +148,13 @@ public void testEndToEndMetrics() throws Exception {
     HandlerKey key = HandlerKey.of(ReportableEntityType.POINT, String.valueOf(proxyPort));
     ((SenderTaskFactoryImpl) proxy.senderTaskFactory).flushNow(key);
     assertEquals(1, successfulSteps.getAndSet(0));
-    AtomicBoolean part1 = new AtomicBoolean(false);
-    AtomicBoolean part2 = new AtomicBoolean(false);
+    // Once the batch is rejected as too large (case 4), the proxy splits it into smaller
+    // sub-batches and retries each independently. The number/size of those sub-batches depends
+    // on the runtime dataPerBatch setting, which can be recalculated in the background while this
+    // test runs - so rather than assuming a fixed split count, collect every line delivered after
+    // the split and check the *set* of lines eventually matches the original payload exactly once
+    // each, regardless of how many pieces it took.
+    List<String> deliveredSplitLines = new CopyOnWriteArrayList<>();
     server.update(
         req -> {
           String content = req.content().toString(CharsetUtil.UTF_8);
@@ -168,24 +176,30 @@ public void testEndToEndMetrics() throws Exception {
               assertEquals(expectedTest1part1 + "\n" + expectedTest1part2, content);
               successfulSteps.incrementAndGet();
               return makeResponse(HttpResponseStatus.REQUEST_ENTITY_TOO_LARGE, "");
-            case 5:
-            case 6:
-              if (content.equals(expectedTest1part1)) part1.set(true);
-              if (content.equals(expectedTest1part2)) part2.set(true);
+            default:
+              deliveredSplitLines.addAll(Arrays.asList(content.split("\n")));
               successfulSteps.incrementAndGet();
               return makeResponse(HttpResponseStatus.OK, "");
           }
-          throw new IllegalStateException();
         });
     gzippedHttpPost("http://localhost:" + proxyPort + "/", payload);
     ((SenderTaskFactoryImpl) proxy.senderTaskFactory).flushNow(key);
     ((QueueingFactoryImpl) proxy.queueingFactory).flushNow(key);
     gzippedHttpPost("http://localhost:" + proxyPort + "/", payload);
     ((SenderTaskFactoryImpl) proxy.senderTaskFactory).flushNow(key);
-    for (int i = 0; i < 3; i++) ((QueueingFactoryImpl) proxy.queueingFactory).flushNow(key);
-    assertEquals(6, successfulSteps.getAndSet(0));
-    assertTrue(part1.get());
-    assertTrue(part2.get());
+    List<String> expectedSplitLines =
+        Arrays.asList((expectedTest1part1 + "\n" + expectedTest1part2).split("\n"));
+    // The queue processor may skip a pass if rate limiter permits aren't immediately available,
+    // deferring the remaining retries to its own background schedule (already running since
+    // proxy.start()); poll passively for it to catch up rather than calling flushNow from this
+    // thread too, since a manual call racing the background scheduler's own concurrent run() can
+    // double-process the same queued batch and deliver it twice.
+    assertTrueWithTimeout(2000, () -> deliveredSplitLines.size() >= expectedSplitLines.size());
+    List<String> sortedDelivered = new ArrayList<>(deliveredSplitLines);
+    Collections.sort(sortedDelivered);
+    List<String> sortedExpected = new ArrayList<>(expectedSplitLines);
+    Collections.sort(sortedExpected);
+    assertEquals(sortedExpected, sortedDelivered);
   }
 
   @Test

proxy/src/test/java/com/wavefront/agent/ProxyConfigTest.java

@@ -295,4 +295,31 @@ public void testOtlpAppTagsOnMetricsIncluded() {
         new String[] {"--otlpAppTagsOnMetricsIncluded", String.valueOf(false)}, "PushAgentTest");
     assertFalse(config.isOtlpAppTagsOnMetricsIncluded());
   }
+
+  @Test
+  public void testMetricQuerySamplingDryRun() {
+    ProxyConfig config = new ProxyConfig();
+
+    // disabled by default when the parameter is not present
+    config.parseArguments(new String[] {"--token", UUID.randomUUID().toString()}, "PushAgentTest");
+    assertFalse(config.getMetricQuerySamplingDryRunEnabled());
+
+    // explicitly enable
+    config.parseArguments(
+        new String[] {"--metricQuerySamplingDryRun", String.valueOf(true)}, "PushAgentTest");
+    assertTrue(config.getMetricQuerySamplingDryRunEnabled());
+
+    // explicitly disable
+    config.parseArguments(
+        new String[] {"--metricQuerySamplingDryRun", String.valueOf(false)}, "PushAgentTest");
+    assertFalse(config.getMetricQuerySamplingDryRunEnabled());
+
+    // arity = 1 requires an explicit value; the bare flag is no longer a valid no-arg switch
+    try {
+      config.parseArguments(new String[] {"--metricQuerySamplingDryRun"}, "PushAgentTest");
+      fail();
+    } catch (ParameterException e) {
+      // noop
+    }
+  }
 }

proxy/src/test/java/com/wavefront/agent/api/APIContainerTest.java

@@ -5,6 +5,7 @@
 import com.wavefront.agent.ProxyConfig;
 import com.wavefront.agent.TokenManager;
 import com.wavefront.agent.TokenWorkerCSP;
+import org.junit.After;
 import org.junit.Before;
 import org.junit.Ignore;
 import org.junit.Test;
@@ -24,6 +25,15 @@ public void setup() {
     }
   }
 
+  @After
+  public void cleanup() {
+    // TokenManager holds its tenant/worker registry in static state; without resetting it here,
+    // the TokenWorkerCSP instances registered above (which implement TokenWorker.Scheduled)
+    // leak into any later test in the same JVM fork that calls TokenManager.start(), causing
+    // unrelated tests to unexpectedly invoke .run() against these leftover fake workers.
+    TokenManager.reset();
+  }
+
   @Test
   public void testAPIContainerInitiationWithDiscardData() {
     APIContainer apiContainer = new APIContainer(this.proxyConfig, true);

proxy/src/test/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticatorTest.java

@@ -56,6 +56,30 @@ public void testIntrospectionUrlInvocation() throws Exception {
     EasyMock.verify(client);
   }
 
+  @Test
+  public void testTokenIsUrlEncodedToPreventUrlInjection() throws Exception {
+    HttpClient client = EasyMock.createMock(HttpClient.class);
+    AtomicLong fakeClock = new AtomicLong(1_000_000);
+    TokenAuthenticator authenticator =
+        new HttpGetTokenIntrospectionAuthenticator(
+            client, "http://acme.corp/{{token}}/something", null, 300, 600, fakeClock::get);
+
+    // a malicious token attempting to redirect the introspection call to another host/path
+    // and/or inject extra query parameters must be treated as an opaque, percent-encoded value.
+    String maliciousToken = "../../evil.com/bypass?force=200#";
+    String expectedEncodedToken =
+        "..%2F..%2Fevil.com%2Fbypass%3Fforce%3D200%23"; // '.' and '-' remain unescaped
+    EasyMock.expect(
+            client.execute(
+                httpEq(
+                    new HttpGet(
+                        "http://acme.corp/" + expectedEncodedToken + "/something"))))
+        .andReturn(new BasicHttpResponse(HttpVersion.HTTP_1_1, 204, ""));
+    EasyMock.replay(client);
+    assertTrue(authenticator.authorize(maliciousToken));
+    EasyMock.verify(client);
+  }
+
   @Test
   public void testIntrospectionUrlCachedLastResultExpires() throws Exception {
     HttpClient client = EasyMock.createMock(HttpClient.class);

proxy/src/test/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticatorTest.java

@@ -132,4 +132,32 @@ public void testIntrospectionUrlInvalidResponseThrows() throws Exception {
     authenticator.authorize(uuid); // should call http
     assertEquals(1, Metrics.newCounter(new MetricName("auth", "", "api-errors")).count() - count);
   }
+
+  @Test
+  public void testTokenIsUrlEncodedToPreventUrlInjection() throws Exception {
+    HttpClient client = EasyMock.createMock(HttpClient.class);
+    AtomicLong fakeClock = new AtomicLong(1_000_000);
+    TokenAuthenticator authenticator =
+        new Oauth2TokenIntrospectionAuthenticator(
+            client, "http://acme.corp/{{token}}/oauth", null, 300, 600, fakeClock::get);
+
+    // a malicious token attempting to redirect the introspection call to another host/path
+    // and/or inject extra query parameters must be treated as an opaque, percent-encoded value
+    // when substituted into the URL (the raw token is still sent, correctly form-encoded, in the
+    // POST body).
+    String maliciousToken = "../../evil.com/bypass?force=200#";
+    String expectedEncodedToken = "..%2F..%2Fevil.com%2Fbypass%3Fforce%3D200%23";
+
+    HttpPost request = new HttpPost("http://acme.corp/" + expectedEncodedToken + "/oauth");
+    request.setHeader("Content-Type", "application/x-www-form-urlencoded");
+    request.setHeader("Accept", "application/json");
+    request.setEntity(
+        new UrlEncodedFormEntity(
+            ImmutableList.of(new BasicNameValuePair("token", maliciousToken))));
+
+    TestUtils.expectHttpResponse(client, request, "{\"active\": true}".getBytes(), 200);
+
+    assertTrue(authenticator.authorize(maliciousToken));
+    EasyMock.verify(client);
+  }
 }

proxy/src/test/java/com/wavefront/agent/sampler/MetricBloomFilterSamplerTest.java

@@ -1,6 +1,7 @@
 package com.wavefront.agent.sampler;
 
 import com.wavefront.api.agent.BloomFilterDTO;
+import com.wavefront.api.agent.preprocessor.ReportPointSampleInclude;
 import com.wavefront.common.bloomfilter.ReadOnlyBloomFilter;
 import org.junit.Test;
 import wavefront.report.ReportPoint;
@@ -79,7 +80,7 @@ public void testShouldSampleOutSkippedWhenSamplingTagSetFalse() throws Exception
     assertTrue(sampler.shouldSampleOut(point));
 
     // Set the sampling tag to "false" — preprocessor rule opted this point out of sampling
-    annotations.put("_wavefront_sampling_eligable", "false");
+    annotations.put(ReportPointSampleInclude.SAMPLING_TAG, "false");
     long counterBefore = sampler.excludedPreprocessorRules.count();
 
     // Point should NOT be sampled out now, even though bloom filter misses