diff --git a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java index 1e5b15f3133a3f..3209bdfd658def 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java +++ b/flink-clients/src/main/java/org/apache/flink/client/ClientUtils.java @@ -149,6 +149,13 @@ public static void executeProgram( /** * This method blocks until the job status is not INITIALIZING anymore. * + *

Only call this when the origin of {@code jobResultSupplier}'s result is trusted + * (for example, Application Mode or a per-job {@code MiniCluster}, where this process itself + * produced the {@link JobResult}). {@link SerializedThrowable#deserializeError} can run + * arbitrary code if the bytes it deserializes did not originate from a trusted source. A {@code + * jobResultSupplier} backed by a remote JobManager (for example any {@code RestClusterClient}) + * should use {@link #waitUntilSafeJobInitializationFinished} instead. + * * @param jobStatusSupplier supplier returning the job status. * @param jobResultSupplier supplier returning the job result. This will only be called if the * job reaches the FAILED state. @@ -186,6 +193,76 @@ public static void waitUntilJobInitializationFinished( } } + /** + * Like {@link #waitUntilJobInitializationFinished}, but never deserializes the failure cause: + * only {@code jobResultSupplier}'s safe, text-only fields are used to detect and rebuild a + * {@link JobInitializationException}. + * + *

The robust choice when the origin of {@code jobResultSupplier}'s result isn't known to be + * trustworthy - for example, one backed by a remote JobManager via {@code RestClusterClient}. A + * caller that needs the original exception object back must call {@link + * SerializedThrowable#deserializeError} explicitly on the returned exception's cause, and only + * when it separately trusts whoever produced that result. + * + * @param jobStatusSupplier supplier returning the job status. + * @param jobResultSupplier supplier returning the job result. This will only be called if the + * job reaches the FAILED state. + * @throws JobInitializationException If the initialization failed + */ + public static void waitUntilSafeJobInitializationFinished( + SupplierWithException jobStatusSupplier, + SupplierWithException jobResultSupplier) + throws JobInitializationException { + LOG.debug("Wait until job initialization is finished"); + WaitStrategy waitStrategy = new ExponentialWaitStrategy(50, 2000); + try { + JobStatus status = jobStatusSupplier.get(); + long attempt = 0; + while (status == JobStatus.INITIALIZING) { + Thread.sleep(waitStrategy.sleepTime(attempt++)); + status = jobStatusSupplier.get(); + } + if (status == JobStatus.FAILED) { + JobResult result = jobResultSupplier.get(); + Optional throwable = result.getSerializedThrowable(); + // Checked via the safe class-name field, not deserializeError(): this result may + // come from a remote JobManager (e.g. when submitting through a shared session + // cluster), and JobInitializationException is the one, fixed, Flink-internal + // type we ever need to reconstruct here. + if (throwable.isPresent() + && JobInitializationException.class + .getName() + .equals(throwable.get().getOriginalErrorClassName())) { + throw new JobInitializationException( + result.getJobId(), + stripOriginalClassNamePrefix(throwable.get()), + throwable.get().getCause()); + } + } + } catch (JobInitializationException initializationException) { + throw initializationException; + } catch (Throwable throwable) { + ExceptionUtils.checkInterrupted(throwable); + throw new RuntimeException("Error while waiting for job to be initialized", throwable); + } + } + + /** + * {@link SerializedThrowable#getMessage()} always returns {@code ": + * "} (see {@link SerializedThrowable}'s class Javadoc), whereas the original + * exception's own {@code getMessage()} did not carry that prefix. Strips it back off so the + * reconstructed {@link JobInitializationException} carries the same message the original + * exception had. + */ + private static String stripOriginalClassNamePrefix(SerializedThrowable throwable) { + String message = throwable.getMessage(); + String prefix = throwable.getOriginalErrorClassName() + ": "; + if (message != null && message.startsWith(prefix)) { + return message.substring(prefix.length()); + } + return message; + } + /** * The client reports the heartbeat to the dispatcher for aliveness. * diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java index 4a8248d3951471..766f5aa2868a27 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java +++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/ClusterClientJobClientAdapter.java @@ -116,6 +116,11 @@ public CompletableFuture> getAccumulators() { public CompletableFuture getJobExecutionResult() { checkNotNull(classLoader); + // This JobClient talks to a RestClusterClient, so the JobResult may come from a remote + // JobManager this process does not fully control. Use toSafeJobExecutionResult(), not + // toJobExecutionResult(): it builds the failure cause from the response's plain text + // fields instead of deserializing it, which is the more robust default whenever the + // input's origin can't be fully validated. return bridgeClientRequest( clusterClientProvider, (clusterClient -> @@ -124,7 +129,8 @@ public CompletableFuture getJobExecutionResult() { .thenApply( (jobResult) -> { try { - return jobResult.toJobExecutionResult(classLoader); + return jobResult.toSafeJobExecutionResult( + classLoader); } catch (Throwable t) { throw new CompletionException( new ProgramInvocationException( diff --git a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java index e1a667d159271c..ee08504b13d718 100644 --- a/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java +++ b/flink-clients/src/main/java/org/apache/flink/client/deployment/executors/AbstractSessionClusterExecutor.java @@ -107,10 +107,9 @@ public CompletableFuture execute( .thenApplyAsync( FunctionUtils.uncheckedFunction( jobId -> { - ClientUtils.waitUntilJobInitializationFinished( + ClientUtils.waitUntilSafeJobInitializationFinished( () -> clusterClient.getJobStatus(jobId).get(), - () -> clusterClient.requestJobResult(jobId).get(), - userCodeClassloader); + () -> clusterClient.requestJobResult(jobId).get()); return jobId; })) .thenApplyAsync( diff --git a/flink-clients/src/test/java/org/apache/flink/client/ClientUtilsTest.java b/flink-clients/src/test/java/org/apache/flink/client/ClientUtilsTest.java index 43fa6971102f77..bdcf1248657e1a 100644 --- a/flink-clients/src/test/java/org/apache/flink/client/ClientUtilsTest.java +++ b/flink-clients/src/test/java/org/apache/flink/client/ClientUtilsTest.java @@ -123,4 +123,79 @@ void testWaitUntilJobInitializationFinished_regular() throws Exception { }, ClassLoader.getSystemClassLoader()); } + + /** + * Ensure that waitUntilSafeJobInitializationFinished() throws JobInitializationException + * without deserializing, using only the safe, text-only fields. + */ + @Test + void testWaitUntilSafeJobInitializationFinished_throwsInitializationException() { + Iterator statusSequenceIterator = + Arrays.asList(JobStatus.INITIALIZING, JobStatus.INITIALIZING, JobStatus.FAILED) + .iterator(); + + assertThatThrownBy( + () -> + ClientUtils.waitUntilSafeJobInitializationFinished( + statusSequenceIterator::next, + () -> { + Throwable throwable = + new JobInitializationException( + TESTING_JOB_ID, + "Something is wrong", + new RuntimeException("Err")); + return buildJobResult(throwable); + })) + .isInstanceOf(JobInitializationException.class) + .hasMessage("Something is wrong"); + } + + /** + * Ensure that waitUntilSafeJobInitializationFinished() does not throw non-initialization + * exceptions. + */ + @Test + void testWaitUntilSafeJobInitializationFinished_doesNotThrowRuntimeException() + throws Exception { + Iterator statusSequenceIterator = + Arrays.asList(JobStatus.INITIALIZING, JobStatus.INITIALIZING, JobStatus.FAILED) + .iterator(); + ClientUtils.waitUntilSafeJobInitializationFinished( + statusSequenceIterator::next, () -> buildJobResult(new RuntimeException("Err"))); + } + + /** Ensure that other errors are thrown. */ + @Test + void testWaitUntilSafeJobInitializationFinished_throwsOtherErrors() { + assertThatThrownBy( + () -> + ClientUtils.waitUntilSafeJobInitializationFinished( + () -> { + throw new RuntimeException("other error"); + }, + () -> { + Throwable throwable = + new JobInitializationException( + TESTING_JOB_ID, + "Something is wrong", + new RuntimeException("Err")); + return buildJobResult(throwable); + })) + .isInstanceOf(RuntimeException.class) + .hasMessage("Error while waiting for job to be initialized"); + } + + /** Test normal operation. */ + @Test + void testWaitUntilSafeJobInitializationFinished_regular() throws Exception { + Iterator statusSequenceIterator = + Arrays.asList(JobStatus.INITIALIZING, JobStatus.INITIALIZING, JobStatus.RUNNING) + .iterator(); + ClientUtils.waitUntilSafeJobInitializationFinished( + statusSequenceIterator::next, + () -> { + fail("unexpected call"); + return null; + }); + } } diff --git a/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientJobResultSafetyTest.java b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientJobResultSafetyTest.java new file mode 100644 index 00000000000000..56bcb15ed4d1be --- /dev/null +++ b/flink-clients/src/test/java/org/apache/flink/client/program/rest/RestClusterClientJobResultSafetyTest.java @@ -0,0 +1,199 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.flink.client.program.rest; + +import org.apache.flink.api.common.JobID; +import org.apache.flink.api.common.JobStatus; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices; +import org.apache.flink.runtime.jobmaster.JobResult; +import org.apache.flink.runtime.rest.messages.json.JobResultSerializer; +import org.apache.flink.util.SerializedThrowable; +import org.apache.flink.util.jackson.JacksonMapperFactory; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.module.SimpleModule; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.OutputStream; +import java.io.Serializable; +import java.net.ServerSocket; +import java.net.Socket; +import java.nio.charset.StandardCharsets; +import java.util.Base64; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies, against the real {@link RestClusterClient}, that a REST response's {@code + * serialized-throwable} bytes are never handed to {@code ObjectInputStream.readObject()} while + * parsing a {@link JobResult} - only an explicit, later {@link + * SerializedThrowable#deserializeError} call may do that. + * + *

{@code RestClusterClient} is used by callers that talk to a remote JobManager over the + * network, so parsing a response must not assume the {@code serialized-throwable} bytes are + * well-formed or otherwise trustworthy. + */ +class RestClusterClientJobResultSafetyTest { + + private static final AtomicBoolean MARKER_TRIGGERED = new AtomicBoolean(false); + + /** Fires {@link #MARKER_TRIGGERED} purely as a side effect of {@code readObject()}. */ + private static final class Marker implements Serializable { + private static final long serialVersionUID = 1L; + + private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundException { + MARKER_TRIGGERED.set(true); + in.defaultReadObject(); + } + } + + @AfterEach + void resetMarker() { + MARKER_TRIGGERED.set(false); + } + + @Test + void parsingJobResultResponseNeverDeserializesSerializedThrowableBytes() throws Exception { + assertThat(MARKER_TRIGGERED).isFalse(); + + final byte[] markerBytes = serialize(new Marker()); + final String body = + "{\"status\":{\"id\":\"COMPLETED\"},\"job-execution-result\":" + + "{\"id\":\"1bb5e8c7df49938733b7c6a73678de6a\",\"net-runtime\":0," + + "\"application-status\":\"FAILED\"," + + "\"failure-cause\":{" + + "\"class\":\"java.lang.RuntimeException\"," + + "\"message\":\"boom\"," + + "\"stack-trace\":\"java.lang.RuntimeException: boom\"," + + "\"serialized-throwable\":\"" + + Base64.getEncoder().encodeToString(markerBytes) + + "\"}}}"; + + final JobResult jobResult = requestJobResultAgainstFakeServer(body); + + assertThat(MARKER_TRIGGERED) + .as( + "parsing the response must not run ObjectInputStream.readObject() over " + + "serialized-throwable's bytes") + .isFalse(); + + assertThat(jobResult.getSerializedThrowable()).isPresent(); + final SerializedThrowable failureCause = jobResult.getSerializedThrowable().get(); + assertThat(failureCause.getMessage()).isEqualTo("boom"); + assertThat(failureCause.getOriginalErrorClassName()) + .isEqualTo("java.lang.RuntimeException"); + + // Positive control: the marker mechanism does work, and the bytes are still there for + // an explicit, later deserializeError() call to use. + failureCause.deserializeError(ClassLoader.getSystemClassLoader()); + assertThat(MARKER_TRIGGERED) + .as("deserializeError() is the one call site allowed to touch these bytes") + .isTrue(); + } + + /** Negative control: a real, legitimate job failure response must not trigger the marker. */ + @Test + void legitimateJobResultResponseDoesNotTriggerMarker() throws Exception { + assertThat(MARKER_TRIGGERED).isFalse(); + + final SerializedThrowable realFailure = + new SerializedThrowable(new RuntimeException("job failed for real")); + final JobResult realResult = + new JobResult.Builder() + .jobId(new JobID()) + .jobStatus(JobStatus.FAILED) + .netRuntime(1234L) + .serializedThrowable(realFailure) + .build(); + + final ObjectMapper objectMapper = JacksonMapperFactory.createObjectMapper(); + final SimpleModule module = new SimpleModule(); + module.addSerializer(JobResult.class, new JobResultSerializer()); + objectMapper.registerModule(module); + final String jobResultJson = objectMapper.writeValueAsString(realResult); + final String body = + "{\"status\":{\"id\":\"COMPLETED\"},\"job-execution-result\":" + + jobResultJson + + "}"; + + final JobResult jobResult = requestJobResultAgainstFakeServer(body); + + assertThat(jobResult.getSerializedThrowable()).isPresent(); + assertThat(jobResult.getSerializedThrowable().get().getMessage()) + .isEqualTo("java.lang.RuntimeException: job failed for real"); + assertThat(MARKER_TRIGGERED).isFalse(); + } + + private static byte[] serialize(Serializable o) throws IOException { + final java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream(); + try (ObjectOutputStream oos = new ObjectOutputStream(bos)) { + oos.writeObject(o); + } + return bos.toByteArray(); + } + + /** + * Mirrors {@code AbstractFlinkService.getClusterClient(conf)} exactly: a fresh {@link + * RestClusterClient} whose {@link StandaloneClientHAServices} points straight at our fake + * server, then calls {@code requestJobResult(jobID)} - the same call {@code + * AbstractFlinkService.requestJobResult()} makes. + */ + private static JobResult requestJobResultAgainstFakeServer(String jsonBody) throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + final String restServerAddress = "http://localhost:" + serverSocket.getLocalPort(); + try (RestClusterClient clusterClient = + new RestClusterClient<>( + new Configuration(), + "safety-test-cluster", + (c, e) -> new StandaloneClientHAServices(restServerAddress))) { + + final var responseFuture = clusterClient.requestJobResult(new JobID()); + + serverSocket.setSoTimeout(10_000); + try (Socket connection = serverSocket.accept()) { + writeHttpResponse(connection.getOutputStream(), jsonBody); + return responseFuture.get(10, TimeUnit.SECONDS); + } + } + } + } + + private static void writeHttpResponse(OutputStream out, String jsonBody) throws Exception { + final byte[] bodyBytes = jsonBody.getBytes(StandardCharsets.UTF_8); + final String headers = + "HTTP/1.1 200 OK\r\n" + + "Content-Type: application/json; charset=UTF-8\r\n" + + "Content-Length: " + + bodyBytes.length + + "\r\n" + + "Connection: close\r\n" + + "\r\n"; + out.write(headers.getBytes(StandardCharsets.UTF_8)); + out.write(bodyBytes); + out.flush(); + } +} diff --git a/flink-core/src/main/java/org/apache/flink/util/SerializedThrowable.java b/flink-core/src/main/java/org/apache/flink/util/SerializedThrowable.java index 7937e2de8a68a7..d26fb3b8f37f08 100644 --- a/flink-core/src/main/java/org/apache/flink/util/SerializedThrowable.java +++ b/flink-core/src/main/java/org/apache/flink/util/SerializedThrowable.java @@ -18,6 +18,8 @@ package org.apache.flink.util; +import javax.annotation.Nullable; + import java.io.PrintStream; import java.io.PrintWriter; import java.io.Serializable; @@ -37,9 +39,18 @@ public class SerializedThrowable extends Exception implements Serializable { private static final long serialVersionUID = 7284183123441947635L; - /** The original exception in serialized form. */ + /** The original exception's own serialized bytes, or {@code null} if unavailable. */ private final byte[] serializedException; + /** + * Bytes of the whole {@code SerializedThrowable} wrapper as it arrived on the wire (see the + * wire-safe constructor below), or {@code null} for a self-produced instance. Used only + * internally by {@link #deserializeError}, as a fallback when {@link #serializedException} is + * {@code null}: unlike that field, this is not, by itself, the original exception's own + * serialized form, so it is never exposed through {@link #getSerializedException()}. + */ + private final byte[] wireWrapperBytes; + /** Name of the original error class. */ private final String originalErrorClassName; @@ -63,9 +74,19 @@ public SerializedThrowable(Throwable exception) { } private SerializedThrowable(Throwable exception, Set alreadySeen) { - super(getClassNameAndMessageOrError(exception)); + // When copying from an already-SerializedThrowable value (the else branch below), its own + // getMessage() is already correctly formatted (": ") from + // whenever it was itself constructed - reuse it verbatim instead of recomputing this via + // getClassNameAndMessageOrError(exception), which would stamp SerializedThrowable's own + // class name instead of the originally wrapped exception's. + super( + exception instanceof SerializedThrowable + ? exception.getMessage() + : getClassNameAndMessageOrError(exception)); if (!(exception instanceof SerializedThrowable)) { + this.wireWrapperBytes = null; + // serialize and memoize the original message byte[] serialized; // introduce the synchronization here to avoid deadlock of multi thread serializing @@ -103,6 +124,7 @@ private SerializedThrowable(Throwable exception, Set alreadySeen) { // copy from that serialized throwable SerializedThrowable other = (SerializedThrowable) exception; this.serializedException = other.serializedException; + this.wireWrapperBytes = other.wireWrapperBytes; this.originalErrorClassName = other.originalErrorClassName; this.fullStringifiedStackTrace = other.fullStringifiedStackTrace; this.cachedException = other.cachedException; @@ -112,17 +134,36 @@ private SerializedThrowable(Throwable exception, Set alreadySeen) { } } + /** + * Caps how many nested {@link SerializedThrowable} layers {@link #deserializeError} will + * unwrap. A well-formed instance never nests more than one or two layers deep (see the + * unwrapping comment in {@link #deserializeError}); this only guards against an unexpectedly + * long chain turning an explicit, trusted {@code deserializeError()} call into unbounded + * recursion. + */ + private static final int MAX_DESERIALIZE_UNWRAP_DEPTH = 100; + public Throwable deserializeError(ClassLoader classloader) { - if (serializedException == null) { - // failed to serialize the original exception + return deserializeError(classloader, 0); + } + + private Throwable deserializeError(ClassLoader classloader, int depth) { + final byte[] bytesToDeserialize = + serializedException != null ? serializedException : wireWrapperBytes; + if (bytesToDeserialize == null) { + // failed to serialize the original exception, and this isn't a wire-reconstructed + // instance either // return this SerializedThrowable as a stand in return this; } + if (depth >= MAX_DESERIALIZE_UNWRAP_DEPTH) { + return this; + } Throwable cached = cachedException == null ? null : cachedException.get(); if (cached == null) { try { - cached = InstantiationUtil.deserializeObject(serializedException, classloader); + cached = InstantiationUtil.deserializeObject(bytesToDeserialize, classloader); cachedException = new WeakReference<>(cached); } catch (Throwable t) { // something went wrong @@ -130,6 +171,15 @@ public Throwable deserializeError(ClassLoader classloader) { return this; } } + // wireWrapperBytes (used above when serializedException is null) is the bytes of the + // whole SerializedThrowable this instance arrived in (kept on the wire for old-client + // compatibility, see SerializedThrowableSerializer), not just the original exception's + // own bytes. Fully deserializing those bytes yields another SerializedThrowable whose own + // serializedException field is the original exception's bytes; unwrap until we reach the + // real object, exactly as old clients did by fully deserializing in one step. + if (cached instanceof SerializedThrowable && cached != this) { + return ((SerializedThrowable) cached).deserializeError(classloader, depth + 1); + } return cached; } @@ -201,6 +251,39 @@ public static Throwable get(Throwable serThrowable, ClassLoader loader) { } } + /** + * Constructs a SerializedThrowable directly from its already-serialized wire representation + * (see {@code SerializedThrowableDeserializer}), without deserializing {@code wireWrapperBytes} + * - that only happens lazily, on an explicit {@link #deserializeError} call. Used when + * reconstructing a SerializedThrowable that arrived over a channel this process does not fully + * control (e.g. a REST response), where those bytes should not be deserialized automatically. + * + * @param message the message to report via {@link #getMessage()}, normally {@code ": + * "} to match {@link #SerializedThrowable(Throwable)} + * @param originalErrorClassName name of the original exception's class + * @param fullStringifiedStackTrace the original exception's stringified stack trace + * @param wireWrapperBytes bytes usable by {@link #deserializeError} to recover the original + * exception, or {@code null} if unavailable; not touched by this constructor. See {@link + * #wireWrapperBytes}'s Javadoc for why these are kept separate from {@link + * #serializedException} and never returned by {@link #getSerializedException()}. + */ + public SerializedThrowable( + @Nullable String message, + String originalErrorClassName, + String fullStringifiedStackTrace, + @Nullable byte[] wireWrapperBytes) { + super(message); + this.serializedException = null; + this.wireWrapperBytes = wireWrapperBytes; + this.originalErrorClassName = originalErrorClassName; + this.fullStringifiedStackTrace = fullStringifiedStackTrace; + this.cachedException = null; + // super(message) fills in this constructor's own call stack by default; the original + // exception's stack trace is only available as text, in fullStringifiedStackTrace, so + // there is nothing meaningful to put here structurally. + setStackTrace(new StackTraceElement[0]); + } + private static String getClassNameAndMessageOrError(Throwable error) { try { String className = error.getClass().getName(); diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobResult.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobResult.java index f6a0e85bd1635b..a5d3a4f01f3816 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobResult.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobResult.java @@ -149,7 +149,16 @@ public long getEndTime() { } /** - * Converts the {@link JobResult} to a {@link JobExecutionResult}. + * Converts the {@link JobResult} to a {@link JobExecutionResult}, deserializing the failure + * cause if the job did not finish successfully. + * + *

Only call this when the origin of this {@link JobResult} is trusted (for example, a + * same-process execution such as {@code MiniCluster} or Application Mode, where this process + * itself produced the {@link SerializedThrowable}). {@link + * SerializedThrowable#deserializeError} can run arbitrary code if the bytes it deserializes did + * not originate from a trusted source. A {@link JobResult} obtained from a client that talks to + * a remote JobManager (for example any {@code RestClusterClient}-backed {@code JobClient}) + * should use {@link #toSafeJobExecutionResult} instead. * * @param classLoader to use for deserialization * @return JobExecutionResult @@ -160,20 +169,46 @@ public long getEndTime() { */ public JobExecutionResult toJobExecutionResult(ClassLoader classLoader) throws JobExecutionException, IOException, ClassNotFoundException { + return toJobExecutionResult( + classLoader, + serializedThrowable == null + ? null + : serializedThrowable.deserializeError(classLoader)); + } + + /** + * Converts the {@link JobResult} to a {@link JobExecutionResult} without deserializing the + * failure cause: the thrown exception's cause is the {@link SerializedThrowable} itself, never + * the original, live exception object. + * + *

The robust choice when the origin of this {@link JobResult} isn't known to be trustworthy + * - for example, one obtained from a {@code RestClusterClient}-backed {@code JobClient} talking + * to a remote JobManager. {@code classLoader} is only used to deserialize accumulators. A + * caller that needs the original exception object back must call {@link + * SerializedThrowable#deserializeError} explicitly on the returned exception's cause, and only + * when it separately trusts whoever produced this {@link JobResult}. + * + * @param classLoader to use for deserializing accumulators + * @return JobExecutionResult + * @throws JobCancellationException if the job was cancelled + * @throws JobExecutionException if the job execution did not succeed + * @throws IOException if the accumulator could not be deserialized + * @throws ClassNotFoundException if the accumulator could not deserialized + */ + public JobExecutionResult toSafeJobExecutionResult(ClassLoader classLoader) + throws JobExecutionException, IOException, ClassNotFoundException { + return toJobExecutionResult(classLoader, serializedThrowable); + } + + private JobExecutionResult toJobExecutionResult( + ClassLoader classLoader, @Nullable Throwable cause) + throws JobExecutionException, IOException, ClassNotFoundException { if (jobStatus == JobStatus.FINISHED) { return new JobExecutionResult( jobId, netRuntime, AccumulatorHelper.deserializeAccumulators(accumulatorResults, classLoader)); } else { - final Throwable cause; - - if (serializedThrowable == null) { - cause = null; - } else { - cause = serializedThrowable.deserializeError(classLoader); - } - final JobExecutionException exception; if (jobStatus == JobStatus.FAILED) { diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableDeserializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableDeserializer.java index cc95b956d44956..70f3e232502ccc 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableDeserializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableDeserializer.java @@ -18,7 +18,6 @@ package org.apache.flink.runtime.rest.messages.json; -import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.SerializedThrowable; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonParser; @@ -28,7 +27,12 @@ import java.io.IOException; +import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_CAUSE; +import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_CLASS; +import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_MESSAGE; import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_SERIALIZED_THROWABLE; +import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_STACK_TRACE; +import static org.apache.flink.runtime.rest.messages.json.SerializedThrowableSerializer.FIELD_NAME_SUPPRESSED; /** JSON deserializer for {@link SerializedThrowable}. */ public class SerializedThrowableDeserializer extends StdDeserializer { @@ -39,18 +43,71 @@ public SerializedThrowableDeserializer() { super(SerializedThrowable.class); } + /** + * Caps how deep {@link #readThrowable} recurses into nested {@code cause}/{@code suppressed} + * fields. Without this, an unexpectedly deeply-nested {@code cause} chain in a response could + * exhaust the parsing thread's stack before Jackson's own tree-depth limits necessarily kick + * in, turning parsing itself into a source of instability. 100 is far beyond any real + * exception's cause/suppressed depth. + */ + static final int MAX_THROWABLE_DEPTH = 100; + @Override public SerializedThrowable deserialize(final JsonParser p, final DeserializationContext ctxt) throws IOException { - final JsonNode root = p.readValueAsTree(); + return readThrowable(p.readValueAsTree(), 0); + } - final byte[] serializedException = root.get(FIELD_NAME_SERIALIZED_THROWABLE).binaryValue(); - try { - return InstantiationUtil.deserializeObject( - serializedException, ClassLoader.getSystemClassLoader()); - } catch (ClassNotFoundException e) { + /** + * Recursively reconstructs a {@link SerializedThrowable} - and its cause and suppressed + * exceptions - from the plain-string {@code class}/{@code message}/{@code stack-trace} fields + * and the raw {@code serialized-throwable} bytes, without ever passing those bytes to {@code + * InstantiationUtil.deserializeObject()}. Deserializing the original exception object only + * happens lazily, on an explicit {@link SerializedThrowable#deserializeError} call, so that + * parsing a response never runs an {@code ObjectInputStream.readObject()} over bytes this + * process has not otherwise validated. + */ + private static SerializedThrowable readThrowable(final JsonNode node, final int depth) + throws IOException { + if (node == null || node.isNull()) { + return null; + } + if (depth >= MAX_THROWABLE_DEPTH) { throw new IOException( - "Failed to deserialize " + SerializedThrowable.class.getCanonicalName(), e); + "Refusing to deserialize a " + + SerializedThrowable.class.getCanonicalName() + + " nested more than " + + MAX_THROWABLE_DEPTH + + " levels deep"); } + + final JsonNode classNode = node.get(FIELD_NAME_CLASS); + final JsonNode messageNode = node.get(FIELD_NAME_MESSAGE); + final JsonNode stackTraceNode = node.get(FIELD_NAME_STACK_TRACE); + final JsonNode serializedNode = node.get(FIELD_NAME_SERIALIZED_THROWABLE); + + final SerializedThrowable throwable = + new SerializedThrowable( + messageNode != null ? messageNode.asText() : null, + classNode != null ? classNode.asText() : null, + stackTraceNode != null ? stackTraceNode.asText() : null, + serializedNode != null ? serializedNode.binaryValue() : null); + + final SerializedThrowable cause = readThrowable(node.get(FIELD_NAME_CAUSE), depth + 1); + if (cause != null) { + throwable.initCause(cause); + } + + final JsonNode suppressedNode = node.get(FIELD_NAME_SUPPRESSED); + if (suppressedNode != null) { + for (JsonNode s : suppressedNode) { + final SerializedThrowable suppressed = readThrowable(s, depth + 1); + if (suppressed != null) { + throwable.addSuppressed(suppressed); + } + } + } + + return throwable; } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializer.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializer.java index 1ec06925a8348a..1a3feb980879e6 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializer.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializer.java @@ -34,10 +34,16 @@ public class SerializedThrowableSerializer extends StdSerializer 0) { + gen.writeArrayFieldStart(FIELD_NAME_SUPPRESSED); + for (Throwable s : suppressed) { + writeThrowable(asSerializedThrowable(s), gen); + } + gen.writeEndArray(); + } + gen.writeEndObject(); } + + private static SerializedThrowable asSerializedThrowable(Throwable t) { + return t instanceof SerializedThrowable + ? (SerializedThrowable) t + : new SerializedThrowable(t); + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java index f335cd26dfd4ea..97dda6971f5b72 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobResultTest.java @@ -124,6 +124,34 @@ void testFailedJobThrowsJobExecutionException() { .isEqualTo(cause); } + @Test + void testSafeFailedJobThrowsJobExecutionExceptionWithoutDeserializing() { + final FlinkException cause = new FlinkException("Test exception"); + final JobResult jobResult = + JobResult.createFrom( + new ArchivedExecutionGraphBuilder() + .setJobID(new JobID()) + .setState(JobStatus.FAILED) + .setFailureCause(new ErrorInfo(cause, 42L)) + .build()); + + // toSafeJobExecutionResult() must not call SerializedThrowable#deserializeError itself: + // the cause is the SerializedThrowable, and only an explicit, separate deserializeError() + // call recovers the original, live exception object. + assertThatThrownBy(() -> jobResult.toSafeJobExecutionResult(getClass().getClassLoader())) + .isInstanceOf(JobExecutionException.class) + .cause() + .isInstanceOf(SerializedThrowable.class); + + assertThatThrownBy(() -> jobResult.toSafeJobExecutionResult(getClass().getClassLoader())) + .extracting(Throwable::getCause) + .extracting( + c -> + ((SerializedThrowable) c) + .deserializeError(getClass().getClassLoader())) + .isEqualTo(cause); + } + @Test void testFailureResultRequiresFailureCause() { assertThatThrownBy( diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobExecutionResultResponseBodyTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobExecutionResultResponseBodyTest.java index 278b30475464d9..78f704cf1bb360 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobExecutionResultResponseBodyTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/job/JobExecutionResultResponseBodyTest.java @@ -139,8 +139,21 @@ protected void assertOriginalEqualsToUnmarshalled( .isEqualTo(expectedFailureCause.getFullStringifiedStackTrace()); assertThat(actualFailureCause.getOriginalErrorClassName()) .isEqualTo(expectedFailureCause.getOriginalErrorClassName()); - assertThat(expectedFailureCause.getSerializedException()) - .isEqualTo(actualFailureCause.getSerializedException()); + // actualFailureCause was reconstructed from a REST response, so + // getSerializedException() correctly reports unavailable (see + // SerializedThrowable#getSerializedException()); the meaningful guarantee + // is that deserializeError() still recovers an equivalent exception. + assertThat(actualFailureCause.getSerializedException()).isNull(); + assertThat( + actualFailureCause + .deserializeError( + ClassLoader.getSystemClassLoader()) + .getMessage()) + .isEqualTo( + expectedFailureCause + .deserializeError( + ClassLoader.getSystemClassLoader()) + .getMessage()); }); if (expectedJobExecutionResult.getAccumulatorResults() != null) { diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializerTest.java index a237bfa50ef902..72a80cfe95e5a2 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/messages/json/SerializedThrowableSerializerTest.java @@ -18,16 +18,21 @@ package org.apache.flink.runtime.rest.messages.json; +import org.apache.flink.util.InstantiationUtil; import org.apache.flink.util.SerializedThrowable; import org.apache.flink.util.jackson.JacksonMapperFactory; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.module.SimpleModule; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import java.util.Base64; + import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; /** Tests for {@link SerializedThrowableSerializer} and {@link SerializedThrowableDeserializer}. */ class SerializedThrowableSerializerTest { @@ -72,4 +77,168 @@ void testSerializationDeserialization() throws Exception { assertThat(deserializedSerializedThrowable.getSuppressed()[0]) .isInstanceOf(SerializedThrowable.class); } + + @Test + void testDeserializationNeverRunsObjectInputStreamOverSerializedThrowableBytes() + throws Exception { + // Not a valid Java serialization stream - if the deserializer ever passed this to + // InstantiationUtil.deserializeObject()/ObjectInputStream.readObject() at parse time, + // that call would throw. Parsing must succeed regardless of what's in this field. + final String garbageBase64 = + Base64.getEncoder().encodeToString("not-a-serialized-object".getBytes()); + final String json = + "{\"class\":\"java.lang.Exception\",\"message\":\"boom\"," + + "\"stack-trace\":\"java.lang.Exception: boom\"," + + "\"serialized-throwable\":\"" + + garbageBase64 + + "\"}"; + + final SerializedThrowable[] deserialized = new SerializedThrowable[1]; + assertThatCode( + () -> + deserialized[0] = + objectMapper.readValue(json, SerializedThrowable.class)) + .doesNotThrowAnyException(); + + assertThat(deserialized[0].getMessage()).isEqualTo("boom"); + assertThat(deserialized[0].getOriginalErrorClassName()).isEqualTo("java.lang.Exception"); + + // Only an explicit deserializeError() call touches the bytes, and it degrades gracefully + // (returns itself as a stand-in) instead of propagating a deserialization failure. + assertThat(deserialized[0].deserializeError(ClassLoader.getSystemClassLoader())) + .isSameAs(deserialized[0]); + } + + /** + * Proves backward compatibility: an old, pre-fix client - which only ever reads the {@code + * serialized-throwable} field and runs it straight through {@code + * InstantiationUtil.deserializeObject()} - must still fully recover message/cause/suppressed + * when talking to this (patched) serializer. The helper below reproduces that old + * deserializer's exact logic (the code this fix removed from {@link + * SerializedThrowableDeserializer}), not an approximation of it, so this test fails if a future + * change ever stops writing the full-wrapper blob into that field for compatibility. + * + *

See {@link #newDeserializerDegradesGracefullyAgainstOldShapedResponse} directly below for + * the other rolling-upgrade direction. + */ + @Test + void oldClientLogicFullyRecoversCauseAndSuppressedFromNewSerializer() throws Exception { + Exception cause = new Exception("cause"); + Exception root = new Exception("message", cause); + Exception suppressed = new Exception("suppressed"); + root.addSuppressed(suppressed); + + final SerializedThrowable serializedThrowable = new SerializedThrowable(root); + final String json = objectMapper.writeValueAsString(serializedThrowable); + + final SerializedThrowable oldClientResult = + oldClientDeserialize(objectMapper.readTree(json)); + + assertThat(oldClientResult.getMessage()).isEqualTo("java.lang.Exception: message"); + assertThat(oldClientResult.getCause()).isInstanceOf(SerializedThrowable.class); + assertThat(oldClientResult.getCause().getMessage()).isEqualTo("java.lang.Exception: cause"); + assertThat(oldClientResult.getSuppressed()).hasSize(1); + assertThat(oldClientResult.getSuppressed()[0].getMessage()) + .isEqualTo("java.lang.Exception: suppressed"); + } + + /** Byte-for-byte the old (pre-fix) {@code SerializedThrowableDeserializer.deserialize()}. */ + private static SerializedThrowable oldClientDeserialize(JsonNode root) throws Exception { + final byte[] serializedException = + root.get(SerializedThrowableSerializer.FIELD_NAME_SERIALIZED_THROWABLE) + .binaryValue(); + return InstantiationUtil.deserializeObject( + serializedException, ClassLoader.getSystemClassLoader()); + } + + /** + * The other rolling-upgrade direction from {@link + * #oldClientLogicFullyRecoversCauseAndSuppressedFromNewSerializer}: this (patched) deserializer + * talking to an old, pre-fix server's response, which never had {@code message}/{@code + * cause}/{@code suppressed} fields - only {@code class}/{@code stack-trace}/{@code + * serialized-throwable}. Parsing must still succeed; message/cause are simply unavailable until + * an explicit {@link SerializedThrowable#deserializeError} call, since that data only exists + * inside the (unparsed) blob on an old server. + */ + @Test + void newDeserializerDegradesGracefullyAgainstOldShapedResponse() throws Exception { + final Exception cause = new Exception("cause"); + final Exception original = new Exception("message", cause); + final SerializedThrowable serializedThrowable = new SerializedThrowable(original); + final byte[] oldStyleBlob = InstantiationUtil.serializeObject(serializedThrowable); + + final String json = + "{\"class\":\"java.lang.Exception\"," + + "\"stack-trace\":\"java.lang.Exception: message\"," + + "\"serialized-throwable\":\"" + + Base64.getEncoder().encodeToString(oldStyleBlob) + + "\"}"; + + final SerializedThrowable[] deserialized = new SerializedThrowable[1]; + assertThatCode( + () -> + deserialized[0] = + objectMapper.readValue(json, SerializedThrowable.class)) + .doesNotThrowAnyException(); + + // The known degradation: no message/cause field on the wire means none available yet. + assertThat(deserialized[0].getMessage()).isNull(); + assertThat(deserialized[0].getCause()).isNull(); + assertThat(deserialized[0].getOriginalErrorClassName()).isEqualTo("java.lang.Exception"); + + // The escape hatch still works: an explicit deserializeError() call recovers everything + // an old client would have had, straight from the still-present binary blob. Parsing no + // longer does an implicit first unwrap the way an old client's own parsing did, so + // deserializeError() now does that plus its own unwrap in one call (see + // SerializedThrowable#deserializeError), landing on the real original exception and its + // real (unwrapped) cause directly - exactly what an old client's + // parse-then-deserializeError + // sequence would have produced. + final Throwable recovered = + deserialized[0].deserializeError(ClassLoader.getSystemClassLoader()); + assertThat(recovered.getMessage()).isEqualTo("message"); + assertThat(recovered.getCause()).isNotInstanceOf(SerializedThrowable.class); + assertThat(recovered.getCause().getMessage()).isEqualTo("cause"); + } + + /** + * The reconstructing constructor must not let {@code super(message)}'s default stack-trace + * capture leak the deserializer's own call stack into {@link + * SerializedThrowable#getStackTrace()} - the original exception's stack trace is only + * meaningfully available as text, via {@link + * SerializedThrowable#getFullStringifiedStackTrace()}. + */ + @Test + void deserializedThrowableDoesNotExposeDeserializersOwnStackTrace() throws Exception { + final String json = + "{\"class\":\"java.lang.Exception\",\"message\":\"boom\"," + + "\"stack-trace\":\"java.lang.Exception: boom\"}"; + + final SerializedThrowable deserialized = + objectMapper.readValue(json, SerializedThrowable.class); + + assertThat(deserialized.getStackTrace()).isEmpty(); + } + + /** + * A {@code cause} chain nested one level past {@code SerializedThrowableDeserializer}'s depth + * cap must fail parsing with an {@link java.io.IOException}, not exhaust the parsing thread's + * stack. Without an explicit cap, a response with an arbitrarily deep {@code cause} chain would + * turn parsing itself into a source of instability (a stack overflow). + */ + @Test + void causeChainNestedPastTheDepthCapFailsParsingInsteadOfOverflowingTheStack() { + final int nestingLevels = SerializedThrowableDeserializer.MAX_THROWABLE_DEPTH + 1; + final StringBuilder json = new StringBuilder("{\"class\":\"java.lang.Exception\""); + for (int i = 0; i < nestingLevels; i++) { + json.append(",\"cause\":{\"class\":\"java.lang.Exception\""); + } + for (int i = 0; i < nestingLevels; i++) { + json.append('}'); + } + json.append('}'); + + assertThatCode(() -> objectMapper.readValue(json.toString(), SerializedThrowable.class)) + .isInstanceOf(java.io.IOException.class); + } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializedThrowableTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializedThrowableTest.java index 64a9340937eede..c534291e540217 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializedThrowableTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/util/SerializedThrowableTest.java @@ -152,10 +152,12 @@ void testCopyPreservesCause() { SerializedThrowable serialized = new SerializedThrowable(parent); assertThat(serialized.getCause()).isNotNull(); + // Copying a SerializedThrowable must preserve its already-correctly-formatted message + // verbatim, not recompute it as if `serialized` (a SerializedThrowable) were itself the + // original exception - that would stamp SerializedThrowable's own class name instead of + // the originally wrapped exception's. SerializedThrowable copy = new SerializedThrowable(serialized); - assertThat(copy) - .hasMessage( - "org.apache.flink.util.SerializedThrowable: java.lang.Exception: parent message"); + assertThat(copy.getMessage()).isEqualTo(serialized.getMessage()); assertThat(copy.getCause()).isNotNull().hasMessage("java.lang.Exception: original message"); }