From 62e559b4eddf90a0be01c4a8ca927eba50eda784 Mon Sep 17 00:00:00 2001
From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com>
Date: Thu, 13 Aug 2026 12:04:35 -0400
Subject: [PATCH 1/9] update GitHub Actions to Node 24 versions (#974)

---
 .github/workflows/ci.yml      | 18 +++++++++---------
 .github/workflows/nightly.yml | 10 +++++-----
 .github/workflows/release.yml | 10 +++++-----
 3 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 32ef50ae..d4f02efe 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -14,12 +14,12 @@ jobs:
         python: ['3.9', '3.10', '3.11', '3.12', '3.13', 'pypy3.9']
 
     steps:
-    - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         persist-credentials: false
 
     - name: Set up Python 3.13 for emulator
-      uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: '3.13'
     - name: Setup functions emulator environment
@@ -29,7 +29,7 @@ jobs:
         pip install -r integration/emulators/functions/requirements.txt
         deactivate
     - name: Set up Python ${{ matrix.python }}
-      uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: ${{ matrix.python }}
     - name: Install dependencies
@@ -38,12 +38,12 @@ jobs:
         pip install -r requirements.txt
     - name: Test with pytest
       run: pytest
-    - name: Set up Node.js 20
-      uses: actions/setup-node@48b55a011bda9f5d6aeb4c2d9c7362e8dae4041e # v6.4.0
+    - name: Set up Node.js 24
+      uses: actions/setup-node@820762786026740c76f36085b0efc47a31fe5020 # v7.0.0
       with:
-        node-version: 20
+        node-version: 24
     - name: Set up Java 21
-      uses: actions/setup-java@0f481fcb613427c0f801b606911222b5b6f3083a # v5.5.0
+      uses: actions/setup-java@b6effb05e454b25005698d916606bdc6ffcbf961 # v5.7.0
       with:
         distribution: 'temurin'
         java-version: '21'
@@ -57,11 +57,11 @@ jobs:
   lint:
     runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+    - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         persist-credentials: false
     - name: Set up Python 3.9
-      uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: 3.9
     - name: Install dependencies
diff --git a/.github/workflows/nightly.yml b/.github/workflows/nightly.yml
index 57240b23..adfc47fd 100644
--- a/.github/workflows/nightly.yml
+++ b/.github/workflows/nightly.yml
@@ -32,13 +32,13 @@ jobs:
 
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         ref: ${{ github.event.client_payload.ref || github.ref }}
         persist-credentials: false
 
     - name: Set up Python
-      uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: 3.9
 
@@ -67,14 +67,14 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: dist
         path: dist
 
     - name: Send email on failure
       if: failure()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
+      uses: firebase/firebase-admin-node/.github/actions/send-email@7343f166ec71c2efee5fe0554c2fb96203faf2a5 # Admin Node SDK v14.2.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
@@ -89,7 +89,7 @@ jobs:
 
     - name: Send email on cancelled
       if: cancelled()
-      uses: firebase/firebase-admin-node/.github/actions/send-email@2e2b36a84ba28679bcb7aecdacabfec0bded2d48 # Admin Node SDK v13.6.0
+      uses: firebase/firebase-admin-node/.github/actions/send-email@7343f166ec71c2efee5fe0554c2fb96203faf2a5 # Admin Node SDK v14.2.0
       with:
         api-key: ${{ secrets.OSS_BOT_MAILGUN_KEY }}
         domain: ${{ secrets.OSS_BOT_MAILGUN_DOMAIN }}
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 510a1352..34f1d848 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -49,12 +49,12 @@ jobs:
 
     steps:
     - name: Checkout source for staging
-      uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         persist-credentials: false
 
     - name: Set up Python
-      uses: actions/setup-python@ece7cb06caefa5fff74198d8649806c4678c61a1 # v6.3.0
+      uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
       with:
         python-version: 3.9
 
@@ -83,7 +83,7 @@ jobs:
     # Attach the packaged artifacts to the workflow output. These can be manually
     # downloaded for later inspection if necessary.
     - name: Archive artifacts
-      uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4.6.2
+      uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
       with:
         name: dist
         path: dist
@@ -110,14 +110,14 @@ jobs:
 
     steps:
     - name: Checkout source for publish
-      uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7.0.0
+      uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
       with:
         # Ignored since the GITHUB_TOKEN needs to be persisted for this workflow.
         persist-credentials: true # zizmor: ignore[artipacked]
 
     # Download the artifacts created by the stage_release job.
     - name: Download release candidates
-      uses: actions/download-artifact@d3f86a106a0bac45b974a628896c90dbdf5c8093 # v4.3.0
+      uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
       with:
         name: dist
         path: dist

From eba8fa00c02930504b093853a056c24480ca5c48 Mon Sep 17 00:00:00 2001
From: Jonathan Edey <145066863+jonathanedey@users.noreply.github.com>
Date: Tue, 1 Sep 2026 10:46:43 -0400
Subject: [PATCH 2/9] chore(ml): Skip flaky ML tests blocking nightly builds
 (#979)

---
 integration/test_ml.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/integration/test_ml.py b/integration/test_ml.py
index ea5b10be..7050ea63 100644
--- a/integration/test_ml.py
+++ b/integration/test_ml.py
@@ -26,6 +26,7 @@
 from firebase_admin import ml
 from tests import testutils
 
+pytestmark = pytest.mark.skip(reason='ML integration tests are skipped.')
 
 # pylint: disable=import-error, no-member
 try:

From a7aafa2d1b3d5020928a39d00333e1d228d30f6d Mon Sep 17 00:00:00 2001
From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com>
Date: Tue, 1 Sep 2026 18:24:16 -0400
Subject: [PATCH 3/9] feat(ml): deprecate Machine Learning APIs (#981)

* add ml api deprecation warnings

* Verify deprecation warnings in unit tests

* add helper function to prevent duplicate warning flooding and unit tests for it
---
 firebase_admin/ml.py | 144 +++++++++++++++++++++++++++++++++++++++++--
 tests/test_ml.py     |  87 ++++++++++++++++++++++++++
 2 files changed, 227 insertions(+), 4 deletions(-)

diff --git a/firebase_admin/ml.py b/firebase_admin/ml.py
index 3a77dd05..2257383b 100644
--- a/firebase_admin/ml.py
+++ b/firebase_admin/ml.py
@@ -14,16 +14,25 @@
 
 """Firebase ML module.
 
+.. deprecated::
+    Firebase ML is deprecated and will be shut down on June 15, 2027.
+    To host custom models, you must migrate to another solution. You can use Cloud Storage for
+    Firebase as an alternative for hosting custom models. For more info, see
+    https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
 This module contains functions for creating, updating, getting, listing,
 deleting, publishing and unpublishing Firebase ML models.
 """
 
+# pylint: disable=too-many-lines
 
 import datetime
+import os
 import re
+import sys
 import time
-import os
 from urllib import parse
+import warnings
 
 import requests
 
@@ -46,6 +55,30 @@
 except ImportError:
     _TF_ENABLED = False
 
+_DEPRECATION_WARNING = (
+    'Firebase ML is deprecated and will be shut down on June 15, 2027. '
+    'To host custom models, you must migrate to another solution. You can use Cloud Storage for '
+    'Firebase as an alternative for hosting custom models. For more info, see '
+    'https://firebase.google.com/docs/ml/migrate-to-cloud-storage'
+)
+
+
+def _is_internal_call():
+    """Checks if the current call originated from within the ML module."""
+    try:
+        # pylint: disable=protected-access
+        frame = sys._getframe(2)
+        while frame:
+            if frame.f_code.co_name in (
+                'from_dict', '_update_from_dict', 'handle_operation', 'list_models'
+            ):
+                return True
+            frame = frame.f_back
+    except (AttributeError, ValueError):
+        pass
+    return False
+
+
 _ML_ATTRIBUTE = '_ml'
 _MAX_PAGE_SIZE = 100
 _MODEL_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]{1,60}$')
@@ -77,6 +110,12 @@ def _get_ml_service(app):
 def create_model(model, app=None):
     """Creates a model in the current Firebase project.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model: An ml.Model to create.
         app: A Firebase app instance (or None to use the default app).
@@ -84,6 +123,8 @@ def create_model(model, app=None):
     Returns:
         Model: The model that was created in Firebase ML.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return Model.from_dict(ml_service.create_model(model), app=app)
 
@@ -91,6 +132,12 @@ def create_model(model, app=None):
 def update_model(model, app=None):
     """Updates a model's metadata or model file.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model: The ml.Model to update.
         app: A Firebase app instance (or None to use the default app).
@@ -98,6 +145,8 @@ def update_model(model, app=None):
     Returns:
         Model: The updated model.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return Model.from_dict(ml_service.update_model(model), app=app)
 
@@ -105,6 +154,12 @@ def update_model(model, app=None):
 def publish_model(model_id, app=None):
     """Publishes a Firebase ML model.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     A published model can be downloaded to client apps.
 
     Args:
@@ -114,6 +169,8 @@ def publish_model(model_id, app=None):
     Returns:
         Model: The published model.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return Model.from_dict(ml_service.set_published(model_id, publish=True), app=app)
 
@@ -121,6 +178,12 @@ def publish_model(model_id, app=None):
 def unpublish_model(model_id, app=None):
     """Unpublishes a Firebase ML model.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model_id: The id of the model to unpublish.
         app: A Firebase app instance (or None to use the default app).
@@ -128,6 +191,8 @@ def unpublish_model(model_id, app=None):
     Returns:
         Model: The unpublished model.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return Model.from_dict(ml_service.set_published(model_id, publish=False), app=app)
 
@@ -135,6 +200,12 @@ def unpublish_model(model_id, app=None):
 def get_model(model_id, app=None):
     """Gets the model specified by the given ID.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model_id: The id of the model to get.
         app: A Firebase app instance (or None to use the default app).
@@ -142,6 +213,8 @@ def get_model(model_id, app=None):
     Returns:
      Model: The requested model.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return Model.from_dict(ml_service.get_model(model_id), app=app)
 
@@ -149,6 +222,12 @@ def get_model(model_id, app=None):
 def list_models(list_filter=None, page_size=None, page_token=None, app=None):
     """Lists the current project's models.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         list_filter: a list filter string such as ``tags:'tag_1'``. None will return all models.
         page_size: A number between 1 and 100 inclusive that specifies the maximum
@@ -160,6 +239,8 @@ def list_models(list_filter=None, page_size=None, page_token=None, app=None):
     Returns:
         ListModelsPage: A (filtered) list of models.
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     return ListModelsPage(
         ml_service.list_models, list_filter, page_size, page_token, app=app)
@@ -168,10 +249,18 @@ def list_models(list_filter=None, page_size=None, page_token=None, app=None):
 def delete_model(model_id, app=None):
     """Deletes a model from the current project.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model_id: The id of the model you wish to delete.
         app: A Firebase app instance (or None to use the default app).
     """
+    if not _is_internal_call():
+        warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
     ml_service = _get_ml_service(app)
     ml_service.delete_model(model_id)
 
@@ -179,12 +268,20 @@ def delete_model(model_id, app=None):
 class Model:
     """A Firebase ML Model object.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         display_name: The display name of your model - used to identify your model in code.
         tags: Optional list of strings associated with your model. Can be used in list queries.
         model_format: A subclass of ModelFormat. (e.g. TFLiteFormat) Specifies the model details.
     """
     def __init__(self, display_name=None, tags=None, model_format=None):
+        if not _is_internal_call():
+            warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
         self._app = None  # Only needed for wait_for_unlo
         self._data = {}
         self._model_format = None
@@ -342,7 +439,14 @@ def as_dict(self, for_upload=False):
 
 
 class ModelFormat:
-    """Abstract base class representing a Model Format such as TFLite."""
+    """Abstract base class representing a Model Format such as TFLite.
+
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+    """
     def as_dict(self, for_upload=False):
         """Returns a serializable representation of the object."""
         raise NotImplementedError
@@ -351,10 +455,18 @@ def as_dict(self, for_upload=False):
 class TFLiteFormat(ModelFormat):
     """Model format representing a TFLite model.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Args:
         model_source: A TFLiteModelSource sub class. Specifies the details of the model source.
     """
     def __init__(self, model_source=None):
+        if not _is_internal_call():
+            warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
         self._data = {}
         self._model_source = None
 
@@ -412,7 +524,14 @@ def as_dict(self, for_upload=False):
 
 
 class TFLiteModelSource:
-    """Abstract base class representing a model source for TFLite format models."""
+    """Abstract base class representing a model source for TFLite format models.
+
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+    """
     def as_dict(self, for_upload=False):
         """Returns a serializable representation of the object."""
         raise NotImplementedError
@@ -466,11 +585,20 @@ def sign_uri(gcs_tflite_uri, app):
 
 
 class TFLiteGCSModelSource(TFLiteModelSource):
-    """TFLite model source representing a tflite model file stored in GCS."""
+    """TFLite model source representing a tflite model file stored in GCS.
+
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+    """
 
     _STORAGE_CLIENT = _CloudStorageClient()
 
     def __init__(self, gcs_tflite_uri, app=None):
+        if not _is_internal_call():
+            warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
         self._app = app
         self._gcs_tflite_uri = _validate_gcs_tflite_uri(gcs_tflite_uri)
 
@@ -600,12 +728,20 @@ def as_dict(self, for_upload=False):
 class ListModelsPage:
     """Represents a page of models in a Firebase project.
 
+    .. deprecated::
+        Firebase ML is deprecated and will be shut down on June 15, 2027.
+        To host custom models, you must migrate to another solution. You can use Cloud Storage for
+        Firebase as an alternative for hosting custom models. For more info, see
+        https://firebase.google.com/docs/ml/migrate-to-cloud-storage
+
     Provides methods for traversing the models included in this page, as well as
     retrieving subsequent pages of models. The iterator returned by
     ``iterate_all()`` can be used to iterate through all the models in the
     Firebase project starting from this page.
     """
     def __init__(self, list_models_func, list_filter, page_size, page_token, app):
+        if not _is_internal_call():
+            warnings.warn(_DEPRECATION_WARNING, DeprecationWarning, stacklevel=2)
         self._list_models_func = list_models_func
         self._list_filter = list_filter
         self._page_size = page_size
diff --git a/tests/test_ml.py b/tests/test_ml.py
index bcc93fd0..af97b174 100644
--- a/tests/test_ml.py
+++ b/tests/test_ml.py
@@ -15,6 +15,7 @@
 """Test cases for the firebase_admin.ml module."""
 
 import json
+import warnings
 
 import pytest
 
@@ -1086,3 +1087,89 @@ def test_list_models_no_models(self):
         assert len(page.models) == 0
         models = list(page.iterate_all())
         assert len(models) == 0
+
+
+class TestMLDeprecation:
+    """Tests for verifying deprecation warnings in the ML module."""
+
+    DEPRECATION_MSG = (
+        r'Firebase ML is deprecated and will be shut down on June 15, 2027\. '
+        r'To host custom models, you must migrate to another solution\. '
+        r'You can use Cloud Storage for Firebase as an alternative for hosting custom models\. '
+        r'For more info, see https://firebase\.google\.com/docs/ml/migrate-to-cloud-storage'
+    )
+
+    @classmethod
+    def setup_class(cls):
+        cred = testutils.MockCredential()
+        firebase_admin.initialize_app(cred, {'projectId': PROJECT_ID})
+        ml._MLService.POLL_BASE_WAIT_TIME_SECONDS = 0.1
+        ml.TFLiteGCSModelSource._STORAGE_CLIENT = _TestStorageClient()
+
+    @classmethod
+    def teardown_class(cls):
+        testutils.cleanup_apps()
+
+    def test_model_init_deprecation_warning(self):
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.Model()
+
+    def test_tflite_format_init_deprecation_warning(self):
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.TFLiteFormat()
+
+    def test_tflite_gcs_model_source_init_deprecation_warning(self):
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.TFLiteGCSModelSource(GCS_TFLITE_URI)
+
+    def test_get_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=DEFAULT_GET_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.get_model(MODEL_ID_1)
+
+    def test_list_models_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=ONE_PAGE_LIST_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.list_models()
+
+    def test_delete_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=EMPTY_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.delete_model(MODEL_ID_1)
+
+    def test_create_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=OPERATION_DONE_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.create_model(ml.Model(display_name=DISPLAY_NAME_1))
+
+    def test_update_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=OPERATION_DONE_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.update_model(ml.Model(display_name=DISPLAY_NAME_1))
+
+    def test_publish_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=OPERATION_DONE_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.publish_model(MODEL_ID_1)
+
+    def test_unpublish_model_deprecation_warning(self):
+        instrument_ml_service(status=200, payload=OPERATION_DONE_RESPONSE)
+        with pytest.warns(DeprecationWarning, match=self.DEPRECATION_MSG):
+            ml.unpublish_model(MODEL_ID_1)
+
+    def test_single_warning_on_get_model(self):
+        instrument_ml_service(status=200, payload=DEFAULT_GET_RESPONSE)
+        with warnings.catch_warnings(record=True) as captured_warnings:
+            warnings.simplefilter('always')
+            ml.get_model(MODEL_ID_1)
+            assert len(captured_warnings) == 1
+            assert issubclass(captured_warnings[0].category, DeprecationWarning)
+
+    def test_single_warning_on_list_models(self):
+        instrument_ml_service(status=200, payload=ONE_PAGE_LIST_RESPONSE)
+        with warnings.catch_warnings(record=True) as captured_warnings:
+            warnings.simplefilter('always')
+            page = ml.list_models()
+            _ = page.models
+            assert len(captured_warnings) == 1
+            assert issubclass(captured_warnings[0].category, DeprecationWarning)

From 2ef3414361289c0b1ca937e4248df9b6d4adfdfe Mon Sep 17 00:00:00 2001
From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com>
Date: Wed, 16 Sep 2026 17:19:15 -0400
Subject: [PATCH 4/9] feat(appcheck): Verify one-time tokens for replay
 protection (#976)

* add one time token verification support

* add unit tests

* fix lint error

* change to use _utils.handle_platform_error_from_requests

* change the function description

* change docstring to state limited-use tokens

* add validation for consume variable

* add check_boolean to _Validators and add unit tests to check non-boolean consume

* raise `exceptions.UnknownError` if body is not a dict

* move the replay verification to a separate helper method and add unit tests

* add test_verify_token_with_consume_false_makes_no_http_call unit test

* replace the broad exceptions.FirebaseError assertion with more specific error types

* Refactor token verification tests to use mock_jwt fixture

* Document ValueError when consume is not a boolean in verify_token docstring
---
 firebase_admin/app_check.py |  62 +++++++++++++---
 tests/test_app_check.py     | 141 ++++++++++++++++++++++++++++++++++--
 2 files changed, 189 insertions(+), 14 deletions(-)

diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py
index 40d857f4..1117e151 100644
--- a/firebase_admin/app_check.py
+++ b/firebase_admin/app_check.py
@@ -15,38 +15,47 @@
 """Firebase App Check module."""
 
 from typing import Any, Dict
+import requests
 import jwt
 from jwt import PyJWKClient, ExpiredSignatureError, InvalidTokenError, DecodeError
 from jwt import InvalidAudienceError, InvalidIssuerError, InvalidSignatureError
-from firebase_admin import _utils
+from firebase_admin import _http_client, _utils, exceptions
 
 _APP_CHECK_ATTRIBUTE = '_app_check'
 
 def _get_app_check_service(app) -> Any:
     return _utils.get_app_service(app, _APP_CHECK_ATTRIBUTE, _AppCheckService)
 
-def verify_token(token: str, app=None) -> Dict[str, Any]:
-    """Verifies a Firebase App Check token.
+def verify_token(token: str, app=None, consume: bool = False) -> Dict[str, Any]:
+    """Verifies a Firebase App Check token, optionally consuming limited-use tokens.
 
     Args:
         token: A token from App Check.
         app: An App instance (optional).
+        consume: Set to ``True`` only if the token is a limited-use (one-time) token
+            that should be consumed upon verification (optional, defaults to ``False``).
 
     Returns:
-        Dict[str, Any]: The token's decoded claims.
+        Dict[str, Any]: The token's decoded claims. If ``consume`` is ``True``, the dictionary
+            also includes an ``already_consumed`` boolean key indicating whether the token was
+            previously consumed.
 
     Raises:
-        ValueError: If the app's ``project_id`` is invalid or unspecified,
-        or if the token's headers or payload are invalid.
+        ValueError: If ``consume`` is not a boolean, or if the app's ``project_id``
+            is invalid or unspecified, or if the token's headers or payload are invalid.
+        FirebaseError: If an error occurs while communicating with the App Check service.
         PyJWKClientError: If PyJWKClient fails to fetch a valid signing key.
     """
-    return _get_app_check_service(app).verify_token(token)
+    return _get_app_check_service(app).verify_token(token, consume=consume)
 
 class _AppCheckService:
     """Service class that implements Firebase App Check functionality."""
 
     _APP_CHECK_ISSUER = 'https://firebaseappcheck.googleapis.com/'
     _JWKS_URL = 'https://firebaseappcheck.googleapis.com/v1/jwks'
+    _VERIFY_URL_FORMAT = (
+        'https://firebaseappcheck.googleapis.com/v1/projects/{project_id}:verifyAppCheckToken'
+    )
     _project_id = None
     _scoped_project_id = None
     _jwks_client = None
@@ -68,11 +77,15 @@ def __init__(self, app):
         # Default lifespan is 300 seconds (5 minutes) so we change it to 21600 seconds (6 hours).
         self._jwks_client = PyJWKClient(
             self._JWKS_URL, lifespan=21600, headers=self._APP_CHECK_HEADERS)
+        timeout = app.options.get('httpTimeout', _http_client.DEFAULT_TIMEOUT_SECONDS)
+        self._http_client = _http_client.JsonHttpClient(
+            credential=app.credential.get_credential(), timeout=timeout)
 
 
-    def verify_token(self, token: str) -> Dict[str, Any]:
-        """Verifies a Firebase App Check token."""
+    def verify_token(self, token: str, consume: bool = False) -> Dict[str, Any]:
+        """Verifies a Firebase App Check token, optionally consuming limited-use tokens."""
         _Validators.check_string("app check token", token)
+        _Validators.check_boolean("consume", consume)
 
         # Obtain the Firebase App Check Public Keys
         # Note: It is not recommended to hard code these keys as they rotate,
@@ -87,8 +100,32 @@ def verify_token(self, token: str) -> Dict[str, Any]:
                 ) from exception
 
         verified_claims['app_id'] = verified_claims.get('sub')
+
+        if consume:
+            verified_claims['already_consumed'] = self._verify_replay_protection(token)
+
         return verified_claims
 
+    def _verify_replay_protection(self, token: str) -> bool:
+        """Verifies replay protection with the backend and returns the alreadyConsumed status."""
+        url = self._VERIFY_URL_FORMAT.format(project_id=self._project_id)
+        try:
+            body = self._http_client.body('post', url, json={'app_check_token': token})
+        except requests.exceptions.RequestException as error:
+            raise _utils.handle_platform_error_from_requests(error)
+        except ValueError as error:
+            raise exceptions.UnknownError(
+                f'Unexpected response from App Check service: {error}'
+            ) from error
+
+        if not isinstance(body, dict):
+            raise exceptions.UnknownError(
+                'Unexpected response from App Check service. '
+                f'Expected a JSON object, but got {type(body).__name__}.'
+            )
+
+        return bool(body.get('alreadyConsumed', False))
+
     def _has_valid_token_headers(self, headers: Any) -> None:
         """Checks whether the token has valid headers for App Check."""
         # Ensure the token's header has type JWT
@@ -159,3 +196,10 @@ def check_string(cls, label: str, value: Any):
             raise ValueError(f'{label} "{value}" must be a non-empty string.')
         if not isinstance(value, str):
             raise ValueError(f'{label} "{value}" must be a string.')
+
+    @classmethod
+    def check_boolean(cls, label: str, value: Any):
+        """Checks if the given value is a boolean."""
+        if not isinstance(value, bool):
+            raise ValueError(f'{label} must be a boolean.')
+        return value
diff --git a/tests/test_app_check.py b/tests/test_app_check.py
index e55ae39d..b7a7bcd3 100644
--- a/tests/test_app_check.py
+++ b/tests/test_app_check.py
@@ -15,11 +15,12 @@
 """Test cases for the firebase_admin.app_check module."""
 import base64
 import pytest
+import requests
 
 from jwt import PyJWK, InvalidAudienceError, InvalidIssuerError
 from jwt import ExpiredSignatureError, InvalidSignatureError
 import firebase_admin
-from firebase_admin import app_check
+from firebase_admin import app_check, exceptions
 from tests import testutils
 
 NON_STRING_ARGS = [[], tuple(), {}, True, False, 1, 0]
@@ -58,6 +59,15 @@ def setup_class(cls):
     def teardown_class(cls):
         testutils.cleanup_apps()
 
+
+@pytest.fixture
+def mock_jwt(mocker):
+    """Mocks JWT decoding and JWKS key retrieval for a valid App Check token."""
+    mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE)
+    mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key))
+    mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers"))
+
+
 class TestVerifyToken(TestBatch):
 
     def test_no_project_id(self):
@@ -74,6 +84,12 @@ def test_verify_token_with_non_string_raises_error(self, token):
         expected = f'app check token "{token}" must be a string.'
         assert str(excinfo.value) == expected
 
+    @pytest.mark.parametrize('consume', [[], tuple(), {}, 1, 0, 'true', 'false', None])
+    def test_verify_token_with_non_boolean_consume_raises_error(self, consume):
+        with pytest.raises(ValueError) as excinfo:
+            app_check.verify_token('app_check_token', consume=consume)
+        assert str(excinfo.value) == 'consume must be a boolean.'
+
     def test_has_valid_token_headers(self):
         app = firebase_admin.get_app()
         app_check_service = app_check._get_app_check_service(app)
@@ -221,10 +237,8 @@ def test_decode_and_verify_with_non_string_sub_raises_error(self, mocker):
             f'"{sub_number}" must be a string.')
         assert str(excinfo.value) == expected
 
-    def test_verify_token(self, mocker):
-        mocker.patch("jwt.decode", return_value=JWT_PAYLOAD_SAMPLE)
-        mocker.patch("jwt.PyJWKClient.get_signing_key_from_jwt", return_value=PyJWK(signing_key))
-        mocker.patch("jwt.get_unverified_header", return_value=JWT_PAYLOAD_SAMPLE.get("headers"))
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token(self):
         app = firebase_admin.get_app()
 
         payload = app_check.verify_token("encoded", app)
@@ -273,3 +287,120 @@ def test_verify_token_with_incorrect_issuer_raises_error(self, mocker):
 
         expected = 'Token does not contain the correct "iss" (issuer).'
         assert str(excinfo.value) == expected
+
+    @pytest.mark.usefixtures("mock_jwt")
+    @pytest.mark.parametrize('consume_kwargs', [{}, {'consume': False}])
+    def test_verify_token_with_consume_false_makes_no_http_call(
+        self, mocker, consume_kwargs
+    ):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+        mock_body = mocker.patch.object(app_check_service._http_client, "body")
+
+        payload = app_check.verify_token("encoded", app=app, **consume_kwargs)
+        expected = JWT_PAYLOAD_SAMPLE.copy()
+        expected["app_id"] = APP_ID
+        assert payload == expected
+        assert 'already_consumed' not in payload
+        mock_body.assert_not_called()
+
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token_with_consume_true_not_consumed(self, mocker):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+        mock_body = mocker.patch.object(
+            app_check_service._http_client, "body", return_value={"alreadyConsumed": False}
+        )
+
+        payload = app_check.verify_token("encoded", app=app, consume=True)
+        expected = JWT_PAYLOAD_SAMPLE.copy()
+        expected["app_id"] = APP_ID
+        expected["already_consumed"] = False
+        assert payload == expected
+
+        expected_url = (
+            f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken"
+        )
+        mock_body.assert_called_once_with(
+            "post", expected_url, json={"app_check_token": "encoded"}
+        )
+
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token_with_consume_true_already_consumed(self, mocker):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+        mock_body = mocker.patch.object(
+            app_check_service._http_client, "body", return_value={"alreadyConsumed": True}
+        )
+
+        payload = app_check.verify_token("encoded", app=app, consume=True)
+        expected = JWT_PAYLOAD_SAMPLE.copy()
+        expected["app_id"] = APP_ID
+        expected["already_consumed"] = True
+        assert payload == expected
+
+        expected_url = (
+            f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken"
+        )
+        mock_body.assert_called_once_with(
+            "post", expected_url, json={"app_check_token": "encoded"}
+        )
+
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token_with_consume_true_backend_error(self, mocker):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+
+        req_exc = requests.exceptions.RequestException("Backend error")
+        mocker.patch.object(app_check_service._http_client, "body", side_effect=req_exc)
+
+        with pytest.raises(exceptions.UnknownError) as excinfo:
+            app_check.verify_token("encoded", app=app, consume=True)
+        assert "Unknown error while making a remote service call" in str(excinfo.value)
+
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token_with_consume_true_http_error(self, mocker):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+
+        response = requests.Response()
+        response.status_code = 403
+        response._content = (
+            b'{"error": {"status": "PERMISSION_DENIED", "message": "Permission denied."}}'
+        )
+        http_exc = requests.exceptions.HTTPError(response=response)
+        mocker.patch.object(app_check_service._http_client, "body", side_effect=http_exc)
+
+        with pytest.raises(exceptions.PermissionDeniedError) as excinfo:
+            app_check.verify_token("encoded", app=app, consume=True)
+        assert "Permission denied." in str(excinfo.value)
+
+    @pytest.mark.usefixtures("mock_jwt")
+    @pytest.mark.parametrize('malformed_body', ['string_response', [1, 2], 123, None])
+    def test_verify_token_with_consume_true_malformed_response_raises_error(
+        self, mocker, malformed_body
+    ):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+        mocker.patch.object(app_check_service._http_client, "body", return_value=malformed_body)
+
+        with pytest.raises(exceptions.UnknownError) as excinfo:
+            app_check.verify_token("encoded", app=app, consume=True)
+        assert 'Unexpected response from App Check service' in str(excinfo.value)
+
+    @pytest.mark.usefixtures("mock_jwt")
+    def test_verify_token_with_consume_true_json_decode_error(self, mocker):
+        app = firebase_admin.get_app()
+        app_check_service = app_check._get_app_check_service(app)
+        mocker.patch.object(
+            app_check_service._http_client,
+            "body",
+            side_effect=ValueError("Expecting value: line 1 column 1 (char 0)"),
+        )
+
+        with pytest.raises(exceptions.UnknownError) as excinfo:
+            app_check.verify_token("encoded", app=app, consume=True)
+        expected_msg = (
+            "Unexpected response from App Check service: Expecting value: line 1 column 1 (char 0)"
+        )
+        assert expected_msg in str(excinfo.value)

From cbacb022095e0249811ba3361f64f32f800f3b40 Mon Sep 17 00:00:00 2001
From: Yvonne Pan <103622026+yvonnep165@users.noreply.github.com>
Date: Thu, 17 Sep 2026 15:57:20 -0400
Subject: [PATCH 5/9] [chore] Release 7.6.0 (#985)

---
 firebase_admin/__about__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/firebase_admin/__about__.py b/firebase_admin/__about__.py
index 096dc67b..5b706a61 100644
--- a/firebase_admin/__about__.py
+++ b/firebase_admin/__about__.py
@@ -14,7 +14,7 @@
 
 """About information (version, etc) for Firebase Admin SDK."""
 
-__version__ = '7.5.0'
+__version__ = '7.6.0'
 __title__ = 'firebase_admin'
 __author__ = 'Firebase'
 __license__ = 'Apache License 2.0'

From 46edf94b04f23338ca71961ba2baa0969de61ce2 Mon Sep 17 00:00:00 2001
From: Lahiru Maramba <llahiru@gmail.com>
Date: Wed, 23 Sep 2026 11:35:47 -0400
Subject: [PATCH 6/9] feat(fcm): Migrate topic management to FCM v1 API (#980)

* feat(fcm): Migrate topic management to FCM v1 API

* fix(fcm): Prevent headers mutation race condition and deduplicate topic error parsing

- Pass a copy of self._fcm_headers in topic management requests to prevent concurrent mutation race conditions in ThreadPoolExecutor.
- Extract common error parsing logic from _build_topic_subscription_result_from_requests_error and _build_topic_subscription_result_from_httpx_error into _build_topic_subscription_result.

* fix(fcm): Improve topic management performance and fix test lint issues

- Hoist URL-encoded topic computation out of the per-token request loops.
- Mount an HTTPAdapter with a connection pool size of 100 on the FCM client session.
- Fix line length and method signature override lint warnings in test_messaging.py.

* fix(fcm): Addressed review improvements for FCM v1 topic management

- Added type annotations to public topic management functions and internal methods
- Updated HTTP status code mapping (401->UNAUTHENTICATED, 503->UNAVAILABLE, 408/504->DEADLINE_EXCEEDED) and prioritized status codes over free-form message strings
- Updated topic regex with \Z to reject trailing newlines
- Passed header copies in send, send_each, and send_each_async to prevent concurrency mutations
- Added test coverage for status code mapping, async argument validation, prefixed topics, and async batch unsubscribe

* chore(fcm): Trigger CI workflows
---
 firebase_admin/messaging.py | 347 ++++++++++++++++++++-
 tests/test_messaging.py     | 582 ++++++++++++++++++++++++++++++++++--
 2 files changed, 903 insertions(+), 26 deletions(-)

diff --git a/firebase_admin/messaging.py b/firebase_admin/messaging.py
index 93ecfda1..a108444a 100644
--- a/firebase_admin/messaging.py
+++ b/firebase_admin/messaging.py
@@ -15,14 +15,17 @@
 """Firebase Cloud Messaging module."""
 
 from __future__ import annotations
-from typing import Any, Callable, Dict, List, Optional, cast
+import asyncio
 import concurrent.futures
 import json
-import asyncio
 import logging
+import re
+from typing import Any, Callable, Dict, List, Optional, Union, cast
+import urllib.parse
 import warnings
-import requests
+
 import httpx
+import requests
 
 import firebase_admin
 from firebase_admin import (
@@ -73,7 +76,11 @@
     'send_each_for_multicast',
     'send_each_for_multicast_async',
     'subscribe_to_topic',
+    'subscribe_to_topic_async',
+    'subscribe_to_topic_legacy',
     'unsubscribe_from_topic',
+    'unsubscribe_from_topic_async',
+    'unsubscribe_from_topic_legacy',
 ]
 
 
@@ -252,9 +259,53 @@ def send_each_for_multicast(multicast_message, dry_run=False, app=None):
     messages = _get_messages_from_multicast(multicast_message)
     return _get_messaging_service(app).send_each(messages, dry_run)
 
-def subscribe_to_topic(tokens, topic, app=None):
+def subscribe_to_topic(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
     """Subscribes a list of registration tokens to an FCM topic.
 
+    Args:
+        tokens: A non-empty list of device registration tokens. List may not have more than 1000
+            elements.
+        topic: Name of the topic to subscribe to. May contain the ``/topics/`` prefix.
+        app: An App instance (optional).
+
+    Returns:
+        TopicManagementResponse: A ``TopicManagementResponse`` instance.
+
+    Raises:
+        FirebaseError: If an error occurs while communicating with the FCM service.
+        ValueError: If the input arguments are invalid.
+    """
+    return _get_messaging_service(app).subscribe_to_topic(tokens, topic)
+
+async def subscribe_to_topic_async(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
+    """Subscribes a list of registration tokens to an FCM topic asynchronously.
+
+    Args:
+        tokens: A non-empty list of device registration tokens. List may not have more than 1000
+            elements.
+        topic: Name of the topic to subscribe to. May contain the ``/topics/`` prefix.
+        app: An App instance (optional).
+
+    Returns:
+        TopicManagementResponse: A ``TopicManagementResponse`` instance.
+
+    Raises:
+        FirebaseError: If an error occurs while communicating with the FCM service.
+        ValueError: If the input arguments are invalid.
+    """
+    return await _get_messaging_service(app).subscribe_to_topic_async(tokens, topic)
+
+def subscribe_to_topic_legacy(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
+    """Subscribes a list of registration tokens to an FCM topic using the legacy Instance ID API.
+
+    subscribe_to_topic_legacy is deprecated. Use subscribe_to_topic instead.
+
     Args:
         tokens: A non-empty list of device registration tokens. List may not have more than 1000
             elements.
@@ -268,12 +319,61 @@ def subscribe_to_topic(tokens, topic, app=None):
         FirebaseError: If an error occurs while communicating with instance ID service.
         ValueError: If the input arguments are invalid.
     """
+    warnings.warn(
+        'subscribe_to_topic_legacy is deprecated. Use subscribe_to_topic instead.',
+        DeprecationWarning,
+        stacklevel=2)
     return _get_messaging_service(app).make_topic_management_request(
         tokens, topic, 'iid/v1:batchAdd')
 
-def unsubscribe_from_topic(tokens, topic, app=None):
+def unsubscribe_from_topic(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
     """Unsubscribes a list of registration tokens from an FCM topic.
 
+    Args:
+        tokens: A non-empty list of device registration tokens. List may not have more than 1000
+            elements.
+        topic: Name of the topic to unsubscribe from. May contain the ``/topics/`` prefix.
+        app: An App instance (optional).
+
+    Returns:
+        TopicManagementResponse: A ``TopicManagementResponse`` instance.
+
+    Raises:
+        FirebaseError: If an error occurs while communicating with the FCM service.
+        ValueError: If the input arguments are invalid.
+    """
+    return _get_messaging_service(app).unsubscribe_from_topic(tokens, topic)
+
+async def unsubscribe_from_topic_async(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
+    """Unsubscribes a list of registration tokens from an FCM topic asynchronously.
+
+    Args:
+        tokens: A non-empty list of device registration tokens. List may not have more than 1000
+            elements.
+        topic: Name of the topic to unsubscribe from. May contain the ``/topics/`` prefix.
+        app: An App instance (optional).
+
+    Returns:
+        TopicManagementResponse: A ``TopicManagementResponse`` instance.
+
+    Raises:
+        FirebaseError: If an error occurs while communicating with the FCM service.
+        ValueError: If the input arguments are invalid.
+    """
+    return await _get_messaging_service(app).unsubscribe_from_topic_async(tokens, topic)
+
+def unsubscribe_from_topic_legacy(
+    tokens: Union[str, List[str]], topic: str, app: Optional[App] = None
+) -> TopicManagementResponse:
+    """Unsubscribes a list of registration tokens from an FCM topic using the legacy
+    Instance ID API.
+
+    unsubscribe_from_topic_legacy is deprecated. Use unsubscribe_from_topic instead.
+
     Args:
         tokens: A non-empty list of device registration tokens. List may not have more than 1000
             elements.
@@ -287,6 +387,10 @@ def unsubscribe_from_topic(tokens, topic, app=None):
         FirebaseError: If an error occurs while communicating with instance ID service.
         ValueError: If the input arguments are invalid.
     """
+    warnings.warn(
+        'unsubscribe_from_topic_legacy is deprecated. Use unsubscribe_from_topic instead.',
+        DeprecationWarning,
+        stacklevel=2)
     return _get_messaging_service(app).make_topic_management_request(
         tokens, topic, 'iid/v1:batchRemove')
 
@@ -410,7 +514,9 @@ def __init__(self, app: App) -> None:
                 'Project ID is required to access Cloud Messaging service. Either set the '
                 'projectId option, or use service account credentials. Alternatively, set the '
                 'GOOGLE_CLOUD_PROJECT environment variable.')
+        self._project_id = project_id
         self._fcm_url = _MessagingService.FCM_URL.format(project_id)
+        self._fcm_topic_url = f'https://fcm.googleapis.com/v1/projects/{project_id}/registrations'
         self._fcm_headers = {
             'X-GOOG-API-FORMAT-VERSION': '2',
             'X-FIREBASE-CLIENT': f'fire-admin-python/{firebase_admin.__version__}',
@@ -418,6 +524,12 @@ def __init__(self, app: App) -> None:
         timeout = app.options.get('httpTimeout', _http_client.DEFAULT_TIMEOUT_SECONDS)
         self._credential = app.credential.get_credential()
         self._client = _http_client.JsonHttpClient(credential=self._credential, timeout=timeout)
+        fcm_adapter = requests.adapters.HTTPAdapter(
+            pool_connections=100,
+            pool_maxsize=100,
+            max_retries=_http_client.DEFAULT_RETRY_CONFIG
+        )
+        self._client.session.mount('https://fcm.googleapis.com', fcm_adapter)
         self._async_client = _http_client.HttpxAsyncClient(
             credential=self._credential, timeout=timeout)
 
@@ -434,7 +546,7 @@ def send(self, message: Message, dry_run: bool = False) -> str:
             resp = self._client.body(
                 'post',
                 url=self._fcm_url,
-                headers=self._fcm_headers,
+                headers=dict(self._fcm_headers),
                 json=data
             )
         except requests.exceptions.RequestException as error:
@@ -453,7 +565,7 @@ def send_data(data):
                 resp = self._client.body(
                     'post',
                     url=self._fcm_url,
-                    headers=self._fcm_headers,
+                    headers=dict(self._fcm_headers),
                     json=data)
             except requests.exceptions.RequestException as exception:
                 return SendResponse(resp=None, exception=self._handle_fcm_error(exception))
@@ -481,7 +593,7 @@ async def send_data(data):
                 resp = await self._async_client.request(
                     'post',
                     url=self._fcm_url,
-                    headers=self._fcm_headers,
+                    headers=dict(self._fcm_headers),
                     json=data)
             except httpx.HTTPError as exception:
                 return SendResponse(resp=None, exception=self._handle_fcm_httpx_error(exception))
@@ -499,6 +611,225 @@ async def send_data(data):
                 message=f'Unknown error while making remote service calls: {error}',
                 cause=error)
 
+    def _validate_topic_management_args(self, tokens, topic):
+        """Validates and formats topic management arguments."""
+        if isinstance(tokens, str):
+            tokens = [tokens]
+        if not isinstance(tokens, list) or not tokens:
+            raise ValueError('Tokens must be a string or a non-empty list of strings.')
+        invalid_str = [t for t in tokens if not isinstance(t, str) or not t]
+        if invalid_str:
+            raise ValueError('Tokens must be non-empty strings.')
+        if len(tokens) > 1000:
+            raise ValueError('tokens must not contain more than 1000 elements.')
+
+        if not isinstance(topic, str) or not topic:
+            raise ValueError('Topic must be a non-empty string.')
+        topic_name = topic
+        if topic_name.startswith('/topics/'):
+            topic_name = topic_name[len('/topics/'):]
+        if not topic_name or not re.match(r'^[a-zA-Z0-9-_\.~%]+\Z', topic_name):
+            raise ValueError('Malformed topic name.')
+
+        return tokens, topic_name
+
+    def subscribe_to_topic(
+        self, tokens: Union[str, List[str]], topic: str
+    ) -> TopicManagementResponse:
+        """Subscribes a list of registration tokens to an FCM topic via the FCM v1 API."""
+        return self._make_topic_management_request_v1(tokens, topic, is_subscribe=True)
+
+    def unsubscribe_from_topic(
+        self, tokens: Union[str, List[str]], topic: str
+    ) -> TopicManagementResponse:
+        """Unsubscribes a list of registration tokens from an FCM topic via the FCM v1 API."""
+        return self._make_topic_management_request_v1(tokens, topic, is_subscribe=False)
+
+    async def subscribe_to_topic_async(
+        self, tokens: Union[str, List[str]], topic: str
+    ) -> TopicManagementResponse:
+        """Subscribes a list of registration tokens to an FCM topic asynchronously
+        via the FCM v1 API."""
+        return await self._make_topic_management_request_v1_async(
+            tokens, topic, is_subscribe=True)
+
+    async def unsubscribe_from_topic_async(
+        self, tokens: Union[str, List[str]], topic: str
+    ) -> TopicManagementResponse:
+        """Unsubscribes a list of registration tokens from an FCM topic asynchronously
+        via the FCM v1 API."""
+        return await self._make_topic_management_request_v1_async(
+            tokens, topic, is_subscribe=False)
+
+    def _make_topic_management_request_v1(
+        self, tokens, topic, is_subscribe: bool
+    ) -> TopicManagementResponse:
+        """Helper method that sends topic subscription requests via FCM v1 API."""
+        tokens_list, topic_name = self._validate_topic_management_args(tokens, topic)
+        encoded_topic = urllib.parse.quote(topic_name, safe='')
+
+        def send_request(token: str):
+            encoded_token = urllib.parse.quote(token, safe='')
+            base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions'
+            if is_subscribe:
+                url = f'{base_url}?topic_name={encoded_topic}'
+                method = 'post'
+                json_data = {}
+            else:
+                url = f'{base_url}/{encoded_topic}?allow_missing=true'
+                method = 'delete'
+                json_data = None
+
+            try:
+                self._client.request(
+                    method,
+                    url=url,
+                    headers=dict(self._fcm_headers),
+                    json=json_data,
+                )
+                return {'success': True}
+            except requests.exceptions.RequestException as error:
+                return self._build_topic_subscription_result_from_requests_error(
+                    error, is_subscribe)
+
+        try:
+            with concurrent.futures.ThreadPoolExecutor(
+                max_workers=min(len(tokens_list), 100)
+            ) as executor:
+                results = list(executor.map(send_request, tokens_list))
+                return self._parse_topic_management_results(results)
+        except Exception as error:
+            raise exceptions.UnknownError(
+                message=f'Unknown error while making remote service calls: {error}',
+                cause=error)
+
+    async def _make_topic_management_request_v1_async(
+        self, tokens, topic, is_subscribe: bool
+    ) -> TopicManagementResponse:
+        """Helper method that sends topic subscription requests asynchronously via FCM v1 API."""
+        tokens_list, topic_name = self._validate_topic_management_args(tokens, topic)
+        encoded_topic = urllib.parse.quote(topic_name, safe='')
+        semaphore = asyncio.Semaphore(100)
+
+        async def send_request_async(token: str):
+            encoded_token = urllib.parse.quote(token, safe='')
+            base_url = f'{self._fcm_topic_url}/{encoded_token}/topicSubscriptions'
+            if is_subscribe:
+                url = f'{base_url}?topic_name={encoded_topic}'
+                method = 'post'
+                json_data = {}
+            else:
+                url = f'{base_url}/{encoded_topic}?allow_missing=true'
+                method = 'delete'
+                json_data = None
+
+            async with semaphore:
+                try:
+                    await self._async_client.request(
+                        method,
+                        url=url,
+                        headers=dict(self._fcm_headers),
+                        json=json_data,
+                    )
+                    return {'success': True}
+                except httpx.HTTPError as error:
+                    return self._build_topic_subscription_result_from_httpx_error(
+                        error, is_subscribe)
+                except requests.exceptions.RequestException as error:
+                    return self._build_topic_subscription_result_from_requests_error(
+                        error, is_subscribe)
+
+        try:
+            results = await asyncio.gather(*[send_request_async(token) for token in tokens_list])
+            return self._parse_topic_management_results(results)
+        except Exception as error:
+            raise exceptions.UnknownError(
+                message=f'Unknown error while making remote service calls: {error}',
+                cause=error)
+
+    @classmethod
+    def _get_topic_error_code(cls, error_dict: dict, status_code: int) -> str:
+        """Extracts the error code for a topic subscription error response."""
+        error_data = error_dict.get('error')
+        if isinstance(error_data, str) and error_data:
+            return error_data
+        if isinstance(error_data, dict):
+            details = error_data.get('details')
+            if isinstance(details, list):
+                fcm_error_type = 'type.googleapis.com/google.firebase.fcm.v1.FcmError'
+                for element in details:
+                    if isinstance(element, dict) and element.get('@type') == fcm_error_type:
+                        code = element.get('errorCode')
+                        if code:
+                            return code
+            status = error_data.get('status')
+            if status:
+                return status
+
+        status_map = {
+            400: 'INVALID_ARGUMENT',
+            401: 'UNAUTHENTICATED',
+            403: 'PERMISSION_DENIED',
+            404: 'NOT_FOUND',
+            408: 'DEADLINE_EXCEEDED',
+            429: 'RESOURCE_EXHAUSTED',
+            500: 'INTERNAL',
+            503: 'UNAVAILABLE',
+            504: 'DEADLINE_EXCEEDED',
+        }
+        if status_code in status_map:
+            return status_map[status_code]
+
+        if isinstance(error_data, dict):
+            message = error_data.get('message')
+            if message:
+                return message
+
+        return 'UNKNOWN_ERROR'
+
+    def _build_topic_subscription_result(self, response, is_subscribe):
+        """Constructs a result dict from a response object."""
+        if response is not None:
+            if is_subscribe and response.status_code == 409:
+                return {'success': True}
+            error_dict = {}
+            try:
+                parsed = response.json()
+                if isinstance(parsed, dict):
+                    error_dict = parsed
+            except ValueError:
+                pass
+
+            error_data = error_dict.get('error')
+            if is_subscribe and isinstance(error_data, dict) and (
+                error_data.get('status') == 'ALREADY_EXISTS'
+            ):
+                return {'success': True}
+
+            error_code = self._get_topic_error_code(error_dict, response.status_code)
+            return {'success': False, 'error': error_code}
+
+        return {'success': False, 'error': 'UNKNOWN_ERROR'}
+
+    def _build_topic_subscription_result_from_requests_error(self, error, is_subscribe):
+        """Constructs a result dict from a requests error."""
+        return self._build_topic_subscription_result(error.response, is_subscribe)
+
+    def _build_topic_subscription_result_from_httpx_error(self, error, is_subscribe):
+        """Constructs a result dict from an httpx error."""
+        response = error.response if isinstance(error, httpx.HTTPStatusError) else None
+        return self._build_topic_subscription_result(response, is_subscribe)
+
+    def _parse_topic_management_results(self, results) -> TopicManagementResponse:
+        """Parses individual request results into a TopicManagementResponse."""
+        formatted_results = []
+        for result in results:
+            if result.get('success'):
+                formatted_results.append({})
+            else:
+                formatted_results.append({'error': result.get('error', 'UNKNOWN_ERROR')})
+        return TopicManagementResponse({'results': formatted_results})
+
     def make_topic_management_request(self, tokens, topic, operation):
         """Invokes the IID service for topic management functionality."""
         if isinstance(tokens, str):
diff --git a/tests/test_messaging.py b/tests/test_messaging.py
index 749e5311..ae2e4a4d 100644
--- a/tests/test_messaging.py
+++ b/tests/test_messaging.py
@@ -14,10 +14,12 @@
 
 """Test cases for the firebase_admin.messaging module."""
 import datetime
+import io
 from itertools import chain, repeat
 import json
 import numbers
 import httpx
+import requests
 import respx
 
 import pytest
@@ -1742,10 +1744,25 @@ def test_topic_management_custom_timeout(self, options, timeout):
         all_options.update(options)
         firebase_admin.initialize_app(cred, all_options)
         recorder = self._instrument_service(
-            'https://iid.googleapis.com', {'results': [{}, {'error': 'error_reason'}]})
+            'https://fcm.googleapis.com', {})
         messaging.subscribe_to_topic(['1'], 'a')
         self._check_timeout(recorder, timeout)
 
+    @pytest.mark.parametrize('options, timeout', [
+        ({'httpTimeout': 4}, 4),
+        ({'httpTimeout': None}, None),
+        ({}, _http_client.DEFAULT_TIMEOUT_SECONDS),
+    ])
+    def test_topic_management_legacy_custom_timeout(self, options, timeout):
+        cred = testutils.MockCredential()
+        all_options = {'projectId': 'explicit-project-id'}
+        all_options.update(options)
+        firebase_admin.initialize_app(cred, all_options)
+        recorder = self._instrument_service(
+            'https://iid.googleapis.com', {'results': [{}, {'error': 'error_reason'}]})
+        messaging.subscribe_to_topic_legacy(['1'], 'a')
+        self._check_timeout(recorder, timeout)
+
 
 class TestSend:
 
@@ -2448,7 +2465,7 @@ def test_send_each_for_multicast_fcm_error_code(self, status):
         check_exception(exception, 'test error', status)
 
 
-class TestTopicManagement:
+class TestTopicManagementLegacy:
 
     _DEFAULT_RESPONSE = json.dumps({'results': [{}, {'error': 'error_reason'}]})
     _DEFAULT_ERROR_RESPONSE = json.dumps({'error': 'error_reason'})
@@ -2502,77 +2519,79 @@ def test_invalid_tokens(self, tokens):
             expected = 'Tokens must be non-empty strings.'
 
         with pytest.raises(ValueError) as excinfo:
-            messaging.subscribe_to_topic(tokens, 'test-topic')
+            messaging.subscribe_to_topic_legacy(tokens, 'test-topic')
         assert str(excinfo.value) == expected
 
         with pytest.raises(ValueError) as excinfo:
-            messaging.unsubscribe_from_topic(tokens, 'test-topic')
+            messaging.unsubscribe_from_topic_legacy(tokens, 'test-topic')
         assert str(excinfo.value) == expected
 
     @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, ''])
     def test_invalid_topic(self, topic):
         expected = 'Topic must be a non-empty string.'
         with pytest.raises(ValueError) as excinfo:
-            messaging.subscribe_to_topic('test-token', topic)
+            messaging.subscribe_to_topic_legacy('test-token', topic)
         assert str(excinfo.value) == expected
 
         with pytest.raises(ValueError) as excinfo:
-            messaging.unsubscribe_from_topic('test-tokens', topic)
+            messaging.unsubscribe_from_topic_legacy('test-tokens', topic)
         assert str(excinfo.value) == expected
 
     @pytest.mark.parametrize('args', _VALID_ARGS)
-    def test_subscribe_to_topic(self, args):
+    def test_subscribe_to_topic_legacy(self, args):
         _, recorder = self._instrument_iid_service()
-        resp = messaging.subscribe_to_topic(args[0], args[1])
+        with pytest.deprecated_call():
+            resp = messaging.subscribe_to_topic_legacy(args[0], args[1])
         self._check_response(resp)
         assert len(recorder) == 1
         self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd'))
         assert json.loads(recorder[0].body.decode()) == args[2]
 
     @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items())
-    def test_subscribe_to_topic_error(self, status, exc_type):
+    def test_subscribe_to_topic_legacy_error(self, status, exc_type):
         _, recorder = self._instrument_iid_service(
             status=status, payload=self._DEFAULT_ERROR_RESPONSE)
         with pytest.raises(exc_type) as excinfo:
-            messaging.subscribe_to_topic('foo', 'test-topic')
+            messaging.subscribe_to_topic_legacy('foo', 'test-topic')
         assert str(excinfo.value) == 'Error while calling the IID service: error_reason'
         assert len(recorder) == 1
         self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd'))
 
     @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items())
-    def test_subscribe_to_topic_non_json_error(self, status, exc_type):
+    def test_subscribe_to_topic_legacy_non_json_error(self, status, exc_type):
         _, recorder = self._instrument_iid_service(status=status, payload='not json')
         with pytest.raises(exc_type) as excinfo:
-            messaging.subscribe_to_topic('foo', 'test-topic')
+            messaging.subscribe_to_topic_legacy('foo', 'test-topic')
         reason = f'Unexpected HTTP response with status: {status}; body: not json'
         assert str(excinfo.value) == reason
         assert len(recorder) == 1
         self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchAdd'))
 
     @pytest.mark.parametrize('args', _VALID_ARGS)
-    def test_unsubscribe_from_topic(self, args):
+    def test_unsubscribe_from_topic_legacy(self, args):
         _, recorder = self._instrument_iid_service()
-        resp = messaging.unsubscribe_from_topic(args[0], args[1])
+        with pytest.deprecated_call():
+            resp = messaging.unsubscribe_from_topic_legacy(args[0], args[1])
         self._check_response(resp)
         assert len(recorder) == 1
         self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchRemove'))
         assert json.loads(recorder[0].body.decode()) == args[2]
 
     @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items())
-    def test_unsubscribe_from_topic_error(self, status, exc_type):
+    def test_unsubscribe_from_topic_legacy_error(self, status, exc_type):
         _, recorder = self._instrument_iid_service(
             status=status, payload=self._DEFAULT_ERROR_RESPONSE)
         with pytest.raises(exc_type) as excinfo:
-            messaging.unsubscribe_from_topic('foo', 'test-topic')
+            messaging.unsubscribe_from_topic_legacy('foo', 'test-topic')
         assert str(excinfo.value) == 'Error while calling the IID service: error_reason'
         assert len(recorder) == 1
         self._assert_request(recorder[0], 'POST', self._get_url('iid/v1:batchRemove'))
 
     @pytest.mark.parametrize('status, exc_type', HTTP_ERROR_CODES.items())
-    def test_unsubscribe_from_topic_non_json_error(self, status, exc_type):
+    def test_unsubscribe_from_topic_legacy_non_json_error(self, status, exc_type):
         _, recorder = self._instrument_iid_service(status=status, payload='not json')
         with pytest.raises(exc_type) as excinfo:
-            messaging.unsubscribe_from_topic('foo', 'test-topic')
+            messaging.unsubscribe_from_topic_legacy('foo', 'test-topic')
         reason = f'Unexpected HTTP response with status: {status}; body: not json'
         assert str(excinfo.value) == reason
         assert len(recorder) == 1
@@ -2584,3 +2603,530 @@ def _check_response(self, resp):
         assert len(resp.errors) == 1
         assert resp.errors[0].index == 1
         assert resp.errors[0].reason == 'error_reason'
+
+
+class TestTopicManagement:
+
+    _CLIENT_VERSION = f'fire-admin-python/{firebase_admin.__version__}'
+
+    @classmethod
+    def setup_class(cls):
+        cred = testutils.MockCredential()
+        firebase_admin.initialize_app(cred, {'projectId': 'explicit-project-id'})
+
+    @classmethod
+    def teardown_class(cls):
+        testutils.cleanup_apps()
+
+    def _instrument_messaging_service(self, app=None, status=200, payload='{}'):
+        if not app:
+            app = firebase_admin.get_app()
+        fcm_service = messaging._get_messaging_service(app)
+        recorder = []
+        fcm_service._client.session.mount(
+            'https://fcm.googleapis.com',
+            testutils.MockAdapter(payload, status, recorder))
+        return fcm_service, recorder
+
+    def _assert_request(self, request, expected_method, expected_url, expected_body=None):
+        assert request.method == expected_method
+        assert request.url == expected_url
+        assert request.headers['X-GOOG-API-FORMAT-VERSION'] == '2'
+        assert request.headers['X-FIREBASE-CLIENT'] == self._CLIENT_VERSION
+        expected_metrics_header = _utils.get_metrics_header() + ' mock-cred-metric-tag'
+        assert request.headers['x-goog-api-client'] == expected_metrics_header
+        if expected_body is None:
+            assert request.body is None
+        else:
+            assert json.loads(request.body.decode()) == expected_body
+
+    @pytest.mark.parametrize('tokens', [None, '', [], {}, tuple()])
+    def test_invalid_tokens(self, tokens):
+        expected = 'Tokens must be a string or a non-empty list of strings.'
+        if isinstance(tokens, str):
+            expected = 'Tokens must be non-empty strings.'
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.subscribe_to_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == expected
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.unsubscribe_from_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == expected
+
+    @pytest.mark.parametrize('tokens', [
+        ['foo', 'bar', ''],
+        ['foo', 123, 'bar'],
+    ])
+    def test_invalid_tokens_in_list(self, tokens):
+        with pytest.raises(ValueError) as excinfo:
+            messaging.subscribe_to_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == 'Tokens must be non-empty strings.'
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.unsubscribe_from_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == 'Tokens must be non-empty strings.'
+
+    def test_tokens_over_1000(self):
+        tokens = [f'token{i}' for i in range(1001)]
+        with pytest.raises(ValueError) as excinfo:
+            messaging.subscribe_to_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.'
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.unsubscribe_from_topic(tokens, 'test-topic')
+        assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.'
+
+    @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, ''])
+    def test_invalid_topic(self, topic):
+        expected = 'Topic must be a non-empty string.'
+        with pytest.raises(ValueError) as excinfo:
+            messaging.subscribe_to_topic('test-token', topic)
+        assert str(excinfo.value) == expected
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.unsubscribe_from_topic('test-token', topic)
+        assert str(excinfo.value) == expected
+
+    @pytest.mark.parametrize('topic', [
+        '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&',
+        '/topics/foo\n', 'foo\n',
+    ])
+    def test_malformed_topic(self, topic):
+        with pytest.raises(ValueError) as excinfo:
+            messaging.subscribe_to_topic('test-token', topic)
+        assert str(excinfo.value) == 'Malformed topic name.'
+
+        with pytest.raises(ValueError) as excinfo:
+            messaging.unsubscribe_from_topic('test-token', topic)
+        assert str(excinfo.value) == 'Malformed topic name.'
+
+    def test_subscribe_to_topic_single(self):
+        _, recorder = self._instrument_messaging_service()
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+        assert len(recorder) == 1
+        expected_url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        self._assert_request(recorder[0], 'POST', expected_url, {})
+
+    def test_subscribe_to_topic_prefixed(self):
+        _, recorder = self._instrument_messaging_service()
+        resp = messaging.subscribe_to_topic('token1', '/topics/test-topic')
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+        assert len(recorder) == 1
+        expected_url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        self._assert_request(recorder[0], 'POST', expected_url, {})
+
+    def test_unsubscribe_from_topic_single(self):
+        _, recorder = self._instrument_messaging_service()
+        resp = messaging.unsubscribe_from_topic('token1', 'test-topic')
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+        assert len(recorder) == 1
+        expected_url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions/test-topic?allow_missing=true'
+        )
+        self._assert_request(recorder[0], 'DELETE', expected_url, None)
+
+    def test_unsubscribe_from_topic_prefixed(self):
+        _, recorder = self._instrument_messaging_service()
+        resp = messaging.unsubscribe_from_topic('token1', '/topics/test-topic')
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+        assert len(recorder) == 1
+        expected_url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions/test-topic?allow_missing=true'
+        )
+        self._assert_request(recorder[0], 'DELETE', expected_url, None)
+
+    def test_subscribe_to_topic_already_exists_409(self):
+        payload = json.dumps({'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}})
+        _, recorder = self._instrument_messaging_service(status=409, payload=payload)
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+        assert len(recorder) == 1
+
+    def test_unsubscribe_from_topic_not_found_404(self):
+        payload = json.dumps({'error': {'status': 'NOT_FOUND', 'message': 'Not found'}})
+        _, recorder = self._instrument_messaging_service(status=404, payload=payload)
+        resp = messaging.unsubscribe_from_topic('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'NOT_FOUND'
+        assert len(recorder) == 1
+
+    def test_topic_management_multiple_tokens(self):
+        fcm_service = messaging._get_messaging_service(firebase_admin.get_app())
+        recorder = []
+
+        class MultiTokenMockAdapter(requests.adapters.HTTPAdapter):
+            def send(self, request, **kwargs):  # pylint: disable=arguments-differ,unused-argument
+                recorder.append(request)
+                resp = requests.models.Response()
+                resp.url = request.url
+                if 'token2' in request.url:
+                    resp.status_code = 404
+                    content = json.dumps({'error': {'status': 'NOT_FOUND'}}).encode()
+                else:
+                    resp.status_code = 200
+                    content = b'{}'
+                resp.raw = io.BytesIO(content)
+                return resp
+
+        fcm_service._client.session.mount(
+            'https://fcm.googleapis.com',
+            MultiTokenMockAdapter())
+
+        resp = messaging.subscribe_to_topic(['token1', 'token2', 'token3'], 'test-topic')
+        assert resp.success_count == 2
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 1
+        assert resp.errors[0].reason == 'NOT_FOUND'
+        assert len(recorder) == 3
+
+    def test_topic_management_fcm_error_details(self):
+        payload = json.dumps({
+            'error': {
+                'status': 'NOT_FOUND',
+                'details': [
+                    {
+                        '@type': 'type.googleapis.com/google.firebase.fcm.v1.FcmError',
+                        'errorCode': 'UNREGISTERED',
+                    },
+                ],
+            }
+        })
+        _, recorder = self._instrument_messaging_service(status=404, payload=payload)
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'UNREGISTERED'
+        assert len(recorder) == 1
+
+    def test_topic_management_500_error(self):
+        _, recorder = self._instrument_messaging_service(status=500, payload='{"error": null}')
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'INTERNAL'
+        assert len(recorder) == 1
+
+    def test_topic_management_non_json_error(self):
+        _, recorder = self._instrument_messaging_service(status=400, payload='not json')
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'INVALID_ARGUMENT'
+        assert len(recorder) == 1
+
+    @pytest.mark.parametrize('status_code, expected_reason', [
+        (401, 'UNAUTHENTICATED'),
+        (403, 'PERMISSION_DENIED'),
+        (408, 'DEADLINE_EXCEEDED'),
+        (429, 'RESOURCE_EXHAUSTED'),
+        (503, 'UNAVAILABLE'),
+        (504, 'DEADLINE_EXCEEDED'),
+    ])
+    def test_topic_management_status_code_mapping(self, status_code, expected_reason):
+        _, recorder = self._instrument_messaging_service(
+            status=status_code, payload=json.dumps({'error': {'message': 'Some message'}}))
+        resp = messaging.subscribe_to_topic('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == expected_reason
+        assert len(recorder) == (3 if status_code == 401 else 1)
+
+
+class TestTopicManagementAsync:
+
+    @classmethod
+    def setup_class(cls):
+        cred = testutils.MockCredential()
+        firebase_admin.initialize_app(cred, {'projectId': 'explicit-project-id'})
+
+    @classmethod
+    def teardown_class(cls):
+        testutils.cleanup_apps()
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_subscribe_to_topic_async_single(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        route = respx.post(url).mock(return_value=respx.MockResponse(200, json={}))
+        resp = await messaging.subscribe_to_topic_async('token1', 'test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_subscribe_to_topic_async_already_exists(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        payload = {'error': {'status': 'ALREADY_EXISTS', 'message': 'Already exists'}}
+        route = respx.post(url).mock(return_value=respx.MockResponse(409, json=payload))
+        resp = await messaging.subscribe_to_topic_async('token1', 'test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_unsubscribe_from_topic_async_single(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions/test-topic?allow_missing=true'
+        )
+        route = respx.delete(url).mock(return_value=respx.MockResponse(200, json={}))
+        resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_unsubscribe_from_topic_async_not_found(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions/test-topic?allow_missing=true'
+        )
+        payload = {'error': {'status': 'NOT_FOUND', 'message': 'Not found'}}
+        route = respx.delete(url).mock(return_value=respx.MockResponse(404, json=payload))
+        resp = await messaging.unsubscribe_from_topic_async('token1', 'test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'NOT_FOUND'
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_subscribe_to_topic_async_multiple(self):
+        base = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations'
+        url1 = f'{base}/token1/topicSubscriptions?topic_name=test-topic'
+        url2 = f'{base}/token2/topicSubscriptions?topic_name=test-topic'
+        url3 = f'{base}/token3/topicSubscriptions?topic_name=test-topic'
+        route1 = respx.post(url1).mock(return_value=respx.MockResponse(200, json={}))
+        route2 = respx.post(url2).mock(
+            return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}}))
+        route3 = respx.post(url3).mock(return_value=respx.MockResponse(200, json={}))
+
+        resp = await messaging.subscribe_to_topic_async(
+            ['token1', 'token2', 'token3'], 'test-topic')
+        assert route1.call_count == 1
+        assert route2.call_count == 1
+        assert route3.call_count == 1
+        assert resp.success_count == 2
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 1
+        assert resp.errors[0].reason == 'NOT_FOUND'
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_subscribe_to_topic_async_prefixed(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        route = respx.post(url).mock(return_value=respx.MockResponse(200, json={}))
+        resp = await messaging.subscribe_to_topic_async('token1', '/topics/test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_unsubscribe_from_topic_async_prefixed(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions/test-topic?allow_missing=true'
+        )
+        route = respx.delete(url).mock(return_value=respx.MockResponse(200, json={}))
+        resp = await messaging.unsubscribe_from_topic_async('token1', '/topics/test-topic')
+        assert route.call_count == 1
+        assert resp.success_count == 1
+        assert resp.failure_count == 0
+        assert resp.errors == []
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_unsubscribe_from_topic_async_multiple(self):
+        base = 'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations'
+        url1 = f'{base}/token1/topicSubscriptions/test-topic?allow_missing=true'
+        url2 = f'{base}/token2/topicSubscriptions/test-topic?allow_missing=true'
+        url3 = f'{base}/token3/topicSubscriptions/test-topic?allow_missing=true'
+        route1 = respx.delete(url1).mock(return_value=respx.MockResponse(200, json={}))
+        route2 = respx.delete(url2).mock(
+            return_value=respx.MockResponse(404, json={'error': {'status': 'NOT_FOUND'}}))
+        route3 = respx.delete(url3).mock(return_value=respx.MockResponse(200, json={}))
+
+        resp = await messaging.unsubscribe_from_topic_async(
+            ['token1', 'token2', 'token3'], 'test-topic')
+        assert route1.call_count == 1
+        assert route2.call_count == 1
+        assert route3.call_count == 1
+        assert resp.success_count == 2
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 1
+        assert resp.errors[0].reason == 'NOT_FOUND'
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('tokens', [None, '', [], {}, tuple()])
+    async def test_invalid_tokens(self, tokens):
+        expected = 'Tokens must be a string or a non-empty list of strings.'
+        if isinstance(tokens, str):
+            expected = 'Tokens must be non-empty strings.'
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.subscribe_to_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == expected
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.unsubscribe_from_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == expected
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('tokens', [
+        ['foo', 'bar', ''],
+        ['foo', 123, 'bar'],
+    ])
+    async def test_invalid_tokens_in_list(self, tokens):
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.subscribe_to_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == 'Tokens must be non-empty strings.'
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.unsubscribe_from_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == 'Tokens must be non-empty strings.'
+
+    @pytest.mark.asyncio
+    async def test_tokens_over_1000(self):
+        tokens = [f'token{i}' for i in range(1001)]
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.subscribe_to_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.'
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.unsubscribe_from_topic_async(tokens, 'test-topic')
+        assert str(excinfo.value) == 'tokens must not contain more than 1000 elements.'
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('topic', NON_STRING_ARGS + [None, ''])
+    async def test_invalid_topic(self, topic):
+        expected = 'Topic must be a non-empty string.'
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.subscribe_to_topic_async('test-token', topic)
+        assert str(excinfo.value) == expected
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.unsubscribe_from_topic_async('test-token', topic)
+        assert str(excinfo.value) == expected
+
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize('topic', [
+        '/topics/', '/foo/bar', 'foo bar', 'f*o*o', '/topics/f+o+o', '$foo', '/topics/foo&',
+        '/topics/foo\n', 'foo\n',
+    ])
+    async def test_malformed_topic(self, topic):
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.subscribe_to_topic_async('test-token', topic)
+        assert str(excinfo.value) == 'Malformed topic name.'
+
+        with pytest.raises(ValueError) as excinfo:
+            await messaging.unsubscribe_from_topic_async('test-token', topic)
+        assert str(excinfo.value) == 'Malformed topic name.'
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_topic_management_async_fcm_error_details(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        payload = {
+            'error': {
+                'status': 'NOT_FOUND',
+                'details': [
+                    {
+                        '@type': 'type.googleapis.com/google.firebase.fcm.v1.FcmError',
+                        'errorCode': 'UNREGISTERED',
+                    },
+                ],
+            }
+        }
+        respx.post(url).mock(return_value=respx.MockResponse(404, json=payload))
+        resp = await messaging.subscribe_to_topic_async('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'UNREGISTERED'
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_topic_management_async_500_error(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        respx.post(url).mock(return_value=respx.MockResponse(500, json={'error': None}))
+        resp = await messaging.subscribe_to_topic_async('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'INTERNAL'
+
+    @pytest.mark.asyncio
+    @respx.mock
+    async def test_topic_management_async_non_json_error(self):
+        url = (
+            'https://fcm.googleapis.com/v1/projects/explicit-project-id/registrations/'
+            'token1/topicSubscriptions?topic_name=test-topic'
+        )
+        respx.post(url).mock(return_value=respx.MockResponse(400, text='not json'))
+        resp = await messaging.subscribe_to_topic_async('token1', 'test-topic')
+        assert resp.success_count == 0
+        assert resp.failure_count == 1
+        assert len(resp.errors) == 1
+        assert resp.errors[0].index == 0
+        assert resp.errors[0].reason == 'INVALID_ARGUMENT'

From 6d5d8b6f075f2446497c4e2d5a62ecb9937a6d3a Mon Sep 17 00:00:00 2001
From: Lahiru Maramba <llahiru@gmail.com>
Date: Wed, 23 Sep 2026 12:01:15 -0400
Subject: [PATCH 7/9] fix(appcheck): Use v1beta endpoint for
 verifyAppCheckToken (#987)

* fix(fac): Use v1beta endpoint for verifyAppCheckToken

Updated _AppCheckService._VERIFY_URL_FORMAT to point to the v1beta endpoint (https://firebaseappcheck.googleapis.com/v1beta/projects/{project_id}:verifyAppCheckToken) for replay protection verification.

* chore(fac): Fix line length lint in test_app_check.py
---
 firebase_admin/app_check.py | 2 +-
 tests/test_app_check.py     | 6 ++++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/firebase_admin/app_check.py b/firebase_admin/app_check.py
index 1117e151..7dbd8fe6 100644
--- a/firebase_admin/app_check.py
+++ b/firebase_admin/app_check.py
@@ -54,7 +54,7 @@ class _AppCheckService:
     _APP_CHECK_ISSUER = 'https://firebaseappcheck.googleapis.com/'
     _JWKS_URL = 'https://firebaseappcheck.googleapis.com/v1/jwks'
     _VERIFY_URL_FORMAT = (
-        'https://firebaseappcheck.googleapis.com/v1/projects/{project_id}:verifyAppCheckToken'
+        'https://firebaseappcheck.googleapis.com/v1beta/projects/{project_id}:verifyAppCheckToken'
     )
     _project_id = None
     _scoped_project_id = None
diff --git a/tests/test_app_check.py b/tests/test_app_check.py
index b7a7bcd3..b88d1d6a 100644
--- a/tests/test_app_check.py
+++ b/tests/test_app_check.py
@@ -319,7 +319,8 @@ def test_verify_token_with_consume_true_not_consumed(self, mocker):
         assert payload == expected
 
         expected_url = (
-            f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken"
+            f"https://firebaseappcheck.googleapis.com/v1beta/projects/{PROJECT_ID}"
+            ":verifyAppCheckToken"
         )
         mock_body.assert_called_once_with(
             "post", expected_url, json={"app_check_token": "encoded"}
@@ -340,7 +341,8 @@ def test_verify_token_with_consume_true_already_consumed(self, mocker):
         assert payload == expected
 
         expected_url = (
-            f"https://firebaseappcheck.googleapis.com/v1/projects/{PROJECT_ID}:verifyAppCheckToken"
+            f"https://firebaseappcheck.googleapis.com/v1beta/projects/{PROJECT_ID}"
+            ":verifyAppCheckToken"
         )
         mock_body.assert_called_once_with(
             "post", expected_url, json={"app_check_token": "encoded"}

From de456071dc7450a6d7715667debf6049f38fd85d Mon Sep 17 00:00:00 2001
From: Lahiru Maramba <llahiru@gmail.com>
Date: Wed, 23 Sep 2026 16:28:06 -0400
Subject: [PATCH 8/9] [chore] Release 7.7.0 (#988)

---
 firebase_admin/__about__.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/firebase_admin/__about__.py b/firebase_admin/__about__.py
index 5b706a61..09b693d7 100644
--- a/firebase_admin/__about__.py
+++ b/firebase_admin/__about__.py
@@ -14,7 +14,7 @@
 
 """About information (version, etc) for Firebase Admin SDK."""
 
-__version__ = '7.6.0'
+__version__ = '7.7.0'
 __title__ = 'firebase_admin'
 __author__ = 'Firebase'
 __license__ = 'Apache License 2.0'

From ae0a284127609272957c594af7c8a6179d87d1d4 Mon Sep 17 00:00:00 2001
From: macastelaz <34776182+macastelaz@users.noreply.github.com>
Date: Wed, 21 Oct 2026 11:56:51 -0500
Subject: [PATCH 9/9] fix(auth): validate tenant and provider IDs and
 URL-encode query parameters (#996)

* fix(auth): validate tenant and provider IDs and URL-encode query parameters

* fix(auth): simplify tenant and provider ID validation checks
---
 firebase_admin/_auth_client.py       |  3 ++
 firebase_admin/_auth_providers.py    | 15 +++++----
 firebase_admin/_auth_utils.py        | 17 ++++++++++
 firebase_admin/_user_mgt.py          |  3 +-
 firebase_admin/project_management.py | 27 +++++++--------
 firebase_admin/tenant_mgt.py         | 15 +++------
 tests/test_auth_providers.py         | 50 +++++++++++++++++++++++++---
 tests/test_project_management.py     | 26 +++++++++++++++
 tests/test_tenant_mgt.py             |  8 +++--
 9 files changed, 124 insertions(+), 40 deletions(-)

diff --git a/firebase_admin/_auth_client.py b/firebase_admin/_auth_client.py
index 74261fa3..254bd865 100644
--- a/firebase_admin/_auth_client.py
+++ b/firebase_admin/_auth_client.py
@@ -37,6 +37,9 @@ def __init__(self, app, tenant_id=None):
             2. set the project ID explicitly via Firebase App options, or
             3. set the project ID via the GOOGLE_CLOUD_PROJECT environment variable.""")
 
+        if tenant_id is not None:
+            _auth_utils.validate_tenant_id(tenant_id)
+
         credential = None
         version_header = f'Python/Admin/{firebase_admin.__version__}'
         timeout = app.options.get('httpTimeout', _http_client.DEFAULT_TIMEOUT_SECONDS)
diff --git a/firebase_admin/_auth_providers.py b/firebase_admin/_auth_providers.py
index cc794952..232cf25e 100644
--- a/firebase_admin/_auth_providers.py
+++ b/firebase_admin/_auth_providers.py
@@ -182,7 +182,8 @@ def __init__(self, http_client, project_id, tenant_id=None, url_override=None):
         self.http_client = http_client
         url_prefix = url_override or self.PROVIDER_CONFIG_URL
         self.base_url = f'{url_prefix}/projects/{project_id}'
-        if tenant_id:
+        if tenant_id is not None:
+            _auth_utils.validate_tenant_id(tenant_id)
             self.base_url += f'/tenants/{tenant_id}'
 
     def get_oidc_provider_config(self, provider_id):
@@ -218,7 +219,7 @@ def create_oidc_provider_config(
         if response_type:
             req['responseType'] = response_type
 
-        params = f'oauthIdpConfigId={provider_id}'
+        params = {'oauthIdpConfigId': provider_id}
         body = self._make_request('post', '/oauthIdpConfigs', json=req, params=params)
         return OIDCProviderConfig(body)
 
@@ -301,7 +302,7 @@ def create_saml_provider_config(
         if enabled is not None:
             req['enabled'] = _auth_utils.validate_boolean(enabled, 'enabled')
 
-        params = f'inboundSamlConfigId={provider_id}'
+        params = {'inboundSamlConfigId': provider_id}
         body = self._make_request('post', '/inboundSamlConfigs', json=req, params=params)
         return SAMLProviderConfig(body)
 
@@ -369,9 +370,9 @@ def _fetch_provider_configs(self, path, page_token=None, max_results=MAX_LIST_CO
                 'Max results must be a positive integer less than or equal to '
                 f'{MAX_LIST_CONFIGS_RESULTS}.')
 
-        params = f'pageSize={max_results}'
+        params = {'pageSize': max_results}
         if page_token:
-            params += f'&pageToken={page_token}'
+            params['pageToken'] = page_token
         return self._make_request('get', path, params=params)
 
     def _make_request(self, method, path, **kwargs):
@@ -386,7 +387,7 @@ def _validate_oidc_provider_id(provider_id):
     if not isinstance(provider_id, str):
         raise ValueError(
             f'Invalid OIDC provider ID: {provider_id}. Provider ID must be a non-empty string.')
-    if not provider_id.startswith('oidc.'):
+    if not provider_id.startswith('oidc.') or not _auth_utils.is_safe_url_segment(provider_id):
         raise ValueError(f'Invalid OIDC provider ID: {provider_id}.')
     return provider_id
 
@@ -395,7 +396,7 @@ def _validate_saml_provider_id(provider_id):
     if not isinstance(provider_id, str):
         raise ValueError(
             f'Invalid SAML provider ID: {provider_id}. Provider ID must be a non-empty string.')
-    if not provider_id.startswith('saml.'):
+    if not provider_id.startswith('saml.') or not _auth_utils.is_safe_url_segment(provider_id):
         raise ValueError(f'Invalid SAML provider ID: {provider_id}.')
     return provider_id
 
diff --git a/firebase_admin/_auth_utils.py b/firebase_admin/_auth_utils.py
index 8f3c419a..4dd39018 100644
--- a/firebase_admin/_auth_utils.py
+++ b/firebase_admin/_auth_utils.py
@@ -30,6 +30,7 @@
     'iss', 'jti', 'nbf', 'nonce', 'sub', 'firebase',
 ])
 VALID_EMAIL_ACTION_TYPES = set(['VERIFY_EMAIL', 'EMAIL_SIGNIN', 'PASSWORD_RESET'])
+_SAFE_URL_SEGMENT_PATTERN = re.compile(r'^[a-zA-Z0-9\-_.~]+\Z')
 
 
 class PageIterator:
@@ -83,6 +84,22 @@ def is_emulated():
     return get_emulator_host() != ''
 
 
+def is_safe_url_segment(segment):
+    if not isinstance(segment, str) or not segment or segment.endswith('.'):
+        return False
+    return bool(_SAFE_URL_SEGMENT_PATTERN.match(segment))
+
+
+def validate_tenant_id(tenant_id):
+    if not isinstance(tenant_id, str) or not tenant_id:
+        raise ValueError(
+            f'Invalid tenant ID: {tenant_id}. Tenant ID must be a non-empty string.')
+    if not is_safe_url_segment(tenant_id):
+        raise ValueError(
+            f'Invalid tenant ID: {tenant_id}. Tenant ID must be a valid URL path segment.')
+    return tenant_id
+
+
 def validate_uid(uid, required=False):
     if uid is None and not required:
         return None
diff --git a/firebase_admin/_user_mgt.py b/firebase_admin/_user_mgt.py
index e7825499..4d852500 100644
--- a/firebase_admin/_user_mgt.py
+++ b/firebase_admin/_user_mgt.py
@@ -602,7 +602,8 @@ def __init__(self, http_client, project_id, tenant_id=None, url_override=None):
         self.http_client = http_client
         url_prefix = url_override or self.ID_TOOLKIT_URL
         self.base_url = f'{url_prefix}/projects/{project_id}'
-        if tenant_id:
+        if tenant_id is not None:
+            _auth_utils.validate_tenant_id(tenant_id)
             self.base_url += f'/tenants/{tenant_id}'
 
     def get_user(self, **kwargs):
diff --git a/firebase_admin/project_management.py b/firebase_admin/project_management.py
index 73c100d3..74696dfb 100644
--- a/firebase_admin/project_management.py
+++ b/firebase_admin/project_management.py
@@ -541,11 +541,9 @@ def list_ios_apps(self):
 
     def _list_apps(self, platform_resource_name, app_class):
         """Lists all the Android or iOS apps within the Firebase project."""
-        path = (
-            f'/v1beta1/projects/{self._project_id}/{platform_resource_name}?pageSize='
-            f'{_ProjectManagementService.MAXIMUM_LIST_APPS_PAGE_SIZE}'
-        )
-        response = self._make_request('get', path)
+        path = f'/v1beta1/projects/{self._project_id}/{platform_resource_name}'
+        params = {'pageSize': _ProjectManagementService.MAXIMUM_LIST_APPS_PAGE_SIZE}
+        response = self._make_request('get', path, params=params)
         apps_list = []
         while True:
             apps = response.get('apps')
@@ -556,12 +554,11 @@ def _list_apps(self, platform_resource_name, app_class):
             if not next_page_token:
                 break
             # Retrieve the next page of apps.
-            path = (
-                f'/v1beta1/projects/{self._project_id}/{platform_resource_name}'
-                f'?pageToken={next_page_token}'
-                f'&pageSize={_ProjectManagementService.MAXIMUM_LIST_APPS_PAGE_SIZE}'
-            )
-            response = self._make_request('get', path)
+            params = {
+                'pageToken': next_page_token,
+                'pageSize': _ProjectManagementService.MAXIMUM_LIST_APPS_PAGE_SIZE,
+            }
+            response = self._make_request('get', path, params=params)
         return apps_list
 
     def create_android_app(self, package_name, display_name=None):
@@ -652,12 +649,12 @@ def delete_sha_certificate(self, certificate_to_delete):
         path = f'/v1beta1/{name}'
         self._make_request('delete', path)
 
-    def _make_request(self, method, url, json=None):
-        body, _ = self._body_and_response(method, url, json)
+    def _make_request(self, method, url, json=None, params=None):
+        body, _ = self._body_and_response(method, url, json=json, params=params)
         return body
 
-    def _body_and_response(self, method, url, json=None):
+    def _body_and_response(self, method, url, json=None, params=None):
         try:
-            return self._client.body_and_response(method=method, url=url, json=json)
+            return self._client.body_and_response(method=method, url=url, json=json, params=params)
         except requests.exceptions.RequestException as error:
             raise _utils.handle_platform_error_from_requests(error)
diff --git a/firebase_admin/tenant_mgt.py b/firebase_admin/tenant_mgt.py
index 9e713d98..f9f78d58 100644
--- a/firebase_admin/tenant_mgt.py
+++ b/firebase_admin/tenant_mgt.py
@@ -246,9 +246,7 @@ def __init__(self, app):
 
     def auth_for_tenant(self, tenant_id):
         """Gets an Auth Client instance scoped to the given tenant ID."""
-        if not isinstance(tenant_id, str) or not tenant_id:
-            raise ValueError(
-                f'Invalid tenant ID: {tenant_id}. Tenant ID must be a non-empty string.')
+        _auth_utils.validate_tenant_id(tenant_id)
 
         with self.lock:
             if tenant_id in self.tenant_clients:
@@ -260,9 +258,7 @@ def auth_for_tenant(self, tenant_id):
 
     def get_tenant(self, tenant_id):
         """Gets the tenant corresponding to the given ``tenant_id``."""
-        if not isinstance(tenant_id, str) or not tenant_id:
-            raise ValueError(
-                f'Invalid tenant ID: {tenant_id}. Tenant ID must be a non-empty string.')
+        _auth_utils.validate_tenant_id(tenant_id)
 
         try:
             body = self.client.body('get', f'/tenants/{tenant_id}')
@@ -292,8 +288,7 @@ def update_tenant(
             self, tenant_id, display_name=None, allow_password_sign_up=None,
             enable_email_link_sign_in=None):
         """Updates the specified tenant with the given parameters."""
-        if not isinstance(tenant_id, str) or not tenant_id:
-            raise ValueError('Tenant ID must be a non-empty string.')
+        _auth_utils.validate_tenant_id(tenant_id)
 
         payload = {}
         if display_name is not None:
@@ -319,9 +314,7 @@ def update_tenant(
 
     def delete_tenant(self, tenant_id):
         """Deletes the tenant corresponding to the given ``tenant_id``."""
-        if not isinstance(tenant_id, str) or not tenant_id:
-            raise ValueError(
-                f'Invalid tenant ID: {tenant_id}. Tenant ID must be a non-empty string.')
+        _auth_utils.validate_tenant_id(tenant_id)
 
         try:
             self.client.request('delete', f'/tenants/{tenant_id}')
diff --git a/tests/test_auth_providers.py b/tests/test_auth_providers.py
index 106e1cae..959ecfde 100644
--- a/tests/test_auth_providers.py
+++ b/tests/test_auth_providers.py
@@ -45,6 +45,14 @@
 }"""
 
 INVALID_PROVIDER_IDS = [None, True, False, 1, 0, [], tuple(), {}, '']
+INVALID_OIDC_PROVIDER_IDS = INVALID_PROVIDER_IDS + [
+    'saml.provider', 'oidc.', 'oidc.provider.', 'oidc.provider.\n', 'oidc.provider\n',
+    'oidc.provider/sub', 'oidc.provider?param=1', 'oidc.provider#frag',
+]
+INVALID_SAML_PROVIDER_IDS = INVALID_PROVIDER_IDS + [
+    'oidc.provider', 'saml.', 'saml.provider.', 'saml.provider.\n', 'saml.provider\n',
+    'saml.provider/sub', 'saml.provider?param=1', 'saml.provider#frag',
+]
 
 
 @pytest.fixture(scope='module', params=[{'emulated': False}, {'emulated': True}])
@@ -105,7 +113,7 @@ class TestOIDCProviderConfig:
         },
     }
 
-    @pytest.mark.parametrize('provider_id', INVALID_PROVIDER_IDS + ['saml.provider'])
+    @pytest.mark.parametrize('provider_id', INVALID_OIDC_PROVIDER_IDS)
     def test_get_invalid_provider_id(self, user_mgt_app, provider_id):
         with pytest.raises(ValueError) as excinfo:
             auth.get_oidc_provider_config(provider_id, app=user_mgt_app)
@@ -124,6 +132,9 @@ def test_get(self, user_mgt_app):
 
     @pytest.mark.parametrize('invalid_opts', [
         {'provider_id': None}, {'provider_id': ''}, {'provider_id': 'saml.provider'},
+        {'provider_id': 'oidc.'}, {'provider_id': 'oidc.provider.'},
+        {'provider_id': 'oidc.provider/sub'}, {'provider_id': 'oidc.provider?param=1'},
+        {'provider_id': 'oidc.provider#frag'},
         {'client_id': None}, {'client_id': ''},
         {'issuer': None}, {'issuer': ''}, {'issuer': 'not a url'},
         {'display_name': True},
@@ -203,6 +214,9 @@ def test_create_empty_values(self, user_mgt_app):
     @pytest.mark.parametrize('invalid_opts', [
         {},
         {'provider_id': None}, {'provider_id': ''}, {'provider_id': 'saml.provider'},
+        {'provider_id': 'oidc.'}, {'provider_id': 'oidc.provider.'},
+        {'provider_id': 'oidc.provider/sub'}, {'provider_id': 'oidc.provider?param=1'},
+        {'provider_id': 'oidc.provider#frag'},
         {'client_id': ''},
         {'issuer': ''}, {'issuer': 'not a url'},
         {'display_name': True},
@@ -265,7 +279,7 @@ def test_update_empty_values(self, user_mgt_app):
         got = json.loads(recorder[0].body.decode())
         assert got == {'displayName': None, 'enabled': False, 'responseType': {'idToken': False}}
 
-    @pytest.mark.parametrize('provider_id', INVALID_PROVIDER_IDS + ['saml.provider'])
+    @pytest.mark.parametrize('provider_id', INVALID_OIDC_PROVIDER_IDS)
     def test_delete_invalid_provider_id(self, user_mgt_app, provider_id):
         with pytest.raises(ValueError) as excinfo:
             auth.delete_oidc_provider_config(provider_id, app=user_mgt_app)
@@ -303,6 +317,17 @@ def test_list_single_page(self, user_mgt_app):
         _assert_request(recorder[0], 'GET',
                         f'{USER_MGT_URLS["PREFIX"]}/oauthIdpConfigs?pageSize=100')
 
+    def test_list_page_token_encoding(self, user_mgt_app):
+        recorder = _instrument_provider_mgt(user_mgt_app, 200, LIST_OIDC_PROVIDER_CONFIGS_RESPONSE)
+        page = auth.list_oidc_provider_configs(page_token='token+1=2&3#4', app=user_mgt_app)
+
+        self._assert_page(page)
+        assert len(recorder) == 1
+        _assert_request(
+            recorder[0], 'GET',
+            f'{USER_MGT_URLS["PREFIX"]}/oauthIdpConfigs'
+            '?pageSize=100&pageToken=token%2B1%3D2%263%234')
+
     def test_list_multiple_pages(self, user_mgt_app):
         sample_response = json.loads(OIDC_PROVIDER_CONFIG_RESPONSE)
         configs = _create_list_response(sample_response)
@@ -441,7 +466,7 @@ class TestSAMLProviderConfig:
         }
     }
 
-    @pytest.mark.parametrize('provider_id', INVALID_PROVIDER_IDS + ['oidc.provider'])
+    @pytest.mark.parametrize('provider_id', INVALID_SAML_PROVIDER_IDS)
     def test_get_invalid_provider_id(self, user_mgt_app, provider_id):
         with pytest.raises(ValueError) as excinfo:
             auth.get_saml_provider_config(provider_id, app=user_mgt_app)
@@ -460,6 +485,9 @@ def test_get(self, user_mgt_app):
 
     @pytest.mark.parametrize('invalid_opts', [
         {'provider_id': None}, {'provider_id': ''}, {'provider_id': 'oidc.provider'},
+        {'provider_id': 'saml.'}, {'provider_id': 'saml.provider.'},
+        {'provider_id': 'saml.provider/sub'}, {'provider_id': 'saml.provider?param=1'},
+        {'provider_id': 'saml.provider#frag'},
         {'idp_entity_id': None}, {'idp_entity_id': ''},
         {'sso_url': None}, {'sso_url': ''}, {'sso_url': 'not a url'},
         {'x509_certificates': None}, {'x509_certificates': []}, {'x509_certificates': 'cert'},
@@ -530,6 +558,9 @@ def test_create_empty_values(self, user_mgt_app):
     @pytest.mark.parametrize('invalid_opts', [
         {},
         {'provider_id': None}, {'provider_id': ''}, {'provider_id': 'oidc.provider'},
+        {'provider_id': 'saml.'}, {'provider_id': 'saml.provider.'},
+        {'provider_id': 'saml.provider/sub'}, {'provider_id': 'saml.provider?param=1'},
+        {'provider_id': 'saml.provider#frag'},
         {'idp_entity_id': ''},
         {'sso_url': ''}, {'sso_url': 'not a url'},
         {'x509_certificates': []}, {'x509_certificates': 'cert'},
@@ -592,7 +623,7 @@ def test_update_empty_values(self, user_mgt_app):
         got = json.loads(recorder[0].body.decode())
         assert got == {'displayName': None, 'enabled': False}
 
-    @pytest.mark.parametrize('provider_id', INVALID_PROVIDER_IDS + ['oidc.provider'])
+    @pytest.mark.parametrize('provider_id', INVALID_SAML_PROVIDER_IDS)
     def test_delete_invalid_provider_id(self, user_mgt_app, provider_id):
         with pytest.raises(ValueError) as excinfo:
             auth.delete_saml_provider_config(provider_id, app=user_mgt_app)
@@ -642,6 +673,17 @@ def test_list_single_page(self, user_mgt_app):
         _assert_request(
             recorder[0], 'GET', f'{USER_MGT_URLS["PREFIX"]}/inboundSamlConfigs?pageSize=100')
 
+    def test_list_page_token_encoding(self, user_mgt_app):
+        recorder = _instrument_provider_mgt(user_mgt_app, 200, LIST_SAML_PROVIDER_CONFIGS_RESPONSE)
+        page = auth.list_saml_provider_configs(page_token='token+1=2&3#4', app=user_mgt_app)
+
+        self._assert_page(page)
+        assert len(recorder) == 1
+        _assert_request(
+            recorder[0], 'GET',
+            f'{USER_MGT_URLS["PREFIX"]}/inboundSamlConfigs'
+            '?pageSize=100&pageToken=token%2B1%3D2%263%234')
+
     def test_list_multiple_pages(self, user_mgt_app):
         sample_response = json.loads(SAML_PROVIDER_CONFIG_RESPONSE)
         configs = _create_list_response(sample_response)
diff --git a/tests/test_project_management.py b/tests/test_project_management.py
index 89e48c2e..0536c291 100644
--- a/tests/test_project_management.py
+++ b/tests/test_project_management.py
@@ -841,6 +841,32 @@ def test_list_android_apps_multiple_pages(self):
         self._assert_request_is_correct(recorder[0], 'GET', TestListAndroidApps._LISTING_URL)
         self._assert_request_is_correct(recorder[1], 'GET', TestListAndroidApps._LISTING_PAGE_2_URL)
 
+    def test_list_android_apps_page_token_encoding(self):
+        page_1_response = json.dumps({
+            'apps': [{
+                'name': 'projects/test-project-id/androidApps/1:12345678:android:deadbeef',
+                'appId': '1:12345678:android:deadbeef',
+                'displayName': 'My Android App',
+                'projectId': 'test-project-id',
+                'packageName': 'com.hello.world.android',
+            }],
+            'nextPageToken': 'token+1=2&3#4',
+        })
+        recorder = self._instrument_service(
+            statuses=[200, 200],
+            responses=[page_1_response, LIST_ANDROID_APPS_PAGE_2_RESPONSE])
+
+        android_apps = project_management.list_android_apps()
+
+        assert len(android_apps) == 2
+        assert len(recorder) == 2
+        self._assert_request_is_correct(recorder[0], 'GET', TestListAndroidApps._LISTING_URL)
+        expected_page_2_url = (
+            'https://firebase.googleapis.com/v1beta1/projects/test-project-id/'
+            'androidApps?pageToken=token%2B1%3D2%263%234&pageSize=100'
+        )
+        self._assert_request_is_correct(recorder[1], 'GET', expected_page_2_url)
+
     def test_list_android_apps_multiple_pages_rpc_error(self):
         recorder = self._instrument_service(
             statuses=[200, 503],
diff --git a/tests/test_tenant_mgt.py b/tests/test_tenant_mgt.py
index 900faa37..491924a3 100644
--- a/tests/test_tenant_mgt.py
+++ b/tests/test_tenant_mgt.py
@@ -107,7 +107,11 @@
 LIST_OIDC_PROVIDER_CONFIGS_RESPONSE = testutils.resource('list_oidc_provider_configs.json')
 LIST_SAML_PROVIDER_CONFIGS_RESPONSE = testutils.resource('list_saml_provider_configs.json')
 
-INVALID_TENANT_IDS = [None, '', 0, 1, True, False, [], tuple(), {}]
+INVALID_TENANT_IDS = [
+    None, '', '.', '..', '..\n', 'tenant.', 'tenant.\n', 'tenant\n',
+    'tenant/sub', 'tenant?param=1', 'tenant#frag',
+    0, 1, True, False, [], tuple(), {},
+]
 INVALID_BOOLEANS = ['', 1, 0, [], tuple(), {}]
 
 USER_MGT_URL_PREFIX = 'https://identitytoolkit.googleapis.com/v1/projects/mock-project-id'
@@ -304,7 +308,7 @@ class TestUpdateTenant:
     def test_invalid_tenant_id(self, tenant_id, tenant_mgt_app):
         with pytest.raises(ValueError) as excinfo:
             tenant_mgt.update_tenant(tenant_id, display_name='My Tenant', app=tenant_mgt_app)
-        assert str(excinfo.value).startswith('Tenant ID must be a non-empty string')
+        assert str(excinfo.value).startswith('Invalid tenant ID')
 
     @pytest.mark.parametrize('display_name', [True, False, 1, 0, [], tuple(), {}])
     def test_invalid_display_name_type(self, display_name, tenant_mgt_app):