diff --git a/server/oauth/models.py b/server/oauth/models.py index fbcfeed6..c71728ad 100644 --- a/server/oauth/models.py +++ b/server/oauth/models.py @@ -55,14 +55,14 @@ class OAuth2Token(models.Model): return self.expires_at > now_unix_timestamp def update_with_refresh_data(self, data): - self.token_type = data['token_type'] - self.access_token = data['access_token'] - self.refresh_token = data['refresh_token'] - payload = self._jwt_payload(data['access_token']) if not payload: return None, False + self.token_type = data['token_type'] + self.access_token = data['access_token'] + self.refresh_token = data['refresh_token'] + self.expires_at = int(payload['exp']) self.save() @@ -70,8 +70,8 @@ class OAuth2Token(models.Model): def _jwt_payload(self, jwt): jwt_parts = jwt.split('.') - payload_bytes = base64.b64decode(jwt_parts[1]) try: + payload_bytes = base64.b64decode(jwt_parts[1]) payload = json.loads(payload_bytes.decode("UTF-8")) except: return None diff --git a/server/oauth/tests/test_oauth2token.py b/server/oauth/tests/test_oauth2token.py new file mode 100644 index 00000000..b791c90f --- /dev/null +++ b/server/oauth/tests/test_oauth2token.py @@ -0,0 +1,51 @@ +from datetime import timedelta + +from django.test import TestCase +from django.utils import timezone + +from core.factories import UserFactory +from oauth.factories import Oauth2TokenFactory + + +REFRESH_DATA = { + "token_type": "Bearer", + "expires_in": 31536000, + "access_token": "eyJ0eXAiOiJKV1QiLCJhbGciOiJSUzI1NiJ9.eyJhdWQiOiI5MzU3NDdmZC1mODM0LTQ0Y2MtODY0NC01YjA0ZGY2N2M4ZTMiLCJqdGkiOiI5ZmI5ZWFmOWM5Y2M4YWM4ZThlMDkyMDlkNWQwZjQxNTRiZTYwNmQ4ZDdkOTU0OTVkNTM0OWU3NWJmNDJiZGQ3MWRhYWU5MTJmMDIwNGMxNiIsImlhdCI6MTYyMzkyNjI1MC4xMzkxNDYsIm5iZiI6MTYyMzkyNjI1MC4xMzkxNTUsImV4cCI6MTY1NTQ2MjI1MC4xMDM2NTksInN1YiI6IjI5Iiwic2NvcGVzIjpbIm9yZGVycyJdLCJncm91cHMiOlsidGVzdHVzZXIiLCJ0ZWFjaGVyIl19.mdj0xP-GpPbwFt6VpnGq1RJND9SbfutcQkVv0I3G8HNEVylf17FuK22CMJRZLN2BW6hV67Kpps7RoCBPh9XWYUkkpLA1lD3RBZEit2IdBOhXf6G8B8go_UV2B8BUgHNn0AyLVWsawtdPkcIXbPkXv0oQAi-tsqFan5OE0XPQCUJfun2Cvhe4Teyl98-5zd_njt6mK_0BNtnDDAWjMTgVh9y_-WTu34S_2xttlh-vCFYMl8JwZPuNpTrCyD_UqfY8sp_dKPyg87BLRk4uR1iFoL399BvMSIUXoLFdh7Hb6eMuSBQH63JM77zuWk2bACBofULE2ajsbQg9a8dL43inNRwtRDlhofaw1NHYF_TrzRBP2pRgbo8FsEONVx9FRocMdfo4-icR1_Pb59Rr9lmiEu5JAi47o0rRCz9lAuUiHdliZtEPyAUQXJ5-y0zOITko83VstsU88OodgGvwZ53yp_aibdDBuX99YOSRvlBFXH0Sst49PEvGWnnNRP_4KOtAOzJ7n9yE0cDWo-VgB97KOVEv_BhiiE0SMbeYe7ByT8u9lNwKGX3AYWQTsbO5IlKn9f86NKBeLAB5bWaXXNnsQreNrTlhky8LoUQBtrSdwNWR7ZUheQOlSBKvqhT_48lJU_CMNxx38rmaoG6qC_WNKcFq_Lb01hLZ_VvYOPaIlWw", + "refresh_token": "def502007c4d524d828b97e4dc75198a482b3cf7db88dcef78b9c2bfe008ec8db48f02d564366b4f6d41115b090411bdc6c185fe06195aad2ef4e01cd6a7e4d87144eccf628c3864d669ec8fcebbaf2eb62fb42b367650e4dea8dbcc9465f5cf1ccb3e44ff3066f2ced4aac455677fd95f3abf36536f1828eb24c0c173858ca593a9d855ede03c7ec5a900930ca10b0a6270358ba114e3f695b5adfc7c51d0a67b4b29a617015213ae4193f28c9c2d8f8504a8573240085c948c6dc9654bc5f0d7447e22bf25278135005ba51f953a056cf8f64f55fe15447ba395160ce1e03f6861684763d0de11bc976d71f545548451539d2025b74693010661d1d3a7886644fc952f3adeb6f05350489f1cebdf947bca96112ccdb8ffd2b5b5d0a843ff2d3835d9cbee1f7442c8c5ba41d028f2e3b83523e510084c3c54f672820c92781aa63ce8a7cb9a938ba30562b8e69e23ab500a6f90eaa98a3e2ab40af88d20fc8a3c23185eab949d81ea9d737798aa022a03b8c1e5f987c77345cf5582debb726171484b1502871a8f95ce90e5d6" +} + + +class OAuth2TokenTestCases(TestCase): + def setUp(self): + user = UserFactory(username='housi') + self.token = Oauth2TokenFactory(user=user) + self.now = timezone.now() + + def test_is_valid(self): + self.assertTrue(self.token.is_valid()) + + def test_has_expired(self): + one_hourish_delta_in_ms = 60*60 + self.token.expires_at -= one_hourish_delta_in_ms + self.token.save() + self.assertFalse(self.token.is_valid()) + + def test_can_update_refresh_data(self): + token, success = self.token.update_with_refresh_data(REFRESH_DATA) + + self.assertTrue(success) + self.assertEqual(REFRESH_DATA['access_token'], token.access_token) + + def test_success_on_update_refresh_data(self): + token, success = self.token.update_with_refresh_data(REFRESH_DATA) + + self.assertTrue(success) + self.assertEqual(REFRESH_DATA['access_token'], token.access_token) + + def test_fail_on_update_refresh_data(self): + data = REFRESH_DATA.copy() + data['access_token'] = '12344' + token, success = self.token.update_with_refresh_data(data) + + self.assertFalse(success) + self.assertIsNone(token)