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() {
1784 let tls_config = resolve_wire_tls_config(config);
1785 match tls_config {
1786 Ok(tls_cfg) => {
1787 let wire_rt = Arc::new(runtime.clone());
1788 tokio::spawn(async move {
1789 if let Err(e) =
1790 crate::wire::start_redwire_tls_listener(&wire_tls_addr, wire_rt, &tls_cfg)
1791 .await
1792 {
1793 tracing::error!(err = %e, "redwire+tls listener error");
1794 }
1795 });
1796 }
1797 Err(e) => tracing::error!(err = %e, "redwire TLS config error"),
1798 }
1799 }
1800 Ok(())
1801}
1802
1803fn spawn_pg_listener(config: &ServerCommandConfig, runtime: &RedDBRuntime) {
1810 if let Some(pg_addr) = config.pg_bind_addr.clone() {
1811 let tls = resolve_pg_wire_tls(config);
1812 let rt = Arc::new(runtime.clone());
1813 tokio::spawn(async move {
1814 let cfg = crate::wire::PgWireConfig {
1815 bind_addr: pg_addr,
1816 tls,
1817 ..Default::default()
1818 };
1819 if let Err(e) = crate::wire::start_pg_wire_listener(cfg, rt).await {
1820 tracing::error!(err = %e, "pg wire listener error");
1821 }
1822 });
1823 }
1824}
1825
1826fn resolve_pg_wire_tls(config: &ServerCommandConfig) -> Option<crate::wire::WireTlsConfig> {
1837 let tls_configured = config.wire_tls_bind_addr.is_some()
1838 || (config.wire_tls_cert.is_some() && config.wire_tls_key.is_some());
1839 if !tls_configured {
1840 return None;
1841 }
1842 match resolve_wire_tls_config(config) {
1843 Ok(cfg) => Some(cfg),
1844 Err(e) => {
1845 tracing::error!(
1846 err = %e,
1847 "pg wire TLS config error; listener will run plaintext"
1848 );
1849 None
1850 }
1851 }
1852}
1853
1854fn resolve_grpc_tls_options(config: &ServerCommandConfig) -> Result<crate::GrpcTlsOptions, String> {
1868 use crate::utils::secret_file::expand_file_env;
1869
1870 for var in [
1874 "REDDB_GRPC_TLS_CERT",
1875 "REDDB_GRPC_TLS_KEY",
1876 "REDDB_GRPC_TLS_CLIENT_CA",
1877 ] {
1878 if let Err(err) = expand_file_env(var) {
1879 tracing::warn!(
1880 target: "reddb::secrets",
1881 env = %var,
1882 err = %err,
1883 "could not expand *_FILE companion for gRPC TLS"
1884 );
1885 }
1886 }
1887
1888 let (cert_pem, key_pem) = match (&config.grpc_tls_cert, &config.grpc_tls_key) {
1889 (Some(cert), Some(key)) => {
1890 let cert_pem = std::fs::read(cert)
1891 .map_err(|e| format!("read grpc cert {}: {e}", cert.display()))?;
1892 let key_pem =
1893 std::fs::read(key).map_err(|e| format!("read grpc key {}: {e}", key.display()))?;
1894 (cert_pem, key_pem)
1895 }
1896 _ => {
1897 let dev = std::env::var("RED_GRPC_TLS_DEV")
1899 .ok()
1900 .map(|v| matches!(v.to_ascii_lowercase().as_str(), "1" | "true" | "yes" | "on"))
1901 .unwrap_or(false);
1902 if !dev {
1903 return Err("gRPC TLS configured but no cert/key supplied — set \
1904 REDDB_GRPC_TLS_CERT / REDDB_GRPC_TLS_KEY (or \
1905 RED_GRPC_TLS_DEV=1 to auto-generate a self-signed cert)"
1906 .to_string());
1907 }
1908 let dir = config
1909 .path
1910 .as_ref()
1911 .and_then(|p| p.parent())
1912 .map(PathBuf::from)
1913 .unwrap_or_else(|| PathBuf::from("."));
1914 let (cert_pem_str, key_pem_str) =
1915 crate::wire::tls::generate_self_signed_cert("localhost")
1916 .map_err(|e| format!("auto-generate dev grpc cert: {e}"))?;
1917
1918 let fp = sha256_pem_fingerprint(cert_pem_str.as_bytes());
1923 tracing::warn!(
1924 target: "reddb::security",
1925 transport = "grpc",
1926 cert_sha256 = %fp,
1927 "RED_GRPC_TLS_DEV=1: using auto-generated self-signed cert; \
1928 DO NOT use in production"
1929 );
1930 let cert_path = dir.join("grpc-tls-cert.pem");
1932 let key_path = dir.join("grpc-tls-key.pem");
1933 if !cert_path.exists() || !key_path.exists() {
1934 let _ = std::fs::create_dir_all(&dir);
1935 std::fs::write(&cert_path, cert_pem_str.as_bytes())
1936 .map_err(|e| format!("write grpc dev cert: {e}"))?;
1937 std::fs::write(&key_path, key_pem_str.as_bytes())
1938 .map_err(|e| format!("write grpc dev key: {e}"))?;
1939 #[cfg(unix)]
1940 {
1941 use std::os::unix::fs::PermissionsExt;
1942 let _ =
1943 std::fs::set_permissions(&key_path, std::fs::Permissions::from_mode(0o600));
1944 }
1945 }
1946 (cert_pem_str.into_bytes(), key_pem_str.into_bytes())
1947 }
1948 };
1949
1950 let client_ca_pem = match &config.grpc_tls_client_ca {
1951 Some(path) => Some(
1952 std::fs::read(path)
1953 .map_err(|e| format!("read grpc client CA {}: {e}", path.display()))?,
1954 ),
1955 None => None,
1956 };
1957
1958 Ok(crate::GrpcTlsOptions {
1959 cert_pem,
1960 key_pem,
1961 client_ca_pem,
1962 })
1963}
1964
1965fn spawn_grpc_tls_listener_if_configured(
1969 config: &ServerCommandConfig,
1970 runtime: RedDBRuntime,
1971 auth_store: Arc<AuthStore>,
1972) {
1973 let Some(tls_bind) = config.grpc_tls_bind_addr.clone() else {
1974 return;
1975 };
1976 let tls_opts = match resolve_grpc_tls_options(config) {
1977 Ok(opts) => opts,
1978 Err(err) => {
1979 tracing::error!(
1980 target: "reddb::security",
1981 transport = "grpc",
1982 err = %err,
1983 "gRPC TLS config error; TLS listener will not start"
1984 );
1985 return;
1986 }
1987 };
1988 tokio::spawn(async move {
1989 let server = RedDBGrpcServer::with_options(
1990 runtime,
1991 GrpcServerOptions {
1992 bind_addr: tls_bind.clone(),
1993 tls: Some(tls_opts),
1994 },
1995 auth_store,
1996 );
1997 tracing::info!(transport = "grpc+tls", bind = %tls_bind, "listener online");
1998 if let Err(err) = server.serve().await {
1999 tracing::error!(transport = "grpc+tls", err = %err, "gRPC TLS listener error");
2000 }
2001 });
2002}
2003
2004fn sha256_pem_fingerprint(pem: &[u8]) -> String {
2007 use sha2::{Digest, Sha256};
2008 let mut h = Sha256::new();
2009 h.update(pem);
2010 let d = h.finalize();
2011 let mut buf = String::with_capacity(64);
2012 for b in d.iter() {
2013 buf.push_str(&format!("{b:02x}"));
2014 }
2015 buf
2016}
2017
2018fn resolve_wire_tls_config(
2020 config: &ServerCommandConfig,
2021) -> Result<crate::wire::WireTlsConfig, String> {
2022 match (&config.wire_tls_cert, &config.wire_tls_key) {
2023 (Some(cert), Some(key)) => Ok(crate::wire::WireTlsConfig {
2024 cert_path: cert.clone(),
2025 key_path: key.clone(),
2026 }),
2027 _ => {
2028 let dir = config
2030 .path
2031 .as_ref()
2032 .and_then(|p| p.parent())
2033 .map(PathBuf::from)
2034 .unwrap_or_else(|| PathBuf::from("."));
2035 crate::wire::tls::auto_generate_cert(&dir).map_err(|e| e.to_string())
2036 }
2037 }
2038}
2039
2040#[inline(never)]
2041fn run_wire_only_server(config: ServerCommandConfig, wire_addr: String) -> Result<(), String> {
2042 let rt_config = detect_runtime_config();
2043 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2044 let db_options = config.to_db_options()?;
2045 let mut transport_readiness = TransportReadiness::default();
2046
2047 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2048 .enable_all()
2049 .worker_threads(workers)
2050 .thread_stack_size(rt_config.stack_size)
2051 .build()
2052 .map_err(|err| format!("tokio runtime: {err}"))?;
2053
2054 let (runtime, _auth_store, _telemetry_guard) =
2058 build_runtime_and_auth_store(&config, db_options.clone())?;
2059 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2060 let signal_runtime = runtime.clone();
2061 tokio_runtime.block_on(async move {
2062 spawn_lifecycle_signal_handler(signal_runtime).await;
2063 spawn_pg_listener(&config, &runtime);
2064 let wire_rt = Arc::new(runtime);
2065 let listener = tokio::net::TcpListener::bind(&wire_addr)
2066 .await
2067 .map_err(|err| {
2068 let reason = format!("wire listener bind {wire_addr}: {err}");
2069 transport_readiness.failed(
2070 "wire",
2071 &wire_addr,
2072 config.wire_bind_explicit,
2073 reason.clone(),
2074 );
2075 if config.wire_bind_explicit {
2076 format!("explicit {reason}")
2077 } else {
2078 reason
2079 }
2080 })?;
2081 transport_readiness.active("wire", &wire_addr, config.wire_bind_explicit);
2082 crate::wire::redwire::listener::start_redwire_listener_on(listener, wire_rt)
2083 .await
2084 .map_err(|e| e.to_string())
2085 })
2086}
2087
2088#[inline(never)]
2096fn run_wire_tls_only_server(
2097 config: ServerCommandConfig,
2098 wire_tls_addr: String,
2099) -> Result<(), String> {
2100 let rt_config = detect_runtime_config();
2101 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2102 let db_options = config.to_db_options()?;
2103 let tls_cfg = resolve_wire_tls_config(&config)?;
2107
2108 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2109 .enable_all()
2110 .worker_threads(workers)
2111 .thread_stack_size(rt_config.stack_size)
2112 .build()
2113 .map_err(|err| format!("tokio runtime: {err}"))?;
2114
2115 let (runtime, _auth_store, _telemetry_guard) =
2116 build_runtime_and_auth_store(&config, db_options.clone())?;
2117 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2118 let signal_runtime = runtime.clone();
2119 tokio_runtime.block_on(async move {
2120 spawn_lifecycle_signal_handler(signal_runtime).await;
2121 spawn_pg_listener(&config, &runtime);
2122 let wire_rt = Arc::new(runtime);
2123 crate::wire::start_redwire_tls_listener(&wire_tls_addr, wire_rt, &tls_cfg)
2124 .await
2125 .map_err(|e| format!("explicit wire-tls listener bind {wire_tls_addr}: {e}"))
2126 })
2127}
2128
2129#[inline(never)]
2130fn run_pg_only_server(config: ServerCommandConfig, pg_addr: String) -> Result<(), String> {
2131 let rt_config = detect_runtime_config();
2132 let workers = config.workers.unwrap_or(rt_config.suggested_workers);
2133 let db_options = config.to_db_options()?;
2134
2135 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
2136 .enable_all()
2137 .worker_threads(workers)
2138 .thread_stack_size(rt_config.stack_size)
2139 .build()
2140 .map_err(|err| format!("tokio runtime: {err}"))?;
2141
2142 let (runtime, _auth_store, _telemetry_guard) =
2143 build_runtime_and_auth_store(&config, db_options.clone())?;
2144 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
2145 let pg_tls = resolve_pg_wire_tls(&config);
2146 let signal_runtime = runtime.clone();
2147 tokio_runtime.block_on(async move {
2148 spawn_lifecycle_signal_handler(signal_runtime).await;
2149 let cfg = crate::wire::PgWireConfig {
2150 bind_addr: pg_addr,
2151 tls: pg_tls,
2152 ..Default::default()
2153 };
2154 crate::wire::start_pg_wire_listener(cfg, Arc::new(runtime))
2155 .await
2156 .map_err(|e| e.to_string())
2157 })
2158}
2159
2160#[inline(never)]
2161fn build_runtime_and_auth_store(
2162 config: &ServerCommandConfig,
2163 db_options: RedDBOptions,
2164) -> Result<
2165 (
2166 RedDBRuntime,
2167 Arc<AuthStore>,
2168 Option<crate::telemetry::TelemetryGuard>,
2169 ),
2170 String,
2171> {
2172 build_runtime_with_bootstrap(db_options, config.telemetry.clone(), &config.bootstrap)
2179}
2180
2181pub(crate) fn build_runtime_with_telemetry(
2191 db_options: RedDBOptions,
2192 cli_telemetry: Option<crate::telemetry::TelemetryConfig>,
2193) -> Result<
2194 (
2195 RedDBRuntime,
2196 Arc<AuthStore>,
2197 Option<crate::telemetry::TelemetryGuard>,
2198 ),
2199 String,
2200> {
2201 let bootstrap = BootstrapConfig::default();
2202 build_runtime_with_bootstrap(db_options, cli_telemetry, &bootstrap)
2203}
2204
2205fn build_runtime_with_bootstrap(
2206 db_options: RedDBOptions,
2207 cli_telemetry: Option<crate::telemetry::TelemetryConfig>,
2208 bootstrap: &BootstrapConfig,
2209) -> Result<
2210 (
2211 RedDBRuntime,
2212 Arc<AuthStore>,
2213 Option<crate::telemetry::TelemetryGuard>,
2214 ),
2215 String,
2216> {
2217 let runtime = RedDBRuntime::with_options(db_options.clone()).map_err(|err| {
2218 let msg = err.to_string();
2224 crate::telemetry::operator_event::OperatorEvent::StartupFailed {
2225 phase: "runtime_construction".to_string(),
2226 error: msg.clone(),
2227 }
2228 .emit_global();
2229 msg
2230 })?;
2231
2232 crate::runtime::lease_loop::start_lease_loop_if_required(&runtime).map_err(|err| {
2237 let msg = err.to_string();
2238 crate::telemetry::operator_event::OperatorEvent::StartupFailed {
2239 phase: "lease_loop".to_string(),
2240 error: msg.clone(),
2241 }
2242 .emit_global();
2243 msg
2244 })?;
2245
2246 if let Some(data_path) = db_options.data_path.as_deref() {
2250 let watch_dir = data_path.parent().unwrap_or(data_path);
2251 crate::runtime::disk_space_monitor::DiskSpaceMonitor::new(watch_dir, 90).spawn();
2252 }
2253
2254 {
2258 let config_path = crate::runtime::config_overlay::config_file_path();
2259 let store = runtime.db().store();
2260 crate::runtime::config_watcher::ConfigWatcher::new(config_path, store).spawn();
2261 }
2262
2263 let merged = merge_telemetry_with_config(
2266 cli_telemetry
2267 .unwrap_or_else(|| default_telemetry_for_path(db_options.data_path.as_deref())),
2268 &runtime,
2269 );
2270 let telemetry_guard = crate::telemetry::init(merged);
2271
2272 let no_auth = no_auth_active(&db_options);
2273 let auth_store =
2274 if db_options.auth.vault_enabled {
2275 let pager =
2276 runtime.db().store().pager().cloned().ok_or_else(|| {
2277 "vault requires a paged database (persistent mode)".to_string()
2278 })?;
2279 let store = AuthStore::with_vault(db_options.auth.clone(), pager)
2280 .map_err(|err| err.to_string())?;
2281 Arc::new(store)
2282 } else {
2283 Arc::new(AuthStore::new(db_options.auth.clone()))
2284 };
2285 auth_store.configure_control_events(
2286 runtime.control_event_ledger(),
2287 runtime.control_event_config(),
2288 );
2289 let bootstrap_already_completed = runtime
2306 .db()
2307 .store()
2308 .get_config(BOOTSTRAP_COMPLETED_KEY)
2309 .is_some();
2310 let disposition = crate::cluster::authorize_cluster_bootstrap(
2311 db_options.storage_profile.deploy_profile,
2312 no_auth,
2313 bootstrap.auth_bootstrap_input(),
2314 bootstrap_already_completed,
2315 )?;
2316 let vault_plan = crate::cluster::plan_vault_bootstrap(disposition);
2325 match disposition {
2326 crate::cluster::BootstrapDisposition::SkipDevBypass => {
2327 debug_assert_eq!(vault_plan, crate::cluster::VaultBootstrapPlan::SkipNoVault);
2328 eprintln!("{NO_AUTH_WARNING}");
2329 tracing::warn!("{NO_AUTH_WARNING}");
2330 }
2331 crate::cluster::BootstrapDisposition::AlreadyComplete => {
2332 debug_assert_eq!(
2339 vault_plan,
2340 crate::cluster::VaultBootstrapPlan::OpenClusterGlobalStore {
2341 consume_secret_inputs: false,
2342 }
2343 );
2344 apply_preset_from_config(&runtime, &auth_store, bootstrap)?;
2345 tracing::info!("bootstrap already completed, unsealing existing cluster auth store");
2346 }
2347 crate::cluster::BootstrapDisposition::ProceedLocal => {
2348 debug_assert_eq!(
2351 vault_plan,
2352 crate::cluster::VaultBootstrapPlan::OpenClusterGlobalStore {
2353 consume_secret_inputs: true,
2354 }
2355 );
2356 apply_preset_from_config(&runtime, &auth_store, bootstrap)?;
2357 maybe_apply_policy_break_glass(&auth_store);
2358 }
2359 }
2360
2361 {
2363 let store = Arc::clone(&auth_store);
2364 std::thread::Builder::new()
2365 .name("reddb-session-purge".into())
2366 .spawn(move || loop {
2367 std::thread::sleep(std::time::Duration::from_secs(300));
2368 store.purge_expired_sessions();
2369 })
2370 .ok();
2371 }
2372
2373 Ok((runtime, auth_store, telemetry_guard))
2374}
2375
2376fn maybe_apply_policy_break_glass(auth_store: &Arc<AuthStore>) {
2380 use crate::auth::self_lock_guard::BREAK_GLASS_ENV;
2381
2382 let enabled = std::env::var(BREAK_GLASS_ENV)
2383 .ok()
2384 .map(|v| {
2385 let trimmed = v.trim().to_ascii_lowercase();
2386 matches!(trimmed.as_str(), "1" | "true" | "yes")
2387 })
2388 .unwrap_or(false);
2389 if !enabled {
2390 return;
2391 }
2392 let now = crate::utils::now_unix_millis() as u128;
2393 match auth_store.apply_policy_break_glass(now) {
2394 Ok(()) => {
2395 tracing::warn!(env = BREAK_GLASS_ENV, "policy break-glass recovery applied");
2396 }
2397 Err(err) => {
2398 tracing::error!(env = BREAK_GLASS_ENV, %err, "policy break-glass recovery failed");
2399 }
2400 }
2401}
2402
2403pub(crate) const BOOTSTRAP_COMPLETED_KEY: &str = "system.bootstrap.completed";
2408pub(crate) const BOOTSTRAP_PRESET_KEY: &str = "system.bootstrap.preset";
2409pub(crate) const BOOTSTRAP_FIRST_ADMIN_KEY: &str = "system.bootstrap.first_admin_id";
2410
2411pub(crate) const BOOTSTRAP_PRESET_ENV: &str = "REDDB_BOOTSTRAP_PRESET";
2414pub(crate) const PRESET_ENV: &str = "REDDB_PRESET";
2415pub(crate) const BOOTSTRAP_CERT_OUT_ENV: &str = "REDDB_BOOTSTRAP_CERT_OUT";
2418pub(crate) const PRESET_SIMPLE: &str = "simple";
2419pub(crate) const PRESET_PRODUCTION: &str = "production";
2420pub(crate) const PRESET_REGULATED: &str = "regulated";
2421pub(crate) const PRESET_CLOUD: &str = "cloud";
2422
2423pub(crate) const FIRST_ADMIN_ALLOW_ALL_POLICY: &str = "system.bootstrap.first-admin-allow-all";
2427pub(crate) const REGULATED_PROTECT_MANAGED_POLICY: &str = "system.regulated.protect-managed";
2428pub(crate) const CLOUD_PROTECT_MANAGED_POLICY: &str = "system.cloud.protect-managed";
2429pub(crate) const CLOUD_CONFIG_NAMESPACE: &str = "red.config.cloud";
2430pub(crate) const REGULATED_AUDIT_CONFIG_NAMESPACE: &str = "red.config.audit";
2431pub(crate) const REGULATED_EVIDENCE_CONFIG_NAMESPACE: &str = "red.config.evidence";
2432pub(crate) const REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE: &str = "red.config.query_audit";
2433
2434pub(crate) fn apply_preset(
2444 runtime: &RedDBRuntime,
2445 auth_store: &Arc<AuthStore>,
2446) -> Result<(), String> {
2447 let bootstrap = BootstrapConfig::default();
2448 apply_preset_from_config(runtime, auth_store, &bootstrap)
2449}
2450
2451fn apply_preset_from_config(
2452 runtime: &RedDBRuntime,
2453 auth_store: &Arc<AuthStore>,
2454 bootstrap: &BootstrapConfig,
2455) -> Result<(), String> {
2456 let store = runtime.db().store();
2457
2458 if store.get_config(BOOTSTRAP_COMPLETED_KEY).is_some() {
2459 crate::cli::bootstrap_manifest::rehydrate_manifest_registry(
2460 runtime,
2461 &runtime.config_registry(),
2462 )?;
2463 tracing::info!("bootstrap state present, skipping preset application");
2464 return Ok(());
2465 }
2466
2467 let preset = bootstrap.resolved_preset();
2468
2469 for var in bootstrap.secret_env_vars_to_expand(&preset) {
2474 crate::utils::expand_file_env(var).map_err(|err| format!("expand {var}_FILE: {err}"))?;
2475 }
2476
2477 if let Some(path) = bootstrap.selected_manifest() {
2478 if let Some(manifest_bootstrap) =
2479 crate::cli::bootstrap_manifest::bootstrap_config_from_manifest_file(&path)?
2480 {
2481 apply_preset_from_config(runtime, auth_store, &manifest_bootstrap)?;
2482 return Ok(());
2483 }
2484 let first_admin_id = crate::cli::bootstrap_manifest::apply_manifest_file(
2485 runtime,
2486 auth_store,
2487 &runtime.config_registry(),
2488 &path,
2489 )?;
2490 persist_bootstrap_state(runtime, "manifest", Some(&first_admin_id));
2491 tracing::info!("bootstrap manifest applied");
2492 return Ok(());
2493 }
2494
2495 let (first_admin_id, certificate) = match preset.as_str() {
2496 PRESET_SIMPLE => {
2497 (None, None)
2501 }
2502 PRESET_PRODUCTION => {
2503 let (id, cert) = apply_production_preset_from_config(auth_store, bootstrap)?;
2504 (Some(id), cert)
2505 }
2506 PRESET_CLOUD => {
2507 let (id, cert) = apply_cloud_preset(runtime, auth_store, bootstrap)?;
2508 (Some(id), cert)
2509 }
2510 PRESET_REGULATED => {
2511 apply_regulated_preset(runtime, auth_store)?;
2512 (None, None)
2513 }
2514 other => {
2515 return Err(format!(
2516 "bootstrap preset {other:?} is not recognised (expected `simple`, `production`, `regulated`, or `cloud`)"
2517 ));
2518 }
2519 };
2520
2521 if let Some(cert) = certificate.as_deref() {
2528 write_bootstrap_certificate(bootstrap, cert)?;
2529 }
2530
2531 persist_bootstrap_state(runtime, &preset, first_admin_id.as_deref());
2532 tracing::info!(preset = %preset, "bootstrap preset applied");
2533 Ok(())
2534}
2535
2536fn write_bootstrap_certificate(
2540 bootstrap: &BootstrapConfig,
2541 certificate_hex: &str,
2542) -> Result<(), String> {
2543 let Some(path) = bootstrap.selected_cert_out() else {
2544 return Ok(());
2545 };
2546 std::fs::write(&path, certificate_hex)
2547 .map_err(|err| format!("write --bootstrap-cert-out {}: {err}", path.display()))?;
2548 tracing::info!(
2549 target: "reddb::bootstrap",
2550 path = %path.display(),
2551 "first-boot self-bootstrap: wrote unseal certificate to --bootstrap-cert-out (consume via REDDB_CERTIFICATE_FILE on the next boot)"
2552 );
2553 Ok(())
2554}
2555
2556fn apply_production_preset_from_config(
2557 auth_store: &Arc<AuthStore>,
2558 bootstrap: &BootstrapConfig,
2559) -> Result<(String, Option<String>), String> {
2560 use crate::auth::store::PrincipalRef;
2561 use crate::auth::UserId;
2562
2563 let username = bootstrap.production_username().ok_or_else(|| {
2564 "production preset requires --bootstrap-admin or REDDB_USERNAME (or REDDB_USERNAME_FILE)"
2565 .to_string()
2566 })?;
2567 let password = bootstrap.production_password().ok_or_else(|| {
2568 "production preset requires --bootstrap-admin-password or REDDB_PASSWORD (or REDDB_PASSWORD_FILE)"
2569 .to_string()
2570 })?;
2571
2572 let result = auth_store
2576 .bootstrap(&username, &password)
2577 .map_err(|err| format!("bootstrap first admin: {err}"))?;
2578 let certificate = result.certificate.clone();
2579 if let Some(cert) = certificate.as_deref() {
2580 eprintln!("[reddb] CERTIFICATE: {}", cert);
2581 tracing::warn!(
2582 "vault certificate issued by REDDB_PRESET=production -- save it and update the runtime secret before restart"
2583 );
2584 }
2585 let first_admin = UserId::platform(result.user.username.clone());
2586
2587 install_allow_all_policy(auth_store)?;
2589
2590 auth_store
2592 .attach_policy(
2593 PrincipalRef::User(first_admin.clone()),
2594 FIRST_ADMIN_ALLOW_ALL_POLICY,
2595 )
2596 .map_err(|err| format!("attach allow-all policy: {err}"))?;
2597
2598 Ok((first_admin.to_string(), certificate))
2599}
2600
2601fn apply_cloud_preset(
2602 runtime: &RedDBRuntime,
2603 auth_store: &Arc<AuthStore>,
2604 bootstrap: &BootstrapConfig,
2605) -> Result<(String, Option<String>), String> {
2606 use crate::auth::store::PrincipalRef;
2607 use crate::auth::{Role, UserId};
2608
2609 let head_username = bootstrap.cloud_head_username().ok_or_else(|| {
2610 "cloud preset requires --cloud-head-admin or REDDB_CLOUD_HEAD_ADMIN (or REDDB_USERNAME)"
2611 .to_string()
2612 })?;
2613 let head_password = bootstrap.cloud_head_password().ok_or_else(|| {
2614 "cloud preset requires --cloud-head-admin-password or REDDB_CLOUD_HEAD_ADMIN_PASSWORD (or REDDB_PASSWORD)"
2615 .to_string()
2616 })?;
2617 let customer_username = bootstrap.customer_username().ok_or_else(|| {
2618 "cloud preset requires --customer-admin or REDDB_CUSTOMER_ADMIN".to_string()
2619 })?;
2620 let customer_password = bootstrap.customer_password().ok_or_else(|| {
2621 "cloud preset requires --customer-admin-password or REDDB_CUSTOMER_ADMIN_PASSWORD"
2622 .to_string()
2623 })?;
2624 if head_username == customer_username {
2625 return Err("cloud preset requires distinct head and customer admin usernames".to_string());
2626 }
2627
2628 let head = auth_store
2629 .bootstrap_system_admin(&head_username, &head_password)
2630 .map_err(|err| format!("bootstrap cloud head admin: {err}"))?;
2631 let certificate = head.certificate.clone();
2632 let head_id = UserId::platform(head.user.username.clone());
2633
2634 auth_store
2635 .create_user(&customer_username, &customer_password, Role::Admin)
2636 .map_err(|err| format!("create cloud customer admin: {err}"))?;
2637 let customer_id = UserId::platform(customer_username.clone());
2638
2639 install_allow_all_policy(auth_store)?;
2640 for user_id in [&head_id, &customer_id] {
2641 auth_store
2642 .attach_policy(
2643 PrincipalRef::User(user_id.clone()),
2644 FIRST_ADMIN_ALLOW_ALL_POLICY,
2645 )
2646 .map_err(|err| format!("attach allow-all policy: {err}"))?;
2647 }
2648 install_cloud_guardrails(runtime, auth_store, &head_id)?;
2649 auth_store
2650 .attach_policy(
2651 PrincipalRef::User(customer_id),
2652 CLOUD_PROTECT_MANAGED_POLICY,
2653 )
2654 .map_err(|err| format!("attach cloud guardrail policy: {err}"))?;
2655
2656 Ok((head_id.to_string(), certificate))
2657}
2658
2659fn install_allow_all_policy(auth_store: &Arc<AuthStore>) -> Result<(), String> {
2660 use crate::auth::policies::Policy;
2661
2662 let policy = Policy::from_json_str(&format!(
2663 r#"{{
2664 "id": "{id}",
2665 "version": 1,
2666 "statements": [{{
2667 "effect": "allow",
2668 "actions": ["*"],
2669 "resources": ["*"]
2670 }}]
2671 }}"#,
2672 id = FIRST_ADMIN_ALLOW_ALL_POLICY
2673 ))
2674 .map_err(|err| format!("compile allow-all policy: {err}"))?;
2675 auth_store
2676 .put_policy(policy)
2677 .map_err(|err| format!("install allow-all policy: {err}"))
2678}
2679
2680fn install_cloud_guardrails(
2681 runtime: &RedDBRuntime,
2682 auth_store: &Arc<AuthStore>,
2683 head_admin: &crate::auth::UserId,
2684) -> Result<(), String> {
2685 use crate::auth::policies::Policy;
2686 use crate::auth::registry::EvidenceRequirement;
2687
2688 let policy = Policy::from_json_str(&format!(
2689 r#"{{
2690 "id": "{id}",
2691 "version": 1,
2692 "statements": [
2693 {{
2694 "effect": "deny",
2695 "actions": ["policy:put", "policy:drop", "policy:attach", "policy:detach"],
2696 "resources": ["policy:{id}"]
2697 }},
2698 {{
2699 "effect": "deny",
2700 "actions": [
2701 "user:delete",
2702 "user:disable",
2703 "user:password:change",
2704 "user:role:update"
2705 ],
2706 "resources": ["user:{head_admin}"]
2707 }},
2708 {{
2709 "effect": "deny",
2710 "actions": ["config:write"],
2711 "resources": ["config:{cloud}.*"]
2712 }}
2713 ]
2714 }}"#,
2715 id = CLOUD_PROTECT_MANAGED_POLICY,
2716 head_admin = head_admin,
2717 cloud = CLOUD_CONFIG_NAMESPACE,
2718 ))
2719 .map_err(|err| format!("compile cloud guardrail policy: {err}"))?;
2720 auth_store
2721 .put_policy(policy)
2722 .map_err(|err| format!("install cloud guardrail policy: {err}"))?;
2723
2724 let now_ms = crate::utils::now_unix_millis() as u128;
2725 let entries = vec![
2726 regulated_registry_entry(
2727 CLOUD_PROTECT_MANAGED_POLICY,
2728 crate::auth::managed_policy::RESOURCE_TYPE_POLICY,
2729 "policy",
2730 "policy:*",
2731 &format!("policy:{CLOUD_PROTECT_MANAGED_POLICY}"),
2732 EvidenceRequirement::Metadata,
2733 now_ms,
2734 ),
2735 regulated_registry_entry(
2736 CLOUD_CONFIG_NAMESPACE,
2737 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2738 "config_namespace",
2739 "config:write",
2740 &format!("config:{CLOUD_CONFIG_NAMESPACE}.*"),
2741 EvidenceRequirement::Metadata,
2742 now_ms,
2743 ),
2744 ];
2745 for entry in entries.iter().cloned() {
2746 runtime
2747 .config_registry()
2748 .restore_bootstrap_entry(entry)
2749 .map_err(|err| format!("install cloud registry entry: {err}"))?;
2750 }
2751 crate::cli::bootstrap_manifest::persist_registry_state(runtime, &entries)?;
2752 Ok(())
2753}
2754
2755fn apply_regulated_preset(
2756 runtime: &RedDBRuntime,
2757 auth_store: &Arc<AuthStore>,
2758) -> Result<(), String> {
2759 use crate::auth::policies::Policy;
2760 use crate::auth::registry::EvidenceRequirement;
2761 use crate::auth::store::{PrincipalRef, PUBLIC_IAM_GROUP};
2762
2763 runtime.query_audit().enable_infrastructure();
2764
2765 let policy = Policy::from_json_str(&format!(
2766 r#"{{
2767 "id": "{id}",
2768 "version": 1,
2769 "statements": [
2770 {{
2771 "effect": "deny",
2772 "actions": ["policy:put", "policy:drop", "policy:attach", "policy:detach"],
2773 "resources": ["policy:{id}"]
2774 }},
2775 {{
2776 "effect": "deny",
2777 "actions": ["config:write"],
2778 "resources": [
2779 "config:{audit}.*",
2780 "config:{evidence}.*",
2781 "config:{query_audit}.*"
2782 ]
2783 }}
2784 ]
2785 }}"#,
2786 id = REGULATED_PROTECT_MANAGED_POLICY,
2787 audit = REGULATED_AUDIT_CONFIG_NAMESPACE,
2788 evidence = REGULATED_EVIDENCE_CONFIG_NAMESPACE,
2789 query_audit = REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE,
2790 ))
2791 .map_err(|err| format!("compile regulated guardrail policy: {err}"))?;
2792 auth_store
2793 .put_policy(policy)
2794 .map_err(|err| format!("install regulated guardrail policy: {err}"))?;
2795 auth_store
2796 .attach_policy(
2797 PrincipalRef::Group(PUBLIC_IAM_GROUP.to_string()),
2798 REGULATED_PROTECT_MANAGED_POLICY,
2799 )
2800 .map_err(|err| format!("attach regulated guardrail policy: {err}"))?;
2801
2802 let now_ms = crate::utils::now_unix_millis() as u128;
2803 let entries = vec![
2804 regulated_registry_entry(
2805 REGULATED_PROTECT_MANAGED_POLICY,
2806 crate::auth::managed_policy::RESOURCE_TYPE_POLICY,
2807 "iam_policy",
2808 "policy:*",
2809 &format!("policy:{REGULATED_PROTECT_MANAGED_POLICY}"),
2810 EvidenceRequirement::Metadata,
2811 now_ms,
2812 ),
2813 regulated_registry_entry(
2814 REGULATED_AUDIT_CONFIG_NAMESPACE,
2815 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2816 "config_namespace",
2817 "config:write",
2818 &format!("config:{REGULATED_AUDIT_CONFIG_NAMESPACE}.*"),
2819 EvidenceRequirement::Metadata,
2820 now_ms,
2821 ),
2822 regulated_registry_entry(
2823 REGULATED_EVIDENCE_CONFIG_NAMESPACE,
2824 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2825 "config_namespace",
2826 "config:write",
2827 &format!("config:{REGULATED_EVIDENCE_CONFIG_NAMESPACE}.*"),
2828 EvidenceRequirement::Metadata,
2829 now_ms,
2830 ),
2831 regulated_registry_entry(
2832 REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE,
2833 crate::auth::managed_config::RESOURCE_TYPE_CONFIG_NAMESPACE,
2834 "config_namespace",
2835 "config:write",
2836 &format!("config:{REGULATED_QUERY_AUDIT_CONFIG_NAMESPACE}.*"),
2837 EvidenceRequirement::Metadata,
2838 now_ms,
2839 ),
2840 ];
2841
2842 for entry in entries.iter().cloned() {
2843 runtime
2844 .config_registry()
2845 .restore_bootstrap_entry(entry)
2846 .map_err(|err| format!("install regulated registry entry: {err}"))?;
2847 }
2848 crate::cli::bootstrap_manifest::persist_registry_state(runtime, &entries)?;
2849 Ok(())
2850}
2851
2852fn regulated_registry_entry(
2853 id: &str,
2854 resource_type: &str,
2855 schema: &str,
2856 required_action: &str,
2857 required_resource: &str,
2858 evidence_requirement: crate::auth::registry::EvidenceRequirement,
2859 updated_at_ms: u128,
2860) -> crate::auth::registry::ConfigRegistryEntry {
2861 crate::auth::registry::ConfigRegistryEntry {
2862 id: id.to_string(),
2863 version: 1,
2864 resource_type: resource_type.to_string(),
2865 schema: schema.to_string(),
2866 mutability: crate::auth::registry::Mutability::Immutable,
2867 sensitivity: crate::auth::registry::Sensitivity::Internal,
2868 managed: true,
2869 required_action: required_action.to_string(),
2870 required_resource: required_resource.to_string(),
2871 evidence_requirement,
2872 updated_by: "system:regulated-preset".to_string(),
2873 updated_at_ms,
2874 }
2875}
2876
2877fn persist_bootstrap_state(runtime: &RedDBRuntime, preset: &str, first_admin_id: Option<&str>) {
2878 let store = runtime.db().store();
2879 let mut tree = crate::serde_json::Map::new();
2880 tree.insert(
2881 BOOTSTRAP_COMPLETED_KEY.to_string(),
2882 crate::serde_json::Value::Bool(true),
2883 );
2884 tree.insert(
2885 BOOTSTRAP_PRESET_KEY.to_string(),
2886 crate::serde_json::Value::String(preset.to_string()),
2887 );
2888 if let Some(id) = first_admin_id {
2889 tree.insert(
2890 BOOTSTRAP_FIRST_ADMIN_KEY.to_string(),
2891 crate::serde_json::Value::String(id.to_string()),
2892 );
2893 }
2894 let json = crate::serde_json::Value::Object(tree);
2895 store.set_config_tree("", &json);
2896}
2897
2898fn merge_telemetry_with_config(
2909 mut cli: crate::telemetry::TelemetryConfig,
2910 runtime: &RedDBRuntime,
2911) -> crate::telemetry::TelemetryConfig {
2912 use crate::storage::schema::Value;
2913
2914 let store = runtime.db().store();
2915
2916 if !cli.level_explicit {
2917 if let Some(Value::Text(v)) = store.get_config("red.logging.level") {
2918 cli.level_filter = v.to_string();
2919 }
2920 }
2921 if !cli.format_explicit {
2922 if let Some(Value::Text(v)) = store.get_config("red.logging.format") {
2923 if let Some(parsed) = crate::telemetry::LogFormat::parse(&v) {
2924 cli.format = parsed;
2925 }
2926 }
2927 }
2928 if !cli.rotation_keep_days_explicit {
2929 match store.get_config("red.logging.keep_days") {
2930 Some(Value::Integer(n)) if n >= 0 && n <= u16::MAX as i64 => {
2931 cli.rotation_keep_days = n as u16
2932 }
2933 Some(Value::UnsignedInteger(n)) if n <= u16::MAX as u64 => {
2934 cli.rotation_keep_days = n as u16
2935 }
2936 Some(Value::Text(v)) => {
2937 if let Ok(n) = v.parse::<u16>() {
2938 cli.rotation_keep_days = n;
2939 }
2940 }
2941 _ => {}
2942 }
2943 }
2944 if !cli.file_prefix_explicit {
2945 if let Some(Value::Text(v)) = store.get_config("red.logging.file_prefix") {
2946 if !v.is_empty() {
2947 cli.file_prefix = v.to_string();
2948 }
2949 }
2950 }
2951 if !cli.log_dir_explicit && !cli.log_file_disabled {
2954 if let Some(Value::Text(v)) = store.get_config("red.logging.dir") {
2955 if !v.is_empty() {
2956 cli.log_dir = Some(std::path::PathBuf::from(v.as_ref()));
2957 }
2958 }
2959 }
2960
2961 cli
2962}
2963
2964#[cfg(test)]
2965mod telemetry_merge_tests {
2966 use super::*;
2967 use crate::telemetry::{LogFormat, TelemetryConfig};
2968
2969 fn fresh_runtime() -> RedDBRuntime {
2970 RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime")
2971 }
2972
2973 fn set_str(runtime: &RedDBRuntime, key: &str, value: &str) {
2974 runtime
2975 .db()
2976 .store()
2977 .set_config_tree(key, &crate::serde_json::Value::String(value.to_string()));
2978 }
2979
2980 fn cli_base() -> TelemetryConfig {
2981 TelemetryConfig {
2984 log_dir: Some(std::path::PathBuf::from("/tmp/reddb-default/logs")),
2985 format: LogFormat::Json,
2986 ..Default::default()
2987 }
2988 }
2989
2990 #[test]
2991 fn config_log_dir_promoted_when_flag_absent() {
2992 let runtime = fresh_runtime();
2993 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
2994 let merged = merge_telemetry_with_config(cli_base(), &runtime);
2995 assert_eq!(
2996 merged.log_dir.as_deref(),
2997 Some(std::path::Path::new("/var/log/reddb"))
2998 );
2999 }
3000
3001 #[test]
3002 fn explicit_log_dir_wins_over_config() {
3003 let runtime = fresh_runtime();
3004 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
3005 let mut cli = cli_base();
3006 cli.log_dir = Some(std::path::PathBuf::from("/custom/dir"));
3007 cli.log_dir_explicit = true;
3008 let merged = merge_telemetry_with_config(cli, &runtime);
3009 assert_eq!(
3010 merged.log_dir.as_deref(),
3011 Some(std::path::Path::new("/custom/dir"))
3012 );
3013 }
3014
3015 #[test]
3016 fn no_log_file_beats_config_log_dir() {
3017 let runtime = fresh_runtime();
3018 set_str(&runtime, "red.logging.dir", "/var/log/reddb");
3019 let mut cli = cli_base();
3020 cli.log_dir = None;
3021 cli.log_file_disabled = true;
3022 let merged = merge_telemetry_with_config(cli, &runtime);
3023 assert!(
3024 merged.log_dir.is_none(),
3025 "--no-log-file must veto config dir"
3026 );
3027 }
3028
3029 #[test]
3030 fn config_format_promoted_on_non_tty_default() {
3031 let runtime = fresh_runtime();
3035 set_str(&runtime, "red.logging.format", "pretty");
3036 let merged = merge_telemetry_with_config(cli_base(), &runtime);
3037 assert_eq!(merged.format, LogFormat::Pretty);
3038 }
3039
3040 #[test]
3041 fn explicit_format_wins_over_config() {
3042 let runtime = fresh_runtime();
3043 set_str(&runtime, "red.logging.format", "pretty");
3044 let mut cli = cli_base();
3045 cli.format = LogFormat::Json;
3046 cli.format_explicit = true;
3047 let merged = merge_telemetry_with_config(cli, &runtime);
3048 assert_eq!(merged.format, LogFormat::Json);
3049 }
3050}
3051
3052#[inline(never)]
3053fn build_http_server(
3054 runtime: RedDBRuntime,
3055 auth_store: Arc<AuthStore>,
3056 bind_addr: String,
3057) -> RedDBServer {
3058 build_http_server_with_transport_readiness(
3059 runtime,
3060 auth_store,
3061 bind_addr,
3062 TransportReadiness::default(),
3063 )
3064}
3065
3066fn apply_http_limits(
3072 server: RedDBServer,
3073 config: &ServerCommandConfig,
3074 runtime: &RedDBRuntime,
3075) -> RedDBServer {
3076 let store = runtime.db().store();
3077 let resolved =
3078 crate::server::http_limits::resolve_http_limits(&config.http_limits_cli, |key| match store
3079 .get_config(key)
3080 {
3081 Some(crate::storage::schema::Value::Text(v)) => Some(v.to_string()),
3082 Some(crate::storage::schema::Value::Integer(n)) if n >= 0 => Some(n.to_string()),
3083 Some(crate::storage::schema::Value::UnsignedInteger(n)) => Some(n.to_string()),
3084 _ => None,
3085 });
3086 tracing::info!(
3087 target: "reddb::http_limits",
3088 max_handlers = resolved.max_handlers,
3089 handler_timeout_ms = resolved.handler_timeout_ms,
3090 retry_after_secs = resolved.retry_after_secs,
3091 max_inflight_per_principal = resolved.max_inflight_per_principal,
3092 "http_limits resolved"
3093 );
3094 server.with_http_limits(resolved)
3095}
3096
3097fn apply_ui_bundle(
3107 server: RedDBServer,
3108 config: &ServerCommandConfig,
3109) -> Result<RedDBServer, String> {
3110 if !config.ui {
3111 return Ok(server);
3112 }
3113 let ui_dir = match &config.ui_dir {
3114 Some(dir) => dir.clone(),
3115 None => {
3116 let cache_root = crate::server::ui_bundle_resolver::reddb_user_cache_root()
3117 .unwrap_or_else(|_| std::env::temp_dir().join("reddb"));
3118 crate::server::ui_bundle_resolver::resolve_ui_bundle(
3119 &cache_root,
3120 &crate::server::ui_bundle_resolver::HttpFetcher,
3121 )
3122 .map_err(|err| format!("resolve red-ui bundle for --ui: {err}"))?
3123 }
3124 };
3125 tracing::info!(target: "reddb::ui", dir = %ui_dir.display(), "serving red-ui bundle on HTTP surface");
3126 Ok(server.with_ui_dir(ui_dir))
3127}
3128
3129#[inline(never)]
3130fn build_http_server_with_transport_readiness(
3131 runtime: RedDBRuntime,
3132 auth_store: Arc<AuthStore>,
3133 bind_addr: String,
3134 transport_readiness: TransportReadiness,
3135) -> RedDBServer {
3136 RedDBServer::with_options(
3137 runtime,
3138 ServerOptions {
3139 bind_addr,
3140 transport_readiness,
3141 ..ServerOptions::default()
3142 },
3143 )
3144 .with_auth(auth_store)
3145}
3146
3147#[inline(never)]
3151fn build_admin_only_server(
3152 runtime: RedDBRuntime,
3153 auth_store: Arc<AuthStore>,
3154 bind_addr: String,
3155) -> RedDBServer {
3156 RedDBServer::with_options(
3157 runtime,
3158 ServerOptions {
3159 bind_addr,
3160 surface: crate::server::ServerSurface::AdminOnly,
3161 ..ServerOptions::default()
3162 },
3163 )
3164 .with_auth(auth_store)
3165}
3166
3167#[inline(never)]
3171fn build_metrics_only_server(
3172 runtime: RedDBRuntime,
3173 auth_store: Arc<AuthStore>,
3174 bind_addr: String,
3175) -> RedDBServer {
3176 RedDBServer::with_options(
3177 runtime,
3178 ServerOptions {
3179 bind_addr,
3180 surface: crate::server::ServerSurface::MetricsOnly,
3181 ..ServerOptions::default()
3182 },
3183 )
3184 .with_auth(auth_store)
3185}
3186
3187fn spawn_admin_metrics_listeners(runtime: &RedDBRuntime, auth_store: &Arc<AuthStore>) {
3191 if let Some(addr) = env_nonempty("RED_ADMIN_BIND") {
3192 let server = build_admin_only_server(runtime.clone(), auth_store.clone(), addr.clone());
3193 let _ = server.serve_in_background();
3194 tracing::info!(transport = "http", surface = "admin", bind = %addr, "listener online");
3195 }
3196 if let Some(addr) = env_nonempty("RED_METRICS_BIND") {
3197 let server = build_metrics_only_server(runtime.clone(), auth_store.clone(), addr.clone());
3198 let _ = server.serve_in_background();
3199 tracing::info!(transport = "http", surface = "metrics", bind = %addr, "listener online");
3200 }
3201}
3202
3203#[inline(never)]
3204fn run_http_server(config: ServerCommandConfig, bind_addr: String) -> Result<(), String> {
3205 let mut transport_readiness = TransportReadiness::default();
3206 let Some(listener) = bind_listener_for_startup(
3207 &mut transport_readiness,
3208 "http",
3209 &bind_addr,
3210 config.http_bind_explicit,
3211 )?
3212 else {
3213 return Err(format!(
3214 "no HTTP listener started; implicit bind {} failed",
3215 bind_addr
3216 ));
3217 };
3218 let db_options = config.to_db_options()?;
3219 let (runtime, auth_store, _telemetry_guard) =
3220 build_runtime_and_auth_store(&config, db_options.clone())?;
3221 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3222 spawn_admin_metrics_listeners(&runtime, &auth_store);
3223 spawn_http_tls_listener(&config, &runtime, &auth_store)?;
3224 let server = build_http_server_with_transport_readiness(
3225 runtime.clone(),
3226 auth_store,
3227 bind_addr.clone(),
3228 transport_readiness,
3229 );
3230 let server = apply_http_limits(server, &config, &runtime);
3231 let server = apply_ui_bundle(server, &config)?;
3232 tracing::info!(transport = "http", bind = %bind_addr, "listener online");
3233 server.serve_on(listener).map_err(|err| err.to_string())
3234}
3235
3236fn spawn_http_tls_listener(
3242 config: &ServerCommandConfig,
3243 runtime: &RedDBRuntime,
3244 auth_store: &Arc<AuthStore>,
3245) -> Result<(), String> {
3246 let Some(addr) = config.http_tls_bind_addr.clone() else {
3247 return Ok(());
3248 };
3249
3250 let tls_config = resolve_http_tls_config(config)?;
3251 let server_config = crate::server::tls::build_server_config(&tls_config)
3252 .map_err(|err| format!("HTTP TLS: {err}"))?;
3253
3254 let server = build_http_server(runtime.clone(), auth_store.clone(), addr.clone());
3255 let server = apply_http_limits(server, config, runtime);
3256 let _handle = server.serve_tls_in_background(server_config);
3257 tracing::info!(
3258 transport = "https",
3259 bind = %addr,
3260 mtls = %tls_config.client_ca_path.is_some(),
3261 "TLS listener online"
3262 );
3263 Ok(())
3264}
3265
3266fn resolve_http_tls_config(
3268 config: &ServerCommandConfig,
3269) -> Result<crate::server::tls::HttpTlsConfig, String> {
3270 match (&config.http_tls_cert, &config.http_tls_key) {
3271 (Some(cert), Some(key)) => Ok(crate::server::tls::HttpTlsConfig {
3272 cert_path: cert.clone(),
3273 key_path: key.clone(),
3274 client_ca_path: config.http_tls_client_ca.clone(),
3275 }),
3276 (None, None) => {
3277 let dir = config
3279 .path
3280 .as_ref()
3281 .and_then(|p| p.parent().map(std::path::PathBuf::from))
3282 .unwrap_or_else(|| std::path::PathBuf::from("."));
3283 let auto = crate::server::tls::auto_generate_dev_cert(&dir)
3284 .map_err(|err| format!("HTTP TLS dev: {err}"))?;
3285 Ok(crate::server::tls::HttpTlsConfig {
3286 cert_path: auto.cert_path,
3287 key_path: auto.key_path,
3288 client_ca_path: config.http_tls_client_ca.clone(),
3289 })
3290 }
3291 _ => Err("HTTP TLS requires both --http-tls-cert and --http-tls-key (or neither, with RED_HTTP_TLS_DEV=1)".to_string()),
3292 }
3293}
3294
3295#[inline(never)]
3296fn run_grpc_server(config: ServerCommandConfig, bind_addr: String) -> Result<(), String> {
3297 let workers = config.workers;
3298 let db_options = config.to_db_options()?;
3299 let rt_config = detect_runtime_config();
3300 let mut transport_readiness = TransportReadiness::default();
3301 let Some(grpc_listener) = bind_listener_for_startup(
3302 &mut transport_readiness,
3303 "grpc",
3304 &bind_addr,
3305 config.grpc_bind_explicit,
3306 )?
3307 else {
3308 return Err(format!(
3309 "no gRPC listener started; implicit bind {} failed",
3310 bind_addr
3311 ));
3312 };
3313
3314 let worker_threads = workers.unwrap_or(rt_config.suggested_workers);
3315
3316 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
3317 .enable_all()
3318 .worker_threads(worker_threads)
3319 .thread_stack_size(rt_config.stack_size)
3320 .build()
3321 .map_err(|err| format!("tokio runtime: {err}"))?;
3322
3323 let (runtime, auth_store, _telemetry_guard) =
3325 build_runtime_and_auth_store(&config, db_options.clone())?;
3326 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3327 let signal_runtime = runtime.clone();
3328 tokio_runtime.block_on(async move {
3329 spawn_lifecycle_signal_handler(signal_runtime).await;
3330 spawn_wire_listeners(&config, &runtime, &mut transport_readiness).await?;
3332
3333 spawn_pg_listener(&config, &runtime);
3335
3336 spawn_grpc_tls_listener_if_configured(&config, runtime.clone(), auth_store.clone());
3340
3341 let server = RedDBGrpcServer::with_options(
3342 runtime,
3343 GrpcServerOptions {
3344 bind_addr: bind_addr.clone(),
3345 tls: None,
3346 },
3347 auth_store,
3348 );
3349
3350 tracing::info!(
3351 transport = "grpc",
3352 bind = %bind_addr,
3353 cpus = rt_config.available_cpus,
3354 workers = worker_threads,
3355 "listener online"
3356 );
3357 server
3358 .serve_on(grpc_listener)
3359 .await
3360 .map_err(|err| err.to_string())
3361 })
3362}
3363
3364#[inline(never)]
3365fn run_dual_server(
3366 config: ServerCommandConfig,
3367 grpc_bind_addr: String,
3368 http_bind_addr: String,
3369) -> Result<(), String> {
3370 let workers = config.workers;
3371 let db_options = config.to_db_options()?;
3372 let rt_config = detect_runtime_config();
3373 let worker_threads = workers.unwrap_or(rt_config.suggested_workers);
3374 let mut transport_readiness = TransportReadiness::default();
3375 let http_listener = bind_listener_for_startup(
3376 &mut transport_readiness,
3377 "http",
3378 &http_bind_addr,
3379 config.http_bind_explicit,
3380 )?;
3381 let grpc_listener = bind_listener_for_startup(
3382 &mut transport_readiness,
3383 "grpc",
3384 &grpc_bind_addr,
3385 config.grpc_bind_explicit,
3386 )?;
3387 if http_listener.is_none() && grpc_listener.is_none() {
3388 return Err("no listener started; implicit HTTP and gRPC binds failed".to_string());
3389 }
3390 let (runtime, auth_store, _telemetry_guard) =
3391 build_runtime_and_auth_store(&config, db_options.clone())?;
3392 let _backup_tasks = spawn_backup_tasks_if_configured(&db_options, &runtime);
3393
3394 spawn_admin_metrics_listeners(&runtime, &auth_store);
3395 spawn_http_tls_listener(&config, &runtime, &auth_store)?;
3396
3397 let http_handle = if let Some(listener) = http_listener {
3398 let http_server = build_http_server_with_transport_readiness(
3399 runtime.clone(),
3400 auth_store.clone(),
3401 http_bind_addr.clone(),
3402 transport_readiness.clone(),
3403 );
3404 let http_server = apply_http_limits(http_server, &config, &runtime);
3405 let http_server = apply_ui_bundle(http_server, &config)?;
3406 Some(http_server.serve_in_background_on(listener))
3407 } else {
3408 None
3409 };
3410
3411 thread::sleep(Duration::from_millis(150));
3412 if let Some(handle) = http_handle.as_ref() {
3413 if handle.is_finished() {
3414 let handle = http_handle.unwrap();
3415 return match handle.join() {
3416 Ok(Ok(())) => Err("HTTP server exited unexpectedly".to_string()),
3417 Ok(Err(err)) => Err(err.to_string()),
3418 Err(_) => Err("HTTP server thread panicked".to_string()),
3419 };
3420 }
3421 }
3422 if grpc_listener.is_none() {
3423 let Some(handle) = http_handle else {
3424 return Err("no listener started".to_string());
3425 };
3426 return match handle.join() {
3427 Ok(Ok(())) => Err("HTTP server exited unexpectedly".to_string()),
3428 Ok(Err(err)) => Err(err.to_string()),
3429 Err(_) => Err("HTTP server thread panicked".to_string()),
3430 };
3431 }
3432 let grpc_listener = grpc_listener.expect("checked above");
3433
3434 let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
3435 .enable_all()
3436 .worker_threads(worker_threads)
3437 .thread_stack_size(rt_config.stack_size)
3438 .build()
3439 .map_err(|err| format!("tokio runtime: {err}"))?;
3440
3441 let signal_runtime = runtime.clone();
3442 tokio_runtime.block_on(async move {
3443 spawn_lifecycle_signal_handler(signal_runtime).await;
3444 spawn_wire_listeners(&config, &runtime, &mut transport_readiness).await?;
3446
3447 spawn_pg_listener(&config, &runtime);
3449
3450 spawn_grpc_tls_listener_if_configured(&config, runtime.clone(), auth_store.clone());
3452
3453 let server = RedDBGrpcServer::with_options(
3454 runtime,
3455 GrpcServerOptions {
3456 bind_addr: grpc_bind_addr.clone(),
3457 tls: None,
3458 },
3459 auth_store,
3460 );
3461
3462 tracing::info!(transport = "http", bind = %http_bind_addr, "listener online");
3463 tracing::info!(
3464 transport = "grpc",
3465 bind = %grpc_bind_addr,
3466 cpus = rt_config.available_cpus,
3467 workers = worker_threads,
3468 "listener online"
3469 );
3470 server
3471 .serve_on(grpc_listener)
3472 .await
3473 .map_err(|err| err.to_string())
3474 })
3475}
3476
3477#[cfg(test)]
3478mod tests {
3479 use super::*;
3480
3481 #[test]
3482 fn render_systemd_unit_contains_expected_execstart() {
3483 let config = SystemdServiceConfig {
3484 service_name: "reddb".to_string(),
3485 binary_path: PathBuf::from("/usr/local/bin/red"),
3486 run_user: "reddb".to_string(),
3487 run_group: "reddb".to_string(),
3488 data_path: reddb_file::default_service_database_path(),
3489 router_bind_addr: None,
3490 grpc_bind_addr: Some("0.0.0.0:55055".to_string()),
3491 http_bind_addr: None,
3492 };
3493
3494 let unit = render_systemd_unit(&config);
3495 assert!(unit.contains("ExecStart=/usr/local/bin/red server --path /var/lib/reddb/data.rdb --grpc-bind 0.0.0.0:55055"));
3496 assert!(unit.contains("ReadWritePaths=/var/lib/reddb"));
3497 }
3498
3499 #[test]
3500 fn systemd_service_config_derives_paths() {
3501 let config = SystemdServiceConfig {
3502 service_name: "reddb-api".to_string(),
3503 binary_path: PathBuf::from("/usr/local/bin/red"),
3504 run_user: "reddb".to_string(),
3505 run_group: "reddb".to_string(),
3506 data_path: PathBuf::from("/srv/reddb/live/data.rdb"),
3507 router_bind_addr: None,
3508 grpc_bind_addr: None,
3509 http_bind_addr: Some("127.0.0.1:5000".to_string()),
3510 };
3511
3512 assert_eq!(config.data_dir(), PathBuf::from("/srv/reddb/live"));
3513 assert_eq!(
3514 config.unit_path(),
3515 PathBuf::from("/etc/systemd/system/reddb-api.service")
3516 );
3517 }
3518
3519 #[test]
3520 fn render_systemd_unit_supports_dual_transport() {
3521 let config = SystemdServiceConfig {
3522 service_name: "reddb".to_string(),
3523 binary_path: PathBuf::from("/usr/local/bin/red"),
3524 run_user: "reddb".to_string(),
3525 run_group: "reddb".to_string(),
3526 data_path: reddb_file::default_service_database_path(),
3527 router_bind_addr: None,
3528 grpc_bind_addr: Some("0.0.0.0:55055".to_string()),
3529 http_bind_addr: Some("0.0.0.0:5000".to_string()),
3530 };
3531
3532 let unit = render_systemd_unit(&config);
3533 assert!(unit.contains("--grpc-bind 0.0.0.0:55055"));
3534 assert!(unit.contains("--http-bind 0.0.0.0:5000"));
3535 }
3536
3537 #[test]
3538 fn render_systemd_unit_supports_router_mode() {
3539 let config = SystemdServiceConfig {
3540 service_name: "reddb".to_string(),
3541 binary_path: PathBuf::from("/usr/local/bin/red"),
3542 run_user: "reddb".to_string(),
3543 run_group: "reddb".to_string(),
3544 data_path: reddb_file::default_service_database_path(),
3545 router_bind_addr: Some(DEFAULT_ROUTER_BIND_ADDR.to_string()),
3546 grpc_bind_addr: None,
3547 http_bind_addr: None,
3548 };
3549
3550 let unit = render_systemd_unit(&config);
3551 assert!(unit.contains("--bind 127.0.0.1:5050"));
3552 assert!(!unit.contains("--grpc-bind"));
3553 assert!(!unit.contains("--http-bind"));
3554 }
3555
3556 #[test]
3557 fn explicit_bind_collision_is_fatal() {
3558 let held = TcpListener::bind("127.0.0.1:0").expect("hold test port");
3559 let addr = held.local_addr().expect("held addr").to_string();
3560 let mut readiness = TransportReadiness::default();
3561
3562 let error = bind_listener_for_startup(&mut readiness, "http", &addr, true).unwrap_err();
3563
3564 assert!(error.contains("explicit http listener bind"));
3565 assert_eq!(readiness.active.len(), 0);
3566 assert_eq!(readiness.failed.len(), 1);
3567 assert!(readiness.failed[0].explicit);
3568 assert_eq!(readiness.failed[0].bind_addr, addr);
3569 }
3570
3571 fn no_auth_env_lock() -> &'static std::sync::Mutex<()> {
3578 static LOCK: std::sync::OnceLock<std::sync::Mutex<()>> = std::sync::OnceLock::new();
3579 LOCK.get_or_init(|| std::sync::Mutex::new(()))
3580 }
3581
3582 fn no_auth_test_config(no_auth: bool) -> ServerCommandConfig {
3583 ServerCommandConfig {
3584 path: None,
3585 router_bind_addr: Some(DEFAULT_ROUTER_BIND_ADDR.to_string()),
3586 router_bind_explicit: false,
3587 grpc_bind_addr: None,
3588 grpc_bind_explicit: false,
3589 grpc_tls_bind_addr: None,
3590 grpc_tls_cert: None,
3591 grpc_tls_key: None,
3592 grpc_tls_client_ca: None,
3593 http_bind_addr: None,
3594 http_bind_explicit: false,
3595 http_tls_bind_addr: None,
3596 http_tls_cert: None,
3597 http_tls_key: None,
3598 http_tls_client_ca: None,
3599 wire_bind_addr: None,
3600 wire_bind_explicit: false,
3601 wire_tls_bind_addr: None,
3602 wire_tls_cert: None,
3603 wire_tls_key: None,
3604 pg_bind_addr: None,
3605 create_if_missing: true,
3606 read_only: false,
3607 role: "standalone".to_string(),
3608 primary_addr: None,
3609 storage_profile: StorageProfileSelection::embedded_single_file(),
3610 auth: false,
3611 require_auth: false,
3612 vault: true,
3615 no_auth,
3616 workers: None,
3617 telemetry: None,
3618 http_limits_cli: crate::server::HttpLimitsCliInput::default(),
3619 ui: false,
3620 ui_dir: None,
3621 bootstrap: BootstrapConfig::default(),
3622 }
3623 }
3624
3625 #[test]
3626 fn no_auth_flag_disables_every_auth_knob_and_stamps_metadata() {
3627 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3628 unsafe {
3633 std::env::set_var("REDDB_USERNAME", "admin");
3634 std::env::set_var("REDDB_PASSWORD", "hunter2");
3635 }
3636 let config = no_auth_test_config(true);
3637 let options = config.to_db_options().expect("to_db_options");
3638
3639 assert!(no_auth_active(&options), "metadata should be stamped");
3640 assert!(!options.auth.enabled, "auth.enabled must be forced off");
3641 assert!(
3642 !options.auth.require_auth,
3643 "require_auth must be forced off"
3644 );
3645 assert!(
3646 !options.auth.vault_enabled,
3647 "vault_enabled must be forced off (overrides --vault)"
3648 );
3649 assert_eq!(
3650 options.metadata.get(NO_AUTH_META).map(String::as_str),
3651 Some("true"),
3652 );
3653
3654 unsafe {
3656 std::env::remove_var("REDDB_USERNAME");
3657 std::env::remove_var("REDDB_PASSWORD");
3658 }
3659 }
3660
3661 #[test]
3662 fn default_behaviour_without_no_auth_flag_is_unchanged() {
3663 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3664 let config = no_auth_test_config(false);
3665 let options = config.to_db_options().expect("to_db_options");
3666
3667 assert!(
3668 !no_auth_active(&options),
3669 "default boot must not be marked no-auth"
3670 );
3671 assert!(
3672 options.metadata.get(NO_AUTH_META).is_none(),
3673 "metadata key must be absent when flag is off"
3674 );
3675 assert!(options.auth.vault_enabled);
3677 }
3678
3679 #[test]
3680 fn no_auth_active_blocks_bootstrap_from_env() {
3681 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3682 unsafe {
3687 std::env::set_var("REDDB_USERNAME", "admin");
3688 std::env::set_var("REDDB_PASSWORD", "hunter2");
3689 }
3690
3691 let options = no_auth_test_config(true)
3692 .to_db_options()
3693 .expect("to_db_options");
3694
3695 let auth_store = AuthStore::new(options.auth.clone());
3699 if !no_auth_active(&options) {
3700 auth_store.bootstrap_from_env();
3701 }
3702
3703 assert!(
3704 auth_store.needs_bootstrap(),
3705 "no admin user must be bootstrapped under --no-auth even with REDDB_USERNAME/PASSWORD set"
3706 );
3707
3708 unsafe {
3710 std::env::remove_var("REDDB_USERNAME");
3711 std::env::remove_var("REDDB_PASSWORD");
3712 }
3713 }
3714
3715 fn clear_preset_env() {
3722 unsafe {
3724 std::env::remove_var(BOOTSTRAP_PRESET_ENV);
3725 std::env::remove_var(PRESET_ENV);
3726 std::env::remove_var("REDDB_BOOTSTRAP_MANIFEST");
3727 std::env::remove_var("REDDB_AUTH");
3728 std::env::remove_var("REDDB_REQUIRE_AUTH");
3729 std::env::remove_var("REDDB_NO_AUTH");
3730 std::env::remove_var("REDDB_DEV");
3731 std::env::remove_var("REDDB_VAULT");
3732 std::env::remove_var("REDDB_USERNAME");
3733 std::env::remove_var("REDDB_PASSWORD");
3734 std::env::remove_var("REDDB_USERNAME_FILE");
3735 std::env::remove_var("REDDB_PASSWORD_FILE");
3736 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN");
3737 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD");
3738 std::env::remove_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD_FILE");
3739 std::env::remove_var("REDDB_CUSTOMER_ADMIN");
3740 std::env::remove_var("REDDB_CUSTOMER_ADMIN_PASSWORD");
3741 std::env::remove_var("REDDB_CUSTOMER_ADMIN_PASSWORD_FILE");
3742 }
3743 }
3744
3745 fn clear_backup_env() {
3746 unsafe {
3748 std::env::remove_var("REDDB_BACKUP_S3_ENDPOINT");
3749 std::env::remove_var("REDDB_BACKUP_S3_BUCKET");
3750 std::env::remove_var("REDDB_BACKUP_S3_PREFIX");
3751 std::env::remove_var("REDDB_BACKUP_S3_ACCESS_KEY_ID");
3752 std::env::remove_var("REDDB_BACKUP_S3_SECRET_ACCESS_KEY");
3753 std::env::remove_var("REDDB_BACKUP_S3_REGION");
3754 std::env::remove_var("REDDB_BACKUP_CHECKPOINT_INTERVAL_SECS");
3755 std::env::remove_var("REDDB_BACKUP_WAL_FLUSH_INTERVAL_SECS");
3756 std::env::remove_var("REDDB_BACKUP_PAUSE_ON_LAG_SECS");
3757 }
3758 }
3759
3760 fn fresh_runtime_and_store() -> (RedDBRuntime, Arc<AuthStore>) {
3761 let runtime = RedDBRuntime::with_options(RedDBOptions::in_memory()).expect("runtime");
3762 let auth_store = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
3763 (runtime, auth_store)
3764 }
3765
3766 #[test]
3767 fn auth_env_knobs_enable_auth_require_auth_and_vault() {
3768 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3769 clear_preset_env();
3770 unsafe {
3772 std::env::set_var("REDDB_AUTH", "true");
3773 std::env::set_var("REDDB_REQUIRE_AUTH", "true");
3774 std::env::set_var("REDDB_VAULT", "true");
3775 }
3776
3777 let mut config = no_auth_test_config(false);
3778 config.vault = false;
3779 let options = config.to_db_options().expect("to_db_options");
3780
3781 assert!(options.auth.enabled);
3782 assert!(options.auth.require_auth);
3783 assert!(options.auth.vault_enabled);
3784
3785 clear_preset_env();
3786 }
3787
3788 #[test]
3789 fn production_and_cloud_presets_force_auth_require_auth_and_vault() {
3790 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3791 clear_preset_env();
3792
3793 for preset in [PRESET_PRODUCTION, PRESET_CLOUD] {
3794 let mut config = no_auth_test_config(false);
3795 config.vault = false;
3796 config.bootstrap.preset = Some(preset.to_string());
3797
3798 let options = config.to_db_options().expect("to_db_options");
3799 assert!(options.auth.enabled, "{preset} should enable auth");
3800 assert!(
3801 options.auth.require_auth,
3802 "{preset} should require authenticated requests"
3803 );
3804 assert!(options.auth.vault_enabled, "{preset} should enable vault");
3805 assert!(!no_auth_active(&options));
3806 }
3807
3808 clear_preset_env();
3809 }
3810
3811 #[test]
3812 fn no_auth_env_overrides_preset_forced_auth() {
3813 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3814 clear_preset_env();
3815 unsafe {
3817 std::env::set_var("REDDB_NO_AUTH", "true");
3818 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_CLOUD);
3819 }
3820
3821 let mut config = no_auth_test_config(false);
3822 config.auth = true;
3823 config.require_auth = true;
3824 let options = config.to_db_options().expect("to_db_options");
3825
3826 assert!(no_auth_active(&options));
3827 assert!(!options.auth.enabled);
3828 assert!(!options.auth.require_auth);
3829 assert!(!options.auth.vault_enabled);
3830
3831 clear_preset_env();
3832 }
3833
3834 #[test]
3835 fn simple_preset_is_default_and_persists_state() {
3836 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3837 clear_preset_env();
3838
3839 let (runtime, auth_store) = fresh_runtime_and_store();
3840 apply_preset(&runtime, &auth_store).expect("simple preset applies cleanly");
3841
3842 assert!(
3844 auth_store.needs_bootstrap(),
3845 "simple preset must not create an admin"
3846 );
3847
3848 let store = runtime.db().store();
3850 let completed = store
3851 .get_config(BOOTSTRAP_COMPLETED_KEY)
3852 .expect("completed key persisted");
3853 assert!(matches!(
3854 completed,
3855 crate::storage::schema::Value::Boolean(true)
3856 ));
3857 let preset = store
3858 .get_config(BOOTSTRAP_PRESET_KEY)
3859 .expect("preset key persisted");
3860 match preset {
3861 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_SIMPLE),
3862 other => panic!("expected Text(simple), got {other:?}"),
3863 }
3864 assert!(
3865 store.get_config(BOOTSTRAP_FIRST_ADMIN_KEY).is_none(),
3866 "simple preset must not record a first admin"
3867 );
3868
3869 clear_preset_env();
3870 }
3871
3872 #[test]
3873 fn production_preset_creates_first_admin_with_allow_all_policy() {
3874 use crate::auth::policies::{EvalContext, ResourceRef};
3875 use crate::auth::UserId;
3876
3877 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3878 clear_preset_env();
3879 unsafe {
3881 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
3882 std::env::set_var("REDDB_USERNAME", "ops");
3883 std::env::set_var("REDDB_PASSWORD", "hunter2");
3884 }
3885
3886 let (runtime, auth_store) = fresh_runtime_and_store();
3887 apply_preset(&runtime, &auth_store).expect("production preset applies cleanly");
3888
3889 assert!(
3891 !auth_store.needs_bootstrap(),
3892 "production preset must seal bootstrap"
3893 );
3894 let users = auth_store.list_users();
3895 assert_eq!(users.len(), 1);
3896 let admin = &users[0];
3897 assert_eq!(admin.username, "ops");
3898 assert!(
3899 admin.tenant_id.is_none(),
3900 "first admin must be platform-scoped (tenant=None)"
3901 );
3902
3903 let policy = auth_store
3905 .get_policy(FIRST_ADMIN_ALLOW_ALL_POLICY)
3906 .expect("allow-all policy installed");
3907 assert!(!policy.statements.is_empty());
3908
3909 let actor = UserId::platform("ops");
3912 let ctx = EvalContext {
3913 principal_tenant: None,
3914 current_tenant: None,
3915 peer_ip: None,
3916 mfa_present: false,
3917 now_ms: 1_700_000_000_000,
3918 principal_is_admin_role: true,
3919 principal_is_platform_scoped: true,
3920 };
3921 let arbitrary_resource = ResourceRef::new("config", "red.config.audit.enabled");
3922 assert!(
3923 auth_store.check_policy_authz(&actor, "config:write", &arbitrary_resource, &ctx),
3924 "allow-all policy must grant arbitrary actions via the evaluator"
3925 );
3926
3927 let store = runtime.db().store();
3929 match store
3930 .get_config(BOOTSTRAP_FIRST_ADMIN_KEY)
3931 .expect("first_admin_id persisted")
3932 {
3933 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), "ops"),
3934 other => panic!("expected Text(ops), got {other:?}"),
3935 }
3936 match store.get_config(BOOTSTRAP_PRESET_KEY).unwrap() {
3937 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_PRODUCTION),
3938 other => panic!("expected Text(production), got {other:?}"),
3939 }
3940
3941 clear_preset_env();
3942 }
3943
3944 #[test]
3945 fn bootstrap_preset_env_takes_precedence_over_legacy_preset_env() {
3946 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3947 clear_preset_env();
3948 unsafe {
3950 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_REGULATED);
3951 std::env::set_var(PRESET_ENV, PRESET_SIMPLE);
3952 }
3953
3954 let options = no_auth_test_config(false)
3955 .to_db_options()
3956 .expect("regulated options");
3957 assert!(
3958 options.control_events.compliance_mode,
3959 "canonical REDDB_BOOTSTRAP_PRESET should win over REDDB_PRESET"
3960 );
3961 assert!(options.query_audit.enabled);
3962
3963 clear_preset_env();
3964 }
3965
3966 #[test]
3967 fn cloud_preset_creates_system_head_and_customer_admins() {
3968 use crate::auth::policies::{EvalContext, ResourceRef};
3969 use crate::auth::UserId;
3970
3971 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
3972 clear_preset_env();
3973 unsafe {
3975 std::env::set_var(BOOTSTRAP_PRESET_ENV, PRESET_CLOUD);
3976 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN", "head");
3977 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD", "head-pass");
3978 std::env::set_var("REDDB_CUSTOMER_ADMIN", "customer");
3979 std::env::set_var("REDDB_CUSTOMER_ADMIN_PASSWORD", "customer-pass");
3980 }
3981
3982 let (runtime, auth_store) = fresh_runtime_and_store();
3983 apply_preset(&runtime, &auth_store).expect("cloud preset applies cleanly");
3984
3985 let head = auth_store
3986 .get_user(None, "head")
3987 .expect("head admin should exist");
3988 assert_eq!(head.tenant_id, None);
3989 let customer = auth_store
3990 .get_user(None, "customer")
3991 .expect("customer admin should exist");
3992 assert_eq!(customer.tenant_id, None);
3993
3994 let ctx = EvalContext {
3995 principal_tenant: None,
3996 current_tenant: None,
3997 peer_ip: None,
3998 mfa_present: false,
3999 now_ms: 1_700_000_000_000,
4000 principal_is_admin_role: true,
4001 principal_is_platform_scoped: true,
4002 };
4003 assert!(auth_store.check_policy_authz(
4004 &UserId::platform("customer"),
4005 "config:write",
4006 &ResourceRef::new("config", "red.config.customer.enabled"),
4007 &ctx,
4008 ));
4009 assert!(
4010 auth_store.check_policy_authz(
4011 &UserId::platform("head"),
4012 "config:write",
4013 &ResourceRef::new("config", "red.config.cloud.enabled"),
4014 &ctx,
4015 ),
4016 "cloud head keeps allow-all authority over managed cloud config"
4017 );
4018 assert!(
4019 !auth_store.check_policy_authz(
4020 &UserId::platform("customer"),
4021 "config:write",
4022 &ResourceRef::new("config", "red.config.cloud.enabled"),
4023 &ctx,
4024 ),
4025 "cloud customer must be denied managed cloud config writes"
4026 );
4027 assert!(
4028 !auth_store.check_policy_authz(
4029 &UserId::platform("customer"),
4030 "policy:drop",
4031 &ResourceRef::new("policy", CLOUD_PROTECT_MANAGED_POLICY),
4032 &ctx,
4033 ),
4034 "cloud customer must be denied mutations of the managed guardrail policy"
4035 );
4036
4037 assert!(auth_store.get_user(None, "head").is_some());
4038 assert!(auth_store
4039 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4040 .is_some());
4041 assert!(runtime
4042 .config_registry()
4043 .get_active(CLOUD_CONFIG_NAMESPACE)
4044 .is_some());
4045
4046 let store = runtime.db().store();
4047 match store
4048 .get_config(BOOTSTRAP_FIRST_ADMIN_KEY)
4049 .expect("head admin id persisted")
4050 {
4051 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), "head"),
4052 other => panic!("expected Text(head), got {other:?}"),
4053 }
4054 match store.get_config(BOOTSTRAP_PRESET_KEY).unwrap() {
4055 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_CLOUD),
4056 other => panic!("expected Text(cloud), got {other:?}"),
4057 }
4058
4059 clear_preset_env();
4060 }
4061
4062 #[test]
4063 fn cloud_preset_customer_cannot_disable_head_admin_or_drop_guardrail_policy() {
4064 use crate::auth::Role;
4065 use crate::runtime::mvcc::{clear_current_auth_identity, set_current_auth_identity};
4066
4067 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4068 clear_preset_env();
4069
4070 let bootstrap = BootstrapConfig {
4071 preset: Some(PRESET_CLOUD.to_string()),
4072 cloud_head_admin: Some("red_admin".to_string()),
4073 cloud_head_admin_password: Some("head-pass".to_string()),
4074 customer_admin: Some("admin".to_string()),
4075 customer_admin_password: Some("customer-pass".to_string()),
4076 ..BootstrapConfig::default()
4077 };
4078 let (runtime, auth_store) = fresh_runtime_and_store();
4079 apply_preset_from_config(&runtime, &auth_store, &bootstrap)
4080 .expect("cloud preset applies cleanly");
4081 runtime.set_auth_store(Arc::clone(&auth_store));
4082
4083 set_current_auth_identity("admin".to_string(), Role::Admin);
4084 let disable_head = runtime.execute_query("ALTER USER red_admin DISABLE");
4085 let drop_guardrail =
4086 runtime.execute_query(&format!("DROP POLICY '{}'", CLOUD_PROTECT_MANAGED_POLICY));
4087 let create_user = runtime.execute_query("CREATE USER app_user WITH PASSWORD 'p' ROLE read");
4088 let disable_user = runtime.execute_query("ALTER USER app_user DISABLE");
4089 clear_current_auth_identity();
4090
4091 let disable_head_err = disable_head.expect_err("customer admin must not disable red_admin");
4092 assert!(
4093 disable_head_err.to_string().contains("user:disable"),
4094 "error should name the denied lifecycle action: {disable_head_err}"
4095 );
4096 assert!(
4097 auth_store
4098 .get_user(None, "red_admin")
4099 .expect("red_admin should still exist")
4100 .enabled,
4101 "denied mutation must leave red_admin enabled"
4102 );
4103
4104 let drop_guardrail_err =
4105 drop_guardrail.expect_err("customer admin must not drop cloud guardrail policy");
4106 assert!(
4107 drop_guardrail_err.to_string().contains("managed policy"),
4108 "error should name the managed policy guardrail: {drop_guardrail_err}"
4109 );
4110 assert!(auth_store
4111 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4112 .is_some());
4113
4114 create_user.expect("customer admin should retain normal create-user authority");
4115 disable_user.expect("customer admin should retain normal user-disable authority");
4116 assert!(
4117 !auth_store
4118 .get_user(None, "app_user")
4119 .expect("app_user should exist")
4120 .enabled,
4121 "normal capability should still apply outside the reserved head admin"
4122 );
4123
4124 clear_preset_env();
4125 }
4126
4127 #[test]
4128 fn cloud_preset_cli_admin_alias_wins_over_cloud_env_password() {
4129 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4130 clear_preset_env();
4131 unsafe {
4133 std::env::set_var("REDDB_CLOUD_HEAD_ADMIN_PASSWORD", "env-head-pass");
4134 }
4135
4136 let bootstrap = BootstrapConfig {
4137 preset: Some(PRESET_CLOUD.to_string()),
4138 admin_username: Some("head".to_string()),
4139 admin_password: Some("cli-head-pass".to_string()),
4140 customer_admin: Some("customer".to_string()),
4141 customer_admin_password: Some("customer-pass".to_string()),
4142 ..BootstrapConfig::default()
4143 };
4144 let (runtime, auth_store) = fresh_runtime_and_store();
4145 apply_preset_from_config(&runtime, &auth_store, &bootstrap)
4146 .expect("cloud preset applies cleanly");
4147
4148 auth_store
4149 .authenticate("head", "cli-head-pass")
4150 .expect("CLI alias password should win");
4151 assert!(
4152 auth_store.authenticate("head", "env-head-pass").is_err(),
4153 "cloud-specific env password must not beat CLI alias"
4154 );
4155
4156 clear_preset_env();
4157 }
4158
4159 #[test]
4160 fn bootstrap_intent_manifest_matches_cloud_flag_path() {
4161 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4162 clear_preset_env();
4163
4164 let manifest_dir = std::env::current_dir()
4165 .expect("current dir")
4166 .join(".red/tmp/bootstrap-manifest-tests");
4167 std::fs::create_dir_all(&manifest_dir).expect("create manifest test dir");
4168 let unique = format!(
4169 "{}-{}",
4170 std::process::id(),
4171 std::time::SystemTime::now()
4172 .duration_since(std::time::UNIX_EPOCH)
4173 .unwrap_or_default()
4174 .as_millis()
4175 );
4176 let head_password = manifest_dir.join(format!("head-{unique}.secret"));
4177 let customer_password = manifest_dir.join(format!("customer-{unique}.secret"));
4178 let manifest_path = manifest_dir.join(format!("intent-{unique}.json"));
4179 std::fs::write(&head_password, "head-pass\n").expect("write head secret");
4180 std::fs::write(&customer_password, "customer-pass\n").expect("write customer secret");
4181 std::fs::write(
4182 &manifest_path,
4183 format!(
4184 r#"{{
4185 "bootstrap": {{
4186 "preset": "cloud",
4187 "cloud_head_admin": {{
4188 "username": "head",
4189 "password_file": "{}"
4190 }},
4191 "customer_admin": {{
4192 "username": "customer",
4193 "password_file": "{}"
4194 }}
4195 }}
4196 }}"#,
4197 head_password.display(),
4198 customer_password.display()
4199 ),
4200 )
4201 .expect("write manifest");
4202
4203 let mut config = no_auth_test_config(false);
4204 config.bootstrap.manifest = Some(manifest_path.clone());
4205 let options = config.to_db_options().expect("manifest options");
4206 assert!(options.auth.enabled);
4207 assert!(options.auth.require_auth);
4208 assert!(options.auth.vault_enabled);
4209
4210 let (runtime, auth_store) = fresh_runtime_and_store();
4211 apply_preset_from_config(&runtime, &auth_store, &config.bootstrap)
4212 .expect("manifest intent applies cleanly");
4213
4214 auth_store
4215 .authenticate("head", "head-pass")
4216 .expect("head password comes from file");
4217 auth_store
4218 .authenticate("customer", "customer-pass")
4219 .expect("customer password comes from file");
4220 assert!(auth_store.get_user(None, "head").is_some());
4221 assert!(auth_store.get_user(None, "customer").is_some());
4222 assert!(auth_store
4223 .get_policy(CLOUD_PROTECT_MANAGED_POLICY)
4224 .is_some());
4225 match runtime
4226 .db()
4227 .store()
4228 .get_config(BOOTSTRAP_PRESET_KEY)
4229 .expect("preset key persisted")
4230 {
4231 crate::storage::schema::Value::Text(s) => assert_eq!(s.as_ref(), PRESET_CLOUD),
4232 other => panic!("expected Text(cloud), got {other:?}"),
4233 }
4234
4235 apply_preset_from_config(&runtime, &auth_store, &config.bootstrap)
4236 .expect("second boot is idempotent");
4237
4238 let _ = std::fs::remove_file(&manifest_path);
4239 let _ = std::fs::remove_file(&head_password);
4240 let _ = std::fs::remove_file(&customer_password);
4241 clear_preset_env();
4242 }
4243
4244 #[test]
4245 fn regulated_preset_enables_query_audit_infrastructure_without_rules() {
4246 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4247 clear_preset_env();
4248 unsafe {
4250 std::env::set_var(PRESET_ENV, PRESET_REGULATED);
4251 }
4252
4253 let (runtime, auth_store) = fresh_runtime_and_store();
4254 apply_preset(&runtime, &auth_store).expect("regulated preset applies cleanly");
4255
4256 assert!(runtime.query_audit().is_enabled());
4257 assert!(runtime.query_audit().rules().is_empty());
4258 assert!(
4259 runtime
4260 .db()
4261 .store()
4262 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4263 .is_some(),
4264 "regulated preset should create the query-audit stream"
4265 );
4266
4267 runtime
4268 .execute_query("CREATE TABLE docs (id INT)")
4269 .expect("create table");
4270 runtime
4271 .execute_query("INSERT INTO docs (id) VALUES (1)")
4272 .expect("insert");
4273 runtime.execute_query("SELECT * FROM docs").expect("select");
4274 let rows = runtime
4275 .db()
4276 .store()
4277 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4278 .expect("query audit collection")
4279 .query_all(|_| true);
4280 assert!(
4281 rows.is_empty(),
4282 "regulated preset must not globally audit every query"
4283 );
4284
4285 clear_preset_env();
4286 }
4287
4288 #[test]
4289 fn managed_backup_env_rejects_primary_replica_single_file_storage() {
4290 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4291 clear_backup_env();
4292 unsafe {
4294 std::env::set_var("REDDB_BACKUP_S3_ENDPOINT", "https://s3.example.test");
4295 std::env::set_var("REDDB_BACKUP_S3_BUCKET", "reddb");
4296 std::env::set_var("REDDB_BACKUP_S3_PREFIX", "clusters/prod");
4297 std::env::set_var("REDDB_BACKUP_S3_ACCESS_KEY_ID", "AK");
4298 std::env::set_var("REDDB_BACKUP_S3_SECRET_ACCESS_KEY", "SK");
4299 }
4300
4301 let mut config = no_auth_test_config(false);
4302 config.role = "primary".to_string();
4303 config.storage_profile = crate::storage::StorageDeployPreset::PrimaryReplicaDev.selection();
4304
4305 let err = config.to_db_options().unwrap_err();
4306 assert!(err.contains("managed backup"), "got: {err}");
4307 assert!(err.contains("operational-directory"), "got: {err}");
4308
4309 clear_backup_env();
4310 }
4311
4312 #[test]
4313 fn regulated_preset_installs_managed_evidence_guardrails_end_to_end() {
4314 use crate::auth::policies::{EvalContext, Policy, ResourceRef};
4315 use crate::auth::store::PrincipalRef;
4316 use crate::auth::{Role, UserId};
4317 use crate::runtime::mvcc::{clear_current_auth_identity, set_current_auth_identity};
4318 use crate::storage::schema::Value;
4319
4320 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4321 clear_preset_env();
4322 unsafe {
4324 std::env::set_var(PRESET_ENV, PRESET_REGULATED);
4325 }
4326
4327 let options = no_auth_test_config(false)
4328 .to_db_options()
4329 .expect("regulated options");
4330 assert!(
4331 options.control_events.compliance_mode,
4332 "regulated preset must enable fail-closed control evidence before runtime boot"
4333 );
4334 assert!(
4335 options.query_audit.enabled && options.query_audit.rules.is_empty(),
4336 "regulated preset must enable query-audit infrastructure without global rules"
4337 );
4338
4339 let runtime = RedDBRuntime::with_options(options).expect("runtime");
4340 let auth_store = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4341 apply_preset(&runtime, &auth_store).expect("regulated preset applies cleanly");
4342 runtime.set_auth_store(Arc::clone(&auth_store));
4343
4344 assert!(runtime.control_events_require_persistence());
4345 assert!(runtime.query_audit().is_enabled());
4346 assert!(runtime.query_audit().rules().is_empty());
4347 assert!(auth_store
4348 .get_policy(REGULATED_PROTECT_MANAGED_POLICY)
4349 .is_some());
4350
4351 let managed_policy = runtime
4352 .config_registry()
4353 .get_active(REGULATED_PROTECT_MANAGED_POLICY)
4354 .expect("regulated managed policy registry entry");
4355 assert!(managed_policy.managed);
4356 assert_eq!(managed_policy.resource_type, "policy");
4357 assert!(
4358 runtime
4359 .config_registry()
4360 .get_active(REGULATED_AUDIT_CONFIG_NAMESPACE)
4361 .expect("regulated audit config namespace")
4362 .managed
4363 );
4364
4365 let registry_rows = runtime
4366 .execute_query(&format!(
4367 "SELECT id, managed FROM red.registry WHERE id = '{}'",
4368 REGULATED_PROTECT_MANAGED_POLICY
4369 ))
4370 .expect("red.registry query");
4371 assert_eq!(registry_rows.result.records.len(), 1);
4372 assert_eq!(
4373 registry_rows.result.records[0].get("managed"),
4374 Some(&Value::Boolean(true))
4375 );
4376
4377 let managed_policy_rows = runtime
4378 .execute_query(&format!(
4379 "SELECT policy_id FROM red.managed_policies WHERE policy_id = '{}'",
4380 REGULATED_PROTECT_MANAGED_POLICY
4381 ))
4382 .expect("red.managed_policies query");
4383 assert_eq!(managed_policy_rows.result.records.len(), 1);
4384
4385 let capability_rows = runtime
4386 .execute_query(
4387 "SELECT action FROM red.control_capabilities WHERE action = 'evidence:export'",
4388 )
4389 .expect("red.control_capabilities query");
4390 assert_eq!(capability_rows.result.records.len(), 1);
4391
4392 auth_store
4393 .create_user("alice", "p", Role::Admin)
4394 .expect("create ordinary admin");
4395 let allow_all = Policy::from_json_str(
4396 r#"{
4397 "id": "alice-allow-all",
4398 "version": 1,
4399 "statements": [{
4400 "effect": "allow",
4401 "actions": ["*"],
4402 "resources": ["*"]
4403 }]
4404 }"#,
4405 )
4406 .expect("allow-all policy");
4407 auth_store.put_policy(allow_all).expect("install allow-all");
4408 auth_store
4409 .attach_policy(
4410 PrincipalRef::User(UserId::platform("alice")),
4411 "alice-allow-all",
4412 )
4413 .expect("attach allow-all");
4414 let ctx = EvalContext {
4415 principal_tenant: None,
4416 current_tenant: None,
4417 peer_ip: None,
4418 mfa_present: false,
4419 now_ms: 1_700_000_000_000,
4420 principal_is_admin_role: true,
4421 principal_is_platform_scoped: true,
4422 };
4423 assert!(
4424 !auth_store.check_policy_authz(
4425 &UserId::platform("alice"),
4426 "policy:drop",
4427 &ResourceRef::new("policy", REGULATED_PROTECT_MANAGED_POLICY),
4428 &ctx,
4429 ),
4430 "managed guardrail deny policy must be effective for ordinary admins"
4431 );
4432
4433 set_current_auth_identity("alice".to_string(), Role::Admin);
4434 let denied = runtime.execute_query(&format!(
4435 "DROP POLICY '{}'",
4436 REGULATED_PROTECT_MANAGED_POLICY
4437 ));
4438 clear_current_auth_identity();
4439 let err = denied.expect_err("managed policy guardrail must deny ordinary admin");
4440 assert!(
4441 err.to_string().contains("managed policy"),
4442 "error should name the managed guardrail: {err}"
4443 );
4444 assert!(
4445 auth_store
4446 .get_policy(REGULATED_PROTECT_MANAGED_POLICY)
4447 .is_some(),
4448 "denied mutation must leave managed policy installed"
4449 );
4450
4451 let denied_events = runtime
4452 .execute_query(&format!(
4453 "SELECT action, resource, outcome FROM red.control_events \
4454 WHERE action = 'policy:drop' AND resource = 'policy:{}'",
4455 REGULATED_PROTECT_MANAGED_POLICY
4456 ))
4457 .expect("red.control_events denied policy drop");
4458 assert_eq!(denied_events.result.records.len(), 1);
4459 assert_eq!(
4460 denied_events.result.records[0].get("outcome"),
4461 Some(&Value::text("denied"))
4462 );
4463
4464 set_current_auth_identity("alice".to_string(), Role::Admin);
4465 let config_denied = runtime.execute_query("SET CONFIG red.config.audit.enabled = true");
4466 clear_current_auth_identity();
4467 let err = config_denied.expect_err("managed config guardrail must deny ordinary admin");
4468 assert!(
4469 err.to_string().contains("managed config"),
4470 "error should name the managed config guardrail: {err}"
4471 );
4472
4473 let denied_config_events = runtime
4474 .execute_query(
4475 "SELECT action, resource, outcome FROM red.control_events \
4476 WHERE action = 'config:write' AND resource = 'config:red.config.audit.enabled'",
4477 )
4478 .expect("red.control_events denied config write");
4479 assert_eq!(denied_config_events.result.records.len(), 1);
4480 assert_eq!(
4481 denied_config_events.result.records[0].get("outcome"),
4482 Some(&Value::text("denied"))
4483 );
4484
4485 runtime
4486 .execute_query("CREATE TABLE regulated_docs (id INT)")
4487 .expect("create user table");
4488 runtime
4489 .execute_query("SELECT * FROM regulated_docs")
4490 .expect("select user table");
4491 let audit_rows = runtime
4492 .db()
4493 .store()
4494 .get_collection(crate::runtime::query_audit::QUERY_AUDIT_COLLECTION)
4495 .expect("query audit collection")
4496 .query_all(|_| true);
4497 assert!(
4498 audit_rows.is_empty(),
4499 "regulated preset must not globally audit data-plane queries"
4500 );
4501
4502 clear_preset_env();
4503 }
4504
4505 #[test]
4506 fn bootstrap_manifest_installs_initial_users_policies_guardrails_and_config() {
4507 use crate::auth::policies::{EvalContext, ResourceRef};
4508 use crate::auth::UserId;
4509 use crate::storage::schema::Value;
4510
4511 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4512 clear_preset_env();
4513
4514 let manifest_dir = std::env::current_dir()
4515 .expect("current dir")
4516 .join(".red/tmp/bootstrap-manifest-tests");
4517 std::fs::create_dir_all(&manifest_dir).expect("create manifest test dir");
4518 let manifest_path = manifest_dir.join(format!(
4519 "reddb-bootstrap-manifest-{}-{}.json",
4520 std::process::id(),
4521 std::time::SystemTime::now()
4522 .duration_since(std::time::UNIX_EPOCH)
4523 .unwrap_or_default()
4524 .as_millis()
4525 ));
4526 std::fs::write(
4527 &manifest_path,
4528 r#"{
4529 "users": [
4530 {
4531 "username": "ops",
4532 "password": "hunter2",
4533 "role": "admin"
4534 }
4535 ],
4536 "policies": [
4537 {
4538 "id": "bootstrap-registry-admin",
4539 "version": 1,
4540 "statements": [
4541 {
4542 "effect": "allow",
4543 "actions": ["red.registry:*", "policy:*", "config:write", "select"],
4544 "resources": ["registry:*", "policy:*", "config:*", "collection:docs"]
4545 }
4546 ]
4547 }
4548 ],
4549 "managed_policies": [
4550 {
4551 "id": "managed-deny-drop",
4552 "version": 1,
4553 "statements": [
4554 {
4555 "effect": "deny",
4556 "actions": ["policy:drop"],
4557 "resources": ["policy:managed-deny-drop"]
4558 }
4559 ],
4560 "required_resource": "policy:managed-deny-drop",
4561 "evidence": "full"
4562 }
4563 ],
4564 "attachments": [
4565 {"user": "ops", "policy": "bootstrap-registry-admin"}
4566 ],
4567 "managed_config_namespaces": [
4568 {
4569 "id": "red.ai",
4570 "required_action": "config:write",
4571 "required_resource": "config:red.ai.*",
4572 "evidence": "metadata"
4573 }
4574 ],
4575 "config": [
4576 {"key": "red.ai.default.provider", "value": "openai"},
4577 {
4578 "key": "red.ai.openai.default.secret_ref",
4579 "secret_ref": {"collection": "red.vault", "key": "openai"}
4580 }
4581 ],
4582 "actor": "ops"
4583 }"#,
4584 )
4585 .expect("write manifest");
4586 unsafe {
4588 std::env::set_var("REDDB_BOOTSTRAP_MANIFEST", &manifest_path);
4589 }
4590
4591 let (runtime, auth_store) = fresh_runtime_and_store();
4592 apply_preset(&runtime, &auth_store).expect("manifest applies cleanly");
4593
4594 let users = auth_store.list_users();
4595 assert_eq!(users.len(), 1);
4596 assert_eq!(users[0].username, "ops");
4597 assert!(users[0].tenant_id.is_none());
4598
4599 let actor = UserId::platform("ops");
4600 let ctx = EvalContext {
4601 principal_tenant: None,
4602 current_tenant: None,
4603 peer_ip: None,
4604 mfa_present: false,
4605 now_ms: 1_700_000_000_000,
4606 principal_is_admin_role: true,
4607 principal_is_platform_scoped: true,
4608 };
4609 assert!(auth_store.check_policy_authz(
4611 &actor,
4612 "select",
4613 &ResourceRef::new("collection", "docs"),
4614 &ctx
4615 ));
4616
4617 let managed_policy = runtime
4618 .config_registry()
4619 .get_active("managed-deny-drop")
4620 .expect("managed policy registry entry");
4621 assert!(managed_policy.managed);
4622 assert_eq!(managed_policy.resource_type, "policy");
4623 let managed_config = runtime
4624 .config_registry()
4625 .get_active("red.ai")
4626 .expect("managed config namespace registry entry");
4627 assert!(managed_config.managed);
4628 assert_eq!(managed_config.resource_type, "config_namespace");
4629
4630 let store = runtime.db().store();
4631 match store
4632 .get_config("red.ai.default.provider")
4633 .expect("plain config persisted")
4634 {
4635 Value::Text(s) => assert_eq!(s.as_ref(), "openai"),
4636 other => panic!("expected provider text, got {other:?}"),
4637 }
4638 let Value::Json(bytes) = store
4639 .get_config("red.ai.openai.default.secret_ref")
4640 .expect("secret ref config persisted")
4641 else {
4642 panic!("secret ref must be stored as structured JSON");
4643 };
4644 let reference: crate::serde_json::Value =
4645 crate::serde_json::from_slice(&bytes).expect("secret ref json");
4646 assert_eq!(
4647 reference.get("type").and_then(|v| v.as_str()),
4648 Some("secret_ref")
4649 );
4650 assert!(
4651 !String::from_utf8_lossy(&bytes).contains("hunter2"),
4652 "manifest password must not leak into secret ref config"
4653 );
4654
4655 let completed = store
4656 .get_config(BOOTSTRAP_COMPLETED_KEY)
4657 .expect("bootstrap completion persisted");
4658 assert!(matches!(completed, Value::Boolean(true)));
4659 assert!(
4660 store
4661 .get_config("system.bootstrap.manifest.registry_entries")
4662 .is_some(),
4663 "managed registry entries must be persisted internally"
4664 );
4665
4666 std::fs::remove_file(&manifest_path).expect("remove manifest after first boot");
4667 let restored_registry = Arc::new(crate::auth::registry::ConfigRegistry::new());
4668 crate::cli::bootstrap_manifest::rehydrate_manifest_registry(&runtime, &restored_registry)
4669 .expect("registry rehydrates without manifest file");
4670 assert!(restored_registry.get_active("managed-deny-drop").is_some());
4671 assert!(restored_registry.get_active("red.ai").is_some());
4672
4673 let fresh = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4674 apply_preset(&runtime, &fresh).expect("re-run must not need manifest file");
4675 assert!(fresh.needs_bootstrap());
4676
4677 clear_preset_env();
4678 }
4679
4680 #[test]
4681 fn production_preset_refuses_to_start_without_password() {
4682 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4683 clear_preset_env();
4684 unsafe {
4686 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4687 std::env::set_var("REDDB_USERNAME", "ops");
4688 }
4689
4690 let (runtime, auth_store) = fresh_runtime_and_store();
4691 let err = apply_preset(&runtime, &auth_store).expect_err("must reject missing password");
4692 assert!(
4693 err.contains("REDDB_PASSWORD"),
4694 "error must name the missing env: {err}"
4695 );
4696
4697 assert!(auth_store.needs_bootstrap());
4699 assert!(runtime
4700 .db()
4701 .store()
4702 .get_config(BOOTSTRAP_COMPLETED_KEY)
4703 .is_none());
4704
4705 clear_preset_env();
4706 }
4707
4708 #[test]
4709 fn re_running_production_after_first_boot_is_a_silent_skip() {
4710 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4711 clear_preset_env();
4712 unsafe {
4714 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4715 std::env::set_var("REDDB_USERNAME", "ops");
4716 std::env::set_var("REDDB_PASSWORD", "hunter2");
4717 }
4718
4719 let (runtime, auth_store) = fresh_runtime_and_store();
4720 apply_preset(&runtime, &auth_store).expect("first apply");
4721 assert_eq!(auth_store.list_users().len(), 1);
4722
4723 let fresh = Arc::new(AuthStore::new(crate::auth::AuthConfig::default()));
4730 apply_preset(&runtime, &fresh).expect("re-run is silent-skip");
4731 assert!(
4732 fresh.needs_bootstrap(),
4733 "re-run must not create a second admin"
4734 );
4735 assert!(
4736 fresh.get_policy(FIRST_ADMIN_ALLOW_ALL_POLICY).is_none(),
4737 "re-run must not re-install the allow-all policy on the fresh store"
4738 );
4739
4740 clear_preset_env();
4741 }
4742
4743 #[test]
4744 fn unrecognised_preset_value_is_rejected() {
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, "weird");
4750 }
4751
4752 let (runtime, auth_store) = fresh_runtime_and_store();
4753 let err = apply_preset(&runtime, &auth_store).expect_err("must reject unknown preset");
4754 assert!(err.contains("weird"), "error must echo the value: {err}");
4755 assert!(auth_store.needs_bootstrap());
4756
4757 clear_preset_env();
4758 }
4759
4760 #[test]
4761 fn no_auth_short_circuits_preset_entirely() {
4762 let _g = no_auth_env_lock().lock().unwrap_or_else(|e| e.into_inner());
4763 clear_preset_env();
4764 unsafe {
4767 std::env::set_var(PRESET_ENV, PRESET_PRODUCTION);
4768 std::env::set_var("REDDB_USERNAME", "ops");
4769 std::env::set_var("REDDB_PASSWORD", "hunter2");
4770 }
4771
4772 let options = no_auth_test_config(true)
4773 .to_db_options()
4774 .expect("to_db_options");
4775 assert!(no_auth_active(&options));
4776
4777 let (runtime, auth_store) = fresh_runtime_and_store();
4780 if !no_auth_active(&options) {
4781 apply_preset(&runtime, &auth_store).expect("would apply preset");
4782 }
4783
4784 assert!(
4785 auth_store.needs_bootstrap(),
4786 "--no-auth must prevent any admin creation"
4787 );
4788 assert!(
4789 runtime
4790 .db()
4791 .store()
4792 .get_config(BOOTSTRAP_COMPLETED_KEY)
4793 .is_none(),
4794 "--no-auth must skip bootstrap-state persistence"
4795 );
4796
4797 clear_preset_env();
4798 }
4799
4800 #[test]
4801 fn implicit_bind_collision_degrades() {
4802 let held = TcpListener::bind("127.0.0.1:0").expect("hold test port");
4803 let addr = held.local_addr().expect("held addr").to_string();
4804 let mut readiness = TransportReadiness::default();
4805
4806 let listener =
4807 bind_listener_for_startup(&mut readiness, "http", &addr, false).expect("nonfatal");
4808
4809 assert!(listener.is_none());
4810 assert_eq!(readiness.active.len(), 0);
4811 assert_eq!(readiness.failed.len(), 1);
4812 assert!(!readiness.failed[0].explicit);
4813 assert_eq!(readiness.failed[0].bind_addr, addr);
4814 }
4815}