Filtra eventi watchdog con USN journal

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

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;
+- watchdog intercetta gli eventi file ma non li considera automaticamente validi;
-- se esiste gia' uno snapshot locale per quel file, l'evento viene registrato solo se dimensione o timestamp risultano diversi dallo snapshot conosciuto;
+- per eventi `created`, `modified` e `moved`, il servizio consulta il journal USN NTFS;
-- i file gia' presenti nel disco entrano quindi in coda solo quando vengono realmente salvati/modificati, non quando vengono solo navigati.
+- 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
 

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",

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(),

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()

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",

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("<q", output, 0)[0]
+        return parse_usn_records(output[8:])
+
+    def close(self) -> 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("<IHH", data, offset)
+        if record_length <= 0 or offset + record_length > length:
+            break
+
+        if major_version == 2 and record_length >= 60:
+            file_ref = struct.unpack_from("<Q", data, offset + 8)[0]
+            reason = struct.unpack_from("<I", data, offset + 40)[0]
+            records.append(UsnRecord(file_ref, reason))
+        elif major_version == 3 and record_length >= 76:
+            file_ref = int.from_bytes(data[offset + 8 : offset + 24], "little")
+            reason = struct.unpack_from("<I", data, offset + 56)[0]
+            records.append(UsnRecord(file_ref, reason))
+
+        offset += record_length
+    return records
+
+
+def volume_name_for_path(path: Path) -> 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("<Q", output, 0)[0]
+    next_usn = struct.unpack_from("<q", output, 16)[0]
+    return journal_id, next_usn
+
+
+def _pack_read_usn_journal_data(start_usn: int, journal_id: int) -> bytes:
+    return struct.pack(
+        "<qIIQQQ",
+        start_usn,
+        0xFFFFFFFF,
+        0,
+        0,
+        0,
+        journal_id,
+    )
+
+
+def _device_io_control(
+    handle: int,
+    control_code: int,
+    input_buffer: bytes | None,
+    output_size: int,
+) -> 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),
+    ]

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(
+        if event_type in {"created", "modified", "moved"} and not self.usn_filter.is_meaningful_change(
-            mtime_ns,
+            target_path,
-            self.config.watch.modified_event_max_age_seconds,
+            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

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("<IHH", record, 0, 64, 2, 0)
+    struct.pack_into("<Q", record, 8, 12345)
+    struct.pack_into("<I", record, 40, USN_REASON_DATA_OVERWRITE)
+
+    records = parse_usn_records(bytes(record))
+
+    assert len(records) == 1
+    assert records[0].file_reference_number == 12345
+    assert records[0].reason == USN_REASON_DATA_OVERWRITE
+
+
+def test_parse_usn_record_v3_reason_and_file_reference() -> None:
+    record = bytearray(80)
+    struct.pack_into("<IHH", record, 0, 80, 3, 0)
+    record[8:24] = (12345).to_bytes(16, "little")
+    struct.pack_into("<I", record, 56, USN_REASON_FILE_CREATE)
+
+    records = parse_usn_records(bytes(record))
+
+    assert len(records) == 1
+    assert records[0].file_reference_number == 12345
+    assert records[0].reason == USN_REASON_FILE_CREATE

tests/test_watcher.py

@@ -1,17 +0,0 @@
-from __future__ import annotations
-
-import time
-
-from bakrest.watcher import _mtime_is_recent
-
-
-def test_mtime_is_recent_rejects_old_timestamps() -> 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)