snix_store/nar/
import.rs

1use nix_compat::{
2    nar::reader::r#async as nar_reader,
3    nixhash::{CAHash, HashAlgo, NixHash, NixHashDigester, Sha256Digester, copy_hashed},
4};
5use snix_castore::{
6    Node, PathBuf,
7    blob_engine::{self, BlobEngine, concurrent_uploads},
8    directoryservice::DirectoryService,
9    import::{IngestionEntry, IngestionError, ingest_entries},
10};
11use tokio::{
12    io::{AsyncBufRead, AsyncRead},
13    sync::mpsc,
14    try_join,
15};
16use tokio_util::io::InspectReader;
17
18/// Represents errors that can happen during nar ingestion.
19#[derive(Debug, thiserror::Error)]
20pub enum NarIngestionError {
21    #[error("{0}")]
22    IngestionError(#[from] IngestionError<Error>),
23
24    #[error("Hash mismatch, expected: {expected}, got: {actual}.")]
25    HashMismatch { expected: NixHash, actual: NixHash },
26
27    #[error("Expected the nar to contain a single file.")]
28    TypeMismatch,
29
30    #[error("Ingestion failed: {0}")]
31    Io(#[from] std::io::Error),
32
33    #[error("Calling open_read() on BlobEngine failed: {0}")]
34    BlobEngine(#[from] blob_engine::Error),
35}
36
37/// Ingests the contents from a [AsyncRead] providing NAR into the snix store,
38/// interacting with a [BlobEngine] and [DirectoryService].
39/// Returns the castore root node, as well as the sha256 and size of the NAR
40/// contents ingested.
41pub async fn ingest_nar_and_hash<R, BE, DS>(
42    blob_engine: BE,
43    directory_service: DS,
44    r: &mut R,
45    expected_cahash: &Option<CAHash>,
46) -> Result<(Node, [u8; 32], u64), NarIngestionError>
47where
48    R: AsyncRead + Unpin + Send,
49    BE: BlobEngine + Clone + 'static,
50    DS: DirectoryService,
51{
52    let mut nar_hash = Sha256Digester::new();
53    let mut nar_size = 0;
54
55    // Assemble NarHash and NarSize as we read bytes.
56    let mut r = tokio_util::io::InspectReader::new(r, |b| {
57        nar_size += b.len() as u64;
58        nar_hash.update(b);
59    });
60
61    match expected_cahash {
62        Some(CAHash::Nar(expected_hash)) => {
63            let (root_node, actual_hash, nar_hash) = if expected_hash.algo() == HashAlgo::Sha256 {
64                // If this is the required CAHash, we're already computing excatly this in `nar_hash` above.
65                let mut r = tokio::io::BufReader::new(&mut r);
66
67                let root_node = ingest_nar(blob_engine, directory_service, &mut r).await?;
68                let nar_hash = nar_hash.finalize();
69                (root_node, NixHash::from(nar_hash), nar_hash)
70            } else {
71                // For the other algos, wrap the reader with another digester.
72                let mut digester = NixHashDigester::new(expected_hash.algo());
73                let mut r =
74                    tokio::io::BufReader::new(InspectReader::new(r, |data| digester.update(data)));
75
76                let root_node = ingest_nar(blob_engine, directory_service, &mut r).await?;
77                (root_node, digester.finalize(), nar_hash.finalize())
78            };
79
80            if actual_hash != *expected_hash {
81                return Err(NarIngestionError::HashMismatch {
82                    expected: expected_hash.clone(),
83                    actual: actual_hash,
84                });
85            }
86            Ok((root_node, nar_hash.into(), nar_size))
87        }
88        Some(CAHash::Flat(expected_hash)) => {
89            // ingest as NAR
90            let mut r = tokio::io::BufReader::new(&mut r);
91            let root_node = ingest_nar(blob_engine.clone(), directory_service, &mut r).await?;
92
93            // The resulting root node must be Node::File, else CAHash::Flat is not applicable
94            if let Node::File { digest, size, .. } = &root_node {
95                if let Some(mut blob_reader) = blob_engine.open_read(digest, Some(*size)).await? {
96                    let (_, actual_hash) = copy_hashed(
97                        &mut blob_reader,
98                        &mut tokio::io::sink(),
99                        expected_hash.algo(),
100                    )
101                    .await?;
102
103                    if actual_hash != *expected_hash {
104                        return Err(NarIngestionError::HashMismatch {
105                            expected: expected_hash.clone(),
106                            actual: actual_hash,
107                        });
108                    }
109                    Ok((root_node, nar_hash.finalize().into(), nar_size))
110                } else {
111                    Err(NarIngestionError::Io(std::io::Error::other(
112                        "Ingested data not found",
113                    )))
114                }
115            } else {
116                Err(NarIngestionError::TypeMismatch)
117            }
118        }
119        // We either got CAHash::Text, or no CAHash at all, so we just don't do any additional
120        // hash calculation/validation.
121        // FUTUREWORK: We should figure out what to do with CAHash::Text, according to nix-cpp
122        // they don't handle it either:
123        // https://github.com/NixOS/nix/blob/3e9cc78eb5e5c4f1e762e201856273809fd92e71/src/libstore/local-store.cc#L1099-L1133
124        _ => {
125            let mut r = tokio::io::BufReader::new(&mut r);
126            let root_node = ingest_nar(blob_engine, directory_service, &mut r).await?;
127            Ok((root_node, nar_hash.finalize().into(), nar_size))
128        }
129    }
130}
131
132/// Ingests the contents from a [AsyncRead] providing NAR into the snix store,
133/// interacting with a [BlobEngine] and [DirectoryService].
134/// It returns the castore root node or an error.
135pub async fn ingest_nar<R, BE, DS>(
136    blob_engine: BE,
137    directory_service: DS,
138    r: &mut R,
139) -> Result<Node, IngestionError<Error>>
140where
141    R: AsyncBufRead + Unpin + Send,
142    BE: BlobEngine + Clone + 'static,
143    DS: DirectoryService,
144{
145    // open the NAR for reading.
146    // The NAR reader emits nodes in DFS preorder.
147    let root_node = nar_reader::open(r).await.map_err(Error::IO)?;
148
149    let (tx, rx) = mpsc::channel(1);
150    let rx = tokio_stream::wrappers::ReceiverStream::new(rx);
151
152    let produce = async move {
153        let mut blob_uploader = concurrent_uploads::ConcurrentBlobUploader::new(blob_engine);
154
155        let res = produce_nar_inner(
156            &mut blob_uploader,
157            root_node,
158            "root".parse().unwrap(), // HACK: the root node sent to ingest_entries may not be ROOT.
159            tx.clone(),
160        )
161        .await;
162
163        if let Err(err) = blob_uploader.join().await {
164            tx.send(Err(err.into()))
165                .await
166                .map_err(|e| Error::IO(std::io::Error::new(std::io::ErrorKind::BrokenPipe, e)))?;
167        }
168
169        tx.send(res)
170            .await
171            .map_err(|e| Error::IO(std::io::Error::new(std::io::ErrorKind::BrokenPipe, e)))?;
172
173        Ok(())
174    };
175
176    let consume = ingest_entries(directory_service, rx);
177
178    let (_, node) = try_join!(produce, consume)?;
179
180    Ok(node)
181}
182
183async fn produce_nar_inner<BE>(
184    blob_uploader: &mut concurrent_uploads::ConcurrentBlobUploader<BE>,
185    node: nar_reader::Node<'_, '_>,
186    path: PathBuf,
187    tx: mpsc::Sender<Result<IngestionEntry, Error>>,
188) -> Result<IngestionEntry, Error>
189where
190    BE: BlobEngine + Clone + 'static,
191{
192    Ok(match node {
193        nar_reader::Node::Symlink { target } => IngestionEntry::Symlink { path, target },
194        nar_reader::Node::File {
195            executable,
196            mut reader,
197        } => {
198            let size = reader.len();
199            let digest = blob_uploader.upload(&path, size, &mut reader).await?;
200
201            IngestionEntry::Regular {
202                path,
203                size,
204                executable,
205                digest,
206            }
207        }
208        nar_reader::Node::Directory(mut dir_reader) => {
209            while let Some(entry) = dir_reader.next().await? {
210                let mut path = path.clone();
211
212                // valid NAR names are valid castore names
213                path.try_push(entry.name)
214                    .expect("Snix bug: failed to join name");
215
216                let entry = Box::pin(produce_nar_inner(
217                    blob_uploader,
218                    entry.node,
219                    path,
220                    tx.clone(),
221                ))
222                .await?;
223
224                tx.send(Ok(entry)).await.map_err(|e| {
225                    Error::IO(std::io::Error::new(std::io::ErrorKind::BrokenPipe, e))
226                })?;
227            }
228
229            IngestionEntry::Dir { path }
230        }
231    })
232}
233
234#[derive(Debug, thiserror::Error)]
235pub enum Error {
236    #[error(transparent)]
237    IO(#[from] std::io::Error),
238
239    #[error(transparent)]
240    BlobUpload(#[from] concurrent_uploads::Error),
241}
242
243#[cfg(test)]
244mod test {
245    use crate::fixtures::{
246        NAR_CONTENTS_COMPLICATED, NAR_CONTENTS_HELLOWORLD, NAR_CONTENTS_SYMLINK,
247    };
248    use crate::nar::{NarIngestionError, ingest_nar, ingest_nar_and_hash};
249    use std::io::Cursor;
250    use std::sync::Arc;
251
252    use hex_literal::hex;
253    use mockall::predicate;
254    use nix_compat::nixhash::{CAHash, NixHash};
255    use rstest::*;
256    use snix_castore::blob_engine::MockBlobEngine;
257    use snix_castore::blobstore::MockWriter;
258    use snix_castore::directoryservice::{MockDirectoryPutter, MockDirectoryService};
259    use snix_castore::fixtures::{
260        DIRECTORY_COMPLICATED, DIRECTORY_WITH_KEEP, EMPTY_BLOB_DIGEST, HELLOWORLD_BLOB_CONTENTS,
261        HELLOWORLD_BLOB_DIGEST,
262    };
263    use snix_castore::utils::gen_test_blob_engine;
264    use snix_castore::{Node, blobstore};
265
266    #[tokio::test]
267    async fn single_symlink() {
268        let root_node = ingest_nar(
269            Arc::new(MockBlobEngine::new()),
270            MockDirectoryService::new(),
271            &mut Cursor::new(&NAR_CONTENTS_SYMLINK),
272        )
273        .await
274        .expect("must parse");
275
276        assert_eq!(
277            Node::Symlink {
278                target: "/nix/store/somewhereelse".try_into().unwrap()
279            },
280            root_node
281        );
282    }
283
284    #[tokio::test]
285    async fn single_file() {
286        let mut blob_engine = MockBlobEngine::new();
287        let mut seq = mockall::Sequence::new();
288        blob_engine
289            .expect_has()
290            .once()
291            .with(predicate::eq(&*HELLOWORLD_BLOB_DIGEST))
292            .return_once(|_| Ok(false))
293            .in_sequence(&mut seq);
294
295        blob_engine
296            .expect_open_write()
297            .once()
298            .return_once(|| {
299                Box::new(MockWriter::new(*HELLOWORLD_BLOB_DIGEST)) as Box<dyn blobstore::Writer>
300            })
301            .in_sequence(&mut seq);
302
303        let root_node = ingest_nar(
304            Arc::new(blob_engine),
305            MockDirectoryService::new(),
306            &mut Cursor::new(&NAR_CONTENTS_HELLOWORLD),
307        )
308        .await
309        .expect("must parse");
310
311        assert_eq!(
312            Node::File {
313                digest: *HELLOWORLD_BLOB_DIGEST,
314                size: HELLOWORLD_BLOB_CONTENTS.len() as u64,
315                executable: false,
316            },
317            root_node
318        );
319    }
320
321    #[tokio::test]
322    async fn complicated() {
323        let mut blob_engine = MockBlobEngine::new();
324        let mut seq = mockall::Sequence::new();
325        blob_engine
326            .expect_has()
327            .once()
328            .with(predicate::eq(&*EMPTY_BLOB_DIGEST))
329            .return_once(|_| Ok(false))
330            .in_sequence(&mut seq);
331        blob_engine
332            .expect_open_write()
333            .once()
334            .return_once(|| {
335                Box::new(MockWriter::new(*EMPTY_BLOB_DIGEST)) as Box<dyn blobstore::Writer>
336            })
337            .in_sequence(&mut seq);
338        blob_engine
339            .expect_has()
340            .once()
341            .with(predicate::eq(&*EMPTY_BLOB_DIGEST))
342            .return_once(|_| Ok(true))
343            .in_sequence(&mut seq);
344        let mut directory_service = MockDirectoryService::new();
345        directory_service
346            .expect_put_multiple_start()
347            .once()
348            .return_once(|| {
349                let mut directory_putter = MockDirectoryPutter::new();
350                let mut seq = mockall::Sequence::new();
351                directory_putter
352                    .expect_put()
353                    .once()
354                    .with(predicate::eq(&*DIRECTORY_WITH_KEEP))
355                    .returning(|_| Ok(()))
356                    .in_sequence(&mut seq);
357                directory_putter
358                    .expect_put()
359                    .once()
360                    .with(predicate::eq(&*DIRECTORY_COMPLICATED))
361                    .returning(|_| Ok(()))
362                    .in_sequence(&mut seq);
363                directory_putter
364                    .expect_close()
365                    .once()
366                    .returning(|| Ok(DIRECTORY_COMPLICATED.digest()))
367                    .in_sequence(&mut seq);
368                Box::new(directory_putter)
369            });
370
371        let root_node = ingest_nar(
372            Arc::new(blob_engine),
373            directory_service,
374            &mut Cursor::new(&NAR_CONTENTS_COMPLICATED),
375        )
376        .await
377        .expect("must parse");
378
379        assert_eq!(
380            Node::Directory {
381                digest: DIRECTORY_COMPLICATED.digest(),
382                size: DIRECTORY_COMPLICATED.size()
383            },
384            root_node,
385        );
386    }
387
388    #[rstest]
389    #[case::nar_sha256(Some(CAHash::Nar(NixHash::Sha256(hex!("fbd52279a8df024c9fd5718de4103bf5e760dc7f2cf49044ee7dea87ab16911a")))), NAR_CONTENTS_COMPLICATED.as_slice())]
390    #[case::nar_sha512(Some(CAHash::Nar(NixHash::Sha512(Box::new(hex!("ff5d43941411f35f09211f8596b426ee6e4dd3af1639e0ed2273cbe44b818fc4a59e3af02a057c5b18fbfcf435497de5f1994206c137f469b3df674966a922f0"))))), NAR_CONTENTS_COMPLICATED.as_slice())]
391    #[case::flat_md5(Some(CAHash::Flat(NixHash::Md5(hex!("fd076287532e86365e841e92bfc50d8c")))), NAR_CONTENTS_HELLOWORLD.as_slice() )]
392    #[case::nar_symlink_sha1(Some(CAHash::Nar(NixHash::Sha1(hex!("f24eeaaa9cc016bab030bf007cb1be6483e7ba9e")))), NAR_CONTENTS_SYMLINK.as_slice())]
393    #[tokio::test]
394    async fn ingest_with_cahash_mismatch(
395        #[case] ca_hash: Option<CAHash>,
396        #[case] nar_content: &[u8],
397    ) {
398        use snix_castore::utils::gen_test_directory_service;
399
400        let err = ingest_nar_and_hash(
401            gen_test_blob_engine(),
402            gen_test_directory_service(),
403            &mut Cursor::new(nar_content),
404            &ca_hash,
405        )
406        .await
407        .expect_err("Ingestion should have failed");
408        assert!(
409            matches!(err, NarIngestionError::HashMismatch { .. }),
410            "CAHash should have mismatched"
411        );
412    }
413
414    #[rstest]
415    #[case::nar_sha256(Some(CAHash::Nar(NixHash::Sha256(hex!("ebd52279a8df024c9fd5718de4103bf5e760dc7f2cf49044ee7dea87ab16911a")))), &NAR_CONTENTS_COMPLICATED.clone())]
416    #[case::nar_sha512(Some(CAHash::Nar(NixHash::Sha512(Box::new(hex!("1f5d43941411f35f09211f8596b426ee6e4dd3af1639e0ed2273cbe44b818fc4a59e3af02a057c5b18fbfcf435497de5f1994206c137f469b3df674966a922f0"))))), &NAR_CONTENTS_COMPLICATED.clone())]
417    #[case::flat_md5(Some(CAHash::Flat(NixHash::Md5(hex!("ed076287532e86365e841e92bfc50d8c")))), &NAR_CONTENTS_HELLOWORLD.clone())]
418    #[case::nar_symlink_sha1(Some(CAHash::Nar(NixHash::Sha1(hex!("424eeaaa9cc016bab030bf007cb1be6483e7ba9e")))), &NAR_CONTENTS_SYMLINK.clone())]
419    #[tokio::test]
420    async fn ingest_with_cahash_correct(
421        #[case] ca_hash: Option<CAHash>,
422        #[case] nar_content: &[u8],
423    ) {
424        ingest_nar_and_hash(
425            gen_test_blob_engine(),
426            snix_castore::utils::gen_test_directory_service(),
427            &mut Cursor::new(nar_content),
428            &ca_hash,
429        )
430        .await
431        .expect("CAHash should have matched");
432    }
433
434    #[rstest]
435    #[case::nar_sha256(Some(CAHash::Flat(NixHash::Sha256(hex!("ebd52279a8df024c9fd5718de4103bf5e760dc7f2cf49044ee7dea87ab16911a")))), &NAR_CONTENTS_COMPLICATED.clone())]
436    #[case::nar_symlink_sha1(Some(CAHash::Flat(NixHash::Sha1(hex!("424eeaaa9cc016bab030bf007cb1be6483e7ba9e")))), &NAR_CONTENTS_SYMLINK.clone())]
437    #[tokio::test]
438    async fn ingest_with_flat_non_file(
439        #[case] ca_hash: Option<CAHash>,
440        #[case] nar_content: &[u8],
441    ) {
442        let err = ingest_nar_and_hash(
443            gen_test_blob_engine(),
444            snix_castore::utils::gen_test_directory_service(),
445            &mut Cursor::new(nar_content),
446            &ca_hash,
447        )
448        .await
449        .expect_err("Ingestion should have failed");
450
451        assert!(
452            matches!(err, NarIngestionError::TypeMismatch),
453            "Flat cahash should only be allowed for single file nars"
454        );
455    }
456}
snix_store/nar/import.rs

snix_store/nar/
import.rs
