Add a fallback to load creds from legacy JSON file if normal load fails

gcalcli/auth.py

@@ -1,5 +1,6 @@
 from google.auth.transport.requests import Request
 from google_auth_oauthlib.flow import InstalledAppFlow
+from google.oauth2.credentials import Credentials
 
 
 def authenticate(client_id: str, client_secret: str):
@@ -10,8 +11,7 @@ def authenticate(client_id: str, client_secret: str):
                 "client_secret": client_secret,
                 "auth_uri": "https://accounts.google.com/o/oauth2/auth",
                 "token_uri": "https://oauth2.googleapis.com/token",
-                "auth_provider_x509_cert_url":
-                    "https://www.googleapis.com/oauth2/v1/certs",
+                "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
                 "redirect_uris": ["http://localhost"],
             }
         },
@@ -24,3 +24,19 @@ def authenticate(client_id: str, client_secret: str):
 def refresh_if_expired(credentials) -> None:
     if credentials.expired:
         credentials.refresh(Request())
+
+
+def creds_from_legacy_json(data):
+    kwargs = {
+        k: v
+        for k, v in data.items()
+        if k
+        in (
+            'client_id',
+            'client_secret',
+            'refresh_token',
+            'token_uri',
+            'scopes',
+        )
+    }
+    return Credentials(data['access_token'], **kwargs)

gcalcli/gcal.py

@@ -131,28 +131,48 @@ def _google_auth(self):
         else:
             oauth_filepath = os.path.expanduser('~/.gcalcli_oauth')
         if os.path.exists(oauth_filepath):
+            needs_write = False
             with open(oauth_filepath, 'rb') as gcalcli_oauth:
                 try:
                     self.credentials = pickle.load(gcalcli_oauth)
-                except pickle.UnpicklingError as e:
-                    self.printer.err_msg(
-                        f"Couldn't parse {oauth_filepath}.\n"
-                        "The file may be corrupt or be incompatible with this "
-                        "version of gcalcli. It probably has to be removed and "
-                        "provisioning done again.\n"
-                    )
-                    raise e
+                except (pickle.UnpicklingError, EOFError) as e:
+                    # Try reading as legacy json format as fallback.
+                    try:
+                        gcalcli_oauth.seek(0)
+                        self.credentials = auth.creds_from_legacy_json(
+                            json.load(gcalcli_oauth)
+                        )
+                        needs_write = True
+                    except (OSError, ValueError, EOFError):
+                        pass
+                    if not self.credentials:
+                        self.printer.err_msg(
+                            f"Couldn't parse {oauth_filepath}.\n"
+                            "The file may be corrupt or be incompatible with this "
+                            "version of gcalcli. It probably has to be removed and "
+                            "provisioning done again.\n"
+                        )
+                        raise e
+            if needs_write:
+                # Save back loaded creds to file (for legacy conversion case).
+                with open(oauth_filepath, 'wb') as gcalcli_oauth:
+                    pickle.dump(self.credentials, gcalcli_oauth)
 
         if not self.credentials:
             # No cached credentials, start auth flow
             self.printer.msg(
-                'Not yet authenticated. Starting auth flow...\n', 'yellow')
+                'Not yet authenticated. Starting auth flow...\n', 'yellow'
+            )
             self.printer.msg(
                 'NOTE: See '
                 'https://github.com/insanum/gcalcli/blob/HEAD/docs/api-auth.md '
-                'for help/troubleshooting.\n')
-            missing_info = [opt for opt in ['client_id', 'client_secret']
-                            if self.options.get(opt) is None]
+                'for help/troubleshooting.\n'
+            )
+            missing_info = [
+                opt
+                for opt in ['client_id', 'client_secret']
+                if self.options.get(opt) is None
+            ]
             if missing_info:
                 self.printer.msg(
                     f"You'll be asked for a {' and '.join(missing_info)} "
@@ -169,10 +189,13 @@ def _google_auth(self):
                 client_secret = input()
             self.printer.msg(
                 'Now click the link below and follow directions to '
-                'authenticate.\n', 'yellow')
+                'authenticate.\n',
+                'yellow',
+            )
             self.printer.msg(
                 'You will likely see a security warning page and need to '
-                'click "Advanced" and "Go to gcalcli (unsafe)" to proceed.\n')
+                'click "Advanced" and "Go to gcalcli (unsafe)" to proceed.\n'
+            )
             self.credentials = auth.authenticate(client_id, client_secret)
             with open(oauth_filepath, 'wb') as gcalcli_oauth:
                 pickle.dump(self.credentials, gcalcli_oauth)

