feat(pnpr): store only the zstd tar for hosted tar registries (#17437)

A hosted registry that declares `tarballs: tar` is only ever served
through tar registries, so the published `.tgz` has no reader there.
Publish now verifies the `.tgz`, records its tar hash as before, and
stores the tar compressed with zstd at level 19, within the window pnpm
decodes, as `<name>-<version>.tar.zst` in place of the `.tgz`. A `.tgz`
that does not decode in full fails the publish with a 400.

The version route serves the stored bytes to a zstd client and the tar
decoded to any other, which the compression layer gzips when asked. These
versions skip the per-replica level-19 variant cache.

Config loading rejects a router without `tarballs: tar` that lists such a
registry, since it would have no `.tgz` to serve. Unpublish and packument
rewrites remove the stored tar, `pnpr verify` counts it as the version's
blob instead of an orphan, and `pnpr export` writes it.

Decided in pnpm/tasks#226.
This commit is contained in:
Zoltan Kochan authored and GitHub committed 2026-10-28 01:40:51 +01:00
1 parent 64588da2d9
commit 5fb5fba3a2
20 files changed
+694 -83

No files matched your search

+7
View File
@@ -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.
+27
View File
@@ -129,6 +129,7 @@ pub(super) fn build_registries(
let defaults = addressed_defaults(defaults, &registries);
let registries = registries.with_defaults(defaults);
registries.validate().map_err(|err| registry_err(&err))?;
check_hosted_tar_sources(&registries, &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<String, HostedConfig>,
tar_registries: &BTreeSet<String>,
) -> 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<bool, RegistryError> {
let (tarballs, ecosystem) = match file {
@@ -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 = "
+7
View File
@@ -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)
+3
View File
@@ -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"))
+5 -4
View File
@@ -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;
@@ -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 {
@@ -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;
+14 -3
View File
@@ -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),
@@ -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<PreparedAttachment>,
tarballs: TarballStorage,
decodes: &Arc<Semaphore>,
) -> Result<WrittenAttachments, RegistryError> {
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<Semaphore>,
) -> Result<(pnpr_storage::BlobSlot, Option<PublishedTar>), 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<std::path::PathBuf>,
}
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<Semaphore>,
) -> Result<
Result<Option<(String, Integrity, pnpm_tarball::TarInspection)>, 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<OwnedSemaphorePermit>,
_decode: Option<&OwnedSemaphorePermit>,
) -> Result<Option<(String, Integrity, pnpm_tarball::TarInspection)>, 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,
@@ -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;
+30 -1
View File
@@ -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<Integrity> {
@@ -137,6 +137,35 @@ pub(super) fn gunzip_body(
derived: DerivedFrom,
decodes: Arc<Semaphore>,
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<Semaphore>) -> 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<Semaphore>,
decode: impl FnOnce(mpsc::Receiver<io::Result<Bytes>>, &mpsc::Sender<io::Result<Bytes>>)
+ Send
+ 'static,
) -> Body {
const CHANNEL_CHUNKS: usize = 16;
let (compressed_tx, compressed_rx) = mpsc::channel::<io::Result<Bytes>>(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 {
+15 -4
View File
@@ -246,11 +246,22 @@ async fn write_variant(job: &VariantJob) -> io::Result<()> {
fn encode_level_19(tar_path: &std::path::Path) -> io::Result<Vec<u8>> {
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<Out: Write>(
mut tar: impl Read,
out: Out,
size: Option<u64>,
) -> io::Result<Out> {
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()
}
@@ -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.
+75 -19
View File
@@ -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<String, Value>,
) -> miette::Result<bool> {
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<String, Value>,
) -> miette::Result<Option<String>> {
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::<ssri::Integrity>().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<Option<axum::body::Bytes>> {
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(
@@ -1,3 +1,5 @@
#[path = "tar_registry/hosted_tars.rs"]
mod hosted_tars;
#[path = "tar_registry/mutations.rs"]
mod mutations;
@@ -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<std::path::PathBuf> {
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<String> {
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,
);
}
+44
View File
@@ -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();
+2 -1
View File
@@ -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 `<registry>/<name>@<version>`. A tar registry serves no `.tgz` path.
- For an upstream package without fresh cached metadata, pnpr reads the version's metadata from `<upstream>/<name>/<version>`. 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.