use std::path::Path;
use bytes::Bytes;
use http_body_util::{BodyExt, Limited};
use crate::compose::types::ComposeFile;
use crate::error::{ComposeError, Result};
use crate::libpod::urlencoded;
use crate::libpod::API_PREFIX;
use super::Engine;
mod archive;
use archive::{extract_archive, pack_path};
const MAX_CP_ARCHIVE_BYTES: usize = 1024 * 1024 * 1024;
#[derive(Default)]
pub struct CpOptions {
pub index: Option<u32>,
pub follow_link: bool,
pub archive: bool,
}
impl Engine {
pub async fn cp(&self, file: &ComposeFile, src: &str, dst: &str) -> Result<()> {
self.cp_with_options(file, src, dst, CpOptions::default())
.await
}
pub async fn cp_with_options(
&self,
file: &ComposeFile,
src: &str,
dst: &str,
opts: CpOptions,
) -> Result<()> {
check_endpoint(src)?;
check_endpoint(dst)?;
match (parse_endpoint(src), parse_endpoint(dst)) {
(Some((service, container_path)), None) => {
self.cp_from_container(file, service, container_path, Path::new(dst), &opts)
.await
}
(None, Some((service, container_path))) => {
self.cp_to_container(file, service, Path::new(src), container_path, &opts)
.await
}
(Some(_), Some(_)) => Err(ComposeError::Unsupported(
"cp: both src and dst cannot be SERVICE:PATH".into(),
)),
(None, None) => Err(ComposeError::Unsupported(
"cp: one of src or dst must be SERVICE:PATH".into(),
)),
}
}
async fn cp_from_container(
&self,
file: &ComposeFile,
service_name: &str,
container_path: &str,
dst: &Path,
opts: &CpOptions,
) -> Result<()> {
let service = file
.services
.get(service_name)
.ok_or_else(|| ComposeError::ServiceNotFound(service_name.into()))?;
let container_name = self
.live_replica_name_at(service_name, service, opts.index)
.await?;
let path = format!(
"{API_PREFIX}/containers/{}/archive?path={}",
urlencoded(&container_name),
urlencoded(container_path),
);
let resp = self
.client
.get_stream(&path)
.await
.map_err(ComposeError::Podman)?;
let tar_bytes = Limited::new(resp.into_body(), MAX_CP_ARCHIVE_BYTES)
.collect()
.await
.map_err(|_| {
ComposeError::Unsupported(format!(
"cp: container archive exceeds {MAX_CP_ARCHIVE_BYTES} bytes; \
copy fewer files or use `podman cp` for very large transfers"
))
})?
.to_bytes()
.to_vec();
let dst = dst.to_path_buf();
tokio::task::spawn_blocking(move || extract_archive(&tar_bytes, &dst))
.await
.map_err(|e| ComposeError::Build(e.to_string()))??;
Ok(())
}
async fn cp_to_container(
&self,
file: &ComposeFile,
service_name: &str,
src: &Path,
container_path: &str,
opts: &CpOptions,
) -> Result<()> {
let service = file
.services
.get(service_name)
.ok_or_else(|| ComposeError::ServiceNotFound(service_name.into()))?;
let container_name = self
.live_replica_name_at(service_name, service, opts.index)
.await?;
let stat_path = format!(
"{API_PREFIX}/containers/{}/archive?path={}",
urlencoded(&container_name),
urlencoded(container_path),
);
let dest_is_dir = self.client.head_path_is_dir(&stat_path).await? == Some(true);
let (extract_dir, rename) = if dest_is_dir || container_path.ends_with('/') {
(container_path.trim_end_matches('/').to_string(), None)
} else {
let trimmed = container_path.trim_end_matches('/');
let (parent, name) = trimmed.rsplit_once('/').unwrap_or(("", trimmed));
let parent = if parent.is_empty() { "/" } else { parent };
(parent.to_string(), Some(name.to_string()))
};
let extract_stat_path = format!(
"{API_PREFIX}/containers/{}/archive?path={}",
urlencoded(&container_name),
urlencoded(&extract_dir),
);
match self.client.head_path_is_dir(&extract_stat_path).await? {
Some(true) => {}
Some(false) => {
return Err(ComposeError::Copy(format!(
"cp: not a directory: {extract_dir}"
)));
}
None => {
return Err(ComposeError::Copy(format!(
"cp: no such directory: {extract_dir}"
)));
}
}
let src_buf = src.to_path_buf();
let follow = opts.follow_link;
let rename_for_pack = rename.clone();
let tar_bytes = tokio::task::spawn_blocking(move || {
pack_path(&src_buf, follow, rename_for_pack.as_deref())
})
.await
.map_err(|e| ComposeError::Build(e.to_string()))??;
let entry = rename.clone().unwrap_or_else(|| {
src.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default()
});
self.put_archive_verified(&container_name, &extract_dir, &entry, tar_bytes)
.await
}
pub(super) async fn put_archive_verified(
&self,
container: &str,
dir: &str,
entry: &str,
tar_bytes: Vec<u8>,
) -> Result<()> {
let path = format!(
"{API_PREFIX}/containers/{}/archive?path={}",
urlencoded(container),
urlencoded(dir),
);
let verify_path = (!entry.is_empty()).then(|| {
format!(
"{API_PREFIX}/containers/{}/archive?path={}",
urlencoded(container),
urlencoded(&join_archive_path(dir, entry)),
)
});
let pre_mtime = match &verify_path {
Some(p) => self.client.head_path_mtime(p).await.ok(),
None => None,
};
let Err(e) = self
.client
.put_bytes_ok(&path, Bytes::from(tar_bytes), "application/gzip")
.await
else {
return Ok(());
};
if !e.is_incomplete_message() {
return Err(ComposeError::Podman(e));
}
let landed = match (&verify_path, pre_mtime) {
(Some(p), Some(pre)) => match self.client.head_path_mtime(p).await {
Ok(post) => copy_landed(pre.as_deref(), post.as_deref()),
Err(stat_err) => {
tracing::debug!(
"cp: could not re-verify {p} after an incomplete PUT: {stat_err}"
);
false
}
},
_ => false,
};
if landed {
return Ok(());
}
Err(ComposeError::Copy(format!(
"the upload to {dir} could not be confirmed — the container runtime closed the \
connection without a response and the destination did not change. The copy may \
or may not have landed; check {dir} in the container."
)))
}
}
fn copy_landed(pre: Option<&str>, post: Option<&str>) -> bool {
post.is_some() && post != pre
}
fn join_archive_path(dir: &str, entry: &str) -> String {
if dir.ends_with('/') {
format!("{dir}{entry}")
} else {
format!("{dir}/{entry}")
}
}
fn check_endpoint(s: &str) -> Result<()> {
if s == "-" {
return Err(ComposeError::Unsupported(
"cp: stdin/stdout ('-') is not supported".into(),
));
}
if let Some((svc, path)) = s.split_once(':') {
#[cfg(windows)]
if svc.len() == 1 && svc.chars().next().is_some_and(|c| c.is_ascii_alphabetic()) {
return Ok(());
}
if !svc.is_empty() && path.is_empty() {
return Err(ComposeError::Copy(format!(
"cp: empty container path in '{s}' (expected SERVICE:PATH)"
)));
}
}
Ok(())
}
fn parse_endpoint(s: &str) -> Option<(&str, &str)> {
if s == "-" {
return None;
}
let (svc, path) = s.split_once(':')?;
if svc.is_empty() || path.is_empty() {
return None;
}
#[cfg(windows)]
if svc.len() == 1 && svc.chars().next().is_some_and(|c| c.is_ascii_alphabetic()) {
return None;
}
Some((svc, path))
}
#[cfg(test)]
mod tests {
use super::{copy_landed, join_archive_path, parse_endpoint};
#[test]
fn join_archive_path_does_not_double_the_separator() {
assert_eq!(join_archive_path("/tmp", "f.txt"), "/tmp/f.txt");
assert_eq!(join_archive_path("/tmp/", "f.txt"), "/tmp/f.txt");
assert_eq!(join_archive_path("/", "f.txt"), "/f.txt");
}
#[test]
fn copy_landed_requires_the_entry_to_exist_and_its_mtime_to_move() {
assert!(copy_landed(None, Some("2026-01-01T00:00:00Z")));
assert!(copy_landed(
Some("2026-01-01T00:00:00Z"),
Some("2026-06-01T00:00:00Z")
));
assert!(!copy_landed(
Some("2026-01-01T00:00:00Z"),
Some("2026-01-01T00:00:00Z")
));
assert!(!copy_landed(Some("2026-01-01T00:00:00Z"), None));
assert!(!copy_landed(None, None));
}
#[test]
fn parse_service_colon_path() {
assert_eq!(parse_endpoint("web:/app/data"), Some(("web", "/app/data")));
}
#[test]
fn parse_local_path_no_colon() {
assert_eq!(parse_endpoint("/tmp/file.txt"), None);
}
#[test]
fn parse_dash_is_local() {
assert_eq!(parse_endpoint("-"), None);
}
#[cfg(windows)]
#[test]
fn parse_windows_drive_letter_is_local() {
assert_eq!(parse_endpoint("C:\\Users\\foo"), None);
}
#[cfg(not(windows))]
#[test]
fn single_char_service_parses_on_unix() {
assert_eq!(parse_endpoint("c:/tmp/file"), Some(("c", "/tmp/file")));
assert_eq!(parse_endpoint("w:data"), Some(("w", "data")));
}
#[test]
fn parse_empty_service_or_path() {
assert_eq!(parse_endpoint(":path"), None);
assert_eq!(parse_endpoint("svc:"), None);
}
#[cfg(windows)]
#[test]
fn parse_windows_drive_letter_forward_slash() {
assert_eq!(parse_endpoint("C:/Users/foo"), None);
}
#[test]
fn parse_service_with_relative_path() {
assert_eq!(
parse_endpoint("web:data/file.txt"),
Some(("web", "data/file.txt"))
);
}
#[test]
fn parse_service_name_with_dots() {
assert_eq!(
parse_endpoint("my.service:/app/config"),
Some(("my.service", "/app/config"))
);
}
#[test]
fn check_endpoint_rejects_dash() {
let err = super::check_endpoint("-").unwrap_err();
assert!(format!("{err}").contains("stdin/stdout"), "got: {err}");
}
#[test]
fn check_endpoint_rejects_empty_container_path() {
let err = super::check_endpoint("web:").unwrap_err();
assert!(
format!("{err}").contains("empty container path"),
"got: {err}"
);
}
#[test]
fn check_endpoint_allows_normal_forms() {
assert!(super::check_endpoint("/tmp/file").is_ok());
assert!(super::check_endpoint("web:/app/data").is_ok());
assert!(super::check_endpoint("./local").is_ok());
}
}