ocilot 0.1.0

cli and library for interacting with OCI registries
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348

use std::str::FromStr;

use crate::error;
use crate::image::Image;
use crate::layer::Layer;
use crate::models::MediaType;
use crate::models::Platform;
use crate::uri::{Reference, Uri};
use bon::Builder;
use futures::future::join_all;
#[cfg(feature = "progress")]
use indicatif::MultiProgress;
use serde::{Deserialize, Serialize};
use snafu::{OptionExt, ResultExt};
use tempfile::tempdir;
use tokio::fs::{File, create_dir_all};
use tokio::io::AsyncWrite;
use tokio::task::JoinHandle;
use tokio_tar::Builder as ArchiveBuilder;

/// Represents an Image Index and handles all operations that require or utilize one.
#[derive(Debug, Serialize, Deserialize, Clone, Builder)]
#[serde(rename_all = "camelCase")]
pub struct Index {
    #[builder(into)]
    schema_version: usize,
    #[builder(into)]
    media_type: MediaType,
    #[builder(into)]
    manifests: Vec<Layer>,
}

impl Index {
    /// Create a new image index with the provided manifests
    pub async fn new(manifests: &[Layer]) -> Self {
        Self {
            schema_version: 2,
            media_type: MediaType::ImageIndex,
            manifests: manifests.to_vec(),
        }
    }

    /// Check if there is an image index at the provided URI.
    ///
    /// Note: This only checks that a manifest exists so it could return a false
    /// positive as it does not verify the media type of the manifest to ensure
    /// it is an index.
    pub async fn check(uri: &Uri) -> crate::Result<bool> {
        uri.registry()
            .check_manifest(uri.repository(), uri.reference().to_string().as_str())
            .await
    }

    /// Fetch an image index from a registry
    pub async fn fetch(uri: &Uri) -> crate::Result<Self> {
        uri.registry()
            .fetch_manifest(uri.repository(), uri.reference().to_string().as_str())
            .await
    }

    /// Schema version
    pub fn schema_version(&self) -> usize {
        self.schema_version
    }

    /// Media type
    pub fn media_type(&self) -> &MediaType {
        &self.media_type
    }

    /// Manifest layers included in this index
    pub fn manifests(&self) -> &[Layer] {
        self.manifests.as_slice()
    }

    /// Fetch an image from this index.
    ///
    /// If a platform is provided, looks for the first matching image. If not
    /// provided, loads an image matching the current running environment or
    /// the first image in the index.
    pub async fn fetch_image(
        &self,
        uri: &Uri,
        platform: Option<Platform>,
    ) -> crate::Result<Option<Image>> {
        if let Some(platform) = platform {
            let oci = self
                .manifests
                .iter()
                .find(|x| x.platform() == Some(platform.clone()))
                .context(error::IndexNoPlatformSnafu {
                    platform: platform.clone(),
                })?;
            // Use the digest
            let new_uri = Uri::builder()
                .registry(uri.registry().clone())
                .repository(uri.repository())
                .reference(Reference::from_str(oci.digest())?)
                .build();
            Ok(Some(Image::fetch(&new_uri, Some(platform)).await?))
        } else {
            // See if we can match by architecture
            let current = Platform::default();
            if let Some(oci) = self
                .manifests
                .iter()
                .find(|x| x.platform() == Some(current.clone()))
            {
                // Use the digest
                let new_uri = Uri::builder()
                    .registry(uri.registry().clone())
                    .repository(uri.repository())
                    .reference(Reference::from_str(oci.digest())?)
                    .build();
                return Ok(Some(Image::fetch(&new_uri, Some(current.clone())).await?));
            }
            // Otherwise we return the first image
            if let Some(oci) = self.manifests.first() {
                // Use the digest
                let new_uri = Uri::builder()
                    .registry(uri.registry().clone())
                    .repository(uri.repository())
                    .reference(Reference::from_str(oci.digest())?)
                    .build();
                Ok(Some(Image::fetch(&new_uri, oci.platform().clone()).await?))
            } else {
                Ok(None)
            }
        }
    }

