Add shared atomic writes and tolerate torn network.json

Consolidate the atomic writer into src/atomic_io.py and route the
remaining runtime/generated writes through it so an interrupted write
cannot leave a truncated file. Preserve destination ownership and
permissions across the rename, and treat an unreadable or non-object
network.json as a validation failure instead of crashing the scheduler.

src/LibreQoS.py

@@ -17,10 +17,10 @@ import warnings
 import psutil
 import argparse
 import logging
-import shutil
 import time
 from deepdiff import DeepDiff
 
+from atomic_io import atomic_copy, atomic_write_text
 from virtual_tree_nodes import (
     build_logical_to_physical_node_map,
     build_physical_network,
@@ -221,26 +221,6 @@ def get_runtime_state_path(category, filename):
     return get_state_path(category, filename)
 
 
-def ensure_parent_dir(path):
-    parent = os.path.dirname(path)
-    if parent:
-        os.makedirs(parent, exist_ok=True)
-
-
-def atomic_write_text(path, text):
-    # Write to a sibling temp file and atomically replace the target, so an
-    # interrupted write (crash, VM snapshot) can never leave a truncated file.
-    # The file is fsync'd, but the directory rename is not; on power loss the
-    # worst case is the previous complete file.
-    ensure_parent_dir(path)
-    temp_path = path + ".tmp"
-    with open(temp_path, 'w') as file:
-        file.write(text)
-        file.flush()
-        os.fsync(file.fileno())
-    os.replace(temp_path, path)
-
-
 def get_network_json_path():
     base_dir = get_libreqos_directory()
     effective_path = get_runtime_state_path("topology", "network.effective.json")
@@ -1036,10 +1016,13 @@ def validateNetworkAndDevices():
             warnings.warn(f"network.json is missing: {network_json_path}", stacklevel=2)
         networkValidatedOrNot = False
     else:
-        with open(network_json_path) as file:
-            try:
+        try:
+            with open(network_json_path) as file:
                 data = json.load(file) # put JSON-data to a variable
-                if data != {}:
+                if not isinstance(data, dict):
+                    warnings.warn(f"network.json at '{network_json_path}' is not a node object", stacklevel=2)
+                    networkValidatedOrNot = False
+                elif data != {}:
                     #Traverse
                     observedNodes = set()
                     duplicateNodes = set()
@@ -1061,9 +1044,9 @@ def validateNetworkAndDevices():
                     if len(observedNodes) < 1:
                         warnings.warn("network.json had 0 valid nodes. Only {} is accepted for that scenario.", stacklevel=2)
                         networkValidatedOrNot = False
-            except json.decoder.JSONDecodeError:
-                warnings.warn("network.json is an invalid JSON file", stacklevel=2) # in case json is invalid
-                networkValidatedOrNot = False
+        except (json.decoder.JSONDecodeError, UnicodeDecodeError, OSError) as err:
+            warnings.warn(f"network.json at '{network_json_path}' is unreadable or invalid: {err}", stacklevel=2)
+            networkValidatedOrNot = False
     if integration_ingress:
         if devicesValidatedOrNot == True:
             print("integration shaping ingress passed validation")
@@ -1491,11 +1474,9 @@ def refreshShapers():
     if (validateNetworkAndDevices() == True):
         if os.path.isfile(shapedDevicesFile):
             last_good_csv_path = get_state_path("shaping", "lastGoodConfig.csv")
-            ensure_parent_dir(last_good_csv_path)
-            shutil.copyfile(shapedDevicesFile, last_good_csv_path)
+            atomic_copy(shapedDevicesFile, last_good_csv_path)
         last_good_json_path = get_state_path("shaping", "lastGoodConfig.json")
-        ensure_parent_dir(last_good_json_path)
-        shutil.copyfile(networkJSONfile, last_good_json_path)
+        atomic_copy(networkJSONfile, last_good_json_path)
         if os.path.isfile(shapedDevicesFile):
             print("Backed up good config as lastGoodConfig.csv and lastGoodConfig.json")
         else:
@@ -2936,11 +2917,11 @@ def refreshShapers():
         if observe_mode:
             linuxTCcommands = []
         linux_tc_path = get_linux_tc_path()
-        ensure_parent_dir(linux_tc_path)
-        with open(linux_tc_path, 'w') as f:
-            for command in linuxTCcommands:
-                logging.info(command)
-                f.write(f"{command}\n")
+        linux_tc_commands = []
+        for command in linuxTCcommands:
+            logging.info(command)
+            linux_tc_commands.append(f"{command}\n")
+        atomic_write_text(linux_tc_path, "".join(linux_tc_commands))
         # if logging.DEBUG <= logging.root.level:
         # 	# Do not --force in debug mode, so we can see any errors
         # 	shell("/sbin/tc -b linux_tc.txt")
@@ -3014,8 +2995,7 @@ def refreshShapers():
         # snapshot there instead of failing the whole scheduler refresh.
         if os.path.isfile(shapedDevicesFile):
             last_loaded_path = get_state_path("shaping", "ShapedDevices.lastLoaded.csv")
-            ensure_parent_dir(last_loaded_path)
-            shutil.copyfile(shapedDevicesFile, last_loaded_path)
+            atomic_copy(shapedDevicesFile, last_loaded_path)
 
         # Save for stats
         stats_by_circuit_path = get_stats_by_circuit_path()

src/atomic_io.py

@@ -0,0 +1,85 @@
+"""Atomic file writes for runtime state and generated shaping inputs.
+
+Each write goes to a sibling temporary file that is flushed and fsync'd before
+an os.replace() swaps it over the destination. A reader therefore never sees a
+truncated or torn file when a write is interrupted by a crash, VM snapshot, or
+power loss; the worst case is the previous complete file.
+
+A single writer is assumed per destination path; concurrent writers to the same
+path are not coordinated.
+"""
+
+import os
+import shutil
+import stat
+
+
+def _ensure_parent_dir(path):
+    parent = os.path.dirname(path)
+    if parent:
+        os.makedirs(parent, exist_ok=True)
+
+
+def _remove_temp_file(temp_path):
+    try:
+        os.remove(temp_path)
+    except OSError:
+        pass
+
+
+def _preserve_destination_metadata(path, temp_path):
+    # Replacing a file creates a new inode, so carry over the destination's
+    # ownership and permissions. Operators edit generated files over SFTP, and
+    # the scheduler runs as root.
+    try:
+        existing = os.stat(path)
+    except OSError:
+        return
+    try:
+        os.chmod(temp_path, stat.S_IMODE(existing.st_mode))
+    except OSError:
+        pass
+    try:
+        os.chown(temp_path, existing.st_uid, existing.st_gid)
+    except OSError:
+        pass
+
+
+def _atomic_replace(path, write_temp):
+    _ensure_parent_dir(path)
+    temp_path = path + ".tmp"
+    replaced = False
+    try:
+        with open(temp_path, "wb") as handle:
+            write_temp(handle)
+            handle.flush()
+            os.fsync(handle.fileno())
+        _preserve_destination_metadata(path, temp_path)
+        os.replace(temp_path, path)
+        replaced = True
+    finally:
+        if not replaced:
+            _remove_temp_file(temp_path)
+
+
+def atomic_write_text(path, text, encoding="utf-8"):
+    """Atomically replace `path` with `text`.
+
+    Side effects: creates parent directories, writes and fsyncs `path + ".tmp"`,
+    preserves the destination's ownership and permissions when it exists, then
+    renames the temporary file over `path`.
+    """
+    _atomic_replace(path, lambda handle: handle.write(text.encode(encoding)))
+
+
+def atomic_copy(src, dest):
+    """Atomically copy `src` to `dest`, preserving the exact bytes.
+
+    Side effects match `atomic_write_text`.
+    """
+
+    def write_temp(handle):
+        with open(src, "rb") as source:
+            shutil.copyfileobj(source, handle)
+
+    _atomic_replace(dest, write_temp)

src/build_dpkg.sh

@@ -20,6 +20,7 @@ ETC_DIR=$DPKG_DIR/etc
 ETC_LIBREQOS_DIR=$DPKG_DIR/etc/libreqos
 MOTD_DIR=$DPKG_DIR/etc/update-motd.d
 LQOS_FILES=(
+  atomic_io.py
   csvToNetworkJSON.py
   configMigrator.py
   integrationCommon.py

src/integrationCommon.py

@@ -13,6 +13,8 @@ import json
 import os
 import time
 
+from atomic_io import atomic_write_text
+
 try:
 	from liblqos_python import get_libreqos_state_directory as _get_state_dir_native
 except Exception:
@@ -548,10 +550,8 @@ class NetworkGraph:
 		return self.nodes[index].type == NodeType.ap or self.nodes[index].type == NodeType.site or self.nodes[index].type == NodeType.clientWithChildren
 
 	def createNetworkJson(self):
-		import json
 		topLevelNode = self.buildNetworkJson()
-		with open('network.json', 'w') as f:
-			json.dump(topLevelNode, f, indent=4)
+		atomic_write_text('network.json', json.dumps(topLevelNode, indent=4))
 
 	def buildNetworkJson(self):
 		topLevelNode = {}
@@ -671,14 +671,10 @@ class NetworkGraph:
 
 	def createShapedDevices(self):
 		shaped_devices_csv, circuit_anchor_file = self.buildShapedDevicesArtifacts()
-		with open('ShapedDevices.csv', 'w', newline='') as csvfile:
-			csvfile.write(shaped_devices_csv)
+		atomic_write_text('ShapedDevices.csv', shaped_devices_csv)
 
 		anchor_path = _state_path("topology", "circuit_anchors.json")
-		os.makedirs(os.path.dirname(anchor_path), exist_ok=True)
-		with open(anchor_path, 'w', encoding='utf-8') as anchorfile:
-			json.dump(circuit_anchor_file, anchorfile, indent=2)
-			anchorfile.write('\n')
+		atomic_write_text(anchor_path, json.dumps(circuit_anchor_file, indent=2) + '\n')
 
 	def buildShapedDevicesArtifacts(self):
 		import csv

src/scheduler.py

@@ -5,6 +5,7 @@ import io
 import json
 import chardet
 from LibreQoS import RefreshFailure, ValidationFailure, refreshShapers, refreshShapersUpdateOnly
+from atomic_io import atomic_write_text
 import subprocess
 import sys
 import tempfile
@@ -716,10 +717,11 @@ def merge_rows_replace_by_device_id(existing_rows, override_rows):
 
 
 def write_shaped_devices_csv(path: str, header, rows):
-    with open(path, 'w', encoding='utf-8', newline='') as f:
-        writer = csv.writer(f)
-        writer.writerow(header)
-        writer.writerows(rows)
+    buffer = io.StringIO()
+    writer = csv.writer(buffer)
+    writer.writerow(header)
+    writer.writerows(rows)
+    atomic_write_text(path, buffer.getvalue())
 
 
 def header_has_sqm(header):
@@ -896,9 +898,7 @@ def load_topology_canonical_state(path: str):
 
 
 def write_topology_canonical_state(path: str, canonical_state: dict):
-    os.makedirs(os.path.dirname(path), exist_ok=True)
-    with open(path, 'w', encoding='utf-8') as f:
-        f.write(json.dumps(canonical_state, indent=4))
+    atomic_write_text(path, json.dumps(canonical_state, indent=4))
 
 
 def _format_attempt_count(attempts: int) -> str:
@@ -1101,8 +1101,7 @@ def apply_network_adjustments_to_canonical_state(canonical_state: dict, adjustme
 
 
 def write_network_json(path: str, network: dict):
-    with open(path, 'w', encoding='utf-8') as f:
-        f.write(json.dumps(network, indent=4))
+    atomic_write_text(path, json.dumps(network, indent=4))
 
 
 def importAndShapeFullReload():

src/test_atomic_io.py

@@ -0,0 +1,91 @@
+import os
+import stat
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import atomic_io
+
+
+class TestAtomicWrite(unittest.TestCase):
+    def setUp(self):
+        temp_dir = tempfile.TemporaryDirectory()  # nosec B108
+        self.addCleanup(temp_dir.cleanup)
+        self.temp_dir = temp_dir.name
+
+    def test_write_text_creates_parents_and_content(self):
+        path = os.path.join(self.temp_dir, "a", "b", "state.txt")
+        atomic_io.atomic_write_text(path, "hello")
+        with open(path, encoding="utf-8") as handle:
+            self.assertEqual(handle.read(), "hello")
+        self.assertFalse(os.path.exists(path + ".tmp"))
+
+    def test_write_text_replaces_existing_content(self):
+        path = os.path.join(self.temp_dir, "state.txt")
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write("old")
+        atomic_io.atomic_write_text(path, "new")
+        with open(path, encoding="utf-8") as handle:
+            self.assertEqual(handle.read(), "new")
+        self.assertFalse(os.path.exists(path + ".tmp"))
+
+    def test_write_text_preserves_existing_permissions(self):
+        path = os.path.join(self.temp_dir, "state.txt")
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write("old")
+        os.chmod(path, 0o600)
+        atomic_io.atomic_write_text(path, "new")
+        self.assertEqual(stat.S_IMODE(os.stat(path).st_mode), 0o600)
+
+    def test_write_text_failure_preserves_existing_content(self):
+        path = os.path.join(self.temp_dir, "state.txt")
+        with open(path, "w", encoding="utf-8") as handle:
+            handle.write("old")
+        with patch("os.replace", side_effect=OSError(5, "replace failed")):
+            with self.assertRaises(OSError):
+                atomic_io.atomic_write_text(path, "new")
+        with open(path, encoding="utf-8") as handle:
+            self.assertEqual(handle.read(), "old")
+        self.assertFalse(os.path.exists(path + ".tmp"))
+
+
+class TestAtomicCopy(unittest.TestCase):
+    def setUp(self):
+        temp_dir = tempfile.TemporaryDirectory()  # nosec B108
+        self.addCleanup(temp_dir.cleanup)
+        self.temp_dir = temp_dir.name
+
+    def test_copy_creates_parents_and_preserves_bytes(self):
+        src = os.path.join(self.temp_dir, "source.bin")
+        with open(src, "wb") as handle:
+            handle.write(b"\x00\xffpayload")
+        dest = os.path.join(self.temp_dir, "nested", "copy.bin")
+        atomic_io.atomic_copy(src, dest)
+        with open(dest, "rb") as handle:
+            self.assertEqual(handle.read(), b"\x00\xffpayload")
+        self.assertFalse(os.path.exists(dest + ".tmp"))
+
+    def test_copy_failure_preserves_existing_destination(self):
+        src = os.path.join(self.temp_dir, "source.bin")
+        with open(src, "wb") as handle:
+            handle.write(b"new")
+        dest = os.path.join(self.temp_dir, "dest.bin")
+        with open(dest, "wb") as handle:
+            handle.write(b"old")
+        with patch("os.replace", side_effect=OSError(5, "replace failed")):
+            with self.assertRaises(OSError):
+                atomic_io.atomic_copy(src, dest)
+        with open(dest, "rb") as handle:
+            self.assertEqual(handle.read(), b"old")
+        self.assertFalse(os.path.exists(dest + ".tmp"))
+
+    def test_copy_missing_source_preserves_existing_destination(self):
+        dest = os.path.join(self.temp_dir, "dest.bin")
+        with open(dest, "wb") as handle:
+            handle.write(b"old")
+        missing = os.path.join(self.temp_dir, "does-not-exist.bin")
+        with self.assertRaises(FileNotFoundError):
+            atomic_io.atomic_copy(missing, dest)
+        with open(dest, "rb") as handle:
+            self.assertEqual(handle.read(), b"old")
+        self.assertFalse(os.path.exists(dest + ".tmp"))

src/test_libreqos_shaping_inputs.py

@@ -569,37 +569,53 @@ class TestFirstRunSinceBoot(unittest.TestCase):
         self.assertTrue(LibreQoS.checkIfFirstRunSinceBoot())
 
 
-class TestAtomicWriteText(unittest.TestCase):
+class TestValidateNetworkJsonTornWrite(unittest.TestCase):
     def setUp(self):
         temp_dir = tempfile.TemporaryDirectory()  # nosec B108
         self.addCleanup(temp_dir.cleanup)
         self.temp_dir = temp_dir.name
+        self.network_path = os.path.join(self.temp_dir, "network.json")
+        self.devices_path = os.path.join(self.temp_dir, "ShapedDevices.csv")
+        with open(self.devices_path, "w", encoding="utf-8", newline="") as handle:
+            handle.write(
+                "Circuit ID,Circuit Name,Device ID,Device Name,Parent Node,MAC,"
+                "IPv4,IPv6,Download Min,Upload Min,Download Max,Upload Max,Comment\n"
+                "circuit-1,Circuit 1,dev-1,Device 1,Parent,00:00:00:00:00:00,"
+                "192.168.1.2,,10,10,100,100,\n"
+            )
 
-    def test_atomic_write_creates_parents_and_content(self):
-        path = os.path.join(self.temp_dir, "a", "b", "state.txt")
-        LibreQoS.atomic_write_text(path, "hello")
-        with open(path, encoding="utf-8") as handle:
-            self.assertEqual(handle.read(), "hello")
-        self.assertFalse(os.path.exists(path + ".tmp"))
+    def _validate(self):
+        with patch.object(
+            LibreQoS, "get_network_json_path", return_value=self.network_path
+        ), patch.object(
+            LibreQoS, "get_shaped_devices_path", return_value=self.devices_path
+        ):
+            return LibreQoS.validateNetworkAndDevices()
 
-    def test_atomic_write_replaces_existing_content(self):
-        path = os.path.join(self.temp_dir, "state.txt")
-        with open(path, "w", encoding="utf-8") as handle:
-            handle.write("old")
-        LibreQoS.atomic_write_text(path, "new")
-        with open(path, encoding="utf-8") as handle:
-            self.assertEqual(handle.read(), "new")
-        self.assertFalse(os.path.exists(path + ".tmp"))
+    def test_undecodable_network_json_fails_validation_without_raising(self):
+        with open(self.network_path, "wb") as handle:
+            handle.write(b"\x80\x81\x82 torn network")
+        with self.assertWarnsRegex(UserWarning, "unreadable or invalid"):
+            self.assertFalse(self._validate())
 
-    def test_atomic_write_failure_preserves_existing_content(self):
-        path = os.path.join(self.temp_dir, "state.txt")
-        with open(path, "w", encoding="utf-8") as handle:
-            handle.write("old")
-        with patch("os.replace", side_effect=OSError(5, "replace failed")):
-            with self.assertRaises(OSError):
-                LibreQoS.atomic_write_text(path, "new")
-        with open(path, encoding="utf-8") as handle:
-            self.assertEqual(handle.read(), "old")
+    def test_truncated_network_json_fails_validation_without_raising(self):
+        with open(self.network_path, "wb") as handle:
+            handle.write(b'{"Root": ')
+        with self.assertWarnsRegex(UserWarning, "unreadable or invalid"):
+            self.assertFalse(self._validate())
+
+    def test_network_json_read_error_fails_validation_without_raising(self):
+        with open(self.network_path, "w", encoding="utf-8") as handle:
+            handle.write("{}")
+        with patch.object(LibreQoS.json, "load", side_effect=PermissionError(13, "denied")):
+            with self.assertWarnsRegex(UserWarning, "unreadable or invalid"):
+                self.assertFalse(self._validate())
+
+    def test_non_object_network_json_fails_validation_without_raising(self):
+        with open(self.network_path, "w", encoding="utf-8") as handle:
+            handle.write("null")
+        with self.assertWarnsRegex(UserWarning, "not a node object"):
+            self.assertFalse(self._validate())
 
 
 if __name__ == "__main__":