feat(auth): Adding SAMLProviderConfig type and the getter method

firebase_admin/_auth_providers.py

@@ -0,0 +1,100 @@
+# Copyright 2020 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 auth providers management sub module."""
+
+import requests
+
+from firebase_admin import _auth_utils
+
+
+class ProviderConfig:
+    """Parent type for all authentication provider config types."""
+
+    def __init__(self, data):
+        self._data = data
+
+    @property
+    def provider_id(self):
+        name = self._data['name']
+        return name.split('/')[-1]
+
+    @property
+    def display_name(self):
+        return self._data.get('displayName')
+
+    @property
+    def enabled(self):
+        return self._data['enabled']
+
+
+class SAMLProviderConfig(ProviderConfig):
+    """Represents he SAML auth provider configuration.
+
+    See http://docs.oasis-open.org/security/saml/Post2.0/sstc-saml-tech-overview-2.0.html."""
+
+    @property
+    def idp_entity_id(self):
+        return self._data.get('idpConfig', {})['idpEntityId']
+
+    @property
+    def sso_url(self):
+        return self._data.get('idpConfig', {})['ssoUrl']
+
+    @property
+    def x509_certificates(self):
+        certs = self._data.get('idpConfig', {})['idpCertificates']
+        return [c['x509Certificate'] for c in certs]
+
+    @property
+    def request_signing_enabled(self):
+        return self._data.get('idpConfig', {})['signRequest']
+
+    @property
+    def callback_url(self):
+        return self._data.get('spConfig', {})['callbackUri']
+
+    @property
+    def rp_entity_id(self):
+        return self._data.get('spConfig', {})['spEntityId']
+
+
+class ProviderConfigClient:
+    """Client for managing Auth provider configurations."""
+
+    PROVIDER_CONFIG_URL = 'https://identitytoolkit.googleapis.com/v2beta1'
+
+    def __init__(self, http_client, project_id, tenant_id=None):
+        self.http_client = http_client
+        self.base_url = '{0}/projects/{1}'.format(self.PROVIDER_CONFIG_URL, project_id)
+        if tenant_id:
+            self.base_url += '/tenants/{0}'.format(tenant_id)
+
+    def get_saml_provider_config(self, provider_id):
+        if not isinstance(provider_id, str):
+            raise ValueError(
+                'Invalid SAML provider ID: {0}. Provider ID must be a non-empty string.'.format(
+                    provider_id))
+        if not provider_id.startswith('saml.'):
+            raise ValueError('Invalid SAML provider ID: {0}.'.format(provider_id))
+
+        body = self._make_request('get', '/inboundSamlConfigs/{0}'.format(provider_id))
+        return SAMLProviderConfig(body)
+
+    def _make_request(self, method, path, body=None):
+        url = '{0}{1}'.format(self.base_url, path)
+        try:
+            return self.http_client.body(method, url, json=body)
+        except requests.exceptions.RequestException as error:
+            raise _auth_utils.handle_auth_backend_error(error)

firebase_admin/_auth_utils.py

@@ -307,7 +307,17 @@ def __init__(self, message):
         exceptions.InvalidArgumentError.__init__(self, message)
 
 
