Merge with Master

geonode/base/auth.py

@@ -18,8 +18,8 @@
 #
 #########################################################################
 
-import datetime
 import base64
+import datetime
 import logging
 import traceback
 
@@ -153,12 +153,7 @@ def delete_old_tokens(user, client='GeoServer'):
 
 def get_token_from_auth_header(auth_header, create_if_not_exists=False):
     if 'Basic' in auth_header:
-        encoded_credentials = auth_header.split(' ')[1]  # Removes "Basic " to isolate credentials
-        decoded_credentials = base64.b64decode(encoded_credentials).decode("utf-8").split(':')
-        username = decoded_credentials[0]
-        password = decoded_credentials[1]
-        # if the credentials are correct, then the feed_bot is not None, but is a User object.
-        user = authenticate(username=username, password=password)
+        user = basic_auth_authenticate_user(auth_header)
         return get_auth_token(user) if not create_if_not_exists else get_or_create_token(user)
     elif 'Bearer' in auth_header:
         return auth_header.replace('Bearer ', '')
@@ -190,3 +185,17 @@ def get_token_object_from_session(session):
 def remove_session_token(session):
     if 'access_token' in session:
         del session['access_token']
+
+
+def basic_auth_authenticate_user(auth_header: str):
+    """
+    Function performing user authentication based on BasicAuth Authorization header
+
+    :param auth_header: Authorization header of the request
+    """
+    encoded_credentials = auth_header.split(' ')[1]  # Removes "Basic " to isolate credentials
+    decoded_credentials = base64.b64decode(encoded_credentials).decode("utf-8").split(':')
+    username = decoded_credentials[0]
+    password = decoded_credentials[1]
+
+    return authenticate(username=username, password=password)

geonode/security/middleware.py

@@ -28,7 +28,7 @@
 
 from geonode import geoserver
 from geonode.utils import check_ogc_backend
-from geonode.base.auth import get_token_object_from_session
+from geonode.base.auth import get_token_object_from_session, basic_auth_authenticate_user
 
 from guardian.shortcuts import get_anonymous_user
 
@@ -62,21 +62,39 @@ class LoginRequiredMiddleware(MiddlewareMixin):
 
     """
     Requires a user to be logged in to access any page that is not white-listed.
+
+    This middleware simply checks user property of a request, to determine whether the query is authenticated or not,
+    but since DRF assumes correlation between session authentication and presence of user property in the request,
+    an additional check was introduced in the middleware, to allow Basic authenticated requests without additional
+    middleware setting this property (otherwise, all DRF views configured with:
+    `authentication_classes = [SessionAuthentication,]`
+    would accept Basic authenticated requests (regardless of presence of `BasicAuthentication` in view's
+    authentication_classes).
     """
 
-    redirect_to = getattr(settings, 'LOGIN_URL', reverse('account_login'))
+    redirect_to = getattr(settings, "LOGIN_URL", reverse("account_login"))
 
     def __init__(self, get_response):
         self.get_response = get_response
 
     def process_request(self, request):
-        if not request.user.is_authenticated or \
-        request.user == get_anonymous_user():
+
+        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):
                 return HttpResponseRedirect(
-                    '{login_path}?next={request_path}'.format(
-                        login_path=self.redirect_to,
-                        request_path=request.path))
+                    "{login_path}?next={request_path}".format(login_path=self.redirect_to, request_path=request.path)
+                )
 
 
 class SessionControlMiddleware(MiddlewareMixin):

geonode/security/tests.py

@@ -165,6 +165,45 @@ def test_login_middleware(self):
             response = middleware.process_request(request)
             self.assertIsNone(response)
 
+    @on_ogc_backend(geoserver.BACKEND_PACKAGE)
+    @dump_func_name
+    def test_login_middleware_with_basic_auth(self):
+        """
+        Tests the Geonode login required authentication middleware with Basic authenticated queries
+        """
+        from geonode.security.middleware import LoginRequiredMiddleware
+        middleware = LoginRequiredMiddleware(None)
+
+        black_listed_url = reverse('maps_browse')
+        white_listed_url = reverse('account_login')
+
+        # unauthorized request to black listed URL should be redirected to `redirect_to` URL
+        request = HttpRequest()
+        request.user = get_anonymous_user()
+
+        request.path = black_listed_url
+        response = middleware.process_request(request)
+        if response:
+            self.assertEqual(response.status_code, 302)
+            self.assertTrue(
+                response.get('Location').startswith(
+                    middleware.redirect_to))
+
+        # unauthorized request to white listed URL should be allowed
+        request.path = white_listed_url
+        response = middleware.process_request(request)
+        self.assertIsNone(
+            response,
+            msg="Middleware activated for white listed path: {0}".format(black_listed_url))
+
+        # Basic authorized request to black listed URL should be allowed
+        request.path = black_listed_url
+        request.META["HTTP_AUTHORIZATION"] = f'Basic {base64.b64encode(b"bobby:bob").decode("utf-8")}'
+        response = middleware.process_request(request)
+        self.assertIsNone(
+            response,
+            msg="Middleware activated for white listed path: {0}".format(black_listed_url))
+
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
     @dump_func_name
     def test_session_ctrl_middleware(self):