lazaret/crates/lazaret-prepared-lower/src/transport.rs

use std::fs::{File, OpenOptions};
use std::io::{Read, Write};
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::os::unix::fs::PermissionsExt;
use std::path::{Path, PathBuf};
use std::str::FromStr;
use std::sync::Arc;

use anyhow::{Context, Result, bail};
use lazaret_core::configuration::ConfinedPathPlan;
use oci_client::Reference;
use oci_client::client::{Client, ClientConfig};
use oci_client::manifest::{IMAGE_LAYER_GZIP_MEDIA_TYPE, IMAGE_LAYER_MEDIA_TYPE, OciImageManifest};
use oci_client::secrets::RegistryAuth;
use rustix::fs::{
    AtFlags, Mode, OFlags, RenameFlags, ResolveFlags, fstat, mkdirat, openat2, renameat_with,
    unlinkat,
};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

const OCI_LAYER_ZSTD: &str = "application/vnd.oci.image.layer.v1.tar+zstd";
const MAX_CONFIG_BYTES: usize = 16 * 1024 * 1024;
const MAX_LAYER_BYTES: u64 = 32 * 1024 * 1024 * 1024;

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct Platform {
    pub os: String,
    pub architecture: String,
    pub variant: Option<String>,
}

impl Platform {
    #[must_use]
    pub fn linux(architecture: impl Into<String>) -> Self {
        Self {
            os: "linux".to_owned(),
            architecture: architecture.into(),
            variant: None,
        }
    }
}

#[derive(Clone, Eq, PartialEq)]
pub enum RegistryCredentials {
    Anonymous,
    Basic { username: String, password: String },
    Bearer(String),
}

impl std::fmt::Debug for RegistryCredentials {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Anonymous => formatter.write_str("Anonymous"),
            Self::Basic { username, .. } => formatter
                .debug_struct("Basic")
                .field("username", username)
                .field("password", &"[redacted]")
                .finish(),
            Self::Bearer(_) => formatter
                .debug_tuple("Bearer")
                .field(&"[redacted]")
                .finish(),
        }
    }
}

#[derive(Debug)]
pub struct TransportedImage {
    pub layout_path: PathBuf,
    pub source_identity: Vec<u8>,
    pub selected_manifest_digest: [u8; 32],
    retained_layout: Option<RetainedLayout>,
}

#[derive(Debug)]
pub(crate) struct RetainedLayout {
    root: Arc<File>,
    parent: Arc<File>,
    name: String,
    identity: (u64, u64),
    cleanup_armed: bool,
}

impl RetainedLayout {
    pub(crate) fn root(&self) -> &File {
        &self.root
    }
}

impl TransportedImage {
    pub(crate) fn retained_layout(&self) -> Option<&RetainedLayout> {
        self.retained_layout.as_ref()
    }

    /// Deletes only the exact descriptor-retained transport staging leaf.
    ///
    /// # Errors
    /// Refuses a replacement path and reports any descriptor-relative cleanup failure.
    pub fn cleanup(mut self) -> Result<()> {
        if let Some(layout) = &mut self.retained_layout {
            cleanup_retained_layout(layout)?;
        }
        Ok(())
    }
}

impl Drop for RetainedLayout {
    fn drop(&mut self) {
        let _ = cleanup_retained_layout(self);
    }
}

#[derive(Clone, Debug, Default)]
pub struct OciRegistryTransport;

impl OciRegistryTransport {
    /// Pulls one explicitly selected platform into a digest-verified private OCI layout.
    ///
    /// # Errors
    ///
    /// Refuses invalid references, authentication failures, unsupported platforms, malformed
    /// descriptors, resource limits, and transport or filesystem errors.
    pub async fn pull(
        &self,
        reference: &str,
        credentials: &RegistryCredentials,
        platform: &Platform,
        layout_path: &Path,
    ) -> Result<TransportedImage> {
        self.pull_with_config(
            reference,
            credentials,
            platform,
            layout_path,
            ClientConfig::default(),
        )
        .await
    }