+class ConfigurationNotFoundError(exceptions.NotFoundError):
+    """No auth provider found for the specified identifier."""
+
+    default_message = 'No auth provider found for the given identifier'
+
+    def __init__(self, message, cause=None, http_response=None):
+        exceptions.NotFoundError.__init__(self, message, cause, http_response)
+
+
 _CODE_TO_EXC_TYPE = {
+    'CONFIGURATION_NOT_FOUND': ConfigurationNotFoundError,
     'DUPLICATE_EMAIL': EmailAlreadyExistsError,
     'DUPLICATE_LOCAL_ID': UidAlreadyExistsError,
     'EMAIL_EXISTS': EmailAlreadyExistsError,

firebase_admin/_token_gen.py

@@ -82,10 +82,13 @@ def from_iam(cls, request, google_cred, service_account):
 class TokenGenerator:
     """Generates custom tokens and session cookies."""
 
-    def __init__(self, app, client):
+    ID_TOOLKIT_URL = 'https://identitytoolkit.googleapis.com/v1'
+
+    def __init__(self, app, http_client):
         self.app = app
-        self.client = client
+        self.http_client = http_client
         self.request = transport.requests.Request()
+        self.base_url = '{0}/projects/{1}'.format(self.ID_TOOLKIT_URL, app.project_id)
         self._signing_provider = None
 
     def _init_signing_provider(self):
@@ -192,13 +195,13 @@ def create_session_cookie(self, id_token, expires_in):
             raise ValueError('Illegal expiry duration: {0}. Duration must be at most {1} '
                              'seconds.'.format(expires_in, MAX_SESSION_COOKIE_DURATION_SECONDS))
 
+        url = '{0}:createSessionCookie'.format(self.base_url)
         payload = {
             'idToken': id_token,
             'validDuration': expires_in,
         }
         try:
-            body, http_resp = self.client.body_and_response(
-                'post', ':createSessionCookie', json=payload)
+            body, http_resp = self.http_client.body_and_response('post', url, json=payload)
         except requests.exceptions.RequestException as error:
             raise _auth_utils.handle_auth_backend_error(error)
         else:

firebase_admin/_user_mgt.py

@@ -454,8 +454,13 @@ def encode_action_code_settings(settings):
 class UserManager:
     """Provides methods for interacting with the Google Identity Toolkit."""
 
-    def __init__(self, client):
-        self._client = client
+    ID_TOOLKIT_URL = 'https://identitytoolkit.googleapis.com/v1'
+
+    def __init__(self, http_client, project_id, tenant_id=None):
+        self.http_client = http_client
+        self.base_url = '{0}/projects/{1}'.format(self.ID_TOOLKIT_URL, project_id)
+        if tenant_id:
+            self.base_url += '/tenants/{0}'.format(tenant_id)
 
     def get_user(self, **kwargs):
         """Gets the user data corresponding to the provided key."""
@@ -471,17 +476,12 @@ def get_user(self, **kwargs):
         else:
             raise TypeError('Unsupported keyword arguments: {0}.'.format(kwargs))
 
-        try:
-            body, http_resp = self._client.body_and_response(
-                'post', '/accounts:lookup', json=payload)
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not body or not body.get('users'):
-                raise _auth_utils.UserNotFoundError(
-                    'No user record found for the provided {0}: {1}.'.format(key_type, key),
-                    http_response=http_resp)
-            return body['users'][0]
+        body, http_resp = self._make_request('post', '/accounts:lookup', json=payload)
+        if not body or not body.get('users'):
+            raise _auth_utils.UserNotFoundError(
+                'No user record found for the provided {0}: {1}.'.format(key_type, key),
+                http_response=http_resp)
+        return body['users'][0]
 
     def list_users(self, page_token=None, max_results=MAX_LIST_USERS_RESULTS):
         """Retrieves a batch of users."""
@@ -498,10 +498,8 @@ def list_users(self, page_token=None, max_results=MAX_LIST_USERS_RESULTS):
         payload = {'maxResults': max_results}
         if page_token:
             payload['nextPageToken'] = page_token
-        try:
-            return self._client.body('get', '/accounts:batchGet', params=payload)
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
+        body, _ = self._make_request('get', '/accounts:batchGet', params=payload)
+        return body
 
     def create_user(self, uid=None, display_name=None, email=None, phone_number=None,
                     photo_url=None, password=None, disabled=None, email_verified=None):
@@ -517,15 +515,11 @@ def create_user(self, uid=None, display_name=None, email=None, phone_number=None
             'disabled': bool(disabled) if disabled is not None else None,
         }
         payload = {k: v for k, v in payload.items() if v is not None}
-        try:
-            body, http_resp = self._client.body_and_response('post', '/accounts', json=payload)
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not body or not body.get('localId'):
-                raise _auth_utils.UnexpectedResponseError(
-                    'Failed to create new user.', http_response=http_resp)
-            return body.get('localId')
+        body, http_resp = self._make_request('post', '/accounts', json=payload)
+        if not body or not body.get('localId'):
+            raise _auth_utils.UnexpectedResponseError(
+                'Failed to create new user.', http_response=http_resp)
+        return body.get('localId')
 
     def update_user(self, uid, display_name=None, email=None, phone_number=None,
                     photo_url=None, password=None, disabled=None, email_verified=None,
@@ -568,29 +562,19 @@ def update_user(self, uid, display_name=None, email=None, phone_number=None,
             payload['customAttributes'] = _auth_utils.validate_custom_claims(json_claims)
 
         payload = {k: v for k, v in payload.items() if v is not None}
-        try:
-            body, http_resp = self._client.body_and_response(
-                'post', '/accounts:update', json=payload)
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not body or not body.get('localId'):
-                raise _auth_utils.UnexpectedResponseError(
-                    'Failed to update user: {0}.'.format(uid), http_response=http_resp)
-            return body.get('localId')
+        body, http_resp = self._make_request('post', '/accounts:update', json=payload)
+        if not body or not body.get('localId'):
+            raise _auth_utils.UnexpectedResponseError(
+                'Failed to update user: {0}.'.format(uid), http_response=http_resp)
+        return body.get('localId')
 
     def delete_user(self, uid):
         """Deletes the user identified by the specified user ID."""
         _auth_utils.validate_uid(uid, required=True)
-        try:
-            body, http_resp = self._client.body_and_response(
-                'post', '/accounts:delete', json={'localId' : uid})
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not body or not body.get('kind'):
-                raise _auth_utils.UnexpectedResponseError(
-                    'Failed to delete user: {0}.'.format(uid), http_response=http_resp)
+        body, http_resp = self._make_request('post', '/accounts:delete', json={'localId' : uid})
+        if not body or not body.get('kind'):
+            raise _auth_utils.UnexpectedResponseError(
+                'Failed to delete user: {0}.'.format(uid), http_response=http_resp)
 
     def import_users(self, users, hash_alg=None):
         """Imports the given list of users to Firebase Auth."""
@@ -609,16 +593,11 @@ def import_users(self, users, hash_alg=None):
             if not isinstance(hash_alg, _user_import.UserImportHash):
                 raise ValueError('A UserImportHash is required to import users with passwords.')
             payload.update(hash_alg.to_dict())
-        try:
-            body, http_resp = self._client.body_and_response(
-                'post', '/accounts:batchCreate', json=payload)
-        except requests.exceptions.RequestException as error:
-            raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not isinstance(body, dict):
-                raise _auth_utils.UnexpectedResponseError(
-                    'Failed to import users.', http_response=http_resp)
-            return body
+        body, http_resp = self._make_request('post', '/accounts:batchCreate', json=payload)
+        if not isinstance(body, dict):
+            raise _auth_utils.UnexpectedResponseError(
+                'Failed to import users.', http_response=http_resp)
+        return body
 
     def generate_email_action_link(self, action_type, email, action_code_settings=None):
         """Fetches the email action links for types
@@ -646,16 +625,18 @@ def generate_email_action_link(self, action_type, email, action_code_settings=No
         if action_code_settings:
             payload.update(encode_action_code_settings(action_code_settings))
 
+        body, http_resp = self._make_request('post', '/accounts:sendOobCode', json=payload)
+        if not body or not body.get('oobLink'):
+            raise _auth_utils.UnexpectedResponseError(
+                'Failed to generate email action link.', http_response=http_resp)
+        return body.get('oobLink')
+
+    def _make_request(self, method, path, **kwargs):
+        url = '{0}{1}'.format(self.base_url, path)
         try:
-            body, http_resp = self._client.body_and_response(
-                'post', '/accounts:sendOobCode', json=payload)
+            return self.http_client.body_and_response(method, url, **kwargs)
         except requests.exceptions.RequestException as error:
             raise _auth_utils.handle_auth_backend_error(error)
-        else:
-            if not body or not body.get('oobLink'):
-                raise _auth_utils.UnexpectedResponseError(
-                    'Failed to generate email action link.', http_response=http_resp)
-            return body.get('oobLink')
 
 
 class _UserIterator: