fix(auth): Suunto - verify cryptographic signatures#540
fix(auth): Suunto - verify cryptographic signatures#540HuziHoic wants to merge 3 commits intothe-momentum:mainfrom
Conversation
HuziHoic
requested review from
KaliszS,
bartmichalak and
czajkub
as code owners
📝 WalkthroughWalkthroughReworks Suunto OAuth user-info extraction to verify JWTs: decodes unverified header to get Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant SuuntoService
participant JWKS as "Suunto JWKS Endpoint"
participant JWKLib as "jose jwk / crypto"
Client->>SuuntoService: provide access token
SuuntoService->>SuuntoService: decode unverified header -> extract kid
SuuntoService->>JWKS: GET /.well-known/jwks.json (optional subscription header)
JWKS-->>SuuntoService: return JWKS (keys)
SuuntoService->>JWKLib: select key by kid, build public key
SuuntoService->>JWKLib: verify JWT (RS256) with audience
JWKLib-->>SuuntoService: claims or verification error
SuuntoService-->>Client: return user info or null on failure
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Tip Try Coding Plans. Let us write the prompt for your AI agent so you can ship faster (with fewer bugs). Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
backend/app/services/providers/suunto/oauth.py (1)
33-45:⚠️ Potential issue | 🟠 MajorAvoid silent catch-all handling in this auth-verification path.
Line 44 catches all exceptions and returns null identifiers, which hides JWKS/config failures and removes actionable observability. Please move to the service exception-handling pattern (
@handle_exceptions) or at least narrow and log failures.As per coding guidelines, "backend/app/services/**/*.py: Services layer must contain business logic and never perform database operations directly; use
@handle_exceptionsdecorator for error handling".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/providers/suunto/oauth.py` around lines 33 - 45, The current try/except around PyJWKClient.get_signing_key_from_jwt and jwt.decode swallows all errors; remove the bare "except Exception" and adopt the service error pattern by adding the `@handle_exceptions` decorator to the function that performs this verification (the function in oauth.py that reads token_response.access_token and returns {"user_id", "username"}), or if you must handle locally, catch specific exceptions (e.g., PyJWKClient/get_signing_key errors, jwt.InvalidTokenError, requests exceptions) and log them with the service logger before returning the fallback; ensure you reference PyJWKClient.get_signing_key_from_jwt, jwt.decode, token_response.access_token and use `@handle_exceptions` on the enclosing function to follow the service layer pattern.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/services/providers/suunto/oauth.py`:
- Line 2: The import of PyJWKClient from jose.jwk is invalid for this project
(we use python-jose), so either switch to PyJWT end-to-end (replace the import
with jwt.PyJWKClient and use PyJWT APIs throughout) or adapt to python-jose by
removing PyJWKClient usage and implement JWKS handling yourself: fetch the JWKS
JSON, locate the key by kid, call jose.jwk.construct(...) to build the key, and
pass that to jose.jwt.decode(...). Also remove the bare `except Exception:`
currently around the JWKS retrieval/verification and instead annotate the
function that does JWKS fetching/validation with the `@handle_exceptions`
decorator (replace the try/except block inside the JWKS-related function with
the decorator) so errors are handled per service guidelines.
---
Outside diff comments:
In `@backend/app/services/providers/suunto/oauth.py`:
- Around line 33-45: The current try/except around
PyJWKClient.get_signing_key_from_jwt and jwt.decode swallows all errors; remove
the bare "except Exception" and adopt the service error pattern by adding the
`@handle_exceptions` decorator to the function that performs this verification
(the function in oauth.py that reads token_response.access_token and returns
{"user_id", "username"}), or if you must handle locally, catch specific
exceptions (e.g., PyJWKClient/get_signing_key errors, jwt.InvalidTokenError,
requests exceptions) and log them with the service logger before returning the
fallback; ensure you reference PyJWKClient.get_signing_key_from_jwt, jwt.decode,
token_response.access_token and use `@handle_exceptions` on the enclosing function
to follow the service layer pattern.
| @@ -1,4 +1,5 @@ | |||
| from jose import jwt | |||
| from jose.jwk import PyJWKClient | |||
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
cat -n backend/app/services/providers/suunto/oauth.pyRepository: the-momentum/open-wearables
Length of output: 2415
🌐 Web query:
Does python-jose expose PyJWKClient or is it a PyJWT API?
💡 Result:
PyJWKClient is a PyJWT API (it’s documented as jwt.PyJWKClient in PyJWT’s API reference). (pyjwt.readthedocs.io)
python-jose does not expose PyJWKClient; instead it has its own jose.jwt / jose.jwk APIs (e.g., jose.jwk.construct(...) for working with JWK material), and if you need JWKS-from-URL behavior you typically fetch the JWKS yourself and then construct/select the key. (python-jose.readthedocs.io)
Citations:
- 1: https://pyjwt.readthedocs.io/en/latest/api.html
- 2: https://python-jose.readthedocs.io/en/latest/?utm_source=openai
🏁 Script executed:
# Check declared dependencies
find . -type f \( -name "pyproject.toml" -o -name "poetry.lock" -o -name "requirements*.txt" -o -name "Pipfile*" \) \
| head -5 \
| xargs -r ls -lh
echo "=== Checking for JWT library declarations ==="
rg -i 'python-jose|pyjwt' -t toml -t lock -t txt 2>/dev/null | head -20 || trueRepository: the-momentum/open-wearables
Length of output: 1289
🏁 Script executed:
# Check the base class and error handling pattern used in services
cat -n backend/app/services/providers/templates/base_oauth.py | head -50Repository: the-momentum/open-wearables
Length of output: 1867
Resolve the JWT library mismatch—this import will fail at runtime.
Line 2 imports PyJWKClient from jose.jwk, but the project declares python-jose (not PyJWT) in backend/pyproject.toml. PyJWKClient is a PyJWT API that doesn't exist in python-jose, so this line will raise an ImportError or AttributeError at import time, preventing the entire module from loading.
Switch to one consistent JWT library: either use PyJWT end-to-end with jwt.PyJWKClient, or use python-jose with its own JWKS key handling approach (e.g., fetch JWKS manually and use jose.jwk.construct(...)).
Additionally, replace the bare except Exception: at line 44–45 with the @handle_exceptions decorator per service layer guidelines.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/providers/suunto/oauth.py` at line 2, The import of
PyJWKClient from jose.jwk is invalid for this project (we use python-jose), so
either switch to PyJWT end-to-end (replace the import with jwt.PyJWKClient and
use PyJWT APIs throughout) or adapt to python-jose by removing PyJWKClient usage
and implement JWKS handling yourself: fetch the JWKS JSON, locate the key by
kid, call jose.jwk.construct(...) to build the key, and pass that to
jose.jwt.decode(...). Also remove the bare `except Exception:` currently around
the JWKS retrieval/verification and instead annotate the function that does JWKS
fetching/validation with the `@handle_exceptions` decorator (replace the
try/except block inside the JWKS-related function with the decorator) so errors
are handled per service guidelines.
There was a problem hiding this comment.
🧹 Nitpick comments (3)
backend/app/services/providers/suunto/oauth.py (3)
56-61: Consider adding issuer validation for defense in depth.The JWT is correctly verified with RS256 signature and audience validation. For additional security hardening, consider also validating the
iss(issuer) claim to ensure the token originates from the expected Suunto authorization server.🛡️ Optional issuer validation
decoded = jwt.decode( token_response.access_token, - public_pem, + matching, algorithms=["RS256"], audience=settings.suunto_client_id, + issuer="https://cloudapi-oauth.suunto.com", )Note: Verify the exact issuer value from Suunto's documentation.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/providers/suunto/oauth.py` around lines 56 - 61, The JWT decode call currently validates signature and audience but lacks issuer checking; update the jwt.decode invocation that processes token_response.access_token (using public_pem and settings.suunto_client_id) to also validate the iss claim by passing the expected issuer (e.g. settings.suunto_issuer) to jwt.decode (or via options/issuer param), and ensure settings.suunto_issuer matches Suunto's documented issuer value.
39-46: Consider caching JWKS to improve performance and reliability.The JWKS is fetched from the remote endpoint on every token verification request. This adds latency and creates a dependency on the external endpoint's availability for each OAuth callback.
Consider caching the JWKS with a reasonable TTL (e.g., 1 hour) to reduce network calls and improve resilience. Libraries like
cachetoolsor a simple module-level cache with expiration could work here.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/providers/suunto/oauth.py` around lines 39 - 46, The JWKS is being fetched on every verification (jwks_url via requests.get / resp.json), causing latency and fragility; implement a module-level cache (e.g., JWKS_CACHE or similar) that stores the JWKS payload and a fetched timestamp and check TTL (e.g., 3600s) before calling requests.get again, returning cached jwks.get("keys", []) when fresh; on fetch failure, fall back to the cached value if present and not expired or raise the original error only if no valid cache exists; update the code paths that use resp, jwks, keys, and matching to read from the cache abstraction instead of always calling requests.get.
48-54: Narrow the exception handling for PEM conversion.The bare
except Exception:at line 52 is overly broad and masks the actual error if key construction fails. Sincepython-jose'sjwt.decode()accepts JWK dicts directly (a supported pattern), the fallback tomatchingwill work. However, catching all exceptions obscures whether the failure is due to an unexpected key type, malformed key, or a legitimate error.Either catch the specific exception (e.g.,
AttributeErrorifto_pem()is unavailable) or simplify by passing the JWK dict directly:♻️ Proposed simplification
- # Construct a key object and obtain a PEM representation for decoding - public_key = jwk.construct(matching) - try: - public_pem = public_key.to_pem() - except Exception: - # Fallback: if to_pem is not available, pass the jwk dict directly - public_pem = matching - decoded = jwt.decode( token_response.access_token, - public_pem, + matching, algorithms=["RS256"], audience=settings.suunto_client_id, )🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/providers/suunto/oauth.py` around lines 48 - 54, The except block that currently catches all exceptions when calling public_key.to_pem() is too broad; instead either catch the specific exception (e.g., AttributeError) thrown when to_pem() is not present, or remove the try/except and always pass the JWK dict `matching` to downstream decode logic (since python-jose accepts JWK dicts). Update the code around jwk.construct(matching) and the public_key.to_pem() call to only handle AttributeError (or remove the fallback entirely) and ensure `public_pem` is either the PEM bytes when to_pem() exists or the original `matching` dict otherwise, so jwt.decode() receives a supported key representation.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Nitpick comments:
In `@backend/app/services/providers/suunto/oauth.py`:
- Around line 56-61: The JWT decode call currently validates signature and
audience but lacks issuer checking; update the jwt.decode invocation that
processes token_response.access_token (using public_pem and
settings.suunto_client_id) to also validate the iss claim by passing the
expected issuer (e.g. settings.suunto_issuer) to jwt.decode (or via
options/issuer param), and ensure settings.suunto_issuer matches Suunto's
documented issuer value.
- Around line 39-46: The JWKS is being fetched on every verification (jwks_url
via requests.get / resp.json), causing latency and fragility; implement a
module-level cache (e.g., JWKS_CACHE or similar) that stores the JWKS payload
and a fetched timestamp and check TTL (e.g., 3600s) before calling requests.get
again, returning cached jwks.get("keys", []) when fresh; on fetch failure, fall
back to the cached value if present and not expired or raise the original error
only if no valid cache exists; update the code paths that use resp, jwks, keys,
and matching to read from the cache abstraction instead of always calling
requests.get.
- Around line 48-54: The except block that currently catches all exceptions when
calling public_key.to_pem() is too broad; instead either catch the specific
exception (e.g., AttributeError) thrown when to_pem() is not present, or remove
the try/except and always pass the JWK dict `matching` to downstream decode
logic (since python-jose accepts JWK dicts). Update the code around
jwk.construct(matching) and the public_key.to_pem() call to only handle
AttributeError (or remove the fallback entirely) and ensure `public_pem` is
either the PEM bytes when to_pem() exists or the original `matching` dict
otherwise, so jwt.decode() receives a supported key representation.
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
backend/app/services/providers/suunto/oauth.py (1)
90-91: Redundant exception types and overly broad catch.The exception tuple
(JWTError, JWSError, requests.RequestException, Exception)is redundant sinceExceptioncatches everything, making the specific types pointless. This also masks unexpected errors. Consider catching only the specific expected exceptions.♻️ Proposed fix
- except (JWTError, JWSError, requests.RequestException, Exception): + except (JWTError, JWSError, requests.RequestException, KeyError, ValueError): return {"user_id": None, "username": None}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@backend/app/services/providers/suunto/oauth.py` around lines 90 - 91, The current except clause in backend/app/services/providers/suunto/oauth.py is too broad (it catches Exception), so change it to catch only the expected exceptions: JWTError, JWSError, and requests.RequestException (e.g. "except (JWTError, JWSError, requests.RequestException) as e:"), remove the generic Exception from the tuple, and handle the error by logging the exception (using the existing logger or logger.exception) before returning {"user_id": None, "username": None} to avoid swallowing unexpected errors.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@backend/app/services/providers/suunto/oauth.py`:
- Around line 40-51: The HS256 branch in oauth.py currently decodes
token_response.access_token with key=None and options={"verify_signature":
False} (the jwt.decode call) which accepts forged tokens; remove that unverified
HS256 path or change it to verify using the shared secret (use the configured
client_secret when algorithm == "HS256") by passing the secret as the key and
not disabling signature verification, and validate claims as you do for RS256;
also replace the bare "except Exception: pass" with explicit error handling (log
the exception and return None or re-raise) so failures in the jwt.decode/token
parsing are not silently swallowed.
---
Nitpick comments:
In `@backend/app/services/providers/suunto/oauth.py`:
- Around line 90-91: The current except clause in
backend/app/services/providers/suunto/oauth.py is too broad (it catches
Exception), so change it to catch only the expected exceptions: JWTError,
JWSError, and requests.RequestException (e.g. "except (JWTError, JWSError,
requests.RequestException) as e:"), remove the generic Exception from the tuple,
and handle the error by logging the exception (using the existing logger or
logger.exception) before returning {"user_id": None, "username": None} to avoid
swallowing unexpected errors.
| if algorithm == "HS256": | ||
| decoded = jwt.decode( | ||
| token_response.access_token, | ||
| key=None, | ||
| options={"verify_signature": False}, | ||
| ) | ||
| provider_username = decoded.get("user") | ||
| provider_user_id = decoded.get("sub") | ||
| if provider_user_id is not None: | ||
| return {"user_id": provider_user_id, "username": provider_username} | ||
| except Exception: | ||
| pass |
There was a problem hiding this comment.
Critical: HS256 branch bypasses signature verification entirely.
This code path decodes HS256 tokens without any signature verification (key=None, verify_signature: False). An attacker can craft a malicious JWT with alg: HS256 and arbitrary claims—the code will accept it. This completely defeats the security fix this PR intends to provide.
Either:
- Remove this unverified HS256 path entirely, or
- Verify HS256 tokens using the appropriate shared secret (client_secret)
Additionally, the bare except Exception: pass at lines 50-51 silently swallows all errors, masking potential issues.
🔒 Option 1: Remove the insecure HS256 path (recommended if Suunto only issues RS256 tokens)
`@handle_exceptions`
def _get_provider_user_info(self, token_response: OAuthTokenResponse, user_id: str) -> dict[str, str | None]:
"""Extracts and verifies Suunto user info from JWT access token."""
- try:
- unverified_header = jwt.get_unverified_header(token_response.access_token)
- algorithm = unverified_header.get("alg")
-
- if algorithm == "HS256":
- decoded = jwt.decode(
- token_response.access_token,
- key=None,
- options={"verify_signature": False},
- )
- provider_username = decoded.get("user")
- provider_user_id = decoded.get("sub")
- if provider_user_id is not None:
- return {"user_id": provider_user_id, "username": provider_username}
- except Exception:
- pass
-
try:
# Get the key ID from token header🔒 Option 2: Properly verify HS256 with the client secret
try:
unverified_header = jwt.get_unverified_header(token_response.access_token)
algorithm = unverified_header.get("alg")
if algorithm == "HS256":
decoded = jwt.decode(
token_response.access_token,
- key=None,
- options={"verify_signature": False},
+ key=self.credentials.client_secret,
+ algorithms=["HS256"],
+ audience=settings.suunto_client_id,
)
provider_username = decoded.get("user")
provider_user_id = decoded.get("sub")
if provider_user_id is not None:
return {"user_id": provider_user_id, "username": provider_username}
- except Exception:
- pass
+ except (JWTError, JWSError):
+ pass # Fall through to RS256 verification🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@backend/app/services/providers/suunto/oauth.py` around lines 40 - 51, The
HS256 branch in oauth.py currently decodes token_response.access_token with
key=None and options={"verify_signature": False} (the jwt.decode call) which
accepts forged tokens; remove that unverified HS256 path or change it to verify
using the shared secret (use the configured client_secret when algorithm ==
"HS256") by passing the secret as the key and not disabling signature
verification, and validate claims as you do for RS256; also replace the bare
"except Exception: pass" with explicit error handling (log the exception and
return None or re-raise) so failures in the jwt.decode/token parsing are not
silently swallowed.
KaliszS
left a comment
KaliszS
left a comment
There was a problem hiding this comment.
We need to fix code quality issues. You can check them locally by running uv run pre-commit run --all-files from backend directory.
Also you have asked for algorithm used in jwkts by Suunto. They didn't provide this information in docs, but we can always decode JWT access token and thats HS256
Thank you, I will fix the code quality issue and decode using HS256. Additionally, the test |
KaliszS
left a comment
KaliszS
left a comment
There was a problem hiding this comment.
Would be great for tests to match real flow, but we don't want to use real secrets from .env within tests in my opinion.
There are also a lot of errors in code-quality Ci step. They need to be fixed after the review changes will be implemented.
| unverified_header = jwt.get_unverified_header(token_response.access_token) | ||
| algorithm = unverified_header.get("alg") | ||
|
|
||
| if algorithm == "HS256": |
There was a problem hiding this comment.
As code rabbit say - HS256 is the only option. So there should be just one path to solve the issue.
There was a problem hiding this comment.
Maybe we can just add something like if algorithm =! "HS256" and then raise exception "Not implemented" so we will know whether Suunto will change their algorithm.
| ) | ||
| provider_username = decoded.get("user") | ||
| provider_user_id = decoded.get("sub") | ||
| if provider_user_id is not None: |
There was a problem hiding this comment.
I guess just if provider_user_id will be enough. If that would be empty string, 0 or [], that also should raise exception. So it's safe to do that this way.
| key="", # Empty key since we're not verifying | ||
| options={"verify_signature": False}, | ||
| matching_key, | ||
| algorithms=["RS256"], |
There was a problem hiding this comment.
That RS256 usage is redundant, please simplify it to one simple workflow.
| if self.credentials.subscription_key: | ||
| headers["Ocp-Apim-Subscription-Key"] = self.credentials.subscription_key | ||
|
|
||
| resp = requests.get(jwks_url, headers=headers, timeout=5) |
There was a problem hiding this comment.
Maybe we should implement some caching strategy to avoid making this request every time user is logging? lru_cache should work fine.
There was a problem hiding this comment.
Nevermind, we should probably drop that support anyway. HS256 is used by Suunto.
| if algorithm == "HS256": | ||
| decoded = jwt.decode( | ||
| token_response.access_token, | ||
| key=None, |
There was a problem hiding this comment.
Correct me if I'm wrong, but HS256 is symmtric algorithm. So we can just probaly use here client_secret? Or subscription key, that should be tested manually.
2 participants
Closes #511
Description
Fixes a security vulnerability in the Suunto OAuth integration where JWT tokens were being accepted without signature verification.
Checklist
General
Backend Changes
You have to be in
backenddirectory to make it work:uv run pre-commit run --all-filespassesFrontend Changes
pnpm run lintpassespnpm run format:checkpassespnpm run buildsucceedsTesting Instructions
Steps to test:
Expected behavior:
Screenshots
Additional Notes
Summary by CodeRabbit