Coverage for src/local_deep_research/security/file_integrity/integrity

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``.

101

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 )

116

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

123

124 def _normalize_path(self, file_path: Path) -> str:

125 """

126 Normalize path for consistent storage and lookup.

127

128 Resolves symlinks, makes absolute, and normalizes separators

129 to ensure the same file is always represented the same way.

130

131 Args:

132 file_path: Path to normalize

133

134 Returns:

135 Normalized path string

136 """

137 return str(file_path.resolve())

138

139 def register_verifier(self, verifier: BaseFileVerifier) -> None:

140 """

141 Register a file type verifier.

142

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 )

150

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.

161

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

166

167 Returns:

168 FileIntegrityRecord instance

169

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}")

176

177 verifier = self._get_verifier_for_file(file_path)

178 if not verifier:

179 raise ValueError(f"No verifier registered for file: {file_path}")

180

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)

185

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 )

194

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 logger.info(

205 f"[FILE_INTEGRITY] Updated record for: {file_path}"

206 )

207 else:

208 # Create new record

209 record = FileIntegrityRecord(

210 file_path=normalized_path,

211 file_type=verifier.get_file_type(),

212 checksum=checksum,

213 algorithm=verifier.get_algorithm(),

214 file_size=file_stat.st_size,

215 file_mtime=file_stat.st_mtime,

216 verify_on_load=True,

217 allow_modifications=verifier.allows_modifications(),

218 related_entity_type=related_entity_type,

219 related_entity_id=related_entity_id,

220 total_verifications=0,

221 consecutive_successes=0,

222 consecutive_failures=0,

223 )

224 session.add(record)

225 logger.info(

226 f"[FILE_INTEGRITY] Created record for: {file_path} (type: {verifier.get_file_type()})"

227 )

228 except Exception as exc:

229 if not nested:

230 # Standalone: let get_user_db_session roll back the failed

231 # write (original behavior — nothing else is pending).

232 raise

233 # Nested: the SAVEPOINT already rolled back JUST this write; the

234 # caller's flushed rows are intact. We must NOT let the exception

235 # escape this block, or get_user_db_session's except would do a

236 # FULL session.rollback() and discard them. Exit cleanly, then

237 # re-raise below so apply()'s retry/failure logic still sees it.

238 nested_error = exc

239

240 if nested_error is None:

241 session.refresh(record)

242

243 if nested_error is not None:

244 raise nested_error

245 return record

246

247 def verify_file(

248 self, file_path: Path, force: bool = False, *, nested: bool = False

249 ) -> Tuple[bool, Optional[str]]:

250 """

251 Verify file integrity with smart checking.

252

253 Only verifies if:

254 - File modification time changed since last verification, OR

255 - force=True

256

257 Args:

258 file_path: Path to file to verify

259 force: Force verification even if file hasn't changed

260

261 Returns:

262 Tuple of (success, reason_if_failed)

