CVE-2023-34110: Generation of Error Message Containing Sensitive Information (CWE-209)
value: Flask-AppBuilder is an application development framework, built on top of Flask. Prior to version 4.3.2, an authenticated malicious actor with Admin privileges, could by adding a special character on the add, edit User forms trigger a database error, this error is surfaced back to this actor on the UI. On certain database engines this error can include the entire user row including the pbkdf2:sha256 hashed password. This vulnerability has been fixed in version 4.3.2.

CWE description: The product generates an error message that includes sensitive information about its environment, users, or associated data.
CVSS v3 base score: 2.7
Published: 2023-06-22T23:15Z
Repository: https://github.com/dpgaspar/Flask-AppBuilder
Language: Python
Files changed: flask_appbuilder/babel/manager.py, flask_appbuilder/models/base.py, flask_appbuilder/models/sqla/interface.py, flask_appbuilder/tests/security/test_mvc_security.py
Commit date: 2023-05-19 15:51:38 +0100
Commit message: fix: CRUD MVC log message (#2045)

* fix: CRUD MVC log message

* lint

* add tests

* fix lint and tests

* fix lint and tests

* revert babel name refactor

* fix lint
Diff: diff --git a/flask_appbuilder/babel/manager.py b/flask_appbuilder/babel/manager.py
index 523d8681e..b011e0102 100644
--- a/flask_appbuilder/babel/manager.py
+++ b/flask_appbuilder/babel/manager.py
@@ -1,13 +1,12 @@
 import os
 
 from flask import has_request_context, request, session
+from flask_appbuilder.babel.views import LocaleView
+from flask_appbuilder.basemanager import BaseManager
 from flask_babel import Babel
 
-from .views import LocaleView
-from ..basemanager import BaseManager
-
 
 class BabelManager(BaseManager):
 
     babel = None
     locale_view = None
diff --git a/flask_appbuilder/models/base.py b/flask_appbuilder/models/base.py
index 6a1529e6e..3d4273038 100644
--- a/flask_appbuilder/models/base.py
+++ b/flask_appbuilder/models/base.py
@@ -39,10 +39,12 @@ class BaseInterface:
     edit_integrity_error_message = lazy_gettext(
         "Integrity error, probably unique constraint"
     )
     general_error_message = lazy_gettext("General Error")
 
+    database_error_message = lazy_gettext("Database Error")
+
     """ Tuple with message and text with severity type ex: ("Added Row", "info") """
     message = ()
 
     def __init__(self, obj: Type[Any]):
         self.obj = obj
@@ -101,17 +103,17 @@ class BaseInterface:
     def get_values_item(self, item, show_columns):
         return [self._get_attr_value(item, col) for col in show_columns]
 
     def _get_values(self, lst, list_columns):
         """
-            Get Values: formats values for list template.
-            returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
+        Get Values: formats values for list template.
+        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
 
-            :param lst:
-                The list of item objects from query
-            :param list_columns:
-                The list of columns to include
+        :param lst:
+            The list of item objects from query
+        :param list_columns:
+            The list of columns to include
         """
         retlst = []
         for item in lst:
             retdict = {}
             for col in list_columns:
@@ -119,27 +121,27 @@ class BaseInterface:
             retlst.append(retdict)
         return retlst
 
     def get_values(self, lst, list_columns):
         """
-            Get Values: formats values for list template.
-            returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
+        Get Values: formats values for list template.
+        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
 
-            :param lst:
-                The list of item objects from query
-            :param list_columns:
-                The list of columns to include
+        :param lst:
+            The list of item objects from query
+        :param list_columns:
+            The list of columns to include
         """
         for item in lst:
             retdict = {}
             for col in list_columns:
                 retdict[col] = self._get_attr_value(item, col)
             yield retdict
 
     def get_values_json(self, lst, list_columns):
         """
-            Converts list of objects from query to JSON
+        Converts list of objects from query to JSON
         """
         result = []
         for item in self.get_values(lst, list_columns):
             for key, value in list(item.items()):
                 if isinstance(value, datetime.datetime) or isinstance(
@@ -262,42 +264,42 @@ class BaseInterface:
     -----------------------------------------
     """
 
     def add(self, item):
         """
-            Adds object
+        Adds object
         """
         raise NotImplementedError
 
     def edit(self, item):
         """
-            Edit (change) object
+        Edit (change) object
         """
         raise NotImplementedError
 
     def delete(self, item):
         """
-            Deletes object
+        Deletes object
         """
         raise NotImplementedError
 
     def get_col_default(self, col_name):
         pass
 
     def get_keys(self, lst):
         """
-            return a list of pk values from object list
+        return a list of pk values from object list
         """
         pk_name = self.get_pk_name()
         if self.is_pk_composite():
             return [[getattr(item, pk) for pk in pk_name] for item in lst]
         else:
             return [getattr(item, pk_name) for item in lst]
 
     def get_pk_name(self):
         """
-            Returns the primary key name
+        Returns the primary key name
         """
         raise NotImplementedError
 
     def get_pk_value(self, item):
         pk_name = self.get_pk_name()
@@ -306,25 +308,25 @@ class BaseInterface:
         else:
             return getattr(item, pk_name)
 
     def get(self, pk, filter=None):
         """
-            return the record from key, you can optionally pass filters
-            if pk exits on the db but filters exclude it it will return none.
+        return the record from key, you can optionally pass filters
+        if pk exits on the db but filters exclude it it will return none.
         """
         pass
 
     def get_related_model(self, prop):
         raise NotImplementedError
 
     def get_related_interface(self, col_name):
         """
-            Returns a BaseInterface for the related model
-            of column name.
+        Returns a BaseInterface for the related model
+        of column name.
 
-            :param col_name: Column name with relation
-            :return: BaseInterface
+        :param col_name: Column name with relation
+        :return: BaseInterface
         """
         raise NotImplementedError
 
     def get_related_obj(self, col_name, value):
         raise NotImplementedError
@@ -332,29 +334,29 @@ class BaseInterface:
     def get_related_fk(self, model):
         raise NotImplementedError
 
     def get_columns_list(self):
         """
-            Returns a list of all the columns names
+        Returns a list of all the columns names
         """
         return []
 
     def get_user_columns_list(self):
         """
-            Returns a list of user viewable columns names
+        Returns a list of user viewable columns names
         """
         return self.get_columns_list()
 
     def get_search_columns_list(self):
         """
-            Returns a list of searchable columns names
+        Returns a list of searchable columns names
         """
         return []
 
     def get_order_columns_list(self, list_columns=None):
         """
-            Returns a list of order columns names
+        Returns a list of order columns names
         """
         return []
 
     def get_relation_fk(self, prop):
         pass
diff --git a/flask_appbuilder/models/sqla/interface.py b/flask_appbuilder/models/sqla/interface.py
index c98372d02..b2e7517f4 100644
--- a/flask_appbuilder/models/sqla/interface.py
+++ b/flask_appbuilder/models/sqla/interface.py
@@ -1,16 +1,13 @@
 # -*- coding: utf-8 -*-
 from contextlib import suppress
 import logging
-import sys
 from typing import Any, Dict, List, Optional, Tuple, Type, Union
 
 from flask_appbuilder._compat import as_unicode
 from flask_appbuilder.const import (
-    LOGMSG_ERR_DBI_ADD_GENERIC,
     LOGMSG_ERR_DBI_DEL_GENERIC,
-    LOGMSG_ERR_DBI_EDIT_GENERIC,
     LOGMSG_WAR_DBI_ADD_INTEGRITY,
     LOGMSG_WAR_DBI_DEL_INTEGRITY,
     LOGMSG_WAR_DBI_EDIT_INTEGRITY,
 )
 from flask_appbuilder.exceptions import InterfaceQueryWithoutSession
@@ -734,15 +731,12 @@ class SQLAInterface(BaseInterface):
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
         except Exception as e:
-            self.message = (
-                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
-                "danger",
-            )
-            log.exception(LOGMSG_ERR_DBI_ADD_GENERIC.format(str(e)))
+            self.message = (as_unicode(self.database_error_message), "danger")
+            log.exception("Database error")
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
 
@@ -758,15 +752,12 @@ class SQLAInterface(BaseInterface):
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
         except Exception as e:
-            self.message = (
-                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
-                "danger",
-            )
-            log.exception(LOGMSG_ERR_DBI_EDIT_GENERIC.format(str(e)))
+            self.message = (as_unicode(self.database_error_message), "danger")
+            log.exception("Database error")
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
 
@@ -783,15 +774,12 @@ class SQLAInterface(BaseInterface):
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
         except Exception as e:
-            self.message = (
-                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
-                "danger",
-            )
-            log.exception(LOGMSG_ERR_DBI_DEL_GENERIC.format(str(e)))
+            self.message = (as_unicode(self.database_error_message), "danger")
+            log.exception("Database error")
             self.session.rollback()
             if raise_exception:
                 raise e
             return False
 
@@ -807,14 +795,11 @@ class SQLAInterface(BaseInterface):
             self.message = (as_unicode(self.delete_integrity_error_message), "warning")
             log.warning(LOGMSG_WAR_DBI_DEL_INTEGRITY.format(str(e)))
             self.session.rollback()
             return False
         except Exception as e:
-            self.message = (
-                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
-                "danger",
-            )
+            self.message = (as_unicode(self.database_error_message), "danger")
             log.exception(LOGMSG_ERR_DBI_DEL_GENERIC.format(str(e)))
             self.session.rollback()
             return False
 
     """
diff --git a/flask_appbuilder/tests/security/test_mvc_security.py b/flask_appbuilder/tests/security/test_mvc_security.py
index e3dc5c6cb..b268211e8 100644
--- a/flask_appbuilder/tests/security/test_mvc_security.py
+++ b/flask_appbuilder/tests/security/test_mvc_security.py
@@ -1,5 +1,7 @@
+from unittest.mock import patch
+
 from flask_appbuilder import ModelView
 from flask_appbuilder.exceptions import PasswordComplexityValidationError
 from flask_appbuilder.models.sqla.filters import FilterEqual
 from flask_appbuilder.models.sqla.interface import SQLAInterface
 from flask_appbuilder.security.sqla.models import User
@@ -420,5 +422,119 @@ class MVCSecurityTestCase(BaseMVCTestCase):
             .filter(User.username == "from test 1-1")
             .one_or_none()
         )
         self.db.session.delete(user)
         self.db.session.commit()
+
+    def test_edit_user(self):
+        """
+        Test edit user
+        """
+        client = self.app.test_client()
+        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)
+
+        _tmp_user = self.create_user(
+            self.appbuilder,
+            "tmp_user",
+            "password1",
+            "",
+            first_name="tmp",
+            last_name="user",
+            email="tmp@fab.org",
+            role_names=["Admin"],
+        )
+
+        # use all required params
+        rv = client.get(f"/users/edit/{_tmp_user.id}", follow_redirects=True)
+        data = rv.data.decode("utf-8")
+        self.assertIn("Edit User", data)
+        rv = client.post(
+            f"/users/edit/{_tmp_user.id}",
+            data=dict(
+                first_name=_tmp_user.first_name,
+                last_name=_tmp_user.last_name,
+                username=_tmp_user.username,
+                email="changed@changed.org",
+                roles=_tmp_user.roles[0].id,
+            ),
+            follow_redirects=True,
+        )
+        data = rv.data.decode("utf-8")
+        self.assertIn("Changed Row", data)
+
+        user = (
+            self.db.session.query(User)
+            .filter(User.username == _tmp_user.username)
+            .one_or_none()
+        )
+
+        assert user.email == "changed@changed.org"
+        self.db.session.delete(user)
+        self.db.session.commit()
+
+    def test_edit_user_email_validation(self):
+        """
+        Test edit user with email not null validation
+        """
+        client = self.app.test_client()
+        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)
+
+        read_ony_user: User = (
+            self.db.session.query(User)
+            .filter(User.username == USERNAME_READONLY)
+            .one_or_none()
+        )
+
+        # use all required params
+        rv = client.get(f"/users/edit/{read_ony_user.id}", follow_redirects=True)
+        data = rv.data.decode("utf-8")
+        self.assertIn("Edit User", data)
+        rv = client.post(
+            f"/users/edit/{read_ony_user.id}",
+            data=dict(
+                first_name=read_ony_user.first_name,
+                last_name=read_ony_user.last_name,
+                username=read_ony_user.username,
+                email=None,
+                roles=read_ony_user.roles[0].id,
+            ),
+            follow_redirects=True,
+        )
+        data = rv.data.decode("utf-8")
+        self.assertIn("This field is required", data)
+
+    def test_edit_user_db_fail(self):
+        """
+        Test edit user with DB fail
+        """
+        client = self.app.test_client()
+        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)
+
+        read_ony_user: User = (
+            self.db.session.query(User)
+            .filter(User.username == USERNAME_READONLY)
+            .one_or_none()
+        )
+
+        # use all required params
+        rv = client.get(f"/users/edit/{read_ony_user.id}", follow_redirects=True)
+        data = rv.data.decode("utf-8")
+        self.assertIn("Edit User", data)
+
+        with patch.object(self.appbuilder.session, "merge") as mock_merge:
+            with patch.object(self.appbuilder.sm, "has_access", return_value=True) as _:
+                mock_merge.side_effect = Exception("BANG!")
+
+                rv = client.post(
+                    f"/users/edit/{read_ony_user.id}",
+                    data=dict(
+                        first_name=read_ony_user.first_name,
+                        last_name=read_ony_user.last_name,
+                        username=read_ony_user.username,
+                        email="changed@changed.org",
+                        roles=read_ony_user.roles[0].id,
+                    ),
+                    follow_redirects=True,
+                )
+
+                data = rv.data.decode("utf-8")
+                self.assertIn("Database Error", data)

