diff --git a/.changeset/pnpr-hosted-tar-storage.md b/.changeset/pnpr-hosted-tar-storage.md new file mode 100644 index 0000000000..8996cc568e --- /dev/null +++ b/.changeset/pnpr-hosted-tar-storage.md @@ -0,0 +1,7 @@ +--- +"@pnpm/pnpr": minor +--- + +A hosted registry that declares `tarballs: tar` now stores each published version as one tar compressed with zstd at level 19, and no `.tgz`. A client that accepts zstd gets those bytes, so even its first download is at level 19. Any other client gets the tar decoded, gzip-encoded if it asks for gzip. + +pnpr now refuses to start when a router without `tarballs: tar` lists such a registry as a source. Versions that an earlier pnpr published to such a registry are no longer served. Publish them again. diff --git a/pnpr/crates/config/src/registry_graph.rs b/pnpr/crates/config/src/registry_graph.rs index c81c585d74..7d78ca3c42 100644 --- a/pnpr/crates/config/src/registry_graph.rs +++ b/pnpr/crates/config/src/registry_graph.rs @@ -129,6 +129,7 @@ pub(super) fn build_registries( let defaults = addressed_defaults(defaults, ®istries); let registries = registries.with_defaults(defaults); registries.validate().map_err(|err| registry_err(&err))?; + check_hosted_tar_sources(®istries, &builder.hosted, &builder.tar_registries)?; Ok(ResolvedRegistryGraph { hosted: builder.hosted, registries, @@ -211,6 +212,32 @@ impl RegistryGraphBuilder { } } +/// A hosted registry that declares `tarballs: tar` stores its versions as +/// tars, with no `.tgz`, so a router that serves `.tgz` files cannot list it. +fn check_hosted_tar_sources( + registries: &Registries, + hosted: &IndexMap, + tar_registries: &BTreeSet, +) -> Result<(), RegistryError> { + let tgz_routers = registries + .names() + .filter(|name| registries.is_router(name) && !tar_registries.contains(*name)); + for router in tgz_routers { + let tar_source = registries + .sources(router, Ecosystem::Npm) + .into_iter() + .find(|source| hosted.contains_key(*source) && tar_registries.contains(*source)); + if let Some(source) = tar_source { + return Err(RegistryError::InvalidConfig { + reason: format!( + "router {router:?} serves `.tgz` files but lists {source:?}, which declares `tarballs: tar` and stores no `.tgz`; declare `tarballs: tar` on {router:?} or remove {source:?} from its sources", + ), + }); + } + } + Ok(()) +} + /// Whether `file` declares `tarballs: tar`, which only an npm registry may. fn declares_tar_tarballs(name: &str, file: &RegistryFile) -> Result { let (tarballs, ecosystem) = match file { diff --git a/pnpr/crates/config/src/tests/registry_graph.rs b/pnpr/crates/config/src/tests/registry_graph.rs index 0c519feef0..e3f7af97b6 100644 --- a/pnpr/crates/config/src/tests/registry_graph.rs +++ b/pnpr/crates/config/src/tests/registry_graph.rs @@ -499,6 +499,32 @@ registries: assert!(!config.routing.tar_registries.contains("npmjs")); } +#[test] +fn a_tgz_router_cannot_list_a_hosted_tar_registry() { + let yaml = " +registries: + local: + type: hosted + tarballs: tar + packages: + '@acme/*': + publish: $authenticated + main: + type: router + sources: [local] +"; + let error = Config::from_yaml_str(yaml, Path::new("/x"), listen(), None).unwrap_err(); + assert!( + error.to_string().contains(r#"router "main" serves `.tgz` files but lists "local""#), + "{error}", + ); + + let tar_router = format!("{yaml} tarballs: tar\n"); + let config = Config::from_yaml_str(&tar_router, Path::new("/x"), listen(), None).unwrap(); + assert!(config.routing.tar_registries.contains("local")); + assert!(config.routing.tar_registries.contains("main")); +} + #[test] fn tarballs_is_refused_outside_npm() { let yaml = " diff --git a/pnpr/crates/package-name/src/lib.rs b/pnpr/crates/package-name/src/lib.rs index 4f834bfb2d..0f3db228b9 100644 --- a/pnpr/crates/package-name/src/lib.rs +++ b/pnpr/crates/package-name/src/lib.rs @@ -196,6 +196,13 @@ impl CanonicalPackageName { format!("{}-{version}.tgz", self.basename) } + /// The blob a hosted tar registry stores `version` in: its tar, zstd + /// compressed, in place of the published `.tgz`. + #[must_use] + pub fn tar_name_for_version(&self, version: &str) -> String { + format!("{}-{version}.tar.zst", self.basename) + } + #[must_use] pub fn sigstore_name_for_version(&self, version: &str) -> String { format!("{}-{version}.sigstore", self.basename) diff --git a/pnpr/crates/pnpr/src/export.rs b/pnpr/crates/pnpr/src/export.rs index 9c281ce6d5..ad2d518227 100644 --- a/pnpr/crates/pnpr/src/export.rs +++ b/pnpr/crates/pnpr/src/export.rs @@ -131,6 +131,9 @@ async fn export_version_blobs( { export_single_blob(storage, canonical, filename, pkg_dir, report).await?; } + // A hosted tar registry stores the version's tar in place of its `.tgz`. + let tar = canonical.tar_name_for_version(ver); + export_single_blob(storage, canonical, &tar, pkg_dir, report).await?; if dist .get("attestations") .and_then(|a| a.get("url")) diff --git a/pnpr/crates/pnpr/src/server.rs b/pnpr/crates/pnpr/src/server.rs index 96e4b5bf17..73673ed07e 100644 --- a/pnpr/crates/pnpr/src/server.rs +++ b/pnpr/crates/pnpr/src/server.rs @@ -48,8 +48,8 @@ use tar_registry::{ mod tar_responses; use tar_responses::{ - CachedTgz, MAX_CONCURRENT_TAR_DECODES, TarSource, TgzTrust, tar_decode_limit, tar_integrity_of, - tar_integrity_within, tar_resource_integrity, tar_response, + CachedTgz, MAX_CONCURRENT_TAR_DECODES, TarSource, TgzTrust, stored_tar_response, + tar_decode_limit, tar_integrity_of, tar_integrity_within, tar_resource_integrity, tar_response, }; mod version_tarballs; @@ -68,8 +68,9 @@ use package_responses::{ mod package_reads; use package_reads::{ HostedGate, RegistrySource, default_registry_target, hosted_gate, hosted_read_namespace, - resolve_ecosystem_source, resolve_registry_source, resolves_to_private_source, serve_packument, - serve_registry_tarball, serve_tarball, serve_version_manifest, + resolve_ecosystem_source, resolve_registry_source, resolves_to_private_source, + serve_hosted_version_blob, serve_packument, serve_registry_tarball, serve_tarball, + serve_version_manifest, }; mod revision_tarballs; diff --git a/pnpr/crates/pnpr/src/server/package_mutation.rs b/pnpr/crates/pnpr/src/server/package_mutation.rs index 53a1d54aa5..a7a3a70b48 100644 --- a/pnpr/crates/pnpr/src/server/package_mutation.rs +++ b/pnpr/crates/pnpr/src/server/package_mutation.rs @@ -267,7 +267,11 @@ async fn remove_unpublished_blobs( let blobs = removed_versions .iter() .flat_map(|version| { - [name.tarball_name_for_version(version), name.sigstore_name_for_version(version)] + [ + name.tarball_name_for_version(version), + name.tar_name_for_version(version), + name.sigstore_name_for_version(version), + ] }); futures_util::stream::iter(blobs) .for_each_concurrent(BLOB_REMOVALS_IN_FLIGHT, |blob| async move { diff --git a/pnpr/crates/pnpr/src/server/package_mutation/unpublish.rs b/pnpr/crates/pnpr/src/server/package_mutation/unpublish.rs index 66149a2424..406d61b4a9 100644 --- a/pnpr/crates/pnpr/src/server/package_mutation/unpublish.rs +++ b/pnpr/crates/pnpr/src/server/package_mutation/unpublish.rs @@ -154,10 +154,14 @@ pub(crate) async fn delete_tarball( Ok(false) => return RegistryError::NotFound.into_response(), Err(err) => return err.into_response(), } - let removed = match storage.remove_blob(&name, &canonical).await { - Ok(removed) => removed, - Err(err) => return err.into_response(), - }; + // A hosted tar registry stores the version's tar in place of its `.tgz`. + let mut removed = false; + for blob in [canonical.clone(), name.tar_name_for_version(&version)] { + match storage.remove_blob(&name, &blob).await { + Ok(found) => removed |= found, + Err(err) => return err.into_response(), + } + } let sigstore_name = name.sigstore_name_for_version(&version); let _ = storage.remove_blob(&name, &sigstore_name).await; diff --git a/pnpr/crates/pnpr/src/server/package_reads.rs b/pnpr/crates/pnpr/src/server/package_reads.rs index 2210fb85a0..422021fd46 100644 --- a/pnpr/crates/pnpr/src/server/package_reads.rs +++ b/pnpr/crates/pnpr/src/server/package_reads.rs @@ -469,12 +469,23 @@ pub(super) async fn serve_hosted_tarball( Ok(parsed) => parsed, Err(err) => return err.into_response(), }; - if let Err(err) = ensure_osv_allowed(state, name, &name_version) { + serve_hosted_version_blob(state, &org, name, &name_version, &filename).await +} + +/// The blob `filename` of `version` in the hosted namespace `org`. +pub(super) async fn serve_hosted_version_blob( + state: &AppState, + org: &str, + name: &CanonicalPackageName, + version: &str, + filename: &str, +) -> Response { + if let Err(err) = ensure_osv_allowed(state, name, version) { return err.into_response(); } match state.inner.storage - .for_hosted(&org, pnpr_registry::Ecosystem::Npm) - .open_hosted_blob(name, &filename) + .for_hosted(org, pnpr_registry::Ecosystem::Npm) + .open_hosted_blob(name, filename) .await { Ok(Some((body, len))) => tarball_response(body, len), diff --git a/pnpr/crates/pnpr/src/server/publishing/attachments.rs b/pnpr/crates/pnpr/src/server/publishing/attachments.rs index b81738002d..574ea022be 100644 --- a/pnpr/crates/pnpr/src/server/publishing/attachments.rs +++ b/pnpr/crates/pnpr/src/server/publishing/attachments.rs @@ -13,6 +13,17 @@ use super::{ use std::sync::Arc; use tokio::sync::{OwnedSemaphorePermit, Semaphore}; +/// What a hosted registry keeps of a published `.tgz`. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(in super::super) enum TarballStorage { + /// The `.tgz` as published. + Tgz, + /// Only the tar it decodes to, compressed at level 19 (see + /// [`CanonicalPackageName::tar_name_for_version`]): a registry that + /// declares `tarballs: tar`. + Tar, +} + /// What kind of payload an attachment carries. #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(in super::super) enum AttachmentKind { @@ -33,6 +44,7 @@ pub(in super::super) struct ValidatedPublish { /// The caller these checks admitted, for the checks that repeat under /// the package lock. pub(in super::super) publisher: Identity, + pub(in super::super) tarballs: TarballStorage, } /// One publish attachment resolved to its canonical on-disk filename and its @@ -79,7 +91,12 @@ pub(in super::super) async fn validate_publish_doc( // canonical storage filename, matching version, and payload kind. let prepared = prepare_attachments(&name, &incoming, attachments)?; let publisher = identity.clone(); - Ok((ValidatedPublish { name, incoming, prepared, publisher }, target)) + let tarballs = if super::super::tar_registry::is_tar_registry(state, &target.source) { + TarballStorage::Tar + } else { + TarballStorage::Tgz + }; + Ok((ValidatedPublish { name, incoming, prepared, publisher, tarballs }, target)) } fn prepare_attachments( @@ -174,7 +191,13 @@ pub(in super::super) async fn stage_publish( )?; check_staged_tombstone(state, &storage, &doc, hosted.is_some()).await?; - let ValidatedPublish { name, mut incoming, prepared, .. } = doc; + let ValidatedPublish { + name, + mut incoming, + prepared, + tarballs, + .. + } = doc; restore_submitted_tgz_integrities(&storage, &name, hosted.as_ref(), &mut incoming).await?; check_publishable_versions(&name, &incoming, hosted.as_ref(), &prepared)?; @@ -190,7 +213,8 @@ pub(in super::super) async fn stage_publish( // 400; any tmp files written before the failure get removed // along the way so a bad upload leaves no on-disk artifact. let written = - write_attachment_slots(&storage, &name, prepared, &state.inner.proxy.tar_decodes).await?; + write_attachment_slots(&storage, &name, prepared, tarballs, &state.inner.proxy.tar_decodes) + .await?; let document = serialize_staged_document(merged, &written.tars); Ok(StagedPublish { name, @@ -221,12 +245,13 @@ pub(super) async fn write_attachment_slots( storage: &Storage, name: &CanonicalPackageName, prepared: Vec, + tarballs: TarballStorage, decodes: &Arc, ) -> Result { let mut written = WrittenAttachments { slots: Vec::with_capacity(prepared.len()), tars: Vec::new() }; for attachment in prepared { - match write_attachment_slot(storage, name, attachment, decodes).await { + match write_attachment_slot(storage, name, attachment, tarballs, decodes).await { Ok((slot, tar)) => { written.slots.push(slot); written.tars.extend(tar); @@ -243,45 +268,90 @@ pub(super) async fn write_attachment_slots( /// Stream-decode, verify and write one tarball into a reserved slot, and /// derive the tar a `.tgz` decodes to. A mismatch, or a missing integrity /// field, fails with a 400. A `.tgz` that does not decode in full has no tar -/// and is published without one. A tarball holds a permit from `decodes` -/// until its tar is hashed. +/// and is published without one, except under [`TarballStorage::Tar`], which +/// stores the tar in the slot and fails such a `.tgz` with a 400. A tarball +/// holds a permit from `decodes` until its tar is hashed. pub(super) async fn write_attachment_slot( storage: &Storage, name: &CanonicalPackageName, prepared: PreparedAttachment, + tarballs: TarballStorage, decodes: &Arc, ) -> Result<(pnpr_storage::BlobSlot, Option), RegistryError> { - let slot = storage.reserve_hosted_blob(name, &prepared.canonical).await?; - let result = write_under_permit( - prepared.attachment, - prepared.kind, - prepared.dist, - slot.tmp_path.clone(), - decodes, - ) - .await; - match result { - Ok(Ok(None)) => Ok((slot, None)), - Ok(Ok(Some((tgz_integrity, integrity, inspection)))) => { - // A tar registry's first tar response records a binding publish - // failed to. - let derived = pnpr_storage::DerivedTar::new( - integrity.to_string(), - inspection.requires_build, - inspection.file_count, - inspection.unpacked_size, - ) - .with_bins(inspection.bins.clone()); - if let Err(err) = storage.record_tar_integrity(name, &tgz_integrity, derived).await { - tracing::warn!(?err, package = %name.as_str(), "recording a tar integrity failed"); - } - Ok((slot, Some(PublishedTar { version: prepared.version, inspection }))) + let stores_tar = tarballs == TarballStorage::Tar && prepared.kind == AttachmentKind::Tarball; + let blob = if stores_tar { + name.tar_name_for_version(&prepared.version) + } else { + prepared.canonical.clone() + }; + let slot = storage.reserve_hosted_blob(name, &blob).await?; + let paths = AttachmentPaths::for_slot(&slot.tmp_path, stores_tar); + let result = + write_under_permit(prepared.attachment, prepared.kind, prepared.dist, paths, decodes).await; + let derived = match result { + Ok(Ok(derived)) => derived, + Ok(Err(err)) => { + let _ = tokio::fs::remove_file(&slot.tmp_path).await; + return Err(err); } - Ok(Err(err)) => Err(err), Err(join_err) => { let _ = tokio::fs::remove_file(&slot.tmp_path).await; - Err(RegistryError::Io(std::io::Error::other(join_err.to_string()))) + return Err(RegistryError::Io(std::io::Error::other(join_err.to_string()))); } + }; + let Some((tgz_integrity, integrity, inspection)) = derived else { return Ok((slot, None)) }; + let recorded = + record_published_tar(storage, name, &tgz_integrity, &integrity, &inspection, stores_tar); + if let Err(err) = recorded.await { + let _ = tokio::fs::remove_file(&slot.tmp_path).await; + return Err(err); + } + Ok((slot, Some(PublishedTar { version: prepared.version, inspection }))) +} + +/// Record the tar a published `.tgz` decodes to. A failure only warns: a tar +/// registry's first tar response records it later. A stored tar is served +/// without its `.tgz`, so with `stores_tar` nothing would, and the failure +/// fails the publish. +async fn record_published_tar( + storage: &Storage, + name: &CanonicalPackageName, + tgz_integrity: &str, + integrity: &Integrity, + inspection: &pnpm_tarball::TarInspection, + stores_tar: bool, +) -> Result<(), RegistryError> { + let derived = pnpr_storage::DerivedTar::new( + integrity.to_string(), + inspection.requires_build, + inspection.file_count, + inspection.unpacked_size, + ) + .with_bins(inspection.bins.clone()); + let Err(err) = storage.record_tar_integrity(name, tgz_integrity, derived).await else { + return Ok(()); + }; + tracing::warn!(?err, package = %name.as_str(), "recording a tar integrity failed"); + if stores_tar { Err(err) } else { Ok(()) } +} + +/// Where an attachment is written, and where its tar is encoded to when the +/// registry stores tars. +struct AttachmentPaths { + attachment: std::path::PathBuf, + tar: Option, +} + +impl AttachmentPaths { + /// The paths for the blob slot at `slot`. With `stores_tar`, the verified + /// `.tgz` waits beside the slot until its tar is encoded into it. + fn for_slot(slot: &std::path::Path, stores_tar: bool) -> Self { + if !stores_tar { + return Self { attachment: slot.to_path_buf(), tar: None }; + } + let mut tgz = slot.as_os_str().to_os_string(); + tgz.push(".tgz"); + Self { attachment: tgz.into(), tar: Some(slot.to_path_buf()) } } } @@ -291,7 +361,7 @@ async fn write_under_permit( attachment: PendingAttachment, kind: AttachmentKind, dist: Value, - tmp_path: std::path::PathBuf, + paths: AttachmentPaths, decodes: &Arc, ) -> Result< Result, RegistryError>, @@ -307,7 +377,19 @@ async fn write_under_permit( AttachmentKind::Sigstore => None, }; tokio::task::spawn_blocking(move || { - write_and_derive_tar(&attachment, kind, &dist, &tmp_path, permit) + // The permit covers the encode too: it decodes the `.tgz` again. + let derived = + write_and_derive_tar(&attachment, kind, &dist, &paths.attachment, permit.as_ref()); + let Some(tar_path) = &paths.tar else { return derived }; + let stored = derived.and_then(|derived| { + let derived = derived.ok_or_else(|| RegistryError::BadRequest { + reason: format!("{} does not decode to a tar", attachment.filename), + })?; + encode_stored_tar(&paths.attachment, tar_path)?; + Ok(Some(derived)) + }); + let _ = std::fs::remove_file(&paths.attachment); + stored }) .await } @@ -320,7 +402,7 @@ fn write_and_derive_tar( kind: AttachmentKind, dist: &Value, tmp_path: &std::path::Path, - _decode: Option, + _decode: Option<&OwnedSemaphorePermit>, ) -> Result, RegistryError> { decode_and_write_attachment(attachment, kind, dist, tmp_path)?; if kind != AttachmentKind::Tarball { @@ -332,6 +414,22 @@ fn write_and_derive_tar( .map(|(tar, inspection)| (tgz_integrity, tar, inspection))) } +/// Compress the tar the verified `.tgz` at `tgz_path` decodes to into +/// `tar_path`. [`tar_integrity_of`] already decoded it within the limit. +fn encode_stored_tar( + tgz_path: &std::path::Path, + tar_path: &std::path::Path, +) -> Result<(), RegistryError> { + let tgz = std::io::BufReader::new(std::fs::File::open(tgz_path)?); + let tar = flate2::read::MultiGzDecoder::new(tgz); + let out = std::io::BufWriter::new(std::fs::File::create(tar_path)?); + super::super::tar_variants::encode_tar(tar, out, None)? + .into_inner() + .map_err(std::io::IntoInnerError::into_error)? + .sync_all()?; + Ok(()) +} + fn decode_and_write_attachment( attachment: &PendingAttachment, kind: AttachmentKind, diff --git a/pnpr/crates/pnpr/src/server/publishing/tests.rs b/pnpr/crates/pnpr/src/server/publishing/tests.rs index b2f0cb72ee..ec512e47a9 100644 --- a/pnpr/crates/pnpr/src/server/publishing/tests.rs +++ b/pnpr/crates/pnpr/src/server/publishing/tests.rs @@ -1,5 +1,5 @@ use super::{ - attachments::{AttachmentKind, PreparedAttachment, write_attachment_slot}, + attachments::{AttachmentKind, PreparedAttachment, TarballStorage, write_attachment_slot}, manifest::record_publisher, }; use base64::Engine; @@ -47,7 +47,7 @@ async fn a_published_tarball_hashes_its_tar_under_a_decode_permit() { kind: AttachmentKind::Tarball, }; let decodes = Arc::new(tokio::sync::Semaphore::new(0)); - let write = write_attachment_slot(&storage, &name, prepared, &decodes); + let write = write_attachment_slot(&storage, &name, prepared, TarballStorage::Tgz, &decodes); tokio::pin!(write); let waited = tokio::time::timeout(Duration::from_millis(200), &mut write).await; diff --git a/pnpr/crates/pnpr/src/server/tar_responses.rs b/pnpr/crates/pnpr/src/server/tar_responses.rs index 5bc0c58189..1583626cb8 100644 --- a/pnpr/crates/pnpr/src/server/tar_responses.rs +++ b/pnpr/crates/pnpr/src/server/tar_responses.rs @@ -6,6 +6,9 @@ //! encodes at its default level. //! - neither: the decoded tar. //! +//! A hosted tar registry stores no `.tgz`, only the level-19 tar (see +//! [`stored_tar_response`]). +//! //! A cached upstream `.tgz` is served without a re-check against the current //! packument, and a client of a tar registry may have no tar hash to check what //! it receives. Its bytes are checked against the version's `.tgz` integrity @@ -19,7 +22,7 @@ pub(super) use decode::{ use super::{AppState, CanonicalPackageName, HeaderMap, Response, TarCopy, header, open_variant}; use axum::body::Body; -use decode::{DerivedFrom, gunzip_body}; +use decode::{DerivedFrom, gunzip_body, unzstd_body}; use negotiation::{TarCoding, preferred_coding}; use pnpr_storage::{Storage, streaming}; use ssri::Integrity; @@ -88,6 +91,32 @@ pub(super) async fn tar_response( } } +/// Serve `stored`, a hosted tar registry's level-19 zstd tar, as the request +/// accepts: the stored bytes to a `zstd` client, else the tar decoded, which +/// the response compression layer encodes for a `gzip` client. +pub(super) fn stored_tar_response( + state: &AppState, + headers: &HeaderMap, + stored: Response, +) -> Response { + match preferred_coding(headers) { + TarCoding::Zstd => { + let len = stored + .headers() + .get(header::CONTENT_LENGTH) + .and_then(|value| value.to_str().ok()?.parse().ok()); + let (parts, body) = stored.into_parts(); + encoded_response(Response::from_parts(parts, Body::empty()), body, len, "zstd") + } + TarCoding::Gzip | TarCoding::Identity => { + let (mut parts, body) = stored.into_parts(); + describe_tar_headers(&mut parts.headers); + let decodes = std::sync::Arc::clone(&state.inner.proxy.tar_decodes); + Response::from_parts(parts, unzstd_body(body, decodes)) + } + } +} + impl TarSource { /// The integrity the served bytes still have to be checked against. fn check(&self) -> Option { diff --git a/pnpr/crates/pnpr/src/server/tar_responses/decode.rs b/pnpr/crates/pnpr/src/server/tar_responses/decode.rs index 1b674917eb..bbb9ad0f14 100644 --- a/pnpr/crates/pnpr/src/server/tar_responses/decode.rs +++ b/pnpr/crates/pnpr/src/server/tar_responses/decode.rs @@ -137,6 +137,35 @@ pub(super) fn gunzip_body( derived: DerivedFrom, decodes: Arc, on_complete: impl FnOnce(Integrity, pnpm_tarball::TarInspection) + Send + 'static, +) -> Body { + decoded_body(body, decodes, move |compressed_rx, tar_tx| { + derive_from_decode(compressed_rx, tar_tx, derived, on_complete); + }) +} + +/// Decode a stored zstd tar as it streams. A body that fails to decode, or +/// decodes past [`MAX_DECODED_TAR_BYTES`], ends the response with an error. +pub(super) fn unzstd_body(body: Body, decodes: Arc) -> Body { + decoded_body(body, decodes, |compressed_rx, tar_tx| { + let decoder = match zstd::stream::read::Decoder::new(ChannelReader::new(compressed_rx)) { + Ok(decoder) => decoder.take(MAX_DECODED_TAR_BYTES.saturating_add(1)), + Err(err) => { + let _ = tar_tx.blocking_send(Err(err)); + return; + } + }; + decode_into(decoder, tar_tx, &mut None, None, false, MAX_DECODED_TAR_BYTES); + }) +} + +/// The body `decode` produces on a blocking thread from `body`'s chunks, +/// once a permit from `decodes` is free. +fn decoded_body( + body: Body, + decodes: Arc, + decode: impl FnOnce(mpsc::Receiver>, &mpsc::Sender>) + + Send + + 'static, ) -> Body { const CHANNEL_CHUNKS: usize = 16; let (compressed_tx, compressed_rx) = mpsc::channel::>(CHANNEL_CHUNKS); @@ -149,7 +178,7 @@ pub(super) fn gunzip_body( }; tokio::task::spawn_blocking(move || { let _permit = permit; - derive_from_decode(compressed_rx, &tar_tx, derived, on_complete); + decode(compressed_rx, &tar_tx); }); }); let chunks = futures_util::stream::unfold(tar_rx, |mut tar_rx| async move { diff --git a/pnpr/crates/pnpr/src/server/tar_variants.rs b/pnpr/crates/pnpr/src/server/tar_variants.rs index fc6ad2adb6..43a99ad168 100644 --- a/pnpr/crates/pnpr/src/server/tar_variants.rs +++ b/pnpr/crates/pnpr/src/server/tar_variants.rs @@ -246,11 +246,22 @@ async fn write_variant(job: &VariantJob) -> io::Result<()> { fn encode_level_19(tar_path: &std::path::Path) -> io::Result> { let mut tar = Vec::new(); std::fs::File::open(tar_path)?.read_to_end(&mut tar)?; - let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), LEVEL)?; + let size = tar.len() as u64; + encode_tar(tar.as_slice(), Vec::new(), Some(size)) +} + +/// Compress `tar` into `out` at level 19, within the window pnpm decodes. +/// `size`, when known, is recorded in the frame. +pub(super) fn encode_tar( + mut tar: impl Read, + out: Out, + size: Option, +) -> io::Result { + let mut encoder = zstd::stream::write::Encoder::new(out, LEVEL)?; encoder.window_log(WINDOW_LOG)?; - encoder.include_contentsize(true)?; - encoder.set_pledged_src_size(Some(tar.len() as u64))?; - encoder.write_all(&tar)?; + encoder.include_contentsize(size.is_some())?; + encoder.set_pledged_src_size(size)?; + io::copy(&mut tar, &mut encoder)?; encoder.finish() } diff --git a/pnpr/crates/pnpr/src/server/version_tarballs.rs b/pnpr/crates/pnpr/src/server/version_tarballs.rs index 6915a243ca..f12237b419 100644 --- a/pnpr/crates/pnpr/src/server/version_tarballs.rs +++ b/pnpr/crates/pnpr/src/server/version_tarballs.rs @@ -7,6 +7,7 @@ //! registry also needs for the `.tgz` integrity: from a fresh cached upstream //! packument, else from the upstream's version manifest, else from the //! packument. A dist read from an upstream is kept in a [`VersionTgzCache`]. +//! A hosted tar registry stores no `.tgz`: its stored tar is served instead. pub(super) use cache::VersionTgzCache; @@ -17,10 +18,11 @@ use super::{ HeaderMap, Identity, PackumentDists, RegistryError, RegistrySource, Response, StatusCode, TarSource, TarballOrigin, TgzTrust, Upstream, UpstreamTarball, Value, addressed_registry, authorize, authorized_upstream, cached_upstream_tarball, caller_scoped, - declared_tarball_integrity, ensure_osv_allowed, is_tar_registry, not_found, - read_source_packument, recover_stale_upstream_packument, resolve_registry_source, - serve_bound_upstream_tarball, serve_registry_tarball, serve_revision_tarball, tar_decode_limit, - tar_response, tarball_basename_for_package, tarball_cache_name, tgz_integrity_key, timed, + declared_tarball_integrity, ensure_osv_allowed, hosted_read_namespace, is_tar_registry, + not_found, read_source_packument, recover_stale_upstream_packument, resolve_registry_source, + serve_bound_upstream_tarball, serve_hosted_version_blob, serve_registry_tarball, + serve_revision_tarball, stored_tar_response, tar_decode_limit, tar_response, + tarball_basename_for_package, tarball_cache_name, tgz_integrity_key, timed, upstream_cache_namespace, version_route_tarball_version, }; use axum::response::IntoResponse; @@ -96,15 +98,32 @@ impl VersionRequest<'_> { } let tar_registry = is_tar_registry(self.state, self.target); let source = resolve_registry_source(self.state, self.target, name.as_str()); + if let RegistrySource::Hosted(hosted) = &source + && is_tar_registry(self.state, hosted) + { + return self.serve_stored_tar(headers, hosted, &name).await; + } if !tar_registry && let Some(response) = self.serve_canonical_tgz(&name, &source).await { return response; } - let tgz = match self.version_tgz(&source, &name).await { + self.serve_from_dist(headers, name, &source, tar_registry).await + } + + /// Serve the `.tgz` the version's dist names, and on a tar registry the + /// tar it decodes to. + async fn serve_from_dist( + &self, + headers: &HeaderMap, + name: CanonicalPackageName, + source: &RegistrySource, + tar_registry: bool, + ) -> Response { + let tgz = match self.version_tgz(source, &name).await { Ok(Some(tgz)) => tgz, Ok(None) => return not_found(), Err(err) => return err.into_response(), }; - let (response, trust) = match &source { + let (response, trust) = match source { RegistrySource::Upstream(upstream) => self.serve_upstream(upstream, &name, &tgz).await, RegistrySource::Hosted(_) | RegistrySource::Unclaimed | RegistrySource::NotFound => { let response = serve_registry_tarball( @@ -121,10 +140,30 @@ impl VersionRequest<'_> { if !tar_registry || response.status() != StatusCode::OK { return response; } - let source = self.tar_source(name, &tgz, &source, trust).await; + let source = self.tar_source(name, &tgz, source, trust).await; tar_response(self.state, headers, response, source).await } + /// Serve the version's stored tar from the hosted tar registry `hosted` + /// (see [`CanonicalPackageName::tar_name_for_version`]). + async fn serve_stored_tar( + &self, + headers: &HeaderMap, + hosted: &str, + name: &CanonicalPackageName, + ) -> Response { + let org = match hosted_read_namespace(self.state, self.identity, hosted, name.as_str()) { + Ok(org) => org, + Err(err) => return err.into_response(), + }; + let blob = name.tar_name_for_version(self.version); + let response = serve_hosted_version_blob(self.state, &org, name, self.version, &blob).await; + if response.status() != StatusCode::OK { + return response; + } + stored_tar_response(self.state, headers, response) + } + /// Serve the version's `.tgz` without reading its packument: a hosted /// blob, which publish names canonically, or a cached upstream `.tgz` /// whose filename [`VersionTgzCache`] already bound to this version. diff --git a/pnpr/crates/pnpr/src/verify.rs b/pnpr/crates/pnpr/src/verify.rs index 319bc7321c..f6e6b09350 100644 --- a/pnpr/crates/pnpr/src/verify.rs +++ b/pnpr/crates/pnpr/src/verify.rs @@ -127,29 +127,85 @@ async fn verify_version_tarball( return Ok(()); }; - let blob_key = format!("{pkg_name}/{filename}"); - ctx.referenced_blobs.insert(blob_key); + // A hosted tar registry stores the version's tar in place of its `.tgz`. + let tar = canonical.tar_name_for_version(ver); + ctx.referenced_blobs.insert(format!("{pkg_name}/{filename}")); + ctx.referenced_blobs.insert(format!("{pkg_name}/{tar}")); + if let Some(bytes) = read_blob(ctx, canonical, filename).await? { + if ctx.options.check_hashes { + check_hash_mismatch(pkg_name, ver, filename, &bytes, dist, ctx.report); + } + } else if !verify_stored_tar(ctx, canonical, ver, dist).await? { + ctx.report.missing_blobs.push(format!("{pkg_name}@{ver}: {filename}")); + } + Ok(()) +} + +/// Check the tar a hosted tar registry stores for `ver` in place of its +/// `.tgz`. `false` when there is none. +async fn verify_stored_tar( + ctx: &mut VerifyContext<'_>, + canonical: &CanonicalPackageName, + ver: &str, + dist: &serde_json::Map, +) -> miette::Result { + let tar = canonical.tar_name_for_version(ver); + let Some(bytes) = read_blob(ctx, canonical, &tar).await? else { return Ok(false) }; + if ctx.options.check_hashes + && let Some(problem) = stored_tar_problem(ctx.storage, canonical, &bytes, dist).await? + { + let pkg_name = canonical.as_str(); + ctx.report.corrupted_blobs.push(format!("{pkg_name}@{ver}: {tar} ({problem})")); + } + Ok(true) +} + +/// What is wrong with a stored tar, checked against the tar integrity +/// recorded at publish: `dist.integrity` names the `.tgz`, which was not kept. +async fn stored_tar_problem( + storage: &Storage, + canonical: &CanonicalPackageName, + stored: &[u8], + dist: &serde_json::Map, +) -> miette::Result> { + let table = storage + .read_tar_integrities(canonical) + .await + .map_err(|err| { + miette::miette!("failed to read tar integrities for '{}': {err}", canonical.as_str()) + })?; + let expected = dist + .get("integrity") + .and_then(Value::as_str) + .and_then(|integrity| integrity.parse::().ok()) + .and_then(|integrity| table.get(&integrity.to_string())); + let Some(expected) = expected else { + return Ok(Some("no recorded tar integrity".to_string())); + }; + let Ok(tar) = zstd::decode_all(stored) else { + return Ok(Some("not a zstd stream".to_string())); + }; + Ok((!verify_blob_integrity(&tar, &expected.tar_integrity)).then(|| { + format!("tar integrity mismatch: {}", expected.tar_integrity) + })) +} + +/// The bytes of the hosted blob `filename`, counted as checked. +async fn read_blob( + ctx: &mut VerifyContext<'_>, + canonical: &CanonicalPackageName, + filename: &str, +) -> miette::Result> { let opened = ctx.storage .open_hosted_blob(canonical, filename) .await - .map_err(|err| miette::miette!("error opening blob for '{pkg_name}': {err}"))?; - - match opened { - None => { - ctx.report.missing_blobs.push(format!("{pkg_name}@{ver}: {filename}")); - } - Some((body, _)) => { - let bytes = to_bytes(body, usize::MAX).await.into_diagnostic()?; - ctx.report.blobs_checked += 1; - ctx.report.bytes_verified += bytes.len() as u64; - - if ctx.options.check_hashes { - check_hash_mismatch(pkg_name, ver, filename, &bytes, dist, ctx.report); - } - } - } - Ok(()) + .map_err(|err| miette::miette!("error opening blob for '{}': {err}", canonical.as_str()))?; + let Some((body, _)) = opened else { return Ok(None) }; + let bytes = to_bytes(body, usize::MAX).await.into_diagnostic()?; + ctx.report.blobs_checked += 1; + ctx.report.bytes_verified += bytes.len() as u64; + Ok(Some(bytes)) } fn check_hash_mismatch( diff --git a/pnpr/crates/pnpr/tests/server/tar_registry.rs b/pnpr/crates/pnpr/tests/server/tar_registry.rs index 7514e04769..00512d86ad 100644 --- a/pnpr/crates/pnpr/tests/server/tar_registry.rs +++ b/pnpr/crates/pnpr/tests/server/tar_registry.rs @@ -1,3 +1,5 @@ +#[path = "tar_registry/hosted_tars.rs"] +mod hosted_tars; #[path = "tar_registry/mutations.rs"] mod mutations; diff --git a/pnpr/crates/pnpr/tests/server/tar_registry/hosted_tars.rs b/pnpr/crates/pnpr/tests/server/tar_registry/hosted_tars.rs new file mode 100644 index 0000000000..fc6438b69f --- /dev/null +++ b/pnpr/crates/pnpr/tests/server/tar_registry/hosted_tars.rs @@ -0,0 +1,212 @@ +use super::{ + AuthState, Body, Config, Ecosystem, PackagePattern, Registries, Registry, Request, ServiceExt, + StatusCode, TempDir, body_bytes, body_json, config_for, get, gzip, header, + hosted_publish_request, hosted_with_access, issue_token, router_with_auth, sha512_integrity, + widget_tar, +}; + +/// A hosted `acme` registry that declares `tarballs: tar`, behind the +/// path-less `zstd` router, which serves tars too. +fn tar_acme_config(storage: std::path::PathBuf) -> Config { + let mut config = config_for("http://127.0.0.1:1", storage); + config.routing.hosted.insert("acme".to_string(), hosted_with_access("acme", "$all")); + let graph = vec![ + ("npmjs".to_string(), Registry::Upstream { patterns: vec![] }), + ( + "acme".to_string(), + Registry::Hosted { + patterns: vec![PackagePattern::parse("@acme/*", Ecosystem::Npm).unwrap()], + }, + ), + ("zstd".to_string(), Registry::Router { sources: vec!["acme".to_string()] }), + ]; + config.routing.registries = + Registries::new(graph.into_iter().collect(), Some("zstd".to_string())); + config.routing.tar_registries.insert("acme".to_string()); + config.routing.tar_registries.insert("zstd".to_string()); + config +} + +/// The path of the file named `name` under `dir`. +fn find_file(dir: &std::path::Path, name: &str) -> Option { + for entry in std::fs::read_dir(dir).unwrap() { + let entry = entry.unwrap(); + if entry.file_type().unwrap().is_dir() { + if let Some(found) = find_file(&entry.path(), name) { + return Some(found); + } + } else if entry.file_name() == name { + return Some(entry.path()); + } + } + None +} + +/// The names of every file under `dir`. +fn file_names(dir: &std::path::Path) -> Vec { + let mut names = Vec::new(); + for entry in std::fs::read_dir(dir).unwrap() { + let entry = entry.unwrap(); + if entry.file_type().unwrap().is_dir() { + names.extend(file_names(&entry.path())); + } else { + names.push( + entry + .file_name() + .to_string_lossy() + .into_owned(), + ); + } + } + names +} + +async fn publish(app: &axum::Router, token: &str, version: &str, tgz: &[u8]) -> StatusCode { + app.clone() + .oneshot(hosted_publish_request( + "/npm/~acme/@acme/widget", + "@acme/widget", + version, + tgz, + token, + )) + .await + .unwrap() + .status() +} + +#[tokio::test] +async fn a_hosted_tar_registry_stores_and_serves_only_the_tar() { + let tmp = TempDir::new().unwrap(); + let auth = AuthState::in_memory(); + let token = issue_token(&auth, "alice").await; + let app = router_with_auth(tar_acme_config(tmp.path().to_path_buf()), auth); + let tar = widget_tar(); + assert_eq!(publish(&app, &token, "1.0.0", &gzip(&tar)).await, StatusCode::CREATED); + + let files = file_names(tmp.path()); + assert!(files.contains(&"widget-1.0.0.tar.zst".to_string()), "{files:?}"); + assert!( + !files + .iter() + .any(|file| file.ends_with(".tgz")), + "{files:?}", + ); + + let packument = body_json(get(&app, "/npm/@acme%2Fwidget", None).await.into_body()).await; + assert_eq!(packument["versions"]["1.0.0"]["dist"]["integrity"], sha512_integrity(&tar)); + + let zstd = get(&app, "/npm/@acme/widget@1.0.0", Some("zstd")).await; + assert_eq!(zstd.status(), StatusCode::OK); + assert_eq!(zstd.headers()[header::CONTENT_ENCODING], "zstd"); + assert_eq!(zstd.headers()[header::CONTENT_TYPE], "application/x-tar"); + assert!(zstd.headers().contains_key(header::CONTENT_LENGTH)); + let decoded = zstd::decode_all(body_bytes(zstd.into_body()).await.as_slice()).unwrap(); + assert_eq!(decoded, tar); + + let identity = get(&app, "/npm/@acme/widget@1.0.0", None).await; + assert_eq!(identity.status(), StatusCode::OK); + assert!( + identity + .headers() + .get(header::CONTENT_ENCODING) + .is_none(), + ); + assert_eq!(body_bytes(identity.into_body()).await, tar); + + let gzip_client = get(&app, "/npm/@acme/widget@1.0.0", Some("gzip")).await; + assert_eq!(gzip_client.status(), StatusCode::OK); + let gzipped = gzip_client + .headers() + .get(header::CONTENT_ENCODING) + .cloned(); + let body = body_bytes(gzip_client.into_body()).await; + let body = if gzipped.is_some_and(|coding| coding == "gzip") { + let mut decoded = Vec::new(); + std::io::Read::read_to_end( + &mut flate2::read::GzDecoder::new(body.as_slice()), + &mut decoded, + ) + .unwrap(); + decoded + } else { + body.clone() + }; + assert_eq!(body, tar); + + let through_hosted = get(&app, "/npm/~acme/@acme/widget@1.0.0", None).await; + assert_eq!(body_bytes(through_hosted.into_body()).await, tar); +} + +#[tokio::test] +async fn a_hosted_tar_registry_refuses_a_tgz_that_does_not_decode() { + let tmp = TempDir::new().unwrap(); + let auth = AuthState::in_memory(); + let token = issue_token(&auth, "alice").await; + let app = router_with_auth(tar_acme_config(tmp.path().to_path_buf()), auth); + + let mut truncated = gzip(&widget_tar()); + truncated.truncate(truncated.len() - 8); + assert_eq!(publish(&app, &token, "1.0.0", &truncated).await, StatusCode::BAD_REQUEST); + let files = file_names(tmp.path()); + assert!( + !files + .iter() + .any(|file| file.contains("widget-1.0.0")), + "{files:?}", + ); +} + +#[tokio::test] +async fn deleting_a_version_of_a_hosted_tar_registry_removes_its_tar() { + let tmp = TempDir::new().unwrap(); + let auth = AuthState::in_memory(); + let token = issue_token(&auth, "alice").await; + let app = router_with_auth(tar_acme_config(tmp.path().to_path_buf()), auth); + let tar = widget_tar(); + assert_eq!(publish(&app, &token, "1.0.0", &gzip(&tar)).await, StatusCode::CREATED); + + let delete = app + .clone() + .oneshot( + Request::delete("/npm/~acme/@acme/widget/-/widget-1.0.0.tgz") + .header(header::AUTHORIZATION, format!("Bearer {token}")) + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(delete.status(), StatusCode::CREATED); + let files = file_names(tmp.path()); + assert!(!files.contains(&"widget-1.0.0.tar.zst".to_string()), "{files:?}"); + let served = get(&app, "/npm/@acme/widget@1.0.0", Some("zstd")).await; + assert_eq!(served.status(), StatusCode::NOT_FOUND); +} + +#[tokio::test] +async fn verify_checks_a_stored_tar_against_its_tar_integrity() { + let tmp = TempDir::new().unwrap(); + let auth = AuthState::in_memory(); + let token = issue_token(&auth, "alice").await; + let app = router_with_auth(tar_acme_config(tmp.path().to_path_buf()), auth); + assert_eq!(publish(&app, &token, "1.0.0", &gzip(&widget_tar())).await, StatusCode::CREATED); + + let config = tar_acme_config(tmp.path().to_path_buf()); + pnpr_storage::layout::write_local_layout_marker(&config.storage.hosted_dir).unwrap(); + let options = + pnpr::verify::VerifyOptions { check_hashes: true, prune_orphans: false, min_age_secs: 0 }; + let report = pnpr::verify::verify_storage(&config, &options).await.unwrap(); + assert_eq!(report.blobs_checked, 1); + assert!(report.corrupted_blobs.is_empty(), "{:?}", report.corrupted_blobs); + assert!(report.orphaned_blobs.is_empty(), "{:?}", report.orphaned_blobs); + + let stored = find_file(tmp.path(), "widget-1.0.0.tar.zst").unwrap(); + std::fs::write(&stored, zstd::encode_all(&b"another tar"[..], 3).unwrap()).unwrap(); + let report = pnpr::verify::verify_storage(&config, &options).await.unwrap(); + assert_eq!(report.corrupted_blobs.len(), 1, "{:?}", report.corrupted_blobs); + assert!( + report.corrupted_blobs[0].contains("tar integrity mismatch"), + "{:?}", + report.corrupted_blobs, + ); +} diff --git a/pnpr/crates/pnpr/tests/tooling.rs b/pnpr/crates/pnpr/tests/tooling.rs index 63cd4c7b09..42cffebdea 100644 --- a/pnpr/crates/pnpr/tests/tooling.rs +++ b/pnpr/crates/pnpr/tests/tooling.rs @@ -143,6 +143,50 @@ async fn test_export_storage() { assert!(export_out.join("demo-pkg/demo-pkg-1.0.0.tgz").exists()); } +#[tokio::test] +async fn verify_and_export_keep_the_stored_tar_of_a_hosted_tar_registry() { + let dir = TempDir::new().unwrap(); + let (config, _cfg_file) = create_test_config(dir.path()); + let pkg = "demo-pkg"; + write_local_layout_marker(&config.storage.hosted_dir).unwrap(); + let doc = serde_json::json!({ + "name": "demo-pkg", + "versions": { + "1.0.0": { + "name": "demo-pkg", + "version": "1.0.0", + "dist": { "tarball": "http://localhost:7677/demo-pkg/-/demo-pkg-1.0.0.tgz" } + } + } + }); + seed_file( + local_document_path(&config.storage.hosted_dir, "", Ecosystem::Npm, pkg), + serde_json::to_vec(&doc).unwrap(), + ); + let stored_tar = local_blob_path( + &config.storage.hosted_dir, + "", + Ecosystem::Npm, + pkg, + "demo-pkg-1.0.0.tar.zst", + ); + seed_file(stored_tar.clone(), b"zstd tar bytes"); + + let prune = VerifyOptions { check_hashes: true, prune_orphans: true, min_age_secs: 0 }; + let report = verify_storage(&config, &prune).await.unwrap(); + assert_eq!(report.blobs_checked, 1); + assert!(report.missing_blobs.is_empty(), "{:?}", report.missing_blobs); + assert!(report.orphaned_blobs.is_empty(), "{:?}", report.orphaned_blobs); + assert!(stored_tar.exists()); + assert_eq!(report.corrupted_blobs.len(), 1, "{:?}", report.corrupted_blobs); + assert!(report.corrupted_blobs[0].contains("no recorded tar integrity")); + + let export_out = dir.path().join("exported"); + let exported = export_storage(&config, &export_out, true).await.unwrap(); + assert_eq!(exported.blobs_exported, 1); + assert!(export_out.join("demo-pkg/demo-pkg-1.0.0.tar.zst").exists()); +} + #[tokio::test] async fn test_verify_storage_health_and_orphan_prune() { let dir = TempDir::new().unwrap(); diff --git a/pnpr/docs/configuration.md b/pnpr/docs/configuration.md index 1a6c3596c6..641682f5ff 100644 --- a/pnpr/docs/configuration.md +++ b/pnpr/docs/configuration.md @@ -358,7 +358,8 @@ npm, Yarn, and pnpm projects that do not opt in keep using `main`. A pnpm projec - Every registry serves a version's tarball at `/@`. A tar registry serves no `.tgz` path. - For an upstream package without fresh cached metadata, pnpr reads the version's metadata from `//`. If the upstream does not serve it, pnpr reads the package's full metadata. -- pnpr keeps the published `.tgz` and derives the tar from it: +- A hosted registry that declares `tarballs: tar` keeps no `.tgz`. At publish, pnpr compresses the tar at level 19 and stores only that. A `zstd` client gets the stored bytes, and any other client gets the tar decoded, compressed as the request accepts. pnpr refuses to start when a router without `tarballs: tar` lists such a registry as a source, because the router would have no `.tgz` to serve. npm signatures and provenance sign the `.tgz`, so these packages have neither. +- For every other registry, pnpr keeps the published `.tgz` and derives the tar from it: - A `gzip` client gets the `.tgz` itself, which is the gzip encoding of its tar. - A `zstd` client gets a level-19 encoding once pnpr has stored one. Until then pnpr compresses the tar at level 3 as it streams, and a background worker stores the level-19 encoding. Tars larger than 256 MiB stay at level 3. The level-19 encodings live in the cache, so they can be evicted and are made again. - Any other client gets the tar uncompressed.