diff --git a/README.rst b/README.rst index de2df03..d6640b5 100644 --- a/README.rst +++ b/README.rst @@ -89,6 +89,19 @@ testing with live email addresses. It works like this: client.destroy_account(acct.email, "MySecretPassword") +CI WAF bypass +============= + +When running CI tests against a Mozilla Accounts environment protected by +WAF, requests may be subject to challenge rules. +To bypass them, set the ``CI_WAF_TOKEN`` environment variable:: + + export CI_WAF_TOKEN= + +When this variable is set, PyFxA automatically includes the ``fxa-ci: `` +header on every outgoing request. When it is absent, no such header is sent. + + Passing tokens to other applications =================================================== diff --git a/fxa/_utils.py b/fxa/_utils.py index e333ce3..7daeace 100644 --- a/fxa/_utils.py +++ b/fxa/_utils.py @@ -147,6 +147,7 @@ class APIClient: * backoff protocol support * sensible request timeouts * timestamp skew tracking with automatic retry on clockskew error + * CI WAF bypass header injection """ @@ -160,6 +161,9 @@ def __init__(self, server_url, session=None): allowed_methods={"DELETE", "GET", "POST", "PUT"}, ) session.mount(server_url, HTTPAdapter(max_retries=retries)) + waf_token = os.environ.get("CI_WAF_TOKEN") + if waf_token: + session.headers["fxa-ci"] = waf_token # Properties that can be customized to change behaviour. self.server_url = server_url self.timeout = 30 diff --git a/fxa/tests/test_core.py b/fxa/tests/test_core.py index f2cc92f..979b98c 100644 --- a/fxa/tests/test_core.py +++ b/fxa/tests/test_core.py @@ -1,16 +1,19 @@ # 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 os import time from urllib.parse import urlparse import pyotp import pytest +import requests from parameterized import parameterized_class import fxa.errors from fxa.core import Client, StretchedPassword +from fxa._utils import APIClient from fxa.tests.utils import ( unittest, @@ -35,6 +38,8 @@ class TestCoreClient(unittest.TestCase): server_url = TEST_SERVER_URL def setUp(self): + if not os.environ.get("FXA_RUN_LIVE_TESTS"): + self.skipTest("Set FXA_RUN_LIVE_TESTS=1 to run live tests against the stage server") self.client_v1 = Client(self.server_url) self.client_v2 = Client(self.server_url, key_stretch_version=2) if self.key_stretch_version == 2: @@ -282,7 +287,8 @@ class TestCoreClientSession(unittest.TestCase): server_url = TEST_SERVER_URL def setUp(self): - + if not os.environ.get("FXA_RUN_LIVE_TESTS"): + self.skipTest("Set FXA_RUN_LIVE_TESTS=1 to run live tests against the stage server") self.client_v2 = Client(self.server_url, key_stretch_version=2) self.client_v1 = Client(self.server_url, key_stretch_version=1) if self.key_stretch_version == 2: @@ -395,6 +401,29 @@ def test_totp(self): self.assertFalse(self.session.totp_exists()) +class TestAPIClientWAFHeader(unittest.TestCase): + """Unit tests for CI_WAF_TOKEN header injection in APIClient.""" + + SERVER_URL = "https://api.example.com/v1/" + + def test_waf_header_set_when_env_var_present(self): + with unittest.mock.patch.dict("os.environ", {"CI_WAF_TOKEN": "sekrit"}): + client = APIClient(self.SERVER_URL) + self.assertEqual(client.headers.get("fxa-ci"), "sekrit") + + def test_waf_header_absent_when_env_var_not_set(self): + env = {k: v for k, v in os.environ.items() if k != "CI_WAF_TOKEN"} + with unittest.mock.patch.dict("os.environ", env, clear=True): + client = APIClient(self.SERVER_URL) + self.assertNotIn("fxa-ci", client.headers) + + def test_waf_header_set_on_caller_supplied_session(self): + supplied = requests.Session() + with unittest.mock.patch.dict("os.environ", {"CI_WAF_TOKEN": "sekrit"}): + APIClient(self.SERVER_URL, session=supplied) + self.assertEqual(supplied.headers.get("fxa-ci"), "sekrit") + + # helpers def verify_account(acct, client): def wait_for_email(m):