Improper Authentication in Flask-AppBuilder

@@ -202,6 +202,9 @@ Use config.py to configure the following parameters. By default it will use SQLL
 | 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      |

@@ -1,24 +1,20 @@
-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):
@@ -35,7 +31,7 @@ def add_apispec_components(self, api_spec):
 
     @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:
@@ -88,20 +84,20 @@ def login(self):
         """
         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()
@@ -111,16 +107,14 @@ def login(self):
         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

@@ -3,7 +3,7 @@
 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 _
@@ -219,6 +219,7 @@ def __init__(self, appbuilder):
         # 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:
@@ -330,6 +331,11 @@ def get_roles_from_keys(self, role_keys: List[str]) -> Set[role_model]:
                         )
         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(
@@ -346,39 +352,43 @@ 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

@@ -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()