Skip to content

Commit 1b0c1f5

Browse files
ok2cok2c
committed
RFC 9113 conformance: corrected H2 stream id generation and verification logic
1 parent 91d5032 commit 1b0c1f5

4 files changed

Lines changed: 62 additions & 59 deletions

File tree

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

Lines changed: 18 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -128,7 +128,6 @@ enum SettingsHandshake { READY, TRANSMITTED, ACKED }
128128

129129
private Continuation continuation;
130130

131-
private int processedRemoteStreamId;
132131
private EndpointDetails endpointDetails;
133132
private boolean goAwayReceived;
134133

@@ -422,7 +421,7 @@ public final void onInput(final ByteBuffer src) throws HttpException, IOExceptio
422421
} else {
423422
if (inputBuffer.isEndOfStream()) {
424423
if (connState == ConnectionHandshake.ACTIVE) {
425-
final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Unexpected end of stream");
424+
final RawFrame goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.NO_ERROR, "Unexpected end of stream");
426425
commitFrame(goAway);
427426
}
428427
connState = ConnectionHandshake.SHUTDOWN;
@@ -512,7 +511,7 @@ public final void onOutput() throws HttpException, IOException {
512511
streams.release(stream);
513512
it.remove();
514513
} else {
515-
if (streams.isSameSide(stream.getId()) || stream.getId() <= processedRemoteStreamId) {
514+
if (streams.isSameSide(stream.getId()) || stream.getId() <= streams.getLastRemoteId()) {
516515
liveStreams++;
517516
}
518517
}
@@ -555,10 +554,10 @@ public final void onTimeout(final Timeout timeout) throws HttpException, IOExcep
555554

556555
final RawFrame goAway;
557556
if (localSettingState != SettingsHandshake.ACKED) {
558-
goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.SETTINGS_TIMEOUT,
557+
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.SETTINGS_TIMEOUT,
559558
"Setting timeout (" + timeout + ")");
560559
} else {
561-
goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR,
560+
goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.NO_ERROR,
562561
"Timeout due to inactivity (" + timeout + ")");
563562
}
564563
commitFrame(goAway);
@@ -588,7 +587,7 @@ private void executeShutdown(final ShutdownCommand shutdownCommand) throws IOExc
588587
connState = ConnectionHandshake.SHUTDOWN;
589588
} else {
590589
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
591-
final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, H2Error.NO_ERROR, "Graceful shutdown");
590+
final RawFrame goAway = frameFactory.createGoAway(streams.getLastRemoteId(), H2Error.NO_ERROR, "Graceful shutdown");
592591
commitFrame(goAway);
593592
connState = streams.isEmpty() ? ConnectionHandshake.SHUTDOWN : ConnectionHandshake.GRACEFUL_SHUTDOWN;
594593
}
@@ -608,8 +607,8 @@ private void executeRequest(final ExecutableCommand executableCommand) throws IO
608607
final H2StreamHandler streamHandler = createLocallyInitiatedStream(
609608
executableCommand, channel, httpProcessor, connMetrics);
610609

611-
final H2Stream stream = new H2Stream(channel, streamHandler, false);
612-
streams.add(streamId, stream);
610+
final H2Stream stream = new H2Stream(channel, streamHandler);
611+
streams.addLocallyInitiated(stream);
613612

614613
if (streamListener != null) {
615614
final int initInputWindow = stream.getInputWindow().get();
@@ -634,8 +633,8 @@ public void executePush(final int promisedStreamId, final AsyncPushProducer push
634633
context.setEndpointDetails(getEndpointDetails());
635634
final H2StreamHandler streamHandler = new ServerPushH2StreamHandler(
636635
channel, httpProcessor, connMetrics, pushProducer, context);
637-
final H2Stream stream = new H2Stream(channel, streamHandler, false);
638-
streams.add(promisedStreamId, stream);
636+
final H2Stream stream = new H2Stream(channel, streamHandler);
637+
streams.addLocallyInitiated(stream);
639638
}
640639

641640
public final void onException(final Exception cause) {
@@ -662,7 +661,7 @@ public final void onException(final Exception cause) {
662661
} else {
663662
errorCode = H2Error.INTERNAL_ERROR;
664663
}
665-
final RawFrame goAway = frameFactory.createGoAway(processedRemoteStreamId, errorCode, cause.getMessage());
664+
final RawFrame goAway = frameFactory.createGoAway(streams.getLastRemoteId(), errorCode, cause.getMessage());
666665
commitFrame(goAway);
667666
}
668667
}
@@ -723,8 +722,6 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
723722
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "GOAWAY received");
724723
}
725724

