From 4d91c53f28ac180ba3c3b92aa9d862e58890c269 Mon Sep 17 00:00:00 2001 From: tallison Date: Fri, 26 Jun 2026 14:29:37 -0400 Subject: [PATCH 1/4] TIKA-4764 and TIKA-4763 merge conflicts --- .../apache/tika/server/core/TikaServerProcess.java | 13 ++++++------- .../tika/server/core/TikaServerProcessTest.java | 10 +++++----- 2 files changed, 11 insertions(+), 12 deletions(-) diff --git a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java index 7803e0a519..d5d9d859a3 100644 --- a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java +++ b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerProcess.java @@ -398,13 +398,12 @@ static List loadCoreProviders(TikaServerConfig tikaServerConfi } // /pipes and /async fork processes and read/write via fetchers/emitters: never expose them - // unless the operator opted into unsecure features, even when explicitly listed. (The - // zero-endpoints branch only enables them when unsecure is on, so this won't false-fire.) - if ((addPipesResource || addAsyncResource) && !tikaServerConfig.isEnableUnsecureFeatures()) { - throw new TikaConfigException("The pipes/async endpoints require " + - "true in the server config: " + - "they fork processes and read/write via configured fetchers and emitters, so " + - "they are disabled by default even when explicitly listed as endpoints."); + // unless the operator set allowPipes, even when explicitly listed. (The zero-endpoints + // branch only enables them when allowPipes is set, so this won't false-fire.) + if ((addPipesResource || addAsyncResource) && !tikaServerConfig.isAllowPipes()) { + throw new TikaConfigException("The pipes/async endpoints require 'allowPipes' to be true " + + "in the server config: they fork processes and read/write via configured fetchers " + + "and emitters, so they are disabled by default even when explicitly listed as endpoints."); } if (addAsyncResource) { diff --git a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerProcessTest.java b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerProcessTest.java index 14cf63e48b..bfc52a4f80 100644 --- a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerProcessTest.java +++ b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerProcessTest.java @@ -28,17 +28,17 @@ public class TikaServerProcessTest { - private static TikaServerConfig config(boolean unsecure, String... endpoints) { + private static TikaServerConfig config(boolean allowPipes, String... endpoints) { TikaServerConfig c = new TikaServerConfig(); c.setEndpoints(new ArrayList<>(List.of(endpoints))); - c.setEnableUnsecureFeatures(unsecure); + c.setAllowPipes(allowPipes); return c; } @Test - public void pipesAndAsyncRequireUnsecureFeatures() { + public void pipesAndAsyncRequireAllowPipes() { // The pipes/async endpoints fork processes and read/write via fetchers/emitters; the - // start-guard must refuse them unless enableUnsecureFeatures is set, even when listed. + // start-guard must refuse them unless allowPipes is set, even when listed. assertThrows(TikaConfigException.class, () -> TikaServerProcess.loadCoreProviders(config(false, "pipes"), null)); assertThrows(TikaConfigException.class, @@ -46,7 +46,7 @@ public void pipesAndAsyncRequireUnsecureFeatures() { } @Test - public void ordinaryEndpointIsAllowedWithoutUnsecureFeatures() { + public void ordinaryEndpointIsAllowedWithoutAllowPipes() { // The guard must not false-fire on a non-forking endpoint. assertDoesNotThrow( () -> TikaServerProcess.loadCoreProviders(config(false, "meta"), null)); From 493814a3de2d7c19ec91de92679b6774b7f07f54 Mon Sep 17 00:00:00 2001 From: tallison Date: Fri, 26 Jun 2026 16:26:29 -0400 Subject: [PATCH 2/4] TIKA-476x - fix merge conflicts, simplify tika.proto --- .../ROOT/pages/using-tika/grpc/index.adoc | 24 ++-- tika-grpc/README.md | 10 +- .../tika/pipes/grpc/TikaGrpcConfig.java | 34 +++-- .../tika/pipes/grpc/TikaGrpcServerImpl.java | 132 +++++++++++------- tika-grpc/src/main/proto/tika.proto | 19 +-- .../tika/pipes/grpc/TikaGrpcServerTest.java | 67 ++++++++- .../pipes/emitter/fs/FileSystemEmitter.java | 13 +- .../emitter/fs/FileSystemEmitterConfig.java | 2 +- .../emitter/fs/FileSystemEmitterTest.java | 85 +++++++++++ 9 files changed, 285 insertions(+), 101 deletions(-) create mode 100644 tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java diff --git a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc index 47ddb21e9f..a7d49bcabd 100644 --- a/docs/modules/ROOT/pages/using-tika/grpc/index.adoc +++ b/docs/modules/ROOT/pages/using-tika/grpc/index.adoc @@ -67,13 +67,16 @@ the server's defaults for that request. Because this can reconfigure any pipelin component (fetcher, parser, timeouts, ...), it is off by default. When `false`, a request carrying either field is rejected with `PERMISSION_DENIED`. -|`allowComponentModifications` -|When `true`, callers may add, modify, or delete fetchers and pipes iterators at -runtime (`SaveFetcher`, `DeleteFetcher`, `SavePipesIterator`, -`DeletePipesIterator`). Because this changes what the server can reach for all -subsequent requests (for example, adding a fetcher that escapes a configured base -path), it is off by default. When `false`, those RPCs are rejected with -`PERMISSION_DENIED`. +|`allowComponentManagement` +|When `true`, callers may manage fetchers and pipes iterators at runtime: add, +modify, or delete them (`SaveFetcher`, `DeleteFetcher`, `SavePipesIterator`, +`DeletePipesIterator`), and read their stored configuration back (`GetFetcher`, +`ListFetchers`, `GetPipesIterator`). It is off by default for two reasons: +mutations change what the server can reach for all subsequent requests (for +example, adding a fetcher that escapes a configured base path), and the stored +configs returned by the read RPCs may contain secrets (passwords, access keys, +tokens). When `false`, the mutating RPCs are rejected with `PERMISSION_DENIED`, +and the read RPCs return only component identity (id and class), never the config. |=== Enable these only for trusted callers over a secured channel: @@ -83,7 +86,7 @@ Enable these only for trusted callers over a secured channel: { "grpc": { "allowPerRequestConfig": true, - "allowComponentModifications": true + "allowComponentManagement": true } } ---- @@ -149,8 +152,9 @@ automatically. Two distinct controls are involved, and both matter: Mesh mTLS authenticates *who opened the connection*; it does not authorize *what that caller may do*, so an authenticated-but-untrusted pod can still invoke whatever RPC surface is enabled — at minimum `FetchAndParse` against your -configured fetchers, and the runtime-mutation RPCs too if you have set -`allowComponentModifications`. Restricting reachability with a `NetworkPolicy` is +configured fetchers, and the runtime component-management RPCs too (including the +config reads that can expose stored secrets) if you have set +`allowComponentManagement`. Restricting reachability with a `NetworkPolicy` is therefore required, not optional — running in Kubernetes without one leaves tika-grpc reachable by every pod in the cluster. diff --git a/tika-grpc/README.md b/tika-grpc/README.md index 986529ca30..2ebea44c00 100644 --- a/tika-grpc/README.md +++ b/tika-grpc/README.md @@ -11,10 +11,12 @@ This server will manage a pool of Tika Pipes clients. * Delete * Fetch + Parse a given Fetch Item -> **Security note:** runtime fetcher/iterator mutations (Create/Update/Delete) and -> per-request parse configuration are **disabled by default**. Enable them -> explicitly via `allowComponentModifications` / `allowPerRequestConfig` in the -> `grpc` section of your tika-config. See the +> **Security note:** runtime fetcher/iterator management — mutations (Create/Update/Delete) +> and reading stored configs back (Read), which may contain secrets — plus per-request parse +> configuration are **disabled by default**. Enable them explicitly via +> `allowComponentManagement` / `allowPerRequestConfig` in the `grpc` section of your +> tika-config; with management off, the Read RPCs return only component id and class, never +> the config. See the > [Tika gRPC security configuration docs](../docs/modules/ROOT/pages/using-tika/grpc/index.adoc). ## Distribution and Maven Artifact diff --git a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcConfig.java b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcConfig.java index 3ef88c19c9..8f06d06d16 100644 --- a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcConfig.java +++ b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcConfig.java @@ -32,15 +32,15 @@ *
    *
  • {@link #isAllowPerRequestConfig()} lets a client reconfigure any * pipeline component for a single request.
    • - *
  • {@link #isAllowComponentModifications()} lets a client change which - * fetchers and iterators the server has at all.
    • + *
  • {@link #isAllowComponentManagement()} lets a client change which + * fetchers and iterators the server has, and read their stored configs.
    • *
*/ public class TikaGrpcConfig { private boolean allowPerRequestConfig = false; - private boolean allowComponentModifications = false; + private boolean allowComponentManagement = false; /** * Loads {@link TikaGrpcConfig} from the {@code "grpc"} section of the JSON @@ -81,22 +81,26 @@ public void setAllowPerRequestConfig(boolean allowPerRequestConfig) { } /** - * Whether clients may add, modify, or delete fetchers and pipes iterators at - * runtime (the SaveFetcher, DeleteFetcher, SavePipesIterator and - * DeletePipesIterator RPCs). + * Whether clients may manage fetchers and pipes iterators at runtime: add, + * modify, or delete them (the SaveFetcher, DeleteFetcher, SavePipesIterator + * and DeletePipesIterator RPCs), and read their stored configuration back + * (the GetFetcher, ListFetchers and GetPipesIterator RPCs). *

- * This is dangerous because it changes what the server can reach for all - * subsequent requests and clients (for example, adding a fetcher that - * escapes a configured base path or points at an internal host). Defaults - * to {@code false}; when {@code false}, those RPCs are rejected. + * This is dangerous on two counts. Writes change what the server can reach + * for all subsequent requests and clients (for example, adding a fetcher + * that escapes a configured base path or points at an internal host). Reads + * return stored component configs verbatim, which may include secrets + * (passwords, access keys, tokens). Defaults to {@code false}; when + * {@code false}, the mutating RPCs are rejected and the read RPCs return + * only component identity (id and class), never the config. * - * @return true if runtime component modifications are permitted + * @return true if runtime component management is permitted */ - public boolean isAllowComponentModifications() { - return allowComponentModifications; + public boolean isAllowComponentManagement() { + return allowComponentManagement; } - public void setAllowComponentModifications(boolean allowComponentModifications) { - this.allowComponentModifications = allowComponentModifications; + public void setAllowComponentManagement(boolean allowComponentManagement) { + this.allowComponentManagement = allowComponentManagement; } } diff --git a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java index 5b3ee9e719..49013b6a6b 100644 --- a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java +++ b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java @@ -147,7 +147,7 @@ class TikaGrpcServerImpl extends TikaGrpc.TikaImplBase { this.configStore = createConfigStore(); fetcherManager = FetcherManager.load(pluginManager, tikaJsonConfig, - tikaGrpcConfig.isAllowComponentModifications(), this.configStore); + tikaGrpcConfig.isAllowComponentManagement(), this.configStore); } private ConfigStore createConfigStore() throws TikaConfigException { @@ -189,7 +189,7 @@ private void startIgniteServer(ExtensionConfig config) { } /** - * If the operator has not opted in to runtime component modifications, closes + * If the operator has not opted in to runtime component management, closes * the call with {@code PERMISSION_DENIED} and returns {@code true}. Guards the * Save/Delete fetcher and pipes-iterator RPCs. The caller must {@code return} * immediately when this returns {@code true}. @@ -199,13 +199,13 @@ private void startIgniteServer(ExtensionConfig config) { * reported to the client as {@code UNKNOWN}; only {@code onError} transmits * the intended status. */ - private boolean denyComponentModifications(StreamObserver responseObserver) { - if (tikaGrpcConfig.isAllowComponentModifications()) { + private boolean denyComponentManagement(StreamObserver responseObserver) { + if (tikaGrpcConfig.isAllowComponentManagement()) { return false; } responseObserver.onError(io.grpc.Status.PERMISSION_DENIED - .withDescription("Runtime component modifications are disabled. Set " - + "'allowComponentModifications' to true in the 'grpc' section of your " + .withDescription("Runtime component management is disabled. Set " + + "'allowComponentManagement' to true in the 'grpc' section of your " + "tika-config to allow SaveFetcher/DeleteFetcher/SavePipesIterator/" + "DeletePipesIterator. Understand the security implications first.") .asRuntimeException()); @@ -332,16 +332,23 @@ private void fetchAndParseImpl(FetchAndParseRequest request, @Override public void saveFetcher(SaveFetcherRequest request, StreamObserver responseObserver) { - if (denyComponentModifications(responseObserver)) { + if (denyComponentManagement(responseObserver)) { + return; + } + String fetcherType = request.getFetcherType(); + if (!isRegisteredFetcherType(fetcherType)) { + responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT + .withDescription("Unknown fetcher type: '" + fetcherType + + "'. Use the short factory name (e.g. 'file-system-fetcher').") + .asRuntimeException()); return; } SaveFetcherReply reply = SaveFetcherReply.newBuilder().setFetcherId(request.getFetcherId()).build(); try { - String factoryName = findFactoryNameForClass(request.getFetcherClass()); - ExtensionConfig config = new ExtensionConfig(request.getFetcherId(), factoryName, request.getFetcherConfigJson()); - - // Save to fetcher manager (updates ConfigStore which is shared with PipesServer) + // The fetcher type is the factory short name, used directly as the ConfigStore key + // (shared with PipesServer) -- no class resolution or construction needed. + ExtensionConfig config = new ExtensionConfig(request.getFetcherId(), fetcherType, request.getFetcherConfigJson()); fetcherManager.saveFetcher(config); } catch (Exception e) { throw new RuntimeException(e); @@ -350,27 +357,14 @@ public void saveFetcher(SaveFetcherRequest request, responseObserver.onCompleted(); } - private String findFactoryNameForClass(String className) throws TikaConfigException { - var factories = pluginManager.getExtensions(org.apache.tika.pipes.api.fetcher.FetcherFactory.class); - LOG.debug("Looking for factory that produces class: {}", className); - LOG.debug("Found {} factories", factories.size()); - for (var factory : factories) { - LOG.debug("Checking factory: {} (name={})", factory.getClass().getName(), factory.getName()); - try { - // Use a permissive config that should allow most factories to create an instance - // FileSystemFetcher requires basePath or allowAbsolutePaths - String tempJson = "{\"allowAbsolutePaths\": true}"; - ExtensionConfig tempConfig = new ExtensionConfig("temp", factory.getName(), tempJson); - Fetcher fetcher = factory.buildExtension(tempConfig); - LOG.debug("Factory {} produced: {}", factory.getName(), fetcher.getClass().getName()); - if (fetcher.getClass().getName().equals(className)) { - return factory.getName(); - } - } catch (Exception e) { - LOG.debug("Could not build fetcher for factory: {} - {}", factory.getName(), e.getMessage()); + private boolean isRegisteredFetcherType(String fetcherType) { + for (var factory : pluginManager.getExtensions( + org.apache.tika.pipes.api.fetcher.FetcherFactory.class)) { + if (factory.getName().equals(fetcherType)) { + return true; } } - throw new TikaConfigException("Could not find factory for class: " + className); + return false; } static Status notFoundStatus(String fetcherId) { return Status.newBuilder() @@ -388,12 +382,15 @@ public void getFetcher(GetFetcherRequest request, ExtensionConfig config = fetcher.getExtensionConfig(); getFetcherReply.setFetcherId(config.id()); - // Return the class name instead of the factory name for backward compatibility - getFetcherReply.setFetcherClass(fetcher.getClass().getName()); + getFetcherReply.setFetcherType(config.name()); - Map paramMap = OBJECT_MAPPER.readValue(config.json(), new TypeReference<>() { - }); - paramMap.forEach((k, v) -> getFetcherReply.putParams(Objects.toString(k), Objects.toString(v))); + // The config may carry secrets (passwords, access keys, tokens). Only return it once the + // operator has opted in to runtime component management; identity is always safe. + if (tikaGrpcConfig.isAllowComponentManagement()) { + Map paramMap = OBJECT_MAPPER.readValue(config.json(), new TypeReference<>() { + }); + paramMap.forEach((k, v) -> getFetcherReply.putParams(Objects.toString(k), Objects.toString(v))); + } responseObserver.onNext(getFetcherReply.build()); responseObserver.onCompleted(); @@ -406,16 +403,20 @@ public void getFetcher(GetFetcherRequest request, public void listFetchers(ListFetchersRequest request, StreamObserver responseObserver) { ListFetchersReply.Builder listFetchersReplyBuilder = ListFetchersReply.newBuilder(); + // The config may carry secrets; only include it once component management is enabled. + boolean includeConfig = tikaGrpcConfig.isAllowComponentManagement(); for (String fetcherId : fetcherManager.getSupported()) { try { Fetcher fetcher = fetcherManager.getFetcher(fetcherId); ExtensionConfig config = fetcher.getExtensionConfig(); - GetFetcherReply.Builder replyBuilder = GetFetcherReply.newBuilder().setFetcherId(config.id()).setFetcherClass(fetcher.getClass().getName()); + GetFetcherReply.Builder replyBuilder = GetFetcherReply.newBuilder().setFetcherId(config.id()).setFetcherType(config.name()); - Map paramMap = OBJECT_MAPPER.readValue(config.json(), new TypeReference<>() { - }); - paramMap.forEach((k, v) -> replyBuilder.putParams(Objects.toString(k), Objects.toString(v))); + if (includeConfig) { + Map paramMap = OBJECT_MAPPER.readValue(config.json(), new TypeReference<>() { + }); + paramMap.forEach((k, v) -> replyBuilder.putParams(Objects.toString(k), Objects.toString(v))); + } listFetchersReplyBuilder.addGetFetcherReplies(replyBuilder.build()); } catch (Exception e) { @@ -429,7 +430,7 @@ public void listFetchers(ListFetchersRequest request, @Override public void deleteFetcher(DeleteFetcherRequest request, StreamObserver responseObserver) { - if (denyComponentModifications(responseObserver)) { + if (denyComponentManagement(responseObserver)) { return; } boolean successfulDelete = deleteFetcher(request.getFetcherId()); @@ -440,11 +441,33 @@ public void deleteFetcher(DeleteFetcherRequest request, @Override public void getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest request, StreamObserver responseObserver) { GetFetcherConfigJsonSchemaReply.Builder builder = GetFetcherConfigJsonSchemaReply.newBuilder(); + String className = request.getFetcherClass(); + // Resolve without running static initializers, and only for real Fetcher implementations -- + // never load an arbitrary classpath class on a client's say-so. + Class fetcherClass; + try { + fetcherClass = Class.forName(className, false, getClass().getClassLoader()); + } catch (ClassNotFoundException | LinkageError e) { + responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT + .withDescription("Unknown fetcher class: " + className) + .asRuntimeException()); + return; + } + if (!Fetcher.class.isAssignableFrom(fetcherClass)) { + responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT + .withDescription("Not a fetcher class: " + className) + .asRuntimeException()); + return; + } try { - JsonSchema jsonSchema = JSON_SCHEMA_GENERATOR.generateSchema(Class.forName(request.getFetcherClass())); + JsonSchema jsonSchema = JSON_SCHEMA_GENERATOR.generateSchema(fetcherClass); builder.setFetcherConfigJsonSchema(OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(jsonSchema)); - } catch (ClassNotFoundException | JsonProcessingException e) { - throw new RuntimeException("Could not create json schema for " + request.getFetcherClass(), e); + } catch (JsonProcessingException e) { + responseObserver.onError(io.grpc.Status.INTERNAL + .withDescription("Could not create json schema for " + className) + .withCause(e) + .asRuntimeException()); + return; } responseObserver.onNext(builder.build()); responseObserver.onCompleted(); @@ -467,17 +490,17 @@ private boolean deleteFetcher(String id) { @Override public void savePipesIterator(SavePipesIteratorRequest request, StreamObserver responseObserver) { - if (denyComponentModifications(responseObserver)) { + if (denyComponentManagement(responseObserver)) { return; } try { String iteratorId = request.getIteratorId(); - String iteratorClass = request.getIteratorClass(); + String iteratorType = request.getIteratorType(); String iteratorConfigJson = request.getIteratorConfigJson(); - LOG.info("Saving pipes iterator: id={}, class={}", iteratorId, iteratorClass); + LOG.info("Saving pipes iterator: id={}, type={}", iteratorId, iteratorType); - ExtensionConfig config = new ExtensionConfig(iteratorId, iteratorClass, iteratorConfigJson); + ExtensionConfig config = new ExtensionConfig(iteratorId, iteratorType, iteratorConfigJson); // Save directly to ConfigStore (shared with PipesServer) configStore.put(PIPES_ITERATOR_PREFIX + iteratorId, config); @@ -516,12 +539,15 @@ public void getPipesIterator(GetPipesIteratorRequest request, return; } - GetPipesIteratorReply reply = GetPipesIteratorReply.newBuilder() + GetPipesIteratorReply.Builder reply = GetPipesIteratorReply.newBuilder() .setIteratorId(config.id()) - .setIteratorClass(config.name()) - .setIteratorConfigJson(config.json()) - .build(); - responseObserver.onNext(reply); + .setIteratorType(config.name()); + // The iterator config may carry secrets; only include it once component management is + // enabled (identity is always safe to return). + if (tikaGrpcConfig.isAllowComponentManagement()) { + reply.setIteratorConfigJson(config.json()); + } + responseObserver.onNext(reply.build()); responseObserver.onCompleted(); LOG.info("Successfully retrieved pipes iterator: {}", iteratorId); @@ -538,7 +564,7 @@ public void getPipesIterator(GetPipesIteratorRequest request, @Override public void deletePipesIterator(DeletePipesIteratorRequest request, StreamObserver responseObserver) { - if (denyComponentModifications(responseObserver)) { + if (denyComponentManagement(responseObserver)) { return; } try { diff --git a/tika-grpc/src/main/proto/tika.proto b/tika-grpc/src/main/proto/tika.proto index 1979b2b162..7fde9a0643 100644 --- a/tika-grpc/src/main/proto/tika.proto +++ b/tika-grpc/src/main/proto/tika.proto @@ -77,9 +77,9 @@ service Tika { message SaveFetcherRequest { // A unique identifier for each fetcher. If this already exists, operation will overwrite existing. string fetcher_id = 1; - // The full java class name of the fetcher class. List of - // fetcher classes is found here: https://cwiki.apache.org/confluence/display/TIKA/tika-pipes - string fetcher_class = 2; + // The registered fetcher type -- the short factory name, e.g. "file-system-fetcher", + // "http-fetcher", "s3-fetcher". These are the same names used as fetcher keys in tika-config. + string fetcher_type = 2; // JSON string of the fetcher config object. To see the json schema from which to build this json, // use the GetFetcherConfigJsonSchema rpc method. string fetcher_config_json = 3; @@ -137,8 +137,8 @@ message GetFetcherRequest { message GetFetcherReply { // Echoes the ID of the fetcher being returned. string fetcher_id = 1; - // The full Java class name of the Fetcher. - string fetcher_class = 2; + // The registered fetcher type (the short factory name, e.g. "file-system-fetcher"). + string fetcher_type = 2; // The configuration parameters. map params = 3; } @@ -170,8 +170,9 @@ message GetFetcherConfigJsonSchemaReply { message SavePipesIteratorRequest { // A unique identifier for each pipes iterator. If this already exists, operation will overwrite existing. string iterator_id = 1; - // The full java class name of the pipes iterator class. - string iterator_class = 2; + // The registered pipes-iterator type -- the short factory name, e.g. + // "file-system-pipes-iterator". These are the same names used as keys in tika-config. + string iterator_type = 2; // JSON string of the pipes iterator config object. string iterator_config_json = 3; } @@ -189,8 +190,8 @@ message GetPipesIteratorRequest { message GetPipesIteratorReply { // The pipes iterator ID string iterator_id = 1; - // The full java class name of the pipes iterator - string iterator_class = 2; + // The registered pipes-iterator type (the short factory name, e.g. "file-system-pipes-iterator"). + string iterator_type = 2; // JSON string of the pipes iterator config object string iterator_config_json = 3; } diff --git a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java index 04d57defce..abd3c95657 100644 --- a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java +++ b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java @@ -65,8 +65,12 @@ import org.apache.tika.DeleteFetcherRequest; import org.apache.tika.FetchAndParseReply; import org.apache.tika.FetchAndParseRequest; +import org.apache.tika.GetFetcherConfigJsonSchemaReply; +import org.apache.tika.GetFetcherConfigJsonSchemaRequest; import org.apache.tika.GetFetcherReply; import org.apache.tika.GetFetcherRequest; +import org.apache.tika.ListFetchersReply; +import org.apache.tika.ListFetchersRequest; import org.apache.tika.SaveFetcherReply; import org.apache.tika.SaveFetcherRequest; import org.apache.tika.TikaGrpc; @@ -116,7 +120,7 @@ static void init() throws Exception { // for the tests that add/modify fetchers at runtime. ObjectNode root = (ObjectNode) OBJECT_MAPPER.readTree(tikaConfig.toFile()); ObjectNode grpc = OBJECT_MAPPER.createObjectNode(); - grpc.put("allowComponentModifications", true); + grpc.put("allowComponentManagement", true); grpc.put("allowPerRequestConfig", true); root.set("grpc", grpc); FileUtils.write(tikaConfigUnlocked.toFile(), @@ -162,7 +166,7 @@ public void testFetcherCrud(Resources resources) throws Exception { SaveFetcherReply reply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(fetcherId) - .setFetcherClass(FileSystemFetcher.class.getName()) + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap .builder() .put("basePath", targetFolder) @@ -177,7 +181,7 @@ public void testFetcherCrud(Resources resources) throws Exception { SaveFetcherReply reply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(fetcherId) - .setFetcherClass(FileSystemFetcher.class.getName()) + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap .builder() .put("basePath", targetFolder) @@ -202,7 +206,7 @@ public void testFetcherCrud(Resources resources) throws Exception { .setFetcherId(fetcherId) .build()); assertEquals(fetcherId, getFetcherReply.getFetcherId()); - assertEquals(FileSystemFetcher.class.getName(), getFetcherReply.getFetcherClass()); + assertEquals("file-system-fetcher", getFetcherReply.getFetcherType()); } // delete fetchers @@ -223,7 +227,7 @@ private static String createFetcherId(int i) { } @Test - public void testComponentModificationsDeniedByDefault(Resources resources) throws Exception { + public void testComponentManagementDeniedByDefault(Resources resources) throws Exception { TikaGrpc.TikaBlockingStub blockingStub = startServer(resources, tikaConfig); String targetFolder = new File("target").getAbsolutePath(); @@ -231,7 +235,7 @@ public void testComponentModificationsDeniedByDefault(Resources resources) throw blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(createFetcherId(0)) - .setFetcherClass(FileSystemFetcher.class.getName()) + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap .builder() .put("basePath", targetFolder) @@ -262,6 +266,55 @@ public void testPerRequestConfigDeniedByDefault(Resources resources) throws Exce assertEquals(Status.Code.PERMISSION_DENIED, ex.getStatus().getCode()); } + @Test + public void testComponentManagementHidesConfigByDefault(Resources resources) throws Exception { + // With component management off (the default), the read RPCs must return component identity + // only -- never the stored config, which can carry secrets (passwords, access keys, ...). + TikaGrpc.TikaBlockingStub blockingStub = startServer(resources, tikaConfig); + + GetFetcherReply reply = blockingStub.getFetcher(GetFetcherRequest + .newBuilder() + .setFetcherId(createFetcherId(1)) + .build()); + assertEquals(createFetcherId(1), reply.getFetcherId()); + Assertions.assertFalse(reply.getFetcherType().isEmpty(), + "identity (type) should still be returned"); + Assertions.assertTrue(reply.getParamsMap().isEmpty(), + "config params must NOT be returned when component management is off"); + + ListFetchersReply listReply = + blockingStub.listFetchers(ListFetchersRequest.newBuilder().build()); + Assertions.assertFalse(listReply.getGetFetcherRepliesList().isEmpty(), + "fetchers should still be listed by identity"); + for (GetFetcherReply f : listReply.getGetFetcherRepliesList()) { + Assertions.assertTrue(f.getParamsMap().isEmpty(), + "listFetchers must not leak config for " + f.getFetcherId()); + } + } + + @Test + public void testFetcherConfigJsonSchemaRejectsNonFetcherClass(Resources resources) + throws Exception { + TikaGrpc.TikaBlockingStub blockingStub = startServer(resources, tikaConfig); + // The schema RPC must never load an arbitrary class on a client's say-so: a non-Fetcher + // class is rejected outright, and no static initializer is ever run for it. + StatusRuntimeException ex = Assertions.assertThrows(StatusRuntimeException.class, () -> + blockingStub.getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest + .newBuilder() + .setFetcherClass("java.lang.Runtime") + .build())); + assertEquals(Status.Code.INVALID_ARGUMENT, ex.getStatus().getCode()); + + // A real fetcher class still yields a schema. + GetFetcherConfigJsonSchemaReply ok = blockingStub.getFetcherConfigJsonSchema( + GetFetcherConfigJsonSchemaRequest + .newBuilder() + .setFetcherClass(FileSystemFetcher.class.getName()) + .build()); + Assertions.assertFalse(ok.getFetcherConfigJsonSchema().isEmpty(), + "a real fetcher class should still produce a schema"); + } + private static TikaGrpc.TikaBlockingStub startServer(Resources resources, Path config) throws Exception { String serverName = InProcessServerBuilder.generateName(); @@ -307,7 +360,7 @@ public void testBiStream(Resources resources) throws Exception { SaveFetcherReply reply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(fetcherId) - .setFetcherClass(FileSystemFetcher.class.getName()) + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(OBJECT_MAPPER.writeValueAsString(ImmutableMap .builder() .put("basePath", targetFolder) diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java index 564beff9af..c14ee77f30 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitter.java @@ -75,6 +75,14 @@ private void checkConfig(FileSystemEmitterConfig fileSystemEmitterConfig) throws if (fileSystemEmitterConfig.onExists() == null) { throw new TikaConfigException("Must configure 'onExists' as 'skip', 'exception' or 'replace'"); } + if (StringUtils.isBlank(fileSystemEmitterConfig.basePath()) + && !fileSystemEmitterConfig.allowAbsolutePaths()) { + throw new TikaConfigException( + "'basePath' must be set, or 'allowAbsolutePaths' must be true. " + + "Without basePath, clients can write to any file this process " + + "has access to. Set 'allowAbsolutePaths: true' to explicitly " + + "allow this behavior and accept the security risks."); + } } @Override @@ -199,9 +207,10 @@ private FileSystemEmitterConfig getConfig(ParseContext parseContext) throws Tika // Load runtime config (excludes basePath for security) FileSystemEmitterRuntimeConfig runtimeConfig = FileSystemEmitterRuntimeConfig.load(configJson.json()); - // Merge runtime config into default config while preserving basePath + // Merge runtime config into default config while preserving basePath and the + // init-time allowAbsolutePaths -- neither may be changed at runtime. config = new FileSystemEmitterConfig(fileSystemEmitterConfig.basePath(), runtimeConfig.getFileExtension(), runtimeConfig.getOnExists(), - runtimeConfig.isPrettyPrint()); + runtimeConfig.isPrettyPrint(), fileSystemEmitterConfig.allowAbsolutePaths()); checkConfig(config); } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java index d7cf621828..62d7f279d5 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterConfig.java @@ -21,7 +21,7 @@ import org.apache.tika.exception.TikaConfigException; -public record FileSystemEmitterConfig(String basePath, String fileExtension, ON_EXISTS onExists, boolean prettyPrint) { +public record FileSystemEmitterConfig(String basePath, String fileExtension, ON_EXISTS onExists, boolean prettyPrint, boolean allowAbsolutePaths) { enum ON_EXISTS { SKIP, EXCEPTION, REPLACE diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java new file mode 100644 index 0000000000..7695a193a9 --- /dev/null +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/test/java/org/apache/tika/pipes/emitter/fs/FileSystemEmitterTest.java @@ -0,0 +1,85 @@ +/* + * 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.tika.pipes.emitter.fs; + +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.List; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import org.apache.tika.exception.TikaConfigException; +import org.apache.tika.metadata.Metadata; +import org.apache.tika.parser.ParseContext; +import org.apache.tika.pipes.api.emitter.Emitter; +import org.apache.tika.plugins.ExtensionConfig; + +public class FileSystemEmitterTest { + + private static final ObjectMapper MAPPER = new ObjectMapper(); + + @TempDir + Path tempDir; + + private Emitter createEmitter(Path basePath, Boolean allowAbsolutePaths) + throws TikaConfigException, IOException { + ObjectNode config = MAPPER.createObjectNode(); + if (basePath != null) { + config.put("basePath", basePath.toAbsolutePath().toString()); + } + if (allowAbsolutePaths != null) { + config.put("allowAbsolutePaths", allowAbsolutePaths); + } + config.put("onExists", "REPLACE"); + ExtensionConfig pluginConfig = new ExtensionConfig("test", "test", config.toString()); + return new FileSystemEmitterFactory().buildExtension(pluginConfig); + } + + @Test + public void testAllowAbsolutePathsRequired() throws Exception { + // Without basePath and without allowAbsolutePaths, the emitter would write client-controlled + // keys to arbitrary paths -- build must refuse it (mirrors FileSystemFetcher). + assertThrows(TikaConfigException.class, () -> createEmitter(null, null)); + } + + @Test + public void testAllowAbsolutePathsWorks() throws Exception { + // With allowAbsolutePaths=true and no basePath, the operator has explicitly accepted the + // risk, so an absolute emit key is written. + Emitter emitter = createEmitter(null, true); + Path out = tempDir.resolve("out/result.json"); + emitter.emit(out.toAbsolutePath().toString(), List.of(new Metadata()), new ParseContext()); + assertTrue(Files.isRegularFile(out), "absolute emit key should have been written"); + } + + @Test + public void testPathTraversalBlocked() throws Exception { + Path basePath = tempDir.resolve("allowed"); + Files.createDirectories(basePath); + Emitter emitter = createEmitter(basePath, null); + // An emit key escaping basePath must be rejected, even with basePath set. + assertThrows(IOException.class, () -> emitter.emit( + "../escaped.json", List.of(new Metadata()), new ParseContext())); + } +} From 2c07998b78b2b5507fc81bdb5d38a9e7ed2bd95f Mon Sep 17 00:00:00 2001 From: tallison Date: Fri, 26 Jun 2026 18:04:22 -0400 Subject: [PATCH 3/4] TIKA-4763 fix e2e --- tika-e2e-tests/pom.xml | 9 +++++ .../filesystem/FileSystemFetcherTest.java | 2 +- .../pipes/filesystem/HandlerTypeTest.java | 2 +- .../pipes/ignite/IgniteConfigStoreTest.java | 2 +- .../tika/pipes/grpc/TikaGrpcServerImpl.java | 37 +++++++++---------- tika-grpc/src/main/proto/tika.proto | 5 ++- .../tika/pipes/grpc/TikaGrpcServerTest.java | 14 +++---- .../pipes/api/fetcher/FetcherFactory.java | 5 +++ .../AtlassianJwtFetcherFactory.java | 6 +++ .../fetcher/azblob/AZBlobFetcherFactory.java | 6 +++ .../fetcher/fs/FileSystemFetcherFactory.java | 5 +++ .../pipes/fetcher/gcs/GCSFetcherFactory.java | 6 +++ .../GoogleDriveFetcherFactory.java | 6 +++ .../fetcher/http/HttpFetcherFactory.java | 6 +++ .../MicrosoftGraphFetcherFactory.java | 6 +++ .../pipes/fetcher/s3/S3FetcherFactory.java | 6 +++ 16 files changed, 91 insertions(+), 32 deletions(-) diff --git a/tika-e2e-tests/pom.xml b/tika-e2e-tests/pom.xml index f506d549c1..06c3e5cb7f 100644 --- a/tika-e2e-tests/pom.xml +++ b/tika-e2e-tests/pom.xml @@ -123,6 +123,15 @@ 3.15.0 17 + + + + org.projectlombok + lombok + ${lombok.version} + + diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java index 9947763e08..92bfc72575 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java @@ -64,7 +64,7 @@ void testFileSystemFetcher() throws Exception { SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(fetcherId) - .setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher") + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(configJson) .build()); diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java index ea1f11f52b..f8e4799331 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java @@ -297,7 +297,7 @@ void testParseContextJson() throws Exception { SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest.newBuilder() .setFetcherId(fetcherId) - .setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher") + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(ExternalTestBase.OBJECT_MAPPER.writeValueAsString(config)) .build()); log.info("Fetcher created: {}", saveReply.getFetcherId()); diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java index fb986362a9..eecfe2a0a2 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ignite/IgniteConfigStoreTest.java @@ -409,7 +409,7 @@ void testIgniteConfigStore() throws Exception { SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() .setFetcherId(fetcherId) - .setFetcherClass("org.apache.tika.pipes.fetcher.fs.FileSystemFetcher") + .setFetcherType("file-system-fetcher") .setFetcherConfigJson(configJson) .build()); diff --git a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java index 49013b6a6b..1f4c842421 100644 --- a/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java +++ b/tika-grpc/src/main/java/org/apache/tika/pipes/grpc/TikaGrpcServerImpl.java @@ -66,6 +66,7 @@ import org.apache.tika.pipes.api.emitter.EmitKey; import org.apache.tika.pipes.api.fetcher.FetchKey; import org.apache.tika.pipes.api.fetcher.Fetcher; +import org.apache.tika.pipes.api.fetcher.FetcherFactory; import org.apache.tika.pipes.core.PipesClient; import org.apache.tika.pipes.core.PipesConfig; import org.apache.tika.pipes.core.config.ConfigStore; @@ -358,13 +359,16 @@ public void saveFetcher(SaveFetcherRequest request, } private boolean isRegisteredFetcherType(String fetcherType) { - for (var factory : pluginManager.getExtensions( - org.apache.tika.pipes.api.fetcher.FetcherFactory.class)) { + return findFetcherFactory(fetcherType) != null; + } + + private FetcherFactory findFetcherFactory(String fetcherType) { + for (FetcherFactory factory : pluginManager.getExtensions(FetcherFactory.class)) { if (factory.getName().equals(fetcherType)) { - return true; + return factory; } } - return false; + return null; } static Status notFoundStatus(String fetcherId) { return Status.newBuilder() @@ -441,30 +445,23 @@ public void deleteFetcher(DeleteFetcherRequest request, @Override public void getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest request, StreamObserver responseObserver) { GetFetcherConfigJsonSchemaReply.Builder builder = GetFetcherConfigJsonSchemaReply.newBuilder(); - String className = request.getFetcherClass(); - // Resolve without running static initializers, and only for real Fetcher implementations -- - // never load an arbitrary classpath class on a client's say-so. - Class fetcherClass; - try { - fetcherClass = Class.forName(className, false, getClass().getClassLoader()); - } catch (ClassNotFoundException | LinkageError e) { - responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT - .withDescription("Unknown fetcher class: " + className) - .asRuntimeException()); - return; - } - if (!Fetcher.class.isAssignableFrom(fetcherClass)) { + String fetcherType = request.getFetcherType(); + // Only resolve config classes from registered fetcher factories -- never load an arbitrary + // classpath class on a client's say-so. + FetcherFactory factory = findFetcherFactory(fetcherType); + if (factory == null) { responseObserver.onError(io.grpc.Status.INVALID_ARGUMENT - .withDescription("Not a fetcher class: " + className) + .withDescription("Unknown fetcher type: '" + fetcherType + + "'. Use the short factory name (e.g. 'file-system-fetcher').") .asRuntimeException()); return; } try { - JsonSchema jsonSchema = JSON_SCHEMA_GENERATOR.generateSchema(fetcherClass); + JsonSchema jsonSchema = JSON_SCHEMA_GENERATOR.generateSchema(factory.getConfigClass()); builder.setFetcherConfigJsonSchema(OBJECT_MAPPER.writerWithDefaultPrettyPrinter().writeValueAsString(jsonSchema)); } catch (JsonProcessingException e) { responseObserver.onError(io.grpc.Status.INTERNAL - .withDescription("Could not create json schema for " + className) + .withDescription("Could not create json schema for fetcher type " + fetcherType) .withCause(e) .asRuntimeException()); return; diff --git a/tika-grpc/src/main/proto/tika.proto b/tika-grpc/src/main/proto/tika.proto index 7fde9a0643..7d365f95d1 100644 --- a/tika-grpc/src/main/proto/tika.proto +++ b/tika-grpc/src/main/proto/tika.proto @@ -156,8 +156,9 @@ message ListFetchersReply { } message GetFetcherConfigJsonSchemaRequest { - // The full java class name of the fetcher config for which to fetch json schema. - string fetcher_class = 1; + // The registered fetcher type -- the short factory name, e.g. "file-system-fetcher" -- + // for which to fetch the config json schema. Same names used as fetcher keys in tika-config. + string fetcher_type = 1; } message GetFetcherConfigJsonSchemaReply { diff --git a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java index abd3c95657..66d6a250fe 100644 --- a/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java +++ b/tika-grpc/src/test/java/org/apache/tika/pipes/grpc/TikaGrpcServerTest.java @@ -293,26 +293,26 @@ public void testComponentManagementHidesConfigByDefault(Resources resources) thr } @Test - public void testFetcherConfigJsonSchemaRejectsNonFetcherClass(Resources resources) + public void testFetcherConfigJsonSchemaRejectsUnknownType(Resources resources) throws Exception { TikaGrpc.TikaBlockingStub blockingStub = startServer(resources, tikaConfig); - // The schema RPC must never load an arbitrary class on a client's say-so: a non-Fetcher - // class is rejected outright, and no static initializer is ever run for it. + // The schema RPC only resolves config classes from registered fetcher factories: an + // unknown type is rejected outright, and no arbitrary class is ever loaded. StatusRuntimeException ex = Assertions.assertThrows(StatusRuntimeException.class, () -> blockingStub.getFetcherConfigJsonSchema(GetFetcherConfigJsonSchemaRequest .newBuilder() - .setFetcherClass("java.lang.Runtime") + .setFetcherType("no-such-fetcher") .build())); assertEquals(Status.Code.INVALID_ARGUMENT, ex.getStatus().getCode()); - // A real fetcher class still yields a schema. + // A registered fetcher type still yields a schema. GetFetcherConfigJsonSchemaReply ok = blockingStub.getFetcherConfigJsonSchema( GetFetcherConfigJsonSchemaRequest .newBuilder() - .setFetcherClass(FileSystemFetcher.class.getName()) + .setFetcherType("file-system-fetcher") .build()); Assertions.assertFalse(ok.getFetcherConfigJsonSchema().isEmpty(), - "a real fetcher class should still produce a schema"); + "a registered fetcher type should still produce a schema"); } private static TikaGrpc.TikaBlockingStub startServer(Resources resources, Path config) diff --git a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/fetcher/FetcherFactory.java b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/fetcher/FetcherFactory.java index 4fef27c4b7..a9abf41db3 100644 --- a/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/fetcher/FetcherFactory.java +++ b/tika-pipes/tika-pipes-api/src/main/java/org/apache/tika/pipes/api/fetcher/FetcherFactory.java @@ -20,4 +20,9 @@ public interface FetcherFactory extends TikaExtensionFactory { + /** + * @return the config POJO class for this fetcher; used to generate the JSON schema that + * describes the {@code fetcher_config_json} a client must supply when saving a fetcher. + */ + Class getConfigClass(); } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-atlassian-jwt/src/main/java/org/apache/tika/pipes/fetcher/atlassianjwt/AtlassianJwtFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-atlassian-jwt/src/main/java/org/apache/tika/pipes/fetcher/atlassianjwt/AtlassianJwtFetcherFactory.java index 11b15fbe2a..145bc80bee 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-atlassian-jwt/src/main/java/org/apache/tika/pipes/fetcher/atlassianjwt/AtlassianJwtFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-atlassian-jwt/src/main/java/org/apache/tika/pipes/fetcher/atlassianjwt/AtlassianJwtFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.atlassianjwt.config.AtlassianJwtFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -55,4 +56,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return AtlassianJwtFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return AtlassianJwtFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-az-blob/src/main/java/org/apache/tika/pipes/fetcher/azblob/AZBlobFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-az-blob/src/main/java/org/apache/tika/pipes/fetcher/azblob/AZBlobFetcherFactory.java index 2598f01aef..5508e69be5 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-az-blob/src/main/java/org/apache/tika/pipes/fetcher/azblob/AZBlobFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-az-blob/src/main/java/org/apache/tika/pipes/fetcher/azblob/AZBlobFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.azblob.config.AZBlobFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -56,4 +57,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return AZBlobFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return AZBlobFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/fetcher/fs/FileSystemFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/fetcher/fs/FileSystemFetcherFactory.java index 30c1244850..6483f32fe7 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/fetcher/fs/FileSystemFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-file-system/src/main/java/org/apache/tika/pipes/fetcher/fs/FileSystemFetcherFactory.java @@ -54,4 +54,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return new FileSystemFetcher(extensionConfig); } + + @Override + public Class getConfigClass() { + return FileSystemFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-gcs/src/main/java/org/apache/tika/pipes/fetcher/gcs/GCSFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-gcs/src/main/java/org/apache/tika/pipes/fetcher/gcs/GCSFetcherFactory.java index d21c9528de..fceca509c5 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-gcs/src/main/java/org/apache/tika/pipes/fetcher/gcs/GCSFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-gcs/src/main/java/org/apache/tika/pipes/fetcher/gcs/GCSFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.gcs.config.GCSFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -56,4 +57,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return GCSFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return GCSFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-google-drive/src/main/java/org/apache/tika/pipes/fetcher/googledrive/GoogleDriveFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-google-drive/src/main/java/org/apache/tika/pipes/fetcher/googledrive/GoogleDriveFetcherFactory.java index c1cc483ca3..a213484b72 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-google-drive/src/main/java/org/apache/tika/pipes/fetcher/googledrive/GoogleDriveFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-google-drive/src/main/java/org/apache/tika/pipes/fetcher/googledrive/GoogleDriveFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.googledrive.config.GoogleDriveFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -54,4 +55,9 @@ public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return GoogleDriveFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return GoogleDriveFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcherFactory.java index 5537314bbd..d2df6bdfb6 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-http/src/main/java/org/apache/tika/pipes/fetcher/http/HttpFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.http.config.HttpFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -56,4 +57,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return HttpFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return HttpFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-microsoft-graph/src/main/java/org/apache/tika/pipes/fetchers/microsoftgraph/MicrosoftGraphFetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-microsoft-graph/src/main/java/org/apache/tika/pipes/fetchers/microsoftgraph/MicrosoftGraphFetcherFactory.java index 3647c0245e..a0b4e88c1b 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-microsoft-graph/src/main/java/org/apache/tika/pipes/fetchers/microsoftgraph/MicrosoftGraphFetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-microsoft-graph/src/main/java/org/apache/tika/pipes/fetchers/microsoftgraph/MicrosoftGraphFetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetchers.microsoftgraph.config.MicrosoftGraphFetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -59,4 +60,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return MicrosoftGraphFetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return MicrosoftGraphFetcherConfig.class; + } } diff --git a/tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3FetcherFactory.java b/tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3FetcherFactory.java index 75c8264399..fd98da1e22 100644 --- a/tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3FetcherFactory.java +++ b/tika-pipes/tika-pipes-plugins/tika-pipes-s3/src/main/java/org/apache/tika/pipes/fetcher/s3/S3FetcherFactory.java @@ -23,6 +23,7 @@ import org.apache.tika.exception.TikaConfigException; import org.apache.tika.pipes.api.fetcher.Fetcher; import org.apache.tika.pipes.api.fetcher.FetcherFactory; +import org.apache.tika.pipes.fetcher.s3.config.S3FetcherConfig; import org.apache.tika.plugins.ExtensionConfig; /** @@ -57,4 +58,9 @@ public String getName() { public Fetcher buildExtension(ExtensionConfig extensionConfig) throws IOException, TikaConfigException { return S3Fetcher.build(extensionConfig); } + + @Override + public Class getConfigClass() { + return S3FetcherConfig.class; + } } From 53bbcb9bb92ce668594083a3c7beeac4b99bf367 Mon Sep 17 00:00:00 2001 From: tallison Date: Thu, 2 Jul 2026 09:29:15 -0400 Subject: [PATCH 4/4] follow on to TIKA-4763 fix e2e --- tika-e2e-tests/pom.xml | 17 ---- tika-e2e-tests/tika-grpc/pom.xml | 7 -- .../apache/tika/pipes/ExternalTestBase.java | 43 ++++++----- .../filesystem/FileSystemFetcherTest.java | 24 +++--- .../pipes/filesystem/HandlerTypeTest.java | 41 +++++----- .../pipes/ignite/IgniteConfigStoreTest.java | 77 ++++++++++--------- tika-parent/pom.xml | 1 - 7 files changed, 95 insertions(+), 115 deletions(-) diff --git a/tika-e2e-tests/pom.xml b/tika-e2e-tests/pom.xml index 06c3e5cb7f..5f60869446 100644 --- a/tika-e2e-tests/pom.xml +++ b/tika-e2e-tests/pom.xml @@ -103,14 +103,6 @@ jackson-databind ${jackson.version} - - - - org.projectlombok - lombok - ${lombok.version} - true - @@ -123,15 +115,6 @@ 3.15.0 17 - - - - org.projectlombok - lombok - ${lombok.version} - - diff --git a/tika-e2e-tests/tika-grpc/pom.xml b/tika-e2e-tests/tika-grpc/pom.xml index 4dd87ed12c..e4c91463e4 100644 --- a/tika-e2e-tests/tika-grpc/pom.xml +++ b/tika-e2e-tests/tika-grpc/pom.xml @@ -70,13 +70,6 @@ jackson-databind - - - org.projectlombok - lombok - true - - org.junit.jupiter diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java index d0d135309d..edcfd9a445 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/ExternalTestBase.java @@ -43,12 +43,13 @@ import com.fasterxml.jackson.databind.ObjectMapper; import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; -import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.TestInstance; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.testcontainers.containers.DockerComposeContainer; import org.testcontainers.containers.output.Slf4jLogConsumer; import org.testcontainers.containers.wait.strategy.Wait; @@ -60,9 +61,9 @@ @TestInstance(TestInstance.Lifecycle.PER_CLASS) @Testcontainers -@Slf4j @Tag("E2ETest") public abstract class ExternalTestBase { + private static final Logger LOG = LoggerFactory.getLogger(ExternalTestBase.class); public static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); public static final int MAX_STARTUP_TIMEOUT = 120; public static final String GOV_DOCS_FOLDER = "/tika/govdocs1"; @@ -88,7 +89,7 @@ static void setup() throws Exception { } private static void startLocalGrpcServer() throws Exception { - log.info("Starting local tika-grpc server using Maven exec"); + LOG.info("Starting local tika-grpc server using Maven exec"); Path tikaGrpcDir = findTikaGrpcDirectory(); Path configFile = Path.of("src/test/resources/tika-config.json").toAbsolutePath(); @@ -97,8 +98,8 @@ private static void startLocalGrpcServer() throws Exception { throw new IllegalStateException("Config file not found: " + configFile); } - log.info("Using tika-grpc from: {}", tikaGrpcDir); - log.info("Using config file: {}", configFile); + LOG.info("Using tika-grpc from: {}", tikaGrpcDir); + LOG.info("Using config file: {}", configFile); String javaHome = System.getProperty("java.home"); boolean isWindows = System.getProperty("os.name").toLowerCase(Locale.ROOT).contains("win"); @@ -131,10 +132,10 @@ private static void startLocalGrpcServer() throws Exception { new InputStreamReader(localGrpcProcess.getInputStream(), StandardCharsets.UTF_8))) { String line; while ((line = reader.readLine()) != null) { - log.info("tika-grpc: {}", line); + LOG.info("tika-grpc: {}", line); } } catch (IOException e) { - log.error("Error reading server output", e); + LOG.error("Error reading server output", e); } }); logThread.setDaemon(true); @@ -142,7 +143,7 @@ private static void startLocalGrpcServer() throws Exception { waitForServerReady(); - log.info("Local tika-grpc server started successfully on port {}", GRPC_PORT); + LOG.info("Local tika-grpc server started successfully on port {}", GRPC_PORT); } private static Path findTikaGrpcDirectory() { @@ -173,10 +174,10 @@ private static void waitForServerReady() throws Exception { try { TikaGrpc.TikaBlockingStub stub = TikaGrpc.newBlockingStub(testChannel); stub.listFetchers(ListFetchersRequest.newBuilder().build()); - log.info("gRPC server is ready"); + LOG.info("gRPC server is ready"); return; } catch (Exception e) { - log.trace("gRPC server not ready yet (attempt {}/{}): {}", i + 1, maxAttempts, e.getMessage()); + LOG.trace("gRPC server not ready yet (attempt {}/{}): {}", i + 1, maxAttempts, e.getMessage()); } finally { testChannel.shutdown(); testChannel.awaitTermination(1, TimeUnit.SECONDS); @@ -191,7 +192,7 @@ private static void waitForServerReady() throws Exception { } private static void startDockerGrpcServer() { - log.info("Starting Docker Compose tika-grpc server"); + LOG.info("Starting Docker Compose tika-grpc server"); String composeFilePath = System.getProperty("tika.docker.compose.file"); if (composeFilePath == null || composeFilePath.isBlank()) { @@ -208,11 +209,11 @@ private static void startDockerGrpcServer() { .withStartupTimeout(Duration.of(MAX_STARTUP_TIMEOUT, ChronoUnit.SECONDS)) .withExposedService("tika-grpc", 50052, Wait.forLogMessage(".*Server started.*\\n", 1)) - .withLogConsumer("tika-grpc", new Slf4jLogConsumer(log)); + .withLogConsumer("tika-grpc", new Slf4jLogConsumer(LOG)); composeContainer.start(); - log.info("Docker Compose containers started successfully"); + LOG.info("Docker Compose containers started successfully"); } private static void loadGovdocs1() throws IOException, InterruptedException { @@ -230,7 +231,7 @@ private static void loadGovdocs1() throws IOException, InterruptedException { if (attempt >= retries) { throw e; } - log.warn("Download attempt {} failed, retrying in 10 seconds...", attempt, e); + LOG.warn("Download attempt {} failed, retrying in 10 seconds...", attempt, e); TimeUnit.SECONDS.sleep(10); } } @@ -253,13 +254,13 @@ public static void copyTestFixtures() throws IOException { Files.copy(in, targetDir.resolve(fixture), StandardCopyOption.REPLACE_EXISTING); } } - log.info("Copied {} test fixtures to {}", fixtures.length, targetDir); + LOG.info("Copied {} test fixtures to {}", fixtures.length, targetDir); } @AfterAll void close() { if (USE_LOCAL_SERVER && localGrpcProcess != null) { - log.info("Stopping local gRPC server"); + LOG.info("Stopping local gRPC server"); localGrpcProcess.destroy(); try { if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) { @@ -284,14 +285,14 @@ public static void downloadAndUnzipGovdocs1(int fromIndex, int toIndex) throws I Path zipPath = targetDir.resolve(zipName); if (Files.exists(zipPath)) { - log.info("{} already exists, skipping download", zipName); + LOG.info("{} already exists, skipping download", zipName); } else { - log.info("Downloading {} from {}...", zipName, url); + LOG.info("Downloading {} from {}...", zipName, url); try (InputStream in = new URL(url).openStream()) { Files.copy(in, zipPath, StandardCopyOption.REPLACE_EXISTING); } } - log.info("Unzipping {}...", zipName); + LOG.info("Unzipping {}...", zipName); try (ZipInputStream zis = new ZipInputStream(new FileInputStream(zipPath.toFile()))) { ZipEntry entry; while ((entry = zis.getNextEntry()) != null) { @@ -309,7 +310,7 @@ public static void downloadAndUnzipGovdocs1(int fromIndex, int toIndex) throws I } } - log.info("Finished downloading and extracting govdocs1 files"); + LOG.info("Finished downloading and extracting govdocs1 files"); } public static void assertAllFilesFetched(Path baseDir, List successes, @@ -337,7 +338,7 @@ public static void assertAllFilesFetched(Path baseDir, List } Assertions.assertNotEquals(0, successes.size(), "Should have some successful fetches"); - log.info("Processed {} files: {} successes, {} errors", allFetchKeys.size(), successes.size(), errors.size()); + LOG.info("Processed {} files: {} successes, {} errors", allFetchKeys.size(), successes.size(), errors.size()); Assertions.assertEquals(keysFromGovdocs1, allFetchKeys, () -> { Set missing = new HashSet<>(keysFromGovdocs1); missing.removeAll(allFetchKeys); diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java index 92bfc72575..065aa7890a 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/FileSystemFetcherTest.java @@ -27,11 +27,12 @@ import io.grpc.ManagedChannel; import io.grpc.stub.StreamObserver; -import lombok.extern.slf4j.Slf4j; import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.condition.DisabledOnOs; import org.junit.jupiter.api.condition.OS; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.tika.FetchAndParseReply; import org.apache.tika.FetchAndParseRequest; @@ -41,9 +42,10 @@ import org.apache.tika.pipes.ExternalTestBase; import org.apache.tika.pipes.fetcher.fs.FileSystemFetcherConfig; -@Slf4j @DisabledOnOs(value = OS.WINDOWS, disabledReason = "exec:exec classpath exceeds Windows CreateProcess command-line length limit") class FileSystemFetcherTest extends ExternalTestBase { + private static final Logger LOG = LoggerFactory.getLogger(FileSystemFetcherTest.class); + @Test void testFileSystemFetcher() throws Exception { @@ -59,7 +61,7 @@ void testFileSystemFetcher() throws Exception { config.setBasePath(basePath); String configJson = OBJECT_MAPPER.writeValueAsString(config); - log.info("Creating fetcher with config (basePath={}): {}", basePath, configJson); + LOG.info("Creating fetcher with config (basePath={}): {}", basePath, configJson); SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() @@ -68,7 +70,7 @@ void testFileSystemFetcher() throws Exception { .setFetcherConfigJson(configJson) .build()); - log.info("Fetcher created: {}", saveReply.getFetcherId()); + LOG.info("Fetcher created: {}", saveReply.getFetcherId()); List successes = Collections.synchronizedList(new ArrayList<>()); List errors = Collections.synchronizedList(new ArrayList<>()); @@ -78,7 +80,7 @@ void testFileSystemFetcher() throws Exception { requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(new StreamObserver<>() { @Override public void onNext(FetchAndParseReply fetchAndParseReply) { - log.debug("Reply from fetch-and-parse - key={}, status={}", + LOG.debug("Reply from fetch-and-parse - key={}, status={}", fetchAndParseReply.getFetchKey(), fetchAndParseReply.getStatus()); if ("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) { errors.add(fetchAndParseReply); @@ -89,14 +91,14 @@ public void onNext(FetchAndParseReply fetchAndParseReply) { @Override public void onError(Throwable throwable) { - log.error("Received an error", throwable); + LOG.error("Received an error", throwable); Assertions.fail(throwable); countDownLatch.countDown(); } @Override public void onCompleted() { - log.info("Finished streaming fetch and parse replies"); + LOG.info("Finished streaming fetch and parse replies"); countDownLatch.countDown(); } }); @@ -122,13 +124,13 @@ public void onCompleted() { } }); } - log.info("Done submitting files to fetcher {}", fetcherId); + LOG.info("Done submitting files to fetcher {}", fetcherId); requestStreamObserver.onCompleted(); try { if (!countDownLatch.await(3, TimeUnit.MINUTES)) { - log.error("Timed out waiting for parse to complete"); + LOG.error("Timed out waiting for parse to complete"); Assertions.fail("Timed out waiting for parsing to complete"); } } catch (InterruptedException e) { @@ -140,14 +142,14 @@ public void onCompleted() { assertAllFilesFetched(TEST_FOLDER.toPath(), successes, errors); } else { int totalProcessed = successes.size() + errors.size(); - log.info("Processed {} documents (limit was {})", totalProcessed, maxDocs); + LOG.info("Processed {} documents (limit was {})", totalProcessed, maxDocs); Assertions.assertTrue(totalProcessed <= maxDocs, "Should not process more than " + maxDocs + " documents"); Assertions.assertTrue(totalProcessed > 0, "Should have processed at least one document"); } - log.info("Test completed successfully - {} successes, {} errors", + LOG.info("Test completed successfully - {} successes, {} errors", successes.size(), errors.size()); } finally { channel.shutdown(); diff --git a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java index f8e4799331..9cf6697389 100644 --- a/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java +++ b/tika-e2e-tests/tika-grpc/src/test/java/org/apache/tika/pipes/filesystem/HandlerTypeTest.java @@ -28,7 +28,6 @@ import io.grpc.ManagedChannel; import io.grpc.ManagedChannelBuilder; -import lombok.extern.slf4j.Slf4j; import org.awaitility.Awaitility; import org.junit.jupiter.api.AfterAll; import org.junit.jupiter.api.Assertions; @@ -38,6 +37,8 @@ import org.junit.jupiter.api.TestInstance; import org.junit.jupiter.api.condition.DisabledOnOs; import org.junit.jupiter.api.condition.OS; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.apache.tika.FetchAndParseReply; import org.apache.tika.FetchAndParseRequest; @@ -58,11 +59,11 @@ * Example: {"basic-content-handler-factory": {"type": "HTML"}} */ @TestInstance(TestInstance.Lifecycle.PER_CLASS) -@Slf4j @Tag("E2ETest") @DisabledOnOs(value = OS.WINDOWS, disabledReason = "exec:exec classpath exceeds Windows CreateProcess command-line length limit") class HandlerTypeTest { + private static final Logger LOG = LoggerFactory.getLogger(HandlerTypeTest.class); private static final File TEST_FOLDER = ExternalTestBase.TEST_FOLDER; private static final int GRPC_PORT = Integer.parseInt(System.getProperty("tika.e2e.grpcPort", "50052")); @@ -75,7 +76,7 @@ void setup() throws Exception { killProcessOnPort(3344); killProcessOnPort(10800); } catch (Exception e) { - log.debug("No orphaned processes to clean up: {}", e.getMessage()); + LOG.debug("No orphaned processes to clean up: {}", e.getMessage()); } ExternalTestBase.copyTestFixtures(); @@ -85,7 +86,7 @@ void setup() throws Exception { @AfterAll void teardown() { if (localGrpcProcess != null) { - log.info("Stopping local gRPC server and child processes"); + LOG.info("Stopping local gRPC server and child processes"); localGrpcProcess.destroy(); try { if (!localGrpcProcess.waitFor(10, TimeUnit.SECONDS)) { @@ -97,14 +98,14 @@ void teardown() { killProcessOnPort(3344); killProcessOnPort(10800); } catch (Exception e) { - log.debug("Error during teardown: {}", e.getMessage()); + LOG.debug("Error during teardown: {}", e.getMessage()); } - log.info("Local gRPC server stopped"); + LOG.info("Local gRPC server stopped"); } } private static void startLocalGrpcServer() throws Exception { - log.info("Starting local tika-grpc server with Ignite config for HandlerType test"); + LOG.info("Starting local tika-grpc server with Ignite config for HandlerType test"); Path currentDir = Path.of("").toAbsolutePath(); Path tikaRootDir = currentDir; @@ -123,8 +124,8 @@ private static void startLocalGrpcServer() throws Exception { throw new IllegalStateException("Config file not found: " + configFile); } - log.info("tika-grpc dir: {}", tikaGrpcDir); - log.info("Config file: {}", configFile); + LOG.info("tika-grpc dir: {}", tikaGrpcDir); + LOG.info("Config file: {}", configFile); String javaHome = System.getProperty("java.home"); boolean isWindows = System.getProperty("os.name").toLowerCase(Locale.ROOT).contains("win"); @@ -172,7 +173,7 @@ private static void startLocalGrpcServer() throws Exception { new InputStreamReader(localGrpcProcess.getInputStream(), StandardCharsets.UTF_8))) { String line; while ((line = reader.readLine()) != null) { - log.info("tika-grpc: {}", line); + LOG.info("tika-grpc: {}", line); if (line.contains("Ignite server started") || (line.contains("Table") && line.contains("created successfully")) || line.contains("Server started, listening on")) { @@ -183,7 +184,7 @@ private static void startLocalGrpcServer() throws Exception { } } } catch (IOException e) { - log.error("Error reading server output", e); + LOG.error("Error reading server output", e); } }); logThread.setDaemon(true); @@ -230,7 +231,7 @@ private static void startLocalGrpcServer() throws Exception { throw new RuntimeException("tika-grpc server with Ignite failed to start within timeout", e); } - log.info("HandlerType test server ready on port {}", GRPC_PORT); + LOG.info("HandlerType test server ready on port {}", GRPC_PORT); } private ManagedChannel getManagedChannel() { @@ -258,10 +259,10 @@ private static void killProcessOnPort(int port) throws IOException, InterruptedE .flatMap(h -> h.info().commandLine()) .orElse(""); if (!cmdLine.contains("tika") && !cmdLine.contains("TikaGrpc") && !cmdLine.contains("ignite")) { - log.debug("Skipping kill of PID {} on port {} — not a tika/ignite process", pid, port); + LOG.debug("Skipping kill of PID {} on port {} — not a tika/ignite process", pid, port); return; } - log.info("Killing tika/ignite process {} on port {}", pid, port); + LOG.info("Killing tika/ignite process {} on port {}", pid, port); new ProcessBuilder("kill", String.valueOf(pid)).start().waitFor(2, TimeUnit.SECONDS); Thread.sleep(1000); new ProcessBuilder("kill", "-9", String.valueOf(pid)).start().waitFor(2, TimeUnit.SECONDS); @@ -280,7 +281,7 @@ private static boolean isParentProcess(long pid) { } } } catch (Exception e) { - log.debug("Error checking parent process", e); + LOG.debug("Error checking parent process", e); } return false; } @@ -300,7 +301,7 @@ void testParseContextJson() throws Exception { .setFetcherType("file-system-fetcher") .setFetcherConfigJson(ExternalTestBase.OBJECT_MAPPER.writeValueAsString(config)) .build()); - log.info("Fetcher created: {}", saveReply.getFetcherId()); + LOG.info("Fetcher created: {}", saveReply.getFetcherId()); // Parse sample.html requesting HTML output FetchAndParseReply htmlReply = blockingStub.fetchAndParse(FetchAndParseRequest.newBuilder() @@ -309,13 +310,13 @@ void testParseContextJson() throws Exception { .setParseContextJson("{\"basic-content-handler-factory\": {\"type\": \"HTML\"}}") .build()); - log.info("HTML parse status: {}", htmlReply.getStatus()); + LOG.info("HTML parse status: {}", htmlReply.getStatus()); Assertions.assertEquals("PARSE_SUCCESS", htmlReply.getStatus(), "Parse should succeed with HTML handler type"); String htmlContent = htmlReply.getFieldsMap().get("X-TIKA:content"); Assertions.assertNotNull(htmlContent, "Content should be present in HTML response"); - log.info("HTML content (first 200 chars): {}", htmlContent.substring(0, Math.min(200, htmlContent.length()))); + LOG.info("HTML content (first 200 chars): {}", htmlContent.substring(0, Math.min(200, htmlContent.length()))); Assertions.assertTrue( htmlContent.contains(" h.info().commandLine()) .orElse(""); if (!cmdLine.contains("tika") && !cmdLine.contains("TikaGrpc") && !cmdLine.contains("ignite")) { - log.debug("Skipping kill of PID {} on port {} — not a tika/ignite process: {}", pid, port, cmdLine); + LOG.debug("Skipping kill of PID {} on port {} — not a tika/ignite process: {}", pid, port, cmdLine); return; } - log.info("Found tika/ignite process {} on port {}, killing it", pid, port); + LOG.info("Found tika/ignite process {} on port {}, killing it", pid, port); ProcessBuilder killPb = new ProcessBuilder("kill", String.valueOf(pid)); Process killProcess = killPb.start(); @@ -385,7 +386,7 @@ private static boolean isParentProcess(long pid) { } } } catch (Exception e) { - log.debug("Error checking parent process", e); + LOG.debug("Error checking parent process", e); } return false; } @@ -404,7 +405,7 @@ void testIgniteConfigStore() throws Exception { config.setBasePath(basePath); String configJson = ExternalTestBase.OBJECT_MAPPER.writeValueAsString(config); - log.info("Creating fetcher with Ignite ConfigStore (basePath={}): {}", basePath, configJson); + LOG.info("Creating fetcher with Ignite ConfigStore (basePath={}): {}", basePath, configJson); SaveFetcherReply saveReply = blockingStub.saveFetcher(SaveFetcherRequest .newBuilder() @@ -413,7 +414,7 @@ void testIgniteConfigStore() throws Exception { .setFetcherConfigJson(configJson) .build()); - log.info("Fetcher saved to Ignite: {}", saveReply.getFetcherId()); + LOG.info("Fetcher saved to Ignite: {}", saveReply.getFetcherId()); List successes = Collections.synchronizedList(new ArrayList<>()); List errors = Collections.synchronizedList(new ArrayList<>()); @@ -423,7 +424,7 @@ void testIgniteConfigStore() throws Exception { requestStreamObserver = tikaStub.fetchAndParseBiDirectionalStreaming(new StreamObserver<>() { @Override public void onNext(FetchAndParseReply fetchAndParseReply) { - log.debug("Reply from fetch-and-parse - key={}, status={}", + LOG.debug("Reply from fetch-and-parse - key={}, status={}", fetchAndParseReply.getFetchKey(), fetchAndParseReply.getStatus()); if ("FETCH_AND_PARSE_EXCEPTION".equals(fetchAndParseReply.getStatus())) { errors.add(fetchAndParseReply); @@ -434,20 +435,20 @@ public void onNext(FetchAndParseReply fetchAndParseReply) { @Override public void onError(Throwable throwable) { - log.error("Received an error", throwable); + LOG.error("Received an error", throwable); Assertions.fail(throwable); countDownLatch.countDown(); } @Override public void onCompleted() { - log.info("Finished streaming fetch and parse replies"); + LOG.info("Finished streaming fetch and parse replies"); countDownLatch.countDown(); } }); int maxDocs = Integer.parseInt(System.getProperty("corpus.numDocs", "-1")); - log.info("Document limit: {}", maxDocs == -1 ? "unlimited" : maxDocs); + LOG.info("Document limit: {}", maxDocs == -1 ? "unlimited" : maxDocs); try (Stream paths = Files.walk(TEST_FOLDER.toPath())) { Stream fileStream = paths @@ -473,13 +474,13 @@ public void onCompleted() { } }); } - log.info("Done submitting files to Ignite-backed fetcher {}", fetcherId); + LOG.info("Done submitting files to Ignite-backed fetcher {}", fetcherId); requestStreamObserver.onCompleted(); try { if (!countDownLatch.await(3, TimeUnit.MINUTES)) { - log.error("Timed out waiting for parse to complete"); + LOG.error("Timed out waiting for parse to complete"); Assertions.fail("Timed out waiting for parsing to complete"); } } catch (InterruptedException e) { @@ -491,7 +492,7 @@ public void onCompleted() { ExternalTestBase.assertAllFilesFetched(TEST_FOLDER.toPath(), successes, errors); } else { int totalProcessed = successes.size() + errors.size(); - log.info("Processed {} documents with Ignite ConfigStore (limit was {})", + LOG.info("Processed {} documents with Ignite ConfigStore (limit was {})", totalProcessed, maxDocs); Assertions.assertTrue(totalProcessed <= maxDocs, "Should not process more than " + maxDocs + " documents"); @@ -499,7 +500,7 @@ public void onCompleted() { "Should have processed at least one document"); } - log.info("Ignite ConfigStore test completed successfully - {} successes, {} errors", + LOG.info("Ignite ConfigStore test completed successfully - {} successes, {} errors", successes.size(), errors.size()); } finally { channel.shutdown(); diff --git a/tika-parent/pom.xml b/tika-parent/pom.xml index bf1d0129af..fbc7ed8d3d 100644 --- a/tika-parent/pom.xml +++ b/tika-parent/pom.xml @@ -397,7 +397,6 @@ 2.4.0 0.9.3 2.26.0 - 1.18.46 9.12.3 3.15.2