Skip to content

Commit

Permalink
[Backport to 3.2.x][Fixes #8858] Support "workspaced" requests on Geo…
Browse files Browse the repository at this point in the history 
…Server proxy view (#8861)

* [Fixes #8858] Support "workspaced" requests on GeoServer proxy view

(cherry picked from commit 85e45c6)

* [LGTM] Polynomial regular expression used on uncontrolled data

* [LGTM] Polynomial regular expression used on uncontrolled data

(cherry picked from commit 2908584)
  • Loading branch information
Alessio Fabiani authored Feb 25, 2022
1 parent bf01d78 commit 89f8508
Show file tree
Hide file tree
Showing 2 changed files with 62 additions and 41 deletions.
37 changes: 31 additions & 6 deletions geonode/geoserver/tests/test_server.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -691,7 +691,7 @@ def test_style_change_on_basic_auth(self):
**valid_auth_headers
)
post_request.user = AnonymousUser()
raw_url, headers, access_token = check_geoserver_access(
raw_url, headers, access_token, downstream_path = check_geoserver_access(
post_request,
'/gs/rest/workspaces',
'rest/workspaces',
Expand All @@ -702,7 +702,19 @@ def test_style_change_on_basic_auth(self):
self.assertIsNotNone(headers)
self.assertIsNotNone(access_token)

authorized = style_change_check(post_request, 'rest/workspaces', access_token=access_token)
authorized = style_change_check(post_request, downstream_path, style_name='styles', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(post_request, 'rest/styles', style_name=f'{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(post_request, f'rest/workspaces/{layer.workspace}/styles/{layer.name}', style_name=f'{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(post_request, f'rest/layers/{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(post_request, f'rest/workspaces/{layer.workspace}/layers/{layer.name}', access_token=access_token)
self.assertTrue(authorized)

put_request = rf.put(
Expand All @@ -712,7 +724,7 @@ def test_style_change_on_basic_auth(self):
**valid_auth_headers
)
put_request.user = AnonymousUser()
raw_url, headers, access_token = check_geoserver_access(
raw_url, headers, access_token, downstream_path = check_geoserver_access(
put_request,
'/gs/rest/workspaces',
'rest/workspaces',
Expand All @@ -728,7 +740,20 @@ def test_style_change_on_basic_auth(self):
# ref: 05b000cdb06b0b6e9b72bd9eb8a8e03abeb204a8
# [Regression] "style_change_check" always fails in the case the style does not exist on GeoNode too, preventing a user editing temporary generated styles
Style.objects.filter(name=layer.name).delete()
authorized = style_change_check(put_request, 'rest/workspaces', access_token=access_token)

authorized = style_change_check(put_request, downstream_path, style_name='styles', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(put_request, 'rest/styles', style_name=f'{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(put_request, f'rest/workspaces/{layer.workspace}/styles/{layer.name}', style_name=f'{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(put_request, f'rest/layers/{layer.name}', access_token=access_token)
self.assertTrue(authorized)

authorized = style_change_check(put_request, f'rest/workspaces/{layer.workspace}/layers/{layer.name}', access_token=access_token)
self.assertTrue(authorized)

# Check is NOT 'authorized'
Expand All @@ -739,7 +764,7 @@ def test_style_change_on_basic_auth(self):
**invalid_auth_headers
)
post_request.user = AnonymousUser()
raw_url, headers, access_token = check_geoserver_access(
raw_url, headers, access_token, downstream_path = check_geoserver_access(
post_request,
'/gs/rest/workspaces',
'rest/workspaces',
Expand All @@ -750,7 +775,7 @@ def test_style_change_on_basic_auth(self):
self.assertIsNotNone(headers)
self.assertIsNone(access_token)

authorized = style_change_check(post_request, 'rest/workspaces', access_token=access_token)
authorized = style_change_check(post_request, downstream_path, access_token=access_token)
self.assertFalse(authorized)

@on_ogc_backend(geoserver.BACKEND_PACKAGE)
Expand Down
66 changes: 31 additions & 35 deletions geonode/geoserver/views.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
#########################################################################

import os
import re
import json
Expand Down Expand Up @@ -345,7 +346,7 @@ def layer_style_manage(request, layername):
)


def style_change_check(request, path, access_token=None):
def style_change_check(request, path, style_name=None, access_token=None):
"""
If the layer has not change_layer_style permission, return a status of
401 (unauthorized)
Expand All @@ -363,16 +364,12 @@ def style_change_check(request, path, access_token=None):
if request.method in ('PUT', 'POST'):
if not request.user.is_authenticated and not access_token:
authorized = False
elif path == 'rest/layers' and request.method == 'PUT':
# layer update, should be safe to always authorize it
authorized = True
else:
elif re.match(r'^.*(?<!/rest/)/rest/.*/?styles.*', path):
# style new/update
# we will iterate all layers (should be just one if not using GS)
# to which the posted style is associated
# and check if the user has change_style_layer permissions on each
# of them
style_name = os.path.splitext(request.path)[0].split('/')[-1]
if style_name == 'styles' and 'raw' in request.GET:
authorized = True
elif re.match(temp_style_name_regex, style_name):
Expand All @@ -394,7 +391,8 @@ def style_change_check(request, path, access_token=None):
authorized = True
break
except Style.DoesNotExist:
logger.warn(f'There is not a style with such a name: {style_name}.')
if request.method != 'POST':
logger.warn(f'There is not a style with such a name: {style_name}.')
except Exception as e:
logger.exception(e)
authorized = (request.method == 'POST') # The user is probably trying to create a new style
Expand Down Expand Up @@ -468,7 +466,7 @@ def strip_prefix(path, prefix):

# Collecting headers and cookies
headers, access_token = get_headers(request, url, unquote(raw_url), allowed_hosts=allowed_hosts)
return (raw_url, headers, access_token)
return (raw_url, headers, access_token, downstream_path)


@csrf_exempt
Expand All @@ -484,7 +482,7 @@ def geoserver_proxy(request,
affected_layers = None
allowed_hosts = [urlsplit(ogc_server_settings.public_url).hostname, ]

raw_url, headers, access_token = check_geoserver_access(
raw_url, headers, access_token, downstream_path = check_geoserver_access(
request,
proxy_path,
downstream_path,
Expand All @@ -493,36 +491,34 @@ def geoserver_proxy(request,
allowed_hosts=allowed_hosts)
url = urlsplit(raw_url)

if request.method in ("POST", "PUT", "DELETE"):
if downstream_path in ('rest/styles', 'rest/layers',
'rest/workspaces'):
if not style_change_check(request, downstream_path, access_token=access_token):
if re.match(r'^.*/rest/', url.path) and request.method in ("POST", "PUT", "DELETE"):
if re.match(r'^.*(?<!/rest/)/rest/.*/?styles.*', url.path):
logger.debug(
f"[geoserver_proxy] Updating Style ---> url {url.geturl()}")
_style_name, _style_ext = os.path.splitext(os.path.basename(urlsplit(url.geturl()).path))
_parsed_get_args = dict(parse_qsl(urlsplit(url.geturl()).query))
if _style_name == 'styles.json' and request.method == "PUT":
if _parsed_get_args.get('name'):
_style_name, _style_ext = os.path.splitext(_parsed_get_args.get('name'))
else:
_style_name, _style_ext = os.path.splitext(_style_name)

if not style_change_check(request, url.path, style_name=_style_name, access_token=access_token):
return HttpResponse(
_("You don't have permissions to change style for this layer"),
content_type="text/plain",
status=401)
elif downstream_path == 'rest/styles':
logger.debug(
f"[geoserver_proxy] Updating Style ---> url {url.geturl()}")
_style_name, _style_ext = os.path.splitext(os.path.basename(urlsplit(url.geturl()).path))
_parsed_get_args = dict(parse_qsl(urlsplit(url.geturl()).query))
if _style_name == 'styles.json' and request.method == "PUT":
if _parsed_get_args.get('name'):
_style_name, _style_ext = os.path.splitext(_parsed_get_args.get('name'))
else:
_style_name, _style_ext = os.path.splitext(_style_name)
if _style_name != 'style-check' and (_style_ext == '.json' or _parsed_get_args.get('raw')) and \
not re.match(temp_style_name_regex, _style_name):
affected_layers = style_update(request, raw_url, workspace)
elif downstream_path == 'rest/layers':
logger.debug(
f"[geoserver_proxy] Updating Layer ---> url {url.geturl()}")
try:
_layer_name = os.path.splitext(os.path.basename(request.path))[0]
_layer = Layer.objects.get(name=_layer_name)
affected_layers = [_layer]
except Exception:
logger.warn(f"Could not find any Layer {os.path.basename(request.path)} on DB")
if _style_name != 'style-check' and (_style_ext == '.json' or _parsed_get_args.get('raw')) and \
not re.match(temp_style_name_regex, _style_name):
affected_layers = style_update(request, raw_url, workspace)
elif re.match(r'^.*(?<!/rest/)/rest/.*/?layers.*', url.path):
logger.debug(f"[geoserver_proxy] Updating Dataset ---> url {url.geturl()}")
try:
_layer_name = os.path.splitext(os.path.basename(request.path))[0]
_layer = Layer.objects.get(name=_layer_name)
affected_layers = [_layer]
except Exception:
logger.warn(f"Could not find any Layer {os.path.basename(request.path)} on DB")

kwargs = {'affected_layers': affected_layers}
raw_url = unquote(raw_url)
Expand Down

0 comments on commit 89f8508

Please sign in to comment.