bug fix for incomplete xml imports

docs/samples/automation/retriever.py

@@ -31,6 +31,7 @@
 import io
 import os
 import xml
+import defusedxml.minidom
 import requests
 from fs import open_fs
 import PIL.Image
@@ -58,7 +59,7 @@ def load_from_ftml(filename):
     _fs = open_fs(_dirname)
     datasource = _fs.open(_name)
     try:
-        doc = xml.dom.minidom.parse(datasource)
+        doc = defusedxml.minidom.parse(datasource)
     except xml.parsers.expat.ExpatError as ex:
         raise SyntaxError(str(ex))
 

mslib/mscolab/file_manager.py

@@ -31,6 +31,7 @@
 import git
 import threading
 from sqlalchemy.exc import IntegrityError
+from mslib.utils.verify_waypoint_data import verify_waypoint_data
 from mslib.mscolab.models import db, Operation, Permission, User, Change, Message
 from mslib.mscolab.conf import mscolab_settings
 
@@ -53,9 +54,19 @@ def _get_operation_lock(self, op_id):
 
     def create_operation(self, path, description, user, last_used=None, content=None, category="default", active=True):
         """
-        path: path to the operation
-        description: description of the operation
+        Creates a new operation in the mscolab system.
+
+        :param path: The path of the operation.
+        :param description: The description of the operation.
+        :param user: The user object creating the operation.
+        :param last_used: The last used datetime of the operation. Default is None.
+        :param content: The content of the operation. Default is None.
+        :param category: The category of the operation. Default is 'default'.
+        :param active: The activity status of the operation. Default is True.
+        :return: True if the operation is created successfully, False otherwise.
         """
+        if content is not None and not verify_waypoint_data(content):
+            return False
         # set codes on these later
         if path.find("/") != -1 or path.find("\\") != -1 or (" " in path):
             logging.debug("malicious request: %s", user)
@@ -263,14 +274,14 @@ def update_operation(self, op_id, attribute, value, user):
             if value.find("/") != -1 or value.find("\\") != -1 or (" " in value):
                 logging.debug("malicious request: %s", user)
                 return False
-            data = fs.open_fs(self.data_dir)
-            if data.exists(value):
-                return False
-            # will be move when operations are introduced
-            # make a directory, else movedir
-            data.makedir(value)
-            data.movedir(operation.path, value)
-            # when renamed to a Group operation
+            with fs.open_fs(self.data_dir) as data:
+                if data.exists(value):
+                    return False
+                # will be move when operations are introduced
+                # make a directory, else movedir
+                data.makedir(value)
+                data.movedir(operation.path, value)
+                # when renamed to a Group operation
             if value.endswith(mscolab_settings.GROUP_POSTFIX):
                 # getting the category
                 category = value.split(mscolab_settings.GROUP_POSTFIX)[0]
