tests.py 37 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927
  1. import time
  2. from datetime import datetime, timedelta
  3. from http import cookies
  4. from io import BytesIO
  5. from itertools import chain
  6. from urllib.parse import urlencode
  7. from django.core.exceptions import DisallowedHost
  8. from django.core.handlers.wsgi import LimitedStream, WSGIRequest
  9. from django.http import (
  10. HttpRequest, HttpResponse, RawPostDataException, UnreadablePostError,
  11. )
  12. from django.http.request import split_domain_port
  13. from django.test import RequestFactory, SimpleTestCase, override_settings
  14. from django.test.client import FakePayload
  15. from django.test.utils import freeze_time
  16. from django.utils.http import cookie_date
  17. from django.utils.timezone import utc
  18. class RequestsTests(SimpleTestCase):
  19. def test_httprequest(self):
  20. request = HttpRequest()
  21. self.assertEqual(list(request.GET), [])
  22. self.assertEqual(list(request.POST), [])
  23. self.assertEqual(list(request.COOKIES), [])
  24. self.assertEqual(list(request.META), [])
  25. # .GET and .POST should be QueryDicts
  26. self.assertEqual(request.GET.urlencode(), '')
  27. self.assertEqual(request.POST.urlencode(), '')
  28. # and FILES should be MultiValueDict
  29. self.assertEqual(request.FILES.getlist('foo'), [])
  30. self.assertIsNone(request.content_type)
  31. self.assertIsNone(request.content_params)
  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. def test_httprequest_full_path_with_query_string_and_fragment(self):
  41. request = HttpRequest()
  42. request.path = '/foo#bar'
  43. request.path_info = '/prefix' + request.path
  44. request.META['QUERY_STRING'] = 'baz#quux'
  45. self.assertEqual(request.get_full_path(), '/foo%23bar?baz#quux')
  46. self.assertEqual(request.get_full_path_info(), '/prefix/foo%23bar?baz#quux')
  47. def test_httprequest_repr(self):
  48. request = HttpRequest()
  49. request.path = '/somepath/'
  50. request.method = 'GET'
  51. request.GET = {'get-key': 'get-value'}
  52. request.POST = {'post-key': 'post-value'}
  53. request.COOKIES = {'post-key': 'post-value'}
  54. request.META = {'post-key': 'post-value'}
  55. self.assertEqual(repr(request), "<HttpRequest: GET '/somepath/'>")
  56. def test_httprequest_repr_invalid_method_and_path(self):
  57. request = HttpRequest()
  58. self.assertEqual(repr(request), "<HttpRequest>")
  59. request = HttpRequest()
  60. request.method = "GET"
  61. self.assertEqual(repr(request), "<HttpRequest>")
  62. request = HttpRequest()
  63. request.path = ""
  64. self.assertEqual(repr(request), "<HttpRequest>")
  65. def test_wsgirequest(self):
  66. request = WSGIRequest({
  67. 'PATH_INFO': 'bogus',
  68. 'REQUEST_METHOD': 'bogus',
  69. 'CONTENT_TYPE': 'text/html; charset=utf8',
  70. 'wsgi.input': BytesIO(b''),
  71. })
  72. self.assertEqual(list(request.GET), [])
  73. self.assertEqual(list(request.POST), [])
  74. self.assertEqual(list(request.COOKIES), [])
  75. self.assertEqual(
  76. set(request.META),
  77. {'PATH_INFO', 'REQUEST_METHOD', 'SCRIPT_NAME', 'CONTENT_TYPE', 'wsgi.input'}
  78. )
  79. self.assertEqual(request.META['PATH_INFO'], 'bogus')
  80. self.assertEqual(request.META['REQUEST_METHOD'], 'bogus')
  81. self.assertEqual(request.META['SCRIPT_NAME'], '')
  82. self.assertEqual(request.content_type, 'text/html')
  83. self.assertEqual(request.content_params, {'charset': 'utf8'})
  84. def test_wsgirequest_with_script_name(self):
  85. """
  86. The request's path is correctly assembled, regardless of whether or
  87. not the SCRIPT_NAME has a trailing slash (#20169).
  88. """
  89. # With trailing slash
  90. request = WSGIRequest({
  91. 'PATH_INFO': '/somepath/',
  92. 'SCRIPT_NAME': '/PREFIX/',
  93. 'REQUEST_METHOD': 'get',
  94. 'wsgi.input': BytesIO(b''),
  95. })
  96. self.assertEqual(request.path, '/PREFIX/somepath/')
  97. # Without trailing slash
  98. request = WSGIRequest({
  99. 'PATH_INFO': '/somepath/',
  100. 'SCRIPT_NAME': '/PREFIX',
  101. 'REQUEST_METHOD': 'get',
  102. 'wsgi.input': BytesIO(b''),
  103. })
  104. self.assertEqual(request.path, '/PREFIX/somepath/')
  105. def test_wsgirequest_script_url_double_slashes(self):
  106. """
  107. WSGI squashes multiple successive slashes in PATH_INFO, WSGIRequest
  108. should take that into account when populating request.path and
  109. request.META['SCRIPT_NAME'] (#17133).
  110. """
  111. request = WSGIRequest({
  112. 'SCRIPT_URL': '/mst/milestones//accounts/login//help',
  113. 'PATH_INFO': '/milestones/accounts/login/help',
  114. 'REQUEST_METHOD': 'get',
  115. 'wsgi.input': BytesIO(b''),
  116. })
  117. self.assertEqual(request.path, '/mst/milestones/accounts/login/help')
  118. self.assertEqual(request.META['SCRIPT_NAME'], '/mst')
  119. def test_wsgirequest_with_force_script_name(self):
  120. """
  121. The FORCE_SCRIPT_NAME setting takes precedence over the request's
  122. SCRIPT_NAME environment parameter (#20169).
  123. """
  124. with override_settings(FORCE_SCRIPT_NAME='/FORCED_PREFIX/'):
  125. request = WSGIRequest({
  126. 'PATH_INFO': '/somepath/',
  127. 'SCRIPT_NAME': '/PREFIX/',
  128. 'REQUEST_METHOD': 'get',
  129. 'wsgi.input': BytesIO(b''),
  130. })
  131. self.assertEqual(request.path, '/FORCED_PREFIX/somepath/')
  132. def test_wsgirequest_path_with_force_script_name_trailing_slash(self):
  133. """
  134. The request's path is correctly assembled, regardless of whether or not
  135. the FORCE_SCRIPT_NAME setting has a trailing slash (#20169).
  136. """
  137. # With trailing slash
  138. with override_settings(FORCE_SCRIPT_NAME='/FORCED_PREFIX/'):
  139. request = WSGIRequest({'PATH_INFO': '/somepath/', 'REQUEST_METHOD': 'get', 'wsgi.input': BytesIO(b'')})
  140. self.assertEqual(request.path, '/FORCED_PREFIX/somepath/')
  141. # Without trailing slash
  142. with override_settings(FORCE_SCRIPT_NAME='/FORCED_PREFIX'):
  143. request = WSGIRequest({'PATH_INFO': '/somepath/', 'REQUEST_METHOD': 'get', 'wsgi.input': BytesIO(b'')})
  144. self.assertEqual(request.path, '/FORCED_PREFIX/somepath/')
  145. def test_wsgirequest_repr(self):
  146. request = WSGIRequest({'REQUEST_METHOD': 'get', 'wsgi.input': BytesIO(b'')})
  147. self.assertEqual(repr(request), "<WSGIRequest: GET '/'>")
  148. request = WSGIRequest({'PATH_INFO': '/somepath/', 'REQUEST_METHOD': 'get', 'wsgi.input': BytesIO(b'')})
  149. request.GET = {'get-key': 'get-value'}
  150. request.POST = {'post-key': 'post-value'}
  151. request.COOKIES = {'post-key': 'post-value'}
  152. request.META = {'post-key': 'post-value'}
  153. self.assertEqual(repr(request), "<WSGIRequest: GET '/somepath/'>")
  154. def test_wsgirequest_path_info(self):
  155. def wsgi_str(path_info, encoding='utf-8'):
  156. path_info = path_info.encode(encoding) # Actual URL sent by the browser (bytestring)
  157. path_info = path_info.decode('iso-8859-1') # Value in the WSGI environ dict (native string)
  158. return path_info
  159. # Regression for #19468
  160. request = WSGIRequest({'PATH_INFO': wsgi_str("/سلام/"), 'REQUEST_METHOD': 'get', 'wsgi.input': BytesIO(b'')})
  161. self.assertEqual(request.path, "/سلام/")
  162. # The URL may be incorrectly encoded in a non-UTF-8 encoding (#26971)
  163. request = WSGIRequest({
  164. 'PATH_INFO': wsgi_str("/café/", encoding='iso-8859-1'),
  165. 'REQUEST_METHOD': 'get',
  166. 'wsgi.input': BytesIO(b''),
  167. })
  168. # Since it's impossible to decide the (wrong) encoding of the URL, it's
  169. # left percent-encoded in the path.
  170. self.assertEqual(request.path, "/caf%E9/")
  171. def test_httprequest_location(self):
  172. request = HttpRequest()
  173. self.assertEqual(
  174. request.build_absolute_uri(location="https://www.example.com/asdf"),
  175. 'https://www.example.com/asdf'
  176. )
  177. request.get_host = lambda: 'www.example.com'
  178. request.path = ''
  179. self.assertEqual(
  180. request.build_absolute_uri(location="/path/with:colons"),
  181. 'http://www.example.com/path/with:colons'
  182. )
  183. def test_near_expiration(self):
  184. "Cookie will expire when an near expiration time is provided"
  185. response = HttpResponse()
  186. # There is a timing weakness in this test; The
  187. # expected result for max-age requires that there be
  188. # a very slight difference between the evaluated expiration
  189. # time, and the time evaluated in set_cookie(). If this
  190. # difference doesn't exist, the cookie time will be
  191. # 1 second larger. To avoid the problem, put in a quick sleep,
  192. # which guarantees that there will be a time difference.
  193. expires = datetime.utcnow() + timedelta(seconds=10)
  194. time.sleep(0.001)
  195. response.set_cookie('datetime', expires=expires)
  196. datetime_cookie = response.cookies['datetime']
  197. self.assertEqual(datetime_cookie['max-age'], 10)
  198. def test_aware_expiration(self):
  199. "Cookie accepts an aware datetime as expiration time"
  200. response = HttpResponse()
  201. expires = (datetime.utcnow() + timedelta(seconds=10)).replace(tzinfo=utc)
  202. time.sleep(0.001)
  203. response.set_cookie('datetime', expires=expires)
  204. datetime_cookie = response.cookies['datetime']
  205. self.assertEqual(datetime_cookie['max-age'], 10)
  206. def test_create_cookie_after_deleting_cookie(self):
  207. """
  208. Setting a cookie after deletion should clear the expiry date.
  209. """
  210. response = HttpResponse()
  211. response.set_cookie('c', 'old-value')
  212. self.assertEqual(response.cookies['c']['expires'], '')
  213. response.delete_cookie('c')
  214. self.assertEqual(response.cookies['c']['expires'], 'Thu, 01-Jan-1970 00:00:00 GMT')
  215. response.set_cookie('c', 'new-value')
  216. self.assertEqual(response.cookies['c']['expires'], '')
  217. def test_far_expiration(self):
  218. "Cookie will expire when a distant expiration time is provided"
  219. response = HttpResponse()
  220. response.set_cookie('datetime', expires=datetime(2028, 1, 1, 4, 5, 6))
  221. datetime_cookie = response.cookies['datetime']
  222. self.assertIn(
  223. datetime_cookie['expires'],
  224. # assertIn accounts for slight time dependency (#23450)
  225. ('Sat, 01-Jan-2028 04:05:06 GMT', 'Sat, 01-Jan-2028 04:05:07 GMT')
  226. )
  227. def test_max_age_expiration(self):
  228. "Cookie will expire if max_age is provided"
  229. response = HttpResponse()
  230. set_cookie_time = time.time()
  231. with freeze_time(set_cookie_time):
  232. response.set_cookie('max_age', max_age=10)
  233. max_age_cookie = response.cookies['max_age']
  234. self.assertEqual(max_age_cookie['max-age'], 10)
  235. self.assertEqual(max_age_cookie['expires'], cookie_date(set_cookie_time + 10))
  236. def test_httponly_cookie(self):
  237. response = HttpResponse()
  238. response.set_cookie('example', httponly=True)
  239. example_cookie = response.cookies['example']
  240. # A compat cookie may be in use -- check that it has worked
  241. # both as an output string, and using the cookie attributes
  242. self.assertIn('; %s' % cookies.Morsel._reserved['httponly'], str(example_cookie))
  243. self.assertTrue(example_cookie['httponly'])
  244. def test_unicode_cookie(self):
  245. "Verify HttpResponse.set_cookie() works with unicode data."
  246. response = HttpResponse()
  247. cookie_value = '清風'
  248. response.set_cookie('test', cookie_value)
  249. self.assertEqual(cookie_value, response.cookies['test'].value)
  250. def test_limited_stream(self):
  251. # Read all of a limited stream
  252. stream = LimitedStream(BytesIO(b'test'), 2)
  253. self.assertEqual(stream.read(), b'te')
  254. # Reading again returns nothing.
  255. self.assertEqual(stream.read(), b'')
  256. # Read a number of characters greater than the stream has to offer
  257. stream = LimitedStream(BytesIO(b'test'), 2)
  258. self.assertEqual(stream.read(5), b'te')
  259. # Reading again returns nothing.
  260. self.assertEqual(stream.readline(5), b'')
  261. # Read sequentially from a stream
  262. stream = LimitedStream(BytesIO(b'12345678'), 8)
  263. self.assertEqual(stream.read(5), b'12345')
  264. self.assertEqual(stream.read(5), b'678')
  265. # Reading again returns nothing.
  266. self.assertEqual(stream.readline(5), b'')
  267. # Read lines from a stream
  268. stream = LimitedStream(BytesIO(b'1234\n5678\nabcd\nefgh\nijkl'), 24)
  269. # Read a full line, unconditionally
  270. self.assertEqual(stream.readline(), b'1234\n')
  271. # Read a number of characters less than a line
  272. self.assertEqual(stream.readline(2), b'56')
  273. # Read the rest of the partial line
  274. self.assertEqual(stream.readline(), b'78\n')
  275. # Read a full line, with a character limit greater than the line length
  276. self.assertEqual(stream.readline(6), b'abcd\n')
  277. # Read the next line, deliberately terminated at the line end
  278. self.assertEqual(stream.readline(4), b'efgh')
  279. # Read the next line... just the line end
  280. self.assertEqual(stream.readline(), b'\n')
  281. # Read everything else.
  282. self.assertEqual(stream.readline(), b'ijkl')
  283. # Regression for #15018
  284. # If a stream contains a newline, but the provided length
  285. # is less than the number of provided characters, the newline
  286. # doesn't reset the available character count
  287. stream = LimitedStream(BytesIO(b'1234\nabcdef'), 9)
  288. self.assertEqual(stream.readline(10), b'1234\n')
  289. self.assertEqual(stream.readline(3), b'abc')
  290. # Now expire the available characters
  291. self.assertEqual(stream.readline(3), b'd')
  292. # Reading again returns nothing.
  293. self.assertEqual(stream.readline(2), b'')
  294. # Same test, but with read, not readline.
  295. stream = LimitedStream(BytesIO(b'1234\nabcdef'), 9)
  296. self.assertEqual(stream.read(6), b'1234\na')
  297. self.assertEqual(stream.read(2), b'bc')
  298. self.assertEqual(stream.read(2), b'd')
  299. self.assertEqual(stream.read(2), b'')
  300. self.assertEqual(stream.read(), b'')
  301. def test_stream(self):
  302. payload = FakePayload('name=value')
  303. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  304. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  305. 'CONTENT_LENGTH': len(payload),
  306. 'wsgi.input': payload})
  307. self.assertEqual(request.read(), b'name=value')
  308. def test_read_after_value(self):
  309. """
  310. Reading from request is allowed after accessing request contents as
  311. POST or body.
  312. """
  313. payload = FakePayload('name=value')
  314. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  315. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  316. 'CONTENT_LENGTH': len(payload),
  317. 'wsgi.input': payload})
  318. self.assertEqual(request.POST, {'name': ['value']})
  319. self.assertEqual(request.body, b'name=value')
  320. self.assertEqual(request.read(), b'name=value')
  321. def test_value_after_read(self):
  322. """
  323. Construction of POST or body is not allowed after reading
  324. from request.
  325. """
  326. payload = FakePayload('name=value')
  327. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  328. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  329. 'CONTENT_LENGTH': len(payload),
  330. 'wsgi.input': payload})
  331. self.assertEqual(request.read(2), b'na')
  332. with self.assertRaises(RawPostDataException):
  333. request.body
  334. self.assertEqual(request.POST, {})
  335. def test_non_ascii_POST(self):
  336. payload = FakePayload(urlencode({'key': 'España'}))
  337. request = WSGIRequest({
  338. 'REQUEST_METHOD': 'POST',
  339. 'CONTENT_LENGTH': len(payload),
  340. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  341. 'wsgi.input': payload,
  342. })
  343. self.assertEqual(request.POST, {'key': ['España']})
  344. def test_alternate_charset_POST(self):
  345. """
  346. Test a POST with non-utf-8 payload encoding.
  347. """
  348. payload = FakePayload(urlencode({'key': 'España'.encode('latin-1')}))
  349. request = WSGIRequest({
  350. 'REQUEST_METHOD': 'POST',
  351. 'CONTENT_LENGTH': len(payload),
  352. 'CONTENT_TYPE': 'application/x-www-form-urlencoded; charset=iso-8859-1',
  353. 'wsgi.input': payload,
  354. })
  355. self.assertEqual(request.POST, {'key': ['España']})
  356. def test_body_after_POST_multipart_form_data(self):
  357. """
  358. Reading body after parsing multipart/form-data is not allowed
  359. """
  360. # Because multipart is used for large amounts of data i.e. file uploads,
  361. # we don't want the data held in memory twice, and we don't want to
  362. # silence the error by setting body = '' either.
  363. payload = FakePayload("\r\n".join([
  364. '--boundary',
  365. 'Content-Disposition: form-data; name="name"',
  366. '',
  367. 'value',
  368. '--boundary--'
  369. '']))
  370. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  371. 'CONTENT_TYPE': 'multipart/form-data; boundary=boundary',
  372. 'CONTENT_LENGTH': len(payload),
  373. 'wsgi.input': payload})
  374. self.assertEqual(request.POST, {'name': ['value']})
  375. with self.assertRaises(RawPostDataException):
  376. request.body
  377. def test_body_after_POST_multipart_related(self):
  378. """
  379. Reading body after parsing multipart that isn't form-data is allowed
  380. """
  381. # Ticket #9054
  382. # There are cases in which the multipart data is related instead of
  383. # being a binary upload, in which case it should still be accessible
  384. # via body.
  385. payload_data = b"\r\n".join([
  386. b'--boundary',
  387. b'Content-ID: id; name="name"',
  388. b'',
  389. b'value',
  390. b'--boundary--'
  391. b''])
  392. payload = FakePayload(payload_data)
  393. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  394. 'CONTENT_TYPE': 'multipart/related; boundary=boundary',
  395. 'CONTENT_LENGTH': len(payload),
  396. 'wsgi.input': payload})
  397. self.assertEqual(request.POST, {})
  398. self.assertEqual(request.body, payload_data)
  399. def test_POST_multipart_with_content_length_zero(self):
  400. """
  401. Multipart POST requests with Content-Length >= 0 are valid and need to be handled.
  402. """
  403. # According to:
  404. # http://www.w3.org/Protocols/rfc2616/rfc2616-sec14.html#sec14.13
  405. # Every request.POST with Content-Length >= 0 is a valid request,
  406. # this test ensures that we handle Content-Length == 0.
  407. payload = FakePayload("\r\n".join([
  408. '--boundary',
  409. 'Content-Disposition: form-data; name="name"',
  410. '',
  411. 'value',
  412. '--boundary--'
  413. '']))
  414. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  415. 'CONTENT_TYPE': 'multipart/form-data; boundary=boundary',
  416. 'CONTENT_LENGTH': 0,
  417. 'wsgi.input': payload})
  418. self.assertEqual(request.POST, {})
  419. def test_POST_binary_only(self):
  420. payload = b'\r\n\x01\x00\x00\x00ab\x00\x00\xcd\xcc,@'
  421. environ = {'REQUEST_METHOD': 'POST',
  422. 'CONTENT_TYPE': 'application/octet-stream',
  423. 'CONTENT_LENGTH': len(payload),
  424. 'wsgi.input': BytesIO(payload)}
  425. request = WSGIRequest(environ)
  426. self.assertEqual(request.POST, {})
  427. self.assertEqual(request.FILES, {})
  428. self.assertEqual(request.body, payload)
  429. # Same test without specifying content-type
  430. environ.update({'CONTENT_TYPE': '', 'wsgi.input': BytesIO(payload)})
  431. request = WSGIRequest(environ)
  432. self.assertEqual(request.POST, {})
  433. self.assertEqual(request.FILES, {})
  434. self.assertEqual(request.body, payload)
  435. def test_read_by_lines(self):
  436. payload = FakePayload('name=value')
  437. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  438. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  439. 'CONTENT_LENGTH': len(payload),
  440. 'wsgi.input': payload})
  441. self.assertEqual(list(request), [b'name=value'])
  442. def test_POST_after_body_read(self):
  443. """
  444. POST should be populated even if body is read first
  445. """
  446. payload = FakePayload('name=value')
  447. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  448. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  449. 'CONTENT_LENGTH': len(payload),
  450. 'wsgi.input': payload})
  451. request.body # evaluate
  452. self.assertEqual(request.POST, {'name': ['value']})
  453. def test_POST_after_body_read_and_stream_read(self):
  454. """
  455. POST should be populated even if body is read first, and then
  456. the stream is read second.
  457. """
  458. payload = FakePayload('name=value')
  459. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  460. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  461. 'CONTENT_LENGTH': len(payload),
  462. 'wsgi.input': payload})
  463. request.body # evaluate
  464. self.assertEqual(request.read(1), b'n')
  465. self.assertEqual(request.POST, {'name': ['value']})
  466. def test_POST_after_body_read_and_stream_read_multipart(self):
  467. """
  468. POST should be populated even if body is read first, and then
  469. the stream is read second. Using multipart/form-data instead of urlencoded.
  470. """
  471. payload = FakePayload("\r\n".join([
  472. '--boundary',
  473. 'Content-Disposition: form-data; name="name"',
  474. '',
  475. 'value',
  476. '--boundary--'
  477. '']))
  478. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  479. 'CONTENT_TYPE': 'multipart/form-data; boundary=boundary',
  480. 'CONTENT_LENGTH': len(payload),
  481. 'wsgi.input': payload})
  482. request.body # evaluate
  483. # Consume enough data to mess up the parsing:
  484. self.assertEqual(request.read(13), b'--boundary\r\nC')
  485. self.assertEqual(request.POST, {'name': ['value']})
  486. def test_POST_immutable_for_mutipart(self):
  487. """
  488. MultiPartParser.parse() leaves request.POST immutable.
  489. """
  490. payload = FakePayload("\r\n".join([
  491. '--boundary',
  492. 'Content-Disposition: form-data; name="name"',
  493. '',
  494. 'value',
  495. '--boundary--',
  496. ]))
  497. request = WSGIRequest({
  498. 'REQUEST_METHOD': 'POST',
  499. 'CONTENT_TYPE': 'multipart/form-data; boundary=boundary',
  500. 'CONTENT_LENGTH': len(payload),
  501. 'wsgi.input': payload,
  502. })
  503. self.assertFalse(request.POST._mutable)
  504. def test_POST_connection_error(self):
  505. """
  506. If wsgi.input.read() raises an exception while trying to read() the
  507. POST, the exception should be identifiable (not a generic IOError).
  508. """
  509. class ExplodingBytesIO(BytesIO):
  510. def read(self, len=0):
  511. raise IOError("kaboom!")
  512. payload = b'name=value'
  513. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  514. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  515. 'CONTENT_LENGTH': len(payload),
  516. 'wsgi.input': ExplodingBytesIO(payload)})
  517. with self.assertRaises(UnreadablePostError):
  518. request.body
  519. def test_set_encoding_clears_POST(self):
  520. payload = FakePayload('name=Hello Günter')
  521. request = WSGIRequest({
  522. 'REQUEST_METHOD': 'POST',
  523. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  524. 'CONTENT_LENGTH': len(payload),
  525. 'wsgi.input': payload,
  526. })
  527. self.assertEqual(request.POST, {'name': ['Hello Günter']})
  528. request.encoding = 'iso-8859-16'
  529. self.assertEqual(request.POST, {'name': ['Hello GĂŒnter']})
  530. def test_set_encoding_clears_GET(self):
  531. request = WSGIRequest({
  532. 'REQUEST_METHOD': 'GET',
  533. 'wsgi.input': '',
  534. 'QUERY_STRING': 'name=Hello%20G%C3%BCnter',
  535. })
  536. self.assertEqual(request.GET, {'name': ['Hello Günter']})
  537. request.encoding = 'iso-8859-16'
  538. self.assertEqual(request.GET, {'name': ['Hello G\u0102\u0152nter']})
  539. def test_FILES_connection_error(self):
  540. """
  541. If wsgi.input.read() raises an exception while trying to read() the
  542. FILES, the exception should be identifiable (not a generic IOError).
  543. """
  544. class ExplodingBytesIO(BytesIO):
  545. def read(self, len=0):
  546. raise IOError("kaboom!")
  547. payload = b'x'
  548. request = WSGIRequest({'REQUEST_METHOD': 'POST',
  549. 'CONTENT_TYPE': 'multipart/form-data; boundary=foo_',
  550. 'CONTENT_LENGTH': len(payload),
  551. 'wsgi.input': ExplodingBytesIO(payload)})
  552. with self.assertRaises(UnreadablePostError):
  553. request.FILES
  554. @override_settings(ALLOWED_HOSTS=['example.com'])
  555. def test_get_raw_uri(self):
  556. factory = RequestFactory(HTTP_HOST='evil.com')
  557. request = factory.get('////absolute-uri')
  558. self.assertEqual(request.get_raw_uri(), 'http://evil.com//absolute-uri')
  559. request = factory.get('/?foo=bar')
  560. self.assertEqual(request.get_raw_uri(), 'http://evil.com/?foo=bar')
  561. request = factory.get('/path/with:colons')
  562. self.assertEqual(request.get_raw_uri(), 'http://evil.com/path/with:colons')
  563. class HostValidationTests(SimpleTestCase):
  564. poisoned_hosts = [
  565. 'example.com@evil.tld',
  566. 'example.com:dr.frankenstein@evil.tld',
  567. 'example.com:dr.frankenstein@evil.tld:80',
  568. 'example.com:80/badpath',
  569. 'example.com: recovermypassword.com',
  570. ]
  571. @override_settings(
  572. USE_X_FORWARDED_HOST=False,
  573. ALLOWED_HOSTS=[
  574. 'forward.com', 'example.com', 'internal.com', '12.34.56.78',
  575. '[2001:19f0:feee::dead:beef:cafe]', 'xn--4ca9at.com',
  576. '.multitenant.com', 'INSENSITIVE.com', '[::ffff:169.254.169.254]',
  577. ])
  578. def test_http_get_host(self):
  579. # Check if X_FORWARDED_HOST is provided.
  580. request = HttpRequest()
  581. request.META = {
  582. 'HTTP_X_FORWARDED_HOST': 'forward.com',
  583. 'HTTP_HOST': 'example.com',
  584. 'SERVER_NAME': 'internal.com',
  585. 'SERVER_PORT': 80,
  586. }
  587. # X_FORWARDED_HOST is ignored.
  588. self.assertEqual(request.get_host(), 'example.com')
  589. # Check if X_FORWARDED_HOST isn't provided.
  590. request = HttpRequest()
  591. request.META = {
  592. 'HTTP_HOST': 'example.com',
  593. 'SERVER_NAME': 'internal.com',
  594. 'SERVER_PORT': 80,
  595. }
  596. self.assertEqual(request.get_host(), 'example.com')
  597. # Check if HTTP_HOST isn't provided.
  598. request = HttpRequest()
  599. request.META = {
  600. 'SERVER_NAME': 'internal.com',
  601. 'SERVER_PORT': 80,
  602. }
  603. self.assertEqual(request.get_host(), 'internal.com')
  604. # Check if HTTP_HOST isn't provided, and we're on a nonstandard port
  605. request = HttpRequest()
  606. request.META = {
  607. 'SERVER_NAME': 'internal.com',
  608. 'SERVER_PORT': 8042,
  609. }
  610. self.assertEqual(request.get_host(), 'internal.com:8042')
  611. legit_hosts = [
  612. 'example.com',
  613. 'example.com:80',
  614. '12.34.56.78',
  615. '12.34.56.78:443',
  616. '[2001:19f0:feee::dead:beef:cafe]',
  617. '[2001:19f0:feee::dead:beef:cafe]:8080',
  618. 'xn--4ca9at.com', # Punycode for öäü.com
  619. 'anything.multitenant.com',
  620. 'multitenant.com',
  621. 'insensitive.com',
  622. 'example.com.',
  623. 'example.com.:80',
  624. '[::ffff:169.254.169.254]',
  625. ]
  626. for host in legit_hosts:
  627. request = HttpRequest()
  628. request.META = {
  629. 'HTTP_HOST': host,
  630. }
  631. request.get_host()
  632. # Poisoned host headers are rejected as suspicious
  633. for host in chain(self.poisoned_hosts, ['other.com', 'example.com..']):
  634. with self.assertRaises(DisallowedHost):
  635. request = HttpRequest()
  636. request.META = {
  637. 'HTTP_HOST': host,
  638. }
  639. request.get_host()
  640. @override_settings(USE_X_FORWARDED_HOST=True, ALLOWED_HOSTS=['*'])
  641. def test_http_get_host_with_x_forwarded_host(self):
  642. # Check if X_FORWARDED_HOST is provided.
  643. request = HttpRequest()
  644. request.META = {
  645. 'HTTP_X_FORWARDED_HOST': 'forward.com',
  646. 'HTTP_HOST': 'example.com',
  647. 'SERVER_NAME': 'internal.com',
  648. 'SERVER_PORT': 80,
  649. }
  650. # X_FORWARDED_HOST is obeyed.
  651. self.assertEqual(request.get_host(), 'forward.com')
  652. # Check if X_FORWARDED_HOST isn't provided.
  653. request = HttpRequest()
  654. request.META = {
  655. 'HTTP_HOST': 'example.com',
  656. 'SERVER_NAME': 'internal.com',
  657. 'SERVER_PORT': 80,
  658. }
  659. self.assertEqual(request.get_host(), 'example.com')
  660. # Check if HTTP_HOST isn't provided.
  661. request = HttpRequest()
  662. request.META = {
  663. 'SERVER_NAME': 'internal.com',
  664. 'SERVER_PORT': 80,
  665. }
  666. self.assertEqual(request.get_host(), 'internal.com')
  667. # Check if HTTP_HOST isn't provided, and we're on a nonstandard port
  668. request = HttpRequest()
  669. request.META = {
  670. 'SERVER_NAME': 'internal.com',
  671. 'SERVER_PORT': 8042,
  672. }
  673. self.assertEqual(request.get_host(), 'internal.com:8042')
  674. # Poisoned host headers are rejected as suspicious
  675. legit_hosts = [
  676. 'example.com',
  677. 'example.com:80',
  678. '12.34.56.78',
  679. '12.34.56.78:443',
  680. '[2001:19f0:feee::dead:beef:cafe]',
  681. '[2001:19f0:feee::dead:beef:cafe]:8080',
  682. 'xn--4ca9at.com', # Punycode for öäü.com
  683. ]
  684. for host in legit_hosts:
  685. request = HttpRequest()
  686. request.META = {
  687. 'HTTP_HOST': host,
  688. }
  689. request.get_host()
  690. for host in self.poisoned_hosts:
  691. with self.assertRaises(DisallowedHost):
  692. request = HttpRequest()
  693. request.META = {
  694. 'HTTP_HOST': host,
  695. }
  696. request.get_host()
  697. @override_settings(USE_X_FORWARDED_PORT=False)
  698. def test_get_port(self):
  699. request = HttpRequest()
  700. request.META = {
  701. 'SERVER_PORT': '8080',
  702. 'HTTP_X_FORWARDED_PORT': '80',
  703. }
  704. # Shouldn't use the X-Forwarded-Port header
  705. self.assertEqual(request.get_port(), '8080')
  706. request = HttpRequest()
  707. request.META = {
  708. 'SERVER_PORT': '8080',
  709. }
  710. self.assertEqual(request.get_port(), '8080')
  711. @override_settings(USE_X_FORWARDED_PORT=True)
  712. def test_get_port_with_x_forwarded_port(self):
  713. request = HttpRequest()
  714. request.META = {
  715. 'SERVER_PORT': '8080',
  716. 'HTTP_X_FORWARDED_PORT': '80',
  717. }
  718. # Should use the X-Forwarded-Port header
  719. self.assertEqual(request.get_port(), '80')
  720. request = HttpRequest()
  721. request.META = {
  722. 'SERVER_PORT': '8080',
  723. }
  724. self.assertEqual(request.get_port(), '8080')
  725. @override_settings(DEBUG=True, ALLOWED_HOSTS=[])
  726. def test_host_validation_in_debug_mode(self):
  727. """
  728. If ALLOWED_HOSTS is empty and DEBUG is True, variants of localhost are
  729. allowed.
  730. """
  731. valid_hosts = ['localhost', '127.0.0.1', '[::1]']
  732. for host in valid_hosts:
  733. request = HttpRequest()
  734. request.META = {'HTTP_HOST': host}
  735. self.assertEqual(request.get_host(), host)
  736. # Other hostnames raise a DisallowedHost.
  737. with self.assertRaises(DisallowedHost):
  738. request = HttpRequest()
  739. request.META = {'HTTP_HOST': 'example.com'}
  740. request.get_host()
  741. @override_settings(ALLOWED_HOSTS=[])
  742. def test_get_host_suggestion_of_allowed_host(self):
  743. """get_host() makes helpful suggestions if a valid-looking host is not in ALLOWED_HOSTS."""
  744. msg_invalid_host = "Invalid HTTP_HOST header: %r."
  745. msg_suggestion = msg_invalid_host + " You may need to add %r to ALLOWED_HOSTS."
  746. msg_suggestion2 = msg_invalid_host + " The domain name provided is not valid according to RFC 1034/1035"
  747. for host in [ # Valid-looking hosts
  748. 'example.com',
  749. '12.34.56.78',
  750. '[2001:19f0:feee::dead:beef:cafe]',
  751. 'xn--4ca9at.com', # Punycode for öäü.com
  752. ]:
  753. request = HttpRequest()
  754. request.META = {'HTTP_HOST': host}
  755. with self.assertRaisesMessage(DisallowedHost, msg_suggestion % (host, host)):
  756. request.get_host()
  757. for domain, port in [ # Valid-looking hosts with a port number
  758. ('example.com', 80),
  759. ('12.34.56.78', 443),
  760. ('[2001:19f0:feee::dead:beef:cafe]', 8080),
  761. ]:
  762. host = '%s:%s' % (domain, port)
  763. request = HttpRequest()
  764. request.META = {'HTTP_HOST': host}
  765. with self.assertRaisesMessage(DisallowedHost, msg_suggestion % (host, domain)):
  766. request.get_host()
  767. for host in self.poisoned_hosts:
  768. request = HttpRequest()
  769. request.META = {'HTTP_HOST': host}
  770. with self.assertRaisesMessage(DisallowedHost, msg_invalid_host % host):
  771. request.get_host()
  772. request = HttpRequest()
  773. request.META = {'HTTP_HOST': "invalid_hostname.com"}
  774. with self.assertRaisesMessage(DisallowedHost, msg_suggestion2 % "invalid_hostname.com"):
  775. request.get_host()
  776. def test_split_domain_port_removes_trailing_dot(self):
  777. domain, port = split_domain_port('example.com.:8080')
  778. self.assertEqual(domain, 'example.com')
  779. self.assertEqual(port, '8080')
  780. class BuildAbsoluteURITestCase(SimpleTestCase):
  781. """
  782. Regression tests for ticket #18314.
  783. """
  784. def setUp(self):
  785. self.factory = RequestFactory()
  786. def test_build_absolute_uri_no_location(self):
  787. """
  788. ``request.build_absolute_uri()`` returns the proper value when
  789. the ``location`` argument is not provided, and ``request.path``
  790. begins with //.
  791. """
  792. # //// is needed to create a request with a path beginning with //
  793. request = self.factory.get('////absolute-uri')
  794. self.assertEqual(
  795. request.build_absolute_uri(),
  796. 'http://testserver//absolute-uri'
  797. )
  798. def test_build_absolute_uri_absolute_location(self):
  799. """
  800. ``request.build_absolute_uri()`` returns the proper value when
  801. an absolute URL ``location`` argument is provided, and ``request.path``
  802. begins with //.
  803. """
  804. # //// is needed to create a request with a path beginning with //
  805. request = self.factory.get('////absolute-uri')
  806. self.assertEqual(
  807. request.build_absolute_uri(location='http://example.com/?foo=bar'),
  808. 'http://example.com/?foo=bar'
  809. )
  810. def test_build_absolute_uri_schema_relative_location(self):
  811. """
  812. ``request.build_absolute_uri()`` returns the proper value when
  813. a schema-relative URL ``location`` argument is provided, and
  814. ``request.path`` begins with //.
  815. """
  816. # //// is needed to create a request with a path beginning with //
  817. request = self.factory.get('////absolute-uri')
  818. self.assertEqual(
  819. request.build_absolute_uri(location='//example.com/?foo=bar'),
  820. 'http://example.com/?foo=bar'
  821. )
  822. def test_build_absolute_uri_relative_location(self):
  823. """
  824. ``request.build_absolute_uri()`` returns the proper value when
  825. a relative URL ``location`` argument is provided, and ``request.path``
  826. begins with //.
  827. """
  828. # //// is needed to create a request with a path beginning with //
  829. request = self.factory.get('////absolute-uri')
  830. self.assertEqual(
  831. request.build_absolute_uri(location='/foo/bar/'),
  832. 'http://testserver/foo/bar/'
  833. )