From deaabb46c09aa92741c15023ae5912d128d0d749 Mon Sep 17 00:00:00 2001 From: allebonvi Date: Thu, 2 Jul 2026 10:20:13 +0200 Subject: [PATCH] Filtra eventi watchdog con USN journal --- README.md | 2 + SPECIFICHE.md | 8 +- config/config.example.toml | 1 - src/bakrest/config.py | 2 - src/bakrest/config_app.py | 21 --- src/bakrest/default_config.py | 1 - src/bakrest/usn_journal.py | 315 ++++++++++++++++++++++++++++++++++ src/bakrest/watcher.py | 27 +-- tests/test_usn_journal.py | 52 ++++++ tests/test_watcher.py | 17 -- 10 files changed, 381 insertions(+), 65 deletions(-) create mode 100644 src/bakrest/usn_journal.py create mode 100644 tests/test_usn_journal.py delete mode 100644 tests/test_watcher.py diff --git a/README.md b/README.md index 00ac895..e637947 100644 --- a/README.md +++ b/README.md @@ -51,6 +51,8 @@ python -m pip install -e . python -m bakrest.watchdog_runner ``` +Il watchdog usa il journal USN NTFS per distinguere modifiche reali da eventi rumorosi generati da navigazione, anteprime o metadati. La lettura del journal puo' richiedere privilegi elevati: il servizio eseguito come `LocalSystem` e' il percorso consigliato. Se USN non e' disponibile o non e' leggibile, il servizio registra un warning e usa un fallback conservativo. + ## Avvio tray app per test ```powershell diff --git a/SPECIFICHE.md b/SPECIFICHE.md index f8e88ca..5e345e9 100644 --- a/SPECIFICHE.md +++ b/SPECIFICHE.md @@ -320,9 +320,11 @@ unchanged_checks_required: 2 Quando si monitora una directory molto ampia, ad esempio un'intera unita' `D:\`, la semplice navigazione con Explorer o altri programmi puo' generare eventi `modified` non rilevanti. Per ridurre questi falsi positivi senza costruire una baseline completa del disco, il servizio deve applicare una strategia pigra: - non deve scansionare all'avvio tutti i file gia' presenti; -- un evento `modified` viene registrato solo se il timestamp di modifica del file e' recente rispetto all'evento; -- se esiste gia' uno snapshot locale per quel file, l'evento viene registrato solo se dimensione o timestamp risultano diversi dallo snapshot conosciuto; -- i file gia' presenti nel disco entrano quindi in coda solo quando vengono realmente salvati/modificati, non quando vengono solo navigati. +- watchdog intercetta gli eventi file ma non li considera automaticamente validi; +- per eventi `created`, `modified` e `moved`, il servizio consulta il journal USN NTFS; +- entrano in coda solo eventi con reason USN di contenuto reale o creazione/rinomina utile, ad esempio `DATA_OVERWRITE`, `DATA_EXTEND`, `DATA_TRUNCATION`, `FILE_CREATE`, `RENAME_NEW_NAME`; +- eventi di soli metadati, chiusura handle, indicizzazione, security o basic info devono essere ignorati; +- se USN non e' disponibile su un volume, il servizio deve fare fallback conservativo e registrare un warning nei log. ## Configurazione diff --git a/config/config.example.toml b/config/config.example.toml index 2aca588..ffbfb27 100644 --- a/config/config.example.toml +++ b/config/config.example.toml @@ -6,7 +6,6 @@ autostart = true [watch] target_user = "" -modified_event_max_age_seconds = 300 include_dirs = [ "%USERPROFILE%\\Desktop", "%USERPROFILE%\\Documents", diff --git a/src/bakrest/config.py b/src/bakrest/config.py index 8798059..3757e28 100644 --- a/src/bakrest/config.py +++ b/src/bakrest/config.py @@ -40,7 +40,6 @@ class WatchConfig: include_extensions: frozenset[str] exclude_dirs: tuple[Path | str, ...] exclude_patterns: tuple[str, ...] - modified_event_max_age_seconds: int @dataclass(frozen=True) @@ -188,7 +187,6 @@ def parse_config(data: dict[str, Any], path: Path) -> AppConfig: include_extensions=include_extensions, exclude_dirs=exclude_dirs, exclude_patterns=tuple(watch.get("exclude_patterns", [])), - modified_event_max_age_seconds=int(watch.get("modified_event_max_age_seconds", 300)), ), backup=BackupConfig( engine=str(backup.get("engine", "robocopy")).lower(), diff --git a/src/bakrest/config_app.py b/src/bakrest/config_app.py index 479e997..7557ea9 100644 --- a/src/bakrest/config_app.py +++ b/src/bakrest/config_app.py @@ -61,24 +61,8 @@ class BakRestConfigApp(ctk.CTk): target_frame = ctk.CTkFrame(self.monitor_tab) target_frame.grid(row=0, column=0, columnspan=2, sticky="ew", padx=8, pady=(8, 0)) target_frame.grid_columnconfigure(1, weight=1) - target_frame.grid_columnconfigure(3, weight=1) self.target_user_var = tk.StringVar() - self.modified_event_max_age_var = tk.StringVar() _label_entry(target_frame, "Utente percorsi", self.target_user_var, 0) - ctk.CTkLabel(target_frame, text="Eta max modifica (sec)", anchor="w").grid( - row=0, - column=2, - sticky="w", - padx=12, - pady=6, - ) - ctk.CTkEntry(target_frame, textvariable=self.modified_event_max_age_var).grid( - row=0, - column=3, - sticky="ew", - padx=12, - pady=6, - ) left = ctk.CTkFrame(self.monitor_tab) left.grid(row=1, column=0, sticky="nsew", padx=(0, 8), pady=8) @@ -197,7 +181,6 @@ class BakRestConfigApp(ctk.CTk): server_check = backup.setdefault("server_check", {}) self.target_user_var.set(str(watch.get("target_user", ""))) - self.modified_event_max_age_var.set(str(watch.get("modified_event_max_age_seconds", 300))) self.include_dirs.set_items(watch.get("include_dirs", [])) self.extensions.set_items(watch.get("include_extensions", [])) self.exclude_dirs.set_items(watch.get("exclude_dirs", [])) @@ -217,10 +200,6 @@ class BakRestConfigApp(ctk.CTk): watch["include_dirs"] = self.include_dirs.items() watch["target_user"] = self.target_user_var.get().strip() - watch["modified_event_max_age_seconds"] = _safe_int( - self.modified_event_max_age_var.get(), - 300, - ) watch["include_extensions"] = self.extensions.items() watch["exclude_dirs"] = self.exclude_dirs.items() watch["exclude_patterns"] = self.exclude_patterns.items() diff --git a/src/bakrest/default_config.py b/src/bakrest/default_config.py index 89427c6..b22efe8 100644 --- a/src/bakrest/default_config.py +++ b/src/bakrest/default_config.py @@ -6,7 +6,6 @@ autostart = true [watch] target_user = "" -modified_event_max_age_seconds = 300 include_dirs = [ "%USERPROFILE%\\\\Desktop", "%USERPROFILE%\\\\Documents", diff --git a/src/bakrest/usn_journal.py b/src/bakrest/usn_journal.py new file mode 100644 index 0000000..f38dd5f --- /dev/null +++ b/src/bakrest/usn_journal.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import ctypes +import os +import struct +import time +from collections.abc import Iterable +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +USN_REASON_DATA_OVERWRITE = 0x00000001 +USN_REASON_DATA_EXTEND = 0x00000002 +USN_REASON_DATA_TRUNCATION = 0x00000004 +USN_REASON_NAMED_DATA_OVERWRITE = 0x00000010 +USN_REASON_NAMED_DATA_EXTEND = 0x00000020 +USN_REASON_NAMED_DATA_TRUNCATION = 0x00000040 +USN_REASON_FILE_CREATE = 0x00000100 +USN_REASON_RENAME_NEW_NAME = 0x00002000 +USN_REASON_BASIC_INFO_CHANGE = 0x00008000 +USN_REASON_CLOSE = 0x80000000 + +MEANINGFUL_REASON_MASK = ( + USN_REASON_DATA_OVERWRITE + | USN_REASON_DATA_EXTEND + | USN_REASON_DATA_TRUNCATION + | USN_REASON_NAMED_DATA_OVERWRITE + | USN_REASON_NAMED_DATA_EXTEND + | USN_REASON_NAMED_DATA_TRUNCATION + | USN_REASON_FILE_CREATE + | USN_REASON_RENAME_NEW_NAME +) + +FSCTL_QUERY_USN_JOURNAL = 0x000900F4 +FSCTL_READ_USN_JOURNAL = 0x000900BB + +GENERIC_READ = 0x80000000 +FILE_READ_ATTRIBUTES = 0x0080 +FILE_SHARE_READ = 0x00000001 +FILE_SHARE_WRITE = 0x00000002 +FILE_SHARE_DELETE = 0x00000004 +OPEN_EXISTING = 3 +FILE_FLAG_BACKUP_SEMANTICS = 0x02000000 +INVALID_HANDLE_VALUE = ctypes.c_void_p(-1).value + +kernel32 = ctypes.windll.kernel32 +kernel32.CreateFileW.restype = ctypes.c_void_p +kernel32.CreateFileW.argtypes = [ + ctypes.c_wchar_p, + ctypes.c_ulong, + ctypes.c_ulong, + ctypes.c_void_p, + ctypes.c_ulong, + ctypes.c_ulong, + ctypes.c_void_p, +] +kernel32.DeviceIoControl.restype = ctypes.c_bool +kernel32.CloseHandle.argtypes = [ctypes.c_void_p] + + +class UsnJournalUnavailable(RuntimeError): + pass + + +@dataclass(frozen=True) +class UsnRecord: + file_reference_number: int + reason: int + + +class UsnChangeFilter: + def __init__(self, enabled: bool = True, cache_ttl_seconds: int = 30) -> None: + self.enabled = enabled and os.name == "nt" + self.cache_ttl_seconds = cache_ttl_seconds + self.readers: dict[str, UsnJournalReader] = {} + self.useful_refs: dict[tuple[str, int], float] = {} + self.unavailable_volumes: set[str] = set() + + def prepare_for_paths(self, paths: Iterable[Path], logger: Any) -> None: + if not self.enabled: + return + + for path in paths: + volume = volume_name_for_path(path) + if not volume or volume in self.readers or volume in self.unavailable_volumes: + continue + try: + self.readers[volume] = UsnJournalReader.open(volume) + logger.info("USN journal filter enabled for volume: %s", volume) + except OSError as exc: + self.unavailable_volumes.add(volume) + logger.warning("USN journal unavailable for %s: %s", volume, exc) + + def is_meaningful_change(self, path: Path, event_type: str, logger: Any) -> bool: + if not self.enabled: + return True + if event_type == "deleted": + return False + + volume = volume_name_for_path(path) + if not volume: + return True + if volume in self.unavailable_volumes: + return True + + reader = self.readers.get(volume) + if reader is None: + try: + reader = UsnJournalReader.open(volume) + self.readers[volume] = reader + except OSError as exc: + self.unavailable_volumes.add(volume) + logger.warning("USN journal unavailable for %s: %s", volume, exc) + return True + + try: + file_ref = file_reference_number(path) + except OSError as exc: + logger.info("USN skipped; cannot read file reference for %s: %s", path, exc) + return False + + for attempt in range(4): + try: + self._drain_reader(volume, reader) + except OSError as exc: + self.unavailable_volumes.add(volume) + logger.warning("USN journal read failed for %s: %s", volume, exc) + return True + self._purge_expired() + if (volume, file_ref) in self.useful_refs: + return True + if attempt < 3: + time.sleep(0.2) + + logger.info("Ignored watchdog event without meaningful USN reason: %s", path) + return False + + def _drain_reader(self, volume: str, reader: UsnJournalReader) -> None: + expires_at = time.time() + self.cache_ttl_seconds + for record in reader.read_available(): + if is_meaningful_usn_reason(record.reason): + self.useful_refs[(volume, record.file_reference_number)] = expires_at + + def _purge_expired(self) -> None: + now = time.time() + expired = [key for key, expires_at in self.useful_refs.items() if expires_at < now] + for key in expired: + self.useful_refs.pop(key, None) + + +class UsnJournalReader: + def __init__(self, volume: str, handle: int, journal_id: int, next_usn: int) -> None: + self.volume = volume + self.handle = handle + self.journal_id = journal_id + self.next_usn = next_usn + + @classmethod + def open(cls, volume: str) -> UsnJournalReader: + handle = _create_handle(volume, GENERIC_READ) + try: + journal_id, next_usn = _query_usn_journal(handle) + return cls(volume, handle, journal_id, next_usn) + except Exception: + _close_handle(handle) + raise + + def read_available(self) -> list[UsnRecord]: + output = _device_io_control( + self.handle, + FSCTL_READ_USN_JOURNAL, + _pack_read_usn_journal_data(self.next_usn, self.journal_id), + 1024 * 1024, + ) + if len(output) < 8: + return [] + + self.next_usn = struct.unpack_from(" None: + _close_handle(self.handle) + + +def is_meaningful_usn_reason(reason: int) -> bool: + return bool(reason & MEANINGFUL_REASON_MASK) + + +def parse_usn_records(data: bytes) -> list[UsnRecord]: + records: list[UsnRecord] = [] + offset = 0 + length = len(data) + while offset + 8 <= length: + record_length, major_version, _minor_version = struct.unpack_from(" length: + break + + if major_version == 2 and record_length >= 60: + file_ref = struct.unpack_from("= 76: + file_ref = int.from_bytes(data[offset + 8 : offset + 24], "little") + reason = struct.unpack_from(" str | None: + drive = path.drive + if not drive: + return None + return f"\\\\.\\{drive.rstrip(':')}:" + + +def file_reference_number(path: Path) -> int: + handle = _create_handle(str(path), FILE_READ_ATTRIBUTES) + try: + info = BY_HANDLE_FILE_INFORMATION() + if not ctypes.windll.kernel32.GetFileInformationByHandle(handle, ctypes.byref(info)): + raise ctypes.WinError() + return (info.nFileIndexHigh << 32) | info.nFileIndexLow + finally: + _close_handle(handle) + + +def _query_usn_journal(handle: int) -> tuple[int, int]: + output = _device_io_control(handle, FSCTL_QUERY_USN_JOURNAL, None, 128) + if len(output) < 24: + raise UsnJournalUnavailable("FSCTL_QUERY_USN_JOURNAL returned too few bytes") + journal_id = struct.unpack_from(" bytes: + return struct.pack( + " bytes: + out_buffer = ctypes.create_string_buffer(output_size) + bytes_returned = ctypes.c_ulong(0) + if input_buffer is None: + in_buffer = None + in_size = 0 + else: + in_buffer = ctypes.create_string_buffer(input_buffer) + in_size = len(input_buffer) + + ok = kernel32.DeviceIoControl( + handle, + control_code, + in_buffer, + in_size, + out_buffer, + output_size, + ctypes.byref(bytes_returned), + None, + ) + if not ok: + raise ctypes.WinError() + return out_buffer.raw[: bytes_returned.value] + + +def _create_handle(path: str, desired_access: int) -> int: + handle = kernel32.CreateFileW( + str(path), + desired_access, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + None, + OPEN_EXISTING, + FILE_FLAG_BACKUP_SEMANTICS, + None, + ) + if handle == INVALID_HANDLE_VALUE: + raise ctypes.WinError() + return handle + + +def _close_handle(handle: int) -> None: + kernel32.CloseHandle(handle) + + +class BY_HANDLE_FILE_INFORMATION(ctypes.Structure): + _fields_ = [ + ("dwFileAttributes", ctypes.c_ulong), + ("ftCreationTime_dwLowDateTime", ctypes.c_ulong), + ("ftCreationTime_dwHighDateTime", ctypes.c_ulong), + ("ftLastAccessTime_dwLowDateTime", ctypes.c_ulong), + ("ftLastAccessTime_dwHighDateTime", ctypes.c_ulong), + ("ftLastWriteTime_dwLowDateTime", ctypes.c_ulong), + ("ftLastWriteTime_dwHighDateTime", ctypes.c_ulong), + ("dwVolumeSerialNumber", ctypes.c_ulong), + ("nFileSizeHigh", ctypes.c_ulong), + ("nFileSizeLow", ctypes.c_ulong), + ("nNumberOfLinks", ctypes.c_ulong), + ("nFileIndexHigh", ctypes.c_ulong), + ("nFileIndexLow", ctypes.c_ulong), + ] diff --git a/src/bakrest/watcher.py b/src/bakrest/watcher.py index af90cfa..5ff8c25 100644 --- a/src/bakrest/watcher.py +++ b/src/bakrest/watcher.py @@ -2,13 +2,13 @@ from __future__ import annotations import logging import threading -import time from pathlib import Path from .config import AppConfig from .filters import WatchFilter from .logging_setup import close_logger from .registry import ChangeEvent, ChangeRegistry +from .usn_journal import UsnChangeFilter class WatchdogRuntime: @@ -18,6 +18,7 @@ class WatchdogRuntime: self.stop_event = threading.Event() self.registry = ChangeRegistry(config.storage.change_registry) self.filter = WatchFilter(config.watch) + self.usn_filter = UsnChangeFilter() def run(self) -> None: try: @@ -34,6 +35,7 @@ class WatchdogRuntime: observer = Observer() watched = 0 + self.usn_filter.prepare_for_paths(self.config.watch.include_dirs, self.logger) for include_dir in self.config.watch.include_dirs: if include_dir.exists(): observer.schedule(Handler(), str(include_dir), recursive=True) @@ -72,19 +74,11 @@ class WatchdogRuntime: return size, mtime_ns = _safe_snapshot(target_path) - if event_type == "modified" and not _mtime_is_recent( - mtime_ns, - self.config.watch.modified_event_max_age_seconds, + if event_type in {"created", "modified", "moved"} and not self.usn_filter.is_meaningful_change( + target_path, + event_type, + self.logger, ): - self.logger.info("Ignored stale modified event: %s", target_path) - return - - if event_type == "modified" and not self.registry.has_snapshot_changed( - str(target_path), - size, - mtime_ns, - ): - self.logger.info("Ignored metadata-only modified event: %s", target_path) return change = ChangeEvent( @@ -108,10 +102,3 @@ def _safe_snapshot(path: Path) -> tuple[int | None, int | None]: return stat.st_size, stat.st_mtime_ns except OSError: return None, None - - -def _mtime_is_recent(mtime_ns: int | None, max_age_seconds: int) -> bool: - if mtime_ns is None: - return False - age_seconds = time.time() - (mtime_ns / 1_000_000_000) - return age_seconds <= max_age_seconds diff --git a/tests/test_usn_journal.py b/tests/test_usn_journal.py new file mode 100644 index 0000000..54e542e --- /dev/null +++ b/tests/test_usn_journal.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import struct + +from bakrest.usn_journal import ( + USN_REASON_BASIC_INFO_CHANGE, + USN_REASON_CLOSE, + USN_REASON_DATA_EXTEND, + USN_REASON_DATA_OVERWRITE, + USN_REASON_FILE_CREATE, + USN_REASON_RENAME_NEW_NAME, + is_meaningful_usn_reason, + parse_usn_records, +) + + +def test_usn_reason_filter_accepts_real_data_changes() -> None: + assert is_meaningful_usn_reason(USN_REASON_DATA_OVERWRITE) + assert is_meaningful_usn_reason(USN_REASON_DATA_EXTEND | USN_REASON_CLOSE) + assert is_meaningful_usn_reason(USN_REASON_FILE_CREATE) + assert is_meaningful_usn_reason(USN_REASON_RENAME_NEW_NAME) + + +def test_usn_reason_filter_rejects_metadata_only_changes() -> None: + assert not is_meaningful_usn_reason(USN_REASON_BASIC_INFO_CHANGE) + assert not is_meaningful_usn_reason(USN_REASON_CLOSE) + + +def test_parse_usn_record_v2_reason_and_file_reference() -> None: + record = bytearray(64) + struct.pack_into(" None: + record = bytearray(80) + struct.pack_into(" None: - old_mtime_ns = int((time.time() - 3600) * 1_000_000_000) - - assert not _mtime_is_recent(old_mtime_ns, 300) - - -def test_mtime_is_recent_accepts_recent_timestamps() -> None: - recent_mtime_ns = int(time.time() * 1_000_000_000) - - assert _mtime_is_recent(recent_mtime_ns, 300)