1. from io import BytesIO

  2. from itertools import chain

  3. from urllib.parse import urlencode

  4. 

  5. from django.core.exceptions import DisallowedHost

  6. from django.core.handlers.wsgi import LimitedStream, WSGIRequest

  7. from django.http import HttpRequest, RawPostDataException, UnreadablePostError

  8. from django.http.multipartparser import MultiPartParserError

  9. from django.http.request import HttpHeaders, split_domain_port

  10. from django.test import RequestFactory, SimpleTestCase, override_settings

  11. from django.test.client import FakePayload

  12. 

  13. 

  14. class RequestsTests(SimpleTestCase):

  15.     def test_httprequest(self):

  16.         request = HttpRequest()

  17.         self.assertEqual(list(request.GET), [])

  18.         self.assertEqual(list(request.POST), [])

  19.         self.assertEqual(list(request.COOKIES), [])

  20.         self.assertEqual(list(request.META), [])

  21. 

  22.         # .GET and .POST should be QueryDicts

  23.         self.assertEqual(request.GET.urlencode(), "")

  24.         self.assertEqual(request.POST.urlencode(), "")

  25. 

  26.         # and FILES should be MultiValueDict

  27.         self.assertEqual(request.FILES.getlist("foo"), [])

  28. 

  29.         self.assertIsNone(request.content_type)

  30.         self.assertIsNone(request.content_params)

  31. 

  32.     def test_httprequest_full_path(self):

  33.         request = HttpRequest()

  34.         request.path = "/;some/?awful/=path/foo:bar/"

  35.         request.path_info = "/prefix" + request.path

  36.         request.META["QUERY_STRING"] = ";some=query&+query=string"

  37.         expected = "/%3Bsome/%3Fawful/%3Dpath/foo:bar/?;some=query&+query=string"

  38.         self.assertEqual(request.get_full_path(), expected)

  39.         self.assertEqual(request.get_full_path_info(), "/prefix" + expected)

  40. 

  41.     def test_httprequest_full_path_with_query_string_and_fragment(self):

  42.         request = HttpRequest()

  43.         request.path = "/foo#bar"

  44.         request.path_info = "/prefix" + request.path

  45.         request.META["QUERY_STRING"] = "baz#quux"

  46.         self.assertEqual(request.get_full_path(), "/foo%23bar?baz#quux")

  47.         self.assertEqual(request.get_full_path_info(), "/prefix/foo%23bar?baz#quux")

  48. 

  49.     def test_httprequest_repr(self):

  50.         request = HttpRequest()

  51.         request.path = "/somepath/"

  52.         request.method = "GET"

  53.         request.GET = {"get-key": "get-value"}

  54.         request.POST = {"post-key": "post-value"}

  55.         request.COOKIES = {"post-key": "post-value"}

  56.         request.META = {"post-key": "post-value"}

  57.         self.assertEqual(repr(request), "<HttpRequest: GET '/somepath/'>")

  58. 

  59.     def test_httprequest_repr_invalid_method_and_path(self):

  60.         request = HttpRequest()

  61.         self.assertEqual(repr(request), "<HttpRequest>")

  62.         request = HttpRequest()

  63.         request.method = "GET"

  64.         self.assertEqual(repr(request), "<HttpRequest>")

  65.         request = HttpRequest()

  66.         request.path = ""

  67.         self.assertEqual(repr(request), "<HttpRequest>")

  68. 

  69.     def test_wsgirequest(self):

  70.         request = WSGIRequest(

  71.             {

  72.                 "PATH_INFO": "bogus",

  73.                 "REQUEST_METHOD": "bogus",

  74.                 "CONTENT_TYPE": "text/html; charset=utf8",

  75.                 "wsgi.input": BytesIO(b""),

  76.             }

  77.         )

  78.         self.assertEqual(list(request.GET), [])

  79.         self.assertEqual(list(request.POST), [])

  80.         self.assertEqual(list(request.COOKIES), [])

  81.         self.assertEqual(

  82.             set(request.META),

  83.             {

  84.                 "PATH_INFO",

  85.                 "REQUEST_METHOD",

  86.                 "SCRIPT_NAME",

  87.                 "CONTENT_TYPE",

  88.                 "wsgi.input",

  89.             },

  90.         )

  91.         self.assertEqual(request.META["PATH_INFO"], "bogus")

  92.         self.assertEqual(request.META["REQUEST_METHOD"], "bogus")

  93.         self.assertEqual(request.META["SCRIPT_NAME"], "")

  94.         self.assertEqual(request.content_type, "text/html")

  95.         self.assertEqual(request.content_params, {"charset": "utf8"})

  96. 

  97.     def test_wsgirequest_with_script_name(self):

  98.         """

  99.         The request's path is correctly assembled, regardless of whether or

  100.         not the SCRIPT_NAME has a trailing slash (#20169).

  101.         """

  102.         # With trailing slash

  103.         request = WSGIRequest(

  104.             {

  105.                 "PATH_INFO": "/somepath/",

  106.                 "SCRIPT_NAME": "/PREFIX/",

  107.                 "REQUEST_METHOD": "get",

  108.                 "wsgi.input": BytesIO(b""),

  109.             }

  110.         )

  111.         self.assertEqual(request.path, "/PREFIX/somepath/")

  112.         # Without trailing slash

  113.         request = WSGIRequest(

  114.             {

  115.                 "PATH_INFO": "/somepath/",

  116.                 "SCRIPT_NAME": "/PREFIX",

  117.                 "REQUEST_METHOD": "get",

  118.                 "wsgi.input": BytesIO(b""),

  119.             }

  120.         )

  121.         self.assertEqual(request.path, "/PREFIX/somepath/")

  122. 

  123.     def test_wsgirequest_script_url_double_slashes(self):

  124.         """

  125.         WSGI squashes multiple successive slashes in PATH_INFO, WSGIRequest

  126.         should take that into account when populating request.path and

  127.         request.META['SCRIPT_NAME'] (#17133).

  128.         """

  129.         request = WSGIRequest(

  130.             {

  131.                 "SCRIPT_URL": "/mst/milestones//accounts/login//help",

  132.                 "PATH_INFO": "/milestones/accounts/login/help",

  133.                 "REQUEST_METHOD": "get",

  134.                 "wsgi.input": BytesIO(b""),

  135.             }

  136.         )

  137.         self.assertEqual(request.path, "/mst/milestones/accounts/login/help")

  138.         self.assertEqual(request.META["SCRIPT_NAME"], "/mst")

  139. 

  140.     def test_wsgirequest_with_force_script_name(self):

  141.         """

  142.         The FORCE_SCRIPT_NAME setting takes precedence over the request's

  143.         SCRIPT_NAME environment parameter (#20169).

  144.         """

  145.         with override_settings(FORCE_SCRIPT_NAME="/FORCED_PREFIX/"):

  146.             request = WSGIRequest(

  147.                 {

  148.                     "PATH_INFO": "/somepath/",

  149.                     "SCRIPT_NAME": "/PREFIX/",

  150.                     "REQUEST_METHOD": "get",

  151.                     "wsgi.input": BytesIO(b""),

  152.                 }

  153.             )

  154.             self.assertEqual(request.path, "/FORCED_PREFIX/somepath/")

  155. 

  156.     def test_wsgirequest_path_with_force_script_name_trailing_slash(self):

  157.         """

  158.         The request's path is correctly assembled, regardless of whether or not

  159.         the FORCE_SCRIPT_NAME setting has a trailing slash (#20169).

  160.         """

  161.         # With trailing slash

  162.         with override_settings(FORCE_SCRIPT_NAME="/FORCED_PREFIX/"):

  163.             request = WSGIRequest(

  164.                 {

  165.                     "PATH_INFO": "/somepath/",

  166.                     "REQUEST_METHOD": "get",

  167.                     "wsgi.input": BytesIO(b""),

  168.                 }

  169.             )

  170.             self.assertEqual(request.path, "/FORCED_PREFIX/somepath/")

  171.         # Without trailing slash

  172.         with override_settings(FORCE_SCRIPT_NAME="/FORCED_PREFIX"):

  173.             request = WSGIRequest(

  174.                 {

  175.                     "PATH_INFO": "/somepath/",

  176.                     "REQUEST_METHOD": "get",

  177.                     "wsgi.input": BytesIO(b""),

  178.                 }

  179.             )

  180.             self.assertEqual(request.path, "/FORCED_PREFIX/somepath/")

  181. 

  182.     def test_wsgirequest_repr(self):

  183.         request = WSGIRequest({"REQUEST_METHOD": "get", "wsgi.input": BytesIO(b"")})

  184.         self.assertEqual(repr(request), "<WSGIRequest: GET '/'>")

  185.         request = WSGIRequest(

  186.             {

  187.                 "PATH_INFO": "/somepath/",

  188.                 "REQUEST_METHOD": "get",

  189.                 "wsgi.input": BytesIO(b""),

  190.             }

  191.         )

  192.         request.GET = {"get-key": "get-value"}

  193.         request.POST = {"post-key": "post-value"}

  194.         request.COOKIES = {"post-key": "post-value"}

  195.         request.META = {"post-key": "post-value"}

  196.         self.assertEqual(repr(request), "<WSGIRequest: GET '/somepath/'>")

  197. 

  198.     def test_wsgirequest_path_info(self):

  199.         def wsgi_str(path_info, encoding="utf-8"):

  200.             path_info = path_info.encode(

  201.                 encoding

  202.             )  # Actual URL sent by the browser (bytestring)

  203.             path_info = path_info.decode(

  204.                 "iso-8859-1"

  205.             )  # Value in the WSGI environ dict (native string)

  206.             return path_info

  207. 

  208.         # Regression for #19468

  209.         request = WSGIRequest(

  210.             {

  211.                 "PATH_INFO": wsgi_str("/سلام/"),

  212.                 "REQUEST_METHOD": "get",

  213.                 "wsgi.input": BytesIO(b""),

  214.             }

  215.         )

  216.         self.assertEqual(request.path, "/سلام/")

  217. 

  218.         # The URL may be incorrectly encoded in a non-UTF-8 encoding (#26971)

  219.         request = WSGIRequest(

  220.             {

  221.                 "PATH_INFO": wsgi_str("/café/", encoding="iso-8859-1"),

  222.                 "REQUEST_METHOD": "get",

  223.                 "wsgi.input": BytesIO(b""),

  224.             }

  225.         )

  226.         # Since it's impossible to decide the (wrong) encoding of the URL, it's

  227.         # left percent-encoded in the path.

  228.         self.assertEqual(request.path, "/caf%E9/")

  229. 

  230.     def test_limited_stream(self):

  231.         # Read all of a limited stream

  232.         stream = LimitedStream(BytesIO(b"test"), 2)

  233.         self.assertEqual(stream.read(), b"te")

  234.         # Reading again returns nothing.

  235.         self.assertEqual(stream.read(), b"")

  236. 

  237.         # Read a number of characters greater than the stream has to offer

  238.         stream = LimitedStream(BytesIO(b"test"), 2)

  239.         self.assertEqual(stream.read(5), b"te")

  240.         # Reading again returns nothing.

  241.         self.assertEqual(stream.readline(5), b"")

  242. 

  243.         # Read sequentially from a stream

  244.         stream = LimitedStream(BytesIO(b"12345678"), 8)

  245.         self.assertEqual(stream.read(5), b"12345")

  246.         self.assertEqual(stream.read(5), b"678")

  247.         # Reading again returns nothing.

  248.         self.assertEqual(stream.readline(5), b"")

  249. 

  250.         # Read lines from a stream

  251.         stream = LimitedStream(BytesIO(b"1234\n5678\nabcd\nefgh\nijkl"), 24)

  252.         # Read a full line, unconditionally

  253.         self.assertEqual(stream.readline(), b"1234\n")

  254.         # Read a number of characters less than a line

  255.         self.assertEqual(stream.readline(2), b"56")

  256.         # Read the rest of the partial line

  257.         self.assertEqual(stream.readline(), b"78\n")

  258.         # Read a full line, with a character limit greater than the line length

  259.         self.assertEqual(stream.readline(6), b"abcd\n")

  260.         # Read the next line, deliberately terminated at the line end

  261.         self.assertEqual(stream.readline(4), b"efgh")

  262.         # Read the next line... just the line end

  263.         self.assertEqual(stream.readline(), b"\n")

  264.         # Read everything else.

  265.         self.assertEqual(stream.readline(), b"ijkl")

  266. 

  267.         # Regression for #15018

  268.         # If a stream contains a newline, but the provided length

  269.         # is less than the number of provided characters, the newline

  270.         # doesn't reset the available character count

  271.         stream = LimitedStream(BytesIO(b"1234\nabcdef"), 9)

  272.         self.assertEqual(stream.readline(10), b"1234\n")

  273.         self.assertEqual(stream.readline(3), b"abc")

  274.         # Now expire the available characters

  275.         self.assertEqual(stream.readline(3), b"d")

  276.         # Reading again returns nothing.

  277.         self.assertEqual(stream.readline(2), b"")

  278. 

  279.         # Same test, but with read, not readline.

  280.         stream = LimitedStream(BytesIO(b"1234\nabcdef"), 9)

  281.         self.assertEqual(stream.read(6), b"1234\na")

  282.         self.assertEqual(stream.read(2), b"bc")

  283.         self.assertEqual(stream.read(2), b"d")

  284.         self.assertEqual(stream.read(2), b"")

  285.         self.assertEqual(stream.read(), b"")

  286. 

  287.     def test_stream(self):

  288.         payload = FakePayload("name=value")

  289.         request = WSGIRequest(

  290.             {

  291.                 "REQUEST_METHOD": "POST",

  292.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  293.                 "CONTENT_LENGTH": len(payload),

  294.                 "wsgi.input": payload,

  295.             },

  296.         )

  297.         self.assertEqual(request.read(), b"name=value")

  298. 

  299.     def test_read_after_value(self):

  300.         """

  301.         Reading from request is allowed after accessing request contents as

  302.         POST or body.

  303.         """

  304.         payload = FakePayload("name=value")

  305.         request = WSGIRequest(

  306.             {

  307.                 "REQUEST_METHOD": "POST",

  308.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  309.                 "CONTENT_LENGTH": len(payload),

  310.                 "wsgi.input": payload,

  311.             }

  312.         )

  313.         self.assertEqual(request.POST, {"name": ["value"]})

  314.         self.assertEqual(request.body, b"name=value")

  315.         self.assertEqual(request.read(), b"name=value")

  316. 

  317.     def test_value_after_read(self):

  318.         """

  319.         Construction of POST or body is not allowed after reading

  320.         from request.

  321.         """

  322.         payload = FakePayload("name=value")

  323.         request = WSGIRequest(

  324.             {

  325.                 "REQUEST_METHOD": "POST",

  326.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  327.                 "CONTENT_LENGTH": len(payload),

  328.                 "wsgi.input": payload,

  329.             }

  330.         )

  331.         self.assertEqual(request.read(2), b"na")

  332.         with self.assertRaises(RawPostDataException):

  333.             request.body

  334.         self.assertEqual(request.POST, {})

  335. 

  336.     def test_non_ascii_POST(self):

  337.         payload = FakePayload(urlencode({"key": "España"}))

  338.         request = WSGIRequest(

  339.             {

  340.                 "REQUEST_METHOD": "POST",

  341.                 "CONTENT_LENGTH": len(payload),

  342.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  343.                 "wsgi.input": payload,

  344.             }

  345.         )

  346.         self.assertEqual(request.POST, {"key": ["España"]})

  347. 

  348.     def test_alternate_charset_POST(self):

  349.         """

  350.         Test a POST with non-utf-8 payload encoding.

  351.         """

  352.         payload = FakePayload(urlencode({"key": "España".encode("latin-1")}))

  353.         request = WSGIRequest(

  354.             {

  355.                 "REQUEST_METHOD": "POST",

  356.                 "CONTENT_LENGTH": len(payload),

  357.                 "CONTENT_TYPE": "application/x-www-form-urlencoded; charset=iso-8859-1",

  358.                 "wsgi.input": payload,

  359.             }

  360.         )

  361.         self.assertEqual(request.POST, {"key": ["España"]})

  362. 

  363.     def test_body_after_POST_multipart_form_data(self):

  364.         """

  365.         Reading body after parsing multipart/form-data is not allowed

  366.         """

  367.         # Because multipart is used for large amounts of data i.e. file uploads,

  368.         # we don't want the data held in memory twice, and we don't want to

  369.         # silence the error by setting body = '' either.

  370.         payload = FakePayload(

  371.             "\r\n".join(

  372.                 [

  373.                     "--boundary",

  374.                     'Content-Disposition: form-data; name="name"',

  375.                     "",

  376.                     "value",

  377.                     "--boundary--",

  378.                 ]

  379.             )

  380.         )

  381.         request = WSGIRequest(

  382.             {

  383.                 "REQUEST_METHOD": "POST",

  384.                 "CONTENT_TYPE": "multipart/form-data; boundary=boundary",

  385.                 "CONTENT_LENGTH": len(payload),

  386.                 "wsgi.input": payload,

  387.             }

  388.         )

  389.         self.assertEqual(request.POST, {"name": ["value"]})

  390.         with self.assertRaises(RawPostDataException):

  391.             request.body

  392. 

  393.     def test_body_after_POST_multipart_related(self):

  394.         """

  395.         Reading body after parsing multipart that isn't form-data is allowed

  396.         """

  397.         # Ticket #9054

  398.         # There are cases in which the multipart data is related instead of

  399.         # being a binary upload, in which case it should still be accessible

  400.         # via body.

  401.         payload_data = b"\r\n".join(

  402.             [

  403.                 b"--boundary",

  404.                 b'Content-ID: id; name="name"',

  405.                 b"",

  406.                 b"value",

  407.                 b"--boundary--",

  408.             ]

  409.         )

  410.         payload = FakePayload(payload_data)

  411.         request = WSGIRequest(

  412.             {

  413.                 "REQUEST_METHOD": "POST",

  414.                 "CONTENT_TYPE": "multipart/related; boundary=boundary",

  415.                 "CONTENT_LENGTH": len(payload),

  416.                 "wsgi.input": payload,

  417.             }

  418.         )

  419.         self.assertEqual(request.POST, {})

  420.         self.assertEqual(request.body, payload_data)

  421. 

  422.     def test_POST_multipart_with_content_length_zero(self):

  423.         """

  424.         Multipart POST requests with Content-Length >= 0 are valid and need to

  425.         be handled.

  426.         """

  427.         # According to:

  428.         # https://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.13

  429.         # Every request.POST with Content-Length >= 0 is a valid request,

  430.         # this test ensures that we handle Content-Length == 0.

  431.         payload = FakePayload(

  432.             "\r\n".join(

  433.                 [

  434.                     "--boundary",

  435.                     'Content-Disposition: form-data; name="name"',

  436.                     "",

  437.                     "value",

  438.                     "--boundary--",

  439.                 ]

  440.             )

  441.         )

  442.         request = WSGIRequest(

  443.             {

  444.                 "REQUEST_METHOD": "POST",

  445.                 "CONTENT_TYPE": "multipart/form-data; boundary=boundary",

  446.                 "CONTENT_LENGTH": 0,

  447.                 "wsgi.input": payload,

  448.             }

  449.         )

  450.         self.assertEqual(request.POST, {})

  451. 

  452.     def test_POST_binary_only(self):

  453.         payload = b"\r\n\x01\x00\x00\x00ab\x00\x00\xcd\xcc,@"

  454.         environ = {

  455.             "REQUEST_METHOD": "POST",

  456.             "CONTENT_TYPE": "application/octet-stream",

  457.             "CONTENT_LENGTH": len(payload),

  458.             "wsgi.input": BytesIO(payload),

  459.         }

  460.         request = WSGIRequest(environ)

  461.         self.assertEqual(request.POST, {})

  462.         self.assertEqual(request.FILES, {})

  463.         self.assertEqual(request.body, payload)

  464. 

  465.         # Same test without specifying content-type

  466.         environ.update({"CONTENT_TYPE": "", "wsgi.input": BytesIO(payload)})

  467.         request = WSGIRequest(environ)

  468.         self.assertEqual(request.POST, {})

  469.         self.assertEqual(request.FILES, {})

  470.         self.assertEqual(request.body, payload)

  471. 

  472.     def test_read_by_lines(self):

  473.         payload = FakePayload("name=value")

  474.         request = WSGIRequest(

  475.             {

  476.                 "REQUEST_METHOD": "POST",

  477.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  478.                 "CONTENT_LENGTH": len(payload),

  479.                 "wsgi.input": payload,

  480.             }

  481.         )

  482.         self.assertEqual(list(request), [b"name=value"])

  483. 

  484.     def test_POST_after_body_read(self):

  485.         """

  486.         POST should be populated even if body is read first

  487.         """

  488.         payload = FakePayload("name=value")

  489.         request = WSGIRequest(

  490.             {

  491.                 "REQUEST_METHOD": "POST",

  492.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  493.                 "CONTENT_LENGTH": len(payload),

  494.                 "wsgi.input": payload,

  495.             }

  496.         )

  497.         request.body  # evaluate

  498.         self.assertEqual(request.POST, {"name": ["value"]})

  499. 

  500.     def test_POST_after_body_read_and_stream_read(self):

  501.         """

  502.         POST should be populated even if body is read first, and then

  503.         the stream is read second.

  504.         """

  505.         payload = FakePayload("name=value")

  506.         request = WSGIRequest(

  507.             {

  508.                 "REQUEST_METHOD": "POST",

  509.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  510.                 "CONTENT_LENGTH": len(payload),

  511.                 "wsgi.input": payload,

  512.             }

  513.         )

  514.         request.body  # evaluate

  515.         self.assertEqual(request.read(1), b"n")

  516.         self.assertEqual(request.POST, {"name": ["value"]})

  517. 

  518.     def test_POST_after_body_read_and_stream_read_multipart(self):

  519.         """

  520.         POST should be populated even if body is read first, and then

  521.         the stream is read second. Using multipart/form-data instead of urlencoded.

  522.         """

  523.         payload = FakePayload(

  524.             "\r\n".join(

  525.                 [

  526.                     "--boundary",

  527.                     'Content-Disposition: form-data; name="name"',

  528.                     "",

  529.                     "value",

  530.                     "--boundary--" "",

  531.                 ]

  532.             )

  533.         )

  534.         request = WSGIRequest(

  535.             {

  536.                 "REQUEST_METHOD": "POST",

  537.                 "CONTENT_TYPE": "multipart/form-data; boundary=boundary",

  538.                 "CONTENT_LENGTH": len(payload),

  539.                 "wsgi.input": payload,

  540.             }

  541.         )

  542.         request.body  # evaluate

  543.         # Consume enough data to mess up the parsing:

  544.         self.assertEqual(request.read(13), b"--boundary\r\nC")

  545.         self.assertEqual(request.POST, {"name": ["value"]})

  546. 

  547.     def test_POST_immutable_for_multipart(self):

  548.         """

  549.         MultiPartParser.parse() leaves request.POST immutable.

  550.         """

  551.         payload = FakePayload(

  552.             "\r\n".join(

  553.                 [

  554.                     "--boundary",

  555.                     'Content-Disposition: form-data; name="name"',

  556.                     "",

  557.                     "value",

  558.                     "--boundary--",

  559.                 ]

  560.             )

  561.         )

  562.         request = WSGIRequest(

  563.             {

  564.                 "REQUEST_METHOD": "POST",

  565.                 "CONTENT_TYPE": "multipart/form-data; boundary=boundary",

  566.                 "CONTENT_LENGTH": len(payload),

  567.                 "wsgi.input": payload,

  568.             }

  569.         )

  570.         self.assertFalse(request.POST._mutable)

  571. 

  572.     def test_multipart_without_boundary(self):

  573.         request = WSGIRequest(

  574.             {

  575.                 "REQUEST_METHOD": "POST",

  576.                 "CONTENT_TYPE": "multipart/form-data;",

  577.                 "CONTENT_LENGTH": 0,

  578.                 "wsgi.input": FakePayload(),

  579.             }

  580.         )

  581.         with self.assertRaisesMessage(

  582.             MultiPartParserError, "Invalid boundary in multipart: None"

  583.         ):

  584.             request.POST

  585. 

  586.     def test_multipart_non_ascii_content_type(self):

  587.         request = WSGIRequest(

  588.             {

  589.                 "REQUEST_METHOD": "POST",

  590.                 "CONTENT_TYPE": "multipart/form-data; boundary = \xe0",

  591.                 "CONTENT_LENGTH": 0,

  592.                 "wsgi.input": FakePayload(),

  593.             }

  594.         )

  595.         msg = (

  596.             "Invalid non-ASCII Content-Type in multipart: multipart/form-data; "

  597.             "boundary = à"

  598.         )

  599.         with self.assertRaisesMessage(MultiPartParserError, msg):

  600.             request.POST

  601. 

  602.     def test_POST_connection_error(self):

  603.         """

  604.         If wsgi.input.read() raises an exception while trying to read() the

  605.         POST, the exception is identifiable (not a generic OSError).

  606.         """

  607. 

  608.         class ExplodingBytesIO(BytesIO):

  609.             def read(self, len=0):

  610.                 raise OSError("kaboom!")

  611. 

  612.         payload = b"name=value"

  613.         request = WSGIRequest(

  614.             {

  615.                 "REQUEST_METHOD": "POST",

  616.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  617.                 "CONTENT_LENGTH": len(payload),

  618.                 "wsgi.input": ExplodingBytesIO(payload),

  619.             }

  620.         )

  621.         with self.assertRaises(UnreadablePostError):

  622.             request.body

  623. 

  624.     def test_set_encoding_clears_POST(self):

  625.         payload = FakePayload("name=Hello Günter")

  626.         request = WSGIRequest(

  627.             {

  628.                 "REQUEST_METHOD": "POST",

  629.                 "CONTENT_TYPE": "application/x-www-form-urlencoded",

  630.                 "CONTENT_LENGTH": len(payload),

  631.                 "wsgi.input": payload,

  632.             }

  633.         )

  634.         self.assertEqual(request.POST, {"name": ["Hello Günter"]})

  635.         request.encoding = "iso-8859-16"

  636.         self.assertEqual(request.POST, {"name": ["Hello GĂŒnter"]})

  637. 

  638.     def test_set_encoding_clears_GET(self):

  639.         request = WSGIRequest(

  640.             {

  641.                 "REQUEST_METHOD": "GET",

  642.                 "wsgi.input": "",

  643.                 "QUERY_STRING": "name=Hello%20G%C3%BCnter",

  644.             }

  645.         )

  646.         self.assertEqual(request.GET, {"name": ["Hello Günter"]})

  647.         request.encoding = "iso-8859-16"

  648.         self.assertEqual(request.GET, {"name": ["Hello G\u0102\u0152nter"]})

  649. 

  650.     def test_FILES_connection_error(self):

  651.         """

  652.         If wsgi.input.read() raises an exception while trying to read() the

  653.         FILES, the exception is identifiable (not a generic OSError).

  654.         """

  655. 

  656.         class ExplodingBytesIO(BytesIO):

  657.             def read(self, len=0):

  658.                 raise OSError("kaboom!")

  659. 

  660.         payload = b"x"

  661.         request = WSGIRequest(

  662.             {

  663.                 "REQUEST_METHOD": "POST",

  664.                 "CONTENT_TYPE": "multipart/form-data; boundary=foo_",

  665.                 "CONTENT_LENGTH": len(payload),

  666.                 "wsgi.input": ExplodingBytesIO(payload),

  667.             }

  668.         )

  669.         with self.assertRaises(UnreadablePostError):

  670.             request.FILES

  671. 

  672. 

  673. class HostValidationTests(SimpleTestCase):

  674.     poisoned_hosts = [

  675.         "[email protected]",

  676.         "example.com:[email protected]",

  677.         "example.com:[email protected]:80",

  678.         "example.com:80/badpath",

  679.         "example.com: recovermypassword.com",

  680.     ]

  681. 

  682.     @override_settings(

  683.         USE_X_FORWARDED_HOST=False,

  684.         ALLOWED_HOSTS=[

  685.             "forward.com",

  686.             "example.com",

  687.             "internal.com",

  688.             "12.34.56.78",

  689.             "[2001:19f0:feee::dead:beef:cafe]",

  690.             "xn--4ca9at.com",

  691.             ".multitenant.com",

  692.             "INSENSITIVE.com",

  693.             "[::ffff:169.254.169.254]",

  694.         ],

  695.     )

  696.     def test_http_get_host(self):

  697.         # Check if X_FORWARDED_HOST is provided.

  698.         request = HttpRequest()

  699.         request.META = {

  700.             "HTTP_X_FORWARDED_HOST": "forward.com",

  701.             "HTTP_HOST": "example.com",

  702.             "SERVER_NAME": "internal.com",

  703.             "SERVER_PORT": 80,

  704.         }

  705.         # X_FORWARDED_HOST is ignored.

  706.         self.assertEqual(request.get_host(), "example.com")

  707. 

  708.         # Check if X_FORWARDED_HOST isn't provided.

  709.         request = HttpRequest()

  710.         request.META = {

  711.             "HTTP_HOST": "example.com",

  712.             "SERVER_NAME": "internal.com",

  713.             "SERVER_PORT": 80,

  714.         }

  715.         self.assertEqual(request.get_host(), "example.com")

  716. 

  717.         # Check if HTTP_HOST isn't provided.

  718.         request = HttpRequest()

  719.         request.META = {

  720.             "SERVER_NAME": "internal.com",

  721.             "SERVER_PORT": 80,

  722.         }

  723.         self.assertEqual(request.get_host(), "internal.com")

  724. 

  725.         # Check if HTTP_HOST isn't provided, and we're on a nonstandard port

  726.         request = HttpRequest()

  727.         request.META = {

  728.             "SERVER_NAME": "internal.com",

  729.             "SERVER_PORT": 8042,

  730.         }

  731.         self.assertEqual(request.get_host(), "internal.com:8042")

  732. 

  733.         legit_hosts = [

  734.             "example.com",

  735.             "example.com:80",

  736.             "12.34.56.78",

  737.             "12.34.56.78:443",

  738.             "[2001:19f0:feee::dead:beef:cafe]",

  739.             "[2001:19f0:feee::dead:beef:cafe]:8080",

  740.             "xn--4ca9at.com",  # Punycode for öäü.com

  741.             "anything.multitenant.com",

  742.             "multitenant.com",

  743.             "insensitive.com",

  744.             "example.com.",

  745.             "example.com.:80",

  746.             "[::ffff:169.254.169.254]",

  747.         ]

  748. 

  749.         for host in legit_hosts:

  750.             request = HttpRequest()

  751.             request.META = {

  752.                 "HTTP_HOST": host,

  753.             }

  754.             request.get_host()

  755. 

  756.         # Poisoned host headers are rejected as suspicious

  757.         for host in chain(self.poisoned_hosts, ["other.com", "example.com.."]):

  758.             with self.assertRaises(DisallowedHost):

  759.                 request = HttpRequest()

  760.                 request.META = {

  761.                     "HTTP_HOST": host,

  762.                 }

  763.                 request.get_host()

  764. 

  765.     @override_settings(USE_X_FORWARDED_HOST=True, ALLOWED_HOSTS=["*"])

  766.     def test_http_get_host_with_x_forwarded_host(self):

  767.         # Check if X_FORWARDED_HOST is provided.

  768.         request = HttpRequest()

  769.         request.META = {

  770.             "HTTP_X_FORWARDED_HOST": "forward.com",

  771.             "HTTP_HOST": "example.com",

  772.             "SERVER_NAME": "internal.com",

  773.             "SERVER_PORT": 80,

  774.         }

  775.         # X_FORWARDED_HOST is obeyed.

  776.         self.assertEqual(request.get_host(), "forward.com")

  777. 

  778.         # Check if X_FORWARDED_HOST isn't provided.

  779.         request = HttpRequest()

  780.         request.META = {

  781.             "HTTP_HOST": "example.com",

  782.             "SERVER_NAME": "internal.com",

  783.             "SERVER_PORT": 80,

  784.         }

  785.         self.assertEqual(request.get_host(), "example.com")

  786. 

  787.         # Check if HTTP_HOST isn't provided.

  788.         request = HttpRequest()

  789.         request.META = {

  790.             "SERVER_NAME": "internal.com",

  791.             "SERVER_PORT": 80,

  792.         }

  793.         self.assertEqual(request.get_host(), "internal.com")

  794. 

  795.         # Check if HTTP_HOST isn't provided, and we're on a nonstandard port

  796.         request = HttpRequest()

  797.         request.META = {

  798.             "SERVER_NAME": "internal.com",

  799.             "SERVER_PORT": 8042,

  800.         }

  801.         self.assertEqual(request.get_host(), "internal.com:8042")

  802. 

  803.         # Poisoned host headers are rejected as suspicious

  804.         legit_hosts = [

  805.             "example.com",

  806.             "example.com:80",

  807.             "12.34.56.78",

  808.             "12.34.56.78:443",

  809.             "[2001:19f0:feee::dead:beef:cafe]",

  810.             "[2001:19f0:feee::dead:beef:cafe]:8080",

  811.             "xn--4ca9at.com",  # Punycode for öäü.com

  812.         ]

  813. 

  814.         for host in legit_hosts:

  815.             request = HttpRequest()

  816.             request.META = {

  817.                 "HTTP_HOST": host,

  818.             }

  819.             request.get_host()

  820. 

  821.         for host in self.poisoned_hosts:

  822.             with self.assertRaises(DisallowedHost):

  823.                 request = HttpRequest()

  824.                 request.META = {

  825.                     "HTTP_HOST": host,

  826.                 }

  827.                 request.get_host()

  828. 

  829.     @override_settings(USE_X_FORWARDED_PORT=False)

  830.     def test_get_port(self):

  831.         request = HttpRequest()

  832.         request.META = {

  833.             "SERVER_PORT": "8080",

  834.             "HTTP_X_FORWARDED_PORT": "80",

  835.         }

  836.         # Shouldn't use the X-Forwarded-Port header

  837.         self.assertEqual(request.get_port(), "8080")

  838. 

  839.         request = HttpRequest()

  840.         request.META = {

  841.             "SERVER_PORT": "8080",

  842.         }

  843.         self.assertEqual(request.get_port(), "8080")

  844. 

  845.     @override_settings(USE_X_FORWARDED_PORT=True)

  846.     def test_get_port_with_x_forwarded_port(self):

  847.         request = HttpRequest()

  848.         request.META = {

  849.             "SERVER_PORT": "8080",

  850.             "HTTP_X_FORWARDED_PORT": "80",

  851.         }

  852.         # Should use the X-Forwarded-Port header

  853.         self.assertEqual(request.get_port(), "80")

  854. 

  855.         request = HttpRequest()

  856.         request.META = {

  857.             "SERVER_PORT": "8080",

  858.         }

  859.         self.assertEqual(request.get_port(), "8080")

  860. 

  861.     @override_settings(DEBUG=True, ALLOWED_HOSTS=[])

  862.     def test_host_validation_in_debug_mode(self):

  863.         """

  864.         If ALLOWED_HOSTS is empty and DEBUG is True, variants of localhost are

  865.         allowed.

  866.         """

  867.         valid_hosts = ["localhost", "subdomain.localhost", "127.0.0.1", "[::1]"]

  868.         for host in valid_hosts:

  869.             request = HttpRequest()

  870.             request.META = {"HTTP_HOST": host}

  871.             self.assertEqual(request.get_host(), host)

  872. 

  873.         # Other hostnames raise a DisallowedHost.

  874.         with self.assertRaises(DisallowedHost):

  875.             request = HttpRequest()

  876.             request.META = {"HTTP_HOST": "example.com"}

  877.             request.get_host()

  878. 

  879.     @override_settings(ALLOWED_HOSTS=[])

  880.     def test_get_host_suggestion_of_allowed_host(self):

  881.         """

  882.         get_host() makes helpful suggestions if a valid-looking host is not in

  883.         ALLOWED_HOSTS.

  884.         """

  885.         msg_invalid_host = "Invalid HTTP_HOST header: %r."

  886.         msg_suggestion = msg_invalid_host + " You may need to add %r to ALLOWED_HOSTS."

  887.         msg_suggestion2 = (

  888.             msg_invalid_host

  889.             + " The domain name provided is not valid according to RFC 1034/1035"

  890.         )

  891. 

  892.         for host in [  # Valid-looking hosts

  893.             "example.com",

  894.             "12.34.56.78",

  895.             "[2001:19f0:feee::dead:beef:cafe]",

  896.             "xn--4ca9at.com",  # Punycode for öäü.com

  897.         ]:

  898.             request = HttpRequest()

  899.             request.META = {"HTTP_HOST": host}

  900.             with self.assertRaisesMessage(

  901.                 DisallowedHost, msg_suggestion % (host, host)

  902.             ):

  903.                 request.get_host()

  904. 

  905.         for domain, port in [  # Valid-looking hosts with a port number

  906.             ("example.com", 80),

  907.             ("12.34.56.78", 443),

  908.             ("[2001:19f0:feee::dead:beef:cafe]", 8080),

  909.         ]:

  910.             host = "%s:%s" % (domain, port)

  911.             request = HttpRequest()

  912.             request.META = {"HTTP_HOST": host}

  913.             with self.assertRaisesMessage(

  914.                 DisallowedHost, msg_suggestion % (host, domain)

  915.             ):

  916.                 request.get_host()

  917. 

  918.         for host in self.poisoned_hosts:

  919.             request = HttpRequest()

  920.             request.META = {"HTTP_HOST": host}

  921.             with self.assertRaisesMessage(DisallowedHost, msg_invalid_host % host):

  922.                 request.get_host()

  923. 

  924.         request = HttpRequest()

  925.         request.META = {"HTTP_HOST": "invalid_hostname.com"}

  926.         with self.assertRaisesMessage(

  927.             DisallowedHost, msg_suggestion2 % "invalid_hostname.com"

  928.         ):

  929.             request.get_host()

  930. 

  931.     def test_split_domain_port_removes_trailing_dot(self):

  932.         domain, port = split_domain_port("example.com.:8080")

  933.         self.assertEqual(domain, "example.com")

  934.         self.assertEqual(port, "8080")

  935. 

  936. 

  937. class BuildAbsoluteURITests(SimpleTestCase):

  938.     factory = RequestFactory()

  939. 

  940.     def test_absolute_url(self):

  941.         request = HttpRequest()

  942.         url = "https://www.example.com/asdf"

  943.         self.assertEqual(request.build_absolute_uri(location=url), url)

  944. 

  945.     def test_host_retrieval(self):

  946.         request = HttpRequest()

  947.         request.get_host = lambda: "www.example.com"

  948.         request.path = ""

  949.         self.assertEqual(

  950.             request.build_absolute_uri(location="/path/with:colons"),

  951.             "http://www.example.com/path/with:colons",

  952.         )

  953. 

  954.     def test_request_path_begins_with_two_slashes(self):

  955.         # //// creates a request with a path beginning with //

  956.         request = self.factory.get("////absolute-uri")

  957.         tests = (

  958.             # location isn't provided

  959.             (None, "http://testserver//absolute-uri"),

  960.             # An absolute URL

  961.             ("http://example.com/?foo=bar", "http://example.com/?foo=bar"),

  962.             # A schema-relative URL

  963.             ("//example.com/?foo=bar", "http://example.com/?foo=bar"),

  964.             # Relative URLs

  965.             ("/foo/bar/", "http://testserver/foo/bar/"),

  966.             ("/foo/./bar/", "http://testserver/foo/bar/"),

  967.             ("/foo/../bar/", "http://testserver/bar/"),

  968.             ("///foo/bar/", "http://testserver/foo/bar/"),

  969.         )

  970.         for location, expected_url in tests:

  971.             with self.subTest(location=location):

  972.                 self.assertEqual(

  973.                     request.build_absolute_uri(location=location), expected_url

  974.                 )

  975. 

  976. 

  977. class RequestHeadersTests(SimpleTestCase):

  978.     ENVIRON = {

  979.         # Non-headers are ignored.

  980.         "PATH_INFO": "/somepath/",

  981.         "REQUEST_METHOD": "get",

  982.         "wsgi.input": BytesIO(b""),

  983.         "SERVER_NAME": "internal.com",

  984.         "SERVER_PORT": 80,

  985.         # These non-HTTP prefixed headers are included.

  986.         "CONTENT_TYPE": "text/html",

  987.         "CONTENT_LENGTH": "100",

  988.         # All HTTP-prefixed headers are included.

  989.         "HTTP_ACCEPT": "*",

  990.         "HTTP_HOST": "example.com",

  991.         "HTTP_USER_AGENT": "python-requests/1.2.0",

  992.     }

  993. 

  994.     def test_base_request_headers(self):

  995.         request = HttpRequest()

  996.         request.META = self.ENVIRON

  997.         self.assertEqual(

  998.             dict(request.headers),

  999.             {

  1000.                 "Content-Type": "text/html",

  1001.                 "Content-Length": "100",

  1002.                 "Accept": "*",

  1003.                 "Host": "example.com",

  1004.                 "User-Agent": "python-requests/1.2.0",

  1005.             },

  1006.         )

  1007. 

  1008.     def test_wsgi_request_headers(self):

  1009.         request = WSGIRequest(self.ENVIRON)

  1010.         self.assertEqual(

  1011.             dict(request.headers),

  1012.             {

  1013.                 "Content-Type": "text/html",

  1014.                 "Content-Length": "100",

  1015.                 "Accept": "*",

  1016.                 "Host": "example.com",

  1017.                 "User-Agent": "python-requests/1.2.0",

  1018.             },

  1019.         )

  1020. 

  1021.     def test_wsgi_request_headers_getitem(self):

  1022.         request = WSGIRequest(self.ENVIRON)

  1023.         self.assertEqual(request.headers["User-Agent"], "python-requests/1.2.0")

  1024.         self.assertEqual(request.headers["user-agent"], "python-requests/1.2.0")

  1025.         self.assertEqual(request.headers["user_agent"], "python-requests/1.2.0")

  1026.         self.assertEqual(request.headers["Content-Type"], "text/html")

  1027.         self.assertEqual(request.headers["Content-Length"], "100")

  1028. 

  1029.     def test_wsgi_request_headers_get(self):

  1030.         request = WSGIRequest(self.ENVIRON)

  1031.         self.assertEqual(request.headers.get("User-Agent"), "python-requests/1.2.0")

  1032.         self.assertEqual(request.headers.get("user-agent"), "python-requests/1.2.0")

  1033.         self.assertEqual(request.headers.get("Content-Type"), "text/html")

  1034.         self.assertEqual(request.headers.get("Content-Length"), "100")

  1035. 

  1036. 

  1037. class HttpHeadersTests(SimpleTestCase):

  1038.     def test_basic(self):

  1039.         environ = {

  1040.             "CONTENT_TYPE": "text/html",

  1041.             "CONTENT_LENGTH": "100",

  1042.             "HTTP_HOST": "example.com",

  1043.         }

  1044.         headers = HttpHeaders(environ)

  1045.         self.assertEqual(sorted(headers), ["Content-Length", "Content-Type", "Host"])

  1046.         self.assertEqual(

  1047.             headers,

  1048.             {

  1049.                 "Content-Type": "text/html",

  1050.                 "Content-Length": "100",

  1051.                 "Host": "example.com",

  1052.             },

  1053.         )

  1054. 

  1055.     def test_parse_header_name(self):

  1056.         tests = (

  1057.             ("PATH_INFO", None),

  1058.             ("HTTP_ACCEPT", "Accept"),

  1059.             ("HTTP_USER_AGENT", "User-Agent"),

  1060.             ("HTTP_X_FORWARDED_PROTO", "X-Forwarded-Proto"),

  1061.             ("CONTENT_TYPE", "Content-Type"),

  1062.             ("CONTENT_LENGTH", "Content-Length"),

  1063.         )

  1064.         for header, expected in tests:

  1065.             with self.subTest(header=header):

  1066.                 self.assertEqual(HttpHeaders.parse_header_name(header), expected)
About Sign Up Leave Feedback Contact