[appsec] Add Remote Config subscription for ASM_SCA product

  Implements Remote Config infrastructure for Supply Chain Analysis (SCA)
  vulnerability detection via dynamic instrumentation.

  This commit adds:
  - New Product.ASM_SCA enum value for Remote Config product type
  - CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION capability flag (bit 47)
  - AppSecSCAConfig data model with InstrumentationTarget (className/methodName)
  - AppSecSCAConfigDeserializer for JSON deserialization
  - subscribeSCA()/unsubscribeSCA() lifecycle methods in AppSecConfigServiceImpl
  - Integration into Remote Config subscription/cleanup flows

  The subscription stores incoming SCA configs in currentSCAConfig field.
  Actual bytecode retransformation will be implemented in future commits
  when AppSecInstrumentationUpdater is added.

Author: jandro996

dd-java-agent/appsec/src/main/java/com/datadog/appsec/config/AppSecConfigServiceImpl.java

@@ -21,6 +21,7 @@
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_RASP_SQLI;
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_RASP_SSRF;
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_REQUEST_BLOCKING;
+import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION;
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SESSION_FINGERPRINT;
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_TRACE_TAGGING_RULES;
 import static datadog.remoteconfig.Capabilities.CAPABILITY_ASM_TRUSTED_IPS;