263 """

264 normalized_path = self._normalize_path(file_path)

265

266 with self._session_for(nested) as session:

267 record = (

268 session.query(FileIntegrityRecord)

269 .filter_by(file_path=normalized_path)

270 .first()

271 )

272

273 if not record:

274 logger.error(

275 f"[FILE_INTEGRITY] No integrity record found for {file_path}, refusing to load"

276 )

277 return False, NO_INTEGRITY_RECORD

278

279 # Check if verification needed

280 if not force and not self._needs_verification(record, file_path):

281 logger.debug(

282 f"[FILE_INTEGRITY] Skipping verification for {file_path} (unchanged)"

283 )

284 return True, None

285

286 # Perform verification (the RESULT below is computed here and is

287 # independent of the stats bookkeeping write that follows).

288 passed, reason = self._do_verification(record, file_path, session)

289

290 try:

291 with self._integrity_write(session, nested):

292 # Update statistics

293 self._update_stats(record, passed, session)

294

295 # Log failure if needed

296 if not passed:

297 self._log_failure(

298 record,

299 file_path,

300 reason or "Unknown failure",

301 session,

302 )

303 except Exception:

304 if not nested:

305 raise

306 # Nested: the SAVEPOINT rolled back only this stats write; the

307 # caller's flushed rows are intact. Do NOT let the exception

308 # escape (get_user_db_session would FULL-rollback the caller's

309 # session). Stats bookkeeping is best-effort — the verification

310 # result is already decided, so log and return it.

311 logger.warning(

312 "[FILE_INTEGRITY] Could not persist verification stats for "

313 f"{file_path} (verification result unaffected)"

314 )

315

316 if passed:

317 logger.info(

318 f"[FILE_INTEGRITY] Verification passed: {file_path}"

319 )

320 else:

321 logger.error(

322 f"[FILE_INTEGRITY] Verification FAILED: {file_path} - {reason}"

323 )

324

325 return passed, reason

326

327 def update_checksum(self, file_path: Path) -> None:

328 """

329 Update checksum after legitimate file modification.

330

331 Use this when you know a file was legitimately modified

332 and want to update the baseline checksum.

333

334 Args:

335 file_path: Path to file

336

337 Raises:

338 FileNotFoundError: If file doesn't exist

339 ValueError: If no record exists for file

340 """

341 if not file_path.exists():

342 raise FileNotFoundError(f"File not found: {file_path}")

343

344 verifier = self._get_verifier_for_file(file_path)

345 if not verifier:

346 raise ValueError(f"No verifier registered for file: {file_path}")

347

348 checksum = verifier.calculate_checksum(file_path)

349 file_stat = file_path.stat()

350

351 with get_user_db_session(self.username, self.password) as session:

352 record = (

353 session.query(FileIntegrityRecord)

354 .filter_by(file_path=str(file_path))

355 .first()

356 )

357

358 if not record:

359 raise ValueError(f"No integrity record exists for: {file_path}")

360

361 record.checksum = checksum

362 record.file_size = file_stat.st_size

363 record.file_mtime = file_stat.st_mtime

364 record.updated_at = datetime.now(UTC)

365

366 session.commit()

367 logger.info(f"[FILE_INTEGRITY] Updated checksum for: {file_path}")

368

369 def get_file_stats(self, file_path: Path) -> Optional[dict]:

370 """

371 Get verification statistics for a file.

372

373 Args:

374 file_path: Path to file

375

376 Returns:

377 Dictionary of stats or None if no record exists

378 """

379 with get_user_db_session(self.username, self.password) as session:

380 record = (

381 session.query(FileIntegrityRecord)

382 .filter_by(file_path=str(file_path))

383 .first()

384 )

385

386 if not record:

387 return None

388

389 return {

390 "total_verifications": record.total_verifications,

391 "last_verified_at": record.last_verified_at,

392 "last_verification_passed": record.last_verification_passed,

393 "consecutive_successes": record.consecutive_successes,

394 "consecutive_failures": record.consecutive_failures,

395 "file_type": record.file_type,

396 "created_at": record.created_at,

397 }

398

399 def get_failure_history(

400 self, file_path: Path, limit: int = 100

401 ) -> List[FileVerificationFailure]:

402 """

403 Get failure history for a file.

404

405 Args:

406 file_path: Path to file

407 limit: Maximum number of failures to return

408

409 Returns:

410 List of failure records

