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

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)

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
 

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

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

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)

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}')

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)

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],

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