fix: optionally return HTTP 403 instead of 401 when unauthorized (#1748)

* fix: optionally return HTTP 403 instead of 401 when unauthorized

* add test

docs/config.rst

@@ -202,8 +202,13 @@ 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_STRICT_RESPONSE_CODES             | When True, protected endpoints will return |   No      |
+|                                        | HTTP 403 instead of 401. This option will  |           |
+|                                        | be removed and default to True on the next |           |
+|                                        | major release. defaults to False           |           |
++----------------------------------------+--------------------------------------------+-----------+
 | AUTH_API_LOGIN_ALLOW_MULTIPLE_PROVIDERS| Allow REST API login with alternative auth |   No      |
-| True|False                             | providers (default False)                  |           |           |
+| True|False                             | providers (default False)                  |           |
 +----------------------------------------+--------------------------------------------+-----------+
 | APP_NAME                               | The name of your application.              |   No      |
 +----------------------------------------+--------------------------------------------+-----------+

flask_appbuilder/security/decorators.py

@@ -1,42 +1,61 @@
 import functools
 import logging
-
-from flask import current_app, flash, jsonify, make_response, redirect, request, url_for
-from flask_jwt_extended import verify_jwt_in_request
-from flask_login import current_user
-
-from .._compat import as_unicode
-from ..const import (
+from typing import TYPE_CHECKING
+
+from flask import (
+    current_app,
+    flash,
+    jsonify,
+    make_response,
+    redirect,
+    request,
+    Response,
+    url_for,
+)
+from flask_appbuilder._compat import as_unicode
+from flask_appbuilder.const import (
     FLAMSG_ERR_SEC_ACCESS_DENIED,
     LOGMSG_ERR_SEC_ACCESS_DENIED,
     PERMISSION_PREFIX,
 )
+from flask_jwt_extended import verify_jwt_in_request
+from flask_login import current_user
+
 
 log = logging.getLogger(__name__)
 
+if TYPE_CHECKING:
+    from flask_appbuilder.api import BaseApi
+
+
+def response_unauthorized(base_class: "BaseApi") -> Response:
+    if current_app.config.get("AUTH_STRICT_RESPONSE_CODES", False):
+        return base_class.response_403()
+    return base_class.response_401()
+
 
 def protect(allow_browser_login=False):
     """
-        Use this decorator to enable granular security permissions
-        to your API methods (BaseApi and child classes).
-        Permissions will be associated to a role, and roles are associated to users.
-
-        allow_browser_login will accept signed cookies obtained from the normal MVC app::
-
-            class MyApi(BaseApi):
-                @expose('/dosonmething', methods=['GET'])
-                @protect(allow_browser_login=True)
-                @safe
-                def do_something(self):
-                    ....
-
-                @expose('/dosonmethingelse', methods=['GET'])
-                @protect()
-                @safe
-                def do_something_else(self):
-                    ....
-
-        By default the permission's name is the methods name.
+    Use this decorator to enable granular security permissions
+    to your API methods (BaseApi and child classes).
+    Permissions will be associated to a role, and roles are associated to users.
+
+    allow_browser_login will accept signed cookies obtained from the normal MVC app::
+
+        class MyApi(BaseApi):
+            @expose('/dosonmething', methods=['GET'])
+            @protect(allow_browser_login=True)
+            @safe
+            def do_something(self):
+                ....
+
+            @expose('/dosonmethingelse', methods=['GET'])
+            @protect()
+            @safe
+            def do_something_else(self):
+                ....
+
+    By default the permission's name is the methods name.
     """
 
     def _protect(f):
@@ -47,14 +66,14 @@ def _protect(f):
 
         def wraps(self, *args, **kwargs):
             # Apply method permission name override if exists
-            permission_str = "{}{}".format(PERMISSION_PREFIX, f._permission_name)
+            permission_str = f"{PERMISSION_PREFIX}{f._permission_name}"
             if self.method_permission_name:
                 _permission_name = self.method_permission_name.get(f.__name__)
                 if _permission_name:
-                    permission_str = "{}{}".format(PERMISSION_PREFIX, _permission_name)
+                    permission_str = f"{PERMISSION_PREFIX}{_permission_name}"
             class_permission_name = self.class_permission_name
             if permission_str not in self.base_permissions:
-                return self.response_401()
+                return response_unauthorized(self)
             if current_app.appbuilder.sm.is_item_public(
                 permission_str, class_permission_name
             ):
@@ -77,7 +96,7 @@ def wraps(self, *args, **kwargs):
                     permission_str, class_permission_name
                 )
             )
-            return self.response_401()
+            return response_unauthorized(self)
 
         f._permission_name = permission_str
         return functools.update_wrapper(wraps, f)
@@ -87,22 +106,22 @@ def wraps(self, *args, **kwargs):
 
 def has_access(f):
     """
-        Use this decorator to enable granular security permissions to your methods.
-        Permissions will be associated to a role, and roles are associated to users.
+    Use this decorator to enable granular security permissions to your methods.
+    Permissions will be associated to a role, and roles are associated to users.
 
-        By default the permission's name is the methods name.
+    By default the permission's name is the methods name.
     """
     if hasattr(f, "_permission_name"):
         permission_str = f._permission_name
     else:
         permission_str = f.__name__
 
     def wraps(self, *args, **kwargs):
-        permission_str = "{}{}".format(PERMISSION_PREFIX, f._permission_name)
+        permission_str = f"{PERMISSION_PREFIX}{f._permission_name}"
         if self.method_permission_name:
             _permission_name = self.method_permission_name.get(f.__name__)
             if _permission_name:
