Basic PRIORITY frame validation. (#71)

Motivation:

In the future we'll want to use PRIORITY frames to provide feedback on
how we emit data. Additionally, RFC 7540 forbids certain kinds of
PRIORITY frame, which right now we tolerate receiving. We can design a
minor PRIORITY shim to ensure that we have a code path for PRIORITY data
to be passed around, and then use this shim to enforce RFC 7540's
requirements.

Modifications:

- Added support for PRIORITY frame data inbound.
- Enforce that frames may not establish self-dependency.

Result:

Better policing of RFC 7540 constraints.

Author: Lukasa

Sources/NIOHTTP2/HTTP2ChannelHandler.swift

@@ -206,7 +206,7 @@ extension NIOHTTP2Handler {
         // All frames have one basic processing step: do we send them on, or drop them?
         // Some frames have further processing steps, regarding triggering user events or other operations.
         // Here we centralise this processing.
-        let result: StateMachineResultWithEffect
+        var result: StateMachineResultWithEffect
 
         switch frame.payload {
         case .alternativeService, .origin:
@@ -218,6 +218,16 @@ extension NIOHTTP2Handler {
             result = self.stateMachine.receiveGoaway(lastStreamID: lastStreamID)
         case .headers(let headerBody):
             result = self.stateMachine.receiveHeaders(streamID: frame.streamID, headers: headerBody.headers, isEndStreamSet: headerBody.endStream)
+
+            // Apply a priority update if one is here. If this fails, it may cause a connection error.
+            if let priorityData = headerBody.priorityData {
+                do {
+                    try self.outboundBuffer.priorityUpdate(streamID: frame.streamID, priorityData: priorityData)
+                } catch {
+                    result = StateMachineResultWithEffect(result: .connectionError(underlyingError: error, type: .protocolError), effect: nil)
+                }
+            }
+
         case .ping(let pingData, let ack):
             let (stateMachineResult, postPingOperation) = self.stateMachine.receivePing(ackFlagSet: ack)
             result = stateMachineResult
@@ -230,8 +240,17 @@ extension NIOHTTP2Handler {
                 self.encodeAndWriteFrame(context: context, frame: responseFrame, promise: nil)
                 self.wroteAutomaticFrame = true
             }
-        case .priority:
+
+        case .priority(let priorityData):
             result = self.stateMachine.receivePriority()
+
+            // Apply a priority update if one is here. If this fails, it may cause a connection error.
+            do {
+                try self.outboundBuffer.priorityUpdate(streamID: frame.streamID, priorityData: priorityData)
+            } catch {
+                result = StateMachineResultWithEffect(result: .connectionError(underlyingError: error, type: .protocolError), effect: nil)
+            }
+
         case .pushPromise(let pushedStreamData):
             result = self.stateMachine.receivePushPromise(originalStreamID: frame.streamID, childStreamID: pushedStreamData.pushedStreamID, headers: pushedStreamData.headers)
         case .rstStream(let reason):

Sources/NIOHTTP2/HTTP2Error.swift

@@ -184,12 +184,22 @@ public enum NIOHTTP2Errors {
 
     /// A :status header was received with an invalid value.
     public struct InvalidStatusValue: NIOHTTP2Error {
-        var value: String
+        public var value: String
 
         public init(_ value: String) {
             self.value = value
         }
     }
+
+    /// A priority update was received that would create a PRIORITY cycle.
+    public struct PriorityCycle: NIOHTTP2Error {
+        /// The affected stream ID.
+        public var streamID: HTTP2StreamID
+
+        public init(streamID: HTTP2StreamID) {
+            self.streamID = streamID
+        }
+    }
 }
 
 

Sources/NIOHTTP2/OutboundFlowControlBuffer.swift

@@ -176,6 +176,20 @@ internal struct OutboundFlowControlBuffer {
 }
 
 
+// MARK: Priority API
+extension OutboundFlowControlBuffer {
+    /// A frame with new priority data has been received that affects prioritisation of outbound frames.
+    internal mutating func priorityUpdate(streamID: HTTP2StreamID, priorityData: HTTP2Frame.StreamPriorityData) throws {
+        // Right now we don't actually do anything with priority information. However, we do want to police some parts of
+        // RFC 7540 § 5.3, where we can, so this hook is already in place for us to extend later.
+        if streamID == priorityData.dependency {
+            // Streams may not depend on themselves!
+            throw NIOHTTP2Errors.PriorityCycle(streamID: streamID)
+        }
+    }
+}
+
+
 private struct StreamFlowControlState {
     let streamID: HTTP2StreamID
     var currentWindowSize: Int

Sources/NIOHTTP2/OutboundFrameBuffer.swift

@@ -181,6 +181,10 @@ extension CompoundOutboundBuffer {
         self.flowControlBuffer.initialWindowSizeChanged(delta)
     }
 
+    mutating func priorityUpdate(streamID: HTTP2StreamID, priorityData: HTTP2Frame.StreamPriorityData) throws {
+        try self.flowControlBuffer.priorityUpdate(streamID: streamID, priorityData: priorityData)
+    }
+
     var maxFrameSize: Int {
         get {
             return self.flowControlBuffer.maxFrameSize

Tests/NIOHTTP2Tests/CompoundOutboundBufferTest+XCTest.swift

@@ -33,6 +33,7 @@ extension CompoundOutboundBufferTest {
                 ("testFlowControlStreamThrows", testFlowControlStreamThrows),
                 ("testDelayedFlowControlStreamErrors", testDelayedFlowControlStreamErrors),
                 ("testBufferedFrameDrops", testBufferedFrameDrops),
+                ("testRejectsPrioritySelfDependency", testRejectsPrioritySelfDependency),
            ]
    }
 }

Tests/NIOHTTP2Tests/CompoundOutboundBufferTest.swift

@@ -208,6 +208,17 @@ final class CompoundOutboundBufferTest: XCTestCase {
         }
         XCTAssertEqual(results, [true, false, false, false])
     }
+
+    func testRejectsPrioritySelfDependency() {
+        var buffer = CompoundOutboundBuffer(mode: .client, initialMaxOutboundStreams: 1)
+
+        XCTAssertThrowsError(try buffer.priorityUpdate(streamID: 1, priorityData: .init(exclusive: false, dependency: 1, weight: 36))) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+        XCTAssertThrowsError(try buffer.priorityUpdate(streamID: 1, priorityData: .init(exclusive: true, dependency: 1, weight: 36))) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+    }
 }
 
 

Tests/NIOHTTP2Tests/OutboundFlowControlBufferTests+XCTest.swift

@@ -39,6 +39,7 @@ extension OutboundFlowControlBufferTests {
                 ("testOverlargeFramesAreSplitOnMaxFrameSizeFileRegion", testOverlargeFramesAreSplitOnMaxFrameSizeFileRegion),
                 ("testChangingStreamWindowSizeToZeroAndBack", testChangingStreamWindowSizeToZeroAndBack),
                 ("testStreamWindowChanges", testStreamWindowChanges),
+                ("testRejectsPrioritySelfDependency", testRejectsPrioritySelfDependency),
            ]
    }
 }

Tests/NIOHTTP2Tests/OutboundFlowControlBufferTests.swift

@@ -371,6 +371,15 @@ class OutboundFlowControlBufferTests: XCTestCase {
         self.buffer.flushReceived()
         XCTAssertNil(self.buffer.nextFlushedWritableFrame())
     }
+
+    func testRejectsPrioritySelfDependency() {
+        XCTAssertThrowsError(try self.buffer.priorityUpdate(streamID: 1, priorityData: .init(exclusive: false, dependency: 1, weight: 36))) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+        XCTAssertThrowsError(try self.buffer.priorityUpdate(streamID: 1, priorityData: .init(exclusive: true, dependency: 1, weight: 36))) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+    }
 }
 
 

