Source code for tornado.http1connection

#!/usr/bin/env python## Copyright 2014 Facebook## Licensed 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."""Client and server implementations of HTTP/1.x... versionadded:: 4.0"""from__future__importabsolute_import,division,print_function,with_statementimportrefromtornado.concurrentimportFuturefromtornado.escapeimportnative_str,utf8fromtornadoimportgenfromtornadoimporthttputilfromtornadoimportiostreamfromtornado.logimportgen_log,app_logfromtornadoimportstack_contextfromtornado.utilimportGzipDecompressor,PY3class_QuietException(Exception):def__init__(self):passclass_ExceptionLoggingContext(object):"""Used with the ``with`` statement when calling delegate methods to log any exceptions with the given logger. Any exceptions caught are converted to _QuietException """def__init__(self,logger):self.logger=loggerdef__enter__(self):passdef__exit__(self,typ,value,tb):ifvalueisnotNone:self.logger.error("Uncaught exception",exc_info=(typ,value,tb))raise_QuietException

[docs]classHTTP1ConnectionParameters(object):"""Parameters for `.HTTP1Connection` and `.HTTP1ServerConnection`. """def__init__(self,no_keep_alive=False,chunk_size=None,max_header_size=None,header_timeout=None,max_body_size=None,body_timeout=None,decompress=False):""" :arg bool no_keep_alive: If true, always close the connection after one request. :arg int chunk_size: how much data to read into memory at once :arg int max_header_size: maximum amount of data for HTTP headers :arg float header_timeout: how long to wait for all headers (seconds) :arg int max_body_size: maximum amount of data for body :arg float body_timeout: how long to wait while reading body (seconds) :arg bool decompress: if true, decode incoming ``Content-Encoding: gzip`` """self.no_keep_alive=no_keep_aliveself.chunk_size=chunk_sizeor65536self.max_header_size=max_header_sizeor65536self.header_timeout=header_timeoutself.max_body_size=max_body_sizeself.body_timeout=body_timeoutself.decompress=decompress

[docs]classHTTP1Connection(httputil.HTTPConnection):"""Implements the HTTP/1.x protocol. This class can be on its own for clients, or via `HTTP1ServerConnection` for servers. """def__init__(self,stream,is_client,params=None,context=None):""" :arg stream: an `.IOStream` :arg bool is_client: client or server :arg params: a `.HTTP1ConnectionParameters` instance or ``None`` :arg context: an opaque application-defined object that can be accessed as ``connection.context``. """self.is_client=is_clientself.stream=streamifparamsisNone:params=HTTP1ConnectionParameters()self.params=paramsself.context=contextself.no_keep_alive=params.no_keep_alive# The body limits can be altered by the delegate, so save them# here instead of just referencing self.params later.self._max_body_size=(self.params.max_body_sizeorself.stream.max_buffer_size)self._body_timeout=self.params.body_timeout# _write_finished is set to True when finish() has been called,# i.e. there will be no more data sent. Data may still be in the# stream's write buffer.self._write_finished=False# True when we have read the entire incoming body.self._read_finished=False# _finish_future resolves when all data has been written and flushed# to the IOStream.self._finish_future=Future()# If true, the connection should be closed after this request# (after the response has been written in the server side,# and after it has been read in the client)self._disconnect_on_finish=Falseself._clear_callbacks()# Save the start lines after we read or write them; they# affect later processing (e.g. 304 responses and HEAD methods# have content-length but no bodies)self._request_start_line=Noneself._response_start_line=Noneself._request_headers=None# True if we are writing output with chunked encoding.self._chunking_output=None# While reading a body with a content-length, this is the# amount left to read.self._expected_content_remaining=None# A Future for our outgoing writes, returned by IOStream.write.self._pending_write=None

[docs]defread_response(self,delegate):"""Read a single HTTP response. Typical client-mode usage is to write a request using `write_headers`, `write`, and `finish`, and then call ``read_response``. :arg delegate: a `.HTTPMessageDelegate` Returns a `.Future` that resolves to None after the full response has been read. """ifself.params.decompress:delegate=_GzipMessageDelegate(delegate,self.params.chunk_size)returnself._read_message(delegate)

