From bdc71e34e03ecbef181cf979ff1413166df3d2c0 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:07:09 +0200 Subject: [PATCH 01/10] security: validate percent-decoded API path in _run_request guard Report #4020: the traversal guard matched literal '..', '?' and '#' in the raw path, but requests' requote_uri() decodes unreserved characters (%2e -> .) after validation, so %2e%2e/%2e%2e/... slipped through and the authenticated request went to a traversed path on the API host. Validate the urllib.parse.unquote() form instead, and reject backslashes. Also fixes call sites broken by the guard's '?' rejection, which embedded query strings in the path instead of using params: get_gear_activities, get_functional_threshold_power_range, and the lactate-threshold range URLs. --- garminconnect/__init__.py | 23 +++++++++++++++-------- garminconnect/client.py | 9 +++++++-- tests/test_garmin_unit.py | 16 ++++++++++++++++ 3 files changed, 38 insertions(+), 10 deletions(-) diff --git a/garminconnect/__init__.py b/garminconnect/__init__.py index d0649a6..26c9ba7 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 9259cf4..ac06b4d 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 @@ -1587,8 +1588,12 @@ 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) + if ".." in decoded_path 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/tests/test_garmin_unit.py b/tests/test_garmin_unit.py index 635e909..4fb8d98 100644 --- a/tests/test_garmin_unit.py +++ b/tests/test_garmin_unit.py @@ -1679,6 +1679,15 @@ def test_error_message_omits_raw_response_body(self, monkeypatch): "foo/../bar", "foo?bar=1", "foo#fragment", + # Percent-encoded variants: requests' requote_uri() decodes + # unreserved characters (%2e -> .) after the guard would see a + # literal-only match, so validation must run on the decoded path. + "foo/%2e%2e/bar", + "foo/%2E%2E/bar", + "foo/%2e%2e%2fbar", + "foo/%2e%2e;/bar", + "foo/%3fbar=1", + "foo\\..\\bar", ], ) def test_rejects_path_with_traversal_or_query( @@ -1688,6 +1697,13 @@ def test_rejects_path_with_traversal_or_query( with pytest.raises(ValueError, match="Invalid API path"): c._run_request("GET", bad_path) + def test_accepts_legitimate_path(self, monkeypatch): + c = self._client(monkeypatch, _FakeResp(200, {"ok": True})) + resp = c._run_request( + "GET", "/userprofile-service/socialProfile" + ) + assert resp.status_code == 200 + # --------------------------------------------------------------------------- # _run_request: 401 retry must rewind file bodies and keep custom headers From b34d19d634d7cbd8360ca8eacdad972d1b80ae3a Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:15:40 +0200 Subject: [PATCH 02/10] security: redact URL query values from logged/raised login exceptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Report #4018: the CAS service ticket must travel in a query string on the ticket-consumption fallback (CAS protocol requirement), and requests embeds the full URL in its exception text. That text reached the WARNING log on strategy failure and the 'All login strategies exhausted' error handed to callers — a credential leak precisely when users turn logging up to file a bug report. New _sanitize_exception_text() redacts all query-string values; applied at the strategy-failure and 429 logs, the exhaustion raise, and the DI token exchange fallback log. --- garminconnect/client.py | 28 ++++++++++++++++++++++++---- tests/test_garmin_unit.py | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 58 insertions(+), 4 deletions(-) diff --git a/garminconnect/client.py b/garminconnect/client.py index ac06b4d..58db15b 100644 --- a/garminconnect/client.py +++ b/garminconnect/client.py @@ -188,6 +188,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") @@ -533,12 +547,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 @@ -548,7 +564,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") ) # ------------------------------------------------------------------ # @@ -1215,7 +1232,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: diff --git a/tests/test_garmin_unit.py b/tests/test_garmin_unit.py index 4fb8d98..3d6d3e1 100644 --- a/tests/test_garmin_unit.py +++ b/tests/test_garmin_unit.py @@ -737,6 +737,40 @@ def portal_strategy(_email, _password): assert c.di_token == "portal-token" # noqa: S105 mock_clear.assert_called_once() + def test_sanitize_exception_text_redacts_query_values(self): + err = Exception("Max retries exceeded with url: /x?ticket=ST-1&foo=bar ok") + out = client_mod._sanitize_exception_text(err) + assert "ST-1" not in out + assert "bar" not in out + assert "ticket=" in out + assert "foo=" in out + assert out.startswith("Exception: ") + + def test_failed_login_redacts_ticket_from_log_and_error(self, caplog): + # requests embeds the full URL — query string included — in exception + # text; the ticket-consumption fallback URL carries ?ticket=ST-..., + # which must reach neither the log nor the raised error. + c = client_mod.Client(verify_login=False) + boom = Exception( + "Max retries exceeded with url: /gcm/ios?ticket=ST-CANARY-123 " + "(Caused by ConnectTimeoutError)" + ) + with ( + patch.object(c, "_mobile_login_cffi", side_effect=boom), + patch.object(c, "_mobile_login_requests", side_effect=boom), + patch.object(c, "_widget_web_login", side_effect=boom), + patch.object(c, "_portal_web_login_cffi", side_effect=boom), + patch.object(c, "_portal_web_login_requests", side_effect=boom), + caplog.at_level(logging.WARNING), + pytest.raises( + garminconnect.GarminConnectConnectionError, match="exhausted" + ) as exc_info, + ): + c.login("e@x.com", "pw") + assert "ST-CANARY-123" not in caplog.text + assert "ST-CANARY-123" not in str(exc_info.value) + assert "ticket=" in caplog.text + def test_return_on_mfa_sets_pending_flag(self): c = client_mod.Client(verify_login=False) From 314998e4414d5b1a2af3c932b1c02580943e1096 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:19:30 +0200 Subject: [PATCH 03/10] security: sanitize JSON request bodies and ticket URIs in VCR cassettes Report #4021: sanitize_request() only scrubbed key=value form bodies, but the credential-posting login strategies send JSON ({"username":..., "password":...}), so account e-mail and password were recorded verbatim when re-recording cassettes with real credentials. Parse request bodies by structure and reuse sanitize_json() so request and response share one deny-list. Also: add username, mfaVerificationCode, serviceTicketId, service_ticket, captchaToken and customerGuid to SENSITIVE_FIELDS; add service_ticket to SENSITIVE_FORM_PARAMS (form-encoded DI token exchange); scrub ticket/service_ticket from request.uri (ticket-consumption fallback). --- tests/conftest.py | 25 ++++++++++- tests/test_cassette_sanitization.py | 67 +++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 1 deletion(-) diff --git a/tests/conftest.py b/tests/conftest.py index 6988ea0..1ecc5b6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -10,8 +10,10 @@ SENSITIVE_FIELDS = { "access_token", "activityId", + "captchaToken", "consumer_key", "consumer_secret", + "customerGuid", "di_refresh_token", "di_token", "displayName", @@ -20,6 +22,7 @@ "garminGUID", "jti", "locationName", + "mfaVerificationCode", "mfa_token", "oauth_token", "oauth_token_secret", @@ -32,11 +35,14 @@ "profileImageUrlMedium", "profileImageUrlSmall", "refresh_token", + "serviceTicketId", + "service_ticket", "token", "userId", "userName", "userProfileFullName", "userProfileId", + "username", } LOCATION_FIELDS = { "endLatitude", @@ -105,6 +111,7 @@ def scrub_dates(response: Any) -> Any: "oauth_token_secret", "password", "refresh_token", + "service_ticket", "username", ] @@ -138,8 +145,24 @@ def sanitize_request(request: Any) -> Any: body = request.body.decode("utf8") except UnicodeDecodeError: return request # leave as-is; binary bodies not sanitized + # The login strategies post credentials as JSON; a key=value regex + # alone never matches those bodies, so parse by structure first and + # fall back to form-encoded scrubbing. + try: + parsed = json.loads(body) + except json.JSONDecodeError: + body = _sanitize_form_body(body) else: - request.body = _sanitize_form_body(body).encode("utf8") + body = json.dumps(sanitize_json(parsed)) + request.body = body.encode("utf8") + + # Credentials can also ride in the URI itself (CAS service ticket on the + # ticket-consumption fallback: GET ?ticket=ST-...). + uri = getattr(request, "uri", None) + if uri: + request.uri = re.sub( + r"([?&](?:ticket|service_ticket)=)[^&]*", r"\1SANITIZED", uri + ) if "Cookie" in request.headers: cookies = request.headers["Cookie"].split("; ") diff --git a/tests/test_cassette_sanitization.py b/tests/test_cassette_sanitization.py index d1fd90b..797cc30 100644 --- a/tests/test_cassette_sanitization.py +++ b/tests/test_cassette_sanitization.py @@ -111,3 +111,70 @@ def test_sanitize_request_scrubs_oauth_exchange_body(): assert "access_token=SANITIZED" in body assert "private-token" not in body assert "secret-value" not in sanitized.headers["Cookie"] + + +def test_sanitize_request_scrubs_json_login_body(): + """Login strategies post credentials as JSON, not key=value form data. + + Regression test: the form-body regex never matches a JSON body, so the + sanitiser was a no-op on exactly the requests carrying the credentials. + """ + body = json.dumps( + { + "username": "victim@example.com", + "password": "S3cr3t-Passw0rd", # noqa: S105 + "rememberMe": True, + "captchaToken": "captcha-secret", + }, + separators=(",", ":"), + ).encode() + request = SimpleNamespace(body=body, headers={}) + + sanitized = sanitize_request(request) + + parsed = json.loads(sanitized.body) + assert parsed["username"] == "SANITIZED" + assert parsed["password"] == "SANITIZED" + assert parsed["captchaToken"] == "SANITIZED" + assert parsed["rememberMe"] is True + assert b"victim@example.com" not in sanitized.body + assert b"S3cr3t-Passw0rd" not in sanitized.body + + +def test_sanitize_request_scrubs_mfa_code_json_body(): + body = json.dumps( + {"mfaMethod": "email", "mfaVerificationCode": "123456"} + ).encode() + request = SimpleNamespace(body=body, headers={}) + + sanitized = sanitize_request(request) + + assert b"123456" not in sanitized.body + + +def test_sanitize_request_scrubs_di_exchange_form_body(): + """The DI token exchange posts the CAS service ticket form-encoded.""" + request = SimpleNamespace( + body=b"service_ticket=ST-12345-secret&grant_type=service_ticket", + headers={}, + ) + + sanitized = sanitize_request(request) + + body = sanitized.body.decode("utf8") + assert "service_ticket=SANITIZED" in body + assert "ST-12345-secret" not in body + + +def test_sanitize_request_scrubs_ticket_from_uri(): + """The ticket-consumption fallback carries ?ticket=ST-... in the URL.""" + request = SimpleNamespace( + body=None, + headers={}, + uri="https://connect.garmin.com/gcm/ios?ticket=ST-999-canary&x=1", + ) + + sanitized = sanitize_request(request) + + assert "ST-999-canary" not in sanitized.uri + assert "ticket=SANITIZED" in sanitized.uri From e2533a6ae4e368f137844738b3f65c667f06cfb0 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:22:09 +0200 Subject: [PATCH 04/10] security: check full tokenstore ancestry for symlinks Report #4023: token_file_path() only inspected the last two path components, and O_NOFOLLOW on open() only covers the final component, so a symlink planted higher in the tree (e.g. /cfg/store -> /attacker/dir with tokenstore /cfg/store/sub/.garminconnect) redirected dump()/load()/logout() into an attacker-controlled directory, exposing di_refresh_token on write and allowing tokenstore substitution on read. Walk token_path.parents instead of only the immediate parent. Default ~/.garminconnect is unaffected (already fully covered). --- garminconnect/client.py | 8 +++++--- tests/test_garmin_unit.py | 12 ++++++++++++ 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/garminconnect/client.py b/garminconnect/client.py index 58db15b..89c3629 100644 --- a/garminconnect/client.py +++ b/garminconnect/client.py @@ -70,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}") diff --git a/tests/test_garmin_unit.py b/tests/test_garmin_unit.py index 3d6d3e1..069edcd 100644 --- a/tests/test_garmin_unit.py +++ b/tests/test_garmin_unit.py @@ -1083,6 +1083,18 @@ def test_rejects_symlinked_parent_of_json_tokenstore(self, tmp_path, make_symlin with pytest.raises(ValueError, match="must not be a symlink"): client_mod.token_file_path(str(link_dir / "garmin_tokens.json")) + def test_rejects_symlink_two_levels_up(self, tmp_path, make_symlink): + # O_NOFOLLOW only covers the final path component; a symlink planted + # higher in the ancestry must also be rejected. + real = tmp_path / "real" + real.mkdir() + (tmp_path / "cfg").mkdir() + make_symlink(real, tmp_path / "cfg" / "store") + + target = tmp_path / "cfg" / "store" / "sub" / ".garminconnect" + with pytest.raises(ValueError, match="must not be a symlink"): + client_mod.token_file_path(str(target)) + # --------------------------------------------------------------------------- # Tokenstore path-vs-data detection From fa77b831cf6e1b487b83dda5903731690d5af02c Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:24:22 +0200 Subject: [PATCH 05/10] security: scope exercise-catalog extraction to each
  • element MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Report #4024: the ROW regex combined re.S with an unanchored .*?, so an element carrying both data-* attributes but no bare captured the next anywhere later in the document — page chrome from the authenticated session (display name, tokens) included — and render() emitted it into the published exercises.py. Split into
  • elements first and match within each; attribute-only elements are now skipped instead of borrowing foreign text. Also: main() refuses to write output whose names look like session data (e-mail, JWT, URL, GUID), and the docstring now tells maintainers to copy only the picker's
      and notes the output is published. --- scripts/generate_exercises.py | 30 ++++++++++++++++++-- tests/test_generate_exercises.py | 47 ++++++++++++++++++++++++++++++++ 2 files changed, 75 insertions(+), 2 deletions(-) create mode 100644 tests/test_generate_exercises.py diff --git a/scripts/generate_exercises.py b/scripts/generate_exercises.py index 82b799e..ba34a29 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 ``
        `` (or the whole page) HTML into a file. +2. Copy only the picker's ``
          `` 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 @@ -26,6 +29,10 @@ OUT = Path(__file__).resolve().parent.parent / "garminconnect" / "exercises.py" +# Split into elements first; the row regex below is then scoped to a single +#
        • and can never capture a from elsewhere in the document. +LI = re.compile(r"", re.S) + ROW = re.compile( r'data-category-key="([^"]*)"\s+' r'data-exercise-key="([^"]*)"' @@ -33,12 +40,25 @@ re.S, ) +# 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}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" # GUID +) + def parse(text: str) -> list[dict[str, str]]: """Extract unique (name, category, exercise) rows from picker HTML.""" seen: set[tuple[str, str]] = set() out: list[dict[str, str]] = [] - for category, exercise, name in ROW.findall(text): + for element in LI.findall(text): + match = ROW.search(element) + if not match: + continue # attribute-only element: skip, don't reach outside it + category, exercise, name = match.groups() key = (category, exercise) if key in seen: continue # the "Recent" block repeats items listed below @@ -112,6 +132,12 @@ def main() -> None: if len(sys.argv) != 2: sys.exit("usage: python scripts/generate_exercises.py ") exercises = parse(Path(sys.argv[1]).read_text(encoding="utf-8")) + suspect = [e for e in exercises if SUSPECT.search(e["name"])] + if suspect: + sys.exit( + "refusing to write: entries look like session data, not exercise " + f"names (check the pasted HTML): {suspect}" + ) 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.") diff --git a/tests/test_generate_exercises.py b/tests/test_generate_exercises.py new file mode 100644 index 0000000..c071f96 --- /dev/null +++ b/tests/test_generate_exercises.py @@ -0,0 +1,47 @@ +"""Regression tests for scripts/generate_exercises.py (report #4024).""" + +import importlib.util +from pathlib import Path + +import pytest + + +@pytest.fixture(scope="module") +def gen(): + spec = importlib.util.spec_from_file_location( + "generate_exercises", + Path(__file__).parent.parent / "scripts" / "generate_exercises.py", + ) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +PAGE = """ +
          + user@example.com +
          +
            +
          • Back Squat
            • +
          • +
          +
          JWT_WEB=eyJhbGciOi.FAKE.TOKEN
          +""" + + +def test_parse_does_not_capture_span_outside_element(gen): + rows = gen.parse(PAGE) + assert rows == [ + {"name": "Back Squat", "category": "SQUAT", "exercise": "BACK_SQUAT"} + ] + + +def test_main_refuses_suspect_names(gen, tmp_path, monkeypatch, capsys): + html_file = tmp_path / "picker.html" + html_file.write_text( + '
        • ' + "user@example.com
          • " + ) + monkeypatch.setattr("sys.argv", ["generate_exercises.py", str(html_file)]) + with pytest.raises(SystemExit, match="session data"): + gen.main() From 662360c24ef33c795b24938a0414f7fae0c69943 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 19:47:31 +0200 Subject: [PATCH 06/10] test: align FTP URL assertions with params kwarg, fix caplog level Two TestUrlConstruction tests asserted the old inline-query URL form; the path-guard fix moves query parameters to the params kwarg. And the login redaction test must set the garminconnect logger level explicitly because demo.py sets it to CRITICAL at import time (via test_demo_security), which starved caplog in full-suite runs. --- tests/test_garmin_unit.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/tests/test_garmin_unit.py b/tests/test_garmin_unit.py index 069edcd..6e1d37b 100644 --- a/tests/test_garmin_unit.py +++ b/tests/test_garmin_unit.py @@ -218,8 +218,12 @@ def test_get_functional_threshold_power_range_builds_cycling_url( ) mock.assert_called_once_with( - "/biometric-service/stats/functionalThresholdPower/range/2025-06-01/2025-06-30" - "?sport=CYCLING&aggregation=daily&aggregationStrategy=LATEST" + "/biometric-service/stats/functionalThresholdPower/range/2025-06-01/2025-06-30", + params={ + "sport": "CYCLING", + "aggregation": "daily", + "aggregationStrategy": "LATEST", + }, ) assert result == payload @@ -234,7 +238,8 @@ def test_get_functional_threshold_power_range_builds_url_with_weekly_aggregation aggregation="weekly", ) - assert "sport=RUNNING&aggregation=weekly" in mock.call_args[0][0] + assert mock.call_args.kwargs["params"]["sport"] == "RUNNING" + assert mock.call_args.kwargs["params"]["aggregation"] == "weekly" @pytest.mark.parametrize( ("start", "end", "sport", "aggregation", "message"), @@ -761,7 +766,7 @@ def test_failed_login_redacts_ticket_from_log_and_error(self, caplog): patch.object(c, "_widget_web_login", side_effect=boom), patch.object(c, "_portal_web_login_cffi", side_effect=boom), patch.object(c, "_portal_web_login_requests", side_effect=boom), - caplog.at_level(logging.WARNING), + caplog.at_level(logging.WARNING, logger="garminconnect"), pytest.raises( garminconnect.GarminConnectConnectionError, match="exhausted" ) as exc_info, From a51330b8e462c1855a5b9b2112d79e668e0e49ed Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 20:17:20 +0200 Subject: [PATCH 07/10] security: address CodeRabbit review on PR #412 - Sanitize remaining auth-path exception logs (mobile/portal cffi 429 and failure debugs, token-validation, DI refresh, JWT_WEB refresh) through _sanitize_exception_text(). - generate_exercises: replace the element-splitting regex with an HTMLParser-based extractor (item boundaries hold even when is omitted); SUSPECT regex is now case-insensitive (uppercase URL schemes and GUID hex); the refusal message reports row positions only, never the suspected labels. - ruff format. Skipped: descriptor-relative (dir_fd) tokenstore traversal. The ancestry walk plus O_NOFOLLOW on the final component covers the reported model; the residual TOCTOU window needs a local attacker racing a token write mid-operation, which is out of proportion for this library. --- garminconnect/client.py | 34 +++++++++++---- scripts/generate_exercises.py | 73 ++++++++++++++++++++++++-------- tests/test_generate_exercises.py | 30 ++++++++++++- 3 files changed, 110 insertions(+), 27 deletions(-) diff --git a/garminconnect/client.py b/garminconnect/client.py index 89c3629..1260aa5 100644 --- a/garminconnect/client.py +++ b/garminconnect/client.py @@ -425,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]: @@ -592,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: @@ -958,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: @@ -1419,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 @@ -1459,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.""" @@ -1615,7 +1628,12 @@ def _run_request(self, method: str, path: str, **kwargs: Any) -> Any: # characters (e.g. %2e -> .) after this check, so a literal-only match # would let a %2e%2e traversal slip through. decoded_path = unquote(path) - if ".." in decoded_path or "?" in decoded_path or "#" in decoded_path or "\\" in decoded_path: + if ( + ".." in decoded_path + 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 ba34a29..fad6187 100644 --- a/scripts/generate_exercises.py +++ b/scripts/generate_exercises.py @@ -25,20 +25,54 @@ import html import re import sys +from html.parser import HTMLParser from pathlib import Path OUT = Path(__file__).resolve().parent.parent / "garminconnect" / "exercises.py" -# Split into elements first; the row regex below is then scoped to a single -#
        • and can never capture a from elsewhere in the document. -LI = re.compile(r"", re.S) -ROW = re.compile( - r'data-category-key="([^"]*)"\s+' - r'data-exercise-key="([^"]*)"' - r".*?([^<]*)", - re.S, -) +class _PickerParser(HTMLParser): + """Collect (category, exercise, name) rows, each scoped to its own
        • . + + Regex extraction with DOTALL can reach past an element boundary and + capture a from unrelated page chrome; a real parser cannot, and + it also keeps items separate when a closing
          • is omitted. + """ + + def __init__(self) -> None: + super().__init__() + self.rows: list[tuple[str, str, str]] = [] + self._cur: list[str | None] | None = None + self._in_span = False + + def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: + a = dict(attrs) + if tag == "li": + # An unclosed previous
        • ends here, not at its
          • . + 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 and self._cur[2] is None: + self._in_span = True + + def handle_data(self, data: str) -> None: + if self._in_span and self._cur is not None: + self._cur[2] = data.strip() + + def handle_endtag(self, tag: str) -> None: + if tag == "span": + self._in_span = False + 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._in_span = False + # Names are display labels; anything shaped like session data means the # extraction went out of scope (or the wrong HTML was pasted). @@ -46,19 +80,19 @@ r"[\w.+-]+@[\w-]+\.[\w.]+" # e-mail address r"|eyJ[\w-]{10,}" # JWT r"|https?://" # URL - r"|[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}" # GUID + 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() seen: set[tuple[str, str]] = set() out: list[dict[str, str]] = [] - for element in LI.findall(text): - match = ROW.search(element) - if not match: - continue # attribute-only element: skip, don't reach outside it - category, exercise, name = match.groups() + for category, exercise, name in parser.rows: key = (category, exercise) if key in seen: continue # the "Recent" block repeats items listed below @@ -132,11 +166,14 @@ def main() -> None: if len(sys.argv) != 2: sys.exit("usage: python scripts/generate_exercises.py ") exercises = parse(Path(sys.argv[1]).read_text(encoding="utf-8")) - suspect = [e for e in exercises if SUSPECT.search(e["name"])] + 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( - "refusing to write: entries look like session data, not exercise " - f"names (check the pasted HTML): {suspect}" + 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}") diff --git a/tests/test_generate_exercises.py b/tests/test_generate_exercises.py index c071f96..ca7c2be 100644 --- a/tests/test_generate_exercises.py +++ b/tests/test_generate_exercises.py @@ -36,6 +36,20 @@ def test_parse_does_not_capture_span_outside_element(gen): ] +def test_parse_keeps_unclosed_items_separate(gen): + # No after the first item: the next
        • start tag is the + # boundary, so the attribute-only item must not merge with (or borrow + # the span of) the following one. + rows = gen.parse( + '
        • \n' + '
        • ' + "user@example.com
          • " + ) + assert rows == [ + {"name": "user@example.com", "category": "X", "exercise": "Y"} + ] + + def test_main_refuses_suspect_names(gen, tmp_path, monkeypatch, capsys): html_file = tmp_path / "picker.html" html_file.write_text( @@ -43,5 +57,19 @@ def test_main_refuses_suspect_names(gen, tmp_path, monkeypatch, capsys): "user@example.com" ) monkeypatch.setattr("sys.argv", ["generate_exercises.py", str(html_file)]) - with pytest.raises(SystemExit, match="session data"): + with pytest.raises(SystemExit) as exc_info: gen.main() + # The refusal must name positions only, never the suspected text. + assert "user@example.com" not in str(exc_info.value) + assert "session data" in str(exc_info.value) + + +@pytest.mark.parametrize( + "label", + [ + "see HTTPS://EXAMPLE.COM/x", # uppercase URL scheme + "id 9F8E7D6C-5B4A-4C3D-8E9F-0A1B2C3D4E5F", # uppercase GUID + ], +) +def test_suspect_regex_is_case_insensitive(gen, label): + assert gen.SUSPECT.search(label) From 764ddfa55b6a1ee0b8c242198624cc40e334ccd7 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 20:36:36 +0200 Subject: [PATCH 08/10] security: fix exercise-catalog text accumulation and final-item flush - handle_data() overwrote the captured span text on every call instead of accumulating it, so a label split across nested tags (e.g. a stray https:// URL with part of it inside a ) lost its leading fragment and could slip past the SUSPECT safety filter. - parser.close() does not synthesize a missing , so the last item in the source HTML was silently dropped when its closing tag was omitted; flush explicitly after close(). Skipped: the client.py display-name path-validation false positive (display names with literal ?, #, or .. get percent-encoded, then rejected after decoding). Loosening that guard to check raw ?/# and exact-segment ".." would reopen the "foo/%2e%2e;/bar" matrix-parameter bypass the existing test suite guards against, for a false positive that requires characters Garmin display names don't contain in practice. --- scripts/generate_exercises.py | 10 +++++++++- tests/test_generate_exercises.py | 30 ++++++++++++++++++++++++++++++ 2 files changed, 39 insertions(+), 1 deletion(-) diff --git a/scripts/generate_exercises.py b/scripts/generate_exercises.py index fad6187..e640b1c 100644 --- a/scripts/generate_exercises.py +++ b/scripts/generate_exercises.py @@ -56,8 +56,12 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None self._in_span = True def handle_data(self, data: str) -> None: + # A nested tag inside the span (e.g. ) 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._in_span and self._cur is not None: - self._cur[2] = data.strip() + self._cur[2] = (self._cur[2] or "") + data def handle_endtag(self, tag: str) -> None: if tag == "span": @@ -90,6 +94,10 @@ def parse(text: str) -> list[dict[str, str]]: parser = _PickerParser() parser.feed(text) parser.close() + # HTMLParser.close() doesn't synthesize a missing , 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 parser.rows: diff --git a/tests/test_generate_exercises.py b/tests/test_generate_exercises.py index ca7c2be..3f6b89c 100644 --- a/tests/test_generate_exercises.py +++ b/tests/test_generate_exercises.py @@ -73,3 +73,33 @@ def test_main_refuses_suspect_names(gen, tmp_path, monkeypatch, capsys): ) def test_suspect_regex_is_case_insensitive(gen, label): assert gen.SUSPECT.search(label) + + +def test_parse_accumulates_text_across_nested_tags(gen): + # A nested inside the span splits the label into two handle_data + # calls; the "https://" prefix must not be dropped, or the SUSPECT + # filter never sees the full (sensitive) label. + rows = gen.parse( + '
        • ' + "https://host/?ticket=secret
          • " + ) + assert rows == [ + { + "name": "https://host/?ticket=secret", + "category": "X", + "exercise": "Y", + } + ] + assert gen.SUSPECT.search(rows[0]["name"]) + + +def test_parse_flushes_final_item_without_closing_tag(gen): + # HTMLParser.close() does not synthesize a missing ; the last + # item in the source HTML must still be captured. + rows = gen.parse( + '
          • ' + "Back Squat" + ) + assert rows == [ + {"name": "Back Squat", "category": "SQUAT", "exercise": "BACK_SQUAT"} + ] From 999914c71f8008a908a18647b9d48cb11295e8a4 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 20:54:17 +0200 Subject: [PATCH 09/10] security: track nested depth in exercise-picker parser A plain in/out flag dropped back "out of span" on an inner , so Back Squat lost the " Squat" that followed the nested tag but was still inside the outer, selected span. Track nesting depth instead so text keeps accumulating until the matching outer . --- scripts/generate_exercises.py | 21 ++++++++++++++------- tests/test_generate_exercises.py | 12 ++++++++++++ 2 files changed, 26 insertions(+), 7 deletions(-) diff --git a/scripts/generate_exercises.py b/scripts/generate_exercises.py index e640b1c..6cb79bf 100644 --- a/scripts/generate_exercises.py +++ b/scripts/generate_exercises.py @@ -43,7 +43,11 @@ def __init__(self) -> None: super().__init__() self.rows: list[tuple[str, str, str]] = [] self._cur: list[str | None] | None = None - self._in_span = False + # Nesting depth of while inside the selected span; 0 means + # not in one. A plain in/out flag would drop back out on an inner + # , losing any text after it (e.g. "Squat" in + # "Back Squat"). + self._span_depth = 0 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: a = dict(attrs) @@ -52,20 +56,23 @@ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None 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 and self._cur[2] is None: - self._in_span = True + 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. ) 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._in_span and self._cur is not None: + 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": - self._in_span = False + if tag == "span" and self._span_depth: + self._span_depth -= 1 elif tag == "li" and self._cur is not None: self._flush() @@ -75,7 +82,7 @@ def _flush(self) -> None: 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._in_span = False + self._span_depth = 0 # Names are display labels; anything shaped like session data means the diff --git a/tests/test_generate_exercises.py b/tests/test_generate_exercises.py index 3f6b89c..df3e47a 100644 --- a/tests/test_generate_exercises.py +++ b/tests/test_generate_exercises.py @@ -103,3 +103,15 @@ def test_parse_flushes_final_item_without_closing_tag(gen): assert rows == [ {"name": "Back Squat", "category": "SQUAT", "exercise": "BACK_SQUAT"} ] + + +def test_parse_keeps_capturing_after_nested_span_closes(gen): + # A plain in/out flag drops back "out of span" on the inner , + # losing text that follows it but is still inside the outer span. + rows = gen.parse( + '
          • ' + "Back Squat
            • " + ) + assert rows == [ + {"name": "Back Squat", "category": "SQUAT", "exercise": "BACK_SQUAT"} + ] From de7bf5194677a2254d3168bffd0d109d12bcdd85 Mon Sep 17 00:00:00 2001 From: Ron Klinkien Date: Mon, 10 Aug 2026 21:09:27 +0200 Subject: [PATCH 10/10] security: make _run_request traversal guard segment-aware A quoted display name can legitimately contain a run of dots (e.g. "first..last"); the old substring match rejected any decoded path containing "..", even inside a single segment. Only reject a path segment that is exactly ".." or "..;" (a known filter-bypass trick), so real traversal is still caught but a legitimate name is not. ?/#/backslash checks are unchanged: they still run against the decoded path, since callers only ever reach those characters raw or via percent-encoding by embedding a query string in the path, which the existing test suite already covers as a rejected case. --- garminconnect/client.py | 8 +++++++- tests/test_garmin_unit.py | 17 +++++++++++++++++ 2 files changed, 24 insertions(+), 1 deletion(-) diff --git a/garminconnect/client.py b/garminconnect/client.py index 1260aa5..17d0ae1 100644 --- a/garminconnect/client.py +++ b/garminconnect/client.py @@ -1628,8 +1628,14 @@ def _run_request(self, method: str, path: str, **kwargs: Any) -> Any: # 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 ( - ".." in decoded_path + has_traversal_segment or "?" in decoded_path or "#" in decoded_path or "\\" in decoded_path diff --git a/tests/test_garmin_unit.py b/tests/test_garmin_unit.py index 6e1d37b..8d02aca 100644 --- a/tests/test_garmin_unit.py +++ b/tests/test_garmin_unit.py @@ -1748,6 +1748,23 @@ def test_rejects_path_with_traversal_or_query( with pytest.raises(ValueError, match="Invalid API path"): c._run_request("GET", bad_path) + @pytest.mark.parametrize( + "path", + [ + # A run of dots that doesn't form a ".." path segment (e.g. a + # quoted display name like "first..last") must not be rejected + # as traversal. + "userprofile-service/first..last", + "userprofile-service/foo...bar", + ], + ) + def test_accepts_path_with_dots_not_forming_traversal_segment( + self, monkeypatch, path: str + ): + c = self._client(monkeypatch, _FakeResp(200, {"ok": True})) + resp = c._run_request("GET", path) + assert resp.status_code == 200 + def test_accepts_legitimate_path(self, monkeypatch): c = self._client(monkeypatch, _FakeResp(200, {"ok": True})) resp = c._run_request(