-                permission_str = "{}{}".format(PERMISSION_PREFIX, _permission_name)
+                permission_str = f"{PERMISSION_PREFIX}{_permission_name}"
         if permission_str in self.base_permissions and self.appbuilder.sm.has_access(
             permission_str, self.class_permission_name
         ):
@@ -127,24 +146,24 @@ def wraps(self, *args, **kwargs):
 
 def has_access_api(f):
     """
-        Use this decorator to enable granular security permissions to your API methods.
-        Permissions will be associated to a role, and roles are associated to users.
+    Use this decorator to enable granular security permissions to your API methods.
+    Permissions will be associated to a role, and roles are associated to users.
 
-        By default the permission's name is the methods name.
+    By default the permission's name is the methods name.
 
-        this will return a message and HTTP 401 is case of unauthorized access.
+    this will return a message and HTTP 403 is case of unauthorized access.
     """
     if hasattr(f, "_permission_name"):
         permission_str = f._permission_name
     else:
         permission_str = f.__name__
 
     def wraps(self, *args, **kwargs):
-        permission_str = "{}{}".format(PERMISSION_PREFIX, f._permission_name)
+        permission_str = f"{PERMISSION_PREFIX}{f._permission_name}"
         if self.method_permission_name:
             _permission_name = self.method_permission_name.get(f.__name__)
             if _permission_name:
-                permission_str = "{}{}".format(PERMISSION_PREFIX, _permission_name)
+                permission_str = f"{PERMISSION_PREFIX}{_permission_name}"
         if permission_str in self.base_permissions and self.appbuilder.sm.has_access(
             permission_str, self.class_permission_name
         ):
@@ -159,7 +178,7 @@ def wraps(self, *args, **kwargs):
                 jsonify(
                     {"message": str(FLAMSG_ERR_SEC_ACCESS_DENIED), "severity": "danger"}
                 ),
-                401,
+                403,
             )
             response.headers["Content-Type"] = "application/json"
             return response
@@ -170,38 +189,38 @@ def wraps(self, *args, **kwargs):
 
 def permission_name(name):
     """
-        Use this decorator to override the name of the permission.
-        has_access will use the methods name has the permission name
-        if you want to override this add this decorator to your methods.
-        This is useful if you want to aggregate methods to permissions
+    Use this decorator to override the name of the permission.
+    has_access will use the methods name has the permission name
+    if you want to override this add this decorator to your methods.
+    This is useful if you want to aggregate methods to permissions
 
-        It will add '_permission_name' attribute to your method
-        that will be inspected by BaseView to collect your view's
-        permissions.
+    It will add '_permission_name' attribute to your method
+    that will be inspected by BaseView to collect your view's
+    permissions.
 
-        Note that you should use @has_access to execute after @permission_name
-        like on the following example.
+    Note that you should use @has_access to execute after @permission_name
+    like on the following example.
 
-        Use it like this to aggregate permissions for your methods::
+    Use it like this to aggregate permissions for your methods::
 
-            class MyModelView(ModelView):
-                datamodel = SQLAInterface(MyModel)
+        class MyModelView(ModelView):
+            datamodel = SQLAInterface(MyModel)
 
-                @has_access
-                @permission_name('GeneralXPTO_Permission')
-                @expose(url='/xpto')
-                def xpto(self):
-                    return "Your on xpto"
+            @has_access
+            @permission_name('GeneralXPTO_Permission')
+            @expose(url='/xpto')
+            def xpto(self):
+                return "Your on xpto"
 
-                @has_access
-                @permission_name('GeneralXPTO_Permission')
-                @expose(url='/xpto2')
-                def xpto2(self):
-                    return "Your on xpto2"
+            @has_access
+            @permission_name('GeneralXPTO_Permission')
+            @expose(url='/xpto2')
+            def xpto2(self):
+                return "Your on xpto2"
 
 
-        :param name:
-            The name of the permission to override
+    :param name:
+        The name of the permission to override
     """
 
     def wraps(f):

flask_appbuilder/tests/config_api.py

@@ -6,7 +6,7 @@
     "SQLALCHEMY_DATABASE_URI"
 ) or "sqlite:///" + os.path.join(basedir, "app.db")
 
-
+AUTH_STRICT_RESPONSE_CODES = False
 SECRET_KEY = "thisismyscretkey"
 SQLALCHEMY_TRACK_MODIFICATIONS = False
 WTF_CSRF_ENABLED = False

flask_appbuilder/tests/test_api.py

@@ -601,8 +601,15 @@ def test_auth_authorization(self):
         # Test unauthorized DELETE
         pk = 1
         uri = f"api/v1/model1apirestrictedpermissions/{pk}"
+
+        self.app.config["AUTH_STRICT_RESPONSE_CODES"] = True
+        rv = self.auth_client_delete(client, token, uri)
+        self.assertEqual(rv.status_code, 403)
+
+        self.app.config["AUTH_STRICT_RESPONSE_CODES"] = False
         rv = self.auth_client_delete(client, token, uri)
         self.assertEqual(rv.status_code, 401)
+
         # Test unauthorized POST
         item = dict(
             field_string="test{}".format(MODEL1_DATA_SIZE + 1),
@@ -611,8 +618,14 @@ def test_auth_authorization(self):
             field_date=None,
         )
         uri = "api/v1/model1apirestrictedpermissions/"
+
+        self.app.config["AUTH_STRICT_RESPONSE_CODES"] = True
+        rv = self.auth_client_post(client, token, uri, item)
+        self.assertEqual(rv.status_code, 403)
+        self.app.config["AUTH_STRICT_RESPONSE_CODES"] = False
         rv = self.auth_client_post(client, token, uri, item)
         self.assertEqual(rv.status_code, 401)
+
         # Test authorized GET
         uri = f"api/v1/model1apirestrictedpermissions/{pk}"
         rv = self.auth_client_get(client, token, uri)