    /// Pulls an image into an exact descriptor-retained child of Storage Root.
    ///
    /// # Errors
    /// Refuses changed Storage Root authority, fixed-child symlinks, staging collisions,
    /// authentication failures, corrupt content, and unsupported images.
    pub async fn pull_confined(
        &self,
        reference: &str,
        credentials: &RegistryCredentials,
        platform: &Platform,
        storage_root: &ConfinedPathPlan,
        staging_id: [u8; 32],
    ) -> Result<TransportedImage> {
        self.pull_to(
            reference,
            credentials,
            platform,
            LayoutDestination::Confined {
                storage_root,
                staging_id,
            },
            ClientConfig::default(),
        )
        .await
    }

    async fn pull_with_config(
        &self,
        reference: &str,
        credentials: &RegistryCredentials,
        platform: &Platform,
        layout_path: &Path,
        config: ClientConfig,
    ) -> Result<TransportedImage> {
        self.pull_to(
            reference,
            credentials,
            platform,
            LayoutDestination::Ambient(layout_path),
            config,
        )
        .await
    }

    async fn pull_to(
        &self,
        reference: &str,
        credentials: &RegistryCredentials,
        platform: &Platform,
        destination: LayoutDestination<'_>,
        mut config: ClientConfig,
    ) -> Result<TransportedImage> {
        validate_credentials(credentials)?;
        let reference = Reference::from_str(reference).context("invalid OCI image reference")?;
        let selected_platform = platform.clone();
        config.max_concurrent_download = 1;
        config.platform_resolver = Some(Box::new(move |entries| {
            entries
                .iter()
                .find(|entry| {
                    entry.platform.as_ref().is_some_and(|candidate| {
                        candidate.os.to_string() == selected_platform.os
                            && candidate.architecture.to_string() == selected_platform.architecture
                            && candidate.variant == selected_platform.variant
                    })
                })
                .map(|entry| entry.digest.clone())
        }));
        let client = Client::new(config);
        let auth = match credentials {
            RegistryCredentials::Anonymous => RegistryAuth::Anonymous,
            RegistryCredentials::Basic { username, password } => {
                RegistryAuth::Basic(username.clone(), password.clone())
            }
            RegistryCredentials::Bearer(token) => RegistryAuth::Bearer(token.clone()),
        };
        let (manifest, manifest_digest, config_json) = client
            .pull_manifest_and_config(&reference, &auth)
            .await
            .context("pulling selected OCI manifest and config")?;
        if config_json.len() > MAX_CONFIG_BYTES {
            bail!("OCI config exceeds {MAX_CONFIG_BYTES} bytes");
        }
        verify_platform(&config_json, platform)?;
        validate_manifest(&manifest)?;
        let layout = LayoutWriter::create(destination)?;
        layout.write_verified_blob(&manifest.config.digest, config_json.as_bytes())?;
        for descriptor in &manifest.layers {
            let size = u64::try_from(descriptor.size).context("negative OCI layer size")?;
            if size > MAX_LAYER_BYTES {
                bail!("OCI layer exceeds {MAX_LAYER_BYTES} bytes");
            }
            let digest_component = digest_component(&descriptor.digest)?;
            if layout.blob_exists(digest_component)? {
                layout.require_existing_blob(&descriptor.digest, size)?;
                continue;
            }
            let (file, temporary) = layout.create_blob(digest_component)?;
            let file = tokio::fs::File::from_std(file);
            if let Err(error) = client.pull_blob(&reference, descriptor, file).await {
                layout.remove_temporary_blob(&temporary);
                return Err(error)
                    .with_context(|| format!("pulling OCI layer {}", descriptor.digest));
            }
            let observed = layout.temporary_blob_size(&temporary)?;
            if observed != size {
                bail!("OCI layer {} size mismatch", descriptor.digest);
            }
            layout.publish_blob(&temporary, digest_component)?;
        }
        let local_manifest = serde_json::to_vec(&manifest)?;
        let local_digest = format!("sha256:{}", hex(&Sha256::digest(&local_manifest)));
        layout.write_verified_blob(&local_digest, &local_manifest)?;
        layout.write_layout_index(&local_digest, local_manifest.len())?;
        let selected_manifest_digest = parse_sha256(&local_digest)?;
        let source_identity = format!(
            "oci-v1\0{}\0{}\0{manifest_digest}",
            reference.registry(),
            reference.repository()
        )
        .into_bytes();
        Ok(TransportedImage {
            layout_path: layout.label().to_owned(),
            source_identity,
            selected_manifest_digest,
            retained_layout: layout.into_retained(),
        })
    }
}

