1use std::sync::Arc;
11
12#[cfg(target_os = "linux")]
21struct PostureKernelVerifier {
22 strict: bool,
23 signing_keys: Vec<String>,
24 allowed_hashes: Vec<String>,
25 daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
26}
27
28#[cfg(target_os = "linux")]
31impl std::fmt::Debug for PostureKernelVerifier {
32 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
33 f.debug_struct("PostureKernelVerifier")
34 .field("strict", &self.strict)
35 .field("signing_keys", &self.signing_keys.len())
36 .field("allowed_hashes", &self.allowed_hashes.len())
37 .field("has_daemon", &self.daemon.is_some())
38 .finish()
39 }
40}
41
42#[cfg(target_os = "linux")]
43impl boatramp_firecracker::KernelVerifier for PostureKernelVerifier {
44 fn verify(&self, bytes: &[u8], expected_hash: &str) -> std::result::Result<(), String> {
46 let sig = self
50 .daemon
51 .as_ref()
52 .and_then(|d| d.effective().default_kernel.clone())
53 .filter(|dk| dk.sha256 == expected_hash)
54 .and_then(|dk| dk.sig);
55 let kref = boatramp_core::daemon_config::KernelRef {
56 source: expected_hash.to_string(),
57 sha256: expected_hash.to_string(),
58 sig,
59 };
60 boatramp_core::kernel_trust::verify_kernel(
61 bytes,
62 &kref,
63 self.strict,
64 &self.signing_keys,
65 &self.allowed_hashes,
66 )
67 .map_err(|e| e.to_string())
68 }
69}
70
71#[cfg(target_os = "macos")]
76struct VzPostureKernelVerifier {
77 strict: bool,
78 signing_keys: Vec<String>,
79 allowed_hashes: Vec<String>,
80 daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
81}
82
83#[cfg(target_os = "macos")]
84impl std::fmt::Debug for VzPostureKernelVerifier {
85 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
86 f.debug_struct("VzPostureKernelVerifier")
87 .field("strict", &self.strict)
88 .field("signing_keys", &self.signing_keys.len())
89 .field("allowed_hashes", &self.allowed_hashes.len())
90 .field("has_daemon", &self.daemon.is_some())
91 .finish()
92 }
93}
94
95#[cfg(target_os = "macos")]
96impl boatramp_vz::KernelVerifier for VzPostureKernelVerifier {
97 fn verify(&self, bytes: &[u8], expected_hash: &str) -> std::result::Result<(), String> {
98 let sig = self
99 .daemon
100 .as_ref()
101 .and_then(|d| d.effective().default_kernel.clone())
102 .filter(|dk| dk.sha256 == expected_hash)
103 .and_then(|dk| dk.sig);
104 let kref = boatramp_core::daemon_config::KernelRef {
105 source: expected_hash.to_string(),
106 sha256: expected_hash.to_string(),
107 sig,
108 };
109 boatramp_core::kernel_trust::verify_kernel(
110 bytes,
111 &kref,
112 self.strict,
113 &self.signing_keys,
114 &self.allowed_hashes,
115 )
116 .map_err(|e| e.to_string())
117 }
118}
119
120#[cfg(target_os = "macos")]
127fn macos_supports_vz() -> bool {
128 if cfg!(not(target_arch = "aarch64")) {
130 return false;
131 }
132 let major = sysctl_string("kern.osproductversion")
133 .and_then(|v| v.split('.').next().and_then(|m| m.parse::<u32>().ok()));
134 matches!(major, Some(m) if m >= 15)
135}
136
137#[cfg(target_os = "macos")]
140fn sysctl_string(name: &str) -> Option<String> {
141 let out = std::process::Command::new("sysctl")
142 .args(["-n", name])
143 .output()
144 .ok()?;
145 if !out.status.success() {
146 return None;
147 }
148 Some(String::from_utf8_lossy(&out.stdout).trim().to_string())
149}
150
151pub async fn build_compute(
158 cfg: Option<&crate::config::ComputeConfig>,
159 storage: std::sync::Arc<dyn boatramp_core::Storage>,
160 data_dir: &std::path::Path,
161 node_id: u64,
162 strict: bool,
163 daemon: Option<Arc<boatramp_server::DaemonRuntime>>,
164 worker_exe: Option<&std::path::Path>,
169) -> (
170 boatramp_core::compute::BackendRegistry,
171 boatramp_core::compute::Node,
172) {
173 use boatramp_core::compute::{BackendKind, BackendRegistry, Node};
174 let mut backends: BackendRegistry = std::collections::BTreeMap::new();
175 let empty_node = |id| Node {
176 id,
177 region: None,
178 labels: std::collections::BTreeMap::new(),
179 free_vcpus: 0,
180 free_mem_mib: 0,
181 backends: Vec::new(),
182 };
183 let Some(cfg) = cfg else {
184 return (backends, empty_node(node_id));
185 };
186
187 match boatramp_docker::DockerBackend::connect() {
189 Ok(docker) => {
190 let docker = docker
194 .with_endpoint(cfg.docker_endpoint)
195 .with_volume_mode(cfg.docker_volume_mode)
196 .with_data_dir(data_dir)
197 .with_writable_root_allowed(!strict)
198 .with_cap_add_allowed(!strict);
199 if docker.reachable().await {
200 backends.insert("docker".to_string(), std::sync::Arc::new(docker));
201 } else {
202 tracing::debug!("no reachable docker daemon; skipping docker backend");
203 }
204 }
205 Err(e) => tracing::debug!(%e, "docker backend unavailable"),
206 }
207
208 #[cfg(target_os = "linux")]
214 let bridge_ready = match boatramp_core::ipam::IpPool::new(&cfg.subnet) {
215 Ok(pool) => {
216 match boatramp_container::ensure_bridge(&cfg.bridge, pool.gateway(), pool.prefix_len())
217 .await
218 {
219 Ok(()) => true,
220 Err(e) => {
221 tracing::warn!(%e, bridge = %cfg.bridge, "could not create the compute bridge (need CAP_NET_ADMIN); container + embedded-VMM backends disabled");
222 false
223 }
224 }
225 }
226 Err(e) => {
227 tracing::warn!(%e, subnet = %cfg.subnet, "bad compute subnet; container + embedded-VMM backends disabled");
228 false
229 }
230 };
231
232 #[cfg(target_os = "linux")]
234 if bridge_ready {
235 match worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())) {
236 Ok(self_exe) => match boatramp_container::ContainerBackend::new(
237 storage.clone(),
238 data_dir.to_path_buf(),
239 cfg.bridge.clone(),
240 &cfg.subnet,
241 self_exe,
242 ) {
243 Ok(c) => {
244 let c = c.with_cap_add_allowed(!strict);
247 backends.insert("container".to_string(), std::sync::Arc::new(c));
248 }
249 Err(e) => tracing::warn!(%e, "container backend unavailable"),
250 },
251 Err(e) => tracing::warn!(%e, "current_exe for container backend"),
252 }
253 }
254 #[cfg(all(target_os = "linux", target_arch = "x86_64"))]
261 if bridge_ready && std::path::Path::new("/dev/kvm").exists() {
262 match (
263 worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())),
264 boatramp_core::ipam::IpPool::new(&cfg.subnet),
265 ) {
266 (Ok(self_exe), Ok(pool)) => {
267 let gateway = pool.gateway().to_string();
268 let verifier: Arc<dyn boatramp_firecracker::KernelVerifier> =
270 Arc::new(PostureKernelVerifier {
271 strict,
272 signing_keys: cfg.kernel_signing_pubkeys.clone(),
273 allowed_hashes: cfg.kernel_allowed_hashes.clone(),
274 daemon: daemon.clone(),
275 });
276 match boatramp_firecracker::EmbeddedVmmBackend::new(
277 storage.clone(),
278 self_exe, data_dir.to_path_buf(),
280 cfg.bridge.clone(),
281 gateway,
282 &cfg.subnet,
283 verifier,
284 ) {
285 Ok(vmm) => {
286 backends.insert("vmm-embedded".to_string(), std::sync::Arc::new(vmm));
287 }
288 Err(e) => tracing::warn!(%e, "embedded VMM backend unavailable"),
289 }
290 }
291 (Err(e), _) => tracing::warn!(%e, "current_exe for VMM backend"),
292 (_, Err(e)) => tracing::warn!(%e, "bad compute subnet for VMM backend"),
293 }
294 } else {
295 tracing::debug!("no /dev/kvm; skipping embedded VMM backend");
296 }
297
298 #[cfg(target_os = "macos")]
304 if macos_supports_vz() {
305 match (
306 worker_exe.map_or_else(std::env::current_exe, |p| Ok(p.to_path_buf())),
307 boatramp_core::ipam::IpPool::new(&cfg.subnet),
308 ) {
309 (Ok(self_exe), Ok(_pool)) => {
310 let verifier: Arc<dyn boatramp_vz::KernelVerifier> =
311 Arc::new(VzPostureKernelVerifier {
312 strict,
313 signing_keys: cfg.kernel_signing_pubkeys.clone(),
314 allowed_hashes: cfg.kernel_allowed_hashes.clone(),
315 daemon: daemon.clone(),
316 });
317 match boatramp_vz::VzBackend::new(
318 storage.clone(),
319 self_exe, data_dir.to_path_buf(),
321 &cfg.subnet, verifier,
323 ) {
324 Ok(vz) => {
326 let vz = vz.with_writable_root_allowed(!strict);
327 backends.insert("vmm-vz".to_string(), std::sync::Arc::new(vz));
328 }
329 Err(e) => tracing::warn!(%e, "macOS VMM backend unavailable"),
330 }
331 }
332 (Err(e), _) => tracing::warn!(%e, "current_exe for macOS VMM backend"),
333 (_, Err(e)) => tracing::warn!(%e, "bad compute subnet for macOS VMM backend"),
334 }
335 } else {
336 tracing::debug!("not Apple silicon + macOS 15+; skipping macOS VMM backend");
337 }
338
339 let _ = (&storage, data_dir); #[cfg(not(any(all(target_os = "linux", target_arch = "x86_64"), target_os = "macos")))]
344 let _ = (strict, &daemon);
345
346 let free_vcpus = if cfg.vcpus > 0 {
347 cfg.vcpus
348 } else {
349 std::thread::available_parallelism()
350 .map(|n| n.get() as u32)
351 .unwrap_or(1)
352 };
353 let free_mem_mib = if cfg.mem_mib > 0 { cfg.mem_mib } else { 1024 };
354 let advertised: Vec<BackendKind> = backends
355 .iter()
356 .map(|(id, b)| {
357 let caps = b.capabilities();
358 BackendKind {
359 id: id.clone(),
360 isolation: caps.isolation,
361 persistent_volumes: caps.persistent_volumes,
362 scale_to_zero: caps.scale_to_zero,
363 }
364 })
365 .collect();
366 tracing::info!(backends = ?advertised, free_vcpus, free_mem_mib, "compute node inventory");
367 let node = Node {
368 id: node_id,
369 region: cfg.region.clone(),
370 labels: std::collections::BTreeMap::new(),
371 free_vcpus,
372 free_mem_mib,
373 backends: advertised,
374 };
375 (backends, node)
376}
377
378pub struct NodeComputeExec {
386 backends: boatramp_core::compute::BackendRegistry,
387 deploy: boatramp_core::deploy::DeployStore,
388}
389
390impl NodeComputeExec {
391 pub fn new(
395 backends: boatramp_core::compute::BackendRegistry,
396 deploy: boatramp_core::deploy::DeployStore,
397 ) -> Self {
398 Self { backends, deploy }
399 }
400}
401
402#[async_trait::async_trait]
403impl boatramp_core::compute::ComputeExec for NodeComputeExec {
404 async fn exec(
405 &self,
406 project: &str,
407 workload: &str,
408 argv: &[String],
409 stdin: Option<&[u8]>,
410 ) -> Result<boatramp_core::compute::ExecOutput, boatramp_core::compute::ExecError> {
411 use boatramp_core::compute::{BackendError, ExecError, ReplicaPhase};
412 use boatramp_core::project::ProjectRef;
413 let states = self
414 .deploy
415 .list_replica_states(ProjectRef::new(project), workload)
416 .await
417 .map_err(|e| ExecError::Other(e.to_string()))?;
418 let target = states
421 .iter()
422 .find(|s| s.phase == ReplicaPhase::Running && s.healthy)
423 .or_else(|| states.iter().find(|s| s.phase == ReplicaPhase::Running))
424 .ok_or_else(|| ExecError::NoReplica(workload.to_string()))?;
425 let backend = self
426 .backends
427 .get(&target.backend)
428 .ok_or_else(|| ExecError::Unsupported(target.backend.clone()))?;
429 match backend.exec(&target.handle, argv, stdin).await {
430 Ok(out) => Ok(out),
431 Err(BackendError::Unsupported) => Err(ExecError::Unsupported(target.backend.clone())),
432 Err(e) => Err(ExecError::Other(e.to_string())),
433 }
434 }
435}
436
437pub struct NodeComputeVolumes {
446 backends: boatramp_core::compute::BackendRegistry,
447 deploy: boatramp_core::deploy::DeployStore,
448}
449
450impl NodeComputeVolumes {
451 pub fn new(
453 backends: boatramp_core::compute::BackendRegistry,
454 deploy: boatramp_core::deploy::DeployStore,
455 ) -> Self {
456 Self { backends, deploy }
457 }
458
459 async fn referenced_volume_names(
466 &self,
467 ) -> Result<std::collections::BTreeSet<String>, boatramp_core::compute::VolumeError> {
468 use boatramp_core::compute::VolumeError;
469 let mut names = std::collections::BTreeSet::new();
470 let workloads = self
471 .deploy
472 .list_compute_workloads_all()
473 .await
474 .map_err(|e| VolumeError::Other(e.to_string()))?;
475 for (_project, workload) in workloads {
476 let spec = self
477 .deploy
478 .get_compute_spec(&workload.active)
479 .await
480 .map_err(|e| VolumeError::Other(e.to_string()))?;
481 if let Some(spec) = spec {
482 for vol in spec.volumes {
483 names.insert(vol.name);
484 }
485 }
486 }
487 Ok(names)
488 }
489}
490
491#[async_trait::async_trait]
492impl boatramp_core::compute::ComputeVolumes for NodeComputeVolumes {
493 async fn list(
494 &self,
495 ) -> Result<Vec<boatramp_core::compute::VolumeStatus>, boatramp_core::compute::VolumeError>
496 {
497 use boatramp_core::compute::{VolumeError, VolumeStatus};
498 let referenced = self.referenced_volume_names().await?;
499 let mut by_name: std::collections::BTreeMap<String, u64> =
503 std::collections::BTreeMap::new();
504 for backend in self.backends.values() {
505 let vols = backend
506 .list_volumes()
507 .await
508 .map_err(|e| VolumeError::Other(e.to_string()))?;
509 for v in vols {
510 let slot = by_name.entry(v.name).or_insert(0);
512 *slot = (*slot).max(v.size_bytes);
513 }
514 }
515 Ok(by_name
516 .into_iter()
517 .map(|(name, size_bytes)| VolumeStatus {
518 in_use: referenced.contains(&name),
519 info: boatramp_core::compute::VolumeInfo { name, size_bytes },
520 })
521 .collect())
522 }
523
524 async fn remove(
525 &self,
526 name: &str,
527 force: bool,
528 ) -> Result<bool, boatramp_core::compute::VolumeError> {
529 use boatramp_core::compute::{BackendError, VolumeError};
530 if !force && self.referenced_volume_names().await?.contains(name) {
534 return Err(VolumeError::InUse(name.to_string()));
535 }
536 let mut existed = false;
539 let mut any_supported = false;
540 for backend in self.backends.values() {
541 match backend.remove_volume(name).await {
542 Ok(removed) => {
543 any_supported = true;
544 existed |= removed;
545 }
546 Err(BackendError::Unsupported) => {}
547 Err(e) => return Err(VolumeError::Other(e.to_string())),
548 }
549 }
550 if !any_supported {
551 return Err(VolumeError::Unsupported);
552 }
553 Ok(existed)
554 }
555}
556
557#[cfg(test)]
558mod tests {
559 use super::*;
560 use async_trait::async_trait;
561 use boatramp_core::compute::{
562 Artifact, BackendError, Capabilities, ComputeBackend, ComputeSpec, ComputeVolumes,
563 ComputeWorkload, Health, Instance, InstanceHandle, IsolationClass, IsolationRequirement,
564 LaunchRequest, RestartPolicy, RootSource, VolumeError, VolumeInfo, VolumeRef,
565 };
566 use boatramp_core::deploy::DeployStore;
567 use boatramp_core::project::ProjectRef;
568 use boatramp_core::{ByteStream, GetObject, ObjectMeta, PutMeta, Storage, StorageError};
569 use std::collections::BTreeMap;
570 use std::sync::{Arc, Mutex};
571
572 struct NullStorage;
575 #[async_trait]
576 impl Storage for NullStorage {
577 async fn get(&self, _: &str) -> Result<GetObject, StorageError> {
578 Err(StorageError::NotFound(String::new()))
579 }
580 async fn get_range(
581 &self,
582 _: &str,
583 _: u64,
584 _: Option<u64>,
585 ) -> Result<GetObject, StorageError> {
586 Err(StorageError::unsupported("range"))
587 }
588 async fn put(
589 &self,
590 _: &str,
591 _: ByteStream,
592 _: PutMeta,
593 ) -> Result<ObjectMeta, StorageError> {
594 Err(StorageError::unsupported("put"))
595 }
596 async fn head(&self, _: &str) -> Result<ObjectMeta, StorageError> {
597 Err(StorageError::NotFound(String::new()))
598 }
599 async fn delete(&self, _: &str) -> Result<(), StorageError> {
600 Ok(())
601 }
602 async fn list(&self, _: &str) -> Result<Vec<ObjectMeta>, StorageError> {
603 Ok(Vec::new())
604 }
605 }
606
607 struct FakeVolumeBackend {
610 vols: Mutex<BTreeMap<String, u64>>,
611 }
612 impl FakeVolumeBackend {
613 fn with(names: &[(&str, u64)]) -> Self {
614 Self {
615 vols: Mutex::new(names.iter().map(|(n, s)| (n.to_string(), *s)).collect()),
616 }
617 }
618 }
619 #[async_trait]
620 impl ComputeBackend for FakeVolumeBackend {
621 fn id(&self) -> &'static str {
622 "container"
623 }
624 fn capabilities(&self) -> Capabilities {
625 Capabilities {
626 isolation: IsolationClass::Namespace,
627 scale_to_zero: false,
628 persistent_volumes: true,
629 max_vcpus: None,
630 max_mem_mib: None,
631 }
632 }
633 async fn materialize(&self, _: &ComputeSpec) -> Result<Artifact, BackendError> {
634 Err(BackendError::Unsupported)
635 }
636 async fn launch(&self, _: &LaunchRequest) -> Result<Instance, BackendError> {
637 Err(BackendError::Unsupported)
638 }
639 async fn stop(&self, _: &InstanceHandle) -> Result<(), BackendError> {
640 Ok(())
641 }
642 async fn health(&self, _: &InstanceHandle) -> Result<Health, BackendError> {
643 Ok(Health::Unknown)
644 }
645 async fn list_volumes(&self) -> Result<Vec<VolumeInfo>, BackendError> {
646 Ok(self
647 .vols
648 .lock()
649 .unwrap()
650 .iter()
651 .map(|(name, size)| VolumeInfo {
652 name: name.clone(),
653 size_bytes: *size,
654 })
655 .collect())
656 }
657 async fn remove_volume(&self, name: &str) -> Result<bool, BackendError> {
658 Ok(self.vols.lock().unwrap().remove(name).is_some())
659 }
660 }
661
662 fn spec_with_volume(vol: Option<&str>) -> ComputeSpec {
663 ComputeSpec {
664 version: 1,
665 root: RootSource::Image("img".into()),
666 kernel: String::new(),
667 kernel_cmdline: None,
668 vcpus: 1,
669 mem_mib: 64,
670 entrypoint: vec![],
671 env: BTreeMap::new(),
672 port: 8080,
673 restart: RestartPolicy::Always,
674 scale_to_zero: false,
675 volumes: vol
676 .map(|n| {
677 vec![VolumeRef {
678 mount: "/data".into(),
679 name: n.into(),
680 size_mib: 128,
681 }]
682 })
683 .unwrap_or_default(),
684 writable_root: false,
685 cap_add: vec![],
686 user: None,
687 isolation: IsolationRequirement::Trusted,
688 prefer_backend: None,
689 bindings: vec![],
690 }
691 }
692
693 async fn setup(referenced: Option<&str>, backend_vols: &[(&str, u64)]) -> NodeComputeVolumes {
697 let store = DeployStore::new(
698 Arc::new(NullStorage),
699 Arc::new(boatramp_core::kv::MemoryKv::new()),
700 );
701 let spec = spec_with_volume(referenced);
702 let hash = store.put_compute_spec(&spec).await.expect("put spec");
703 let workload = ComputeWorkload {
704 version: 1,
705 name: "wl".into(),
706 active: hash,
707 replicas: 1,
708 placement: Default::default(),
709 };
710 store
711 .set_compute_workload(ProjectRef::DEFAULT, &workload)
712 .await
713 .expect("set workload");
714 let mut backends: boatramp_core::compute::BackendRegistry = BTreeMap::new();
715 backends.insert(
716 "container".into(),
717 Arc::new(FakeVolumeBackend::with(backend_vols)) as Arc<dyn ComputeBackend>,
718 );
719 NodeComputeVolumes::new(backends, store)
720 }
721
722 #[tokio::test]
723 async fn list_flags_referenced_volume_in_use_and_orphan_free() {
724 let vols = setup(Some("data"), &[("data", 100), ("old", 50)]).await;
726 let listed = vols.list().await.expect("list");
727 assert_eq!(listed.len(), 2);
728 let data = listed.iter().find(|v| v.info.name == "data").unwrap();
729 let old = listed.iter().find(|v| v.info.name == "old").unwrap();
730 assert!(data.in_use, "spec-referenced volume is in use");
731 assert_eq!(data.info.size_bytes, 100);
732 assert!(!old.in_use, "unreferenced volume is orphaned");
733 assert_eq!(old.info.size_bytes, 50);
734 }
735
736 #[tokio::test]
737 async fn remove_refuses_in_use_without_force_and_allows_with_force() {
738 let vols = setup(Some("data"), &[("data", 100)]).await;
739 assert!(matches!(
741 vols.remove("data", false).await,
742 Err(VolumeError::InUse(n)) if n == "data"
743 ));
744 assert!(vols
746 .list()
747 .await
748 .unwrap()
749 .iter()
750 .any(|v| v.info.name == "data"));
751 assert!(vols.remove("data", true).await.expect("forced remove"));
753 assert!(vols.list().await.unwrap().is_empty());
754 }
755
756 #[tokio::test]
757 async fn remove_orphan_succeeds_and_absent_reports_false() {
758 let vols = setup(None, &[("old", 50)]).await;
760 assert!(vols.remove("old", false).await.expect("remove orphan"));
761 assert!(!vols.remove("gone", false).await.expect("remove absent"));
763 }
764}