@gen.coroutinedef_read_message(self,delegate):need_delegate_close=Falsetry:header_future=self.stream.read_until_regex(b"\r?\n\r?\n",max_bytes=self.params.max_header_size)ifself.params.header_timeoutisNone:header_data=yieldheader_futureelse:try:header_data=yieldgen.with_timeout(self.stream.io_loop.time()+self.params.header_timeout,header_future,io_loop=self.stream.io_loop,quiet_exceptions=iostream.StreamClosedError)exceptgen.TimeoutError:self.close()raisegen.Return(False)start_line,headers=self._parse_headers(header_data)ifself.is_client:start_line=httputil.parse_response_start_line(start_line)self._response_start_line=start_lineelse:start_line=httputil.parse_request_start_line(start_line)self._request_start_line=start_lineself._request_headers=headersself._disconnect_on_finish=notself._can_keep_alive(start_line,headers)need_delegate_close=Truewith_ExceptionLoggingContext(app_log):header_future=delegate.headers_received(start_line,headers)ifheader_futureisnotNone:yieldheader_futureifself.streamisNone:# We've been detached.need_delegate_close=Falseraisegen.Return(False)skip_body=Falseifself.is_client:if(self._request_start_lineisnotNoneandself._request_start_line.method=='HEAD'):skip_body=Truecode=start_line.codeifcode==304:# 304 responses may include the content-length header# but do not actually have a body.# http://tools.ietf.org/html/rfc7230#section-3.3skip_body=Trueifcode>=100andcode<200:# 1xx responses should never indicate the presence of# a body.if('Content-Length'inheadersor'Transfer-Encoding'inheaders):raisehttputil.HTTPInputError("Response code %d cannot have body"%code)# TODO: client delegates will get headers_received twice# in the case of a 100-continue. Document or change?yieldself._read_message(delegate)else:if(headers.get("Expect")=="100-continue"andnotself._write_finished):self.stream.write(b"HTTP/1.1 100 (Continue)\r\n\r\n")ifnotskip_body:body_future=self._read_body(start_line.codeifself.is_clientelse0,headers,delegate)ifbody_futureisnotNone:ifself._body_timeoutisNone:yieldbody_futureelse:try:yieldgen.with_timeout(self.stream.io_loop.time()+self._body_timeout,body_future,self.stream.io_loop,quiet_exceptions=iostream.StreamClosedError)exceptgen.TimeoutError:gen_log.info("Timeout reading body from %s",self.context)self.stream.close()raisegen.Return(False)self._read_finished=Trueifnotself._write_finishedorself.is_client:need_delegate_close=Falsewith_ExceptionLoggingContext(app_log):delegate.finish()# If we're waiting for the application to produce an asynchronous# response, and we're not detached, register a close callback# on the stream (we didn't need one while we were reading)if(notself._finish_future.done()andself.streamisnotNoneandnotself.stream.closed()):self.stream.set_close_callback(self._on_connection_close)yieldself._finish_futureifself.is_clientandself._disconnect_on_finish:self.close()ifself.streamisNone:raisegen.Return(False)excepthttputil.HTTPInputErrorase:gen_log.info("Malformed HTTP message from %s: %s",self.context,e)self.close()raisegen.Return(False)finally:ifneed_delegate_close:with_ExceptionLoggingContext(app_log):delegate.on_connection_close()self._clear_callbacks()raisegen.Return(True)def_clear_callbacks(self):"""Clears the callback attributes. This allows the request handler to be garbage collected more quickly in CPython by breaking up reference cycles. """self._write_callback=Noneself._write_future=Noneself._close_callback=Noneifself.streamisnotNone:self.stream.set_close_callback(None)

[docs]defset_close_callback(self,callback):"""Sets a callback that will be run when the connection is closed. .. deprecated:: 4.0 Use `.HTTPMessageDelegate.on_connection_close` instead. """self._close_callback=stack_context.wrap(callback)

def_on_connection_close(self):# Note that this callback is only registered on the IOStream# when we have finished reading the request and are waiting for# the application to produce its response.ifself._close_callbackisnotNone:callback=self._close_callbackself._close_callback=Nonecallback()ifnotself._finish_future.done():self._finish_future.set_result(None)self._clear_callbacks()defclose(self):ifself.streamisnotNone:self.stream.close()self._clear_callbacks()ifnotself._finish_future.done():self._finish_future.set_result(None)