#[derive(Deserialize)]
struct ImageConfigPlatform {
    os: String,
    architecture: String,
    #[serde(default)]
    variant: Option<String>,
}

fn verify_platform(config: &str, expected: &Platform) -> Result<()> {
    let observed: ImageConfigPlatform = serde_json::from_str(config)?;
    if observed.os != expected.os
        || observed.architecture != expected.architecture
        || observed.variant != expected.variant
    {
        bail!(
            "selected image platform {}/{} does not match requested {}/{}",
            observed.os,
            observed.architecture,
            expected.os,
            expected.architecture
        );
    }
    Ok(())
}

fn validate_credentials(credentials: &RegistryCredentials) -> Result<()> {
    match credentials {
        RegistryCredentials::Anonymous => Ok(()),
        RegistryCredentials::Basic { username, password }
            if !username.is_empty() && !password.is_empty() =>
        {
            Ok(())
        }
        RegistryCredentials::Bearer(token) if !token.is_empty() => Ok(()),
        RegistryCredentials::Basic { .. } | RegistryCredentials::Bearer(_) => {
            bail!("registry credentials are incomplete")
        }
    }
}

fn validate_manifest(manifest: &OciImageManifest) -> Result<()> {
    if manifest.schema_version != 2 || manifest.layers.len() > 256 {
        bail!("unsupported OCI manifest schema or layer count");
    }
    for descriptor in &manifest.layers {
        if !matches!(
            descriptor.media_type.as_str(),
            IMAGE_LAYER_MEDIA_TYPE | IMAGE_LAYER_GZIP_MEDIA_TYPE | OCI_LAYER_ZSTD
        ) {
            bail!("unsupported OCI layer media type {}", descriptor.media_type);
        }
        digest_component(&descriptor.digest)?;
        u64::try_from(descriptor.size).context("negative OCI layer size")?;
    }
    digest_component(&manifest.config.digest)?;
    Ok(())
}

#[derive(Clone, Copy)]
enum LayoutDestination<'a> {
    Ambient(&'a Path),
    Confined {
        storage_root: &'a ConfinedPathPlan,
        staging_id: [u8; 32],
    },
}

enum LayoutWriter {
    Ambient(PathBuf),
    Confined {
        layout: RetainedLayout,
        label: PathBuf,
    },
}

impl LayoutWriter {
    fn create(destination: LayoutDestination<'_>) -> Result<Self> {
        match destination {
            LayoutDestination::Ambient(path) => {
                create_layout(path)?;
                Ok(Self::Ambient(path.to_owned()))
            }
            LayoutDestination::Confined {
                storage_root,
                staging_id,
            } => {
                let storage = File::from(storage_root.open_authority()?);
                create_directory(&storage, "image-transports")?;
                let parent = Arc::new(open_directory(&storage, Path::new("image-transports"))?);
                let name = hex(&staging_id);
                mkdirat(&parent, name.as_str(), Mode::from_raw_mode(0o700))?;
                let root = Arc::new(open_directory(&parent, Path::new(&name))?);
                let stat = fstat(&root)?;
                create_directory(&root, "blobs")?;
                let blobs = open_directory(&root, Path::new("blobs"))?;
                create_directory(&blobs, "sha256")?;
                write_relative_atomic(
                    &root,
                    Path::new("oci-layout"),
                    b"{\"imageLayoutVersion\":\"1.0.0\"}\n",
                )?;
                let label = PathBuf::from(std::ffi::OsString::from_vec(
                    storage_root.canonical_label().as_bytes().to_vec(),
                ))
                .join("image-transports")
                .join(&name);
                Ok(Self::Confined {
                    layout: RetainedLayout {
                        root,
                        parent,
                        name,
                        identity: (stat.st_dev, stat.st_ino),
                        cleanup_armed: true,
                    },
                    label,
                })
            }
        }
    }

    fn label(&self) -> &Path {
        match self {
            Self::Ambient(path) => path,
            Self::Confined { label, .. } => label,
        }
    }

    fn into_retained(self) -> Option<RetainedLayout> {
        match self {
            Self::Ambient(_) => None,
            Self::Confined { layout, .. } => Some(layout),
        }
    }

