The native DI OAuth token writer stored credentials with umask-dependent permissions. Version 0.3.5 repairs successful owner-controlled writes; untouched older stores still require rewriting or manual restriction.

-This example demonstrates the basic usage of python-garminconnect:-- Authentication with email/password-- Token storage and automatic reuse-- MFA (Multi-Factor Authentication) support-- Comprehensive error handling for all API calls-- Basic API calls for user stats+Demonstrates authentication, token storage, and basic API calls. -For a comprehensive demo of all available API calls, see demo.py+For a comprehensive demo of all 127+ API methods, see demo.py  Dependencies:-pip3 install garth requests+    pip install garminconnect[example]+    pip install curl_cffi  Environment Variables (optional):-export EMAIL=<your garmin email address>-export PASSWORD=<your garmin password>-export GARMINTOKENS=<path to token storage>+    export EMAIL=<your garmin email address>+    export PASSWORD=<your garmin password>+    export GARMINTOKENS=<path to token storage, default ~/.garminconnect> """ +import contextlib import logging import os import sys from datetime import date from getpass import getpass from pathlib import Path -import requests-from garth.exc import GarthException, GarthHTTPError- from garminconnect import (     Garmin,     GarminConnectAuthenticationError,     GarminConnectConnectionError,     GarminConnectTooManyRequestsError, ) -# Suppress garminconnect library logging to avoid tracebacks in normal operation logging.getLogger("garminconnect").setLevel(logging.CRITICAL)   def safe_api_call(api_method, *args, **kwargs):-    """Safe API call wrapper with comprehensive error handling.--    This demonstrates the error handling patterns used throughout the library.-    Returns (success: bool, result: Any, error_message: str)-    """+    """Call an API method and return (success, result, error_message)."""     try:         result = api_method(*args, **kwargs)         return True, result, None -    except GarthHTTPError as e:-        # Handle specific HTTP errors gracefully+    except GarminConnectAuthenticationError as e:+        return False, None, f"Authentication error: {e}"+    except GarminConnectTooManyRequestsError as e:+        return False, None, f"Rate limit exceeded: {e}"+    except GarminConnectConnectionError as e:         error_str = str(e)-        status_code = getattr(getattr(e, "response", None), "status_code", None)--        if status_code == 400 or "400" in error_str:-            return (-                False,-                None,-                "Endpoint not available (400 Bad Request) - Feature may not be enabled for your account",-            )-        if status_code == 401 or "401" in error_str:+        if "400" in error_str:             return (                 False,                 None,-                "Authentication required (401 Unauthorized) - Please re-authenticate",+                "Not available (400) — feature may not be enabled for your account",             )-        if status_code == 403 or "403" in error_str:-            return (

+import json+import logging+from pathlib import Path+from typing import Any++import requests++try:+    from curl_cffi import requests as cffi_requests++    HAS_CFFI = True+except ImportError:+    HAS_CFFI = False++from .exceptions import (+    GarminConnectAuthenticationError,+    GarminConnectConnectionError,+    GarminConnectTooManyRequestsError,+)++_LOGGER = logging.getLogger(__name__)++# Auth constants (matching Android GCM app)+MOBILE_SSO_CLIENT_ID = "GCM_ANDROID_DARK"+MOBILE_SSO_SERVICE_URL = "https://mobile.integration.garmin.com/gcm/android"+MOBILE_SSO_USER_AGENT = (+    "Mozilla/5.0 (Linux; Android 13; sdk_gphone64_arm64 Build/TE1A.220922.025; wv) "+    "AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/132.0.0.0 Mobile Safari/537.36"+)+NATIVE_API_USER_AGENT = "GCM-Android-5.23"+NATIVE_X_GARMIN_USER_AGENT = (+    "com.garmin.android.apps.connectmobile/5.23; ; Google/sdk_gphone64_arm64/google; "+    "Android/33; Dalvik/2.1.0"+)+DI_TOKEN_URL = "https://diauth.garmin.com/di-oauth2-service/oauth/token"  # noqa: S105+IT_TOKEN_URL = "https://services.garmin.com/api/oauth/token"  # noqa: S105+DI_GRANT_TYPE = (+    "https://connectapi.garmin.com/di-oauth2-service/oauth/grant/service_ticket"+)+DI_CLIENT_IDS = (+    "GARMIN_CONNECT_MOBILE_ANDROID_DI_2025Q2",+    "GARMIN_CONNECT_MOBILE_ANDROID_DI_2024Q4",+    "GARMIN_CONNECT_MOBILE_ANDROID_DI",+)+IT_CLIENT_IDS = (+    "GARMIN_CONNECT_MOBILE_ANDROID_2025Q2",+    "GARMIN_CONNECT_MOBILE_ANDROID_2024Q4",+    "GARMIN_CONNECT_MOBILE_ANDROID",+)+++def _build_basic_auth(client_id: str) -> str:+    return "Basic " + base64.b64encode(f"{client_id}:".encode()).decode()+++def _native_headers(extra: dict[str, str] | None = None) -> dict[str, str]:+    headers: dict[str, str] = {+        "User-Agent": NATIVE_API_USER_AGENT,+        "X-Garmin-User-Agent": NATIVE_X_GARMIN_USER_AGENT,+        "X-Garmin-Paired-App-Version": "10861",+        "X-Garmin-Client-Platform": "Android",+        "X-App-Ver": "10861",+        "X-Lang": "en",+        "X-GCExperience": "GC5",+        "Accept-Language": "en-US,en;q=0.9",+    }+    if extra:+        headers.update(extra)+    return headers+++class Client:+    """A client to communicate with Garmin Connect."""++    def __init__(self, domain: str = "garmin.com", **kwargs: Any) -> None:+        self.domain = domain+        self._sso = f"https://sso.{domain}"+        self._connect = f"https://connect.{domain}"+        self._connectapi = f"https://connectapi.{domain}"++        # Native Bearer tokens (primary auth)+        self.di_token: str | None = None+        self.di_refresh_token: str | None = None+        self.di_client_id: str | None = None+        self.it_token: str | None = None

 from pathlib import Path from typing import Any -import garth-import garth.http import requests-from garth.exc import GarthException, GarthHTTPError-from requests import HTTPError, Session+from requests import HTTPError +from . import client from .fit import FitEncoderWeight  # type: ignore -# Workaround: reset the requests Session on each garth Client.request call.-# A stale/corrupted session object causes auth failures for some users.-# Reported fix by user on garth 0.5.21: https://github.com/cyberjunky/python-garminconnect/issues-_orig_garth_request = garth.http.Client.request---def _fresh_session_request(-    self: garth.http.Client,-    method: str,-    subdomain: str,-    path: str,-    /,-    api: bool = False,-    referrer: str | bool = False,-    headers: dict | None = None,-    **kwargs: Any,-) -> requests.Response:-    # Only reset for connectapi calls — SSO login is stateful and depends on-    # cookies persisting across sequential requests (embed→signin→MFA→ticket).-    if headers is None:-        headers = {}-    if subdomain == "connectapi":-        self.sess = Session()-        self.sess.headers.update(garth.http.USER_AGENT)-        self.configure()-    return _orig_garth_request(-        self,-        method,-        subdomain,-        path,-        api=api,-        referrer=referrer,-        headers=headers,-        **kwargs,-    )---garth.http.Client.request = _fresh_session_request  # type: ignore[method-assign]- logger = logging.getLogger(__name__)  # Constants for validation@@ -347,7 +307,7 @@ def __init__(             "/lifestylelogging-service/dailyLog"         ) -        self.garth = garth.Client(+        self.client = client.Client(             domain="garmin.cn" if is_cn else "garmin.com",             pool_connections=20,             pool_maxsize=20,@@ -358,28 +318,14 @@ def __init__(         self.unit_system = None      def connectapi(self, path: str, **kwargs: Any) -> Any:-        """Wrapper for garth connectapi with error handling."""+        """Wrapper for native connectapi with error handling."""         try:-            return self.garth.connectapi(path, **kwargs)-        except AssertionError as e:-            # Handle Windows-specific OAuth token refresh issue-            # This can occur when garth tries to refresh tokens during API calls-            error_msg = str(e).lower()-            if "oauth" in error_msg and (-                "oauth1" in error_msg or "oauth2" in error_msg-            ):-                logger.exception("OAuth token refresh failed during API call.")-                raise GarminConnectAuthenticationError(-                    f"Token refresh failed. Please re-authenticate. Original error: {e}"

@@ -1191,12 +1192,33 @@ def dumps(self) -> str:         return json.dumps(data)      def dump(self, path: str) -> None:-        """Write tokens safely to disk."""+        """Write tokens safely to disk with owner-only permissions.++        The token file contains the DI refresh token, which grants persistent+        account access. It is written as 0o600 inside a 0o700 directory so a+        permissive process umask can't leave it world-readable on a shared host+        (GHSA-wjhr-76vg-2hvc).+        """         p = Path(path).expanduser()         if p.is_dir() or not p.name.endswith(".json"):             p = p / "garmin_tokens.json"-        p.parent.mkdir(parents=True, exist_ok=True)-        p.write_text(self.dumps())+        p.parent.mkdir(mode=0o700, parents=True, exist_ok=True)+        # mkdir's mode is subject to umask and a no-op if the dir already+        # exists; chmod enforces 0o700 unconditionally.+        with contextlib.suppress(OSError):+            p.parent.chmod(0o700)+        # Open with O_CREAT mode 0o600 (and O_NOFOLLOW where available so a+        # pre-planted symlink can't redirect the write) instead of write_text,+        # which would create the file under the umask first.+        flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC+        if hasattr(os, "O_NOFOLLOW"):+            flags |= os.O_NOFOLLOW+        fd = os.open(p, flags, 0o600)+        with os.fdopen(fd, "w", encoding="utf-8") as token_file:+            token_file.write(self.dumps())+        # Enforce 0o600 even if the file pre-existed with looser permissions.+        with contextlib.suppress(OSError):+            p.chmod(0o600)      def load(self, path: str) -> None:         try:

