Coverage for src/local_deep_research/security/file_integrity/integrity_manager.py: 98%
223 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2File Integrity Manager - Main service for file integrity verification.
4Provides smart verification with embedded statistics and sparse failure logging.
5"""
7from contextlib import contextmanager
8from datetime import datetime, UTC
9from pathlib import Path
10from typing import Optional, Tuple, List
11from loguru import logger
13from .base_verifier import BaseFileVerifier
14from ...database.models.file_integrity import (
15 FileIntegrityRecord,
16 FileVerificationFailure,
17)
19# The database session context is mandatory for integrity verification. Keep
20# this as a direct import: a circular import or other bootstrap/load-order
21# failure must propagate with its original traceback instead of being caught
22# and later replaced by misleading optional-Flask guidance.
23from ...database.session_context import get_user_db_session
26# Reason returned by verify_file when a file has no integrity record. Unknown
27# files are rejected (fail closed) rather than auto-registered, to avoid a
28# trust-on-first-use gap where deleted records or newly injected files would
29# otherwise bypass integrity verification.
30NO_INTEGRITY_RECORD = "no_integrity_record"
33class FileIntegrityManager:
34 """
35 Central service for file integrity verification.
37 Features:
38 - Smart verification (only verify if file modified)
39 - Embedded statistics (low overhead)
40 - Sparse failure logging (audit trail)
41 - Multi-verifier support (different file types)
42 - Automatic cleanup of old failure records
43 """
45 # Configuration for automatic cleanup
46 MAX_FAILURES_PER_FILE = 100 # Keep at most this many failures per file
47 MAX_TOTAL_FAILURES = 10000 # Global limit across all files
49 def __init__(self, username: str, password: Optional[str] = None):
50 """
51 Initialize file integrity manager.
53 Args:
54 username: Username for database access
55 password: Optional password for encrypted database
57 """
58 self.username = username
59 self.password = password
60 self.verifiers: List[BaseFileVerifier] = []
62 # Run startup cleanup to remove old failures
63 try:
64 deleted = self.cleanup_all_old_failures()
65 if deleted > 0:
66 logger.info(
67 f"[FILE_INTEGRITY] Startup cleanup: removed {deleted} old failure records"
68 )
69 except Exception:
70 logger.warning("[FILE_INTEGRITY] Startup cleanup failed")
72 @contextmanager
73 def _integrity_write(self, session, nested: bool):
74 """Wrap an integrity write so it settles correctly for the caller.
76 ``nested=False`` (default, standalone callers): commit the
77 (thread-local) session — immediate durability.
79 ``nested=True``: run the write inside a SAVEPOINT and DON'T commit, so
80 the caller's outer transaction commits it. REQUIRED when record_file /
81 verify_file run reentrantly INSIDE another open transaction — the vector
82 store calls them during apply(), which has just-flushed, uncommitted
83 DocumentChunk rows on the SAME thread-local session. A plain commit
84 would settle those rows early, and a transient-failure rollback would
85 DISCARD them; a SAVEPOINT scopes any failure to just the integrity
86 write, leaving the caller's rows intact, and both commit together.
87 Same connection, so no second-writer SQLite contention.
88 """
89 if nested:
90 with session.begin_nested():
91 yield
92 else:
93 yield
94 session.commit()
96 @contextmanager
97 def _session_for(self, nested: bool):
98 """Yield the session for an integrity op.
100 Standalone (nested=False): a normal ``get_user_db_session``.
102 Reentrant (nested=True): the caller's LIVE thread-local session,
103 obtained via ``get_current_thread_session`` — NOT ``get_user_db_session``.
104 Re-entering ``get_user_db_session`` in a NON-request (background/
105 scheduler) context routes through ``ThreadSessionManager.get_session``,
106 which on re-acquisition runs a validation ``SELECT 1`` and then
107 ``session.rollback()`` to release the DEFERRED lock — that rollback
108 would discard the CALLER's just-flushed, uncommitted DocumentChunk rows
109 before our SAVEPOINT ever runs (silent data loss: vectors persisted,
110 text rows gone). The live-session accessor has no such side effect.
111 """
112 if nested:
113 from ...database.thread_local_session import (
114 get_current_thread_session,
115 )
117 existing = get_current_thread_session()
118 if existing is not None:
119 yield existing
120 return
121 with get_user_db_session(self.username, self.password) as session:
122 yield session
124 def _normalize_path(self, file_path: Path) -> str:
125 """
126 Normalize path for consistent storage and lookup.
128 Resolves symlinks, makes absolute, and normalizes separators
129 to ensure the same file is always represented the same way.
131 Args:
132 file_path: Path to normalize
134 Returns:
135 Normalized path string
136 """
137 return str(file_path.resolve())
139 def register_verifier(self, verifier: BaseFileVerifier) -> None:
140 """
141 Register a file type verifier.
143 Args:
144 verifier: Verifier instance to register
145 """
146 self.verifiers.append(verifier)
147 logger.debug(
148 f"[FILE_INTEGRITY] Registered verifier for type: {verifier.get_file_type()}"
149 )
151 def record_file(
152 self,
153 file_path: Path,
154 related_entity_type: Optional[str] = None,
155 related_entity_id: Optional[int] = None,
156 *,
157 nested: bool = False,
158 ) -> FileIntegrityRecord:
159 """
160 Create or update integrity record for a file.
162 Args:
163 file_path: Path to file to record
164 related_entity_type: Optional related entity type (e.g., 'rag_index')
165 related_entity_id: Optional related entity ID
167 Returns:
168 FileIntegrityRecord instance
170 Raises:
171 FileNotFoundError: If file doesn't exist
172 ValueError: If no verifier handles this file type
173 """
174 if not file_path.exists():
175 raise FileNotFoundError(f"File not found: {file_path}")
177 verifier = self._get_verifier_for_file(file_path)
178 if not verifier:
179 raise ValueError(f"No verifier registered for file: {file_path}")
181 # Calculate checksum and get file stats
182 checksum = verifier.calculate_checksum(file_path)
183 file_stat = file_path.stat()
184 normalized_path = self._normalize_path(file_path)
186 nested_error: Optional[Exception] = None
187 with self._session_for(nested) as session:
188 # Check if record exists (using normalized path)
189 record = (
190 session.query(FileIntegrityRecord)
191 .filter_by(file_path=normalized_path)
192 .first()
193 )
195 try:
196 with self._integrity_write(session, nested):
197 if record:
198 # Update existing record
199 record.checksum = checksum
200 record.file_size = file_stat.st_size
201 record.file_mtime = file_stat.st_mtime
202 record.algorithm = verifier.get_algorithm()
203 record.updated_at = datetime.now(UTC)
204 # Re-recording installs a NEW baseline, so both
205 # verification streaks -- accumulated against the
206 # PREVIOUS checksum -- describe a file that no
207 # longer exists. Left in place they are charged to
208 # the new bytes: an index rebuilt after a
209 # quarantine started out still counted as the one
210 # that had just been thrown away, and
211 # get_file_stats reported statistics about a
212 # baseline the new one had never been verified
213 # against. consecutive_successes is reset alongside
214 # -- the two are a mutually exclusive pair in
215 # _update_stats, so resetting only one would leave
216 # the other crediting (or charging) the new
217 # baseline with verifications it never had.
218 # total_verifications / last_verified_at /
219 # last_verification_passed stay: they are history
220 # about the path, not about one baseline.
221 record.consecutive_failures = 0
222 record.consecutive_successes = 0
223 logger.info(
224 f"[FILE_INTEGRITY] Updated record for: {file_path}"
225 )
226 else:
227 # Create new record
228 record = FileIntegrityRecord(
229 file_path=normalized_path,
230 file_type=verifier.get_file_type(),
231 checksum=checksum,
232 algorithm=verifier.get_algorithm(),
233 file_size=file_stat.st_size,
234 file_mtime=file_stat.st_mtime,
235 verify_on_load=True,
236 allow_modifications=verifier.allows_modifications(),
237 related_entity_type=related_entity_type,
238 related_entity_id=related_entity_id,
239 total_verifications=0,
240 consecutive_successes=0,
241 consecutive_failures=0,
242 )
243 session.add(record)
244 logger.info(
245 f"[FILE_INTEGRITY] Created record for: {file_path} (type: {verifier.get_file_type()})"
246 )
247 except Exception as exc:
248 if not nested:
249 # Standalone: let get_user_db_session roll back the failed
250 # write (original behavior — nothing else is pending).
251 raise
252 # Nested: the SAVEPOINT already rolled back JUST this write; the
253 # caller's flushed rows are intact. We must NOT let the exception
254 # escape this block, or get_user_db_session's except would do a
255 # FULL session.rollback() and discard them. Exit cleanly, then
256 # re-raise below so apply()'s retry/failure logic still sees it.
257 nested_error = exc
259 if nested_error is None:
260 session.refresh(record)
262 if nested_error is not None:
263 raise nested_error
264 return record
266 def verify_file(
267 self, file_path: Path, force: bool = False, *, nested: bool = False
268 ) -> Tuple[bool, Optional[str]]:
269 """
270 Verify file integrity with smart checking.
272 Only verifies if:
273 - File modification time changed since last verification, OR
274 - force=True
276 Args:
277 file_path: Path to file to verify
278 force: Force verification even if file hasn't changed
280 Returns:
281 Tuple of (success, reason_if_failed)
282 """
283 normalized_path = self._normalize_path(file_path)
285 with self._session_for(nested) as session:
286 record = (
287 session.query(FileIntegrityRecord)
288 .filter_by(file_path=normalized_path)
289 .first()
290 )
292 if not record:
293 logger.error(
294 f"[FILE_INTEGRITY] No integrity record found for {file_path}, refusing to load"
295 )
296 return False, NO_INTEGRITY_RECORD
298 # Check if verification needed
299 if not force and not self._needs_verification(record, file_path):
300 logger.debug(
301 f"[FILE_INTEGRITY] Skipping verification for {file_path} (unchanged)"
302 )
303 return True, None
305 # Perform verification (the RESULT below is computed here and is
306 # independent of the stats bookkeeping write that follows).
307 passed, reason = self._do_verification(record, file_path, session)
309 try:
310 with self._integrity_write(session, nested):
311 # Update statistics
312 self._update_stats(record, passed, session)
314 # Log failure if needed
315 if not passed:
316 self._log_failure(
317 record,
318 file_path,
319 reason or "Unknown failure",
320 session,
321 )
322 except Exception:
323 if not nested:
324 raise
325 # Nested: the SAVEPOINT rolled back only this stats write; the
326 # caller's flushed rows are intact. Do NOT let the exception
327 # escape (get_user_db_session would FULL-rollback the caller's
328 # session). Stats bookkeeping is best-effort — the verification
329 # result is already decided, so log and return it.
330 logger.warning(
331 "[FILE_INTEGRITY] Could not persist verification stats for "
332 f"{file_path} (verification result unaffected)"
333 )
335 if passed:
336 logger.info(
337 f"[FILE_INTEGRITY] Verification passed: {file_path}"
338 )
339 else:
340 logger.error(
341 f"[FILE_INTEGRITY] Verification FAILED: {file_path} - {reason}"
342 )
344 return passed, reason
346 def update_checksum(self, file_path: Path) -> None:
347 """
348 Update checksum after legitimate file modification.
350 Use this when you know a file was legitimately modified
351 and want to update the baseline checksum.
353 Like ``record_file``, this installs a NEW baseline, so the same
354 argument applies to the verification streaks: the pair left in
355 place would describe the checksum that has just been replaced.
356 ``record_file`` owns that invariant and resets them (see the
357 comment there); this method currently has no production callers,
358 and the parity fix is tracked in #7019 rather than shipped
359 untested here.
361 Args:
362 file_path: Path to file
364 Raises:
365 FileNotFoundError: If file doesn't exist
366 ValueError: If no record exists for file
367 """
368 if not file_path.exists():
369 raise FileNotFoundError(f"File not found: {file_path}")
371 verifier = self._get_verifier_for_file(file_path)
372 if not verifier:
373 raise ValueError(f"No verifier registered for file: {file_path}")
375 checksum = verifier.calculate_checksum(file_path)
376 file_stat = file_path.stat()
378 with get_user_db_session(self.username, self.password) as session:
379 record = (
380 session.query(FileIntegrityRecord)
381 .filter_by(file_path=str(file_path))
382 .first()
383 )
385 if not record:
386 raise ValueError(f"No integrity record exists for: {file_path}")
388 record.checksum = checksum
389 record.file_size = file_stat.st_size
390 record.file_mtime = file_stat.st_mtime
391 record.updated_at = datetime.now(UTC)
393 session.commit()
394 logger.info(f"[FILE_INTEGRITY] Updated checksum for: {file_path}")
396 def get_file_stats(self, file_path: Path) -> Optional[dict]:
397 """
398 Get verification statistics for a file.
400 Args:
401 file_path: Path to file
403 Returns:
404 Dictionary of stats or None if no record exists
405 """
406 with get_user_db_session(self.username, self.password) as session:
407 record = (
408 session.query(FileIntegrityRecord)
409 .filter_by(file_path=str(file_path))
410 .first()
411 )
413 if not record:
414 return None
416 return {
417 "total_verifications": record.total_verifications,
418 "last_verified_at": record.last_verified_at,
419 "last_verification_passed": record.last_verification_passed,
420 "consecutive_successes": record.consecutive_successes,
421 "consecutive_failures": record.consecutive_failures,
422 "file_type": record.file_type,
423 "created_at": record.created_at,
424 }
426 def get_failure_history(
427 self, file_path: Path, limit: int = 100
428 ) -> List[FileVerificationFailure]:
429 """
430 Get failure history for a file.
432 Args:
433 file_path: Path to file
434 limit: Maximum number of failures to return
436 Returns:
437 List of failure records
438 """
439 with get_user_db_session(self.username, self.password) as session:
440 record = (
441 session.query(FileIntegrityRecord)
442 .filter_by(file_path=str(file_path))
443 .first()
444 )
446 if not record:
447 return []
449 failures = (
450 session.query(FileVerificationFailure)
451 .filter_by(file_record_id=record.id)
452 .order_by(FileVerificationFailure.verified_at.desc())
453 .limit(limit)
454 .all()
455 )
457 # Detach from session
458 for f in failures:
459 session.expunge(f)
461 return failures
463 # Internal methods
465 def _get_verifier_for_file(
466 self, file_path: Path
467 ) -> Optional[BaseFileVerifier]:
468 """Find verifier that handles this file type."""
469 for verifier in self.verifiers:
470 if verifier.should_verify(file_path):
471 return verifier
472 return None
474 def _needs_verification(
475 self, record: FileIntegrityRecord, file_path: Path
476 ) -> bool:
477 """
478 Check if file needs verification.
480 Only verify if file modification time changed since last verification.
481 """
482 if not file_path.exists():
483 return True # File missing needs verification
485 if not record.last_verified_at:
486 return True # Never verified
488 current_mtime = file_path.stat().st_mtime
490 # Compare with stored mtime
491 if record.file_mtime is None:
492 return True # No mtime stored
494 # Verify if file was modified (allow small floating point differences)
495 return abs(current_mtime - record.file_mtime) > 0.001
497 def _do_verification(
498 self, record: FileIntegrityRecord, file_path: Path, session
499 ) -> Tuple[bool, Optional[str]]:
500 """
501 Perform actual verification.
503 Returns:
504 Tuple of (success, reason_if_failed)
505 """
506 # Check file exists
507 if not file_path.exists():
508 return False, "file_missing"
510 # Get verifier
511 verifier = self._get_verifier_for_file(file_path)
512 if not verifier:
513 return False, "no_verifier"
515 # Calculate current checksum
516 try:
517 current_checksum = verifier.calculate_checksum(file_path)
518 except Exception as e:
519 logger.exception("[FILE_INTEGRITY] Failed to calculate checksum")
520 return False, f"checksum_calculation_failed: {str(e)}"
522 # Compare checksums
523 if current_checksum != record.checksum:
524 return False, "checksum_mismatch"
526 # Update file mtime in record
527 record.file_mtime = file_path.stat().st_mtime
529 return True, None
531 def _update_stats(
532 self, record: FileIntegrityRecord, passed: bool, session
533 ) -> None:
534 """Update verification statistics."""
535 record.total_verifications += 1
536 record.last_verified_at = datetime.now(UTC)
537 record.last_verification_passed = passed
539 if passed:
540 record.consecutive_successes += 1
541 record.consecutive_failures = 0
542 else:
543 record.consecutive_failures += 1
544 record.consecutive_successes = 0
546 def _log_failure(
547 self,
548 record: FileIntegrityRecord,
549 file_path: Path,
550 reason: str,
551 session,
552 ) -> None:
553 """Log verification failure to audit trail."""
554 # Get current checksum if possible
555 actual_checksum = None
556 file_size = None
558 if file_path.exists():
559 try:
560 verifier = self._get_verifier_for_file(file_path)
561 if verifier: 561 ↛ 571line 561 didn't jump to line 571 because the condition on line 561 was always true
562 actual_checksum = verifier.calculate_checksum(file_path)
563 file_size = file_path.stat().st_size
564 except Exception:
565 logger.debug(
566 "Checksum calculation failed for {}",
567 file_path,
568 exc_info=True,
569 )
571 failure = FileVerificationFailure(
572 file_record_id=record.id,
573 expected_checksum=record.checksum,
574 actual_checksum=actual_checksum,
575 file_size=file_size,
576 failure_reason=reason,
577 )
578 session.add(failure)
580 logger.warning(
581 f"[FILE_INTEGRITY] Logged failure for {file_path}: {reason}"
582 )
584 # Cleanup old failures for this file
585 self._cleanup_old_failures(record, session)
587 # Periodically check if global cleanup needed (every 100th file to avoid overhead)
588 if record.id % 100 == 0:
589 self._check_global_cleanup_needed(session)
591 def _cleanup_old_failures(
592 self, record: FileIntegrityRecord, session
593 ) -> None:
594 """
595 Clean up old failure records to prevent unbounded growth.
597 Keeps only the most recent MAX_FAILURES_PER_FILE failures per file.
598 """
599 # Count failures for this file
600 failure_count = (
601 session.query(FileVerificationFailure)
602 .filter_by(file_record_id=record.id)
603 .count()
604 )
606 if failure_count > self.MAX_FAILURES_PER_FILE:
607 # Delete oldest failures, keeping only the most recent MAX_FAILURES_PER_FILE
608 failures_to_delete = (
609 session.query(FileVerificationFailure)
610 .filter_by(file_record_id=record.id)
611 .order_by(FileVerificationFailure.verified_at.asc())
612 .limit(failure_count - self.MAX_FAILURES_PER_FILE)
613 .all()
614 )
616 for failure in failures_to_delete:
617 session.delete(failure)
619 logger.info(
620 f"[FILE_INTEGRITY] Cleaned up {len(failures_to_delete)} old failures for file_record {record.id}"
621 )
623 def _check_global_cleanup_needed(self, session) -> None:
624 """
625 Check if global cleanup is needed and run it if threshold exceeded.
627 Only runs cleanup if failure count exceeds MAX_TOTAL_FAILURES by 20%.
628 This prevents constant cleanup while allowing some buffer.
629 """
630 threshold = int(self.MAX_TOTAL_FAILURES * 1.2) # 20% over limit
631 total_failures = session.query(FileVerificationFailure).count()
633 if total_failures > threshold:
634 logger.info(
635 f"[FILE_INTEGRITY] Global failure count ({total_failures}) exceeds threshold ({threshold}), "
636 f"running cleanup..."
637 )
639 # Delete oldest failures to get under limit
640 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES
642 failures_to_delete = (
643 session.query(FileVerificationFailure)
644 .order_by(FileVerificationFailure.verified_at.asc())
645 .limit(failures_to_delete_count)
646 .all()
647 )
649 for failure in failures_to_delete:
650 session.delete(failure)
652 logger.info(
653 f"[FILE_INTEGRITY] Threshold cleanup: deleted {len(failures_to_delete)} old failures"
654 )
656 def cleanup_all_old_failures(self) -> int:
657 """
658 Global cleanup of failure records across all files.
660 Enforces MAX_TOTAL_FAILURES limit by removing oldest failures.
662 Returns:
663 Number of records deleted
664 """
665 with get_user_db_session(self.username, self.password) as session:
666 total_failures = session.query(FileVerificationFailure).count()
668 if total_failures <= self.MAX_TOTAL_FAILURES:
669 return 0
671 # Delete oldest failures to get under limit
672 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES
674 failures_to_delete = (
675 session.query(FileVerificationFailure)
676 .order_by(FileVerificationFailure.verified_at.asc())
677 .limit(failures_to_delete_count)
678 .all()
679 )
681 for failure in failures_to_delete:
682 session.delete(failure)
684 session.commit()
686 logger.info(
687 f"[FILE_INTEGRITY] Global cleanup: deleted {len(failures_to_delete)} old failures "
688 f"(total was {total_failures}, now {total_failures - len(failures_to_delete)})"
689 )
691 return len(failures_to_delete)
693 def get_total_failure_count(self) -> int:
694 """
695 Get total number of failure records across all files.
697 Returns:
698 Total count of failure records
699 """
700 with get_user_db_session(self.username, self.password) as session:
701 return session.query(FileVerificationFailure).count()