[CDAP-21270] Handle JEXL security violations and fail pipeline immediately

wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java

@@ -30,6 +30,7 @@
 import io.cdap.wrangler.api.ReportErrorAndProceed;
 import io.cdap.wrangler.api.Row;
 import io.cdap.wrangler.api.TransientVariableScope;
+import io.cdap.wrangler.expression.ELPermissionException;
 import io.cdap.wrangler.schema.DirectiveOutputSchemaGenerator;
 import io.cdap.wrangler.schema.DirectiveSchemaResolutionContext;
 import io.cdap.wrangler.schema.TransientStoreKeys;
@@ -159,7 +160,7 @@ public List<Row> execute(List<Row> rows) throws RecipeException {
         }
         ++i;
       }
-    } catch (DirectiveExecutionException e) {
+    } catch (DirectiveExecutionException | ELPermissionException e) {
       throw new RecipeException(e.getMessage(), e, i, directiveIndex);
     }
     // Schema generation

wrangler-core/src/main/java/io/cdap/wrangler/expression/EL.java

@@ -66,6 +66,7 @@ public final class EL {
 
   private final Set<String> variables;
   private final JexlScript script;
+  private final boolean allowlistEnabled;
 
   /**
    * Returns {@code true} if this class has been used to execute JEXL script.
@@ -105,6 +106,7 @@ public static void reset() {
    * @param expression to be compiled.
    * @return a compiled {@link EL} object
    * @throws ELException if failed to compile
+   * @throws ELPermissionException if the expression violates the JEXL sandbox permission rules
    */
   public static EL compile(String expression) throws ELException {
     return compile(new DefaultFunctions(), expression);
@@ -120,12 +122,14 @@ public static EL compile(String expression) throws ELException {
    * @param expression   to be compiled.
    * @return a compiled {@link EL} object
    * @throws ELException if failed to compile
+   * @throws ELPermissionException if the expression violates the JEXL sandbox permission rules
    */
   public static EL compile(final ELRegistration registration, final String expression)
       throws ELException {
     used = true;
     CompileOptions options = COMPILE_OPTIONS.get();
-    JexlSandbox sandbox = createSandbox(options.getJexlAllowlist(), options.isAllowlistEnabled());
+    boolean allowlistEnabled = options.isAllowlistEnabled();
+    JexlSandbox sandbox = createSandbox(options.getJexlAllowlist(), allowlistEnabled);
     JexlEngine engine = new JexlBuilder()
       .sandbox(sandbox)
       .namespaces(registration.functions())
@@ -138,9 +142,9 @@ public static EL compile(final ELRegistration registration, final String express
     try {
       JexlScript script = engine.createScript(expression);
       Set<String> variables = extractVariables(script);
-      return new EL(script, variables);
+      return new EL(script, variables, allowlistEnabled);
     } catch (Exception e) {
-      throw handleException(e, expression);
+      throw handleException(e, expression, allowlistEnabled);
     }
   }
 
@@ -156,27 +160,53 @@ private static Set<String> extractVariables(final JexlScript script) {
         .collect(Collectors.toSet());
   }
 
-  private static ELException handleException(Exception e, String expression) {
+  private static ELException handleException(
+      Exception e, String expression, boolean allowlistEnabled) {
     if (e instanceof JexlException) {
-      JexlException jexlException = (JexlException) e;
-      JexlInfo info = jexlException.getInfo();
-      return new ELException(
-          String.format("Error encountered while evaluating '%s', at line '%d' and column '%d'. " +
-              "Make sure the JEXL transformation is valid and uses only allowlisted classes, methods, and properties.",
-              info == null || info.getDetail() == null ? expression : info.getDetail(),
-              info == null ? 0 : info.getLine(), info == null ? 0 : info.getColumn()),
-          e);
-    } else if (e instanceof NumberFormatException) {
-      return new ELException("Type mismatch. Change type of constant " +
-                              "or convert to right data type using conversion functions available. Reason : "
-                              + e.getMessage(), e);
-    } else {
-      if (e.getCause() != null) {
-        return new ELException(e.getCause().getMessage(), e);
-      } else {
-        return new ELException(e);
-      }
+      return handleJexlException((JexlException) e, expression, allowlistEnabled);
     }
+
+    if (e instanceof NumberFormatException) {
+      return new ELException("Type mismatch. Change type of constant "
+          + "or convert to right data type using conversion functions available. Reason : "
+          + e.getMessage(), e);
+    }
+
+    return e.getCause() != null
+        ? new ELException(e.getCause().getMessage(), e)
+        : new ELException(e);
+  }
+
+  /**
+   * Translates a {@link JexlException} into an {@link ELException}, or throws an
+   * {@link ELPermissionException} if the exception represents a JEXL sandbox permission violation.
+   *
+   * @param e                the JEXL exception to handle
+   * @param expression       the source expression being evaluated
+   * @param allowlistEnabled whether the JEXL allowlist sandbox is enabled
+   * @return an {@link ELException} describing the evaluation error
+   * @throws ELPermissionException if the exception is caused by a JEXL sandbox permission violation
+   */
+  private static ELException handleJexlException(
+      JexlException e, String expression, boolean allowlistEnabled) {
+    JexlInfo info = e.getInfo();
+    Object detail = (info == null || info.getDetail() == null) ? expression : info.getDetail();
+    int line = info == null ? 0 : info.getLine();
+    int column = info == null ? 0 : info.getColumn();
+    String baseErrorMessage = String.format(
+        "Error encountered while evaluating '%s', at line '%d' and column '%d'.",
+        detail, line, column);
+
+    if (isJexlPermissionViolation(e, allowlistEnabled)) {
+      throw new ELPermissionException(baseErrorMessage, e);
+    }
+
+    return new ELException(baseErrorMessage, e);
+  }
+
+  private static boolean isJexlPermissionViolation(JexlException e, boolean allowlistEnabled) {
+    return allowlistEnabled
+        && (e instanceof JexlException.Method || e instanceof JexlException.Property);
   }
 
   /**
@@ -241,9 +271,10 @@ private static void applyPropertyPermissions(JexlAllowlist rule, JexlSandbox.Per
     }
   }
 
-  private EL(JexlScript script, Set<String> variables) {
+  private EL(JexlScript script, Set<String> variables, boolean allowlistEnabled) {
     this.script = script;
     this.variables = Collections.unmodifiableSet(variables);
+    this.allowlistEnabled = allowlistEnabled;
   }
 
   public Set<String> variables() {
@@ -254,6 +285,14 @@ public String getScriptParsedText() {
     return script.getParsedText();
   }
 
+  /**
+   * Executes the compiled JEXL script against the given {@link ELContext}.
+   *
+   * @param context the execution context containing variable bindings
+   * @return the {@link ELResult} holding the evaluated value
+   * @throws ELException         if an error occurs while evaluating the expression
+   * @throws ELPermissionException if the expression violates JEXL sandbox permissions
+   */
   public ELResult execute(ELContext context) throws ELException {
     try {
       // Null the missing fields
@@ -265,7 +304,7 @@ public ELResult execute(ELContext context) throws ELException {
       Object value = script.execute(context);
       return new ELResult(value);
     } catch (Exception e) {
-      throw handleException(e, script.getSourceText());
+      throw handleException(e, script.getSourceText(), allowlistEnabled);
     }
   }
 

wrangler-core/src/main/java/io/cdap/wrangler/expression/ELPermissionException.java

@@ -0,0 +1,34 @@
+/*
+ *  Copyright © 2026 Cask Data, Inc.
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ *  use this file except in compliance with the License. You may obtain a copy of
+ *  the License at
+ *
+ *  http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ *  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ *  License for the specific language governing permissions and limitations under
+ *  the License.
+ */
+
+package io.cdap.wrangler.expression;
+
+/**
+ * Exception thrown when a JEXL expression violates sandbox permissions
+ * (e.g., accessing unallowlisted classes, methods, or properties).
+ */
+public class ELPermissionException extends RuntimeException {
+
+  /**
+   * Creates an {@link ELPermissionException} with the specified error message and underlying cause.
+   *
+   * @param message the detail message describing the permission violation
+   * @param cause   the cause of the permission violation
+   */
+  public ELPermissionException(String message, Throwable cause) {
+    super(message, cause);
+  }
+}

wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java

@@ -20,6 +20,7 @@
 import io.cdap.wrangler.api.CompileException;
 import io.cdap.wrangler.api.CompileStatus;
 import io.cdap.wrangler.api.Compiler;
+import io.cdap.wrangler.api.DirectiveContext;
 import io.cdap.wrangler.api.DirectiveLoadException;
 import io.cdap.wrangler.api.DirectiveNotFoundException;
 import io.cdap.wrangler.api.DirectiveParseException;
@@ -35,6 +36,7 @@
 import io.cdap.wrangler.executor.RecipePipelineExecutor;
 import io.cdap.wrangler.parser.GrammarBasedParser;
 import io.cdap.wrangler.parser.MigrateToV2;
+import io.cdap.wrangler.parser.NoOpDirectiveContext;
 import io.cdap.wrangler.parser.RecipeCompiler;
 import io.cdap.wrangler.proto.Contexts;
 import io.cdap.wrangler.registry.CompositeDirectiveRegistry;
@@ -119,13 +121,18 @@ public static Pair<List<Row>, List<Row>> executeWithErrors(String[] recipe, List
   }
 
   public static RecipePipeline execute(String[] recipe)
+    throws RecipeException, DirectiveParseException, DirectiveLoadException, DirectiveNotFoundException {
+    return execute(recipe, new NoOpDirectiveContext());
+  }
+
+  public static RecipePipeline execute(String[] recipe, DirectiveContext directiveContext)
     throws RecipeException, DirectiveParseException, DirectiveLoadException, DirectiveNotFoundException {
     CompositeDirectiveRegistry registry = new CompositeDirectiveRegistry(
       SystemDirectiveRegistry.INSTANCE
     );
 
     String migrate = new MigrateToV2(recipe).migrate();
-    RecipeParser parser = new GrammarBasedParser(Contexts.SYSTEM, migrate, registry);
+    RecipeParser parser = new GrammarBasedParser(Contexts.SYSTEM, migrate, registry, directiveContext);
     return new RecipePipelineExecutor(parser, new TestingPipelineContext());
   }
 

wrangler-core/src/test/java/io/cdap/wrangler/executor/RecipePipelineExecutorTest.java

@@ -19,12 +19,21 @@
 import io.cdap.cdap.api.data.format.StructuredRecord;
 import io.cdap.cdap.api.data.schema.Schema;
 import io.cdap.wrangler.TestingRig;
+import io.cdap.wrangler.api.DirectiveConfig;
+import io.cdap.wrangler.api.DirectiveContext;
+import io.cdap.wrangler.api.ErrorRecord;
+import io.cdap.wrangler.api.JexlConfiguration;
+import io.cdap.wrangler.api.RecipeException;
 import io.cdap.wrangler.api.RecipePipeline;
 import io.cdap.wrangler.api.Row;
+import io.cdap.wrangler.expression.ELPermissionException;
+import io.cdap.wrangler.parser.ConfigDirectiveContext;
 import org.junit.Assert;
 import org.junit.Test;
 
+import java.util.Arrays;
 import java.util.Collections;
+import java.util.List;
 
 /**
  * Tests {@link RecipePipelineExecutor}.
@@ -96,4 +105,45 @@ public void testPipelineWithMoreSimpleTypes() throws Exception {
     Assert.assertEquals(1481666448L, record.<Long>get("timestamp").longValue());
     Assert.assertEquals(186.66f, record.get("weight"), 0.0001f);
   }
+
+  @Test
+  public void testELPermissionExceptionFailsOnFirstRow() throws Exception {
+    DirectiveConfig config =
+        new DirectiveConfig(null, null, new JexlConfiguration(true, Collections.emptyList()));
+    DirectiveContext directiveContext = new ConfigDirectiveContext(config, true);
+    RecipePipeline<Row, StructuredRecord, ErrorRecord> pipeline =
+        TestingRig.execute(
+            new String[] {"send-to-error exp:{ a.toUpperCase() == 'VAL1' }"}, directiveContext);
+    try {
+      pipeline.execute(Arrays.asList(new Row("a", "val1"), new Row("a", "val2")));
+      Assert.fail("Expected RecipeException on first row when JEXL permission violation occurs");
+    } catch (RecipeException e) {
+      Assert.assertEquals(0, e.getRowIndex());
+      Assert.assertEquals(1, e.getDirectiveIndex());
+      Assert.assertTrue(e.getCause() instanceof ELPermissionException);
+      Assert.assertTrue(pipeline.errors().isEmpty());
+    }
+  }
+
+  @Test
+  public void testRowExceptionsProceedThroughAllRows() throws Exception {
+    DirectiveConfig config =
+        new DirectiveConfig(null, null, new JexlConfiguration(true, Collections.emptyList()));
+    DirectiveContext directiveContext = new ConfigDirectiveContext(config, true);
+    RecipePipeline<Row, StructuredRecord, ErrorRecord> pipeline =
+        TestingRig.execute(
+            new String[] {
+              "send-to-error exp:{ a == 'val1' }",
+              "send-to-error-and-continue exp:{ a == 'val2' }"
+            },
+            directiveContext);
+    List<Row> results =
+        pipeline.execute(
+            Arrays.asList(new Row("a", "val1"), new Row("a", "val2"), new Row("a", "val3")));
+    Assert.assertEquals(1, results.size());
+    Assert.assertEquals("val3", results.get(0).getValue("a"));
+    Assert.assertEquals(2, pipeline.errors().size());
+    Assert.assertEquals("val1", pipeline.errors().get(0).getRow().getValue("a"));
+    Assert.assertEquals("val2", pipeline.errors().get(1).getRow().getValue("a"));
+  }
 }

wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java

@@ -502,13 +502,7 @@ && checkPreconditionNotEmpty(false)) {
       }
     } catch (Exception e) {
       getContext().getMetrics().count("failure", 1);
-      if (onErrorStrategy.equalsIgnoreCase(ON_ERROR_PROCEED)) {
-        // Emit error record, if the Error flattener or error handlers are not connected, then
-        // the record is automatically omitted.
-        emitter.emitError(new InvalidEntry<>(0, e.getMessage(), input));
-        return;
-      }
-      if (onErrorStrategy.equalsIgnoreCase(ON_ERROR_FAIL_PIPELINE)) {
+      if (shouldFailPipeline(e)) {
         emitter.emitAlert(ImmutableMap.of(
           "stage", getContext().getStageName(),
           "code", String.valueOf(1),
@@ -523,6 +517,11 @@ && checkPreconditionNotEmpty(false)) {
         throw WranglerErrorUtil.getProgramFailureExceptionDetailsFromChain(e, errorReason,
             errorMessage, ErrorType.UNKNOWN);
       }
+      if (ON_ERROR_PROCEED.equalsIgnoreCase(onErrorStrategy)) {
+        // Emit error record, if the Error flattener or error handlers are not connected, then
+        // the record is automatically omitted.
+        emitter.emitError(new InvalidEntry<>(0, e.getMessage(), input));
+      }
       // If it's 'skip-on-error' we continue processing and don't emit any error records.
       return;
     } finally {
@@ -551,6 +550,18 @@ && checkPreconditionNotEmpty(false)) {
     }
   }
 
+  /**
+   * Determines whether the pipeline should fail immediately for the given exception.
+   * Critical exceptions always fail the pipeline regardless of the configured on-error strategy.
+   *
+   * @param e the exception thrown during record transformation
+   * @return {@code true} if the pipeline should fail immediately, {@code false} otherwise
+   */
+  private boolean shouldFailPipeline(Exception e) {
+    return WranglerErrorUtil.isCriticalException(e)
+        || ON_ERROR_FAIL_PIPELINE.equalsIgnoreCase(onErrorStrategy);
+  }
+
   /**
    * Validates input schema.
    *

wrangler-transform/src/main/java/io/cdap/wrangler/WranglerErrorUtil.java

@@ -28,6 +28,7 @@
 import io.cdap.wrangler.api.DirectiveParseException;
 import io.cdap.wrangler.api.RecipeException;
 import io.cdap.wrangler.expression.ELException;
+import io.cdap.wrangler.expression.ELPermissionException;
 import io.cdap.wrangler.utils.RecordConvertorException;
 import java.util.List;
 import java.util.Map;
@@ -45,7 +46,9 @@ public final class WranglerErrorUtil {
       .put(DirectiveLoadException.class.getName(), "Loading-Directive")
       .put(DirectiveNotFoundException.class.getName(), "Directive-Not-Found")
       .put(RecordConvertorException.class.getName(), "Record-Conversion")
-      .put(ELException.class.getName(), "ExpressionLanguage-Parsing").build();
+      .put(ELException.class.getName(), "ExpressionLanguage-Parsing")
+      .put(ELPermissionException.class.getName(), "ExpressionLanguage-Permission")
+      .build();
 
   private static final Map<String, String> NON_TERMINAL_EXCEPTIONS = ImmutableMap.<String, String>builder()
       .put(RecipeException.class.getName(), "Executing-Recipe").build();
@@ -61,6 +64,19 @@ private WranglerErrorUtil() {
     throw new IllegalStateException("Utility class");
   }
 
+  /**
+   * Checks whether the given {@link Throwable} or any cause in its causal chain is an
+   * {@link ELPermissionException}.
+   *
+   * @param e the {@link Throwable} to inspect
+   * @return {@code true} if {@link ELPermissionException} is present in the causal chain,
+       *     {@code false} otherwise
+   */
+  public static boolean isCriticalException(Throwable e) {
+    return e != null && Throwables.getCausalChain(e).stream()
+        .anyMatch(ELPermissionException.class::isInstance);
+  }
+
   /**
    * Traverses the causal chain of the given Throwable to find specific exceptions. If a terminal
    * exception is found, it returns a corresponding ProgramFailureException. If a non-terminal