[docs]defdetach(self):"""Take control of the underlying stream. Returns the underlying `.IOStream` object and stops all further HTTP processing. May only be called during `.HTTPMessageDelegate.headers_received`. Intended for implementing protocols like websockets that tunnel over an HTTP handshake. """self._clear_callbacks()stream=self.streamself.stream=Noneifnotself._finish_future.done():self._finish_future.set_result(None)returnstream

[docs]defset_body_timeout(self,timeout):"""Sets the body timeout for a single request. Overrides the value from `.HTTP1ConnectionParameters`. """self._body_timeout=timeout

[docs]defset_max_body_size(self,max_body_size):"""Sets the body size limit for a single request. Overrides the value from `.HTTP1ConnectionParameters`. """self._max_body_size=max_body_size

[docs]defwrite_headers(self,start_line,headers,chunk=None,callback=None):"""Implements `.HTTPConnection.write_headers`."""lines=[]ifself.is_client:self._request_start_line=start_linelines.append(utf8('%s%s HTTP/1.1'%(start_line[0],start_line[1])))# Client requests with a non-empty body must have either a# Content-Length or a Transfer-Encoding.self._chunking_output=(start_line.methodin('POST','PUT','PATCH')and'Content-Length'notinheadersand'Transfer-Encoding'notinheaders)else:self._response_start_line=start_linelines.append(utf8('HTTP/1.1 %d%s'%(start_line[1],start_line[2])))self._chunking_output=(# TODO: should this use# self._request_start_line.version or# start_line.version?self._request_start_line.version=='HTTP/1.1'and# 304 responses have no body (not even a zero-length body), and so# should not have either Content-Length or Transfer-Encoding.# headers.start_line.codenotin(204,304)and# No need to chunk the output if a Content-Length is specified.'Content-Length'notinheadersand# Applications are discouraged from touching Transfer-Encoding,# but if they do, leave it alone.'Transfer-Encoding'notinheaders)# If a 1.0 client asked for keep-alive, add the header.if(self._request_start_line.version=='HTTP/1.0'and(self._request_headers.get('Connection','').lower()=='keep-alive')):headers['Connection']='Keep-Alive'ifself._chunking_output:headers['Transfer-Encoding']='chunked'if(notself.is_clientand(self._request_start_line.method=='HEAD'orstart_line.code==304)):self._expected_content_remaining=0elif'Content-Length'inheaders:self._expected_content_remaining=int(headers['Content-Length'])else:self._expected_content_remaining=None# TODO: headers are supposed to be of type str, but we still have some# cases that let bytes slip through. Remove these native_str calls when those# are fixed.header_lines=(native_str(n)+": "+native_str(v)forn,vinheaders.get_all())ifPY3:lines.extend(l.encode('latin1')forlinheader_lines)else:lines.extend(header_lines)forlineinlines:ifb'\n'inline:raiseValueError('Newline in header: '+repr(line))future=Noneifself.stream.closed():future=self._write_future=Future()future.set_exception(iostream.StreamClosedError())future.exception()else:ifcallbackisnotNone:self._write_callback=stack_context.wrap(callback)else:future=self._write_future=Future()data=b"\r\n".join(lines)+b"\r\n\r\n"ifchunk:data+=self._format_chunk(chunk)self._pending_write=self.stream.write(data)self._pending_write.add_done_callback(self._on_write_complete)returnfuture

def_format_chunk(self,chunk):ifself._expected_content_remainingisnotNone:self._expected_content_remaining-=len(chunk)ifself._expected_content_remaining<0:# Close the stream now to stop further framing errors.self.stream.close()raisehttputil.HTTPOutputError("Tried to write more data than Content-Length")ifself._chunking_outputandchunk:# Don't write out empty chunks because that means END-OF-STREAM# with chunked encodingreturnutf8("%x"%len(chunk))+b"\r\n"+chunk+b"\r\n"else:returnchunk

[docs]defwrite(self,chunk,callback=None):"""Implements `.HTTPConnection.write`. For backwards compatibility is is allowed but deprecated to skip `write_headers` and instead call `write()` with a pre-encoded header block. """future=Noneifself.stream.closed():future=self._write_future=Future()self._write_future.set_exception(iostream.StreamClosedError())self._write_future.exception()else:ifcallbackisnotNone:self._write_callback=stack_context.wrap(callback)else:future=self._write_future=Future()self._pending_write=self.stream.write(self._format_chunk(chunk))self._pending_write.add_done_callback(self._on_write_complete)returnfuture

