feat(serve): compile-time integrity witness on the streaming serve path

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<R> VerifyingReader<R> {
     }
 }
 
+// #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<R: tokio::io::AsyncRead + Send + Unpin + 'static>
+    nora_registry::verified::stream_sealed::Sealed for VerifyingReader<R>
+{
+}
+impl<R: tokio::io::AsyncRead + Send + Unpin + 'static> nora_registry::verified::VerifiedByteReader
+    for VerifyingReader<R>
+{
+}
+
 impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for VerifyingReader<R> {
     fn poll_read(
         self: std::pin::Pin<&mut Self>,
@@ -1247,15 +1258,15 @@ 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")
             .header(header::CONTENT_LENGTH, size)
             .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()
                                     });

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()),
     )
 }

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

nora-registry/src/verified.rs

@@ -293,6 +293,94 @@ pub fn verified_body<T>(blob: Blob<Verified, T>) -> 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<Verified>`). 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<VerifyingReader<_>>`).
+/// `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<Verified>`], 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<Item = Result<axum::body::Bytes, std::io::Error>>
+    + 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<S: VerifiedByteStream>(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<R: VerifiedByteReader>(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<S>(stream: S) -> axum::body::Body
+where
+    S: futures::Stream<Item = Result<axum::body::Bytes, std::io::Error>> + Send + 'static,
+{
+    axum::body::Body::from_stream(stream)
+}
+
 // ---------------------------------------------------------------------------
 // Write-side witness: "store unpinned" — the S3 hole, named in the type.
 // ---------------------------------------------------------------------------