Harden Agentless Feature Flags EVP delivery

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

@@ -1,27 +1,61 @@
 package datadog.communication;
 
+import static java.util.Collections.emptyMap;
+import static java.util.Collections.unmodifiableMap;
+
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.http.HttpRetryPolicy;
 import datadog.trace.api.Config;
 import datadog.trace.api.intake.Intake;
 import datadog.trace.util.throwable.FatalAgentMisconfigurationError;
+import java.util.HashMap;
+import java.util.Map;
 import javax.annotation.Nullable;
 import okhttp3.HttpUrl;
 import okhttp3.OkHttpClient;
+import okhttp3.Request;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class BackendApiFactory {
 
   private static final Logger log = LoggerFactory.getLogger(BackendApiFactory.class);
+  private static final int MAX_DNS_LABEL_LENGTH = 63;
+  private static final int MAX_DNS_HOST_LENGTH = 253;
 
   private final Config config;
   private final SharedCommunicationObjects sharedCommunicationObjects;
+  private final Map<String, String> requestHeaders;
+  private final boolean sendOnce;
 
   public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommunicationObjects) {
+    this(config, sharedCommunicationObjects, emptyMap());
+  }
+
+  public BackendApiFactory(
+      Config config,
+      SharedCommunicationObjects sharedCommunicationObjects,
+      Map<String, String> requestHeaders) {
+    this(config, sharedCommunicationObjects, requestHeaders, false);
+  }
+
+  /**
+   * Creates a backend factory with per-request headers and optional send-once transport semantics.
+   *
+   * <p>When {@code sendOnce} is true, both the explicit HTTP retry policy and OkHttp's automatic
+   * connection retry are disabled. This is required for event payloads that do not carry an
+   * idempotency key.
+   */
+  public BackendApiFactory(
+      Config config,
+      SharedCommunicationObjects sharedCommunicationObjects,
+      Map<String, String> requestHeaders,
+      boolean sendOnce) {
     this.config = config;
     this.sharedCommunicationObjects = sharedCommunicationObjects;
+    this.requestHeaders = unmodifiableMap(new HashMap<>(requestHeaders));
+    this.sendOnce = sendOnce;
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake) {
@@ -67,7 +101,9 @@ public BackendApi createDirectIntakeApi(
         apiKey,
         traceId,
         retryPolicyFactory(),
-        directIntakeHttpClient(sharedCommunicationObjects.getIntakeHttpClient(), followRedirects),
+        configureHttpClient(
+            directIntakeHttpClient(
+                sharedCommunicationObjects.getIntakeHttpClient(), followRedirects)),
         responseCompression);
   }
 
