Add license checking to the redact processor (#95477)

docs/changelog/95477.yaml

@@ -0,0 +1,5 @@
+pr: 95477
+summary: Add license checking to the redact processor
+area: Ingest Node
+type: feature
+issues: []

server/src/main/java/org/elasticsearch/ingest/IngestService.java

@@ -627,6 +627,14 @@ public class IngestService implements ClusterStateApplier, ReportingService<Inge
         Pipeline pipeline = Pipeline.create(pipelineId, pipelineConfig, processorFactories, scriptService);
         List<Exception> exceptions = new ArrayList<>();
         for (Processor processor : pipeline.flattenAllProcessors()) {
+
+            // run post-construction extra validation (if any, the default implementation from the Processor interface is a no-op)
+            try {
+                processor.extraValidation();
+            } catch (Exception e) {
+                exceptions.add(e);
+            }
+
             for (Map.Entry<DiscoveryNode, IngestInfo> entry : ingestInfos.entrySet()) {
                 String type = processor.getType();
                 if (entry.getValue().containsProcessor(type) == false && ConditionalProcessor.TYPE.equals(type) == false) {

server/src/main/java/org/elasticsearch/ingest/Processor.java

@@ -78,6 +78,20 @@ public interface Processor {
         return false;
     }
 
+    /**
+     * Validate a processor after it has been constructed by a factory.
+     *
+     * Override this method to perform additional post-construction validation that should be performed at the rest/transport level.
+     * If there's an issue with the processor, then indicate that by throwing an exception. See
+     * {@link IngestService#validatePipeline(Map, String, Map)}} for the call site where there is invoked in a try/catch.
+     *
+     * An example of where this would be needed is a processor that interacts with external state like the license state -- it may
+     * be okay to create that processor on day 1 with license state A, but later illegal to create a similar processor on day 2 with
+     * state B. We want to reject put requests on day 2 (at the rest/transport level), but still allow for restarting nodes in the
+     * cluster (so we can't throw exceptions from {@link Factory#create(Map, String, String, Map)}).
+     */
+    default void extraValidation() throws Exception {}
+
     /**
      * A factory that knows how to construct a processor based on a map of maps.
      */

x-pack/plugin/core/src/main/java/org/elasticsearch/license/XPackLicenseState.java

@@ -81,6 +81,7 @@ public class XPackLicenseState {
                 "The CCR monitoring endpoint will be blocked",
                 "Existing follower indices will continue to replicate data" }
         );
+        messages.put(XPackField.REDACT_PROCESSOR, new String[] { "Executing a redact processor in an ingest pipeline will fail." });
         EXPIRATION_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -101,6 +102,7 @@ public class XPackLicenseState {
         messages.put(XPackField.SQL, XPackLicenseState::sqlAcknowledgementMessages);
         messages.put(XPackField.CCR, XPackLicenseState::ccrAcknowledgementMessages);
         messages.put(XPackField.ENTERPRISE_SEARCH, XPackLicenseState::enterpriseSearchAcknowledgementMessages);
+        messages.put(XPackField.REDACT_PROCESSOR, XPackLicenseState::redactProcessorAcknowledgementMessages);
         ACKNOWLEDGMENT_MESSAGES = Collections.unmodifiableMap(messages);
     }
 
@@ -304,6 +306,22 @@ public class XPackLicenseState {
         return Strings.EMPTY_ARRAY;
     }
 
+    private static String[] redactProcessorAcknowledgementMessages(OperationMode currentMode, OperationMode newMode) {
+        switch (newMode) {
+            case BASIC:
+            case STANDARD:
+            case GOLD:
+                switch (currentMode) {
+                    case TRIAL:
+                    case PLATINUM:
+                    case ENTERPRISE:
+                        return new String[] { "Redact ingest pipeline processors will be disabled" };
+                }
+                break;
+        }
+        return Strings.EMPTY_ARRAY;
+    }
+
     private static boolean isBasic(OperationMode mode) {
         return mode == OperationMode.BASIC;
     }

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/XPackField.java

@@ -79,6 +79,9 @@ public final class XPackField {
     public static final String ENTERPRISE_SEARCH = "enterprise_search";
     public static final String REMOTE_CLUSTERS = "remote_clusters";
 
+    /** Name constant for the redact processor feature. */
+    public static final String REDACT_PROCESSOR = "redact_processor";
+
     private XPackField() {}
 
     public static String featureSettingPrefix(String featureName) {

x-pack/plugin/redact/src/main/java/org/elasticsearch/xpack/redact/RedactPlugin.java

@@ -25,7 +25,7 @@ public class RedactPlugin extends Plugin implements IngestPlugin {
 
     @Override
     public Map<String, Processor.Factory> getProcessors(Processor.Parameters parameters) {
-        return Map.of(RedactProcessor.TYPE, new RedactProcessor.Factory(parameters.matcherWatchdog));
+        return Map.of(RedactProcessor.TYPE, new RedactProcessor.Factory(getLicenseState(), parameters.matcherWatchdog));
     }
 
     protected XPackLicenseState getLicenseState() {

x-pack/plugin/redact/src/main/java/org/elasticsearch/xpack/redact/RedactProcessor.java

@@ -18,6 +18,11 @@ import org.elasticsearch.ingest.AbstractProcessor;
 import org.elasticsearch.ingest.ConfigurationUtils;
 import org.elasticsearch.ingest.IngestDocument;
 import org.elasticsearch.ingest.Processor;
+import org.elasticsearch.license.License;
+import org.elasticsearch.license.LicenseUtils;
+import org.elasticsearch.license.LicensedFeature;
+import org.elasticsearch.license.XPackLicenseState;
+import org.elasticsearch.xpack.core.XPackField;
 import org.joni.Matcher;
 import org.joni.Option;
 import org.joni.Region;
@@ -37,6 +42,12 @@ import static org.elasticsearch.ingest.ConfigurationUtils.newConfigurationExcept
  */
 public class RedactProcessor extends AbstractProcessor {
 
+    public static final LicensedFeature.Momentary REDACT_PROCESSOR_FEATURE = LicensedFeature.momentary(
+        null,
+        XPackField.REDACT_PROCESSOR,
+        License.OperationMode.PLATINUM
+    );
+
     public static final String TYPE = "redact";
 
     private static final Logger logger = LogManager.getLogger(RedactProcessor.class);
@@ -47,9 +58,13 @@ public class RedactProcessor extends AbstractProcessor {
     private final String redactField;
     private final List<Grok> groks;
     private final boolean ignoreMissing;
+
     private final String redactedStartToken;
     private final String redactedEndToken;
 
+    private final XPackLicenseState licenseState;
+    private final boolean skipIfUnlicensed;
+
     RedactProcessor(
         String tag,
         String description,
@@ -59,7 +74,9 @@ public class RedactProcessor extends AbstractProcessor {
         boolean ignoreMissing,
         String redactedStartToken,
         String redactedEndToken,
-        MatcherWatchdog matcherWatchdog
+        MatcherWatchdog matcherWatchdog,
+        XPackLicenseState licenseState,
+        boolean skipIfUnlicensed
     ) {
         super(tag, description);
         this.redactField = redactField;
@@ -75,10 +92,30 @@ public class RedactProcessor extends AbstractProcessor {
         if (matchPatterns.isEmpty() == false) {
             new Grok(patternBank, matchPatterns.get(0), matcherWatchdog, logger::warn).match("___nomatch___");
         }
+        this.licenseState = licenseState;
+        this.skipIfUnlicensed = skipIfUnlicensed;
+    }
+
+    @Override
+    public void extraValidation() throws Exception {
+        // post-creation license check, this is exercised at rest/transport time
+        if (skipIfUnlicensed == false && REDACT_PROCESSOR_FEATURE.check(licenseState) == false) {
+            String message = LicenseUtils.newComplianceException(REDACT_PROCESSOR_FEATURE.getName()).getMessage();
+            throw newConfigurationException(TYPE, tag, "skip_if_unlicensed", message);
+        }
     }
 
     @Override
     public IngestDocument execute(IngestDocument ingestDocument) {
+        // runtime license check, this runs for every document processed
+        if (REDACT_PROCESSOR_FEATURE.check(licenseState) == false) {
+            if (skipIfUnlicensed) {
+                return ingestDocument;
+            } else {
+                throw LicenseUtils.newComplianceException(REDACT_PROCESSOR_FEATURE.getName());
+            }
+        }
+
         // Call with ignoreMissing = true so getFieldValue does not throw
         final String fieldValue = ingestDocument.getFieldValue(redactField, String.class, true);
 
@@ -107,6 +144,10 @@ public class RedactProcessor extends AbstractProcessor {
         return groks;
     }
 
+    boolean getSkipIfUnlicensed() {
+        return skipIfUnlicensed;
+    }
+
     // exposed for testing
     static String matchRedact(String fieldValue, List<Grok> groks) {
         return matchRedact(fieldValue, groks, DEFAULT_REDACTED_START, DEFAULT_REDACTED_END);
@@ -325,9 +366,11 @@ public class RedactProcessor extends AbstractProcessor {
 
     public static final class Factory implements Processor.Factory {
 
+        private final XPackLicenseState licenseState;
         private final MatcherWatchdog matcherWatchdog;
 
-        public Factory(MatcherWatchdog matcherWatchdog) {
+        public Factory(XPackLicenseState licenseState, MatcherWatchdog matcherWatchdog) {
+            this.licenseState = licenseState;
             this.matcherWatchdog = matcherWatchdog;
         }
 
@@ -341,6 +384,7 @@ public class RedactProcessor extends AbstractProcessor {
             String matchField = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "field");
             List<String> matchPatterns = ConfigurationUtils.readList(TYPE, processorTag, config, "patterns");
             boolean ignoreMissing = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "ignore_missing", true);
+            boolean skipIfUnlicensed = ConfigurationUtils.readBooleanProperty(TYPE, processorTag, config, "skip_if_unlicensed", true);
 
             String redactStart = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "prefix", DEFAULT_REDACTED_START);
             String redactEnd = ConfigurationUtils.readStringProperty(TYPE, processorTag, config, "suffix", DEFAULT_REDACTED_END);
@@ -360,7 +404,9 @@ public class RedactProcessor extends AbstractProcessor {
                     ignoreMissing,
                     redactStart,
                     redactEnd,
-                    matcherWatchdog
+                    matcherWatchdog,
+                    licenseState,
+                    skipIfUnlicensed
                 );
             } catch (Exception e) {
                 throw newConfigurationException(

x-pack/plugin/redact/src/test/java/org/elasticsearch/xpack/redact/RedactProcessorFactoryTests.java

@@ -8,6 +8,9 @@ package org.elasticsearch.xpack.redact;
 
 import org.elasticsearch.ElasticsearchException;
 import org.elasticsearch.grok.MatcherWatchdog;
+import org.elasticsearch.license.MockLicenseState;
+import org.elasticsearch.license.TestUtils;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.HashMap;
@@ -19,11 +22,24 @@ import static org.hamcrest.Matchers.empty;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.not;
+import static org.mockito.Mockito.when;
 
 public class RedactProcessorFactoryTests extends ESTestCase {
 
+    private static XPackLicenseState mockLicenseState() {
+        MockLicenseState licenseState = TestUtils.newMockLicenceState();
+        when(licenseState.isAllowed(RedactProcessor.REDACT_PROCESSOR_FEATURE)).thenReturn(true);
+        return licenseState;
+    }
+
+    private static XPackLicenseState mockNotAllowedLicenseState() {
+        MockLicenseState licenseState = TestUtils.newMockLicenceState();
+        when(licenseState.isAllowed(RedactProcessor.REDACT_PROCESSOR_FEATURE)).thenReturn(false);
+        return licenseState;
+    }
+
     public void testPatternNotSet() {
-        RedactProcessor.Factory factory = new RedactProcessor.Factory(MatcherWatchdog.noop());
+        RedactProcessor.Factory factory = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop());
 
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
@@ -33,7 +49,7 @@ public class RedactProcessorFactoryTests extends ESTestCase {
     }
 
     public void testCreateWithCustomPatterns() throws Exception {
-        RedactProcessor.Factory factory = new RedactProcessor.Factory(MatcherWatchdog.noop());
+        RedactProcessor.Factory factory = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop());
 
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
@@ -45,7 +61,7 @@ public class RedactProcessorFactoryTests extends ESTestCase {
     }
 
     public void testConfigKeysRemoved() throws Exception {
-        RedactProcessor.Factory factory = new RedactProcessor.Factory(MatcherWatchdog.noop());
+        RedactProcessor.Factory factory = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop());
 
         Map<String, Object> config = new HashMap<>();
         config.put("field", "_field");
@@ -58,4 +74,49 @@ public class RedactProcessorFactoryTests extends ESTestCase {
         assertThat(config.entrySet(), hasSize(1));
         assertEquals("unused", config.get("extra"));
     }
+
+    public void testSkipIfUnlicensed() throws Exception {
+        {
+            Map<String, Object> config = new HashMap<>();
+            config.put("field", "_field");
+            config.put("patterns", List.of("%{MY_PATTERN:name}!"));
+            config.put("pattern_definitions", Map.of("MY_PATTERN", "foo"));
+
+            // the default value for skip_if_unlicensed is true, and so the license state doesn't matter
+            XPackLicenseState licenseState = randomBoolean() ? mockLicenseState() : mockNotAllowedLicenseState();
+            RedactProcessor.Factory factory = new RedactProcessor.Factory(licenseState, MatcherWatchdog.noop());
+            RedactProcessor processor = factory.create(null, null, null, config);
+            processor.extraValidation();
+            assertThat(processor.getSkipIfUnlicensed(), equalTo(true));
+        }
+
+        {
+            Map<String, Object> config = new HashMap<>();
+            config.put("field", "_field");
+            config.put("patterns", List.of("%{MY_PATTERN:name}!"));
+            config.put("pattern_definitions", Map.of("MY_PATTERN", "foo"));
+
+            // but it can be set to false if you wish, in which case the license check must pass
+            config.put("skip_if_unlicensed", false);
+            RedactProcessor.Factory factory = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop());
+            RedactProcessor processor = factory.create(null, null, null, config);
+            processor.extraValidation();
+            assertThat(processor.getSkipIfUnlicensed(), equalTo(false));
+        }
+
+        {
+            Map<String, Object> config = new HashMap<>();
+            config.put("field", "_field");
+            config.put("patterns", List.of("%{MY_PATTERN:name}!"));
+            config.put("pattern_definitions", Map.of("MY_PATTERN", "foo"));
+
+            // if skip_if_unlicensed is false, then the license must allow for redact to be used in order to pass the extra validation
+            config.put("skip_if_unlicensed", false);
+            RedactProcessor.Factory factory = new RedactProcessor.Factory(mockNotAllowedLicenseState(), MatcherWatchdog.noop());
+            RedactProcessor processor = factory.create(null, null, null, config);
+            ElasticsearchException e = expectThrows(ElasticsearchException.class, () -> processor.extraValidation());
+            assertThat(e.getMessage(), containsString("[skip_if_unlicensed] current license is non-compliant for [redact_processor]"));
+        }
+    }
+
 }

x-pack/plugin/redact/src/test/java/org/elasticsearch/xpack/redact/RedactProcessorTests.java

@@ -6,9 +6,14 @@
  */
 package org.elasticsearch.xpack.redact;
 
+import org.elasticsearch.ElasticsearchSecurityException;
+import org.elasticsearch.grok.GrokBuiltinPatterns;
 import org.elasticsearch.grok.MatcherWatchdog;
 import org.elasticsearch.index.VersionType;
 import org.elasticsearch.ingest.IngestDocument;
+import org.elasticsearch.license.MockLicenseState;
+import org.elasticsearch.license.TestUtils;
+import org.elasticsearch.license.XPackLicenseState;
 import org.elasticsearch.test.ESTestCase;
 
 import java.util.Arrays;
@@ -18,17 +23,31 @@ import java.util.Map;
 
 import static org.hamcrest.Matchers.contains;
 import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
 import static org.hamcrest.Matchers.sameInstance;
+import static org.mockito.Mockito.when;
 
 public class RedactProcessorTests extends ESTestCase {
 
+    private static XPackLicenseState mockLicenseState() {
+        MockLicenseState licenseState = TestUtils.newMockLicenceState();
+        when(licenseState.isAllowed(RedactProcessor.REDACT_PROCESSOR_FEATURE)).thenReturn(true);
+        return licenseState;
+    }
+
+    private static XPackLicenseState mockNotAllowedLicenseState() {
+        MockLicenseState licenseState = TestUtils.newMockLicenceState();
+        when(licenseState.isAllowed(RedactProcessor.REDACT_PROCESSOR_FEATURE)).thenReturn(false);
+        return licenseState;
+    }
+
     public void testMatchRedact() throws Exception {
         {
             var config = new HashMap<String, Object>();
             config.put("field", "to_redact");
             config.put("patterns", List.of("%{EMAILADDRESS:EMAIL}"));
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var groks = processor.getGroks();
 
             {
@@ -52,7 +71,7 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("field", "to_redact");
             config.put("patterns", List.of("%{CREDIT_CARD:CREDIT_CARD}"));
             config.put("pattern_definitions", Map.of("CREDIT_CARD", "\\b(?:\\d[ -]*?){13,16}\\b"));
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var groks = processor.getGroks();
 
             {
@@ -82,7 +101,7 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("field", "to_redact");
             config.put("patterns", List.of("%{CREDIT_CARD:CREDIT_CARD}"));
             config.put("pattern_definitions", Map.of("CREDIT_CARD", "\\d{4}[ -]\\d{4}[ -]\\d{4}[ -]\\d{4}"));
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var grok = processor.getGroks().get(0);
 
             String input = "1001-1002-1003-1004 2001-1002-1003-1004 3001-1002-1003-1004 4001-1002-1003-1004";
@@ -96,7 +115,7 @@ public class RedactProcessorTests extends ESTestCase {
         config.put("field", "to_redact");
         config.put("patterns", List.of("%{EMAILADDRESS:EMAIL}", "%{IP:IP_ADDRESS}", "%{CREDIT_CARD:CREDIT_CARD}"));
         config.put("pattern_definitions", Map.of("CREDIT_CARD", "\\d{4}[ -]\\d{4}[ -]\\d{4}[ -]\\d{4}"));
-        var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+        var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
         var groks = processor.getGroks();
 
         {
@@ -111,7 +130,7 @@ public class RedactProcessorTests extends ESTestCase {
         config.put("field", "to_redact");
         config.put("patterns", List.of("%{EMAILADDRESS:EMAIL}", "%{IP:IP_ADDRESS}", "%{CREDIT_CARD:CREDIT_CARD}"));
         config.put("pattern_definitions", Map.of("CREDIT_CARD", "\\d{4}[ -]\\d{4}[ -]\\d{4}[ -]\\d{4}"));
-        var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+        var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
 
         {
             var ingestDoc = createIngestDoc(Map.of("to_redact", "This is ok nothing to redact"));
@@ -140,7 +159,7 @@ public class RedactProcessorTests extends ESTestCase {
         config.put("field", "to_redact");
         config.put("patterns", List.of("%{EMAILADDRESS:REDACTED}", "%{IP:REDACTED}", "%{CREDIT_CARD:REDACTED}"));
         config.put("pattern_definitions", Map.of("CREDIT_CARD", "\\d{4}[ -]\\d{4}[ -]\\d{4}[ -]\\d{4}"));
-        var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+        var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
 
         {
             var ingestDoc = createIngestDoc(Map.of("to_redact", "look a credit card number! 0001-0002-0003-0004 from david@email.com"));
@@ -157,7 +176,7 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("prefix", "?--");
             config.put("suffix", "}");
 
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var ingestDoc = createIngestDoc(Map.of("to_redact", "0.0.0.1 will be redacted"));
             var redacted = processor.execute(ingestDoc);
             assertEquals("?--IP_ADDRESS} will be redacted", redacted.getFieldValue("to_redact", String.class));
@@ -168,7 +187,7 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("patterns", List.of("%{IP:IP_ADDRESS}"));
             config.put("prefix", "?--");
 
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var ingestDoc = createIngestDoc(Map.of("to_redact", "0.0.0.1 will be redacted"));
             var redacted = processor.execute(ingestDoc);
             assertEquals("?--IP_ADDRESS> will be redacted", redacted.getFieldValue("to_redact", String.class));
@@ -179,7 +198,7 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("patterns", List.of("%{IP:IP_ADDRESS}"));
             config.put("suffix", "++");
 
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var ingestDoc = createIngestDoc(Map.of("to_redact", "0.0.0.1 will be redacted"));
             var redacted = processor.execute(ingestDoc);
             assertEquals("<IP_ADDRESS++ will be redacted", redacted.getFieldValue("to_redact", String.class));
@@ -191,7 +210,7 @@ public class RedactProcessorTests extends ESTestCase {
             var config = new HashMap<String, Object>();
             config.put("field", "to_redact");
             config.put("patterns", List.of("foo"));
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var ingestDoc = createIngestDoc(Map.of("not_the_field", "fieldValue"));
             var processed = processor.execute(ingestDoc);
             assertThat(ingestDoc, sameInstance(processed));
@@ -203,13 +222,56 @@ public class RedactProcessorTests extends ESTestCase {
             config.put("patterns", List.of("foo"));
             config.put("ignore_missing", false);   // this time the missing field should error
 
-            var processor = new RedactProcessor.Factory(MatcherWatchdog.noop()).create(null, "t", "d", config);
+            var processor = new RedactProcessor.Factory(mockLicenseState(), MatcherWatchdog.noop()).create(null, "t", "d", config);
             var ingestDoc = createIngestDoc(Map.of("not_the_field", "fieldValue"));
             IllegalArgumentException e = expectThrows(IllegalArgumentException.class, () -> processor.execute(ingestDoc));
             assertThat(e.getMessage(), containsString("field [to_redact] is null or missing"));
         }
     }
 
+    public void testLicenseChecks() throws Exception {
+        var notAllowed = mockNotAllowedLicenseState();
+        {
+            var config = new HashMap<String, Object>();
+            config.put("field", "to_redact");
+            config.put("patterns", List.of("foo"));
+            config.put("ignore_missing", false); // usually, this would throw, but here it doesn't because of the license check
+            if (randomBoolean()) {
+                config.put("skip_if_unlicensed", true); // set the value to true (versus just using the default, also true)
+            }
+            var processor = new RedactProcessor.Factory(notAllowed, MatcherWatchdog.noop()).create(null, "t", "d", config);
+            assertThat(processor.getSkipIfUnlicensed(), equalTo(true));
+            var ingestDoc = createIngestDoc(Map.of("not_the_field", "fieldValue"));
+
+            // since skip_if_unlicensed is true, the same document is returned to us unchanged
+            var processed = processor.execute(ingestDoc);
+            assertThat(ingestDoc, sameInstance(processed));
+            assertEquals(ingestDoc, processed);
+        }
+        {
+            // bypassing the factory, because it won't construct a processor under these circumstances
+            var processor = new RedactProcessor(
+                "t",
+                "d",
+                GrokBuiltinPatterns.ecsV1Patterns(),
+                List.of("foo"),
+                "to_redact",
+                false, // set ignore_missing to false. usually, this would throw, but here it doesn't because of the license check
+                "<",
+                ">",
+                MatcherWatchdog.noop(),
+                notAllowed,
+                false // set skip_if_unlicensed to false, we do not want to skip, we do want to fail
+            );
+            assertThat(processor.getSkipIfUnlicensed(), equalTo(false));
+            var ingestDoc = createIngestDoc(Map.of("not_the_field", "fieldValue"));
+
+            // since skip_if_unlicensed is false, and the license is not sufficient, we throw on execute
+            ElasticsearchSecurityException e = expectThrows(ElasticsearchSecurityException.class, () -> processor.execute(ingestDoc));
+            assertThat(e.getMessage(), containsString("current license is non-compliant for [redact_processor]"));
+        }
+    }
+
     public void testMergeLongestRegion() {
         var r = List.of(
             new RedactProcessor.RegionTrackingMatchExtractor.Replacement(10, 20, "first"),