1use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
2use std::path::PathBuf;
3use std::process::Command;
4use std::sync::Arc;
5use std::thread;
6use std::time::Duration;
7
8use crate::auth::store::AuthStore;
9use crate::replication::ReplicationConfig;
10use crate::runtime::RedDBRuntime;
11use crate::service_router::{serve_tcp_router, InProcessRouterConfig};
12use crate::storage::StorageProfileSelection;
13use crate::{
14 GrpcServerOptions, RedDBGrpcServer, RedDBOptions, RedDBServer, ServerOptions, StorageMode,
15};
16
17pub const DEFAULT_ROUTER_BIND_ADDR: &str = "127.0.0.1:5050";
18
19pub fn detect_runtime_config() -> RuntimeConfig {
21 let cpus = thread::available_parallelism()
22 .map(|n| n.get())
23 .unwrap_or(1);
24
25 let suggested_workers = cpus.saturating_sub(1).max(1);
27
28 RuntimeConfig {
29 available_cpus: cpus,
30 suggested_workers,
31 stack_size: 8 * 1024 * 1024, }
33}
34
35#[derive(Debug, Clone)]
36pub struct RuntimeConfig {
37 pub available_cpus: usize,
38 pub suggested_workers: usize,
39 pub stack_size: usize,
40}
41
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum ServerTransport {
44 Grpc,
45 Http,
46 Wire,
47}
48
49impl ServerTransport {
50 pub const fn as_str(self) -> &'static str {
51 match self {
52 Self::Grpc => "gRPC",
53 Self::Http => "HTTP",
54 Self::Wire => "wire",
55 }
56 }
57
58 pub const fn default_bind_addr(self) -> &'static str {
59 match self {
60 Self::Grpc => "127.0.0.1:55055",
61 Self::Http => "127.0.0.1:5000",
62 Self::Wire => "127.0.0.1:5050",
63 }
64 }
65}
66
67#[derive(Debug, Clone)]
68pub struct ServerCommandConfig {
69 pub path: Option<PathBuf>,
70 pub router_bind_addr: Option<String>,
71 pub router_bind_explicit: bool,
72 pub grpc_bind_addr: Option<String>,
73 pub grpc_bind_explicit: bool,
74 pub grpc_tls_bind_addr: Option<String>,
78 pub grpc_tls_cert: Option<PathBuf>,
84 pub grpc_tls_key: Option<PathBuf>,
87 pub grpc_tls_client_ca: Option<PathBuf>,
92 pub http_bind_addr: Option<String>,
93 pub http_bind_explicit: bool,
94 pub http_tls_bind_addr: Option<String>,
98 pub http_tls_cert: Option<PathBuf>,
101 pub http_tls_key: Option<PathBuf>,
104 pub http_tls_client_ca: Option<PathBuf>,
108 pub wire_bind_addr: Option<String>,
109 pub wire_bind_explicit: bool,
110 pub wire_tls_bind_addr: Option<String>,
112 pub wire_tls_cert: Option<PathBuf>,
114 pub wire_tls_key: Option<PathBuf>,
116 pub pg_bind_addr: Option<String>,
120 pub create_if_missing: bool,
121 pub read_only: bool,
122 pub role: String,
123 pub primary_addr: Option<String>,
124 pub storage_profile: StorageProfileSelection,
125 pub auth: bool,
128 pub require_auth: bool,
130 pub vault: bool,
131 pub no_auth: bool,
142 pub workers: Option<usize>,
144 pub telemetry: Option<crate::telemetry::TelemetryConfig>,
147 pub http_limits_cli: crate::server::HttpLimitsCliInput,
152 pub ui: bool,
157 pub ui_dir: Option<PathBuf>,
161 pub bootstrap: BootstrapConfig,
164}
165
166#[derive(Debug, Clone, Default)]
167pub struct BootstrapConfig {
168 pub preset: Option<String>,
169 pub manifest: Option<PathBuf>,
170 pub admin_username: Option<String>,
171 pub admin_password: Option<String>,
172 pub cloud_head_admin: Option<String>,
173 pub cloud_head_admin_password: Option<String>,
174 pub customer_admin: Option<String>,
175 pub customer_admin_password: Option<String>,
176 pub cert_out: Option<PathBuf>,
185}
186
187#[derive(Debug, Clone, PartialEq, Eq)]
188pub struct TransportListenerState {
189 pub transport: String,
190 pub bind_addr: String,
191 pub explicit: bool,
192}
193
194#[derive(Debug, Clone, PartialEq, Eq)]
195pub struct TransportListenerFailure {
196 pub transport: String,
197 pub bind_addr: String,
198 pub explicit: bool,
199 pub reason: String,
200}
201
202#[derive(Debug, Clone, Default, PartialEq, Eq)]
203pub struct TransportReadiness {
204 pub active: Vec<TransportListenerState>,
205 pub failed: Vec<TransportListenerFailure>,
206}
207
208impl TransportReadiness {
209 fn active(&mut self, transport: &str, bind_addr: &str, explicit: bool) {
210 self.active.push(TransportListenerState {
211 transport: transport.to_string(),
212 bind_addr: bind_addr.to_string(),
213 explicit,
214 });
215 }
216
217 fn failed(&mut self, transport: &str, bind_addr: &str, explicit: bool, reason: String) {
218 self.failed.push(TransportListenerFailure {
219 transport: transport.to_string(),
220 bind_addr: bind_addr.to_string(),
221 explicit,
222 reason,
223 });
224 }
225}
226
227#[derive(Debug, Clone)]
228pub struct SystemdServiceConfig {
229 pub service_name: String,
230 pub binary_path: PathBuf,
231 pub run_user: String,
232 pub run_group: String,
233 pub data_path: PathBuf,
234 pub router_bind_addr: Option<String>,
235 pub grpc_bind_addr: Option<String>,
236 pub http_bind_addr: Option<String>,
237}
238
239impl SystemdServiceConfig {
240 pub fn data_dir(&self) -> PathBuf {
241 self.data_path
242 .parent()
243 .map(PathBuf::from)
244 .unwrap_or_else(|| PathBuf::from("."))
245 }
246
247 pub fn unit_path(&self) -> PathBuf {
248 PathBuf::from(format!("/etc/systemd/system/{}.service", self.service_name))
249 }
250}
251
252pub fn default_telemetry_for_path(
257 path: Option<&std::path::Path>,
258) -> crate::telemetry::TelemetryConfig {
259 let log_dir = match path {
260 Some(p) => p
261 .parent()
262 .map(|parent| parent.join("logs"))
263 .or_else(|| Some(std::path::PathBuf::from("./logs"))),
264 None => None, };
266 crate::telemetry::TelemetryConfig {
267 log_dir,
268 file_prefix: "reddb.log".to_string(),
269 level_filter: std::env::var("RUST_LOG").unwrap_or_else(|_| "info".to_string()),
270 format: if std::io::IsTerminal::is_terminal(&std::io::stderr()) {
271 crate::telemetry::LogFormat::Pretty
272 } else {
273 crate::telemetry::LogFormat::Json
274 },
275 rotation_keep_days: 14,
276 service_name: "reddb",
277 level_explicit: false,
279 format_explicit: false,
280 rotation_keep_days_explicit: false,
281 file_prefix_explicit: false,
282 log_dir_explicit: false,
283 log_file_disabled: false,
284 }
285}
286
287const BACKUP_INTERVAL_META_CHECKPOINT: &str = "red.boot.backup.checkpoint_interval_secs";
294const BACKUP_INTERVAL_META_WAL_FLUSH: &str = "red.boot.backup.wal_flush_interval_secs";
295const BACKUP_KIND_META: &str = "red.boot.backup.backend_kind";
296const BACKUP_PAUSE_ON_LAG_META: &str = "red.boot.backup.pause_on_lag_secs";
300
301pub(crate) const NO_AUTH_META: &str = "red.boot.no_auth";
309
310pub(crate) fn no_auth_active(options: &RedDBOptions) -> bool {
313 options
314 .metadata
315 .get(NO_AUTH_META)
316 .map(|v| v == "true")
317 .unwrap_or(false)
318}
319
320const NO_AUTH_WARNING: &str =
325 "⚠ auth disabled (--no-auth) — anonymous access, do NOT use in production";
326
327impl ServerCommandConfig {
328 fn to_db_options(&self) -> Result<RedDBOptions, String> {
329 let mut options = match &self.path {
330 Some(path) => RedDBOptions::persistent(path),
331 None => RedDBOptions::in_memory(),
332 };
333
334 options.mode = StorageMode::Persistent;
335 options.create_if_missing = self.create_if_missing;
336 options.read_only = self.read_only
343 || env_nonempty("RED_READONLY")
344 .or_else(|| env_nonempty("REDDB_READONLY"))
345 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
346 .unwrap_or(false)
347 || self.path.as_ref().is_some_and(|data_path| {
348 crate::server::handlers_admin::load_runtime_readonly(std::path::Path::new(
349 data_path,
350 ))
351 .unwrap_or(false)
352 });
353
354 options.replication = match self.role.as_str() {
355 "primary" => ReplicationConfig::primary(),
356 "replica" => {
357 let primary_addr = self
358 .primary_addr
359 .clone()
360 .unwrap_or_else(|| "http://127.0.0.1:55055".to_string());
361 ReplicationConfig::replica(primary_addr)
368 }
369 _ => ReplicationConfig::standalone(),
370 };
371 options.storage_profile = self.storage_profile.validate()?;
372
373 let no_auth = self.no_auth || env_truthy("REDDB_NO_AUTH") || env_truthy("REDDB_DEV");
374 let preset = self.bootstrap.resolved_preset_for_options()?;
375 let preset_requires_auth = matches!(preset.as_str(), PRESET_PRODUCTION | PRESET_CLOUD);
376
377 if self.auth || env_truthy("REDDB_AUTH") || preset_requires_auth {
378 options.auth.enabled = true;
379 }
380 if self.require_auth || env_truthy("REDDB_REQUIRE_AUTH") || preset_requires_auth {
381 options.auth.enabled = true;
382 options.auth.require_auth = true;
383 }
384 if self.vault || env_truthy("REDDB_VAULT") || preset_requires_auth {
385 options.auth.vault_enabled = true;
386 }
387
388 if no_auth {
399 options.auth.enabled = false;
400 options.auth.require_auth = false;
401 options.auth.vault_enabled = false;
402 options
403 .metadata
404 .insert(NO_AUTH_META.to_string(), "true".to_string());
405 }
406
407 let bootstrap_intent_present = !matches!(
427 self.bootstrap.auth_bootstrap_input(),
428 crate::cluster::AuthBootstrapInput::None
429 );
430 let path_absent = self.path.as_ref().is_some_and(|path| !path.exists());
431 let embedded_single_file = options.storage_profile.deploy_profile
434 == crate::storage::DeployProfile::Embedded
435 && options.storage_profile.packaging == crate::storage::StoragePackaging::SingleFile;
436 if options.auth.vault_enabled
437 && bootstrap_intent_present
438 && path_absent
439 && embedded_single_file
440 {
441 options.storage_profile.packaging =
442 crate::storage::StoragePackaging::OperationalDirectory;
443 options.create_if_missing = true;
444 tracing::info!(
445 target: "reddb::bootstrap",
446 "first-boot self-bootstrap: creating paged vault in place for a fresh database with a bootstrap intent"
447 );
448 }
449
450 if let Some(raw) = env_nonempty("REDDB_COMPLIANCE_MODE") {
455 options.control_events.compliance_mode = matches!(
456 raw.to_ascii_lowercase().as_str(),
457 "1" | "true" | "yes" | "on"
458 );
459 }
460 if preset == PRESET_REGULATED {
461 options.control_events.compliance_mode = true;
462 options.query_audit = crate::runtime::query_audit::QueryAuditConfig::regulated();
463 }
464
465 match crate::backup_bootstrap::from_env(|k| std::env::var(k).ok()) {
470 Err(msg) => {
471 return Err(format!("backup bootstrap: {msg}"));
472 }
473 Ok(Some(cfg)) => {
474 apply_backup_config(&mut options, &cfg);
475 }
476 Ok(None) => {
477 configure_remote_backend_from_env(&mut options);
478 }
479 }
480
481 if options.remote_backend.is_some()
482 || options
483 .metadata
484 .contains_key(BACKUP_INTERVAL_META_CHECKPOINT)
485 {
486 let mut selection = options.storage_profile;
487 selection.managed_backup = true;
488 options.storage_profile = selection.validate()?;
489 }
490
491 Ok(options)
492 }
493
494 pub fn enabled_transports(&self) -> Vec<ServerTransport> {
495 let mut transports = Vec::with_capacity(3);
496 if self.router_bind_addr.is_some() || self.grpc_bind_addr.is_some() {
497 transports.push(ServerTransport::Grpc);
498 }
499 if self.router_bind_addr.is_some() || self.http_bind_addr.is_some() {
500 transports.push(ServerTransport::Http);
501 }
502 if self.router_bind_addr.is_some() || self.wire_bind_addr.is_some() {
503 transports.push(ServerTransport::Wire);
504 }
505 transports
506 }
507}
508
509fn env_nonempty(name: &str) -> Option<String> {
514 crate::utils::env_with_file_fallback(name)
515}
516
517fn env_truthy(name: &str) -> bool {
518 env_nonempty(name)
519 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
520 .unwrap_or(false)
521}
522
523impl BootstrapConfig {
524 fn resolved_preset(&self) -> String {
525 self.preset
526 .as_deref()
527 .map(str::trim)
528 .filter(|s| !s.is_empty())
529 .map(str::to_string)
530 .or_else(|| env_nonempty(BOOTSTRAP_PRESET_ENV).or_else(|| env_nonempty(PRESET_ENV)))
531 .unwrap_or_else(|| PRESET_SIMPLE.to_string())
532 }
533
534 fn resolved_preset_for_options(&self) -> Result<String, String> {
535 if let Some(path) = self.selected_manifest() {
536 if let Some(manifest_bootstrap) =
537 crate::cli::bootstrap_manifest::bootstrap_config_from_manifest_file(&path)?
538 {
539 return Ok(manifest_bootstrap.resolved_preset());
540 }
541 }
542 Ok(self.resolved_preset())
543 }
544
545 fn selected_manifest(&self) -> Option<PathBuf> {
546 self.manifest.clone().or_else(|| {
547 env_nonempty(crate::cli::bootstrap_manifest::MANIFEST_ENV).map(PathBuf::from)
548 })
549 }
550
551 fn selected_cert_out(&self) -> Option<PathBuf> {
556 self.cert_out
557 .clone()
558 .or_else(|| env_nonempty(BOOTSTRAP_CERT_OUT_ENV).map(PathBuf::from))
559 }
560
561 fn auth_bootstrap_input(&self) -> crate::cluster::AuthBootstrapInput {
565 use crate::cluster::AuthBootstrapInput;
566 if self.selected_manifest().is_some() {
567 return AuthBootstrapInput::Manifest;
568 }
569 let preset = self.resolved_preset();
570 let preset_creates_auth = matches!(
571 preset.as_str(),
572 PRESET_PRODUCTION | PRESET_CLOUD | PRESET_REGULATED
573 );
574 let credentials_present =
579 self.production_username().is_some() && self.production_password().is_some();
580 if preset_creates_auth || credentials_present {
581 AuthBootstrapInput::Env
582 } else {
583 AuthBootstrapInput::None
584 }
585 }
586
587 fn explicit_value(value: &Option<String>) -> Option<String> {
588 value
589 .as_deref()
590 .map(str::trim)
591 .filter(|s| !s.is_empty())
592 .map(str::to_string)
593 }
594
595 fn value_or_env(value: &Option<String>, env: &str) -> Option<String> {
596 Self::explicit_value(value).or_else(|| env_nonempty(env))
597 }
598
599 fn production_username(&self) -> Option<String> {
600 Self::value_or_env(&self.admin_username, "REDDB_USERNAME")
601 }
602
603 fn production_password(&self) -> Option<String> {
604 Self::value_or_env(&self.admin_password, "REDDB_PASSWORD")
605 }
606
607 fn cloud_head_username(&self) -> Option<String> {
608 Self::explicit_value(&self.cloud_head_admin)
609 .or_else(|| Self::explicit_value(&self.admin_username))
610 .or_else(|| env_nonempty("REDDB_CLOUD_HEAD_ADMIN"))
611 .or_else(|| env_nonempty("REDDB_USERNAME"))
612 }
613
614 fn cloud_head_password(&self) -> Option<String> {
615 Self::explicit_value(&self.cloud_head_admin_password)
616 .or_else(|| Self::explicit_value(&self.admin_password))
617 .or_else(|| env_nonempty("REDDB_CLOUD_HEAD_ADMIN_PASSWORD"))
618 .or_else(|| env_nonempty("REDDB_PASSWORD"))
619 }
620
621 fn customer_username(&self) -> Option<String> {
622 Self::value_or_env(&self.customer_admin, "REDDB_CUSTOMER_ADMIN")
623 }
624
625 fn customer_password(&self) -> Option<String> {
626 Self::value_or_env(
627 &self.customer_admin_password,
628 "REDDB_CUSTOMER_ADMIN_PASSWORD",
629 )
630 }
631
632 fn secret_env_vars_to_expand(&self, preset: &str) -> Vec<&'static str> {
633 let mut vars = Vec::new();
634 match preset {
635 PRESET_PRODUCTION => {
636 if Self::explicit_value(&self.admin_username).is_none() {
637 vars.push("REDDB_USERNAME");
638 }
639 if Self::explicit_value(&self.admin_password).is_none() {
640 vars.push("REDDB_PASSWORD");
641 }
642 }
643 PRESET_CLOUD => {
644 if Self::explicit_value(&self.cloud_head_admin).is_none()
645 && Self::explicit_value(&self.admin_username).is_none()
646 {
647 vars.push("REDDB_USERNAME");
648 }
649 if Self::explicit_value(&self.cloud_head_admin_password).is_none()
650 && Self::explicit_value(&self.admin_password).is_none()
651 {
652 vars.push("REDDB_CLOUD_HEAD_ADMIN_PASSWORD");
653 vars.push("REDDB_PASSWORD");
654 }
655 if Self::explicit_value(&self.customer_admin_password).is_none() {
656 vars.push("REDDB_CUSTOMER_ADMIN_PASSWORD");
657 }
658 }
659 _ => {}
660 }
661 vars
662 }
663}
664
665fn apply_backup_config(options: &mut RedDBOptions, cfg: &crate::backup_bootstrap::BackupConfig) {
671 let endpoint_host = endpoint_host(&cfg.endpoint);
672
673 options.metadata.insert(
674 BACKUP_INTERVAL_META_CHECKPOINT.to_string(),
675 cfg.checkpoint_interval_secs.to_string(),
676 );
677 options.metadata.insert(
678 BACKUP_INTERVAL_META_WAL_FLUSH.to_string(),
679 cfg.wal_flush_interval_secs.to_string(),
680 );
681 options
682 .metadata
683 .insert(BACKUP_KIND_META.to_string(), "s3".to_string());
684 options.metadata.insert(
685 BACKUP_PAUSE_ON_LAG_META.to_string(),
686 cfg.pause_on_lag_secs.to_string(),
687 );
688
689 #[cfg(feature = "backend-s3")]
690 {
691 let s3_cfg = crate::storage::backend::S3Config {
692 endpoint: cfg.endpoint.clone(),
693 bucket: cfg.bucket.clone(),
694 key_prefix: cfg.prefix.clone(),
695 access_key: cfg.access_key_id.clone(),
696 secret_key: cfg.secret_access_key.clone(),
697 region: cfg.region.clone(),
698 path_style: true,
699 };
700 let backend = Arc::new(crate::storage::backend::S3Backend::new(s3_cfg));
701 options.remote_backend = Some(backend.clone());
702 options.remote_backend_atomic = Some(backend);
703 let trimmed = cfg.prefix.trim_end_matches('/');
708 options.remote_key = Some(reddb_file::remote_database_key(trimmed));
709
710 tracing::info!(
711 backend = "s3",
712 endpoint = %endpoint_host,
713 bucket = %cfg.bucket,
714 prefix = %cfg.prefix,
715 checkpoint_interval_secs = cfg.checkpoint_interval_secs,
716 wal_flush_interval_secs = cfg.wal_flush_interval_secs,
717 "backup backend configured from REDDB_BACKUP_* env"
718 );
719 }
720
721 #[cfg(not(feature = "backend-s3"))]
722 {
723 tracing::warn!(
724 backend = "s3",
725 endpoint = %endpoint_host,
726 bucket = %cfg.bucket,
727 prefix = %cfg.prefix,
728 "REDDB_BACKUP_S3_* configured but binary built without `backend-s3` feature; \
729 backend wiring skipped (archiver/checkpointer also disabled)"
730 );
731 }
732}
733
734fn endpoint_host(endpoint: &str) -> &str {
735 let after_scheme = endpoint
736 .split_once("://")
737 .map(|(_, r)| r)
738 .unwrap_or(endpoint);
739 after_scheme.split('/').next().unwrap_or(after_scheme)
740}
741
742fn spawn_backup_tasks_if_configured(
748 options: &RedDBOptions,
749 runtime: &RedDBRuntime,
750) -> Option<BackupTasksHandle> {
751 let checkpoint_secs: u64 = options
752 .metadata
753 .get(BACKUP_INTERVAL_META_CHECKPOINT)?
754 .parse()
755 .ok()?;
756 let wal_secs: u64 = options
757 .metadata
758 .get(BACKUP_INTERVAL_META_WAL_FLUSH)?
759 .parse()
760 .ok()?;
761 let pause_on_lag_secs: u64 = options
764 .metadata
765 .get(BACKUP_PAUSE_ON_LAG_META)
766 .and_then(|raw| raw.parse().ok())
767 .unwrap_or(0);
768 options.remote_backend.as_ref()?;
769
770 let stop = Arc::new(std::sync::atomic::AtomicBool::new(false));
771
772 if pause_on_lag_secs > 0 {
777 let now_ms = std::time::SystemTime::now()
778 .duration_since(std::time::UNIX_EPOCH)
779 .map(|d| d.as_millis() as u64)
780 .unwrap_or(0);
781 runtime
782 .write_gate()
783 .configure_archive_lag_pause(pause_on_lag_secs, now_ms);
784 tracing::info!(
785 pause_on_lag_secs,
786 "archive-lag pause enabled — engine will transition to read-only after threshold seconds of archiver silence"
787 );
788 }
789
790 let checkpoint_handle = {
791 let stop = Arc::clone(&stop);
792 let runtime = runtime.clone();
793 let interval = Duration::from_secs(checkpoint_secs);
794 thread::Builder::new()
795 .name("red-checkpointer".into())
796 .spawn(move || {
797 periodic_loop(stop, interval, move || {
798 if let Err(err) = runtime.checkpoint() {
799 tracing::warn!(error = %err, "periodic checkpoint failed");
800 }
801 })
802 })
803 .ok()
804 };
805
806 let archiver_handle = {
807 let stop = Arc::clone(&stop);
808 let runtime = runtime.clone();
809 let interval = Duration::from_secs(wal_secs);
810 let lag_enabled = pause_on_lag_secs > 0;
811 thread::Builder::new()
812 .name("red-wal-archiver".into())
813 .spawn(move || {
814 periodic_loop(stop, interval, move || match runtime.trigger_backup() {
815 Ok(_) if lag_enabled => {
816 let now_ms = std::time::SystemTime::now()
817 .duration_since(std::time::UNIX_EPOCH)
818 .map(|d| d.as_millis() as u64)
819 .unwrap_or(0);
820 runtime.write_gate().record_archive_success(now_ms);
821 runtime.write_gate().evaluate_archive_lag(now_ms);
825 }
826 Ok(_) => {}
827 Err(err) => {
828 tracing::warn!(error = %err, "periodic WAL archive/backup failed");
829 }
830 })
831 })
832 .ok()
833 };
834
835 let lag_monitor_handle = if pause_on_lag_secs > 0 {
840 let stop = Arc::clone(&stop);
841 let runtime = runtime.clone();
842 let interval = Duration::from_secs(5);
846 thread::Builder::new()
847 .name("red-archive-lag-monitor".into())
848 .spawn(move || {
849 periodic_loop(stop, interval, move || {
850 let now_ms = std::time::SystemTime::now()
851 .duration_since(std::time::UNIX_EPOCH)
852 .map(|d| d.as_millis() as u64)
853 .unwrap_or(0);
854 let was_paused = runtime.write_gate().is_auto_paused();
855 let now_paused = runtime.write_gate().evaluate_archive_lag(now_ms);
856 if now_paused && !was_paused {
857 tracing::warn!(
858 pause_on_lag_secs,
859 last_archive_at_ms = runtime.write_gate().last_archive_at_ms(),
860 "WAL archive lag exceeded threshold — entering graceful read-only mode (issue #519)"
861 );
862 } else if !now_paused && was_paused {
863 tracing::info!(
864 "WAL archive caught up — exiting graceful read-only mode (issue #519)"
865 );
866 }
867 })
868 })
869 .ok()
870 } else {
871 None
872 };
873
874 tracing::info!(
875 checkpoint_interval_secs = checkpoint_secs,
876 wal_flush_interval_secs = wal_secs,
877 "backup tasks spawned (checkpointer + WAL archiver)"
878 );
879
880 Some(BackupTasksHandle {
881 stop,
882 _checkpoint_handle: checkpoint_handle,
883 _archiver_handle: archiver_handle,
884 _lag_monitor_handle: lag_monitor_handle,
885 })
886}
887
888pub struct BackupTasksHandle {
891 stop: Arc<std::sync::atomic::AtomicBool>,
892 _checkpoint_handle: Option<thread::JoinHandle<()>>,
893 _archiver_handle: Option<thread::JoinHandle<()>>,
894 _lag_monitor_handle: Option<thread::JoinHandle<()>>,
897}
898
899impl Drop for BackupTasksHandle {
900 fn drop(&mut self) {
901 self.stop.store(true, std::sync::atomic::Ordering::Release);
902 }
903}
904
905fn periodic_loop<F: FnMut()>(
906 stop: Arc<std::sync::atomic::AtomicBool>,
907 interval: Duration,
908 mut tick: F,
909) {
910 let wake = Duration::from_secs(1);
913 let mut elapsed = Duration::ZERO;
914 while !stop.load(std::sync::atomic::Ordering::Acquire) {
915 thread::sleep(wake);
916 elapsed += wake;
917 if elapsed >= interval {
918 tick();
919 elapsed = Duration::ZERO;
920 }
921 }
922}
923
924fn configure_remote_backend_from_env(options: &mut RedDBOptions) {
925 let backend = env_nonempty("RED_BACKEND")
931 .or_else(|| env_nonempty("REDDB_REMOTE_BACKEND"))
932 .unwrap_or_else(|| "none".to_string())
933 .to_ascii_lowercase();
934
935 match backend.as_str() {
936 "s3" | "minio" | "r2" => {
941 #[cfg(feature = "backend-s3")]
942 {
943 if let Some(config) = s3_config_from_env() {
944 let remote_key = env_nonempty("RED_REMOTE_KEY")
945 .or_else(|| env_nonempty("REDDB_REMOTE_KEY"))
946 .unwrap_or_else(|| reddb_file::remote_database_key("clusters/dev"));
947 let backend = Arc::new(crate::storage::backend::S3Backend::new(config));
948 options.remote_backend = Some(backend.clone());
949 options.remote_backend_atomic = Some(backend);
950 options.remote_key = Some(remote_key);
951 }
952 }
953 #[cfg(not(feature = "backend-s3"))]
954 {
955 tracing::warn!(
956 backend = %backend,
957 "RED_BACKEND={backend} requested but binary was built without `backend-s3` feature"
958 );
959 }
960 }
961 "fs" | "local" => {
966 let base_path = env_nonempty("RED_FS_PATH").or_else(|| env_nonempty("REDDB_FS_PATH"));
967 let remote_key = match (
968 base_path,
969 env_nonempty("RED_REMOTE_KEY").or_else(|| env_nonempty("REDDB_REMOTE_KEY")),
970 ) {
971 (Some(base), Some(rel)) => Some(format!(
972 "{}/{}",
973 base.trim_end_matches('/'),
974 rel.trim_start_matches('/')
975 )),
976 (Some(base), None) => Some(reddb_file::remote_database_key(&format!(
977 "{}/clusters/dev",
978 base.trim_end_matches('/')
979 ))),
980 (None, Some(rel)) => Some(rel),
981 (None, None) => None,
982 };
983 if let Some(key) = remote_key {
984 let backend = Arc::new(crate::storage::backend::LocalBackend);
985 options.remote_backend = Some(backend.clone());
986 options.remote_backend_atomic = Some(backend);
987 options.remote_key = Some(key);
988 }
989 }
990 "http" => {
995 let base_url = match env_nonempty("RED_HTTP_BACKEND_URL")
996 .or_else(|| env_nonempty("REDDB_HTTP_BACKEND_URL"))
997 {
998 Some(u) => u,
999 None => {
1000 tracing::warn!(
1001 "RED_BACKEND=http requires RED_HTTP_BACKEND_URL — backend disabled"
1002 );
1003 return;
1004 }
1005 };
1006 let prefix = env_nonempty("RED_HTTP_BACKEND_PREFIX")
1007 .or_else(|| env_nonempty("REDDB_HTTP_BACKEND_PREFIX"))
1008 .unwrap_or_default();
1009 let auth_header = if let Some(path) = env_nonempty("RED_HTTP_BACKEND_AUTH_HEADER_FILE")
1010 .or_else(|| env_nonempty("REDDB_HTTP_BACKEND_AUTH_HEADER_FILE"))
1011 {
1012 std::fs::read_to_string(&path)
1013 .ok()
1014 .map(|s| s.trim().to_string())
1015 .filter(|s| !s.is_empty())
1016 } else {
1017 env_nonempty("RED_HTTP_BACKEND_AUTH_HEADER")
1018 .or_else(|| env_nonempty("REDDB_HTTP_BACKEND_AUTH_HEADER"))
1019 };
1020
1021 let mut config =
1022 crate::storage::backend::HttpBackendConfig::new(base_url).with_prefix(prefix);
1023 if let Some(auth) = auth_header {
1024 config = config.with_auth_header(auth);
1025 }
1026 let conditional_writes = env_truthy("RED_HTTP_CONDITIONAL_WRITES")
1027 || env_truthy("RED_HTTP_BACKEND_CONDITIONAL_WRITES")
1028 || env_truthy("REDDB_HTTP_BACKEND_CONDITIONAL_WRITES");
1029 config = config.with_conditional_writes(conditional_writes);
1030 if conditional_writes {
1035 match crate::storage::backend::AtomicHttpBackend::try_new(config.clone()) {
1036 Ok(atomic) => {
1037 let atomic_arc = Arc::new(atomic);
1038 options.remote_backend = Some(atomic_arc.clone());
1039 options.remote_backend_atomic = Some(atomic_arc);
1040 }
1041 Err(err) => {
1042 tracing::warn!(error = %err, "AtomicHttpBackend init failed; falling back to plain HTTP (no CAS)");
1043 options.remote_backend =
1044 Some(Arc::new(crate::storage::backend::HttpBackend::new(config)));
1045 }
1046 }
1047 } else {
1048 options.remote_backend =
1049 Some(Arc::new(crate::storage::backend::HttpBackend::new(config)));
1050 }
1051 options.remote_key = env_nonempty("RED_REMOTE_KEY")
1052 .or_else(|| env_nonempty("REDDB_REMOTE_KEY"))
1053 .or_else(|| Some(reddb_file::remote_database_key("clusters/dev")));
1054 }
1055 "none" | "" => {}
1058 other => {
1059 tracing::warn!(
1060 backend = %other,
1061 "unknown RED_BACKEND value — supported: s3 | fs | http | none"
1062 );
1063 }
1064 }
1065}
1066
1067#[cfg(feature = "backend-s3")]
1072fn env_s3(suffix: &str) -> Option<String> {
1073 env_nonempty(&format!("RED_S3_{suffix}"))
1074 .or_else(|| env_nonempty(&format!("REDDB_S3_{suffix}")))
1075}
1076
1077#[cfg(feature = "backend-s3")]
1083fn env_s3_secret(suffix: &str) -> Option<String> {
1084 let file_key_red = format!("RED_S3_{suffix}_FILE");
1085 let file_key_legacy = format!("REDDB_S3_{suffix}_FILE");
1086 if let Some(path) = env_nonempty(&file_key_red).or_else(|| env_nonempty(&file_key_legacy)) {
1087 return std::fs::read_to_string(&path)
1088 .ok()
1089 .map(|s| s.trim().to_string())
1090 .filter(|s| !s.is_empty());
1091 }
1092 env_s3(suffix)
1093}
1094
1095#[cfg(feature = "backend-s3")]
1096fn s3_config_from_env() -> Option<crate::storage::backend::S3Config> {
1097 let endpoint = env_s3("ENDPOINT")?;
1098 let bucket = env_s3("BUCKET")?;
1099 let access_key = env_s3_secret("ACCESS_KEY")?;
1100 let secret_key = env_s3_secret("SECRET_KEY")?;
1101 let region = env_s3("REGION").unwrap_or_else(|| "us-east-1".to_string());
1102 let key_prefix = env_s3("KEY_PREFIX")
1103 .or_else(|| env_s3("PREFIX"))
1104 .unwrap_or_default();
1105 let path_style = env_s3("PATH_STYLE")
1106 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
1107 .unwrap_or(true);
1108 Some(crate::storage::backend::S3Config {
1109 endpoint,
1110 bucket,
1111 key_prefix,
1112 access_key,
1113 secret_key,
1114 region,
1115 path_style,
1116 })
1117}
1118
1119pub fn render_systemd_unit(config: &SystemdServiceConfig) -> String {
1120 let data_dir = config.data_dir();
1121 let exec_start = render_systemd_exec_start(config);
1122 format!(
1123 "[Unit]\n\
1124Description=RedDB unified database service\n\
1125After=network-online.target\n\
1126Wants=network-online.target\n\
1127\n\
1128[Service]\n\
1129Type=simple\n\
1130User={user}\n\
1131Group={group}\n\
1132WorkingDirectory={workdir}\n\
1133ExecStart={exec_start}\n\
1134Restart=always\n\
1135RestartSec=2\n\
1136LimitSTACK=16M\n\
1137NoNewPrivileges=true\n\
1138PrivateTmp=true\n\
1139ProtectSystem=strict\n\
1140ProtectHome=true\n\
1141ProtectControlGroups=true\n\
1142ProtectKernelTunables=true\n\
1143ProtectKernelModules=true\n\
1144RestrictNamespaces=true\n\
1145LockPersonality=true\n\
1146MemoryDenyWriteExecute=true\n\
1147ReadWritePaths={workdir}\n\
1148\n\
1149[Install]\n\
1150WantedBy=multi-user.target\n",
1151 user = config.run_user,
1152 group = config.run_group,
1153 workdir = data_dir.display(),
1154 exec_start = exec_start,
1155 )
1156}
1157
1158#[cfg(target_os = "linux")]
1167pub fn install_systemd_service(config: &SystemdServiceConfig) -> Result<(), String> {
1168 ensure_root()?;
1169 ensure_command_available("systemctl")?;
1170 ensure_command_available("getent")?;
1171 ensure_command_available("groupadd")?;
1172 ensure_command_available("useradd")?;
1173 ensure_command_available("install")?;
1174 ensure_executable(&config.binary_path)?;
1175
1176 if !command_success("getent", ["group", config.run_group.as_str()])? {
1177 run_command("groupadd", ["--system", config.run_group.as_str()])?;
1178 }
1179
1180 if !command_success("id", ["-u", config.run_user.as_str()])? {
1181 let data_dir = config.data_dir();
1182 run_command(
1183 "useradd",
1184 [
1185 "--system",
1186 "--gid",
1187 config.run_group.as_str(),
1188 "--home-dir",
1189 data_dir.to_string_lossy().as_ref(),
1190 "--shell",
1191 "/usr/sbin/nologin",
1192 config.run_user.as_str(),
1193 ],
1194 )?;
1195 }
1196
1197 let data_dir = config.data_dir();
1198 run_command(
1199 "install",
1200 [
1201 "-d",
1202 "-o",
1203 config.run_user.as_str(),
1204 "-g",
1205 config.run_group.as_str(),
1206 "-m",
1207 "0750",
1208 data_dir.to_string_lossy().as_ref(),
1209 ],
1210 )?;
1211
1212 std::fs::write(config.unit_path(), render_systemd_unit(config))
1213 .map_err(|err| format!("failed to write systemd unit: {err}"))?;
1214
1215 run_command("systemctl", ["daemon-reload"])?;
1216 run_command(
1217 "systemctl",
1218 [
1219 "enable",
1220 "--now",
1221 format!("{}.service", config.service_name).as_str(),
1222 ],
1223 )?;
1224
1225 Ok(())
1226}
1227
1228#[cfg(not(target_os = "linux"))]
1233pub fn install_systemd_service(_config: &SystemdServiceConfig) -> Result<(), String> {
1234 Err("systemd install is Linux-only — use sc.exe (Windows) or \
1235 launchd (macOS) to install the service manually using the \
1236 unit printed by `red service print-unit`"
1237 .to_string())
1238}
1239
1240#[cfg(target_os = "linux")]
1241fn ensure_root() -> Result<(), String> {
1242 let output = Command::new("id")
1243 .arg("-u")
1244 .output()
1245 .map_err(|err| format!("failed to determine current uid: {err}"))?;
1246 if !output.status.success() {
1247 return Err("failed to determine current uid".to_string());
1248 }
1249 let uid = String::from_utf8_lossy(&output.stdout);
1250 if uid.trim() != "0" {
1251 return Err("run this command as root (sudo)".to_string());
1252 }
1253 Ok(())
1254}
1255
1256#[cfg(target_os = "linux")]
1257fn ensure_command_available(command: &str) -> Result<(), String> {
1258 let status = Command::new("sh")
1259 .args(["-lc", &format!("command -v {command} >/dev/null 2>&1")])
1260 .status()
1261 .map_err(|err| format!("failed to check command '{command}': {err}"))?;
1262 if status.success() {
1263 Ok(())
1264 } else {
1265 Err(format!("required command not found: {command}"))
1266 }
1267}
1268
1269#[cfg(target_os = "linux")]
1270fn ensure_executable(path: &std::path::Path) -> Result<(), String> {
1271 let metadata = std::fs::metadata(path)
1272 .map_err(|err| format!("binary not found '{}': {err}", path.display()))?;
1273 #[cfg(unix)]
1274 {
1275 use std::os::unix::fs::PermissionsExt;
1276 if metadata.permissions().mode() & 0o111 == 0 {
1277 return Err(format!("binary is not executable: {}", path.display()));
1278 }
1279 }
1280 #[cfg(not(unix))]
1281 {
1282 if !metadata.is_file() {
1283 return Err(format!("binary is not a file: {}", path.display()));
1284 }
1285 }
1286 Ok(())
1287}
1288
1289#[cfg(target_os = "linux")]
1290fn command_success<const N: usize>(program: &str, args: [&str; N]) -> Result<bool, String> {
1291 Command::new(program)
1292 .args(args)
1293 .status()
1294 .map(|status| status.success())
1295 .map_err(|err| format!("failed to run {program}: {err}"))
1296}
1297
1298#[cfg(target_os = "linux")]
1299fn run_command<const N: usize>(program: &str, args: [&str; N]) -> Result<(), String> {
1300 let output = Command::new(program)
1301 .args(args)
1302 .output()
1303 .map_err(|err| format!("failed to run {program}: {err}"))?;
1304 if output.status.success() {
1305 return Ok(());
1306 }
1307
1308 let stderr = String::from_utf8_lossy(&output.stderr).trim().to_string();
1309 let stdout = String::from_utf8_lossy(&output.stdout).trim().to_string();
1310 let detail = if !stderr.is_empty() {
1311 stderr
1312 } else if !stdout.is_empty() {
1313 stdout
1314 } else {
1315 format!("exit status {}", output.status)
1316 };
1317 Err(format!("{program} failed: {detail}"))
1318}
1319
1320pub fn run_server_with_large_stack(config: ServerCommandConfig) -> Result<(), String> {
1321 let has_any = config.router_bind_addr.is_some()
1322 || config.grpc_bind_addr.is_some()
1323 || config.http_bind_addr.is_some()
1324 || config.wire_bind_addr.is_some()
1325 || config.wire_tls_bind_addr.is_some()
1326 || config.pg_bind_addr.is_some();
1327 if !has_any {
1328 return Err("at least one server bind address must be configured".into());
1329 }
1330 let thread_name = if config.router_bind_addr.is_some() {
1331 "red-server-router"
1332 } else {
1333 match (
1334 config.grpc_bind_addr.is_some(),
1335 config.http_bind_addr.is_some(),
1336 ) {
1337 (true, true) => "red-server-dual",
1338 (true, false) => "red-server-grpc",
1339 (false, true) => "red-server-http",
1340 (false, false) if config.wire_bind_addr.is_some() => "red-server-wire",
1341 (false, false) if config.wire_tls_bind_addr.is_some() => "red-server-wire-tls",
1342 (false, false) => "red-server-pg-wire",
1343 }
1344 };
1345
1346 let handle = thread::Builder::new()
1347 .name(thread_name.into())
1348 .stack_size(8 * 1024 * 1024)
1349 .spawn(move || run_configured_servers(config))
1350 .map_err(|err| format!("failed to spawn server thread: {err}"))?;
1351
1352 match handle.join() {
1353 Ok(result) => result,
1354 Err(_) => Err("server thread panicked".to_string()),
1355 }
1356}
1357
1358fn render_systemd_exec_start(config: &SystemdServiceConfig) -> String {
1359 let mut parts = vec![
1360 config.binary_path.display().to_string(),
1361 "server".to_string(),
1362 "--path".to_string(),
1363 config.data_path.display().to_string(),
1364 ];
1365
1366 if let Some(bind_addr) = &config.router_bind_addr {
1367 parts.push("--bind".to_string());
1368 parts.push(bind_addr.clone());
1369 } else if let Some(bind_addr) = &config.grpc_bind_addr {
1370 parts.push("--grpc-bind".to_string());
1371 parts.push(bind_addr.clone());
1372 }
1373 if let Some(bind_addr) = &config.http_bind_addr {
1374 parts.push("--http-bind".to_string());
1375 parts.push(bind_addr.clone());
1376 }
1377
1378 parts.join(" ")
1379}
1380
1381pub fn probe_listener(target: &str, timeout: Duration) -> bool {
1382 let addresses: Vec<SocketAddr> = match target.to_socket_addrs() {
1383 Ok(addresses) => addresses.collect(),
1384 Err(_) => return false,
1385 };
1386
1387 addresses
1388 .into_iter()
1389 .any(|address| TcpStream::connect_timeout(&address, timeout).is_ok())
1390}
1391
1392#[inline(never)]
1393fn run_configured_servers(config: ServerCommandConfig) -> Result<(), String> {
1394 if let Some(router_bind_addr) = config.router_bind_addr.clone() {
1400 return run_routed_server(config, router_bind_addr);
1401 }
1402
1403 match (config.grpc_bind_addr.clone(), config.http_bind_addr.clone()) {
1404 (Some(grpc_bind_addr), Some(http_bind_addr)) => {
1405 run_dual_server(config, grpc_bind_addr, http_bind_addr)
1406 }
1407 (Some(grpc_bind_addr), None) => run_grpc_server(config, grpc_bind_addr),
1408 (None, Some(http_bind_addr)) => run_http_server(config, http_bind_addr),
1409 (None, None) => {
1410 if let Some(wire_addr) = config.wire_bind_addr.clone() {
1411 run_wire_only_server(config, wire_addr)
1412 } else if let Some(wire_tls_addr) = config.wire_tls_bind_addr.clone() {
1413 run_wire_tls_only_server(config, wire_tls_addr)
1414 } else if let Some(pg_addr) = config.pg_bind_addr.clone() {
1415 run_pg_only_server(config, pg_addr)
1416 } else {
1417 Err("at least one server bind address must be configured".to_string())
1418 }
1419 }
1420 }
1421}
1422
1423pub fn bind_listener_for_startup(
1441 readiness: &mut TransportReadiness,
1442 transport: &str,
1443 bind_addr: &str,
1444 explicit: bool,
1445) -> Result<Option<TcpListener>, String> {
1446 match TcpListener::bind(bind_addr) {
1447 Ok(listener) => {
1448 readiness.active(transport, bind_addr, explicit);
1449 Ok(Some(listener))
1450 }
1451 Err(err) => {
1452 let reason = format!("{transport} listener bind {bind_addr}: {err}");
1453 readiness.failed(transport, bind_addr, explicit, reason.clone());
1454 if explicit {
1455 tracing::error!(
1456 transport,
1457 bind = %bind_addr,
1458 error = %err,
1459 "fatal explicit bind failure"
1460 );
1461 Err(format!("explicit {reason}"))
1462 } else {
1463 tracing::warn!(
1464 transport,
1465 bind = %bind_addr,
1466 error = %err,
1467 "non-fatal implicit bind failure; listener degraded"
1468 );
1469 Ok(None)
1470 }
1471 }
1472 }
1473}
1474
1475async fn spawn_lifecycle_signal_handler(runtime: RedDBRuntime) {
1498 let backup_on_shutdown = std::env::var("RED_BACKUP_ON_SHUTDOWN")
1499 .ok()
1500 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
1501 .unwrap_or(true);
1502
1503 #[cfg(unix)]
1504 {
1505 use tokio::signal::unix::{signal, SignalKind};
1506
1507 let mut sigterm = match signal(SignalKind::terminate()) {
1508 Ok(s) => s,
1509 Err(err) => {
1510 tracing::warn!(
1511 error = %err,
1512 "could not install SIGTERM handler; orchestrator graceful shutdown will fall back to SIGKILL"
1513 );
1514 return;
1515 }
1516 };
1517 let mut sigint = match signal(SignalKind::interrupt()) {
1518 Ok(s) => s,
1519 Err(err) => {
1520 tracing::warn!(error = %err, "could not install SIGINT handler");
1521 return;
1522 }
1523 };
1524 let mut sighup = match signal(SignalKind::hangup()) {
1530 Ok(s) => Some(s),
1531 Err(err) => {
1532 tracing::warn!(error = %err, "could not install SIGHUP handler; secret reload via signal disabled");
1533 None
1534 }
1535 };
1536
1537 let reload_runtime = runtime.clone();
1538 tokio::spawn(async move {
1539 loop {
1540 let signal_name = match &mut sighup {
1541 Some(hup) => tokio::select! {
1542 _ = sigterm.recv() => "SIGTERM",
1543 _ = sigint.recv() => "SIGINT",
1544 _ = hup.recv() => "SIGHUP",
1545 },
1546 None => tokio::select! {
1547 _ = sigterm.recv() => "SIGTERM",
1548 _ = sigint.recv() => "SIGINT",
1549 },
1550 };
1551
1552 if signal_name == "SIGHUP" {
1553 handle_sighup_reload(&reload_runtime);
1554 continue; }
1556
1557 tracing::info!(
1558 signal = signal_name,
1559 "lifecycle signal received; shutting down"
1560 );
1561 match runtime.graceful_shutdown(backup_on_shutdown) {
1562 Ok(report) => {
1563 tracing::info!(
1564 duration_ms = report.duration_ms,
1565 flushed_wal = report.flushed_wal,
1566 final_checkpoint = report.final_checkpoint,
1567 backup_uploaded = report.backup_uploaded,
1568 "graceful shutdown complete"
1569 );
1570 }
1571 Err(err) => {
1572 tracing::error!(error = %err, "graceful shutdown failed");
1573 crate::telemetry::operator_event::OperatorEvent::ShutdownForced {
1579 reason: format!("graceful shutdown failed: {err}"),
1580 }
1581 .emit_global();
1582 }
1583 }
1584 std::process::exit(0);
1585 }
1586 });
1587 }
1588
1589 #[cfg(not(unix))]
1590 {
1591 tokio::spawn(async move {
1592 let interrupted = tokio::signal::ctrl_c().await;
1593 if let Err(err) = interrupted {
1594 tracing::warn!(error = %err, "could not install Ctrl+C handler");
1595 return;
1596 }
1597
1598 tracing::info!(
1599 signal = "Ctrl+C",
1600 "lifecycle signal received; shutting down"
1601 );
1602 match runtime.graceful_shutdown(backup_on_shutdown) {
1603 Ok(report) => {
1604 tracing::info!(
1605 duration_ms = report.duration_ms,
1606 flushed_wal = report.flushed_wal,
1607 final_checkpoint = report.final_checkpoint,
1608 backup_uploaded = report.backup_uploaded,
1609 "graceful shutdown complete"
1610 );
1611 }
1612 Err(err) => {
1613 tracing::error!(error = %err, "graceful shutdown failed");
1614 }
1615 }
1616 std::process::exit(0);
1617 });
1618 }
1619}
1620
1621fn handle_sighup_reload(runtime: &RedDBRuntime) {
1630 let now_ms = std::time::SystemTime::now()
1631 .duration_since(std::time::UNIX_EPOCH)
1632 .map(|d| d.as_millis() as u64)
1633 .unwrap_or(0);
1634 tracing::info!(
1635 target: "reddb::secrets",
1636 ts_unix_ms = now_ms,
1637 "SIGHUP received; secrets will be re-read from *_FILE on next access"
1638 );
1639 use crate::runtime::audit_log::{AuditAuthSource, AuditEvent, AuditFieldEscaper, Outcome};
1644 runtime.audit_log().record_event(
1645 AuditEvent::builder("config/sighup_reload")
1646 .source(AuditAuthSource::System)
1647 .resource("secrets")
1648 .outcome(Outcome::Success)
1649 .field(AuditFieldEscaper::field("ts_unix_ms", now_ms))
1650 .build(),
1651 );
1652}
1653
1654#[inline(never)]
1655fn run_routed_server(config: ServerCommandConfig, router_bind_addr: String) -> Result<(), String> {
1656 let workers = config.workers;
1657 let db_options = config.to_db_options()?;
1658 let rt_config = detect_runtime_config();
1659 let worker_threads = workers.unwrap_or(rt_config.suggested_workers);
1660 let (runtime, auth_store, _telemetry_guard) =
1661 build_runtime_and_auth_store(&config, db_options.clone())?;
1662 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
1663
1664 spawn_admin_metrics_listeners(&runtime, &auth_store);
1665
1666 let http_server = build_http_server(
1672 runtime.clone(),
1673 auth_store.clone(),
1674 router_bind_addr.clone(),
1675 );
1676 let http_server = apply_http_limits(http_server, &config, &runtime);
1677 let http_server = apply_ui_bundle(http_server, &config)?;
1678
1679 let grpc_server = RedDBGrpcServer::with_options(
1680 runtime.clone(),
1681 GrpcServerOptions {
1682 bind_addr: router_bind_addr.clone(),
1683 tls: None,
1684 },
1685 auth_store,
1686 );
1687
1688 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
1689 .enable_all()
1690 .worker_threads(worker_threads)
1691 .thread_stack_size(rt_config.stack_size)
1692 .build()
1693 .map_err(|err| format!("tokio runtime: {err}"))?;
1694
1695 let signal_runtime = runtime.clone();
1696 let wire_runtime = Arc::new(runtime);
1697 tokio_runtime.block_on(async move {
1698 spawn_lifecycle_signal_handler(signal_runtime).await;
1699 tracing::info!(
1700 bind = %router_bind_addr,
1701 cpus = rt_config.available_cpus,
1702 workers = worker_threads,
1703 "router bootstrapping"
1704 );
1705 serve_tcp_router(InProcessRouterConfig {
1706 bind_addr: router_bind_addr,
1707 http_server,
1708 grpc_server,
1709 wire_runtime,
1710 })
1711 .await
1712 .map_err(|err| err.to_string())
1713 })
1714}
1715
1716async fn spawn_wire_listeners(
1718 config: &ServerCommandConfig,
1719 runtime: &RedDBRuntime,
1720 readiness: &mut TransportReadiness,
1721) -> Result<(), String> {
1722 if let Some(wire_addr) = config.wire_bind_addr.clone() {
1724 let wire_rt = Arc::new(runtime.clone());
1725 #[cfg(unix)]
1728 {
1729 if wire_addr.starts_with("unix://") || wire_addr.starts_with('/') {
1730 readiness.active("wire", &wire_addr, config.wire_bind_explicit);
1731 tokio::spawn(async move {
1732 if let Err(e) = crate::wire::redwire::listener::start_redwire_unix_listener(
1733 &wire_addr, wire_rt,
1734 )
1735 .await
1736 {
1737 tracing::error!(err = %e, "redwire unix listener error");
1738 }
1739 });
1740 return Ok(());
1741 }
1742 }
1743 match tokio::net::TcpListener::bind(&wire_addr).await {
1744 Ok(listener) => {
1745 readiness.active("wire", &wire_addr, config.wire_bind_explicit);
1746 tokio::spawn(async move {
1747 if let Err(e) =
1748 crate::wire::redwire::listener::start_redwire_listener_on(listener, wire_rt)
1749 .await
1750 {
1751 tracing::error!(err = %e, "redwire listener error");
1752 }
1753 });
1754 }
1755 Err(err) => {
1756 let reason = format!("wire listener bind {wire_addr}: {err}");
1757 readiness.failed(
1758 "wire",
1759 &wire_addr,
1760 config.wire_bind_explicit,
1761 reason.clone(),
1762 );
1763 if config.wire_bind_explicit {
1764 tracing::error!(
1765 transport = "wire",
1766 bind = %wire_addr,
1767 error = %err,
1768 "fatal explicit bind failure"
1769 );
1770 return Err(format!("explicit {reason}"));
1771 }
1772 tracing::warn!(
1773 transport = "wire",
1774 bind = %wire_addr,
1775 error = %err,
1776 "non-fatal implicit bind failure; listener degraded"
1777 );
1778 }
1779 }
1780 }
1781
1782 if let Some(wire_tls_addr) = config.wire_tls_bind_addr.clone() {
1791 let tls_cfg = match resolve_wire_tls_config(config) {
1792 Ok(cfg) => cfg,
1793 Err(e) => {
1794 let reason = format!("redwire TLS config error: {e}");
1795 readiness.failed("wire-tls", &wire_tls_addr, true, reason.clone());
1796 tracing::error!(
1797 transport = "wire-tls",
1798 bind = %wire_tls_addr,
1799 error = %e,
1800 "fatal explicit redwire TLS config error"
1801 );
1802 return Err(format!("explicit {reason}"));
1803 }
1804 };
1805 let listener = match tokio::net::TcpListener::bind(&wire_tls_addr).await {
1808 Ok(listener) => listener,
1809 Err(err) => {
1810 let reason = format!("wire-tls listener bind {wire_tls_addr}: {err}");
1811 readiness.failed("wire-tls", &wire_tls_addr, true, reason.clone());
1812 tracing::error!(
1813 transport = "wire-tls",
1814 bind = %wire_tls_addr,
1815 error = %err,
1816 "fatal explicit bind failure"
1817 );
1818 return Err(format!("explicit {reason}"));
1819 }
1820 };
1821 readiness.active("wire-tls", &wire_tls_addr, true);
1822 tracing::info!(transport = "redwire+tls", bind = %wire_tls_addr, "listener online");
1826 let wire_rt = Arc::new(runtime.clone());
1827 tokio::spawn(async move {
1828 if let Err(e) =
1829 crate::wire::start_redwire_tls_listener_on(listener, wire_rt, &tls_cfg).await
1830 {
1831 tracing::error!(err = %e, "redwire+tls listener error");
1832 }
1833 });
1834 }
1835 Ok(())
1836}
1837
1838fn spawn_pg_listener(config: &ServerCommandConfig, runtime: &RedDBRuntime) {
1845 if let Some(pg_addr) = config.pg_bind_addr.clone() {
1846 let tls = resolve_pg_wire_tls(config);
1847 let rt = Arc::new(runtime.clone());
1848 tokio::spawn(async move {
1849 let cfg = crate::wire::PgWireConfig {
1850 bind_addr: pg_addr,
1851 tls,
1852 ..Default::default()
1853 };
1854 if let Err(e) = crate::wire::start_pg_wire_listener(cfg, rt).await {
1855 tracing::error!(err = %e, "pg wire listener error");
1856 }
1857 });
1858 }
1859}
1860
1861fn resolve_pg_wire_tls(config: &ServerCommandConfig) -> Option<crate::wire::WireTlsConfig> {
1872 let tls_configured = config.wire_tls_bind_addr.is_some()
1873 || (config.wire_tls_cert.is_some() && config.wire_tls_key.is_some());
1874 if !tls_configured {
1875 return None;
1876 }
1877 match resolve_wire_tls_config(config) {
1878 Ok(cfg) => Some(cfg),
1879 Err(e) => {
1880 tracing::error!(
1881 err = %e,
1882 "pg wire TLS config error; listener will run plaintext"
1883 );
1884 None
1885 }
1886 }
1887}
1888
1889fn resolve_grpc_tls_options(config: &ServerCommandConfig) -> Result<crate::GrpcTlsOptions, String> {
1903 use crate::utils::secret_file::expand_file_env;
1904
1905 for var in [
1909 "REDDB_GRPC_TLS_CERT",
1910 "REDDB_GRPC_TLS_KEY",
1911 "REDDB_GRPC_TLS_CLIENT_CA",
1912 ] {
1913 if let Err(err) = expand_file_env(var) {
1914 tracing::warn!(
1915 target: "reddb::secrets",
1916 env = %var,
1917 err = %err,
1918 "could not expand *_FILE companion for gRPC TLS"
1919 );
1920 }
1921 }
1922
1923 let (cert_pem, key_pem) = match (&config.grpc_tls_cert, &config.grpc_tls_key) {
1924 (Some(cert), Some(key)) => {
1925 let cert_pem = std::fs::read(cert)
1926 .map_err(|e| format!("read grpc cert {}: {e}", cert.display()))?;
1927 let key_pem =
1928 std::fs::read(key).map_err(|e| format!("read grpc key {}: {e}", key.display()))?;
1929 (cert_pem, key_pem)
1930 }
1931 _ => {
1932 let dev = std::env::var("RED_GRPC_TLS_DEV")
1934 .ok()
1935 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
1936 .unwrap_or(false);
1937 if !dev {
1938 return Err("gRPC TLS configured but no cert/key supplied — set \
1939 REDDB_GRPC_TLS_CERT / REDDB_GRPC_TLS_KEY (or \
1940 RED_GRPC_TLS_DEV=1 to auto-generate a self-signed cert)"
1941 .to_string());
1942 }
1943 let dir = config
1944 .path
1945 .as_ref()
1946 .and_then(|p| p.parent())
1947 .map(PathBuf::from)
1948 .unwrap_or_else(|| PathBuf::from("."));
1949 let (cert_pem_str, key_pem_str) =
1950 crate::wire::tls::generate_self_signed_cert("localhost")
1951 .map_err(|e| format!("auto-generate dev grpc cert: {e}"))?;
1952
1953 let fp = sha256_pem_fingerprint(cert_pem_str.as_bytes());
1958 tracing::warn!(
1959 target: "reddb::security",
1960 transport = "grpc",
1961 cert_sha256 = %fp,
1962 "RED_GRPC_TLS_DEV=1: using auto-generated self-signed cert; \
1963 DO NOT use in production"
1964 );
1965 let cert_path = dir.join("grpc-tls-cert.pem");
1967 let key_path = dir.join("grpc-tls-key.pem");
1968 if !cert_path.exists() || !key_path.exists() {
1969 let _ = std::fs::create_dir_all(&dir);
1970 std::fs::write(&cert_path, cert_pem_str.as_bytes())
1971 .map_err(|e| format!("write grpc dev cert: {e}"))?;
1972 std::fs::write(&key_path, key_pem_str.as_bytes())
1973 .map_err(|e| format!("write grpc dev key: {e}"))?;
1974 #[cfg(unix)]
1975 {
1976 use std::os::unix::fs::PermissionsExt;
1977 let _ =
1978 std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600));
1979 }
1980 }
1981 (cert_pem_str.into_bytes(), key_pem_str.into_bytes())
1982 }
1983 };
1984
1985 let client_ca_pem = match &config.grpc_tls_client_ca {
1986 Some(path) => Some(
1987 std::fs::read(path)
1988 .map_err(|e| format!("read grpc client CA {}: {e}", path.display()))?,
1989 ),
1990 None => None,
1991 };
1992
1993 Ok(crate::GrpcTlsOptions {
1994 cert_pem,
1995 key_pem,
1996 client_ca_pem,
1997 })
1998}
1999
2000fn spawn_grpc_tls_listener_if_configured(
2004 config: &ServerCommandConfig,
2005 runtime: RedDBRuntime,
2006 auth_store: Arc<AuthStore>,
2007) {
2008 let Some(tls_bind) = config.grpc_tls_bind_addr.clone() else {
2009 return;
2010 };
2011 let tls_opts = match resolve_grpc_tls_options(config) {
2012 Ok(opts) => opts,
2013 Err(err) => {
2014 tracing::error!(
2015 target: "reddb::security",
2016 transport = "grpc",
2017 err = %err,
2018 "gRPC TLS config error; TLS listener will not start"
2019 );
2020 return;
2021 }
2022 };
2023 tokio::spawn(async move {
2024 let server = RedDBGrpcServer::with_options(
2025 runtime,
2026 GrpcServerOptions {
2027 bind_addr: tls_bind.clone(),
2028 tls: Some(tls_opts),
2029 },
2030 auth_store,
2031 );
2032 tracing::info!(transport = "grpc+tls", bind = %tls_bind, "listener online");
2033 if let Err(err) = server.serve().await {
2034 tracing::error!(transport = "grpc+tls", err = %err, "gRPC TLS listener error");
2035 }
2036 });
2037}
2038
2039fn sha256_pem_fingerprint(pem: &[u8]) -> String {
2042 use sha2::{Digest, Sha256};
2043 let mut h = Sha256::new();
2044 h.update(pem);
2045 let d = h.finalize();
2046 let mut buf = String::with_capacity(64);
2047 for b in d.iter() {
2048 buf.push_str(&format!("{b:02x}"));
2049 }
2050 buf
2051}
2052
2053fn resolve_wire_tls_config(
2055 config: &ServerCommandConfig,
2056) -> Result<crate::wire::WireTlsConfig, String> {
2057 match (&config.wire_tls_cert, &config.wire_tls_key) {
2058 (Some(cert), Some(key)) => Ok(crate::wire::WireTlsConfig {
2059 cert_path: cert.clone(),
2060 key_path: key.clone(),
2061 }),
2062 _ => {
2063 let dir = config
2065 .path
2066 .as_ref()
2067 .and_then(|p| p.parent())
2068 .map(PathBuf::from)
2069 .unwrap_or_else(|| PathBuf::from("."));
2070 crate::wire::tls::auto_generate_cert(&dir).map_err(|e| e.to_string())
2071 }
2072 }
2073}
2074
2075#[inline(never)]
2076fn run_wire_only_server(config: ServerCommandConfig, wire_addr: String) -> Result<(), String> {
2077 let rt_config = detect_runtime_config();
2078 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2079 let db_options = config.to_db_options()?;
2080 let mut transport_readiness = TransportReadiness::default();
2081
2082 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2083 .enable_all()
2084 .worker_threads(workers)
2085 .thread_stack_size(rt_config.stack_size)
2086 .build()
2087 .map_err(|err| format!("tokio runtime: {err}"))?;
2088
2089 let (runtime, _auth_store, _telemetry_guard) =
2093 build_runtime_and_auth_store(&config, db_options.clone())?;
2094 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2095 let signal_runtime = runtime.clone();
2096 tokio_runtime.block_on(async move {
2097 spawn_lifecycle_signal_handler(signal_runtime).await;
2098 spawn_pg_listener(&config, &runtime);
2099 let wire_rt = Arc::new(runtime);
2100 let listener = tokio::net::TcpListener::bind(&wire_addr)
2101 .await
2102 .map_err(|err| {
2103 let reason = format!("wire listener bind {wire_addr}: {err}");
2104 transport_readiness.failed(
2105 "wire",
2106 &wire_addr,
2107 config.wire_bind_explicit,
2108 reason.clone(),
2109 );
2110 if config.wire_bind_explicit {
2111 format!("explicit {reason}")
2112 } else {
2113 reason
2114 }
2115 })?;
2116 transport_readiness.active("wire", &wire_addr, config.wire_bind_explicit);
2117 crate::wire::redwire::listener::start_redwire_listener_on(listener, wire_rt)
2118 .await
2119 .map_err(|e| e.to_string())
2120 })
2121}
2122
2123#[inline(never)]
2131fn run_wire_tls_only_server(
2132 config: ServerCommandConfig,
2133 wire_tls_addr: String,
2134) -> Result<(), String> {
2135 let rt_config = detect_runtime_config();
2136 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2137 let db_options = config.to_db_options()?;
2138 let tls_cfg = resolve_wire_tls_config(&config)?;
2142
2143 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2144 .enable_all()
2145 .worker_threads(workers)
2146 .thread_stack_size(rt_config.stack_size)
2147 .build()
2148 .map_err(|err| format!("tokio runtime: {err}"))?;
2149
2150 let (runtime, _auth_store, _telemetry_guard) =
2151 build_runtime_and_auth_store(&config, db_options.clone())?;
2152 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2153 let signal_runtime = runtime.clone();
2154 tokio_runtime.block_on(async move {
2155 spawn_lifecycle_signal_handler(signal_runtime).await;
2156 spawn_pg_listener(&config, &runtime);
2157 let wire_rt = Arc::new(runtime);
2158 crate::wire::start_redwire_tls_listener(&wire_tls_addr, wire_rt, &tls_cfg)
2159 .await
2160 .map_err(|e| format!("explicit wire-tls listener bind {wire_tls_addr}: {e}"))
2161 })
2162}
2163
2164#[inline(never)]
2165fn run_pg_only_server(config: ServerCommandConfig, pg_addr: String) -> Result<(), String> {
2166 let rt_config = detect_runtime_config();
2167 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2168 let db_options = config.to_db_options()?;
2169
2170 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2171 .enable_all()
2172 .worker_threads(workers)
2173 .thread_stack_size(rt_config.stack_size)
2174 .build()
2175 .map_err(|err| format!("tokio runtime: {err}"))?;
2176
2177 let (runtime, _auth_store, _telemetry_guard) =
2178 build_runtime_and_auth_store(&config, db_options.clone())?;
2179 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2180 let pg_tls = resolve_pg_wire_tls(&config);
2181 let signal_runtime = runtime.clone();
2182 tokio_runtime.block_on(async move {
2183 spawn_lifecycle_signal_handler(signal_runtime).await;
2184 let cfg = crate::wire::PgWireConfig {
2185 bind_addr: pg_addr,
2186 tls: pg_tls,
2187 ..Default::default()
2188 };
2189 crate::wire::start_pg_wire_listener(cfg, Arc::new(runtime))
2190 .await
2191 .map_err(|e| e.to_string())
2192 })
2193}
2194
2195#[inline(never)]
2196fn build_runtime_and_auth_store(
2197 config: &ServerCommandConfig,
2198 db_options: RedDBOptions,
2199) -> Result<
2200 (
2201 RedDBRuntime,
2202 Arc<AuthStore>,
2203 Option<crate::telemetry::TelemetryGuard>,
2204 ),
2205 String,
2206> {
2207 build_runtime_with_bootstrap(db_options, config.telemetry.clone(), &config.bootstrap)
2214}
2215
2216pub(crate) fn build_runtime_with_telemetry(
2226 db_options: RedDBOptions,
2227 cli_telemetry: Option<crate::telemetry::TelemetryConfig>,
2228) -> Result<
2229 (
2230 RedDBRuntime,
2231 Arc<AuthStore>,
2232 Option<crate::telemetry::TelemetryGuard>,
2233 ),
2234 String,
2235> {
2236 let bootstrap = BootstrapConfig::default();
2237 build_runtime_with_bootstrap(db_options, cli_telemetry, &bootstrap)
2238}
2239
2240fn build_runtime_with_bootstrap(
2241 db_options: RedDBOptions,
2242 cli_telemetry: Option<crate::telemetry::TelemetryConfig>,
2243 bootstrap: &BootstrapConfig,
2244) -> Result<
2245 (
2246 RedDBRuntime,
2247 Arc<AuthStore>,
2248 Option<crate::telemetry::TelemetryGuard>,
2249 ),
2250 String,
2251> {
2252 let runtime = RedDBRuntime::with_options(db_options.clone()).map_err(|err| {
2253 let msg = err.to_string();
2259 crate::telemetry::operator_event::OperatorEvent::StartupFailed {
2260 phase: "runtime_construction".to_string(),
2261 error: msg.clone(),
2262 }
2263 .emit_global();
2264 msg
2265 })?;
2266
2267 crate::runtime::lease_loop::start_lease_loop_if_required(&runtime).map_err(|err| {
2272 let msg = err.to_string();
2273 crate::telemetry::operator_event::OperatorEvent::StartupFailed {
2274 phase: "lease_loop".to_string(),
2275 error: msg.clone(),
2276 }
2277 .emit_global();
2278 msg
2279 })?;
2280
2281 if let Some(data_path) = db_options.data_path.as_deref() {
2285 let watch_dir = data_path.parent().unwrap_or(data_path);
2286 crate::runtime::disk_space_monitor::DiskSpaceMonitor::new(watch_dir, 90).spawn();
2287 }
2288
2289 {
2293 let config_path = crate::runtime::config_overlay::config_file_path();
2294 let store = runtime.db().store();
2295 crate::runtime::config_watcher::ConfigWatcher::new(config_path, store).spawn();
2296 }
2297
2298 let merged = merge_telemetry_with_config(
2301 cli_telemetry
2302 .unwrap_or_else(|| default_telemetry_for_path(db_options.data_path.as_deref())),
2303 &runtime,
2304 );
2305 let telemetry_guard = crate::telemetry::init(merged);
2306
2307 let no_auth = no_auth_active(&db_options);
2308 let auth_store =
2309 if db_options.auth.vault_enabled {
2310 let pager =
2311 runtime.db().store().pager().cloned().ok_or_else(|| {
2312 "vault requires a paged database (persistent mode)".to_string()
2313 })?;
2314 let store = AuthStore::with_vault(db_options.auth.clone(), pager)
2315 .map_err(|err| err.to_string())?;
2316 Arc::new(store)
2317 } else {
2318 Arc::new(AuthStore::new(db_options.auth.clone()))
2319 };
2320 auth_store.configure_control_events(
2321 runtime.control_event_ledger(),
2322 runtime.control_event_config(),
2323 );
2324 let bootstrap_already_completed = runtime
2341 .db()
2342 .store()
2343 .get_config(BOOTSTRAP_COMPLETED_KEY)
2344 .is_some();
2345 let disposition = crate::cluster::authorize_cluster_bootstrap(
2346 db_options.storage_profile.deploy_profile,
2347 no_auth,
2348 bootstrap.auth_bootstrap_input(),
2349 bootstrap_already_completed,
2350 )?;
2351 let vault_plan = crate::cluster::plan_vault_bootstrap(disposition);
2360 match disposition {
2361 crate::cluster::BootstrapDisposition::SkipDevBypass => {
2362 debug_assert_eq!(vault_plan, crate::cluster::VaultBootstrapPlan::SkipNoVault);
2363 eprintln!("{NO_AUTH_WARNING}");
2364 tracing::warn!("{NO_AUTH_WARNING}");
2365 }
2366 crate::cluster::BootstrapDisposition::AlreadyComplete => {
2367 debug_assert_eq!(
2374 vault_plan,
2375 crate::cluster::VaultBootstrapPlan::OpenClusterGlobalStore {
2376 consume_secret_inputs: false,
2377 }
2378 );
2379 apply_preset_from_config(&runtime, &auth_store, bootstrap)?;
2380 tracing::info!("bootstrap already completed, unsealing existing cluster auth store");
2381 }
2382 crate::cluster::BootstrapDisposition::ProceedLocal => {
2383 debug_assert_eq!(
2386 vault_plan,
2387 crate::cluster::VaultBootstrapPlan::OpenClusterGlobalStore {
2388 consume_secret_inputs: true,
2389 }
2390 );
2391 apply_preset_from_config(&runtime, &auth_store, bootstrap)?;
2392 maybe_apply_policy_break_glass(&auth_store);
2393 }
2394 }
2395
2396 {
2398 let store = Arc::clone(&auth_store);
2399 std::thread::Builder::new()
2400 .name("reddb-session-purge".into())
2401 .spawn(move || loop {
2402 std::thread::sleep(std::time::Duration::from_secs(300));
2403 store.purge_expired_sessions();
2404 })
2405 .ok();
2406 }
2407
2408 Ok((runtime, auth_store, telemetry_guard))
2409}
2410
2411fn maybe_apply_policy_break_glass(auth_store: &Arc<AuthStore>) {
2415 use crate::auth::self_lock_guard::BREAK_GLASS_ENV;
2416
2417 let enabled = std::env::var(BREAK_GLASS_ENV)
2418 .ok()
2419 .map(|v| {
2420 let trimmed = v.trim().to_ascii_lowercase();
2421 matches!(trimmed.as_str(), "1" | "true" | "yes")
2422 })
2423 .unwrap_or(false);
2424 if !enabled {
2425 return;
2426 }
2427 let now = crate::utils::now_unix_millis() as u128;
2428 match auth_store.apply_policy_break_glass(now) {
2429 Ok(()) => {
2430 tracing::warn!(env = BREAK_GLASS_ENV, "policy break-glass recovery applied");
2431 }
2432 Err(err) => {
2433 tracing::error!(env = BREAK_GLASS_ENV, %err, "policy break-glass recovery failed");
2434 }
2435 }
2436}
2437
2438pub(crate) const BOOTSTRAP_COMPLETED_KEY: &str = "system.bootstrap.completed";
2443pub(crate) const BOOTSTRAP_PRESET_KEY: &str = "system.bootstrap.preset";
2444pub(crate) const BOOTSTRAP_FIRST_ADMIN_KEY: &str = "system.bootstrap.first_admin_id";
2445
2446pub(crate) const BOOTSTRAP_PRESET_ENV: &str = "REDDB_BOOTSTRAP_PRESET";
2449pub(crate) const PRESET_ENV: &str = "REDDB_PRESET";
2450pub(crate) const BOOTSTRAP_CERT_OUT_ENV: &str = "REDDB_BOOTSTRAP_CERT_OUT";
2453pub(crate) const PRESET_SIMPLE: &str = "simple";
2454pub(crate) const PRESET_PRODUCTION: &str = "production";
2455pub(crate) const PRESET_REGULATED: &str = "regulated";
2456pub(crate) const PRESET_CLOUD: &str = "cloud";
2457
2458pub(crate) const FIRST_ADMIN_ALLOW_ALL_POLICY: &str = "system.bootstrap.first-admin-allow-all";
2462pub(crate) const REGULATED_PROTECT_MANAGED_POLICY: &str = "system.regulated.protect-managed";
2463pub(crate) const CLOUD_PROTECT_MANAGED_POLICY: &str = "system.cloud.protect-managed";
2464pub(crate) const CLOUD_CONFIG_NAMESPACE: &str = "red.config.cloud";
2465pub(crate) const REGULATED_AUDIT_CONFIG_NAMESPACE: &str = "red.config.audit";
2466pub(crate) const REGULATED_EVIDENCE_CONFIG_NAMESPACE: &str = "red.config.evidence";
2467pub(crate) const REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE: &str = "red.config.query_audit";
2468
2469pub(crate) fn apply_preset(
2479 runtime: &RedDBRuntime,
2480 auth_store: &Arc<AuthStore>,
2481) -> Result<(), String> {
2482 let bootstrap = BootstrapConfig::default();
2483 apply_preset_from_config(runtime, auth_store, &bootstrap)
2484}
2485
2486fn apply_preset_from_config(
2487 runtime: &RedDBRuntime,
2488 auth_store: &Arc<AuthStore>,
2489 bootstrap: &BootstrapConfig,
2490) -> Result<(), String> {
2491 let store = runtime.db().store();
2492
2493 if store.get_config(BOOTSTRAP_COMPLETED_KEY).is_some() {
2494 crate::cli::bootstrap_manifest::rehydrate_manifest_registry(
2495 runtime,
2496 &runtime.config_registry(),
2497 )?;
2498 tracing::info!("bootstrap state present, skipping preset application");
2499 return Ok(());
2500 }
2501
2502 let preset = bootstrap.resolved_preset();
2503
2504 for var in bootstrap.secret_env_vars_to_expand(&preset) {
2509 crate::utils::expand_file_env(var).map_err(|err| format!("expand {var}_FILE: {err}"))?;
2510 }
2511
2512 if let Some(path) = bootstrap.selected_manifest() {
2513 if let Some(manifest_bootstrap) =
2514 crate::cli::bootstrap_manifest::bootstrap_config_from_manifest_file(&path)?
2515 {
2516 apply_preset_from_config(runtime, auth_store, &manifest_bootstrap)?;
2517 return Ok(());
2518 }
2519 let first_admin_id = crate::cli::bootstrap_manifest::apply_manifest_file(
2520 runtime,
2521 auth_store,
2522 &runtime.config_registry(),
2523 &path,
2524 )?;
2525 persist_bootstrap_state(runtime, "manifest", Some(&first_admin_id));
2526 tracing::info!("bootstrap manifest applied");
2527 return Ok(());
2528 }
2529
2530 let (first_admin_id, certificate) = match preset.as_str() {
2531 PRESET_SIMPLE => {
2532 (None, None)
2536 }
2537 PRESET_PRODUCTION => {
2538 let (id, cert) = apply_production_preset_from_config(auth_store, bootstrap)?;
2539 (Some(id), cert)
2540 }
2541 PRESET_CLOUD => {
2542 let (id, cert) = apply_cloud_preset(runtime, auth_store, bootstrap)?;
2543 (Some(id), cert)
2544 }
2545 PRESET_REGULATED => {
2546 apply_regulated_preset(runtime, auth_store)?;
2547 (None, None)
2548 }
2549 other => {
2550 return Err(format!(
2551 "bootstrap preset {other:?} is not recognised (expected `simple`, `production`, `regulated`, or `cloud`)"
2552 ));
2553 }
2554 };
2555
2556 if let Some(cert) = certificate.as_deref() {
2563 write_bootstrap_certificate(bootstrap, cert)?;
2564 }
2565
2566 persist_bootstrap_state(runtime, &preset, first_admin_id.as_deref());
2567 tracing::info!(preset = %preset, "bootstrap preset applied");
2568 Ok(())
2569}
2570
2571fn write_bootstrap_certificate(
2575 bootstrap: &BootstrapConfig,
2576 certificate_hex: &str,
2577) -> Result<(), String> {
2578 let Some(path) = bootstrap.selected_cert_out() else {
2579 return Ok(());
2580 };
2581 std::fs::write(&path, certificate_hex)
2582 .map_err(|err| format!("write --bootstrap-cert-out {}: {err}", path.display()))?;
2583 tracing::info!(
2584 target: "reddb::bootstrap",
2585 path = %path.display(),
2586 "first-boot self-bootstrap: wrote unseal certificate to --bootstrap-cert-out (consume via REDDB_CERTIFICATE_FILE on the next boot)"
2587 );
2588 Ok(())
2589}
2590
2591fn apply_production_preset_from_config(
2592 auth_store: &Arc<AuthStore>,
2593 bootstrap: &BootstrapConfig,
2594) -> Result<(String, Option<String>), String> {
2595 use crate::auth::store::PrincipalRef;
2596 use crate::auth::UserId;
2597
2598 let username = bootstrap.production_username().ok_or_else(|| {
2599 "production preset requires --bootstrap-admin or REDDB_USERNAME (or REDDB_USERNAME_FILE)"
2600 .to_string()
2601 })?;
2602 let password = bootstrap.production_password().ok_or_else(|| {
2603 "production preset requires --bootstrap-admin-password or REDDB_PASSWORD (or REDDB_PASSWORD_FILE)"
2604 .to_string()
2605 })?;
2606
2607 let result = auth_store
2611 .bootstrap(&username, &password)
2612 .map_err(|err| format!("bootstrap first admin: {err}"))?;
2613 let certificate = result.certificate.clone();
2614 if let Some(cert) = certificate.as_deref() {
2615 eprintln!("[reddb] CERTIFICATE: {}", cert);
2616 tracing::warn!(
2617 "vault certificate issued by REDDB_PRESET=production -- save it and update the runtime secret before restart"
2618 );
2619 }
2620 let first_admin = UserId::platform(result.user.username.clone());
2621
2622 install_allow_all_policy(auth_store)?;
2624
2625 auth_store
2627 .attach_policy(
2628 PrincipalRef::User(first_admin.clone()),
2629 FIRST_ADMIN_ALLOW_ALL_POLICY,
2630 )
2631 .map_err(|err| format!("attach allow-all policy: {err}"))?;
2632
2633 Ok((first_admin.to_string(), certificate))
2634}
2635
2636fn apply_cloud_preset(
2637 runtime: &RedDBRuntime,
2638 auth_store: &Arc<AuthStore>,
2639 bootstrap: &BootstrapConfig,
2640) -> Result<(String, Option<String>), String> {
2641 use crate::auth::store::PrincipalRef;
2642 use crate::auth::{Role, UserId};
2643
2644 let head_username = bootstrap.cloud_head_username().ok_or_else(|| {
2645 "cloud preset requires --cloud-head-admin or REDDB_CLOUD_HEAD_ADMIN (or REDDB_USERNAME)"
2646 .to_string()
2647 })?;
2648 let head_password = bootstrap.cloud_head_password().ok_or_else(|| {
2649 "cloud preset requires --cloud-head-admin-password or REDDB_CLOUD_HEAD_ADMIN_PASSWORD (or REDDB_PASSWORD)"
2650 .to_string()
2651 })?;
2652 let customer_username = bootstrap.customer_username().ok_or_else(|| {
2653 "cloud preset requires --customer-admin or REDDB_CUSTOMER_ADMIN".to_string()
2654 })?;
2655 let customer_password = bootstrap.customer_password().ok_or_else(|| {
2656 "cloud preset requires --customer-admin-password or REDDB_CUSTOMER_ADMIN_PASSWORD"
2657 .to_string()
2658 })?;
2659 if head_username == customer_username {
2660 return Err("cloud preset requires distinct head and customer admin usernames".to_string());
2661 }
2662
2663 let head = auth_store
2664 .bootstrap_system_admin(&head_username, &head_password)
2665 .map_err(|err| format!("bootstrap cloud head admin: {err}"))?;
2666 let certificate = head.certificate.clone();
2667 let head_id = UserId::platform(head.user.username.clone());
2668
2669 auth_store
2670 .create_user(&customer_username, &customer_password, Role::Admin)
2671 .map_err(|err| format!("create cloud customer admin: {err}"))?;
2672 let customer_id = UserId::platform(customer_username.clone());
2673
2674 install_allow_all_policy(auth_store)?;
2675 for user_id in [&head_id, &customer_id] {
2676 auth_store
2677 .attach_policy(
2678 PrincipalRef::User(user_id.clone()),
2679 FIRST_ADMIN_ALLOW_ALL_POLICY,
2680 )
2681 .map_err(|err| format!("attach allow-all policy: {err}"))?;
2682 }
2683 install_cloud_guardrails(runtime, auth_store, &head_id)?;
2684 auth_store
2685 .attach_policy(
2686 PrincipalRef::User(customer_id),
2687 CLOUD_PROTECT_MANAGED_POLICY,
2688 )
2689 .map_err(|err| format!("attach cloud guardrail policy: {err}"))?;
2690
2691 Ok((head_id.to_string(), certificate))
2692}
2693
2694fn install_allow_all_policy(auth_store: &Arc<AuthStore>) -> Result<(), String> {
2695 use crate::auth::policies::Policy;
2696
2697 let policy = Policy::from_json_str(&format!(
2698 r#"{{
2699 "id": "{id}",
2700 "version": 1,
2701 "statements": [{{
2702 "effect": "allow",
2703 "actions": ["*"],
2704 "resources": ["*"]
2705 }}]
2706 }}"#,
2707 id = FIRST_ADMIN_ALLOW_ALL_POLICY
2708 ))
2709 .map_err(|err| format!("compile allow-all policy: {err}"))?;
2710 auth_store
2711 .put_policy(policy)
2712 .map_err(|err| format!("install allow-all policy: {err}"))
2713}
2714
2715fn install_cloud_guardrails(
2716 runtime: &RedDBRuntime,
2717 auth_store: &Arc<AuthStore>,
2718 head_admin: &crate::auth::UserId,
2719) -> Result<(), String> {
2720 use crate::auth::policies::Policy;
2721 use crate::auth::registry::EvidenceRequirement;
2722
2723 let policy = Policy::from_json_str(&format!(
2724 r#"{{
2725 "id": "{id}",
2726 "version": 1,
2727 "statements": [
2728 {{
2729 "effect": "deny",
2730 "actions": ["policy:put", "policy:drop", "policy:attach", "policy:detach"],
2731 "resources": ["policy:{id}"]
2732 }},
2733 {{
2734 "effect": "deny",
2735 "actions": [
2736 "user:delete",
2737 "user:disable",
2738 "user:password:change",
2739 "user:role:update"
2740 ],
2741 "resources": ["user:{head_admin}"]
2742 }},
2743 {{
2744 "effect": "deny",
2745 "actions": ["config:write"],
2746 "resources": ["config:{cloud}.*"]
2747 }}
2748 ]
2749 }}"#,
2750 id = CLOUD_PROTECT_MANAGED_POLICY,
2751 head_admin = head_admin,
2752 cloud = CLOUD_CONFIG_NAMESPACE,
2753 ))
2754 .map_err(|err| format!("compile cloud guardrail policy: {err}"))?;
2755 auth_store
2756 .put_policy(policy)
2757 .map_err(|err| format!("install cloud guardrail policy: {err}"))?;
2758
2759 let now_ms = crate::utils::now_unix_millis() as u128;
2760 let entries = vec![
2761 regulated_registry_entry(
2762 CLOUD_PROTECT_MANAGED_POLICY,
2763 crate::auth::managed_policy::RESOURCE_TYPE_POLICY,
2764 "policy",
2765 "policy:*",
2766 &format!("policy:{CLOUD_PROTECT_MANAGED_POLICY}"),
2767 EvidenceRequirement::Metadata,
2768 now_ms,
2769 ),
2770 regulated_registry_entry(
2771 CLOUD_CONFIG_NAMESPACE,
2772 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2773 "config_namespace",
2774 "config:write",
2775 &format!("config:{CLOUD_CONFIG_NAMESPACE}.*"),
2776 EvidenceRequirement::Metadata,
2777 now_ms,
2778 ),
2779 ];
2780 for entry in entries.iter().cloned() {
2781 runtime
2782 .config_registry()
2783 .restore_bootstrap_entry(entry)
2784 .map_err(|err| format!("install cloud registry entry: {err}"))?;
2785 }
2786 crate::cli::bootstrap_manifest::persist_registry_state(runtime, &entries)?;
2787 Ok(())
2788}
2789
2790fn apply_regulated_preset(
2791 runtime: &RedDBRuntime,
2792 auth_store: &Arc<AuthStore>,
2793) -> Result<(), String> {
2794 use crate::auth::policies::Policy;
2795 use crate::auth::registry::EvidenceRequirement;
2796 use crate::auth::store::{PrincipalRef, PUBLIC_IAM_GROUP};
2797
2798 runtime.query_audit().enable_infrastructure();
2799
2800 let policy = Policy::from_json_str(&format!(
2801 r#"{{
2802 "id": "{id}",
2803 "version": 1,
2804 "statements": [
2805 {{
2806 "effect": "deny",
2807 "actions": ["policy:put", "policy:drop", "policy:attach", "policy:detach"],
2808 "resources": ["policy:{id}"]
2809 }},
2810 {{
2811 "effect": "deny",
2812 "actions": ["config:write"],
2813 "resources": [
2814 "config:{audit}.*",
2815 "config:{evidence}.*",
2816 "config:{query_audit}.*"
2817 ]
2818 }}
2819 ]
2820 }}"#,
2821 id = REGULATED_PROTECT_MANAGED_POLICY,
2822 audit = REGULATED_AUDIT_CONFIG_NAMESPACE,
2823 evidence = REGULATED_EVIDENCE_CONFIG_NAMESPACE,
2824 query_audit = REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE,
2825 ))
2826 .map_err(|err| format!("compile regulated guardrail policy: {err}"))?;
2827 auth_store
2828 .put_policy(policy)
2829 .map_err(|err| format!("install regulated guardrail policy: {err}"))?;
2830 auth_store
2831 .attach_policy(
2832 PrincipalRef::Group(PUBLIC_IAM_GROUP.to_string()),
2833 REGULATED_PROTECT_MANAGED_POLICY,
2834 )
2835 .map_err(|err| format!("attach regulated guardrail policy: {err}"))?;
2836
2837 let now_ms = crate::utils::now_unix_millis() as u128;
2838 let entries = vec![
2839 regulated_registry_entry(
2840 REGULATED_PROTECT_MANAGED_POLICY,
2841 crate::auth::managed_policy::RESOURCE_TYPE_POLICY,
2842 "iam_policy",
2843 "policy:*",
2844 &format!("policy:{REGULATED_PROTECT_MANAGED_POLICY}"),
2845 EvidenceRequirement::Metadata,
2846 now_ms,
2847 ),
2848 regulated_registry_entry(
2849 REGULATED_AUDIT_CONFIG_NAMESPACE,
2850 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2851 "config_namespace",
2852 "config:write",
2853 &format!("config:{REGULATED_AUDIT_CONFIG_NAMESPACE}.*"),
2854 EvidenceRequirement::Metadata,
2855 now_ms,
2856 ),
2857 regulated_registry_entry(
2858 REGULATED_EVIDENCE_CONFIG_NAMESPACE,
2859 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2860 "config_namespace",
2861 "config:write",
2862 &format!("config:{REGULATED_EVIDENCE_CONFIG_NAMESPACE}.*"),
2863 EvidenceRequirement::Metadata,
2864 now_ms,
2865 ),
2866 regulated_registry_entry(
2867 REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE,
2868 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2869 "config_namespace",
2870 "config:write",
2871 &format!("config:{REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE}.*"),
2872 EvidenceRequirement::Metadata,
2873 now_ms,
2874 ),
2875 ];
2876
2877 for entry in entries.iter().cloned() {
2878 runtime
2879 .config_registry()
2880 .restore_bootstrap_entry(entry)
2881 .map_err(|err| format!("install regulated registry entry: {err}"))?;
2882 }
2883 crate::cli::bootstrap_manifest::persist_registry_state(runtime, &entries)?;
2884 Ok(())
2885}
2886
2887fn regulated_registry_entry(
2888 id: &str,
2889 resource_type: &str,
2890 schema: &str,
2891 required_action: &str,
2892 required_resource: &str,
2893 evidence_requirement: crate::auth::registry::EvidenceRequirement,
2894 updated_at_ms: u128,
2895) -> crate::auth::registry::ConfigRegistryEntry {
2896 crate::auth::registry::ConfigRegistryEntry {
2897 id: id.to_string(),
2898 version: 1,
2899 resource_type: resource_type.to_string(),
2900 schema: schema.to_string(),
2901 mutability: crate::auth::registry::Mutability::Immutable,
2902 sensitivity: crate::auth::registry::Sensitivity::Internal,
2903 managed: true,
2904 required_action: required_action.to_string(),
2905 required_resource: required_resource.to_string(),
2906 evidence_requirement,
2907 updated_by: "system:regulated-preset".to_string(),
2908 updated_at_ms,
2909 }
2910}
2911
2912fn persist_bootstrap_state(runtime: &RedDBRuntime, preset: &str, first_admin_id: Option<&str>) {
2913 let store = runtime.db().store();
2914 let mut tree = crate::serde_json::Map::new();
2915 tree.insert(
2916 BOOTSTRAP_COMPLETED_KEY.to_string(),
2917 crate::serde_json::Value::Bool(true),
2918 );
2919 tree.insert(
2920 BOOTSTRAP_PRESET_KEY.to_string(),
2921 crate::serde_json::Value::String(preset.to_string()),
2922 );
2923 if let Some(id) = first_admin_id {
2924 tree.insert(
2925 BOOTSTRAP_FIRST_ADMIN_KEY.to_string(),
2926 crate::serde_json::Value::String(id.to_string()),
2927 );
2928 }
2929 let json = crate::serde_json::Value::Object(tree);
2930 store.set_config_tree("", &json);
2931}
2932
2933fn merge_telemetry_with_config(
2944 mut cli: crate::telemetry::TelemetryConfig,
2945 runtime: &RedDBRuntime,
2946) -> crate::telemetry::TelemetryConfig {
2947 use crate::storage::schema::Value;
2948
2949 let store = runtime.db().store();
2950
2951 if !cli.level_explicit {
2952 if let Some(Value::Text(v)) = store.get_config("red.logging.level") {
2953 cli.level_filter = v.to_string();
2954 }
2955 }
2956 if !cli.format_explicit {
2957 if let Some(Value::Text(v)) = store.get_config("red.logging.format") {
2958 if let Some(parsed) = crate::telemetry::LogFormat::parse(&v) {
2959 cli.format = parsed;
2960 }
2961 }
2962 }
2963 if !cli.rotation_keep_days_explicit {
2964 match store.get_config("red.logging.keep_days") {
2965 Some(Value::Integer(n)) if n >= 0 && n <= u16::MAX as i64 => {
2966 cli.rotation_keep_days = n as u16
2967 }
2968 Some(Value::UnsignedInteger(n)) if n <= u16::MAX as u64 => {
2969 cli.rotation_keep_days = n as u16
2970 }
2971 Some(Value::Text(v)) => {
2972 if let Ok(n) = v.parse::<u16>() {
2973 cli.rotation_keep_days = n;
2974 }
2975 }
2976 _ => {}
2977 }
2978 }
2979 if !cli.file_prefix_explicit {
2980 if let Some(Value::Text(v)) = store.get_config("red.logging.file_prefix") {
2981 if !v.is_empty() {
2982 cli.file_prefix = v.to_string();
2983 }
2984 }
2985 }
2986 if !cli.log_dir_explicit && !cli.log_file_disabled {
2989 if let Some(Value::Text(v)) = store.get_config("red.logging.dir") {
2990 if !v.is_empty() {
2991 cli.log_dir = Some(std::path::PathBuf::from(v.as_ref()));
2992 }
2993 }
2994 }
2995
2996 cli
2997}
2998
2999#[cfg(test)]
3000mod telemetry_merge_tests {
3001 use super::*;
3002 use crate::telemetry::{LogFormat, TelemetryConfig};
3003
3004 fn fresh_runtime() -> RedDBRuntime {
3005 RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime")
3006 }
3007
3008 fn set_str(runtime: &RedDBRuntime, key: &str, value: &str) {
3009 runtime
3010 .db()
3011 .store()
3012 .set_config_tree(key, &crate::serde_json::Value::String(value.to_string()));
3013 }
3014
3015 fn cli_base() -> TelemetryConfig {
3016 TelemetryConfig {
3019 log_dir: Some(std::path::PathBuf::from("/tmp/reddb-default/logs")),
3020 format: LogFormat::Json,
3021 ..Default::default()
3022 }
3023 }
3024
3025 #[test]
3026 fn config_log_dir_promoted_when_flag_absent() {
3027 let runtime = fresh_runtime();
3028 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
3029 let merged = merge_telemetry_with_config(cli_base(), &runtime);
3030 assert_eq!(
3031 merged.log_dir.as_deref(),
3032 Some(std::path::Path::new("/var/log/reddb"))
3033 );
3034 }
3035
3036 #[test]
3037 fn explicit_log_dir_wins_over_config() {
3038 let runtime = fresh_runtime();
3039 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
3040 let mut cli = cli_base();
3041 cli.log_dir = Some(std::path::PathBuf::from("/custom/dir"));
3042 cli.log_dir_explicit = true;
3043 let merged = merge_telemetry_with_config(cli, &runtime);
3044 assert_eq!(
3045 merged.log_dir.as_deref(),
3046 Some(std::path::Path::new("/custom/dir"))
3047 );
3048 }
3049
3050 #[test]
3051 fn no_log_file_beats_config_log_dir() {
3052 let runtime = fresh_runtime();
3053 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
3054 let mut cli = cli_base();
3055 cli.log_dir = None;
3056 cli.log_file_disabled = true;
3057 let merged = merge_telemetry_with_config(cli, &runtime);
3058 assert!(
3059 merged.log_dir.is_none(),
3060 "--no-log-file must veto config dir"
3061 );
3062 }
3063
3064 #[test]
3065 fn config_format_promoted_on_non_tty_default() {
3066 let runtime = fresh_runtime();
3070 set_str(&runtime, "red.logging.format", "pretty");
3071 let merged = merge_telemetry_with_config(cli_base(), &runtime);
3072 assert_eq!(merged.format, LogFormat::Pretty);
3073 }
3074
3075 #[test]
3076 fn explicit_format_wins_over_config() {
3077 let runtime = fresh_runtime();
3078 set_str(&runtime, "red.logging.format", "pretty");
3079 let mut cli = cli_base();
3080 cli.format = LogFormat::Json;
3081 cli.format_explicit = true;
3082 let merged = merge_telemetry_with_config(cli, &runtime);
3083 assert_eq!(merged.format, LogFormat::Json);
3084 }
3085}
3086
3087#[inline(never)]
3088fn build_http_server(
3089 runtime: RedDBRuntime,
3090 auth_store: Arc<AuthStore>,
3091 bind_addr: String,
3092) -> RedDBServer {
3093 build_http_server_with_transport_readiness(
3094 runtime,
3095 auth_store,
3096 bind_addr,
3097 TransportReadiness::default(),
3098 )
3099}
3100
3101fn apply_http_limits(
3107 server: RedDBServer,
3108 config: &ServerCommandConfig,
3109 runtime: &RedDBRuntime,
3110) -> RedDBServer {
3111 let store = runtime.db().store();
3112 let resolved =
3113 crate::server::http_limits::resolve_http_limits(&config.http_limits_cli, |key| match store
3114 .get_config(key)
3115 {
3116 Some(crate::storage::schema::Value::Text(v)) => Some(v.to_string()),
3117 Some(crate::storage::schema::Value::Integer(n)) if n >= 0 => Some(n.to_string()),
3118 Some(crate::storage::schema::Value::UnsignedInteger(n)) => Some(n.to_string()),
3119 _ => None,
3120 });
3121 tracing::info!(
3122 target: "reddb::http_limits",
3123 max_handlers = resolved.max_handlers,
3124 handler_timeout_ms = resolved.handler_timeout_ms,
3125 retry_after_secs = resolved.retry_after_secs,
3126 max_inflight_per_principal = resolved.max_inflight_per_principal,
3127 "http_limits resolved"
3128 );
3129 server.with_http_limits(resolved)
3130}
3131
3132fn apply_ui_bundle(
3142 server: RedDBServer,
3143 config: &ServerCommandConfig,
3144) -> Result<RedDBServer, String> {
3145 if !config.ui {
3146 return Ok(server);
3147 }
3148 let ui_dir = match &config.ui_dir {
3149 Some(dir) => dir.clone(),
3150 None => {
3151 let cache_root = crate::server::ui_bundle_resolver::reddb_user_cache_root()
3152 .unwrap_or_else(|_| std::env::temp_dir().join("reddb"));
3153 crate::server::ui_bundle_resolver::resolve_ui_bundle(
3154 &cache_root,
3155 &crate::server::ui_bundle_resolver::HttpFetcher,
3156 )
3157 .map_err(|err| format!("resolve red-ui bundle for --ui: {err}"))?
3158 }
3159 };
3160 tracing::info!(target: "reddb::ui", dir = %ui_dir.display(), "serving red-ui bundle on HTTP surface");
3161 Ok(server.with_ui_dir(ui_dir))
3162}
3163
3164#[inline(never)]
3165fn build_http_server_with_transport_readiness(
3166 runtime: RedDBRuntime,
3167 auth_store: Arc<AuthStore>,
3168 bind_addr: String,
3169 transport_readiness: TransportReadiness,
3170) -> RedDBServer {
3171 RedDBServer::with_options(
3172 runtime,
3173 ServerOptions {
3174 bind_addr,
3175 transport_readiness,
3176 ..ServerOptions::default()
3177 },
3178 )
3179 .with_auth(auth_store)
3180}
3181
3182#[inline(never)]
3186fn build_admin_only_server(
3187 runtime: RedDBRuntime,
3188 auth_store: Arc<AuthStore>,
3189 bind_addr: String,
3190) -> RedDBServer {
3191 RedDBServer::with_options(
3192 runtime,
3193 ServerOptions {
3194 bind_addr,
3195 surface: crate::server::ServerSurface::AdminOnly,
3196 ..ServerOptions::default()
3197 },
3198 )
3199 .with_auth(auth_store)
3200}
3201
3202#[inline(never)]
3206fn build_metrics_only_server(
3207 runtime: RedDBRuntime,
3208 auth_store: Arc<AuthStore>,
3209 bind_addr: String,
3210) -> RedDBServer {
3211 RedDBServer::with_options(
3212 runtime,
3213 ServerOptions {
3214 bind_addr,
3215 surface: crate::server::ServerSurface::MetricsOnly,
3216 ..ServerOptions::default()
3217 },
3218 )
3219 .with_auth(auth_store)
3220}
3221
3222fn spawn_admin_metrics_listeners(runtime: &RedDBRuntime, auth_store: &Arc<AuthStore>) {
3226 if let Some(addr) = env_nonempty("RED_ADMIN_BIND") {
3227 let server = build_admin_only_server(runtime.clone(), auth_store.clone(), addr.clone());
3228 let _ = server.serve_in_background();
3229 tracing::info!(transport = "http", surface = "admin", bind = %addr, "listener online");
3230 }
3231 if let Some(addr) = env_nonempty("RED_METRICS_BIND") {
3232 let server = build_metrics_only_server(runtime.clone(), auth_store.clone(), addr.clone());
3233 let _ = server.serve_in_background();
3234 tracing::info!(transport = "http", surface = "metrics", bind = %addr, "listener online");
3235 }
3236}
3237
3238#[inline(never)]
3239fn run_http_server(config: ServerCommandConfig, bind_addr: String) -> Result<(), String> {
3240 let mut transport_readiness = TransportReadiness::default();
3241 let Some(listener) = bind_listener_for_startup(
3242 &mut transport_readiness,
3243 "http",
3244 &bind_addr,
3245 config.http_bind_explicit,
3246 )?
3247 else {
3248 return Err(format!(
3249 "no HTTP listener started; implicit bind {} failed",
3250 bind_addr
3251 ));
3252 };
3253 let db_options = config.to_db_options()?;
3254 let (runtime, auth_store, _telemetry_guard) =
3255 build_runtime_and_auth_store(&config, db_options.clone())?;
3256 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3257 spawn_admin_metrics_listeners(&runtime, &auth_store);
3258 spawn_http_tls_listener(&config, &runtime, &auth_store)?;
3259 let server = build_http_server_with_transport_readiness(
3260 runtime.clone(),
3261 auth_store,
3262 bind_addr.clone(),
3263 transport_readiness,
3264 );
3265 let server = apply_http_limits(server, &config, &runtime);
3266 let server = apply_ui_bundle(server, &config)?;
3267 tracing::info!(transport = "http", bind = %bind_addr, "listener online");
3268 server.serve_on(listener).map_err(|err| err.to_string())
3269}
3270
3271fn spawn_http_tls_listener(
3277 config: &ServerCommandConfig,
3278 runtime: &RedDBRuntime,
3279 auth_store: &Arc<AuthStore>,
3280) -> Result<(), String> {
3281 let Some(addr) = config.http_tls_bind_addr.clone() else {
3282 return Ok(());
3283 };
3284
3285 let tls_config = resolve_http_tls_config(config)?;
3286 let server_config = crate::server::tls::build_server_config(&tls_config)
3287 .map_err(|err| format!("HTTP TLS: {err}"))?;
3288
3289 let server = build_http_server(runtime.clone(), auth_store.clone(), addr.clone());
3290 let server = apply_http_limits(server, config, runtime);
3291 let _handle = server.serve_tls_in_background(server_config);
3292 tracing::info!(
3293 transport = "https",
3294 bind = %addr,
3295 mtls = %tls_config.client_ca_path.is_some(),
3296 "TLS listener online"
3297 );
3298 Ok(())
3299}
3300
3301fn resolve_http_tls_config(
3303 config: &ServerCommandConfig,
3304) -> Result<crate::server::tls::HttpTlsConfig, String> {
3305 match (&config.http_tls_cert, &config.http_tls_key) {
3306 (Some(cert), Some(key)) => Ok(crate::server::tls::HttpTlsConfig {
3307 cert_path: cert.clone(),
3308 key_path: key.clone(),
3309 client_ca_path: config.http_tls_client_ca.clone(),
3310 }),
3311 (None, None) => {
3312 let dir = config
3314 .path
3315 .as_ref()
3316 .and_then(|p| p.parent().map(std::path::PathBuf::from))
3317 .unwrap_or_else(|| std::path::PathBuf::from("."));
3318 let auto = crate::server::tls::auto_generate_dev_cert(&dir)
3319 .map_err(|err| format!("HTTP TLS dev: {err}"))?;
3320 Ok(crate::server::tls::HttpTlsConfig {
3321 cert_path: auto.cert_path,
3322 key_path: auto.key_path,
3323 client_ca_path: config.http_tls_client_ca.clone(),
3324 })
3325 }
3326 _ => Err("HTTP TLS requires both --http-tls-cert and --http-tls-key (or neither, with RED_HTTP_TLS_DEV=1)".to_string()),
3327 }
3328}
3329
3330#[inline(never)]
3331fn run_grpc_server(config: ServerCommandConfig, bind_addr: String) -> Result<(), String> {
3332 let workers = config.workers;
3333 let db_options = config.to_db_options()?;
3334 let rt_config = detect_runtime_config();
3335 let mut transport_readiness = TransportReadiness::default();
3336 let Some(grpc_listener) = bind_listener_for_startup(
3337 &mut transport_readiness,
3338 "grpc",
3339 &bind_addr,
3340 config.grpc_bind_explicit,
3341 )?
3342 else {
3343 return Err(format!(
3344 "no gRPC listener started; implicit bind {} failed",
3345 bind_addr
3346 ));
3347 };
3348
3349 let worker_threads = workers.unwrap_or(rt_config.suggested_workers);
3350
3351 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
3352 .enable_all()
3353 .worker_threads(worker_threads)
3354 .thread_stack_size(rt_config.stack_size)
3355 .build()
3356 .map_err(|err| format!("tokio runtime: {err}"))?;
3357
3358 let (runtime, auth_store, _telemetry_guard) =
3360 build_runtime_and_auth_store(&config, db_options.clone())?;
3361 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3362 let signal_runtime = runtime.clone();
3363 tokio_runtime.block_on(async move {
3364 spawn_lifecycle_signal_handler(signal_runtime).await;
3365 spawn_wire_listeners(&config, &runtime, &mut transport_readiness).await?;
3367
3368 spawn_pg_listener(&config, &runtime);
3370
3371 spawn_grpc_tls_listener_if_configured(&config, runtime.clone(), auth_store.clone());
3375
3376 let server = RedDBGrpcServer::with_options(
3377 runtime,
3378 GrpcServerOptions {
3379 bind_addr: bind_addr.clone(),
3380 tls: None,
3381 },
3382 auth_store,
3383 );
3384
3385 tracing::info!(
3386 transport = "grpc",
3387 bind = %bind_addr,
3388 cpus = rt_config.available_cpus,
3389 workers = worker_threads,
3390 "listener online"
3391 );
3392 server
3393 .serve_on(grpc_listener)
3394 .await
3395 .map_err(|err| err.to_string())
3396 })
3397}
3398
3399#[inline(never)]
3400fn run_dual_server(
3401 config: ServerCommandConfig,
3402 grpc_bind_addr: String,
3403 http_bind_addr: String,
3404) -> Result<(), String> {
3405 let workers = config.workers;
3406 let db_options = config.to_db_options()?;
3407 let rt_config = detect_runtime_config();
3408 let worker_threads = workers.unwrap_or(rt_config.suggested_workers);
3409 let mut transport_readiness = TransportReadiness::default();
3410 let http_listener = bind_listener_for_startup(
3411 &mut transport_readiness,
3412 "http",
3413 &http_bind_addr,
3414 config.http_bind_explicit,
3415 )?;
3416 let grpc_listener = bind_listener_for_startup(
3417 &mut transport_readiness,
3418 "grpc",
3419 &grpc_bind_addr,
3420 config.grpc_bind_explicit,
3421 )?;
3422 if http_listener.is_none() && grpc_listener.is_none() {
3423 return Err("no listener started; implicit HTTP and gRPC binds failed".to_string());
3424 }
3425 let (runtime, auth_store, _telemetry_guard) =
3426 build_runtime_and_auth_store(&config, db_options.clone())?;
3427 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3428
3429 spawn_admin_metrics_listeners(&runtime, &auth_store);
3430 spawn_http_tls_listener(&config, &runtime, &auth_store)?;
3431
3432 let http_handle = if let Some(listener) = http_listener {
3433 let http_server = build_http_server_with_transport_readiness(
3434 runtime.clone(),
3435 auth_store.clone(),
3436 http_bind_addr.clone(),
3437 transport_readiness.clone(),
3438 );
3439 let http_server = apply_http_limits(http_server, &config, &runtime);
3440 let http_server = apply_ui_bundle(http_server, &config)?;
3441 Some(http_server.serve_in_background_on(listener))
3442 } else {
3443 None
3444 };
3445
3446 thread::sleep(Duration::from_millis(150));
3447 if let Some(handle) = http_handle.as_ref() {
3448 if handle.is_finished() {
3449 let handle = http_handle.unwrap();
3450 return match handle.join() {
3451 Ok(Ok(())) => Err("HTTP server exited unexpectedly".to_string()),
3452 Ok(Err(err)) => Err(err.to_string()),
3453 Err(_) => Err("HTTP server thread panicked".to_string()),
3454 };
3455 }
3456 }
3457 if grpc_listener.is_none() {
3458 let Some(handle) = http_handle else {
3459 return Err("no listener started".to_string());
3460 };
3461 return match handle.join() {
3462 Ok(Ok(())) => Err("HTTP server exited unexpectedly".to_string()),
3463 Ok(Err(err)) => Err(err.to_string()),
3464 Err(_) => Err("HTTP server thread panicked".to_string()),
3465 };
3466 }
3467 let grpc_listener = grpc_listener.expect("checked above");
3468
3469 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
3470 .enable_all()
3471 .worker_threads(worker_threads)
3472 .thread_stack_size(rt_config.stack_size)
3473 .build()
3474 .map_err(|err| format!("tokio runtime: {err}"))?;
3475
3476 let signal_runtime = runtime.clone();
3477 tokio_runtime.block_on(async move {
3478 spawn_lifecycle_signal_handler(signal_runtime).await;
3479 spawn_wire_listeners(&config, &runtime, &mut transport_readiness).await?;
3481
3482 spawn_pg_listener(&config, &runtime);
3484
3485 spawn_grpc_tls_listener_if_configured(&config, runtime.clone(), auth_store.clone());
3487
3488 let server = RedDBGrpcServer::with_options(
3489 runtime,
3490 GrpcServerOptions {
3491 bind_addr: grpc_bind_addr.clone(),
3492 tls: None,
3493 },
3494 auth_store,
3495 );
3496
3497 tracing::info!(transport = "http", bind = %http_bind_addr, "listener online");
3498 tracing::info!(
3499 transport = "grpc",
3500 bind = %grpc_bind_addr,
3501 cpus = rt_config.available_cpus,
3502 workers = worker_threads,
3503 "listener online"
3504 );
3505 server
3506 .serve_on(grpc_listener)
3507 .await
3508 .map_err(|err| err.to_string())
3509 })
3510}
3511
3512#[cfg(test)]
3513mod tests {
3514 use super::*;
3515
3516 #[test]
3517 fn render_systemd_unit_contains_expected_execstart() {
3518 let config = SystemdServiceConfig {
3519 service_name: "reddb".to_string(),
3520 binary_path: PathBuf::from("/usr/local/bin/red"),
3521 run_user: "reddb".to_string(),
3522 run_group: "reddb".to_string(),
3523 data_path: reddb_file::default_service_database_path(),
3524 router_bind_addr: None,
3525 grpc_bind_addr: Some("0.0.0.0:55055".to_string()),
3526 http_bind_addr: None,
3527 };
3528
3529 let unit = render_systemd_unit(&config);
3530 assert!(unit.contains("ExecStart=/usr/local/bin/red server --path /var/lib/reddb/data.rdb --grpc-bind 0.0.0.0:55055"));
3531 assert!(unit.contains("ReadWritePaths=/var/lib/reddb"));
3532 }
3533
3534 #[test]
3535 fn systemd_service_config_derives_paths() {
3536 let config = SystemdServiceConfig {
3537 service_name: "reddb-api".to_string(),
3538 binary_path: PathBuf::from("/usr/local/bin/red"),
3539 run_user: "reddb".to_string(),
3540 run_group: "reddb".to_string(),
3541 data_path: PathBuf::from("/srv/reddb/live/data.rdb"),
3542 router_bind_addr: None,
3543 grpc_bind_addr: None,
3544 http_bind_addr: Some("127.0.0.1:5000".to_string()),
3545 };
3546
3547 assert_eq!(config.data_dir(), PathBuf::from("/srv/reddb/live"));
3548 assert_eq!(
3549 config.unit_path(),
3550 PathBuf::from("/etc/systemd/system/reddb-api.service")
3551 );
3552 }
3553
3554 #[test]
3555 fn render_systemd_unit_supports_dual_transport() {
3556 let config = SystemdServiceConfig {
3557 service_name: "reddb".to_string(),
3558 binary_path: PathBuf::from("/usr/local/bin/red"),
3559 run_user: "reddb".to_string(),
3560 run_group: "reddb".to_string(),
3561 data_path: reddb_file::default_service_database_path(),
3562 router_bind_addr: None,
3563 grpc_bind_addr: Some("0.0.0.0:55055".to_string()),
3564 http_bind_addr: Some("0.0.0.0:5000".to_string()),
3565 };
3566
3567 let unit = render_systemd_unit(&config);
3568 assert!(unit.contains("--grpc-bind 0.0.0.0:55055"));
3569 assert!(unit.contains("--http-bind 0.0.0.0:5000"));
3570 }
3571
3572 #[test]
3573 fn render_systemd_unit_supports_router_mode() {
3574 let config = SystemdServiceConfig {
3575 service_name: "reddb".to_string(),
3576 binary_path: PathBuf::from("/usr/local/bin/red"),
3577 run_user: "reddb".to_string(),
3578 run_group: "reddb".to_string(),
3579 data_path: reddb_file::default_service_database_path(),
3580 router_bind_addr: Some(DEFAULT_ROUTER_BIND_ADDR.to_string()),
3581 grpc_bind_addr: None,
3582 http_bind_addr: None,
3583 };
3584
3585 let unit = render_systemd_unit(&config);
3586 assert!(unit.contains("--bind 127.0.0.1:5050"));
3587 assert!(!unit.contains("--grpc-bind"));
3588 assert!(!unit.contains("--http-bind"));
3589 }
3590
3591 #[test]
3592 fn explicit_bind_collision_is_fatal() {
3593 let held = TcpListener::bind("127.0.0.1:0").expect("hold test port");
3594 let addr = held.local_addr().expect("held addr").to_string();
3595 let mut readiness = TransportReadiness::default();
3596
3597 let error = bind_listener_for_startup(&mut readiness, "http", &addr, true).unwrap_err();
3598
3599 assert!(error.contains("explicit http listener bind"));
3600 assert_eq!(readiness.active.len(), 0);
3601 assert_eq!(readiness.failed.len(), 1);
3602 assert!(readiness.failed[0].explicit);
3603 assert_eq!(readiness.failed[0].bind_addr, addr);
3604 }
3605
3606 fn no_auth_env_lock() -> &'static std::sync::Mutex<()> {
3613 static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
3614 LOCK.get_or_init(|| std::sync::Mutex::new(()))
3615 }
3616
3617 fn no_auth_test_config(no_auth: bool) -> ServerCommandConfig {
3618 ServerCommandConfig {
3619 path: None,
3620 router_bind_addr: Some(DEFAULT_ROUTER_BIND_ADDR.to_string()),
3621 router_bind_explicit: false,
3622 grpc_bind_addr: None,
3623 grpc_bind_explicit: false,
3624 grpc_tls_bind_addr: None,
3625 grpc_tls_cert: None,
3626 grpc_tls_key: None,
3627 grpc_tls_client_ca: None,
3628 http_bind_addr: None,
3629 http_bind_explicit: false,
3630 http_tls_bind_addr: None,
3631 http_tls_cert: None,
3632 http_tls_key: None,
3633 http_tls_client_ca: None,
3634 wire_bind_addr: None,
3635 wire_bind_explicit: false,
3636 wire_tls_bind_addr: None,
3637 wire_tls_cert: None,
3638 wire_tls_key: None,
3639 pg_bind_addr: None,
3640 create_if_missing: true,
3641 read_only: false,
3642 role: "standalone".to_string(),
3643 primary_addr: None,
3644 storage_profile: StorageProfileSelection::embedded_single_file(),
3645 auth: false,
3646 require_auth: false,
3647 vault: true,
3650 no_auth,
3651 workers: None,
3652 telemetry: None,
3653 http_limits_cli: crate::server::HttpLimitsCliInput::default(),
3654 ui: false,
3655 ui_dir: None,
3656 bootstrap: BootstrapConfig::default(),
3657 }
3658 }
3659
3660 #[test]
3661 fn no_auth_flag_disables_every_auth_knob_and_stamps_metadata() {
3662 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3663 unsafe {
3668 std::env::set_var("REDDB_USERNAME", "admin");
3669 std::env::set_var("REDDB_PASSWORD", "hunter2");
3670 }
3671 let config = no_auth_test_config(true);
3672 let options = config.to_db_options().expect("to_db_options");
3673
3674 assert!(no_auth_active(&options), "metadata should be stamped");
3675 assert!(!options.auth.enabled, "auth.enabled must be forced off");
3676 assert!(
3677 !options.auth.require_auth,
3678 "require_auth must be forced off"
3679 );
3680 assert!(
3681 !options.auth.vault_enabled,
3682 "vault_enabled must be forced off (overrides --vault)"
3683 );
3684 assert_eq!(
3685 options.metadata.get(NO_AUTH_META).map(String::as_str),
3686 Some("true"),
3687 );
3688
3689 unsafe {
3691 std::env::remove_var("REDDB_USERNAME");
3692 std::env::remove_var("REDDB_PASSWORD");
3693 }
3694 }
3695
3696 #[test]
3697 fn default_behaviour_without_no_auth_flag_is_unchanged() {
3698 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3699 let config = no_auth_test_config(false);
3700 let options = config.to_db_options().expect("to_db_options");
3701
3702 assert!(
3703 !no_auth_active(&options),
3704 "default boot must not be marked no-auth"
3705 );
3706 assert!(
3707 options.metadata.get(NO_AUTH_META).is_none(),
3708 "metadata key must be absent when flag is off"
3709 );
3710 assert!(options.auth.vault_enabled);
3712 }
3713
3714 #[test]
3715 fn no_auth_active_blocks_bootstrap_from_env() {
3716 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3717 unsafe {
3722 std::env::set_var("REDDB_USERNAME", "admin");
3723 std::env::set_var("REDDB_PASSWORD", "hunter2");
3724 }
3725
3726 let options = no_auth_test_config(true)
3727 .to_db_options()
3728 .expect("to_db_options");
3729
3730 let auth_store = AuthStore::new(options.auth.clone());
3734 if !no_auth_active(&options) {
3735 auth_store.bootstrap_from_env();
3736 }
3737
3738 assert!(
3739 auth_store.needs_bootstrap(),
3740 "no admin user must be bootstrapped under --no-auth even with REDDB_USERNAME/PASSWORD set"
3741 );
3742
3743 unsafe {
3745 std::env::remove_var("REDDB_USERNAME");
3746 std::env::remove_var("REDDB_PASSWORD");
3747 }
3748 }
3749
3750 fn clear_preset_env() {
3757 unsafe {
3759 std::env::remove_var(BOOTSTRAP_PRESET_ENV);
3760 std::env::remove_var(PRESET_ENV);
3761 std::env::remove_var("REDDB_BOOTSTRAP_MANIFEST");
3762 std::env::remove_var("REDDB_AUTH");
3763 std::env::remove_var("REDDB_REQUIRE_AUTH");
3764 std::env::remove_var("REDDB_NO_AUTH");
3765 std::env::remove_var("REDDB_DEV");
3766 std::env::remove_var("REDDB_VAULT");
3767 std::env::remove_var("REDDB_USERNAME");
3768 std::env::remove_var("REDDB_PASSWORD");
3769 std::env::remove_var("REDDB_USERNAME_FILE");
3770 std::env::remove_var("REDDB_PASSWORD_FILE");
3771 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN");
3772 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD");
3773 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD_FILE");
3774 std::env::remove_var("REDDB_CUSTOMER_ADMIN");
3775 std::env::remove_var("REDDB_CUSTOMER_ADMIN_PASSWORD");
3776 std::env::remove_var("REDDB_CUSTOMER_ADMIN_PASSWORD_FILE");
3777 }
3778 }
3779
3780 fn clear_backup_env() {
3781 unsafe {
3783 std::env::remove_var("REDDB_BACKUP_S3_ENDPOINT");
3784 std::env::remove_var("REDDB_BACKUP_S3_BUCKET");
3785 std::env::remove_var("REDDB_BACKUP_S3_PREFIX");
3786 std::env::remove_var("REDDB_BACKUP_S3_ACCESS_KEY_ID");
3787 std::env::remove_var("REDDB_BACKUP_S3_SECRET_ACCESS_KEY");
3788 std::env::remove_var("REDDB_BACKUP_S3_REGION");
3789 std::env::remove_var("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS");
3790 std::env::remove_var("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS");
3791 std::env::remove_var("REDDB_BACKUP_PAUSE_ON_LAG_SECS");
3792 }
3793 }
3794
3795 fn fresh_runtime_and_store() -> (RedDBRuntime, Arc<AuthStore>) {
3796 let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
3797 let auth_store = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
3798 (runtime, auth_store)
3799 }
3800
3801 #[test]
3802 fn auth_env_knobs_enable_auth_require_auth_and_vault() {
3803 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3804 clear_preset_env();
3805 unsafe {
3807 std::env::set_var("REDDB_AUTH", "true");
3808 std::env::set_var("REDDB_REQUIRE_AUTH", "true");
3809 std::env::set_var("REDDB_VAULT", "true");
3810 }
3811
3812 let mut config = no_auth_test_config(false);
3813 config.vault = false;
3814 let options = config.to_db_options().expect("to_db_options");
3815
3816 assert!(options.auth.enabled);
3817 assert!(options.auth.require_auth);
3818 assert!(options.auth.vault_enabled);
3819
3820 clear_preset_env();
3821 }
3822
3823 #[test]
3824 fn production_and_cloud_presets_force_auth_require_auth_and_vault() {
3825 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3826 clear_preset_env();
3827
3828 for preset in [PRESET_PRODUCTION, PRESET_CLOUD] {
3829 let mut config = no_auth_test_config(false);
3830 config.vault = false;
3831 config.bootstrap.preset = Some(preset.to_string());
3832
3833 let options = config.to_db_options().expect("to_db_options");
3834 assert!(options.auth.enabled, "{preset} should enable auth");
3835 assert!(
3836 options.auth.require_auth,
3837 "{preset} should require authenticated requests"
3838 );
3839 assert!(options.auth.vault_enabled, "{preset} should enable vault");
3840 assert!(!no_auth_active(&options));
3841 }
3842
3843 clear_preset_env();
3844 }
3845
3846 #[test]
3847 fn no_auth_env_overrides_preset_forced_auth() {
3848 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3849 clear_preset_env();
3850 unsafe {
3852 std::env::set_var("REDDB_NO_AUTH", "true");
3853 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_CLOUD);
3854 }
3855
3856 let mut config = no_auth_test_config(false);
3857 config.auth = true;
3858 config.require_auth = true;
3859 let options = config.to_db_options().expect("to_db_options");
3860
3861 assert!(no_auth_active(&options));
3862 assert!(!options.auth.enabled);
3863 assert!(!options.auth.require_auth);
3864 assert!(!options.auth.vault_enabled);
3865
3866 clear_preset_env();
3867 }
3868
3869 #[test]
3870 fn simple_preset_is_default_and_persists_state() {
3871 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3872 clear_preset_env();
3873
3874 let (runtime, auth_store) = fresh_runtime_and_store();
3875 apply_preset(&runtime, &auth_store).expect("simple preset applies cleanly");
3876
3877 assert!(
3879 auth_store.needs_bootstrap(),
3880 "simple preset must not create an admin"
3881 );
3882
3883 let store = runtime.db().store();
3885 let completed = store
3886 .get_config(BOOTSTRAP_COMPLETED_KEY)
3887 .expect("completed key persisted");
3888 assert!(matches!(
3889 completed,
3890 crate::storage::schema::Value::Boolean(true)
3891 ));
3892 let preset = store
3893 .get_config(BOOTSTRAP_PRESET_KEY)
3894 .expect("preset key persisted");
3895 match preset {
3896 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_SIMPLE),
3897 other => panic!("expected Text(simple), got {other:?}"),
3898 }
3899 assert!(
3900 store.get_config(BOOTSTRAP_FIRST_ADMIN_KEY).is_none(),
3901 "simple preset must not record a first admin"
3902 );
3903
3904 clear_preset_env();
3905 }
3906
3907 #[test]
3908 fn production_preset_creates_first_admin_with_allow_all_policy() {
3909 use crate::auth::policies::{EvalContext, ResourceRef};
3910 use crate::auth::UserId;
3911
3912 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3913 clear_preset_env();
3914 unsafe {
3916 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
3917 std::env::set_var("REDDB_USERNAME", "ops");
3918 std::env::set_var("REDDB_PASSWORD", "hunter2");
3919 }
3920
3921 let (runtime, auth_store) = fresh_runtime_and_store();
3922 apply_preset(&runtime, &auth_store).expect("production preset applies cleanly");
3923
3924 assert!(
3926 !auth_store.needs_bootstrap(),
3927 "production preset must seal bootstrap"
3928 );
3929 let users = auth_store.list_users();
3930 assert_eq!(users.len(), 1);
3931 let admin = &users[0];
3932 assert_eq!(admin.username, "ops");
3933 assert!(
3934 admin.tenant_id.is_none(),
3935 "first admin must be platform-scoped (tenant=None)"
3936 );
3937
3938 let policy = auth_store
3940 .get_policy(FIRST_ADMIN_ALLOW_ALL_POLICY)
3941 .expect("allow-all policy installed");
3942 assert!(!policy.statements.is_empty());
3943
3944 let actor = UserId::platform("ops");
3947 let ctx = EvalContext {
3948 principal_tenant: None,
3949 current_tenant: None,
3950 peer_ip: None,
3951 mfa_present: false,
3952 now_ms: 1_700_000_000_000,
3953 principal_is_admin_role: true,
3954 principal_is_platform_scoped: true,
3955 };
3956 let arbitrary_resource = ResourceRef::new("config", "red.config.audit.enabled");
3957 assert!(
3958 auth_store.check_policy_authz(&actor, "config:write", &arbitrary_resource, &ctx),
3959 "allow-all policy must grant arbitrary actions via the evaluator"
3960 );
3961
3962 let store = runtime.db().store();
3964 match store
3965 .get_config(BOOTSTRAP_FIRST_ADMIN_KEY)
3966 .expect("first_admin_id persisted")
3967 {
3968 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), "ops"),
3969 other => panic!("expected Text(ops), got {other:?}"),
3970 }
3971 match store.get_config(BOOTSTRAP_PRESET_KEY).unwrap() {
3972 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_PRODUCTION),
3973 other => panic!("expected Text(production), got {other:?}"),
3974 }
3975
3976 clear_preset_env();
3977 }
3978
3979 #[test]
3980 fn bootstrap_preset_env_takes_precedence_over_legacy_preset_env() {
3981 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3982 clear_preset_env();
3983 unsafe {
3985 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_REGULATED);
3986 std::env::set_var(PRESET_ENV, PRESET_SIMPLE);
3987 }
3988
3989 let options = no_auth_test_config(false)
3990 .to_db_options()
3991 .expect("regulated options");
3992 assert!(
3993 options.control_events.compliance_mode,
3994 "canonical REDDB_BOOTSTRAP_PRESET should win over REDDB_PRESET"
3995 );
3996 assert!(options.query_audit.enabled);
3997
3998 clear_preset_env();
3999 }
4000
4001 #[test]
4002 fn cloud_preset_creates_system_head_and_customer_admins() {
4003 use crate::auth::policies::{EvalContext, ResourceRef};
4004 use crate::auth::UserId;
4005
4006 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4007 clear_preset_env();
4008 unsafe {
4010 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_CLOUD);
4011 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN", "head");
4012 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD", "head-pass");
4013 std::env::set_var("REDDB_CUSTOMER_ADMIN", "customer");
4014 std::env::set_var("REDDB_CUSTOMER_ADMIN_PASSWORD", "customer-pass");
4015 }
4016
4017 let (runtime, auth_store) = fresh_runtime_and_store();
4018 apply_preset(&runtime, &auth_store).expect("cloud preset applies cleanly");
4019
4020 let head = auth_store
4021 .get_user(None, "head")
4022 .expect("head admin should exist");
4023 assert_eq!(head.tenant_id, None);
4024 let customer = auth_store
4025 .get_user(None, "customer")
4026 .expect("customer admin should exist");
4027 assert_eq!(customer.tenant_id, None);
4028
4029 let ctx = EvalContext {
4030 principal_tenant: None,
4031 current_tenant: None,
4032 peer_ip: None,
4033 mfa_present: false,
4034 now_ms: 1_700_000_000_000,
4035 principal_is_admin_role: true,
4036 principal_is_platform_scoped: true,
4037 };
4038 assert!(auth_store.check_policy_authz(
4039 &UserId::platform("customer"),
4040 "config:write",
4041 &ResourceRef::new("config", "red.config.customer.enabled"),
4042 &ctx,
4043 ));
4044 assert!(
4045 auth_store.check_policy_authz(
4046 &UserId::platform("head"),
4047 "config:write",
4048 &ResourceRef::new("config", "red.config.cloud.enabled"),
4049 &ctx,
4050 ),
4051 "cloud head keeps allow-all authority over managed cloud config"
4052 );
4053 assert!(
4054 !auth_store.check_policy_authz(
4055 &UserId::platform("customer"),
4056 "config:write",
4057 &ResourceRef::new("config", "red.config.cloud.enabled"),
4058 &ctx,
4059 ),
4060 "cloud customer must be denied managed cloud config writes"
4061 );
4062 assert!(
4063 !auth_store.check_policy_authz(
4064 &UserId::platform("customer"),
4065 "policy:drop",
4066 &ResourceRef::new("policy", CLOUD_PROTECT_MANAGED_POLICY),
4067 &ctx,
4068 ),
4069 "cloud customer must be denied mutations of the managed guardrail policy"
4070 );
4071
4072 assert!(auth_store.get_user(None, "head").is_some());
4073 assert!(auth_store
4074 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4075 .is_some());
4076 assert!(runtime
4077 .config_registry()
4078 .get_active(CLOUD_CONFIG_NAMESPACE)
4079 .is_some());
4080
4081 let store = runtime.db().store();
4082 match store
4083 .get_config(BOOTSTRAP_FIRST_ADMIN_KEY)
4084 .expect("head admin id persisted")
4085 {
4086 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), "head"),
4087 other => panic!("expected Text(head), got {other:?}"),
4088 }
4089 match store.get_config(BOOTSTRAP_PRESET_KEY).unwrap() {
4090 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_CLOUD),
4091 other => panic!("expected Text(cloud), got {other:?}"),
4092 }
4093
4094 clear_preset_env();
4095 }
4096
4097 #[test]
4098 fn cloud_preset_customer_cannot_disable_head_admin_or_drop_guardrail_policy() {
4099 use crate::auth::Role;
4100 use crate::runtime::mvcc::{clear_current_auth_identity, set_current_auth_identity};
4101
4102 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4103 clear_preset_env();
4104
4105 let bootstrap = BootstrapConfig {
4106 preset: Some(PRESET_CLOUD.to_string()),
4107 cloud_head_admin: Some("red_admin".to_string()),
4108 cloud_head_admin_password: Some("head-pass".to_string()),
4109 customer_admin: Some("admin".to_string()),
4110 customer_admin_password: Some("customer-pass".to_string()),
4111 ..BootstrapConfig::default()
4112 };
4113 let (runtime, auth_store) = fresh_runtime_and_store();
4114 apply_preset_from_config(&runtime, &auth_store, &bootstrap)
4115 .expect("cloud preset applies cleanly");
4116 runtime.set_auth_store(Arc::clone(&auth_store));
4117
4118 set_current_auth_identity("admin".to_string(), Role::Admin);
4119 let disable_head = runtime.execute_query("ALTER USER red_admin DISABLE");
4120 let drop_guardrail =
4121 runtime.execute_query(&format!("DROP POLICY '{}'", CLOUD_PROTECT_MANAGED_POLICY));
4122 let create_user = runtime.execute_query("CREATE USER app_user WITH PASSWORD 'p' ROLE read");
4123 let disable_user = runtime.execute_query("ALTER USER app_user DISABLE");
4124 clear_current_auth_identity();
4125
4126 let disable_head_err = disable_head.expect_err("customer admin must not disable red_admin");
4127 assert!(
4128 disable_head_err.to_string().contains("user:disable"),
4129 "error should name the denied lifecycle action: {disable_head_err}"
4130 );
4131 assert!(
4132 auth_store
4133 .get_user(None, "red_admin")
4134 .expect("red_admin should still exist")
4135 .enabled,
4136 "denied mutation must leave red_admin enabled"
4137 );
4138
4139 let drop_guardrail_err =
4140 drop_guardrail.expect_err("customer admin must not drop cloud guardrail policy");
4141 assert!(
4142 drop_guardrail_err.to_string().contains("managed policy"),
4143 "error should name the managed policy guardrail: {drop_guardrail_err}"
4144 );
4145 assert!(auth_store
4146 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4147 .is_some());
4148
4149 create_user.expect("customer admin should retain normal create-user authority");
4150 disable_user.expect("customer admin should retain normal user-disable authority");
4151 assert!(
4152 !auth_store
4153 .get_user(None, "app_user")
4154 .expect("app_user should exist")
4155 .enabled,
4156 "normal capability should still apply outside the reserved head admin"
4157 );
4158
4159 clear_preset_env();
4160 }
4161
4162 #[test]
4163 fn cloud_preset_cli_admin_alias_wins_over_cloud_env_password() {
4164 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4165 clear_preset_env();
4166 unsafe {
4168 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD", "env-head-pass");
4169 }
4170
4171 let bootstrap = BootstrapConfig {
4172 preset: Some(PRESET_CLOUD.to_string()),
4173 admin_username: Some("head".to_string()),
4174 admin_password: Some("cli-head-pass".to_string()),
4175 customer_admin: Some("customer".to_string()),
4176 customer_admin_password: Some("customer-pass".to_string()),
4177 ..BootstrapConfig::default()
4178 };
4179 let (runtime, auth_store) = fresh_runtime_and_store();
4180 apply_preset_from_config(&runtime, &auth_store, &bootstrap)
4181 .expect("cloud preset applies cleanly");
4182
4183 auth_store
4184 .authenticate("head", "cli-head-pass")
4185 .expect("CLI alias password should win");
4186 assert!(
4187 auth_store.authenticate("head", "env-head-pass").is_err(),
4188 "cloud-specific env password must not beat CLI alias"
4189 );
4190
4191 clear_preset_env();
4192 }
4193
4194 #[test]
4195 fn bootstrap_intent_manifest_matches_cloud_flag_path() {
4196 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4197 clear_preset_env();
4198
4199 let manifest_dir = std::env::current_dir()
4200 .expect("current dir")
4201 .join(".red/tmp/bootstrap-manifest-tests");
4202 std::fs::create_dir_all(&manifest_dir).expect("create manifest test dir");
4203 let unique = format!(
4204 "{}-{}",
4205 std::process::id(),
4206 std::time::SystemTime::now()
4207 .duration_since(std::time::UNIX_EPOCH)
4208 .unwrap_or_default()
4209 .as_millis()
4210 );
4211 let head_password = manifest_dir.join(format!("head-{unique}.secret"));
4212 let customer_password = manifest_dir.join(format!("customer-{unique}.secret"));
4213 let manifest_path = manifest_dir.join(format!("intent-{unique}.json"));
4214 std::fs::write(&head_password, "head-pass\n").expect("write head secret");
4215 std::fs::write(&customer_password, "customer-pass\n").expect("write customer secret");
4216 std::fs::write(
4217 &manifest_path,
4218 format!(
4219 r#"{{
4220 "bootstrap": {{
4221 "preset": "cloud",
4222 "cloud_head_admin": {{
4223 "username": "head",
4224 "password_file": "{}"
4225 }},
4226 "customer_admin": {{
4227 "username": "customer",
4228 "password_file": "{}"
4229 }}
4230 }}
4231 }}"#,
4232 head_password.display(),
4233 customer_password.display()
4234 ),
4235 )
4236 .expect("write manifest");
4237
4238 let mut config = no_auth_test_config(false);
4239 config.bootstrap.manifest = Some(manifest_path.clone());
4240 let options = config.to_db_options().expect("manifest options");
4241 assert!(options.auth.enabled);
4242 assert!(options.auth.require_auth);
4243 assert!(options.auth.vault_enabled);
4244
4245 let (runtime, auth_store) = fresh_runtime_and_store();
4246 apply_preset_from_config(&runtime, &auth_store, &config.bootstrap)
4247 .expect("manifest intent applies cleanly");
4248
4249 auth_store
4250 .authenticate("head", "head-pass")
4251 .expect("head password comes from file");
4252 auth_store
4253 .authenticate("customer", "customer-pass")
4254 .expect("customer password comes from file");
4255 assert!(auth_store.get_user(None, "head").is_some());
4256 assert!(auth_store.get_user(None, "customer").is_some());
4257 assert!(auth_store
4258 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4259 .is_some());
4260 match runtime
4261 .db()
4262 .store()
4263 .get_config(BOOTSTRAP_PRESET_KEY)
4264 .expect("preset key persisted")
4265 {
4266 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_CLOUD),
4267 other => panic!("expected Text(cloud), got {other:?}"),
4268 }
4269
4270 apply_preset_from_config(&runtime, &auth_store, &config.bootstrap)
4271 .expect("second boot is idempotent");
4272
4273 let _ = std::fs::remove_file(&manifest_path);
4274 let _ = std::fs::remove_file(&head_password);
4275 let _ = std::fs::remove_file(&customer_password);
4276 clear_preset_env();
4277 }
4278
4279 #[test]
4280 fn regulated_preset_enables_query_audit_infrastructure_without_rules() {
4281 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4282 clear_preset_env();
4283 unsafe {
4285 std::env::set_var(PRESET_ENV, PRESET_REGULATED);
4286 }
4287
4288 let (runtime, auth_store) = fresh_runtime_and_store();
4289 apply_preset(&runtime, &auth_store).expect("regulated preset applies cleanly");
4290
4291 assert!(runtime.query_audit().is_enabled());
4292 assert!(runtime.query_audit().rules().is_empty());
4293 assert!(
4294 runtime
4295 .db()
4296 .store()
4297 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4298 .is_some(),
4299 "regulated preset should create the query-audit stream"
4300 );
4301
4302 runtime
4303 .execute_query("CREATE TABLE docs (id INT)")
4304 .expect("create table");
4305 runtime
4306 .execute_query("INSERT INTO docs (id) VALUES (1)")
4307 .expect("insert");
4308 runtime.execute_query("SELECT * FROM docs").expect("select");
4309 let rows = runtime
4310 .db()
4311 .store()
4312 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4313 .expect("query audit collection")
4314 .query_all(|_| true);
4315 assert!(
4316 rows.is_empty(),
4317 "regulated preset must not globally audit every query"
4318 );
4319
4320 clear_preset_env();
4321 }
4322
4323 #[test]
4324 fn managed_backup_env_rejects_primary_replica_single_file_storage() {
4325 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4326 clear_backup_env();
4327 unsafe {
4329 std::env::set_var("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.test");
4330 std::env::set_var("REDDB_BACKUP_S3_BUCKET", "reddb");
4331 std::env::set_var("REDDB_BACKUP_S3_PREFIX", "clusters/prod");
4332 std::env::set_var("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK");
4333 std::env::set_var("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK");
4334 }
4335
4336 let mut config = no_auth_test_config(false);
4337 config.role = "primary".to_string();
4338 config.storage_profile = crate::storage::StorageDeployPreset::PrimaryReplicaDev.selection();
4339
4340 let err = config.to_db_options().unwrap_err();
4341 assert!(err.contains("managed backup"), "got: {err}");
4342 assert!(err.contains("operational-directory"), "got: {err}");
4343
4344 clear_backup_env();
4345 }
4346
4347 #[test]
4348 fn regulated_preset_installs_managed_evidence_guardrails_end_to_end() {
4349 use crate::auth::policies::{EvalContext, Policy, ResourceRef};
4350 use crate::auth::store::PrincipalRef;
4351 use crate::auth::{Role, UserId};
4352 use crate::runtime::mvcc::{clear_current_auth_identity, set_current_auth_identity};
4353 use crate::storage::schema::Value;
4354
4355 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4356 clear_preset_env();
4357 unsafe {
4359 std::env::set_var(PRESET_ENV, PRESET_REGULATED);
4360 }
4361
4362 let options = no_auth_test_config(false)
4363 .to_db_options()
4364 .expect("regulated options");
4365 assert!(
4366 options.control_events.compliance_mode,
4367 "regulated preset must enable fail-closed control evidence before runtime boot"
4368 );
4369 assert!(
4370 options.query_audit.enabled && options.query_audit.rules.is_empty(),
4371 "regulated preset must enable query-audit infrastructure without global rules"
4372 );
4373
4374 let runtime = RedDBRuntime::with_options(options).expect("runtime");
4375 let auth_store = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4376 apply_preset(&runtime, &auth_store).expect("regulated preset applies cleanly");
4377 runtime.set_auth_store(Arc::clone(&auth_store));
4378
4379 assert!(runtime.control_events_require_persistence());
4380 assert!(runtime.query_audit().is_enabled());
4381 assert!(runtime.query_audit().rules().is_empty());
4382 assert!(auth_store
4383 .get_policy(REGULATED_PROTECT_MANAGED_POLICY)
4384 .is_some());
4385
4386 let managed_policy = runtime
4387 .config_registry()
4388 .get_active(REGULATED_PROTECT_MANAGED_POLICY)
4389 .expect("regulated managed policy registry entry");
4390 assert!(managed_policy.managed);
4391 assert_eq!(managed_policy.resource_type, "policy");
4392 assert!(
4393 runtime
4394 .config_registry()
4395 .get_active(REGULATED_AUDIT_CONFIG_NAMESPACE)
4396 .expect("regulated audit config namespace")
4397 .managed
4398 );
4399
4400 let registry_rows = runtime
4401 .execute_query(&format!(
4402 "SELECT id, managed FROM red.registry WHERE id = '{}'",
4403 REGULATED_PROTECT_MANAGED_POLICY
4404 ))
4405 .expect("red.registry query");
4406 assert_eq!(registry_rows.result.records.len(), 1);
4407 assert_eq!(
4408 registry_rows.result.records[0].get("managed"),
4409 Some(&Value::Boolean(true))
4410 );
4411
4412 let managed_policy_rows = runtime
4413 .execute_query(&format!(
4414 "SELECT policy_id FROM red.managed_policies WHERE policy_id = '{}'",
4415 REGULATED_PROTECT_MANAGED_POLICY
4416 ))
4417 .expect("red.managed_policies query");
4418 assert_eq!(managed_policy_rows.result.records.len(), 1);
4419
4420 let capability_rows = runtime
4421 .execute_query(
4422 "SELECT action FROM red.control_capabilities WHERE action = 'evidence:export'",
4423 )
4424 .expect("red.control_capabilities query");
4425 assert_eq!(capability_rows.result.records.len(), 1);
4426
4427 auth_store
4428 .create_user("alice", "p", Role::Admin)
4429 .expect("create ordinary admin");
4430 let allow_all = Policy::from_json_str(
4431 r#"{
4432 "id": "alice-allow-all",
4433 "version": 1,
4434 "statements": [{
4435 "effect": "allow",
4436 "actions": ["*"],
4437 "resources": ["*"]
4438 }]
4439 }"#,
4440 )
4441 .expect("allow-all policy");
4442 auth_store.put_policy(allow_all).expect("install allow-all");
4443 auth_store
4444 .attach_policy(
4445 PrincipalRef::User(UserId::platform("alice")),
4446 "alice-allow-all",
4447 )
4448 .expect("attach allow-all");
4449 let ctx = EvalContext {
4450 principal_tenant: None,
4451 current_tenant: None,
4452 peer_ip: None,
4453 mfa_present: false,
4454 now_ms: 1_700_000_000_000,
4455 principal_is_admin_role: true,
4456 principal_is_platform_scoped: true,
4457 };
4458 assert!(
4459 !auth_store.check_policy_authz(
4460 &UserId::platform("alice"),
4461 "policy:drop",
4462 &ResourceRef::new("policy", REGULATED_PROTECT_MANAGED_POLICY),
4463 &ctx,
4464 ),
4465 "managed guardrail deny policy must be effective for ordinary admins"
4466 );
4467
4468 set_current_auth_identity("alice".to_string(), Role::Admin);
4469 let denied = runtime.execute_query(&format!(
4470 "DROP POLICY '{}'",
4471 REGULATED_PROTECT_MANAGED_POLICY
4472 ));
4473 clear_current_auth_identity();
4474 let err = denied.expect_err("managed policy guardrail must deny ordinary admin");
4475 assert!(
4476 err.to_string().contains("managed policy"),
4477 "error should name the managed guardrail: {err}"
4478 );
4479 assert!(
4480 auth_store
4481 .get_policy(REGULATED_PROTECT_MANAGED_POLICY)
4482 .is_some(),
4483 "denied mutation must leave managed policy installed"
4484 );
4485
4486 let denied_events = runtime
4487 .execute_query(&format!(
4488 "SELECT action, resource, outcome FROM red.control_events \
4489 WHERE action = 'policy:drop' AND resource = 'policy:{}'",
4490 REGULATED_PROTECT_MANAGED_POLICY
4491 ))
4492 .expect("red.control_events denied policy drop");
4493 assert_eq!(denied_events.result.records.len(), 1);
4494 assert_eq!(
4495 denied_events.result.records[0].get("outcome"),
4496 Some(&Value::text("denied"))
4497 );
4498
4499 set_current_auth_identity("alice".to_string(), Role::Admin);
4500 let config_denied = runtime.execute_query("SET CONFIG red.config.audit.enabled = true");
4501 clear_current_auth_identity();
4502 let err = config_denied.expect_err("managed config guardrail must deny ordinary admin");
4503 assert!(
4504 err.to_string().contains("managed config"),
4505 "error should name the managed config guardrail: {err}"
4506 );
4507
4508 let denied_config_events = runtime
4509 .execute_query(
4510 "SELECT action, resource, outcome FROM red.control_events \
4511 WHERE action = 'config:write' AND resource = 'config:red.config.audit.enabled'",
4512 )
4513 .expect("red.control_events denied config write");
4514 assert_eq!(denied_config_events.result.records.len(), 1);
4515 assert_eq!(
4516 denied_config_events.result.records[0].get("outcome"),
4517 Some(&Value::text("denied"))
4518 );
4519
4520 runtime
4521 .execute_query("CREATE TABLE regulated_docs (id INT)")
4522 .expect("create user table");
4523 runtime
4524 .execute_query("SELECT * FROM regulated_docs")
4525 .expect("select user table");
4526 let audit_rows = runtime
4527 .db()
4528 .store()
4529 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4530 .expect("query audit collection")
4531 .query_all(|_| true);
4532 assert!(
4533 audit_rows.is_empty(),
4534 "regulated preset must not globally audit data-plane queries"
4535 );
4536
4537 clear_preset_env();
4538 }
4539
4540 #[test]
4541 fn bootstrap_manifest_installs_initial_users_policies_guardrails_and_config() {
4542 use crate::auth::policies::{EvalContext, ResourceRef};
4543 use crate::auth::UserId;
4544 use crate::storage::schema::Value;
4545
4546 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4547 clear_preset_env();
4548
4549 let manifest_dir = std::env::current_dir()
4550 .expect("current dir")
4551 .join(".red/tmp/bootstrap-manifest-tests");
4552 std::fs::create_dir_all(&manifest_dir).expect("create manifest test dir");
4553 let manifest_path = manifest_dir.join(format!(
4554 "reddb-bootstrap-manifest-{}-{}.json",
4555 std::process::id(),
4556 std::time::SystemTime::now()
4557 .duration_since(std::time::UNIX_EPOCH)
4558 .unwrap_or_default()
4559 .as_millis()
4560 ));
4561 std::fs::write(
4562 &manifest_path,
4563 r#"{
4564 "users": [
4565 {
4566 "username": "ops",
4567 "password": "hunter2",
4568 "role": "admin"
4569 }
4570 ],
4571 "policies": [
4572 {
4573 "id": "bootstrap-registry-admin",
4574 "version": 1,
4575 "statements": [
4576 {
4577 "effect": "allow",
4578 "actions": ["red.registry:*", "policy:*", "config:write", "select"],
4579 "resources": ["registry:*", "policy:*", "config:*", "collection:docs"]
4580 }
4581 ]
4582 }
4583 ],
4584 "managed_policies": [
4585 {
4586 "id": "managed-deny-drop",
4587 "version": 1,
4588 "statements": [
4589 {
4590 "effect": "deny",
4591 "actions": ["policy:drop"],
4592 "resources": ["policy:managed-deny-drop"]
4593 }
4594 ],
4595 "required_resource": "policy:managed-deny-drop",
4596 "evidence": "full"
4597 }
4598 ],
4599 "attachments": [
4600 {"user": "ops", "policy": "bootstrap-registry-admin"}
4601 ],
4602 "managed_config_namespaces": [
4603 {
4604 "id": "red.ai",
4605 "required_action": "config:write",
4606 "required_resource": "config:red.ai.*",
4607 "evidence": "metadata"
4608 }
4609 ],
4610 "config": [
4611 {"key": "red.ai.default.provider", "value": "openai"},
4612 {
4613 "key": "red.ai.openai.default.secret_ref",
4614 "secret_ref": {"collection": "red.vault", "key": "openai"}
4615 }
4616 ],
4617 "actor": "ops"
4618 }"#,
4619 )
4620 .expect("write manifest");
4621 unsafe {
4623 std::env::set_var("REDDB_BOOTSTRAP_MANIFEST", &manifest_path);
4624 }
4625
4626 let (runtime, auth_store) = fresh_runtime_and_store();
4627 apply_preset(&runtime, &auth_store).expect("manifest applies cleanly");
4628
4629 let users = auth_store.list_users();
4630 assert_eq!(users.len(), 1);
4631 assert_eq!(users[0].username, "ops");
4632 assert!(users[0].tenant_id.is_none());
4633
4634 let actor = UserId::platform("ops");
4635 let ctx = EvalContext {
4636 principal_tenant: None,
4637 current_tenant: None,
4638 peer_ip: None,
4639 mfa_present: false,
4640 now_ms: 1_700_000_000_000,
4641 principal_is_admin_role: true,
4642 principal_is_platform_scoped: true,
4643 };
4644 assert!(auth_store.check_policy_authz(
4646 &actor,
4647 "select",
4648 &ResourceRef::new("collection", "docs"),
4649 &ctx
4650 ));
4651
4652 let managed_policy = runtime
4653 .config_registry()
4654 .get_active("managed-deny-drop")
4655 .expect("managed policy registry entry");
4656 assert!(managed_policy.managed);
4657 assert_eq!(managed_policy.resource_type, "policy");
4658 let managed_config = runtime
4659 .config_registry()
4660 .get_active("red.ai")
4661 .expect("managed config namespace registry entry");
4662 assert!(managed_config.managed);
4663 assert_eq!(managed_config.resource_type, "config_namespace");
4664
4665 let store = runtime.db().store();
4666 match store
4667 .get_config("red.ai.default.provider")
4668 .expect("plain config persisted")
4669 {
4670 Value::Text(s) => assert_eq!(s.as_ref(), "openai"),
4671 other => panic!("expected provider text, got {other:?}"),
4672 }
4673 let Value::Json(bytes) = store
4674 .get_config("red.ai.openai.default.secret_ref")
4675 .expect("secret ref config persisted")
4676 else {
4677 panic!("secret ref must be stored as structured JSON");
4678 };
4679 let reference: crate::serde_json::Value =
4680 crate::serde_json::from_slice(&bytes).expect("secret ref json");
4681 assert_eq!(
4682 reference.get("type").and_then(|v| v.as_str()),
4683 Some("secret_ref")
4684 );
4685 assert!(
4686 !String::from_utf8_lossy(&bytes).contains("hunter2"),
4687 "manifest password must not leak into secret ref config"
4688 );
4689
4690 let completed = store
4691 .get_config(BOOTSTRAP_COMPLETED_KEY)
4692 .expect("bootstrap completion persisted");
4693 assert!(matches!(completed, Value::Boolean(true)));
4694 assert!(
4695 store
4696 .get_config("system.bootstrap.manifest.registry_entries")
4697 .is_some(),
4698 "managed registry entries must be persisted internally"
4699 );
4700
4701 std::fs::remove_file(&manifest_path).expect("remove manifest after first boot");
4702 let restored_registry = Arc::new(crate::auth::registry::ConfigRegistry::new());
4703 crate::cli::bootstrap_manifest::rehydrate_manifest_registry(&runtime, &restored_registry)
4704 .expect("registry rehydrates without manifest file");
4705 assert!(restored_registry.get_active("managed-deny-drop").is_some());
4706 assert!(restored_registry.get_active("red.ai").is_some());
4707
4708 let fresh = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4709 apply_preset(&runtime, &fresh).expect("re-run must not need manifest file");
4710 assert!(fresh.needs_bootstrap());
4711
4712 clear_preset_env();
4713 }
4714
4715 #[test]
4716 fn production_preset_refuses_to_start_without_password() {
4717 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4718 clear_preset_env();
4719 unsafe {
4721 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4722 std::env::set_var("REDDB_USERNAME", "ops");
4723 }
4724
4725 let (runtime, auth_store) = fresh_runtime_and_store();
4726 let err = apply_preset(&runtime, &auth_store).expect_err("must reject missing password");
4727 assert!(
4728 err.contains("REDDB_PASSWORD"),
4729 "error must name the missing env: {err}"
4730 );
4731
4732 assert!(auth_store.needs_bootstrap());
4734 assert!(runtime
4735 .db()
4736 .store()
4737 .get_config(BOOTSTRAP_COMPLETED_KEY)
4738 .is_none());
4739
4740 clear_preset_env();
4741 }
4742
4743 #[test]
4744 fn re_running_production_after_first_boot_is_a_silent_skip() {
4745 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4746 clear_preset_env();
4747 unsafe {
4749 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4750 std::env::set_var("REDDB_USERNAME", "ops");
4751 std::env::set_var("REDDB_PASSWORD", "hunter2");
4752 }
4753
4754 let (runtime, auth_store) = fresh_runtime_and_store();
4755 apply_preset(&runtime, &auth_store).expect("first apply");
4756 assert_eq!(auth_store.list_users().len(), 1);
4757
4758 let fresh = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4765 apply_preset(&runtime, &fresh).expect("re-run is silent-skip");
4766 assert!(
4767 fresh.needs_bootstrap(),
4768 "re-run must not create a second admin"
4769 );
4770 assert!(
4771 fresh.get_policy(FIRST_ADMIN_ALLOW_ALL_POLICY).is_none(),
4772 "re-run must not re-install the allow-all policy on the fresh store"
4773 );
4774
4775 clear_preset_env();
4776 }
4777
4778 #[test]
4779 fn unrecognised_preset_value_is_rejected() {
4780 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4781 clear_preset_env();
4782 unsafe {
4784 std::env::set_var(PRESET_ENV, "weird");
4785 }
4786
4787 let (runtime, auth_store) = fresh_runtime_and_store();
4788 let err = apply_preset(&runtime, &auth_store).expect_err("must reject unknown preset");
4789 assert!(err.contains("weird"), "error must echo the value: {err}");
4790 assert!(auth_store.needs_bootstrap());
4791
4792 clear_preset_env();
4793 }
4794
4795 #[test]
4796 fn no_auth_short_circuits_preset_entirely() {
4797 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4798 clear_preset_env();
4799 unsafe {
4802 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4803 std::env::set_var("REDDB_USERNAME", "ops");
4804 std::env::set_var("REDDB_PASSWORD", "hunter2");
4805 }
4806
4807 let options = no_auth_test_config(true)
4808 .to_db_options()
4809 .expect("to_db_options");
4810 assert!(no_auth_active(&options));
4811
4812 let (runtime, auth_store) = fresh_runtime_and_store();
4815 if !no_auth_active(&options) {
4816 apply_preset(&runtime, &auth_store).expect("would apply preset");
4817 }
4818
4819 assert!(
4820 auth_store.needs_bootstrap(),
4821 "--no-auth must prevent any admin creation"
4822 );
4823 assert!(
4824 runtime
4825 .db()
4826 .store()
4827 .get_config(BOOTSTRAP_COMPLETED_KEY)
4828 .is_none(),
4829 "--no-auth must skip bootstrap-state persistence"
4830 );
4831
4832 clear_preset_env();
4833 }
4834
4835 #[test]
4836 fn implicit_bind_collision_degrades() {
4837 let held = TcpListener::bind("127.0.0.1:0").expect("hold test port");
4838 let addr = held.local_addr().expect("held addr").to_string();
4839 let mut readiness = TransportReadiness::default();
4840
4841 let listener =
4842 bind_listener_for_startup(&mut readiness, "http", &addr, false).expect("nonfatal");
4843
4844 assert!(listener.is_none());
4845 assert_eq!(readiness.active.len(), 0);
4846 assert_eq!(readiness.failed.len(), 1);
4847 assert!(!readiness.failed[0].explicit);
4848 assert_eq!(readiness.failed[0].bind_addr, addr);
4849 }
4850}