[HIGH] Bound peer-created stream state

lib/http/2/connection.rb

@@ -90,7 +90,9 @@ def initialize(settings = {})
       @last_stream_id = 0
       @streams = {}
       @streams_recently_closed = {}
+      @idle_stream_priorities = {}
       @oldest_stream_recently_closed = nil
+      @last_promised_stream_id = 0
       @pending_settings = []
 
       @framer = Framer.new(@local_settings[:settings_max_frame_size])
@@ -316,11 +318,19 @@ def receive(data)
               verify_pseudo_headers(frame)
 
               verify_stream_order(stream_id)
+              priority = @idle_stream_priorities.delete(stream_id) || {}
+              if frame[:flags].anybits?(PRIORITY)
+                priority = {
+                  weight: frame[:weight],
+                  dependency: frame[:dependency],
+                  exclusive: frame[:exclusive]
+                }
+              end
               stream = activate_stream(
                 id: stream_id,
-                weight: frame[:weight] || DEFAULT_WEIGHT,
-                dependency: frame[:dependency] || 0,
-                exclusive: frame[:exclusive] || false
+                weight: priority.fetch(:weight, DEFAULT_WEIGHT),
+                dependency: priority.fetch(:dependency, 0),
+                exclusive: priority.fetch(:exclusive, false)
               )
               emit(:stream, stream)
             end
@@ -352,6 +362,8 @@ def receive(data)
             parent = @streams[stream_id]
             pid = frame[:promise_stream]
 
+            validate_push_promise(pid)
+
             # if PUSH parent is recently closed, RST_STREAM the push
             if @streams_recently_closed[stream_id]
               send(type: :rst_stream, stream: pid, error: :refused_stream)
@@ -377,6 +389,13 @@ def receive(data)
 
             _verify_pseudo_headers(frame, REQUEST_MANDATORY_HEADERS)
             verify_stream_order(pid)
+
+            promised_stream_count = @streams.each_value.count(&:parent)
+            if promised_stream_count >= @local_settings[:settings_max_concurrent_streams]
+              send(type: :rst_stream, stream: pid, error: :refused_stream)
+              next
+            end
+
             stream = activate_stream(id: pid, parent: parent)
             emit(:promise, stream)
             stream << frame
@@ -399,12 +418,17 @@ def receive(data)
                 # the draining connection open.
                 next if closed? && stream_id <= @last_stream_id
 
-                stream = activate_stream(
-                  id: stream_id,
+                unless @idle_stream_priorities.key?(stream_id)
+                  next if @idle_stream_priorities.size >= @local_settings[:settings_max_concurrent_streams]
+                end
+
+                priority = {
                   weight: frame[:weight] || DEFAULT_WEIGHT,
                   dependency: frame[:dependency] || 0,
                   exclusive: frame[:exclusive] || false
-                )
+                }
+                @idle_stream_priorities[stream_id] = priority
+                stream = Stream.new(connection: self, id: stream_id, **priority)
 
                 emit(:stream, stream)
                 stream << frame
@@ -577,6 +601,18 @@ def ping_management(frame)
       end
     end
 
+    def validate_push_promise(stream_id)
+      connection_error(:protocol_error, msg: "clients cannot send PUSH_PROMISE") if @local_role == :server
+      if @local_settings[:settings_enable_push].zero?
+        connection_error(:protocol_error, msg: "received PUSH_PROMISE while push is disabled")
+      end
+      unless stream_id.positive? && stream_id.even? && stream_id > @last_promised_stream_id
+        connection_error(:protocol_error, msg: "invalid promised stream ID")
+      end
+
+      @last_promised_stream_id = stream_id
+    end
+
     # Validate settings parameters.  See sepc Section 6.5.2.
     #
     # @param role [Symbol] The sender's role: :client or :server

sig/connection.rbs

@@ -33,7 +33,9 @@ module HTTP2
 
     @streams: Hash[Integer, Stream]
     @streams_recently_closed: Hash[Integer, Numeric]
+    @idle_stream_priorities: Hash[Integer, Hash[Symbol, Integer | bool]]
     @oldest_stream_recently_closed: Numeric
+    @last_promised_stream_id: Integer
 
     @framer: Framer
 
@@ -91,6 +93,8 @@ module HTTP2
 
     def ping_management: (frame) -> void
 
+    def validate_push_promise: (Integer stream_id) -> void
+
     def validate_settings: (role_type, settings_enum) -> void
 
     def connection_settings: (settings_frame) -> void

spec/server_spec.rb

@@ -153,6 +153,19 @@
       expect(stream.weight).to eq 20
     end
 
+    it "should preserve updated PRIORITY values when an idle stream opens" do
+      srv << CONNECTION_PREFACE_MAGIC
+      srv << f.generate(settings_frame)
+
+      streams = []
+      srv.on(:stream) { |stream| streams << stream }
+      srv << f.generate(priority_frame.merge(stream: 1, weight: 20))
+      srv << f.generate(priority_frame.merge(stream: 1, weight: 30))
+      srv << f.generate(headers_frame)
+
+      expect(streams.last.weight).to eq 30
+    end
+
     it "should process connection management frames after GOAWAY" do
       srv << CONNECTION_PREFACE_MAGIC
       srv << f.generate(settings_frame)