diff --git a/garminconnect/__init__.py b/garminconnect/__init__.py index d0649a68..26c9ba78 100644 --- a/garminconnect/__init__.py +++ b/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: diff --git a/garminconnect/client.py b/garminconnect/client.py index 9259cf43..17d0ae10 100644 --- a/garminconnect/client.py +++ b/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", 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 + # "..;", 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('/')}" diff --git a/scripts/generate_exercises.py b/scripts/generate_exercises.py index 82b799ed..6cb79bf0 100644 --- a/scripts/generate_exercises.py +++ b/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 ``