codekingpro/portable-devtools

1from hashlib import md5
2import unittest
3
4from tornado.escape import utf8
5from tornado.testing import AsyncHTTPTestCase
6from tornado.test import httpclient_test
7from tornado.web import Application, RequestHandler
8
9
10try:
11    import pycurl
12except ImportError:
13    pycurl = None  # type: ignore
14
15if pycurl is not None:
16    from tornado.curl_httpclient import CurlAsyncHTTPClient
17
18
19@unittest.skipIf(pycurl is None, "pycurl module not present")
20class CurlHTTPClientCommonTestCase(httpclient_test.HTTPClientCommonTestCase):
21    def get_http_client(self):
22        client = CurlAsyncHTTPClient(defaults=dict(allow_ipv6=False))
23        # make sure AsyncHTTPClient magic doesn't give us the wrong class
24        self.assertTrue(isinstance(client, CurlAsyncHTTPClient))
25        return client
26
27
28class DigestAuthHandler(RequestHandler):
29    def initialize(self, username, password):
30        self.username = username
31        self.password = password
32
33    def get(self):
34        realm = "test"
35        opaque = "asdf"
36        # Real implementations would use a random nonce.
37        nonce = "1234"
38
39        auth_header = self.request.headers.get("Authorization", None)
40        if auth_header is not None:
41            auth_mode, params = auth_header.split(" ", 1)
42            assert auth_mode == "Digest"
43            param_dict = {}
44            for pair in params.split(","):
45                k, v = pair.strip().split("=", 1)
46                if v[0] == '"' and v[-1] == '"':
47                    v = v[1:-1]
48                param_dict[k] = v
49            assert param_dict["realm"] == realm
50            assert param_dict["opaque"] == opaque
51            assert param_dict["nonce"] == nonce
52            assert param_dict["username"] == self.username
53            assert param_dict["uri"] == self.request.path
54            h1 = md5(utf8(f"{self.username}:{realm}:{self.password}")).hexdigest()
55            h2 = md5(utf8(f"{self.request.method}:{self.request.path}")).hexdigest()
56            digest = md5(utf8(f"{h1}:{nonce}:{h2}")).hexdigest()
57            if digest == param_dict["response"]:
58                self.write("ok")
59            else:
60                self.write("fail")
61        else:
62            self.set_status(401)
63            self.set_header(
64                "WWW-Authenticate",
65                f'Digest realm="{realm}", nonce="{nonce}", opaque="{opaque}"',
66            )
67
68
69class CustomReasonHandler(RequestHandler):
70    def get(self):
71        self.set_status(200, "Custom reason")
72
73
74class CustomFailReasonHandler(RequestHandler):
75    def get(self):
76        self.set_status(400, "Custom reason")
77
78
79@unittest.skipIf(pycurl is None, "pycurl module not present")
80class CurlHTTPClientTestCase(AsyncHTTPTestCase):
81    def setUp(self):
82        super().setUp()
83        self.http_client = self.create_client()
84
85    def get_app(self):
86        return Application(
87            [
88                ("/digest", DigestAuthHandler, {"username": "foo", "password": "bar"}),
89                (
90                    "/digest_non_ascii",
91                    DigestAuthHandler,
92                    {"username": "foo", "password": "barユ£"},
93                ),
94                ("/custom_reason", CustomReasonHandler),
95                ("/custom_fail_reason", CustomFailReasonHandler),
96            ]
97        )
98
99    def create_client(self, **kwargs):
100        return CurlAsyncHTTPClient(
101            force_instance=True, defaults=dict(allow_ipv6=False), **kwargs
102        )
103
104    def test_digest_auth(self):
105        response = self.fetch(
106            "/digest", auth_mode="digest", auth_username="foo", auth_password="bar"
107        )
108        self.assertEqual(response.body, b"ok")
109
110    def test_custom_reason(self):
111        response = self.fetch("/custom_reason")
112        self.assertEqual(response.reason, "Custom reason")
113
114    def test_fail_custom_reason(self):
115        response = self.fetch("/custom_fail_reason")
116        self.assertEqual(str(response.error), "HTTP 400: Custom reason")
117
118    def test_digest_auth_non_ascii(self):
119        response = self.fetch(
120            "/digest_non_ascii",
121            auth_mode="digest",
122            auth_username="foo",
123            auth_password="barユ£",
124        )
125        self.assertEqual(response.body, b"ok")
126