    fn write_verified_blob(&self, digest: &str, bytes: &[u8]) -> Result<()> {
        if parse_sha256(digest)? != Sha256::digest(bytes).as_slice() {
            bail!("OCI descriptor digest mismatch for {digest}");
        }
        let component = digest_component(digest)?;
        if self.blob_exists(component)? {
            return self.require_existing_blob(digest, bytes.len() as u64);
        }
        let (mut file, temporary) = self.create_blob(component)?;
        file.write_all(bytes)?;
        file.sync_all()?;
        self.publish_blob(&temporary, component)
    }

    fn blob_exists(&self, component: &str) -> Result<bool> {
        match self {
            Self::Ambient(path) => Ok(path.join("blobs/sha256").join(component).exists()),
            Self::Confined { layout, .. } => match open_relative_file(
                layout.root(),
                &Path::new("blobs/sha256").join(component),
                OFlags::RDONLY,
            ) {
                Ok(_) => Ok(true),
                Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
                Err(error) => Err(error.into()),
            },
        }
    }

    fn require_existing_blob(&self, digest: &str, size: u64) -> Result<()> {
        let mut file = self.open_blob(digest_component(digest)?)?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes)?;
        if bytes.len() as u64 != size || parse_sha256(digest)? != Sha256::digest(&bytes).as_slice()
        {
            bail!("existing OCI blob does not match its descriptor");
        }
        Ok(())
    }

    fn open_blob(&self, component: &str) -> Result<File> {
        match self {
            Self::Ambient(path) => Ok(File::open(path.join("blobs/sha256").join(component))?),
            Self::Confined { layout, .. } => Ok(open_relative_file(
                layout.root(),
                &Path::new("blobs/sha256").join(component),
                OFlags::RDONLY,
            )?),
        }
    }

    fn create_blob(&self, component: &str) -> Result<(File, String)> {
        let temporary = format!(".{component}.partial");
        match self {
            Self::Ambient(path) => Ok((
                OpenOptions::new()
                    .create_new(true)
                    .write(true)
                    .open(path.join("blobs/sha256").join(&temporary))?,
                temporary,
            )),
            Self::Confined { layout, .. } => Ok((
                open_relative_file(
                    layout.root(),
                    &Path::new("blobs/sha256").join(&temporary),
                    OFlags::CREATE | OFlags::EXCL | OFlags::WRONLY,
                )?,
                temporary,
            )),
        }
    }

    fn temporary_blob_size(&self, temporary: &str) -> Result<u64> {
        Ok(fstat(&self.open_blob(temporary)?)?.st_size.cast_unsigned())
    }

    fn publish_blob(&self, temporary: &str, component: &str) -> Result<()> {
        match self {
            Self::Ambient(path) => {
                std::fs::rename(
                    path.join("blobs/sha256").join(temporary),
                    path.join("blobs/sha256").join(component),
                )?;
            }
            Self::Confined { layout, .. } => {
                let blobs = open_directory(layout.root(), Path::new("blobs/sha256"))?;
                renameat_with(&blobs, temporary, &blobs, component, RenameFlags::NOREPLACE)?;
                blobs.sync_all()?;
            }
        }
        Ok(())
    }

    fn remove_temporary_blob(&self, temporary: &str) {
        match self {
            Self::Ambient(path) => {
                let _ = std::fs::remove_file(path.join("blobs/sha256").join(temporary));
            }
            Self::Confined { layout, .. } => {
                let _ = unlinkat(
                    layout.root(),
                    Path::new("blobs/sha256").join(temporary),
                    AtFlags::empty(),
                );
            }
        }
    }

    fn write_layout_index(&self, digest: &str, size: usize) -> Result<()> {
        let bytes = layout_index(digest, size)?;
        match self {
            Self::Ambient(path) => std::fs::write(path.join("index.json"), bytes)?,
            Self::Confined { layout, .. } => {
                write_relative_atomic(layout.root(), Path::new("index.json"), &bytes)?;
            }
        }
        Ok(())
    }
}

fn create_directory(parent: &File, path: &str) -> Result<()> {
    match mkdirat(parent, path, Mode::from_raw_mode(0o700)) {
        Ok(()) | Err(rustix::io::Errno::EXIST) => {}
        Err(error) => return Err(error.into()),
    }
    drop(open_directory(parent, Path::new(path))?);
    Ok(())
}

