snix_store/nar/
import.rs

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

snix_store/nar/
import.rs
