feat(totp): move TOTP management to the /mfa/totp/* routes

CHANGES.txt

@@ -3,6 +3,15 @@ CHANGELOG
 
 This document describes changes between each past release.
 
+0.9.0 (unreleased)
+==================
+
+- Move TOTP management to the ``/mfa/totp/*`` routes. ``/totp/destroy`` was
+  removed server-side in January 2026 and ``/totp/create`` is being retired.
+  ``Session.totp_create`` and ``totp_delete`` now take an MFA token from the
+  new ``mfa_request_otp`` / ``mfa_verify_otp`` helpers, and setup finishes
+  with ``totp_setup_verify`` and ``totp_setup_complete``.
+
 0.8.2 (2026-05-14)
 ==================
 

fxa/core.py

@@ -9,6 +9,7 @@
 from fxa.errors import ClientError
 from fxa._utils import (
     APIClient,
+    BearerTokenAuth,
     FxATokenBearerAuth,
     exactly_one_of,
     hexstr
@@ -500,18 +501,64 @@ def resend_email_code(self, **kwds):
         url = "/recovery_email/resend_code"
         self.apiclient.post(url, body, auth=self._auth)
 
-    def totp_create(self):
-        url = "/totp/create"
-        return self.apiclient.post(url, {}, auth=self._auth)
+    def mfa_request_otp(self, action):
+        """Ask the server to email a one-time code for a sensitive ``action``.
+
+        The code arrives in the ``X-Account-Change-Verify-Code`` header of the
+        email and is exchanged for a short-lived MFA token with
+        :meth:`mfa_verify_otp`. Requires a verified session.
+        """
+        url = "/mfa/otp/request"
+        return self.apiclient.post(url, {"action": action}, auth=self._auth)
+
+    def mfa_verify_otp(self, code, action):
+        """Exchange an emailed one-time code for an MFA token.
+
+        The returned token is a JWT scoped to ``action`` and is what the
+        ``/mfa/*`` routes accept in place of the session token.
+        """
+        url = "/mfa/otp/verify"
+        body = {
+            "code": code,
+            "action": action,
+        }
+        resp = self.apiclient.post(url, body, auth=self._auth)
+        return resp["accessToken"]
+
+    def totp_create(self, mfa_token):
+        """Start TOTP setup and return the shared secret and QR code URL.
+
+        ``mfa_token`` comes from :meth:`mfa_verify_otp` with the ``"2fa"``
+        action. Nothing is stored on the account until
+        :meth:`totp_setup_complete` succeeds.
+        """
+        url = "/mfa/totp/create"
+        return self.apiclient.post(url, {}, auth=BearerTokenAuth(mfa_token))
+
+    def totp_setup_verify(self, mfa_token, code):
+        """Prove possession of the pending TOTP secret with a current code."""
+        url = "/mfa/totp/setup/verify"
+        body = {
+            "code": code,
+        }
+        resp = self.apiclient.post(url, body, auth=BearerTokenAuth(mfa_token))
+        return resp["success"]
+
+    def totp_setup_complete(self, mfa_token):
+        """Enable TOTP on the account once :meth:`totp_setup_verify` passed."""
+        url = "/mfa/totp/setup/complete"
+        resp = self.apiclient.post(url, {}, auth=BearerTokenAuth(mfa_token))
+        return resp["success"]
 
     def totp_exists(self):
         url = "/totp/exists"
         resp = self.apiclient.get(url, auth=self._auth)
         return resp["exists"]
 
-    def totp_delete(self):
-        url = "/totp/destroy"
-        return self.apiclient.post(url, {}, auth=self._auth)
+    def totp_delete(self, mfa_token):
+        """Remove TOTP from the account. ``mfa_token`` needs the ``"2fa"`` action."""
+        url = "/mfa/totp/destroy"
+        return self.apiclient.post(url, {}, auth=BearerTokenAuth(mfa_token))
 
     def totp_verify(self, code):
         url = "/session/verify/totp"

fxa/tests/test_core.py

@@ -372,31 +372,37 @@ def has_verify_code(m):
         self.assertEqual(self.session.keys[1], keys[1])
 
     def test_totp(self):
-        resp = self.session.totp_create()
+        # TOTP setup is guarded by a short-lived MFA token, obtained by
+        # verifying a code the server emails to the account.
+        self.session.mfa_request_otp("2fa")
+        m = self.acct.wait_for_email(
+            lambda m: "x-account-change-verify-code" in m["headers"])
+        if not m:
+            raise RuntimeError("MFA code email was not received")
+        self.acct.clear()
+        mfa_token = self.session.mfa_verify_otp(
+            m["headers"]["x-account-change-verify-code"], "2fa")
 
-        # Should exist even if not verified
-        self.assertTrue(self.session.totp_exists())
+        resp = self.session.totp_create(mfa_token)
 
-        # Creating again should work unless verified
-        resp = self.session.totp_create()
+        # Nothing is stored on the account until setup completes.
+        self.assertFalse(self.session.totp_exists())
 
-        # Set session unverified to test next call
-        self.session.verified = False
+        # Creating again re-issues the same pending secret.
+        resp2 = self.session.totp_create(mfa_token)
+        self.assertEqual(resp2["secret"], resp["secret"])
 
-        # Verify the code
         code = pyotp.TOTP(resp["secret"]).now()
-        self.assertTrue(self.session.totp_verify(code))
-        self.assertTrue(self.session.verified)
-
-        # Should exist
+        self.assertTrue(self.session.totp_setup_verify(mfa_token, code))
+        self.assertTrue(self.session.totp_setup_complete(mfa_token))
         self.assertTrue(self.session.totp_exists())
 
-        # Double create causes a client error
+        # Creating again once TOTP is enabled is a client error.
         with self.assertRaises(fxa.errors.ClientError):
-            self.session.totp_create()
+            self.session.totp_create(mfa_token)
 
         # Remove the code
-        resp = self.session.totp_delete()
+        self.session.totp_delete(mfa_token)
 
         # And now should not exist
         self.assertFalse(self.session.totp_exists())
@@ -456,6 +462,19 @@ def test_password_forgot_token_call_site_sends_fxpf_bearer(self):
         authz = responses.calls[0].request.headers["Authorization"]
         self.assertRegex(authz, r"^Bearer fxpf_[0-9a-f]{64}$")
 
+    @responses.activate
+    def test_totp_setup_sends_plain_bearer_mfa_token(self):
+        responses.add(responses.POST, self.server_url + "/mfa/totp/create",
+                      json={"secret": "s", "qrCodeUrl": "data:"},
+                      content_type="application/json")
+        session = Session(
+            client=self.client, email="test@example.com",
+            stretchpwd=b"\x00" * 32, uid="abc123", token="1234",
+        )
+        session.totp_create("eyJ.mfa.jwt")
+        authz = responses.calls[0].request.headers["Authorization"]
+        self.assertEqual(authz, "Bearer eyJ.mfa.jwt")
+
 
 # helpers
 def verify_account(acct, client):