    /// Push this image index to a registry
    pub async fn push(&self, uri: &Uri) -> crate::Result<()> {
        uri.registry()
            .push_manifest(
                &self.media_type,
                uri.repository(),
                uri.reference().to_string().as_str(),
                self,
                None,
            )
            .await?;
        Ok(())
    }

    /// Create an OCI tar archive that contains either all of the index images (if no platform provided)
    /// or only the platforms specified
    pub async fn to_oci<W>(
        &self,
        uri: &Uri,
        platform: Option<Platform>,
        output: W,
    ) -> crate::Result<()>
    where
        W: AsyncWrite + Unpin + Send + 'static,
    {
        let tmp_dir = tempdir().context(error::TempSnafu)?;
        tokio::fs::write(
            tmp_dir.path().join("oci-layout"),
            r#"{ "imageLayoutVersion": "1.0.0" }"#,
        )
        .await
        .context(error::FileSnafu)?;

        let blob_dir = tmp_dir.path().join("blobs/sha256");
        create_dir_all(&blob_dir)
            .await
            .context(error::DirectorySnafu)?;

        // Start with ourselves for the index
        let mut index = self.clone();
        if let Some(platform) = platform {
            // If we are selecting only a single platform then filter the manifests down
            index.manifests = index
                .manifests
                .iter()
                .filter(|x| x.platform() == Some(platform.clone()))
                .cloned()
                .collect::<Vec<Layer>>();
            if index.manifests.is_empty() {
                return error::IndexNoPlatformSnafu { platform }.fail();
            }
        }
        let index_content = serde_json::to_string(&index).context(error::SerializeSnafu)?;
        tokio::fs::write(tmp_dir.path().join("index.json"), &index_content)
            .await
            .context(error::FileSnafu)?;

        // Now for every manifest we are working with we need to store it out
        for manifest in index.manifests.iter() {
            let image_uri = Uri::builder()
                .registry(uri.registry().clone())
                .repository(uri.repository())
                .reference(Reference::from_str(manifest.digest())?)
                .build();
            let image = Image::fetch(&image_uri, manifest.platform().clone()).await?;
            // Write the image manifest as a blob
            let manifest_bytes = serde_json::to_string(&image).context(error::SerializeSnafu)?;
            tokio::fs::write(
                blob_dir.join(manifest.digest().strip_prefix("sha256:").unwrap()),
                &manifest_bytes,
            )
            .await
            .context(error::FileSnafu)?;
            // Copy the image config
            let mut config_reader = image.config().open(uri).await?;
            let mut config_file = File::create(
                blob_dir.join(image.config().digest().strip_prefix("sha256:").unwrap()),
            )
            .await
            .context(error::FileSnafu)?;
            Layer::copy(&mut config_reader, &mut config_file, image.config().size()).await?;

            let mut tasks: Vec<JoinHandle<crate::Result<()>>> = Vec::new();
            for layer in image.layers().iter() {
                let layer = layer.clone();
                let uri = uri.clone();
                let blob_dir = blob_dir.clone();
                tasks.push(tokio::spawn(async move {
                    let mut reader = layer.open(&uri).await?;
                    let mut blob_file = File::create(
                        blob_dir.join(layer.digest().strip_prefix("sha256:").unwrap()),
                    )
                    .await
                    .context(error::FileSnafu)?;
                    Layer::copy(&mut reader, &mut blob_file, layer.size()).await?;
                    Ok(())
                }));
            }
            for result in join_all(tasks).await {
                let result = result.context(error::LayerWaitSnafu)?;
                result?;
            }
        }

        let mut archive = ArchiveBuilder::new(output);
        archive
            .append_dir_all(".", tmp_dir.path().to_path_buf())
            .await
            .context(error::ArchiveSnafu)?;
        archive.finish().await.context(error::ArchiveSnafu)?;

        Ok(())
    }