[docs]deffinish(self):"""Implements `.HTTPConnection.finish`."""if(self._expected_content_remainingisnotNoneandself._expected_content_remaining!=0andnotself.stream.closed()):self.stream.close()raisehttputil.HTTPOutputError("Tried to write %d bytes less than Content-Length"%self._expected_content_remaining)ifself._chunking_output:ifnotself.stream.closed():self._pending_write=self.stream.write(b"0\r\n\r\n")self._pending_write.add_done_callback(self._on_write_complete)self._write_finished=True# If the app finished the request while we're still reading,# divert any remaining data away from the delegate and# close the connection when we're done sending our response.# Closing the connection is the only way to avoid reading the# whole input body.ifnotself._read_finished:self._disconnect_on_finish=True# No more data is coming, so instruct TCP to send any remaining# data immediately instead of waiting for a full packet or ack.self.stream.set_nodelay(True)ifself._pending_writeisNone:self._finish_request(None)else:self._pending_write.add_done_callback(self._finish_request)

def_on_write_complete(self,future):exc=future.exception()ifexcisnotNoneandnotisinstance(exc,iostream.StreamClosedError):future.result()ifself._write_callbackisnotNone:callback=self._write_callbackself._write_callback=Noneself.stream.io_loop.add_callback(callback)ifself._write_futureisnotNone:future=self._write_futureself._write_future=Nonefuture.set_result(None)def_can_keep_alive(self,start_line,headers):ifself.params.no_keep_alive:returnFalseconnection_header=headers.get("Connection")ifconnection_headerisnotNone:connection_header=connection_header.lower()ifstart_line.version=="HTTP/1.1":returnconnection_header!="close"elif("Content-Length"inheadersorheaders.get("Transfer-Encoding","").lower()=="chunked"orgetattr(start_line,'method',None)in("HEAD","GET")):# start_line may be a request or reponse start line; only# the former has a method attribute.returnconnection_header=="keep-alive"returnFalsedef_finish_request(self,future):self._clear_callbacks()ifnotself.is_clientandself._disconnect_on_finish:self.close()return# Turn Nagle's algorithm back on, leaving the stream in its# default state for the next request.self.stream.set_nodelay(False)ifnotself._finish_future.done():self._finish_future.set_result(None)def_parse_headers(self,data):# The lstrip removes newlines that some implementations sometimes# insert between messages of a reused connection. Per RFC 7230,# we SHOULD ignore at least one empty line before the request.# http://tools.ietf.org/html/rfc7230#section-3.5data=native_str(data.decode('latin1')).lstrip("\r\n")# RFC 7230 section allows for both CRLF and bare LF.eol=data.find("\n")start_line=data[:eol].rstrip("\r")try:headers=httputil.HTTPHeaders.parse(data[eol:])exceptValueError:# probably form split() if there was no ':' in the lineraisehttputil.HTTPInputError("Malformed HTTP headers: %r"%data[eol:100])returnstart_line,headersdef_read_body(self,code,headers,delegate):if"Content-Length"inheaders:if"Transfer-Encoding"inheaders:# Response cannot contain both Content-Length and# Transfer-Encoding headers.# http://tools.ietf.org/html/rfc7230#section-3.3.3raisehttputil.HTTPInputError("Response with both Transfer-Encoding and Content-Length")if","inheaders["Content-Length"]:# Proxies sometimes cause Content-Length headers to get# duplicated. If all the values are identical then we can# use them but if they differ it's an error.pieces=re.split(r',\s*',headers["Content-Length"])ifany(i!=pieces[0]foriinpieces):raisehttputil.HTTPInputError("Multiple unequal Content-Lengths: %r"%headers["Content-Length"])headers["Content-Length"]=pieces[0]try:content_length=int(headers["Content-Length"])exceptValueError:# Handles non-integer Content-Length value.raisehttputil.HTTPInputError("Only integer Content-Length is allowed: %s"%headers["Content-Length"])ifcontent_length>self._max_body_size:raisehttputil.HTTPInputError("Content-Length too long")else:content_length=Noneifcode==204:# This response code is not allowed to have a non-empty body,# and has an implicit length of zero instead of read-until-close.# http://www.w3.org/Protocols/rfc2616/rfc2616-sec4.html#sec4.3if("Transfer-Encoding"inheadersorcontent_lengthnotin(None,0)):raisehttputil.HTTPInputError("Response with code %d should not have body"%code)content_length=0ifcontent_lengthisnotNone:returnself._read_fixed_body(content_length,delegate)ifheaders.get("Transfer-Encoding","").lower()=="chunked":returnself._read_chunked_body(delegate)ifself.is_client:returnself._read_body_until_close(delegate)returnNone@gen.coroutinedef_read_fixed_body(self,content_length,delegate):whilecontent_length>0:body=yieldself.stream.read_bytes(min(self.params.chunk_size,content_length),partial=True)content_length-=len(body)ifnotself._write_finishedorself.is_client:with_ExceptionLoggingContext(app_log):ret=delegate.data_received(body)ifretisnotNone:yieldret@gen.coroutinedef_read_chunked_body(self,delegate):# TODO: "chunk extensions" http://tools.ietf.org/html/rfc2616#section-3.6.1total_size=0whileTrue:chunk_len=yieldself.stream.read_until(b"\r\n",max_bytes=64)chunk_len=int(chunk_len.strip(),16)ifchunk_len==0:returntotal_size+=chunk_leniftotal_size>self._max_body_size:raisehttputil.HTTPInputError("chunked body too large")bytes_to_read=chunk_lenwhilebytes_to_read:chunk=yieldself.stream.read_bytes(min(bytes_to_read,self.params.chunk_size),partial=True)bytes_to_read-=len(chunk)ifnotself._write_finishedorself.is_client:with_ExceptionLoggingContext(app_log):ret=delegate.data_received(chunk)ifretisnotNone:yieldret# chunk ends with \r\ncrlf=yieldself.stream.read_bytes(2)assertcrlf==b"\r\n"@gen.coroutinedef_read_body_until_close(self,delegate):body=yieldself.stream.read_until_close()ifnotself._write_finishedorself.is_client:with_ExceptionLoggingContext(app_log):delegate.data_received(body)

