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

firebase_admin/multi_factor_config_mgt.py

@@ -0,0 +1,225 @@
+# 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 various multifactor configurations 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}'.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:
+    """Represents a TOTP Provider Configuration to be specified for a tenant or project."""
+
+    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 a given totp_provider_config object.
+
+        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(
+                    'totp_provider_config.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:
+    """Represents a provider configuration for tenant or project.
+    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 a provider_config object.
+
+        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('provider_config.state must be defined.')
+        if not isinstance(self.state, ProviderConfig.State):
+            raise ValueError(
+                'provider_config.state must be of type ProviderConfig.State.')
+        if self.totp_provider_config is None:
+            raise ValueError(
+                'provider_config.totp_provider_config must be defined.')
+        if not isinstance(self.totp_provider_config, TOTPProviderConfig):
+            raise ValueError(
+                'provider_configs.totp_provider_config must be of type TOTPProviderConfig.')
+
+    def build_server_request(self):
+        self.validate()
+        return self.to_dict()
+
+
+class MultiFactorConfig:
+    """Represents a multi factor configuration for tenant or project
+    """
+
+    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 a given multi_factor_config object.
+
+        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,134 @@
+# 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 various project operations like update and create
+"""
+
+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 given 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)