This is an automated email from the ASF dual-hosted git repository.
raboof pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-grpc.git
The following commit(s) were added to refs/heads/main by this push:
new 6477f539 apply a max message len (#818)
6477f539 is described below
commit 6477f539792c1a96fe799fc9b3e7abd9990d3a55
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Aug 30 08:38:53 2026 +0100
apply a max message len (#818)
* apply a max message len
Motivation:
gRPC message size limits protect both clients and servers from decompression
bombs and memory exhaustion. grpc-java defaults to 4 MiB but Pekko gRPC had
no
limit, leaving it vulnerable to oversized frames and gzip decompression
attacks.
Modification:
Protocol layer - `GrpcFramingDecoderStage` and the strict decoder validate
the
declared frame length against `maxInboundMessageSize` before the payload is
buffered. `Codec.uncompress` gained overloads that bound the decompressed
size
and throw `RESOURCE_EXHAUSTED` when it is exceeded; the compression-bit
overload
delegates to the size-bounded one so `Gzip` can fail fast while
decompressing
rather than after. `Gzip` checks cumulative output during streaming
decompression using a `Long` accumulator and a clamped initial buffer.
Negative
frame lengths are reported as `INTERNAL` on both the streaming and the
strict
path.
Client side - added `max-inbound-message-size` to `pekko.grpc.client."*"`
(default 4 MiB) and `GrpcClientSettings.withMaxInboundMessageSize`. Both
backends honour it.
Server side - added `pekko.grpc.server.max-inbound-message-size` and a new
`GrpcServerSettings` class. Generated Scala and Java handlers resolve the
setting once per `partial`/`handler` call rather than per request.
Result:
Both clients and servers reject inbound messages exceeding the configured
limit
with `RESOURCE_EXHAUSTED`. Oversized frames are rejected from the declared
length before the payload is buffered, and gzip frames that inflate past the
limit are rejected while decompressing rather than after.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
* Deprecate the unbounded Codec.uncompress overloads
Motivation:
`uncompress(bytes)` and `uncompress(compressedBitSet, bytes)` place no
bound on
the decompressed output, so a caller that reaches for them is exposed to the
decompression bomb this branch otherwise closes. No production call site
uses
them any more - the framing decoder went through the size-bounded overloads
-
but they remain public and are the shape through which a future caller would
reintroduce the problem.
Modification:
Marked both overloads `@deprecated` in `Codec`, `Gzip` and `Identity`,
pointing
at the size-bounded replacements. The default `uncompress(bytes, max)` in
`Codec` and `Gzip`'s compression-bit overload still delegate to them, so
both
carry `@nowarn("cat=deprecation")`; the build runs with `-Xfatal-warnings`.
Documented on the default `uncompress(bytes, max)` that it checks after
decompressing and that codecs able to fail fast should override it, as
`Gzip`
does.
Result:
The unbounded entry points still work for source compatibility but now warn,
and the bounded overloads are the discoverable ones.
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---------
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../twirl/templates/JavaServer/Handler.scala.txt | 21 ++-
.../twirl/templates/ScalaServer/Handler.scala.txt | 14 +-
docs/src/main/paradox/client/configuration.md | 29 +++
docs/src/main/paradox/server/details.md | 57 ++++++
.../max-inbound-message-size.excludes | 31 ++++
runtime/src/main/resources/reference.conf | 18 ++
.../org/apache/pekko/grpc/GrpcClientSettings.scala | 26 ++-
.../scala/org/apache/pekko/grpc/GrpcProtocol.scala | 28 ++-
.../org/apache/pekko/grpc/GrpcServerSettings.scala | 72 +++++++
.../pekko/grpc/internal/AbstractGrpcProtocol.scala | 66 +++++--
.../org/apache/pekko/grpc/internal/Codec.scala | 60 ++++++
.../pekko/grpc/internal/GrpcProtocolNative.scala | 4 +-
.../pekko/grpc/internal/GrpcProtocolWeb.scala | 4 +-
.../org/apache/pekko/grpc/internal/Gzip.scala | 39 ++++
.../org/apache/pekko/grpc/internal/Identity.scala | 21 +++
.../pekko/grpc/internal/NettyClientUtils.scala | 2 +
.../pekko/grpc/internal/PekkoHttpClientUtils.scala | 12 +-
.../pekko/grpc/javadsl/GrpcMarshalling.scala | 21 +++
.../pekko/grpc/scaladsl/GrpcMarshalling.scala | 18 ++
.../apache/pekko/grpc/GrpcClientSettingsSpec.scala | 43 +++++
.../apache/pekko/grpc/GrpcServerSettingsSpec.scala | 64 +++++++
.../grpc/internal/MaxInboundMessageSizeSpec.scala | 206 +++++++++++++++++++++
22 files changed, 825 insertions(+), 31 deletions(-)
diff --git a/codegen/src/main/twirl/templates/JavaServer/Handler.scala.txt
b/codegen/src/main/twirl/templates/JavaServer/Handler.scala.txt
index d8796868..18194324 100644
--- a/codegen/src/main/twirl/templates/JavaServer/Handler.scala.txt
+++ b/codegen/src/main/twirl/templates/JavaServer/Handler.scala.txt
@@ -27,6 +27,7 @@ import org.apache.pekko.annotation.ApiMayChange;
import org.apache.pekko.stream.Materializer;
import org.apache.pekko.stream.SystemMaterializer;
+import org.apache.pekko.grpc.GrpcServerSettings;
import org.apache.pekko.grpc.Trailers;
import org.apache.pekko.grpc.javadsl.GrpcMarshalling;
import org.apache.pekko.grpc.javadsl.GrpcExceptionHandler;
@@ -143,14 +144,28 @@ public class @{serviceName}HandlerFactory {
* Use {@@link
org.apache.pekko.grpc.javadsl.ServiceHandler#concatOrNotFound} when combining
several services.
*/
public static Function<org.apache.pekko.http.javadsl.model.HttpRequest,
CompletionStage<org.apache.pekko.http.javadsl.model.HttpResponse>>
partial(@serviceName implementation, String prefix, Materializer mat,
org.apache.pekko.japi.function.Function<ActorSystem,
org.apache.pekko.japi.function.Function<Throwable, Trailers>> eHandler,
ClassicActorSystemProvider system) {
+ return partial(implementation, prefix, mat, eHandler,
GrpcServerSettings.create(system), system);
+ }
+
+ /**
+ * Creates a `HttpRequest` to `HttpResponse` handler that can be used in
for example
+ * `Http.get(system).bindAndHandleAsync`. It ends with
`StatusCodes.NotFound` if the request is not matching.
+ *
+ * Use {@@link
org.apache.pekko.grpc.javadsl.ServiceHandler#concatOrNotFound} when combining
several services.
+ *
+ * @@param settings server settings, including the maximum allowed inbound
message size
+ */
+ public static Function<org.apache.pekko.http.javadsl.model.HttpRequest,
CompletionStage<org.apache.pekko.http.javadsl.model.HttpResponse>>
partial(@serviceName implementation, String prefix, Materializer mat,
org.apache.pekko.japi.function.Function<ActorSystem,
org.apache.pekko.japi.function.Function<Throwable, Trailers>> eHandler,
GrpcServerSettings settings, ClassicActorSystemProvider system) {
TelemetrySpi spi = TelemetryExtension.get(system).spi();
+ // resolved once, outside the returned function, so config is not
re-read per request
+ int maxInboundMessageSize = settings.maxInboundMessageSize();
return (req -> {
Iterator<String> segments = req.getUri().pathSegments().iterator();
if (segments.hasNext() && segments.next().equals(prefix) &&
segments.hasNext()) {
String method = segments.next();
if (segments.hasNext()) return notFound; // we don't allow any
random `/prefix/Method/anything/here
else {
- return handle(spi.onRequest(prefix, method, req), method,
implementation, mat, eHandler, system);
+ return handle(spi.onRequest(prefix, method, req), method,
implementation, mat, eHandler, maxInboundMessageSize, system);
}
} else {
return notFound;
@@ -162,8 +177,8 @@ public class @{serviceName}HandlerFactory {
return @{service.name}.name;
}
- private static
CompletionStage<org.apache.pekko.http.javadsl.model.HttpResponse>
handle(org.apache.pekko.http.javadsl.model.HttpRequest request, String method,
@serviceName implementation, Materializer mat,
org.apache.pekko.japi.function.Function<ActorSystem,
org.apache.pekko.japi.function.Function<Throwable, Trailers>> eHandler,
ClassicActorSystemProvider system) {
- return GrpcMarshalling.negotiated(request, (reader, writer) -> {
+ private static
CompletionStage<org.apache.pekko.http.javadsl.model.HttpResponse>
handle(org.apache.pekko.http.javadsl.model.HttpRequest request, String method,
@serviceName implementation, Materializer mat,
org.apache.pekko.japi.function.Function<ActorSystem,
org.apache.pekko.japi.function.Function<Throwable, Trailers>> eHandler, int
maxInboundMessageSize, ClassicActorSystemProvider system) {
+ return GrpcMarshalling.negotiatedWithMaxSize(request,
maxInboundMessageSize, (reader, writer) -> {
CompletionStage<org.apache.pekko.http.javadsl.model.HttpResponse>
response;
@{if(powerApis) { "Metadata metadata =
MetadataBuilder.fromHeaders(request.getHeaders());" } else { "" }}
switch(method) {
diff --git a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
index d7c2c183..130c663a 100644
--- a/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
+++ b/codegen/src/main/twirl/templates/ScalaServer/Handler.scala.txt
@@ -20,7 +20,7 @@ import scala.concurrent.ExecutionContext
import org.apache.pekko
import pekko.grpc.scaladsl.{ GrpcExceptionHandler, GrpcMarshalling }
-import pekko.grpc.Trailers
+import pekko.grpc.{ GrpcServerSettings, Trailers }
import pekko.actor.ActorSystem
import pekko.actor.ClassicActorSystemProvider
@@ -115,7 +115,9 @@ object @{serviceName}Handler {
null
}
- private def handler(implementation: @serviceName, prefix: String,
eHandler: ActorSystem => PartialFunction[Throwable,
Trailers])(@{service.scalaCompatConstants.ImplicitParameter} system:
ClassicActorSystemProvider): model.HttpRequest =>
scala.concurrent.Future[model.HttpResponse] = {
+ private def handler(implementation: @serviceName, prefix: String,
eHandler: ActorSystem => PartialFunction[Throwable, Trailers], settings:
Option[GrpcServerSettings] =
None)(@{service.scalaCompatConstants.ImplicitParameter} system:
ClassicActorSystemProvider): model.HttpRequest =>
scala.concurrent.Future[model.HttpResponse] = {
+ // resolved once, outside the returned function, so config is not
re-read per request
+ val maxInboundMessageSize =
settings.getOrElse(GrpcServerSettings.create(system)).maxInboundMessageSize
@{service.scalaCompatConstants.ImplicitVal} mat: Materializer =
SystemMaterializer(system).materializer
@{service.scalaCompatConstants.ImplicitVal} ec: ExecutionContext =
mat.executionContext
val spi = TelemetryExtension(system).spi
@@ -123,7 +125,7 @@ object @{serviceName}Handler {
import
@{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
def handle(request: model.HttpRequest, method: String):
scala.concurrent.Future[model.HttpResponse] =
- GrpcMarshalling.negotiated(request, (reader, writer) =>
+ GrpcMarshalling.negotiatedWithMaxSize(request, maxInboundMessageSize,
(reader, writer) =>
method match {
@for(method <- service.methods) {
case "@method.grpcName" =>
@@ -158,7 +160,9 @@ object @{serviceName}Handler {
*
* Registering a gRPC service under a custom prefix is not widely
supported and strongly discouraged by the specification.
*/
- def partial(implementation: @serviceName, prefix: String =
@{service.name}.name, eHandler: ActorSystem => PartialFunction[Throwable,
Trailers] =
GrpcExceptionHandler.defaultMapper)(@{service.scalaCompatConstants.ImplicitParameter}
system: ClassicActorSystemProvider): PartialFunction[model.HttpRequest,
scala.concurrent.Future[model.HttpResponse]] = {
+ def partial(implementation: @serviceName, prefix: String =
@{service.name}.name, eHandler: ActorSystem => PartialFunction[Throwable,
Trailers] = GrpcExceptionHandler.defaultMapper, settings:
Option[GrpcServerSettings] =
None)(@{service.scalaCompatConstants.ImplicitParameter} system:
ClassicActorSystemProvider): PartialFunction[model.HttpRequest,
scala.concurrent.Future[model.HttpResponse]] = {
+ // resolved once, outside the returned partial function, so config is
not re-read per request
+ val maxInboundMessageSize =
settings.getOrElse(GrpcServerSettings.create(system)).maxInboundMessageSize
@{service.scalaCompatConstants.ImplicitVal} mat: Materializer =
SystemMaterializer(system).materializer
@{service.scalaCompatConstants.ImplicitVal} ec: ExecutionContext =
mat.executionContext
val spi = TelemetryExtension(system).spi
@@ -166,7 +170,7 @@ object @{serviceName}Handler {
import
@{service.name}.Serializers.@{service.scalaCompatConstants.WildcardImport}
def handle(request: model.HttpRequest, method: String):
scala.concurrent.Future[model.HttpResponse] =
- GrpcMarshalling.negotiated(request, (reader, writer) =>
+ GrpcMarshalling.negotiatedWithMaxSize(request, maxInboundMessageSize,
(reader, writer) =>
method match {
@for(method <- service.methods) {
case "@method.grpcName" =>
diff --git a/docs/src/main/paradox/client/configuration.md
b/docs/src/main/paradox/client/configuration.md
index bbe48862..6a0d7f75 100644
--- a/docs/src/main/paradox/client/configuration.md
+++ b/docs/src/main/paradox/client/configuration.md
@@ -82,6 +82,35 @@ effect and is logged as a warning.
@@@
+## Maximum inbound message size
+
+Inbound gRPC messages are limited to 4 MiB by default, matching the grpc-java
default; larger messages
+are rejected with `RESOURCE_EXHAUSTED`. The limit applies to both the `netty`
and the `pekko-http`
+backend, and covers the decompressed size, so a compressed response that
inflates past the limit is
+rejected while it is being decompressed.
+
+Set it in configuration:
+
+```hocon
+pekko.grpc.client."*" {
+ max-inbound-message-size = 8388608 # 8 MiB
+}
+```
+
+or programmatically:
+
+Scala
+: ```scala
+ val settings = GrpcClientSettings.connectToServiceAt("localhost", 8080)
+ .withMaxInboundMessageSize(8 * 1024 * 1024)
+ ```
+
+Java
+: ```java
+ GrpcClientSettings settings =
GrpcClientSettings.connectToServiceAt("localhost", 8080, system)
+ .withMaxInboundMessageSize(8 * 1024 * 1024);
+ ```
+
## Using Pekko Discovery for Endpoint Discovery
The examples above all use a hard coded host and port for the location of the
gRPC service which is the default if you do not configure a
`service-discovery-mechanism`.
diff --git a/docs/src/main/paradox/server/details.md
b/docs/src/main/paradox/server/details.md
index 839a185a..85cbee9b 100644
--- a/docs/src/main/paradox/server/details.md
+++ b/docs/src/main/paradox/server/details.md
@@ -55,3 +55,60 @@ Java
:
@@snip[RichErrorModelTest](/interop-tests/src/test/java/example/myapp/helloworld/grpc/RichErrorNativeImpl.java)
{ #rich_error_model_unary }
Please look @ref[here](../client/details.md) how to handle this on the client.
+
+## Maximum inbound message size
+
+Inbound gRPC messages are limited to 4 MiB by default, matching the grpc-java
default. A frame whose
+declared length exceeds the limit is rejected before its payload is read, and
a compressed frame that
+inflates past the limit is rejected while it is being decompressed, so neither
an oversized frame nor a
+decompression bomb needs to be buffered in full. In both cases the peer sees a
`RESOURCE_EXHAUSTED` status.
+
+The limit is read from `pekko.grpc.server`:
+
+`reference.conf`
+: @@snip [reference](/runtime/src/main/resources/reference.conf) {
#server-defaults }
+
+To raise it for every service in the actor system, override the setting in
your `application.conf`:
+
+```hocon
+pekko.grpc.server {
+ max-inbound-message-size = 8388608 # 8 MiB
+}
+```
+
+To use a different limit for a single service, pass
@apidoc[GrpcServerSettings] to the generated
+handler's `partial` method:
+
+Scala
+: ```scala
+ val settings = GrpcServerSettings(system).withMaxInboundMessageSize(8 *
1024 * 1024)
+ val handler = GreeterServiceHandler.partial(
+ new GreeterServiceImpl(),
+ settings = Some(settings))
+ ```
+
+Java
+: ```java
+ GrpcServerSettings settings =
+ GrpcServerSettings.create(system).withMaxInboundMessageSize(8 * 1024 *
1024);
+ Function<HttpRequest, CompletionStage<HttpResponse>> handler =
+ GreeterServiceHandlerFactory.partial(
+ new GreeterServiceImpl(),
+ GreeterService.name,
+ SystemMaterializer.get(system).materializer(),
+ GrpcExceptionHandler.defaultMapper(),
+ settings,
+ system);
+ ```
+
+@@@ note
+
+The limit did not exist before 2.0.0, so a server that previously accepted
messages larger than 4 MiB
+will start rejecting them after upgrading. Raise
`pekko.grpc.server.max-inbound-message-size` to keep
+the old behaviour.
+
+Handlers **generated before 2.0.0** call an entry point that has no access to
the actor system's
+configuration, so they fall back on the 4 MiB default and
`pekko.grpc.server.max-inbound-message-size`
+has no effect on them. Regenerate your sources against 2.0.0 to make the
setting apply.
+
+@@@
diff --git
a/runtime/src/main/mima-filters/2.0.x.backwards.excludes/max-inbound-message-size.excludes
b/runtime/src/main/mima-filters/2.0.x.backwards.excludes/max-inbound-message-size.excludes
new file mode 100644
index 00000000..689859db
--- /dev/null
+++
b/runtime/src/main/mima-filters/2.0.x.backwards.excludes/max-inbound-message-size.excludes
@@ -0,0 +1,31 @@
+# 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.
+
+# add max-inbound-message-size
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol.newReader")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.grpc.GrpcProtocol.newReader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol.newReader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol.reader")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol.reader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol.reader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.AbstractGrpcProtocol#GrpcFramingDecoderStage.this")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.GrpcProtocolNative.newReader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.GrpcProtocolNative.reader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.GrpcProtocolWeb.newReader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.GrpcProtocolWebBase.reader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.GrpcProtocolWebText.newReader")
+ProblemFilters.exclude[DirectMissingMethodProblem]("org.apache.pekko.grpc.internal.PekkoHttpClientUtils.responseToSource")
diff --git a/runtime/src/main/resources/reference.conf
b/runtime/src/main/resources/reference.conf
index 8aaecc39..2f8c1a36 100644
--- a/runtime/src/main/resources/reference.conf
+++ b/runtime/src/main/resources/reference.conf
@@ -46,6 +46,13 @@ pekko.grpc.client."*" {
# TODO: Enforce HTTP/2 TLS restrictions:
https://tools.ietf.org/html/draft-ietf-httpbis-http2-17#section-9.2
+ # Maximum allowed size for inbound gRPC messages (in bytes).
+ # Applies to the decompressed message size. Messages exceeding this limit
+ # will be rejected with RESOURCE_EXHAUSTED status.
+ # Applies to both the netty and the pekko-http backend.
+ # Default is 4 MiB, matching grpc-java's default.
+ max-inbound-message-size = 4194304
+
# The number of times to try connecting before giving up.
# '-1': means retry indefinitely, '0' is invalid, '1' means fail
# after the first failed attempt.
@@ -59,4 +66,15 @@ pekko.grpc.client."*" {
# Any of the mechanisms described in
https://pekko.apache.org/docs/pekko-management/current/discovery/index.html can
be used
# including Kubernetes, Consul, AWS API
}
+
//#defaults
+
+//#server-defaults
+pekko.grpc.server {
+ # Maximum allowed size for inbound gRPC messages (in bytes).
+ # Applies to the decompressed message size. Messages exceeding this limit
+ # will be rejected with RESOURCE_EXHAUSTED status.
+ # Default is 4 MiB, matching grpc-java's default.
+ max-inbound-message-size = 4194304
+}
+//#server-defaults
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcClientSettings.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcClientSettings.scala
index 24bd84c3..2aaef4c1 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcClientSettings.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcClientSettings.scala
@@ -18,7 +18,7 @@ import pekko.actor.ClassicActorSystemProvider
import pekko.annotation.{ ApiMayChange, InternalApi }
import pekko.discovery.{ Discovery, ServiceDiscovery }
import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget }
-import pekko.grpc.internal.HardcodedServiceDiscovery
+import pekko.grpc.internal.{ AbstractGrpcProtocol, HardcodedServiceDiscovery }
import pekko.util.Helpers
import com.typesafe.config.{ Config, ConfigValueFactory }
import io.grpc.CallCredentials
@@ -159,8 +159,17 @@ object GrpcClientSettings {
clientConfiguration.getBoolean("use-tls"),
getOptionalString(clientConfiguration, "load-balancing-policy"),
clientConfiguration.getString("backend"),
+ maxInboundMessageSize = getIntWithDefault(
+ clientConfiguration,
+ "max-inbound-message-size",
+ AbstractGrpcProtocol.DefaultMaxInboundMessageSize),
verifyHostname = clientConfiguration.getBoolean("verify-hostname"))
+ // `fromConfig(Config)` is public API and is called with hand-assembled
Configs that do not
+ // necessarily fall back on `pekko.grpc.client."*"`, so a missing key must
not fail.
+ private def getIntWithDefault(config: Config, path: String, default: Int):
Int =
+ if (config.hasPath(path)) config.getInt(path) else default
+
private def getOptionalString(config: Config, path: String): Option[String] =
config.getString(path) match {
case "" => None
@@ -208,6 +217,7 @@ final class GrpcClientSettings private (
val loadBalancingPolicy: Option[String],
val backend: String,
val channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder =
identity,
+ val maxInboundMessageSize: Int,
val verifyHostname: Boolean) {
require(
sslContext.isEmpty || trustManager.isEmpty,
@@ -216,6 +226,9 @@ final class GrpcClientSettings private (
if (sslContext.isDefined) sslProvider.forall(_ == SslProvider.JDK) else
true,
"When sslContext is configured, sslProvider must not set to something
different than JDK")
require(backend == "netty" || backend == "pekko-http", "backend should be
'netty' or 'pekko-http'")
+ require(
+ maxInboundMessageSize > 0,
+ s"maxInboundMessageSize must be positive, was [$maxInboundMessageSize]")
/**
* If using ServiceDiscovery and no port is returned use this one.
@@ -291,6 +304,15 @@ final class GrpcClientSettings private (
def withBackend(value: String): GrpcClientSettings =
copy(backend = value)
+ /**
+ * Maximum allowed size for inbound gRPC messages (in bytes).
+ * Applies to the decompressed message size. Messages exceeding this limit
+ * will be rejected with RESOURCE_EXHAUSTED status.
+ * @since 2.0.0
+ */
+ def withMaxInboundMessageSize(value: Int): GrpcClientSettings =
+ copy(maxInboundMessageSize = value)
+
/**
* Whether to verify the server's hostname against its TLS certificate (RFC
2818).
* When false, the client accepts any valid certificate regardless of
hostname.
@@ -319,6 +341,7 @@ final class GrpcClientSettings private (
loadBalancingPolicy: Option[String] = loadBalancingPolicy,
backend: String = backend,
channelBuilderOverrides: NettyChannelBuilder => NettyChannelBuilder =
channelBuilderOverrides,
+ maxInboundMessageSize: Int = maxInboundMessageSize,
verifyHostname: Boolean = verifyHostname)
: GrpcClientSettings =
new GrpcClientSettings(
@@ -340,5 +363,6 @@ final class GrpcClientSettings private (
loadBalancingPolicy = loadBalancingPolicy,
backend = backend,
channelBuilderOverrides = channelBuilderOverrides,
+ maxInboundMessageSize = maxInboundMessageSize,
verifyHostname = verifyHostname)
}
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
index 1d8e2f04..948f2779 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcProtocol.scala
@@ -18,7 +18,14 @@ import pekko.NotUsed
import pekko.annotation.InternalApi
import pekko.annotation.InternalStableApi
import pekko.grpc.GrpcProtocol.{ GrpcProtocolReader, GrpcProtocolWriter }
-import pekko.grpc.internal.{ Codec, Codecs, GrpcProtocolNative,
GrpcProtocolWeb, GrpcProtocolWebText }
+import pekko.grpc.internal.{
+ AbstractGrpcProtocol,
+ Codec,
+ Codecs,
+ GrpcProtocolNative,
+ GrpcProtocolWeb,
+ GrpcProtocolWebText
+}
import pekko.http.javadsl.{ model => jmodel }
import pekko.http.scaladsl.model.{ ContentType, HttpHeader, HttpResponse,
Trailer }
import pekko.http.scaladsl.model.HttpEntity.ChunkStreamPart
@@ -63,9 +70,11 @@ trait GrpcProtocol {
*
* Constructs a protocol reader for reading gRPC protocol frames for this
variant.
* @param codec the compression codec to decode data frame bodies with.
+ * @param maxInboundMessageSize the maximum allowed inbound message size in
bytes.
*/
@InternalStableApi
- def newReader(codec: Codec): GrpcProtocolReader
+ def newReader(codec: Codec, maxInboundMessageSize: Int =
AbstractGrpcProtocol.DefaultMaxInboundMessageSize)
+ : GrpcProtocolReader
}
/**
@@ -148,8 +157,21 @@ object GrpcProtocol {
* @return the protocol reader for the request, and a protocol writer for
the response.
*/
def negotiate(request: jmodel.HttpRequest): Option[(Try[GrpcProtocolReader],
GrpcProtocolWriter)] =
+ negotiate(request, AbstractGrpcProtocol.DefaultMaxInboundMessageSize)
+
+ /**
+ * Calculates the gRPC protocol encoding to use for an interaction with a
gRPC client.
+ *
+ * @param request the client request to respond to.
+ * @param maxInboundMessageSize the maximum allowed inbound message size in
bytes.
+ * @return the protocol reader for the request, and a protocol writer for
the response.
+ */
+ def negotiate(
+ request: jmodel.HttpRequest,
+ maxInboundMessageSize: Int): Option[(Try[GrpcProtocolReader],
GrpcProtocolWriter)] =
detect(request).map { variant =>
- (Codecs.detect(request).map(variant.newReader),
variant.newWriter(Codecs.negotiate(request)))
+ (Codecs.detect(request).map(variant.newReader(_, maxInboundMessageSize)),
+ variant.newWriter(Codecs.negotiate(request)))
}
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/GrpcServerSettings.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcServerSettings.scala
new file mode 100644
index 00000000..d742745c
--- /dev/null
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/GrpcServerSettings.scala
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * license agreements; and to You under the Apache License, version 2.0:
+ *
+ * https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * This file is part of the Apache Pekko project, which was derived from Akka.
+ */
+
+package org.apache.pekko.grpc
+
+import org.apache.pekko
+import pekko.actor.ClassicActorSystemProvider
+import pekko.annotation.ApiMayChange
+import pekko.grpc.internal.AbstractGrpcProtocol
+import com.typesafe.config.Config
+
+object GrpcServerSettings {
+
+ /**
+ * Scala API: Create settings from the actor system's default configuration
(`pekko.grpc.server`).
+ */
+ def apply(implicit actorSystem: ClassicActorSystemProvider):
GrpcServerSettings =
+
fromConfig(actorSystem.classicSystem.settings.config.getConfig("pekko.grpc.server"))
+
+ /**
+ * Java API: Create settings from the actor system's default configuration
(`pekko.grpc.server`).
+ */
+ def create(actorSystem: ClassicActorSystemProvider): GrpcServerSettings =
+ apply(actorSystem)
+
+ /**
+ * Create settings from a custom Config (must contain the same keys as
`pekko.grpc.server`).
+ *
+ * Keys that are absent fall back on the built-in defaults, so a
hand-assembled `Config` that
+ * does not resolve against `reference.conf` keeps working.
+ */
+ def fromConfig(config: Config): GrpcServerSettings =
+ new GrpcServerSettings(
+ maxInboundMessageSize =
+ if (config.hasPath("max-inbound-message-size"))
config.getInt("max-inbound-message-size")
+ else AbstractGrpcProtocol.DefaultMaxInboundMessageSize)
+}
+
+/**
+ * Settings for gRPC server services.
+ *
+ * Read from `pekko.grpc.server` in the actor system's configuration.
+ *
+ * @since 2.0.0
+ */
+@ApiMayChange
+final class GrpcServerSettings private (
+ val maxInboundMessageSize: Int) {
+
+ require(
+ maxInboundMessageSize > 0,
+ s"maxInboundMessageSize must be positive, was [$maxInboundMessageSize]")
+
+ /**
+ * Maximum allowed size for inbound gRPC messages (in bytes).
+ * Applies to the decompressed message size. Messages exceeding this limit
+ * will be rejected with RESOURCE_EXHAUSTED status.
+ * @since 2.0.0
+ */
+ def withMaxInboundMessageSize(value: Int): GrpcServerSettings =
+ new GrpcServerSettings(
+ maxInboundMessageSize = value)
+
+ override def toString: String =
+ s"GrpcServerSettings(maxInboundMessageSize=$maxInboundMessageSize)"
+}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
index 3067c894..daace3f6 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/AbstractGrpcProtocol.scala
@@ -25,7 +25,7 @@ import pekko.stream.impl.io.ByteStringParser.{ ByteReader,
ParseResult, ParseSte
import pekko.stream.scaladsl.Flow
import pekko.stream.stage.GraphStageLogic
import pekko.util.ByteString
-import io.grpc.StatusException
+import io.grpc.{ Status, StatusException }
import scala.collection.immutable
@@ -38,7 +38,17 @@ abstract class AbstractGrpcProtocol(subType: String) extends
GrpcProtocol {
Set(contentType.mediaType, MediaType.applicationBinary(subType,
MediaType.Compressible))
private lazy val knownWriters = Codecs.supportedCodecs.map(c => c ->
writer(c)).toMap.withDefault(writer)
- private lazy val knownReaders = Codecs.supportedCodecs.map(c => c ->
reader(c)).toMap.withDefault(reader)
+
+ /**
+ * Readers for the default message size limit. `newReader` is called once
per inbound request, so
+ * the common case (no per-service override) is served from this cache
rather than rebuilding the
+ * reader, its framing stage and the surrounding flows every time.
+ */
+ private lazy val knownReaders =
+ Codecs.supportedCodecs
+ .map(c => c -> reader(c,
AbstractGrpcProtocol.DefaultMaxInboundMessageSize))
+ .toMap
+ .withDefault(reader(_,
AbstractGrpcProtocol.DefaultMaxInboundMessageSize))
/**
* Obtains a writer for this protocol:
@@ -50,12 +60,17 @@ abstract class AbstractGrpcProtocol(subType: String)
extends GrpcProtocol {
* Obtains a reader for this protocol.
*
* @param codec the codec to use for compressed frames.
+ * @param maxInboundMessageSize the maximum allowed inbound message size in
bytes.
*/
- override def newReader(codec: Codec): GrpcProtocolReader =
knownReaders(codec)
+ override def newReader(
+ codec: Codec,
+ maxInboundMessageSize: Int =
AbstractGrpcProtocol.DefaultMaxInboundMessageSize): GrpcProtocolReader =
+ if (maxInboundMessageSize ==
AbstractGrpcProtocol.DefaultMaxInboundMessageSize) knownReaders(codec)
+ else reader(codec, maxInboundMessageSize)
protected def writer(codec: Codec): GrpcProtocolWriter
- protected def reader(codec: Codec): GrpcProtocolReader
+ protected def reader(codec: Codec, maxInboundMessageSize: Int):
GrpcProtocolReader
}
object AbstractGrpcProtocol {
@@ -118,11 +133,17 @@ object AbstractGrpcProtocol {
encodeDataToResponse,
Flow[Frame].map(encodeFrame))
+ /**
+ * The default maximum inbound message size (4 MiB), matching grpc-java's
default.
+ */
+ val DefaultMaxInboundMessageSize: Int = 4 * 1024 * 1024
+
def reader(
codec: Codec,
decodeFrame: (Int, ByteString) => Frame,
preDecodeStrict: ByteString => ByteString = null,
- preDecodeFlow: Flow[ByteString, ByteString, NotUsed] = null):
GrpcProtocolReader = {
+ preDecodeFlow: Flow[ByteString, ByteString, NotUsed] = null,
+ maxInboundMessageSize: Int = DefaultMaxInboundMessageSize):
GrpcProtocolReader = {
val strictAdapter: ByteString => ByteString = if (preDecodeStrict eq null)
identity else preDecodeStrict
val adapter: Flow[ByteString, Frame, NotUsed] => Flow[ByteString, Frame,
NotUsed] =
if (preDecodeFlow eq null) identity
@@ -134,30 +155,51 @@ object AbstractGrpcProtocol {
val reader = new ByteReader(strictAdapter(bs))
val frameType = reader.readByte()
val length = reader.readIntBE()
- if (length < 0) throw new IllegalStateException(s"Frame length must
not be negative, was $length")
+ if (length < 0)
+ throw new StatusException(Status.INTERNAL.withDescription(s"Frame
length must not be negative, was $length"))
+ if (length > maxInboundMessageSize)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Frame length $length exceeds maximum inbound message size
$maxInboundMessageSize"))
val data = reader.take(length)
if (reader.hasRemaining) throw new IllegalStateException("Unexpected
data")
- if ((frameType & 0x80) == 0) codec.uncompress((frameType & 1) == 1,
data)
+ if ((frameType & 0x80) == 0) codec.uncompress((frameType & 1) == 1,
data, maxInboundMessageSize)
else throw new IllegalStateException("Cannot read unknown frame")
} catch { case ByteStringParser.NeedMoreData => throw new
MissingParameterException }
- GrpcProtocolReader(codec, decoder, adapter(Flow.fromGraph(new
GrpcFramingDecoderStage(codec, decodeFrame))))
+ GrpcProtocolReader(
+ codec,
+ decoder,
+ adapter(Flow.fromGraph(new GrpcFramingDecoderStage(codec, decodeFrame,
maxInboundMessageSize))))
}
- class GrpcFramingDecoderStage(codec: Codec, deframe: (Int, ByteString) =>
Frame) extends ByteStringParser[Frame] {
+ class GrpcFramingDecoderStage(codec: Codec, deframe: (Int, ByteString) =>
Frame, maxInboundMessageSize: Int)
+ extends ByteStringParser[Frame] {
override def createLogic(inheritedAttributes: Attributes): GraphStageLogic
=
new ParsingLogic {
startWith(ReadFrameHeader)
trait Step extends ParseStep[Frame]
+ // handle explicitly to avoid noisy log: a peer sending oversized or
malformed frame
+ // headers should not cost us a stack trace per attempt
+ private def failWith(status: Status): ParseResult[Frame] = {
+ failStage(new StatusException(status))
+ ParseResult(None, Failed)
+ }
+
object ReadFrameHeader extends Step {
override def parse(reader: ByteReader): ParseResult[Frame] = {
val frameType = reader.readByte()
val length = reader.readIntBE()
- if (length < 0) throw new IllegalStateException(s"Frame length
must not be negative, was $length")
- if (length == 0) ParseResult(Some(deframe(frameType,
ByteString.empty)), ReadFrameHeader)
+ if (length < 0)
+ failWith(Status.INTERNAL.withDescription(s"Frame length must not
be negative, was $length"))
+ else if (length > maxInboundMessageSize)
+ failWith(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Frame length $length exceeds maximum inbound message size
$maxInboundMessageSize"))
+ else if (length == 0) ParseResult(Some(deframe(frameType,
ByteString.empty)), ReadFrameHeader)
else ParseResult(None, ReadFrame(frameType, length),
acceptUpstreamFinish = false)
}
}
@@ -167,7 +209,7 @@ object AbstractGrpcProtocol {
override def parse(reader: ByteReader): ParseResult[Frame] =
try ParseResult(
- Some(deframe(frameType, codec.uncompress(compression,
reader.take(length)))),
+ Some(deframe(frameType, codec.uncompress(compression,
reader.take(length), maxInboundMessageSize))),
ReadFrameHeader)
catch {
case s: StatusException =>
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
index b01d2403..90c88f59 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Codec.scala
@@ -14,18 +14,78 @@
package org.apache.pekko.grpc.internal
import org.apache.pekko.util.ByteString
+import io.grpc.{ Status, StatusException }
+
+import scala.annotation.nowarn
abstract class Codec {
val name: String
def compress(bytes: ByteString): ByteString
+
+ /**
+ * Decompress the given bytes with no bound on the output size.
+ *
+ * A compressed frame can inflate to arbitrarily many bytes, so a caller
that does not
+ * impose a limit is exposed to a decompression bomb. Prefer
+ * `uncompress(bytes, maxDecompressedSize)`, which fails with
`RESOURCE_EXHAUSTED`
+ * instead of allocating without bound.
+ */
+ @deprecated("Use uncompress(bytes, maxDecompressedSize), which bounds the
decompressed size", "2.0.0")
def uncompress(bytes: ByteString): ByteString
/**
* Process the given frame bytes, uncompress if the compression bit is set.
Identity
* codec will fail with a `io.grpc.StatusException` if the compressedBit is
set.
+ *
+ * Places no bound on the decompressed size; prefer
+ * `uncompress(compressedBitSet, bytes, maxDecompressedSize)`.
*/
+ @deprecated(
+ "Use uncompress(compressedBitSet, bytes, maxDecompressedSize), which
bounds the decompressed size",
+ "2.0.0")
def uncompress(compressedBitSet: Boolean, bytes: ByteString): ByteString
+ /**
+ * Decompress the given bytes, enforcing a maximum decompressed size.
+ * Throws a `StatusException` with `RESOURCE_EXHAUSTED` if the decompressed
+ * output exceeds `maxDecompressedSize`.
+ *
+ * This default implementation decompresses in full and checks afterwards,
so it bounds
+ * what a caller receives but not what is allocated along the way. Codecs
that can enforce
+ * the limit while decompressing should override it; `Gzip` does.
+ *
+ * @param bytes the compressed bytes
+ * @param maxDecompressedSize the maximum allowed decompressed size in bytes
+ */
+ @nowarn("cat=deprecation")
+ def uncompress(bytes: ByteString, maxDecompressedSize: Int): ByteString = {
+ val result = uncompress(bytes)
+ if (result.length > maxDecompressedSize)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Decompressed message size ${result.length} exceeds maximum allowed
$maxDecompressedSize"))
+ result
+ }
+
+ /**
+ * Process the given frame bytes, uncompress if the compression bit is set,
+ * enforcing a maximum decompressed size.
+ *
+ * Delegates to `uncompress(bytes, maxDecompressedSize)` so that codecs
which can enforce
+ * the limit while decompressing (rather than after) get the chance to fail
fast.
+ *
+ * @param compressedBitSet whether the compression bit is set
+ * @param bytes the frame bytes
+ * @param maxDecompressedSize the maximum allowed decompressed size in bytes
+ */
+ def uncompress(compressedBitSet: Boolean, bytes: ByteString,
maxDecompressedSize: Int): ByteString =
+ if (compressedBitSet) uncompress(bytes, maxDecompressedSize)
+ else if (bytes.length > maxDecompressedSize)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Message size ${bytes.length} exceeds maximum allowed
$maxDecompressedSize bytes"))
+ else bytes
+
def isCompressed: Boolean = this != Identity
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
index 4338f937..e7745703 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolNative.scala
@@ -43,8 +43,8 @@ object GrpcProtocolNative extends
AbstractGrpcProtocol("grpc") {
override protected def writer(codec: Codec) =
AbstractGrpcProtocol.writer(this, codec, encodeFrame(codec, _),
encodeDataToResponse(codec))
- override protected def reader(codec: Codec): GrpcProtocolReader =
- AbstractGrpcProtocol.reader(codec, decodeFrame)
+ override protected def reader(codec: Codec, maxInboundMessageSize: Int):
GrpcProtocolReader =
+ AbstractGrpcProtocol.reader(codec, decodeFrame, maxInboundMessageSize =
maxInboundMessageSize)
@inline
private def decodeFrame(@nowarn("msg=is never used") frameType: Int, data:
ByteString) = DataFrame(data)
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
index 1912ae77..2b3fdcbd 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/GrpcProtocolWeb.scala
@@ -32,8 +32,8 @@ abstract class GrpcProtocolWebBase(subType: String) extends
AbstractGrpcProtocol
override protected def writer(codec: Codec): GrpcProtocolWriter =
AbstractGrpcProtocol.writer(this, codec, frame => encodeFrame(codec,
frame), encodeDataToResponse(codec))
- override protected def reader(codec: Codec): GrpcProtocolReader =
- AbstractGrpcProtocol.reader(codec, decodeFrame, preDecodeStrict,
preDecodeFlow)
+ override protected def reader(codec: Codec, maxInboundMessageSize: Int):
GrpcProtocolReader =
+ AbstractGrpcProtocol.reader(codec, decodeFrame, preDecodeStrict,
preDecodeFlow, maxInboundMessageSize)
private def encodeFrame(codec: Codec, frame: Frame): ChunkStreamPart =
Chunk(postEncode(encodeFrameToBytes(codec, frame)))
diff --git a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
index e8f80f99..21d3687b 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Gzip.scala
@@ -17,6 +17,9 @@ import java.io.ByteArrayOutputStream
import java.util.zip.{ GZIPInputStream, GZIPOutputStream }
import org.apache.pekko.util.ByteString
+import io.grpc.{ Status, StatusException }
+
+import scala.annotation.nowarn
object Gzip extends Codec {
override val name: String = "gzip"
@@ -29,6 +32,7 @@ object Gzip extends Codec {
ByteString.fromArrayUnsafe(baos.toByteArray)
}
+ @deprecated("Use uncompress(bytes, maxDecompressedSize), which bounds the
decompressed size", "2.0.0")
override def uncompress(compressed: ByteString): ByteString = {
val gzis = new GZIPInputStream(compressed.asInputStream)
@@ -44,6 +48,41 @@ object Gzip extends Codec {
ByteString.fromArrayUnsafe(baos.toByteArray)
}
+ /**
+ * Decompress with a maximum decompressed size limit.
+ * Checks cumulative output size during decompression to fail fast
+ * before allocating excessive memory.
+ */
+ override def uncompress(compressed: ByteString, maxDecompressedSize: Int):
ByteString = {
+ val limit = maxDecompressedSize.toLong
+ // clamp: maxDecompressedSize is validated to be positive by the settings
classes, but this
+ // method is also reachable with a hand-constructed limit, and a negative
initial size would
+ // make ByteArrayOutputStream throw IllegalArgumentException rather than a
gRPC status.
+ val initialSize = Math.max(0L, Math.min(compressed.size.toLong,
limit)).toInt
+ val gzis = new GZIPInputStream(compressed.asInputStream)
+ val baos = new ByteArrayOutputStream(initialSize)
+ val buffer = new Array[Byte](32 * 1024)
+ // Long, so that a limit close to Int.MaxValue cannot be passed by an
overflowing counter
+ var totalBytes = 0L
+ try {
+ var read = gzis.read(buffer)
+ while (read != -1) {
+ totalBytes += read
+ if (totalBytes > limit)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Decompressed message size exceeds maximum allowed
$maxDecompressedSize bytes"))
+ baos.write(buffer, 0, read)
+ read = gzis.read(buffer)
+ }
+ } finally gzis.close()
+ ByteString.fromArrayUnsafe(baos.toByteArray)
+ }
+
+ @deprecated(
+ "Use uncompress(compressedBitSet, bytes, maxDecompressedSize), which
bounds the decompressed size",
+ "2.0.0")
+ @nowarn("cat=deprecation")
override def uncompress(compressedBitSet: Boolean, bytes: ByteString):
ByteString =
if (compressedBitSet) uncompress(bytes)
else bytes
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Identity.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Identity.scala
index 281f5b6b..e236096f 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/internal/Identity.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/internal/Identity.scala
@@ -21,11 +21,32 @@ object Identity extends Codec {
override def compress(bytes: ByteString): ByteString = bytes
+ @deprecated("Use uncompress(bytes, maxDecompressedSize), which bounds the
decompressed size", "2.0.0")
override def uncompress(bytes: ByteString): ByteString = bytes
+ @deprecated(
+ "Use uncompress(compressedBitSet, bytes, maxDecompressedSize), which
bounds the decompressed size",
+ "2.0.0")
override def uncompress(compressedBitSet: Boolean, bytes: ByteString):
ByteString =
if (compressedBitSet)
throw new StatusException(
Status.INTERNAL.withDescription("Compressed-Flag bit is set, but a
compression encoding is not specified"))
else bytes
+
+ override def uncompress(bytes: ByteString, maxDecompressedSize: Int):
ByteString =
+ if (bytes.length > maxDecompressedSize)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Message size ${bytes.length} exceeds maximum allowed
$maxDecompressedSize bytes"))
+ else bytes
+
+ override def uncompress(compressedBitSet: Boolean, bytes: ByteString,
maxDecompressedSize: Int): ByteString =
+ if (compressedBitSet)
+ throw new StatusException(
+ Status.INTERNAL.withDescription("Compressed-Flag bit is set, but a
compression encoding is not specified"))
+ else if (bytes.length > maxDecompressedSize)
+ throw new StatusException(
+ Status.RESOURCE_EXHAUSTED.withDescription(
+ s"Message size ${bytes.length} exceeds maximum allowed
$maxDecompressedSize bytes"))
+ else bytes
}
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/NettyClientUtils.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/NettyClientUtils.scala
index d8d89091..a078af9b 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/NettyClientUtils.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/NettyClientUtils.scala
@@ -101,6 +101,8 @@ object NettyClientUtils {
builder =
settings.loadBalancingPolicy.map(builder.defaultLoadBalancingPolicy(_)).getOrElse(builder)
builder =
settings.overrideAuthority.map(builder.overrideAuthority(_)).getOrElse(builder)
builder = settings.userAgent.map(builder.userAgent(_)).getOrElse(builder)
+ builder = builder.maxInboundMessageSize(settings.maxInboundMessageSize)
+ // applied last so that explicit overrides win over anything derived from
settings
builder = settings.channelBuilderOverrides(builder)
val connectionAttempts = settings.loadBalancingPolicy match {
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
index 91b8ab15..068c8f34 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/internal/PekkoHttpClientUtils.scala
@@ -187,7 +187,9 @@ object PekkoHttpClientUtils {
descriptor.getFullMethodName),
GrpcEntityHelpers.metadataHeaders(headers.entries),
source)
- applyDeadline(responseToSource(singleRequest(httpRequest),
deserializer), options)
+ applyDeadline(
+ responseToSource(singleRequest(httpRequest), deserializer,
settings.maxInboundMessageSize),
+ options)
}
}
}
@@ -286,7 +288,10 @@ object PekkoHttpClientUtils {
* INTERNAL API
*/
@InternalApi
- def responseToSource[O](response: Future[HttpResponse], deserializer:
ProtobufSerializer[O])(
+ def responseToSource[O](
+ response: Future[HttpResponse],
+ deserializer: ProtobufSerializer[O],
+ maxInboundMessageSize: Int =
AbstractGrpcProtocol.DefaultMaxInboundMessageSize)(
implicit ec: ExecutionContext,
mat: Materializer): Source[O, Future[GrpcResponseMetadata]] = {
Source.lazyFutureSource[O, Future[GrpcResponseMetadata]](() => {
@@ -299,7 +304,8 @@ object PekkoHttpClientUtils {
} else {
Codecs.detect(response) match {
case Success(codec) =>
- implicit val reader: GrpcProtocolReader =
GrpcProtocolNative.newReader(codec)
+ implicit val reader: GrpcProtocolReader =
+ GrpcProtocolNative.newReader(codec, maxInboundMessageSize)
val trailerPromise = Promise[immutable.Seq[HttpHeader]]()
// Completed with success or failure based on grpc-status and
grpc-message trailing headers
val completionFuture: Future[Unit] =
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
index 18c49637..1d685036 100644
--- a/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
+++ b/runtime/src/main/scala/org/apache/pekko/grpc/javadsl/GrpcMarshalling.scala
@@ -46,6 +46,27 @@ object GrpcMarshalling {
}
.fold(Optional.empty[CompletionStage[T]])(Optional.of)
+ /**
+ * INTERNAL API
+ *
+ * Negotiates the gRPC protocol with a custom maximum inbound message size.
+ *
+ * Deliberately not an overload of `negotiated`: a second overload would
stop Scala from
+ * inferring the parameter types of the `(reader, writer) => ...` lambda at
existing call sites.
+ */
+ @InternalApi
+ def negotiatedWithMaxSize[T](
+ req: HttpRequest,
+ maxInboundMessageSize: Int,
+ f: (GrpcProtocolReader, GrpcProtocolWriter) => CompletionStage[T]):
Optional[CompletionStage[T]] =
+ GrpcProtocol
+ .negotiate(req, maxInboundMessageSize)
+ .map {
+ case (maybeReader, writer) =>
+ maybeReader.map(reader => f(reader,
writer)).fold[CompletionStage[T]](failure, identity)
+ }
+ .fold(Optional.empty[CompletionStage[T]])(Optional.of)
+
def unmarshal[T](
data: Source[ByteString, AnyRef],
u: ProtobufSerializer[T],
diff --git
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
index 42b248fe..dd316325 100644
---
a/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
+++
b/runtime/src/main/scala/org/apache/pekko/grpc/scaladsl/GrpcMarshalling.scala
@@ -60,6 +60,24 @@ object GrpcMarshalling {
case (Failure(ex), _) => Future.failed(ex)
}
+ /**
+ * INTERNAL API
+ *
+ * Negotiates the gRPC protocol with a custom maximum inbound message size.
+ *
+ * Deliberately not an overload of `negotiated`: a second overload would
stop Scala from
+ * inferring the parameter types of the `(reader, writer) => ...` lambda at
existing call sites.
+ */
+ @InternalApi
+ def negotiatedWithMaxSize[T](
+ req: HttpRequest,
+ maxInboundMessageSize: Int,
+ f: (GrpcProtocolReader, GrpcProtocolWriter) => Future[T]):
Option[Future[T]] =
+ GrpcProtocol.negotiate(req, maxInboundMessageSize).map {
+ case (Success(reader), writer) => f(reader, writer)
+ case (Failure(ex), _) => Future.failed(ex)
+ }
+
def unmarshal[T](data: Source[ByteString, Any])(
implicit u: ProtobufSerializer[T],
mat: Materializer,
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/GrpcClientSettingsSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcClientSettingsSpec.scala
index 22f87a7b..bd40b43c 100644
--- a/runtime/src/test/scala/org/apache/pekko/grpc/GrpcClientSettingsSpec.scala
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcClientSettingsSpec.scala
@@ -122,6 +122,49 @@ class GrpcClientSettingsSpec extends AnyWordSpec with
Matchers with ScalaFutures
ActorSystem("test",
clientConfig.withFallback(defaultConfig).withFallback(clientWithServiceDiscovery))
}
+ "default to the same maximum inbound message size as grpc-java" in {
+ GrpcClientSettings.connectToServiceAt("host.com",
8080).maxInboundMessageSize should be(4 * 1024 * 1024)
+ }
+
+ "support overriding the maximum inbound message size" in {
+ GrpcClientSettings
+ .connectToServiceAt("host.com", 8080)
+ .withMaxInboundMessageSize(8388608)
+ .maxInboundMessageSize should be(8388608)
+ }
+
+ "reject a non-positive maximum inbound message size" in {
+ intercept[IllegalArgumentException](
+ GrpcClientSettings.connectToServiceAt("host.com",
8080).withMaxInboundMessageSize(0))
+ intercept[IllegalArgumentException](
+ GrpcClientSettings.connectToServiceAt("host.com",
8080).withMaxInboundMessageSize(-1))
+ }
+
+ "read the maximum inbound message size from config" in {
+ val config = ConfigFactory
+ .parseString("""
+ host = "host.com"
+ port = 8080
+ max-inbound-message-size = 8388608
+ """)
+
.withFallback(sys.settings.config.getConfig("""pekko.grpc.client."*""""))
+
+ GrpcClientSettings.fromConfig(config).maxInboundMessageSize should
be(8388608)
+ }
+
+ "fall back on the default maximum inbound message size for a Config
without the key" in {
+ // fromConfig(Config) is public API and is called with hand-assembled
Configs
+ val config = ConfigFactory
+ .parseString("""
+ host = "host.com"
+ port = 8080
+ """)
+
.withFallback(sys.settings.config.getConfig("""pekko.grpc.client."*""""))
+ .withoutPath("max-inbound-message-size")
+
+ GrpcClientSettings.fromConfig(config).maxInboundMessageSize should be(4
* 1024 * 1024)
+ }
+
"use static service discovery for connectToServiceAt" in {
val settings = GrpcClientSettings.connectToServiceAt("host.com", 8080)
val resolved = settings.serviceDiscovery.lookup("any",
1.second).futureValue
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/GrpcServerSettingsSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcServerSettingsSpec.scala
new file mode 100644
index 00000000..259359e5
--- /dev/null
+++ b/runtime/src/test/scala/org/apache/pekko/grpc/GrpcServerSettingsSpec.scala
@@ -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.pekko.grpc
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.grpc.internal.AbstractGrpcProtocol
+import pekko.testkit.TestKit
+import com.typesafe.config.ConfigFactory
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class GrpcServerSettingsSpec extends TestKit(ActorSystem()) with
AnyWordSpecLike with Matchers {
+
+ "GrpcServerSettings" should {
+
+ "default to the same maximum inbound message size as grpc-java" in {
+ GrpcServerSettings(system).maxInboundMessageSize shouldBe 4 * 1024 * 1024
+ GrpcServerSettings(system).maxInboundMessageSize shouldBe
AbstractGrpcProtocol.DefaultMaxInboundMessageSize
+ }
+
+ "read the maximum inbound message size from config" in {
+ val config = ConfigFactory.parseString("max-inbound-message-size =
8388608")
+
+ GrpcServerSettings.fromConfig(config).maxInboundMessageSize shouldBe
8388608
+ }
+
+ "fall back on the default when the key is absent" in {
+ GrpcServerSettings
+ .fromConfig(ConfigFactory.empty())
+ .maxInboundMessageSize shouldBe
AbstractGrpcProtocol.DefaultMaxInboundMessageSize
+ }
+
+ "support overriding the maximum inbound message size" in {
+
GrpcServerSettings(system).withMaxInboundMessageSize(1234).maxInboundMessageSize
shouldBe 1234
+ }
+
+ "reject a non-positive maximum inbound message size" in {
+
intercept[IllegalArgumentException](GrpcServerSettings(system).withMaxInboundMessageSize(0))
+
intercept[IllegalArgumentException](GrpcServerSettings(system).withMaxInboundMessageSize(-1))
+ intercept[IllegalArgumentException](
+
GrpcServerSettings.fromConfig(ConfigFactory.parseString("max-inbound-message-size
= 0")))
+ }
+
+ "expose the same value via the Java API" in {
+ GrpcServerSettings.create(system).maxInboundMessageSize shouldBe
GrpcServerSettings(system).maxInboundMessageSize
+ }
+ }
+}
diff --git
a/runtime/src/test/scala/org/apache/pekko/grpc/internal/MaxInboundMessageSizeSpec.scala
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/MaxInboundMessageSizeSpec.scala
new file mode 100644
index 00000000..a484c662
--- /dev/null
+++
b/runtime/src/test/scala/org/apache/pekko/grpc/internal/MaxInboundMessageSizeSpec.scala
@@ -0,0 +1,206 @@
+/*
+ * 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.pekko.grpc.internal
+
+import java.io.ByteArrayOutputStream
+import java.util.zip.GZIPOutputStream
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.grpc.GrpcProtocol.{ DataFrame, Frame }
+import pekko.stream.scaladsl.Source
+import pekko.stream.testkit.scaladsl.TestSink
+import pekko.testkit.TestKit
+import pekko.util.ByteString
+import io.grpc.{ Status, StatusException }
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+class MaxInboundMessageSizeSpec extends TestKit(ActorSystem()) with
AnyWordSpecLike with Matchers {
+
+ private val Limit = 64 * 1024
+
+ /** Inflates to far more than `Limit`, but compresses to far less, so only a
bounded decompressor catches it. */
+ private val BombSize = 4 * 1024 * 1024
+
+ private val TooLargeWhenDecompressed = s"Decompressed message size exceeds
maximum allowed $Limit bytes"
+
+ /** A frame header: 1 byte flags + 4 bytes big-endian length. */
+ private def frameHeader(flags: Byte, length: Int): ByteString = {
+ val header = new Array[Byte](5)
+ header(0) = flags
+ header(1) = (length >>> 24).toByte
+ header(2) = (length >>> 16).toByte
+ header(3) = (length >>> 8).toByte
+ header(4) = length.toByte
+ ByteString.fromArrayUnsafe(header, 0, 5)
+ }
+
+ private def gzipped(bytes: ByteString): ByteString = {
+ val baos = new ByteArrayOutputStream()
+ val gzos = new GZIPOutputStream(baos)
+ try gzos.write(bytes.toArrayUnsafe())
+ finally gzos.close()
+ ByteString.fromArrayUnsafe(baos.toByteArray)
+ }
+
+ /** Highly compressible payload. */
+ private def zeros(size: Int): ByteString = ByteString.fromArrayUnsafe(new
Array[Byte](size))
+
+ private def bomb: ByteString = gzipped(zeros(BombSize))
+
+ private def streamingError(codec: Codec, bytes: ByteString): StatusException
=
+ Source
+ .single(bytes)
+ .via(GrpcProtocolNative.newReader(codec, Limit).frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+ .expectError() match {
+ case s: StatusException => s
+ case other => fail(s"expected a StatusException, got
[$other]")
+ }
+
+ private def strictError(codec: Codec, bytes: ByteString): StatusException =
+ intercept[StatusException](GrpcProtocolNative.newReader(codec,
Limit).decodeSingleFrame(bytes))
+
+ "The streaming frame decoder" should {
+
+ "reject a frame whose declared length exceeds the limit" in {
+ // header only: the oversized payload is never sent, so this can only be
detected
+ // from the declared length, before any of it is buffered
+ val status = streamingError(Identity, frameHeader(0, Limit +
1)).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ status.getDescription should include(s"Frame length ${Limit + 1}")
+ }
+
+ "reject a negative frame length with INTERNAL" in {
+ streamingError(Identity, frameHeader(0, -1)).getStatus.getCode shouldBe
Status.Code.INTERNAL
+ }
+
+ "accept a frame exactly at the limit" in {
+ val data = zeros(Limit)
+
+ Source
+ .single(frameHeader(0, Limit) ++ data)
+ .via(GrpcProtocolNative.newReader(Identity, Limit).frameDecoder)
+ .runWith(TestSink[Frame]())
+ .request(1)
+ .expectNext(DataFrame(data))
+ .expectComplete()
+ }
+
+ "reject a gzip bomb while decompressing rather than after" in {
+ val compressed = bomb
+ // the frame itself is well within the limit, so the frame length check
cannot catch this
+ compressed.length should be < Limit
+
+ val status = streamingError(Gzip, frameHeader(1, compressed.length) ++
compressed).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ // This message is produced only by Gzip's streaming, size-bounded
decoder. The generic
+ // Codec fallback decompresses in full first and reports the actual
size, so asserting
+ // on the exact text pins the fail-fast path rather than merely the
outcome.
+ status.getDescription shouldBe TooLargeWhenDecompressed
+ }
+ }
+
+ "The strict frame decoder" should {
+
+ "reject a frame whose declared length exceeds the limit" in {
+ val status = strictError(Identity, frameHeader(0, Limit + 1) ++
zeros(Limit + 1)).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ status.getDescription should include(s"Frame length ${Limit + 1}")
+ }
+
+ "reject a negative frame length with INTERNAL" in {
+ strictError(Identity, frameHeader(0, -1)).getStatus.getCode shouldBe
Status.Code.INTERNAL
+ }
+
+ "reject a gzip bomb while decompressing rather than after" in {
+ val compressed = bomb
+ val status = strictError(Gzip, frameHeader(1, compressed.length) ++
compressed).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ status.getDescription shouldBe TooLargeWhenDecompressed
+ }
+
+ "decode a frame within the limit" in {
+ val data = zeros(16)
+
+ GrpcProtocolNative.newReader(Identity,
Limit).decodeSingleFrame(frameHeader(0, 16) ++ data) shouldBe data
+ }
+ }
+
+ "Gzip" should {
+
+ "route the compression-bit overload to the size-bounded decoder" in {
+ val status = intercept[StatusException](Gzip.uncompress(compressedBitSet
= true, bomb, Limit)).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ status.getDescription shouldBe TooLargeWhenDecompressed
+ }
+
+ "decompress payloads within the limit" in {
+ val data = zeros(Limit)
+
+ Gzip.uncompress(compressedBitSet = true, gzipped(data), Limit) shouldBe
data
+ }
+
+ "not reject a payload when the limit is Int.MaxValue" in {
+ val data = zeros(64 * 1024)
+
+ Gzip.uncompress(gzipped(data), Int.MaxValue) shouldBe data
+ }
+
+ "report a non-positive limit as a gRPC status rather than
IllegalArgumentException" in {
+ val status =
intercept[StatusException](Gzip.uncompress(gzipped(zeros(16)), -1)).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ }
+ }
+
+ "Identity" should {
+
+ "reject a payload larger than the limit" in {
+ val status =
+ intercept[StatusException](Identity.uncompress(compressedBitSet =
false, zeros(Limit + 1), Limit)).getStatus
+
+ status.getCode shouldBe Status.Code.RESOURCE_EXHAUSTED
+ }
+
+ "still reject a set compression bit with INTERNAL" in {
+ val status =
intercept[StatusException](Identity.uncompress(compressedBitSet = true,
zeros(1), Limit)).getStatus
+
+ status.getCode shouldBe Status.Code.INTERNAL
+ }
+ }
+
+ "newReader" should {
+
+ "reuse a cached reader for the default limit" in {
+ assert(GrpcProtocolNative.newReader(Identity) eq
GrpcProtocolNative.newReader(Identity))
+ assert(GrpcProtocolWeb.newReader(Gzip) eq
GrpcProtocolWeb.newReader(Gzip))
+ }
+
+ "build a fresh reader for a custom limit" in {
+ assert(GrpcProtocolNative.newReader(Identity, Limit) ne
GrpcProtocolNative.newReader(Identity))
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
