Use PKCE for the Trakt OAuth sign-in

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
 

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."""

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"]