726-
streams.updateLastStreamId(streamId);
727-
728725
final H2StreamChannel channel = createChannel(streamId, false);
729726
final H2StreamHandler streamHandler;
730727
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
@@ -734,11 +731,11 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
734731
channel.markLocalClosed();
735732
}
736733

737-
stream = new H2Stream(channel, streamHandler, true);
734+
stream = new H2Stream(channel, streamHandler);
738735
if (stream.isOutputReady()) {
739736
stream.produceOutput();
740737
}
741-
streams.add(streamId, stream);
738+
streams.addRemotelyInitiated(stream);
742739
} else if (stream.isLocalClosed() && stream.isRemoteClosed()) {
743740
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
744741
}
@@ -921,8 +918,6 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
921918
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Stream already open: " + promisedStreamId);
922919
}
923920

924-
streams.updateLastStreamId(promisedStreamId);
925-
926921
final H2StreamChannel channel = createChannel(promisedStreamId, false);
927922
final H2StreamHandler streamHandler;
928923
if (connState.compareTo(ConnectionHandshake.ACTIVE) <= 0) {
@@ -933,8 +928,8 @@ private void consumeFrame(final RawFrame frame) throws HttpException, IOExceptio
933928
channel.markLocalClosed();
934929
}
935930

936-
final H2Stream promisedStream = new H2Stream(channel, streamHandler, true);
937-
streams.add(promisedStreamId, promisedStream);
931+
final H2Stream promisedStream = new H2Stream(channel, streamHandler);
932+
streams.addRemotelyInitiated(promisedStream);
938933

