codekingpro/portable-devtools
114k
1# These tests do not currently do much to verify the correct implementation
2# of the openid/oauth protocols, they just exercise the major code paths
3# and ensure that it doesn't blow up (e.g. with unicode/bytes issues in
4# python 3)
5
6import unittest
7
8from tornado.auth import (
9 OpenIdMixin,
10 OAuthMixin,
11 OAuth2Mixin,
12 GoogleOAuth2Mixin,
13 FacebookGraphMixin,
14 TwitterMixin,
15)
16from tornado.escape import json_decode
17from tornado import gen
18from tornado.httpclient import HTTPClientError
19from tornado.httputil import url_concat
20from tornado.log import app_log
21from tornado.testing import AsyncHTTPTestCase, ExpectLog
22from tornado.web import RequestHandler, Application, HTTPError
23
24try:
25 from unittest import mock
26except ImportError:
27 mock = None # type: ignore
28
29
30class OpenIdClientLoginHandler(RequestHandler, OpenIdMixin):
31 def initialize(self, test):
32 self._OPENID_ENDPOINT = test.get_url("/openid/server/authenticate")
33
34 @gen.coroutine
35 def get(self):
36 if self.get_argument("openid.mode", None):
37 user = yield self.get_authenticated_user(
38 http_client=self.settings["http_client"]
39 )
40 if user is None:
41 raise Exception("user is None")
42 self.finish(user)
43 return
44 res = self.authenticate_redirect() # type: ignore
45 assert res is None
46
47
48class OpenIdServerAuthenticateHandler(RequestHandler):
49 def post(self):
50 if self.get_argument("openid.mode") != "check_authentication":
51 raise Exception("incorrect openid.mode %r")
52 self.write("is_valid:true")
53
54
55class OAuth1ClientLoginHandler(RequestHandler, OAuthMixin):
56 def initialize(self, test, version):
57 self._OAUTH_VERSION = version
58 self._OAUTH_REQUEST_TOKEN_URL = test.get_url("/oauth1/server/request_token")
59 self._OAUTH_AUTHORIZE_URL = test.get_url("/oauth1/server/authorize")
60 self._OAUTH_ACCESS_TOKEN_URL = test.get_url("/oauth1/server/access_token")
61
62 def _oauth_consumer_token(self):
63 return dict(key="asdf", secret="qwer")
64
65 @gen.coroutine
66 def get(self):
67 if self.get_argument("oauth_token", None):
68 user = yield self.get_authenticated_user(
69 http_client=self.settings["http_client"]
70 )
71 if user is None:
72 raise Exception("user is None")
73 self.finish(user)
74 return
75 yield self.authorize_redirect(http_client=self.settings["http_client"])
76
77 @gen.coroutine
78 def _oauth_get_user_future(self, access_token):
79 if self.get_argument("fail_in_get_user", None):
80 raise Exception("failing in get_user")
81 if access_token != dict(key="uiop", secret="5678"):
82 raise Exception("incorrect access token %r" % access_token)
83 return dict(email="foo@example.com")
84
85
86class OAuth1ClientLoginCoroutineHandler(OAuth1ClientLoginHandler):
87 """Replaces OAuth1ClientLoginCoroutineHandler's get() with a coroutine."""
88
89 @gen.coroutine
90 def get(self):
91 if self.get_argument("oauth_token", None):
92 # Ensure that any exceptions are set on the returned Future,
93 # not simply thrown into the surrounding StackContext.
94 try:
95 yield self.get_authenticated_user()
96 except Exception as e:
97 self.set_status(503)
98 self.write("got exception: %s" % e)
99 else:
100 yield self.authorize_redirect()
101
102
103class OAuth1ClientRequestParametersHandler(RequestHandler, OAuthMixin):
104 def initialize(self, version):
105 self._OAUTH_VERSION = version
106
107 def _oauth_consumer_token(self):
108 return dict(key="asdf", secret="qwer")
109
110 def get(self):
111 params = self._oauth_request_parameters(
112 "http://www.example.com/api/asdf",
113 dict(key="uiop", secret="5678"),
114 parameters=dict(foo="bar"),
115 )
116 self.write(params)
117
118
119class OAuth1ServerRequestTokenHandler(RequestHandler):
120 def get(self):
121 self.write("oauth_token=zxcv&oauth_token_secret=1234")
122
123
124class OAuth1ServerAccessTokenHandler(RequestHandler):
125 def get(self):
126 self.write("oauth_token=uiop&oauth_token_secret=5678")
127
128
129class OAuth2ClientLoginHandler(RequestHandler, OAuth2Mixin):
130 def initialize(self, test):
131 self._OAUTH_AUTHORIZE_URL = test.get_url("/oauth2/server/authorize")
132
133 def get(self):
134 res = self.authorize_redirect() # type: ignore
135 assert res is None
136
137
138class FacebookClientLoginHandler(RequestHandler, FacebookGraphMixin):
139 def initialize(self, test):
140 self._OAUTH_AUTHORIZE_URL = test.get_url("/facebook/server/authorize")
141 self._OAUTH_ACCESS_TOKEN_URL = test.get_url("/facebook/server/access_token")
142 self._FACEBOOK_BASE_URL = test.get_url("/facebook/server")
143
144 @gen.coroutine
145 def get(self):
146 if self.get_argument("code", None):
147 user = yield self.get_authenticated_user(
148 redirect_uri=self.request.full_url(),
149 client_id=self.settings["facebook_api_key"],
150 client_secret=self.settings["facebook_secret"],
151 code=self.get_argument("code"),
152 )
153 self.write(user)
154 else:
155 self.authorize_redirect(
156 redirect_uri=self.request.full_url(),
157 client_id=self.settings["facebook_api_key"],
158 extra_params={"scope": "read_stream,offline_access"},
159 )
160
161
162class FacebookServerAccessTokenHandler(RequestHandler):
163 def get(self):
164 self.write(dict(access_token="asdf", expires_in=3600))
165
166
167class FacebookServerMeHandler(RequestHandler):
168 def get(self):
169 self.write("{}")
170
171
172class TwitterClientHandler(RequestHandler, TwitterMixin):
173 def initialize(self, test):
174 self._OAUTH_REQUEST_TOKEN_URL = test.get_url("/oauth1/server/request_token")
175 self._OAUTH_ACCESS_TOKEN_URL = test.get_url("/twitter/server/access_token")
176 self._OAUTH_AUTHORIZE_URL = test.get_url("/oauth1/server/authorize")
177 self._OAUTH_AUTHENTICATE_URL = test.get_url("/twitter/server/authenticate")
178 self._TWITTER_BASE_URL = test.get_url("/twitter/api")
179
180 def get_auth_http_client(self):
181 return self.settings["http_client"]
182
183
184class TwitterClientLoginHandler(TwitterClientHandler):
185 @gen.coroutine
186 def get(self):
187 if self.get_argument("oauth_token", None):
188 user = yield self.get_authenticated_user()
189 if user is None:
190 raise Exception("user is None")
191 self.finish(user)
192 return
193 yield self.authorize_redirect()
194
195
196class TwitterClientAuthenticateHandler(TwitterClientHandler):
197 # Like TwitterClientLoginHandler, but uses authenticate_redirect
198 # instead of authorize_redirect.
199 @gen.coroutine
200 def get(self):
201 if self.get_argument("oauth_token", None):
202 user = yield self.get_authenticated_user()
203 if user is None:
204 raise Exception("user is None")
205 self.finish(user)
206 return
207 yield self.authenticate_redirect()
208
209
210class TwitterClientLoginGenCoroutineHandler(TwitterClientHandler):
211 @gen.coroutine
212 def get(self):
213 if self.get_argument("oauth_token", None):
214 user = yield self.get_authenticated_user()
215 self.finish(user)
216 else:
217 # New style: with @gen.coroutine the result must be yielded
218 # or else the request will be auto-finished too soon.
219 yield self.authorize_redirect()
220
221
222class TwitterClientShowUserHandler(TwitterClientHandler):
223 @gen.coroutine
224 def get(self):
225 # TODO: would be nice to go through the login flow instead of
226 # cheating with a hard-coded access token.
227 try:
228 response = yield self.twitter_request(
229 "/users/show/%s" % self.get_argument("name"),
230 access_token=dict(key="hjkl", secret="vbnm"),
231 )
232 except HTTPClientError:
233 # TODO(bdarnell): Should we catch HTTP errors and
234 # transform some of them (like 403s) into AuthError?
235 self.set_status(500)
236 self.finish("error from twitter request")
237 else:
238 self.finish(response)
239
240
241class TwitterServerAccessTokenHandler(RequestHandler):
242 def get(self):
243 self.write("oauth_token=hjkl&oauth_token_secret=vbnm&screen_name=foo")
244
245
246class TwitterServerShowUserHandler(RequestHandler):
247 def get(self, screen_name):
248 if screen_name == "error":
249 raise HTTPError(500)
250 assert "oauth_nonce" in self.request.arguments
251 assert "oauth_timestamp" in self.request.arguments
252 assert "oauth_signature" in self.request.arguments
253 assert self.get_argument("oauth_consumer_key") == "test_twitter_consumer_key"
254 assert self.get_argument("oauth_signature_method") == "HMAC-SHA1"
255 assert self.get_argument("oauth_version") == "1.0"
256 assert self.get_argument("oauth_token") == "hjkl"
257 self.write(dict(screen_name=screen_name, name=screen_name.capitalize()))
258
259
260class TwitterServerVerifyCredentialsHandler(RequestHandler):
261 def get(self):
262 assert "oauth_nonce" in self.request.arguments
263 assert "oauth_timestamp" in self.request.arguments
264 assert "oauth_signature" in self.request.arguments
265 assert self.get_argument("oauth_consumer_key") == "test_twitter_consumer_key"
266 assert self.get_argument("oauth_signature_method") == "HMAC-SHA1"
267 assert self.get_argument("oauth_version") == "1.0"
268 assert self.get_argument("oauth_token") == "hjkl"
269 self.write(dict(screen_name="foo", name="Foo"))
270
271
272class AuthTest(AsyncHTTPTestCase):
273 def get_app(self):
274 return Application(
275 [
276 # test endpoints
277 ("/openid/client/login", OpenIdClientLoginHandler, dict(test=self)),
278 (
279 "/oauth10/client/login",
280 OAuth1ClientLoginHandler,
281 dict(test=self, version="1.0"),
282 ),
283 (
284 "/oauth10/client/request_params",
285 OAuth1ClientRequestParametersHandler,
286 dict(version="1.0"),
287 ),
288 (
289 "/oauth10a/client/login",
290 OAuth1ClientLoginHandler,
291 dict(test=self, version="1.0a"),
292 ),
293 (
294 "/oauth10a/client/login_coroutine",
295 OAuth1ClientLoginCoroutineHandler,
296 dict(test=self, version="1.0a"),
297 ),
298 (
299 "/oauth10a/client/request_params",
300 OAuth1ClientRequestParametersHandler,
301 dict(version="1.0a"),
302 ),
303 ("/oauth2/client/login", OAuth2ClientLoginHandler, dict(test=self)),
304 ("/facebook/client/login", FacebookClientLoginHandler, dict(test=self)),
305 ("/twitter/client/login", TwitterClientLoginHandler, dict(test=self)),
306 (
307 "/twitter/client/authenticate",
308 TwitterClientAuthenticateHandler,
309 dict(test=self),
310 ),
311 (
312 "/twitter/client/login_gen_coroutine",
313 TwitterClientLoginGenCoroutineHandler,
314 dict(test=self),
315 ),
316 (
317 "/twitter/client/show_user",
318 TwitterClientShowUserHandler,
319 dict(test=self),
320 ),
321 # simulated servers
322 ("/openid/server/authenticate", OpenIdServerAuthenticateHandler),
323 ("/oauth1/server/request_token", OAuth1ServerRequestTokenHandler),
324 ("/oauth1/server/access_token", OAuth1ServerAccessTokenHandler),
325 ("/facebook/server/access_token", FacebookServerAccessTokenHandler),
326 ("/facebook/server/me", FacebookServerMeHandler),
327 ("/twitter/server/access_token", TwitterServerAccessTokenHandler),
328 (r"/twitter/api/users/show/(.*)\.json", TwitterServerShowUserHandler),
329 (
330 r"/twitter/api/account/verify_credentials\.json",
331 TwitterServerVerifyCredentialsHandler,
332 ),
333 ],
334 http_client=self.http_client,
335 twitter_consumer_key="test_twitter_consumer_key",
336 twitter_consumer_secret="test_twitter_consumer_secret",
337 facebook_api_key="test_facebook_api_key",
338 facebook_secret="test_facebook_secret",
339 )
340
341 def test_openid_redirect(self):
342 response = self.fetch("/openid/client/login", follow_redirects=False)
343 self.assertEqual(response.code, 302)
344 self.assertIn("/openid/server/authenticate?", response.headers["Location"])
345
346 def test_openid_get_user(self):
347 response = self.fetch(
348 "/openid/client/login?openid.mode=blah"
349 "&openid.ns.ax=http://openid.net/srv/ax/1.0"
350 "&openid.ax.type.email=http://axschema.org/contact/email"
351 "&openid.ax.value.email=foo@example.com"
352 )
353 response.rethrow()
354 parsed = json_decode(response.body)
355 self.assertEqual(parsed["email"], "foo@example.com")
356
357 def test_oauth10_redirect(self):
358 response = self.fetch("/oauth10/client/login", follow_redirects=False)
359 self.assertEqual(response.code, 302)
360 self.assertTrue(
361 response.headers["Location"].endswith(
362 "/oauth1/server/authorize?oauth_token=zxcv"
363 )
364 )
365 # the cookie is base64('zxcv')|base64('1234')
366 self.assertIn(
367 '_oauth_request_token="enhjdg==|MTIzNA=="',
368 response.headers["Set-Cookie"],
369 response.headers["Set-Cookie"],
370 )
371
372 def test_oauth10_get_user(self):
373 response = self.fetch(
374 "/oauth10/client/login?oauth_token=zxcv",
375 headers={"Cookie": "_oauth_request_token=enhjdg==|MTIzNA=="},
376 )
377 response.rethrow()
378 parsed = json_decode(response.body)
379 self.assertEqual(parsed["email"], "foo@example.com")
380 self.assertEqual(parsed["access_token"], dict(key="uiop", secret="5678"))
381
382 def test_oauth10_request_parameters(self):
383 response = self.fetch("/oauth10/client/request_params")
384 response.rethrow()
385 parsed = json_decode(response.body)
386 self.assertEqual(parsed["oauth_consumer_key"], "asdf")
387 self.assertEqual(parsed["oauth_token"], "uiop")
388 self.assertIn("oauth_nonce", parsed)
389 self.assertIn("oauth_signature", parsed)
390
391 def test_oauth10a_redirect(self):
392 response = self.fetch("/oauth10a/client/login", follow_redirects=False)
393 self.assertEqual(response.code, 302)
394 self.assertTrue(
395 response.headers["Location"].endswith(
396 "/oauth1/server/authorize?oauth_token=zxcv"
397 )
398 )
399 # the cookie is base64('zxcv')|base64('1234')
400 self.assertTrue(
401 '_oauth_request_token="enhjdg==|MTIzNA=="'
402 in response.headers["Set-Cookie"],
403 response.headers["Set-Cookie"],
404 )
405
406 @unittest.skipIf(mock is None, "mock package not present")
407 def test_oauth10a_redirect_error(self):
408 with mock.patch.object(OAuth1ServerRequestTokenHandler, "get") as get:
409 get.side_effect = Exception("boom")
410 with ExpectLog(app_log, "Uncaught exception"):
411 response = self.fetch("/oauth10a/client/login", follow_redirects=False)
412 self.assertEqual(response.code, 500)
413
414 def test_oauth10a_get_user(self):
415 response = self.fetch(
416 "/oauth10a/client/login?oauth_token=zxcv",
417 headers={"Cookie": "_oauth_request_token=enhjdg==|MTIzNA=="},
418 )
419 response.rethrow()
420 parsed = json_decode(response.body)
421 self.assertEqual(parsed["email"], "foo@example.com")
422 self.assertEqual(parsed["access_token"], dict(key="uiop", secret="5678"))
423
424 def test_oauth10a_request_parameters(self):
425 response = self.fetch("/oauth10a/client/request_params")
426 response.rethrow()
427 parsed = json_decode(response.body)
428 self.assertEqual(parsed["oauth_consumer_key"], "asdf")
429 self.assertEqual(parsed["oauth_token"], "uiop")
430 self.assertIn("oauth_nonce", parsed)
431 self.assertIn("oauth_signature", parsed)
432
433 def test_oauth10a_get_user_coroutine_exception(self):
434 response = self.fetch(
435 "/oauth10a/client/login_coroutine?oauth_token=zxcv&fail_in_get_user=true",
436 headers={"Cookie": "_oauth_request_token=enhjdg==|MTIzNA=="},
437 )
438 self.assertEqual(response.code, 503)
439
440 def test_oauth2_redirect(self):
441 response = self.fetch("/oauth2/client/login", follow_redirects=False)
442 self.assertEqual(response.code, 302)
443 self.assertIn("/oauth2/server/authorize?", response.headers["Location"])
444
445 def test_facebook_login(self):
446 response = self.fetch("/facebook/client/login", follow_redirects=False)
447 self.assertEqual(response.code, 302)
448 self.assertTrue("/facebook/server/authorize?" in response.headers["Location"])
449 response = self.fetch(
450 "/facebook/client/login?code=1234", follow_redirects=False
451 )
452 self.assertEqual(response.code, 200)
453 user = json_decode(response.body)
454 self.assertEqual(user["access_token"], "asdf")
455 self.assertEqual(user["session_expires"], "3600")
456
457 def base_twitter_redirect(self, url):
458 # Same as test_oauth10a_redirect
459 response = self.fetch(url, follow_redirects=False)
460 self.assertEqual(response.code, 302)
461 self.assertTrue(
462 response.headers["Location"].endswith(
463 "/oauth1/server/authorize?oauth_token=zxcv"
464 )
465 )
466 # the cookie is base64('zxcv')|base64('1234')
467 self.assertIn(
468 '_oauth_request_token="enhjdg==|MTIzNA=="',
469 response.headers["Set-Cookie"],
470 response.headers["Set-Cookie"],
471 )
472
473 def test_twitter_redirect(self):
474 self.base_twitter_redirect("/twitter/client/login")
475
476 def test_twitter_redirect_gen_coroutine(self):
477 self.base_twitter_redirect("/twitter/client/login_gen_coroutine")
478
479 def test_twitter_authenticate_redirect(self):
480 response = self.fetch("/twitter/client/authenticate", follow_redirects=False)
481 self.assertEqual(response.code, 302)
482 self.assertTrue(
483 response.headers["Location"].endswith(
484 "/twitter/server/authenticate?oauth_token=zxcv"
485 ),
486 response.headers["Location"],
487 )
488 # the cookie is base64('zxcv')|base64('1234')
489 self.assertIn(
490 '_oauth_request_token="enhjdg==|MTIzNA=="',
491 response.headers["Set-Cookie"],
492 response.headers["Set-Cookie"],
493 )
494
495 def test_twitter_get_user(self):
496 response = self.fetch(
497 "/twitter/client/login?oauth_token=zxcv",
498 headers={"Cookie": "_oauth_request_token=enhjdg==|MTIzNA=="},
499 )
500 response.rethrow()
501 parsed = json_decode(response.body)
502 self.assertEqual(
503 parsed,
504 {
505 "access_token": {
506 "key": "hjkl",
507 "screen_name": "foo",
508 "secret": "vbnm",
509 },
510 "name": "Foo",
511 "screen_name": "foo",
512 "username": "foo",
513 },
514 )
515
516 def test_twitter_show_user(self):
517 response = self.fetch("/twitter/client/show_user?name=somebody")
518 response.rethrow()
519 self.assertEqual(
520 json_decode(response.body), {"name": "Somebody", "screen_name": "somebody"}
521 )
522
523 def test_twitter_show_user_error(self):
524 response = self.fetch("/twitter/client/show_user?name=error")
525 self.assertEqual(response.code, 500)
526 self.assertEqual(response.body, b"error from twitter request")
527
528
529class GoogleLoginHandler(RequestHandler, GoogleOAuth2Mixin):
530 def initialize(self, test):
531 self.test = test
532 self._OAUTH_REDIRECT_URI = test.get_url("/client/login")
533 self._OAUTH_AUTHORIZE_URL = test.get_url("/google/oauth2/authorize")
534 self._OAUTH_ACCESS_TOKEN_URL = test.get_url("/google/oauth2/token")
535
536 @gen.coroutine
537 def get(self):
538 code = self.get_argument("code", None)
539 if code is not None:
540 # retrieve authenticate google user
541 access = yield self.get_authenticated_user(self._OAUTH_REDIRECT_URI, code)
542 user = yield self.oauth2_request(
543 self.test.get_url("/google/oauth2/userinfo"),
544 access_token=access["access_token"],
545 )
546 # return the user and access token as json
547 user["access_token"] = access["access_token"]
548 self.write(user)
549 else:
550 self.authorize_redirect(
551 redirect_uri=self._OAUTH_REDIRECT_URI,
552 client_id=self.settings["google_oauth"]["key"],
553 scope=["profile", "email"],
554 response_type="code",
555 extra_params={"prompt": "select_account"},
556 )
557
558
559class GoogleOAuth2AuthorizeHandler(RequestHandler):
560 def get(self):
561 # issue a fake auth code and redirect to redirect_uri
562 code = "fake-authorization-code"
563 self.redirect(url_concat(self.get_argument("redirect_uri"), dict(code=code)))
564
565
566class GoogleOAuth2TokenHandler(RequestHandler):
567 def post(self):
568 assert self.get_argument("code") == "fake-authorization-code"
569 # issue a fake token
570 self.finish(
571 {"access_token": "fake-access-token", "expires_in": "never-expires"}
572 )
573
574
575class GoogleOAuth2UserinfoHandler(RequestHandler):
576 def get(self):
577 assert self.get_argument("access_token") == "fake-access-token"
578 # return a fake user
579 self.finish({"name": "Foo", "email": "foo@example.com"})
580
581
582class GoogleOAuth2Test(AsyncHTTPTestCase):
583 def get_app(self):
584 return Application(
585 [
586 # test endpoints
587 ("/client/login", GoogleLoginHandler, dict(test=self)),
588 # simulated google authorization server endpoints
589 ("/google/oauth2/authorize", GoogleOAuth2AuthorizeHandler),
590 ("/google/oauth2/token", GoogleOAuth2TokenHandler),
591 ("/google/oauth2/userinfo", GoogleOAuth2UserinfoHandler),
592 ],
593 google_oauth={
594 "key": "fake_google_client_id",
595 "secret": "fake_google_client_secret",
596 },
597 )
598
599 def test_google_login(self):
600 response = self.fetch("/client/login")
601 self.assertDictEqual(
602 {
603 "name": "Foo",
604 "email": "foo@example.com",
605 "access_token": "fake-access-token",
606 },
607 json_decode(response.body),
608 )
609