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
 
@@ -57,43 +58,50 @@ def create_operation(self, path, description, user, last_used=None, content=None
         description: description of the operation
         """
         # set codes on these later
-        if path.find("/") != -1 or path.find("\\") != -1 or (" " in path):
-            logging.debug("malicious request: %s", user)
-            return False
-        proj_available = Operation.query.filter_by(path=path).first()
-        if proj_available is not None:
+        if content is None or verify_waypoint_data(content):
+            if path.find("/") != -1 or path.find("\\") != -1 or (" " in path):
+                logging.debug("malicious request: %s", user)
+                return False
+            proj_available = Operation.query.filter_by(path=path).first()
+            if proj_available is not None:
+                return False
+            if last_used is None:
+                last_used = datetime.datetime.now(tz=datetime.timezone.utc)
+            operation = Operation(path, description, last_used, category, active=active)
+            db.session.add(operation)
+            db.session.flush()
+            operation_id = operation.id
+
+            op_lock = self._get_operation_lock(operation_id)
+            with op_lock:
+                # this is the only insertion with "creator" access_level
+                perm = Permission(user.id, operation_id, "creator")
+                db.session.add(perm)
+                db.session.commit()
+                # here we can import the permissions from Group file
+                if not path.endswith(mscolab_settings.GROUP_POSTFIX):
+                    import_op = Operation.query.filter_by(path=f"{category}{mscolab_settings.GROUP_POSTFIX}").first()
+                    if import_op is not None:
+                        self.import_permissions(import_op.id, operation_id, user.id)
+                data = fs.open_fs(self.data_dir)
+                data.makedir(operation.path)
+                with data.open(fs.path.combine(operation.path, 'main.ftml'), 'w') as operation_file:
+                    if content is not None:
+                        if verify_waypoint_data(content):
+                            operation_file.write(content)
+                        else:
+                            logging.debug("Invalid content: %s", path)
+                    else:
+                        operation_file.write(mscolab_settings.STUB_CODE)
+
+                    operation_path = fs.path.combine(self.data_dir, operation.path)
+                    r = git.Repo.init(operation_path)
+                    r.git.clear_cache()
+                    r.index.add(['main.ftml'])
+                    r.index.commit("initial commit")
+                    return True
+        else:
             return False
-        if last_used is None:
-            last_used = datetime.datetime.now(tz=datetime.timezone.utc)
-        operation = Operation(path, description, last_used, category, active=active)
-        db.session.add(operation)
-        db.session.flush()
-        operation_id = operation.id
-
-        op_lock = self._get_operation_lock(operation_id)
-        with op_lock:
-            # this is the only insertion with "creator" access_level
-            perm = Permission(user.id, operation_id, "creator")
-            db.session.add(perm)
-            db.session.commit()
-            # here we can import the permissions from Group file
-            if not path.endswith(mscolab_settings.GROUP_POSTFIX):
-                import_op = Operation.query.filter_by(path=f"{category}{mscolab_settings.GROUP_POSTFIX}").first()
-                if import_op is not None:
-                    self.import_permissions(import_op.id, operation_id, user.id)
-            data = fs.open_fs(self.data_dir)
-            data.makedir(operation.path)
-            operation_file = data.open(fs.path.combine(operation.path, 'main.ftml'), 'w')
-            if content is not None:
-                operation_file.write(content)
-            else:
-                operation_file.write(mscolab_settings.STUB_CODE)
-            operation_path = fs.path.combine(self.data_dir, operation.path)
-            r = git.Repo.init(operation_path)
-            r.git.clear_cache()
-            r.index.add(['main.ftml'])
-            r.index.commit("initial commit")
-            return True
 
     def get_operation_details(self, op_id, user):
         """
@@ -263,14 +271,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 +326,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
@@ -644,13 +645,19 @@ def load_from_ftml(self, filename):
         _dirname, _name = os.path.split(filename)
         _fs = fs.open_fs(_dirname)
         xml_content = _fs.readtext(_name)
-        name = os.path.basename(filename.replace(".ftml", "").strip())
-        self.load_from_xml_data(xml_content, name)
+        if verify_waypoint_data(xml_content):
+            name = os.path.basename(filename.replace(".ftml", "").strip())
+            self.load_from_xml_data(xml_content, name)
+        else:
+            raise SyntaxError(f"Invalid flight track filename: {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
@@ -1977,6 +1978,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")