Harden Agentless Feature Flags EVP delivery

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

@@ -1,5 +1,7 @@
 package datadog.communication;
 
+import static java.util.Collections.emptyList;
+
 import datadog.communication.ddagent.DDAgentFeaturesDiscovery;
 import datadog.communication.ddagent.SharedCommunicationObjects;
 import datadog.communication.http.HttpRetryPolicy;
@@ -15,13 +17,30 @@
 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 boolean sendOnce;
 
   public BackendApiFactory(Config config, SharedCommunicationObjects sharedCommunicationObjects) {
+    this(config, sharedCommunicationObjects, false);
+  }
+
+  /**
+   * Creates a backend factory with optional send-once transport semantics.
+   *
+   * <p>When {@code sendOnce} is true, automatic connection retries and redirects are disabled, and
+   * the default HTTP retry policy is {@link HttpRetryPolicy.Factory#NEVER_RETRY}. Explicit caller
+   * policies may retry proven pre-send failures, but must not replay an ambiguously accepted event
+   * payload without an idempotency key.
+   */
+  public BackendApiFactory(
+      Config config, SharedCommunicationObjects sharedCommunicationObjects, boolean sendOnce) {
     this.config = config;
     this.sharedCommunicationObjects = sharedCommunicationObjects;
+    this.sendOnce = sendOnce;
   }
 
   public @Nullable BackendApi createBackendApi(Intake intake) {
@@ -67,7 +86,9 @@ public BackendApi createDirectIntakeApi(
         apiKey,
         traceId,
         retryPolicyFactory(),
-        directIntakeHttpClient(sharedCommunicationObjects.getIntakeHttpClient(), followRedirects),
+        configureHttpClient(
+            directIntakeHttpClient(
+                sharedCommunicationObjects.getIntakeHttpClient(), followRedirects)),
         responseCompression);
   }
 
@@ -87,11 +108,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 +130,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,14 +175,63 @@ 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, emptyList());
+  }
+
+  /**
+   * Creates an EVP proxy client after Agent discovery, optionally forcing a fresh discovery and
+   * requiring the Agent to advertise every specified forwarding header.
+   *
+   * <p>Callers supply the forwarding capabilities their event protocol requires.
+   *
+   * <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,
+      Iterable<String> requiredProxyHeaders) {
+    String endpoint = discoverEvpProxyEndpoint(forceDiscovery, requiredProxyHeaders);
+    return endpoint == null
+        ? null
+        : createEvpProxyApi(intake, responseCompression, retryPolicyFactory, endpoint);
+  }
+
+  /**
+   * Discovers a capability-validated endpoint that related senders can share without re-probing.
+   */
+  public @Nullable String discoverEvpProxyEndpoint(
+      boolean forceDiscovery, Iterable<String> requiredProxyHeaders) {
     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
+        && !featuresDiscovery.supportsEvpProxyHeaders(requiredProxyHeaders)) {
+      evpProxyEndpoint = null;
+    }
+    return evpProxyEndpoint;
+  }
+
+  /** Creates a client for a fixed compatibility or previously discovered endpoint, without I/O. */
+  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={}",
@@ -140,11 +245,24 @@ static HttpUrl buildEventPlatformIntakeUrl(String site) {
         evpProxyUrl,
         subdomain,
         retryPolicyFactory,
-        sharedCommunicationObjects.agentHttpClient,
+        configureHttpClient(sharedCommunicationObjects.agentHttpClient),
         responseCompression);
   }
 
-  private static HttpRetryPolicy.Factory retryPolicyFactory() {
-    return new HttpRetryPolicy.Factory(5, 100, 2.0, true);
+  OkHttpClient configureHttpClient(final OkHttpClient httpClient) {
+    return sendOnce
+        ? httpClient
+            .newBuilder()
+            .retryOnConnectionFailure(false)
+            .followRedirects(false)
+            .followSslRedirects(false)
+            .build()
+        : httpClient;
+  }
+
+  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,12 @@ 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";
+
   /**
    * 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();
@@ -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)) {
@@ -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;