fix(dataconverter): harden custom converter payload handling

Signed-off-by: “Kevin” <kevlar_ksb@yahoo.com>

Author: Bueller87

src/main/java/com/uber/cadence/samples/dataconverter/CompressedJsonDataConverter.java

@@ -34,11 +34,28 @@
  * <p>For repetitive JSON payloads this typically achieves 60-80% size reduction, lowering storage
  * cost and bandwidth without changing any workflow or activity code. Apply by setting it on the
  * {@code WorkflowClientOptions} used by both the worker and any client that triggers the workflow.
+ * The decode path caps decompressed payloads to avoid unbounded memory growth on malformed input.
  */
 public final class CompressedJsonDataConverter implements DataConverter {
 
+  /** Production code should choose a limit appropriate for its workflow payload contract. */
+  public static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 10 * 1024 * 1024;
+
   private static final DataConverter delegate = JsonDataConverter.getInstance();
 
+  private final int maxDecompressedBytes;
+
+  public CompressedJsonDataConverter() {
+    this(DEFAULT_MAX_DECOMPRESSED_BYTES);
+  }
+
+  public CompressedJsonDataConverter(int maxDecompressedBytes) {
+    if (maxDecompressedBytes <= 0) {
+      throw new IllegalArgumentException("maxDecompressedBytes must be positive");
+    }
+    this.maxDecompressedBytes = maxDecompressedBytes;
+  }
+
   @Override
   public byte[] toData(Object... values) throws DataConverterException {
     if (values == null || values.length == 0) {
@@ -65,23 +82,27 @@ public <T> T fromData(byte[] content, Class<T> valueClass, Type valueType)
     if (content == null || content.length == 0) {
       return delegate.fromData(content, valueClass, valueType);
     }
-    return delegate.fromData(decompress(content), valueClass, valueType);
+    return delegate.fromData(decompress(content, maxDecompressedBytes), valueClass, valueType);
   }
 
   @Override
   public Object[] fromDataArray(byte[] content, Type... valueTypes) throws DataConverterException {
     if (content == null || content.length == 0) {
       return delegate.fromDataArray(content, valueTypes);
     }
-    return delegate.fromDataArray(decompress(content), valueTypes);
+    return delegate.fromDataArray(decompress(content, maxDecompressedBytes), valueTypes);
   }
 
-  private static byte[] decompress(byte[] content) throws DataConverterException {
+  private static byte[] decompress(byte[] content, int maxBytes) throws DataConverterException {
     try (GZIPInputStream gzip = new GZIPInputStream(new ByteArrayInputStream(content));
         ByteArrayOutputStream out = new ByteArrayOutputStream()) {
       byte[] buf = new byte[4096];
       int read;
       while ((read = gzip.read(buf)) != -1) {
+        if (out.size() > maxBytes - read) {
+          throw new DataConverterException(
+              "Gunzip payload exceeds maximum size of " + maxBytes + " bytes", null);
+        }
         out.write(buf, 0, read);
       }
       return out.toByteArray();

src/main/java/com/uber/cadence/samples/dataconverter/LocalFsBlobStore.java

@@ -18,15 +18,18 @@
 package com.uber.cadence.samples.dataconverter;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 
 /**
  * {@link BlobStore} implementation backed by the local filesystem.
  *
- * <p>The default zero-config implementation used by {@link S3OffloadDataConverter} when running
- * the demo without real AWS. Files are written under {@code
+ * <p>The default zero-config implementation used by {@link S3OffloadDataConverter} when running the
+ * demo without real AWS. Files are written under {@code
  * ${java.io.tmpdir}/cadence-java-samples-data-s3/}.
  */
 public final class LocalFsBlobStore implements BlobStore {
@@ -38,11 +41,14 @@ public LocalFsBlobStore() {
   }
 
   public LocalFsBlobStore(Path baseDir) {
-    this.baseDir = baseDir;
+    if (baseDir == null) {
+      throw new IllegalArgumentException("baseDir must not be null");
+    }
+    this.baseDir = baseDir.toAbsolutePath().normalize();
     try {
-      Files.createDirectories(baseDir);
+      Files.createDirectories(this.baseDir);
     } catch (IOException e) {
-      throw new IllegalStateException("Failed to create blob store dir " + baseDir, e);
+      throw new IllegalStateException("Failed to create blob store dir " + this.baseDir, e);
     }
   }
 
@@ -53,22 +59,33 @@ public Path baseDir() {
 
   @Override
   public void put(String key, byte[] data) throws IOException {
-    Files.write(baseDir.resolve(sanitizeKey(key)), data);
+    Files.write(baseDir.resolve(filenameForKey(key)), data);
   }
 
   @Override
   public byte[] get(String key) throws IOException {
-    return Files.readAllBytes(baseDir.resolve(sanitizeKey(key)));
+    return Files.readAllBytes(baseDir.resolve(filenameForKey(key)));
   }
 
   /**
-   * Turns a {@code bucket/sha256hex} key into a single safe filename. Keys are always generated
-   * internally by the DataConverter, but this provides a belt-and-suspenders guarantee against
-   * directory traversal in case a future caller passes a user-controlled key.
+   * Turns any blob-store key into a fixed safe filename. Keys are usually generated internally by
+   * the DataConverter, but hashing prevents directory traversal even if a future caller passes a
+   * user-controlled key.
    */
-  private static String sanitizeKey(String key) {
-    String flat = key.replace('/', '_').replace('\\', '_');
-    int slash = Math.max(flat.lastIndexOf('/'), flat.lastIndexOf('\\'));
-    return slash >= 0 ? flat.substring(slash + 1) : flat;
+  private static String filenameForKey(String key) throws IOException {
+    if (key == null || key.isEmpty()) {
+      throw new IOException("BlobStore key must not be null or empty");
+    }
+    try {
+      MessageDigest md = MessageDigest.getInstance("SHA-256");
+      byte[] digest = md.digest(key.getBytes(StandardCharsets.UTF_8));
+      StringBuilder sb = new StringBuilder(digest.length * 2);
+      for (byte b : digest) {
+        sb.append(String.format("%02x", b & 0xff));
+      }
+      return sb.toString();
+    } catch (NoSuchAlgorithmException e) {
+      throw new IOException("SHA-256 is not available in this JVM", e);
+    }
   }
 }

src/main/java/com/uber/cadence/samples/dataconverter/S3OffloadDataConverter.java

@@ -22,7 +22,6 @@
 import com.uber.cadence.converter.JsonDataConverter;
 import java.io.IOException;
 import java.lang.reflect.Type;
-import java.nio.charset.StandardCharsets;
 import java.security.MessageDigest;
 import java.security.NoSuchAlgorithmException;
 
@@ -39,15 +38,15 @@
  *
  * <ul>
  *   <li>{@code 0x00 || json} — payload is small enough to inline.
- *   <li>{@code 0x01 || jsonEnvelope} — payload was offloaded; the envelope JSON has the form
- *       {@code {"__s3_ref":"<bucket>/<sha256hex>"}}.
+ *   <li>{@code 0x01 || jsonEnvelope} — payload was offloaded; the envelope JSON has the form {@code
+ *       {"s3Ref":"<bucket>/<sha256hex>"}}.
  * </ul>
  *
  * <p>Keys are derived from the SHA-256 of the payload so {@code toData} is idempotent across
  * Cadence workflow replays. Using a fresh UUID per call would write a new orphaned blob on every
  * replay because the SDK calls {@code toData} again each time the workflow re-executes from the
- * top. If the workflow needs to control the key (e.g. to encode routing metadata), generate it
- * with {@code Workflow.sideEffect} and pass it alongside the payload instead.
+ * top. If the workflow needs to control the key (e.g. to encode routing metadata), generate it with
+ * {@code Workflow.sideEffect} and pass it alongside the payload instead.
  */
 /*
  * =============================================================================
@@ -104,12 +103,31 @@ public final class S3OffloadDataConverter implements DataConverter {
   private final String bucket;
   private final int thresholdBytes;
 
+  static final class BlobReference {
+    public String s3Ref;
+
+    public BlobReference() {}
+
+    BlobReference(String s3Ref) {
+      this.s3Ref = s3Ref;
+    }
+  }
+
   /**
    * @param store the BlobStore backend (use {@link LocalFsBlobStore} for zero-config demo).
    * @param bucket logical bucket / prefix name embedded in the reference key.
    * @param thresholdBytes max inline payload size; larger payloads are offloaded.
    */
   public S3OffloadDataConverter(BlobStore store, String bucket, int thresholdBytes) {
+    if (store == null) {
+      throw new IllegalArgumentException("store must not be null");
+    }
+    if (bucket == null || bucket.trim().isEmpty()) {
+      throw new IllegalArgumentException("bucket must not be null or empty");
+    }
+    if (thresholdBytes < 0) {
+      throw new IllegalArgumentException("thresholdBytes must not be negative");
+    }
     this.store = store;
     this.bucket = bucket;
     this.thresholdBytes = thresholdBytes;
@@ -140,8 +158,7 @@ public byte[] toData(Object... values) throws DataConverterException {
           "Failed to offload payload to blob store (key=" + key + ")", e);
     }
 
-    String envelope = "{\"__s3_ref\":\"" + key + "\"}";
-    byte[] envBytes = envelope.getBytes(StandardCharsets.UTF_8);
+    byte[] envBytes = delegate.toData(new BlobReference(key));
     byte[] result = new byte[1 + envBytes.length];
     result[0] = OFFLOAD_PREFIX;
     System.arraycopy(envBytes, 0, result, 1, envBytes.length);
@@ -173,7 +190,7 @@ private byte[] unwrap(byte[] content) throws DataConverterException {
       case INLINE_PREFIX:
         return body;
       case OFFLOAD_PREFIX:
-        String key = extractS3Ref(new String(body, StandardCharsets.UTF_8));
+        String key = extractS3Ref(body);
         try {
           return store.get(key);
         } catch (IOException e) {
@@ -186,24 +203,13 @@ private byte[] unwrap(byte[] content) throws DataConverterException {
     }
   }
 
-  /**
-   * Extracts the value of {@code __s3_ref} from the envelope JSON without bringing in a JSON
-   * parser. The envelope is produced by this class, so the format is fixed and trivially parseable.
-   */
-  private static String extractS3Ref(String envelopeJson) throws DataConverterException {
-    String marker = "\"__s3_ref\":\"";
-    int start = envelopeJson.indexOf(marker);
-    if (start < 0) {
-      throw new DataConverterException(
-          "s3 offload: envelope missing __s3_ref field: " + envelopeJson, null);
-    }
-    start += marker.length();
-    int end = envelopeJson.indexOf('"', start);
-    if (end < 0) {
-      throw new DataConverterException(
-          "s3 offload: envelope __s3_ref field is unterminated: " + envelopeJson, null);
+  private static String extractS3Ref(byte[] envelopeJson) throws DataConverterException {
+    BlobReference reference =
+        delegate.fromData(envelopeJson, BlobReference.class, BlobReference.class);
+    if (reference == null || reference.s3Ref == null || reference.s3Ref.isEmpty()) {
+      throw new DataConverterException("s3 offload: envelope missing s3Ref field", null);
     }
-    return envelopeJson.substring(start, end);
+    return reference.s3Ref;
   }
 
   private static String sha256Hex(byte[] data) throws DataConverterException {