services_hardening_process_qualification.rs (16653B)
1 #![forbid(unsafe_code)] 2 #![cfg(any(target_os = "linux", target_os = "macos"))] 3 4 use std::{ 5 collections::BTreeSet, 6 fs, 7 io::Write as _, 8 os::unix::fs::PermissionsExt as _, 9 path::{Path, PathBuf}, 10 process::{Child, Command, Output, Stdio}, 11 thread, 12 time::{Duration, Instant}, 13 }; 14 15 use serde_json::Value; 16 use sha2::{Digest, Sha256}; 17 use sqlx::{ConnectOptions, Connection, SqliteConnection, sqlite::SqliteConnectOptions}; 18 19 const CONTRACT: &str = 20 include_str!("../contracts/services_hardening/process_qualification.v1.json"); 21 const CONFIG_EXAMPLE: &str = include_str!("../contracts/services_hardening/config.v1.example.toml"); 22 const PROCESS_DEADLINE: Duration = Duration::from_millis(30_000); 23 const POLL_INTERVAL: Duration = Duration::from_millis(2); 24 const PARALLEL_INSPECTIONS: usize = 8; 25 const SOAK_ITERATIONS: usize = 32; 26 const CRASH_FIXTURE_BYTES: i64 = 67_108_864; 27 const MAXIMUM_CAPTURED_OUTPUT_BYTES: usize = 8_192; 28 29 struct ProcessFixture { 30 root: tempfile::TempDir, 31 config: PathBuf, 32 } 33 34 impl ProcessFixture { 35 fn new() -> Self { 36 let root = tempfile::tempdir().expect("repo-local root"); 37 fs::set_permissions(root.path(), fs::Permissions::from_mode(0o700)) 38 .expect("secure repo-local root"); 39 let config = root.path().join("myc.toml"); 40 Self { root, config } 41 } 42 43 fn command(&self, command: &[&str]) -> Command { 44 let mut process = Command::new(env!("CARGO_BIN_EXE_myc")); 45 process 46 .args(["--profile", "repo-local", "--instance", "primary"]) 47 .arg("--repo-local-root") 48 .arg(self.root.path()) 49 .arg("--config") 50 .arg(&self.config) 51 .args(command) 52 .stdin(Stdio::null()) 53 .stdout(Stdio::piped()) 54 .stderr(Stdio::piped()); 55 process 56 } 57 58 fn run(&self, command: &[&str]) -> Output { 59 let child = self.command(command).spawn().expect("spawn Myc process"); 60 wait_bounded(child) 61 } 62 63 fn run_with_stdin(&self, command: &[&str], bytes: &[u8]) -> Output { 64 let mut process = self.command(command); 65 process.stdin(Stdio::piped()); 66 let mut child = process.spawn().expect("spawn Myc process"); 67 child 68 .stdin 69 .take() 70 .expect("stdin") 71 .write_all(bytes) 72 .expect("write bounded stdin"); 73 wait_bounded(child) 74 } 75 76 fn state_directory(&self) -> PathBuf { 77 self.root.path().join("data/services/myc/primary") 78 } 79 80 fn state_database(&self) -> PathBuf { 81 self.state_directory().join("state.sqlite") 82 } 83 84 fn initialize(&self) { 85 let initialized = self.run_with_stdin(&["config", "init"], repo_local_config().as_bytes()); 86 assert_success(&initialized); 87 fs::create_dir_all(self.state_directory()).expect("state directory"); 88 fs::set_permissions(self.state_directory(), fs::Permissions::from_mode(0o700)) 89 .expect("secure state directory"); 90 assert_success(&self.run(&["state", "init"])); 91 } 92 } 93 94 fn repo_local_config() -> String { 95 CONFIG_EXAMPLE 96 .replace("wss://relay-primary.example.test/", "ws://127.0.0.1:9/") 97 .replace("wss://relay-secondary.example.test/", "ws://127.0.0.1:10/") 98 .replace("connect_deadline_ms = 10000", "connect_deadline_ms = 100") 99 } 100 101 fn wait_bounded(mut child: Child) -> Output { 102 let deadline = Instant::now() + PROCESS_DEADLINE; 103 loop { 104 if child.try_wait().expect("poll Myc process").is_some() { 105 let output = child.wait_with_output().expect("collect Myc process"); 106 assert!(output.stdout.len() <= MAXIMUM_CAPTURED_OUTPUT_BYTES); 107 assert!(output.stderr.len() <= MAXIMUM_CAPTURED_OUTPUT_BYTES); 108 return output; 109 } 110 if Instant::now() >= deadline { 111 let _ = child.kill(); 112 let output = child 113 .wait_with_output() 114 .expect("reap timed-out Myc process"); 115 panic!( 116 "Myc process exceeded the qualification deadline: {:?}", 117 output.stderr 118 ); 119 } 120 thread::sleep(POLL_INTERVAL); 121 } 122 } 123 124 fn wait_for_created_member(child: &mut Child, path: &Path) { 125 let deadline = Instant::now() + PROCESS_DEADLINE; 126 loop { 127 if path 128 .metadata() 129 .map(|metadata| metadata.is_file() && metadata.len() > 0) 130 .unwrap_or(false) 131 { 132 return; 133 } 134 assert!( 135 child.try_wait().expect("poll crash target").is_none(), 136 "process exited before the crash boundary was observable" 137 ); 138 assert!( 139 Instant::now() < deadline, 140 "crash boundary was not observed before the deadline" 141 ); 142 thread::sleep(POLL_INTERVAL); 143 } 144 } 145 146 fn diagnostic_code(output: &Output) -> String { 147 serde_json::from_slice::<Value>(&output.stderr).expect("bounded diagnostic JSON")["code"] 148 .as_str() 149 .expect("diagnostic code") 150 .to_owned() 151 } 152 153 fn assert_success(output: &Output) { 154 assert_eq!(output.status.code(), Some(0), "stderr: {:?}", output.stderr); 155 assert_eq!(diagnostic_code(output), "success"); 156 } 157 158 async fn inflate_database(path: &Path) { 159 let options = SqliteConnectOptions::new() 160 .filename(path) 161 .create_if_missing(false) 162 .disable_statement_logging(); 163 let mut connection = SqliteConnection::connect_with(&options) 164 .await 165 .expect("open qualification database"); 166 sqlx::query("CREATE TABLE process_qualification_padding (payload BLOB NOT NULL)") 167 .execute(&mut connection) 168 .await 169 .expect("create temporary padding table"); 170 sqlx::query("INSERT INTO process_qualification_padding(payload) VALUES (zeroblob(?))") 171 .bind(CRASH_FIXTURE_BYTES) 172 .execute(&mut connection) 173 .await 174 .expect("write bounded padding"); 175 sqlx::query("DROP TABLE process_qualification_padding") 176 .execute(&mut connection) 177 .await 178 .expect("remove temporary padding schema"); 179 connection 180 .close() 181 .await 182 .expect("close qualification database"); 183 assert!( 184 fs::metadata(path).expect("database metadata").len() 185 >= u64::try_from(CRASH_FIXTURE_BYTES).expect("positive fixture bound") 186 ); 187 } 188 189 #[test] 190 fn qualification_contract_freezes_the_exact_process_and_component_corpus() { 191 let contract: Value = serde_json::from_str(CONTRACT).expect("qualification contract"); 192 assert_eq!( 193 contract 194 .as_object() 195 .expect("qualification object") 196 .keys() 197 .map(String::as_str) 198 .collect::<BTreeSet<_>>(), 199 BTreeSet::from([ 200 "actual_process_corpus", 201 "binary", 202 "bounds", 203 "component_corpus", 204 "contract_version", 205 "deferred", 206 "invariants", 207 "schema", 208 "service", 209 "source_lock", 210 "source_locked_shared_sqlite_corpus", 211 "step", 212 ]) 213 ); 214 assert_eq!(contract["schema"], "radroots.myc.process-qualification.v1"); 215 assert_eq!(contract["contract_version"], 1); 216 assert_eq!(contract["step"], 161); 217 assert_eq!(contract["service"], "myc"); 218 assert_eq!(contract["binary"], "myc"); 219 assert_eq!( 220 contract["source_lock"], 221 serde_json::json!({ 222 "schema": "radroots.service.source-lock.v2", 223 "lib_revision": "053d0c750bf9cd683c6ea37cefe7e79617ba629f" 224 }) 225 ); 226 assert_eq!( 227 contract["bounds"], 228 serde_json::json!({ 229 "process_deadline_ms": 30_000, 230 "poll_interval_ms": 2, 231 "parallel_inspection_processes": 8, 232 "soak_iterations": 32, 233 "crash_fixture_bytes": 67_108_864, 234 "maximum_captured_output_bytes": 8_192 235 }) 236 ); 237 assert_eq!( 238 contract["actual_process_corpus"], 239 serde_json::json!([ 240 "offline_state_and_diagnostics", 241 "parallel_inspection_saturation", 242 "bounded_reopen_soak", 243 "backup_copy_sigkill_collision", 244 "pre_marker_restore_sigkill_refusal", 245 "durable_restore_recovery", 246 "required_dependency_outage", 247 "secret_and_path_redaction" 248 ]) 249 ); 250 assert_eq!( 251 contract["component_corpus"], 252 serde_json::json!({ 253 "myc_atomicity": [ 254 "request_admission_is_atomic_idempotent_conflict_aware_and_restart_stable", 255 "delivery_jobs_are_config_bound_idempotent_restart_safe_and_unknown_aware", 256 "backup_integrity_and_offline_restore_obey_one_exact_myc_authority" 257 ], 258 "backlog_and_saturation": [ 259 "concurrent_identical_admission_creates_one_request_and_bounded_replay_evidence", 260 "configured_denial_precedes_saturated_unknown_client_rate_windows", 261 "challenge_creation_and_authorization_use_distinct_durable_rate_budgets" 262 ], 263 "recovery_and_crash": [ 264 "restart_recovery_is_bounded_jittered_and_idempotent", 265 "terminal_unknown_is_not_relabelled_as_failure_and_sql_guards_preserve_evidence", 266 "early_success_and_panic_are_fatal_and_join_their_cancelled_peer" 267 ], 268 "property_and_adversarial": [ 269 "duplicate_replay_conflict_and_distinct_relations_are_total", 270 "event_object_is_closed_duplicate_free_nonnull_and_exactly_consumed", 271 "exact_open_rejects_migration_history_drift_without_repair" 272 ] 273 }) 274 ); 275 assert_eq!( 276 contract["source_locked_shared_sqlite_corpus"], 277 serde_json::json!([ 278 "every_initialization_durability_edge_fails_once_and_rolls_back", 279 "transaction_durability_edges_preserve_exact_commit_semantics", 280 "backup_durability_edges_fail_once_clean_exact_stage_and_recover", 281 "close_durability_edges_are_once_only_retryable_or_terminal", 282 "every_marker_and_restore_durability_edge_is_wired_once", 283 "sigkill_restore_boundaries_recover_exact_topologies_and_preserve_permissions" 284 ]) 285 ); 286 assert_eq!( 287 contract["invariants"], 288 serde_json::json!({ 289 "actual_executable_required": true, 290 "production_failpoint_surface": false, 291 "test_environment_selector": false, 292 "detached_test_worker": false, 293 "live_database_unchanged_by_failed_backup": true, 294 "orphan_restore_evidence_fails_closed": true, 295 "durable_restore_evidence_reconciled_on_writable_reopen": true, 296 "captured_output_bounded": true, 297 "diagnostics_path_secret_free": true 298 }) 299 ); 300 assert_eq!( 301 contract["deferred"], 302 serde_json::json!([ 303 "rcld_promotion", 304 "parent_pin_alignment", 305 "nix", 306 "oci", 307 "signing", 308 "publication", 309 "deployment" 310 ]) 311 ); 312 } 313 314 #[test] 315 fn actual_process_is_bounded_under_parallel_inspection_soak_and_outage() { 316 let fixture = ProcessFixture::new(); 317 fixture.initialize(); 318 319 let children = (0..PARALLEL_INSPECTIONS) 320 .map(|_| { 321 fixture 322 .command(&["state", "status"]) 323 .spawn() 324 .expect("inspection process") 325 }) 326 .collect::<Vec<_>>(); 327 for child in children { 328 assert_success(&wait_bounded(child)); 329 } 330 331 for iteration in 0..SOAK_ITERATIONS { 332 let command = if iteration % 2 == 0 { 333 &["state", "status"][..] 334 } else { 335 &["state", "verify"][..] 336 }; 337 assert_success(&fixture.run(command)); 338 } 339 340 let outage = fixture.run(&["run"]); 341 assert_eq!(outage.status.code(), Some(3)); 342 assert!(outage.stdout.is_empty()); 343 assert_eq!( 344 diagnostic_code(&outage), 345 "service_or_dependency_unavailable" 346 ); 347 let diagnostic = String::from_utf8(outage.stderr).expect("diagnostic UTF-8"); 348 assert!(!diagnostic.contains(fixture.root.path().to_str().expect("UTF-8 root"))); 349 assert!(!diagnostic.contains("relay-primary")); 350 } 351 352 #[tokio::test] 353 async fn actual_process_sigkill_boundaries_fail_closed_and_recover_exactly() { 354 let fixture = ProcessFixture::new(); 355 fixture.initialize(); 356 inflate_database(&fixture.state_database()).await; 357 assert_success(&fixture.run(&["state", "verify"])); 358 359 let interrupted_bundle = fixture.root.path().join("interrupted-backup"); 360 let mut backup = fixture 361 .command(&["state", "backup", "--operation-id", "crash-backup-01"]) 362 .arg("--target") 363 .arg(&interrupted_bundle) 364 .args(["--expected-generation", "1", "--confirm"]) 365 .spawn() 366 .expect("spawn backup process"); 367 wait_for_created_member(&mut backup, &interrupted_bundle.join("state.sqlite")); 368 backup.kill().expect("SIGKILL backup process"); 369 let backup_output = backup.wait_with_output().expect("reap backup process"); 370 assert!(!backup_output.status.success()); 371 assert_success(&fixture.run(&["state", "verify"])); 372 assert!(interrupted_bundle.exists(), "crash collision evidence"); 373 fs::remove_dir_all(&interrupted_bundle).expect("remove owned crash fixture"); 374 375 let bundle = fixture.root.path().join("backup"); 376 let backup = fixture 377 .command(&["state", "backup", "--operation-id", "crash-backup-02"]) 378 .arg("--target") 379 .arg(&bundle) 380 .args(["--expected-generation", "1", "--confirm"]) 381 .spawn() 382 .expect("spawn complete backup"); 383 let backup_output = wait_bounded(backup); 384 assert_success(&backup_output); 385 let manifest_bytes = backup_output.stdout; 386 let manifest_digest = hex::encode(Sha256::digest(&manifest_bytes)); 387 let manifest_path = fixture.root.path().join("manifest.json"); 388 fs::write(&manifest_path, &manifest_bytes).expect("manifest file"); 389 fs::set_permissions(&manifest_path, fs::Permissions::from_mode(0o600)) 390 .expect("secure manifest"); 391 392 let staged_path = fixture 393 .state_directory() 394 .join("state.restore-staged.sqlite"); 395 let mut restore = fixture 396 .command(&["state", "restore"]) 397 .arg("--manifest") 398 .arg(&manifest_path) 399 .arg("--manifest-sha256") 400 .arg(&manifest_digest) 401 .arg("--bundle") 402 .arg(&bundle) 403 .args(["--maximum-state-bytes", "134217728", "--confirm"]) 404 .spawn() 405 .expect("spawn restore process"); 406 wait_for_created_member(&mut restore, &staged_path); 407 restore.kill().expect("SIGKILL restore process"); 408 let restore_output = restore.wait_with_output().expect("reap restore process"); 409 assert!(!restore_output.status.success()); 410 assert!(staged_path.exists(), "orphan stage is retained as evidence"); 411 412 let refused = fixture.run(&["state", "verify"]); 413 assert_eq!(refused.status.code(), Some(4)); 414 assert_eq!(diagnostic_code(&refused), "state_or_identity_unavailable"); 415 fs::remove_file(&staged_path).expect("remove owned orphan stage fixture"); 416 assert_success(&fixture.run(&["state", "verify"])); 417 418 let mut corrupted = fs::OpenOptions::new() 419 .write(true) 420 .truncate(true) 421 .open(fixture.state_database()) 422 .expect("open live database"); 423 corrupted 424 .write_all(b"corrupt-live-database") 425 .and_then(|()| corrupted.sync_all()) 426 .expect("persist corruption fixture"); 427 drop(corrupted); 428 429 let restored = fixture 430 .command(&["state", "restore"]) 431 .arg("--manifest") 432 .arg(&manifest_path) 433 .arg("--manifest-sha256") 434 .arg(&manifest_digest) 435 .arg("--bundle") 436 .arg(&bundle) 437 .args(["--maximum-state-bytes", "134217728", "--confirm"]) 438 .spawn() 439 .expect("spawn final restore"); 440 assert_success(&wait_bounded(restored)); 441 assert!( 442 fixture 443 .state_directory() 444 .join("state.restore-marker.v1") 445 .exists() 446 ); 447 assert!( 448 fixture 449 .state_directory() 450 .join("state.restore-backup.sqlite") 451 .exists() 452 ); 453 assert_success(&fixture.run(&["state", "verify"])); 454 for artifact in [ 455 "state.restore-marker.v1", 456 "state.restore-marker.v1.next", 457 "state.restore-staged.sqlite", 458 "state.restore-backup.sqlite", 459 ] { 460 assert!(!fixture.state_directory().join(artifact).exists()); 461 } 462 }