Coverage for src/local_deep_research/security/file_integrity/integrity_manager.py: 97%
228 statements
« prev ^ index » next coverage.py v7.15.1, created at 2026-07-19 23:35 +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# Import session context conditionally (requires Flask)
20try:
21 from ...database.session_context import get_user_db_session
23 _has_session_context = True
24except ImportError:
25 _has_session_context = False
26 # Provide stub for type checking
27 get_user_db_session = None # type: ignore
30# Reason returned by verify_file when a file has no integrity record. Unknown
31# files are rejected (fail closed) rather than auto-registered, to avoid a
32# trust-on-first-use gap where deleted records or newly injected files would
33# otherwise bypass integrity verification.
34NO_INTEGRITY_RECORD = "no_integrity_record"
37class FileIntegrityManager:
38 """
39 Central service for file integrity verification.
41 Features:
42 - Smart verification (only verify if file modified)
43 - Embedded statistics (low overhead)
44 - Sparse failure logging (audit trail)
45 - Multi-verifier support (different file types)
46 - Automatic cleanup of old failure records
47 """
49 # Configuration for automatic cleanup
50 MAX_FAILURES_PER_FILE = 100 # Keep at most this many failures per file
51 MAX_TOTAL_FAILURES = 10000 # Global limit across all files
53 def __init__(self, username: str, password: Optional[str] = None):
54 """
55 Initialize file integrity manager.
57 Args:
58 username: Username for database access
59 password: Optional password for encrypted database
61 Raises:
62 ImportError: If Flask/session_context not available
63 """
64 if not _has_session_context:
65 raise ImportError(
66 "FileIntegrityManager requires Flask and database session context. "
67 "Install Flask to use this feature."
68 )
70 self.username = username
71 self.password = password
72 self.verifiers: List[BaseFileVerifier] = []
74 # Run startup cleanup to remove old failures
75 try:
76 deleted = self.cleanup_all_old_failures()
77 if deleted > 0:
78 logger.info(
79 f"[FILE_INTEGRITY] Startup cleanup: removed {deleted} old failure records"
80 )
81 except Exception:
82 logger.warning("[FILE_INTEGRITY] Startup cleanup failed")
84 @contextmanager
85 def _integrity_write(self, session, nested: bool):
86 """Wrap an integrity write so it settles correctly for the caller.
88 ``nested=False`` (default, standalone callers): commit the
89 (thread-local) session — immediate durability.
91 ``nested=True``: run the write inside a SAVEPOINT and DON'T commit, so
92 the caller's outer transaction commits it. REQUIRED when record_file /
93 verify_file run reentrantly INSIDE another open transaction — the vector
94 store calls them during apply(), which has just-flushed, uncommitted
95 DocumentChunk rows on the SAME thread-local session. A plain commit
96 would settle those rows early, and a transient-failure rollback would
97 DISCARD them; a SAVEPOINT scopes any failure to just the integrity
98 write, leaving the caller's rows intact, and both commit together.
99 Same connection, so no second-writer SQLite contention.
100 """
101 if nested:
102 with session.begin_nested():
103 yield
104 else:
105 yield
106 session.commit()
108 @contextmanager
109 def _session_for(self, nested: bool):
110 """Yield the session for an integrity op.
112 Standalone (nested=False): a normal ``get_user_db_session``.
114 Reentrant (nested=True): the caller's LIVE thread-local session,
115 obtained via ``get_current_thread_session`` — NOT ``get_user_db_session``.
116 Re-entering ``get_user_db_session`` in a NON-request (background/
117 scheduler) context routes through ``ThreadSessionManager.get_session``,
118 which on re-acquisition runs a validation ``SELECT 1`` and then
119 ``session.rollback()`` to release the DEFERRED lock — that rollback
120 would discard the CALLER's just-flushed, uncommitted DocumentChunk rows
121 before our SAVEPOINT ever runs (silent data loss: vectors persisted,
122 text rows gone). The live-session accessor has no such side effect.
123 """
124 if nested:
125 from ...database.thread_local_session import (
126 get_current_thread_session,
127 )
129 existing = get_current_thread_session()
130 if existing is not None:
131 yield existing
132 return
133 with get_user_db_session(self.username, self.password) as session:
134 yield session
136 def _normalize_path(self, file_path: Path) -> str:
137 """
138 Normalize path for consistent storage and lookup.
140 Resolves symlinks, makes absolute, and normalizes separators
141 to ensure the same file is always represented the same way.
143 Args:
144 file_path: Path to normalize
146 Returns:
147 Normalized path string
148 """
149 return str(file_path.resolve())
151 def register_verifier(self, verifier: BaseFileVerifier) -> None:
152 """
153 Register a file type verifier.
155 Args:
156 verifier: Verifier instance to register
157 """
158 self.verifiers.append(verifier)
159 logger.debug(
160 f"[FILE_INTEGRITY] Registered verifier for type: {verifier.get_file_type()}"
161 )
163 def record_file(
164 self,
165 file_path: Path,
166 related_entity_type: Optional[str] = None,
167 related_entity_id: Optional[int] = None,
168 *,
169 nested: bool = False,
170 ) -> FileIntegrityRecord:
171 """
172 Create or update integrity record for a file.
174 Args:
175 file_path: Path to file to record
176 related_entity_type: Optional related entity type (e.g., 'rag_index')
177 related_entity_id: Optional related entity ID
179 Returns:
180 FileIntegrityRecord instance
182 Raises:
183 FileNotFoundError: If file doesn't exist
184 ValueError: If no verifier handles this file type
185 """
186 if not file_path.exists():
187 raise FileNotFoundError(f"File not found: {file_path}")
189 verifier = self._get_verifier_for_file(file_path)
190 if not verifier:
191 raise ValueError(f"No verifier registered for file: {file_path}")
193 # Calculate checksum and get file stats
194 checksum = verifier.calculate_checksum(file_path)
195 file_stat = file_path.stat()
196 normalized_path = self._normalize_path(file_path)
198 nested_error: Optional[Exception] = None
199 with self._session_for(nested) as session:
200 # Check if record exists (using normalized path)
201 record = (
202 session.query(FileIntegrityRecord)
203 .filter_by(file_path=normalized_path)
204 .first()
205 )
207 try:
208 with self._integrity_write(session, nested):
209 if record:
210 # Update existing record
211 record.checksum = checksum
212 record.file_size = file_stat.st_size
213 record.file_mtime = file_stat.st_mtime
214 record.algorithm = verifier.get_algorithm()
215 record.updated_at = datetime.now(UTC)
216 logger.info(
217 f"[FILE_INTEGRITY] Updated record for: {file_path}"
218 )
219 else:
220 # Create new record
221 record = FileIntegrityRecord(
222 file_path=normalized_path,
223 file_type=verifier.get_file_type(),
224 checksum=checksum,
225 algorithm=verifier.get_algorithm(),
226 file_size=file_stat.st_size,
227 file_mtime=file_stat.st_mtime,
228 verify_on_load=True,
229 allow_modifications=verifier.allows_modifications(),
230 related_entity_type=related_entity_type,
231 related_entity_id=related_entity_id,
232 total_verifications=0,
233 consecutive_successes=0,
234 consecutive_failures=0,
235 )
236 session.add(record)
237 logger.info(
238 f"[FILE_INTEGRITY] Created record for: {file_path} (type: {verifier.get_file_type()})"
239 )
240 except Exception as exc:
241 if not nested:
242 # Standalone: let get_user_db_session roll back the failed
243 # write (original behavior — nothing else is pending).
244 raise
245 # Nested: the SAVEPOINT already rolled back JUST this write; the
246 # caller's flushed rows are intact. We must NOT let the exception
247 # escape this block, or get_user_db_session's except would do a
248 # FULL session.rollback() and discard them. Exit cleanly, then
249 # re-raise below so apply()'s retry/failure logic still sees it.
250 nested_error = exc
252 if nested_error is None:
253 session.refresh(record)
255 if nested_error is not None:
256 raise nested_error
257 return record
259 def verify_file(
260 self, file_path: Path, force: bool = False, *, nested: bool = False
261 ) -> Tuple[bool, Optional[str]]:
262 """
263 Verify file integrity with smart checking.
265 Only verifies if:
266 - File modification time changed since last verification, OR
267 - force=True
269 Args:
270 file_path: Path to file to verify
271 force: Force verification even if file hasn't changed
273 Returns:
274 Tuple of (success, reason_if_failed)
275 """
276 normalized_path = self._normalize_path(file_path)
278 with self._session_for(nested) as session:
279 record = (
280 session.query(FileIntegrityRecord)
281 .filter_by(file_path=normalized_path)
282 .first()
283 )
285 if not record:
286 logger.error(
287 f"[FILE_INTEGRITY] No integrity record found for {file_path}, refusing to load"
288 )
289 return False, NO_INTEGRITY_RECORD
291 # Check if verification needed
292 if not force and not self._needs_verification(record, file_path):
293 logger.debug(
294 f"[FILE_INTEGRITY] Skipping verification for {file_path} (unchanged)"
295 )
296 return True, None
298 # Perform verification (the RESULT below is computed here and is
299 # independent of the stats bookkeeping write that follows).
300 passed, reason = self._do_verification(record, file_path, session)
302 try:
303 with self._integrity_write(session, nested):
304 # Update statistics
305 self._update_stats(record, passed, session)
307 # Log failure if needed
308 if not passed:
309 self._log_failure(
310 record,
311 file_path,
312 reason or "Unknown failure",
313 session,
314 )
315 except Exception:
316 if not nested:
317 raise
318 # Nested: the SAVEPOINT rolled back only this stats write; the
319 # caller's flushed rows are intact. Do NOT let the exception
320 # escape (get_user_db_session would FULL-rollback the caller's
321 # session). Stats bookkeeping is best-effort — the verification
322 # result is already decided, so log and return it.
323 logger.warning(
324 "[FILE_INTEGRITY] Could not persist verification stats for "
325 f"{file_path} (verification result unaffected)"
326 )
328 if passed:
329 logger.info(
330 f"[FILE_INTEGRITY] Verification passed: {file_path}"
331 )
332 else:
333 logger.error(
334 f"[FILE_INTEGRITY] Verification FAILED: {file_path} - {reason}"
335 )
337 return passed, reason
339 def update_checksum(self, file_path: Path) -> None:
340 """
341 Update checksum after legitimate file modification.
343 Use this when you know a file was legitimately modified
344 and want to update the baseline checksum.
346 Args:
347 file_path: Path to file
349 Raises:
350 FileNotFoundError: If file doesn't exist
351 ValueError: If no record exists for file
352 """
353 if not file_path.exists():
354 raise FileNotFoundError(f"File not found: {file_path}")
356 verifier = self._get_verifier_for_file(file_path)
357 if not verifier:
358 raise ValueError(f"No verifier registered for file: {file_path}")
360 checksum = verifier.calculate_checksum(file_path)
361 file_stat = file_path.stat()
363 with get_user_db_session(self.username, self.password) as session:
364 record = (
365 session.query(FileIntegrityRecord)
366 .filter_by(file_path=str(file_path))
367 .first()
368 )
370 if not record:
371 raise ValueError(f"No integrity record exists for: {file_path}")
373 record.checksum = checksum
374 record.file_size = file_stat.st_size
375 record.file_mtime = file_stat.st_mtime
376 record.updated_at = datetime.now(UTC)
378 session.commit()
379 logger.info(f"[FILE_INTEGRITY] Updated checksum for: {file_path}")
381 def get_file_stats(self, file_path: Path) -> Optional[dict]:
382 """
383 Get verification statistics for a file.
385 Args:
386 file_path: Path to file
388 Returns:
389 Dictionary of stats or None if no record exists
390 """
391 with get_user_db_session(self.username, self.password) as session:
392 record = (
393 session.query(FileIntegrityRecord)
394 .filter_by(file_path=str(file_path))
395 .first()
396 )
398 if not record:
399 return None
401 return {
402 "total_verifications": record.total_verifications,
403 "last_verified_at": record.last_verified_at,
404 "last_verification_passed": record.last_verification_passed,
405 "consecutive_successes": record.consecutive_successes,
406 "consecutive_failures": record.consecutive_failures,
407 "file_type": record.file_type,
408 "created_at": record.created_at,
409 }
411 def get_failure_history(
412 self, file_path: Path, limit: int = 100
413 ) -> List[FileVerificationFailure]:
414 """
415 Get failure history for a file.
417 Args:
418 file_path: Path to file
419 limit: Maximum number of failures to return
421 Returns:
422 List of failure records
423 """
424 with get_user_db_session(self.username, self.password) as session:
425 record = (
426 session.query(FileIntegrityRecord)
427 .filter_by(file_path=str(file_path))
428 .first()
429 )
431 if not record:
432 return []
434 failures = (
435 session.query(FileVerificationFailure)
436 .filter_by(file_record_id=record.id)
437 .order_by(FileVerificationFailure.verified_at.desc())
438 .limit(limit)
439 .all()
440 )
442 # Detach from session
443 for f in failures:
444 session.expunge(f)
446 return failures
448 # Internal methods
450 def _get_verifier_for_file(
451 self, file_path: Path
452 ) -> Optional[BaseFileVerifier]:
453 """Find verifier that handles this file type."""
454 for verifier in self.verifiers:
455 if verifier.should_verify(file_path):
456 return verifier
457 return None
459 def _needs_verification(
460 self, record: FileIntegrityRecord, file_path: Path
461 ) -> bool:
462 """
463 Check if file needs verification.
465 Only verify if file modification time changed since last verification.
466 """
467 if not file_path.exists():
468 return True # File missing needs verification
470 if not record.last_verified_at:
471 return True # Never verified
473 current_mtime = file_path.stat().st_mtime
475 # Compare with stored mtime
476 if record.file_mtime is None:
477 return True # No mtime stored
479 # Verify if file was modified (allow small floating point differences)
480 return abs(current_mtime - record.file_mtime) > 0.001
482 def _do_verification(
483 self, record: FileIntegrityRecord, file_path: Path, session
484 ) -> Tuple[bool, Optional[str]]:
485 """
486 Perform actual verification.
488 Returns:
489 Tuple of (success, reason_if_failed)
490 """
491 # Check file exists
492 if not file_path.exists():
493 return False, "file_missing"
495 # Get verifier
496 verifier = self._get_verifier_for_file(file_path)
497 if not verifier:
498 return False, "no_verifier"
500 # Calculate current checksum
501 try:
502 current_checksum = verifier.calculate_checksum(file_path)
503 except Exception as e:
504 logger.exception("[FILE_INTEGRITY] Failed to calculate checksum")
505 return False, f"checksum_calculation_failed: {str(e)}"
507 # Compare checksums
508 if current_checksum != record.checksum:
509 return False, "checksum_mismatch"
511 # Update file mtime in record
512 record.file_mtime = file_path.stat().st_mtime
514 return True, None
516 def _update_stats(
517 self, record: FileIntegrityRecord, passed: bool, session
518 ) -> None:
519 """Update verification statistics."""
520 record.total_verifications += 1
521 record.last_verified_at = datetime.now(UTC)
522 record.last_verification_passed = passed
524 if passed:
525 record.consecutive_successes += 1
526 record.consecutive_failures = 0
527 else:
528 record.consecutive_failures += 1
529 record.consecutive_successes = 0
531 def _log_failure(
532 self,
533 record: FileIntegrityRecord,
534 file_path: Path,
535 reason: str,
536 session,
537 ) -> None:
538 """Log verification failure to audit trail."""
539 # Get current checksum if possible
540 actual_checksum = None
541 file_size = None
543 if file_path.exists():
544 try:
545 verifier = self._get_verifier_for_file(file_path)
546 if verifier: 546 ↛ 556line 546 didn't jump to line 556 because the condition on line 546 was always true
547 actual_checksum = verifier.calculate_checksum(file_path)
548 file_size = file_path.stat().st_size
549 except Exception:
550 logger.debug(
551 "Checksum calculation failed for {}",
552 file_path,
553 exc_info=True,
554 )
556 failure = FileVerificationFailure(
557 file_record_id=record.id,
558 expected_checksum=record.checksum,
559 actual_checksum=actual_checksum,
560 file_size=file_size,
561 failure_reason=reason,
562 )
563 session.add(failure)
565 logger.warning(
566 f"[FILE_INTEGRITY] Logged failure for {file_path}: {reason}"
567 )
569 # Cleanup old failures for this file
570 self._cleanup_old_failures(record, session)
572 # Periodically check if global cleanup needed (every 100th file to avoid overhead)
573 if record.id % 100 == 0:
574 self._check_global_cleanup_needed(session)
576 def _cleanup_old_failures(
577 self, record: FileIntegrityRecord, session
578 ) -> None:
579 """
580 Clean up old failure records to prevent unbounded growth.
582 Keeps only the most recent MAX_FAILURES_PER_FILE failures per file.
583 """
584 # Count failures for this file
585 failure_count = (
586 session.query(FileVerificationFailure)
587 .filter_by(file_record_id=record.id)
588 .count()
589 )
591 if failure_count > self.MAX_FAILURES_PER_FILE:
592 # Delete oldest failures, keeping only the most recent MAX_FAILURES_PER_FILE
593 failures_to_delete = (
594 session.query(FileVerificationFailure)
595 .filter_by(file_record_id=record.id)
596 .order_by(FileVerificationFailure.verified_at.asc())
597 .limit(failure_count - self.MAX_FAILURES_PER_FILE)
598 .all()
599 )
601 for failure in failures_to_delete:
602 session.delete(failure)
604 logger.info(
605 f"[FILE_INTEGRITY] Cleaned up {len(failures_to_delete)} old failures for file_record {record.id}"
606 )
608 def _check_global_cleanup_needed(self, session) -> None:
609 """
610 Check if global cleanup is needed and run it if threshold exceeded.
612 Only runs cleanup if failure count exceeds MAX_TOTAL_FAILURES by 20%.
613 This prevents constant cleanup while allowing some buffer.
614 """
615 threshold = int(self.MAX_TOTAL_FAILURES * 1.2) # 20% over limit
616 total_failures = session.query(FileVerificationFailure).count()
618 if total_failures > threshold:
619 logger.info(
620 f"[FILE_INTEGRITY] Global failure count ({total_failures}) exceeds threshold ({threshold}), "
621 f"running cleanup..."
622 )
624 # Delete oldest failures to get under limit
625 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES
627 failures_to_delete = (
628 session.query(FileVerificationFailure)
629 .order_by(FileVerificationFailure.verified_at.asc())
630 .limit(failures_to_delete_count)
631 .all()
632 )
634 for failure in failures_to_delete:
635 session.delete(failure)
637 logger.info(
638 f"[FILE_INTEGRITY] Threshold cleanup: deleted {len(failures_to_delete)} old failures"
639 )
641 def cleanup_all_old_failures(self) -> int:
642 """
643 Global cleanup of failure records across all files.
645 Enforces MAX_TOTAL_FAILURES limit by removing oldest failures.
647 Returns:
648 Number of records deleted
649 """
650 with get_user_db_session(self.username, self.password) as session:
651 total_failures = session.query(FileVerificationFailure).count()
653 if total_failures <= self.MAX_TOTAL_FAILURES:
654 return 0
656 # Delete oldest failures to get under limit
657 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES
659 failures_to_delete = (
660 session.query(FileVerificationFailure)
661 .order_by(FileVerificationFailure.verified_at.asc())
662 .limit(failures_to_delete_count)
663 .all()
664 )
666 for failure in failures_to_delete:
667 session.delete(failure)
669 session.commit()
671 logger.info(
672 f"[FILE_INTEGRITY] Global cleanup: deleted {len(failures_to_delete)} old failures "
673 f"(total was {total_failures}, now {total_failures - len(failures_to_delete)})"
674 )
676 return len(failures_to_delete)
678 def get_total_failure_count(self) -> int:
679 """
680 Get total number of failure records across all files.
682 Returns:
683 Total count of failure records
684 """
685 with get_user_db_session(self.username, self.password) as session:
686 return session.query(FileVerificationFailure).count()