feat: add initial support for syncing files to the enclave

The new `files` key in the manifest specifies the list of file paths that should be synced
to the enclave. Only absolute paths without any relative components are accepted.

This is a simplified implementation to support passing Kubernetes service account credentials
and other ConfigMap and Secret data mounted as pod volumes. It comes with the following
limitations:

* All files must exist on the host (pod) when `enclaver-run` starts. If a file does not exist
  when the enclave is launched, the enclave will hang forever waiting for its initial sync.
* If the path specified in a manifest is a symlink and one of path components of its target
  is removed or replaced non-atomically (i.e. removed and recreated slower than the inotify
  debounce interval of 1 second), the file will stop being synced.
* Files are synced sequentially, so the startup time of the enclave may be slow if they are large.

Author: sangwa

enclaver/Cargo.lock

@@ -34,6 +34,7 @@ base64 = "0.13"
 bollard = { git = "https://github.com/fussybeaver/bollard.git", branch = "master" }
 bytes = "1.6"
 cbc = { version = "0.1", features = [ "std", "block-padding" ] }
+chrono = "0.4"
 circbuf = "0.2"
 clap = { version = "4.5.31", features = ["derive"] }
 console-subscriber = { version = "0.1.10", optional = true }
@@ -52,6 +53,7 @@ json = "0.12"
 lazy_static = "1.5"
 log = "0.4"
 nix = "0.24"
+notify-debouncer-full = "0.5"
 pkcs8 = { version = "0.9", features = ["pem"] }
 pretty_env_logger = "0.5"
 rand = { version = "0.8", features = ["std", "std_rng"] }

enclaver/Cargo.toml

@@ -0,0 +1,211 @@
+use anyhow::{anyhow, Context, Result};
+use futures::{Stream, StreamExt};
+use log::{error, info};
+use std::collections::{HashMap, HashSet};
+use std::path::PathBuf;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
+use tokio::sync::{mpsc, oneshot};
+use tokio::task::JoinHandle;
+use tokio_util::codec::{BytesCodec, FramedRead};
+use tokio_vsock::VsockStream;
+
+use crate::config::Configuration;
+use enclaver::constants::FILE_SYNC_PORT;
+use enclaver::files;
+use enclaver::json_transport::JsonTransport;
+use enclaver::vsock;
+
+struct FileSyncServer {
+    files: Arc<HashMap<String, PathBuf>>,
+    incoming: Box<dyn Stream<Item = VsockStream> + Unpin + Send>,
+    initial_sync_done: Arc<AtomicBool>,
+    initial_sync_notifier: oneshot::Sender<()>,
+}
+
+impl FileSyncServer {
+    pub fn new(sender: oneshot::Sender<()>, files: &HashSet<String>) -> Result<Self> {
+        let incoming = Box::new(vsock::serve(FILE_SYNC_PORT)?);
+
+        Ok(Self {
+            files: Arc::new(
+                files
+                    .iter()
+                    .map(|file| (file.clone(), PathBuf::from(file)))
+                    .collect::<HashMap<_, _>>(),
+            ),
+            incoming: incoming,
+            initial_sync_done: Arc::new(AtomicBool::new(false)),
+            initial_sync_notifier: sender,
+        })
+    }
+
+    pub async fn serve(self) -> Result<()> {
+        for path in self.files.values() {
+            if let Some(parent) = path.parent() {
+                fs::create_dir_all(parent).await.context(format!(
+                    "cannot create parent directory for {}",
+                    path.to_string_lossy()
+                ))?;
+            } else {
+                return Err(anyhow!(
+                    "cannot find parent directory for {}",
+                    path.to_string_lossy()
+                ));
+            }
+        }
+
+        info!("Waiting for the host to connect for file sync");
+
+        let mut incoming = Box::into_pin(self.incoming);
+        loop {
+            let initial_conn = incoming.next().await;
+            if let Some(_) = initial_conn {
+                break;
+            }
+        }
+
+        info!("Accepting file sync connections");
+
+        let (event_tx, mut event_rx) = mpsc::channel(self.files.len());
+        let mut initial_sync_files = self.files.keys().cloned().collect::<HashSet<_>>();
+        let initial_sync_done = self.initial_sync_done.clone();
+
+        let server = tokio::task::spawn(async move {
+            while let Some(stream) = incoming.next().await {
+                let initial_sync_done = initial_sync_done.clone();
+                let files = self.files.clone();
+                let tx = event_tx.clone();
+                tokio::task::spawn(async move {
+                    match FileSyncServer::service_conn(stream, files).await {
+                        Ok(file) => {
+                            if !initial_sync_done.load(Ordering::SeqCst) {
+                                let _ = tx.send(file).await;
+                            }
+                        }
+                        Err(err) => error!("{err}"),
+                    }
+                });
+            }
+        });
+
+        let initial_sync_done = self.initial_sync_done.clone();
+
+        while !initial_sync_done.load(Ordering::SeqCst) {
+            if let Some(file) = event_rx.recv().await {
+                initial_sync_files.remove(&file);
+                if initial_sync_files.is_empty() {
+                    initial_sync_done.store(true, Ordering::SeqCst);
+                    break;
+                }
+            }
+        }
+
+        let _ = self.initial_sync_notifier.send(());
+        server.await?;
+
+        Ok(())
+    }
+
+    async fn service_conn(
+        mut vsock: VsockStream,
+        files: Arc<HashMap<String, PathBuf>>,
+    ) -> Result<String> {
+        let file_meta = files::Metadata::recv(&mut vsock).await?;
+
+        if let Some(file) = files.get(&file_meta.path) {
+            info!(
+                "Syncing file: {} ({} bytes)",
+                file_meta.path, file_meta.size
+            );
+
+            let now = chrono::Utc::now();
+            let ts = now.timestamp_millis().to_string();
+
+            let mut tmp_file = file.as_os_str().to_os_string();
+            tmp_file.push(std::ffi::OsString::from(".".to_string() + &ts));
+
+            {
+                let mut file = fs::File::create(&tmp_file).await?;
+
+                if file_meta.size > 0 {
+                    let mut reader = FramedRead::new(vsock, BytesCodec::new());
+                    let mut bytes_written: u64 = 0;
+
+                    while let Some(Ok(chunk)) = reader.next().await {
+                        let bytes_pending = if bytes_written + chunk.len() as u64 > file_meta.size {
+                            bytes_written + chunk.len() as u64 - file_meta.size
+                        } else {
+                            chunk.len() as u64
+                        };
+
+                        file.write_all(&chunk[..bytes_pending as usize]).await?;
+                        bytes_written += bytes_pending;
+                    }
+
+                    file.flush().await?;
+                }
+            }
+
+            fs::rename(&tmp_file, &file).await?;
+
+            info!("File sync done: {}", file_meta.path);
+        } else {
+            return Err(anyhow!(
+                "received sync request for unknown file {}",
+                file_meta.path
+            ));
+        }
+
+        Ok(file_meta.path)
+    }
+}
+
+pub struct FileSyncService {
+    server: Option<JoinHandle<()>>,
+}
+
+impl FileSyncService {
+    pub async fn start(config: &Configuration) -> Result<Self> {
+        let mut files = HashSet::new();
+
+        if let Some(ref manifest_files) = config.manifest.files {
+            for file_path in manifest_files {
+                files.insert(file_path.clone());
+            }
+        }
+
+        let task = if files.is_empty() {
+            None
+        } else {
+            info!("Starting file sync server");
+
+            let (sync_tx, sync_rx) = oneshot::channel();
+            let server = FileSyncServer::new(sync_tx, &files)?;
+
+            let handle = tokio::task::spawn(async move {
+                if let Err(err) = server.serve().await {
+                    error!("{err}");
+                }
+            });
+
+            info!("Waiting for initial file sync");
+            let _ = sync_rx.await;
+            info!("Initial file sync complete");
+
+            Some(handle)
+        };
+
+        Ok(Self { server: task })
+    }
+
+    pub async fn stop(self) {
+        if let Some(server) = self.server {
+            server.abort();
+            _ = server.await;
+        }
+    }
+}

