CVE-2023-34110 (Python, published 2023-06-22T23:15Z)
Description: 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.

Severity: nan (CVSS v3 base score: 2.7)
Weakness: CWE-209 \u2014 Generation of Error Message Containing Sensitive Information
The product generates an error message that includes sensitive information about its environment, users, or associated data.
Repository: https://github.com/dpgaspar/Flask-AppBuilder
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 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

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)