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