Vulnerable code: from .views import LocaleView
from ..basemanager import BaseManager

            Get Values: formats values for list template.
            returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
            :param lst:
                The list of item objects from query
            :param list_columns:
                The list of columns to include
            Get Values: formats values for list template.
            returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
            :param lst:
                The list of item objects from query
            :param list_columns:
                The list of columns to include
            Converts list of objects from query to JSON
            Adds object
            Edit (change) object
            Deletes object
            return a list of pk values from object list
            Returns the primary key name
            return the record from key, you can optionally pass filters
            if pk exits on the db but filters exclude it it will return none.
            Returns a BaseInterface for the related model
            of column name.
            :param col_name: Column name with relation
            :return: BaseInterface
            Returns a list of all the columns names
            Returns a list of user viewable columns names
            Returns a list of searchable columns names
            Returns a list of order columns names
import sys
    LOGMSG_ERR_DBI_ADD_GENERIC,
    LOGMSG_ERR_DBI_EDIT_GENERIC,
            self.message = (
                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
                "danger",
            )
            log.exception(LOGMSG_ERR_DBI_ADD_GENERIC.format(str(e)))
            self.message = (
                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
                "danger",
            )
            log.exception(LOGMSG_ERR_DBI_EDIT_GENERIC.format(str(e)))
            self.message = (
                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
                "danger",
            )
            log.exception(LOGMSG_ERR_DBI_DEL_GENERIC.format(str(e)))
            self.message = (
                as_unicode(self.general_error_message + " " + str(sys.exc_info()[0])),
                "danger",
            )
