re-serialize JSON metadata written into the ES and OpenSearch _bulk body

tika-pipes/tika-pipes-plugins/tika-pipes-es/src/main/java/org/apache/tika/pipes/emitter/es/ESClient.java

@@ -501,7 +501,10 @@ private static void writeMetadata(Metadata metadata, JsonGenerator jsonGenerator
             if (vals.length == 1) {
                 if (isJsonField(n) && isValidJson(vals[0])) {
                     jsonGenerator.writeFieldName(n);
-                    jsonGenerator.writeRawValue(vals[0]);
+                    // re-serialize compact rather than passing vals[0] through raw: an embedded
+                    // newline in an otherwise valid value would split the NDJSON _bulk body into
+                    // extra action lines
+                    jsonGenerator.writeRawValue(CHUNK_MAPPER.readTree(vals[0]).toString());
                 } else {
                     jsonGenerator.writeStringField(n, vals[0]);
                 }

tika-pipes/tika-pipes-plugins/tika-pipes-es/src/test/java/org/apache/tika/pipes/emitter/es/ESClientTest.java

@@ -343,4 +343,16 @@ public void testChunkStrategyConfig() throws Exception {
         assertThrows(TikaConfigException.class, () -> ESEmitterConfig.load(
                 String.format(Locale.ROOT, base, "PARENT_CHILD", ", \"chunkStrategy\": \"DOCUMENTS\"")));
     }
+
+    /** A valid tk:chunks value with newlines around its tokens cannot split the NDJSON _bulk body. */
+    @Test
+    public void testChunksWithNewlinesStayOnOneLine() throws Exception {
+        CapturingClient client = new CapturingClient(config(
+                ESEmitterConfig.UpdateStrategy.OVERWRITE, ESEmitterConfig.ChunkStrategy.INLINE));
+        client.emitDocument("k", List.of(document("[\n{\"text\":\"a\"},\n{\"text\":\"b\"}\n]")));
+        List<JsonNode> lines = lines(client.bodies.get(0));
+        assertEquals(2, lines.size(), "one action and one source document, not split by newlines");
+        assertTrue(lines.get(1).get("tk:chunks").isArray());
+        assertEquals(2, lines.get(1).get("tk:chunks").size());
+    }
 }

tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/main/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClient.java

@@ -33,6 +33,7 @@
 
 import com.fasterxml.jackson.core.JsonFactory;
 import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.core.JsonParser;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ArrayNode;
@@ -485,7 +486,10 @@ private static void writeMetadata(Metadata metadata, JsonGenerator jsonGenerator
                 if (isJsonField(n)
                         && isValidJson(vals[0])) {
                     jsonGenerator.writeFieldName(n);
-                    jsonGenerator.writeRawValue(vals[0]);
+                    // re-serialize compact rather than passing vals[0] through raw: an embedded
+                    // newline in an otherwise valid value would split the NDJSON _bulk body into
+                    // extra action lines
+                    jsonGenerator.writeRawValue(VALIDATION_MAPPER.readTree(vals[0]).toString());
                 } else {
                     jsonGenerator.writeStringField(n, vals[0]);
                 }
@@ -501,10 +505,17 @@ && isValidJson(vals[0])) {
 
     private static final ObjectMapper VALIDATION_MAPPER = new ObjectMapper();
 
+    private static final JsonFactory STRICT_JSON_FACTORY = new JsonFactory();
+
     /**
-     * Validates that the value is well-formed JSON (array or object)
-     * before writing it as raw JSON. This prevents injection of
-     * arbitrary content into the bulk request payload.
+     * Validates that the value is well-formed JSON (array or object) with
+     * no trailing content before writing it as raw JSON.
+     *
+     * <p>{@code ObjectMapper.readTree()} silently ignores trailing content,
+     * so a value like {@code [1,2,3], "injected": true} would pass a simple
+     * readTree check and then inject extra fields into the document via
+     * {@code writeRawValue}. We use a {@link JsonParser} directly and
+     * assert that the stream is fully consumed after the root value.
      */
     private static boolean isValidJson(String value) {
         if (value == null || value.isEmpty()) {
@@ -514,8 +525,14 @@ private static boolean isValidJson(String value) {
         if (first != '[' && first != '{') {
             return false;
         }
-        try {
-            VALIDATION_MAPPER.readTree(value);
+        try (JsonParser parser = STRICT_JSON_FACTORY.createParser(value)) {
+            parser.nextToken();
+            parser.skipChildren();
+            if (parser.nextToken() != null) {
+                LOG.warn("Field value has trailing content after root JSON value; "
+                        + "writing as escaped string");
+                return false;
+            }
             return true;
         } catch (IOException e) {
             LOG.warn("Field value starts with '{}' but is not valid JSON; "

tika-pipes/tika-pipes-plugins/tika-pipes-opensearch/src/test/java/org/apache/tika/pipes/emitter/opensearch/OpenSearchClientTest.java

@@ -237,4 +237,33 @@ public void testChunkStrategyConfig() throws Exception {
         assertThrows(TikaConfigException.class, () -> OpenSearchEmitterConfig.load(
                 String.format(Locale.ROOT, base, "PARENT_CHILD", ", \"chunkStrategy\": \"DOCUMENTS\"")));
     }
+
+    /** A crafted tk:chunks value cannot add or split action lines in the NDJSON _bulk body. */
+    @Test
+    public void testChunksCannotInjectBulkActions() throws Exception {
+        // Trailing content after a valid array plus embedded newlines. Written raw, the "}" closed
+        // the source document early and the newlines added a delete and an index action.
+        String attack = "[1]}\n{\"delete\":{\"_index\":\"victim\",\"_id\":\"1\"}}\n"
+                + "{\"index\":{\"_index\":\"junk\"}}\n{\"pad\":[1]";
+        CapturingClient client = new CapturingClient(config(
+                OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE,
+                OpenSearchEmitterConfig.ChunkStrategy.INLINE));
+        client.emitDocument("k", List.of(document(attack)));
+        List<JsonNode> lines = lines(client.bodies.get(0));
+        assertEquals(2, lines.size(), "one action and one source document, nothing injected");
+        assertTrue(lines.get(0).has("index"));
+        assertTrue(lines.get(1).get("tk:chunks").isTextual(),
+                "a value that is not a single clean JSON document is escaped, not written raw");
+        assertFalse(lines.get(1).has("delete"));
+
+        // A legitimate chunks array still indexes natively, even with newlines around its tokens.
+        CapturingClient ok = new CapturingClient(config(
+                OpenSearchEmitterConfig.UpdateStrategy.OVERWRITE,
+                OpenSearchEmitterConfig.ChunkStrategy.INLINE));
+        ok.emitDocument("k", List.of(document("[\n  {\"text\":\"hi\"}\n]")));
+        List<JsonNode> okLines = lines(ok.bodies.get(0));
+        assertEquals(2, okLines.size());
+        assertTrue(okLines.get(1).get("tk:chunks").isArray());
+        assertEquals("hi", okLines.get(1).get("tk:chunks").get(0).get("text").asText());
+    }
 }