test_oauth1_session.py 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331
  1. import unittest
  2. import requests
  3. from io import StringIO
  4. from unittest import mock
  5. from oauthlib.oauth1 import SIGNATURE_TYPE_QUERY, SIGNATURE_TYPE_BODY
  6. from oauthlib.oauth1 import SIGNATURE_RSA, SIGNATURE_PLAINTEXT
  7. from requests_oauthlib import OAuth1Session
  8. try:
  9. import cryptography
  10. except ImportError:
  11. cryptography = None
  12. try:
  13. import jwt
  14. except ImportError:
  15. jwt = None
  16. TEST_RSA_KEY = (
  17. "-----BEGIN RSA PRIVATE KEY-----\n"
  18. "MIIEogIBAAKCAQEApF1JaMSN8TEsh4N4O/5SpEAVLivJyLH+Cgl3OQBPGgJkt8cg\n"
  19. "49oasl+5iJS+VdrILxWM9/JCJyURpUuslX4Eb4eUBtQ0x5BaPa8+S2NLdGTaL7nB\n"
  20. "OO8o8n0C5FEUU+qlEip79KE8aqOj+OC44VsIquSmOvWIQD26n3fCVlgwoRBD1gzz\n"
  21. "sDOeaSyzpKrZR851Kh6rEmF2qjJ8jt6EkxMsRNACmBomzgA4M1TTsisSUO87444p\n"
  22. "e35Z4/n5c735o2fZMrGgMwiJNh7rT8SYxtIkxngioiGnwkxGQxQ4NzPAHg+XSY0J\n"
  23. "04pNm7KqTkgtxyrqOANJLIjXlR+U9SQ90NjHVQIDAQABAoIBABuBPOKaWcJt3yzC\n"
  24. "NGGduoif7KtwSnEaUA+v69KPGa2Zju8uFHPssKD+4dZYRc2qMeunKJLpaGaSjnRh\n"
  25. "yHyvvOBJCN1nr3lhz6gY5kzJTfwpUFXCOPJlGy4Q+2Xnp4YvcvYqQ9n5DVovDiZ8\n"
  26. "vJOBn16xqpudMPLHIa7D5LJ8SY76HBjE+imTXw1EShdh5TOV9bmPFQqH6JFzowRH\n"
  27. "hyH2DPHuyHJj6cl8FyqJw5lVWzG3n6Prvk7bYHsjmGjurN35UsumNAp6VouNyUP1\n"
  28. "RAEcUJega49aIs6/FJ0ENJzQjlsAzVbTleHkpez2aIok+wsWJGJ4SVxAjADOWAaZ\n"
  29. "uEJPc3UCgYEA1g4ZGrXOuo75p9/MRIepXGpBWxip4V7B9XmO9WzPCv8nMorJntWB\n"
  30. "msYV1I01aITxadHatO4Gl2xLniNkDyrEQzJ7w38RQgsVK+CqbnC0K9N77QPbHeC1\n"
  31. "YQd9RCNyUohOimKvb7jyv798FBU1GO5QI2eNgfnnfteSVXhD2iOoTOsCgYEAxJJ+\n"
  32. "8toxJdnLa0uUsAbql6zeNXGbUBMzu3FomKlyuWuq841jS2kIalaO/TRj5hbnE45j\n"
  33. "mCjeLgTVO6Ach3Wfk4zrqajqfFJ0zUg/Wexp49lC3RWiV4icBb85Q6bzeJD9Dn9v\n"
  34. "hjpfWVkczf/NeA1fGH/pcgfkT6Dm706GFFttLL8CgYBl/HeXk1H47xAiHO4dJKnb\n"
  35. "v0B+X8To/RXamF01r+8BpUoOubOQetdyX7ic+d6deuHu8i6LD/GSCeYJZYFR/KVg\n"
  36. "AtiW757QYalnq3ZogkhFrVCZP8IRfTPOFBxp752TlyAcrSI7T9pQ47IBe4094KXM\n"
  37. "CJWSfPgAJkOxd0iU0XJpmwKBgGfQxuMTgSlwYRKFlD1zKap5TdID8fbUbVnth0Q5\n"
  38. "GbH7vwlp/qrxCdS/aj0n0irOpbOaW9ccnlrHiqY25VpVMLYIkt3DrDOEiNNx+KNR\n"
  39. "TItdTwbcSiTYrS4L0/56ydM/H6bsfsXxRjI18hSJqMZiqXqS84OZz2aOn+h7HCzc\n"
  40. "LEiZAoGASk20wFvilpRKHq79xxFWiDUPHi0x0pp82dYIEntGQkKUWkbSlhgf3MAi\n"
  41. "5NEQTDmXdnB+rVeWIvEi+BXfdnNgdn8eC4zSdtF4sIAhYr5VWZo0WVWDhT7u2ccv\n"
  42. "ZBFymiz8lo3gN57wGUCi9pbZqzV1+ZppX6YTNDdDCE0q+KO3Cec=\n"
  43. "-----END RSA PRIVATE KEY-----"
  44. )
  45. TEST_RSA_OAUTH_SIGNATURE = (
  46. "j8WF8PGjojT82aUDd2EL%2Bz7HCoHInFzWUpiEKMCy%2BJ2cYHWcBS7mXlmFDLgAKV0"
  47. "P%2FyX4TrpXODYnJ6dRWdfghqwDpi%2FlQmB2jxCiGMdJoYxh3c5zDf26gEbGdP6D7O"
  48. "Ssp5HUnzH6sNkmVjuE%2FxoJcHJdc23H6GhOs7VJ2LWNdbhKWP%2FMMlTrcoQDn8lz"
  49. "%2Fb24WsJ6ae1txkUzpFOOlLM8aTdNtGL4OtsubOlRhNqnAFq93FyhXg0KjzUyIZzmMX"
  50. "9Vx90jTks5QeBGYcLE0Op2iHb2u%2FO%2BEgdwFchgEwE5LgMUyHUI4F3Wglp28yHOAM"
  51. "jPkI%2FkWMvpxtMrU3Z3KN31WQ%3D%3D"
  52. )
  53. class OAuth1SessionTest(unittest.TestCase):
  54. def test_signature_types(self):
  55. def verify_signature(getter):
  56. def fake_send(r, **kwargs):
  57. signature = getter(r)
  58. if isinstance(signature, bytes):
  59. signature = signature.decode("utf-8")
  60. self.assertIn("oauth_signature", signature)
  61. resp = mock.MagicMock(spec=requests.Response)
  62. resp.cookies = []
  63. return resp
  64. return fake_send
  65. header = OAuth1Session("foo")
  66. header.send = verify_signature(lambda r: r.headers["Authorization"])
  67. header.post("https://i.b")
  68. query = OAuth1Session("foo", signature_type=SIGNATURE_TYPE_QUERY)
  69. query.send = verify_signature(lambda r: r.url)
  70. query.post("https://i.b")
  71. body = OAuth1Session("foo", signature_type=SIGNATURE_TYPE_BODY)
  72. headers = {"Content-Type": "application/x-www-form-urlencoded"}
  73. body.send = verify_signature(lambda r: r.body)
  74. body.post("https://i.b", headers=headers, data="")
  75. @mock.patch("oauthlib.oauth1.rfc5849.generate_timestamp")
  76. @mock.patch("oauthlib.oauth1.rfc5849.generate_nonce")
  77. def test_signature_methods(self, generate_nonce, generate_timestamp):
  78. if not cryptography:
  79. raise unittest.SkipTest("cryptography module is required")
  80. if not jwt:
  81. raise unittest.SkipTest("pyjwt module is required")
  82. generate_nonce.return_value = "abc"
  83. generate_timestamp.return_value = "123"
  84. signature = 'OAuth oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", oauth_signature_method="HMAC-SHA1", oauth_consumer_key="foo", oauth_signature="h2sRqLArjhlc5p3FTkuNogVHlKE%3D"'
  85. auth = OAuth1Session("foo")
  86. auth.send = self.verify_signature(signature)
  87. auth.post("https://i.b")
  88. signature = 'OAuth oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", oauth_signature_method="PLAINTEXT", oauth_consumer_key="foo", oauth_signature="%26"'
  89. auth = OAuth1Session("foo", signature_method=SIGNATURE_PLAINTEXT)
  90. auth.send = self.verify_signature(signature)
  91. auth.post("https://i.b")
  92. signature = (
  93. "OAuth "
  94. 'oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", '
  95. 'oauth_signature_method="RSA-SHA1", oauth_consumer_key="foo", '
  96. 'oauth_signature="{sig}"'
  97. ).format(sig=TEST_RSA_OAUTH_SIGNATURE)
  98. auth = OAuth1Session(
  99. "foo", signature_method=SIGNATURE_RSA, rsa_key=TEST_RSA_KEY
  100. )
  101. auth.send = self.verify_signature(signature)
  102. auth.post("https://i.b")
  103. @mock.patch("oauthlib.oauth1.rfc5849.generate_timestamp")
  104. @mock.patch("oauthlib.oauth1.rfc5849.generate_nonce")
  105. def test_binary_upload(self, generate_nonce, generate_timestamp):
  106. generate_nonce.return_value = "abc"
  107. generate_timestamp.return_value = "123"
  108. fake_xml = StringIO("hello world")
  109. headers = {"Content-Type": "application/xml"}
  110. signature = 'OAuth oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", oauth_signature_method="HMAC-SHA1", oauth_consumer_key="foo", oauth_signature="h2sRqLArjhlc5p3FTkuNogVHlKE%3D"'
  111. auth = OAuth1Session("foo")
  112. auth.send = self.verify_signature(signature)
  113. auth.post("https://i.b", headers=headers, files=[("fake", fake_xml)])
  114. @mock.patch("oauthlib.oauth1.rfc5849.generate_timestamp")
  115. @mock.patch("oauthlib.oauth1.rfc5849.generate_nonce")
  116. def test_nonascii(self, generate_nonce, generate_timestamp):
  117. generate_nonce.return_value = "abc"
  118. generate_timestamp.return_value = "123"
  119. signature = 'OAuth oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", oauth_signature_method="HMAC-SHA1", oauth_consumer_key="foo", oauth_signature="W0haoue5IZAZoaJiYCtfqwMf8x8%3D"'
  120. auth = OAuth1Session("foo")
  121. auth.send = self.verify_signature(signature)
  122. auth.post("https://i.b?cjk=%E5%95%A6%E5%95%A6")
  123. def test_authorization_url(self):
  124. auth = OAuth1Session("foo")
  125. url = "https://example.comm/authorize"
  126. token = "asluif023sf"
  127. auth_url = auth.authorization_url(url, request_token=token)
  128. self.assertEqual(auth_url, url + "?oauth_token=" + token)
  129. def test_parse_response_url(self):
  130. url = "https://i.b/callback?oauth_token=foo&oauth_verifier=bar"
  131. auth = OAuth1Session("foo")
  132. resp = auth.parse_authorization_response(url)
  133. self.assertEqual(resp["oauth_token"], "foo")
  134. self.assertEqual(resp["oauth_verifier"], "bar")
  135. for k, v in resp.items():
  136. self.assertIsInstance(k, str)
  137. self.assertIsInstance(v, str)
  138. def test_fetch_request_token(self):
  139. auth = OAuth1Session("foo")
  140. auth.send = self.fake_body("oauth_token=foo")
  141. resp = auth.fetch_request_token("https://example.com/token")
  142. self.assertEqual(resp["oauth_token"], "foo")
  143. for k, v in resp.items():
  144. self.assertIsInstance(k, str)
  145. self.assertIsInstance(v, str)
  146. def test_fetch_request_token_with_optional_arguments(self):
  147. auth = OAuth1Session("foo")
  148. auth.send = self.fake_body("oauth_token=foo")
  149. resp = auth.fetch_request_token(
  150. "https://example.com/token", verify=False, stream=True
  151. )
  152. self.assertEqual(resp["oauth_token"], "foo")
  153. for k, v in resp.items():
  154. self.assertIsInstance(k, str)
  155. self.assertIsInstance(v, str)
  156. def test_fetch_access_token(self):
  157. auth = OAuth1Session("foo", verifier="bar")
  158. auth.send = self.fake_body("oauth_token=foo")
  159. resp = auth.fetch_access_token("https://example.com/token")
  160. self.assertEqual(resp["oauth_token"], "foo")
  161. for k, v in resp.items():
  162. self.assertIsInstance(k, str)
  163. self.assertIsInstance(v, str)
  164. def test_fetch_access_token_with_optional_arguments(self):
  165. auth = OAuth1Session("foo", verifier="bar")
  166. auth.send = self.fake_body("oauth_token=foo")
  167. resp = auth.fetch_access_token(
  168. "https://example.com/token", verify=False, stream=True
  169. )
  170. self.assertEqual(resp["oauth_token"], "foo")
  171. for k, v in resp.items():
  172. self.assertIsInstance(k, str)
  173. self.assertIsInstance(v, str)
  174. def _test_fetch_access_token_raises_error(self, auth):
  175. """Assert that an error is being raised whenever there's no verifier
  176. passed in to the client.
  177. """
  178. auth.send = self.fake_body("oauth_token=foo")
  179. with self.assertRaises(ValueError) as cm:
  180. auth.fetch_access_token("https://example.com/token")
  181. self.assertEqual("No client verifier has been set.", str(cm.exception))
  182. def test_fetch_token_invalid_response(self):
  183. auth = OAuth1Session("foo")
  184. auth.send = self.fake_body("not valid urlencoded response!")
  185. self.assertRaises(
  186. ValueError, auth.fetch_request_token, "https://example.com/token"
  187. )
  188. for code in (400, 401, 403):
  189. auth.send = self.fake_body("valid=response", code)
  190. with self.assertRaises(ValueError) as cm:
  191. auth.fetch_request_token("https://example.com/token")
  192. self.assertEqual(cm.exception.status_code, code)
  193. self.assertIsInstance(cm.exception.response, requests.Response)
  194. def test_fetch_access_token_missing_verifier(self):
  195. self._test_fetch_access_token_raises_error(OAuth1Session("foo"))
  196. def test_fetch_access_token_has_verifier_is_none(self):
  197. auth = OAuth1Session("foo")
  198. del auth._client.client.verifier
  199. self._test_fetch_access_token_raises_error(auth)
  200. def test_token_proxy_set(self):
  201. token = {
  202. "oauth_token": "fake-key",
  203. "oauth_token_secret": "fake-secret",
  204. "oauth_verifier": "fake-verifier",
  205. }
  206. sess = OAuth1Session("foo")
  207. self.assertIsNone(sess._client.client.resource_owner_key)
  208. self.assertIsNone(sess._client.client.resource_owner_secret)
  209. self.assertIsNone(sess._client.client.verifier)
  210. self.assertEqual(sess.token, {})
  211. sess.token = token
  212. self.assertEqual(sess._client.client.resource_owner_key, "fake-key")
  213. self.assertEqual(sess._client.client.resource_owner_secret, "fake-secret")
  214. self.assertEqual(sess._client.client.verifier, "fake-verifier")
  215. def test_token_proxy_get(self):
  216. token = {
  217. "oauth_token": "fake-key",
  218. "oauth_token_secret": "fake-secret",
  219. "oauth_verifier": "fake-verifier",
  220. }
  221. sess = OAuth1Session(
  222. "foo",
  223. resource_owner_key=token["oauth_token"],
  224. resource_owner_secret=token["oauth_token_secret"],
  225. verifier=token["oauth_verifier"],
  226. )
  227. self.assertEqual(sess.token, token)
  228. sess._client.client.resource_owner_key = "different-key"
  229. token["oauth_token"] = "different-key"
  230. self.assertEqual(sess.token, token)
  231. def test_authorized_false(self):
  232. sess = OAuth1Session("foo")
  233. self.assertIs(sess.authorized, False)
  234. def test_authorized_false_rsa(self):
  235. signature = (
  236. "OAuth "
  237. 'oauth_nonce="abc", oauth_timestamp="123", oauth_version="1.0", '
  238. 'oauth_signature_method="RSA-SHA1", oauth_consumer_key="foo", '
  239. 'oauth_signature="{sig}"'
  240. ).format(sig=TEST_RSA_OAUTH_SIGNATURE)
  241. sess = OAuth1Session(
  242. "foo", signature_method=SIGNATURE_RSA, rsa_key=TEST_RSA_KEY
  243. )
  244. sess.send = self.verify_signature(signature)
  245. self.assertIs(sess.authorized, False)
  246. def test_authorized_true(self):
  247. sess = OAuth1Session("key", "secret", verifier="bar")
  248. sess.send = self.fake_body("oauth_token=foo&oauth_token_secret=bar")
  249. sess.fetch_access_token("https://example.com/token")
  250. self.assertIs(sess.authorized, True)
  251. @mock.patch("oauthlib.oauth1.rfc5849.generate_timestamp")
  252. @mock.patch("oauthlib.oauth1.rfc5849.generate_nonce")
  253. def test_authorized_true_rsa(self, generate_nonce, generate_timestamp):
  254. if not cryptography:
  255. raise unittest.SkipTest("cryptography module is required")
  256. if not jwt:
  257. raise unittest.SkipTest("pyjwt module is required")
  258. generate_nonce.return_value = "abc"
  259. generate_timestamp.return_value = "123"
  260. sess = OAuth1Session(
  261. "key",
  262. "secret",
  263. signature_method=SIGNATURE_RSA,
  264. rsa_key=TEST_RSA_KEY,
  265. verifier="bar",
  266. )
  267. sess.send = self.fake_body("oauth_token=foo&oauth_token_secret=bar")
  268. sess.fetch_access_token("https://example.com/token")
  269. self.assertIs(sess.authorized, True)
  270. def verify_signature(self, signature):
  271. def fake_send(r, **kwargs):
  272. auth_header = r.headers["Authorization"]
  273. if isinstance(auth_header, bytes):
  274. auth_header = auth_header.decode("utf-8")
  275. self.assertEqual(auth_header, signature)
  276. resp = mock.MagicMock(spec=requests.Response)
  277. resp.cookies = []
  278. return resp
  279. return fake_send
  280. def fake_body(self, body, status_code=200):
  281. def fake_send(r, **kwargs):
  282. resp = mock.MagicMock(spec=requests.Response)
  283. resp.cookies = []
  284. resp.text = body
  285. resp.status_code = status_code
  286. return resp
  287. return fake_send