Something went wrong. Try again.
Runtime standardisation for decentralised apps on the Atmosphere
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814use async_trait::async_trait;use at_run_core::{ ArchiveLimits, ArchiveSummary, ArchiveValidationError, PinnedImageRef, RuntimeTarget, validate_archive,};use bollard::{ Docker, body_full, models::{ContainerCreateBody, HostConfig, Mount, MountType, VolumeCreateRequest}, query_parameters::{ CreateImageOptionsBuilder, DownloadFromContainerOptionsBuilder, RemoveContainerOptionsBuilder, RemoveVolumeOptionsBuilder, UploadToContainerOptionsBuilder, },};use futures_util::StreamExt;use std::{collections::HashMap, path::Path};use thiserror::Error;
const MATERIALIZATION_TARGET: &str = "/at-run-output";
/// Small portability boundary around the container backend.////// Bollard selects Unix sockets, Windows named pipes, or HTTP transports from/// the local Docker environment. Runner orchestration should depend on this/// trait rather than on Bollard's generated request models directly.#[async_trait]pub trait ContainerEngine: Send + Sync { fn backend_name(&self) -> &'static str;
async fn ping(&self) -> Result<(), EngineError>;
/// Return the OCI platform used by containers created by this daemon. async fn runtime_target(&self) -> Result<RuntimeTarget, EngineError>;
/// Pull and safely materialize one exported directory from a public image /// into a named Docker volume without starting that image. async fn materialize_image_export( &self, image: &PinnedImageRef, export_path: &str, volume_name: &str, limits: ArchiveLimits, ) -> Result<MaterializedVolume, EngineError>;
/// Copy a validated tar into a volume through a stopped carrier /// container. The carrier image is never started by this operation. async fn materialize_archive( &self, carrier_image: &PinnedImageRef, archive: &[u8], volume_name: &str, limits: ArchiveLimits, ) -> Result<MaterializedVolume, EngineError>;
/// Create, but do not start, the runner-owned sandbox container. async fn create_sandbox( &self, specification: &SandboxSpec, ) -> Result<StoppedSandbox, EngineError>;}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct MaterializedVolume { pub name: String, pub archive: ArchiveSummary,}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct SandboxMount { pub volume_name: String, pub target: String,}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct SandboxSpec { pub image: PinnedImageRef, pub user: String, pub supervisor: String, pub command: Vec<String>, pub environment: Vec<String>, pub mounts: Vec<SandboxMount>, pub memory_bytes: i64, pub pids: i64, pub work_bytes: u64,}
#[derive(Clone, Debug, Eq, PartialEq)]pub struct StoppedSandbox { pub id: String, pub warnings: Vec<String>,}
#[derive(Clone)]pub struct DockerEngine { client: Docker,}
impl DockerEngine { /// Connect using the platform's local Docker defaults. pub fn connect_local() -> Result<Self, EngineError> { if std::env::var_os("DOCKER_HOST").is_some() { return Ok(Self { client: Docker::connect_with_defaults()?, }); }
match Docker::connect_with_local_defaults() { Ok(client) => Ok(Self { client }), Err(default_error) => { #[cfg(unix)] if let Some(client) = docker_desktop_client() { return Ok(Self { client }); }
Err(default_error.into()) } } }
pub fn client(&self) -> &Docker { &self.client }
async fn ensure_image(&self, image: &str) -> Result<(), EngineError> { match self.client.inspect_image(image).await { Ok(_) => return Ok(()), Err(bollard::errors::Error::DockerResponseServerError { status_code: 404, .. }) => {} Err(error) => return Err(error.into()), }
let mut progress = self.client.create_image( Some( CreateImageOptionsBuilder::default() .from_image(image) .build(), ), None, None, ); while let Some(result) = progress.next().await { result?; } Ok(()) }
async fn materialize_existing_image_export( &self, image: &str, export_path: &str, volume_name: &str, limits: ArchiveLimits, ) -> Result<MaterializedVolume, EngineError> { validate_export_path(export_path)?;
let mut labels = HashMap::new(); labels.insert("dev.at-run.managed".to_owned(), "true".to_owned()); labels.insert("dev.at-run.source.image".to_owned(), image.to_owned()); labels.insert("dev.at-run.source.path".to_owned(), export_path.to_owned());
let volume = self .client .create_volume(VolumeCreateRequest { name: Some(volume_name.to_owned()), labels: Some(labels), ..Default::default() }) .await?;
let result = self .populate_volume(image, export_path, &volume.name, limits) .await;
if result.is_err() { let _ = self .client .remove_volume( &volume.name, Some(RemoveVolumeOptionsBuilder::default().force(true).build()), ) .await; }
result.map(|archive| MaterializedVolume { name: volume.name, archive, }) }
async fn populate_volume( &self, image: &str, export_path: &str, volume_name: &str, limits: ArchiveLimits, ) -> Result<ArchiveSummary, EngineError> { let created = self .client .create_container( None, ContainerCreateBody { image: Some(image.to_owned()), host_config: Some(HostConfig { mounts: Some(vec![Mount { target: Some(MATERIALIZATION_TARGET.to_owned()), source: Some(volume_name.to_owned()), typ: Some(MountType::VOLUME), read_only: Some(false), ..Default::default() }]), ..Default::default() }), ..Default::default() }, ) .await?;
let operation = async { let archive = self .archive_from_container(&created.id, export_path) .await?; let summary = validate_archive(&archive, limits)?;
self.client .upload_to_container( &created.id, Some( UploadToContainerOptionsBuilder::default() .path(MATERIALIZATION_TARGET) .build(), ), body_full(archive.into()), ) .await?;
Ok::<_, EngineError>(summary) } .await;
let cleanup = self .client .remove_container( &created.id, Some(RemoveContainerOptionsBuilder::default().force(true).build()), ) .await;
match (operation, cleanup) { (Ok(summary), Ok(())) => Ok(summary), (Err(error), _) => Err(error), (Ok(_), Err(error)) => Err(error.into()), } }
async fn populate_volume_from_archive( &self, carrier_image: &str, archive: Vec<u8>, volume_name: &str, ) -> Result<(), EngineError> { let created = self .client .create_container( None, ContainerCreateBody { image: Some(carrier_image.to_owned()), host_config: Some(HostConfig { mounts: Some(vec![Mount { target: Some(MATERIALIZATION_TARGET.to_owned()), source: Some(volume_name.to_owned()), typ: Some(MountType::VOLUME), read_only: Some(false), ..Default::default() }]), ..Default::default() }), ..Default::default() }, ) .await?;
let operation = self .client .upload_to_container( &created.id, Some( UploadToContainerOptionsBuilder::default() .path(MATERIALIZATION_TARGET) .build(), ), body_full(archive.into()), ) .await; let cleanup = self .client .remove_container( &created.id, Some(RemoveContainerOptionsBuilder::default().force(true).build()), ) .await;
match (operation, cleanup) { (Ok(()), Ok(())) => Ok(()), (Err(error), _) | (Ok(()), Err(error)) => Err(error.into()), } }
async fn archive_from_container( &self, container_id: &str, path: &str, ) -> Result<Vec<u8>, EngineError> { let mut stream = self.client.download_from_container( container_id, Some( DownloadFromContainerOptionsBuilder::default() .path(path) .build(), ), ); let mut archive = Vec::new(); while let Some(chunk) = stream.next().await { archive.extend_from_slice(&chunk?); } Ok(archive) }}
#[cfg(unix)]fn docker_desktop_client() -> Option<Docker> { let socket = std::env::var_os("HOME") .map(std::path::PathBuf::from)? .join(".docker/run/docker.sock"); if !socket.exists() { return None; }
let host = format!("unix://{}", socket.display()); Docker::connect_with_host(&host).ok()}
#[async_trait]impl ContainerEngine for DockerEngine { fn backend_name(&self) -> &'static str { "docker" }
async fn ping(&self) -> Result<(), EngineError> { self.client.ping().await?; Ok(()) }
async fn runtime_target(&self) -> Result<RuntimeTarget, EngineError> { let version = self.client.version().await?; let os = version.os.ok_or(EngineError::MissingDaemonPlatform("os"))?; let architecture = version .arch .ok_or(EngineError::MissingDaemonPlatform("architecture"))?; Ok(RuntimeTarget::new(os, architecture)) }
async fn materialize_image_export( &self, image: &PinnedImageRef, export_path: &str, volume_name: &str, limits: ArchiveLimits, ) -> Result<MaterializedVolume, EngineError> { self.ensure_image(&image.to_string()).await?; self.materialize_existing_image_export(&image.to_string(), export_path, volume_name, limits) .await }
async fn materialize_archive( &self, carrier_image: &PinnedImageRef, archive: &[u8], volume_name: &str, limits: ArchiveLimits, ) -> Result<MaterializedVolume, EngineError> { let summary = validate_archive(archive, limits)?; self.ensure_image(&carrier_image.to_string()).await?;
let mut labels = HashMap::new(); labels.insert("dev.at-run.managed".to_owned(), "true".to_owned()); labels.insert( "dev.at-run.carrier.image".to_owned(), carrier_image.to_string(), );
let volume = self .client .create_volume(VolumeCreateRequest { name: Some(volume_name.to_owned()), labels: Some(labels), ..Default::default() }) .await?; let result = self .populate_volume_from_archive( &carrier_image.to_string(), archive.to_vec(), &volume.name, ) .await; if result.is_err() { let _ = self .client .remove_volume( &volume.name, Some(RemoveVolumeOptionsBuilder::default().force(true).build()), ) .await; } result?;
Ok(MaterializedVolume { name: volume.name, archive: summary, }) }
async fn create_sandbox( &self, specification: &SandboxSpec, ) -> Result<StoppedSandbox, EngineError> { validate_sandbox_spec(specification)?; self.ensure_image(&specification.image.to_string()).await?;
let mounts = specification .mounts .iter() .map(|mount| Mount { target: Some(mount.target.clone()), source: Some(mount.volume_name.clone()), typ: Some(MountType::VOLUME), read_only: Some(true), ..Default::default() }) .collect(); let mut tmpfs = HashMap::new(); tmpfs.insert( "/opt/at-run/work".to_owned(), format!("rw,nosuid,nodev,size={}", specification.work_bytes), ); let mut labels = HashMap::new(); labels.insert("dev.at-run.managed".to_owned(), "true".to_owned()); labels.insert("dev.at-run.role".to_owned(), "sandbox".to_owned());
let created = self .client .create_container( None, ContainerCreateBody { user: Some(specification.user.clone()), attach_stdin: Some(true), attach_stdout: Some(true), attach_stderr: Some(true), tty: Some(false), open_stdin: Some(true), stdin_once: Some(true), env: Some(specification.environment.clone()), cmd: Some(specification.command.clone()), healthcheck: Some(bollard::models::HealthConfig { test: Some(vec!["NONE".to_owned()]), ..Default::default() }), image: Some(specification.image.to_string()), working_dir: Some("/opt/at-run/work".to_owned()), entrypoint: Some(vec![specification.supervisor.clone()]), network_disabled: Some(true), labels: Some(labels), host_config: Some(HostConfig { memory: Some(specification.memory_bytes), memory_swap: Some(specification.memory_bytes), pids_limit: Some(specification.pids), init: Some(false), network_mode: Some("none".to_owned()), auto_remove: Some(false), mounts: Some(mounts), cap_drop: Some(vec!["ALL".to_owned()]), ipc_mode: Some("none".to_owned()), privileged: Some(false), readonly_rootfs: Some(true), security_opt: Some(vec![ "no-new-privileges=true".to_owned(), "seccomp=builtin".to_owned(), ]), tmpfs: Some(tmpfs), shm_size: Some(16 * 1024 * 1024), ..Default::default() }), ..Default::default() }, ) .await?;
Ok(StoppedSandbox { id: created.id, warnings: created.warnings, }) }}
fn validate_sandbox_spec(specification: &SandboxSpec) -> Result<(), EngineError> { if specification.user.is_empty() || specification.supervisor.is_empty() || !specification.supervisor.starts_with('/') || specification.command.is_empty() || specification.memory_bytes <= 0 || specification.pids <= 0 || specification.work_bytes == 0 || specification .environment .iter() .any(|value| value.contains('\0')) || specification.mounts.iter().any(|mount| { mount.volume_name.is_empty() || !mount.target.starts_with("/opt/at-run/") || mount.target.contains("/../") }) { return Err(EngineError::InvalidSandboxSpec); } Ok(())}
fn validate_export_path(path: &str) -> Result<(), EngineError> { let path = Path::new(path); if !path.is_absolute() || path .components() .any(|component| matches!(component, std::path::Component::ParentDir)) { return Err(EngineError::InvalidExportPath( path.to_string_lossy().into_owned(), )); } Ok(())}
#[derive(Debug, Error)]pub enum EngineError { #[error("Docker Engine operation failed: {0}")] Docker(#[from] bollard::errors::Error), #[error(transparent)] Archive(#[from] ArchiveValidationError), #[error("runtime export path must be an absolute path without parent traversal: {0:?}")] InvalidExportPath(String), #[error("Docker daemon did not report its {0}")] MissingDaemonPlatform(&'static str), #[error("invalid sandbox container specification")] InvalidSandboxSpec,}
#[cfg(test)]mod tests { use std::io::Cursor; use std::time::{SystemTime, UNIX_EPOCH};
use bollard::query_parameters::RemoveVolumeOptionsBuilder;
use super::*;
#[tokio::test] #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"] async fn materializes_export_without_starting_selected_image() { let engine = DockerEngine::connect_local().unwrap(); engine.ping().await.unwrap();
let fixture = engine .client .inspect_image("at-run-runtime-fixture:local") .await .unwrap(); let image: PinnedImageRef = fixture .repo_digests .unwrap() .into_iter() .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:")) .unwrap() .parse() .unwrap();
let nonce = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_nanos(); let volume_name = format!("at-run-test-{}-{nonce}", std::process::id());
let result = engine .materialize_image_export( &image, "/at-run/runtime", &volume_name, ArchiveLimits::default(), ) .await;
let verification_container = engine .client .create_container( None, ContainerCreateBody { image: Some(image.to_string()), host_config: Some(HostConfig { mounts: Some(vec![Mount { target: Some("/verify".to_owned()), source: Some(volume_name.clone()), typ: Some(MountType::VOLUME), read_only: Some(true), ..Default::default() }]), ..Default::default() }), ..Default::default() }, ) .await .unwrap(); let copied_file = engine .archive_from_container(&verification_container.id, "/verify/runtime/runtime.txt") .await; engine .client .remove_container( &verification_container.id, Some(RemoveContainerOptionsBuilder::default().force(true).build()), ) .await .unwrap();
let cleanup = engine .client .remove_volume( &volume_name, Some(RemoveVolumeOptionsBuilder::default().force(true).build()), ) .await;
let materialized = result.unwrap(); let copied_file = copied_file.unwrap(); cleanup.unwrap(); assert_eq!(materialized.name, volume_name); assert_eq!(materialized.archive.root, Path::new("runtime")); assert!(materialized.archive.entries >= 2); assert_eq!( validate_archive(&copied_file, ArchiveLimits::default()) .unwrap() .root, Path::new("runtime.txt") ); }
#[tokio::test] #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"] async fn materializes_pds_archive_without_starting_carrier() { let engine = DockerEngine::connect_local().unwrap(); engine.ping().await.unwrap(); let fixture = engine .client .inspect_image("at-run-runtime-fixture:local") .await .unwrap(); let image: PinnedImageRef = fixture .repo_digests .unwrap() .into_iter() .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:")) .unwrap() .parse() .unwrap(); let nonce = SystemTime::now() .duration_since(UNIX_EPOCH) .unwrap() .as_nanos(); let volume_name = format!("at-run-code-test-{}-{nonce}", std::process::id()); let archive = archive_with_file("app/main.py", b"print('hello')");
let materialized = engine .materialize_archive(&image, &archive, &volume_name, ArchiveLimits::default()) .await .unwrap(); let verification_container = engine .client .create_container( None, ContainerCreateBody { image: Some(image.to_string()), host_config: Some(HostConfig { mounts: Some(vec![Mount { target: Some("/verify".to_owned()), source: Some(volume_name.clone()), typ: Some(MountType::VOLUME), read_only: Some(true), ..Default::default() }]), ..Default::default() }), ..Default::default() }, ) .await .unwrap(); let copied = engine .archive_from_container(&verification_container.id, "/verify/app/main.py") .await .unwrap(); engine .client .remove_container( &verification_container.id, Some(RemoveContainerOptionsBuilder::default().force(true).build()), ) .await .unwrap(); engine .client .remove_volume( &volume_name, Some(RemoveVolumeOptionsBuilder::default().force(true).build()), ) .await .unwrap();
assert_eq!(materialized.archive.root, Path::new("app")); assert_eq!( validate_archive(&copied, ArchiveLimits::default()) .unwrap() .root, Path::new("main.py") ); }
#[tokio::test] #[ignore = "requires Docker and the local at-run-runtime-fixture:local image"] async fn creates_hardened_runner_owned_sandbox_in_stopped_state() { let engine = DockerEngine::connect_local().unwrap(); engine.ping().await.unwrap(); let fixture = engine .client .inspect_image("at-run-runtime-fixture:local") .await .unwrap(); let image: PinnedImageRef = fixture .repo_digests .unwrap() .into_iter() .find(|digest| digest.starts_with("at-run-runtime-fixture@sha256:")) .unwrap() .parse() .unwrap(); let specification = SandboxSpec { image, user: "65532:65532".to_owned(), supervisor: "/at-run/runtime/never-run".to_owned(), command: vec!["/opt/at-run/runtime/runtime/never-run".to_owned()], environment: vec!["AT_RUN_TEST=true".to_owned()], mounts: vec![], memory_bytes: 64 * 1024 * 1024, pids: 16, work_bytes: 1024 * 1024, };
let sandbox = engine.create_sandbox(&specification).await.unwrap(); let inspected = engine .client .inspect_container(&sandbox.id, None) .await .unwrap(); engine .client .remove_container( &sandbox.id, Some(RemoveContainerOptionsBuilder::default().force(true).build()), ) .await .unwrap();
let state = inspected.state.unwrap(); let config = inspected.config.unwrap(); let host = inspected.host_config.unwrap(); assert_eq!(state.running, Some(false)); assert_eq!(config.user.as_deref(), Some("65532:65532")); assert_eq!(host.readonly_rootfs, Some(true)); assert_eq!(host.network_mode.as_deref(), Some("none")); assert_eq!(host.cap_drop, Some(vec!["ALL".to_owned()])); assert_eq!(host.memory, Some(64 * 1024 * 1024)); let security = host.security_opt.unwrap(); assert!( security .iter() .any(|option| option.starts_with("no-new-privileges")) ); assert!(security.iter().any(|option| option == "seccomp=builtin")); }
fn archive_with_file(path: &str, contents: &[u8]) -> Vec<u8> { let mut bytes = Vec::new(); { let mut builder = tar::Builder::new(&mut bytes); let mut header = tar::Header::new_gnu(); header.set_mode(0o644); header.set_size(contents.len() as u64); header.set_cksum(); builder .append_data(&mut header, path, Cursor::new(contents)) .unwrap(); builder.finish().unwrap(); } bytes }}