codekingpro/portable-devtools
114k
1import http
2
3from tornado.concurrent import Future
4from tornado import gen
5from tornado.escape import (
6 json_decode,
7 utf8,
8 to_unicode,
9 recursive_unicode,
10 native_str,
11 to_basestring,
12)
13from tornado.httpclient import HTTPClientError
14from tornado.httputil import format_timestamp
15from tornado.iostream import IOStream
16from tornado import locale
17from tornado.locks import Event
18from tornado.log import app_log, gen_log
19from tornado.simple_httpclient import SimpleAsyncHTTPClient
20from tornado.template import DictLoader
21from tornado.testing import AsyncHTTPTestCase, AsyncTestCase, ExpectLog, gen_test
22from tornado.test.util import ignore_deprecation
23from tornado.util import ObjectDict, unicode_type
24from tornado.web import (
25 Application,
26 RequestHandler,
27 StaticFileHandler,
28 RedirectHandler as WebRedirectHandler,
29 HTTPError,
30 MissingArgumentError,
31 ErrorHandler,
32 authenticated,
33 url,
34 _create_signature_v1,
35 create_signed_value,
36 decode_signed_value,
37 get_signature_key_version,
38 UIModule,
39 Finish,
40 stream_request_body,
41 removeslash,
42 addslash,
43 GZipContentEncoding,
44)
45
46import binascii
47import contextlib
48import copy
49import datetime
50import email.utils
51import gzip
52from io import BytesIO
53import itertools
54import logging
55import os
56import re
57import socket
58import typing # noqa: F401
59import unittest
60import urllib.parse
61
62
63def relpath(*a):
64 return os.path.join(os.path.dirname(__file__), *a)
65
66
67class WebTestCase(AsyncHTTPTestCase):
68 """Base class for web tests that also supports WSGI mode.
69
70 Override get_handlers and get_app_kwargs instead of get_app.
71 This class is deprecated since WSGI mode is no longer supported.
72 """
73
74 def get_app(self):
75 self.app = Application(self.get_handlers(), **self.get_app_kwargs())
76 return self.app
77
78 def get_handlers(self):
79 raise NotImplementedError()
80
81 def get_app_kwargs(self):
82 return {}
83
84
85class SimpleHandlerTestCase(WebTestCase):
86 """Simplified base class for tests that work with a single handler class.
87
88 To use, define a nested class named ``Handler``.
89 """
90
91 Handler = None
92
93 def get_handlers(self):
94 return [("/", self.Handler)]
95
96
97class HelloHandler(RequestHandler):
98 def get(self):
99 self.write("hello")
100
101
102class CookieTestRequestHandler(RequestHandler):
103 # stub out enough methods to make the signed_cookie functions work
104 def __init__(self, cookie_secret="0123456789", key_version=None):
105 # don't call super.__init__
106 self._cookies = {} # type: typing.Dict[str, bytes]
107 if key_version is None:
108 self.application = ObjectDict( # type: ignore
109 settings=dict(cookie_secret=cookie_secret)
110 )
111 else:
112 self.application = ObjectDict( # type: ignore
113 settings=dict(cookie_secret=cookie_secret, key_version=key_version)
114 )
115
116 def get_cookie(self, name):
117 return self._cookies.get(name)
118
119 def set_cookie(self, name, value, expires_days=None):
120 self._cookies[name] = value
121
122
123# See SignedValueTest below for more.
124class SecureCookieV1Test(unittest.TestCase):
125 def test_round_trip(self):
126 handler = CookieTestRequestHandler()
127 handler.set_signed_cookie("foo", b"bar", version=1)
128 self.assertEqual(handler.get_signed_cookie("foo", min_version=1), b"bar")
129
130 def test_cookie_tampering_future_timestamp(self):
131 handler = CookieTestRequestHandler()
132 # this string base64-encodes to '12345678'
133 handler.set_signed_cookie("foo", binascii.a2b_hex(b"d76df8e7aefc"), version=1)
134 cookie = handler._cookies["foo"]
135 match = re.match(rb"12345678\|([0-9]+)\|([0-9a-f]+)", cookie)
136 self.assertIsNotNone(match)
137 assert match is not None # for mypy
138 timestamp = match.group(1)
139 sig = match.group(2)
140 self.assertEqual(
141 _create_signature_v1(
142 handler.application.settings["cookie_secret"],
143 "foo",
144 "12345678",
145 timestamp,
146 ),
147 sig,
148 )
149 # shifting digits from payload to timestamp doesn't alter signature
150 # (this is not desirable behavior, just confirming that that's how it
151 # works)
152 self.assertEqual(
153 _create_signature_v1(
154 handler.application.settings["cookie_secret"],
155 "foo",
156 "1234",
157 b"5678" + timestamp,
158 ),
159 sig,
160 )
161 # tamper with the cookie
162 handler._cookies["foo"] = utf8(
163 f"1234|5678{to_basestring(timestamp)}|{to_basestring(sig)}"
164 )
165 # it gets rejected
166 with ExpectLog(gen_log, "Cookie timestamp in future"):
167 self.assertIsNone(handler.get_signed_cookie("foo", min_version=1))
168
169 def test_arbitrary_bytes(self):
170 # Secure cookies accept arbitrary data (which is base64 encoded).
171 # Note that normal cookies accept only a subset of ascii.
172 handler = CookieTestRequestHandler()
173 handler.set_signed_cookie("foo", b"\xe9", version=1)
174 self.assertEqual(handler.get_signed_cookie("foo", min_version=1), b"\xe9")
175
176
177# See SignedValueTest below for more.
178class SecureCookieV2Test(unittest.TestCase):
179 KEY_VERSIONS = {0: "ajklasdf0ojaisdf", 1: "aslkjasaolwkjsdf"}
180
181 def test_round_trip(self):
182 handler = CookieTestRequestHandler()
183 handler.set_signed_cookie("foo", b"bar", version=2)
184 self.assertEqual(handler.get_signed_cookie("foo", min_version=2), b"bar")
185
186 def test_key_version_roundtrip(self):
187 handler = CookieTestRequestHandler(
188 cookie_secret=self.KEY_VERSIONS, key_version=0
189 )
190 handler.set_signed_cookie("foo", b"bar")
191 self.assertEqual(handler.get_signed_cookie("foo"), b"bar")
192
193 def test_key_version_roundtrip_differing_version(self):
194 handler = CookieTestRequestHandler(
195 cookie_secret=self.KEY_VERSIONS, key_version=1
196 )
197 handler.set_signed_cookie("foo", b"bar")
198 self.assertEqual(handler.get_signed_cookie("foo"), b"bar")
199
200 def test_key_version_increment_version(self):
201 handler = CookieTestRequestHandler(
202 cookie_secret=self.KEY_VERSIONS, key_version=0
203 )
204 handler.set_signed_cookie("foo", b"bar")
205 new_handler = CookieTestRequestHandler(
206 cookie_secret=self.KEY_VERSIONS, key_version=1
207 )
208 new_handler._cookies = handler._cookies
209 self.assertEqual(new_handler.get_signed_cookie("foo"), b"bar")
210
211 def test_key_version_invalidate_version(self):
212 handler = CookieTestRequestHandler(
213 cookie_secret=self.KEY_VERSIONS, key_version=0
214 )
215 handler.set_signed_cookie("foo", b"bar")
216 new_key_versions = self.KEY_VERSIONS.copy()
217 new_key_versions.pop(0)
218 new_handler = CookieTestRequestHandler(
219 cookie_secret=new_key_versions, key_version=1
220 )
221 new_handler._cookies = handler._cookies
222 self.assertEqual(new_handler.get_signed_cookie("foo"), None)
223
224
225class FinalReturnTest(WebTestCase):
226 final_return = None # type: Future
227
228 def get_handlers(self):
229 test = self
230
231 class FinishHandler(RequestHandler):
232 @gen.coroutine
233 def get(self):
234 test.final_return = self.finish()
235 yield test.final_return
236
237 @gen.coroutine
238 def post(self):
239 self.write("hello,")
240 yield self.flush()
241 test.final_return = self.finish("world")
242 yield test.final_return
243
244 class RenderHandler(RequestHandler):
245 def create_template_loader(self, path):
246 return DictLoader({"foo.html": "hi"})
247
248 @gen.coroutine
249 def get(self):
250 test.final_return = self.render("foo.html")
251
252 return [("/finish", FinishHandler), ("/render", RenderHandler)]
253
254 def get_app_kwargs(self):
255 return dict(template_path="FinalReturnTest")
256
257 def test_finish_method_return_future(self):
258 response = self.fetch(self.get_url("/finish"))
259 self.assertEqual(response.code, 200)
260 self.assertIsInstance(self.final_return, Future)
261 self.assertTrue(self.final_return.done())
262
263 response = self.fetch(self.get_url("/finish"), method="POST", body=b"")
264 self.assertEqual(response.code, 200)
265 self.assertIsInstance(self.final_return, Future)
266 self.assertTrue(self.final_return.done())
267
268 def test_render_method_return_future(self):
269 response = self.fetch(self.get_url("/render"))
270 self.assertEqual(response.code, 200)
271 self.assertIsInstance(self.final_return, Future)
272
273
274class CookieTest(WebTestCase):
275 def get_handlers(self):
276 class SetCookieHandler(RequestHandler):
277 def get(self):
278 # Try setting cookies with different argument types
279 # to ensure that everything gets encoded correctly
280 self.set_cookie("str", "asdf")
281 self.set_cookie("unicode", "qwer")
282 self.set_cookie("bytes", b"zxcv")
283
284 class GetCookieHandler(RequestHandler):
285 def get(self):
286 cookie = self.get_cookie("foo", "default")
287 assert cookie is not None
288 self.write(cookie)
289
290 class SetCookieDomainHandler(RequestHandler):
291 def get(self):
292 # unicode domain and path arguments shouldn't break things
293 # either (see bug #285)
294 self.set_cookie("unicode_args", "blah", domain="foo.com", path="/foo")
295
296 class SetCookieSpecialCharHandler(RequestHandler):
297 # "Special" characters are allowed in cookie values, but trigger special quoting.
298 def get(self):
299 self.set_cookie("equals", "a=b")
300 self.set_cookie("semicolon", "a;b")
301 self.set_cookie("quote", 'a"b')
302
303 class SetCookieForbiddenCharHandler(RequestHandler):
304 def get(self):
305 # Control characters and semicolons raise errors in cookie names and attributes
306 # (but not values, which are tested in SetCookieSpecialCharHandler)
307 for char in list(map(chr, range(0x20))) + [chr(0x7F), ";"]:
308 try:
309 self.set_cookie("foo" + char, "bar")
310 self.write(
311 "Didn't get expected exception for char %r in name\n" % char
312 )
313 except http.cookies.CookieError as e:
314 if "Invalid cookie attribute name" not in str(e):
315 self.write(
316 "unexpected exception for char %r in name: %s\n"
317 % (char, e)
318 )
319
320 try:
321 self.set_cookie("foo", "bar", domain="example" + char + ".com")
322 self.write(
323 "Didn't get expected exception for char %r in domain\n"
324 % char
325 )
326 except http.cookies.CookieError as e:
327 if "Invalid cookie attribute domain" not in str(e):
328 self.write(
329 "unexpected exception for char %r in domain: %s\n"
330 % (char, e)
331 )
332
333 try:
334 self.set_cookie("foo", "bar", path="/" + char)
335 self.write(
336 "Didn't get expected exception for char %r in path\n" % char
337 )
338 except http.cookies.CookieError as e:
339 if "Invalid cookie attribute path" not in str(e):
340 self.write(
341 "unexpected exception for char %r in path: %s\n"
342 % (char, e)
343 )
344
345 try:
346 self.set_cookie("foo", "bar", samesite="a" + char)
347 self.write(
348 "Didn't get expected exception for char %r in samesite\n"
349 % char
350 )
351 except http.cookies.CookieError as e:
352 if "Invalid cookie attribute samesite" not in str(e):
353 self.write(
354 "unexpected exception for char %r in samesite: %s\n"
355 % (char, e)
356 )
357
358 class SetCookieOverwriteHandler(RequestHandler):
359 def get(self):
360 self.set_cookie("a", "b", domain="example.com")
361 self.set_cookie("c", "d", domain="example.com")
362 # A second call with the same name clobbers the first.
363 # Attributes from the first call are not carried over.
364 self.set_cookie("a", "e")
365
366 class SetCookieMaxAgeHandler(RequestHandler):
367 def get(self):
368 self.set_cookie("foo", "bar", max_age=10)
369
370 class SetCookieExpiresDaysHandler(RequestHandler):
371 def get(self):
372 self.set_cookie("foo", "bar", expires_days=10)
373
374 class SetCookieFalsyFlags(RequestHandler):
375 def get(self):
376 self.set_cookie("a", "1", secure=True)
377 self.set_cookie("b", "1", secure=False)
378 self.set_cookie("c", "1", httponly=True)
379 self.set_cookie("d", "1", httponly=False)
380
381 class SetCookieDeprecatedArgs(RequestHandler):
382 def get(self):
383 # Mixed case is supported, but deprecated
384 self.set_cookie("a", "b", HttpOnly=True, pATH="/foo")
385
386 return [
387 ("/set", SetCookieHandler),
388 ("/get", GetCookieHandler),
389 ("/set_domain", SetCookieDomainHandler),
390 ("/special_char", SetCookieSpecialCharHandler),
391 ("/forbidden_char", SetCookieForbiddenCharHandler),
392 ("/set_overwrite", SetCookieOverwriteHandler),
393 ("/set_max_age", SetCookieMaxAgeHandler),
394 ("/set_expires_days", SetCookieExpiresDaysHandler),
395 ("/set_falsy_flags", SetCookieFalsyFlags),
396 ("/set_deprecated", SetCookieDeprecatedArgs),
397 ]
398
399 def test_set_cookie(self):
400 response = self.fetch("/set")
401 self.assertEqual(
402 sorted(response.headers.get_list("Set-Cookie")),
403 ["bytes=zxcv; Path=/", "str=asdf; Path=/", "unicode=qwer; Path=/"],
404 )
405
406 def test_get_cookie(self):
407 response = self.fetch("/get", headers={"Cookie": "foo=bar"})
408 self.assertEqual(response.body, b"bar")
409
410 response = self.fetch("/get", headers={"Cookie": 'foo="bar"'})
411 self.assertEqual(response.body, b"bar")
412
413 response = self.fetch("/get", headers={"Cookie": "/=exception;"})
414 self.assertEqual(response.body, b"default")
415
416 def test_set_cookie_domain(self):
417 response = self.fetch("/set_domain")
418 self.assertEqual(
419 response.headers.get_list("Set-Cookie"),
420 ["unicode_args=blah; Domain=foo.com; Path=/foo"],
421 )
422
423 def test_cookie_special_char(self):
424 response = self.fetch("/special_char")
425 headers = sorted(response.headers.get_list("Set-Cookie"))
426 self.assertEqual(len(headers), 3)
427 self.assertEqual(headers[0], 'equals="a=b"; Path=/')
428 self.assertEqual(headers[1], 'quote="a\\"b"; Path=/')
429 # Semicolons are octal-escaped
430 self.assertIn(
431 headers[2],
432 ('semicolon="a;b"; Path=/', 'semicolon="a\\073b"; Path=/'),
433 headers[2],
434 )
435
436 data = [
437 ("foo=a=b", "a=b"),
438 ('foo="a=b"', "a=b"),
439 ('foo="a;b"', '"a'), # even quoted, ";" is a delimiter
440 ("foo=a\\073b", "a\\073b"), # escapes only decoded in quotes
441 ('foo="a\\073b"', "a;b"),
442 ('foo="a\\"b"', 'a"b'),
443 ]
444 for header, expected in data:
445 logging.debug("trying %r", header)
446 response = self.fetch("/get", headers={"Cookie": header})
447 self.assertEqual(response.body, utf8(expected))
448
449 def test_set_cookie_forbidden_char(self):
450 response = self.fetch("/forbidden_char")
451 self.assertEqual(response.code, 200)
452 self.maxDiff = 10000
453 self.assertMultiLineEqual(to_unicode(response.body), "")
454
455 def test_set_cookie_overwrite(self):
456 response = self.fetch("/set_overwrite")
457 headers = response.headers.get_list("Set-Cookie")
458 self.assertEqual(
459 sorted(headers), ["a=e; Path=/", "c=d; Domain=example.com; Path=/"]
460 )
461
462 def test_set_cookie_max_age(self):
463 response = self.fetch("/set_max_age")
464 headers = response.headers.get_list("Set-Cookie")
465 self.assertEqual(sorted(headers), ["foo=bar; Max-Age=10; Path=/"])
466
467 def test_set_cookie_expires_days(self):
468 response = self.fetch("/set_expires_days")
469 header = response.headers.get("Set-Cookie")
470 self.assertIsNotNone(header)
471 assert header is not None # for mypy
472 match = re.match("foo=bar; expires=(?P<expires>.+); Path=/", header)
473 self.assertIsNotNone(match)
474 assert match is not None # for mypy
475
476 expires = datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(
477 days=10
478 )
479 header_expires = email.utils.parsedate_to_datetime(match.groupdict()["expires"])
480 self.assertLess(abs((expires - header_expires).total_seconds()), 10)
481
482 def test_set_cookie_false_flags(self):
483 response = self.fetch("/set_falsy_flags")
484 headers = sorted(response.headers.get_list("Set-Cookie"))
485 self.assertEqual(headers[0], "a=1; Path=/; Secure")
486 self.assertEqual(headers[1], "b=1; Path=/")
487 self.assertEqual(headers[2], "c=1; HttpOnly; Path=/")
488 self.assertEqual(headers[3], "d=1; Path=/")
489
490 def test_set_cookie_deprecated(self):
491 with ignore_deprecation():
492 response = self.fetch("/set_deprecated")
493 header = response.headers.get("Set-Cookie")
494 self.assertEqual(header, "a=b; HttpOnly; Path=/foo")
495
496
497class AuthRedirectRequestHandler(RequestHandler):
498 def initialize(self, login_url):
499 self.login_url = login_url
500
501 def get_login_url(self):
502 return self.login_url
503
504 @authenticated
505 def get(self):
506 # we'll never actually get here because the test doesn't follow redirects
507 self.send_error(500)
508
509
510class AuthRedirectTest(WebTestCase):
511 def get_handlers(self):
512 return [
513 ("/relative", AuthRedirectRequestHandler, dict(login_url="/login")),
514 (
515 "/absolute",
516 AuthRedirectRequestHandler,
517 dict(login_url="http://example.com/login"),
518 ),
519 ]
520
521 def test_relative_auth_redirect(self):
522 response = self.fetch(self.get_url("/relative"), follow_redirects=False)
523 self.assertEqual(response.code, 302)
524 self.assertEqual(response.headers["Location"], "/login?next=%2Frelative")
525
526 def test_absolute_auth_redirect(self):
527 response = self.fetch(self.get_url("/absolute"), follow_redirects=False)
528 self.assertEqual(response.code, 302)
529 self.assertTrue(
530 re.match(
531 r"http://example.com/login\?next=http%3A%2F%2F127.0.0.1%3A[0-9]+%2Fabsolute",
532 response.headers["Location"],
533 ),
534 response.headers["Location"],
535 )
536
537
538class ConnectionCloseHandler(RequestHandler):
539 def initialize(self, test):
540 self.test = test
541
542 @gen.coroutine
543 def get(self):
544 self.test.on_handler_waiting()
545 yield self.test.cleanup_event.wait()
546
547 def on_connection_close(self):
548 self.test.on_connection_close()
549
550
551class ConnectionCloseTest(WebTestCase):
552 def get_handlers(self):
553 self.cleanup_event = Event()
554 return [("/", ConnectionCloseHandler, dict(test=self))]
555
556 def test_connection_close(self):
557 s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0)
558 s.connect(("127.0.0.1", self.get_http_port()))
559 self.stream = IOStream(s)
560 self.stream.write(b"GET / HTTP/1.0\r\n\r\n")
561 self.wait()
562 # Let the hanging coroutine clean up after itself
563 self.cleanup_event.set()
564 self.io_loop.run_sync(lambda: gen.sleep(0))
565
566 def on_handler_waiting(self):
567 logging.debug("handler waiting")
568 self.stream.close()
569
570 def on_connection_close(self):
571 logging.debug("connection closed")
572 self.stop()
573
574
575class EchoHandler(RequestHandler):
576 def get(self, *path_args):
577 # Type checks: web.py interfaces convert argument values to
578 # unicode strings (by default, but see also decode_argument).
579 # In httpserver.py (i.e. self.request.arguments), they're left
580 # as bytes. Keys are always native strings.
581 for key in self.request.arguments:
582 if type(key) is not str:
583 raise Exception("incorrect type for key: %r" % type(key))
584 for bvalue in self.request.arguments[key]:
585 if type(bvalue) is not bytes:
586 raise Exception("incorrect type for value: %r" % type(bvalue))
587 for svalue in self.get_arguments(key):
588 if type(svalue) is not unicode_type:
589 raise Exception("incorrect type for value: %r" % type(svalue))
590 for arg in path_args:
591 if type(arg) is not unicode_type:
592 raise Exception("incorrect type for path arg: %r" % type(arg))
593 self.write(
594 dict(
595 path=self.request.path,
596 path_args=path_args,
597 args=recursive_unicode(self.request.arguments),
598 )
599 )
600
601
602class RequestEncodingTest(WebTestCase):
603 def get_handlers(self):
604 return [("/group/(.*)", EchoHandler), ("/slashes/([^/]*)/([^/]*)", EchoHandler)]
605
606 def fetch_json(self, path):
607 return json_decode(self.fetch(path).body)
608
609 def test_group_question_mark(self):
610 # Ensure that url-encoded question marks are handled properly
611 self.assertEqual(
612 self.fetch_json("/group/%3F"),
613 dict(path="/group/%3F", path_args=["?"], args={}),
614 )
615 self.assertEqual(
616 self.fetch_json("/group/%3F?%3F=%3F"),
617 dict(path="/group/%3F", path_args=["?"], args={"?": ["?"]}),
618 )
619
620 def test_group_encoding(self):
621 # Path components and query arguments should be decoded the same way
622 self.assertEqual(
623 self.fetch_json("/group/%C3%A9?arg=%C3%A9"),
624 {
625 "path": "/group/%C3%A9",
626 "path_args": ["\u00e9"],
627 "args": {"arg": ["\u00e9"]},
628 },
629 )
630
631 def test_slashes(self):
632 # Slashes may be escaped to appear as a single "directory" in the path,
633 # but they are then unescaped when passed to the get() method.
634 self.assertEqual(
635 self.fetch_json("/slashes/foo/bar"),
636 dict(path="/slashes/foo/bar", path_args=["foo", "bar"], args={}),
637 )
638 self.assertEqual(
639 self.fetch_json("/slashes/a%2Fb/c%2Fd"),
640 dict(path="/slashes/a%2Fb/c%2Fd", path_args=["a/b", "c/d"], args={}),
641 )
642
643 def test_error(self):
644 # Percent signs (encoded as %25) should not mess up printf-style
645 # messages in logs
646 with ExpectLog(gen_log, ".*Invalid unicode"):
647 self.fetch("/group/?arg=%25%e9")
648
649
650class TypeCheckHandler(RequestHandler):
651 def prepare(self):
652 self.errors = {} # type: typing.Dict[str, str]
653
654 self.check_type("status", self.get_status(), int)
655
656 # get_argument is an exception from the general rule of using
657 # type str for non-body data mainly for historical reasons.
658 self.check_type("argument", self.get_argument("foo"), unicode_type)
659 self.check_type("cookie_key", list(self.cookies.keys())[0], str)
660 self.check_type("cookie_value", list(self.cookies.values())[0].value, str)
661
662 # Secure cookies return bytes because they can contain arbitrary
663 # data, but regular cookies are native strings.
664 if list(self.cookies.keys()) != ["asdf"]:
665 raise Exception(
666 "unexpected values for cookie keys: %r" % self.cookies.keys()
667 )
668 self.check_type("get_signed_cookie", self.get_signed_cookie("asdf"), bytes)
669 self.check_type("get_cookie", self.get_cookie("asdf"), str)
670
671 self.check_type("xsrf_token", self.xsrf_token, bytes)
672 self.check_type("xsrf_form_html", self.xsrf_form_html(), str)
673
674 self.check_type("reverse_url", self.reverse_url("typecheck", "foo"), str)
675
676 self.check_type("request_summary", self._request_summary(), str)
677
678 def get(self, path_component):
679 # path_component uses type unicode instead of str for consistency
680 # with get_argument()
681 self.check_type("path_component", path_component, unicode_type)
682 self.write(self.errors)
683
684 def post(self, path_component):
685 self.check_type("path_component", path_component, unicode_type)
686 self.write(self.errors)
687
688 def check_type(self, name, obj, expected_type):
689 actual_type = type(obj)
690 if expected_type != actual_type:
691 self.errors[name] = f"expected {expected_type}, got {actual_type}"
692
693
694class DecodeArgHandler(RequestHandler):
695 def decode_argument(self, value, name=None):
696 if type(value) is not bytes:
697 raise Exception("unexpected type for value: %r" % type(value))
698 # use self.request.arguments directly to avoid recursion
699 if "encoding" in self.request.arguments:
700 return value.decode(to_unicode(self.request.arguments["encoding"][0]))
701 else:
702 return value
703
704 def get(self, arg):
705 def describe(s):
706 if type(s) is bytes:
707 return ["bytes", native_str(binascii.b2a_hex(s))]
708 elif type(s) is unicode_type:
709 return ["unicode", s]
710 raise Exception("unknown type")
711
712 self.write({"path": describe(arg), "query": describe(self.get_argument("foo"))})
713
714
715class LinkifyHandler(RequestHandler):
716 def get(self):
717 self.render("linkify.html", message="http://example.com")
718
719
720class UIModuleResourceHandler(RequestHandler):
721 def get(self):
722 self.render("page.html", entries=[1, 2])
723
724
725class OptionalPathHandler(RequestHandler):
726 def get(self, path):
727 self.write({"path": path})
728
729
730class MultiHeaderHandler(RequestHandler):
731 def get(self):
732 self.set_header("x-overwrite", "1")
733 self.set_header("X-Overwrite", 2)
734 self.add_header("x-multi", 3)
735 self.add_header("X-Multi", "4")
736
737
738class RedirectHandler(RequestHandler):
739 def get(self):
740 if self.get_argument("permanent", None) is not None:
741 self.redirect("/", permanent=bool(int(self.get_argument("permanent"))))
742 elif self.get_argument("status", None) is not None:
743 self.redirect("/", status=int(self.get_argument("status")))
744 else:
745 raise Exception("didn't get permanent or status arguments")
746
747
748class EmptyFlushCallbackHandler(RequestHandler):
749 @gen.coroutine
750 def get(self):
751 # Ensure that the flush callback is run whether or not there
752 # was any output. The gen.Task and direct yield forms are
753 # equivalent.
754 yield self.flush() # "empty" flush, but writes headers
755 yield self.flush() # empty flush
756 self.write("o")
757 yield self.flush() # flushes the "o"
758 yield self.flush() # empty flush
759 self.finish("k")
760
761
762class HeaderInjectionHandler(RequestHandler):
763 def get(self):
764 try:
765 self.set_header("X-Foo", "foo\r\nX-Bar: baz")
766 raise Exception("Didn't get expected exception")
767 except ValueError as e:
768 if "Unsafe header value" in str(e):
769 self.finish(b"ok")
770 else:
771 raise
772
773
774class SetHeaderHandler(RequestHandler):
775 def get(self):
776 # tests the validity of web.RequestHandler._VALID_HEADER_CHARS
777 illegal_chars = [chr(o) for o in range(0, 0x20)]
778 illegal_chars.append(chr(0x7F))
779 illegal_chars.remove("\t")
780 for char in illegal_chars:
781 try:
782 self.set_header("X-Foo", "foo" + char + "bar")
783 raise Exception("Didn't get expected exception")
784 except ValueError as e:
785 if "Unsafe header value" not in str(e):
786 raise
787
788 # an empty header value is valid as well
789 self.set_header("X-Foo", "")
790
791 self.finish(b"ok")
792
793
794class GetArgumentHandler(RequestHandler):
795 def prepare(self):
796 if self.get_argument("source", None) == "query":
797 method = self.get_query_argument
798 elif self.get_argument("source", None) == "body":
799 method = self.get_body_argument
800 else:
801 method = self.get_argument # type: ignore
802 self.finish(method("foo", "default"))
803
804
805class GetArgumentsHandler(RequestHandler):
806 def prepare(self):
807 self.finish(
808 dict(
809 default=self.get_arguments("foo"),
810 query=self.get_query_arguments("foo"),
811 body=self.get_body_arguments("foo"),
812 )
813 )
814
815
816# This test was shared with wsgi_test.py; now the name is meaningless.
817class WSGISafeWebTest(WebTestCase):
818 COOKIE_SECRET = "WebTest.COOKIE_SECRET"
819
820 def get_app_kwargs(self):
821 loader = DictLoader(
822 {
823 "linkify.html": "{% module linkify(message) %}",
824 "page.html": """\
825<html><head></head><body>
826{% for e in entries %}
827{% module Template("entry.html", entry=e) %}
828{% end %}
829</body></html>""",
830 "entry.html": """\
831{{ set_resources(embedded_css=".entry { margin-bottom: 1em; }",
832 embedded_javascript="js_embed()",
833 css_files=["/base.css", "/foo.css"],
834 javascript_files="/common.js",
835 html_head="<meta>",
836 html_body='<script src="/analytics.js"/>') }}
837<div class="entry">...</div>""",
838 }
839 )
840 return dict(
841 template_loader=loader,
842 autoescape="xhtml_escape",
843 cookie_secret=self.COOKIE_SECRET,
844 )
845
846 def tearDown(self):
847 super().tearDown()
848 RequestHandler._template_loaders.clear()
849
850 def get_handlers(self):
851 urls = [
852 url("/typecheck/(.*)", TypeCheckHandler, name="typecheck"),
853 url("/decode_arg/(.*)", DecodeArgHandler, name="decode_arg"),
854 url("/decode_arg_kw/(?P<arg>.*)", DecodeArgHandler),
855 url("/linkify", LinkifyHandler),
856 url("/uimodule_resources", UIModuleResourceHandler),
857 url("/optional_path/(.+)?", OptionalPathHandler),
858 url("/multi_header", MultiHeaderHandler),
859 url("/redirect", RedirectHandler),
860 url(
861 "/web_redirect_permanent",
862 WebRedirectHandler,
863 {"url": "/web_redirect_newpath"},
864 ),
865 url(
866 "/web_redirect",
867 WebRedirectHandler,
868 {"url": "/web_redirect_newpath", "permanent": False},
869 ),
870 url(
871 "//web_redirect_double_slash",
872 WebRedirectHandler,
873 {"url": "/web_redirect_newpath"},
874 ),
875 url("/header_injection", HeaderInjectionHandler),
876 url("/get_argument", GetArgumentHandler),
877 url("/get_arguments", GetArgumentsHandler),
878 url("/set_header", SetHeaderHandler),
879 ]
880 return urls
881
882 def fetch_json(self, *args, **kwargs):
883 response = self.fetch(*args, **kwargs)
884 response.rethrow()
885 return json_decode(response.body)
886
887 def test_types(self):
888 cookie_value = to_unicode(
889 create_signed_value(self.COOKIE_SECRET, "asdf", "qwer")
890 )
891 response = self.fetch(
892 "/typecheck/asdf?foo=bar", headers={"Cookie": "asdf=" + cookie_value}
893 )
894 data = json_decode(response.body)
895 self.assertEqual(data, {})
896
897 response = self.fetch(
898 "/typecheck/asdf?foo=bar",
899 method="POST",
900 headers={"Cookie": "asdf=" + cookie_value},
901 body="foo=bar",
902 )
903
904 def test_decode_argument(self):
905 # These urls all decode to the same thing
906 urls = [
907 "/decode_arg/%C3%A9?foo=%C3%A9&encoding=utf-8",
908 "/decode_arg/%E9?foo=%E9&encoding=latin1",
909 "/decode_arg_kw/%E9?foo=%E9&encoding=latin1",
910 ]
911 for req_url in urls:
912 response = self.fetch(req_url)
913 response.rethrow()
914 data = json_decode(response.body)
915 self.assertEqual(
916 data,
917 {"path": ["unicode", "\u00e9"], "query": ["unicode", "\u00e9"]},
918 )
919
920 response = self.fetch("/decode_arg/%C3%A9?foo=%C3%A9")
921 response.rethrow()
922 data = json_decode(response.body)
923 self.assertEqual(data, {"path": ["bytes", "c3a9"], "query": ["bytes", "c3a9"]})
924
925 def test_decode_argument_invalid_unicode(self):
926 # test that invalid unicode in URLs causes 400, not 500
927 with ExpectLog(gen_log, ".*Invalid unicode.*"):
928 response = self.fetch("/typecheck/invalid%FF")
929 self.assertEqual(response.code, 400)
930 response = self.fetch("/typecheck/invalid?foo=%FF")
931 self.assertEqual(response.code, 400)
932
933 def test_decode_argument_plus(self):
934 # These urls are all equivalent.
935 urls = [
936 "/decode_arg/1%20%2B%201?foo=1%20%2B%201&encoding=utf-8",
937 "/decode_arg/1%20+%201?foo=1+%2B+1&encoding=utf-8",
938 ]
939 for req_url in urls:
940 response = self.fetch(req_url)
941 response.rethrow()
942 data = json_decode(response.body)
943 self.assertEqual(
944 data,
945 {"path": ["unicode", "1 + 1"], "query": ["unicode", "1 + 1"]},
946 )
947
948 def test_reverse_url(self):
949 self.assertEqual(self.app.reverse_url("decode_arg", "foo"), "/decode_arg/foo")
950 self.assertEqual(self.app.reverse_url("decode_arg", 42), "/decode_arg/42")
951 self.assertEqual(self.app.reverse_url("decode_arg", b"\xe9"), "/decode_arg/%E9")
952 self.assertEqual(
953 self.app.reverse_url("decode_arg", "\u00e9"), "/decode_arg/%C3%A9"
954 )
955 self.assertEqual(
956 self.app.reverse_url("decode_arg", "1 + 1"), "/decode_arg/1%20%2B%201"
957 )
958
959 def test_uimodule_unescaped(self):
960 response = self.fetch("/linkify")
961 self.assertEqual(
962 response.body, b'<a href="http://example.com">http://example.com</a>'
963 )
964
965 def test_uimodule_resources(self):
966 response = self.fetch("/uimodule_resources")
967 self.assertEqual(
968 response.body,
969 b"""\
970<html><head><link href="/base.css" type="text/css" rel="stylesheet"/><link href="/foo.css" type="text/css" rel="stylesheet"/>
971<style type="text/css">
972.entry { margin-bottom: 1em; }
973</style>
974<meta>
975</head><body>
976
977
978<div class="entry">...</div>
979
980
981<div class="entry">...</div>
982
983<script src="/common.js" type="text/javascript"></script>
984<script type="text/javascript">
985//<![CDATA[
986js_embed()
987//]]>
988</script>
989<script src="/analytics.js"/>
990</body></html>""", # noqa: E501
991 )
992
993 def test_optional_path(self):
994 self.assertEqual(self.fetch_json("/optional_path/foo"), {"path": "foo"})
995 self.assertEqual(self.fetch_json("/optional_path/"), {"path": None})
996
997 def test_multi_header(self):
998 response = self.fetch("/multi_header")
999 self.assertEqual(response.headers["x-overwrite"], "2")
1000 self.assertEqual(response.headers.get_list("x-multi"), ["3", "4"])
1001
1002 def test_redirect(self):
1003 response = self.fetch("/redirect?permanent=1", follow_redirects=False)
1004 self.assertEqual(response.code, 301)
1005 response = self.fetch("/redirect?permanent=0", follow_redirects=False)
1006 self.assertEqual(response.code, 302)
1007 response = self.fetch("/redirect?status=307", follow_redirects=False)
1008 self.assertEqual(response.code, 307)
1009
1010 def test_web_redirect(self):
1011 response = self.fetch("/web_redirect_permanent", follow_redirects=False)
1012 self.assertEqual(response.code, 301)
1013 self.assertEqual(response.headers["Location"], "/web_redirect_newpath")
1014 response = self.fetch("/web_redirect", follow_redirects=False)
1015 self.assertEqual(response.code, 302)
1016 self.assertEqual(response.headers["Location"], "/web_redirect_newpath")
1017
1018 def test_web_redirect_double_slash(self):
1019 response = self.fetch("//web_redirect_double_slash", follow_redirects=False)
1020 self.assertEqual(response.code, 301)
1021 self.assertEqual(response.headers["Location"], "/web_redirect_newpath")
1022
1023 def test_header_injection(self):
1024 response = self.fetch("/header_injection")
1025 self.assertEqual(response.body, b"ok")
1026
1027 def test_set_header(self):
1028 response = self.fetch("/set_header")
1029 self.assertEqual(response.body, b"ok")
1030
1031 def test_get_argument(self):
1032 response = self.fetch("/get_argument?foo=bar")
1033 self.assertEqual(response.body, b"bar")
1034 response = self.fetch("/get_argument?foo=")
1035 self.assertEqual(response.body, b"")
1036 response = self.fetch("/get_argument")
1037 self.assertEqual(response.body, b"default")
1038
1039 # Test merging of query and body arguments.
1040 # In singular form, body arguments take precedence over query arguments.
1041 body = urllib.parse.urlencode(dict(foo="hello"))
1042 response = self.fetch("/get_argument?foo=bar", method="POST", body=body)
1043 self.assertEqual(response.body, b"hello")
1044 # In plural methods they are merged.
1045 response = self.fetch("/get_arguments?foo=bar", method="POST", body=body)
1046 self.assertEqual(
1047 json_decode(response.body),
1048 dict(default=["bar", "hello"], query=["bar"], body=["hello"]),
1049 )
1050
1051 def test_get_query_arguments(self):
1052 # send as a post so we can ensure the separation between query
1053 # string and body arguments.
1054 body = urllib.parse.urlencode(dict(foo="hello"))
1055 response = self.fetch(
1056 "/get_argument?source=query&foo=bar", method="POST", body=body
1057 )
1058 self.assertEqual(response.body, b"bar")
1059 response = self.fetch(
1060 "/get_argument?source=query&foo=", method="POST", body=body
1061 )
1062 self.assertEqual(response.body, b"")
1063 response = self.fetch("/get_argument?source=query", method="POST", body=body)
1064 self.assertEqual(response.body, b"default")
1065
1066 def test_get_body_arguments(self):
1067 body = urllib.parse.urlencode(dict(foo="bar"))
1068 response = self.fetch(
1069 "/get_argument?source=body&foo=hello", method="POST", body=body
1070 )
1071 self.assertEqual(response.body, b"bar")
1072
1073 body = urllib.parse.urlencode(dict(foo=""))
1074 response = self.fetch(
1075 "/get_argument?source=body&foo=hello", method="POST", body=body
1076 )
1077 self.assertEqual(response.body, b"")
1078
1079 body = urllib.parse.urlencode(dict())
1080 response = self.fetch(
1081 "/get_argument?source=body&foo=hello", method="POST", body=body
1082 )
1083 self.assertEqual(response.body, b"default")
1084
1085 def test_no_gzip(self):
1086 response = self.fetch("/get_argument")
1087 self.assertNotIn("Accept-Encoding", response.headers.get("Vary", ""))
1088 self.assertNotIn("gzip", response.headers.get("Content-Encoding", ""))
1089
1090
1091class NonWSGIWebTests(WebTestCase):
1092 def get_handlers(self):
1093 return [("/empty_flush", EmptyFlushCallbackHandler)]
1094
1095 def test_empty_flush(self):
1096 response = self.fetch("/empty_flush")
1097 self.assertEqual(response.body, b"ok")
1098
1099
1100class ErrorResponseTest(WebTestCase):
1101 def get_handlers(self):
1102 class DefaultHandler(RequestHandler):
1103 def get(self):
1104 if self.get_argument("status", None):
1105 raise HTTPError(int(self.get_argument("status")))
1106 1 / 0
1107
1108 class WriteErrorHandler(RequestHandler):
1109 def get(self):
1110 if self.get_argument("status", None):
1111 self.send_error(int(self.get_argument("status")))
1112 else:
1113 1 / 0
1114
1115 def write_error(self, status_code, **kwargs):
1116 self.set_header("Content-Type", "text/plain")
1117 if "exc_info" in kwargs:
1118 self.write("Exception: %s" % kwargs["exc_info"][0].__name__)
1119 else:
1120 self.write("Status: %d" % status_code)
1121
1122 class FailedWriteErrorHandler(RequestHandler):
1123 def get(self):
1124 1 / 0
1125
1126 def write_error(self, status_code, **kwargs):
1127 raise Exception("exception in write_error")
1128
1129 return [
1130 url("/default", DefaultHandler),
1131 url("/write_error", WriteErrorHandler),
1132 url("/failed_write_error", FailedWriteErrorHandler),
1133 ]
1134
1135 def test_default(self):
1136 with ExpectLog(app_log, "Uncaught exception"):
1137 response = self.fetch("/default")
1138 self.assertEqual(response.code, 500)
1139 self.assertIn(b"500: Internal Server Error", response.body)
1140
1141 response = self.fetch("/default?status=503")
1142 self.assertEqual(response.code, 503)
1143 self.assertIn(b"503: Service Unavailable", response.body)
1144
1145 response = self.fetch("/default?status=435")
1146 self.assertEqual(response.code, 435)
1147 self.assertIn(b"435: Unknown", response.body)
1148
1149 def test_write_error(self):
1150 with ExpectLog(app_log, "Uncaught exception"):
1151 response = self.fetch("/write_error")
1152 self.assertEqual(response.code, 500)
1153 self.assertEqual(b"Exception: ZeroDivisionError", response.body)
1154
1155 response = self.fetch("/write_error?status=503")
1156 self.assertEqual(response.code, 503)
1157 self.assertEqual(b"Status: 503", response.body)
1158
1159 def test_failed_write_error(self):
1160 with ExpectLog(app_log, "Uncaught exception"):
1161 response = self.fetch("/failed_write_error")
1162 self.assertEqual(response.code, 500)
1163 self.assertEqual(b"", response.body)
1164
1165
1166class StaticFileTest(WebTestCase):
1167 # The expected SHA-512 hash of robots.txt, used in tests that call
1168 # StaticFileHandler.get_version
1169 robots_txt_hash = (
1170 b"63a36e950e134b5217e33c763e88840c10a07d80e6057d92b9ac97508de7fb1f"
1171 b"a6f0e9b7531e169657165ea764e8963399cb6d921ffe6078425aaafe54c04563"
1172 )
1173 static_dir = os.path.join(os.path.dirname(__file__), "static")
1174
1175 def get_handlers(self):
1176 class StaticUrlHandler(RequestHandler):
1177 def get(self, path):
1178 with_v = int(self.get_argument("include_version", "1"))
1179 self.write(self.static_url(path, include_version=with_v))
1180
1181 class AbsoluteStaticUrlHandler(StaticUrlHandler):
1182 include_host = True
1183
1184 class OverrideStaticUrlHandler(RequestHandler):
1185 def get(self, path):
1186 do_include = bool(self.get_argument("include_host"))
1187 self.include_host = not do_include
1188
1189 regular_url = self.static_url(path)
1190 override_url = self.static_url(path, include_host=do_include)
1191 if override_url == regular_url:
1192 return self.write(str(False))
1193
1194 protocol = self.request.protocol + "://"
1195 protocol_length = len(protocol)
1196 check_regular = regular_url.find(protocol, 0, protocol_length)
1197 check_override = override_url.find(protocol, 0, protocol_length)
1198
1199 if do_include:
1200 result = check_override == 0 and check_regular == -1