diff --git a/src/integrations/imports/trakt.py b/src/integrations/imports/trakt.py index ed2793645..eac49949b 100644 --- a/src/integrations/imports/trakt.py +++ b/src/integrations/imports/trakt.py @@ -1,6 +1,9 @@ +import base64 +import hashlib import json import logging import re +import secrets import zipfile from collections import defaultdict @@ -49,7 +52,29 @@ } -def handle_oauth_callback(request, redirect_uri=None): +def generate_pkce_pair(): + """Return a PKCE ``(code_verifier, code_challenge)`` pair for one sign-in. + + Trakt only supports the S256 method: the challenge is the base64url-encoded + SHA-256 hash of the verifier, without padding. + """ + code_verifier = secrets.token_urlsafe(64) + digest = hashlib.sha256(code_verifier.encode("ascii")).digest() + code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode("ascii") + return code_verifier, code_challenge + + +def add_client_secret(params): + """Add the client secret to token request params when one is configured. + + Trakt apps created for PKCE are not issued a secret, and Trakt asks not to + send one with them. + """ + if settings.TRAKT_API_SECRET: + params["client_secret"] = settings.TRAKT_API_SECRET + + +def handle_oauth_callback(request, redirect_uri=None, code_verifier=None): """View for getting the Trakt OAuth2 token.""" code = request.GET["code"] @@ -61,11 +86,13 @@ def handle_oauth_callback(request, redirect_uri=None): params = { "client_id": settings.TRAKT_API, - "client_secret": settings.TRAKT_API_SECRET, "code": code, "grant_type": "authorization_code", "redirect_uri": redirect_uri, } + add_client_secret(params) + if code_verifier: + params["code_verifier"] = code_verifier try: token_response = app.providers.services.api_request( @@ -125,10 +152,10 @@ def get_access_token(encrypted_refresh_token, redirect_uri=None): params = { "client_id": settings.TRAKT_API, - "client_secret": settings.TRAKT_API_SECRET, "refresh_token": decrypted_token, "grant_type": "refresh_token", } + add_client_secret(params) if redirect_uri: params["redirect_uri"] = redirect_uri diff --git a/src/integrations/tests/test_oauth_views.py b/src/integrations/tests/test_oauth_views.py index 457496c9c..a293703d3 100644 --- a/src/integrations/tests/test_oauth_views.py +++ b/src/integrations/tests/test_oauth_views.py @@ -1,11 +1,15 @@ +import base64 +import hashlib from unittest.mock import patch from urllib.parse import parse_qs, urlparse from django.contrib.auth import get_user_model from django.core.files.uploadedfile import SimpleUploadedFile -from django.test import TestCase, override_settings +from django.test import RequestFactory, TestCase, override_settings from django.urls import reverse +from integrations.imports import trakt + class TraktOAuthViewTests(TestCase): """Test Trakt OAuth redirect URL handling.""" @@ -28,7 +32,7 @@ def test_trakt_oauth_uses_configured_public_url(self): redirect = urlparse(response["Location"]) query = parse_qs(redirect.query) self.assertEqual(redirect.scheme, "https") - self.assertEqual(redirect.netloc, "trakt.tv") + self.assertEqual(redirect.netloc, "auth.trakt.tv") self.assertEqual(query["client_id"], ["client"]) self.assertEqual( query["redirect_uri"], @@ -92,6 +96,135 @@ def test_trakt_callback_rejects_missing_state(self, mock_oauth_callback): self.assertRedirects(response, reverse("import_data")) mock_oauth_callback.assert_not_called() + @override_settings(TRAKT_API="client") + def test_trakt_oauth_sends_pkce_challenge(self): + """The authorization request carries an S256 challenge for a stored verifier.""" + response = self.client.post( + reverse("trakt_oauth"), + {"mode": "new", "frequency": "once", "time": "14:30"}, + ) + + query = parse_qs(urlparse(response["Location"]).query) + self.assertEqual(query["code_challenge_method"], ["S256"]) + verifier = self.client.session[query["state"][0]]["code_verifier"] + # RFC 7636: 43 to 128 characters + self.assertGreaterEqual(len(verifier), 43) + self.assertLessEqual(len(verifier), 128) + expected = ( + base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()) + .rstrip(b"=") + .decode() + ) + self.assertEqual(query["code_challenge"], [expected]) + + def test_trakt_oauth_uses_a_new_verifier_each_time(self): + """Every sign-in gets its own PKCE verifier.""" + verifiers = set() + for _ in range(2): + response = self.client.post( + reverse("trakt_oauth"), + {"mode": "new", "frequency": "once", "time": "14:30"}, + ) + state_token = parse_qs(urlparse(response["Location"]).query)["state"][0] + verifiers.add(self.client.session[state_token]["code_verifier"]) + + self.assertEqual(len(verifiers), 2) + + @patch("integrations.views.tasks.import_trakt.delay") + @patch("integrations.views.trakt.handle_oauth_callback") + def test_trakt_callback_passes_stored_code_verifier( + self, + mock_oauth_callback, + _mock_import_trakt, + ): + """The token exchange receives the verifier stored when sign-in started.""" + session = self.client.session + session["state-token"] = { + "mode": "new", + "frequency": "once", + "time": "14:30", + "redirect_uri": "https://yamtrack.example.com/import/trakt/private", + "code_verifier": "stored-verifier", + } + session.save() + mock_oauth_callback.return_value = { + "refresh_token": "refresh-token", + "username": "trakt-user", + } + + self.client.get( + reverse("import_trakt_private"), + {"code": "code", "state": "state-token"}, + ) + + self.assertEqual( + mock_oauth_callback.call_args.kwargs["code_verifier"], + "stored-verifier", + ) + + +class TraktTokenExchangeTests(TestCase): + """Test the parameters sent to Trakt's token endpoint.""" + + def exchange(self, code_verifier): + """Run the token exchange and return the params sent to Trakt.""" + request = RequestFactory().get("/import/trakt/private", {"code": "code"}) + with ( + patch( + "integrations.imports.trakt.app.providers.services.api_request", + return_value={"refresh_token": "refresh", "access_token": "access"}, + ) as mock_request, + patch( + "integrations.imports.trakt.get_username_from_oauth", + return_value="trakt-user", + ), + ): + result = trakt.handle_oauth_callback( + request, + redirect_uri="https://yamtrack.example.com/import/trakt/private", + code_verifier=code_verifier, + ) + self.assertEqual(result["username"], "trakt-user") + return mock_request.call_args.kwargs["params"] + + def test_sends_code_verifier(self): + """The PKCE verifier is sent with the authorization code.""" + params = self.exchange("stored-verifier") + self.assertEqual(params["code_verifier"], "stored-verifier") + self.assertEqual(params["grant_type"], "authorization_code") + + @override_settings(TRAKT_API_SECRET="") + def test_omits_client_secret_when_not_configured(self): + """PKCE apps have no client secret, so none is sent.""" + params = self.exchange("stored-verifier") + self.assertNotIn("client_secret", params) + + @override_settings(TRAKT_API_SECRET="secret") # noqa: S106 + def test_sends_configured_client_secret(self): + """Apps that were issued a secret keep sending it.""" + params = self.exchange("stored-verifier") + self.assertEqual(params["client_secret"], "secret") + + @override_settings(TRAKT_API_SECRET="") + def test_refresh_omits_client_secret_when_not_configured(self): + """Scheduled imports renew their token without a secret for PKCE apps.""" + with ( + patch("integrations.imports.trakt.helpers.decrypt", return_value="r"), + patch( + "integrations.imports.trakt.app.providers.services.api_request", + return_value={"refresh_token": "new", "access_token": "access"}, + ) as mock_request, + ): + trakt.get_access_token("encrypted", redirect_uri="https://x/cb") + params = mock_request.call_args.kwargs["params"] + self.assertNotIn("client_secret", params) + self.assertEqual(params["grant_type"], "refresh_token") + + def test_omits_code_verifier_without_one(self): + """A sign-in started before PKCE was added still exchanges its code.""" + params = self.exchange(None) + self.assertNotIn("code_verifier", params) + class TraktExportUploadViewTests(TestCase): """Test the Trakt export archive upload view.""" diff --git a/src/integrations/views.py b/src/integrations/views.py index d571c6f72..6294ab81e 100644 --- a/src/integrations/views.py +++ b/src/integrations/views.py @@ -33,13 +33,15 @@ def trakt_oauth(request): request, reverse("import_trakt_private"), ) - url = "https://trakt.tv/oauth/authorize" + url = "https://auth.trakt.tv/oauth/authorize" state = { "mode": request.POST["mode"], "frequency": request.POST["frequency"], "time": request.POST["time"], "redirect_uri": redirect_uri, } + code_verifier, code_challenge = trakt.generate_pkce_pair() + state["code_verifier"] = code_verifier state_token = secrets.token_urlsafe(32) request.session[state_token] = state return redirect( @@ -50,6 +52,8 @@ def trakt_oauth(request): 'redirect_uri': redirect_uri, 'response_type': 'code', 'state': state_token, + 'code_challenge': code_challenge, + 'code_challenge_method': 'S256', } ) }", @@ -73,7 +77,11 @@ def import_trakt_private(request): request, reverse("import_trakt_private"), ) - oauth_callback = trakt.handle_oauth_callback(request, redirect_uri=redirect_uri) + oauth_callback = trakt.handle_oauth_callback( + request, + redirect_uri=redirect_uri, + code_verifier=state.get("code_verifier"), + ) enc_token = helpers.encrypt(oauth_callback["refresh_token"]) frequency = state["frequency"]