From b14a6870e0d1dd970c710d51fdd117afa2e07d67 Mon Sep 17 00:00:00 2001 From: Herbert Wolverson Date: Tue, 15 Sep 2026 14:47:22 -0500 Subject: [PATCH] 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 | 56 ++++++------------ src/atomic_io.py | 85 +++++++++++++++++++++++++++ src/build_dpkg.sh | 1 + src/integrationCommon.py | 14 ++--- src/scheduler.py | 17 +++--- src/test_atomic_io.py | 91 +++++++++++++++++++++++++++++ src/test_libreqos_shaping_inputs.py | 64 ++++++++++++-------- 7 files changed, 248 insertions(+), 80 deletions(-) create mode 100644 src/atomic_io.py create mode 100644 src/test_atomic_io.py diff --git a/src/LibreQoS.py b/src/LibreQoS.py index a0b9e08c..cd004d71 100755 --- a/src/LibreQoS.py +++ b/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() diff --git a/src/atomic_io.py b/src/atomic_io.py new file mode 100644 index 00000000..2d0497eb --- /dev/null +++ b/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) diff --git a/src/build_dpkg.sh b/src/build_dpkg.sh index 024bab10..440154b8 100755 --- a/src/build_dpkg.sh +++ b/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 diff --git a/src/integrationCommon.py b/src/integrationCommon.py index 0570c074..db5159e9 100644 --- a/src/integrationCommon.py +++ b/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 diff --git a/src/scheduler.py b/src/scheduler.py index f0ccbd33..d5bb722b 100644 --- a/src/scheduler.py +++ b/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(): diff --git a/src/test_atomic_io.py b/src/test_atomic_io.py new file mode 100644 index 00000000..29197f85 --- /dev/null +++ b/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")) diff --git a/src/test_libreqos_shaping_inputs.py b/src/test_libreqos_shaping_inputs.py index 83f681b6..41a3c58e 100644 --- a/src/test_libreqos_shaping_inputs.py +++ b/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__":