@@ -103,13 +104,15 @@ public class AppSecConfigServiceImpl implements AppSecConfigService {
   private boolean hasUserWafConfig;
   private boolean defaultConfigActivated;
   private final AtomicBoolean subscribedToRulesAndData = new AtomicBoolean();
+  private final AtomicBoolean subscribedToSCA = new AtomicBoolean();
   private final Set<String> usedDDWafConfigKeys =
       Collections.newSetFromMap(new ConcurrentHashMap<>());
   private final Set<String> ignoredConfigKeys =
       Collections.newSetFromMap(new ConcurrentHashMap<>());
   private final String DEFAULT_WAF_CONFIG_RULE = "ASM_DD/default";
   private String currentRuleVersion;
   private List<AppSecModule> modulesToUpdateVersionIn;
+  private volatile AppSecSCAConfig currentSCAConfig;
 
   public AppSecConfigServiceImpl(
       Config tracerConfig,
@@ -134,6 +137,8 @@ private void subscribeConfigurationPoller() {
       log.debug("Will not subscribe to ASM, ASM_DD and ASM_DATA (AppSec custom rules in use)");
     }
 
+    subscribeSCA();
+
     this.configurationPoller.addConfigurationEndListener(applyRemoteConfigListener);
   }
 
@@ -345,6 +350,51 @@ private void subscribeAsmFeatures() {
     this.configurationPoller.addCapabilities(CAPABILITY_ASM_AUTO_USER_INSTRUM_MODE);
   }
 
+  /**
+   * Subscribes to Supply Chain Analysis (SCA) configuration from Remote Config.
+   * Receives instrumentation targets for vulnerability detection in third-party dependencies.
+   */
+  private void subscribeSCA() {
+    if (subscribedToSCA.compareAndSet(false, true)) {
+      log.debug("Subscribing to ASM_SCA Remote Config product");
+      this.configurationPoller.addListener(
+          Product.ASM_SCA,
+          AppSecSCAConfigDeserializer.INSTANCE,
+          (configKey, newConfig, hinter) -> {
+            if (newConfig == null) {
+              log.debug("Received removal for SCA config key: {}", configKey);
+              currentSCAConfig = null;
+              // TODO: Trigger retransformation to remove instrumentation when updater exists
+            } else {
+              log.debug(
+                  "Received SCA config update for key: {} - enabled: {}, targets: {}",
+                  configKey,
+                  newConfig.enabled,
+                  newConfig.instrumentationTargets != null
+                      ? newConfig.instrumentationTargets.size()
+                      : 0);
+              currentSCAConfig = newConfig;
+              // TODO: Trigger retransformation when AppSecInstrumentationUpdater exists
+            }
+          });
+      this.configurationPoller.addCapabilities(CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION);
+      log.info("Successfully subscribed to ASM_SCA Remote Config product");
+    }
+  }
+
+  /**
+   * Unsubscribes from SCA Remote Config product and clears current configuration.
+   */
+  private void unsubscribeSCA() {
+    if (subscribedToSCA.compareAndSet(true, false)) {
+      log.debug("Unsubscribing from ASM_SCA Remote Config product");
+      this.configurationPoller.removeListeners(Product.ASM_SCA);
+      this.configurationPoller.removeCapabilities(CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION);
+      currentSCAConfig = null;
+      log.info("Successfully unsubscribed from ASM_SCA Remote Config product");
+    }
+  }
+
   private void distributeSubConfigurations(
       String key, AppSecModuleConfigurer.Reconfiguration reconfiguration) {
     maybeInitializeDefaultConfig();
@@ -547,6 +597,7 @@ public void close() {
     this.configurationPoller.removeListeners(Product.ASM_DATA);
     this.configurationPoller.removeListeners(Product.ASM);
     this.configurationPoller.removeListeners(Product.ASM_FEATURES);
+    unsubscribeSCA();
     this.configurationPoller.removeConfigurationEndListener(applyRemoteConfigListener);
     this.subscribedToRulesAndData.set(false);
     this.configurationPoller.stop();

dd-java-agent/appsec/src/main/java/com/datadog/appsec/config/AppSecSCAConfig.java

@@ -0,0 +1,44 @@
+package com.datadog.appsec.config;
+
+import com.squareup.moshi.Json;
+import java.util.List;
+
+/**
+ * Configuration model for Supply Chain Analysis (SCA) vulnerability detection.
+ * Received via Remote Config in the ASM_SCA product.
+ *
+ * <p>This configuration enables dynamic instrumentation of third-party dependencies
+ * to detect and report known vulnerabilities at runtime.
+ */
+public class AppSecSCAConfig {
+
+  /**
+   * Whether SCA vulnerability detection is enabled.
+   */
+  @Json(name = "enabled")
+  public Boolean enabled;
+
+  /**
+   * List of instrumentation targets for SCA analysis.
+   * Each target specifies a class/method to instrument for vulnerability detection.
+   */
+  @Json(name = "instrumentation_targets")
+  public List<InstrumentationTarget> instrumentationTargets;
+
+  /**
+   * Represents a single instrumentation target for SCA.
+   */
+  public static class InstrumentationTarget {
+    /**
+     * Fully qualified class name in internal format (e.g., "org/springframework/web/client/RestTemplate").
+     */
+    @Json(name = "class_name")
+    public String className;
+
+    /**
+     * Method name to instrument (e.g., "execute").
+     */
+    @Json(name = "method_name")
+    public String methodName;
+  }
+}

dd-java-agent/appsec/src/main/java/com/datadog/appsec/config/AppSecSCAConfigDeserializer.java

@@ -0,0 +1,30 @@
+package com.datadog.appsec.config;
+
+import com.squareup.moshi.JsonAdapter;
+import com.squareup.moshi.Moshi;
+import datadog.remoteconfig.ConfigurationDeserializer;
+import okio.Okio;
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+
+/**
+ * Deserializer for Supply Chain Analysis (SCA) configuration from Remote Config.
+ * Converts JSON payload from ASM_SCA product into typed AppSecSCAConfig objects.
+ */
+public class AppSecSCAConfigDeserializer implements ConfigurationDeserializer<AppSecSCAConfig> {
+
+  public static final AppSecSCAConfigDeserializer INSTANCE = new AppSecSCAConfigDeserializer();
+
+  private static final JsonAdapter<AppSecSCAConfig> ADAPTER =
+      new Moshi.Builder().build().adapter(AppSecSCAConfig.class);
+
+  private AppSecSCAConfigDeserializer() {}
+
+  @Override
+  public AppSecSCAConfig deserialize(byte[] content) throws IOException {
+    if (content == null || content.length == 0) {
+      return null;
+    }
+    return ADAPTER.fromJson(Okio.buffer(Okio.source(new ByteArrayInputStream(content))));
+  }
+}

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/config/AppSecConfigServiceImplSpecification.groovy

@@ -99,6 +99,7 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
     1 * poller.addListener(Product.ASM_FEATURES, _, _)
     1 * poller.addListener(Product.ASM, _)
     1 * poller.addListener(Product.ASM_DATA, _)
+    1 * poller.addListener(Product.ASM_SCA, _, _)
     1 * poller.addConfigurationEndListener(_)
     0 * poller.addListener(*_)
     0 * poller.addCapabilities(CAPABILITY_ASM_ACTIVATION)
@@ -135,6 +136,7 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
     then:
     2 * config.getAppSecActivation() >> ProductActivation.ENABLED_INACTIVE
     1 * poller.addListener(Product.ASM_FEATURES, _, _)
+    1 * poller.addListener(Product.ASM_SCA, _, _)
     1 * poller.addConfigurationEndListener(_)
     0 * poller.addListener(*_)
   }
@@ -211,11 +213,13 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
       listeners.savedFeaturesDeserializer = it[1]
       listeners.savedFeaturesListener = it[2]
     }
+    1 * poller.addListener(Product.ASM_SCA, _, _)
     1 * poller.addConfigurationEndListener(_) >> {
       listeners.savedConfEndListener = it[0]
     }
     1 * poller.addCapabilities(CAPABILITY_ASM_ACTIVATION)
     1 * poller.addCapabilities(CAPABILITY_ASM_AUTO_USER_INSTRUM_MODE)
+    1 * poller.addCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
     0 * poller._
 
     when:
@@ -252,11 +256,13 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
       listeners.savedFeaturesDeserializer = it[1]
       listeners.savedFeaturesListener = it[2]
     }