fn open_directory(parent: &File, path: &Path) -> std::io::Result<File> {
    openat2(
        parent,
        path,
        OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        Mode::empty(),
        confined_resolve(),
    )
    .map(File::from)
    .map_err(Into::into)
}

fn open_relative_file(parent: &File, path: &Path, flags: OFlags) -> std::io::Result<File> {
    openat2(
        parent,
        path,
        flags | OFlags::NOFOLLOW | OFlags::CLOEXEC,
        Mode::from_raw_mode(0o600),
        confined_resolve(),
    )
    .map(File::from)
    .map_err(Into::into)
}

fn write_relative_atomic(root: &File, path: &Path, bytes: &[u8]) -> Result<()> {
    let name = path.file_name().context("layout file has no name")?;
    let parent_path = path.parent().unwrap_or(Path::new("."));
    let parent = open_directory(root, parent_path)?;
    let temporary = format!(".{}.partial", name.to_string_lossy());
    let mut file = open_relative_file(
        &parent,
        Path::new(&temporary),
        OFlags::CREATE | OFlags::EXCL | OFlags::WRONLY,
    )?;
    file.write_all(bytes)?;
    file.sync_all()?;
    renameat_with(&parent, temporary, &parent, name, RenameFlags::NOREPLACE)?;
    parent.sync_all()?;
    Ok(())
}

fn confined_resolve() -> ResolveFlags {
    ResolveFlags::BENEATH
        | ResolveFlags::NO_MAGICLINKS
        | ResolveFlags::NO_SYMLINKS
        | ResolveFlags::NO_XDEV
}

fn cleanup_retained_layout(layout: &mut RetainedLayout) -> Result<()> {
    if !layout.cleanup_armed {
        return Ok(());
    }
    let current = open_directory(&layout.parent, Path::new(&layout.name))?;
    let stat = fstat(&current)?;
    if (stat.st_dev, stat.st_ino) != layout.identity {
        bail!("transport staging path no longer names the retained layout");
    }
    remove_layout_children(&layout.root)?;
    unlinkat(&layout.parent, layout.name.as_str(), AtFlags::REMOVEDIR)?;
    layout.cleanup_armed = false;
    Ok(())
}

fn remove_layout_children(directory: &File) -> Result<()> {
    let mut buffer = [std::mem::MaybeUninit::<u8>::uninit(); 4096];
    let mut entries = rustix::fs::RawDir::new(rustix::io::dup(directory)?, &mut buffer);
    while let Some(entry) = entries.next() {
        let entry = entry?;
        let name = entry.file_name();
        if name.to_bytes() == b"." || name.to_bytes() == b".." {
            continue;
        }
        if entry.file_type() == rustix::fs::FileType::Directory {
            let child = open_directory(
                directory,
                Path::new(std::ffi::OsStr::from_bytes(name.to_bytes())),
            )?;
            remove_layout_children(&child)?;
            unlinkat(directory, name, AtFlags::REMOVEDIR)?;
        } else {
            unlinkat(directory, name, AtFlags::empty())?;
        }
    }
    Ok(())
}

fn create_layout(path: &Path) -> Result<()> {
    std::fs::create_dir(path).context("creating private OCI layout")?;
    std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o700))?;
    std::fs::create_dir_all(path.join("blobs/sha256"))?;
    std::fs::set_permissions(path.join("blobs"), std::fs::Permissions::from_mode(0o700))?;
    std::fs::set_permissions(
        path.join("blobs/sha256"),
        std::fs::Permissions::from_mode(0o700),
    )?;
    std::fs::write(
        path.join("oci-layout"),
        b"{\"imageLayoutVersion\":\"1.0.0\"}\n",
    )?;
    Ok(())
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct LayoutIndex<'a> {
    schema_version: u8,
    manifests: [LayoutDescriptor<'a>; 1],
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct LayoutDescriptor<'a> {
    media_type: &'a str,
    digest: &'a str,
    size: usize,
}

fn layout_index(digest: &str, size: usize) -> Result<Vec<u8>> {
    let index = LayoutIndex {
        schema_version: 2,
        manifests: [LayoutDescriptor {
            media_type: "application/vnd.oci.image.manifest.v1+json",
            digest,
            size,
        }],
    };
    Ok(serde_json::to_vec(&index)?)
}

