Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_parser_request_2(self):
p = httptools.HttpRequestParser(None)
with self.assertRaises(httptools.HttpParserInvalidMethodError):
p.feed_data(b'SPAM /test.php?a=b+c HTTP/1.2')
def data_received(self, data: bytes):
"""
Called when data is received.
This is the bulk of the processing.
"""
# Feed the data to the parser.
try:
self.parser.feed_data(data)
except httptools.HttpParserInvalidMethodError as e:
ctx = HTTPRequestContext(None, self.app, self.parent_context)
# Transform it into a 405.
exc = exc_from(e)
exc.code = 405
self.loop.create_task(self._safe_handle_error(ctx, exc))
return
except httptools.HttpParserError as e:
ctx = HTTPRequestContext(None, self.app, self.parent_context)
# Transform it into a 400.
exc = exc_from(e)
exc.code = 400
self.loop.create_task(self._safe_handle_error(ctx, exc))
return
# Wait on the event.
if self.waiter is None:
def handle_parser_exception(self, exc: Exception):
"""
Handles an exception when parsing.
This will not call into the app (hence why it is a normal function, and not a coroutine).
It will also close the connection when it's done.
:param exc: The exception to handle.
"""
if isinstance(exc, httptools.HttpParserInvalidMethodError):
# 405 method not allowed
r = MethodNotAllowed()
elif isinstance(exc, httptools.HttpParserError):
# 400 bad request
r = BadRequest()
elif isinstance(exc, httptools.HttpParserUpgrade):
r = BadRequest(description="Invalid upgrade header.")
else:
# internal server error
r = InternalServerError()
# Make a fake environment.
new_environ = to_wsgi_environment(headers=self.headers, method="", path="/",
http_version="1.0", body=None)
new_environ["SERVER_NAME"] = self.component.get_server_name()
new_environ["SERVER_PORT"] = str(self.server_port)