@@ -318,6 +329,8 @@ def save_file(self, op_id, content, user, comment=""):
         content: content of the file to be saved
         # ToDo save change in schema
         """
+        if not verify_waypoint_data(content):
+            return False
         # ToDo use comment
         operation = Operation.query.filter_by(id=op_id).first()
         if not operation:

mslib/msui/flighttrack.py

@@ -48,6 +48,7 @@
 from mslib.utils.units import units
 from mslib.utils.coordinate import find_location, path_points, get_distance
 from mslib.utils import thermolib
+from mslib.utils.verify_waypoint_data import verify_waypoint_data
 from mslib.utils.config import config_loader, save_settings_qsettings, load_settings_qsettings
 from mslib.utils.config import MSUIDefaultConfig as mss_default
 from mslib.utils.qt import variant_to_string, variant_to_float
@@ -649,8 +650,11 @@ def load_from_ftml(self, filename):
 
     def load_from_xml_data(self, xml_content, name="Flight track"):
         self.name = name
-        _waypoints_list = load_from_xml_data(xml_content, name)
-        self.replace_waypoints(_waypoints_list)
+        if verify_waypoint_data(xml_content):
+            _waypoints_list = load_from_xml_data(xml_content, name)
+            self.replace_waypoints(_waypoints_list)
+        else:
+            raise SyntaxError(f"Invalid flight track filename: {name}")
 
     def get_filename(self):
         return self.filename

mslib/msui/mscolab.py

@@ -57,6 +57,7 @@
 
 from mslib.utils.auth import get_password_from_keyring, save_password_to_keyring
 from mslib.utils.verify_user_token import verify_user_token
+from mslib.utils.verify_waypoint_data import verify_waypoint_data
 from mslib.utils.qt import get_open_filename, get_save_filename, dropEvent, dragEnterEvent, show_popup
 from mslib.msui.qt5 import ui_mscolab_help_dialog as msc_help_dialog
 from mslib.msui.qt5 import ui_add_operation_dialog as add_operation_ui
@@ -1962,10 +1963,12 @@ def handle_import_msc(self, file_path, extension, function, pickertype):
                 return
             dir_path, file_name = fs.path.split(file_path)
             file_name = fs.path.basename(file_path)
-            name, file_ext = fs.path.splitext(file_name)
             if function is None:
                 with open_fs(dir_path) as file_dir:
                     xml_content = file_dir.readtext(file_name)
+                    if not verify_waypoint_data(xml_content):
+                        show_popup(self.ui, "Import Failed", f"The file - {file_name}, does not contain valid XML")
+                        return
                 try:
                     model = ft.WaypointsTableModel(xml_content=xml_content)
                 except SyntaxError:
@@ -1977,6 +1980,9 @@ def handle_import_msc(self, file_path, extension, function, pickertype):
                 model = ft.WaypointsTableModel(waypoints=new_waypoints)
                 xml_doc = self.waypoints_model.get_xml_doc()
                 xml_content = xml_doc.toprettyxml(indent="  ", newl="\n")
+            if not verify_waypoint_data(xml_content):
+                show_popup(self.ui, "Import Failed", f"The file - {file_name}, was not imported!", 0)
+                return
             self.waypoints_model.dataChanged.disconnect(self.handle_waypoints_changed)
             self.waypoints_model = model
             self.handle_waypoints_changed()

mslib/utils/verify_waypoint_data.py

@@ -0,0 +1,53 @@
+# -*- coding: utf-8 -*-
+"""
+
+    mslib.utils.verify_waypoint_data
+    ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+    basic checks for xml waypoint data.
+
+    This file is part of MSS.
+
+    :copyright: Copyright 2024 Reimar Bauer
+    :license: APACHE-2.0, see LICENSE for details.
+
+    Licensed under the Apache License, Version 2.0 (the "License");
+    you may not use this file except in compliance with the License.
+    You may obtain a copy of the License at
+
+       http://www.apache.org/licenses/LICENSE-2.0
+
+    Unless required by applicable law or agreed to in writing, software
+    distributed under the License is distributed on an "AS IS" BASIS,
+    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+    See the License for the specific language governing permissions and
+    limitations under the License.
+"""
+
+
+import defusedxml.minidom
+import xml.parsers.expat
+
+
+def verify_waypoint_data(xml_content):
+    try:
+        doc = defusedxml.minidom.parseString(xml_content)
+    except xml.parsers.expat.ExpatError:
+        return False
+
+    ft_el = doc.getElementsByTagName("FlightTrack")[0]
+    waypoints = ft_el.getElementsByTagName("Waypoint")
+    if (len(waypoints)) < 2:
+        return False
+
+    for wp_el in ft_el.getElementsByTagName("Waypoint"):
+        try:
+            wp_el.getAttribute("location")
+            float(wp_el.getAttribute("lat"))
+            float(wp_el.getAttribute("lon"))
+            float(wp_el.getAttribute("flightlevel"))
+            wp_el.getElementsByTagName("Comments")[0]
+        except ValueError:
+            return False
+
+    return True

tests/_test_mscolab/test_files.py

@@ -32,6 +32,7 @@
 from mslib.mscolab.models import User, Operation, Permission, Change, Message
 from mslib.mscolab.seed import add_user, get_user
 from mslib.mscolab.utils import get_recent_op_id
+from tests.utils import XML_CONTENT1, XML_CONTENT2, XML_CONTENT3
 
 
 class Test_Files:
@@ -94,25 +95,54 @@ def test_is_creator(self):
     def test_file_save(self):
         with self.app.test_client():
             flight_path, operation = self._create_operation(flight_path="operation77")
-            assert self.fm.save_file(operation.id, "beta", self.user)
-            assert self.fm.get_file(operation.id, self.user) == "beta"
-            assert self.fm.save_file(operation.id, "gamma", self.user)
-            assert self.fm.get_file(operation.id, self.user) == "gamma"
+            assert self.fm.save_file(operation.id, XML_CONTENT1, self.user)
+            assert self.fm.get_file(operation.id, self.user) == XML_CONTENT1
+            assert self.fm.save_file(operation.id, XML_CONTENT2, self.user)
+            assert self.fm.get_file(operation.id, self.user) == XML_CONTENT2
             # check if change is saved properly
             changes = self.fm.get_all_changes(operation.id, self.user)
             assert len(changes) == 2
 
+    def test_cant_save(self):
+        with self.app.test_client():
+            flight_path, operation = self._create_operation(flight_path="operation911")
+            assert self.fm.save_file(operation.id, "text", self.user) is False
+            incomplete = """<?xml version="1.0" encoding="utf-8"?>
+  <FlightTrack version="9.1.0">
+    <ListOfWaypoints/>
+  </FlightTrack>"""
+            assert self.fm.save_file(operation.id, incomplete, self.user) is False
+            incomplete = """<?xml version="1.0" encoding="utf-8"?>
+  <FlightTrack version="9.1.0.">
+    <ListOfWaypoints>
+      <Waypoint flightlevel="350">
+        <Comments></Comments>
+      </Waypoint>
+      <Waypoint flightlevel="350">
+        <Comments></Comments>
+      </Waypoint>
+    </ListOfWaypoints>
+  </FlightTrack>"""
+            assert self.fm.save_file(operation.id, incomplete, self.user) is False
+
+    def test_stub_data(self):
+        with self.app.test_client():
+            flight_path, operation = self._create_operation(flight_path="operationstub")
+            content = self.fm.get_file(operation.id, self.user)
+            assert flight_path == "operationstub"
+            assert content == mscolab_settings.STUB_CODE
+
     def test_undo(self):
         with self.app.test_client():
-            flight_path, operation = self._create_operation(flight_path="operation7", content="alpha")
-            assert self.fm.save_file(operation.id, "beta", self.user)
-            assert self.fm.save_file(operation.id, "gamma", self.user)
+            flight_path, operation = self._create_operation(flight_path="operation7", content=XML_CONTENT1)
+            assert self.fm.save_file(operation.id, XML_CONTENT2, self.user)
+            assert self.fm.save_file(operation.id, XML_CONTENT3, self.user)
             changes = Change.query.filter_by(op_id=operation.id).all()
             assert changes is not None
             assert changes[0].id == 1
             assert self.fm.undo_changes(changes[0].id, self.user) is True
             assert len(self.fm.get_all_changes(operation.id, self.user)) == 3
-            assert "beta" in self.fm.get_file(operation.id, self.user)
+            assert XML_CONTENT2 == self.fm.get_file(operation.id, self.user)
 
     def test_get_operation(self):
         with self.app.test_client():
@@ -156,8 +186,7 @@ def test_delete_operation(self):
             assert len(messages) == 0
 
     def _example_data(self):
-        self.content1 = """\
-    <?xml version="1.0" encoding="utf-8"?>
+        self.content1 = """<?xml version="1.0" encoding="utf-8"?>
       <FlightTrack>
         <Name>new flight track (1)</Name>
         <ListOfWaypoints>
@@ -178,8 +207,7 @@ def _example_data(self):
           </Waypoint>
         </ListOfWaypoints>
       </FlightTrack>"""