411 """

412 with get_user_db_session(self.username, self.password) as session:

413 record = (

414 session.query(FileIntegrityRecord)

415 .filter_by(file_path=str(file_path))

416 .first()

417 )

418

419 if not record:

420 return []

421

422 failures = (

423 session.query(FileVerificationFailure)

424 .filter_by(file_record_id=record.id)

425 .order_by(FileVerificationFailure.verified_at.desc())

426 .limit(limit)

427 .all()

428 )

429

430 # Detach from session

431 for f in failures:

432 session.expunge(f)

433

434 return failures

435

436 # Internal methods

437

438 def _get_verifier_for_file(

439 self, file_path: Path

440 ) -> Optional[BaseFileVerifier]:

441 """Find verifier that handles this file type."""

442 for verifier in self.verifiers:

443 if verifier.should_verify(file_path):

444 return verifier

445 return None

446

447 def _needs_verification(

448 self, record: FileIntegrityRecord, file_path: Path

449 ) -> bool:

450 """

451 Check if file needs verification.

452

453 Only verify if file modification time changed since last verification.

454 """

455 if not file_path.exists():

456 return True # File missing needs verification

457

458 if not record.last_verified_at:

459 return True # Never verified

460

461 current_mtime = file_path.stat().st_mtime

462

463 # Compare with stored mtime

464 if record.file_mtime is None:

465 return True # No mtime stored

466

467 # Verify if file was modified (allow small floating point differences)

468 return abs(current_mtime - record.file_mtime) > 0.001

469

470 def _do_verification(

471 self, record: FileIntegrityRecord, file_path: Path, session

472 ) -> Tuple[bool, Optional[str]]:

473 """

474 Perform actual verification.

475

476 Returns:

477 Tuple of (success, reason_if_failed)

478 """

479 # Check file exists

480 if not file_path.exists():

481 return False, "file_missing"

482

483 # Get verifier

484 verifier = self._get_verifier_for_file(file_path)

485 if not verifier:

486 return False, "no_verifier"

487

488 # Calculate current checksum

489 try:

490 current_checksum = verifier.calculate_checksum(file_path)

491 except Exception as e:

492 logger.exception("[FILE_INTEGRITY] Failed to calculate checksum")

493 return False, f"checksum_calculation_failed: {str(e)}"

494

495 # Compare checksums

496 if current_checksum != record.checksum:

497 return False, "checksum_mismatch"

498

499 # Update file mtime in record

500 record.file_mtime = file_path.stat().st_mtime

501

502 return True, None

503

504 def _update_stats(

505 self, record: FileIntegrityRecord, passed: bool, session

506 ) -> None:

507 """Update verification statistics."""

508 record.total_verifications += 1

509 record.last_verified_at = datetime.now(UTC)

510 record.last_verification_passed = passed

511

512 if passed:

513 record.consecutive_successes += 1

514 record.consecutive_failures = 0

515 else:

516 record.consecutive_failures += 1

517 record.consecutive_successes = 0

518

519 def _log_failure(

520 self,

521 record: FileIntegrityRecord,

522 file_path: Path,

523 reason: str,

524 session,

525 ) -> None:

526 """Log verification failure to audit trail."""

527 # Get current checksum if possible

528 actual_checksum = None

529 file_size = None

530

531 if file_path.exists():

532 try:

533 verifier = self._get_verifier_for_file(file_path)

534 if verifier: 534 ↛ 544line 534 didn't jump to line 544 because the condition on line 534 was always true

535 actual_checksum = verifier.calculate_checksum(file_path)

536 file_size = file_path.stat().st_size

537 except Exception:

538 logger.debug(

539 "Checksum calculation failed for {}",

540 file_path,

541 exc_info=True,

542 )

543

544 failure = FileVerificationFailure(

545 file_record_id=record.id,

546 expected_checksum=record.checksum,

547 actual_checksum=actual_checksum,

548 file_size=file_size,

549 failure_reason=reason,

550 )

551 session.add(failure)

552

553 logger.warning(

554 f"[FILE_INTEGRITY] Logged failure for {file_path}: {reason}"

555 )

556

557 # Cleanup old failures for this file

558 self._cleanup_old_failures(record, session)

559

560 # Periodically check if global cleanup needed (every 100th file to avoid overhead)

561 if record.id % 100 == 0:

562 self._check_global_cleanup_needed(session)

563

564 def _cleanup_old_failures(

565 self, record: FileIntegrityRecord, session

566 ) -> None:

567 """

