http11.rb in async-http-0.25.0

- old
+ new

@@ -19,10 +19,11 @@
 # THE SOFTWARE.
 
 require 'async/io/protocol/line'
 
 require_relative 'request_failed'
+require_relative 'bad_request'
 
 require_relative '../request'
 require_relative '../response'
 require_relative '../headers'
 
@@ -30,10 +31,14 @@
 require_relative '../body/fixed'
 
 module Async
 	module HTTP
 		module Protocol
+			TRANSFER_ENCODING = 'transfer-encoding'.freeze
+			CONTENT_LENGTH = 'content-length'.freeze
+			CHUNKED = 'chunked'.freeze
+			
 			# Implements basic HTTP/1.1 request/response.
 			class HTTP11 < Async::IO::Protocol::Line
 				CRLF = "\r\n".freeze
 				CONNECTION = 'connection'.freeze
 				HOST = 'host'.freeze
@@ -78,29 +83,66 @@
 					else
 						return true
 					end
 				end
 				
+				def hijack
+					@persistent = false
+					
+					@stream.flush
+					
+					return @stream.io
+				end
+				
+				class Request < HTTP::Request
+					def initialize(protocol)
+						super(*protocol.read_request)
+						
+						@protocol = protocol
+					end
+					
+					attr :protocol
+					
+					def hijack?
+						true
+					end
+					
+					def hijack
+						@protocol.hijack
+					end
+				end
+				
+				def next_request
+					# The default is true.
+					return nil unless @persistent
+					
+					request = Request.new(self)
+					
+					unless persistent?(request.headers)
+						@persistent = false
+					end
+					
+					return request
+				rescue
+					# Bad Request
+					write_response(self.version, 400, {}, nil)
+					
+					raise
+				end
+				
 				# Server loop.
 				def receive_requests(task: Task.current)
-					while @persistent
-						request = Request.new(*read_request)
-						
-						unless persistent?(request.headers)
-							@persistent = false
+					while request = next_request
+						if response = yield(request, self)
+							write_response(response.version || self.version, response.status, response.headers, response.body)
+							request.finish
+							
+							# This ensures we yield at least once every iteration of the loop and allow other fibers to execute.
+							task.yield
+						else
+							break
 						end
-						
-						response = yield request
-						
-						response.version ||= request.version
-						
-						write_response(response.version, response.status, response.headers, response.body)
-						
-						request.finish
-						
-						# This ensures we yield at least once every iteration of the loop and allow other fibers to execute.
-						task.yield
 					end
 				end
 				
 				def call(request)
 					request.version ||= self.version
@@ -116,43 +158,49 @@
 					end
 					
 					# Once we start writing the body, we can't recover if the request fails. That's because the body might be generated dynamically, streaming, etc.
 					write_body(request.body)
 					
-					return Response.new(*read_response)
+					return Response.new(*read_response(request))
 				rescue
 					# This will ensure that #reusable? returns false.
 					@stream.close
 					
 					raise
 				end
 				
 				def write_request(authority, method, path, version, headers)
 					@stream.write("#{method} #{path} #{version}\r\n")
-					@stream.write("Host: #{authority}\r\n")
+					@stream.write("host: #{authority}\r\n")
 					write_headers(headers)
 					
 					@stream.flush
 				end
 				
-				def read_response
+				def read_response(request)
 					version, status, reason = read_line.split(/\s+/, 3)
+					Async.logger.debug(self) {"#{version} #{status} #{reason}"}
+					
 					headers = read_headers
-					body = read_body(headers)
 					
-					@count += 1
-					
 					@persistent = persistent?(headers)
 					
+					body = read_response_body(request, status, headers)
+					
+					@count += 1
+					
 					return version, Integer(status), reason, headers, body
 				end
 				
 				def read_request
 					method, path, version = read_line.split(/\s+/, 3)
 					headers = read_headers
-					body = read_body(headers)
 					
+					@persistent = persistent?(headers)
+					
+					body = read_request_body(headers)
+					
 					@count += 1
 					
 					return headers.delete(HOST), method, path, version, headers, body
 				end
 				
@@ -165,19 +213,17 @@
 				end
 				
 				protected
 				
 				def write_persistent_header
-					@stream.write("Connection: close\r\n") unless @persistent
+					@stream.write("connection: close\r\n") unless @persistent
 				end
 				
 				def write_headers(headers)
 					headers.each do |name, value|
 						@stream.write("#{name}: #{value}\r\n")
 					end
-					
-					write_persistent_header
 				end
 				
 				def read_headers
 					fields = []
 					
@@ -190,63 +236,157 @@
 					end
 					
 					return Headers.new(fields)
 				end
 				
+				def write_empty_body(body)
+					# Write empty body:
+					write_persistent_header
+					@stream.write("content-length: 0\r\n\r\n")
+					
+					body.read if body
+					
+					@stream.flush
+				end
+				
+				def write_fixed_length_body(body, length)
+					write_persistent_header
+					@stream.write("content-length: #{length}\r\n\r\n")
+					
+					body.each do |chunk|
+						@stream.write(chunk)
+					end
+					
+					@stream.flush
+				end
+				
+				def write_chunked_body(body)
+					write_persistent_header
+					@stream.write("transfer-encoding: chunked\r\n\r\n")
+					
+					body.each do |chunk|
+						next if chunk.size == 0
+						
+						@stream.write("#{chunk.bytesize.to_s(16).upcase}\r\n")
+						@stream.write(chunk)
+						@stream.write(CRLF)
+						@stream.flush
+					end
+					
+					@stream.write("0\r\n\r\n")
+					@stream.flush
+				end
+				
+				def write_body_and_close(body)
+					# We can't be persistent because we don't know the data length:
+					@persistent = false
+					write_persistent_header
+					
+					@stream.write("\r\n")
+					
+					body.each do |chunk|
+						@stream.write(chunk)
+						@stream.flush
+					end
+					
+					@stream.io.close_write
+				end
+				
 				def write_body(body, chunked = true)
 					if body.nil? or body.empty?
