[Fixes #9212] Login with basic auth or token headers or apikey GET pa…

…rameter (#9214)

* login with headers or apikey

* nerw middleware setting

* loop over all backends

* login through account adapter

* check access_token before using it

* avoid exceptions in case token doesn't exist

* fix queryset

geonode/base/auth.py

@@ -101,12 +101,11 @@ def get_auth_token(user, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     return None
 
 
-def get_auth_user(access_token, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
+def get_auth_user(token, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     try:
-        Application = get_application_model()
-        app = Application.objects.get(name=client)
-        user = AccessToken.objects.filter(token=access_token, application=app).order_by('-expires').first().user
-        return user
+        access_token = AccessToken.objects.filter(token=token).first()
+        if access_token and access_token.is_valid():
+            return access_token.user
     except Exception:
         tb = traceback.format_exc()
         if tb:
@@ -170,7 +169,10 @@ def get_token_from_auth_header(auth_header, create_if_not_exists=False):
         if user and user.is_active:
             return get_auth_token(user) if not create_if_not_exists else get_or_create_token(user)
     elif re.search('Bearer', auth_header, re.IGNORECASE):
-        return re.compile(re.escape('Bearer '), re.IGNORECASE).sub('', auth_header)
+        token = re.compile(re.escape('Bearer '), re.IGNORECASE).sub('', auth_header)
+        access_token = AccessToken.objects.filter(token=token).first()
+        if access_token and access_token.is_valid():
+            return access_token
     return None
 
 
@@ -213,3 +215,8 @@ def basic_auth_authenticate_user(auth_header: str):
     password = decoded_credentials[1]
 
     return authenticate(username=username, password=password)
+
+
+def token_header_authenticate_user(auth_header: str):
+    token = get_token_from_auth_header(auth_header)
+    return get_auth_user(token)

geonode/security/middleware.py

@@ -25,9 +25,15 @@
 from django.http import HttpResponseRedirect
 from django.utils.deprecation import MiddlewareMixin
 
+from allauth.account.adapter import get_adapter
+
 from geonode import geoserver
 from geonode.utils import check_ogc_backend
-from geonode.base.auth import get_token_object_from_session, basic_auth_authenticate_user
+from geonode.base.auth import (
+    get_token_object_from_session,
+    basic_auth_authenticate_user,
+    token_header_authenticate_user,
+    get_auth_user)
 
 from guardian.shortcuts import get_anonymous_user
 
@@ -87,15 +93,12 @@ def __init__(self, get_response):
     def process_request(self, request):
 
         if not request.user.is_authenticated or request.user == get_anonymous_user():
-
             if "HTTP_AUTHORIZATION" in request.META:
                 auth_header = request.META.get("HTTP_AUTHORIZATION", request.META.get("HTTP_AUTHORIZATION2"))
 
                 if auth_header and "Basic" in auth_header:
                     user = basic_auth_authenticate_user(auth_header)
-
                     if user:
-                        # allow Basic Auth authenticated requests with valid credentials
                         return
 
             if not any(path.match(request.path) for path in white_list):
@@ -143,3 +146,28 @@ def do_logout(self, request):
             if not any(path.match(request.path) for path in white_list):
                 return HttpResponseRedirect(
                     f'{self.redirect_to}?next={request.path}')
+
+
+class LoginWithHeaderOrKeyMiddleware(MiddlewareMixin):
+    """
+    Middelware that creates a session from valid authentication headers (Basic Auth or Bearer Token)
+    or apikey parameter (token). Notice that forcing the creation of a session makes any DRF authentication class,
+    outisde SessionAuthentication, useless.
+    This middleware permits to expose ebedded views and APIs when LCKDOWN_GEONODE is active
+    """
+    def process_request(self, request):
+        if not request.user.is_authenticated or request.user == get_anonymous_user():
+            user = None
+            if "HTTP_AUTHORIZATION" in request.META:
+                auth_header = request.META.get("HTTP_AUTHORIZATION", request.META.get("HTTP_AUTHORIZATION2"))
+
+                if auth_header and "Basic" in auth_header:
+                    user = basic_auth_authenticate_user(auth_header)
+                elif auth_header and "Bearer" in auth_header:
+                    user = token_header_authenticate_user(auth_header)
+
+            if "apikey" in request.GET:
+                user = get_auth_user(request.GET.get('apikey'))
+
+            if user is not None:
+                get_adapter().login(request, user)

geonode/security/tests.py

@@ -294,6 +294,60 @@ def test_session_ctrl_middleware(self):
         response = self.client.get('/admin')
         self.assertEqual(response.status_code, 302)
 
+    @on_ogc_backend(geoserver.BACKEND_PACKAGE)
+    def test_login_with_headers_middleware(self):
+        """
+        Tests the Geonode login with auth headers and apikey parameter
+        """
+        from django.contrib.sessions.middleware import SessionMiddleware
+        from geonode.security.middleware import LoginWithHeaderOrKeyMiddleware
+        from geonode.base.auth import create_auth_token
+        session_middleware = SessionMiddleware(None)
+        middleware = LoginWithHeaderOrKeyMiddleware(None)
+
+        standard_user = get_user_model().objects.get(username="bobby")
+        access_token = create_auth_token(standard_user)
+
+        request = HttpRequest()
+        request.user = get_anonymous_user()
+
+        request.path = reverse('maps_browse')
+        session_middleware.process_request(request)
+        middleware.process_request(request)
+        self.assertEqual(request.user, get_anonymous_user())
+
+        request = HttpRequest()
+        request.path = reverse('maps_browse')
+        request.user = get_anonymous_user()
+        request.META["HTTP_AUTHORIZATION"] = f'Basic {base64.b64encode(b"fake:fake").decode("utf-8")}'
+        session_middleware.process_request(request)
+        middleware.process_request(request)
+        self.assertEqual(request.user, get_anonymous_user())
+
+        request = HttpRequest()
+        request.path = reverse('maps_browse')
+        request.user = get_anonymous_user()
+        request.META["HTTP_AUTHORIZATION"] = f'Basic {base64.b64encode(b"bobby:bob").decode("utf-8")}'
+        session_middleware.process_request(request)
+        middleware.process_request(request)
+        self.assertEqual(request.user, standard_user)
+
+        request = HttpRequest()
+        request.path = reverse('maps_browse')
+        request.user = get_anonymous_user()
+        request.META["HTTP_AUTHORIZATION"] = f'Bearer {access_token.token}'
+        session_middleware.process_request(request)
+        middleware.process_request(request)
+        self.assertEqual(request.user, standard_user)
+
+        request = HttpRequest()
+        request.path = reverse('maps_browse')
+        request.user = get_anonymous_user()
+        request.GET['apikey'] = access_token.token
+        session_middleware.process_request(request)
+        middleware.process_request(request)
+        self.assertEqual(request.user, standard_user)
+
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
     def test_attributes_sats_refresh(self):
         layers = Layer.objects.all()[:2].values_list('id', flat=True)

geonode/settings.py

@@ -863,6 +863,8 @@
 # Require users to authenticate before using Geonode
 LOCKDOWN_GEONODE = ast.literal_eval(os.getenv('LOCKDOWN_GEONODE', 'False'))
 
+LOGIN_WITH_HEADER_OR_KEY = ast.literal_eval(os.getenv('LOGIN_WITH_HEADER_OR_KEY', 'False'))
+
 # Add additional paths (as regular expressions) that don't require
 # authentication.
 # - authorized exempt urls needed for oauth when GeoNode is set to lockdown
@@ -1935,6 +1937,11 @@ def get_geonode_catalogue_service():
 # SECURITY SETTINGS
 # ########################################################################### #
 
+# Creates a session from valid authentication headers (Basic Auth or Bearer Token) or apikey parameter (token)
+if LOGIN_WITH_HEADER_OR_KEY:
+    MIDDLEWARE += \
+        ('geonode.security.middleware.LoginWithHeaderOrKeyMiddleware',)
+
 # Require users to authenticate before using Geonode
 if LOCKDOWN_GEONODE:
     MIDDLEWARE += \