Fixed code: from flask_appbuilder.babel.views import LocaleView
from flask_appbuilder.basemanager import BaseManager
    database_error_message = lazy_gettext("Database Error")

        Get Values: formats values for list template.
        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
        :param lst:
            The list of item objects from query
        :param list_columns:
            The list of columns to include
        Get Values: formats values for list template.
        returns [{'col_name':'col_value',....},{'col_name':'col_value',....}]
        :param lst:
            The list of item objects from query
        :param list_columns:
            The list of columns to include
        Converts list of objects from query to JSON
        Adds object
        Edit (change) object
        Deletes object
        return a list of pk values from object list
        Returns the primary key name
        return the record from key, you can optionally pass filters
        if pk exits on the db but filters exclude it it will return none.
        Returns a BaseInterface for the related model
        of column name.
        :param col_name: Column name with relation
        :return: BaseInterface
        Returns a list of all the columns names
        Returns a list of user viewable columns names
        Returns a list of searchable columns names
        Returns a list of order columns names
            self.message = (as_unicode(self.database_error_message), "danger")
            log.exception("Database error")
            self.message = (as_unicode(self.database_error_message), "danger")
            log.exception("Database error")
            self.message = (as_unicode(self.database_error_message), "danger")
            log.exception("Database error")
            self.message = (as_unicode(self.database_error_message), "danger")
