Validate JSON complex input file paths

.github/workflows/main.yml

@@ -68,6 +68,7 @@ jobs:
         env:
           TOX_GH_MAJOR_MINOR: ${{ matrix.python-version }}
       - name: Report Coverage
+        if: ${{ matrix.os == 'ubuntu' }}
         uses: coverallsapp/github-action@v2
         with:
           flag-name: run-${{ matrix.python-version }}

pyproject.toml

@@ -92,6 +92,7 @@ replace = "{new_version}"
 
 [tool.coverage.run]
 relative_files = true
+omit = ["pywps/templates/*"]
 
 [tool.pytest.ini_options]
 addopts = [
@@ -121,4 +122,3 @@ lint.select = ["E", "W", "F", "C90"]   # Flake8-equivalent rule families
 lint.ignore = ["F401", "E402", "C901"]
 line-length = 120
 exclude = ["tests"]
-

pywps/app/Process.py

@@ -317,7 +317,8 @@ def launch_next_process(self):
             request_json = request_json.decode('utf-8')
             LOGGER.debug("Launching the stored request {}".format(str(uuid)))
             new_wps_request = WPSRequest()
-            new_wps_request.json = json.loads(request_json)
+            # This request was saved by PyWPS, so it is trusted.
+            new_wps_request.restore_json(json.loads(request_json))
             process_identifier = new_wps_request.identifier
             process = self.service.prepare_process_for_execution(process_identifier)
             process._set_uuid(uuid)

pywps/app/WPSRequest.py

@@ -17,6 +17,7 @@
 from pywps.app.basic import get_xpath_ns, parse_http_url
 from pywps.configuration import wps_strict
 from pywps.exceptions import (
+    FileURLNotSupported,
     FileSizeExceeded,
     InvalidParameterValue,
     MissingParameterValue,
@@ -145,7 +146,8 @@ def _post_request(self):
 
             if self.preprocess_request is not None:
                 jdoc = self.preprocess_request(jdoc, http_request=self.http_request)
-            self.json = jdoc
+            # Load client input and validate any local file paths.
+            self.load_json(jdoc)
 
             version = jdoc.get('version')
             self.set_version(version)
@@ -456,6 +458,21 @@ def json(self, value):
 
         :param value: the json (not string) representation
         """
+        self.load_json(value)
+
+    def load_json(self, value):
+        """Load a request and validate its input file paths."""
+        self._set_json(value, validate_file=True)
+
+    def restore_json(self, value):
+        """Restore a request previously saved by PyWPS.
+
+        Input files are already in the job work directory and do not need to
+        be checked against ``allowedinputpaths`` again.
+        """
+        self._set_json(value, validate_file=False)
+
+    def _set_json(self, value, validate_file=True):
 
         self.operation = value.get('operation')
         self.version = value.get('version')
@@ -482,8 +499,10 @@ def json(self, value):
                 if 'identifier' not in inpt_def:
                     inpt_def['identifier'] = identifier
                 try:
-                    inpt = input_from_json(inpt_def)
+                    inpt = input_from_json(inpt_def, validate_file=validate_file)
                     self.inputs[identifier].append(inpt)
+                except FileURLNotSupported:
+                    raise
                 except Exception as e:
                     LOGGER.warning(e)
                     LOGGER.warning(f'skipping input: {identifier}')

pywps/inout/inputs.py

@@ -5,6 +5,8 @@
 import base64
 import re
 from copy import deepcopy
+from pathlib import Path
+from urllib.parse import urlparse
 
 from pywps import xml_util as etree
 from pywps.app.Common import Metadata
@@ -185,7 +187,7 @@ def json(self):
         return data
 
     @classmethod
-    def from_json(cls, json_input):
+    def from_json(cls, json_input, validate_file=True):
         data_format = json_input.get('data_format')
         if data_format is not None:
             data_format = Format(
@@ -215,9 +217,14 @@ def from_json(cls, json_input):
         )
         instance.as_reference = json_input.get('asreference', False)
         if json_input.get('file'):
+            if validate_file:
+                instance._validate_file_input(Path(json_input['file']).absolute().as_uri())
             instance.file = json_input['file']
         elif json_input.get('href'):
-            instance.url = json_input['href']
+            href = json_input['href']
+            if validate_file and urlparse(href).scheme == 'file':
+                instance._validate_file_input(href)
+            instance.url = href
         elif json_input.get('data'):
             data = json_input['data']
             if data_format.encoding == 'base64':
@@ -377,10 +384,10 @@ def clone(self):
         return deepcopy(self)
 
 
-def input_from_json(json_data):
+def input_from_json(json_data, validate_file=True):
     data_type = json_data.get('type', 'literal')
     if data_type in ['complex', 'reference']:
-        inpt = ComplexInput.from_json(json_data)
+        inpt = ComplexInput.from_json(json_data, validate_file=validate_file)
     elif data_type == 'literal':
         inpt = LiteralInput.from_json(json_data)
     elif data_type == 'bbox':

pywps/processing/job.py

@@ -56,7 +56,8 @@ def from_json(cls, value):
         """
         process = Process.from_json(value['process'])
         wps_request = WPSRequest()
-        wps_request.json = json.loads(value['wps_request'])
+        # This request was saved by PyWPS, so it is trusted.
+        wps_request.restore_json(json.loads(value['wps_request']))
         wps_response = ExecuteResponse(
             wps_request=wps_request,
             uuid=process.uuid,

tests/basic.py

@@ -21,6 +21,7 @@ def setUp(self) -> None:
         set('server', 'temp_path', f"{self.tmpdir.name}/temp_path")
         set('server', 'outputpath', f"{self.tmpdir.name}/outputpath")
         set('server', 'workdir', f"{self.tmpdir.name}/workdir")
+        set('server', 'allowedinputpaths', self.tmpdir.name)
 
         set('logging', 'level', 'DEBUG')
         set('logging', 'file', f"{self.tmpdir.name}/logging-file.log")

tests/test_execute.py

@@ -142,6 +142,24 @@ def complex_proces(request, response):
                    ])
 
 
+def create_complex_input_process():
+    def handler(request, response):
+        response.outputs['text'].data = request.inputs['complex'][0].data
+        return response
+
+    return Process(
+        handler=handler,
+        identifier='complex_input_process',
+        title='Complex input process',
+        inputs=[ComplexInput(
+            'complex',
+            'Complex input',
+            supported_formats=[FORMATS.TEXT],
+        )],
+        outputs=[LiteralOutput('text', 'Text output')],
+    )
+
+
 def create_complex_nc_process():
     def complex_proces(request, response):
         from pywps.dependencies import netCDF4 as nc
@@ -447,6 +465,27 @@ def test_post_with_no_inputs(self):
         resp = client.post_json(doc=request)
         assert_response_success_json(resp, result)
 
+    def test_json_file_is_validated(self):
+        request = {
+            'identifier': 'complex_input_process',
+            'version': '1.0.0',
+            'inputs': {
+                'complex': {
+                    'type': 'complex',
+                    'file': '/etc/passwd',
+                    'mimeType': 'text/plain',
+                    'data_format': {'mime_type': 'text/plain'},
+                    'supported_formats': [{'mime_type': 'text/plain'}],
+                },
+            },
+        }
+        client = client_for(Service(processes=[create_complex_input_process()]))
+
+        resp = client.post_json(doc=request)
+
+        assert resp.status_code == 400
+        assert json.loads(resp.data)['name'] == 'FileURLNotSupported'
+
     def test_post_with_string_input(self):
         request = {
             'identifier': 'greeter',

tests/test_inout.py

@@ -28,7 +28,7 @@
 from io import StringIO
 from urllib.parse import urlparse
 from pywps.validator.base import emptyvalidator
-from pywps.exceptions import InvalidParameterValue
+from pywps.exceptions import FileURLNotSupported, InvalidParameterValue
 from pywps.validator.mode import MODE
 from pywps.inout.basic import UOM
 from pywps.inout.storage.file import FileStorageBuilder
@@ -273,6 +273,28 @@ def test_complex_input_file(self):
         self.assertEqual(complex.file, complex2.file)
         self.assertEqual(complex.data, complex2.data)
 
+    def test_complex_input_file_outside_allowed_paths(self):
+        complex = self.make_complex_input()
+        with tempfile.TemporaryDirectory(prefix="pywps_not_allowed_") as outside_dir:
+            some_file = os.path.join(outside_dir, "some_file.txt")
+            with open(some_file, "w") as f:
+                f.write("some data")
+            complex.file = some_file
+
+            with self.assertRaises(FileURLNotSupported):
+                inout.inputs.ComplexInput.from_json(complex.json)
+
+    def test_complex_input_file_url_outside_allowed_paths(self):
+        complex = self.make_complex_input()
+        with tempfile.TemporaryDirectory(prefix="pywps_not_allowed_") as outside_dir:
+            some_file = os.path.join(outside_dir, "some_file.txt")
+            with open(some_file, "w") as f:
+                f.write("some data")
+            complex.url = f"file:{some_file}"
+
+            with self.assertRaises(FileURLNotSupported):
+                inout.inputs.ComplexInput.from_json(complex.json)
+
     def test_complex_input_data(self):
         complex = self.make_complex_input()
         complex.data = "some data"