-        self.content2 = """\
-    <?xml version="1.0" encoding="utf-8"?>
+        self.content2 = """<?xml version="1.0" encoding="utf-8"?>
       <FlightTrack>
         <Name>new flight track (1)</Name>
         <ListOfWaypoints>

tests/_test_mscolab/test_files_api.py

@@ -29,6 +29,7 @@
 
 from mslib.mscolab.models import Operation
 from mslib.mscolab.seed import add_user, get_user
+from tests.utils import XML_CONTENT1, XML_CONTENT2, XML_CONTENT3
 
 
 class Test_Files:
@@ -170,8 +171,8 @@ def test_delete_operation(self):
     def test_get_all_changes(self):
         with self.app.test_client():
             flight_path, operation = self._create_operation(flight_path="V11")
-            assert self.fm.save_file(operation.id, "content1", self.user)
-            assert self.fm.save_file(operation.id, "content2", self.user)
+            assert self.fm.save_file(operation.id, XML_CONTENT1, self.user)
+            assert self.fm.save_file(operation.id, XML_CONTENT2, self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
             # the newest change is on index 0, because it has a recent created_at time
             assert len(all_changes) == 2
@@ -181,20 +182,20 @@ def test_get_all_changes(self):
 
     def test_get_change_content(self):
         with self.app.test_client():
-            flight_path, operation = self._create_operation(flight_path="V12", content='initial')
-            assert self.fm.save_file(operation.id, "content1", self.user)
-            assert self.fm.save_file(operation.id, "content2", self.user)
-            assert self.fm.save_file(operation.id, "content3", self.user)
+            flight_path, operation = self._create_operation(flight_path="V12", content=XML_CONTENT3)
+            assert self.fm.save_file(operation.id, XML_CONTENT1, self.user)
+            assert self.fm.save_file(operation.id, XML_CONTENT2, self.user)
+            assert self.fm.save_file(operation.id, XML_CONTENT3, self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
             previous_change = self.fm.get_change_content(all_changes[2]["id"], self.user)
-            assert previous_change == "content1"
+            assert previous_change == XML_CONTENT1
             previous_change = self.fm.get_change_content(all_changes[1]["id"], self.user)
-            assert previous_change == "content2"
+            assert previous_change == XML_CONTENT2
 
     def test_set_version_name(self):
         with self.app.test_client():
-            flight_path, operation = self._create_operation(flight_path="V13", content='initial')
-            assert self.fm.save_file(operation.id, "content1", self.user)
+            flight_path, operation = self._create_operation(flight_path="V13", content=XML_CONTENT3)
+            assert self.fm.save_file(operation.id, XML_CONTENT1, self.user)
             all_changes = self.fm.get_all_changes(operation.id, self.user)
             ch_id = all_changes[-1]["id"]
             self.fm.set_version_name(ch_id, operation.id, self.user.id, "berlin")