This is an automated email from the ASF dual-hosted git repository.
tballison pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/tika.git
The following commit(s) were added to refs/heads/main by this push:
new 918b0d0aa1 TIKA-4991: bound tika-server transfers in time and total
bytes (#3402)
918b0d0aa1 is described below
commit 918b0d0aa17231ba6a9836b65ce6f44ccef0b76f
Author: Tim Allison <[email protected]>
AuthorDate: Fri Oct 23 07:45:31 2026 -0400
TIKA-4991: bound tika-server transfers in time and total bytes (#3402)
Co-Authored-By: Claude Fable 5 <[email protected]>
---
CHANGES.txt | 11 +
.../ROOT/pages/using-tika/server/index.adoc | 29 +-
.../ROOT/pages/using-tika/server/monitoring.adoc | 15 +-
.../tika/server/e2e/TikaServerHttp2Test.java | 5 +
tika-parent/pom.xml | 8 +
tika-server/README.md | 3 +-
tika-server/tika-server-core/pom.xml | 4 +
.../org/apache/tika/server/core/JettyRequests.java | 45 ++
.../tika/server/core/MaxRequestSizeFilter.java | 174 +++++-
.../apache/tika/server/core/RequestBodyBudget.java | 131 +++++
.../core/RequestReadDeadlineInterceptor.java | 68 +++
.../core/ResponseWriteDeadlineInterceptor.java | 249 +++++++++
.../apache/tika/server/core/TikaServerConfig.java | 103 ++++
.../apache/tika/server/core/TikaServerProcess.java | 55 +-
.../apache/tika/server/core/TransferLimitLog.java | 64 +++
.../server/core/metrics/TikaServerMetrics.java | 12 +-
.../server/core/resource/LanguageResource.java | 6 +-
.../tika/server/core/resource/TikaWelcome.java | 6 +-
.../tika/server/core/IntegrationTestBase.java | 2 +-
.../tika/server/core/RequestBodyBudgetTest.java | 127 +++++
.../core/ResponseWriteDeadlineInterceptorTest.java | 178 ++++++
.../core/TikaServerBasicIntegrationTest.java | 6 +-
.../tika/server/core/TikaServerConfigTest.java | 38 +-
.../tika/server/core/TikaServerProcessTest.java | 16 +
.../tika/server/core/TransferLimitLogTest.java | 46 ++
.../standard/MaxRequestSizeIntegrationTest.java | 606 +++++++++++++++++++++
26 files changed, 1973 insertions(+), 34 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 7399f40d50..27646c9141 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -54,6 +54,14 @@ Release 4.2.0 - 10/22/2026
_min-width and min-width is an error, and gated entries load as
GatedEnricher,
not SizeGatedEnricher (TIKA-4972).
+ * tika-server adds transfer limits: maxRequestReadMillis (408) and
+ maxResponseWriteMillis, both 5 min; idleTimeoutMillis (120s); and
+ maxInFlightRequestBytes (429). Raise the deadlines for very large or very
slow
+ transfers; cut-offs are logged at INFO by the TransferLimitLog logger
(TIKA-4991).
+
+ * tika-server accepts cleartext HTTP/2 (h2c) only when allowHttp2 is set;
it had
+ been on by default since 4.0.0 (TIKA-4991).
+
* embedded-limits maxDepth defaults to 20 (was unlimited) and replaces
output-limits
maxPackageEntryDepth (#3319). To be removed in 4.3.0:
maxPackageEntryDepth and its
accessors on OutputLimits and SecureContentHandler, and
tika-parser-mail-commons
@@ -99,6 +107,9 @@ Release 4.2.0 - 10/22/2026
SERVER, PIPES AND OPERATIONS
+ * tika-server's welcome page lists endpoints inherited from superclasses
(XMP on
+ /meta) under their real paths (TIKA-4991).
+
* The file-system emitter can gzip its output ("compression": "GZIP");
tika-eval reads
.json.gz extracts and measures them uncompressed (TIKA-4989).
diff --git a/docs/modules/ROOT/pages/using-tika/server/index.adoc
b/docs/modules/ROOT/pages/using-tika/server/index.adoc
index 7930aa5a01..fcb41e95b3 100644
--- a/docs/modules/ROOT/pages/using-tika/server/index.adoc
+++ b/docs/modules/ROOT/pages/using-tika/server/index.adoc
@@ -52,7 +52,9 @@ Forks still exit after `socketTimeoutMillis` idle and restart
on the next reques
A saturated `tika-server` says so. When every fork in a pool is in use and
none frees up within
`pipes.maxWaitForClientMillis`, the request comes back as `429 Too Many
Requests` with a
`Retry-After` header and `status: CLIENT_UNAVAILABLE_WITHIN_MS` — deliberately
distinct from the
-`503` you get when a fork actually crashed, OOM'd or timed out.
+`503` you get when a fork actually crashed, OOM'd or timed out. A request
whose body would take
+the bytes all requests hold past `maxInFlightRequestBytes` also gets `429`
with `Retry-After`,
+but with a plain-text body rather than a `status`.
That distinction is worth building on. A client can tell "the server is busy,
come back in a
moment" apart from "this document broke a fork" and act accordingly: back off
and retry the
@@ -567,6 +569,11 @@ backpressure; see
<<_endpoints_and_forked_process_groups,Endpoints and Forked-Pr
Carries `Retry-After`. Distinct from `503` on purpose: a `429` spike means
"raise `numClients` or
add capacity", a `503` spike means "something is actually crashing".
+|`429 Too Many Requests`
+|none (`text/plain` body)
+|The request bodies in flight would exceed `maxInFlightRequestBytes`. Carries
`Retry-After`;
+retry as for the other `429`.
+
|`400 Bad Request`
|`FETCHER_NOT_FOUND`, `EMITTER_NOT_FOUND`
|The request named a fetcher or emitter that does not exist, or one reserved
for the server's
@@ -699,6 +706,26 @@ independent forked-process groups you run — see
|`1 GiB`
|Maximum request body in bytes; larger requests are rejected with `413`.
Enforced on bytes actually read, so a chunked upload with no `Content-Length`
is bounded too. Bodies above
xref:pipes/configuration.adoc#payload-limits[`maxInlineBytes`] are spooled to
disk, so this also bounds how much one request can write. Raise it for larger
documents, or set a negative value to disable the limit.
+|`maxRequestReadMillis`
+|`300000` (5 min)
+|How long, from the start of a request, its body may take to arrive; a request
still sending after that is rejected with `408`. At the default, a body of the
default `maxRequestSizeBytes` (1 GiB) needs about 29 Mbit/s. This bounds how
long a slow client can hold a request thread; a client that sends nothing at
all is dropped by the connector's idle timeout instead. Set a negative value to
disable it; zero is rejected.
+
+|`maxResponseWriteMillis`
+|`300000` (5 min)
+|How long, from when its status line is sent, a response body may take to
write; a response still being written after that is cut off (the status line
has already been sent, so the client sees a truncated body). This bounds how
long a client that reads slowly can hold a request thread. Set a negative value
to disable it; zero is rejected.
+
+|`maxInFlightRequestBytes`
+|`0` (`maxRequestSizeBytes` × 2 × `pipes.numClients`)
+|How many request-body bytes all requests together may hold at once. Bytes
count as they are read, not as declared, and are freed when each response
completes; a request whose body would go over is rejected with `429`
(`text/plain`, with `Retry-After`) partway through its upload. This bounds the
temp disk and memory request bodies take across all requests, while a client
that sends slowly holds only the bytes it has actually sent, so slow
connections cannot lock others out. A client with [...]
+
+|`idleTimeoutMillis`
+|`120000` (2 min)
+|How long a connection may go without any bytes moving before it is closed.
This is what ends a client that stops sending or reading altogether, which the
read and write deadlines cannot. A request waiting on its parse is not
affected, however long the parse takes. Must be positive: it cannot be
disabled. Every request cut off by this or by the size and deadline limits
above is logged once, at `INFO`, by the
`org.apache.tika.server.core.TransferLimitLog` logger, so these client-caused
ev [...]
+
+|`allowHttp2`
+|`false`
+|Whether to accept cleartext HTTP/2 (h2c) as well as HTTP/1.1, for example
behind Google Cloud Run, whose HTTP/1.1 requests are capped at 32 MB. HTTP/2
requests get every limit above except `maxResponseWriteMillis`, whose cut-off
would close the connection and so every request multiplexed on it; a client
that reads an HTTP/2 response slowly is bounded only by `idleTimeoutMillis`.
HTTP/2 over TLS is not offered either way.
+
|`maxQueuePauseMillis`
|`60000`
|How long a POST to `/async` blocks when the queue is full before it is
rejected with `429`
diff --git a/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
b/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
index 6374aee862..cbecdbc3e0 100644
--- a/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
+++ b/docs/modules/ROOT/pages/using-tika/server/monitoring.adoc
@@ -119,9 +119,18 @@ chunked uploads have none and are not recorded. Bucketed
at 1KB..1GB by decades.
|counter
|`reason`
|Requests refused for capacity reasons, by the status the server already uses
to encode
-them: `busy_429` (the `/async` queue was full, or no fork was free within
-`maxWaitForClientMillis`), `crash_503` (the fork serving the request OOM'd,
timed out or
-crashed), `payload_413` (body over `maxRequestSizeBytes` or the IPC payload
limit).
+them: `busy_429` (the `/async` queue was full, no fork was free within
+`maxWaitForClientMillis`, or the request's body would exceed
`maxInFlightRequestBytes`), `crash_503` (the fork serving the request OOM'd,
timed out or
+crashed), `payload_413` (body over `maxRequestSizeBytes` or the IPC payload
limit),
+`slow_408` (body still arriving after `maxRequestReadMillis`, or stalled past
+`idleTimeoutMillis`). A response cut off by `maxResponseWriteMillis` keeps the
status it
+was sent with, so it is not counted here; `TransferLimitLog` records it.
+
+|`tika_server_request_body_in_flight_bytes`
+|gauge
+|
+|Request-body bytes held by in-flight requests, counted against
`maxInFlightRequestBytes`;
+present only when that budget is enabled. It should return to 0 when the
server is idle.
|`tika_server_tasks_active`
|gauge
diff --git
a/tika-e2e-tests/tika-server/src/test/java/org/apache/tika/server/e2e/TikaServerHttp2Test.java
b/tika-e2e-tests/tika-server/src/test/java/org/apache/tika/server/e2e/TikaServerHttp2Test.java
index cd5b55742e..43e2a56974 100644
---
a/tika-e2e-tests/tika-server/src/test/java/org/apache/tika/server/e2e/TikaServerHttp2Test.java
+++
b/tika-e2e-tests/tika-server/src/test/java/org/apache/tika/server/e2e/TikaServerHttp2Test.java
@@ -91,8 +91,13 @@ public class TikaServerHttp2Test {
"mvn test -pl tika-e2e-tests/tika-server -Pe2e");
log.info("Starting tika-server from: {}", serverJar);
+ // h2c is opt-in
+ Path config = Files.writeString(Files.createTempFile("tika-http2-",
".json"),
+ "{\"server\": {\"allowHttp2\": true}}");
+ config.toFile().deleteOnExit();
ProcessBuilder pb = new ProcessBuilder(
"java", "-jar", serverJar.getFileName().toString(),
+ "-c", config.toAbsolutePath().toString(),
"-p", String.valueOf(port),
"-h", "localhost"
);
diff --git a/tika-parent/pom.xml b/tika-parent/pom.xml
index 02fe49bc6a..3e922396cc 100644
--- a/tika-parent/pom.xml
+++ b/tika-parent/pom.xml
@@ -540,6 +540,14 @@
<type>pom</type>
<scope>import</scope>
</dependency>
+ <!-- keeps the ee11 servlet layer CXF pulls in on the same Jetty as
jetty-server -->
+ <dependency>
+ <groupId>org.eclipse.jetty.ee11</groupId>
+ <artifactId>jetty-ee11-bom</artifactId>
+ <version>${jetty.version}</version>
+ <type>pom</type>
+ <scope>import</scope>
+ </dependency>
<dependency>
<groupId>io.grpc</groupId>
<artifactId>grpc-bom</artifactId>
diff --git a/tika-server/README.md b/tika-server/README.md
index f09e2555c4..c55293c276 100644
--- a/tika-server/README.md
+++ b/tika-server/README.md
@@ -101,8 +101,9 @@ HTTP Return Codes
`204` - No content (for example when we are unpacking file without
attachments)
`400` - Bad request (unknown or invalid handler type, or a reserved/unknown
fetcher or emitter was named)
`403` - Forbidden (per-request configuration was supplied but
`allowPerRequestConfig` is off)
+`408` - Request timeout (the request body was still arriving after
`maxRequestReadMillis`, or stalled past `idleTimeoutMillis`)
`413` - Payload too large (the request body exceeds `maxRequestSizeBytes`, or
a pipes payload limit was exceeded)
`422` - Unparsable document of known type (password protected documents and
unsupported versions like Biff5 Excel)
-`429` - Too many requests (all forked workers were busy for longer than
`maxWaitForClientMillis`; retry with backoff)
+`429` - Too many requests (all forked workers were busy for longer than
`maxWaitForClientMillis`, or the request bodies in flight would exceed
`maxInFlightRequestBytes`; retry with backoff)
`500` - Internal error
`503` - Service unavailable (the forked worker hit a timeout, ran out of
memory, or crashed)
diff --git a/tika-server/tika-server-core/pom.xml
b/tika-server/tika-server-core/pom.xml
index 9e66d48693..62c4c59f49 100644
--- a/tika-server/tika-server-core/pom.xml
+++ b/tika-server/tika-server-core/pom.xml
@@ -84,6 +84,10 @@
<groupId>org.eclipse.jetty.http2</groupId>
<artifactId>jetty-http2-server</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.eclipse.jetty.ee11</groupId>
+ <artifactId>jetty-ee11-servlet</artifactId>
+ </dependency>
<dependency>
<groupId>jakarta.servlet</groupId>
<artifactId>jakarta.servlet-api</artifactId>
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/JettyRequests.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/JettyRequests.java
new file mode 100644
index 0000000000..79afd1fddc
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/JettyRequests.java
@@ -0,0 +1,45 @@
+/*
+ * 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.server.core;
+
+import jakarta.servlet.http.HttpServletRequest;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.transport.http.AbstractHTTPDestination;
+import org.eclipse.jetty.ee11.servlet.ServletContextRequest;
+
+/** The Jetty request under a CXF message, for what JAX-RS does not expose. */
+final class JettyRequests {
+
+ private JettyRequests() {
+ }
+
+ /** Null when there is no Jetty HTTP request, as under CXF's local
transport. */
+ static ServletContextRequest of(Message message) {
+ if (message == null || message.getExchange() == null) {
+ return null;
+ }
+ Object request =
message.getExchange().getInMessage().get(AbstractHTTPDestination.HTTP_REQUEST);
+ if (!(request instanceof HttpServletRequest httpRequest)) {
+ return null;
+ }
+ try {
+ return ServletContextRequest.getServletContextRequest(httpRequest);
+ } catch (IllegalStateException e) {
+ return null;
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java
index 9021e6edbb..97db65b9fa 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/MaxRequestSizeFilter.java
@@ -19,47 +19,84 @@ package org.apache.tika.server.core;
import java.io.FilterInputStream;
import java.io.IOException;
import java.io.InputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import jakarta.ws.rs.container.ContainerRequestContext;
import jakarta.ws.rs.container.ContainerRequestFilter;
+import jakarta.ws.rs.core.HttpHeaders;
import jakarta.ws.rs.core.MediaType;
import jakarta.ws.rs.core.Response;
import jakarta.ws.rs.ext.ExceptionMapper;
import jakarta.ws.rs.ext.Provider;
+import org.apache.cxf.jaxrs.utils.JAXRSUtils;
+import org.apache.cxf.message.Message;
/**
- * Rejects request bodies larger than {@code maxRequestSizeBytes}.
+ * Rejects request bodies larger than {@code maxRequestSizeBytes} (413), still
arriving
+ * after {@code maxRequestReadMillis} (408), or that would take the bytes all
requests hold
+ * past {@code maxInFlightRequestBytes} (429, see {@link RequestBodyBudget}).
* <p>
* A declared Content-Length over the limit is refused before the body is
read. Requests
* without a usable Content-Length -- chunked transfer encoding, in particular
-- are
* counted as they are consumed, so the limit holds whether or not the client
is honest
- * about the size.
+ * about the size. The wire under the entity is bounded by
+ * {@link RequestReadDeadlineInterceptor}, whose start time this deadline
shares.
*/
@Provider
public class MaxRequestSizeFilter implements ContainerRequestFilter {
static final String TOO_LARGE_MESSAGE = "Request body exceeds
maxRequestSizeBytes";
+ static final String TOO_SLOW_MESSAGE = "Request body not received within
maxRequestReadMillis";
+ static final String IDLE_MESSAGE = "Request body stalled past
idleTimeoutMillis";
+ static final String BUSY_MESSAGE = "Request body would exceed
maxInFlightRequestBytes across requests in flight";
+ // budget frees as soon as any in-flight request completes
+ private static final long RETRY_AFTER_SECONDS = 1;
private final long maxRequestSizeBytes;
+ private final long maxRequestReadMillis;
+ private final RequestBodyBudget budget;
/**
* @param maxRequestSizeBytes maximum request body in bytes; negative
disables the limit
*/
public MaxRequestSizeFilter(long maxRequestSizeBytes) {
+ this(maxRequestSizeBytes, -1, null);
+ }
+
+ /**
+ * @param maxRequestSizeBytes maximum request body in bytes; negative
disables the limit
+ * @param maxRequestReadMillis how long reading the body may take;
negative disables it
+ * @param budget bytes all requests may hold at once; null disables it
+ */
+ public MaxRequestSizeFilter(long maxRequestSizeBytes, long
maxRequestReadMillis,
+ RequestBodyBudget budget) {
this.maxRequestSizeBytes = maxRequestSizeBytes;
+ this.maxRequestReadMillis = maxRequestReadMillis;
+ this.budget = budget;
}
@Override
public void filter(ContainerRequestContext requestContext) {
- if (maxRequestSizeBytes < 0) {
- return;
- }
- if (requestContext.getLength() > maxRequestSizeBytes) {
+ if (maxRequestSizeBytes >= 0 && requestContext.getLength() >
maxRequestSizeBytes) {
+ TransferLimitLog.log("declared Content-Length " +
requestContext.getLength()
+ + " exceeds maxRequestSizeBytes (" + maxRequestSizeBytes +
")");
requestContext.abortWith(tooLarge());
return;
}
- requestContext.setEntityStream(
- new BoundedInputStream(requestContext.getEntityStream(),
maxRequestSizeBytes));
+ long maxReadNanos = maxRequestReadMillis < 0 ? -1 :
TimeUnit.MILLISECONDS.toNanos(maxRequestReadMillis);
+ RequestBodyBudget.Reservation reservation =
+ budget == null ? null :
budget.open(JAXRSUtils.getCurrentMessage());
+ // Entity bytes: after any Content-Encoding is undone.
+ requestContext.setEntityStream(new
BoundedInputStream(requestContext.getEntityStream(),
+ maxRequestSizeBytes, startNanos(), maxReadNanos,
maxRequestReadMillis, reservation));
+ }
+
+ private static long startNanos() {
+ Message message = JAXRSUtils.getCurrentMessage();
+ Object start = message == null || message.getExchange() == null
+ ? null :
message.getExchange().get(RequestReadDeadlineInterceptor.START_NANOS);
+ return start instanceof Long l ? l : System.nanoTime();
}
private static Response tooLarge() {
@@ -71,23 +108,39 @@ public class MaxRequestSizeFilter implements
ContainerRequestFilter {
}
/**
- * Throws once more than {@code limit} bytes have been read. Deliberately
not
- * silent truncation: a caller that sent too much must not receive a 200
describing
- * a prefix of their document.
+ * Throws once more than {@code limit} bytes have been read, or on a read
after the
+ * deadline. Deliberately not silent truncation: a caller that sent too
much must not
+ * receive a 200 describing a prefix of their document. A client that
sends nothing at
+ * all is left to the connector's idle timeout.
*/
- private static final class BoundedInputStream extends FilterInputStream {
+ static final class BoundedInputStream extends FilterInputStream {
private final long limit;
+ private final long startNanos;
+ private final long maxReadNanos;
+ private final long maxReadMillis;
+ private final RequestBodyBudget.Reservation reservation;
private long count;
- private BoundedInputStream(InputStream in, long limit) {
+ BoundedInputStream(InputStream in, long limit, long startNanos, long
maxReadNanos,
+ long maxReadMillis, RequestBodyBudget.Reservation
reservation) {
super(in);
this.limit = limit;
+ this.startNanos = startNanos;
+ this.maxReadNanos = maxReadNanos;
+ this.maxReadMillis = maxReadMillis;
+ this.reservation = reservation;
}
@Override
public int read() throws IOException {
- int c = super.read();
+ checkDeadline();
+ int c;
+ try {
+ c = super.read();
+ } catch (IOException e) {
+ throw idleOrRethrow(e);
+ }
if (c != -1) {
add(1);
}
@@ -96,17 +149,60 @@ public class MaxRequestSizeFilter implements
ContainerRequestFilter {
@Override
public int read(byte[] b, int off, int len) throws IOException {
- int read = super.read(b, off, len);
+ checkDeadline();
+ int read;
+ try {
+ read = super.read(b, off, len);
+ } catch (IOException e) {
+ throw idleOrRethrow(e);
+ }
if (read > 0) {
add(read);
}
return read;
}
+ // Jetty's idle timeout surfaces as an IOException caused by a
TimeoutException
+ private static IOException idleOrRethrow(IOException e) {
+ for (Throwable t = e; t != null; t = t.getCause()) {
+ if (t instanceof TimeoutException) {
+ throw new RequestTimeoutException(IDLE_MESSAGE + " (" +
t.getMessage() + ")");
+ }
+ }
+ return e;
+ }
+
+ // FilterInputStream.skip delegates, and the skipped bytes would go
uncounted
+ @Override
+ public long skip(long n) throws IOException {
+ if (n <= 0) {
+ return 0;
+ }
+ byte[] buffer = new byte[(int) Math.min(n, 8192)];
+ long skipped = 0;
+ while (skipped < n) {
+ int read = read(buffer, 0, (int) Math.min(buffer.length, n -
skipped));
+ if (read == -1) {
+ break;
+ }
+ skipped += read;
+ }
+ return skipped;
+ }
+
+ private void checkDeadline() {
+ if (maxReadNanos >= 0 && System.nanoTime() - startNanos >
maxReadNanos) {
+ throw new RequestTimeoutException(TOO_SLOW_MESSAGE + " (" +
maxReadMillis + " ms)");
+ }
+ }
+
private void add(int n) {
count += n;
- if (count > limit) {
- throw new RequestTooLargeException(TOO_LARGE_MESSAGE + " (" +
limit + ")");
+ if (limit >= 0 && count > limit) {
+ throw new RequestTooLargeException(TOO_LARGE_MESSAGE + " (" +
limit + " bytes)");
+ }
+ if (reservation != null && !reservation.ensure(count)) {
+ throw new RequestBudgetExceededException(BUSY_MESSAGE);
}
}
}
@@ -123,11 +219,55 @@ public class MaxRequestSizeFilter implements
ContainerRequestFilter {
}
}
+ /** Unchecked for the same reason; {@link RequestTimeoutExceptionMapper}
makes it a 408. */
+ public static final class RequestTimeoutException extends RuntimeException
{
+ RequestTimeoutException(String message) {
+ super(message);
+ }
+ }
+
+ /** Unchecked for the same reason; {@link
RequestBudgetExceededExceptionMapper} makes it a 429. */
+ public static final class RequestBudgetExceededException extends
RuntimeException {
+ RequestBudgetExceededException(String message) {
+ super(message);
+ }
+ }
+
+ @Provider
+ public static final class RequestBudgetExceededExceptionMapper
+ implements ExceptionMapper<RequestBudgetExceededException> {
+ @Override
+ public Response toResponse(RequestBudgetExceededException e) {
+ TransferLimitLog.log(e.getMessage());
+ return Response
+ .status(Response.Status.TOO_MANY_REQUESTS)
+ .header(HttpHeaders.RETRY_AFTER, RETRY_AFTER_SECONDS)
+ .entity(e.getMessage())
+ .type(MediaType.TEXT_PLAIN)
+ .build();
+ }
+ }
+
+ @Provider
+ public static final class RequestTimeoutExceptionMapper
+ implements ExceptionMapper<RequestTimeoutException> {
+ @Override
+ public Response toResponse(RequestTimeoutException e) {
+ TransferLimitLog.log(e.getMessage());
+ return Response
+ .status(Response.Status.REQUEST_TIMEOUT)
+ .entity(e.getMessage())
+ .type(MediaType.TEXT_PLAIN)
+ .build();
+ }
+ }
+
@Provider
public static final class RequestTooLargeExceptionMapper
implements ExceptionMapper<RequestTooLargeException> {
@Override
public Response toResponse(RequestTooLargeException e) {
+ TransferLimitLog.log(e.getMessage());
return Response
.status(Response.Status.REQUEST_ENTITY_TOO_LARGE)
.entity(e.getMessage())
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestBodyBudget.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestBodyBudget.java
new file mode 100644
index 0000000000..465f37ac73
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestBodyBudget.java
@@ -0,0 +1,131 @@
+/*
+ * 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.server.core;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.cxf.message.Message;
+import org.eclipse.jetty.ee11.servlet.ServletContextRequest;
+import org.eclipse.jetty.server.Request;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Caps the request-body bytes all in-flight requests hold at once ({@code
maxInFlightRequestBytes}).
+ * <p>
+ * Bytes are charged as they are read, not as declared, so a client that sends
slowly holds only
+ * what it has sent. Each request charges through its own {@link Reservation},
which is the only
+ * way the total changes; it is released exactly once, when Jetty completes
the request.
+ */
+public final class RequestBodyBudget {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(RequestBodyBudget.class);
+
+ private final long maxBytes;
+ private final AtomicLong inFlight = new AtomicLong();
+ private final AtomicBoolean warnedNegative = new AtomicBoolean();
+
+ /**
+ * @param maxBytes the most request-body bytes in flight at once; must be
positive
+ */
+ public RequestBodyBudget(long maxBytes) {
+ if (maxBytes <= 0) {
+ throw new IllegalArgumentException("maxBytes must be positive: " +
maxBytes);
+ }
+ this.maxBytes = maxBytes;
+ }
+
+ public long getMaxBytes() {
+ return maxBytes;
+ }
+
+ public long getInFlightBytes() {
+ return inFlight.get();
+ }
+
+ /**
+ * Opens the reservation for the request behind {@code message}, released
when Jetty
+ * completes it. Fails rather than open one that could never be released.
+ */
+ Reservation open(Message message) {
+ ServletContextRequest request = JettyRequests.of(message);
+ if (request == null) {
+ throw new IllegalStateException("no Jetty request to release the
body budget on");
+ }
+ Reservation reservation = new Reservation();
+ // registered before anything is charged, so a failure here leaks
nothing
+ Request.addCompletionListener(request, failure ->
reservation.release());
+ return reservation;
+ }
+
+ Reservation newReservation() {
+ return new Reservation();
+ }
+
+ private boolean tryCharge(long bytes) {
+ while (true) {
+ long current = inFlight.get();
+ if (bytes > maxBytes - current) {
+ return false;
+ }
+ if (inFlight.compareAndSet(current, current + bytes)) {
+ return true;
+ }
+ }
+ }
+
+ private void refund(long bytes) {
+ if (inFlight.addAndGet(-bytes) < 0) {
+ if (warnedNegative.compareAndSet(false, true)) {
+ LOG.warn("request body budget went negative: accounting bug,
clamped to 0");
+ }
+ inFlight.updateAndGet(v -> Math.max(v, 0));
+ }
+ }
+
+ /** One request's share of the budget. */
+ final class Reservation {
+
+ private long charged;
+ private boolean released;
+
+ /**
+ * Grows this request's charge to {@code bytes} read so far; false,
with nothing
+ * charged, if that would exceed the budget. After release, charges
nothing.
+ */
+ synchronized boolean ensure(long bytes) {
+ if (released || bytes <= charged) {
+ return true;
+ }
+ if (!tryCharge(bytes - charged)) {
+ return false;
+ }
+ charged = bytes;
+ return true;
+ }
+
+ synchronized void release() {
+ if (released) {
+ return;
+ }
+ released = true;
+ refund(charged);
+ charged = 0;
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestReadDeadlineInterceptor.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestReadDeadlineInterceptor.java
new file mode 100644
index 0000000000..cf20dd3fe4
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/RequestReadDeadlineInterceptor.java
@@ -0,0 +1,68 @@
+/*
+ * 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.server.core;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.cxf.io.DelegatingInputStream;
+import org.apache.cxf.message.Exchange;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.phase.AbstractPhaseInterceptor;
+import org.apache.cxf.phase.Phase;
+
+/**
+ * Runs on every request as it is received, before resource matching, so it
also covers
+ * requests CXF answers itself (OPTIONS, 404, 405, ...). Stops CXF spooling
the unread body
+ * to disk once a response starts, and puts the {@code maxRequestReadMillis}
deadline on the
+ * wire, so the up to 16 MiB CXF then discards cannot be dripped in forever.
+ * {@link MaxRequestSizeFilter} bounds the entity stream of matched requests.
+ */
+public class RequestReadDeadlineInterceptor extends
AbstractPhaseInterceptor<Message> {
+
+ static final String START_NANOS =
RequestReadDeadlineInterceptor.class.getName() + ".startNanos";
+
+ // must be on the exchange: unset there, CXF's OutgoingChainInterceptor
sets it true on the
+ // in-message and the unread body of a 2xx response is spooled to disk
+ private static final String CXF_CACHE_INPUT = "cxf.io.cacheinput";
+
+ private final long maxRequestReadMillis;
+
+ /**
+ * @param maxRequestReadMillis how long reading the body may take;
negative disables it
+ */
+ public RequestReadDeadlineInterceptor(long maxRequestReadMillis) {
+ super(Phase.RECEIVE);
+ this.maxRequestReadMillis = maxRequestReadMillis;
+ }
+
+ @Override
+ public void handleMessage(Message message) {
+ Exchange exchange = message.getExchange();
+ if (exchange == null) {
+ return;
+ }
+ long startNanos = System.nanoTime();
+ exchange.put(CXF_CACHE_INPUT, Boolean.FALSE);
+ exchange.put(START_NANOS, startNanos);
+ DelegatingInputStream wire =
message.getContent(DelegatingInputStream.class);
+ if (maxRequestReadMillis >= 0 && wire != null) {
+ // deadline only: a size cap here would cut off a client that
reads only after sending
+ wire.setInputStream(new
MaxRequestSizeFilter.BoundedInputStream(wire.getInputStream(), -1,
+ startNanos,
TimeUnit.MILLISECONDS.toNanos(maxRequestReadMillis), maxRequestReadMillis,
null));
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptor.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptor.java
new file mode 100644
index 0000000000..e7029522fa
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptor.java
@@ -0,0 +1,249 @@
+/*
+ * 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.server.core;
+
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.BooleanSupplier;
+
+import jakarta.servlet.http.HttpServletResponse;
+import jakarta.ws.rs.core.Response;
+import jakarta.ws.rs.ext.ExceptionMapper;
+import jakarta.ws.rs.ext.Provider;
+import jakarta.ws.rs.ext.WriterInterceptor;
+import jakarta.ws.rs.ext.WriterInterceptorContext;
+import org.apache.cxf.jaxrs.utils.JAXRSUtils;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.transport.http.AbstractHTTPDestination;
+import org.eclipse.jetty.ee11.servlet.ServletContextRequest;
+import org.eclipse.jetty.server.Request;
+import org.eclipse.jetty.util.thread.Scheduler;
+
+/**
+ * Aborts a response still being written {@code maxResponseWriteMillis} after
it is committed;
+ * the idle timeout alone resets on every trickle a slow reader accepts.
+ * <p>
+ * A timer, not a check between writes: one blocking write to a client that
accepts a byte at
+ * a time can otherwise outlast the deadline indefinitely. From commit, not
the first write:
+ * CXF drains the unread request body inside that write, which the read
deadline bounds.
+ * Aborted, not closed: closing would end a chunked body cleanly, passing a
truncated document
+ * off as complete.
+ */
+@Provider
+public class ResponseWriteDeadlineInterceptor implements WriterInterceptor {
+
+ private final long maxResponseWriteMillis;
+
+ /**
+ * @param maxResponseWriteMillis how long writing a response body may
take; negative disables it
+ */
+ public ResponseWriteDeadlineInterceptor(long maxResponseWriteMillis) {
+ this.maxResponseWriteMillis = maxResponseWriteMillis;
+ }
+
+ @Override
+ public void aroundWriteTo(WriterInterceptorContext context) throws
IOException {
+ Message message = JAXRSUtils.getCurrentMessage();
+ ServletContextRequest request = JettyRequests.of(message);
+ // without a Jetty connection there is nothing to abort; on HTTP/2 the
abort would end
+ // every stream on the connection (see allowHttp2)
+ if (maxResponseWriteMillis >= 0 && request != null
+ &&
request.getConnectionMetaData().getHttpVersion().getVersion() < 20) {
+ Message inMessage = message.getExchange().getInMessage();
+ BooleanSupplier committed =
inMessage.get(AbstractHTTPDestination.HTTP_RESPONSE)
+ instanceof HttpServletResponse response ?
response::isCommitted : () -> true;
+ DeadlineOutputStream out = new
DeadlineOutputStream(context.getOutputStream(),
+ maxResponseWriteMillis,
request.getComponents().getScheduler(), committed,
+ () ->
request.getConnectionMetaData().getConnection().getEndPoint()
+ .close(new
ResponseWriteTimeoutException("maxResponseWriteMillis exceeded")),
+ inMessage);
+ Request.addCompletionListener(request, failure -> out.complete());
+ context.setOutputStream(out);
+ // CXF closes the message's stream after the writer; ours skips
that once aborted
+ if (message.getContent(OutputStream.class) == out.getDelegate()) {
+ message.setContent(OutputStream.class, out);
+ }
+ try {
+ context.proceed();
+ } catch (ResponseWriteTimeoutException e) {
+ // the timer logged it and closed the connection; CXF would
log it again as an error
+ if (!out.isAborted()) {
+ throw e;
+ }
+ }
+ return;
+ }
+ context.proceed();
+ }
+
+ /**
+ * From the first write, waits for the response to commit, then arms the
deadline. The
+ * timer and the request's completion race to end it, and exactly one
wins: once completed
+ * it never aborts; once aborted, writes fail and flush and close do
nothing.
+ */
+ static final class DeadlineOutputStream extends FilterOutputStream {
+
+ // how often to look for the commit; the deadline may start this much
late
+ private static final long COMMIT_POLL_MILLIS = 1000;
+
+ private static final int WRITING = 0;
+ private static final int DONE = 1;
+ private static final int ABORTED = 2;
+
+ private final long maxWriteMillis;
+ private final Scheduler scheduler;
+ private final BooleanSupplier committed;
+ private final Runnable abort;
+ private final Message request;
+ private final AtomicInteger state = new AtomicInteger(WRITING);
+ private volatile Scheduler.Task timer;
+
+ /**
+ * @param abort closes the connection under any blocked write
+ * @param request the inbound message, for the log line the timer
thread writes
+ */
+ DeadlineOutputStream(OutputStream out, long maxWriteMillis, Scheduler
scheduler,
+ BooleanSupplier committed, Runnable abort,
Message request) {
+ super(out);
+ this.maxWriteMillis = maxWriteMillis;
+ this.scheduler = scheduler;
+ this.committed = committed;
+ this.abort = abort;
+ this.request = request;
+ }
+
+ private OutputStream getDelegate() {
+ return out;
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ start();
+ try {
+ out.write(b);
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ start();
+ try {
+ out.write(b, off, len);
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (state.get() == ABORTED) {
+ return;
+ }
+ try {
+ out.flush();
+ } catch (IOException e) {
+ throw timedOutOr(e);
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (state.get() == ABORTED) {
+ return;
+ }
+ try {
+ out.close();
+ } catch (IOException e) {
+ // aborted during the final flush: already logged, and the
connection is gone
+ if (state.get() != ABORTED) {
+ throw e;
+ }
+ }
+ }
+
+ boolean isAborted() {
+ return state.get() == ABORTED;
+ }
+
+ void complete() {
+ if (state.compareAndSet(WRITING, DONE)) {
+ Scheduler.Task t = timer;
+ if (t != null) {
+ t.cancel();
+ }
+ }
+ }
+
+ private void start() throws IOException {
+ if (state.get() == ABORTED) {
+ throw timeout();
+ }
+ if (timer == null && state.get() == WRITING) {
+ awaitCommit();
+ }
+ }
+
+ private void awaitCommit() {
+ if (state.get() != WRITING) {
+ return;
+ }
+ // a task scheduled after complete() cancels the old one is
harmless: expire() CAS-fails
+ timer = committed.getAsBoolean()
+ ? scheduler.schedule(this::expire, maxWriteMillis,
TimeUnit.MILLISECONDS)
+ : scheduler.schedule(this::awaitCommit,
Math.min(COMMIT_POLL_MILLIS, maxWriteMillis),
+ TimeUnit.MILLISECONDS);
+ }
+
+ private void expire() {
+ if (state.compareAndSet(WRITING, ABORTED)) {
+ TransferLimitLog.log(request, timeout().getMessage());
+ abort.run();
+ }
+ }
+
+ private IOException timedOutOr(IOException e) {
+ return state.get() == ABORTED ? timeout() : e;
+ }
+
+ private ResponseWriteTimeoutException timeout() {
+ return new ResponseWriteTimeoutException(
+ "Response not written within maxResponseWriteMillis (" +
maxWriteMillis + " ms)");
+ }
+ }
+
+ /** Distinct type so its mapper, not the catch-all, handles it. */
+ public static final class ResponseWriteTimeoutException extends
IOException {
+ ResponseWriteTimeoutException(String message) {
+ super(message);
+ }
+ }
+
+ /** Entity-less: the connection is gone, so CXF has nothing to write; the
timer already logged it. */
+ @Provider
+ public static final class ResponseWriteTimeoutExceptionMapper
+ implements ExceptionMapper<ResponseWriteTimeoutException> {
+ @Override
+ public Response toResponse(ResponseWriteTimeoutException e) {
+ return Response.status(Response.Status.REQUEST_TIMEOUT).build();
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
index 6747f953ac..16513ef886 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TikaServerConfig.java
@@ -66,14 +66,31 @@ public class TikaServerConfig {
*/
public static final long DEFAULT_MAX_REQUEST_SIZE_BYTES = 1024L * 1024L *
1024L;
+ /** Default time allowed to receive a request body: 5 minutes, so 1 GiB
needs about 29 Mbit/s. */
+ public static final long DEFAULT_MAX_REQUEST_READ_MILLIS = 5L * 60L *
1000L;
+
+ /** Default time allowed to write a response body: 5 minutes. */
+ public static final long DEFAULT_MAX_RESPONSE_WRITE_MILLIS = 5L * 60L *
1000L;
+
+ /** Default connector idle timeout: 2 minutes. Ends clients that stop
sending or reading entirely. */
+ public static final int DEFAULT_IDLE_TIMEOUT_MILLIS = 120_000;
+
+ /** Default cap on request-body bytes in flight: 0, meaning
maxRequestSizeBytes x 2 x pipes clients. */
+ public static final long DEFAULT_MAX_IN_FLIGHT_REQUEST_BYTES = 0;
+
/** How long a full /async queue blocks a POST before it is rejected with
429. */
public static final long DEFAULT_MAX_QUEUE_PAUSE_MILLIS = 60000;
private boolean allowPipes = false;
private boolean allowPerRequestConfig = false;
+ private boolean allowHttp2 = false;
private String cors = "";
private long maxQueuePauseMillis = DEFAULT_MAX_QUEUE_PAUSE_MILLIS;
private long maxRequestSizeBytes = DEFAULT_MAX_REQUEST_SIZE_BYTES;
+ private long maxRequestReadMillis = DEFAULT_MAX_REQUEST_READ_MILLIS;
+ private long maxResponseWriteMillis = DEFAULT_MAX_RESPONSE_WRITE_MILLIS;
+ private int idleTimeoutMillis = DEFAULT_IDLE_TIMEOUT_MILLIS;
+ private long maxInFlightRequestBytes = DEFAULT_MAX_IN_FLIGHT_REQUEST_BYTES;
private String id = UUID
.randomUUID()
.toString();
@@ -225,6 +242,20 @@ public class TikaServerConfig {
this.allowPerRequestConfig = allowPerRequestConfig;
}
+ /**
+ * Whether to accept cleartext HTTP/2 (h2c) as well as HTTP/1.1; off by
default. HTTP/2
+ * requests get no {@code maxResponseWriteMillis}: its abort closes the
connection, which
+ * would end every request multiplexed on it. HTTP/2 over TLS is not
offered either way
+ * (tika-server ships no ALPN server provider).
+ */
+ public boolean isAllowHttp2() {
+ return allowHttp2;
+ }
+
+ public void setAllowHttp2(boolean allowHttp2) {
+ this.allowHttp2 = allowHttp2;
+ }
+
private void validateConsistency(Set<String> settings) throws
TikaConfigException {
if (host == null) {
throw new TikaConfigException("Must specify 'host'");
@@ -248,6 +279,11 @@ public class TikaServerConfig {
}
}
tlsConfig.checkInitialization();
+ if (maxInFlightRequestBytes > 0 && maxRequestSizeBytes >= 0
+ && maxInFlightRequestBytes < maxRequestSizeBytes) {
+ throw new TikaConfigException("maxInFlightRequestBytes (" +
maxInFlightRequestBytes
+ + ") must be at least maxRequestSizeBytes (" +
maxRequestSizeBytes + ")");
+ }
if (metricsPort != null) {
if (metricsPort < 0 || metricsPort > 65535) {
throw new TikaConfigException("metricsPort must be 0-65535,
got " + metricsPort);
@@ -349,6 +385,73 @@ public class TikaServerConfig {
this.maxRequestSizeBytes = maxRequestSizeBytes;
}
+ /**
+ * How long, from the start of the request, reading its body may take;
defaults to
+ * {@link #DEFAULT_MAX_REQUEST_READ_MILLIS} (5 minutes). A request still
sending after
+ * that is rejected with 408. A negative value disables the deadline; zero
is rejected.
+ */
+ public long getMaxRequestReadMillis() {
+ return maxRequestReadMillis;
+ }
+
+ public void setMaxRequestReadMillis(long maxRequestReadMillis) throws
TikaConfigException {
+ this.maxRequestReadMillis = nonZero("maxRequestReadMillis",
maxRequestReadMillis);
+ }
+
+ /**
+ * How long, from when its status line is sent, writing a response body
may take; defaults to
+ * {@link #DEFAULT_MAX_RESPONSE_WRITE_MILLIS} (5 minutes). A response
still being written
+ * after that is cut off. A negative value disables the deadline; zero is
rejected.
+ */
+ public long getMaxResponseWriteMillis() {
+ return maxResponseWriteMillis;
+ }
+
+ public void setMaxResponseWriteMillis(long maxResponseWriteMillis) throws
TikaConfigException {
+ this.maxResponseWriteMillis = nonZero("maxResponseWriteMillis",
maxResponseWriteMillis);
+ }
+
+ // zero would fail every transfer, not disable the deadline
+ private static long nonZero(String name, long millis) throws
TikaConfigException {
+ if (millis == 0) {
+ throw new TikaConfigException(name + " must be positive, or
negative to disable it");
+ }
+ return millis;
+ }
+
+ /**
+ * How long a connection may go without any bytes moving before it is
closed; defaults to
+ * {@link #DEFAULT_IDLE_TIMEOUT_MILLIS} (2 minutes). Must be positive:
Jetty treats zero
+ * as "use its own default", not "never".
+ */
+ public int getIdleTimeoutMillis() {
+ return idleTimeoutMillis;
+ }
+
+ public void setIdleTimeoutMillis(int idleTimeoutMillis) throws
TikaConfigException {
+ if (idleTimeoutMillis <= 0) {
+ throw new TikaConfigException("idleTimeoutMillis must be positive:
" + idleTimeoutMillis);
+ }
+ this.idleTimeoutMillis = idleTimeoutMillis;
+ }
+
+ /**
+ * How many request-body bytes all requests together may hold at once,
counted as they
+ * are read and freed when each response completes; a request that would
go over gets 429.
+ * This bounds the temp disk and memory request bodies take, without
letting a client that
+ * sends slowly hold capacity it has not used. 0 (the default) means
+ * {@code maxRequestSizeBytes} x 2 x the pipes clients; negative disables
it; a positive
+ * value below {@code maxRequestSizeBytes} is rejected, as a request of
that size could
+ * never succeed.
+ */
+ public long getMaxInFlightRequestBytes() {
+ return maxInFlightRequestBytes;
+ }
+
+ public void setMaxInFlightRequestBytes(long maxInFlightRequestBytes) {
+ this.maxInFlightRequestBytes = maxInFlightRequestBytes;
+ }
+
/**
* How long a POST to /async blocks when the queue is full before it is
rejected with
* 429 (Retry-After); defaults to {@link #DEFAULT_MAX_QUEUE_PAUSE_MILLIS}.
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 a986773f0b..dbc9f0ac12 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
@@ -52,6 +52,7 @@ import
org.apache.cxf.rs.security.cors.CrossOriginResourceSharingFilter;
import org.apache.cxf.service.factory.ServiceConstructionException;
import org.apache.cxf.transport.common.gzip.GZIPInInterceptor;
import org.apache.cxf.transport.common.gzip.GZIPOutInterceptor;
+import org.apache.cxf.transport.http.HttpServerEngineSupport;
import org.apache.cxf.transport.http_jetty.JettyHTTPServerEngineFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -248,6 +249,11 @@ public class TikaServerProcess {
if ("*".equals(config.getCors())) {
LOG.warn("cors is '*': any origin may call this server from a
browser.");
}
+ if (config.isAllowHttp2() && config.getMaxResponseWriteMillis() >= 0) {
+ LOG.warn("allowHttp2 is enabled: maxResponseWriteMillis does not
apply to HTTP/2 "
+ + "responses; a client reading an HTTP/2 response slowly
is bounded only "
+ + "by idleTimeoutMillis.");
+ }
if ("0.0.0.0".equals(config.getHost()) ||
"*".equals(config.getHost())) {
LOG.warn("host is {}: the server is reachable on every interface.
tika-server "
+ "performs no authentication; restrict access at the
network layer.",
@@ -295,7 +301,11 @@ public class TikaServerProcess {
List<ResourceProvider> resourceProviders = new ArrayList<>();
List<Object> providers = new ArrayList<>();
- loadAllProviders(tikaServerConfig, serverStatus, tikaResource,
resourceProviders, providers);
+ long maxInFlight = maxInFlightRequestBytes(tikaServerConfig,
+ pipesParsingHelper == null ? 1 :
pipesParsingHelper.getPipesConfig().getNumClients());
+ RequestBodyBudget bodyBudget = maxInFlight > 0 ? new
RequestBodyBudget(maxInFlight) : null;
+ LOG.info("maxInFlightRequestBytes: {}", maxInFlight > 0 ? maxInFlight
: "disabled");
+ loadAllProviders(tikaServerConfig, serverStatus, tikaResource,
resourceProviders, providers, bodyBudget);
TikaServerMetrics metrics = null;
MetricsServer metricsServer = null;
@@ -303,6 +313,9 @@ public class TikaServerProcess {
metrics = new TikaServerMetrics();
metrics.bindJvm();
metrics.bindServerStatus(serverStatus);
+ if (bodyBudget != null) {
+ metrics.bindRequestBodyBudget(bodyBudget);
+ }
if (pipesParsingHelper != null) {
metrics.bindSyncPool(pipesParsingHelper.getPipesParser());
}
@@ -321,7 +334,8 @@ public class TikaServerProcess {
//set compression interceptors
sf.setOutInterceptors(Collections.singletonList(new
GZIPOutInterceptor()));
- sf.setInInterceptors(Collections.singletonList(new
GZIPInInterceptor()));
+ sf.setInInterceptors(List.of(new GZIPInInterceptor(),
+ new
RequestReadDeadlineInterceptor(tikaServerConfig.getMaxRequestReadMillis())));
warnOnPermissiveConfig(tikaServerConfig);
@@ -350,6 +364,16 @@ public class TikaServerProcess {
factory.setBus(sf.getBus());
manager.registerBindingFactory(JAXRSBindingFactory.JAXRS_BINDING_ID, factory);
}
+ // CXF enables HTTP/2 whenever its jar is present; "true" would also
make it required,
+ // and TLS startup fails without an ALPN provider, so allowing it
leaves the property unset
+ if (!tikaServerConfig.isAllowHttp2()) {
+ sf.getBus().setProperty(HttpServerEngineSupport.ENABLE_HTTP2,
false);
+ }
+ // CXF keys engines by port and reuses this one when the endpoint binds
+ sf.getBus()
+ .getExtension(JettyHTTPServerEngineFactory.class)
+ .createJettyHTTPServerEngine(host, port, protocol)
+ .setMaxIdleTime(tikaServerConfig.getIdleTimeoutMillis());
ServerDetails details = new ServerDetails();
details.sf = sf;
details.url = url;
@@ -361,6 +385,23 @@ public class TikaServerProcess {
return details;
}
+ /** 0 resolves to maxRequestSizeBytes x 2 x the pipes clients (off if that
size is unbounded); negative is off. */
+ static long maxInFlightRequestBytes(TikaServerConfig config, int
numClients) {
+ long configured = config.getMaxInFlightRequestBytes();
+ if (configured != 0) {
+ return configured;
+ }
+ long perRequest = config.getMaxRequestSizeBytes();
+ if (perRequest < 0) {
+ return -1;
+ }
+ try {
+ return Math.multiplyExact(perRequest, 2L * numClients);
+ } catch (ArithmeticException e) {
+ return Long.MAX_VALUE;
+ }
+ }
+
private static AsyncResource findAsyncResource(List<ResourceProvider>
resourceProviders) {
for (ResourceProvider p : resourceProviders) {
if (p instanceof SingletonResourceProvider s
@@ -428,7 +469,8 @@ public class TikaServerProcess {
}
private static void loadAllProviders(TikaServerConfig tikaServerConfig,
ServerStatus serverStatus,
- TikaResource tikaResource,
List<ResourceProvider> resourceProviders, List<Object> writers)
+ TikaResource tikaResource,
List<ResourceProvider> resourceProviders, List<Object> writers,
+ RequestBodyBudget bodyBudget)
throws TikaException, SAXException, IOException {
List<ResourceProvider> tmpCoreProviders =
loadCoreProviders(tikaServerConfig, serverStatus, tikaResource);
@@ -451,9 +493,14 @@ public class TikaServerProcess {
// Add ConfigEndpointSecurityFilter to gate /config endpoints
writers.add(new
ConfigEndpointSecurityFilter(tikaServerConfig.isAllowPerRequestConfig()));
- writers.add(new
MaxRequestSizeFilter(tikaServerConfig.getMaxRequestSizeBytes()));
+ writers.add(new
MaxRequestSizeFilter(tikaServerConfig.getMaxRequestSizeBytes(),
+ tikaServerConfig.getMaxRequestReadMillis(), bodyBudget));
// 413 for over-limit chunked bodies, which bypass the Content-Length
pre-check.
writers.add(new MaxRequestSizeFilter.RequestTooLargeExceptionMapper());
+ writers.add(new MaxRequestSizeFilter.RequestTimeoutExceptionMapper());
+ writers.add(new
MaxRequestSizeFilter.RequestBudgetExceededExceptionMapper());
+ writers.add(new
ResponseWriteDeadlineInterceptor(tikaServerConfig.getMaxResponseWriteMillis()));
+ writers.add(new
ResponseWriteDeadlineInterceptor.ResponseWriteTimeoutExceptionMapper());
// setRequestLogLevel rejects anything but debug/info, so no
validation needed here.
TikaLoggingFilter logFilter = null;
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TransferLimitLog.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TransferLimitLog.java
new file mode 100644
index 0000000000..902e84179b
--- /dev/null
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/TransferLimitLog.java
@@ -0,0 +1,64 @@
+/*
+ * 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.server.core;
+
+import org.apache.cxf.jaxrs.utils.JAXRSUtils;
+import org.apache.cxf.message.Message;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * One line, no stack trace, per request refused or cut off by a transfer
limit (413, 408,
+ * 429, aborted response): its own logger, so these client-caused events can
be routed or
+ * silenced apart from server errors.
+ */
+public final class TransferLimitLog {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(TransferLimitLog.class);
+ private static final String LOGGED = TransferLimitLog.class.getName() +
".logged";
+
+ private TransferLimitLog() {
+ }
+
+ static void log(String reason) {
+ Message current = JAXRSUtils.getCurrentMessage();
+ // while the response is written, the current message is the outbound
one
+ log(current == null || current.getExchange() == null
+ ? current : current.getExchange().getInMessage(), reason);
+ }
+
+ /**
+ * For a thread with no current CXF message: {@code request} is the
inbound one, or null.
+ * Once per request, as CXF may map one failure twice (multipart); returns
whether it logged.
+ */
+ static boolean log(Message request, String reason) {
+ if (!LOG.isInfoEnabled()) {
+ return false;
+ }
+ if (request == null) {
+ LOG.info("{}", reason);
+ return true;
+ }
+ // the exchange is a ConcurrentHashMap, so this holds across threads
too
+ if (request.getExchange() != null && request.getExchange().put(LOGGED,
Boolean.TRUE) != null) {
+ return false;
+ }
+ LOG.info("{} {}: {}", request.get(Message.HTTP_REQUEST_METHOD),
+ request.get(Message.REQUEST_URI), reason);
+ return true;
+ }
+}
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
index bd98312302..6da0ddf65c 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/metrics/TikaServerMetrics.java
@@ -38,6 +38,7 @@ import
io.micrometer.prometheusmetrics.PrometheusMeterRegistry;
import org.apache.tika.pipes.core.PipesParser;
import org.apache.tika.pipes.core.RestartReason;
import org.apache.tika.pipes.core.async.AsyncProcessor;
+import org.apache.tika.server.core.RequestBodyBudget;
import org.apache.tika.server.core.ServerStatus;
/**
@@ -73,6 +74,7 @@ public final class TikaServerMetrics implements AutoCloseable
{
static final String REQUEST_SIZE = "tika_server_request_size_bytes";
static final String REJECTED = "tika_server_rejected_total";
static final String TASKS_ACTIVE = "tika_server_tasks_active";
+ static final String REQUEST_BYTES_IN_FLIGHT =
"tika_server_request_body_in_flight_bytes";
static final String PIPES_WORKERS = "tika_pipes_workers";
static final String PIPES_RESTARTS = "tika_pipes_worker_restarts_total";
static final String PIPES_QUEUE_DEPTH = "tika_pipes_queue_depth";
@@ -113,6 +115,13 @@ public final class TikaServerMetrics implements
AutoCloseable {
.register(registry);
}
+ public void bindRequestBodyBudget(RequestBodyBudget budget) {
+ Gauge.builder(REQUEST_BYTES_IN_FLIGHT, budget,
RequestBodyBudget::getInFlightBytes)
+ .description("Request-body bytes held by in-flight requests,
against maxInFlightRequestBytes")
+ .baseUnit("bytes")
+ .register(registry);
+ }
+
/** The sync endpoints' forks: restart counters plus busy/idle slots. */
public void bindSyncPool(PipesParser pipesParser) {
bindRestarts(POOL_SYNC, pipesParser::getRestartCount);
@@ -187,11 +196,12 @@ public final class TikaServerMetrics implements
AutoCloseable {
return (status / 100) + "xx";
}
- /** By status, as tika-server already maps them: 429 capacity, 503 fork
OOM/timeout/crash, 413 body limit. */
+ /** By status, as tika-server already maps them: 429 capacity, 503 fork
OOM/timeout/crash, 413 body limit, 408 body too slow. */
static String rejectedReason(int status) {
return switch (status) {
case 429 -> "busy_429";
case 413 -> "payload_413";
+ case 408 -> "slow_408";
case 503 -> "crash_503";
default -> null;
};
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/LanguageResource.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/LanguageResource.java
index cc6aee3182..03be9c4954 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/LanguageResource.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/LanguageResource.java
@@ -29,6 +29,7 @@ import jakarta.ws.rs.POST;
import jakarta.ws.rs.PUT;
import jakarta.ws.rs.Path;
import jakarta.ws.rs.Produces;
+import org.apache.commons.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -67,7 +68,10 @@ public class LanguageResource {
public static final int MAX_DETECT_CHARS = 100_000;
private String detectStream(InputStream is) throws IOException {
- return detectString(readAtMost(is, MAX_DETECT_CHARS));
+ String text = readAtMost(is, MAX_DETECT_CHARS);
+ // so maxRequestSizeBytes applies to the whole body, not just the
sample
+ IOUtils.consume(is);
+ return detectString(text);
}
/** Reads up to maxChars without materializing the rest of the stream. */
diff --git
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaWelcome.java
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaWelcome.java
index 164498435b..0202ecd7d0 100644
---
a/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaWelcome.java
+++
b/tika-server/tika-server-core/src/main/java/org/apache/tika/server/core/resource/TikaWelcome.java
@@ -78,7 +78,11 @@ public class TikaWelcome {
protected List<Endpoint> identifyEndpoints() {
List<Endpoint> found = new ArrayList<>();
for (Class<?> endpoint : endpoints) {
- Path p = endpoint.getAnnotation(Path.class);
+ // @Path is not @Inherited, but JAX-RS resolves it from
superclasses
+ Path p = null;
+ for (Class<?> c = endpoint; p == null && c != null; c =
c.getSuperclass()) {
+ p = c.getAnnotation(Path.class);
+ }
String basePath = null;
if (p != null) {
basePath = p
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
index bebc5d8267..44fb8953e6 100644
---
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/IntegrationTestBase.java
@@ -76,7 +76,7 @@ public class IntegrationTestBase extends TikaTest {
// `static final`, computed once for the whole JVM fork and shared by every
// subclass; harmless serially, but a guaranteed bind collision if two
// subclasses' tests ever ran concurrently in the same fork.
- final int integrationTestPort = TestPortAllocator.findFreePort();
+ protected final int integrationTestPort = TestPortAllocator.findFreePort();
final String INTEGRATION_TEST_PORT = String.valueOf(integrationTestPort);
protected final String endPoint = "http://localhost:"; +
INTEGRATION_TEST_PORT;
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/RequestBodyBudgetTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/RequestBodyBudgetTest.java
new file mode 100644
index 0000000000..cb8182c649
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/RequestBodyBudgetTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.junit.jupiter.api.Test;
+
+/** The accounting contract: the total changes only through reservations, each
released once. */
+public class RequestBodyBudgetTest {
+
+ @Test
+ public void testChargeGrowsToBytesReadAndReleaseRefundsAll() {
+ RequestBodyBudget budget = new RequestBodyBudget(100);
+ RequestBodyBudget.Reservation r = budget.newReservation();
+ assertTrue(r.ensure(30));
+ assertTrue(r.ensure(20));
+ assertEquals(30, budget.getInFlightBytes());
+ assertTrue(r.ensure(70));
+ assertEquals(70, budget.getInFlightBytes());
+ r.release();
+ assertEquals(0, budget.getInFlightBytes());
+ }
+
+ @Test
+ public void testOverBudgetChargesNothing() {
+ RequestBodyBudget budget = new RequestBodyBudget(100);
+ RequestBodyBudget.Reservation a = budget.newReservation();
+ RequestBodyBudget.Reservation b = budget.newReservation();
+ assertTrue(a.ensure(60));
+ assertTrue(b.ensure(10));
+ assertFalse(b.ensure(50));
+ assertEquals(70, budget.getInFlightBytes());
+ a.release();
+ assertTrue(b.ensure(50));
+ assertEquals(50, budget.getInFlightBytes());
+ b.release();
+ assertEquals(0, budget.getInFlightBytes());
+ }
+
+ @Test
+ public void testReleaseIsIdempotentAndCloses() {
+ RequestBodyBudget budget = new RequestBodyBudget(100);
+ RequestBodyBudget.Reservation a = budget.newReservation();
+ RequestBodyBudget.Reservation b = budget.newReservation();
+ assertTrue(a.ensure(40));
+ assertTrue(b.ensure(40));
+ a.release();
+ a.release();
+ assertEquals(40, budget.getInFlightBytes());
+ // a late read after completion must not charge again
+ assertTrue(a.ensure(90));
+ assertEquals(40, budget.getInFlightBytes());
+ b.release();
+ assertEquals(0, budget.getInFlightBytes());
+ }
+
+ @Test
+ public void testNonPositiveBudgetRejected() {
+ assertThrows(IllegalArgumentException.class, () -> new
RequestBodyBudget(0));
+ assertThrows(IllegalArgumentException.class, () -> new
RequestBodyBudget(-1));
+ }
+
+ /** Concurrent charges never exceed the budget, and everything drains to
zero. */
+ @Test
+ public void testConcurrentChargesNeverExceedAndDrain() throws Exception {
+ long max = 10_000;
+ RequestBodyBudget budget = new RequestBodyBudget(max);
+ ExecutorService pool = Executors.newFixedThreadPool(8);
+ try {
+ List<Future<?>> futures = new ArrayList<>();
+ for (int t = 0; t < 8; t++) {
+ long seed = t;
+ futures.add(pool.submit(() -> {
+ Random random = new Random(seed);
+ for (int i = 0; i < 5_000; i++) {
+ RequestBodyBudget.Reservation r =
budget.newReservation();
+ long read = 0;
+ for (int k = 0; k < 4; k++) {
+ read += random.nextInt(1_000);
+ if (!r.ensure(read)) {
+ break;
+ }
+ if (budget.getInFlightBytes() > max) {
+ throw new AssertionError("over budget: " +
budget.getInFlightBytes());
+ }
+ }
+ r.release();
+ if (random.nextBoolean()) {
+ r.release();
+ }
+ }
+ }));
+ }
+ for (Future<?> f : futures) {
+ f.get();
+ }
+ } finally {
+ pool.shutdownNow();
+ }
+ assertEquals(0, budget.getInFlightBytes());
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptorTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptorTest.java
new file mode 100644
index 0000000000..ec446bbf13
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/ResponseWriteDeadlineInterceptorTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTimeoutPreemptively;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.DeadlineOutputStream;
+import
org.apache.tika.server.core.ResponseWriteDeadlineInterceptor.ResponseWriteTimeoutException;
+
+/**
+ * The deadline as a timer: a write blocked by a client that reads a trickle
(each byte resetting
+ * the idle timeout) must still end at the deadline, which a check between
writes cannot do.
+ */
+public class ResponseWriteDeadlineInterceptorTest {
+
+ private static final long DEADLINE_MILLIS = 200;
+
+ private final ScheduledExecutorScheduler scheduler = new
ScheduledExecutorScheduler();
+
+ @BeforeEach
+ public void start() throws Exception {
+ scheduler.start();
+ }
+
+ @AfterEach
+ public void stop() throws Exception {
+ scheduler.stop();
+ }
+
+ @Test
+ public void testBlockedWriteAbortedAtDeadline() {
+ BlockingSink sink = new BlockingSink();
+ AtomicInteger aborts = new AtomicInteger();
+ DeadlineOutputStream out = new DeadlineOutputStream(sink,
DEADLINE_MILLIS, scheduler, () -> true, () -> {
+ aborts.incrementAndGet();
+ sink.closeConnection();
+ }, null);
+ long start = System.nanoTime();
+ assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> assertThrows(ResponseWriteTimeoutException.class, () ->
out.write(new byte[10])));
+ assertTrue(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start) >=
DEADLINE_MILLIS);
+ assertEquals(1, aborts.get());
+ }
+
+ /** Once aborted, CXF's flush and close must not finish the body as if it
were complete. */
+ @Test
+ public void testCloseAfterAbortIsSilentNoOp() throws Exception {
+ BlockingSink sink = new BlockingSink();
+ DeadlineOutputStream out = new DeadlineOutputStream(sink,
DEADLINE_MILLIS, scheduler,
+ () -> true, sink::closeConnection, null);
+ assertTimeoutPreemptively(Duration.ofSeconds(10),
+ () -> assertThrows(ResponseWriteTimeoutException.class, () ->
out.write(1)));
+ out.flush();
+ out.close();
+ assertFalse(sink.closedByStream);
+ assertThrows(ResponseWriteTimeoutException.class, () -> out.write(1));
+ }
+
+ /** CXF drains the unread request body before committing; that wait is not
the write's. */
+ @Test
+ public void testDeadlineStartsAtCommit() throws Exception {
+ BlockingSink sink = new BlockingSink();
+ AtomicBoolean committed = new AtomicBoolean();
+ AtomicInteger aborts = new AtomicInteger();
+ DeadlineOutputStream out = new DeadlineOutputStream(sink,
DEADLINE_MILLIS, scheduler,
+ committed::get, () -> {
+ aborts.incrementAndGet();
+ sink.closeConnection();
+ }, null);
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(1);
+ } catch (IOException e) {
+ // expected once aborted
+ }
+ });
+ writer.start();
+ Thread.sleep(4 * DEADLINE_MILLIS);
+ assertEquals(0, aborts.get(), "aborted before the response committed");
+ committed.set(true);
+ writer.join(10_000);
+ assertFalse(writer.isAlive());
+ assertEquals(1, aborts.get());
+ }
+
+ @Test
+ public void testCompletedResponseNeverAborted() throws Exception {
+ AtomicInteger aborts = new AtomicInteger();
+ CountingSink sink = new CountingSink();
+ DeadlineOutputStream out = new DeadlineOutputStream(sink,
DEADLINE_MILLIS, scheduler,
+ () -> true, aborts::incrementAndGet, null);
+ out.write(new byte[100]);
+ out.close();
+ out.complete();
+ Thread.sleep(3 * DEADLINE_MILLIS);
+ assertEquals(0, aborts.get());
+ assertEquals(100, sink.count);
+ }
+
+ /** A write that blocks until the connection is closed under it, as a full
socket does. */
+ private static final class BlockingSink extends OutputStream {
+ private final CountDownLatch connectionClosed = new CountDownLatch(1);
+ private volatile boolean closedByStream;
+
+ void closeConnection() {
+ connectionClosed.countDown();
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ block();
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ block();
+ }
+
+ private void block() throws IOException {
+ try {
+ connectionClosed.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new IOException("connection closed");
+ }
+
+ @Override
+ public void close() {
+ closedByStream = true;
+ }
+ }
+
+ private static final class CountingSink extends OutputStream {
+ private long count;
+
+ @Override
+ public void write(int b) {
+ count++;
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) {
+ count += len;
+ }
+ }
+}
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerBasicIntegrationTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerBasicIntegrationTest.java
index 9ba39665a3..f3512510de 100644
---
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerBasicIntegrationTest.java
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerBasicIntegrationTest.java
@@ -75,9 +75,9 @@ public class TikaServerBasicIntegrationTest extends
IntegrationTestBase {
assertTrue(body.contains("Valid types"), body);
}
+ /** h2c is opt-in (allowHttp2): a client asking to upgrade is answered
over HTTP/1.1. */
@Test
- public void testH2c() throws Exception {
- // Using HttpClient in order to check Http2 Version
+ public void testH2cOffByDefault() throws Exception {
HttpClient httpClient = HttpClient.newBuilder()
.version(HttpClient.Version.HTTP_2)
.build();
@@ -88,7 +88,7 @@ public class TikaServerBasicIntegrationTest extends
IntegrationTestBase {
.build();
HttpResponse<String> response = httpClient.send(request,
HttpResponse.BodyHandlers.ofString(UTF_8));
assertEquals(200, response.statusCode());
- assertEquals(HttpClient.Version.HTTP_2, response.version());
+ assertEquals(HttpClient.Version.HTTP_1_1, response.version());
}
@Test
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
index 7ce1bbb390..ffb3009950 100644
---
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TikaServerConfigTest.java
@@ -23,6 +23,7 @@ import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.HashSet;
import java.util.List;
@@ -36,6 +37,7 @@ import org.apache.commons.cli.DefaultParser;
import org.apache.commons.cli.Option;
import org.apache.commons.cli.Options;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.CsvSource;
@@ -76,15 +78,49 @@ public class TikaServerConfigTest extends TikaTest {
assertTrue(config.isAllowPerRequestConfig());
}
- /** Documented defaults: 1 GiB request cap, 60s /async queue pause. */
+ /** Documented defaults: 1 GiB request cap, 5 min transfer deadlines, 2
min idle, derived in-flight budget, 60s /async queue pause. */
@Test
public void testDocumentedDefaults() {
TikaServerConfig config = new TikaServerConfig();
assertEquals(1024L * 1024 * 1024, config.getMaxRequestSizeBytes());
+ assertEquals(5L * 60 * 1000, config.getMaxRequestReadMillis());
+ assertEquals(5L * 60 * 1000, config.getMaxResponseWriteMillis());
+ assertEquals(120_000, config.getIdleTimeoutMillis());
+ assertEquals(0, config.getMaxInFlightRequestBytes());
assertEquals(TikaServerConfig.DEFAULT_MAX_QUEUE_PAUSE_MILLIS,
config.getMaxQueuePauseMillis());
assertEquals(60000L, TikaServerConfig.DEFAULT_MAX_QUEUE_PAUSE_MILLIS);
}
+ /** Jetty reads a non-positive idle timeout as "use my default", not
"never". */
+ @Test
+ public void testIdleTimeoutMustBePositive() {
+ TikaServerConfig config = new TikaServerConfig();
+ assertThrows(TikaConfigException.class, () ->
config.setIdleTimeoutMillis(0));
+ assertThrows(TikaConfigException.class, () ->
config.setIdleTimeoutMillis(-1));
+ }
+
+ /** Zero would fail every transfer rather than disable the deadline. */
+ @Test
+ public void testZeroDeadlinesRejected() throws Exception {
+ TikaServerConfig config = new TikaServerConfig();
+ assertThrows(TikaConfigException.class, () ->
config.setMaxRequestReadMillis(0));
+ assertThrows(TikaConfigException.class, () ->
config.setMaxResponseWriteMillis(0));
+ config.setMaxRequestReadMillis(-1);
+ config.setMaxResponseWriteMillis(-1);
+ }
+
+ /** A budget below one maximum-size request would refuse that request
forever. */
+ @Test
+ public void testInFlightBudgetBelowRequestSizeRejected(@TempDir Path dir)
throws Exception {
+ Path path = Files.writeString(dir.resolve("budget.json"),
+ "{\"server\": {\"maxRequestSizeBytes\": 1000,
\"maxInFlightRequestBytes\": 999}}");
+ Options options = new
Options().addOption(Option.builder("c").longOpt("config").hasArg().get());
+ String[] args = {"-c",
ProcessUtils.escapeCommandLine(path.toAbsolutePath().toString())};
+ TikaConfigException ex = assertThrows(TikaConfigException.class,
+ () -> TikaServerConfig.load(new DefaultParser().parse(options,
args), null));
+ assertContains("maxInFlightRequestBytes", ex.getMessage());
+ }
+
@Test
public void testPresetsFromCommandLineAndEnvironment() throws Exception {
Options options = new
Options().addOption(Option.builder().longOpt("preset").hasArg().get());
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 5622021f5b..3525193d06 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
@@ -34,6 +34,22 @@ import org.apache.tika.server.core.resource.MetadataResource;
public class TikaServerProcessTest {
+ /** 0 is maxRequestSizeBytes x 2 x clients, off when that size is
unbounded; explicit values pass through. */
+ @Test
+ public void testMaxInFlightRequestBytesResolution() {
+ TikaServerConfig config = new TikaServerConfig();
+ config.setMaxRequestSizeBytes(1000);
+ assertEquals(6000, TikaServerProcess.maxInFlightRequestBytes(config,
3));
+ config.setMaxRequestSizeBytes(Long.MAX_VALUE / 2);
+ assertEquals(Long.MAX_VALUE,
TikaServerProcess.maxInFlightRequestBytes(config, 3));
+ config.setMaxRequestSizeBytes(-1);
+ assertTrue(TikaServerProcess.maxInFlightRequestBytes(config, 3) < 0);
+ config.setMaxInFlightRequestBytes(5000);
+ assertEquals(5000, TikaServerProcess.maxInFlightRequestBytes(config,
3));
+ config.setMaxInFlightRequestBytes(-1);
+ assertTrue(TikaServerProcess.maxInFlightRequestBytes(config, 3) < 0);
+ }
+
private static TikaServerConfig config(boolean allowPipes, String...
endpoints) {
TikaServerConfig c = new TikaServerConfig();
c.setEndpoints(new ArrayList<>(List.of(endpoints)));
diff --git
a/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TransferLimitLogTest.java
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TransferLimitLogTest.java
new file mode 100644
index 0000000000..a6297e1048
--- /dev/null
+++
b/tika-server/tika-server-core/src/test/java/org/apache/tika/server/core/TransferLimitLogTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.server.core;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.cxf.message.ExchangeImpl;
+import org.apache.cxf.message.Message;
+import org.apache.cxf.message.MessageImpl;
+import org.junit.jupiter.api.Test;
+
+public class TransferLimitLogTest {
+
+ /** CXF maps a multipart 413 twice; the second mapping must not log again.
*/
+ @Test
+ public void testOncePerRequest() {
+ Message first = request();
+ assertTrue(TransferLimitLog.log(first, "too large"));
+ assertFalse(TransferLimitLog.log(first, "too large"));
+ assertTrue(TransferLimitLog.log(request(), "too large"));
+ }
+
+ private static Message request() {
+ Message message = new MessageImpl();
+ ExchangeImpl exchange = new ExchangeImpl();
+ exchange.setInMessage(message);
+ message.put(Message.HTTP_REQUEST_METHOD, "POST");
+ message.put(Message.REQUEST_URI, "/meta/form");
+ return message;
+ }
+}
diff --git
a/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java
b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java
new file mode 100644
index 0000000000..82f2625557
--- /dev/null
+++
b/tika-server/tika-server-standard/src/test/java/org/apache/tika/server/standard/MaxRequestSizeIntegrationTest.java
@@ -0,0 +1,606 @@
+/*
+ * 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.server.standard;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.io.BufferedReader;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.net.URI;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+import jakarta.ws.rs.core.Response;
+import org.apache.cxf.jaxrs.client.WebClient;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import org.apache.tika.server.core.IntegrationTestBase;
+import org.apache.tika.server.core.TestPortAllocator;
+
+/**
+ * Transfer limits on a real server, across every body-accepting endpoint its
welcome page
+ * lists.
+ */
+public class MaxRequestSizeIntegrationTest extends IntegrationTestBase {
+
+ // above what /language reads, so that endpoint cannot trip the limit by
reading alone
+ private static final int MAX_BYTES = 200_000;
+ private static final int MAX_READ_MILLIS = 3_000;
+ private static final int MAX_WRITE_MILLIS = 3_000;
+ private static final int IDLE_TIMEOUT_MILLIS = 2_000;
+ // room for two maximum-size bodies and part of a third
+ private static final int MAX_IN_FLIGHT_BYTES = 3 * MAX_BYTES;
+ // past the 16 MiB CXF discards after a response, so a reply proves
reading stopped
+ private static final int SENT_BYTES = 20 * 1024 * 1024;
+ private static final String BOUNDARY = "tika-limit-boundary";
+ // from tika-server-core's test preset catalog
+ private static final String PRESET = "catalog-xml-content";
+ // text in bigTextZip(): far more than the socket buffers hold
+ private static final int TEXT_BYTES = 16 * 1024 * 1024;
+ private static final String[] TIKA_TEXT = {"PUT", "/tika/text",
"text/plain"};
+
+ private static final Map<String, byte[]> PREAMBLES = new LinkedHashMap<>();
+
+ static {
+ PREAMBLES.put("text/plain", new byte[0]);
+ // an open JSON string, so a JSON reader keeps consuming
+ PREAMBLES.put("application/json", "{\"a\":\"".getBytes(UTF_8));
+ PREAMBLES.put("multipart/form-data; boundary=" + BOUNDARY,
+ ("--" + BOUNDARY + "\r\nContent-Disposition: form-data;
name=\"file\"; "
+ + "filename=\"a.txt\"\r\nContent-Type:
text/plain\r\n\r\n").getBytes(UTF_8));
+ }
+
+ @TempDir
+ static Path classWorkingDir;
+
+ private final int metricsPort = TestPortAllocator.findFreePort();
+
+ @BeforeAll
+ public void startServer() throws Exception {
+ // every endpoint: the defaults, plus pipes/async via allowPipes and
the config routes
+ String config = """
+ {
+ "server": {
+ "allowPipes": true,
+ "allowPerRequestConfig": true,
+ "allowHttp2": true,
+ "maxRequestSizeBytes": %d,
+ "maxRequestReadMillis": %r,
+ "maxResponseWriteMillis": %w,
+ "idleTimeoutMillis": %i,
+ "maxInFlightRequestBytes": %c,
+ "metricsPort": %m
+ },
+ "pipes": { "numClients": 1, "forkedJvmArgs": ["-Xmx256m"] },
+ "plugin-roots": "%s"
+ }
+ """
+ .replace("%d", Integer.toString(MAX_BYTES))
+ .replace("%r", Integer.toString(MAX_READ_MILLIS))
+ .replace("%w", Integer.toString(MAX_WRITE_MILLIS))
+ .replace("%i", Integer.toString(IDLE_TIMEOUT_MILLIS))
+ .replace("%c", Integer.toString(MAX_IN_FLIGHT_BYTES))
+ .replace("%m", Integer.toString(metricsPort))
+ .replace("%s",
Paths.get("target/plugins").toAbsolutePath().toString().replace("\\", "/"));
+ Path configFile =
Files.writeString(classWorkingDir.resolve("request-limit.json"), config);
+ startClassProcess(new String[]{"-config", configFile.toString(),
"--preset", PRESET},
+ classWorkingDir);
+ }
+
+ /** Whatever a test did to the server, every request's share of the body
budget comes back. */
+ @AfterEach
+ public void assertBudgetDrained() throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(10);
+ long inFlight;
+ do {
+ inFlight = bodyBytesInFlight();
+ if (inFlight != 0) {
+ Thread.sleep(50);
+ }
+ } while (inFlight != 0 && System.nanoTime() < deadline);
+ assertEquals(0, inFlight, "request body budget not released");
+ }
+
+ @Test
+ public void testEveryEndpointAnswersOverLimitChunkedBeforeBodyEnds()
throws Exception {
+ List<String[]> endpoints = bodyEndpoints();
+ assertTrue(endpoints.size() > 40, "welcome page listed too few
endpoints: " + endpoints.size());
+ List<String> failures = new ArrayList<>();
+ for (String[] endpoint : endpoints) {
+ boolean anyTooLarge = false;
+ List<String> statuses = new ArrayList<>();
+ for (Map.Entry<String, byte[]> e : PREAMBLES.entrySet()) {
+ String status = sendChunked(endpoint, e.getKey(),
e.getValue());
+ statuses.add(e.getKey() + "=" + status);
+ // fail fast: each further hang would cost another socket
timeout
+ assertNotNull(status, String.join(" ", endpoint) + " " +
e.getKey()
+ + ": no response before the body ended");
+ if (status.startsWith("2")) {
+ failures.add(String.join(" ", endpoint) + " " + e.getKey()
+ " -> " + status);
+ }
+ anyTooLarge |= "413".equals(status);
+ }
+ if (!anyTooLarge) {
+ failures.add(String.join(" ", endpoint) + " never reached the
limit: " + statuses);
+ }
+ }
+ assertTrue(failures.isEmpty(), String.join("\n", failures));
+ }
+
+ @Test
+ public void testEveryEndpointRejectsOverLimitContentLength() throws
Exception {
+ List<String> failures = new ArrayList<>();
+ for (String[] endpoint : bodyEndpoints()) {
+ List<String> statuses = new ArrayList<>();
+ for (Map.Entry<String, byte[]> e : PREAMBLES.entrySet()) {
+ statuses.add(sendDeclared(endpoint, e.getKey(), e.getValue()));
+ }
+ // other types may be refused (415) before the size is looked at
+ if (!statuses.contains("413") || statuses.stream().anyMatch(s -> s
== null || s.startsWith("2"))) {
+ failures.add(String.join(" ", endpoint) + " -> " + statuses);
+ }
+ }
+ assertTrue(failures.isEmpty(), String.join("\n", failures));
+ }
+
+ /** A body dripped in under the size limit is cut off at the read
deadline. */
+ @Test
+ public void testSlowBodyTimesOut() throws Exception {
+ assertWithinReadDeadline("408", sendThenDrip(TIKA_TEXT, 0));
+ }
+
+ /** CXF discards up to 16 MiB left over after a 413; a drip must not
stretch that out. */
+ @Test
+ public void testSlowRemainderAfterOverLimitTimesOut() throws Exception {
+ assertWithinReadDeadline("413", sendThenDrip(TIKA_TEXT, 2 *
MAX_BYTES));
+ }
+
+ /** Refused before any resource runs, the body is still discarded under
the read deadline. */
+ @Test
+ public void testSlowBodyToUnmatchedPathTimesOut() throws Exception {
+ assertWithinReadDeadline("404", sendThenDrip(new String[]{"PUT",
"/nonexistent", "*/*"}, 0));
+ }
+
+ /**
+ * CXF answers OPTIONS itself, before any resource or filter runs; the
body must still be
+ * discarded, not spooled to disk without limit.
+ */
+ @Test
+ public void testOptionsBodyNotSpooled() throws Exception {
+ for (String path : new String[]{"/tika", "/rmeta"}) {
+ String[] reply = sendThenDrip(new String[]{"OPTIONS", path,
"*/*"}, SENT_BYTES);
+ assertEquals("200", reply[0]);
+ // spooling reads on until the read deadline; discarding stops at
16 MiB and answers
+ assertTrue(Long.parseLong(reply[1]) < MAX_READ_MILLIS / 2,
"answered after " + reply[1] + " ms");
+ }
+ }
+
+ private static void assertWithinReadDeadline(String expected, String[]
reply) {
+ assertEquals(expected, reply[0], "after " + reply[1] + " ms");
+ // the idle timeout alone would answer only once the drip ends
+ assertTrue(Long.parseLong(reply[1]) < 2 * MAX_READ_MILLIS, "answered
after " + reply[1] + " ms");
+ }
+
+ /** Bodies held in flight fill the budget; the request that would go over
gets 429. */
+ @Test
+ public void testOverBudgetRefused() throws Exception {
+ int held = MAX_BYTES - 10_000;
+ List<Socket> uploads = new ArrayList<>();
+ try {
+ for (int i = 0; i < 3; i++) {
+ uploads.add(startUpload(held));
+ }
+ awaitBodyBytesInFlight(3L * held);
+ try (Socket socket = new Socket("localhost", integrationTestPort))
{
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(TIKA_TEXT, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ writeChunk(out, new byte[MAX_IN_FLIGHT_BYTES - 3 * held + 1]);
+ out.write("0\r\n\r\n".getBytes(UTF_8));
+ out.flush();
+ assertEquals("429", readStatus(socket));
+ }
+ } finally {
+ for (Socket upload : uploads) {
+ upload.close();
+ }
+ }
+ }
+
+ /** Clients that send slowly hold only what they have sent, so they cannot
lock others out. */
+ @Test
+ public void testDripsDoNotExhaustBudget() throws Exception {
+ List<Socket> uploads = new ArrayList<>();
+ try {
+ for (int i = 0; i < 16; i++) {
+ uploads.add(startUpload(0));
+ }
+ awaitBodyBytesInFlight(16);
+ assertEquals("200", sendSmall());
+ } finally {
+ for (Socket upload : uploads) {
+ upload.close();
+ }
+ }
+ }
+
+ private void awaitBodyBytesInFlight(long atLeast) throws Exception {
+ long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(2);
+ while (bodyBytesInFlight() < atLeast) {
+ assertTrue(System.nanoTime() < deadline, "uploads not in flight: "
+ bodyBytesInFlight());
+ Thread.sleep(20);
+ }
+ }
+
+ private long bodyBytesInFlight() {
+ String scrape = WebClient.create("http://localhost:"; + metricsPort +
"/metrics").get()
+ .readEntity(String.class);
+ for (String line : scrape.split("\n")) {
+ if (line.startsWith("tika_server_request_body_in_flight_bytes ")) {
+ return (long) Double.parseDouble(line.substring(line.indexOf('
') + 1).trim());
+ }
+ }
+ throw new AssertionError("no body budget gauge in:\n" + scrape);
+ }
+
+ private String sendSmall() throws IOException {
+ byte[] body = "hello world".getBytes(UTF_8);
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "text/plain")
+ .append("Content-Length:
").append(body.length).append("\r\n\r\n")
+ .toString().getBytes(UTF_8));
+ out.write(body);
+ out.flush();
+ return readStatus(socket);
+ }
+ }
+
+ /** An upload of {@code headBytes}, then kept in flight a byte at a time
until its socket is closed. */
+ private Socket startUpload(int headBytes) throws IOException {
+ Socket socket = new Socket("localhost", integrationTestPort);
+ OutputStream out = socket.getOutputStream();
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(requestHead(TIKA_TEXT, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ writeChunk(out, new byte[headBytes]);
+ while (!socket.isClosed()) {
+ writeChunk(out, new byte[]{'a'});
+ out.flush();
+ Thread.sleep(100);
+ }
+ } catch (IOException | InterruptedException e) {
+ // closed
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ return socket;
+ }
+
+ /** A client that stops sending is dropped at the idle timeout, with the
same 408. */
+ @Test
+ public void testStalledBodyTimesOut() throws Exception {
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(5 * IDLE_TIMEOUT_MILLIS);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "text/plain")
+ .append("Content-Length:
100\r\n\r\nabc").toString().getBytes(UTF_8));
+ out.flush();
+ assertEquals("408", readStatus(socket));
+ }
+ }
+
+ /**
+ * Sends {@code headBytes}, then a byte at a time, well inside the idle
timeout, for over
+ * three times the read deadline. Returns {status, millis from start to
status}.
+ */
+ private String[] sendThenDrip(String[] endpoint, int headBytes) throws
IOException {
+ byte[] filler = new byte[64 * 1024];
+ Arrays.fill(filler, (byte) 'a');
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(5 * MAX_READ_MILLIS);
+ OutputStream out = socket.getOutputStream();
+ long start = System.nanoTime();
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(requestHead(endpoint, "text/plain")
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ for (int sent = 0; sent < headBytes; sent +=
filler.length) {
+ writeChunk(out, Arrays.copyOf(filler,
Math.min(filler.length, headBytes - sent)));
+ }
+ out.flush();
+ for (int i = 0; i < 100; i++) {
+ writeChunk(out, new byte[]{'a'});
+ out.flush();
+ Thread.sleep(100);
+ }
+ } catch (IOException | InterruptedException e) {
+ // the server stopped reading and closed
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ String status = readStatus(socket);
+ return new String[]{status,
+
Long.toString(TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start))};
+ }
+ }
+
+ private static byte[] bigTextZip() throws IOException {
+ ByteArrayOutputStream zipped = new ByteArrayOutputStream();
+ try (ZipOutputStream zip = new ZipOutputStream(zipped)) {
+ zip.putNextEntry(new ZipEntry("big.txt"));
+ byte[] line = "the quick brown fox jumps over the lazy
dog\n".getBytes(UTF_8);
+ for (int written = 0; written < TEXT_BYTES; written +=
line.length) {
+ zip.write(line);
+ }
+ }
+ assertTrue(zipped.size() < MAX_BYTES);
+ return zipped.toByteArray();
+ }
+
+ /**
+ * A client reading too slowly to finish within the write deadline is cut
off. It keeps
+ * reading, so the server's writes progress to the deadline check; the
response, from a
+ * small zip, is far larger than the socket buffers.
+ */
+ @Test
+ public void testSlowReaderCutOff() throws Exception {
+ byte[] body = bigTextZip();
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(30_000);
+ OutputStream out = socket.getOutputStream();
+ // keep-alive, so the body is chunked: a close-delimited one
cannot show a truncation
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "application/zip")
+ .append("Content-Length:
").append(body.length).append("\r\n\r\n")
+ .toString().replace("Connection: close\r\n",
"").getBytes(UTF_8));
+ out.write(body);
+ out.flush();
+ InputStream in = socket.getInputStream();
+ byte[] buffer = new byte[64 * 1024];
+ long read = 0;
+ String tail = "";
+ // about 1.3 MB/s: the whole body would take over 4 times the
deadline
+ while (true) {
+ int n;
+ try {
+ n = in.read(buffer);
+ } catch (IOException e) {
+ break;
+ }
+ if (n == -1) {
+ break;
+ }
+ read += n;
+ tail = (tail + new String(buffer, 0, n, UTF_8));
+ tail = tail.substring(Math.max(0, tail.length() - 16));
+ Thread.sleep(50);
+ }
+ assertTrue(read < TEXT_BYTES, "the whole response arrived: " +
read);
+ // a cleanly ended chunked body would pass a truncated document
off as complete
+ assertFalse(tail.endsWith("\r\n0\r\n\r\n"), "the cut-off response
ended cleanly");
+ }
+ }
+
+ /**
+ * With allowHttp2, the body limits act per stream: an over-limit body is
refused and the
+ * connection carries on.
+ */
+ @Test
+ public void testH2cLimitsPerStream() throws Exception {
+ HttpClient client =
HttpClient.newBuilder().version(HttpClient.Version.HTTP_2).build();
+ // a request with a body cannot upgrade to h2c, so upgrade first
+ HttpResponse<String> upgrade =
client.send(HttpRequest.newBuilder(URI.create(endPoint + "/version"))
+ .GET().build(), HttpResponse.BodyHandlers.ofString());
+ assertEquals(HttpClient.Version.HTTP_2, upgrade.version());
+ HttpResponse<String> tooLarge =
client.send(HttpRequest.newBuilder(URI.create(endPoint + "/tika/text"))
+ .header("Content-Type", "text/plain")
+ .PUT(HttpRequest.BodyPublishers.ofByteArray(new byte[2 *
MAX_BYTES])).build(),
+ HttpResponse.BodyHandlers.ofString());
+ assertEquals(413, tooLarge.statusCode());
+ assertEquals(HttpClient.Version.HTTP_2, tooLarge.version());
+ HttpResponse<String> small =
client.send(HttpRequest.newBuilder(URI.create(endPoint + "/tika/text"))
+ .header("Content-Type", "text/plain")
+ .PUT(HttpRequest.BodyPublishers.ofString("hello
world")).build(),
+ HttpResponse.BodyHandlers.ofString());
+ assertEquals(200, small.statusCode());
+ assertEquals(HttpClient.Version.HTTP_2, small.version());
+ }
+
+ /** The limit counts decompressed bytes, so a small gzip body cannot
smuggle in a large one. */
+ @Test
+ public void testGzipBodyCountedDecompressed() throws Exception {
+ ByteArrayOutputStream compressed = new ByteArrayOutputStream();
+ try (GZIPOutputStream gz = new GZIPOutputStream(compressed)) {
+ gz.write(new byte[4 * MAX_BYTES]);
+ }
+ byte[] body = compressed.toByteArray();
+ assertTrue(body.length < MAX_BYTES);
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(new String[]{"PUT", "/tika/text",
"text/plain"}, "text/plain")
+ .append("Content-Encoding: gzip\r\nContent-Length:
").append(body.length)
+ .append("\r\n\r\n").toString().getBytes(UTF_8));
+ out.write(body);
+ out.flush();
+ assertEquals("413", readStatus(socket));
+ }
+ }
+
+ /**
+ * {method, path, accept} per PUT/POST endpoint and type it produces, from
the server's
+ * own listing; Accept is what selects among methods sharing a path (XMP
on /meta).
+ */
+ private List<String[]> bodyEndpoints() {
+ Response response = WebClient.create(endPoint +
"/").accept("text/plain").get();
+ String listing = response.readEntity(String.class);
+ List<String[]> endpoints = new ArrayList<>();
+ String[] current = null;
+ boolean producesListed = false;
+ for (String line : listing.split("\n")) {
+ line = line.trim();
+ if (line.startsWith("=> ")) {
+ if (current != null) {
+ endpoints.add(new String[]{current[0], current[1],
line.substring(3)});
+ producesListed = true;
+ }
+ continue;
+ }
+ if (current != null && !producesListed) {
+ endpoints.add(new String[]{current[0], current[1], "*/*"});
+ }
+ current = null;
+ producesListed = false;
+ String[] parts = line.split(" ", 2);
+ if (parts.length == 2 && ("PUT".equals(parts[0]) ||
"POST".equals(parts[0]))) {
+ current = new String[]{parts[0], fillTemplate(parts[1])};
+ }
+ }
+ if (current != null && !producesListed) {
+ endpoints.add(new String[]{current[0], current[1], "*/*"});
+ }
+ assertFalse(endpoints.isEmpty(), "no PUT/POST endpoints in:\n" +
listing);
+ return endpoints;
+ }
+
+ private static String fillTemplate(String path) {
+ // {name} or {name : regex}; regexes may themselves hold braces, e.g.
\w{1,3}
+ StringBuilder sb = new StringBuilder();
+ StringBuilder param = new StringBuilder();
+ int depth = 0;
+ for (char c : path.toCharArray()) {
+ if (c == '{') {
+ depth++;
+ } else if (c == '}') {
+ if (--depth == 0) {
+ String name = param.toString().split(":")[0].trim();
+ sb.append("presetName".equals(name) ? PRESET : "text");
+ param.setLength(0);
+ }
+ } else if (depth > 0) {
+ param.append(c);
+ } else {
+ sb.append(c);
+ }
+ }
+ return sb.toString();
+ }
+
+ /** Status code, "closed" if the connection closed first, or null if
nothing came in time. */
+ private String sendChunked(String[] endpoint, String contentType, byte[]
preamble)
+ throws IOException {
+ byte[] filler = new byte[64 * 1024];
+ Arrays.fill(filler, (byte) 'a');
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ Thread writer = new Thread(() -> {
+ try {
+ out.write(requestHead(endpoint, contentType)
+ .append("Transfer-Encoding:
chunked\r\n\r\n").toString().getBytes(UTF_8));
+ writeChunk(out, preamble);
+ for (int sent = 0; sent < SENT_BYTES; sent +=
filler.length) {
+ writeChunk(out, filler);
+ }
+ out.flush();
+ // the terminating chunk is never sent
+ } catch (IOException e) {
+ // the server stopped reading and closed
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ return readStatus(socket);
+ }
+ }
+
+ private String sendDeclared(String[] endpoint, String contentType, byte[]
preamble)
+ throws IOException {
+ byte[] body = new byte[2 * MAX_BYTES];
+ Arrays.fill(body, (byte) 'a');
+ System.arraycopy(preamble, 0, body, 0, preamble.length);
+ try (Socket socket = new Socket("localhost", integrationTestPort)) {
+ socket.setSoTimeout(10_000);
+ OutputStream out = socket.getOutputStream();
+ out.write(requestHead(endpoint, contentType)
+ .append("Content-Length:
").append(body.length).append("\r\n\r\n")
+ .toString().getBytes(UTF_8));
+ // sent in full: CXF reads past a rejected body before answering
+ out.write(body);
+ out.flush();
+ return readStatus(socket);
+ }
+ }
+
+ private StringBuilder requestHead(String[] endpoint, String contentType) {
+ return new StringBuilder(endpoint[0]).append('
').append(endpoint[1]).append(" HTTP/1.1\r\n")
+ .append("Host: localhost\r\nConnection: close\r\n")
+ .append("Accept: ").append(endpoint[2]).append("\r\n")
+ .append("Content-Type: ").append(contentType).append("\r\n");
+ }
+
+ private static void writeChunk(OutputStream out, byte[] bytes) throws
IOException {
+ if (bytes.length == 0) {
+ return;
+ }
+ out.write((Integer.toHexString(bytes.length) +
"\r\n").getBytes(UTF_8));
+ out.write(bytes);
+ out.write("\r\n".getBytes(UTF_8));
+ }
+
+ private static String readStatus(Socket socket) throws IOException {
+ try {
+ String line = new BufferedReader(
+ new InputStreamReader(socket.getInputStream(),
UTF_8)).readLine();
+ // "HTTP/1.1 413 Payload Too Large"
+ return line == null ? "closed" : line.split(" ")[1];
+ } catch (SocketTimeoutException e) {
+ return null;
+ }
+ }
+}
