1use async_trait::async_trait;
16use boatramp_core::compute::{
17 Artifact, BackendError, Capabilities, ComputeBackend, ComputeSpec, Endpoint, Health, Instance,
18 InstanceHandle, IsolationClass, LaunchRequest, RestartPolicy, RootSource, Scheme, VolumeRef,
19};
20use bollard::container::{
21 Config, CreateContainerOptions, RemoveContainerOptions, StopContainerOptions,
22};
23use bollard::image::CreateImageOptions;
24use bollard::models::{
25 HostConfig, Mount, MountTypeEnum, PortBinding, RestartPolicy as DockerRestartPolicy,
26 RestartPolicyNameEnum,
27};
28use bollard::Docker;
29use futures::StreamExt;
30use serde::{Deserialize, Serialize};
31use std::collections::HashMap;
32use std::path::{Path, PathBuf};
33
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
48#[serde(rename_all = "lowercase")]
49pub enum DockerEndpoint {
50 #[default]
52 Published,
53 Bridge,
55}
56
57#[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
67#[serde(rename_all = "lowercase")]
68pub enum DockerVolumeMode {
69 #[default]
71 Named,
72 Bind,
74}
75
76fn docker_volume_name(name: &str) -> String {
79 format!("boatramp-{name}")
80}
81
82fn volume_dir(data_dir: &Path, name: &str) -> PathBuf {
85 data_dir.join("compute").join("volumes").join(name)
86}
87
88fn validate_volume(name: &str, mount: &str) -> Result<(), BackendError> {
94 use std::path::Component;
95 let name_ok = matches!(
96 Path::new(name).components().collect::<Vec<_>>().as_slice(),
97 [Component::Normal(_)]
98 );
99 if !name_ok {
100 return Err(BackendError::Launch(format!(
101 "invalid volume name {name:?}: must be a single path component"
102 )));
103 }
104 let m = Path::new(mount);
105 let mount_ok = m.is_absolute()
106 && m.components()
107 .all(|c| matches!(c, Component::RootDir | Component::Normal(_)));
108 if !mount_ok {
109 return Err(BackendError::Launch(format!(
110 "invalid volume mount {mount:?}: must be an absolute path with no `..`"
111 )));
112 }
113 Ok(())
114}
115
116fn volume_mount(vol: &VolumeRef, mode: DockerVolumeMode, data_dir: &Path) -> Mount {
120 let (typ, source) = match mode {
121 DockerVolumeMode::Named => (MountTypeEnum::VOLUME, docker_volume_name(&vol.name)),
122 DockerVolumeMode::Bind => (
123 MountTypeEnum::BIND,
124 volume_dir(data_dir, &vol.name).display().to_string(),
125 ),
126 };
127 Mount {
128 target: Some(vol.mount.clone()),
129 source: Some(source),
130 typ: Some(typ),
131 read_only: Some(false),
132 ..Default::default()
133 }
134}
135
136pub struct DockerBackend {
138 docker: Docker,
139 endpoint: DockerEndpoint,
140 volume_mode: DockerVolumeMode,
142 data_dir: PathBuf,
144 writable_root_allowed: bool,
148}
149
150impl DockerBackend {
151 pub fn connect() -> Result<Self, BackendError> {
154 let docker = Docker::connect_with_defaults()
155 .map_err(|e| BackendError::Other(format!("connect to docker: {e}")))?;
156 Ok(Self {
157 docker,
158 endpoint: DockerEndpoint::default(),
159 volume_mode: DockerVolumeMode::default(),
160 data_dir: PathBuf::from("."),
161 writable_root_allowed: false,
162 })
163 }
164
165 pub fn with_client(docker: Docker) -> Self {
167 Self {
168 docker,
169 endpoint: DockerEndpoint::default(),
170 volume_mode: DockerVolumeMode::default(),
171 data_dir: PathBuf::from("."),
172 writable_root_allowed: false,
173 }
174 }
175
176 pub fn with_endpoint(mut self, endpoint: DockerEndpoint) -> Self {
178 self.endpoint = endpoint;
179 self
180 }
181
182 pub fn with_writable_root_allowed(mut self, allowed: bool) -> Self {
185 self.writable_root_allowed = allowed;
186 self
187 }
188
189 pub fn with_volume_mode(mut self, mode: DockerVolumeMode) -> Self {
191 self.volume_mode = mode;
192 self
193 }
194
195 pub fn with_data_dir(mut self, data_dir: impl Into<PathBuf>) -> Self {
197 self.data_dir = data_dir.into();
198 self
199 }
200
201 pub async fn reachable(&self) -> bool {
204 self.docker.ping().await.is_ok()
205 }
206
207 async fn stage_volumes(&self, spec: &ComputeSpec) -> Result<Vec<Mount>, BackendError> {
213 let mut mounts = Vec::with_capacity(spec.volumes.len());
214 for vol in &spec.volumes {
215 validate_volume(&vol.name, &vol.mount)?;
216 if self.volume_mode == DockerVolumeMode::Bind {
217 let dir = volume_dir(&self.data_dir, &vol.name);
218 tokio::fs::create_dir_all(&dir).await.map_err(|e| {
219 BackendError::Launch(format!("create volume {} dir: {e}", vol.name))
220 })?;
221 }
222 mounts.push(volume_mount(vol, self.volume_mode, &self.data_dir));
223 }
224 Ok(mounts)
225 }
226}
227
228fn container_name(workload: &str, replica: u32) -> String {
230 format!("boatramp-{workload}-{replica}")
231}
232
233fn encode_ref(name: &str, ip: &str, port: u16) -> String {
236 format!("{name}@{ip}:{port}")
237}
238
239fn decode_ref(s: &str) -> Option<(String, String, u16)> {
241 let (name, rest) = s.split_once('@')?;
242 let (ip, port) = rest.rsplit_once(':')?;
243 Some((name.to_string(), ip.to_string(), port.parse().ok()?))
244}
245
246fn restart_policy(policy: RestartPolicy) -> DockerRestartPolicy {
248 let name = match policy {
249 RestartPolicy::Never => RestartPolicyNameEnum::NO,
250 RestartPolicy::OnFailure => RestartPolicyNameEnum::ON_FAILURE,
251 RestartPolicy::Always => RestartPolicyNameEnum::ALWAYS,
252 };
253 DockerRestartPolicy {
254 name: Some(name),
255 maximum_retry_count: None,
256 }
257}
258
259const MAX_PIDS: i64 = 512;
262
263fn hardened_host_config(
276 mem_mib: u32,
277 vcpus: u32,
278 restart: RestartPolicy,
279 writable_root: bool,
280) -> HostConfig {
281 let tmpfs = std::collections::HashMap::from([
282 ("/tmp".to_string(), "rw,noexec,nosuid,size=64m".to_string()),
283 ("/run".to_string(), "rw,noexec,nosuid,size=16m".to_string()),
284 ]);
285 HostConfig {
286 memory: Some(i64::from(mem_mib) * 1024 * 1024),
287 nano_cpus: Some(i64::from(vcpus.max(1)) * 1_000_000_000),
288 restart_policy: Some(restart_policy(restart)),
289 security_opt: Some(vec!["no-new-privileges:true".to_string()]),
291 cap_drop: Some(vec!["ALL".to_string()]),
292 readonly_rootfs: Some(!writable_root),
293 tmpfs: Some(tmpfs),
294 pids_limit: Some(MAX_PIDS),
295 ..Default::default()
296 }
297}
298
299#[async_trait]
300impl ComputeBackend for DockerBackend {
301 fn id(&self) -> &'static str {
302 "docker"
303 }
304
305 fn capabilities(&self) -> Capabilities {
306 Capabilities {
307 isolation: IsolationClass::Container,
308 scale_to_zero: false,
309 persistent_volumes: true,
310 max_vcpus: None,
311 max_mem_mib: None,
312 }
313 }
314
315 async fn materialize(&self, spec: &ComputeSpec) -> Result<Artifact, BackendError> {
316 let reference = match &spec.root {
319 RootSource::Image(reference) => reference.clone(),
320 RootSource::Tar(_) | RootSource::Rootfs(_) => {
321 return Err(BackendError::Materialize(
322 "docker backend requires an image reference (RootSource::Image)".into(),
323 ))
324 }
325 };
326 let options = CreateImageOptions {
327 from_image: reference.clone(),
328 ..Default::default()
329 };
330 let mut pull = self.docker.create_image(Some(options), None, None);
331 while let Some(step) = pull.next().await {
332 step.map_err(|e| BackendError::Materialize(format!("pull {reference}: {e}")))?;
333 }
334 Ok(Artifact::Image { reference })
335 }
336
337 async fn launch(&self, req: &LaunchRequest) -> Result<Instance, BackendError> {
338 let reference = match &req.artifact {
339 Artifact::Image { reference } => reference.clone(),
340 _ => {
341 return Err(BackendError::Launch(
342 "docker backend requires an Image artifact".into(),
343 ))
344 }
345 };
346 let name = container_name(&req.workload, req.replica);
347 let env: Vec<String> = req
348 .spec
349 .env
350 .iter()
351 .map(|(k, v)| format!("{k}={v}"))
352 .collect();
353 let port = req.spec.port;
354 let port_key = format!("{port}/tcp");
355 let writable_root = req.spec.writable_root && self.writable_root_allowed;
358 let mut host_config = hardened_host_config(
359 req.spec.mem_mib,
360 req.spec.vcpus,
361 req.spec.restart,
362 writable_root,
363 );
364 let mounts = self.stage_volumes(&req.spec).await?;
366 if !mounts.is_empty() {
367 host_config.mounts = Some(mounts);
368 }
369 let mut config = Config {
370 image: Some(reference),
371 cmd: Some(req.spec.entrypoint.clone()),
372 env: Some(env),
373 ..Default::default()
374 };
375 if self.endpoint == DockerEndpoint::Published {
380 config.exposed_ports = Some(HashMap::from([(port_key.clone(), HashMap::new())]));
381 host_config.port_bindings = Some(HashMap::from([(
382 port_key.clone(),
383 Some(vec![PortBinding {
384 host_ip: Some("127.0.0.1".to_string()),
385 host_port: Some("0".to_string()),
386 }]),
387 )]));
388 }
389 config.host_config = Some(host_config);
390
391 let _ = self
393 .docker
394 .remove_container(
395 &name,
396 Some(RemoveContainerOptions {
397 force: true,
398 ..Default::default()
399 }),
400 )
401 .await;
402 let created = self
403 .docker
404 .create_container(
405 Some(CreateContainerOptions {
406 name: name.clone(),
407 platform: None,
408 }),
409 config,
410 )
411 .await
412 .map_err(|e| BackendError::Launch(format!("create {name}: {e}")))?;
413 self.docker
414 .start_container::<String>(&created.id, None)
415 .await
416 .map_err(|e| BackendError::Launch(format!("start {name}: {e}")))?;
417
418 let (host, endpoint_port) = match self.endpoint {
421 DockerEndpoint::Published => (
422 "127.0.0.1".to_string(),
423 self.published_host_port(&created.id, &port_key).await?,
424 ),
425 DockerEndpoint::Bridge => (self.container_ip(&created.id).await?, port),
426 };
427 Ok(Instance {
428 handle: InstanceHandle {
429 workload: req.workload.clone(),
430 replica: req.replica,
431 backend_ref: encode_ref(&name, &host, endpoint_port),
432 },
433 endpoint: Endpoint {
434 scheme: Scheme::Http,
435 host,
436 port: endpoint_port,
437 },
438 })
439 }
440
441 async fn stop(&self, handle: &InstanceHandle) -> Result<(), BackendError> {
442 let name = decode_ref(&handle.backend_ref)
443 .map(|(n, _, _)| n)
444 .unwrap_or_else(|| container_name(&handle.workload, handle.replica));
445 let _ = self
447 .docker
448 .stop_container(&name, None::<StopContainerOptions>)
449 .await;
450 self.docker
451 .remove_container(
452 &name,
453 Some(RemoveContainerOptions {
454 force: true,
455 ..Default::default()
456 }),
457 )
458 .await
459 .map_err(|e| BackendError::Stop(format!("remove {name}: {e}")))?;
460 Ok(())
461 }
462
463 async fn health(&self, handle: &InstanceHandle) -> Result<Health, BackendError> {
464 let name = match decode_ref(&handle.backend_ref) {
465 Some((n, _, _)) => n,
466 None => container_name(&handle.workload, handle.replica),
467 };
468 let info = match self.docker.inspect_container(&name, None).await {
469 Ok(info) => info,
470 Err(_) => return Ok(Health::Unhealthy),
471 };
472 let running = info.state.and_then(|s| s.running).unwrap_or(false);
473 Ok(if running {
474 Health::Healthy
475 } else {
476 Health::Unhealthy
477 })
478 }
479}
480
481impl DockerBackend {
482 async fn container_ip(&self, id: &str) -> Result<String, BackendError> {
485 let info = self
486 .docker
487 .inspect_container(id, None)
488 .await
489 .map_err(|e| BackendError::Launch(format!("inspect {id}: {e}")))?;
490 let networks = info
491 .network_settings
492 .ok_or_else(|| BackendError::Launch("container has no network settings".into()))?;
493 if let Some(ip) = networks.ip_address.filter(|s| !s.is_empty()) {
495 return Ok(ip);
496 }
497 if let Some(nets) = networks.networks {
498 for net in nets.values() {
499 if let Some(ip) = net.ip_address.as_ref().filter(|s| !s.is_empty()) {
500 return Ok(ip.clone());
501 }
502 }
503 }
504 Err(BackendError::Launch("container has no IP address".into()))
505 }
506
507 async fn published_host_port(&self, id: &str, port_key: &str) -> Result<u16, BackendError> {
510 let info = self
511 .docker
512 .inspect_container(id, None)
513 .await
514 .map_err(|e| BackendError::Launch(format!("inspect {id}: {e}")))?;
515 info.network_settings
516 .and_then(|ns| ns.ports)
517 .and_then(|mut ports| ports.remove(port_key).flatten())
518 .and_then(|bindings| bindings.into_iter().next())
519 .and_then(|b| b.host_port)
520 .and_then(|hp| hp.parse::<u16>().ok())
521 .ok_or_else(|| {
522 BackendError::Launch(format!("no published host port for {port_key} on {id}"))
523 })
524 }
525}
526
527#[cfg(test)]
528mod tests {
529 use super::*;
530
531 #[test]
532 fn docker_endpoint_defaults_to_published_and_parses_lowercase() {
533 assert_eq!(DockerEndpoint::default(), DockerEndpoint::Published);
535 assert_eq!(
537 serde_json::from_str::<DockerEndpoint>("\"published\"").unwrap(),
538 DockerEndpoint::Published
539 );
540 assert_eq!(
541 serde_json::from_str::<DockerEndpoint>("\"bridge\"").unwrap(),
542 DockerEndpoint::Bridge
543 );
544 }
545
546 #[test]
547 fn name_and_ref_round_trip() {
548 assert_eq!(container_name("web", 0), "boatramp-web-0");
549 let r = encode_ref("boatramp-web-0", "172.17.0.3", 8080);
550 assert_eq!(r, "boatramp-web-0@172.17.0.3:8080");
551 assert_eq!(
552 decode_ref(&r),
553 Some(("boatramp-web-0".to_string(), "172.17.0.3".to_string(), 8080))
554 );
555 assert_eq!(decode_ref("garbage"), None);
556 }
557
558 #[test]
559 fn host_config_is_hardened_by_default() {
560 let hc = hardened_host_config(256, 2, RestartPolicy::Never, false);
561 assert_eq!(hc.memory, Some(256 * 1024 * 1024));
563 assert_eq!(hc.nano_cpus, Some(2_000_000_000));
564 assert_eq!(hc.pids_limit, Some(MAX_PIDS));
565 assert_eq!(
567 hc.security_opt.as_deref(),
568 Some(["no-new-privileges:true".to_string()].as_slice())
569 );
570 assert_eq!(hc.cap_drop.as_deref(), Some(["ALL".to_string()].as_slice()));
571 assert_eq!(hc.readonly_rootfs, Some(true));
572 let tmpfs = hc.tmpfs.expect("tmpfs mounts for a read-only rootfs");
574 assert!(tmpfs.get("/tmp").is_some_and(|o| o.contains("noexec")));
575 assert!(tmpfs.contains_key("/run"));
576 assert_eq!(
578 hardened_host_config(64, 0, RestartPolicy::Never, false).nano_cpus,
579 Some(1_000_000_000)
580 );
581 }
582
583 #[test]
584 fn writable_root_relaxes_only_the_read_only_root() {
585 let hc = hardened_host_config(256, 2, RestartPolicy::Never, true);
586 assert_eq!(hc.readonly_rootfs, Some(false));
588 assert_eq!(
590 hc.security_opt.as_deref(),
591 Some(["no-new-privileges:true".to_string()].as_slice())
592 );
593 assert_eq!(hc.cap_drop.as_deref(), Some(["ALL".to_string()].as_slice()));
594 assert_eq!(hc.pids_limit, Some(MAX_PIDS));
595 }
596
597 #[test]
598 fn writable_root_is_off_by_default_on_the_backend() {
599 let docker = Docker::connect_with_defaults().unwrap();
601 let backend = DockerBackend::with_client(docker);
602 assert!(!backend.writable_root_allowed);
603 assert!(
604 backend
605 .with_writable_root_allowed(true)
606 .writable_root_allowed,
607 "the single-tenant posture opts in"
608 );
609 }
610
611 #[test]
612 fn named_volume_mode_builds_a_prefixed_daemon_volume_mount() {
613 let vol = VolumeRef {
614 name: "db".into(),
615 mount: "/data".into(),
616 size_mib: 64,
617 };
618 let m = volume_mount(&vol, DockerVolumeMode::Named, Path::new("/srv/data"));
619 assert_eq!(m.typ, Some(MountTypeEnum::VOLUME));
620 assert_eq!(m.source.as_deref(), Some("boatramp-db"));
622 assert_eq!(m.target.as_deref(), Some("/data"));
623 assert_eq!(m.read_only, Some(false), "a persistent volume is writable");
624 }
625
626 #[test]
627 fn bind_volume_mode_builds_a_host_path_mount() {
628 let vol = VolumeRef {
629 name: "db".into(),
630 mount: "/data".into(),
631 size_mib: 64,
632 };
633 let m = volume_mount(&vol, DockerVolumeMode::Bind, Path::new("/srv/data"));
634 assert_eq!(m.typ, Some(MountTypeEnum::BIND));
635 assert_eq!(m.source.as_deref(), Some("/srv/data/compute/volumes/db"));
636 assert_eq!(m.target.as_deref(), Some("/data"));
637 assert_eq!(m.read_only, Some(false));
638 }
639
640 #[test]
641 fn validate_volume_rejects_traversal_in_name_and_mount() {
642 assert!(validate_volume("db", "/data").is_ok());
643 assert!(validate_volume("cache-1", "/var/lib/app").is_ok());
644 assert!(validate_volume("../etc", "/data").is_err());
646 assert!(validate_volume("a/b", "/data").is_err());
647 assert!(validate_volume("db", "relative").is_err());
649 assert!(validate_volume("db", "/data/../etc").is_err());
650 }
651
652 #[test]
653 fn volume_mode_defaults_to_named_and_parses_lowercase() {
654 assert_eq!(DockerVolumeMode::default(), DockerVolumeMode::Named);
655 assert_eq!(
656 serde_json::from_str::<DockerVolumeMode>("\"named\"").unwrap(),
657 DockerVolumeMode::Named
658 );
659 assert_eq!(
660 serde_json::from_str::<DockerVolumeMode>("\"bind\"").unwrap(),
661 DockerVolumeMode::Bind
662 );
663 }
664
665 #[test]
666 fn restart_policy_maps_to_docker() {
667 assert_eq!(
668 restart_policy(RestartPolicy::Always).name,
669 Some(RestartPolicyNameEnum::ALWAYS)
670 );
671 assert_eq!(
672 restart_policy(RestartPolicy::OnFailure).name,
673 Some(RestartPolicyNameEnum::ON_FAILURE)
674 );
675 assert_eq!(
676 restart_policy(RestartPolicy::Never).name,
677 Some(RestartPolicyNameEnum::NO)
678 );
679 }
680}