fix(feature-flags): add safe agentless EVP fallback

communication/src/main/java/datadog/communication/BackendApiFactory.java

@@ -8,6 +8,7 @@
 import datadog.trace.util.throwable.FatalAgentMisconfigurationError;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
+import okhttp3.OkHttpClient;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -48,7 +49,13 @@ public BackendApi createDirectIntakeApi(Intake intake) {
 
   /** Creates an authenticated API client that sends data directly to a Datadog intake. */
   public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompression) {
-    HttpUrl agentlessUrl = HttpUrl.get(intake.getAgentlessUrl(config));
+    return createDirectIntakeApi(intake, responseCompression, true);
+  }
+
+  /** Creates an authenticated API client that sends data directly to a Datadog intake. */
+  public BackendApi createDirectIntakeApi(
+      Intake intake, boolean responseCompression, boolean followRedirects) {
+    HttpUrl agentlessUrl = buildDirectIntakeUrl(intake, config);
     String apiKey = config.getApiKey();
     if (apiKey == null || apiKey.isEmpty()) {
       throw new FatalAgentMisconfigurationError(
@@ -60,10 +67,52 @@ public BackendApi createDirectIntakeApi(Intake intake, boolean responseCompressi
         apiKey,
         traceId,
         retryPolicyFactory(),
-        sharedCommunicationObjects.getIntakeHttpClient(),
+        directIntakeHttpClient(sharedCommunicationObjects.getIntakeHttpClient(), followRedirects),
         responseCompression);
   }
 
+  static OkHttpClient directIntakeHttpClient(
+      final OkHttpClient intakeHttpClient, final boolean followRedirects) {
+    if (followRedirects) {
+      return intakeHttpClient;
+    }
+    return intakeHttpClient.newBuilder().followRedirects(false).followSslRedirects(false).build();
+  }
+
+  private static HttpUrl buildDirectIntakeUrl(Intake intake, Config config) {
+    if (intake != Intake.EVENT_PLATFORM) {
+      return HttpUrl.get(intake.getAgentlessUrl(config));
+    }
+    return buildEventPlatformIntakeUrl(config.getSite());
+  }
+
+  static HttpUrl buildEventPlatformIntakeUrl(String site) {
+    if (site == null || site.isEmpty()) {
+      throw new IllegalArgumentException("Invalid Datadog site");
+    }
+
+    String expectedHost = Intake.EVENT_PLATFORM.getUrlPrefix() + "." + site;
+    HttpUrl url =
+        new HttpUrl.Builder()
+            .scheme("https")
+            .host(expectedHost)
+            .addPathSegment("api")
+            .addPathSegment(Intake.EVENT_PLATFORM.getVersion())
+            .addPathSegment("")
+            .build();
+    if (!url.isHttps()
+        || !url.username().isEmpty()
+        || !url.password().isEmpty()
+        || !url.host().equalsIgnoreCase(expectedHost)
+        || url.port() != 443
+        || !url.encodedPath().equals("/api/" + Intake.EVENT_PLATFORM.getVersion() + "/")
+        || url.encodedQuery() != null
+        || url.encodedFragment() != null) {
+      throw new IllegalArgumentException("Invalid Datadog site");
+    }
+    return url;
+  }
+
   /** Creates an API client that uses the specified retry policy with a compatible local proxy. */
   public @Nullable BackendApi createEvpProxyApi(Intake intake) {
     return createEvpProxyApi(intake, true);

communication/src/main/java/datadog/communication/ddagent/SharedCommunicationObjects.java

@@ -11,13 +11,25 @@
 import datadog.remoteconfig.DefaultConfigurationPoller;
 import datadog.trace.api.Config;
 import datadog.trace.api.civisibility.config.BazelMode;
+import datadog.trace.util.AgentProxySelector;
 import datadog.trace.util.AgentTaskScheduler;
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.net.Proxy;
+import java.net.ProxySelector;
+import java.net.SocketAddress;
+import java.net.URI;
 import java.security.Security;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
+import java.util.Locale;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
+import javax.annotation.Nullable;
+import okhttp3.Credentials;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
 import org.slf4j.Logger;
@@ -44,6 +56,9 @@ public class SharedCommunicationObjects {
    */
   private volatile OkHttpClient intakeHttpClient;
 
+  private volatile HttpUrl intakeHttpsProxy;
+  private volatile Set<String> intakeNoProxyHosts = Collections.emptySet();
+
   @SuppressFBWarnings("PA_PUBLIC_PRIMITIVE_ATTRIBUTE")
   public long httpClientTimeout;
 
@@ -78,6 +93,8 @@ public void createRemaining(Config config) {
             : TimeUnit.SECONDS.toMillis(config.getAgentTimeout());
 
     forceClearTextHttpForIntakeClient = config.isForceClearTextHttpForIntakeClient();
+    intakeHttpsProxy = parseHttpsProxy(config.getHttpsProxy());
+    intakeNoProxyHosts = config.getNoProxyHosts();
 
     if (agentUrl == null) {
       agentUrl = parseAgentUrl(config);
@@ -269,11 +286,94 @@ public OkHttpClient getIntakeHttpClient() {
 
     synchronized (this) {
       if (this.intakeHttpClient == null) {
-        this.intakeHttpClient =
+        OkHttpClient intakeClient =
             OkHttpUtils.buildHttpClient(
                 forceClearTextHttpForIntakeClient, null, null, httpClientTimeout);
+        if (intakeHttpsProxy != null) {
+          final Proxy proxy =
+              new Proxy(
+                  Proxy.Type.HTTP,
+                  new InetSocketAddress(intakeHttpsProxy.host(), intakeHttpsProxy.port()));
+          final OkHttpClient.Builder builder =
+              intakeClient
+                  .newBuilder()
+                  .proxySelector(new IntakeProxySelector(proxy, intakeNoProxyHosts));
+          if (!intakeHttpsProxy.username().isEmpty()) {
+            final String credential =
+                Credentials.basic(intakeHttpsProxy.username(), intakeHttpsProxy.password());
+            builder.proxyAuthenticator(
+                (route, response) ->
+                    response
+                        .request()
+                        .newBuilder()
+                        .header("Proxy-Authorization", credential)
+                        .build());
+          }
+          intakeClient = builder.build();
+        }
+        this.intakeHttpClient = intakeClient;
       }
       return this.intakeHttpClient;
     }
   }
+
+  @Nullable
+  static HttpUrl parseHttpsProxy(@Nullable final String configuredProxy) {
+    if (configuredProxy == null || configuredProxy.trim().isEmpty()) {
+      return null;
+    }
+    final String candidate =
+        configuredProxy.contains("://") ? configuredProxy : "http://" + configuredProxy;
+    final HttpUrl proxy = HttpUrl.parse(candidate);
+    if (proxy == null || !"http".equalsIgnoreCase(proxy.scheme())) {
+      log.warn("Ignoring invalid HTTPS proxy configuration");
+      return null;
+    }
+    return proxy;
+  }
+
+  static final class IntakeProxySelector extends ProxySelector {
+    private static final List<Proxy> DIRECT = Collections.singletonList(Proxy.NO_PROXY);
+
+    private final Proxy proxy;
+    private final Set<String> noProxyHosts;
+
+    IntakeProxySelector(final Proxy proxy, final Set<String> noProxyHosts) {
+      this.proxy = proxy;
+      this.noProxyHosts = noProxyHosts;
+    }
+
+    @Override
+    public List<Proxy> select(final URI uri) {
+      final String host = uri.getHost();
+      if (host != null && shouldBypassProxy(host)) {
+        return DIRECT;
+      }
+      if ("https".equalsIgnoreCase(uri.getScheme())) {
+        return Collections.singletonList(proxy);
+      }
+      return AgentProxySelector.INSTANCE.select(uri);
+    }
+
+    @Override
+    public void connectFailed(
+        final URI uri, final SocketAddress address, final IOException failure) {
+      AgentProxySelector.INSTANCE.connectFailed(uri, address, failure);
+    }
+
+    private boolean shouldBypassProxy(final String host) {
+      final String normalizedHost = host.toLowerCase(Locale.ROOT);
+      for (final String configuredHost : noProxyHosts) {
+        final String normalized = configuredHost.trim().toLowerCase(Locale.ROOT);
+        if ("*".equals(normalized)
+            || normalizedHost.equals(normalized)
+            || (normalized.startsWith(".")
+                && (normalizedHost.equals(normalized.substring(1))
+                    || normalizedHost.endsWith(normalized)))) {
+          return true;
+        }
+      }
+      return false;
+    }
+  }
 }

communication/src/test/groovy/datadog/communication/ddagent/SharedCommunicationsObjectsSpecification.groovy

@@ -90,6 +90,8 @@ class SharedCommunicationsObjectsSpecification extends DDSpecification {
     1 * config.isCiVisibilityEnabled()
     1 * config.getAgentTimeout()
     1 * config.isForceClearTextHttpForIntakeClient()
+    1 * config.getHttpsProxy()
+    1 * config.getNoProxyHosts()
     0 * _
     sco.agentUrl.is(url)
     sco.agentHttpClient.is(okHttpClient)
@@ -136,4 +138,46 @@ class SharedCommunicationsObjectsSpecification extends DDSpecification {
     then:
     client != null
   }
+
+  void 'configures standard HTTPS proxy for intake while preserving no-proxy hosts'() {
+    given:
+    Config config = Mock()
+    sco.agentUrl = HttpUrl.get("http://example.com")
+    sco.agentHttpClient = Mock(OkHttpClient)
+    sco.monitoring = Monitoring.DISABLED
+    sco.featuresDiscovery = Mock(DDAgentFeaturesDiscovery)
+
+    when:
+    sco.createRemaining(config)
+    def selector = sco.getIntakeHttpClient().proxySelector()
+
+    then:
+    1 * config.isCiVisibilityEnabled() >> false
+    1 * config.getAgentTimeout() >> 1
+    1 * config.isForceClearTextHttpForIntakeClient() >> false
+    1 * config.getHttpsProxy() >> "http://proxy.example:8181"
+    1 * config.getNoProxyHosts() >> (["direct.example", ".internal.example"] as Set)
+    0 * _
+
+    and:
+    def selected = selector.select(new URI("https://event-platform-intake.datadoghq.com"))
+    selected.size() == 1
+    selected[0].type() == Proxy.Type.HTTP
+    selected[0].address() == new InetSocketAddress("proxy.example", 8181)
+    selector.select(new URI("https://direct.example")) == [Proxy.NO_PROXY]
+    selector.select(new URI("https://service.internal.example")) == [Proxy.NO_PROXY]
+  }
+
+  void 'parses supported HTTPS proxy forms without logging credentials'() {
+    expect:
+    SharedCommunicationObjects.parseHttpsProxy(configured)?.toString() == expected
+
+    where:
+    configured                      | expected
+    null                            | null
+    ""                              | null
+    "proxy.example:8080"            | "http://proxy.example:8080/"
+    "http://user:pass@proxy:3128"   | "http://user:pass@proxy:3128/"
+    "https://unsupported.example"   | null
+  }
 }

communication/src/test/java/datadog/communication/BackendApiFactoryTest.java

@@ -13,7 +13,9 @@
 import datadog.trace.api.Config;
 import datadog.trace.api.ProtocolVersion;
 import datadog.trace.api.intake.Intake;
+import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Locale;
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
@@ -22,11 +24,98 @@
 import okhttp3.mockwebserver.MockWebServer;
 import okhttp3.mockwebserver.RecordedRequest;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.NullAndEmptySource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 class BackendApiFactoryTest {
 
   private static final MediaType JSON = MediaType.parse("application/json");
 
+  @ParameterizedTest
+  @ValueSource(strings = {"datadoghq.com", "custom.example", "DATADOGHQ.EU"})
+  void eventPlatformDirectIntakeUsesExactHttpsHost(String site) {
+    final HttpUrl url = BackendApiFactory.buildEventPlatformIntakeUrl(site);
+
+    assertEquals("https", url.scheme());
+    assertEquals("event-platform-intake." + site.toLowerCase(Locale.ROOT), url.host());
+    assertEquals(443, url.port());
+    assertEquals("/api/v2/", url.encodedPath());
+    assertEquals("", url.username());
+    assertEquals("", url.password());
+    assertNull(url.encodedQuery());
+    assertNull(url.encodedFragment());
+  }
+
+  @ParameterizedTest
+  @NullAndEmptySource
+  @ValueSource(
+      strings = {
+        "datadoghq.com@evil.example",
+        "datadoghq.com:password@evil.example",
+        "https://datadoghq.com",
+        "datadoghq.com:443",
+        "datadoghq.com:8443",
+        "datadoghq.com/path",
+        "datadoghq.com?query=value",
+        "datadoghq.com#fragment",
+        "data doghq.com",
+        " datadoghq.com",
+        "datadoghq.com ",
+        "datadoghq.com\\evil.example"
+      })
+  void eventPlatformDirectIntakeRejectsUnsafeSite(String site) {
+    assertThrows(
+        IllegalArgumentException.class, () -> BackendApiFactory.buildEventPlatformIntakeUrl(site));
+  }
+
+  @ParameterizedTest
+  @ValueSource(ints = {301, 302, 307, 308})
+  void featureFlagDirectIntakeDoesNotFollowRedirects(final int statusCode) throws Exception {
+    final MockWebServer intake = new MockWebServer();
+    final MockWebServer redirectTarget = new MockWebServer();
+    final OkHttpClient sharedClient = new OkHttpClient.Builder().build();
+    final OkHttpClient directClient = BackendApiFactory.directIntakeHttpClient(sharedClient, false);
+    redirectTarget.start();
+    intake.enqueue(
+        new MockResponse()
+            .setResponseCode(statusCode)
+            .setHeader("Location", redirectTarget.url("/redirected")));
+    intake.start();
+    try {
+      final IntakeApi api =
+          new IntakeApi(
+              intake.url("/api/v2/"),
+              "api-key",
+              "123",
+              HttpRetryPolicy.Factory.NEVER_RETRY,
+              directClient,
+              false);
+
+      assertThrows(
+          IOException.class,
+          () ->
+              api.post(
+                  "flagevaluation",
+                  RequestBody.create(JSON, "{}".getBytes(StandardCharsets.UTF_8)),
+                  stream -> null,
+                  null,
+                  false));
+
+      final RecordedRequest request = intake.takeRequest();
+      assertEquals("api-key", request.getHeader("DD-API-KEY"));
+      assertEquals(1, intake.getRequestCount());
+      assertEquals(0, redirectTarget.getRequestCount());
+    } finally {
+      directClient.dispatcher().executorService().shutdownNow();
+      directClient.connectionPool().evictAll();
+      sharedClient.dispatcher().executorService().shutdownNow();
+      sharedClient.connectionPool().evictAll();
+      intake.shutdown();
+      redirectTarget.shutdown();
+    }
+  }
+
   @Test
   void noBackendApiWhenAgentDoesNotAdvertiseEvpProxy() {
     final FakeFeaturesDiscovery discovery = new FakeFeaturesDiscovery(null);

dd-trace-api/src/main/java/datadog/trace/api/config/TracerConfig.java

@@ -29,6 +29,7 @@ public final class TracerConfig {
   public static final String AGENT_TIMEOUT = "trace.agent.timeout";
   public static final String FORCE_CLEAR_TEXT_HTTP_FOR_INTAKE_CLIENT =
       "force.clear.text.http.for.intake.client";
+  public static final String PROXY_HTTPS = "proxy.https";
   public static final String PROXY_NO_PROXY = "proxy.no_proxy";
   public static final String TRACE_AGENT_PATH = "trace.agent.path";
   public static final String TRACE_AGENT_ARGS = "trace.agent.args";