@@ -87,11 +123,14 @@ private static HttpUrl buildDirectIntakeUrl(Intake intake, Config config) {
   }
 
   static HttpUrl buildEventPlatformIntakeUrl(String site) {
-    if (site == null || site.isEmpty()) {
+    if (!isValidDnsSuffix(site)) {
       throw new IllegalArgumentException("Invalid Datadog site");
     }
 
     String expectedHost = Intake.EVENT_PLATFORM.getUrlPrefix() + "." + site;
+    if (expectedHost.length() > MAX_DNS_HOST_LENGTH) {
+      throw new IllegalArgumentException("Invalid Datadog site");
+    }
     HttpUrl url =
         new HttpUrl.Builder()
             .scheme("https")
@@ -106,6 +145,38 @@ static HttpUrl buildEventPlatformIntakeUrl(String site) {
     return url;
   }
 
+  private static boolean isValidDnsSuffix(@Nullable String site) {
+    if (site == null || site.isEmpty()) {
+      return false;
+    }
+
+    int labelLength = 0;
+    for (int i = 0; i < site.length(); i++) {
+      final char character = site.charAt(i);
+      if (character == '.') {
+        if (labelLength == 0 || labelLength > MAX_DNS_LABEL_LENGTH || site.charAt(i - 1) == '-') {
+          return false;
+        }
+        labelLength = 0;
+      } else {
+        if ((!isAsciiLetterOrDigit(character) && character != '-')
+            || (labelLength == 0 && character == '-')) {
+          return false;
+        }
+        labelLength++;
+      }
+    }
+    return labelLength > 0
+        && labelLength <= MAX_DNS_LABEL_LENGTH
+        && site.charAt(site.length() - 1) != '-';
+  }
+
+  private static boolean isAsciiLetterOrDigit(final char character) {
+    return (character >= 'a' && character <= 'z')
+        || (character >= 'A' && character <= 'Z')
+        || (character >= '0' && character <= '9');
+  }
+
   /** 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);
@@ -119,32 +190,104 @@ static HttpUrl buildEventPlatformIntakeUrl(String site) {
   /** Creates an API client that sends data through a compatible local EVP proxy. */
   public @Nullable BackendApi createEvpProxyApi(
       Intake intake, boolean responseCompression, HttpRetryPolicy.Factory retryPolicyFactory) {
+    return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, false, false);
+  }
+
+  /**
+   * Creates an EVP proxy client after Agent discovery, optionally forcing a fresh discovery and
+   * requiring the Agent to advertise every configured request header.
+   *
+   * <p>The {@code forceDiscovery} form is intended for bounded unavailable-route recovery probes.
+   */
+  public @Nullable BackendApi createEvpProxyApi(
+      Intake intake,
+      boolean responseCompression,
+      HttpRetryPolicy.Factory retryPolicyFactory,
+      boolean forceDiscovery,
+      boolean requireConfiguredRequestHeaders) {
     DDAgentFeaturesDiscovery featuresDiscovery =
         sharedCommunicationObjects.featuresDiscovery(config);
-    featuresDiscovery.discoverIfOutdated();
-    if (!featuresDiscovery.supportsEvpProxy()) {
-      return null;
+    if (forceDiscovery) {
+      featuresDiscovery.discover();
+    } else {
+      featuresDiscovery.discoverIfOutdated();
     }
     String evpProxyEndpoint = featuresDiscovery.getEvpProxyEndpoint();
+    if (evpProxyEndpoint != null
+        && requireConfiguredRequestHeaders
+        && !featuresDiscovery.supportsEvpProxyHeaders(requestHeaders.keySet())) {
+      evpProxyEndpoint = null;
+    }
+    if (evpProxyEndpoint == null) {
+      return null;
+    }
+
+    return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, evpProxyEndpoint);
+  }
+
+  /** Creates an EVP proxy client for a fixed compatibility endpoint without Agent discovery. */
+  public BackendApi createEvpProxyApiForEndpoint(
+      Intake intake,
+      boolean responseCompression,
+      HttpRetryPolicy.Factory retryPolicyFactory,
+      String evpProxyEndpoint) {
+    return createEvpProxyApi(intake, responseCompression, retryPolicyFactory, evpProxyEndpoint);
+  }
 
+  private BackendApi createEvpProxyApi(
+      Intake intake,
+      boolean responseCompression,
+      HttpRetryPolicy.Factory retryPolicyFactory,
+      String evpProxyEndpoint) {
     String traceId = config.getIdGenerationStrategy().generateTraceId().toString();
     log.debug(
         "Creating EVP proxy client for {} using endpoint {} with responseCompression={}",
         intake,
         evpProxyEndpoint,
         responseCompression);
-    HttpUrl evpProxyUrl = sharedCommunicationObjects.agentUrl.resolve(evpProxyEndpoint);
+    HttpUrl evpProxyUrl = appendPath(sharedCommunicationObjects.agentUrl, evpProxyEndpoint);
     String subdomain = intake.getUrlPrefix();
     return new EvpProxyApi(
         traceId,
         evpProxyUrl,
         subdomain,
-        retryPolicyFactory,
-        sharedCommunicationObjects.agentHttpClient,
+        sendOnce ? HttpRetryPolicy.Factory.NEVER_RETRY : retryPolicyFactory,
+        configureHttpClient(sharedCommunicationObjects.agentHttpClient),
         responseCompression);
   }
 
-  private static HttpRetryPolicy.Factory retryPolicyFactory() {
-    return new HttpRetryPolicy.Factory(5, 100, 2.0, true);
+  static HttpUrl appendPath(final HttpUrl baseUrl, final String path) {
+    int firstCharacter = 0;
+    while (firstCharacter < path.length() && path.charAt(firstCharacter) == '/') {
+      firstCharacter++;
+    }
+    return baseUrl.newBuilder().addPathSegments(path.substring(firstCharacter)).build();
+  }
+
+  OkHttpClient configureHttpClient(final OkHttpClient httpClient) {
+    if (requestHeaders.isEmpty() && !sendOnce) {
+      return httpClient;
+    }
+    final OkHttpClient.Builder builder = httpClient.newBuilder();
+    if (sendOnce) {
+      builder.retryOnConnectionFailure(false);
+    }
+    if (!requestHeaders.isEmpty()) {
+      builder.addInterceptor(
+          chain -> {
+            final Request.Builder requestBuilder = chain.request().newBuilder();
+            for (Map.Entry<String, String> header : requestHeaders.entrySet()) {
+              requestBuilder.header(header.getKey(), header.getValue());
+            }
+            return chain.proceed(requestBuilder.build());
+          });
+    }
+    return builder.build();
+  }
+
+  private HttpRetryPolicy.Factory retryPolicyFactory() {
+    return sendOnce
+        ? HttpRetryPolicy.Factory.NEVER_RETRY
+        : new HttpRetryPolicy.Factory(5, 100, 2.0, true);
   }
 }

communication/src/main/java/datadog/communication/EvpProxy.java

@@ -5,6 +5,15 @@ public final class EvpProxy {
 
   public static final String SUBDOMAIN_HEADER = "X-Datadog-EVP-Subdomain";
 
+  /** Identifies the SDK that produced an EVP request. */
+  public static final String ORIGIN_HEADER = "DD-EVP-ORIGIN";
+
+  /** Identifies the version of the SDK that produced an EVP request. */
+  public static final String ORIGIN_VERSION_HEADER = "DD-EVP-ORIGIN-VERSION";
+
+  /** Origin header value identifying this tracing library. */
+  public static final String JAVA_TRACING_LIBRARY = "dd-trace-java";
+
   /**
    * Default SDK-side target for uncompressed EVP request bodies. Writers may split batches at or
    * below this size to keep Agent proxy requests comfortably bounded.

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

@@ -24,6 +24,7 @@
 import java.nio.ByteBuffer;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Set;
 import okhttp3.HttpUrl;
@@ -98,6 +99,7 @@ private static class State {
     String debuggerSnapshotEndpoint;
     String debuggerDiagnosticsEndpoint;
     String evpProxyEndpoint;
+    Set<String> evpProxyAllowedHeaders = emptySet();
     String version;
     String telemetryProxyEndpoint;
     Set<String> peerTags = emptySet();
@@ -156,7 +158,7 @@ private void doDiscovery(State newState) {
     try (Recording recording = discoveryTimer.start()) {
       boolean fallback = true;
       final Request request =
-          prepareRequest(agentBaseUrl.resolve("info"), emptyMap()).get().build();
+          prepareRequest(appendPath(agentBaseUrl, "info"), emptyMap()).get().build();
       try (Response response = client.newCall(request).execute()) {
         if (response.isSuccessful()) {
           processInfoResponseHeaders(response);
@@ -206,7 +208,7 @@ private String probeTracesEndpoint(State newState, List<String> endpoints) {
       try (Response response =
           client
               .newCall(
-                  prepareRequest(agentBaseUrl.resolve(candidate), emptyMap())
+                  prepareRequest(appendPath(agentBaseUrl, candidate), emptyMap())
                       .put(msgpackRequestBodyOf(singletonList(ByteBuffer.wrap(PROBE_MESSAGE))))
                       .build())
               .execute()) {
@@ -289,6 +291,16 @@ private boolean processInfoResponse(State newState, String response) {
           break;
         }
       }
+      final Object allowedHeadersObj = map.get("evp_proxy_allowed_headers");
+      if (allowedHeadersObj instanceof List) {
+        final Set<String> allowedHeaders = new HashSet<>();
+        for (Object header : (List<?>) allowedHeadersObj) {
+          if (header instanceof String) {
+            allowedHeaders.add(((String) header).toLowerCase(Locale.ROOT));
+          }
+        }
+        newState.evpProxyAllowedHeaders = unmodifiableSet(allowedHeaders);
+      }
 
       for (String endpoint : telemetryProxyEndpoints) {
         if (containsEndpoint(endpoints, endpoint)) {
@@ -424,7 +436,7 @@ public String getEvpProxyEndpoint() {
   }
 
   public HttpUrl buildUrl(String endpoint) {
-    return agentBaseUrl.resolve(endpoint);
+    return appendPath(agentBaseUrl, endpoint);
   }
 
   public boolean supportsDataStreams() {
@@ -435,6 +447,17 @@ public boolean supportsEvpProxy() {
     return discoveryState.evpProxyEndpoint != null;
   }
 
+  /** Returns whether the Agent advertises forwarding every required EVP request header. */
+  public boolean supportsEvpProxyHeaders(final Iterable<String> requiredHeaders) {
+    final Set<String> allowedHeaders = discoveryState.evpProxyAllowedHeaders;
+    for (String requiredHeader : requiredHeaders) {
+      if (!allowedHeaders.contains(requiredHeader.toLowerCase(Locale.ROOT))) {
+        return false;
+      }
+    }
+    return true;
+  }
+
   public boolean supportsContentEncodingHeadersWithEvpProxy() {
     // content encoding headers are supported in /v4 and above
     final String evpProxyEndpoint = discoveryState.evpProxyEndpoint;
@@ -469,4 +492,12 @@ public boolean active() {
   public boolean supportsTelemetryProxy() {
     return discoveryState.telemetryProxyEndpoint != null;
   }
+
+  private static HttpUrl appendPath(final HttpUrl baseUrl, final String path) {
+    int firstCharacter = 0;
+    while (firstCharacter < path.length() && path.charAt(firstCharacter) == '/') {
+      firstCharacter++;
+    }
+    return baseUrl.newBuilder().addPathSegments(path.substring(firstCharacter)).build();
+  }
 }