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