+    1 * poller.addListener(Product.ASM_SCA, _, _)
     1 * poller.addConfigurationEndListener(_) >> {
       listeners.savedConfEndListener = it[0]
     }
     1 * poller.addCapabilities(CAPABILITY_ASM_ACTIVATION)
     1 * poller.addCapabilities(CAPABILITY_ASM_AUTO_USER_INSTRUM_MODE)
+    1 * poller.addCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
     0 * poller._
 
     when:
@@ -416,11 +422,13 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
       listeners.savedFeaturesDeserializer = it[1]
       listeners.savedFeaturesListener = it[2]
     }
+    1 * poller.addListener(Product.ASM_SCA, _, _)
     1 * poller.addConfigurationEndListener(_) >> {
       listeners.savedConfEndListener = it[0]
     }
     1 * poller.addCapabilities(CAPABILITY_ASM_ACTIVATION)
     1 * poller.addCapabilities(CAPABILITY_ASM_AUTO_USER_INSTRUM_MODE)
+    1 * poller.addCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
     0 * poller._
 
     when:
@@ -553,6 +561,8 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
       | CAPABILITY_ASM_HEADER_FINGERPRINT
       | CAPABILITY_ASM_TRACE_TAGGING_RULES
       | CAPABILITY_ASM_EXTENDED_DATA_COLLECTION)
+    1 * poller.removeListeners(Product.ASM_SCA)
+    1 * poller.removeCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
     4 * poller.removeListeners(_)
     1 * poller.removeConfigurationEndListener(_)
     1 * poller.stop()
@@ -776,6 +786,35 @@ class AppSecConfigServiceImplSpecification extends DDSpecification {
     noExceptionThrown()
   }
 
