CVE-2021-41265: Improper Authentication (CWE-287)
value: Flask-AppBuilder is a development framework built on top of Flask. Verions prior to 3.3.4 contain an improper authentication vulnerability in the REST API. The issue allows for a malicious actor with a carefully crafted request to successfully authenticate and gain access to existing protected REST API endpoints. This only affects non database authentication types and new REST API endpoints. Users should upgrade to Flask-AppBuilder 3.3.4 to receive a patch.
CWE description: When an actor claims to have a given identity, the product does not prove or insufficiently proves that the claim is correct.
CVSS v3 base score: 8.8
Published: 2021-12-09T17:15Z
Repository: https://github.com/dpgaspar/Flask-AppBuilder
Language: Python
Files changed: docs/config.rst, flask_appbuilder/security/api.py, flask_appbuilder/security/manager.py, flask_appbuilder/security/schemas.py
Commit date: 2021-10-12 15:31:20 +0100
Commit message: chore: improve schema validation (#1712)


Diff: diff --git a/docs/config.rst b/docs/config.rst
index d338bcaa3..90ca8eb23 100755
--- a/docs/config.rst
+++ b/docs/config.rst
@@ -200,10 +200,13 @@ Use config.py to configure the following parameters. By default it will use SQLL
 | AUTH_ROLE_ADMIN                        | Configure the name of the admin role.      |   No      |
 +----------------------------------------+--------------------------------------------+-----------+
 | AUTH_ROLE_PUBLIC                       | Special Role that holds the public         |   No      |
 |                                        | permissions, no authentication needed.     |           |
 +----------------------------------------+--------------------------------------------+-----------+
+| AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS| Allow REST API login with alternative auth |   No      |
+| True|False                             | providers (default False)                  |           |           |
++----------------------------------------+--------------------------------------------+-----------+
 | APP_NAME                               | The name of your application.              |   No      |
 +----------------------------------------+--------------------------------------------+-----------+
 | APP_THEME                              | Various themes for you to choose           |   No      |
 |                                        | from (bootwatch).                          |           |
 +----------------------------------------+--------------------------------------------+-----------+
diff --git a/flask_appbuilder/security/api.py b/flask_appbuilder/security/api.py
index 2e2dfd612..966968a28 100644
--- a/flask_appbuilder/security/api.py
+++ b/flask_appbuilder/security/api.py
@@ -1,26 +1,22 @@
-from flask import request
+from flask import request, Response
+from flask_appbuilder.api import BaseApi, safe
+from flask_appbuilder.const import (
+    API_SECURITY_ACCESS_TOKEN_KEY,
+    API_SECURITY_PROVIDER_DB,
+    API_SECURITY_PROVIDER_LDAP,
+    API_SECURITY_VERSION,
+)
+from flask_appbuilder.security.schemas import login_post
+from flask_appbuilder.views import expose
 from flask_jwt_extended import (
     create_access_token,
     create_refresh_token,
     get_jwt_identity,
     jwt_refresh_token_required,
 )
-
-from ..api import BaseApi, safe
-from ..const import (
-    API_SECURITY_ACCESS_TOKEN_KEY,
-    API_SECURITY_PASSWORD_KEY,
-    API_SECURITY_PROVIDER_DB,
-    API_SECURITY_PROVIDER_KEY,
-    API_SECURITY_PROVIDER_LDAP,
-    API_SECURITY_REFRESH_KEY,
-    API_SECURITY_REFRESH_TOKEN_KEY,
-    API_SECURITY_USERNAME_KEY,
-    API_SECURITY_VERSION,
-)
-from ..views import expose
+from marshmallow import ValidationError
 
 
 class SecurityApi(BaseApi):
 
     resource_name = "security"
@@ -33,11 +29,11 @@ class SecurityApi(BaseApi):
         api_spec.components.security_scheme("jwt", jwt_scheme)
         api_spec.components.security_scheme("jwt_refresh", jwt_scheme)
 
     @expose("/login", methods=["POST"])
     @safe
-    def login(self):
+    def login(self) -> Response:
         """Login endpoint for the API returns a JWT and optionally a refresh token
         ---
         post:
           description: >-
             Authenticate and get a JWT access and refresh token
@@ -86,43 +82,41 @@ class SecurityApi(BaseApi):
             500:
               $ref: '#/components/responses/500'
         """
         if not request.is_json:
             return self.response_400(message="Request payload is not JSON")
-        username = request.json.get(API_SECURITY_USERNAME_KEY, None)
-        password = request.json.get(API_SECURITY_PASSWORD_KEY, None)
-        provider = request.json.get(API_SECURITY_PROVIDER_KEY, None)
-        refresh = request.json.get(API_SECURITY_REFRESH_KEY, False)
-        if not username or not password or not provider:
-            return self.response_400(message="Missing required parameter")
+        try:
+            login_payload = login_post.load(request.json)
+        except ValidationError as error:
+            return self.response_400(message=error.messages)
+
         # AUTH
-        if provider == API_SECURITY_PROVIDER_DB:
-            user = self.appbuilder.sm.auth_user_db(username, password)
-        elif provider == API_SECURITY_PROVIDER_LDAP:
-            user = self.appbuilder.sm.auth_user_ldap(username, password)
-        else:
-            return self.response_400(
-                message="Provider {} not supported".format(provider)
+        user = None
+        if login_payload["provider"] == API_SECURITY_PROVIDER_DB:
+            user = self.appbuilder.sm.auth_user_db(
+                login_payload["username"], login_payload["password"]
+            )
+        elif login_payload["provider"] == API_SECURITY_PROVIDER_LDAP:
+            user = self.appbuilder.sm.auth_user_ldap(
+                login_payload["username"], login_payload["password"]
             )
         if not user:
             return self.response_401()
 
         # Identity can be any data that is json serializable
         resp = dict()
         resp[API_SECURITY_ACCESS_TOKEN_KEY] = create_access_token(
             identity=user.id, fresh=True
         )
-        if refresh:
-            resp[API_SECURITY_REFRESH_TOKEN_KEY] = create_refresh_token(
-                identity=user.id
-            )
+        if "refresh" in login_payload:
+            login_payload["refresh"] = create_refresh_token(identity=user.id)
         return self.response(200, **resp)
 
     @expose("/refresh", methods=["POST"])
     @jwt_refresh_token_required
     @safe
-    def refresh(self):
+    def refresh(self) -> Response:
         """
             Security endpoint for the refresh token, so we can obtain a new
             token without forcing the user to login again
         ---
         post:
diff --git a/flask_appbuilder/security/manager.py b/flask_appbuilder/security/manager.py
index 81d868e28..43b66fc5f 100644
--- a/flask_appbuilder/security/manager.py
+++ b/flask_appbuilder/security/manager.py
@@ -1,11 +1,11 @@
 import base64
 import datetime
 import json
 import logging
 import re
-from typing import Dict, List, Optional, Set, Tuple
+from typing import Any, Dict, List, Optional, Set, Tuple
 
 from flask import g, session, url_for
 from flask_babel import lazy_gettext as _
 from flask_jwt_extended import current_user as current_user_jwt
 from flask_jwt_extended import JWTManager
@@ -217,10 +217,11 @@ class BaseSecurityManager(AbstractSecurityManager):
         app.config.setdefault("AUTH_USER_REGISTRATION_ROLE", self.auth_role_public)
         app.config.setdefault("AUTH_USER_REGISTRATION_ROLE_JMESPATH", None)
         # Role Mapping
         app.config.setdefault("AUTH_ROLES_MAPPING", {})
         app.config.setdefault("AUTH_ROLES_SYNC_AT_LOGIN", False)
+        app.config.setdefault("AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS", False)
 
         # LDAP Config
         if self.auth_type == AUTH_LDAP:
             if "AUTH_LDAP_SERVER" not in app.config:
                 raise Exception(
@@ -328,10 +329,15 @@ class BaseSecurityManager(AbstractSecurityManager):
                                 fab_role_name
                             )
                         )
         return _roles
 
+    @property
+    def auth_type_provider_name(self) -> Optional[str]:
+        provider_to_auth_type = {AUTH_DB: "db", AUTH_LDAP: "ldap"}
+        return provider_to_auth_type.get(self.auth_type)
+
     @property
     def get_url_for_registeruser(self):
         return url_for(
             "%s.%s"
             % (self.registeruser_view.endpoint, self.registeruser_view.default_view)
@@ -344,43 +350,47 @@ class BaseSecurityManager(AbstractSecurityManager):
     @property
     def get_register_user_datamodel(self):
         return self.registerusermodelview.datamodel
 
     @property
-    def builtin_roles(self):
+    def builtin_roles(self) -> Dict[str, Any]:
         return self._builtin_roles
 
     @property
-    def auth_type(self):
+    def api_login_allow_multiple_providers(self):
+        return self.appbuilder.get_app.config["AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS"]
+
+    @property
+    def auth_type(self) -> int:
         return self.appbuilder.get_app.config["AUTH_TYPE"]
 
     @property
-    def auth_username_ci(self):
+    def auth_username_ci(self) -> str:
         return self.appbuilder.get_app.config.get("AUTH_USERNAME_CI", True)
 
     @property
-    def auth_role_admin(self):
+    def auth_role_admin(self) -> str:
         return self.appbuilder.get_app.config["AUTH_ROLE_ADMIN"]
 
     @property
-    def auth_role_public(self):
+    def auth_role_public(self) -> str:
         return self.appbuilder.get_app.config["AUTH_ROLE_PUBLIC"]
 
     @property
-    def auth_ldap_server(self):
+    def auth_ldap_server(self) -> str:
         return self.appbuilder.get_app.config["AUTH_LDAP_SERVER"]
 
     @property
-    def auth_ldap_use_tls(self):
+    def auth_ldap_use_tls(self) -> bool:
         return self.appbuilder.get_app.config["AUTH_LDAP_USE_TLS"]
 
     @property
-    def auth_user_registration(self):
+    def auth_user_registration(self) -> bool:
         return self.appbuilder.get_app.config["AUTH_USER_REGISTRATION"]
 
     @property
-    def auth_user_registration_role(self):
+    def auth_user_registration_role(self) -> str:
         return self.appbuilder.get_app.config["AUTH_USER_REGISTRATION_ROLE"]
 
     @property
     def auth_user_registration_role_jmespath(self) -> str:
         return self.appbuilder.get_app.config["AUTH_USER_REGISTRATION_ROLE_JMESPATH"]
diff --git a/flask_appbuilder/security/schemas.py b/flask_appbuilder/security/schemas.py
new file mode 100644
index 000000000..debe63d2e
--- /dev/null
+++ b/flask_appbuilder/security/schemas.py
@@ -0,0 +1,45 @@
+from typing import Union
+
+from flask import current_app
+from flask_appbuilder.const import (
+    API_SECURITY_PROVIDER_DB,
+    API_SECURITY_PROVIDER_LDAP,
+    AUTH_DB,
+    AUTH_LDAP,
+)
+from marshmallow import fields, Schema, ValidationError
+from marshmallow.validate import Length, OneOf
+
+
+provider_to_auth_type = {"db": AUTH_DB, "ldap": AUTH_LDAP}
+
+
+def validate_password(value: Union[bytes, bytearray, str]) -> None:
+    if not value:
+        raise ValidationError("Password is required")
+    if len(value) == 1 and value.encode()[0] == 0:
+        raise ValidationError("Password null is not allowed")
+
+
+def validate_provider(value: Union[bytes, bytearray, str]) -> None:
+    if not current_app.appbuilder.sm.api_login_allow_multiple_providers:
+        provider_name = current_app.appbuilder.sm.auth_type_provider_name
+        if provider_name and provider_name != value:
+            raise ValidationError("Alternative authentication provider is not allowed")
+
+
+class LoginPost(Schema):
+    username = fields.String(required=True, allow_none=False, validate=Length(min=1))
+    password = fields.String(
+        validate=validate_password, required=True, allow_none=False
+    )
+    provider = fields.String(
+        validate=[
+            OneOf([API_SECURITY_PROVIDER_DB, API_SECURITY_PROVIDER_LDAP]),
+            validate_provider,
+        ]
+    )
+    refresh = fields.Boolean(required=False)
+
+
+login_post = LoginPost()

Vulnerable code: from flask import request

from ..api import BaseApi, safe
from ..const import (
    API_SECURITY_ACCESS_TOKEN_KEY,
    API_SECURITY_PASSWORD_KEY,
    API_SECURITY_PROVIDER_DB,
    API_SECURITY_PROVIDER_KEY,
    API_SECURITY_PROVIDER_LDAP,
    API_SECURITY_REFRESH_KEY,
    API_SECURITY_REFRESH_TOKEN_KEY,
    API_SECURITY_USERNAME_KEY,
    API_SECURITY_VERSION,
)
from ..views import expose
    def login(self):
        username = request.json.get(API_SECURITY_USERNAME_KEY, None)
        password = request.json.get(API_SECURITY_PASSWORD_KEY, None)
        provider = request.json.get(API_SECURITY_PROVIDER_KEY, None)
        refresh = request.json.get(API_SECURITY_REFRESH_KEY, False)
        if not username or not password or not provider:
            return self.response_400(message="Missing required parameter")
        if provider == API_SECURITY_PROVIDER_DB:
            user = self.appbuilder.sm.auth_user_db(username, password)
        elif provider == API_SECURITY_PROVIDER_LDAP:
            user = self.appbuilder.sm.auth_user_ldap(username, password)
        else:
            return self.response_400(
                message="Provider {} not supported".format(provider)
        if refresh:
            resp[API_SECURITY_REFRESH_TOKEN_KEY] = create_refresh_token(
                identity=user.id
            )
    def refresh(self):
from typing import Dict, List, Optional, Set, Tuple
    def builtin_roles(self):
    def auth_type(self):
    def auth_username_ci(self):
    def auth_role_admin(self):
    def auth_role_public(self):
    def auth_ldap_server(self):
    def auth_ldap_use_tls(self):
    def auth_user_registration(self):
    def auth_user_registration_role(self):
Fixed code: | AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS| Allow REST API login with alternative auth |   No      |
| True|False                             | providers (default False)                  |           |           |
+----------------------------------------+--------------------------------------------+-----------+
from flask import request, Response
from flask_appbuilder.api import BaseApi, safe
from flask_appbuilder.const import (
    API_SECURITY_ACCESS_TOKEN_KEY,
    API_SECURITY_PROVIDER_DB,
    API_SECURITY_PROVIDER_LDAP,
    API_SECURITY_VERSION,
)
from flask_appbuilder.security.schemas import login_post
from flask_appbuilder.views import expose
from marshmallow import ValidationError
    def login(self) -> Response:
        try:
            login_payload = login_post.load(request.json)
        except ValidationError as error:
            return self.response_400(message=error.messages)

        user = None
        if login_payload["provider"] == API_SECURITY_PROVIDER_DB:
            user = self.appbuilder.sm.auth_user_db(
                login_payload["username"], login_payload["password"]
            )
        elif login_payload["provider"] == API_SECURITY_PROVIDER_LDAP:
            user = self.appbuilder.sm.auth_user_ldap(
                login_payload["username"], login_payload["password"]
        if "refresh" in login_payload:
            login_payload["refresh"] = create_refresh_token(identity=user.id)
    def refresh(self) -> Response:
from typing import Any, Dict, List, Optional, Set, Tuple
        app.config.setdefault("AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS", False)
    @property
    def auth_type_provider_name(self) -> Optional[str]:
        provider_to_auth_type = {AUTH_DB: "db", AUTH_LDAP: "ldap"}
        return provider_to_auth_type.get(self.auth_type)

    def builtin_roles(self) -> Dict[str, Any]:
    def api_login_allow_multiple_providers(self):
        return self.appbuilder.get_app.config["AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS"]

    @property
    def auth_type(self) -> int:
    def auth_username_ci(self) -> str:
    def auth_role_admin(self) -> str:
    def auth_role_public(self) -> str:
    def auth_ldap_server(self) -> str:
    def auth_ldap_use_tls(self) -> bool:
    def auth_user_registration(self) -> bool:
    def auth_user_registration_role(self) -> str:
from typing import Union

from flask import current_app
from flask_appbuilder.const import (
    API_SECURITY_PROVIDER_DB,
    API_SECURITY_PROVIDER_LDAP,
    AUTH_DB,
    AUTH_LDAP,
)
from marshmallow import fields, Schema, ValidationError
from marshmallow.validate import Length, OneOf


provider_to_auth_type = {"db": AUTH_DB, "ldap": AUTH_LDAP}


def validate_password(value: Union[bytes, bytearray, str]) -> None:
    if not value:
        raise ValidationError("Password is required")
    if len(value) == 1 and value.encode()[0] == 0:
        raise ValidationError("Password null is not allowed")


def validate_provider(value: Union[bytes, bytearray, str]) -> None:
    if not current_app.appbuilder.sm.api_login_allow_multiple_providers:
        provider_name = current_app.appbuilder.sm.auth_type_provider_name
        if provider_name and provider_name != value:
            raise ValidationError("Alternative authentication provider is not allowed")


class LoginPost(Schema):
    username = fields.String(required=True, allow_none=False, validate=Length(min=1))
    password = fields.String(
        validate=validate_password, required=True, allow_none=False
    )
    provider = fields.String(
        validate=[
            OneOf([API_SECURITY_PROVIDER_DB, API_SECURITY_PROVIDER_LDAP]),
            validate_provider,
        ]
    )
    refresh = fields.Boolean(required=False)


login_post = LoginPost()