diff --git a/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java b/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java index 159d6d512..92c12b10f 100644 --- a/wrangler-core/src/main/java/io/cdap/wrangler/executor/RecipePipelineExecutor.java +++ b/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 execute(List rows) throws RecipeException { } ++i; } - } catch (DirectiveExecutionException e) { + } catch (DirectiveExecutionException | ELPermissionException e) { throw new RecipeException(e.getMessage(), e, i, directiveIndex); } // Schema generation diff --git a/wrangler-core/src/main/java/io/cdap/wrangler/expression/EL.java b/wrangler-core/src/main/java/io/cdap/wrangler/expression/EL.java index 2c913152a..0457567a9 100644 --- a/wrangler-core/src/main/java/io/cdap/wrangler/expression/EL.java +++ b/wrangler-core/src/main/java/io/cdap/wrangler/expression/EL.java @@ -66,6 +66,7 @@ public final class EL { private final Set 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 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 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 variables) { + private EL(JexlScript script, Set variables, boolean allowlistEnabled) { this.script = script; this.variables = Collections.unmodifiableSet(variables); + this.allowlistEnabled = allowlistEnabled; } public Set 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); } } diff --git a/wrangler-core/src/main/java/io/cdap/wrangler/expression/ELPermissionException.java b/wrangler-core/src/main/java/io/cdap/wrangler/expression/ELPermissionException.java new file mode 100644 index 000000000..598b0744b --- /dev/null +++ b/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); + } +} diff --git a/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java b/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java index a56d12504..b82f68371 100644 --- a/wrangler-core/src/test/java/io/cdap/wrangler/TestingRig.java +++ b/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> 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()); } diff --git a/wrangler-core/src/test/java/io/cdap/wrangler/executor/RecipePipelineExecutorTest.java b/wrangler-core/src/test/java/io/cdap/wrangler/executor/RecipePipelineExecutorTest.java index 7f4572434..e64be8d94 100644 --- a/wrangler-core/src/test/java/io/cdap/wrangler/executor/RecipePipelineExecutorTest.java +++ b/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.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 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 pipeline = + TestingRig.execute( + new String[] { + "send-to-error exp:{ a == 'val1' }", + "send-to-error-and-continue exp:{ a == 'val2' }" + }, + directiveContext); + List 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")); + } } diff --git a/wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java b/wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java index a9e605dd0..681bb40ae 100644 --- a/wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java +++ b/wrangler-transform/src/main/java/io/cdap/wrangler/Wrangler.java @@ -469,13 +469,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), @@ -486,6 +480,11 @@ && checkPreconditionNotEmpty(false)) { throw new Exception(String.format("Stage:%s - Failing pipeline due to error : %s", getContext().getStageName(), e.getMessage()), e); } + 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 { @@ -514,6 +513,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. * diff --git a/wrangler-transform/src/main/java/io/cdap/wrangler/WranglerErrorUtil.java b/wrangler-transform/src/main/java/io/cdap/wrangler/WranglerErrorUtil.java new file mode 100644 index 000000000..98c2a33a4 --- /dev/null +++ b/wrangler-transform/src/main/java/io/cdap/wrangler/WranglerErrorUtil.java @@ -0,0 +1,50 @@ +/* + * Copyright © 2025 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; + +import com.google.common.base.Throwables; +import io.cdap.wrangler.expression.ELPermissionException; + +/** + * Error util file to handle exceptions caught in Wrangler plugin. + */ +public final class WranglerErrorUtil { + + /** + * Private constructor to prevent instantiation of this utility class. + *

+ * This class is designed to contain only static utility methods for handling exceptions and + * should not be instantiated. Any attempt to create an instance of this class will result in an + * {@link IllegalStateException}. + */ + 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); + } +}