+  void 'subscribes to ASM_SCA product when configuration poller is active'() {
+    setup:
+    appSecConfigService.init()
+    AppSecSystem.active = false
+    config.getAppSecActivation() >> ProductActivation.ENABLED_INACTIVE
+
+    when:
+    appSecConfigService.maybeSubscribeConfigPolling()
+
+    then:
+    1 * poller.addListener(Product.ASM_SCA, AppSecSCAConfigDeserializer.INSTANCE, _)
+    1 * poller.addCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
+  }
+
+  void 'unsubscribes from ASM_SCA product on close'() {
+    setup:
+    appSecConfigService.init()
+    AppSecSystem.active = false
+    config.getAppSecActivation() >> ProductActivation.ENABLED_INACTIVE
+    appSecConfigService.maybeSubscribeConfigPolling()
+
+    when:
+    appSecConfigService.close()
+
+    then:
+    1 * poller.removeListeners(Product.ASM_SCA)
+    1 * poller.removeCapabilities({ it & datadog.remoteconfig.Capabilities.CAPABILITY_ASM_SCA_VULNERABILITY_DETECTION })
+  }
+
 
   private static AppSecFeatures autoUserInstrum(String mode) {
     return new AppSecFeatures().tap { features ->

dd-java-agent/appsec/src/test/groovy/com/datadog/appsec/config/AppSecSCAConfigDeserializerTest.groovy

@@ -0,0 +1,108 @@
+package com.datadog.appsec.config
+
+import spock.lang.Specification
+
+class AppSecSCAConfigDeserializerTest extends Specification {
+
+  def "deserializes valid JSON byte array"() {
+    given:
+    def json = '''
+      {
+        "enabled": true,
+        "instrumentation_targets": [
+          {
+            "class_name": "org/springframework/web/client/RestTemplate",
+            "method_name": "execute"
+          }
+        ]
+      }
+    '''
+    def bytes = json.bytes
+
+    when:
+    def config = AppSecSCAConfigDeserializer.INSTANCE.deserialize(bytes)
+
+    then:
+    config != null
+    config.enabled == true
+    config.instrumentationTargets.size() == 1
+    config.instrumentationTargets[0].className == "org/springframework/web/client/RestTemplate"
+    config.instrumentationTargets[0].methodName == "execute"
+  }
+
+  def "returns null for null content"() {
+    when:
+    def config = AppSecSCAConfigDeserializer.INSTANCE.deserialize(null)
+
+    then:
+    config == null
+  }
+
+  def "returns null for empty byte array"() {
+    when:
+    def config = AppSecSCAConfigDeserializer.INSTANCE.deserialize(new byte[0])
+
+    then:
+    config == null
+  }
+
+  def "deserializes minimal configuration"() {
+    given:
+    def json = '{"enabled": false}'
+    def bytes = json.bytes
+
+    when:
+    def config = AppSecSCAConfigDeserializer.INSTANCE.deserialize(bytes)
+
+    then:
+    config != null
+    config.enabled == false
+    config.instrumentationTargets == null
+  }
+
+  def "handles multiple instrumentation targets"() {
+    given:
+    def json = '''
+      {
+        "enabled": true,
+        "instrumentation_targets": [
+          {
+            "class_name": "com/example/Class1",
+            "method_name": "method1"
+          },
+          {
+            "class_name": "com/example/Class2",
+            "method_name": "method2"
+          },
+          {
+            "class_name": "com/example/Class3",
+            "method_name": "method3"
+          }
+        ]
+      }
+    '''
+    def bytes = json.bytes
+
+    when:
+    def config = AppSecSCAConfigDeserializer.INSTANCE.deserialize(bytes)
+
+    then:
+    config != null
+    config.enabled == true
+    config.instrumentationTargets.size() == 3
+
+    config.instrumentationTargets[0].className == "com/example/Class1"
+    config.instrumentationTargets[0].methodName == "method1"
+
+    config.instrumentationTargets[1].className == "com/example/Class2"
+    config.instrumentationTargets[1].methodName == "method2"
+
+    config.instrumentationTargets[2].className == "com/example/Class3"
+    config.instrumentationTargets[2].methodName == "method3"
+  }
+
+  def "INSTANCE is a singleton"() {
+    expect:
+    AppSecSCAConfigDeserializer.INSTANCE === AppSecSCAConfigDeserializer.INSTANCE
+  }
+}