Security/audit 20260810

garminconnect/__init__.py

@@ -1488,15 +1488,19 @@ def get_functional_threshold_power_range(
         url = (
             f"{self.garmin_connect_biometric_stats_url}"
             f"/functionalThresholdPower/range/{start}/{end}"
-            f"?sport={normalized_sport}&aggregation={aggregation}&aggregationStrategy=LATEST"
         )
+        params = {
+            "sport": normalized_sport,
+            "aggregation": aggregation,
+            "aggregationStrategy": "LATEST",
+        }
         logger.debug(
             "Requesting functional threshold power from %s to %s for sport %s",
             start,
             end,
             normalized_sport,
         )
-        return self.connectapi(url)
+        return self.connectapi(url, params=params)
 
     def get_lactate_threshold(
         self,
@@ -1592,20 +1596,23 @@ def get_lactate_threshold(
             aggregation=aggregation,
         )
 
+        params = {
+            "sport": "RUNNING",
+            "aggregation": aggregation,
+            "aggregationStrategy": "LATEST",
+        }
         speed_url = (
             f"{self.garmin_connect_biometric_stats_url}"
             f"/lactateThresholdSpeed/range/{start_date}/{end_date}"
-            f"?sport=RUNNING&aggregation={aggregation}&aggregationStrategy=LATEST"
         )
 
         heart_rate_url = (
             f"{self.garmin_connect_biometric_stats_url}"
             f"/lactateThresholdHeartRate/range/{start_date}/{end_date}"
-            f"?sport=RUNNING&aggregation={aggregation}&aggregationStrategy=LATEST"
         )
 
-        speed = self.connectapi(speed_url)
-        heart_rate = self.connectapi(heart_rate_url)
+        speed = self.connectapi(speed_url, params=params)
+        heart_rate = self.connectapi(heart_rate_url, params=params)
 
         return {"speed": speed, "heart_rate": heart_rate, "power": power}
 
@@ -2991,11 +2998,11 @@ def get_gear_activities(
         limit = _validate_positive_integer(limit, "limit")
         # Optional: enforce a reasonable ceiling to avoid heavy responses
         limit = min(limit, MAX_ACTIVITY_LIMIT)
-        url = f"{self.garmin_connect_activities_baseurl}{gearUUID}/gear?start=0&limit={limit}"
+        url = f"{self.garmin_connect_activities_baseurl}{gearUUID}/gear"
         logger.debug("Requesting activities for gearUUID %s", gearUUID)
 
         try:
-            return self.connectapi(url)
+            return self.connectapi(url, params={"start": 0, "limit": limit})
         except GarminConnectConnectionError as e:
             status = getattr(getattr(e, "response", None), "status_code", None)
             if status == 404:

garminconnect/client.py

@@ -22,6 +22,7 @@
 from collections.abc import Iterator, Mapping
 from pathlib import Path
 from typing import Any, cast
+from urllib.parse import unquote
 
 import requests
 from requests.adapters import HTTPAdapter
@@ -69,9 +70,11 @@ def token_file_path(path: str) -> Path:
             f"Token path must not reference another user's home directory: {path!r}"
         )
     token_path = Path(path).expanduser()
-    # Reject symlinks on the tokenstore path or its immediate parent
-    # (e.g. ~/.garminconnect -> /attacker/dir).
-    for check_path in (token_path, token_path.parent):
+    # Reject symlinks anywhere in the tokenstore ancestry (e.g.
+    # ~/.garminconnect -> /attacker/dir). O_NOFOLLOW on the final open()
+    # only covers the last component; an intermediate symlinked directory
+    # would still redirect load/dump/logout into an attacker-controlled tree.
+    for check_path in (token_path, *token_path.parents):
         try:
             if check_path.is_symlink():
                 raise ValueError(f"Token path must not be a symlink: {path!r}")
@@ -187,6 +190,20 @@ def _build_basic_auth(client_id: str) -> str:
     return "Basic " + base64.b64encode(f"{client_id}:".encode()).decode()
 
 
+_QUERY_VALUE_RE = re.compile(r"([?&][\w.-]+=)[^&\s)'\"]+")
+
+
+def _sanitize_exception_text(err: Exception) -> str:
+    """Render an exception for logs/messages with URL query values redacted.
+
+    ``requests`` embeds the full request URL — query string included — in its
+    exception text. On the login fallback path that URL carries the CAS
+    service ticket (``?ticket=ST-...``), so logging or re-raising the raw
+    exception would leak a credential into application logs and bug reports.
+    """
+    return _QUERY_VALUE_RE.sub(r"\1<redacted>", f"{type(err).__name__}: {err}")
+
+
 def _iter_file_objects(kwargs: dict[str, Any]) -> Iterator[Any]:
     """Yield file-like objects referenced by request kwargs (files/data)."""
     files = kwargs.get("files")
@@ -408,7 +425,10 @@ def _verify_token(self) -> bool:
             _LOGGER.debug("Token validation inconclusive (kept): %s", msg)
             return True
         except Exception as e:
-            _LOGGER.debug("Token validation inconclusive (kept): %s", e)
+            _LOGGER.debug(
+                "Token validation inconclusive (kept): %s",
+                _sanitize_exception_text(e),
+            )
             return True
 
     def get_api_headers(self) -> dict[str, str]:
@@ -532,12 +552,14 @@ def resolve_mfa(name: str) -> tuple[str | None, Any]:
                     last_err = e
                     continue
             except GarminConnectTooManyRequestsError as e:
-                _LOGGER.warning("%s returned 429: %s", name, e)
+                _LOGGER.warning(
+                    "%s returned 429: %s", name, _sanitize_exception_text(e)
+                )
                 rate_limited_count += 1
                 last_err = e
                 continue
             except Exception as e:
-                _LOGGER.warning("%s failed: %s", name, e)
+                _LOGGER.warning("%s failed: %s", name, _sanitize_exception_text(e))
                 last_err = e
                 continue
 
@@ -547,7 +569,8 @@ def resolve_mfa(name: str) -> tuple[str | None, Any]:
                 "Try again later or check your IP/network."
             )
         raise GarminConnectConnectionError(
-            f"All login strategies exhausted: {last_err}"
+            "All login strategies exhausted: "
+            + (_sanitize_exception_text(last_err) if last_err else "no strategies ran")
         )
 
     # ------------------------------------------------------------------ #
@@ -572,11 +595,15 @@ def _mobile_login_cffi(self, email: str, password: str) -> None:
             except (GarminConnectAuthenticationError, _MFARequired):
                 raise
             except GarminConnectTooManyRequestsError as e:
-                _LOGGER.debug("mobile+cffi(%s) 429: %s", imp, e)
+                _LOGGER.debug(
+                    "mobile+cffi(%s) 429: %s", imp, _sanitize_exception_text(e)
+                )
                 last_err = e
                 continue
             except Exception as e:
-                _LOGGER.debug("mobile+cffi(%s) failed: %s", imp, e)
+                _LOGGER.debug(
+                    "mobile+cffi(%s) failed: %s", imp, _sanitize_exception_text(e)
+                )
                 last_err = e
                 continue
         if last_err:
@@ -938,11 +965,15 @@ def _portal_web_login_cffi(self, email: str, password: str) -> None:
             except (GarminConnectAuthenticationError, _MFARequired):
                 raise
             except GarminConnectTooManyRequestsError as e:
-                _LOGGER.debug("portal+cffi(%s) 429: %s", imp, e)
+                _LOGGER.debug(
+                    "portal+cffi(%s) 429: %s", imp, _sanitize_exception_text(e)
+                )
                 last_err = e
                 continue
             except Exception as e:
-                _LOGGER.debug("portal+cffi(%s) failed: %s", imp, e)
+                _LOGGER.debug(
+                    "portal+cffi(%s) failed: %s", imp, _sanitize_exception_text(e)
+                )
                 last_err = e
                 continue
         if last_err:
@@ -1214,7 +1245,10 @@ def _establish_session(
             self._exchange_service_ticket(ticket, service_url=service_url)
             return
         except Exception as e:
-            _LOGGER.warning("DI token exchange failed (%s), falling back to JWT_WEB", e)
+            _LOGGER.warning(
+                "DI token exchange failed (%s), falling back to JWT_WEB",
+                _sanitize_exception_text(e),
+            )
 
         # Fallback: consume ticket via connect.garmin.com for JWT_WEB cookie
         if sess is not None:
@@ -1396,7 +1430,9 @@ def _refresh_session(self) -> None:
                         with contextlib.suppress(Exception):
                             self.dump(self._tokenstore_path)
                 except Exception as err:
-                    _LOGGER.debug("DI token refresh failed: %s", err)
+                    _LOGGER.debug(
+                        "DI token refresh failed: %s", _sanitize_exception_text(err)
+                    )
                 return
 
             # JWT_WEB refresh via CAS TGT
@@ -1436,7 +1472,7 @@ def _refresh_session(self) -> None:
                         self.jwt_web = c.value
                         break
             except Exception as err:
-                _LOGGER.debug("Refresh failed: %s", err)
+                _LOGGER.debug("Refresh failed: %s", _sanitize_exception_text(err))
 
     def dumps(self) -> str:
         """Serialize session state to JSON string."""
@@ -1587,8 +1623,23 @@ def _run_request(self, method: str, path: str, **kwargs: Any) -> Any:
                 self._refresh_session()
 
         # Defense-in-depth: callers must pass clean path components; query strings
-        # belong in the `params` kwarg, not embedded in the path.
-        if ".." in path or "?" in path or "#" in path:
+        # belong in the `params` kwarg, not embedded in the path. Validate the
+        # percent-decoded form: requests' requote_uri() decodes unreserved
+        # characters (e.g. %2e -> .) after this check, so a literal-only match
+        # would let a %2e%2e traversal slip through.
+        decoded_path = unquote(path)
+        # A quoted display name may legitimately contain a run of dots (e.g.
+        # "first..last"); only a path *segment* that is exactly ".." (or
+        # "..;<matrix-params>", a known filter-bypass trick) is traversal.
+        has_traversal_segment = any(
+            segment.split(";", 1)[0] == ".." for segment in decoded_path.split("/")
+        )
+        if (
+            has_traversal_segment
+            or "?" in decoded_path
+            or "#" in decoded_path
+            or "\\" in decoded_path
+        ):
             raise ValueError(f"Invalid API path: {path!r}")
 
         url = f"{self._connectapi}/{path.lstrip('/')}"

scripts/generate_exercises.py

@@ -6,7 +6,10 @@
 
 1. Open the workout editor at https://connect.garmin.com, add a strength
    exercise, and open the exercise picker.
-2. Copy the picker's ``<ul>`` (or the whole page) HTML into a file.
+2. Copy only the picker's ``<ul>`` HTML into a file.  Do not paste the
+   whole page: it comes from an authenticated session and may contain
+   account data, and the generated file is published to a public
+   repository.
 3. Run::
 
        python scripts/generate_exercises.py path/to/picker.html
@@ -22,23 +25,89 @@
 import html
 import re
 import sys
+from html.parser import HTMLParser
 from pathlib import Path
 
 OUT = Path(__file__).resolve().parent.parent / "garminconnect" / "exercises.py"
 
-ROW = re.compile(
-    r'data-category-key="([^"]*)"\s+'
-    r'data-exercise-key="([^"]*)"'
-    r".*?<span>([^<]*)</span>",
-    re.S,
+
+class _PickerParser(HTMLParser):
+    """Collect (category, exercise, name) rows, each scoped to its own <li>.
+
+    Regex extraction with DOTALL can reach past an element boundary and
+    capture a <span> from unrelated page chrome; a real parser cannot, and
+    it also keeps items separate when a closing </li> is omitted.
+    """
+
+    def __init__(self) -> None:
+        super().__init__()
+        self.rows: list[tuple[str, str, str]] = []
+        self._cur: list[str | None] | None = None
+        # Nesting depth of <span> while inside the selected span; 0 means
+        # not in one. A plain in/out flag would drop back out on an inner
+        # </span>, losing any text after it (e.g. "Squat" in
+        # "<span><span>Back</span> Squat</span>").
+        self._span_depth = 0
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        a = dict(attrs)
+        if tag == "li":
+            # An unclosed previous <li> ends here, not at its </li>.
+            self._flush()
+            if "data-category-key" in a and "data-exercise-key" in a:
+                self._cur = [a["data-category-key"], a["data-exercise-key"], None]
+        elif tag == "span" and self._cur is not None:
+            if self._span_depth:
+                self._span_depth += 1
+            elif self._cur[2] is None:
+                self._span_depth = 1
+
+    def handle_data(self, data: str) -> None:
+        # A nested tag inside the span (e.g. <b>) triggers another
+        # handle_data call for its own text; accumulate rather than
+        # overwrite, or a fragment like the "https://" prefix of a stray
+        # URL is dropped and the SUSPECT filter never sees it.
+        if self._span_depth and self._cur is not None:
+            self._cur[2] = (self._cur[2] or "") + data
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "span" and self._span_depth:
+            self._span_depth -= 1
+        elif tag == "li" and self._cur is not None:
+            self._flush()
+
+    def _flush(self) -> None:
+        # Attribute-only element (no span of its own): skip rather than
+        # borrowing text from elsewhere in the document.
+        if self._cur is not None and self._cur[2]:
+            self.rows.append((self._cur[0] or "", self._cur[1] or "", self._cur[2]))
+        self._cur = None
+        self._span_depth = 0
+
+
+# Names are display labels; anything shaped like session data means the
+# extraction went out of scope (or the wrong HTML was pasted).
+SUSPECT = re.compile(
+    r"[\w.+-]+@[\w-]+\.[\w.]+"  # e-mail address
+    r"|eyJ[\w-]{10,}"  # JWT
+    r"|https?://"  # URL
+    r"|[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}",  # GUID
+    re.IGNORECASE,
 )
 
 
 def parse(text: str) -> list[dict[str, str]]:
     """Extract unique (name, category, exercise) rows from picker HTML."""
+    parser = _PickerParser()
+    parser.feed(text)
+    parser.close()
+    # HTMLParser.close() doesn't synthesize a missing </li>, so the last
+    # item (if the source HTML omits its closing tag) needs an explicit
+    # flush here.
+    parser._flush()
     seen: set[tuple[str, str]] = set()
     out: list[dict[str, str]] = []
-    for category, exercise, name in ROW.findall(text):
+    for category, exercise, name in parser.rows:
         key = (category, exercise)
         if key in seen:
             continue  # the "Recent" block repeats items listed below
@@ -112,6 +181,15 @@ def main() -> None:
     if len(sys.argv) != 2:
         sys.exit("usage: python scripts/generate_exercises.py <picker.html>")
     exercises = parse(Path(sys.argv[1]).read_text(encoding="utf-8"))
+    suspect = [i for i, e in enumerate(exercises) if SUSPECT.search(e["name"])]
+    if suspect:
+        # Report only row positions — the matched labels are exactly the
+        # potentially sensitive text we refuse to write out.
+        sys.exit(
+            f"refusing to write: {len(suspect)} entr{'ies' if len(suspect) != 1 else 'y'} "
+            "look like session data, not exercise names (rows "
+            f"{suspect}; check the pasted HTML)"
+        )
     OUT.write_text(render(exercises), encoding="utf-8")
     print(f"Wrote {len(exercises)} exercises to {OUT}")
     print("Run `pdm run format` to normalize quoting/formatting.")