    /// Create an OCI tar archive that contains either all of the index images (if no platform provided)
    /// or only the platforms specified
    #[cfg(feature = "progress")]
    pub async fn to_oci_progress<W>(
        &self,
        uri: &Uri,
        platform: Option<Platform>,
        output: W,
        multi: &mut MultiProgress,
    ) -> crate::Result<()>
    where
        W: AsyncWrite + Unpin + Send + 'static,
    {
        let tmp_dir = tempdir().context(error::TempSnafu)?;
        tokio::fs::write(
            tmp_dir.path().join("oci-layout"),
            r#"{ "imageLayoutVersion": "1.0.0" }"#,
        )
        .await
        .context(error::FileSnafu)?;

        let blob_dir = tmp_dir.path().join("blobs/sha256");
        create_dir_all(&blob_dir)
            .await
            .context(error::DirectorySnafu)?;

        // Start with ourselves for the index
        let mut index = self.clone();
        if let Some(platform) = platform {
            // If we are selecting only a single platform then filter the manifests down
            index.manifests = index
                .manifests
                .iter()
                .filter(|x| x.platform() == Some(platform.clone()))
                .cloned()
                .collect::<Vec<Layer>>();
            if index.manifests.is_empty() {
                return error::IndexNoPlatformSnafu { platform }.fail();
            }
        }
        let index_content = serde_json::to_string(&index).context(error::SerializeSnafu)?;
        tokio::fs::write(tmp_dir.path().join("index.json"), &index_content)
            .await
            .context(error::FileSnafu)?;

        // Now for every manifest we are working with we need to store it out
        for manifest in index.manifests.iter() {
            let image_uri = Uri::builder()
                .registry(uri.registry().clone())
                .repository(uri.repository())
                .reference(Reference::from_str(manifest.digest())?)
                .build();
            let image = Image::fetch(&image_uri, manifest.platform().clone()).await?;
            // Write the image manifest as a blob
            let manifest_bytes = serde_json::to_string(&image).context(error::SerializeSnafu)?;
            tokio::fs::write(
                blob_dir.join(manifest.digest().strip_prefix("sha256:").unwrap()),
                &manifest_bytes,
            )
            .await
            .context(error::FileSnafu)?;
            // Copy the image config
            let mut config_reader = image.config().open_progress(uri, multi).await?;
            let mut config_file = File::create(
                blob_dir.join(image.config().digest().strip_prefix("sha256:").unwrap()),
            )
            .await
            .context(error::FileSnafu)?;
            Layer::copy(&mut config_reader, &mut config_file, image.config().size()).await?;

            let mut tasks: Vec<JoinHandle<crate::Result<()>>> = Vec::new();
            for layer in image.layers().iter() {
                let layer = layer.clone();
                let uri = uri.clone();
                let mut multi = multi.clone();
                let blob_dir = blob_dir.clone();
                tasks.push(tokio::spawn(async move {
                    let mut reader = layer.open_progress(&uri, &mut multi).await?;
                    let mut blob_file = File::create(
                        blob_dir.join(layer.digest().strip_prefix("sha256:").unwrap()),
                    )
                    .await
                    .context(error::FileSnafu)?;
                    Layer::copy(&mut reader, &mut blob_file, layer.size()).await?;
                    Ok(())
                }));
            }
            for result in join_all(tasks).await {
                let result = result.context(error::LayerWaitSnafu)?;
                result?;
            }
        }

        let mut archive = ArchiveBuilder::new(output);
        archive
            .append_dir_all(".", tmp_dir.path().to_path_buf())
            .await
            .context(error::ArchiveSnafu)?;
        archive.finish().await.context(error::ArchiveSnafu)?;

        Ok(())
    }
}