sdk: add thread-safety to local_file backend

Until now the `LocalFileIdentifiableStore` had a TOCTOU race condition
in the `add()` and `commit()` methods. After a file was checked
for existance a concurrent running `add()` or `discard()` could still
create or remove the file before the original invocation accesses it.

To fix this, all write accesses to a file need to be mutual exclusive.
In this implementation I chose to use a global `_writing_lock` to
lock write operations to all files of the store, as locking individual
files has no performance benefit, due to serialization of multiple threads
through Pythons GIL.

This implementation prevents the following concurrency issues:
 1. Two concurrent `add()`: Instead of silent overwriting, first call will
    succeed, second failes with `KeyError`.
 2. `commit()` + `discard()`: Instead of `commit()` creating file after
    `discard()` deleted it, the file is now certainly discarded.

Author: hpoeche

sdk/basyx/aas/backend/local_file.py

@@ -175,6 +175,9 @@ def __init__(self, directory_path: str):
             = weakref.WeakValueDictionary()
         self._object_cache_lock = threading.Lock()
 
+        # Prevent concurrent write operations to avoid TOCTOU problems
+        self._write_lock = threading.Lock()
+
         # We need to prevent multiple instances of LocalFileIdentifiableStore performing R/W operations on the same
         # directory in order to ensure cache validity. The directory is locked as soon as it exists.
         self._dir_lock = DirectoryLock(self.directory_path)
@@ -266,10 +269,11 @@ def add(self, x: model.Identifiable) -> None:
         :raises KeyError: If an object with the same id exists already in the object store
         """
         logger.debug("Adding object %s to Local File Store ...", repr(x))
-        with self._dir_lock.ensure_locked():
-            if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
-                raise KeyError("Identifiable with id {} already exists in local file database".format(x.id))
-            self._write_atomic(x)
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                if os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
+                    raise KeyError("Identifiable with id {} already exists in local file database".format(x.id))
+                self._write_atomic(x)
         with self._object_cache_lock:
             self._object_cache[x.id] = x
 
@@ -280,10 +284,11 @@ def commit(self, x: model.Identifiable) -> None:
         :param x: The object to persist
         :raises KeyError: If the object is not present in the store
         """
-        with self._dir_lock.ensure_locked():
-            if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
-                raise KeyError("No AAS object with id {} exists in local file database".format(x.id))
-            self._write_atomic(x)
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                if not os.path.exists("{}/{}.json".format(self.directory_path, self._transform_id(x.id))):
+                    raise KeyError("No AAS object with id {} exists in local file database".format(x.id))
+                self._write_atomic(x)
 
     def discard(self, x: model.Identifiable) -> None:
         """
@@ -293,11 +298,12 @@ def discard(self, x: model.Identifiable) -> None:
         :raises KeyError: If the object does not exist in the database
         """
         logger.debug("Deleting object %s from Local File Store database ...", repr(x))
-        with self._dir_lock.ensure_locked():
-            try:
-                os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id)))
-            except FileNotFoundError as e:
-                raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e
+        with self._write_lock:
+            with self._dir_lock.ensure_locked():
+                try:
+                    os.remove("{}/{}.json".format(self.directory_path, self._transform_id(x.id)))
+                except FileNotFoundError as e:
+                    raise KeyError("No AAS object with id {} exists in local file database".format(x.id)) from e
         with self._object_cache_lock:
             self._object_cache.pop(x.id, None)
 

sdk/test/backend/test_local_file.py

@@ -10,7 +10,7 @@
 import tempfile
 import threading
 import concurrent.futures
-from typing import Callable
+from typing import Callable, cast
 
 from unittest import TestCase
 
@@ -307,3 +307,119 @@ def test_reload_discard(self) -> None:
         self.identifiable_store = local_file.LocalFileIdentifiableStore(store_path)
         self.identifiable_store.discard(example_submodel)
         self.assertNotIn(example_submodel, self.identifiable_store)
+
+
+class _GatedLocalFileStore(local_file.LocalFileIdentifiableStore):
+    """
+    Patch :meth:`_write_atomic()` to control and enforce concurrent writes.
+    """
+
+    def __init__(self, directory_path: str, *args, **kwargs):
+        super().__init__(directory_path)
+        self._gate = threading.Event()  # gate to run _write_atomic()
+        self._gate.set()
+        self._inside = threading.Event()  # signal _write_atomic() entry
+
+    def _write_atomic(self, x: model.Identifiable) -> None:
+        self._inside.set()
+        self._gate.wait(timeout=5)
+        super()._write_atomic(x)
+
+
+class LocalFileBackendConcurrencyTest(TestCase):
+    def setUp(self):
+        self.store = _GatedLocalFileStore(store_path)
+        self.store.check_directory(create=True)
+
+    def tearDown(self):
+        try:
+            self.store.clear()
+        finally:
+            self.store.close()
+            shutil.rmtree(store_path)
+
+    def _example_submodels(self) -> tuple[model.Submodel, model.Submodel]:
+        first_submodel = model.Submodel(
+            id_='https://example.org/BackendTest',
+            submodel_element={
+                model.Property(id_short='Prop', value_type=model.datatypes.String, value='first')
+            }
+        )
+
+        second_submodel = model.Submodel(
+            id_='https://example.org/BackendTest',
+            submodel_element={
+                model.Property(id_short='Prop', value_type=model.datatypes.String, value='second')
+            }
+        )
+
+        return first_submodel, second_submodel
+
+    def test_concurrent_add(self):
+        """Checks that second add for same ID fails"""
+        first_submodel, second_submodel = self._example_submodels()
+
+        self.store._gate.clear()
+        self.store._inside.clear()
+        barrier = threading.Barrier(2, timeout=5)
+        all_done = threading.Barrier(3, timeout=5)
+
+        def first():
+            self.store.add(first_submodel)
+            all_done.wait()
+
+        def second():
+            self.store._inside.wait()
+            barrier.wait()
+            with self.assertRaises(KeyError) as ex:
+                self.store.add(second_submodel)
+
+            all_done.wait()
+            self.assertIn("already exists", ex.exception.args[0])
+
+        def control():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store._gate.set()
+
+            all_done.wait()
+            submodel = self.store.get_item("https://example.org/BackendTest")
+            self.assertIsInstance(submodel, model.Submodel)
+            sm_property = submodel.get_referable("Prop")
+            self.assertIsInstance(sm_property, model.Property)
+            self.assertEqual(sm_property.value, "first")
+
+        run_threads([first, second, control])
+
+    def test_concurrent_commit_discard(self):
+        """Checks that discard is not overwritten by concurrent commit"""
+        submodel, altered_submodel = self._example_submodels()
+        self.store.add(submodel)
+
+        self.store._gate.clear()
+        self.store._inside.clear()
+        barrier = threading.Barrier(2, timeout=5)
+        all_done = threading.Barrier(3, timeout=5)
+
+        def first():
+            self.store.commit(submodel)
+            all_done.wait()
+
+        def second():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store.discard(submodel)
+            all_done.wait()
+
+        def control():
+            self.store._inside.wait()
+            barrier.wait()
+            self.store._gate.set()
+
+            all_done.wait()
+            with self.assertRaises(KeyError) as ex:
+                self.store.get_item("https://example.org/BackendTest")
+
+            self.assertIn("No Identifiable", ex.exception.args[0])
+
+        run_threads([first, second, control])