+owner-only directory (0o700), regardless of the process umask, so it can't be+left world-readable on a shared host.++POSIX-only — file mode bits are not meaningful on Windows.+"""++import os+import stat+import sys+from pathlib import Path++import pytest++from garminconnect.client import Client++pytestmark = pytest.mark.skipif(sys.platform == "win32", reason="POSIX file modes only")+++def _make_client() -> Client:+    c = Client()+    c.di_token = "ACCESS_TOKEN_EXAMPLE"+    c.di_refresh_token = "REFRESH_TOKEN_EXAMPLE"+    c.di_client_id = "CLIENT_ID_EXAMPLE"+    return c+++def _mode(path) -> int:+    return stat.S_IMODE(Path(path).stat().st_mode)+++def test_dump_creates_owner_only_token_file(tmp_path):+    """Under a permissive umask the token file must still be 0o600."""+    old_umask = os.umask(0o022)+    try:+        token_dir = tmp_path / "tokens"+        _make_client().dump(str(token_dir))+        token_file = token_dir / "garmin_tokens.json"++        assert _mode(token_file) == 0o600, oct(_mode(token_file))+        assert _mode(token_dir) == 0o700, oct(_mode(token_dir))+        # World/group bits must be clear+        assert not (_mode(token_file) & (stat.S_IRWXG | stat.S_IRWXO))+        assert not (_mode(token_dir) & (stat.S_IRWXG | stat.S_IRWXO))+    finally:+        os.umask(old_umask)+++def test_dump_tightens_preexisting_loose_permissions(tmp_path):+    """A pre-existing world-readable dir/file is tightened on dump."""+    old_umask = os.umask(0o022)+    try:+        token_dir = tmp_path / "tokens"+        token_dir.mkdir(mode=0o755)+        token_file = token_dir / "garmin_tokens.json"+        token_file.write_text("{}")+        token_file.chmod(0o644)++        _make_client().dump(str(token_dir))++        assert _mode(token_file) == 0o600, oct(_mode(token_file))+        assert _mode(token_dir) == 0o700, oct(_mode(token_dir))+    finally:+        os.umask(old_umask)+++def test_dump_explicit_json_path(tmp_path):+    """A direct ``*.json`` path is also written owner-only."""+    old_umask = os.umask(0o022)+    try:+        token_file = tmp_path / "store" / "garmin_tokens.json"+        _make_client().dump(str(token_file))+        assert _mode(token_file) == 0o600, oct(_mode(token_file))+    finally:+        os.umask(old_umask)

@@ -195,6 +196,7 @@ unfixable = []  # Allow all fixes, including unsafe ones "demo.py" = ["T20", "S101", "ERA", "RUF001", "PTH", "PERF401", "PERF203", "PLR0911", "D103"]  # Demo script - various user-facing patterns "example.py" = ["S110", "PLR0911", "T20", "PERF203"]  # Example script - simple error handling is intentional "test_strategy.py" = ["T20", "D103"]  # Interactive login-strategy diagnostic script+"tests/test_token_permissions.py" = ["S105"]  # synthetic token literals in fixtures  [tool.coverage.run] source = ["garminconnect"]