feat(auth): Add TOTP support in Project and Tenant config

firebase_admin/multi_factor_config_mgt.py

@@ -0,0 +1,224 @@
+# Copyright 2023 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Firebase multifactor configuration management module.
+
+This module contains functions for managing multifactor auth configuration at
+the project and tenant level.
+"""
+from enum import Enum
+from typing import List
+
+__all__ = [
+ 'validate_keys',
+ 'MultiFactorServerConfig',
+ 'TOTPProviderConfig',
+ 'ProviderConfig',
+ 'MultiFactorConfig',
+]
+
+
+def validate_keys(keys, valid_keys, config_name):
+ for key in keys:
+ if key not in valid_keys:
+ raise ValueError(
+ '"{0}" is not a valid "{1}" parameter.'.format(
+ key, config_name))
+
+
+class MultiFactorServerConfig:
+ """Represents multi factor configuration response received from the server and
+ converts it to user format.
+ """
+
+ def __init__(self, data):
+ if not isinstance(data, dict):
+ raise ValueError(
+ 'Invalid data argument in MultiFactorServerConfig constructor: {0}, must be a valid dict'.format(data))
+ self._data = data
+
+ @property
+ def provider_configs(self):
+ data = self._data.get('providerConfigs', None)
+ if data is not None:
+ return [self.ProviderServerConfig(d) for d in data]
+ return None
+
+ class ProviderServerConfig:
+ """Represents provider configuration response received from the server and converts
+ it to user format.
+ """
+
+ def __init__(self, data):
+ if not isinstance(data, dict):
+ raise ValueError(
+ 'Invalid data argument in ProviderServerConfig constructor: {0}'.format(data))
+ self._data = data
+
+ @property
+ def state(self):
+ return self._data.get('state', None)
+
+ @property
+ def totp_provider_config(self):
+ data = self._data.get('totpProviderConfig', None)
+ if data is not None:
+ return self.TOTPProviderServerConfig(data)
+ return None
+
+ class TOTPProviderServerConfig:
+ """Represents TOTP provider configuration response received from the server and converts
+ it to user format.
+ """
+
+ def __init__(self, data):
+ if not isinstance(data, dict):
+ raise ValueError(
+ 'Invalid data argument in TOTPProviderServerConfig'
+ ' constructor: {0}'.format(data))
+ self._data = data
+
+ @property
+ def adjacent_intervals(self):
+ return self._data.get('adjacentIntervals', None)
+
+
+class TOTPProviderConfig:
+ """A tenant or project's TOTP provider configuration."""
+
+ def __init__(self, adjacent_intervals: int = None):
+ self.adjacent_intervals: int = adjacent_intervals
+
+ def to_dict(self) -> dict:
+ data = {}
+ if self.adjacent_intervals is not None:
+ data['adjacentIntervals'] = self.adjacent_intervals
+ return data
+
+ def validate(self):
+ """Validates the configuration.
+
+ Raises:
+ ValueError: In case of an unsuccessful validation.
+ """
+ validate_keys(
+ keys=vars(self).keys(),
+ valid_keys={'adjacent_intervals'},
+ config_name='TOTPProviderConfig')
+ if self.adjacent_intervals is not None:
+ # Because bool types get converted to int here
+ # pylint: disable=C0123
+ if type(self.adjacent_intervals) is not int:
+ raise ValueError(
+ 'TOTPProviderConfig.adjacent_intervals must be an integer between'
+ ' 1 and 10 (inclusive).')
+ if not 1 <= self.adjacent_intervals <= 10:
+ raise ValueError(
+ 'totp_provider_config.adjacent_intervals must be an integer between'
+ ' 1 and 10 (inclusive).')
+
+ def build_server_request(self):
+ self.validate()
+ return self.to_dict()
+
+
+class ProviderConfig:
+ """A tenant or project's multifactor provider configuration.
+ Currently, only TOTP can be configured."""
+
+ class State(Enum):
+ ENABLED = 'ENABLED'
+ DISABLED = 'DISABLED'
+
+ def __init__(self,
+ state: State = None,
+ totp_provider_config: TOTPProviderConfig = None):
+ self.state: self.State = state
+ self.totp_provider_config: TOTPProviderConfig = totp_provider_config
+
+ def to_dict(self) -> dict:
+ data = {}
+ if self.state:
+ data['state'] = self.state.value
+ if self.totp_provider_config:
+ data['totpProviderConfig'] = self.totp_provider_config.to_dict()
+ return data
+
+ def validate(self):
+ """Validates the provider configuration.
+
+ Raises:
+ ValueError: In case of an unsuccessful validation.
+ """
+ validate_keys(
+ keys=vars(self).keys(),
+ valid_keys={
+ 'state',
+ 'totp_provider_config'},
+ config_name='ProviderConfig')
+ if self.state is None:
+ raise ValueError('ProviderConfig.state must be defined.')
+ if not isinstance(self.state, ProviderConfig.State):
+ raise ValueError(
+ 'ProviderConfig.state must be of type ProviderConfig.State.')
+ if self.totp_provider_config is None:
+ raise ValueError(
+ 'ProviderConfig.totp_provider_config must be defined.')
+ if not isinstance(self.totp_provider_config, TOTPProviderConfig):
+ raise ValueError(
+ 'ProviderConfig.totp_provider_config must be of type TOTPProviderConfig.')
+
+ def build_server_request(self):
+ self.validate()
+ return self.to_dict()
+
+
+class MultiFactorConfig:
+ """A tenant or project's multi factor configuration."""
+
+ def __init__(self,
+ provider_configs: List[ProviderConfig] = None):
+ self.provider_configs: List[ProviderConfig] = provider_configs
+
+ def to_dict(self) -> dict:
+ data = {}
+ if self.provider_configs is not None:
+ data['providerConfigs'] = [d.to_dict()
+ for d in self.provider_configs]
+ return data
+
+ def validate(self):
+ """Validates the configuration.
+
+ Raises:
+ ValueError: In case of an unsuccessful validation.
+ """
+ validate_keys(
+ keys=vars(self).keys(),
+ valid_keys={'provider_configs'},
+ config_name='MultiFactorConfig')
+ if self.provider_configs is None:
+ raise ValueError(
+ 'multi_factor_config.provider_configs must be specified')
+ if not isinstance(self.provider_configs, list) or not self.provider_configs:
+ raise ValueError(
+ 'provider_configs must be an array of type ProviderConfigs.')
+ for provider_config in self.provider_configs:
+ if not isinstance(provider_config, ProviderConfig):
+ raise ValueError(
+ 'provider_configs must be an array of type ProviderConfigs.')
+ provider_config.validate()
+
+ def build_server_request(self):
+ self.validate()
+ return self.to_dict()

