diff --git a/nora-registry/src/registry/docker.rs b/nora-registry/src/registry/docker.rs index c82e771..7132380 100644 --- a/nora-registry/src/registry/docker.rs +++ b/nora-registry/src/registry/docker.rs @@ -32,7 +32,6 @@ use serde_json::{json, Value}; use std::collections::HashMap; use std::sync::Arc; use std::time::{Duration, SystemTime, UNIX_EPOCH}; -use tokio_util::io::ReaderStream; // ============================================================================ // Namespaced key builders (issue #323) @@ -280,6 +279,18 @@ impl VerifyingReader { } } +// #849: `VerifyingReader` verifies the blob digest at EOF, so it may be served through the +// `reader_stream_body` sole-sink. A raw reader carries neither impl, so it cannot be handed +// to it — a compile-time serve-integrity witness on the streaming path. +impl + nora_registry::verified::stream_sealed::Sealed for VerifyingReader +{ +} +impl nora_registry::verified::VerifiedByteReader + for VerifyingReader +{ +} + impl tokio::io::AsyncRead for VerifyingReader { fn poll_read( self: std::pin::Pin<&mut Self>, @@ -1247,7 +1258,7 @@ async fn download_blob( crate::registry_type::RegistryType::Docker, "LOCAL", )); - let stream = ReaderStream::new(VerifyingReader::new(reader, &digest)); + let stream = VerifyingReader::new(reader, &digest); return Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -1255,7 +1266,7 @@ async fn download_blob( .header(header::ACCEPT_RANGES, "bytes") .header(header::CACHE_CONTROL, "public, max-age=31536000, immutable") .header("docker-content-digest", &digest) - .body(Body::from_stream(stream)) + .body(nora_registry::verified::reader_stream_body(stream)) .unwrap_or_else(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response()); } @@ -1374,8 +1385,7 @@ async fn download_blob( // Successfully stored — stream from storage match state.storage.get_reader(&key).await { Ok((size, _pin, reader)) => { - let stream = - ReaderStream::new(VerifyingReader::new(reader, &digest)); + let stream = VerifyingReader::new(reader, &digest); return Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") @@ -1385,7 +1395,7 @@ async fn download_blob( "public, max-age=31536000, immutable", ) .header("docker-content-digest", &digest) - .body(Body::from_stream(stream)) + .body(nora_registry::verified::reader_stream_body(stream)) .unwrap_or_else(|_| { StatusCode::INTERNAL_SERVER_ERROR.into_response() }); @@ -1399,13 +1409,13 @@ async fn download_blob( // put_from_path failed — stream from temp file directly match tokio::fs::File::open(&fetched.path).await { Ok(file) => { - let stream = ReaderStream::new(VerifyingReader::new(file, &digest)); + let stream = VerifyingReader::new(file, &digest); return Response::builder() .status(StatusCode::OK) .header(header::CONTENT_TYPE, "application/octet-stream") .header(header::CONTENT_LENGTH, file_size) .header("docker-content-digest", &digest) - .body(Body::from_stream(stream)) + .body(nora_registry::verified::reader_stream_body(stream)) .unwrap_or_else(|_| { StatusCode::INTERNAL_SERVER_ERROR.into_response() }); diff --git a/nora-registry/src/registry/range.rs b/nora-registry/src/registry/range.rs index 76b8476..e2a2d68 100644 --- a/nora-registry/src/registry/range.rs +++ b/nora-registry/src/registry/range.rs @@ -177,7 +177,9 @@ pub(crate) async fn range_response( } Some( response - .body(Body::from_stream(ReaderStream::new(reader?))) + .body(nora_registry::verified::open_world_stream_body( + ReaderStream::new(reader?), + )) .unwrap_or_else(|_| StatusCode::INTERNAL_SERVER_ERROR.into_response()), ) } diff --git a/nora-registry/src/registry/raw.rs b/nora-registry/src/registry/raw.rs index eff6731..be750b6 100644 --- a/nora-registry/src/registry/raw.rs +++ b/nora-registry/src/registry/raw.rs @@ -176,11 +176,9 @@ async fn download( builder = builder.header(header::ETAG, format!("\"{}\"", hash)); } builder - .body(axum::body::Body::from_stream(verify_while_streaming( - reader, - pin, - key.clone(), - ))) + .body(nora_registry::verified::stream_body( + verify_while_streaming(reader, pin, key.clone()), + )) .expect("valid response") .into_response() } @@ -281,6 +279,12 @@ fn verify_while_streaming( } } +// #849: `VerifyingStream` is an EOF-verifying stream, so it may be served through the +// `stream_body` sole-sink. A raw `ReaderStream` (no digest check) does not carry these +// impls, so it cannot be handed to `stream_body` — a compile-time serve-integrity witness. +impl nora_registry::verified::stream_sealed::Sealed for VerifyingStream {} +impl nora_registry::verified::VerifiedByteStream for VerifyingStream {} + /// Verify an RFC 9530 `Repr-Digest` header against the server-computed sha-256. /// /// The header gates the commit but never sets the pin, so the stored pin is diff --git a/nora-registry/src/verified.rs b/nora-registry/src/verified.rs index d232b94..c028d47 100644 --- a/nora-registry/src/verified.rs +++ b/nora-registry/src/verified.rs @@ -293,6 +293,94 @@ pub fn verified_body(blob: Blob) -> T { blob.into_inner() } +// --------------------------------------------------------------------------- +// Streaming serve sink: the type-level witness on the STREAMING path (#849). +// +// The buffered path above discharges a compile-time witness (`verified_body` +// takes only `Blob`). Streaming can't carry a whole-`Blob` value, so +// the guarantee is restored one level up: only a stream that was wired through an +// EOF-verifying wrapper implements `VerifiedByteStream`, and `stream_body` — the +// single verified `Body::from_stream` site — accepts only that. Handing a raw +// reader stream to it is a compile error, exactly like `verified_body(raw)`. +// --------------------------------------------------------------------------- + +/// Seal for [`VerifiedByteStream`]. NORA's own EOF-verifying stream types attest it where +/// they are defined (raw `VerifyingStream`, docker `ReaderStream>`). +/// `pub` (but doc-hidden) only because those types live in the `nora` binary crate while +/// this trait lives in the `nora_registry` library — the handlers must be able to impl it. +/// The guarantee that matters holds regardless: a *new* NORA stream does not fit +/// [`stream_body`] without a deliberate, visible `impl` (a raw reader stream is a compile +/// error), and the single `Body::from_stream` serve site is [`stream_body`]. +#[doc(hidden)] +pub mod stream_sealed { + /// The seal. + pub trait Sealed {} +} + +/// A byte stream whose bytes flow through an EOF-verifying wrapper — it hashes while +/// streaming and aborts the body on a digest mismatch (tamper-evident). The streaming +/// analogue of holding a [`Blob`], and, like it, unforgeable outside NORA: +/// [`stream_body`] accepts only this type, so serving a raw reader stream on an +/// integrity path is a **compile error**. +pub trait VerifiedByteStream: + futures::Stream> + + Send + + 'static + + stream_sealed::Sealed +{ +} + +/// The sole streaming serve sink for integrity-checked bytes — the compile-time +/// counterpart of [`verified_body`]. A raw reader stream does not implement +/// [`VerifiedByteStream`], so it cannot be turned into a response body here. +/// +/// # A raw stream does NOT compile on the verified streaming path +/// +/// ```compile_fail +/// use nora_registry::verified::stream_body; +/// // a plain ReaderStream (no EOF digest check) is not a VerifiedByteStream +/// let raw = tokio_util::io::ReaderStream::new(tokio::io::empty()); +/// let _ = stream_body(raw); // the trait bound `VerifiedByteStream` is not satisfied +/// ``` +pub fn stream_body(stream: S) -> axum::body::Body { + axum::body::Body::from_stream(stream) +} + +/// An `AsyncRead` that verifies its content digest at EOF — docker's `VerifyingReader`, +/// which recomputes SHA-256 as bytes flow and errors the read on mismatch. The reader +/// analogue of [`VerifiedByteStream`]; accepted only by [`reader_stream_body`]. +pub trait VerifiedByteReader: + tokio::io::AsyncRead + Send + Unpin + 'static + stream_sealed::Sealed +{ +} + +/// The docker-blob streaming sink: frames a [`VerifiedByteReader`] into a response body. +/// Like [`stream_body`], a raw reader (no EOF digest check) does not implement the trait, +/// so serving it here is a **compile error**. This is where the single `Body::from_stream` +/// for the reader path lives. +/// +/// # A raw reader does NOT compile on the verified streaming path +/// +/// ```compile_fail +/// use nora_registry::verified::reader_stream_body; +/// // a plain AsyncRead (no EOF digest check) is not a VerifiedByteReader +/// let _ = reader_stream_body(tokio::io::empty()); // trait bound `VerifiedByteReader` unmet +/// ``` +pub fn reader_stream_body(reader: R) -> axum::body::Body { + axum::body::Body::from_stream(tokio_util::io::ReaderStream::new(reader)) +} + +/// The explicit open-world streaming arm: partial-content / byte-range serves, where a +/// partial range has no whole-file digest to check against. Named — never silent — the +/// streaming counterpart of [`GateOutcome::Unpinned`]: the call site states that these +/// bytes are served without a NORA-side cryptographic check. +pub fn open_world_stream_body(stream: S) -> axum::body::Body +where + S: futures::Stream> + Send + 'static, +{ + axum::body::Body::from_stream(stream) +} + // --------------------------------------------------------------------------- // Write-side witness: "store unpinned" — the S3 hole, named in the type. // ---------------------------------------------------------------------------