CVE-2024-25128: Insufficient Information (NVD-CWE-noinfo)
value: Flask-AppBuilder is an application development framework, built on top of Flask. When Flask-AppBuilder is set to AUTH_TYPE AUTH_OID, it allows an attacker to forge an HTTP request, that could deceive the backend into using any requested OpenID service. This vulnerability could grant an attacker unauthorised privilege access if a custom OpenID service is deployed by the attacker and accessible by the backend. This vulnerability is only exploitable when the application is using the OpenID 2.0 authorization protocol. Upgrade to Flask-AppBuilder 4.3.11 to fix the vulnerability.
CWE description: There is insufficient information about the issue to classify it; details are unkown or unspecified.
Published: 2024-02-29T01:44Z
Repository: https://github.com/dpgaspar/Flask-AppBuilder
Language: Python
Files changed: flask_appbuilder/security/manager.py, flask_appbuilder/security/views.py, flask_appbuilder/templates/appbuilder/general/security/login_oid.html, tests/config_oid.py, tests/test_mvc_oauth.py, tests/test_mvc_oid.py, tests/test_security_api.py
Commit date: 2024-02-06 20:51:56 +0000
Commit message: fix: openID provider validation flow (#2186)

* fix: openID provider validation flow

* remove test cleanup
Diff: diff --git a/flask_appbuilder/security/manager.py b/flask_appbuilder/security/manager.py
index 80a91b8cf..8f86b7ee5 100644
--- a/flask_appbuilder/security/manager.py
+++ b/flask_appbuilder/security/manager.py
@@ -1445,10 +1445,18 @@ class BaseSecurityManager(AbstractSecurityManager):
                 db_role_ids.append(role.id)
 
         # If it's not a builtin role check against database store roles
         return self.exist_permission_on_roles(view_name, permission_name, db_role_ids)
 
+    def get_oid_identity_url(self, provider_name: str) -> Optional[str]:
+        """
+        Returns the OIDC identity provider URL
+        """
+        for provider in self.openid_providers:
+            if provider.get("name") == provider_name:
+                return provider.get("url")
+
     def get_user_roles(self, user) -> List[object]:
         """
         Get current user roles, if user is not authenticated returns the public role
         """
         if not user.is_authenticated:
diff --git a/flask_appbuilder/security/views.py b/flask_appbuilder/security/views.py
index fd4db6741..c38e23f39 100644
--- a/flask_appbuilder/security/views.py
+++ b/flask_appbuilder/security/views.py
@@ -563,12 +563,16 @@ class AuthOIDView(AuthView):
             if g.user is not None and g.user.is_authenticated:
                 return redirect(self.appbuilder.get_url_for_index)
             form = LoginForm_oid()
             if form.validate_on_submit():
                 session["remember_me"] = form.remember_me.data
+                identity_url = self.appbuilder.sm.get_oid_identity_url(form.openid.data)
+                if identity_url is None:
+                    flash(as_unicode(self.invalid_login_message), "warning")
+                    return redirect(self.appbuilder.get_url_for_login)
                 return self.appbuilder.sm.oid.try_login(
-                    form.openid.data,
+                    identity_url,
                     ask_for=self.oid_ask_for,
                     ask_for_optional=self.oid_ask_for_optional,
                 )
             return self.render_template(
                 self.login_template,
diff --git a/flask_appbuilder/templates/appbuilder/general/security/login_oid.html b/flask_appbuilder/templates/appbuilder/general/security/login_oid.html
index 9e7a97280..77d8457bb 100644
--- a/flask_appbuilder/templates/appbuilder/general/security/login_oid.html
+++ b/flask_appbuilder/templates/appbuilder/general/security/login_oid.html
@@ -34,17 +34,13 @@
                                     <br>
                                 {% endfor %}
                                 <label class="hidden control-label" id="label-username"
                                        for="username">{{ _("Enter your OpenID Username") }}:</label>
                                 {{ form.username(size = 80, class = "hidden form-control", autofocus = true) }}
-                            </div>
-                        </div>
-                        <div class="control-group">
-                            <div class="controls">
                                 <label class="checkbox" for="remember_me">
-                                    {{ form.remember_me }} Remember Me
                                 </label>
+                                {{ form.remember_me }} Remember Me
                             </div>
                         </div>
                         <input
                                 id="btn-oid-before-submit"
                                 class="btn btn-primary btn-block"
@@ -131,11 +127,11 @@
             }
         }
         {% for pr in providers %}
             document.getElementById("btn-oid-provider-{{ pr.name }}")
                 .addEventListener("click", function () {
-                    set_openid("{{ pr.url | safe }}", "{{ pr.name }}");
+                    set_openid("{{ pr.name | safe }}", "{{ pr.name }}");
                 });
         {% endfor %}
         document.getElementById("btn-oid-before-submit")
             .addEventListener("click", function () {
                 beforeSubmit()
diff --git a/tests/config_oid.py b/tests/config_oid.py
new file mode 100644
index 000000000..140e863f7
--- /dev/null
+++ b/tests/config_oid.py
@@ -0,0 +1,29 @@
+import os
+
+from flask_appbuilder.security.manager import AUTH_OID
+
+basedir = os.path.abspath(os.path.dirname(__file__))
+
+SQLALCHEMY_DATABASE_URI = os.environ.get(
+    "SQLALCHEMY_DATABASE_URI"
+) or "sqlite:///" + os.path.join(basedir, "app.db")
+
+SECRET_KEY = "thisismyscretkey"
+
+AUTH_TYPE = AUTH_OID
+
+OPENID_PROVIDERS = [
+    {"name": "Google", "url": "https://www.google.com/accounts/o8/id"},
+    {"name": "Yahoo", "url": "https://me.yahoo.com"},
+    {"name": "AOL", "url": "http://openid.aol.com/<username>"},
+    {"name": "Flickr", "url": "http://www.flickr.com/<username>"},
+    {"name": "OpenStack", "url": "https://openstackid.org/"},
+]
+
+WTF_CSRF_ENABLED = False
+
+# Will allow user self registration
+AUTH_USER_REGISTRATION = True
+
+# The default user self registration role for all users
+AUTH_USER_REGISTRATION_ROLE = "Admin"
diff --git a/tests/test_mvc_oauth.py b/tests/test_mvc_oauth.py
index b51d64717..6e9a5227b 100644
--- a/tests/test_mvc_oauth.py
+++ b/tests/test_mvc_oauth.py
@@ -24,11 +24,11 @@ class OAuthRemoteMock:
     def get(self, item):
         if item == "userinfo":
             return UserInfoReponseMock()
 
 
-class APICSRFTestCase(FABTestCase):
+class MVCOAuthTestCase(FABTestCase):
     def setUp(self):
         from flask import Flask
         from flask_wtf import CSRFProtect
         from flask_appbuilder import AppBuilder
 
diff --git a/tests/test_mvc_oid.py b/tests/test_mvc_oid.py
new file mode 100644
index 000000000..2965f4369
--- /dev/null
+++ b/tests/test_mvc_oid.py
@@ -0,0 +1,53 @@
+from unittest.mock import MagicMock
+
+from flask_appbuilder import SQLA
+from tests.base import FABTestCase
+
+
+class MVCOIDTestCase(FABTestCase):
+    def setUp(self):
+        from flask import Flask
+        from flask_appbuilder import AppBuilder
+
+        self.app = Flask(__name__)
+        self.app.config.from_object("tests.config_oid")
+        self.db = SQLA(self.app)
+        self.appbuilder = AppBuilder(self.app, self.db.session)
+
+    def test_oid_login_get(self):
+        """
+        OID: Test login get
+        """
+        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Login ok")
+
+        with self.app.test_client() as client:
+            response = client.get("/login/")
+        self.assertEqual(response.status_code, 200)
+        for provider in self.app.config["OPENID_PROVIDERS"]:
+            self.assertIn(provider["name"], response.data.decode("utf-8"))
+
+    def test_oid_login_post(self):
+        """
+        OID: Test login post with a valid provider
+        """
+        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Login ok")
+
+        with self.app.test_client() as client:
+            response = client.post("/login/", data=dict(openid="OpenStack"))
+            self.assertEqual(response.status_code, 200)
+            self.assertEqual(response.data, b"Login ok")
+        self.appbuilder.sm.oid.try_login.assert_called_with(
+            "https://openstackid.org/", ask_for=["email"], ask_for_optional=[]
+        )
+
+    def test_oid_login_post_invalid_provider(self):
+        """
+        OID: Test login post with an invalid provider
+        """
+        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Not Ok")
+
+        with self.app.test_client() as client:
+            response = client.post("/login/", data=dict(openid="DoesNotExist"))
+            self.assertEqual(response.status_code, 302)
+            self.assertEqual(response.location, "/login/")
+        self.appbuilder.sm.oid.try_login.assert_not_called()
diff --git a/tests/test_security_api.py b/tests/test_security_api.py
index b0e4b419e..18606e9f6 100644
--- a/tests/test_security_api.py
+++ b/tests/test_security_api.py
@@ -442,10 +442,12 @@ class RolePermissionAPITestCase(FABTestCase):
 
         for b in self.appbuilder.baseviews:
             if hasattr(b, "datamodel") and b.datamodel.session is not None:
                 b.datamodel.session = self.db.session
 
+        self.create_default_users(self.appbuilder)
+
     def tearDown(self):
         self.appbuilder.session.close()
         engine = self.appbuilder.session.get_bind(mapper=None, clause=None)
         for baseview in self.appbuilder.baseviews:
             if hasattr(baseview, "datamodel"):

Vulnerable code:                     form.openid.data,
                            </div>
                        </div>
                        <div class="control-group">
                            <div class="controls">
                                    {{ form.remember_me }} Remember Me
                    set_openid("{{ pr.url | safe }}", "{{ pr.name }}");
class APICSRFTestCase(FABTestCase):
Fixed code:     def get_oid_identity_url(self, provider_name: str) -> Optional[str]:
        """
        Returns the OIDC identity provider URL
        """
        for provider in self.openid_providers:
            if provider.get("name") == provider_name:
                return provider.get("url")

                identity_url = self.appbuilder.sm.get_oid_identity_url(form.openid.data)
                if identity_url is None:
                    flash(as_unicode(self.invalid_login_message), "warning")
                    return redirect(self.appbuilder.get_url_for_login)
                    identity_url,
                                {{ form.remember_me }} Remember Me
                    set_openid("{{ pr.name | safe }}", "{{ pr.name }}");
import os

from flask_appbuilder.security.manager import AUTH_OID

basedir = os.path.abspath(os.path.dirname(__file__))

SQLALCHEMY_DATABASE_URI = os.environ.get(
    "SQLALCHEMY_DATABASE_URI"
) or "sqlite:///" + os.path.join(basedir, "app.db")

SECRET_KEY = "thisismyscretkey"

AUTH_TYPE = AUTH_OID

OPENID_PROVIDERS = [
    {"name": "Google", "url": "https://www.google.com/accounts/o8/id"},
    {"name": "Yahoo", "url": "https://me.yahoo.com"},
    {"name": "AOL", "url": "http://openid.aol.com/<username>"},
    {"name": "Flickr", "url": "http://www.flickr.com/<username>"},
    {"name": "OpenStack", "url": "https://openstackid.org/"},
]

WTF_CSRF_ENABLED = False

# Will allow user self registration
AUTH_USER_REGISTRATION = True

# The default user self registration role for all users
AUTH_USER_REGISTRATION_ROLE = "Admin"
class MVCOAuthTestCase(FABTestCase):
from unittest.mock import MagicMock

from flask_appbuilder import SQLA
from tests.base import FABTestCase


class MVCOIDTestCase(FABTestCase):
    def setUp(self):
        from flask import Flask
        from flask_appbuilder import AppBuilder

        self.app = Flask(__name__)
        self.app.config.from_object("tests.config_oid")
        self.db = SQLA(self.app)
        self.appbuilder = AppBuilder(self.app, self.db.session)

    def test_oid_login_get(self):
        """
        OID: Test login get
        """
        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Login ok")

        with self.app.test_client() as client:
            response = client.get("/login/")
        self.assertEqual(response.status_code, 200)
        for provider in self.app.config["OPENID_PROVIDERS"]:
            self.assertIn(provider["name"], response.data.decode("utf-8"))

    def test_oid_login_post(self):
        """
        OID: Test login post with a valid provider
        """
        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Login ok")

        with self.app.test_client() as client:
            response = client.post("/login/", data=dict(openid="OpenStack"))
            self.assertEqual(response.status_code, 200)
            self.assertEqual(response.data, b"Login ok")
        self.appbuilder.sm.oid.try_login.assert_called_with(
            "https://openstackid.org/", ask_for=["email"], ask_for_optional=[]
        )

    def test_oid_login_post_invalid_provider(self):
        """
        OID: Test login post with an invalid provider
        """
        self.appbuilder.sm.oid.try_login = MagicMock(return_value="Not Ok")

        with self.app.test_client() as client:
            response = client.post("/login/", data=dict(openid="DoesNotExist"))
            self.assertEqual(response.status_code, 302)
            self.assertEqual(response.location, "/login/")
        self.appbuilder.sm.oid.try_login.assert_not_called()
        self.create_default_users(self.appbuilder)