enclaver/src/bin/odyn/file_sync.rs

@@ -5,6 +5,7 @@ pub mod config;
 pub mod console;
 pub mod egress;
 pub mod enclave;
+pub mod file_sync;
 pub mod ingress;
 pub mod kms_proxy;
 pub mod launcher;
@@ -24,6 +25,7 @@ use api::ApiService;
 use config::Configuration;
 use console::{AppLog, AppStatus};
 use egress::EgressService;
+use file_sync::FileSyncService;
 use ingress::IngressService;
 use kms_proxy::KmsProxyService;
 
@@ -62,13 +64,15 @@ async fn launch(args: &CliArgs) -> Result<launcher::ExitStatus> {
     let kms_proxy = KmsProxyService::start(config.clone(), nsm.clone()).await?;
     let api = ApiService::start(&config, nsm.clone()).await?;
     let env = sync_environment(&config).await?;
+    let file_sync = FileSyncService::start(&config).await?;
 
     let creds = launcher::Credentials { uid: 0, gid: 0 };
 
     info!("Starting {:?}", args.entrypoint);
     let exit_status = launcher::start_child(args.entrypoint.clone(), creds, env).await??;
     info!("Entrypoint {}", exit_status);
 
+    file_sync.stop().await;
     api.stop().await;
     kms_proxy.stop().await;
     ingress.stop().await;

enclaver/src/bin/odyn/main.rs

@@ -81,6 +81,20 @@ impl EnclaveArtifactBuilder {
 
         self.analyze_manifest(&manifest);
 
+        if let Some(ref files) = &manifest.files {
+            for path in files.iter() {
+                if PathBuf::from(path).components().any(|component| {
+                    component == std::path::Component::ParentDir
+                        || component == std::path::Component::CurDir
+                }) {
+                    return Err(anyhow!(
+                        "File path with relative components in manifest: {}",
+                        path,
+                    ));
+                }
+            }
+        }
+
         let resolved_sources = self.resolve_sources(&manifest).await?;
 
         let amended_img = self

enclaver/src/build.rs

@@ -14,6 +14,7 @@ pub const STATUS_PORT: u32 = 17000;
 pub const APP_LOG_PORT: u32 = 17001;
 pub const HTTP_EGRESS_VSOCK_PORT: u32 = 17002;
 pub const ENV_SYNC_PORT: u32 = 17003;
+pub const FILE_SYNC_PORT: u32 = 17004;
 
 // Default TCP Port that the egress proxy listens on inside the enclave, if not
 // specified in the manifest.