wsgi.py 7.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197
  1. import sys
  2. from collections import defaultdict
  3. from tempfile import SpooledTemporaryFile
  4. from asgiref.sync import AsyncToSync, sync_to_async
  5. class WsgiToAsgi:
  6. """
  7. Wraps a WSGI application to make it into an ASGI application.
  8. """
  9. def __init__(self, wsgi_application, duplicate_header_limit=100):
  10. self.wsgi_application = wsgi_application
  11. self.duplicate_header_limit = duplicate_header_limit
  12. async def __call__(self, scope, receive, send):
  13. """
  14. ASGI application instantiation point.
  15. We return a new WsgiToAsgiInstance here with the WSGI app
  16. and the scope, ready to respond when it is __call__ed.
  17. """
  18. await WsgiToAsgiInstance(self.wsgi_application, self.duplicate_header_limit)(
  19. scope, receive, send
  20. )
  21. class WsgiToAsgiInstance:
  22. """
  23. Per-socket instance of a wrapped WSGI application
  24. """
  25. def __init__(self, wsgi_application, duplicate_header_limit=100):
  26. self.wsgi_application = wsgi_application
  27. self.duplicate_header_limit = duplicate_header_limit
  28. self.response_started = False
  29. self.response_content_length = None
  30. async def __call__(self, scope, receive, send):
  31. if scope["type"] != "http":
  32. raise ValueError("WSGI wrapper received a non-HTTP scope")
  33. self.scope = scope
  34. with SpooledTemporaryFile(max_size=65536) as body:
  35. # Alright, wait for the http.request messages
  36. while True:
  37. message = await receive()
  38. if message["type"] != "http.request":
  39. raise ValueError("WSGI wrapper received a non-HTTP-request message")
  40. body.write(message.get("body", b""))
  41. if not message.get("more_body"):
  42. break
  43. body.seek(0)
  44. # Wrap send so it can be called from the subthread
  45. self.sync_send = AsyncToSync(send)
  46. # Call the WSGI app
  47. await self.run_wsgi_app(body)
  48. def build_environ(self, scope, body):
  49. """
  50. Builds a scope and request body into a WSGI environ object.
  51. """
  52. script_name = scope.get("root_path", "").encode("utf8").decode("latin1")
  53. path_info = scope["path"].encode("utf8").decode("latin1")
  54. if path_info.startswith(script_name):
  55. path_info = path_info[len(script_name) :]
  56. environ = {
  57. "REQUEST_METHOD": scope["method"],
  58. "SCRIPT_NAME": script_name,
  59. "PATH_INFO": path_info,
  60. "QUERY_STRING": scope["query_string"].decode("ascii"),
  61. "SERVER_PROTOCOL": "HTTP/%s" % scope["http_version"],
  62. "wsgi.version": (1, 0),
  63. "wsgi.url_scheme": scope.get("scheme", "http"),
  64. "wsgi.input": body,
  65. "wsgi.errors": sys.stderr,
  66. "wsgi.multithread": True,
  67. "wsgi.multiprocess": True,
  68. "wsgi.run_once": False,
  69. }
  70. # Get server name and port - required in WSGI, not in ASGI
  71. if "server" in scope:
  72. environ["SERVER_NAME"] = scope["server"][0]
  73. environ["SERVER_PORT"] = str(scope["server"][1])
  74. else:
  75. environ["SERVER_NAME"] = "localhost"
  76. environ["SERVER_PORT"] = "80"
  77. if scope.get("client") is not None:
  78. environ["REMOTE_ADDR"] = scope["client"][0]
  79. # Go through headers and make them into environ entries
  80. _headers = defaultdict(list)
  81. for name, value in self.scope.get("headers", []):
  82. name = name.decode("latin1")
  83. if name == "content-length":
  84. corrected_name = "CONTENT_LENGTH"
  85. elif name == "content-type":
  86. corrected_name = "CONTENT_TYPE"
  87. else:
  88. corrected_name = "HTTP_%s" % name.upper().replace("-", "_")
  89. # HTTPbis say only ASCII chars are allowed in headers, but we latin1 just in case
  90. value = value.decode("latin1")
  91. if (
  92. self.duplicate_header_limit
  93. and len(_headers[corrected_name]) >= self.duplicate_header_limit
  94. ):
  95. raise ValueError(
  96. f"Too many duplicate headers: {corrected_name} exceeds limit of"
  97. f"{self.duplicate_header_limit}"
  98. )
  99. _headers[corrected_name].append(value)
  100. for name, values in _headers.items():
  101. environ[name] = ",".join(values)
  102. return environ
  103. def start_response(self, status, response_headers, exc_info=None):
  104. """
  105. WSGI start_response callable.
  106. """
  107. # Don't allow re-calling once response has begun
  108. if self.response_started:
  109. raise exc_info[1].with_traceback(exc_info[2])
  110. # Don't allow re-calling without exc_info
  111. if hasattr(self, "response_start") and exc_info is None:
  112. raise ValueError(
  113. "You cannot call start_response a second time without exc_info"
  114. )
  115. # Extract status code
  116. status_code, _ = status.split(" ", 1)
  117. status_code = int(status_code)
  118. # Extract headers
  119. headers = [
  120. (name.lower().encode("ascii"), value.encode("ascii"))
  121. for name, value in response_headers
  122. ]
  123. # Extract content-length
  124. self.response_content_length = None
  125. for name, value in response_headers:
  126. if name.lower() == "content-length":
  127. self.response_content_length = int(value)
  128. # Build and send response start message.
  129. self.response_start = {
  130. "type": "http.response.start",
  131. "status": status_code,
  132. "headers": headers,
  133. }
  134. @sync_to_async
  135. def run_wsgi_app(self, body):
  136. """
  137. Called in a subthread to run the WSGI app. We encapsulate like
  138. this so that the start_response callable is called in the same thread.
  139. """
  140. # Translate the scope and incoming request body into a WSGI environ
  141. try:
  142. environ = self.build_environ(self.scope, body)
  143. except ValueError:
  144. # Return 400 Bad Request if header limit exceeded
  145. self.sync_send(
  146. {
  147. "type": "http.response.start",
  148. "status": 400,
  149. "headers": [(b"content-type", b"text/plain")],
  150. }
  151. )
  152. self.sync_send(
  153. {
  154. "type": "http.response.body",
  155. "body": b"Bad Request: Too many duplicate headers",
  156. }
  157. )
  158. return
  159. # Run the WSGI app
  160. bytes_sent = 0
  161. for output in self.wsgi_application(environ, self.start_response):
  162. # If this is the first response, include the response headers
  163. if not self.response_started:
  164. self.response_started = True
  165. self.sync_send(self.response_start)
  166. # If the application supplies a Content-Length header
  167. if self.response_content_length is not None:
  168. # The server should not transmit more bytes to the client than the header allows
  169. bytes_allowed = self.response_content_length - bytes_sent
  170. if len(output) > bytes_allowed:
  171. output = output[:bytes_allowed]
  172. self.sync_send(
  173. {"type": "http.response.body", "body": output, "more_body": True}
  174. )
  175. bytes_sent += len(output)
  176. # The server should stop iterating over the response when enough data has been sent
  177. if bytes_sent == self.response_content_length:
  178. break
  179. # Close connection
  180. if not self.response_started:
  181. self.response_started = True
  182. self.sync_send(self.response_start)
  183. self.sync_send({"type": "http.response.body"})