From 5d9337935492ff4b213cd7dd636843805a819b2e Mon Sep 17 00:00:00 2001 From: Florian Achleitner Date: Sun, 11 Feb 2024 17:51:07 +0100 Subject: [PATCH 1/2] Remove remaining proxy/ paths from URLs --- garminexport/garminclient.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/garminexport/garminclient.py b/garminexport/garminclient.py index dddd6eed..2e431f7e 100755 --- a/garminexport/garminclient.py +++ b/garminexport/garminclient.py @@ -452,7 +452,7 @@ def _poll_upload_completion(self, uuid, creation_date): :obj:`None` if upload is still processing. :rtype: int """ - response = self.session.get("https://connect.garmin.com/proxy/activity-service/activity/status/{}/{}?_={}".format( + response = self.session.get("https://connect.garmin.com/activity-service/activity/status/{}/{}?_={}".format( creation_date[:10], uuid.replace("-",""), int(datetime.now().timestamp()*1000)), headers={"nk": "NT"}) if response.status_code == 201 and response.headers["location"]: # location should be https://connectapi.garmin.com/activity-service/activity/ACTIVITY_ID @@ -495,7 +495,7 @@ def upload_activity(self, file, format=None, name=None, description=None, activi # upload it files = dict(data=(fn, file)) - response = self.session.post("https://connect.garmin.com/proxy/upload-service/upload/.{}".format(format), + response = self.session.post("https://connect.garmin.com/upload-service/upload/.{}".format(format), files=files, headers={"nk": "NT"}) # check response and get activity ID @@ -548,7 +548,7 @@ def upload_activity(self, file, format=None, name=None, description=None, activi data['activityId'] = activity_id encoding_headers = {"Content-Type": "application/json; charset=UTF-8"} # see Tapiriik response = self.session.put( - "https://connect.garmin.com/proxy/activity-service/activity/{}".format(activity_id), + "https://connect.garmin.com/activity-service/activity/{}".format(activity_id), data=json.dumps(data), headers=encoding_headers) if response.status_code != 204: raise Exception(u"failed to set metadata for activity {}: {}\n{}".format( From a53435ad2f971e97925963524e10fcb419cd43dc Mon Sep 17 00:00:00 2001 From: Florian Achleitner Date: Sun, 11 Feb 2024 22:04:46 +0100 Subject: [PATCH 2/2] Use garth for auth and api Replaces the login dance and api calls, and fixes the broken login. Tokens are long-lived and can be stored, thus username and password are optional. The token store is currently hardcoded and insecure. --- garminexport/cli/backup.py | 5 +- garminexport/cli/get_activity.py | 4 +- garminexport/cli/upload_activity.py | 4 +- garminexport/garminclient.py | 215 +++++----------------------- setup.cfg | 7 +- 5 files changed, 44 insertions(+), 191 deletions(-) diff --git a/garminexport/cli/backup.py b/garminexport/cli/backup.py index 87ab0619..e6824d90 100644 --- a/garminexport/cli/backup.py +++ b/garminexport/cli/backup.py @@ -31,10 +31,9 @@ def parse_args() -> argparse.Namespace: "given Garmin Connect account. Only activities that " "aren't already stored in the backup directory will " "be downloaded.")) - # positional args - parser.add_argument( - "username", metavar="", type=str, help="Account user name.") # optional args + parser.add_argument( + "--username", type=str, help="Account user name.") parser.add_argument( "--password", type=str, help="Account password.") parser.add_argument( diff --git a/garminexport/cli/get_activity.py b/garminexport/cli/get_activity.py index c7ceff95..713efad9 100755 --- a/garminexport/cli/get_activity.py +++ b/garminexport/cli/get_activity.py @@ -24,8 +24,6 @@ def main(): description="Downloads one particular activity for a given Garmin Connect account.") # positional args - parser.add_argument( - "username", metavar="", type=str, help="Account user name.") parser.add_argument( "activity", metavar="", type=int, help="Activity ID.") parser.add_argument( @@ -33,6 +31,8 @@ def main(): help="Export format (one of: {}).".format(garminexport.backup.supported_export_formats)) # optional args + parser.add_argument( + "--username", type=str, help="Account user name.") parser.add_argument( "--password", type=str, help="Account password.") parser.add_argument( diff --git a/garminexport/cli/upload_activity.py b/garminexport/cli/upload_activity.py index 1d6d69e1..9ec8e419 100755 --- a/garminexport/cli/upload_activity.py +++ b/garminexport/cli/upload_activity.py @@ -17,13 +17,13 @@ def main(): description="Uploads an activity file to a Garmin Connect account.") # positional args - parser.add_argument( - "username", metavar="", type=str, help="Account user name.") parser.add_argument( "activity", nargs='+', metavar="", type=argparse.FileType("rb"), help="Activity file (.gpx, .tcx, or .fit).") # optional args + parser.add_argument( + "--username", type=str, help="Account user name.") parser.add_argument( "--password", type=str, help="Account password.") parser.add_argument( diff --git a/garminexport/garminclient.py b/garminexport/garminclient.py index 2e431f7e..99502121 100755 --- a/garminexport/garminclient.py +++ b/garminexport/garminclient.py @@ -12,17 +12,13 @@ import logging import os import os.path -import requests +import garth import sys import zipfile from garminexport.retryer import Retryer, ExponentialBackoffDelayStrategy, MaxRetriesStopStrategy - log = logging.getLogger(__name__) -# reduce logging noise from requests library -logging.getLogger("requests").setLevel(logging.ERROR) - PORTAL_LOGIN_URL = "https://sso.garmin.com/portal/api/login" """Garmin Connect's Single-Sign On login URL.""" @@ -41,7 +37,7 @@ def require_session(client_function): @wraps(client_function) def check_session(*args, **kwargs): client_object = args[0] - if not client_object.session: + if not client_object.gc: raise Exception("Attempt to use GarminClient without being connected. Call connect() before first use.'") return client_function(*args, **kwargs) @@ -66,7 +62,7 @@ class GarminClient(object): """ - def __init__(self, username, password): + def __init__(self, username=None, password=None): """Initialize a :class:`GarminClient` instance. :param username: Garmin Connect user name or email address. @@ -76,8 +72,9 @@ def __init__(self, username, password): """ self.username = username self.password = password + self.tokenpath = "~/.garminauth" - self.session = None + self.gc = None def __enter__(self): @@ -88,13 +85,13 @@ def __exit__(self, exc_type, exc_value, traceback): self.disconnect() def connect(self): - self.session = new_http_session() + self.gc = garth.client self._authenticate() def disconnect(self): - if self.session: - self.session.close() - self.session = None + if self.gc: + self.gc.dump(self.tokenpath) + self.gc = None def _authenticate(self): """ @@ -105,132 +102,15 @@ def _authenticate(self): following a sign-in. """ log.info("authenticating user ...") - - auth_ticket_url = self._login(self.username, self.password) - log.debug("auth ticket url: '%s'", auth_ticket_url) - - self._claim_auth_ticket(auth_ticket_url) - - # we need to touch base with the main page to complete the login ceremony. - self.session.get('https://connect.garmin.com/modern') - # This header appears to be needed on subsequent session requests or we - # end up with a 402 response from Garmin. - self.session.headers.update({'NK': 'NT'}) - - # We need to pass an Authorization oauth token with subsequent requests. - auth_token = self._get_oauth_token() - token_type = auth_token['token_type'] - access_token = auth_token['access_token'] - self.session.headers.update( - { - 'Authorization': f'{token_type} {access_token}', - 'Di-Backend': 'connectapi.garmin.com', - }) - - - def _get_oauth_token(self): - """Retrieve an OAuth token to use for the session. - - Typically looks something like the following. The 'access_token' - needs to be passed in the 'Authorization' header for remaining session - requests. - - { - "scope": "...", - "jti": "...", - "access_token": "...", - "token_type": "Bearer", - "refresh_token": "...", - "expires_in": 3599, - "refresh_token_expires_in": 7199 - } - """ - log.info("getting oauth token ...") - headers = { - 'authority': 'connect.garmin.com', - 'origin': 'https://connect.garmin.com', - 'referer': 'https://connect.garmin.com/modern/', - } - resp = self.session.post('https://connect.garmin.com/modern/di-oauth/exchange', - headers=headers) - if resp.status_code != 200: - raise ValueError(f'get oauth token failed with {resp.status_code}: {resp.text}') - return resp.json() - - - def _login(self, username, password): - """Logs in with the supplied account credentials. - The return value is a URL where the created authentication ticket can be claimed. - For example, "https://connect.garmin.com/modern?ticket=ST-2550833-30KdiEJ3jqvFzLNGi2C7-sso" - - The response message looks typically something like this: - { - "serviceURL":"https://connect.garmin.com/modern/", - "serviceTicketId":"ST-2550833-30KdiEJ3jqvFzLNGi2C7-sso", - "responseStatus":{"type":"SUCCESSFUL","message":"","httpStatus":"OK"}, - "customerMfaInfo":null, - "consentTypeList":null - } - """ - headers = { - 'authority': 'sso.garmin.com', - 'origin': 'https://sso.garmin.com', - 'referer': 'https://sso.garmin.com/portal/sso/en-US/sign-in?clientId=GarminConnect&service=https%3A%2F%2Fconnect.garmin.com%2Fmodern', - } - params = { - "clientId": "GarminConnect", - "service": "https://connect.garmin.com/modern/", - "gauthHost": "https://sso.garmin.com/sso", - } - form_data = {'username': username, 'password': password} - - log.info("passing login credentials ...") - resp = self.session.post(PORTAL_LOGIN_URL, headers=headers, params=params, json=form_data) - log.debug("got auth response %d: %s", resp.status_code, resp.text) - if resp.status_code != 200: - raise ValueError(f'authentication attempt failed with {resp.status_code}: {resp.text}') - return self._extract_auth_ticket_url(resp.json()) - - def _claim_auth_ticket(self, auth_ticket_url): - # Note: first we bump the login URL. - p = { - 'clientId': 'GarminConnect', - 'service': 'https://connect.garmin.com/modern/', - 'webhost': 'https://connect.garmin.com', - 'gateway': 'true', - 'generateExtraServiceTicket': 'true', - 'generateTwoExtraServiceTickets': 'true', - } - self.session.get(SSO_LOGIN_URL, headers={}, params=p) - - log.info("claiming auth ticket %s ...", auth_ticket_url) - response = self.session.get(auth_ticket_url) - if response.status_code != 200: - raise RuntimeError( - "auth failure: failed to claim auth ticket: {}: {}\n{}".format( - auth_ticket_url, response.status_code, response.text)) - - - @staticmethod - def _extract_auth_ticket_url(auth_response): - """Extracts an authentication ticket URL from the response of an - authentication form submission. The auth ticket URL is typically - of form: - - https://connect.garmin.com/modern?ticket=ST-0123456-aBCDefgh1iJkLmN5opQ9R-cas - - :param auth_response: JSON response from a login form submission. - """ - if auth_response['responseStatus']['type'] == 'INVALID_USERNAME_PASSWORD': - RuntimeError("authentication failure: did you provide a correct username/password?") - service_url = auth_response.get('serviceURL') - auth_ticket = auth_response.get('serviceTicketId') - if not service_url: - raise RuntimeError("auth failure: unable to extract serviceURL") - if not auth_ticket: - raise RuntimeError("auth failure: unable to extract serviceTicketId") - auth_ticket_url = service_url.rstrip('/') + '?ticket=' + auth_ticket - return auth_ticket_url + try: + log.info("Checking for stored tokens at %s", self.tokenpath) + self.gc.load(self.tokenpath) + log.info("Tokens loaded.") + return + except Exception as e: + log.info("No tokens loaded (%s)", str(e)) + self.gc.login(self.username, self.password) + log.info("Logged in with username and password.") @require_session def list_activities(self): @@ -268,8 +148,8 @@ def _fetch_activity_ids_and_ts(self, start_index, max_limit=100): :rtype: tuples of (int, datetime) """ log.debug("fetching activities %d through %d ...", start_index, start_index + max_limit - 1) - response = self.session.get( - "https://connect.garmin.com/activitylist-service/activities/search/activities", + response = self.gc.get( + "connectapi", "/activitylist-service/activities/search/activities", api=True, params={"start": start_index, "limit": max_limit}) if response.status_code != 200: raise Exception( @@ -301,8 +181,8 @@ def get_activity_summary(self, activity_id): :returns: The activity summary as a JSON dict. :rtype: dict """ - response = self.session.get( - "https://connect.garmin.com/activity-service/activity/{}".format(activity_id)) + response = self.gc.get( + "connectapi", "/activity-service/activity/{}".format(activity_id), api=True) if response.status_code != 200: log.error(u"failed to fetch json summary for activity %s: %d\n%s", activity_id, response.status_code, response.text) @@ -322,8 +202,8 @@ def get_activity_details(self, activity_id): :rtype: dict """ # mounted at xml or json depending on result encoding - response = self.session.get( - "https://connect.garmin.com/activity-service/activity/{}/details".format(activity_id)) + response = self.gc.get( + "connectapi", "/activity-service/activity/{}/details".format(activity_id), api=True) if response.status_code != 200: raise Exception(u"failed to fetch json activityDetails for {}: {}\n{}".format( activity_id, response.status_code, response.text)) @@ -342,12 +222,12 @@ def get_activity_gpx(self, activity_id): or ``None`` if the activity couldn't be exported to GPX. :rtype: str """ - response = self.session.get( - "https://connect.garmin.com/download-service/export/gpx/activity/{}".format(activity_id)) + response = self.gc.get( + "connectapi", "/download-service/export/gpx/activity/{}".format(activity_id), api=True) # An alternate URL that seems to produce the same results # and is the one used when exporting through the Garmin # Connect web page. - # response = self.session.get("https://connect.garmin.com/proxy/activity-service-1.1/gpx/activity/{}?full=true".format(activity_id)) + # response = self.gc.get("connectapi", "/proxy/activity-service-1.1/gpx/activity/{}?full=true".format(activity_id)) # A 404 (Not Found) or 204 (No Content) response are both indicators # of a gpx file not being available for the activity. It may, for @@ -374,8 +254,8 @@ def get_activity_tcx(self, activity_id): :rtype: str """ - response = self.session.get( - "https://connect.garmin.com/download-service/export/tcx/activity/{}".format(activity_id)) + response = self.gc.get( + "connectapi", "/download-service/export/tcx/activity/{}".format(activity_id), api=True) if response.status_code == 404: return None if response.status_code != 200: @@ -395,8 +275,8 @@ def get_original_activity(self, activity_id): its contents, or :obj:`(None,None)` if no file is found. :rtype: (str, str) """ - response = self.session.get( - "https://connect.garmin.com/download-service/files/activity/{}".format(activity_id)) + response = self.gc.get( + "connectapi", "/download-service/files/activity/{}".format(activity_id), api=True) # A 404 (Not Found) response is a clear indicator of a missing .fit # file. As of lately, the endpoint appears to have started to # respond with 500 "NullPointerException" on attempts to download a @@ -452,8 +332,8 @@ def _poll_upload_completion(self, uuid, creation_date): :obj:`None` if upload is still processing. :rtype: int """ - response = self.session.get("https://connect.garmin.com/activity-service/activity/status/{}/{}?_={}".format( - creation_date[:10], uuid.replace("-",""), int(datetime.now().timestamp()*1000)), headers={"nk": "NT"}) + response = self.gc.get("connectapi", "/activity-service/activity/status/{}/{}?_={}".format( + creation_date[:10], uuid.replace("-",""), int(datetime.now().timestamp()*1000)), api=True) if response.status_code == 201 and response.headers["location"]: # location should be https://connectapi.garmin.com/activity-service/activity/ACTIVITY_ID return int(response.headers["location"].split("/")[-1]) @@ -495,8 +375,8 @@ def upload_activity(self, file, format=None, name=None, description=None, activi # upload it files = dict(data=(fn, file)) - response = self.session.post("https://connect.garmin.com/upload-service/upload/.{}".format(format), - files=files, headers={"nk": "NT"}) + response = self.gc.post("connectapi", "/upload-service/upload/.{}".format(format), + files=files, api=True) # check response and get activity ID try: @@ -547,32 +427,11 @@ def upload_activity(self, file, format=None, name=None, description=None, activi if data: data['activityId'] = activity_id encoding_headers = {"Content-Type": "application/json; charset=UTF-8"} # see Tapiriik - response = self.session.put( - "https://connect.garmin.com/activity-service/activity/{}".format(activity_id), + response = self.gc.put( + "connectapi", "/activity-service/activity/{}".format(activity_id), api=True, data=json.dumps(data), headers=encoding_headers) if response.status_code != 204: raise Exception(u"failed to set metadata for activity {}: {}\n{}".format( activity_id, response.status_code, response.text)) return activity_id - - -def new_http_session(): - """Returns a requests-compatible HTTP Session. - See https://requests.readthedocs.io/en/latest/user/advanced/#session-objects. - - By default it uses the requests library to create http sessions. If built with - the 'impersonate-browser' extra, it will use curl_cffi and a patched libcurl to - produce identical TLS fingerprints as a real web browsers to circumvent - Cloudflare's bot protection. - """ - session_factory_func = requests.session - try: - import curl_cffi.requests - # For supported browsers: see https://github.com/lwthiker/curl-impersonate#supported-browsers - browser = os.getenv("GARMINEXPORT_IMPERSONATE_BROWSER", "chrome110") - log.info("using 'curl_cffi' to create HTTP sessions that impersonate web browser '%s' ...", browser) - session_factory_func = partial(curl_cffi.requests.Session, impersonate=browser) - except (ImportError): - pass - return session_factory_func() diff --git a/setup.cfg b/setup.cfg index ac8a88dd..731ef3df 100644 --- a/setup.cfg +++ b/setup.cfg @@ -38,15 +38,10 @@ packages = garminexport.cli python_requires = >=3.5 install_requires = - requests>=2.0,<3 + garth~=0.4.44 python-dateutil~=2.4 [options.extras_require] -# Note: needed to impersonate web browsers. Garmin Connect uses Cloudflare's bot -# protection which looks at TLS fingerprints to determine if the caller is a -# script (like curl) or a web browser. curl_cffi uses a patched libcurl to -# produce identical TLS fingerprints as real web browsers. -impersonate_browser = curl_cffi==0.5.9 test = pytest~=7.3; pytest-cov~=4.0 [options.entry_points]