from unittest.mock import patch


    def test_edit_user(self):
        """
        Test edit user
        """
        client = self.app.test_client()
        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)

        _tmp_user = self.create_user(
            self.appbuilder,
            "tmp_user",
            "password1",
            "",
            first_name="tmp",
            last_name="user",
            email="tmp@fab.org",
            role_names=["Admin"],
        )

        # use all required params
        rv = client.get(f"/users/edit/{_tmp_user.id}", follow_redirects=True)
        data = rv.data.decode("utf-8")
        self.assertIn("Edit User", data)
        rv = client.post(
            f"/users/edit/{_tmp_user.id}",
            data=dict(
                first_name=_tmp_user.first_name,
                last_name=_tmp_user.last_name,
                username=_tmp_user.username,
                email="changed@changed.org",
                roles=_tmp_user.roles[0].id,
            ),
            follow_redirects=True,
        )
        data = rv.data.decode("utf-8")
        self.assertIn("Changed Row", data)

        user = (
            self.db.session.query(User)
            .filter(User.username == _tmp_user.username)
            .one_or_none()
        )

        assert user.email == "changed@changed.org"
        self.db.session.delete(user)
        self.db.session.commit()

    def test_edit_user_email_validation(self):
        """
        Test edit user with email not null validation
        """
        client = self.app.test_client()
        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)

        read_ony_user: User = (
            self.db.session.query(User)
            .filter(User.username == USERNAME_READONLY)
            .one_or_none()
        )

        # use all required params
        rv = client.get(f"/users/edit/{read_ony_user.id}", follow_redirects=True)
        data = rv.data.decode("utf-8")
        self.assertIn("Edit User", data)
        rv = client.post(
            f"/users/edit/{read_ony_user.id}",
            data=dict(
                first_name=read_ony_user.first_name,
                last_name=read_ony_user.last_name,
                username=read_ony_user.username,
                email=None,
                roles=read_ony_user.roles[0].id,
            ),
            follow_redirects=True,
        )
        data = rv.data.decode("utf-8")
        self.assertIn("This field is required", data)

    def test_edit_user_db_fail(self):
        """
        Test edit user with DB fail
        """
        client = self.app.test_client()
        _ = self.browser_login(client, USERNAME_ADMIN, PASSWORD_ADMIN)

        read_ony_user: User = (
            self.db.session.query(User)
            .filter(User.username == USERNAME_READONLY)
            .one_or_none()
        )

        # use all required params
        rv = client.get(f"/users/edit/{read_ony_user.id}", follow_redirects=True)
        data = rv.data.decode("utf-8")
        self.assertIn("Edit User", data)

        with patch.object(self.appbuilder.session, "merge") as mock_merge:
            with patch.object(self.appbuilder.sm, "has_access", return_value=True) as _:
                mock_merge.side_effect = Exception("BANG!")

                rv = client.post(
                    f"/users/edit/{read_ony_user.id}",
                    data=dict(
                        first_name=read_ony_user.first_name,
                        last_name=read_ony_user.last_name,
                        username=read_ony_user.username,
                        email="changed@changed.org",
                        roles=read_ony_user.roles[0].id,
                    ),
                    follow_redirects=True,
                )

                data = rv.data.decode("utf-8")
                self.assertIn("Database Error", data)