Add per-connection protection against excessive early HTTP/2 stream resets.

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/AbstractH2StreamMultiplexer.java

@@ -104,7 +104,7 @@
 
 abstract class AbstractH2StreamMultiplexer implements Identifiable, HttpConnection {
 
-    private final H2ControlFrameRateLimiter controlFrameRateLimiter;
+    private final FrameLimitController frameLimitController;
 
     private static final long CONNECTION_WINDOW_LOW_MARK = 10 * 1024 * 1024;
 
@@ -191,7 +191,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
             final Timeout validateAfterInactivity,
             final Timeout pingAckTimeout) {
         this(ioSession, frameFactory, idGenerator, httpProcessor, charCodingConfig, h2Config, streamListener,
-                validateAfterInactivity, pingAckTimeout, new H2ControlFrameRateLimiter());
+                validateAfterInactivity, pingAckTimeout, new FrameLimitController());
     }
 
     AbstractH2StreamMultiplexer(
@@ -204,7 +204,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
             final H2StreamListener streamListener,
             final Timeout validateAfterInactivity,
             final Timeout pingAckTimeout,
-            final H2ControlFrameRateLimiter controlFrameRateLimiter) {
+            final FrameLimitController frameLimitController) {
         this.ioSession = Args.notNull(ioSession, "IO session");
         this.frameFactory = Args.notNull(frameFactory, "Frame factory");
         this.httpProcessor = Args.notNull(httpProcessor, "HTTP processor");
@@ -233,7 +233,7 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
         this.lastActivityTime = System.currentTimeMillis();
         this.validateAfterInactivity = validateAfterInactivity;
         this.pingAckTimeout = Args.notNull(pingAckTimeout, "PING ACK timeout");
-        this.controlFrameRateLimiter = Args.notNull(controlFrameRateLimiter, "Control frame rate limiter");
+        this.frameLimitController = Args.notNull(frameLimitController, "Frame limit controller");
     }
 
     @Override
@@ -970,6 +970,12 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                 final H2Stream stream = streams.lookupSeen(streamId);
                 if (stream != null) {
                     final int errorCode = payload.getInt();
+                    // Only newly activated peer-initiated streams count. Valid resets of
+                    // long-lived or locally initiated streams remain unaffected.
+                    if (streams.isOtherSide(streamId) && !stream.isReserved() && !stream.isClosed()
+                            && !frameLimitController.acceptRapidReset(stream.getActivatedAtNanos(), nowNanos)) {
+                        throw new H2ConnectionException(H2Error.ENHANCE_YOUR_CALM, "Excessive rapid HTTP/2 stream resets");
+                    }
                     if (errorCode == H2Error.NO_ERROR.getCode() && allowGracefulAbort(stream)) {
                         stream.abortGracefully();
                         requestSessionOutput();
@@ -994,10 +1000,10 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                         pingHandler.consumeResponse(ping);
                     }
                 } else {
-                    if (!controlFrameRateLimiter.accept(System.nanoTime())) {
-                    throw new H2ConnectionException(H2Error.ENHANCE_YOUR_CALM, "Excessive HTTP/2 PING / SETTINGS frames");
-                }
-                final ByteBuffer pong = ByteBuffer.allocate(ping.remaining());
+                    if (!frameLimitController.acceptControlFrame(System.nanoTime())) {
+                        throw new H2ConnectionException(H2Error.ENHANCE_YOUR_CALM, "Excessive HTTP/2 PING / SETTINGS frames");
+                    }
+                    final ByteBuffer pong = ByteBuffer.allocate(ping.remaining());
                     pong.put(ping);
                     pong.flip();
                     final RawFrame response = frameFactory.createPingAck(pong);
@@ -1030,7 +1036,7 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
                         throw new H2ConnectionException(H2Error.FRAME_SIZE_ERROR,
                                 "Invalid SETTINGS payload");
                     }
-                    if (!controlFrameRateLimiter.accept(System.nanoTime())) {
+                    if (!frameLimitController.acceptControlFrame(System.nanoTime())) {
                         throw new H2ConnectionException(H2Error.ENHANCE_YOUR_CALM,
                                 "Excessive HTTP/2 PING / SETTINGS frames");
                     }

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/FrameLimitController.java

@@ -0,0 +1,145 @@
+/*
+ * ====================================================================
+ * 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.
+ * ====================================================================
+ *
+ * This software consists of voluntary contributions made by many
+ * individuals on behalf of the Apache Software Foundation.  For more
+ * information on the Apache Software Foundation, please see
+ * <http://www.apache.org/>.
+ *
+ */
+package org.apache.hc.core5.http2.impl.nio;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hc.core5.util.Args;
+
+/**
+ * Per-connection resource policy for unsolicited HTTP/2 control frames. Non-ACK PING and
+ * SETTINGS frames share one token bucket, exactly as before this class existed; rapid
+ * cancellation (RST_STREAM) of peer-initiated streams is charged against its own,
+ * independent bucket.
+ *
+ * <p>ACK frames, DATA and WINDOW_UPDATE are not charged, as they can represent legitimate
+ * high-throughput or highly concurrent exchanges. A stream reset is only charged against the
+ * RST_STREAM bucket when it arrives shortly after the stream was activated; resets of
+ * long-lived streams are not charged.</p>
+ *
+ * <p>Parameters are hardwired for now; they may become user configurable in the future, at
+ * which point this is the single place that would need a public constructor or setters.</p>
+ *
+ * <p>This is a resource safeguard, not an RFC 9113 protocol-validity rule.</p>
+ */
+final class FrameLimitController {
+
+    static final int CONTROL_FRAME_CAPACITY = 256;
+    static final int CONTROL_FRAME_REFILL_PER_SECOND = 128;
+
+    static final int RST_STREAM_CAPACITY = 128;
+    static final int RST_STREAM_REFILL_PER_SECOND = 64;
+    static final long RST_STREAM_MAX_AGE_NANOS = TimeUnit.SECONDS.toNanos(1);
+
+    private final TokenBucket controlFrameBucket;
+    private final TokenBucket rstStreamBucket;
+    private final long rstStreamMaxAgeNanos;
+
+    FrameLimitController() {
+        this(CONTROL_FRAME_CAPACITY, CONTROL_FRAME_REFILL_PER_SECOND,
+                RST_STREAM_CAPACITY, RST_STREAM_REFILL_PER_SECOND,
+                RST_STREAM_MAX_AGE_NANOS);
+    }
+
+    FrameLimitController(
+            final int controlFrameCapacity,
+            final int controlFrameRefillPerSecond,
+            final int rstStreamCapacity,
+            final int rstStreamRefillPerSecond,
+            final long rstStreamMaxAgeNanos) {
+        this.controlFrameBucket = new TokenBucket(controlFrameCapacity, controlFrameRefillPerSecond);
+        this.rstStreamBucket = new TokenBucket(rstStreamCapacity, rstStreamRefillPerSecond);
+        this.rstStreamMaxAgeNanos = Args.positive(rstStreamMaxAgeNanos, "Maximum reset age");
+    }
+
+    boolean acceptControlFrame(final long nowNanos) {
+        return controlFrameBucket.accept(nowNanos);
+    }
+
+    boolean acceptRapidReset(final long activatedAtNanos, final long nowNanos) {
+        final long age = nowNanos - activatedAtNanos;
+        return age < 0 || age >= rstStreamMaxAgeNanos || rstStreamBucket.accept(nowNanos);
+    }
+
+    /**
+     * Constant-time token bucket, with no locks or per-frame allocation, used to tolerate
+     * bursts and recover over time. Shared by every bucket kept by {@link FrameLimitController};
+     * a new kind of limited frame is just another instance of this class.
+     */
+    static final class TokenBucket {
+
+        private static final long NANOS_PER_SECOND = 1_000_000_000L;
+
+        private final int capacity;
+        private final int refillPerSecond;
+        private final long fullRefillNanos;
+
+        private int tokens;
+        private long previousNanos;
+        private long remainder;
+        private boolean initialized;
+
+        TokenBucket(final int capacity, final int refillPerSecond) {
+            this.capacity = Args.positive(capacity, "Capacity");
+            this.refillPerSecond = Args.positive(refillPerSecond, "Refill per second");
+            this.fullRefillNanos = (NANOS_PER_SECOND * capacity + refillPerSecond - 1) / refillPerSecond;
+            this.tokens = capacity;
+        }
+
+        boolean accept(final long nowNanos) {
+            if (!initialized) {
+                initialized = true;
+                previousNanos = nowNanos;
+            } else {
+                final long elapsed = nowNanos - previousNanos;
+                // Avoid overflow and do not refill on a backward clock jump.
+                if (elapsed > 0) {
+                    previousNanos = nowNanos;
+                    if (elapsed >= fullRefillNanos) {
+                        tokens = capacity;
+                        remainder = 0;
+                    } else {
+                        final long numerator = elapsed * refillPerSecond + remainder;
+                        final long refill = numerator / NANOS_PER_SECOND;
+                        remainder = numerator % NANOS_PER_SECOND;
+                        if (refill > 0) {
+                            tokens = (int) Math.min(capacity, (long) tokens + refill);
+                            if (tokens == capacity) {
+                                remainder = 0;
+                            }
+                        }
+                    }
+                }
+            }
+            if (tokens == 0) {
+                return false;
+            }
+            tokens--;
+            return true;
+        }
+    }
+}

httpcore5-h2/src/main/java/org/apache/hc/core5/http2/impl/nio/H2Stream.java

@@ -63,6 +63,7 @@ class H2Stream implements StreamControl {
     private volatile boolean remoteClosed;
 
     private volatile long lastActivityNanos;
+    private volatile long activatedAtNanos;
 
     private volatile Timeout idleTimeout;
     private volatile PriorityValue priorityValue;
@@ -112,9 +113,14 @@ private void triggerClosed() {
     void activate() {
         reserved = false;
         touch();
+        activatedAtNanos = lastActivityNanos;
         triggerOpen();
     }
 
+    long getActivatedAtNanos() {
+        return activatedAtNanos;
+    }
+
     AtomicInteger getOutputWindow() {
         return channel.getOutputWindow();
     }