568 Clean up old failure records to prevent unbounded growth.

569

570 Keeps only the most recent MAX_FAILURES_PER_FILE failures per file.

571 """

572 # Count failures for this file

573 failure_count = (

574 session.query(FileVerificationFailure)

575 .filter_by(file_record_id=record.id)

576 .count()

577 )

578

579 if failure_count > self.MAX_FAILURES_PER_FILE:

580 # Delete oldest failures, keeping only the most recent MAX_FAILURES_PER_FILE

581 failures_to_delete = (

582 session.query(FileVerificationFailure)

583 .filter_by(file_record_id=record.id)

584 .order_by(FileVerificationFailure.verified_at.asc())

585 .limit(failure_count - self.MAX_FAILURES_PER_FILE)

586 .all()

587 )

588

589 for failure in failures_to_delete:

590 session.delete(failure)

591

592 logger.info(

593 f"[FILE_INTEGRITY] Cleaned up {len(failures_to_delete)} old failures for file_record {record.id}"

594 )

595

596 def _check_global_cleanup_needed(self, session) -> None:

597 """

598 Check if global cleanup is needed and run it if threshold exceeded.

599

600 Only runs cleanup if failure count exceeds MAX_TOTAL_FAILURES by 20%.

601 This prevents constant cleanup while allowing some buffer.

602 """

603 threshold = int(self.MAX_TOTAL_FAILURES * 1.2) # 20% over limit

604 total_failures = session.query(FileVerificationFailure).count()

605

606 if total_failures > threshold:

607 logger.info(

608 f"[FILE_INTEGRITY] Global failure count ({total_failures}) exceeds threshold ({threshold}), "

609 f"running cleanup..."

610 )

611

612 # Delete oldest failures to get under limit

613 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES

614

615 failures_to_delete = (

616 session.query(FileVerificationFailure)

617 .order_by(FileVerificationFailure.verified_at.asc())

618 .limit(failures_to_delete_count)

619 .all()

620 )

621

622 for failure in failures_to_delete:

623 session.delete(failure)

624

625 logger.info(

626 f"[FILE_INTEGRITY] Threshold cleanup: deleted {len(failures_to_delete)} old failures"

627 )

628

629 def cleanup_all_old_failures(self) -> int:

630 """

631 Global cleanup of failure records across all files.

632

633 Enforces MAX_TOTAL_FAILURES limit by removing oldest failures.

634

635 Returns:

636 Number of records deleted

637 """

638 with get_user_db_session(self.username, self.password) as session:

639 total_failures = session.query(FileVerificationFailure).count()

640

641 if total_failures <= self.MAX_TOTAL_FAILURES:

642 return 0

643

644 # Delete oldest failures to get under limit

645 failures_to_delete_count = total_failures - self.MAX_TOTAL_FAILURES

646

647 failures_to_delete = (

648 session.query(FileVerificationFailure)

649 .order_by(FileVerificationFailure.verified_at.asc())

650 .limit(failures_to_delete_count)

651 .all()

652 )

653

654 for failure in failures_to_delete:

655 session.delete(failure)

656

657 session.commit()

658

659 logger.info(

660 f"[FILE_INTEGRITY] Global cleanup: deleted {len(failures_to_delete)} old failures "

661 f"(total was {total_failures}, now {total_failures - len(failures_to_delete)})"

662 )

663

664 return len(failures_to_delete)

665

666 def get_total_failure_count(self) -> int:

667 """

668 Get total number of failure records across all files.

669

670 Returns:

671 Total count of failure records

672 """

673 with get_user_db_session(self.username, self.password) as session:

674 return session.query(FileVerificationFailure).count()

Coverage for src/local_deep_research/security/file_integrity/integrity_manager.py: 98%

221 statements