feat: AUTH_REMOTE_USER_ENV_VAR config key for auth REMOTE_USER type (#…

…2193)

docs/config.rst

@@ -46,6 +46,11 @@ Use config.py to configure the following parameters. By default it will use SQLL
 |                                        | Requires ``jmespath`` to be installed.     |           |
 |                                        | See :ref:`jmespath-examples` for examples  |           |
 +----------------------------------------+--------------------------------------------+-----------+
+| AUTH_REMOTE_USER_ENV_VAR               | When using AUTH_TYPE = AUTH_REMOTE_USER    |   No      |
+|                                        | Optionally set the wsgi environment var    |           |
+|                                        | that holds the current logged in user      |           |
+|                                        | Default: REMOTE_USER                       |           |
++----------------------------------------+--------------------------------------------+-----------+
 | AUTH_ROLES_SYNC_AT_LOGIN               | Sets if user's roles are replaced each     |   No      |
 |                                        | login with those received from LDAP/OAUTH  |           |
 |                                        | Default: False                             |           |

flask_appbuilder/security/manager.py

@@ -258,6 +258,9 @@ def __init__(self, appbuilder):
             app.config.setdefault("AUTH_LDAP_LASTNAME_FIELD", "sn")
             app.config.setdefault("AUTH_LDAP_EMAIL_FIELD", "mail")
 
+        if self.auth_type == AUTH_REMOTE_USER:
+            app.config.setdefault("AUTH_REMOTE_USER_ENV_VAR", "REMOTE_USER")
+
         # Rate limiting
         app.config.setdefault("AUTH_RATE_LIMITED", False)
         app.config.setdefault("AUTH_RATE_LIMIT", "10 per 20 second")
@@ -415,6 +418,10 @@ def auth_user_registration_role(self) -> str:
     def auth_user_registration_role_jmespath(self) -> str:
         return self.appbuilder.get_app.config["AUTH_USER_REGISTRATION_ROLE_JMESPATH"]
 
+    @property
+    def auth_remote_user_env_var(self) -> str:
+        return self.appbuilder.get_app.config["AUTH_REMOTE_USER_ENV_VAR"]
+
     @property
     def auth_roles_mapping(self) -> Dict[str, List[str]]:
         return self.appbuilder.get_app.config["AUTH_ROLES_MAPPING"]

flask_appbuilder/security/views.py

@@ -716,7 +716,7 @@ class AuthRemoteUserView(AuthView):
 
     @expose("/login/")
     def login(self) -> WerkzeugResponse:
-        username = request.environ.get("REMOTE_USER")
+        username = request.environ.get(self.appbuilder.sm.auth_remote_user_env_var)
         if g.user is not None and g.user.is_authenticated:
             next_url = request.args.get("next", "")
             return redirect(get_safe_redirect(next_url))

tests/security/test_auth_oauth.py

@@ -1,4 +1,3 @@
-import logging
 import os
 import unittest
 from unittest.mock import MagicMock
@@ -12,10 +11,6 @@
 from tests.const import USERNAME_ADMIN, USERNAME_READONLY
 from tests.fixtures.users import create_default_users
 
-logging.basicConfig(format="%(asctime)s:%(levelname)s:%(name)s:%(message)s")
-logging.getLogger().setLevel(logging.DEBUG)
-log = logging.getLogger(__name__)
-
 
 class OAuthRegistrationRoleTestCase(unittest.TestCase):
     def setUp(self):

tests/security/test_auth_remote_user.py

@@ -0,0 +1,75 @@
+import unittest
+
+from flask import Flask
+from flask_appbuilder import AppBuilder, SQLA
+from flask_appbuilder.const import AUTH_REMOTE_USER
+
+
+class AuthRemoteUserTestCase(unittest.TestCase):
+    def setUp(self):
+        # start Flask
+        self.app = Flask(__name__)
+        self.app.config["AUTH_TYPE"] = AUTH_REMOTE_USER
+
+        # start Database
+        self.db = SQLA(self.app)
+
+    def tearDown(self):
+        # Remove test user
+        user_alice = self.appbuilder.sm.find_user("alice")
+        if user_alice:
+            self.db.session.delete(user_alice)
+            self.db.session.commit()
+
+        # stop Flask
+        self.app = None
+
+        # stop Flask-AppBuilder
+        self.appbuilder = None
+
+        # stop Database
+        self.db.session.remove()
+        self.db = None
+
+    def test_unset_remote_user_env_var(self):
+        self.appbuilder = AppBuilder(self.app, self.db.session)
+        sm = self.appbuilder.sm
+
+        self.assertEqual(sm.auth_remote_user_env_var, "REMOTE_USER")
+
+    def test_set_remote_user_env_var(self):
+        self.app.config["AUTH_REMOTE_USER_ENV_VAR"] = "HTTP_REMOTE_USER"
+        self.appbuilder = AppBuilder(self.app, self.db.session)
+        sm = self.appbuilder.sm
+
+        self.assertEqual(sm.auth_remote_user_env_var, "HTTP_REMOTE_USER")
+
+    def test_normal_login(self):
+        self.appbuilder = AppBuilder(self.app, self.db.session)
+        sm = self.appbuilder.sm
+
+        # register a user
+        _ = sm.add_user(
+            username="alice",
+            first_name="Alice",
+            last_name="Doe",
+            email="alice@example.com",
+            role=[],
+        )
+
+        self.assertTrue(sm.auth_user_remote_user("alice"))
+
+    def test_inactive_user_login(self):
+        self.appbuilder = AppBuilder(self.app, self.db.session)
+        sm = self.appbuilder.sm
+
+        # register a user
+        alice_user = sm.add_user(
+            username="alice",
+            first_name="Alice",
+            last_name="Doe",
+            email="alice@example.com",
+            role=[],
+        )
+        alice_user.active = False
+        self.assertFalse(sm.auth_user_remote_user("alice"))