[Backport to 3.2.x][Fixes #8760] Support Basic Auth for proxied… (#8769)

* [Backport to 3.3.x/3.2.x][Fixes #8760] Support Basic Auth for proxied requests to change styles (#8761)

* [Fixes #8760] Support Basic Auth for proxied requests to change styles

* - Getting rid of the GeoServer BASIC auth-pnly proxified endpoint

* - Test Cases

* [Pep8] Fix flake8 issues

* - Test Cases

(cherry picked from commit f40d9b1)

# Conflicts:
#	geonode/api/tests.py
#	geonode/geoserver/tests/test_server.py
#	geonode/geoserver/views.py
#	geonode/groups/tests.py
#	geonode/proxy/views.py
#	geonode/social/tests.py
(cherry picked from commit e74dc71)

* [LGTM] Fix issues

* [CircleCi] Fix CSW tests

(cherry picked from commit 47f6f4d)

geonode/base/auth.py

@@ -17,7 +17,7 @@
 # along with this program. If not, see <http://www.gnu.org/licenses/>.
 #
 #########################################################################
-
+import re
 import base64
 import datetime
 import logging
@@ -56,7 +56,7 @@ def make_token_expiration(seconds=86400):
     return _expire_time + _expire_delta
 
 
-def create_auth_token(user, client="GeoServer"):
+def create_auth_token(user, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     if not user or user.is_anonymous:
         return None
     expires = make_token_expiration()
@@ -87,7 +87,7 @@ def extend_token(token):
             logger.debug(tb)
 
 
-def get_auth_token(user, client="GeoServer"):
+def get_auth_token(user, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     if not user or user.is_anonymous or not user.is_authenticated:
         return None
     try:
@@ -99,9 +99,23 @@ def get_auth_token(user, client="GeoServer"):
         tb = traceback.format_exc()
         if tb:
             logger.debug(tb)
+    return None
+
+
+def get_auth_user(access_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
+    except Exception:
+        tb = traceback.format_exc()
+        if tb:
+            logger.debug(tb)
+    return None
 
 
-def get_or_create_token(user, client="GeoServer"):
+def get_or_create_token(user, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     if not user or user.is_anonymous:
         return None
     try:
@@ -133,7 +147,7 @@ def get_or_create_token(user, client="GeoServer"):
             logger.debug(tb)
 
 
-def delete_old_tokens(user, client='GeoServer'):
+def delete_old_tokens(user, client=settings.OAUTH2_DEFAULT_BACKEND_CLIENT_NAME):
     if not user or user.is_anonymous:
         return None
     try:
@@ -152,11 +166,12 @@ 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:
+    if re.search('Basic', auth_header, re.IGNORECASE):
         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 ', '')
+        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)
     return None
 
 

geonode/documents/tests.py

@@ -124,7 +124,7 @@ def test_create_document_with_rel(self):
             owner=superuser,
             title='theimg')
 
-        m = Map.objects.all()[0]
+        m = Map.objects.first()
         ctype = ContentType.objects.get_for_model(m)
         _d = DocumentResourceLink.objects.create(
             document_id=c.id,
@@ -273,7 +273,7 @@ def test_document_isuploaded(self):
             'test_img_file.gif',
             self.imgfile.read(),
             'image/gif')
-        m = Map.objects.all()[0]
+        m = Map.objects.first()
 
         self.client.login(username='admin', password='admin')
         response = self.client.post(

geonode/geoserver/tests/integration.py

@@ -133,7 +133,7 @@ def test_capabilities(self):
         """
 
         # a category
-        category = TopicCategory.objects.all()[0]
+        category = TopicCategory.objects.first()
 
         # some users
         norman = get_user_model().objects.get(username="norman")

geonode/geoserver/tests/test_server.py

@@ -25,17 +25,22 @@
 import shutil
 import tempfile
 
-from urllib.parse import urljoin, urlencode
-from django.core.management import call_command
 from os.path import basename, splitext
+from urllib.parse import urljoin, urlencode, urlsplit
 
 from django.conf import settings
 from django.urls import reverse
+from django.test.client import RequestFactory
 from django.contrib.auth import get_user_model
+from django.core.management import call_command
 from django.test.utils import override_settings
+from django.contrib.auth.models import AnonymousUser
 
 from guardian.shortcuts import assign_perm
 
+from geonode.geoserver.helpers import ogc_server_settings
+from geonode.geoserver.views import check_geoserver_access, style_change_check
+
 from geonode import geoserver
 from geonode.base.models import Configuration
 from geonode.decorators import on_ogc_backend
@@ -564,8 +569,7 @@ def test_style_manager(self):
         bob = get_user_model().objects.get(username='bobby')
         assign_perm('change_layer_style', bob, layer)
 
-        logged_in = self.client.login(username='bobby', password='bob')
-        self.assertEqual(logged_in, True)
+        self.assertTrue(self.client.login(username='bobby', password='bob'))
         response = self.client.get(
             reverse(
                 'layer_style_manage', args=(
@@ -648,6 +652,81 @@ def test_style_validity_and_name(self):
             if d is not None:
                 shutil.rmtree(d, ignore_errors=True)
 
+    @on_ogc_backend(geoserver.BACKEND_PACKAGE)
+    def test_style_change_on_basic_auth(self):
+        """
+        Ensures we are able to update the style through a BASIC auth call only.
+        """
+        layer = Layer.objects.filter(default_style__isnull=False).first()
+
+        bob = get_user_model().objects.get(username='bobby')
+        assign_perm('change_layer_style', bob, layer)
+
+        self.assertTrue(bob.has_perm('change_layer_style', obj=layer))
+
+        # Test that HTTP_AUTHORIZATION in request.META is working properly
+        valid_uname_pw = b"bobby:bob"
+        invalid_uname_pw = b"n0t:v@l1d"
+
+        valid_auth_headers = {
+            'HTTP_AUTHORIZATION': f"BASIC {base64.b64encode(valid_uname_pw).decode()}",
+        }
+
+        invalid_auth_headers = {
+            'HTTP_AUTHORIZATION': f"BASIC {base64.b64encode(invalid_uname_pw).decode()}",
+        }
+
+        change_style_url = urljoin(
+            settings.SITEURL,
+            f"/gs/rest/workspaces/{settings.DEFAULT_WORKSPACE}/styles/{layer.name}?raw=true")
+        logger.debug(f"{change_style_url}")
+
+        rf = RequestFactory()
+
+        # Check is 'authorized'
+        post_request = rf.post(
+            change_style_url,
+            data=san_andres_y_providencia_sld,
+            content_type='application/vnd.ogc.sld+xml',
+            **valid_auth_headers
+        )
+        post_request.user = AnonymousUser()
+        raw_url, headers, access_token = check_geoserver_access(
+            post_request,
+            '/gs/rest/workspaces',
+            'rest/workspaces',
+            workspace='geonode',
+            layername=layer.name,
+            allowed_hosts=[urlsplit(ogc_server_settings.public_url).hostname, ])
+        self.assertIsNotNone(raw_url)
+        self.assertIsNotNone(headers)
+        self.assertIsNotNone(access_token)
+
+        authorized = style_change_check(post_request, 'rest/workspaces', access_token=access_token)
+        self.assertTrue(authorized)
+
+        # Check is NOT 'authorized'
+        post_request = rf.post(
+            change_style_url,
+            data=san_andres_y_providencia_sld,
+            content_type='application/vnd.ogc.sld+xml',
+            **invalid_auth_headers
+        )
+        post_request.user = AnonymousUser()
+        raw_url, headers, access_token = check_geoserver_access(
+            post_request,
+            '/gs/rest/workspaces',
+            'rest/workspaces',
+            workspace='geonode',
+            layername=layer.name,
+            allowed_hosts=[urlsplit(ogc_server_settings.public_url).hostname, ])
+        self.assertIsNotNone(raw_url)
+        self.assertIsNotNone(headers)
+        self.assertIsNone(access_token)
+
+        authorized = style_change_check(post_request, 'rest/workspaces', access_token=access_token)
+        self.assertFalse(authorized)
+
     @on_ogc_backend(geoserver.BACKEND_PACKAGE)
     def test_layer_acls(self):
         """ Verify that the layer_acls view is behaving as expected
@@ -678,13 +757,15 @@ def test_layer_acls(self):
             'is_anonymous': False,
             'is_superuser': False,
             'name': 'bobby',
-            'ro': ['geonode:layer2',
-                     'geonode:mylayer',
-                     'geonode:foo',
-                     'geonode:whatever',
-                     'geonode:fooey',
-                     'geonode:quux',
-                     'geonode:fleem'],
+            'ro': [
+                'geonode:layer2',
+                'geonode:mylayer',
+                'geonode:foo',
+                'geonode:whatever',
+                'geonode:fooey',
+                'geonode:quux',
+                'geonode:fleem'
+            ],
             'rw': ['geonode:CA']
         }
         response = self.client.get(reverse('layer_acls'), **valid_auth_headers)

geonode/geoserver/urls.py

@@ -41,12 +41,12 @@
                                              downstream_path='ows'), name='ows_endpoint'),
     url(r'^gwc', views.geoserver_proxy, dict(proxy_path='/gs/gwc',
                                              downstream_path='gwc'), name='gwc_endpoint'),
-    url(r'^wms', views.geoserver_protected_proxy, dict(proxy_path='/gs/wms',
-                                                       downstream_path='wms'), name='wms_endpoint'),
-    url(r'^wfs', views.geoserver_protected_proxy, dict(proxy_path='/gs/wfs',
-                                                       downstream_path='wfs'), name='wfs_endpoint'),
-    url(r'^wcs', views.geoserver_protected_proxy, dict(proxy_path='/gs/wcs',
-                                                       downstream_path='wcs'), name='wcs_endpoint'),
+    url(r'^wms', views.geoserver_proxy, dict(proxy_path='/gs/wms',
+                                             downstream_path='wms'), name='wms_endpoint'),
+    url(r'^wfs', views.geoserver_proxy, dict(proxy_path='/gs/wfs',
+                                             downstream_path='wfs'), name='wfs_endpoint'),
+    url(r'^wcs', views.geoserver_proxy, dict(proxy_path='/gs/wcs',
+                                             downstream_path='wcs'), name='wcs_endpoint'),
     url(r'^wps', views.geoserver_proxy, dict(proxy_path='/gs/wps',
                                              downstream_path='wps'), name='wps_endpoint'),
     url(r'^pdf', views.geoserver_proxy, dict(proxy_path='/gs/pdf',