firebase_admin/project_config_mgt.py

@@ -0,0 +1,135 @@
+# Copyright 2023 Google Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Firebase project configuration management module.
+
+This module contains functions for managing projects.
+"""
+
+import requests
+
+import firebase_admin
+from firebase_admin import _auth_utils
+from firebase_admin import _http_client
+from firebase_admin import _utils
+from firebase_admin.multi_factor_config_mgt import MultiFactorConfig
+from firebase_admin.multi_factor_config_mgt import MultiFactorServerConfig
+
+_PROJECT_CONFIG_MGT_ATTRIBUTE = '_project_config_mgt'
+
+__all__ = [
+ 'ProjectConfig',
+ 'get_project_config',
+ 'update_project_config',
+]
+
+
+def get_project_config(app=None):
+ """Gets the project config corresponding to the current project_id.
+
+ Args:
+ app: An App instance (optional).
+
+ Returns:
+ Project: A project object.
+
+ Raises:
+ ValueError: If the project ID is None, empty or not a string.
+ ProjectNotFoundError: If no project exists by the given ID.
+ FirebaseError: If an error occurs while retrieving the project.
+ """
+ project_config_mgt_service = _get_project_config_mgt_service(app)
+ return project_config_mgt_service.get_project_config()
+
+def update_project_config(multi_factor_config: MultiFactorConfig = None, app=None):
+ """Update the project config with the given options.
+
+ Args:
+ multi_factor_config: Updated multi-factor authentication configuration
+ (optional)
+ app: An App instance (optional).
+ Returns:
+ Project: An updated ProjectConfig object.
+ Raises:
+ ValueError: If any of the given arguments are invalid.
+ FirebaseError: If an error occurs while updating the project.
+ """
+ project_config_mgt_service = _get_project_config_mgt_service(app)
+ return project_config_mgt_service.update_project_config(multi_factor_config=multi_factor_config)
+
+
+def _get_project_config_mgt_service(app):
+ return _utils.get_app_service(app, _PROJECT_CONFIG_MGT_ATTRIBUTE,
+ _ProjectConfigManagementService)
+
+class ProjectConfig:
+ """Represents a project config in an application.
+ """
+
+ def __init__(self, data):
+ if not isinstance(data, dict):
+ raise ValueError(
+ 'Invalid data argument in Project constructor: {0}'.format(data))
+ self._data = data
+
+ @property
+ def multi_factor_config(self):
+ data = self._data.get('mfa')
+ if data:
+ return MultiFactorServerConfig(data)
+ return None
+
+class _ProjectConfigManagementService:
+ """Firebase project management service."""
+
+ PROJECT_CONFIG_MGT_URL = 'https://identitytoolkit.googleapis.com/v2/projects'
+
+ def __init__(self, app):
+ credential = app.credential.get_credential()
+ version_header = 'Python/Admin/{0}'.format(firebase_admin.__version__)
+ base_url = '{0}/{1}/config'.format(
+ self.PROJECT_CONFIG_MGT_URL, app.project_id)
+ self.app = app
+ self.client = _http_client.JsonHttpClient(
+ credential=credential, base_url=base_url, headers={'X-Client-Version': version_header})
+
+ def get_project_config(self) -> ProjectConfig:
+ """Gets the project config"""
+ try:
+ body = self.client.body('get', url='')
+ except requests.exceptions.RequestException as error:
+ raise _auth_utils.handle_auth_backend_error(error)
+ else:
+ return ProjectConfig(body)
+
+ def update_project_config(self, multi_factor_config: MultiFactorConfig = None) -> ProjectConfig:
+ """Updates the specified project with the given parameters."""
+
+ payload = {}
+ if multi_factor_config is not None:
+ if not isinstance(multi_factor_config, MultiFactorConfig):
+ raise ValueError('multi_factor_config must be of type MultiFactorConfig.')
+ payload['mfa'] = multi_factor_config.build_server_request()
+ if not payload:
+ raise ValueError(
+ 'At least one parameter must be specified for update.')
+
+ update_mask = ','.join(_auth_utils.build_update_mask(payload))
+ params = 'updateMask={0}'.format(update_mask)
+ try:
+ body = self.client.body(
+ 'patch', url='', json=payload, params=params)
+ except requests.exceptions.RequestException as error:
+ raise _auth_utils.handle_auth_backend_error(error)
+ else:
+ return ProjectConfig(body)