tests/conftest.py

@@ -1,9 +1,11 @@
-from datetime import datetime
+from datetime import datetime, timedelta
 import os
 import sys
+from types import SimpleNamespace
 
 from dateutil.tz import tzlocal
 from googleapiclient.discovery import build, HttpMock
+import google.oauth2.reauth
 import pytest
 
 from gcalcli.argparsers import (get_cal_query_parser, get_color_parser,
@@ -60,74 +62,44 @@ def default_options():
 
 
 @pytest.fixture
-def PatchedGCalIForEvents(monkeypatch):
+def PatchedGCalIForEvents(PatchedGCalI, monkeypatch):
     def mocked_search_for_events(self, start, end, search_text):
         return mock_event
 
-    def mocked_calendar_service(self):
-        http = HttpMock(
-                TEST_DATA_DIR + '/cal_service_discovery.json',
-                {'status': '200'})
-        if not self.cal_service:
-            self.cal_service = build(
-                    serviceName='calendar', version='v3', http=http)
-        return self.cal_service
-
-    def mocked_calendar_list(self):
-        http = HttpMock(
-                TEST_DATA_DIR + '/cal_list.json', {'status': '200'})
-        request = self.get_cal_service().calendarList().list()
-        cal_list = request.execute(http=http)
-        self.all_cals = [cal for cal in cal_list['items']]
-        if not self.cal_service:
-            self.cal_service = build(
-                 serviceName='calendar', version='v3', http=http)
-        return self.cal_service
-
-    def mocked_msg(self, msg, colorname='default', file=sys.stdout):
-        # ignores file and always writes to stdout
-        if self.use_color:
-            msg = self.colors[colorname] + msg + self.colors['default']
-        sys.stdout.write(msg)
-
     monkeypatch.setattr(
-            GoogleCalendarInterface, '_search_for_events',
-            mocked_search_for_events
+        GoogleCalendarInterface, '_search_for_events', mocked_search_for_events
     )
-    monkeypatch.setattr(
-            GoogleCalendarInterface, 'get_cal_service', mocked_calendar_service
-    )
-    monkeypatch.setattr(
-            GoogleCalendarInterface, '_get_cached', mocked_calendar_list
-    )
-    monkeypatch.setattr(Printer, 'msg', mocked_msg)
 
-    def _init(**opts):
-        return GoogleCalendarInterface(use_cache=False, **opts)
+    return PatchedGCalI
 
-    return _init
+
+@pytest.fixture
+def PatchedGCalI(gcali_patches):
+    gcali_patches.stub_out_cal_service()
+    return gcali_patches.GCalI
 
 
 @pytest.fixture
-def PatchedGCalI(monkeypatch):
-    def mocked_calendar_service(self):
+def gcali_patches(monkeypatch):
+    def mocked_cal_service(self):
         http = HttpMock(
-                TEST_DATA_DIR + '/cal_service_discovery.json',
-                {'status': '200'})
+            TEST_DATA_DIR + '/cal_service_discovery.json', {'status': '200'}
+        )
         if not self.cal_service:
             self.cal_service = build(
-                    serviceName='calendar', version='v3', http=http)
+                serviceName='calendar', version='v3', http=http
+            )
         return self.cal_service
 
     def mocked_calendar_list(self):
