services_hardening_process.rs (25233B)
1 #![forbid(unsafe_code)] 2 #![cfg(any(target_os = "linux", target_os = "macos"))] 3 4 use std::{ 5 collections::BTreeSet, 6 fs, 7 io::{Read as _, Write as _}, 8 net::{TcpListener, TcpStream}, 9 os::unix::fs::PermissionsExt as _, 10 path::PathBuf, 11 process::{Child, Command, ExitStatus, Output, Stdio}, 12 sync::{ 13 Arc, 14 atomic::{AtomicBool, Ordering}, 15 }, 16 thread, 17 time::{Duration, Instant}, 18 }; 19 20 use nostr::{Keys, SecretKey}; 21 use rhi::{ 22 RadrootsHostEnvironment, RadrootsPathResolver, RadrootsPlatform, parse_rhi_cli_v1_from, 23 resolve_rhi_runtime_context, 24 }; 25 use sha2::{Digest as _, Sha256}; 26 use tungstenite::{Error as WebSocketError, Message, accept}; 27 28 const CONFIG_EXAMPLE: &str = include_str!("../contracts/services_hardening/config.v1.example.toml"); 29 const PROCESS_QUALIFICATION_CONTRACT: &str = 30 include_str!("../contracts/services_hardening/process_qualification.v1.json"); 31 const FAILURE_QUALIFICATION_CONTRACT: &[u8] = 32 include_bytes!("../contracts/services_hardening/failure_qualification.v1.json"); 33 const SOURCE_LOCK: &str = include_str!("../radroots.service.source-lock.v3.toml"); 34 const PROCESS_DEADLINE: Duration = Duration::from_secs(30); 35 const POLL_INTERVAL: Duration = Duration::from_millis(2); 36 const CONNECT_DEADLINE_MILLISECONDS: u64 = 5_000; 37 const RELAY_HANDSHAKE_TIMEOUT: Duration = Duration::from_millis(5_000); 38 const RELAY_IO_TIMEOUT: Duration = Duration::from_millis(100); 39 const PARALLEL_INSPECTIONS: usize = 8; 40 const SOAK_ITERATIONS: usize = 32; 41 const MAXIMUM_STDOUT_BYTES: usize = 1_048_576; 42 const MAXIMUM_STDERR_BYTES: usize = 8_192; 43 #[cfg(target_os = "linux")] 44 const PROCESS_TEMPORARY_ROOT: &str = "/tmp"; 45 #[cfg(target_os = "macos")] 46 const PROCESS_TEMPORARY_ROOT: &str = "/private/tmp"; 47 48 struct RelayHarness { 49 address: std::net::SocketAddr, 50 stop: Arc<AtomicBool>, 51 thread: Option<thread::JoinHandle<()>>, 52 } 53 54 impl RelayHarness { 55 fn start() -> Self { 56 let listener = TcpListener::bind("127.0.0.1:0").expect("test relay listener"); 57 listener 58 .set_nonblocking(true) 59 .expect("nonblocking test relay"); 60 let address = listener.local_addr().expect("test relay address"); 61 let stop = Arc::new(AtomicBool::new(false)); 62 let thread_stop = Arc::clone(&stop); 63 let thread = thread::spawn(move || { 64 let mut sessions = Vec::new(); 65 while !thread_stop.load(Ordering::SeqCst) { 66 match listener.accept() { 67 Ok((stream, _)) => { 68 let session_stop = Arc::clone(&thread_stop); 69 sessions.push(thread::spawn(move || relay_session(stream, &session_stop))); 70 } 71 Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => { 72 thread::sleep(Duration::from_millis(2)); 73 } 74 Err(error) => panic!("test relay accept failed: {error}"), 75 } 76 } 77 for session in sessions { 78 session.join().expect("test relay session"); 79 } 80 }); 81 Self { 82 address, 83 stop, 84 thread: Some(thread), 85 } 86 } 87 88 fn url(&self, path: &str) -> String { 89 format!("ws://{}/{path}", self.address) 90 } 91 } 92 93 impl Drop for RelayHarness { 94 fn drop(&mut self) { 95 self.stop.store(true, Ordering::SeqCst); 96 let _ = TcpStream::connect(self.address); 97 if let Some(thread) = self.thread.take() { 98 thread.join().expect("test relay"); 99 } 100 } 101 } 102 103 fn relay_session(stream: TcpStream, stop: &AtomicBool) { 104 stream 105 .set_read_timeout(Some(RELAY_HANDSHAKE_TIMEOUT)) 106 .expect("relay handshake read timeout"); 107 stream 108 .set_write_timeout(Some(RELAY_HANDSHAKE_TIMEOUT)) 109 .expect("relay handshake write timeout"); 110 let mut websocket = match accept(stream) { 111 Ok(websocket) => websocket, 112 Err(_) => return, 113 }; 114 websocket 115 .get_mut() 116 .set_read_timeout(Some(RELAY_IO_TIMEOUT)) 117 .expect("relay read timeout"); 118 websocket 119 .get_mut() 120 .set_write_timeout(Some(RELAY_IO_TIMEOUT)) 121 .expect("relay write timeout"); 122 while !stop.load(Ordering::SeqCst) { 123 let message = match websocket.read() { 124 Ok(message) => message, 125 Err(WebSocketError::Io(error)) 126 if matches!( 127 error.kind(), 128 std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut 129 ) => 130 { 131 continue; 132 } 133 Err(WebSocketError::ConnectionClosed | WebSocketError::AlreadyClosed) => break, 134 Err(_) => break, 135 }; 136 match message { 137 Message::Text(text) => relay_text(&mut websocket, text.as_str()), 138 Message::Ping(bytes) => { 139 if websocket.send(Message::Pong(bytes)).is_err() { 140 break; 141 } 142 } 143 Message::Close(_) => break, 144 _ => {} 145 } 146 } 147 } 148 149 fn relay_text(websocket: &mut tungstenite::WebSocket<TcpStream>, text: &str) { 150 let Ok(message) = serde_json::from_str::<serde_json::Value>(text) else { 151 return; 152 }; 153 let Some(parts) = message.as_array() else { 154 return; 155 }; 156 match parts.first().and_then(serde_json::Value::as_str) { 157 Some("REQ") => { 158 let Some(subscription) = parts.get(1).and_then(serde_json::Value::as_str) else { 159 return; 160 }; 161 let response = serde_json::json!(["EOSE", subscription]).to_string(); 162 let _ = websocket.send(Message::Text(response.into())); 163 } 164 Some("EVENT") => { 165 let Some(event_id) = parts 166 .get(1) 167 .and_then(|event| event.get("id")) 168 .and_then(serde_json::Value::as_str) 169 else { 170 return; 171 }; 172 let response = serde_json::json!(["OK", event_id, true, ""]).to_string(); 173 let _ = websocket.send(Message::Text(response.into())); 174 } 175 _ => {} 176 } 177 } 178 179 struct ProcessFixture { 180 root: tempfile::TempDir, 181 config: PathBuf, 182 runtime: rhi::RhiRuntimeContext, 183 } 184 185 impl ProcessFixture { 186 fn new() -> Self { 187 let root = tempfile::Builder::new() 188 .prefix("rhi-") 189 .tempdir_in(PROCESS_TEMPORARY_ROOT) 190 .expect("short repo-local root"); 191 fs::set_permissions(root.path(), fs::Permissions::from_mode(0o700)) 192 .expect("secure repo-local root"); 193 let config = root.path().join("rhi.toml"); 194 let invocation = parse_rhi_cli_v1_from([ 195 "rhi", 196 "--profile", 197 "repo-local", 198 "--instance", 199 "primary", 200 "--repo-local-root", 201 root.path().to_str().expect("UTF-8 test root"), 202 "--config", 203 config.to_str().expect("UTF-8 config path"), 204 "run", 205 ]) 206 .expect("runtime invocation"); 207 let runtime = resolve_rhi_runtime_context( 208 &RadrootsPathResolver::new( 209 RadrootsPlatform::current(), 210 RadrootsHostEnvironment::default(), 211 ), 212 &invocation, 213 ) 214 .expect("runtime context"); 215 Self { 216 root, 217 config, 218 runtime, 219 } 220 } 221 222 fn command(&self, command: &[&str]) -> Command { 223 let mut process = Command::new(env!("CARGO_BIN_EXE_rhi")); 224 process 225 .args(["--profile", "repo-local", "--instance", "primary"]) 226 .arg("--repo-local-root") 227 .arg(self.root.path()) 228 .arg("--config") 229 .arg(&self.config) 230 .args(command) 231 .stdin(Stdio::null()) 232 .stdout(Stdio::piped()) 233 .stderr(Stdio::piped()); 234 process 235 } 236 237 fn run(&self, command: &[&str]) -> Output { 238 BoundedProcess::spawn(self.command(command)).wait() 239 } 240 241 fn run_with_stdin(&self, command: &[&str], bytes: &[u8]) -> Output { 242 let mut process = self.command(command); 243 process.stdin(Stdio::piped()); 244 let mut child = BoundedProcess::spawn(process); 245 child 246 .child 247 .stdin 248 .take() 249 .expect("process stdin") 250 .write_all(bytes) 251 .expect("bounded stdin"); 252 child.wait() 253 } 254 } 255 256 struct BoundedProcess { 257 child: Child, 258 stdout_reader: Option<thread::JoinHandle<Vec<u8>>>, 259 stderr_reader: Option<thread::JoinHandle<Vec<u8>>>, 260 completed: bool, 261 } 262 263 impl BoundedProcess { 264 fn spawn(mut command: Command) -> Self { 265 let child = command.spawn().expect("RHI process"); 266 Self::from_child(child) 267 } 268 269 fn from_child(mut child: Child) -> Self { 270 let stdout = child.stdout.take().expect("captured process stdout"); 271 let stderr = child.stderr.take().expect("captured process stderr"); 272 Self { 273 child, 274 stdout_reader: Some(read_bounded(stdout, MAXIMUM_STDOUT_BYTES)), 275 stderr_reader: Some(read_bounded(stderr, MAXIMUM_STDERR_BYTES)), 276 completed: false, 277 } 278 } 279 280 fn id(&self) -> u32 { 281 self.child.id() 282 } 283 284 fn try_wait(&mut self) -> Option<ExitStatus> { 285 self.child.try_wait().expect("poll RHI process") 286 } 287 288 fn collect(&mut self, status: ExitStatus) -> Output { 289 self.completed = true; 290 let stdout = self 291 .stdout_reader 292 .take() 293 .expect("stdout reader available") 294 .join() 295 .expect("join stdout reader"); 296 let stderr = self 297 .stderr_reader 298 .take() 299 .expect("stderr reader available") 300 .join() 301 .expect("join stderr reader"); 302 assert!(stdout.len() <= MAXIMUM_STDOUT_BYTES); 303 assert!(stderr.len() <= MAXIMUM_STDERR_BYTES); 304 Output { 305 status, 306 stdout, 307 stderr, 308 } 309 } 310 311 fn wait(mut self) -> Output { 312 let deadline = Instant::now() + PROCESS_DEADLINE; 313 loop { 314 if let Some(status) = self.try_wait() { 315 return self.collect(status); 316 } 317 if Instant::now() >= deadline { 318 let _ = self.child.kill(); 319 let status = self.child.wait().expect("reap RHI process"); 320 let output = self.collect(status); 321 panic!("RHI process exceeded deadline: {:?}", output.stderr); 322 } 323 thread::sleep(POLL_INTERVAL); 324 } 325 } 326 } 327 328 impl Drop for BoundedProcess { 329 fn drop(&mut self) { 330 if !self.completed { 331 let _ = self.child.kill(); 332 let _ = self.child.wait(); 333 } 334 if let Some(reader) = self.stdout_reader.take() { 335 let _ = reader.join(); 336 } 337 if let Some(reader) = self.stderr_reader.take() { 338 let _ = reader.join(); 339 } 340 } 341 } 342 343 fn read_bounded( 344 reader: impl std::io::Read + Send + 'static, 345 maximum_bytes: usize, 346 ) -> thread::JoinHandle<Vec<u8>> { 347 thread::spawn(move || { 348 let limit = u64::try_from(maximum_bytes) 349 .expect("output bound fits u64") 350 .checked_add(1) 351 .expect("output bound plus sentinel fits u64"); 352 let mut output = Vec::with_capacity(maximum_bytes.min(8_192)); 353 reader 354 .take(limit) 355 .read_to_end(&mut output) 356 .expect("read bounded process output"); 357 output 358 }) 359 } 360 361 fn identity_secret() -> [u8; 32] { 362 let mut candidate: [u8; 32] = 363 Sha256::digest(b"radroots.rhi.step-213.process-identity.v1").into(); 364 while SecretKey::from_slice(&candidate).is_err() { 365 candidate = Sha256::digest(candidate).into(); 366 } 367 candidate 368 } 369 370 fn provisioning_document(secret: [u8; 32]) -> [u8; 117] { 371 let mut document = [0_u8; 117]; 372 document[..4].copy_from_slice(b"RHIP"); 373 document[4] = 1; 374 document[5..37].copy_from_slice(&secret); 375 document[37..69].copy_from_slice(&Sha256::digest(b"rhi-step-213-data-key")); 376 document[69..93].copy_from_slice(&[3; 24]); 377 document[93..117].copy_from_slice(&[4; 24]); 378 document 379 } 380 381 fn configuration( 382 fixture: &ProcessFixture, 383 expected_public_key: &str, 384 primary_relay: &str, 385 secondary_relay: &str, 386 ) -> String { 387 CONFIG_EXAMPLE 388 .replace( 389 "/var/lib/radroots/services/rhi/default/secrets/service.identity.ncrypt", 390 fixture 391 .runtime 392 .identity_path() 393 .to_str() 394 .expect("UTF-8 identity path"), 395 ) 396 .replace(&"2".repeat(64), expected_public_key) 397 .replace("wss://relay.example.com/", primary_relay) 398 .replace("wss://relay-secondary.example.com/", secondary_relay) 399 .replace( 400 "connect_deadline_ms = 10000", 401 &format!("connect_deadline_ms = {CONNECT_DEADLINE_MILLISECONDS}"), 402 ) 403 .replace("request_deadline_ms = 15000", "request_deadline_ms = 1000") 404 } 405 406 fn diagnostic_code(output: &Output) -> String { 407 let value: serde_json::Value = serde_json::from_slice(&output.stderr).expect("diagnostic JSON"); 408 value["code"].as_str().expect("diagnostic code").to_owned() 409 } 410 411 fn assert_success(output: &Output) { 412 assert_eq!(output.status.code(), Some(0), "stderr: {:?}", output.stderr); 413 assert_eq!(diagnostic_code(output), "success"); 414 } 415 416 fn bootstrap(fixture: &ProcessFixture, configuration: &str, secret: [u8; 32]) -> String { 417 let expected_public_key = Keys::new(SecretKey::from_slice(&secret).expect("secret")) 418 .public_key() 419 .to_hex(); 420 let config = fixture.run_with_stdin(&["config", "init"], configuration.as_bytes()); 421 assert_success(&config); 422 assert_eq!(config.stdout, b"config_initialized\n"); 423 424 for directory in [ 425 fixture.runtime.context().paths().state(), 426 fixture.runtime.context().paths().secrets(), 427 fixture.runtime.context().paths().run(), 428 ] { 429 fs::create_dir_all(directory).expect("secure runtime directory"); 430 fs::set_permissions(directory, fs::Permissions::from_mode(0o700)) 431 .expect("secure runtime mode"); 432 } 433 let credential = fixture 434 .runtime 435 .context() 436 .paths() 437 .secrets() 438 .join("service_wrapping_key"); 439 fs::write(&credential, Sha256::digest(b"rhi-step-213-wrapping-key")) 440 .expect("wrapping credential"); 441 fs::set_permissions(&credential, fs::Permissions::from_mode(0o600)).expect("credential mode"); 442 443 let state = fixture.run(&["state", "init"]); 444 assert_success(&state); 445 assert_eq!(state.stdout, b"state_initialized\n"); 446 let identity = fixture.run_with_stdin( 447 &["--output", "json", "identity", "init"], 448 &provisioning_document(secret), 449 ); 450 assert_success(&identity); 451 let identity_value: serde_json::Value = 452 serde_json::from_slice(&identity.stdout).expect("identity result"); 453 assert_eq!(identity_value["public_key"], expected_public_key); 454 455 for command in [ 456 &["config", "validate"][..], 457 &["state", "verify"][..], 458 &["state", "migrate"][..], 459 ] { 460 assert_success(&fixture.run(command)); 461 } 462 expected_public_key 463 } 464 465 fn wait_for_live_status(fixture: &ProcessFixture, daemon: &mut BoundedProcess) -> Output { 466 let deadline = Instant::now() + PROCESS_DEADLINE; 467 let mut last_diagnostic = Vec::new(); 468 loop { 469 if let Some(status) = daemon.try_wait() { 470 let output = daemon.collect(status); 471 panic!( 472 "RHI daemon exited before admin became ready: {status}; stderr={:?}", 473 output.stderr 474 ); 475 } 476 if fixture.runtime.artifacts().admin_socket().exists() { 477 let status = fixture.run(&["status"]); 478 if status.status.success() { 479 return status; 480 } 481 last_diagnostic = status.stderr; 482 } 483 if Instant::now() >= deadline { 484 panic!( 485 "RHI admin did not become ready before deadline; socket={}; stderr={last_diagnostic:?}", 486 fixture.runtime.artifacts().admin_socket().exists() 487 ); 488 } 489 thread::sleep(POLL_INTERVAL); 490 } 491 } 492 493 fn interrupt_and_wait(daemon: BoundedProcess) -> Output { 494 let signal = Command::new("/bin/kill") 495 .arg("-INT") 496 .arg(daemon.id().to_string()) 497 .status() 498 .expect("send interrupt"); 499 assert!(signal.success()); 500 daemon.wait() 501 } 502 503 #[test] 504 fn process_qualification_contract_freezes_the_exact_wave_closure() { 505 let contract: serde_json::Value = 506 serde_json::from_str(PROCESS_QUALIFICATION_CONTRACT).expect("process contract"); 507 assert_eq!( 508 contract 509 .as_object() 510 .expect("process contract object") 511 .keys() 512 .map(String::as_str) 513 .collect::<BTreeSet<_>>(), 514 BTreeSet::from([ 515 "actual_process_corpus", 516 "binary", 517 "bounds", 518 "component_qualification", 519 "contract_version", 520 "deferred", 521 "invariants", 522 "schema", 523 "service", 524 "source_lock", 525 "step", 526 ]) 527 ); 528 assert_eq!(contract["schema"], "radroots.rhi.process-qualification.v1"); 529 assert_eq!(contract["contract_version"], 1); 530 assert_eq!(contract["step"], 215); 531 assert_eq!(contract["service"], "rhi"); 532 assert_eq!(contract["binary"], "rhi"); 533 assert_eq!( 534 contract["source_lock"], 535 serde_json::json!({ 536 "schema": "radroots.service.source-lock.v2", 537 "lib_revision": "053d0c750bf9cd683c6ea37cefe7e79617ba629f" 538 }) 539 ); 540 assert_eq!( 541 contract["component_qualification"], 542 serde_json::json!({ 543 "schema": "radroots.rhi.failure-qualification.v1", 544 "step": 214, 545 "sha256": "f05da8e559f463f99c67c3c7e22eafa57935be91fb0094f2aa2f98a58544b8b9" 546 }) 547 ); 548 assert_eq!( 549 contract["bounds"], 550 serde_json::json!({ 551 "process_deadline_ms": 30_000, 552 "poll_interval_ms": 2, 553 "connect_deadline_ms": 5_000, 554 "relay_handshake_timeout_ms": 5_000, 555 "relay_io_timeout_ms": 100, 556 "parallel_inspection_processes": 8, 557 "soak_iterations": 32, 558 "maximum_stdout_bytes": 1_048_576, 559 "maximum_stderr_bytes": 8_192 560 }) 561 ); 562 assert_eq!( 563 PROCESS_DEADLINE, 564 Duration::from_millis( 565 contract["bounds"]["process_deadline_ms"] 566 .as_u64() 567 .expect("process deadline"), 568 ) 569 ); 570 assert_eq!( 571 POLL_INTERVAL, 572 Duration::from_millis( 573 contract["bounds"]["poll_interval_ms"] 574 .as_u64() 575 .expect("poll interval"), 576 ) 577 ); 578 assert_eq!( 579 RELAY_HANDSHAKE_TIMEOUT, 580 Duration::from_millis( 581 contract["bounds"]["relay_handshake_timeout_ms"] 582 .as_u64() 583 .expect("relay handshake timeout"), 584 ) 585 ); 586 assert_eq!( 587 RELAY_IO_TIMEOUT, 588 Duration::from_millis( 589 contract["bounds"]["relay_io_timeout_ms"] 590 .as_u64() 591 .expect("relay timeout"), 592 ) 593 ); 594 assert_eq!( 595 CONNECT_DEADLINE_MILLISECONDS, 596 contract["bounds"]["connect_deadline_ms"] 597 ); 598 assert_eq!( 599 u64::try_from(PARALLEL_INSPECTIONS).expect("inspection bound"), 600 contract["bounds"]["parallel_inspection_processes"] 601 ); 602 assert_eq!( 603 u64::try_from(SOAK_ITERATIONS).expect("soak bound"), 604 contract["bounds"]["soak_iterations"] 605 ); 606 assert_eq!( 607 u64::try_from(MAXIMUM_STDOUT_BYTES).expect("stdout bound"), 608 contract["bounds"]["maximum_stdout_bytes"] 609 ); 610 assert_eq!( 611 u64::try_from(MAXIMUM_STDERR_BYTES).expect("stderr bound"), 612 contract["bounds"]["maximum_stderr_bytes"] 613 ); 614 assert_eq!( 615 contract["actual_process_corpus"], 616 serde_json::json!([ 617 "actual_binary_executes_offline_bootstrap_and_reaches_real_runtime_dependency_boundary", 618 "actual_binary_runs_the_task_graph_serves_admin_and_shuts_down_on_interrupt", 619 "actual_binary_is_bounded_under_parallel_inspection_and_reopen_soak" 620 ]) 621 ); 622 assert_eq!( 623 contract["invariants"], 624 serde_json::json!({ 625 "actual_executable_required": true, 626 "loopback_relay_only": true, 627 "production_failpoint_surface": false, 628 "test_environment_selector": false, 629 "detached_test_worker": false, 630 "parallel_inspection_is_read_only": true, 631 "every_daemon_is_interrupted_joined_and_reaped": true, 632 "admin_socket_absent_after_shutdown": true, 633 "state_verifies_after_every_reopen": true, 634 "captured_output_bounded": true, 635 "diagnostics_path_secret_free": true 636 }) 637 ); 638 assert_eq!( 639 contract["deferred"], 640 serde_json::json!([ 641 "native_release_artifacts_step_216", 642 "rcld_promotion_step_217", 643 "parent_pin_alignment_step_217", 644 "nix", 645 "oci", 646 "signing", 647 "publication", 648 "deployment" 649 ]) 650 ); 651 assert_eq!( 652 lower_hex(&Sha256::digest(FAILURE_QUALIFICATION_CONTRACT)), 653 contract["component_qualification"]["sha256"] 654 ); 655 assert!(SOURCE_LOCK.contains("revision = \"055096853fca95e15d0f813d33a14aca13be3881\"")); 656 } 657 658 #[test] 659 fn actual_binary_executes_offline_bootstrap_and_reaches_real_runtime_dependency_boundary() { 660 let fixture = ProcessFixture::new(); 661 let secret = identity_secret(); 662 let expected_public_key = Keys::new(SecretKey::from_slice(&secret).expect("secret")) 663 .public_key() 664 .to_hex(); 665 let configuration = configuration( 666 &fixture, 667 &expected_public_key, 668 "ws://127.0.0.1:9/", 669 "ws://127.0.0.1:10/", 670 ); 671 bootstrap(&fixture, &configuration, secret); 672 673 let run = fixture.run(&["run"]); 674 assert_eq!(run.status.code(), Some(3)); 675 assert!(run.stdout.is_empty()); 676 assert_eq!(diagnostic_code(&run), "service_or_dependency_unavailable"); 677 let diagnostic = String::from_utf8(run.stderr).expect("diagnostic UTF-8"); 678 assert!(!diagnostic.contains(fixture.root.path().to_str().expect("UTF-8 root"))); 679 assert!(!diagnostic.contains("relay.example")); 680 assert!(!diagnostic.contains("service_wrapping_key")); 681 } 682 683 #[test] 684 fn actual_binary_runs_the_task_graph_serves_admin_and_shuts_down_on_interrupt() { 685 let relay = RelayHarness::start(); 686 let fixture = ProcessFixture::new(); 687 let secret = identity_secret(); 688 let expected_public_key = Keys::new(SecretKey::from_slice(&secret).expect("secret")) 689 .public_key() 690 .to_hex(); 691 let configuration = configuration( 692 &fixture, 693 &expected_public_key, 694 &relay.url("primary"), 695 &relay.url("secondary"), 696 ); 697 bootstrap(&fixture, &configuration, secret); 698 699 let mut daemon = BoundedProcess::spawn(fixture.command(&["run"])); 700 let status = wait_for_live_status(&fixture, &mut daemon); 701 assert_success(&status); 702 let status_value: serde_json::Value = 703 serde_json::from_slice(&status.stdout).expect("live status JSON"); 704 assert_eq!(status_value["service"], "rhi"); 705 assert_eq!(status_value["phase"], "ready"); 706 assert_eq!(status_value["ready"], true); 707 assert_eq!(status_value["persistence"]["schema_version"], 11); 708 709 let shutdown = interrupt_and_wait(daemon); 710 assert_success(&shutdown); 711 assert!(shutdown.stdout.is_empty()); 712 assert!(!fixture.runtime.artifacts().admin_socket().exists()); 713 } 714 715 #[test] 716 fn actual_binary_is_bounded_under_parallel_inspection_and_reopen_soak() { 717 let fixture = ProcessFixture::new(); 718 let secret = identity_secret(); 719 let expected_public_key = Keys::new(SecretKey::from_slice(&secret).expect("secret")) 720 .public_key() 721 .to_hex(); 722 let configuration = configuration( 723 &fixture, 724 &expected_public_key, 725 "ws://127.0.0.1:9/", 726 "ws://127.0.0.1:10/", 727 ); 728 bootstrap(&fixture, &configuration, secret); 729 730 let inspections = (0..PARALLEL_INSPECTIONS) 731 .map(|_| BoundedProcess::spawn(fixture.command(&["config", "validate"]))) 732 .collect::<Vec<_>>(); 733 for inspection in inspections { 734 assert_success(&inspection.wait()); 735 } 736 737 for _ in 0..SOAK_ITERATIONS { 738 assert_success(&fixture.run(&["state", "verify"])); 739 } 740 assert!(!fixture.runtime.artifacts().admin_socket().exists()); 741 } 742 743 fn lower_hex(bytes: &[u8]) -> String { 744 const DIGITS: &[u8; 16] = b"0123456789abcdef"; 745 let mut output = String::with_capacity(bytes.len() * 2); 746 for byte in bytes { 747 output.push(char::from(DIGITS[usize::from(byte >> 4)])); 748 output.push(char::from(DIGITS[usize::from(byte & 0x0f)])); 749 } 750 output 751 }