208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
from io import IOBase
|
|
|
|
from django.conf import settings
|
|
from django.core import signals
|
|
from django.core.handlers import base
|
|
from django.http import HttpRequest, QueryDict, parse_cookie
|
|
from django.urls import set_script_prefix
|
|
from django.utils.encoding import repercent_broken_unicode
|
|
from django.utils.functional import cached_property
|
|
from django.utils.regex_helper import _lazy_re_compile
|
|
|
|
_slashes_re = _lazy_re_compile(rb"/+")
|
|
|
|
|
|
class LimitedStream(IOBase):
|
|
"""
|
|
Wrap another stream to disallow reading it past a number of bytes.
|
|
|
|
Based on the implementation from werkzeug.wsgi.LimitedStream
|
|
See https://github.com/pallets/werkzeug/blob/dbf78f67/src/werkzeug/wsgi.py#L828
|
|
"""
|
|
|
|
def __init__(self, stream, limit):
|
|
self._read = stream.read
|
|
self._readline = stream.readline
|
|
self._pos = 0
|
|
self.limit = limit
|
|
|
|
def read(self, size=-1, /):
|
|
_pos = self._pos
|
|
limit = self.limit
|
|
if _pos >= limit:
|
|
return b""
|
|
if size == -1 or size is None:
|
|
size = limit - _pos
|
|
else:
|
|
size = min(size, limit - _pos)
|
|
data = self._read(size)
|
|
self._pos += len(data)
|
|
return data
|
|
|
|
def readline(self, size=-1, /):
|
|
_pos = self._pos
|
|
limit = self.limit
|
|
if _pos >= limit:
|
|
return b""
|
|
if size == -1 or size is None:
|
|
size = limit - _pos
|
|
else:
|
|
size = min(size, limit - _pos)
|
|
line = self._readline(size)
|
|
self._pos += len(line)
|
|
return line
|
|
|
|
|
|
class WSGIRequest(HttpRequest):
|
|
def __init__(self, environ):
|
|
script_name = get_script_name(environ)
|
|
# If PATH_INFO is empty (e.g. accessing the SCRIPT_NAME URL without a
|
|
# trailing slash), operate as if '/' was requested.
|
|
path_info = get_path_info(environ) or "/"
|
|
self.environ = environ
|
|
self.path_info = path_info
|
|
# be careful to only replace the first slash in the path because of
|
|
# http://test/something and http://test//something being different as
|
|
# stated in RFC 3986.
|
|
self.path = "%s/%s" % (script_name.rstrip("/"), path_info.replace("/", "", 1))
|
|
self.META = environ
|
|
self.META["PATH_INFO"] = path_info
|
|
self.META["SCRIPT_NAME"] = script_name
|
|
self.method = environ["REQUEST_METHOD"].upper()
|
|
# Set content_type, content_params, and encoding.
|
|
self._set_content_type_params(environ)
|
|
try:
|
|
content_length = int(environ.get("CONTENT_LENGTH"))
|
|
except (ValueError, TypeError):
|
|
content_length = 0
|
|
self._stream = LimitedStream(self.environ["wsgi.input"], content_length)
|
|
self._read_started = False
|
|
self.resolver_match = None
|
|
|
|
def _get_scheme(self):
|
|
return self.environ.get("wsgi.url_scheme")
|
|
|
|
@cached_property
|
|
def GET(self):
|
|
# The WSGI spec says 'QUERY_STRING' may be absent.
|
|
raw_query_string = get_bytes_from_wsgi(self.environ, "QUERY_STRING", "")
|
|
return QueryDict(raw_query_string, encoding=self._encoding)
|
|
|
|
def _get_post(self):
|
|
if not hasattr(self, "_post"):
|
|
self._load_post_and_files()
|
|
return self._post
|
|
|
|
def _set_post(self, post):
|
|
self._post = post
|
|
|
|
@cached_property
|
|
def COOKIES(self):
|
|
raw_cookie = get_str_from_wsgi(self.environ, "HTTP_COOKIE", "")
|
|
return parse_cookie(raw_cookie)
|
|
|
|
@property
|
|
def FILES(self):
|
|
if not hasattr(self, "_files"):
|
|
self._load_post_and_files()
|
|
return self._files
|
|
|
|
POST = property(_get_post, _set_post)
|
|
|
|
|
|
class WSGIHandler(base.BaseHandler):
|
|
request_class = WSGIRequest
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.load_middleware()
|
|
|
|
def __call__(self, environ, start_response):
|
|
set_script_prefix(get_script_name(environ))
|
|
signals.request_started.send(sender=self.__class__, environ=environ)
|
|
request = self.request_class(environ)
|
|
response = self.get_response(request)
|
|
|
|
response._handler_class = self.__class__
|
|
|
|
status = "%d %s" % (response.status_code, response.reason_phrase)
|
|
response_headers = [
|
|
*response.items(),
|
|
*(("Set-Cookie", c.output(header="")) for c in response.cookies.values()),
|
|
]
|
|
start_response(status, response_headers)
|
|
if getattr(response, "file_to_stream", None) is not None and environ.get(
|
|
"wsgi.file_wrapper"
|
|
):
|
|
# If `wsgi.file_wrapper` is used the WSGI server does not call
|
|
# .close on the response, but on the file wrapper. Patch it to use
|
|
# response.close instead which takes care of closing all files.
|
|
response.file_to_stream.close = response.close
|
|
response = environ["wsgi.file_wrapper"](
|
|
response.file_to_stream, response.block_size
|
|
)
|
|
return response
|
|
|
|
|
|
def get_path_info(environ):
|
|
"""Return the HTTP request's PATH_INFO as a string."""
|
|
path_info = get_bytes_from_wsgi(environ, "PATH_INFO", "/")
|
|
|
|
return repercent_broken_unicode(path_info).decode()
|
|
|
|
|
|
def get_script_name(environ):
|
|
"""
|
|
Return the equivalent of the HTTP request's SCRIPT_NAME environment
|
|
variable. If Apache mod_rewrite is used, return what would have been
|
|
the script name prior to any rewriting (so it's the script name as seen
|
|
from the client's perspective), unless the FORCE_SCRIPT_NAME setting is
|
|
set (to anything).
|
|
"""
|
|
if settings.FORCE_SCRIPT_NAME is not None:
|
|
return settings.FORCE_SCRIPT_NAME
|
|
|
|
# If Apache's mod_rewrite had a whack at the URL, Apache set either
|
|
# SCRIPT_URL or REDIRECT_URL to the full resource URL before applying any
|
|
# rewrites. Unfortunately not every web server (lighttpd!) passes this
|
|
# information through all the time, so FORCE_SCRIPT_NAME, above, is still
|
|
# needed.
|
|
script_url = get_bytes_from_wsgi(environ, "SCRIPT_URL", "") or get_bytes_from_wsgi(
|
|
environ, "REDIRECT_URL", ""
|
|
)
|
|
|
|
if script_url:
|
|
if b"//" in script_url:
|
|
# mod_wsgi squashes multiple successive slashes in PATH_INFO,
|
|
# do the same with script_url before manipulating paths (#17133).
|
|
script_url = _slashes_re.sub(b"/", script_url)
|
|
path_info = get_bytes_from_wsgi(environ, "PATH_INFO", "")
|
|
script_name = script_url[: -len(path_info)] if path_info else script_url
|
|
else:
|
|
script_name = get_bytes_from_wsgi(environ, "SCRIPT_NAME", "")
|
|
|
|
return script_name.decode()
|
|
|
|
|
|
def get_bytes_from_wsgi(environ, key, default):
|
|
"""
|
|
Get a value from the WSGI environ dictionary as bytes.
|
|
|
|
key and default should be strings.
|
|
"""
|
|
value = environ.get(key, default)
|
|
# Non-ASCII values in the WSGI environ are arbitrarily decoded with
|
|
# ISO-8859-1. This is wrong for Django websites where UTF-8 is the default.
|
|
# Re-encode to recover the original bytestring.
|
|
return value.encode("iso-8859-1")
|
|
|
|
|
|
def get_str_from_wsgi(environ, key, default):
|
|
"""
|
|
Get a value from the WSGI environ dictionary as str.
|
|
|
|
key and default should be str objects.
|
|
"""
|
|
value = get_bytes_from_wsgi(environ, key, default)
|
|
return value.decode(errors="replace")
|