From 3faeb7d108010eb5686ea194d3ac835d9cdf7a5a Mon Sep 17 00:00:00 2001 From: Wil Clouser Date: Fri, 25 Sep 2026 15:23:48 -0700 Subject: [PATCH] fix(password): move password reset to the /password/forgot OTP routes Because: - The auth-server removed /password/forgot/{send_code,resend_code,status} in January 2026 (mozilla/fxa 985f565dc9), so Client.send_reset_code and the PasswordForgotToken flow have returned 404 since then. - Only the opt-in stage live test exercised the flow, so it went unnoticed. This commit: - Point send_reset_code at /password/forgot/send_otp and add Client.verify_reset_otp for /password/forgot/verify_otp. - Make PasswordForgotToken.verify_code chain verify_otp then verify_code and populate token, uid and email_to_hash_with on success. - Drop resend_reset_code, get_reset_code_status and get_status, which have no OTP equivalent; resend_code now re-sends the OTP. - Update the live test and add mocked tests for the request shapes. - Add a CHANGES entry. Fixes FXA-14639 --- CHANGES.txt | 12 +++++ fxa/core.py | 93 +++++++++++++++++------------------- fxa/tests/test_core.py | 106 +++++++++++++++++++++++++++++++++++------ 3 files changed, 147 insertions(+), 64 deletions(-) diff --git a/CHANGES.txt b/CHANGES.txt index 0119ea0..568010a 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -3,6 +3,18 @@ CHANGELOG This document describes changes between each past release. +0.9.0 (unreleased) +================== + +- Move password reset to the ``/password/forgot/send_otp`` and + ``/password/forgot/verify_otp`` routes. The ``send_code``, ``resend_code`` + and ``status`` routes were removed server-side in January 2026 and had been + returning 404. ``Client.send_reset_code`` now takes only ``service`` and + returns a ``PasswordForgotToken`` whose ``token`` and ``uid`` are set by + ``verify_code``. ``PasswordForgotToken.get_status``, + ``Client.resend_reset_code`` and ``Client.get_reset_code_status`` are + removed; ``Client.verify_reset_otp`` is new. + 0.8.2 (2026-05-14) ================== diff --git a/fxa/core.py b/fxa/core.py index 8ca3dd3..c3bad68 100644 --- a/fxa/core.py +++ b/fxa/core.py @@ -320,39 +320,35 @@ def reset_account(self, email, token, password=None, stretchpwd=None): auth = FxATokenBearerAuth(token, "accountResetToken", self.apiclient) self.apiclient.post(url, body, auth=auth) - def send_reset_code(self, email, **kwds): + def send_reset_code(self, email, service=None): + """Email a one-time code that starts a password reset for ``email``. + + The server issues no token until the code is verified, so the + returned :class:`PasswordForgotToken` is only usable through + :meth:`PasswordForgotToken.verify_code`. + """ body = { "email": email, } - for extra in kwds: - if extra in ("service", "redirectTo", "resume"): - body[extra] = kwds[extra] - else: - msg = f"Unexpected keyword argument: {extra}" - raise TypeError(msg) - url = "/password/forgot/send_code" - resp = self.apiclient.post(url, body) - return PasswordForgotToken( - self, email, - resp["passwordForgotToken"], - resp["ttl"], - resp["codeLength"], - resp["tries"], - ) - - def resend_reset_code(self, email, token, **kwds): + if service is not None: + body["service"] = service + url = "/password/forgot/send_otp" + self.apiclient.post(url, body) + return PasswordForgotToken(self, email, service=service) + + def verify_reset_otp(self, email, code): + """Exchange the emailed one-time code for a ``passwordForgotToken``. + + Returns the raw response: ``token`` (the passwordForgotToken), the + server-issued ``code`` that :meth:`verify_reset_code` expects, + ``uid`` and ``emailToHashWith``. + """ body = { "email": email, + "code": code, } - for extra in kwds: - if extra in ("service", "redirectTo", "resume"): - body[extra] = kwds[extra] - else: - msg = f"Unexpected keyword argument: {extra}" - raise TypeError(msg) - url = "/password/forgot/resend_code" - auth = FxATokenBearerAuth(token, "passwordForgotToken", self.apiclient) - return self.apiclient.post(url, body, auth=auth) + url = "/password/forgot/verify_otp" + return self.apiclient.post(url, body) def verify_reset_code(self, token, code): body = { @@ -362,11 +358,6 @@ def verify_reset_code(self, token, code): auth = FxATokenBearerAuth(token, "passwordForgotToken", self.apiclient) return self.apiclient.post(url, body, auth=auth) - def get_reset_code_status(self, token): - url = "/password/forgot/status" - auth = FxATokenBearerAuth(token, "passwordForgotToken", self.apiclient) - return self.apiclient.get(url, auth=auth) - def verify_email_code(self, uid, code): body = { "uid": uid, @@ -541,31 +532,33 @@ def get_random_bytes(self): class PasswordForgotToken: + """A password reset in progress, started by :meth:`Client.send_reset_code`. + + The auth server emails an 8-digit one-time code. Pass it to + :meth:`verify_code` to obtain the ``accountResetToken`` that + :meth:`Client.reset_account` needs. ``token``, ``uid`` and + ``email_to_hash_with`` are populated once the code has been verified. + """ - def __init__(self, client, email, token, ttl=0, code_length=16, - tries_remaining=1): + def __init__(self, client, email, service=None): self.client = client self.email = email - self.token = token - self.ttl = ttl - self.code_length = code_length - self.tries_remaining = tries_remaining + self.service = service + self.token = None + self.uid = None + self.email_to_hash_with = None def verify_code(self, code): - resp = self.client.verify_reset_code(self.token, code) + otp = self.client.verify_reset_otp(self.email, code) + self.token = otp["token"] + self.uid = otp["uid"] + self.email_to_hash_with = otp["emailToHashWith"] + resp = self.client.verify_reset_code(self.token, otp["code"]) return resp["accountResetToken"] - def resend_code(self, **kwds): - resp = self.client.resend_reset_code(self.email, self.token, **kwds) - self.ttl = resp["ttl"] - self.code_length = resp["codeLength"] - self.tries_remaining = resp["tries"] - - def get_status(self): - resp = self.client.get_reset_code_status(self.token) - self.ttl = resp["ttl"] - self.tries_remaining = resp["tries"] - return resp + def resend_code(self): + """Email a fresh one-time code.""" + self.client.send_reset_code(self.email, service=self.service) class StretchedPassword: diff --git a/fxa/tests/test_core.py b/fxa/tests/test_core.py index e829618..a517f7f 100644 --- a/fxa/tests/test_core.py +++ b/fxa/tests/test_core.py @@ -1,6 +1,7 @@ # This Source Code Form is subject to the terms of the Mozilla Public # License, v. 2.0. If a copy of the MPL was not distributed with this file, # You can obtain one at http://mozilla.org/MPL/2.0/. +import json import os import time @@ -13,7 +14,7 @@ from parameterized import parameterized_class import fxa.errors -from fxa.core import Client, Session, StretchedPassword +from fxa.core import Client, PasswordForgotToken, Session, StretchedPassword from fxa._utils import APIClient from fxa.tests.utils import ( @@ -147,31 +148,30 @@ def test_forgot_password_flow(self): ) self.add_account_to_delete(acct, session) - # Initiate the password reset flow, and grab the verification code. + # Initiate the password reset flow, and grab the one-time code. pftok = self.client.send_reset_code(acct.email, service="foobar") - m = acct.wait_for_email(lambda m: "x-recovery-code" in m["headers"]) + m = acct.wait_for_email(lambda m: "x-password-forgot-otp" in m["headers"]) if not m: raise RuntimeError("Password reset email was not received") acct.clear() - code = m["headers"]["x-recovery-code"] + code = m["headers"]["x-password-forgot-otp"] # Try with an invalid code to test error handling. - tries = pftok.tries_remaining - self.assertTrue(tries > 1) - with self.assertRaises(Exception): + with self.assertRaises(fxa.errors.ClientError): pftok.verify_code(mutate_one_byte(code)) - pftok.get_status() - self.assertEqual(pftok.tries_remaining, tries - 1) + self.assertIsNone(pftok.token) # Re-send the code, as if we've lost the email. pftok.resend_code() - m = acct.wait_for_email(lambda m: "x-recovery-code" in m["headers"]) + m = acct.wait_for_email(lambda m: "x-password-forgot-otp" in m["headers"]) if not m: raise RuntimeError("Password reset email was not received") - self.assertEqual(m["headers"]["x-recovery-code"], code) + code = m["headers"]["x-password-forgot-otp"] # Now verify with the actual code, and reset the account. artok = pftok.verify_code(code) + self.assertIsNotNone(pftok.token) + self.assertEqual(pftok.uid, session.uid) self.client.reset_account( email=acct.email, token=artok, @@ -450,13 +450,91 @@ def test_session_token_call_site_sends_fxs_bearer(self): @responses.activate def test_password_forgot_token_call_site_sends_fxpf_bearer(self): - responses.add(responses.GET, self.server_url + "/password/forgot/status", - json={}, content_type="application/json") - self.client.get_reset_code_status("1234") + responses.add(responses.POST, self.server_url + "/password/forgot/verify_code", + json={"accountResetToken": "ab" * 32}, + content_type="application/json") + self.client.verify_reset_code("1234", "deadbeef") authz = responses.calls[0].request.headers["Authorization"] self.assertRegex(authz, r"^Bearer fxpf_[0-9a-f]{64}$") +class TestCorePasswordReset(unittest.TestCase): + """Mocked coverage of the OTP password-reset flow and its request shapes.""" + + server_url = "https://server/v1" + + def setUp(self): + self.client = Client(self.server_url) + + @responses.activate + def test_send_reset_code_posts_otp_request(self): + responses.add(responses.POST, self.server_url + "/password/forgot/send_otp", + json={}, content_type="application/json") + pftok = self.client.send_reset_code("test@example.com", service="sync") + body = json.loads(responses.calls[0].request.body) + self.assertEqual(body, {"email": "test@example.com", "service": "sync"}) + self.assertNotIn("Authorization", responses.calls[0].request.headers) + self.assertEqual(pftok.email, "test@example.com") + self.assertEqual(pftok.service, "sync") + self.assertIsNone(pftok.token) + + @responses.activate + def test_send_reset_code_omits_service_when_unset(self): + responses.add(responses.POST, self.server_url + "/password/forgot/send_otp", + json={}, content_type="application/json") + self.client.send_reset_code("test@example.com") + body = json.loads(responses.calls[0].request.body) + self.assertEqual(body, {"email": "test@example.com"}) + + @responses.activate + def test_verify_code_chains_otp_and_code_verification(self): + responses.add(responses.POST, self.server_url + "/password/forgot/verify_otp", + json={ + "code": "c0de" * 8, + "token": "12" * 32, + "uid": "abc123", + "emailToHashWith": "primary@example.com", + }, content_type="application/json") + responses.add(responses.POST, self.server_url + "/password/forgot/verify_code", + json={"accountResetToken": "ab" * 32}, + content_type="application/json") + pftok = PasswordForgotToken(self.client, "test@example.com") + + artok = pftok.verify_code("12345678") + + self.assertEqual(artok, "ab" * 32) + otp_req, code_req = responses.calls[0].request, responses.calls[1].request + self.assertEqual(json.loads(otp_req.body), + {"email": "test@example.com", "code": "12345678"}) + self.assertNotIn("Authorization", otp_req.headers) + self.assertEqual(json.loads(code_req.body), {"code": "c0de" * 8}) + self.assertRegex(code_req.headers["Authorization"], r"^Bearer fxpf_[0-9a-f]{64}$") + self.assertEqual(pftok.token, "12" * 32) + self.assertEqual(pftok.uid, "abc123") + self.assertEqual(pftok.email_to_hash_with, "primary@example.com") + + @responses.activate + def test_invalid_otp_leaves_token_unset(self): + responses.add(responses.POST, self.server_url + "/password/forgot/verify_otp", + json={"code": 400, "errno": 105, "error": "Bad Request", + "message": "Invalid verification code"}, + status=400, content_type="application/json") + pftok = PasswordForgotToken(self.client, "test@example.com") + with self.assertRaises(fxa.errors.ClientError): + pftok.verify_code("00000000") + self.assertIsNone(pftok.token) + self.assertEqual(len(responses.calls), 1) + + @responses.activate + def test_resend_code_posts_otp_request_again(self): + responses.add(responses.POST, self.server_url + "/password/forgot/send_otp", + json={}, content_type="application/json") + pftok = PasswordForgotToken(self.client, "test@example.com", service="sync") + pftok.resend_code() + body = json.loads(responses.calls[0].request.body) + self.assertEqual(body, {"email": "test@example.com", "service": "sync"}) + + # helpers def verify_account(acct, client): def wait_for_email(m):