diff --git a/proxy/pom.xml b/proxy/pom.xml index b00036b9f..e6d735e8f 100644 --- a/proxy/pom.xml +++ b/proxy/pom.xml @@ -4,7 +4,7 @@ com.wavefront proxy - 15.1 + 16.0 Wavefront Proxy Service for batching and relaying metric traffic to Wavefront diff --git a/proxy/src/main/java/com/wavefront/agent/ProxyConfigDef.java b/proxy/src/main/java/com/wavefront/agent/ProxyConfigDef.java index 171950926..f8468b7a7 100644 --- a/proxy/src/main/java/com/wavefront/agent/ProxyConfigDef.java +++ b/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; } diff --git a/proxy/src/main/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticator.java b/proxy/src/main/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticator.java index 379b28431..72949ba5c 100644 --- a/proxy/src/main/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticator.java +++ b/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); } diff --git a/proxy/src/main/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticator.java b/proxy/src/main/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticator.java index eb28c0ca7..f00c866af 100644 --- a/proxy/src/main/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticator.java +++ b/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) { diff --git a/proxy/src/main/java/com/wavefront/agent/auth/TokenIntrospectionAuthenticator.java b/proxy/src/main/java/com/wavefront/agent/auth/TokenIntrospectionAuthenticator.java index ee1b36318..0c8d1a810 100644 --- a/proxy/src/main/java/com/wavefront/agent/auth/TokenIntrospectionAuthenticator.java +++ b/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) { diff --git a/proxy/src/main/java/com/wavefront/agent/listeners/DataDogPortUnificationHandler.java b/proxy/src/main/java/com/wavefront/agent/listeners/DataDogPortUnificationHandler.java index 3545805c7..ad1967a14 100644 --- a/proxy/src/main/java/com/wavefront/agent/listeners/DataDogPortUnificationHandler.java +++ b/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 diff --git a/proxy/src/test/java/com/wavefront/agent/HttpEndToEndTest.java b/proxy/src/test/java/com/wavefront/agent/HttpEndToEndTest.java index fbdd9b439..6b493f645 100644 --- a/proxy/src/test/java/com/wavefront/agent/HttpEndToEndTest.java +++ b/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 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 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 sortedDelivered = new ArrayList<>(deliveredSplitLines); + Collections.sort(sortedDelivered); + List sortedExpected = new ArrayList<>(expectedSplitLines); + Collections.sort(sortedExpected); + assertEquals(sortedExpected, sortedDelivered); } @Test diff --git a/proxy/src/test/java/com/wavefront/agent/ProxyConfigTest.java b/proxy/src/test/java/com/wavefront/agent/ProxyConfigTest.java index 60d0d29e2..53d6b5af8 100644 --- a/proxy/src/test/java/com/wavefront/agent/ProxyConfigTest.java +++ b/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 + } + } } diff --git a/proxy/src/test/java/com/wavefront/agent/api/APIContainerTest.java b/proxy/src/test/java/com/wavefront/agent/api/APIContainerTest.java index e065b2654..5b1261a7f 100644 --- a/proxy/src/test/java/com/wavefront/agent/api/APIContainerTest.java +++ b/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); diff --git a/proxy/src/test/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticatorTest.java b/proxy/src/test/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticatorTest.java index 8a057693b..4c6a3b554 100644 --- a/proxy/src/test/java/com/wavefront/agent/auth/HttpGetTokenIntrospectionAuthenticatorTest.java +++ b/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); diff --git a/proxy/src/test/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticatorTest.java b/proxy/src/test/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticatorTest.java index 2d85b8679..fbf91555e 100644 --- a/proxy/src/test/java/com/wavefront/agent/auth/Oauth2TokenIntrospectionAuthenticatorTest.java +++ b/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); + } } diff --git a/proxy/src/test/java/com/wavefront/agent/sampler/MetricBloomFilterSamplerTest.java b/proxy/src/test/java/com/wavefront/agent/sampler/MetricBloomFilterSamplerTest.java index a8b72663d..3b1d8d421 100644 --- a/proxy/src/test/java/com/wavefront/agent/sampler/MetricBloomFilterSamplerTest.java +++ b/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