-        http = HttpMock(
-                TEST_DATA_DIR + '/cal_list.json', {'status': '200'})
+        http = HttpMock(TEST_DATA_DIR + '/cal_list.json', {'status': '200'})
         request = self.get_cal_service().calendarList().list()
         cal_list = request.execute(http=http)
         self.all_cals = [cal for cal in cal_list['items']]
         if not self.cal_service:
             self.cal_service = build(
-                 serviceName='calendar', version='v3', http=http)
+                serviceName='calendar', version='v3', http=http
+            )
         return self.cal_service
 
     def mocked_msg(self, msg, colorname='default', file=sys.stdout):
@@ -137,14 +109,35 @@ def mocked_msg(self, msg, colorname='default', file=sys.stdout):
         sys.stdout.write(msg)
 
     monkeypatch.setattr(
-            GoogleCalendarInterface, 'get_cal_service', mocked_calendar_service
-    )
-    monkeypatch.setattr(
-            GoogleCalendarInterface, '_get_cached', mocked_calendar_list
+        GoogleCalendarInterface, '_get_cached', mocked_calendar_list
     )
     monkeypatch.setattr(Printer, 'msg', mocked_msg)
 
     def _init(**opts):
         return GoogleCalendarInterface(use_cache=False, **opts)
 
-    return _init
+    return SimpleNamespace(
+        GCalI=_init,
+        stub_out_cal_service=lambda: monkeypatch.setattr(
+            GoogleCalendarInterface, 'get_cal_service', mocked_cal_service
+        ),
+    )
+
+
+@pytest.fixture
+def patched_google_reauth(monkeypatch):
+    def mocked_refresh_grant(*args, **kw):
+        expiry = datetime.now() + timedelta(minutes=60)
+        grant_response = {}
+        return (
+            'some_access_token',
+            'some_refresh_token',
+            expiry,
+            grant_response,
+            'some_rapt_token',
+        )
+
+    monkeypatch.setattr(
+        google.oauth2.reauth, 'refresh_grant', mocked_refresh_grant
+    )
+    return monkeypatch

tests/data/legacy_oauth_creds.json

@@ -0,0 +1 @@
+{"access_token": "d3adb33f", "client_id": "someclient.apps.googleusercontent.com", "client_secret": "SECRET-secret-SECRET", "refresh_token": "SOME-REFRESH-TOKEN", "token_expiry": "2099-01-01T00:00:00Z", "token_uri": "https://oauth2.googleapis.com/token", "user_agent": "gcalcli/v4.3.0", "revoke_uri": "https://oauth2.googleapis.com/revoke", "id_token": null, "id_token_jwt": null, "token_response": {"access_token": "some.ACCESS-tOKeN", "expires_in": 3599, "refresh_token": "SOME-REFRESH-TOKEN", "scope": "https://www.googleapis.com/auth/calendar", "token_type": "Bearer"}, "scopes": ["https://www.googleapis.com/auth/calendar"], "token_info_uri": "https://oauth2.googleapis.com/tokeninfo", "invalid": false, "_class": "OAuth2Credentials", "_module": "oauth2client.client"}

tests/test_auth.py

@@ -0,0 +1,28 @@
+import pathlib
+import shutil
+
+try:
+    import cPickle as pickle  # type: ignore
+except Exception:
+    import pickle
+
+import googleapiclient.discovery
+
+TEST_DATA_DIR = pathlib.Path(__file__).parent / 'data'
+
+
+def test_legacy_certs(tmpdir, gcali_patches, patched_google_reauth):
+    tmpdir = pathlib.Path(tmpdir)
+    oauth_filepath = tmpdir / 'oauth'
+    shutil.copy(TEST_DATA_DIR / 'legacy_oauth_creds.json', oauth_filepath)
+    gcal = gcali_patches.GCalI(config_folder=tmpdir, refresh_cache=False)
+    assert isinstance(
+        gcal.get_cal_service(), googleapiclient.discovery.Resource
+    )
+    with open(oauth_filepath, 'rb') as gcalcli_oauth:
+        try:
+            pickle.load(gcalcli_oauth)
+        except Exception as e:
+            raise AssertionError(
+                f"Couldn't load oauth file as updated pickle format: {e}"
+            )