inspection.rs (19810B)
1 //! Explicit, bounded integrity inspection over one governed SQLite snapshot. 2 3 use serde::Serialize; 4 5 use crate::StorageIntegrity; 6 7 /// Caller-injected wall-clock time for one completed integrity inspection. 8 #[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize)] 9 #[serde(transparent)] 10 pub struct IntegrityCheckedAtUnixMs(u64); 11 12 impl IntegrityCheckedAtUnixMs { 13 /// Constructs a positive timestamp that SQLite can represent exactly. 14 #[must_use] 15 pub const fn new(value: u64) -> Option<Self> { 16 if value > 0 && value <= i64::MAX as u64 { 17 Some(Self(value)) 18 } else { 19 None 20 } 21 } 22 23 /// Returns the validated Unix timestamp in milliseconds. 24 #[must_use] 25 pub const fn get(self) -> u64 { 26 self.0 27 } 28 } 29 30 /// Closed result of one completed bounded database check. 31 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] 32 #[serde(rename_all = "snake_case")] 33 pub enum IntegrityCheckOutcome { 34 Verified, 35 Failed, 36 } 37 38 /// Stable, content-free diagnostic code for a completed failed check. 39 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] 40 #[serde(rename_all = "snake_case")] 41 pub enum IntegrityDiagnosticCode { 42 SqliteIntegrityFailed, 43 ForeignKeyViolation, 44 } 45 46 /// Safe bounded result of an explicit host integrity inspection. 47 #[derive(Clone, Debug, PartialEq, Eq, Serialize)] 48 pub struct ServiceSqliteIntegrityReport { 49 checked_at_unix_ms: IntegrityCheckedAtUnixMs, 50 sqlite: IntegrityCheckOutcome, 51 foreign_keys: IntegrityCheckOutcome, 52 diagnostics: Box<[IntegrityDiagnosticCode]>, 53 } 54 55 impl ServiceSqliteIntegrityReport { 56 #[cfg(any(test, target_os = "linux", target_os = "macos"))] 57 pub(crate) fn new( 58 checked_at_unix_ms: IntegrityCheckedAtUnixMs, 59 sqlite: IntegrityCheckOutcome, 60 foreign_keys: IntegrityCheckOutcome, 61 ) -> Self { 62 let diagnostics: Box<[IntegrityDiagnosticCode]> = match (sqlite, foreign_keys) { 63 (IntegrityCheckOutcome::Verified, IntegrityCheckOutcome::Verified) => Box::new([]), 64 (IntegrityCheckOutcome::Failed, IntegrityCheckOutcome::Verified) => { 65 Box::new([IntegrityDiagnosticCode::SqliteIntegrityFailed]) 66 } 67 (IntegrityCheckOutcome::Verified, IntegrityCheckOutcome::Failed) => { 68 Box::new([IntegrityDiagnosticCode::ForeignKeyViolation]) 69 } 70 (IntegrityCheckOutcome::Failed, IntegrityCheckOutcome::Failed) => Box::new([ 71 IntegrityDiagnosticCode::SqliteIntegrityFailed, 72 IntegrityDiagnosticCode::ForeignKeyViolation, 73 ]), 74 }; 75 Self { 76 checked_at_unix_ms, 77 sqlite, 78 foreign_keys, 79 diagnostics, 80 } 81 } 82 83 /// Returns the caller-injected completion time. 84 #[must_use] 85 pub const fn checked_at_unix_ms(&self) -> IntegrityCheckedAtUnixMs { 86 self.checked_at_unix_ms 87 } 88 89 /// Returns the completed SQLite integrity-check outcome. 90 #[must_use] 91 pub const fn sqlite(&self) -> IntegrityCheckOutcome { 92 self.sqlite 93 } 94 95 /// Returns the completed foreign-key-check outcome. 96 #[must_use] 97 pub const fn foreign_keys(&self) -> IntegrityCheckOutcome { 98 self.foreign_keys 99 } 100 101 /// Returns zero to two stable diagnostic codes in canonical order. 102 #[must_use] 103 pub fn diagnostics(&self) -> &[IntegrityDiagnosticCode] { 104 &self.diagnostics 105 } 106 107 /// Projects this active result into the passive storage-status vocabulary. 108 #[must_use] 109 pub const fn storage_integrity(&self) -> StorageIntegrity { 110 if matches!(self.sqlite, IntegrityCheckOutcome::Verified) 111 && matches!(self.foreign_keys, IntegrityCheckOutcome::Verified) 112 { 113 StorageIntegrity::Verified 114 } else { 115 StorageIntegrity::Failed 116 } 117 } 118 } 119 120 #[cfg(any(target_os = "linux", target_os = "macos"))] 121 mod native { 122 use sqlx::{Connection, SqliteConnection}; 123 124 use super::{IntegrityCheckOutcome, IntegrityCheckedAtUnixMs, ServiceSqliteIntegrityReport}; 125 use crate::{ServiceSqliteError, ServiceSqliteErrorKind}; 126 127 const FOREIGN_KEY_SQL: &str = "SELECT 1 FROM pragma_foreign_key_check LIMIT 1"; 128 129 pub(crate) async fn inspect_database_integrity( 130 connection: &mut SqliteConnection, 131 checked_at: IntegrityCheckedAtUnixMs, 132 mut validate: impl FnMut() -> Result<(), ServiceSqliteError>, 133 ) -> Result<ServiceSqliteIntegrityReport, ServiceSqliteError> { 134 validate()?; 135 let transaction = connection.begin().await; 136 validate()?; 137 let mut transaction = transaction.map_err(|_| integrity_error())?; 138 139 #[cfg(test)] 140 if super::test_seam::real_sqlite_probe_enabled() { 141 super::test_seam::observe(super::test_seam::PHASE_SQLITE_EXECUTION_AWAITING); 142 let probe = sqlx::query_scalar::<_, i64>( 143 "WITH RECURSIVE counter(value) AS ( 144 VALUES(0) UNION ALL SELECT value + 1 FROM counter WHERE value < 5000000 145 ) SELECT sum(value) FROM counter", 146 ) 147 .fetch_one(&mut *transaction) 148 .await; 149 validate()?; 150 if probe.is_err() { 151 return rollback_error(transaction, &mut validate).await; 152 } 153 } 154 155 #[cfg(test)] 156 super::test_seam::pause(super::test_seam::PHASE_BEFORE_SQLITE).await; 157 let sqlite_rows = sqlx::query(crate::persisted_value::INTEGRITY_CHECK_SQL) 158 .fetch_all(&mut *transaction) 159 .await; 160 validate()?; 161 let sqlite = match sqlite_rows { 162 Ok(rows) if rows.len() == 1 => { 163 match crate::persisted_value::integrity_result_failed(&rows[0]) { 164 Some(failed) => classify_integrity_failure(failed), 165 None => return rollback_error(transaction, &mut validate).await, 166 } 167 } 168 Ok(_) | Err(_) => return rollback_error(transaction, &mut validate).await, 169 }; 170 171 #[cfg(test)] 172 super::test_seam::pause(super::test_seam::PHASE_BEFORE_FOREIGN_KEYS).await; 173 let foreign_key_row = sqlx::query_scalar::<_, i64>(FOREIGN_KEY_SQL) 174 .fetch_optional(&mut *transaction) 175 .await; 176 validate()?; 177 let foreign_keys = match foreign_key_row { 178 Ok(None) => IntegrityCheckOutcome::Verified, 179 Ok(Some(1)) => IntegrityCheckOutcome::Failed, 180 Ok(Some(_)) | Err(_) => return rollback_error(transaction, &mut validate).await, 181 }; 182 183 #[cfg(test)] 184 super::test_seam::pause(super::test_seam::PHASE_BEFORE_ROLLBACK).await; 185 let rollback = transaction.rollback().await; 186 validate()?; 187 rollback.map_err(|_| integrity_error())?; 188 Ok(ServiceSqliteIntegrityReport::new( 189 checked_at, 190 sqlite, 191 foreign_keys, 192 )) 193 } 194 195 async fn rollback_error( 196 transaction: sqlx::Transaction<'_, sqlx::Sqlite>, 197 validate: &mut impl FnMut() -> Result<(), ServiceSqliteError>, 198 ) -> Result<ServiceSqliteIntegrityReport, ServiceSqliteError> { 199 let rollback = transaction.rollback().await; 200 validate()?; 201 rollback.map_err(|_| integrity_error())?; 202 Err(integrity_error()) 203 } 204 205 fn integrity_error() -> ServiceSqliteError { 206 ServiceSqliteError::new(ServiceSqliteErrorKind::Integrity) 207 } 208 209 fn classify_integrity_failure(failed: bool) -> IntegrityCheckOutcome { 210 if failed { 211 IntegrityCheckOutcome::Failed 212 } else { 213 IntegrityCheckOutcome::Verified 214 } 215 } 216 217 #[cfg(test)] 218 pub(super) fn classify_test_value(value: &str) -> IntegrityCheckOutcome { 219 classify_integrity_failure(value != "ok") 220 } 221 } 222 223 #[cfg(any(target_os = "linux", target_os = "macos"))] 224 pub(crate) use native::inspect_database_integrity; 225 226 #[cfg(all(test, any(target_os = "linux", target_os = "macos")))] 227 pub(crate) mod test_seam { 228 use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; 229 230 pub(crate) const PHASE_BEFORE_SQLITE: u8 = 1; 231 pub(crate) const PHASE_BEFORE_FOREIGN_KEYS: u8 = 2; 232 pub(crate) const PHASE_BEFORE_ROLLBACK: u8 = 3; 233 pub(crate) const PHASE_SQLITE_EXECUTION_AWAITING: u8 = 4; 234 pub(crate) const PHASE_CONNECTION_CLOSE_AWAITING: u8 = 5; 235 236 static BLOCKED: AtomicU8 = AtomicU8::new(0); 237 static REACHED: AtomicU8 = AtomicU8::new(0); 238 static RELEASED: AtomicBool = AtomicBool::new(true); 239 static REAL_SQLITE_PROBE: AtomicBool = AtomicBool::new(false); 240 static CONNECTION_CLOSE_FAILURE: AtomicBool = AtomicBool::new(false); 241 pub(crate) static LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); 242 243 pub(crate) async fn pause(phase: u8) { 244 REACHED.store(phase, Ordering::Release); 245 while BLOCKED.load(Ordering::Acquire) == phase && !RELEASED.load(Ordering::Acquire) { 246 tokio::task::yield_now().await; 247 } 248 } 249 250 pub(crate) fn block(phase: u8) { 251 REACHED.store(0, Ordering::Release); 252 BLOCKED.store(phase, Ordering::Release); 253 RELEASED.store(false, Ordering::Release); 254 } 255 256 pub(crate) fn observe(phase: u8) { 257 REACHED.store(phase, Ordering::Release); 258 } 259 260 pub(crate) fn enable_real_sqlite_probe(enabled: bool) { 261 REACHED.store(0, Ordering::Release); 262 REAL_SQLITE_PROBE.store(enabled, Ordering::Release); 263 } 264 265 pub(crate) fn real_sqlite_probe_enabled() -> bool { 266 REAL_SQLITE_PROBE.load(Ordering::Acquire) 267 } 268 269 pub(crate) fn inject_connection_close_failure(enabled: bool) { 270 CONNECTION_CLOSE_FAILURE.store(enabled, Ordering::Release); 271 } 272 273 pub(crate) fn take_connection_close_failure() -> bool { 274 CONNECTION_CLOSE_FAILURE.swap(false, Ordering::AcqRel) 275 } 276 277 pub(crate) fn reached() -> u8 { 278 REACHED.load(Ordering::Acquire) 279 } 280 281 pub(crate) fn release() { 282 RELEASED.store(true, Ordering::Release); 283 BLOCKED.store(0, Ordering::Release); 284 } 285 } 286 287 #[cfg(test)] 288 mod tests { 289 use super::*; 290 291 #[cfg(any(target_os = "linux", target_os = "macos"))] 292 use std::{ 293 fs::OpenOptions, 294 io::{Seek, SeekFrom, Write}, 295 }; 296 297 #[cfg(any(target_os = "linux", target_os = "macos"))] 298 use sqlx::{Connection, SqliteConnection, sqlite::SqliteConnectOptions}; 299 300 #[test] 301 fn timestamp_bounds_and_report_wire_vocabulary_are_exact() { 302 assert!(IntegrityCheckedAtUnixMs::new(0).is_none()); 303 let maximum = IntegrityCheckedAtUnixMs::new(i64::MAX as u64).expect("maximum timestamp"); 304 assert_eq!(maximum.get(), i64::MAX as u64); 305 assert!(IntegrityCheckedAtUnixMs::new(i64::MAX as u64 + 1).is_none()); 306 307 let checked_at = IntegrityCheckedAtUnixMs::new(1_700_000_000_000).unwrap(); 308 let verified = ServiceSqliteIntegrityReport::new( 309 checked_at, 310 IntegrityCheckOutcome::Verified, 311 IntegrityCheckOutcome::Verified, 312 ); 313 assert!(verified.diagnostics().is_empty()); 314 assert_eq!(verified.storage_integrity(), StorageIntegrity::Verified); 315 assert_eq!( 316 serde_json::to_string(&verified).unwrap(), 317 r#"{"checked_at_unix_ms":1700000000000,"sqlite":"verified","foreign_keys":"verified","diagnostics":[]}"# 318 ); 319 320 let failed = ServiceSqliteIntegrityReport::new( 321 checked_at, 322 IntegrityCheckOutcome::Failed, 323 IntegrityCheckOutcome::Failed, 324 ); 325 assert_eq!( 326 failed.diagnostics(), 327 [ 328 IntegrityDiagnosticCode::SqliteIntegrityFailed, 329 IntegrityDiagnosticCode::ForeignKeyViolation, 330 ] 331 ); 332 assert_eq!(failed.storage_integrity(), StorageIntegrity::Failed); 333 assert_eq!( 334 serde_json::to_string(&failed).unwrap(), 335 r#"{"checked_at_unix_ms":1700000000000,"sqlite":"failed","foreign_keys":"failed","diagnostics":["sqlite_integrity_failed","foreign_key_violation"]}"# 336 ); 337 assert!(!format!("{failed:?}").contains("sqlite_schema")); 338 #[cfg(any(target_os = "linux", target_os = "macos"))] 339 assert_eq!( 340 native::classify_test_value( 341 "a completed SQLite diagnostic that is intentionally much longer than sixty-four bytes" 342 ), 343 IntegrityCheckOutcome::Failed 344 ); 345 } 346 347 #[cfg(any(target_os = "linux", target_os = "macos"))] 348 async fn in_memory_database() -> SqliteConnection { 349 SqliteConnection::connect_with(&SqliteConnectOptions::new().filename(":memory:")) 350 .await 351 .expect("in-memory database") 352 } 353 354 #[cfg(any(target_os = "linux", target_os = "macos"))] 355 #[tokio::test(flavor = "current_thread")] 356 async fn native_inspection_reports_healthy_and_foreign_key_failure() { 357 let _serial = test_seam::LOCK.lock().await; 358 test_seam::release(); 359 let checked_at = IntegrityCheckedAtUnixMs::new(1).unwrap(); 360 let mut healthy = in_memory_database().await; 361 let healthy = inspect_database_integrity(&mut healthy, checked_at, || Ok(())) 362 .await 363 .expect("healthy inspection"); 364 assert_eq!(healthy.sqlite(), IntegrityCheckOutcome::Verified); 365 assert_eq!(healthy.foreign_keys(), IntegrityCheckOutcome::Verified); 366 assert!(healthy.diagnostics().is_empty()); 367 368 let mut foreign_keys = in_memory_database().await; 369 sqlx::raw_sql( 370 "PRAGMA foreign_keys=OFF; 371 CREATE TABLE parent (id INTEGER PRIMARY KEY) STRICT; 372 CREATE TABLE child (parent_id INTEGER REFERENCES parent(id)) STRICT; 373 INSERT INTO child(parent_id) VALUES (99);", 374 ) 375 .execute(&mut foreign_keys) 376 .await 377 .expect("seed foreign-key violation"); 378 let failed = inspect_database_integrity(&mut foreign_keys, checked_at, || Ok(())) 379 .await 380 .expect("completed foreign-key inspection"); 381 assert_eq!(failed.sqlite(), IntegrityCheckOutcome::Verified); 382 assert_eq!(failed.foreign_keys(), IntegrityCheckOutcome::Failed); 383 assert_eq!( 384 failed.diagnostics(), 385 [IntegrityDiagnosticCode::ForeignKeyViolation] 386 ); 387 } 388 389 #[cfg(any(target_os = "linux", target_os = "macos"))] 390 #[tokio::test(flavor = "current_thread")] 391 async fn native_inspection_keeps_completed_failure_diagnostics_bounded() { 392 let _serial = test_seam::LOCK.lock().await; 393 test_seam::release(); 394 let mut corrupt = in_memory_database().await; 395 sqlx::raw_sql( 396 "PRAGMA foreign_keys=OFF; 397 CREATE TABLE parent (id INTEGER PRIMARY KEY) STRICT; 398 CREATE TABLE child (parent_id INTEGER REFERENCES parent(id)) STRICT; 399 CREATE INDEX parent_index ON parent(id); 400 INSERT INTO child(parent_id) VALUES (99); 401 PRAGMA writable_schema=ON; 402 UPDATE sqlite_schema SET rootpage=0 WHERE name='parent_index'; 403 PRAGMA writable_schema=OFF; 404 PRAGMA schema_version=99;", 405 ) 406 .execute(&mut corrupt) 407 .await 408 .expect("seed bounded corruption"); 409 let report = inspect_database_integrity( 410 &mut corrupt, 411 IntegrityCheckedAtUnixMs::new(2).unwrap(), 412 || Ok(()), 413 ) 414 .await 415 .expect("completed corruption inspection"); 416 assert_eq!(report.sqlite(), IntegrityCheckOutcome::Failed); 417 assert_eq!(report.foreign_keys(), IntegrityCheckOutcome::Failed); 418 assert_eq!(report.diagnostics().len(), 2); 419 let rendered = format!("{report:?}"); 420 assert!(!rendered.contains("parent")); 421 assert!(!rendered.contains("rootpage")); 422 } 423 424 #[cfg(any(target_os = "linux", target_os = "macos"))] 425 #[tokio::test(flavor = "current_thread")] 426 async fn physical_corruption_and_query_failure_remain_redacted_and_typed() { 427 let _serial = test_seam::LOCK.lock().await; 428 test_seam::release(); 429 let directory = tempfile::tempdir().expect("temporary database directory"); 430 let path = directory.path().join("sensitive-state-name.sqlite"); 431 let options = SqliteConnectOptions::new() 432 .filename(&path) 433 .create_if_missing(true); 434 let mut connection = SqliteConnection::connect_with(&options) 435 .await 436 .expect("create database"); 437 sqlx::query("CREATE TABLE integrity_probe (value BLOB NOT NULL) STRICT") 438 .execute(&mut connection) 439 .await 440 .expect("create probe table"); 441 sqlx::query("INSERT INTO integrity_probe(value) VALUES (zeroblob(4096))") 442 .execute(&mut connection) 443 .await 444 .expect("allocate probe page"); 445 let page_size = sqlx::query_scalar::<_, i64>("PRAGMA page_size") 446 .fetch_one(&mut connection) 447 .await 448 .expect("page size"); 449 let root_page = sqlx::query_scalar::<_, i64>( 450 "SELECT rootpage FROM sqlite_schema WHERE name='integrity_probe'", 451 ) 452 .fetch_one(&mut connection) 453 .await 454 .expect("probe root page"); 455 connection.close().await.expect("close database"); 456 let offset = u64::try_from(root_page - 1) 457 .ok() 458 .and_then(|page| page.checked_mul(u64::try_from(page_size).ok()?)) 459 .expect("corrupt page offset"); 460 let mut file = OpenOptions::new() 461 .write(true) 462 .open(&path) 463 .expect("open database bytes"); 464 file.seek(SeekFrom::Start(offset)).expect("seek root page"); 465 file.write_all(&[0xff]).expect("corrupt page type"); 466 file.sync_all().expect("sync corrupt database"); 467 drop(file); 468 469 let options = SqliteConnectOptions::new() 470 .filename(&path) 471 .create_if_missing(false); 472 let mut corrupt = SqliteConnection::connect_with(&options) 473 .await 474 .expect("open corrupt database shell"); 475 let report = inspect_database_integrity( 476 &mut corrupt, 477 IntegrityCheckedAtUnixMs::new(3).unwrap(), 478 || Ok(()), 479 ) 480 .await 481 .expect("completed physical-corruption result"); 482 assert_eq!(report.sqlite(), IntegrityCheckOutcome::Failed); 483 assert_eq!( 484 report.diagnostics(), 485 [IntegrityDiagnosticCode::SqliteIntegrityFailed] 486 ); 487 let rendered = format!("{report:?}"); 488 assert!(!rendered.contains("sensitive-state-name")); 489 assert!(!rendered.contains("integrity_probe")); 490 assert!(!rendered.contains("database disk image")); 491 492 let mut malformed = in_memory_database().await; 493 sqlx::raw_sql( 494 "CREATE TABLE secret_schema_name (value INTEGER) STRICT; 495 PRAGMA writable_schema=ON; 496 UPDATE sqlite_schema SET sql='CREATE TABLE secret_schema_name(' 497 WHERE name='secret_schema_name'; 498 PRAGMA writable_schema=OFF; 499 PRAGMA schema_version=99;", 500 ) 501 .execute(&mut malformed) 502 .await 503 .expect("seed malformed schema"); 504 let error = inspect_database_integrity( 505 &mut malformed, 506 IntegrityCheckedAtUnixMs::new(4).unwrap(), 507 || Ok(()), 508 ) 509 .await 510 .expect_err("query failure is not a completed report"); 511 assert_eq!(error.kind(), crate::ServiceSqliteErrorKind::Integrity); 512 let rendered = format!("{error:?} {}", error); 513 assert!(!rendered.contains("secret_schema_name")); 514 assert!(!rendered.contains("incomplete input")); 515 } 516 }