Tests/NIOHTTP2Tests/SimpleClientServerTests+XCTest.swift

@@ -53,6 +53,8 @@ extension SimpleClientServerTests {
                 ("testSettingsAckNotifiesAboutChangedFlowControl", testSettingsAckNotifiesAboutChangedFlowControl),
                 ("testStreamMultiplexerAcknowledgesSettingsBasedFlowControlChanges", testStreamMultiplexerAcknowledgesSettingsBasedFlowControlChanges),
                 ("testChangingMaxFrameSize", testChangingMaxFrameSize),
+                ("testStreamErrorOnSelfDependentPriorityFrames", testStreamErrorOnSelfDependentPriorityFrames),
+                ("testStreamErrorOnSelfDependentHeadersFrames", testStreamErrorOnSelfDependentHeadersFrames),
            ]
    }
 }

Tests/NIOHTTP2Tests/SimpleClientServerTests.swift

@@ -1193,4 +1193,67 @@ class SimpleClientServerTests: XCTestCase {
         XCTAssertNoThrow(try self.clientChannel.finish())
         XCTAssertNoThrow(try self.serverChannel.finish())
     }
+
+    func testStreamErrorOnSelfDependentPriorityFrames() throws {
+        // Begin by getting the connection up.
+        try self.basicHTTP2Connection()
+
+        // Send a PRIORITY frame that has a self-dependency. Note that we aren't policing these outbound:
+        // as we can't detect cycles generally without maintaining a bunch of state, we simply don't. That
+        // allows us to emit this.
+        let frame = HTTP2Frame(streamID: 1, payload: .priority(.init(exclusive: false, dependency: 1, weight: 32)))
+        self.clientChannel.writeAndFlush(frame, promise: nil)
+
+        // We treat this as a connection error.
+        guard let frameData = try assertNoThrowWithValue(self.clientChannel.readOutbound(as: ByteBuffer.self)) else {
+            XCTFail("Did not receive frame")
+            return
+        }
+        XCTAssertThrowsError(try self.serverChannel.writeInbound(frameData)) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+
+        guard let responseFrame = try assertNoThrowWithValue(self.serverChannel.readOutbound(as: ByteBuffer.self)) else {
+            XCTFail("Did not receive response frame")
+            return
+        }
+        XCTAssertNoThrow(try self.clientChannel.writeInbound(responseFrame))
+
+        try self.clientChannel.assertReceivedFrame().assertGoAwayFrame(lastStreamID: .maxID, errorCode: UInt32(http2ErrorCode: .protocolError), opaqueData: nil)
+
+        XCTAssertNoThrow(XCTAssertTrue(try self.clientChannel.finish().isClean))
+        _ = try? self.serverChannel.finish()
+    }
+
+    func testStreamErrorOnSelfDependentHeadersFrames() throws {
+        // Begin by getting the connection up.
+        try self.basicHTTP2Connection()
+
+        // Send a HEADERS frame that has a self-dependency. Note that we aren't policing these outbound:
+        // as we can't detect cycles generally without maintaining a bunch of state, we simply don't. That
+        // allows us to emit this.
+        let headers = HPACKHeaders([(":path", "/"), (":method", "POST"), (":scheme", "https"), (":authority", "localhost")])
+        let frame = HTTP2Frame(streamID: 1, payload: .headers(.init(headers: headers, priorityData: .init(exclusive: true, dependency: 1, weight: 64))))
+        self.clientChannel.writeAndFlush(frame, promise: nil)
+
+        // We treat this as a connection error.
+        guard let frameData = try assertNoThrowWithValue(self.clientChannel.readOutbound(as: ByteBuffer.self)) else {
+            XCTFail("Did not receive frame")
+            return
+        }
+        XCTAssertThrowsError(try self.serverChannel.writeInbound(frameData)) { error in
+            XCTAssertEqual(error as? NIOHTTP2Errors.PriorityCycle, NIOHTTP2Errors.PriorityCycle(streamID: 1))
+        }
+
+        guard let responseFrame = try assertNoThrowWithValue(self.serverChannel.readOutbound(as: ByteBuffer.self)) else {
+            XCTFail("Did not receive response frame")
+            return
+        }
+        XCTAssertNoThrow(try self.clientChannel.writeInbound(responseFrame))
+
+        try self.clientChannel.assertReceivedFrame().assertGoAwayFrame(lastStreamID: .maxID, errorCode: UInt32(http2ErrorCode: .protocolError), opaqueData: nil)
+
+        XCTAssertNoThrow(XCTAssertTrue(try self.clientChannel.finish().isClean))
+        _ = try? self.serverChannel.finish()
+    }
 }