-						@stream.write("Content-Length: 0\r\n\r\n")
-						body.read if body
+						write_empty_body(body)
 					elsif length = body.length
-						@stream.write("Content-Length: #{length}\r\n\r\n")
-						
-						body.each do |chunk|
-							@stream.write(chunk)
-						end
+						write_fixed_length_body(body, length)
 					elsif chunked
-						@stream.write("Transfer-Encoding: chunked\r\n\r\n")
-						
-						body.each do |chunk|
-							next if chunk.size == 0
-							
-							@stream.write("#{chunk.bytesize.to_s(16).upcase}\r\n")
-							@stream.write(chunk)
-							@stream.write(CRLF)
-							@stream.flush
-						end
-						
-						@stream.write("0\r\n\r\n")
+						write_chunked_body(body)
 					else
-						body = Body::Buffered.for(body)
-						
-						@stream.write("Content-Length: #{body.bytesize}\r\n\r\n")
-						
-						body.each do |chunk|
-							@stream.write(chunk)
-						end
+						write_body_and_close(body)
 					end
+				end
+				
+				def read_response_body(request, status, headers)
+					# RFC 7230 3.3.3
+					# 1.  Any response to a HEAD request and any response with a 1xx
+					# (Informational), 204 (No Content), or 304 (Not Modified) status
+					# code is always terminated by the first empty line after the
+					# header fields, regardless of the header fields present in the
+					# message, and thus cannot contain a message body.
+					if request.head? or status == 204 or status == 304
+						return nil
+					end
 					
-					@stream.flush
+					# 2.  Any 2xx (Successful) response to a CONNECT request implies that
+					# the connection will become a tunnel immediately after the empty
+					# line that concludes the header fields.  A client MUST ignore any
+					# Content-Length or Transfer-Encoding header fields received in
+					# such a message.
+					if request.connect? and status == 200
+						return Body::Remainder.new(@stream)
+					end
+					
+					if body = read_body(headers)
+						return body
+					else
+						# 7.  Otherwise, this is a response message without a declared message
+						# body length, so the message body length is determined by the
+						# number of octets received prior to the server closing the
+						# connection.
+						return Body::Remainder.new(@stream)
+					end
 				end
 				
-				TRANSFER_ENCODING = 'transfer-encoding'.freeze
-				CONTENT_LENGTH = 'content-length'.freeze
-				CHUNKED = 'chunked'.freeze
-				
-				def chunked?(headers)
-					if transfer_encoding = headers[TRANSFER_ENCODING]
-						if transfer_encoding.count == 1
-							return transfer_encoding.first == CHUNKED
-						end
+				def read_request_body(headers)
+					# 6.  If this is a request message and none of the above are true, then
+					# the message body length is zero (no message body is present).
+					if body = read_body(headers)
+						return body
 					end
 				end
 				
 				def read_body(headers)
-					if chunked?(headers)
-						return Body::Chunked.new(self)
-					elsif content_length = headers[CONTENT_LENGTH]
-						if content_length != 0
-							return Body::Fixed.new(@stream, Integer(content_length))
+					# 3.  If a Transfer-Encoding header field is present and the chunked
+					# transfer coding (Section 4.1) is the final encoding, the message
+					# body length is determined by reading and decoding the chunked
+					# data until the transfer coding indicates the data is complete.
+					if transfer_encoding = headers[TRANSFER_ENCODING]
+						# If a message is received with both a Transfer-Encoding and a
+						# Content-Length header field, the Transfer-Encoding overrides the
+						# Content-Length.  Such a message might indicate an attempt to
+						# perform request smuggling (Section 9.5) or response splitting
+						# (Section 9.4) and ought to be handled as an error.  A sender MUST
+						# remove the received Content-Length field prior to forwarding such
+						# a message downstream.
+						if headers[CONTENT_LENGTH]
+							raise BadRequest, "Message contains both transfer encoding and content length!"
+						end
+						
+						if transfer_encoding.last == CHUNKED
+							return Body::Chunked.new(self)
+						else
+							# If a Transfer-Encoding header field is present in a response and
+							# the chunked transfer coding is not the final encoding, the
+							# message body length is determined by reading the connection until
+							# it is closed by the server.  If a Transfer-Encoding header field
+							# is present in a request and the chunked transfer coding is not
+							# the final encoding, the message body length cannot be determined
+							# reliably; the server MUST respond with the 400 (Bad Request)
+							# status code and then close the connection.
+							return Body::Remainder.new(@stream)
+						end
+					end
+
+					# 5.  If a valid Content-Length header field is present without
+					# Transfer-Encoding, its decimal value defines the expected message
+					# body length in octets.  If the sender closes the connection or
+					# the recipient times out before the indicated number of octets are
+					# received, the recipient MUST consider the message to be
+					# incomplete and close the connection.
+					if content_length = headers[CONTENT_LENGTH]
+						length = Integer(content_length)
+						if length >= 0
+							return Body::Fixed.new(@stream, length)
+						else
+							raise BadRequest, "Invalid content length: #{content_length}"
 						end
 					end
 				end
 			end
 		end