939934
try {
940935
consumePushPromiseFrame(frame, payload, promisedStream);
@@ -1019,9 +1014,6 @@ private void consumePushPromiseFrame(final RawFrame frame, final ByteBuffer payl
10191014
}
10201015
if (continuation == null) {
10211016
final List<Header> headers = hPackDecoder.decodeHeaders(payload);
1022-
if (promisedStreamId > processedRemoteStreamId) {
1023-
processedRemoteStreamId = promisedStreamId;
1024-
}
10251017
if (streamListener != null) {
10261018
streamListener.onHeaderInput(this, promisedStreamId, headers);
10271019
}
@@ -1052,9 +1044,6 @@ private void consumeHeaderFrame(final RawFrame frame, final H2Stream stream) thr
10521044
}
10531045
if (continuation == null) {
10541046
final List<Header> headers = decodeHeaders(payload);
1055-
if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
1056-
processedRemoteStreamId = streamId;
1057-
}
10581047
if (streamListener != null) {
10591048
streamListener.onHeaderInput(this, streamId, headers);
10601049
}
@@ -1073,9 +1062,6 @@ private void consumeContinuationFrame(final RawFrame frame, final H2Stream strea
10731062
continuation.copyPayload(payload);
10741063
if (frame.isFlagSet(FrameFlag.END_HEADERS)) {
10751064
final List<Header> headers = decodeHeaders(continuation.getContent());
1076-
if (stream.isRemoteInitiated() && streamId > processedRemoteStreamId) {
1077-
processedRemoteStreamId = streamId;
1078-
}
10791065
if (streamListener != null) {
10801066
streamListener.onHeaderInput(this, streamId, headers);
10811067
}
@@ -1272,7 +1258,8 @@ void appendState(final StringBuilder buf) {
12721258
.append(", connOutputWindow=").append(connOutputWindow)
12731259
.append(", outputQueue=").append(outputQueue.size())
12741260
.append(", streams.size=").append(streams.size())
1275-
.append(", processedRemoteStreamId=").append(processedRemoteStreamId);
1261+
.append(", streams.lastLocal=").append(streams.getLastLocalId())
1262+
.append(", streams.lastRemote=").append(streams.getLastRemoteId());
12761263
}
12771264

12781265
private static class Continuation {
@@ -1320,8 +1307,8 @@ H2StreamChannel createChannel(final int streamId, final boolean idle) {
13201307
return new H2StreamChannelImpl(streamId, idle, initInputWinSize, initOutputWinSize);
13211308
}
13221309

1323-
void addStream(final H2Stream stream) {
1324-
streams.add(stream);
1310+
void addStream(final H2Stream stream) throws H2ConnectionException {
1311+
streams.addLocallyInitiated(stream);
13251312
}
13261313

13271314
class H2StreamChannelImpl implements H2StreamChannel {

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

Lines changed: 1 addition & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -49,29 +49,20 @@ class H2Stream {
4949

5050
private final H2StreamChannel channel;
5151
private final H2StreamHandler handler;
52-
private final boolean remoteInitiated;
5352
private final AtomicBoolean released;
5453

5554
private volatile boolean remoteClosed;
5655

57-
H2Stream(
58-
final H2StreamChannel channel,
59-
final H2StreamHandler handler,
60-
final boolean remoteInitiated) {
56+
H2Stream(final H2StreamChannel channel, final H2StreamHandler handler) {
6157
this.channel = channel;
6258
this.handler = handler;
63-
this.remoteInitiated = remoteInitiated;
6459
this.released = new AtomicBoolean();
6560
}
6661

6762
int getId() {
6863
return channel.getId();
6964
}
7065

71-
boolean isRemoteInitiated() {
72-
return remoteInitiated;
73-
}
74-
7566
AtomicInteger getOutputWindow() {
7667
return channel.getOutputWindow();
7768
}

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

Lines changed: 39 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -45,13 +45,15 @@ class H2Streams {
4545
private final StreamIdGenerator idGenerator;
4646
private final Map<Integer, H2Stream> streamMap;
4747
private final Queue<H2Stream> streams;
48-
private final AtomicInteger lastStreamId;
48+
private final AtomicInteger lastLocalId;
49+
private final AtomicInteger lastRemoteId;
4950

5051
public H2Streams(final StreamIdGenerator idGenerator) {
5152
this.idGenerator = Args.notNull(idGenerator, "Stream id generator");
5253
this.streamMap = new ConcurrentHashMap<>();
5354
this.streams = new ConcurrentLinkedQueue<>();
54-
this.lastStreamId = new AtomicInteger(0);
55+
this.lastLocalId = new AtomicInteger(0);
56+
this.lastRemoteId = new AtomicInteger(0);
5557
}
5658

5759
public int size() {
@@ -66,13 +68,33 @@ public Iterator<H2Stream> iterator() {
6668
return streams.iterator();
6769
}
6870

69-
public void add(final int streamId, final H2Stream stream) {
71+
public int getLastLocalId() {
72+
return lastLocalId.get();
73+
}
74+
75+
public int getLastRemoteId() {
76+
return lastRemoteId.get();
77+
}
78+
79+
public void addLocallyInitiated(final H2Stream stream) throws H2ConnectionException {
80+
final int streamId = stream.getId();
81+
if (isOtherSide(streamId)) {
82+
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
83+
}
7084
streamMap.put(streamId, stream);
7185
streams.add(stream);
7286
}
7387

74-
public void add(final H2Stream stream) {
75-
streamMap.put(stream.getId(), stream);
88+
public void addRemotelyInitiated(final H2Stream stream) throws H2ConnectionException {
89+
final int streamId = stream.getId();
90+
if (isSameSide(streamId)) {
91+
throw new H2ConnectionException(H2Error.PROTOCOL_ERROR, "Illegal stream id");
92+
}
93+
final int currentId = lastRemoteId.get();
94+
if (streamId > currentId) {
95+
lastRemoteId.compareAndSet(currentId, streamId);
96+
}
97+
streamMap.put(streamId, stream);
7698
streams.add(stream);
7799
}
78100

@@ -101,8 +123,14 @@ public H2Stream lookup(final int streamId) {
101123
public H2Stream lookupValidOrNull(final int streamId) throws H2ConnectionException {
102124
final H2Stream stream = streamMap.get(streamId);
103125
if (stream == null) {
104-
if (streamId <= lastStreamId.get()) {
105-
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
126+
if (idGenerator.isSameSide(streamId)) {
127+
if (streamId <= lastLocalId.get()) {
128+
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
129+
}
130+
} else {
131+
if (streamId <= lastRemoteId.get()) {
132+
throw new H2ConnectionException(H2Error.STREAM_CLOSED, "Stream closed");
133+
}
106134
}
107135
} else {
108136
if (stream.isLocalClosed() && stream.isRemoteClosed()) {
@@ -124,18 +152,15 @@ public boolean isSameSide(final int streamId) {
124152
return idGenerator.isSameSide(streamId);
125153
}
126154

127-
public void updateLastStreamId(final int streamId) {
128-
final int currentId = lastStreamId.get();
129-
if (streamId > currentId) {
130-
lastStreamId.compareAndSet(currentId, streamId);
131-
}
155+
public boolean isOtherSide(final int streamId) {
156+
return !idGenerator.isSameSide(streamId);
132157
}
133158

134159
public int generateStreamId() {
135160
for (;;) {
136-
final int currentId = lastStreamId.get();
161+
final int currentId = lastLocalId.get();
137162
final int newStreamId = idGenerator.generate(currentId);
138-
if (lastStreamId.compareAndSet(currentId, newStreamId)) {
163+
if (lastLocalId.compareAndSet(currentId, newStreamId)) {
139164
return newStreamId;
140165
}
141166
}

‎httpcore5-h2/src/test/java/org/apache/hc/core5/http2/impl/nio/TestAbstractH2StreamMultiplexer.java‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -677,15 +677,15 @@ void testStreamRemoteReset() throws Exception {
677677
final AbstractH2StreamMultiplexer streamMultiplexer = new H2StreamMultiplexerImpl(
678678
protocolIOSession,
679679
FRAME_FACTORY,
680-
StreamIdGenerator.EVEN,
680+
StreamIdGenerator.ODD,
681681
httpProcessor,
682682
CharCodingConfig.DEFAULT,
683683
h2Config,
684684
h2StreamListener,
685685
() -> streamHandler);
686686

687687
final H2StreamChannel channel = streamMultiplexer.createChannel(1, false);
688-
final H2Stream stream = new H2Stream(channel, streamHandler, false);
688+
final H2Stream stream = new H2Stream(channel, streamHandler);
689689
streamMultiplexer.addStream(stream);
690690

691691
final ByteArrayBuffer buf = new ByteArrayBuffer(19);
@@ -724,15 +724,15 @@ void testStreamRemoteResetNoErrorRemoteAlreadyClosed() throws Exception {
724724
final AbstractH2StreamMultiplexer streamMultiplexer = new H2StreamMultiplexerImpl(
725725
protocolIOSession,
726726
FRAME_FACTORY,
727-
StreamIdGenerator.EVEN,
727+
StreamIdGenerator.ODD,
728728
httpProcessor,
729729
CharCodingConfig.DEFAULT,
730730
h2Config,
731731
h2StreamListener,
732732
() -> streamHandler);
733733

734734
final H2StreamChannel channel = streamMultiplexer.createChannel(1, false);
735-
final H2Stream stream = new H2Stream(channel, streamHandler, false);
735+
final H2Stream stream = new H2Stream(channel, streamHandler);
736736
streamMultiplexer.addStream(stream);
737737

738738
final ByteArrayBuffer buf = new ByteArrayBuffer(19);

0 commit comments

Comments
 (0)