feat: send WAF bypass header when env var is set

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=<token>
+
+When this variable is set, PyFxA automatically includes the ``fxa-ci: <token>``
+header on every outgoing request. When it is absent, no such header is sent.
+
+
 Passing tokens to other applications
 ===================================================
 

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

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):