forge
ExploreLog in

Tahinli/forge

public
0 watching
★ 0
15.7 KiB
1//! Resolving an image reference, pulling its manifest and layers, and unpacking them.
2
3use std::fs::{self, Permissions};
4use std::io::{self, Read};
5use std::os::unix::fs::PermissionsExt;
6use std::path::PathBuf;
7
8use oci_client::Reference;
9use oci_client::client::{Client, ClientConfig};
10use oci_client::manifest::OciDescriptor;
11use oci_client::secrets::RegistryAuth as OciAuth;
12use serde::Deserialize;
13use sha2::{Digest, Sha256};
14use tokio::io::AsyncWriteExt;
15use tokio::sync::mpsc;
16
17use super::helper::{UnpackRequest, run_unpack_helper};
18use super::store::{ImageMeta, ImageStore, lock_file};
19use super::unpack::{Ownership, UnpackStats, unpack_layers};
20use super::{ImageError, PreparedImage, RegistryAuth};
21
22impl ImageStore {
23 /// Makes `image` available as an unpacked rootfs and returns it with its config.
24 ///
25 /// The manifest is always fetched, so the digest is current. When the digest already has an
26 /// unpacked rootfs, nothing else is downloaded. Missing layer blobs are downloaded and checked
27 /// against their digest. When the registry cannot be reached and the tag was pulled before,
28 /// the cached image is used and a warning is logged. Progress lines go to `log`.
29 pub async fn ensure(
30 &self,
31 image: &str,
32 auth: Option<&RegistryAuth>,
33 log: &mpsc::Sender<Vec<u8>>,
34 ) -> Result<PreparedImage, ImageError> {
35 let reference = parse_reference(image)?;
36 let whole = reference.whole();
37 log_line(log, format!("pulling {whole}")).await;
38
39 let oci_auth = match auth {
40 Some(auth) => OciAuth::Basic(auth.username.clone(), auth.password.clone()),
41 None => OciAuth::Anonymous,
42 };
43 let client = Client::new(ClientConfig::default());
44
45 let (manifest, digest) = match client.pull_image_manifest(&reference, &oci_auth).await {
46 Ok(found) => found,
47 Err(source) => {
48 if let Some(prepared) = self.cached_for_tag(&whole)? {
49 log_line(
50 log,
51 format!(
52 "warning: registry unavailable ({source}), using cached image {}",
53 prepared.digest
54 ),
55 )
56 .await;
57 self.touch(hex_of(&prepared.digest)?)?;
58 return Ok(prepared);
59 }
60 return Err(ImageError::Registry {
61 image: whole,
62 source,
63 });
64 }
65 };
66
67 let hex = hex_of(&digest)?.to_owned();
68 let lock_path = self.lock_path(&hex);
69 let _lock = tokio::task::spawn_blocking(move || lock_file(&lock_path)).await??;
70
71 if let Some(prepared) = self.cached(&hex)? {
72 self.write_tag(&whole, &prepared.digest)?;
73 self.touch(&hex)?;
74 return Ok(prepared);
75 }
76
77 let mut config_bytes = Vec::new();
78 client
79 .pull_blob(&reference, &manifest.config, &mut config_bytes)
80 .await
81 .map_err(|source| ImageError::Registry {
82 image: whole.clone(),
83 source,
84 })?;
85 verify_bytes(&config_bytes, &manifest.config.digest)?;
86 let config: ConfigFile = serde_json::from_slice(&config_bytes)?;
87 let container = config.config.unwrap_or_default();
88
89 let mut layer_hexes = Vec::with_capacity(manifest.layers.len());
90 for layer in &manifest.layers {
91 let layer_hex = hex_of(&layer.digest)?.to_owned();
92 if !self.blob_path(&layer_hex).is_file() {
93 log_line(log, format!("fetching layer {}", layer.digest)).await;
94 self.download_blob(&client, &reference, layer, &layer_hex)
95 .await?;
96 }
97 layer_hexes.push(layer_hex);
98 }
99
100 log_line(log, format!("unpacking {} layers", layer_hexes.len())).await;
101 let rootfs_parent = self.root_dir().join("rootfs");
102 fs::create_dir_all(&rootfs_parent)?;
103 let staging = tempfile::Builder::new()
104 .prefix(".unpack-")
105 .tempdir_in(&rootfs_parent)?;
106 let staging_path = staging.path().to_path_buf();
107 let layer_paths: Vec<PathBuf> = layer_hexes.iter().map(|h| self.blob_path(h)).collect();
108 let max_bytes = self.max_unpacked_bytes();
109 let stats = match self
110 .unpack_into(staging_path.clone(), layer_paths, max_bytes)
111 .await
112 {
113 Ok(stats) => stats,
114 Err(err) => {
115 // Subordinate-owned files stay behind if the runner removes them directly.
116 let _ = self.remove_entry(&staging_path);
117 return Err(err);
118 }
119 };
120 if stats.rejected > 0 {
121 log_line(
122 log,
123 format!(
124 "warning: {} layer entries were outside the rootfs rules and were skipped",
125 stats.rejected
126 ),
127 )
128 .await;
129 }
130 if stats.skipped > 0 {
131 log_line(
132 log,
133 format!("skipped {} device or special entries", stats.skipped),
134 )
135 .await;
136 }
137
138 // The rootfs must be readable by the container user, whatever the tempdir default is.
139 fs::set_permissions(staging.path(), Permissions::from_mode(0o755))?;
140 let unpacked = staging.keep();
141 let final_dir = self.rootfs_path(&hex);
142 if let Err(err) = fs::rename(&unpacked, &final_dir) {
143 let _ = self.remove_entry(&unpacked);
144 return Err(err.into());
145 }
146
147 let meta = ImageMeta {
148 digest: digest.clone(),
149 env: container.env.unwrap_or_default(),
150 working_dir: container.working_dir.filter(|dir| !dir.is_empty()),
151 user: container.user.filter(|user| !user.is_empty()),
152 layers: layer_hexes,
153 };
154 self.write_meta(&hex, &meta)?;
155 self.write_tag(&whole, &digest)?;
156 self.touch(&hex)?;
157 Ok(self.prepared(&hex, meta))
158 }
159
160 /// Unpacks `layers` into the staging dir `staging`. A subuid map unpacks in the helper process,
161 /// so the files get subordinate owners. A single map unpacks in this process as before.
162 async fn unpack_into(
163 &self,
164 staging: PathBuf,
165 layers: Vec<PathBuf>,
166 max_bytes: u64,
167 ) -> Result<UnpackStats, ImageError> {
168 let map = self.idmap.clone();
169 let helper_exe = self.helper_exe.clone();
170 let stats = tokio::task::spawn_blocking(move || {
171 if map.is_subuid() {
172 let req = UnpackRequest {
173 map,
174 rootfs: staging,
175 layers,
176 max_bytes,
177 };
178 run_unpack_helper(&helper_exe, &req)
179 } else {
180 unpack_layers(&staging, &layers, max_bytes, Ownership::Flatten)
181 }
182 })
183 .await??;
184 Ok(stats)
185 }
186
187 /// Downloads one layer blob into a temp file, checks its digest, and renames it into place.
188 async fn download_blob(
189 &self,
190 client: &Client,
191 reference: &Reference,
192 layer: &OciDescriptor,
193 hex: &str,
194 ) -> Result<(), ImageError> {
195 let blob_dir = self.root_dir().join("blobs/sha256");
196 fs::create_dir_all(&blob_dir)?;
197 let tmp = tempfile::NamedTempFile::new_in(&blob_dir)?;
198 {
199 let mut out = tokio::fs::File::from_std(tmp.reopen()?);
200 client
201 .pull_blob(reference, layer, &mut out)
202 .await
203 .map_err(|source| ImageError::Registry {
204 image: reference.whole(),
205 source,
206 })?;
207 out.flush().await?;
208 }
209 let actual = sha256_file(tmp.path().to_path_buf()).await?;
210 if actual != hex {
211 return Err(ImageError::DigestMismatch {
212 expected: format!("sha256:{hex}"),
213 actual: format!("sha256:{actual}"),
214 });
215 }
216 tmp.persist(self.blob_path(hex)).map_err(|err| err.error)?;
217 Ok(())
218 }
219}
220
221/// The parts of an image config that the runner uses.
222#[derive(Deserialize)]
223struct ConfigFile {
224 #[serde(default)]
225 config: Option<ContainerConfig>,
226}
227
228#[derive(Default, Deserialize)]
229struct ContainerConfig {
230 #[serde(rename = "Env", default)]
231 env: Option<Vec<String>>,
232 #[serde(rename = "WorkingDir", default)]
233 working_dir: Option<String>,
234 #[serde(rename = "User", default)]
235 user: Option<String>,
236}
237
238/// Parses an image reference, applying the Docker Hub defaults: `alpine` is
239/// `docker.io/library/alpine:latest`.
240pub(super) fn parse_reference(image: &str) -> Result<Reference, ImageError> {
241 image
242 .parse::<Reference>()
243 .map_err(|err| ImageError::InvalidReference {
244 image: image.to_owned(),
245 reason: err.to_string(),
246 })
247}
248
249/// Returns the hex part of a `sha256:<hex>` digest. The digest comes from a registry and becomes
250/// a file name, so anything other than 64 lowercase hex digits is refused.
251pub(super) fn hex_of(digest: &str) -> Result<&str, ImageError> {
252 let hex = digest
253 .strip_prefix("sha256:")
254 .ok_or_else(|| ImageError::UnsupportedDigest(digest.to_owned()))?;
255 let valid = hex.len() == 64
256 && hex
257 .bytes()
258 .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b));
259 if valid {
260 Ok(hex)
261 } else {
262 Err(ImageError::UnsupportedDigest(digest.to_owned()))
263 }
264}
265
266fn verify_bytes(bytes: &[u8], digest: &str) -> Result<(), ImageError> {
267 let expected = hex_of(digest)?;
268 let actual = hex::encode(Sha256::digest(bytes));
269 if actual == expected {
270 Ok(())
271 } else {
272 Err(ImageError::DigestMismatch {
273 expected: digest.to_owned(),
274 actual: format!("sha256:{actual}"),
275 })
276 }
277}
278
279/// Hashes a file on a blocking thread and returns the hex digest.
280async fn sha256_file(path: PathBuf) -> Result<String, ImageError> {
281 let hash = tokio::task::spawn_blocking(move || -> io::Result<String> {
282 let mut file = fs::File::open(&path)?;
283 let mut hasher = Sha256::new();
284 let mut buf = vec![0u8; 64 * 1024];
285 loop {
286 let n = file.read(&mut buf)?;
287 if n == 0 {
288 break;
289 }
290 hasher.update(&buf[..n]);
291 }
292 Ok(hex::encode(hasher.finalize()))
293 })
294 .await??;
295 Ok(hash)
296}
297
298/// Sends one line to the job log. A closed log channel is not an error for the pull.
299async fn log_line(log: &mpsc::Sender<Vec<u8>>, line: String) {
300 let mut bytes = line.into_bytes();
301 bytes.push(b'\n');
302 let _ = log.send(bytes).await;
303}
304
305#[cfg(test)]
306mod tests {
307 use super::*;
308
309 const DIGEST: &str = "sha256:0000000000000000000000000000000000000000000000000000000000000001";
310
311 #[test]
312 fn short_names_normalize_to_docker_hub() {
313 assert_eq!(
314 parse_reference("alpine").expect("parse").whole(),
315 "docker.io/library/alpine:latest"
316 );
317 assert_eq!(
318 parse_reference("rust:1").expect("parse").whole(),
319 "docker.io/library/rust:1"
320 );
321 assert_eq!(
322 parse_reference("ghcr.io/o/i:2").expect("parse").whole(),
323 "ghcr.io/o/i:2"
324 );
325 }
326
327 #[test]
328 fn digest_references_keep_their_digest() {
329 let image = format!("ghcr.io/o/i@{DIGEST}");
330 let reference = parse_reference(&image).expect("parse");
331 assert_eq!(reference.whole(), image);
332 assert_eq!(reference.digest(), Some(DIGEST));
333 }
334
335 #[test]
336 fn invalid_references_are_errors() {
337 assert!(parse_reference("Bad Name!!").is_err());
338 }
339
340 #[test]
341 fn only_clean_sha256_digests_are_accepted() {
342 assert_eq!(hex_of(DIGEST).expect("ok").len(), 64);
343 assert!(hex_of("sha512:00").is_err());
344 assert!(hex_of("sha256:../../etc/passwd").is_err());
345 assert!(
346 hex_of("sha256:ABCDEF0000000000000000000000000000000000000000000000000000000000")
347 .is_err()
348 );
349 assert!(hex_of("sha256:abc").is_err());
350 }
351
352 #[test]
353 fn verify_bytes_checks_the_digest() {
354 let digest = format!("sha256:{}", hex::encode(Sha256::digest(b"layer")));
355 verify_bytes(b"layer", &digest).expect("matches");
356 assert!(matches!(
357 verify_bytes(b"other", &digest),
358 Err(ImageError::DigestMismatch { .. })
359 ));
360 }
361
362 #[tokio::test]
363 async fn unreachable_registry_uses_the_cached_image() {
364 let dir = tempfile::tempdir().expect("tempdir");
365 let store = ImageStore::new(dir.path().join("oci"));
366 let reference = "127.0.0.1:1/forge-test/cached:1";
367 let hex = "1".repeat(64);
368 let digest = format!("sha256:{hex}");
369
370 fs::create_dir_all(store.rootfs_path(&hex)).expect("rootfs");
371 fs::write(store.rootfs_path(&hex).join("marker"), b"x").expect("marker");
372 store
373 .write_meta(
374 &hex,
375 &ImageMeta {
376 digest: digest.clone(),
377 env: vec!["PATH=/bin".to_owned()],
378 working_dir: Some("/work".to_owned()),
379 user: None,
380 layers: Vec::new(),
381 },
382 )
383 .expect("meta");
384 store.write_tag(reference, &digest).expect("tag");
385
386 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(64);
387 let drain = tokio::spawn(async move {
388 let mut lines = Vec::new();
389 while let Some(line) = rx.recv().await {
390 lines.push(String::from_utf8_lossy(&line).into_owned());
391 }
392 lines
393 });
394 let prepared = store.ensure(reference, None, &tx).await.expect("cached");
395 drop(tx);
396 let lines = drain.await.expect("drain");
397
398 assert_eq!(prepared.digest, digest);
399 assert_eq!(prepared.env, vec!["PATH=/bin".to_owned()]);
400 assert_eq!(prepared.working_dir.as_deref(), Some("/work"));
401 assert!(prepared.rootfs.join("marker").is_file());
402 assert!(lines.iter().any(|l| l.contains("using cached image")));
403 }
404
405 #[tokio::test]
406 async fn unreachable_registry_without_cache_is_an_error() {
407 let dir = tempfile::tempdir().expect("tempdir");
408 let store = ImageStore::new(dir.path().join("oci"));
409 let (tx, _rx) = mpsc::channel::<Vec<u8>>(64);
410 let result = store
411 .ensure("127.0.0.1:1/forge-test/none:1", None, &tx)
412 .await;
413 assert!(matches!(result, Err(ImageError::Registry { .. })));
414 }
415
416 /// Pulls the real alpine image. Needs network access to Docker Hub.
417 /// Run with: cargo test -p forge-runner oci -- --ignored
418 #[tokio::test]
419 #[ignore]
420 async fn pulls_alpine_for_real() {
421 let dir = tempfile::tempdir().expect("tempdir");
422 let store = ImageStore::new(dir.path().join("oci"));
423 let (tx, mut rx) = mpsc::channel::<Vec<u8>>(256);
424 let drain = tokio::spawn(async move { while rx.recv().await.is_some() {} });
425
426 let first = store
427 .ensure("docker.io/library/alpine:latest", None, &tx)
428 .await
429 .expect("pull alpine");
430 assert!(first.digest.starts_with("sha256:"));
431 assert!(first.rootfs.join("bin/sh").symlink_metadata().is_ok());
432 assert!(first.env.iter().any(|e| e.starts_with("PATH=")));
433
434 // A second ensure with the same digest reuses the unpacked rootfs.
435 let second = store
436 .ensure("alpine", None, &tx)
437 .await
438 .expect("reuse alpine");
439 assert_eq!(second.digest, first.digest);
440 assert_eq!(second.rootfs, first.rootfs);
441
442 drop(tx);
443 let _ = drain.await;
444 }
445}