fn digest_component(digest: &str) -> Result<&str> {
    let component = digest
        .strip_prefix("sha256:")
        .context("only sha256 OCI descriptors are supported")?;
    if component.len() != 64 || !component.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        bail!("invalid sha256 OCI descriptor");
    }
    Ok(component)
}

fn parse_sha256(digest: &str) -> Result<[u8; 32]> {
    let component = digest_component(digest)?;
    let mut bytes = [0; 32];
    for (index, pair) in component.as_bytes().chunks_exact(2).enumerate() {
        bytes[index] = u8::from_str_radix(std::str::from_utf8(pair)?, 16)?;
    }
    Ok(bytes)
}

fn hex(bytes: &[u8]) -> String {
    use std::fmt::Write as _;

    bytes.iter().fold(
        String::with_capacity(bytes.len() * 2),
        |mut output, byte| {
            write!(output, "{byte:02x}").expect("writing to String cannot fail");
            output
        },
    )
}

#[cfg(test)]
mod tests {
    use std::io::{Read, Write};
    use std::net::TcpListener;

    use oci_client::client::ClientProtocol;

    use super::*;

    #[test]
    fn refuses_mismatched_and_incomplete_platforms() {
        assert!(
            verify_platform(
                r#"{"os":"linux","architecture":"amd64"}"#,
                &Platform::linux("amd64")
            )
            .is_ok()
        );
        assert!(
            verify_platform(
                r#"{"os":"linux","architecture":"arm64"}"#,
                &Platform::linux("amd64")
            )
            .is_err()
        );
        assert!(
            verify_platform(
                r#"{"os":"linux","architecture":"amd64","variant":"v2"}"#,
                &Platform::linux("amd64")
            )
            .is_err()
        );
        assert!(verify_platform(r#"{"architecture":"amd64"}"#, &Platform::linux("amd64")).is_err());
    }

    #[test]
    fn incomplete_authentication_refuses_without_layout_or_secret_disclosure() {
        let runtime = tokio::runtime::Builder::new_current_thread()
            .build()
            .unwrap();
        let temp = tempfile::tempdir().unwrap();
        let result = runtime.block_on(OciRegistryTransport.pull_with_config(
            "registry.invalid/private/image:latest",
            &RegistryCredentials::Basic {
                username: "developer".to_owned(),
                password: String::new(),
            },
            &Platform::linux("amd64"),
            &temp.path().join("layout"),
            ClientConfig::default(),
        ));
        let error = result.unwrap_err().to_string();
        assert!(error.contains("credentials are incomplete"));
        assert!(!error.contains("developer"));
        assert!(!temp.path().join("layout").exists());
        assert_eq!(
            format!(
                "{:?}",
                RegistryCredentials::Bearer("not-a-real-token".to_owned())
            ),
            "Bearer(\"[redacted]\")"
        );
    }

    #[test]
    fn registry_authentication_failure_refuses_before_layout_creation() {
        let listener = match TcpListener::bind("127.0.0.1:0") {
            Ok(listener) => listener,
            Err(error) if error.kind() == std::io::ErrorKind::PermissionDenied => return,
            Err(error) => panic!("binding test registry: {error}"),
        };
        let address = listener.local_addr().unwrap();
        let server = std::thread::spawn(move || {
            let (mut stream, _) = listener.accept().unwrap();
            let mut request = [0; 4096];
            assert!(stream.read(&mut request).unwrap() > 0);
            stream
                .write_all(
                    b"HTTP/1.1 401 Unauthorized\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
                )
                .unwrap();
        });
        let runtime = tokio::runtime::Builder::new_current_thread()
            .enable_all()
            .build()
            .unwrap();
        let temp = tempfile::tempdir().unwrap();
        let result = runtime.block_on(OciRegistryTransport.pull_with_config(
            &format!("{address}/private/image:latest"),
            &RegistryCredentials::Basic {
                username: "wrong".to_owned(),
                password: "wrong".to_owned(),
            },
            &Platform::linux("amd64"),
            &temp.path().join("layout"),
            ClientConfig {
                protocol: ClientProtocol::Http,
                ..ClientConfig::default()
            },
        ));
        assert!(result.is_err());
        assert!(!temp.path().join("layout").exists());
        server.join().unwrap();
    }
}