class_GzipMessageDelegate(httputil.HTTPMessageDelegate):"""Wraps an `HTTPMessageDelegate` to decode ``Content-Encoding: gzip``. """def__init__(self,delegate,chunk_size):self._delegate=delegateself._chunk_size=chunk_sizeself._decompressor=Nonedefheaders_received(self,start_line,headers):ifheaders.get("Content-Encoding")=="gzip":self._decompressor=GzipDecompressor()# Downstream delegates will only see uncompressed data,# so rename the content-encoding header.# (but note that curl_httpclient doesn't do this).headers.add("X-Consumed-Content-Encoding",headers["Content-Encoding"])delheaders["Content-Encoding"]returnself._delegate.headers_received(start_line,headers)@gen.coroutinedefdata_received(self,chunk):ifself._decompressor:compressed_data=chunkwhilecompressed_data:decompressed=self._decompressor.decompress(compressed_data,self._chunk_size)ifdecompressed:ret=self._delegate.data_received(decompressed)ifretisnotNone:yieldretcompressed_data=self._decompressor.unconsumed_tailelse:ret=self._delegate.data_received(chunk)ifretisnotNone:yieldretdeffinish(self):ifself._decompressorisnotNone:tail=self._decompressor.flush()iftail:# I believe the tail will always be empty (i.e.# decompress will return all it can). The purpose# of the flush call is to detect errors such# as truncated input. But in case it ever returns# anything, treat it as an extra chunkself._delegate.data_received(tail)returnself._delegate.finish()defon_connection_close(self):returnself._delegate.on_connection_close()

[docs]defclose(self):"""Closes the connection. Returns a `.Future` that resolves after the serving loop has exited. """self.stream.close()# Block until the serving loop is done, but ignore any exceptions# (start_serving is already responsible for logging them).try:yieldself._serving_futureexceptException:pass

[docs]defstart_serving(self,delegate):"""Starts serving requests on this connection. :arg delegate: a `.HTTPServerConnectionDelegate` """assertisinstance(delegate,httputil.HTTPServerConnectionDelegate)self._serving_future=self._server_request_loop(delegate)# Register the future on the IOLoop so its errors get logged.self.stream.io_loop.add_future(self._serving_future,lambdaf:f.result())