1use super::{PathInfo, PathInfoService};
2use crate::{
3 nar::{NarIngestionError, ingest_nar_and_hash},
4 pathinfoservice::{self, nix_http::castore_infused::try_infused_nar_path},
5};
6use futures::{TryStreamExt, stream::BoxStream};
7use nix_compat::{
8 narinfo::{self, NarInfo, Signature},
9 nixbase32,
10 nixhash::NixHash,
11};
12use reqwest::StatusCode;
13use snix_castore::{
14 blob_engine,
15 blobstore::{self, BlobStore},
16 chunkstore::{self, ChunkStore},
17 directoryservice::{self, DirectoryService},
18 proto::{
19 blob_store_service_client::BlobStoreServiceClient,
20 chunk_store_service_client::ChunkStoreServiceClient,
21 directory_service_client::DirectoryServiceClient,
22 },
23};
24use snix_castore::{
25 composition::{CompositionContext, ServiceBuilder},
26 directoryservice::GRPCDirectoryService,
27};
28use std::sync::Arc;
29use tokio::io::{self, AsyncRead};
30use tonic::{async_trait, transport::Channel};
31use tracing::{Span, instrument, warn};
32use url::Url;
33
34mod castore_infused;
35
36pub struct NixHTTPPathInfoService<BS, CS, DS> {
52 instance_name: String,
53 base_url: url::Url,
54 http_client: reqwest_middleware::ClientWithMiddleware,
55
56 directory_service: DS,
57
58 blob_engine: Arc<blob_engine::ChunkingBlobEngine<BS, CS>>,
62
63 #[allow(clippy::type_complexity)]
71 layered_blob_engine: blob_engine::ChunkingBlobEngine<
72 Arc<blobstore::combinators::Cache<BS, blobstore::grpc::GrpcBlobStore<Channel>>>,
73 Arc<chunkstore::combinators::Cache<CS, chunkstore::grpc::GrpcChunkStore<Channel>>>,
74 >,
75
76 layered_directory_service:
80 directoryservice::combinators::Cache<DS, GRPCDirectoryService<Channel>>,
81
82 trusted_public_keys: Vec<narinfo::VerifyingKey>,
86
87 force_download_nar: bool,
89}
90
91impl<BS, CS, DS> NixHTTPPathInfoService<BS, CS, DS>
92where
93 BS: BlobStore + Clone,
94 CS: ChunkStore + Clone,
95 DS: DirectoryService + Clone,
96{
97 pub fn try_build(
98 instance_name: String,
99 config: NixHTTPPathInfoServiceConfig,
100 blob_store: BS,
101 chunk_store: CS,
102 directory_service: DS,
103 ) -> Result<Self, Error> {
104 let mut trusted_public_keys = Vec::new();
105 for s in config.params.trusted_public_keys {
106 trusted_public_keys.push(
107 narinfo::VerifyingKey::parse(&s).map_err(|e| Error::ParseTrustedPublicKey(s, e))?,
108 )
109 }
110
111 let blob_engine = Arc::new(
116 snix_castore::blob_engine::ChunkingBlobEngine::builder()
117 .blob_store(blob_store.clone())
118 .chunk_store(chunk_store.clone())
119 .build(),
120 );
121
122 let (layered_blob_engine, layered_directory_service) = {
123 let grpc_url = {
124 let url_str = format!("grpc+{}", config.base_url);
125 let mut url: Url = url_str.parse().expect("url to parse");
126 url.set_path("");
127 url
128 };
129
130 let channel =
131 snix_castore::tonic::TonicConnector::from_url(&grpc_url)?.connect_expect_lazy();
132
133 let instance_name_layered = format!("{}-layered", instance_name);
134 let instance_name_grpc = format!("{}-grpc", instance_name);
135
136 let layered_blob_store = blobstore::combinators::Cache::new(
137 instance_name_layered.clone(),
138 blob_store.clone(),
139 blobstore::grpc::GrpcBlobStore::from_client(
140 instance_name_grpc.clone(),
141 BlobStoreServiceClient::new(channel.clone()),
142 ),
143 );
144 let layered_chunk_store = chunkstore::combinators::Cache::new(
145 instance_name_layered.clone(),
146 chunk_store.clone(),
147 chunkstore::grpc::GrpcChunkStore::from_client(
148 instance_name_grpc.clone(),
149 ChunkStoreServiceClient::new(channel.clone()),
150 ),
151 );
152
153 let layered_directory_service = directoryservice::combinators::Cache::new(
154 instance_name_layered,
155 directory_service.clone(),
156 {
157 GRPCDirectoryService::from_client(
158 instance_name_grpc,
159 DirectoryServiceClient::new(channel).max_decoding_message_size(
160 directoryservice::GRPC_MAX_DECODING_MESSAGE_SIZE,
161 ),
162 )
163 },
164 );
165
166 (
167 snix_castore::blob_engine::ChunkingBlobEngine::builder()
168 .blob_store(Arc::new(layered_blob_store))
169 .chunk_store(Arc::new(layered_chunk_store))
170 .build(),
171 layered_directory_service,
172 )
173 };
174
175 Ok(Self {
176 instance_name,
177 base_url: {
178 let mut base_url = config.base_url;
180 if !base_url.path().ends_with('/') {
181 let with_slash = format!("{}/", base_url.path());
182 base_url.set_path(&with_slash);
183 }
184 base_url
185 },
186 http_client: reqwest_middleware::ClientBuilder::new(
187 reqwest::Client::builder()
188 .user_agent(crate::USER_AGENT)
189 .build()
190 .map_err(reqwest_middleware::Error::Reqwest)?,
191 )
192 .with(snix_tracing::propagate::reqwest::tracing_middleware())
193 .build(),
194 blob_engine,
195 directory_service,
196
197 layered_blob_engine,
198 layered_directory_service,
199
200 trusted_public_keys,
201 force_download_nar: config.params.force_download_nar,
202 })
203 }
204
205 #[instrument(level=tracing::Level::TRACE, skip_all,fields(path.digest=nixbase32::encode(&digest)),err)]
206 fn derive_narinfo_url(&self, digest: [u8; 20]) -> Result<Url, Error> {
207 let s = format!("{}.narinfo", nixbase32::encode(&digest));
208 self.base_url
209 .join(&s)
210 .map_err(|e| Error::JoinUrl(self.base_url.to_owned(), s.to_owned(), e))
211 }
212}
213
214#[derive(Debug, thiserror::Error)]
215pub enum Error {
216 #[error("wrong arguments: {0}")]
217 WrongConfig(&'static str),
218 #[error("serde-qs error: {0}")]
219 SerdeQS(#[from] serde_qs::Error),
220 #[error("unable to parse pubkey {0}")]
221 ParseTrustedPublicKey(String, nix_compat::narinfo::VerifyingKeyError),
222 #[error("unable to construct tonic channel: {0}")]
223 TonicChannel(#[from] snix_castore::tonic::Error),
224
225 #[error("unable to join URL {0} with {1}")]
226 JoinUrl(Url, String, url::ParseError),
227 #[error("reqwest error")]
228 Reqwest(#[from] reqwest_middleware::Error),
229 #[error("unable to decode NARInfo response as string")]
230 DecodeBody(reqwest::Error),
231 #[error("unable to parse NARInfo")]
232 ParseNARInfo(nix_compat::narinfo::Error),
233 #[error("no valid signature found")]
234 NoValidSignature,
235 #[error("failed to request NAR, status {0}")]
236 FailedToRequestNAR(reqwest::StatusCode),
237 #[error("unsupported NAR compression: {0}")]
238 UnsupportedNARCompression(String),
239 #[error("failed to ingest NAR")]
240 IngestNAR(NarIngestionError),
241 #[error("NARSize mismatch, narinfo size {narinfo_size}, actual size {actual_size}")]
242 NARSizeMismatch { narinfo_size: u64, actual_size: u64 },
243 #[error("NARHash mismatch, narinfo NARHash {exp}, actual NARHash {act}",
244 exp = NixHash::Sha256(*.narinfo_nar_sha256),
245 act = NixHash::Sha256(*.actual_nar_sha256))]
246 NARHashMismatch {
247 narinfo_nar_sha256: [u8; 32],
248 actual_nar_sha256: [u8; 32],
249 },
250
251 #[error("put not supported")]
252 PutNotSupported,
253 #[error("list not supported")]
254 ListNotSupported,
255}
256
257#[async_trait]
258impl<BS, CS, DS> PathInfoService for NixHTTPPathInfoService<BS, CS, DS>
259where
260 BS: BlobStore + Send + Sync + Clone + 'static,
261 CS: ChunkStore + Send + Sync + Clone + 'static,
262 DS: DirectoryService + Send + Sync + Clone + 'static,
263{
264 #[instrument(skip_all, err, fields(
265 path.digest=nixbase32::encode(&digest),
266 instance_name=%self.instance_name,
267 narinfo.url=tracing::field::Empty,
268 nar.url=tracing::field::Empty,
269 ))]
270 async fn get(&self, digest: [u8; 20]) -> Result<Option<PathInfo>, pathinfoservice::Error> {
271 let narinfo_url = self.derive_narinfo_url(digest)?;
272
273 let span = Span::current();
274 span.record("narinfo.url", narinfo_url.to_string());
275
276 let resp = self
277 .http_client
278 .get(narinfo_url)
279 .send()
280 .await
281 .map_err(Error::Reqwest)?;
282
283 if resp.status() == StatusCode::NOT_FOUND || resp.status() == StatusCode::FORBIDDEN {
287 return Ok(None);
288 }
289
290 let narinfo_str = resp.text().await.map_err(Error::DecodeBody)?;
291
292 let narinfo = NarInfo::parse(&narinfo_str).map_err(Error::ParseNARInfo)?;
294
295 if narinfo.store_path.digest() != &digest {
297 return Err("Store path digest in NARInfo doesn't match".into());
298 }
299
300 if !self.trusted_public_keys.is_empty() {
302 let fingerprint = narinfo.fingerprint();
303
304 if !self.trusted_public_keys.iter().any(|pubkey| {
305 narinfo
306 .signatures
307 .iter()
308 .any(|sig| pubkey.verify(&fingerprint, sig))
309 }) {
310 Err(Error::NoValidSignature)?
311 }
312 }
313
314 let root_node = if !self.force_download_nar
324 && let Some(root_node) = try_infused_nar_path(
325 &narinfo,
326 &self.layered_blob_engine,
327 &self.layered_directory_service,
328 )
329 .await
330 .unwrap_or_else(|err| {
331 warn!(%err, "unable to use infused store path");
332 None
333 }) {
334 root_node
335 } else {
336 let nar_url = self
338 .base_url
339 .join(narinfo.url)
340 .map_err(|e| Error::JoinUrl(self.base_url.clone(), narinfo.url.to_owned(), e))?;
341 span.record("nar.url", nar_url.to_string());
342
343 let resp = self
344 .http_client
345 .get(nar_url.clone())
346 .send()
347 .await
348 .map_err(Error::Reqwest)?;
349
350 if !resp.status().is_success() {
352 Err(Error::FailedToRequestNAR(resp.status()))?;
353 }
354
355 let r = tokio_util::io::StreamReader::new(resp.bytes_stream().map_err(|e| {
357 let e = e.without_url();
358 warn!(e=%e, "failed to get response body");
359 io::Error::new(io::ErrorKind::BrokenPipe, e.to_string())
360 }));
361
362 let mut r: Box<dyn AsyncRead + Send + Unpin> = match narinfo.compression {
364 None => Box::new(r) as Box<dyn AsyncRead + Send + Unpin>,
365 Some("bzip2") => Box::new(async_compression::tokio::bufread::BzDecoder::new(r))
366 as Box<dyn AsyncRead + Send + Unpin>,
367 Some("gzip") => Box::new(async_compression::tokio::bufread::GzipDecoder::new(r))
368 as Box<dyn AsyncRead + Send + Unpin>,
369 Some("xz") => Box::new(async_compression::tokio::bufread::XzDecoder::new(r))
370 as Box<dyn AsyncRead + Send + Unpin>,
371 Some("zstd") => {
372 let mut decoder = async_compression::tokio::bufread::ZstdDecoder::new(r);
375 decoder.multiple_members(true);
376 Box::new(decoder) as Box<dyn AsyncRead + Send + Unpin>
377 }
378 Some(comp_str) => Err(Error::UnsupportedNARCompression(comp_str.to_owned()))?,
379 };
380
381 let (root_node, nar_hash, nar_size) = ingest_nar_and_hash(
382 self.blob_engine.clone(),
383 &self.directory_service,
384 &mut r,
385 &narinfo.ca,
386 )
387 .await
388 .map_err(Error::IngestNAR)?;
389
390 if narinfo.nar_size != nar_size {
392 Err(Error::NARSizeMismatch {
393 narinfo_size: narinfo.nar_size,
394 actual_size: nar_size,
395 })?
396 }
397 if narinfo.nar_hash != nar_hash {
398 Err(Error::NARHashMismatch {
399 narinfo_nar_sha256: narinfo.nar_hash,
400 actual_nar_sha256: nar_hash,
401 })?
402 }
403 root_node
404 };
405
406 Ok(Some(PathInfo {
407 store_path: narinfo.store_path.to_owned(),
408 node: root_node,
409 references: narinfo.references.iter().map(|sp| sp.to_owned()).collect(),
410 nar_size: narinfo.nar_size,
411 nar_sha256: narinfo.nar_hash,
412 deriver: narinfo.deriver.as_ref().map(|sp| sp.to_owned()),
413 signatures: narinfo
414 .signatures
415 .into_iter()
416 .map(|s| Signature::<String>::new(s.name().to_string(), s.bytes().to_owned()))
417 .collect(),
418 ca: narinfo.ca,
419 }))
420 }
421
422 #[instrument(skip_all, err, fields(
423 path.digest=nixbase32::encode(&digest),
424 instance_name=%self.instance_name,
425 narinfo.url=tracing::field::Empty,
426 ))]
427 async fn has(&self, digest: [u8; 20]) -> Result<bool, pathinfoservice::Error> {
428 let narinfo_url = self.derive_narinfo_url(digest)?;
429
430 let span = Span::current();
431 span.record("narinfo.url", narinfo_url.to_string());
432
433 let resp = self
434 .http_client
435 .head(narinfo_url)
436 .send()
437 .await
438 .map_err(Error::Reqwest)?;
439
440 if resp.status() == StatusCode::NOT_FOUND || resp.status() == StatusCode::FORBIDDEN {
444 Ok(false)
445 } else {
446 Ok(true)
447 }
448 }
449
450 #[instrument(skip_all, fields(path_info=?_path_info, instance_name=%self.instance_name))]
451 async fn put(&self, _path_info: PathInfo) -> Result<PathInfo, pathinfoservice::Error> {
452 Err(Box::new(Error::PutNotSupported))
453 }
454
455 fn list(&self) -> BoxStream<'static, Result<PathInfo, pathinfoservice::Error>> {
456 Box::pin(futures::stream::once(async {
457 Err(Error::ListNotSupported)?
458 }))
459 }
460}
461
462#[derive(serde::Deserialize, Clone, Debug, PartialEq, Eq)]
463#[serde(deny_unknown_fields)]
464pub struct NixHTTPPathInfoServiceConfig {
465 base_url: Url,
466
467 #[serde(flatten)]
468 params: NixHTTPPathInfoServiceParams,
469}
470
471#[derive(serde::Deserialize, Clone, Debug, PartialEq, Eq)]
472#[serde(deny_unknown_fields)]
473struct NixHTTPPathInfoServiceParams {
474 #[serde(default = "default_blob_store")]
475 blob_store: String,
476 #[serde(default = "default_chunk_store")]
477 chunk_store: String,
478 #[serde(default = "default_directory_service")]
479 directory_service: String,
480 #[serde(default)]
481 trusted_public_keys: Vec<String>,
484
485 #[serde(default)]
486 force_download_nar: bool,
488}
489
490fn default_blob_store() -> String {
491 "&root".to_string()
492}
493fn default_chunk_store() -> String {
494 "&root".to_string()
495}
496fn default_directory_service() -> String {
497 "&root".to_string()
498}
499
500impl TryFrom<Url> for NixHTTPPathInfoServiceConfig {
501 type Error = Box<dyn std::error::Error + Send + Sync>;
502 fn try_from(url: Url) -> Result<Self, Self::Error> {
503 let scheme = url
504 .scheme()
505 .strip_prefix("nix+")
506 .ok_or_else(|| Error::WrongConfig("scheme must start with nix+"))?;
507
508 if !url.has_authority() {
509 Err(Error::WrongConfig("url must have authority component"))?
510 }
511 if !url.has_host() {
512 Err(Error::WrongConfig("url must have host component"))?
513 }
514 if !["http", "https"].contains(&scheme) {
515 Err(Error::WrongConfig("unknown scheme"))?
516 }
517
518 Ok(NixHTTPPathInfoServiceConfig {
519 base_url: {
525 let mut url: Url = url
526 .to_string()
527 .strip_prefix("nix+")
528 .unwrap()
529 .parse()
530 .expect("stripped URL to parse again");
531 url.set_query(None);
532 url
533 },
534 params: serde_qs::from_str(url.query().unwrap_or_default())?,
535 })
536 }
537}
538
539#[async_trait]
540impl ServiceBuilder for NixHTTPPathInfoServiceConfig {
541 type Output = dyn PathInfoService;
542 async fn build<'a>(
543 &'a self,
544 instance_name: &str,
545 context: &CompositionContext,
546 ) -> Result<Arc<Self::Output>, Box<dyn std::error::Error + Send + Sync + 'static>> {
547 let (blob_store, chunk_store, directory_service) = futures::join!(
548 context.resolve::<dyn BlobStore>(&self.params.blob_store),
549 context.resolve::<dyn ChunkStore>(&self.params.chunk_store),
550 context.resolve::<dyn DirectoryService>(&self.params.directory_service)
551 );
552 let svc = NixHTTPPathInfoService::try_build(
553 instance_name.to_string(),
554 self.to_owned(),
555 blob_store?,
556 chunk_store?,
557 directory_service?,
558 )?;
559 Ok(Arc::new(svc))
560 }
561}
562
563#[cfg(test)]
564mod tests {
565 use super::{NixHTTPPathInfoServiceConfig, NixHTTPPathInfoServiceParams};
566 use rstest::rstest;
567 use url::Url;
568
569 #[rstest]
570 #[case::correct_nix_https("nix+https://cache.nixos.org", Some(
572 NixHTTPPathInfoServiceConfig {
573 base_url: "https://cache.nixos.org".try_into().unwrap(),
574 params: NixHTTPPathInfoServiceParams {
575 blob_store: "&root".to_string(),
576 chunk_store: "&root".to_string(),
577 directory_service: "&root".to_string(),
578 trusted_public_keys: vec![],
579 force_download_nar: false,
580 }
581 }
582 ))]
583 #[case::correct_nix_http("nix+http://cache.nixos.org", Some(
585 NixHTTPPathInfoServiceConfig {
586 base_url: "http://cache.nixos.org".try_into().unwrap(),
587 params: NixHTTPPathInfoServiceParams {
588 blob_store: "&root".to_string(),
589 chunk_store: "&root".to_string(),
590 directory_service: "&root".to_string(),
591 trusted_public_keys: vec![],
592 force_download_nar: false,
593 }
594 }
595 ))]
596 #[case::correct_nix_http_with_subpath("nix+http://192.0.2.1/foo", Some(
598 NixHTTPPathInfoServiceConfig {
599 base_url: "http://192.0.2.1/foo".try_into().unwrap(),
600 params: NixHTTPPathInfoServiceParams {
601 blob_store: "&root".to_string(),
602 chunk_store: "&root".to_string(),
603 directory_service: "&root".to_string(),
604 trusted_public_keys: vec![],
605 force_download_nar: false,
606 }
607 }
608 ))]
609 #[case::correct_nix_http_with_subpath_and_port("nix+http://[::1]:8080/foo", Some(
611 NixHTTPPathInfoServiceConfig {
612 base_url: "http://[::1]:8080/foo".try_into().unwrap(),
613 params: NixHTTPPathInfoServiceParams {
614 blob_store: "&root".to_string(),
615 chunk_store: "&root".to_string(),
616 directory_service: "&root".to_string(),
617 trusted_public_keys: vec![],
618 force_download_nar: false,
619 }
620 }
621
622 ))]
623 #[case::correct_nix_https_with_trusted_public_key(
625 "nix+https://cache.nixos.org?trusted_public_keys[0]=cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=", Some(
626 NixHTTPPathInfoServiceConfig {
627 base_url: "https://cache.nixos.org".try_into().unwrap(),
628 params: NixHTTPPathInfoServiceParams {
629 blob_store: "&root".to_string(),
630 chunk_store: "&root".to_string(),
631 directory_service: "&root".to_string(),
632 trusted_public_keys: vec![
633 "cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=".to_string()
634 ],
635 force_download_nar: false,
636 }
637 }
638 ))]
639 #[case::correct_nix_https_with_two_trusted_public_keys(
641 "nix+https://cache.nixos.org?trusted_public_keys[0]=cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=&trusted_public_keys[1]=foo:jp4fCEx9tBEId/L0ZsVJ26k0wC0fu7vJqLjjIGFkup8=", Some(
642 NixHTTPPathInfoServiceConfig {
643 base_url: "https://cache.nixos.org".try_into().unwrap(),
644 params: NixHTTPPathInfoServiceParams {
645 blob_store: "&root".to_string(),
646 chunk_store: "&root".to_string(),
647 directory_service: "&root".to_string(),
648 trusted_public_keys: vec![
649 "cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=".to_string(),
650 "foo:jp4fCEx9tBEId/L0ZsVJ26k0wC0fu7vJqLjjIGFkup8=".to_string()
651 ],
652 force_download_nar: false,
653 }
654 }
655 ))]
656 #[case::wrong_scheme("nix+grpc://example.com", None)]
657 #[case::missing_host("nix+http:///", None)]
658 #[case::missing_authority("nix+http:", None)]
659 #[case::trusted_public_keys_no_sequence(
661 "nix+https://cache.nixos.org?trusted_public_keys=cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=",
662 None
663 )]
664 #[case::trusted_public_keys_wrong_pubkey(
666 "nix+https://cache.nixos.org?trustedpublickeys=cache.nixos.org-1:6NCHdD59X431o0gWypbMrAURkbJ16ZPMQFGspcDShjY=",
667 None
668 )]
669 fn parse_url(#[case] url_str: &str, #[case] exp_config: Option<NixHTTPPathInfoServiceConfig>) {
670 let url: Url = url_str.parse().expect("url to parse");
671
672 match (NixHTTPPathInfoServiceConfig::try_from(url), exp_config) {
673 (Ok(_), None) => panic!("parsing url unexpectedly succeeded"),
674 (Ok(config), Some(exp_config)) => assert_eq!(exp_config, config),
675 (Err(_), None) => {}
676 (Err(e), Some(_)) => panic!("parsing url unexpectedly failed: {e}"),
677 }
678 }
679}