feat: implement encrypted read

Author: alanshaw

DESIGN_NOTES.md

@@ -100,6 +100,20 @@ by the location commitment. The indexing-service query path is implemented
 but unwired. The full trace is the
 [GetObject diagram](./docs/diagrams.md#getobject-version-resolution-local-tiers-network-retrieval).
 
+**Encrypted blobs** (the FilOne encryption design's read side) decrypt inside
+the per-blob open, leaving every other read-path value plaintext: the
+manifest's `BlobRef.Offset/Length`, `Body.Size`, ETag and Content-Length are
+plaintext coordinates, while `BlobRef.Digest` names the stored FEE envelope.
+A `blob_encryption_params` row marks a blob encrypted and carries what its
+decryptor needs; the read unwraps the region-wrapped CEK through
+`regionkey.Provider` (OpenBao transit in production, bound to the blob's
+(space, digest)), maps the plaintext range to one contiguous ciphertext span
+(`aesstream.CiphertextRange`), fetches only that span (ranged from the spool
+or piri via `OpenBlobRange`), and decrypts it as it streams
+(`aesstream.SpanReader`). A tampered chunk fails authentication mid-stream.
+With no region key provider configured the lookups are skipped entirely.
+HEAD never decrypts. See `s3frontend/decrypt.go`.
+
 ## Identity & auth
 
 - **agent**: `ServiceIdentity.Signer` (daemon: `identity.key_file` PEM), the

blockstore/cache.go

@@ -54,8 +54,9 @@ func NewCached(base BlockReader, maxBytes int64) BlockReader {
 }
 
 var (
-	_ BlockReader = (*Cached)(nil)
-	_ BlobReader  = (*Cached)(nil)
+	_ BlockReader     = (*Cached)(nil)
+	_ BlobReader      = (*Cached)(nil)
+	_ BlobRangeReader = (*Cached)(nil)
 )
 
 // OpenBlob streams a body blob straight from the base reader, bypassing the LRU.
@@ -69,6 +70,15 @@ func (c *Cached) OpenBlob(ctx context.Context, space did.DID, digest mh.Multihas
 	return nil, ErrNotFound
 }
 
+// OpenBlobRange streams stored bytes [off, off+n) of a body blob straight
+// from the base reader, bypassing the LRU exactly like OpenBlob.
+func (c *Cached) OpenBlobRange(ctx context.Context, space did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error) {
+	if br, ok := c.base.(BlobReader); ok {
+		return OpenBlobRangeOf(ctx, br, space, digest, off, n)
+	}
+	return nil, ErrNotFound
+}
+
 // GetBlock returns the cached block if present, otherwise fetches it from the
 // base reader and caches it (subject to the byte budget). The LRU is keyed by
 // CID only — blocks are content-addressed, so a hit is valid whatever space

blockstore/forge.go

@@ -1,13 +1,16 @@
 package blockstore
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io"
 	"net/http"
 	"net/url"
 
+	"github.com/fil-forge/libforge/blobindex"
+
 	indexclient "github.com/fil-forge/indexing-service/pkg/client"
 	contentcmds "github.com/fil-forge/libforge/commands/content"
 	"github.com/fil-forge/libforge/ucan/retrieval"
@@ -55,8 +58,9 @@ type Forge struct {
 }
 
 var (
-	_ BlockReader = (*Forge)(nil)
-	_ BlobReader  = (*Forge)(nil)
+	_ BlockReader     = (*Forge)(nil)
+	_ BlobReader      = (*Forge)(nil)
+	_ BlobRangeReader = (*Forge)(nil)
 )
 
 // ForgeConfig wires a read-only Forge block reader.
@@ -141,16 +145,19 @@ func NewForge(cfg ForgeConfig) (*Forge, error) {
 // empty 200 when it can't resolve a did:plc principal in the proof chain)
 // otherwise surfaces only as a bare client-side EOF mid-body, with nothing
 // attributable in ingot's own logs.
-func (f *Forge) retrieve(ctx context.Context, space did.DID, c cid.Cid) (io.ReadCloser, int64, error) {
-	rc, n, err := f.doRetrieve(ctx, space, c)
+// off/n select a sub-range of the located bytes: stored bytes [off, off+n)
+// of the blob. n < 0 retrieves the locator's whole range (the entire blob,
+// for a body blob).
+func (f *Forge) retrieve(ctx context.Context, space did.DID, c cid.Cid, off, n int64) (io.ReadCloser, int64, error) {
+	rc, got, err := f.doRetrieve(ctx, space, c, off, n)
 	if err != nil && !errors.Is(err, ErrNotFound) {
 		f.logger.Warn("forge: retrieve failed",
 			zap.Stringer("space", space), zap.Stringer("cid", c), zap.Error(err))
 	}
-	return rc, n, err
+	return rc, got, err
 }
 
-func (f *Forge) doRetrieve(ctx context.Context, space did.DID, c cid.Cid) (io.ReadCloser, int64, error) {
+func (f *Forge) doRetrieve(ctx context.Context, space did.DID, c cid.Cid, off, n int64) (io.ReadCloser, int64, error) {
 	locations, err := f.locator.Locate(ctx, []did.DID{space}, c.Hash())
 	if err != nil {
 		var nf locator.NotFoundError
@@ -170,6 +177,23 @@ func (f *Forge) doRetrieve(ctx context.Context, space did.DID, c cid.Cid) (io.Re
 	}
 	target := cm.Location[0]
 
+	// Narrow the located range to the caller's sub-range. loc.Range is
+	// inclusive and, for a whole blob, starts at 0 — so the sub-range is
+	// [Start+off, Start+off+n-1], clamped to the located end. The decrypting
+	// read path uses this to fetch only the ciphertext chunks it needs.
+	if n >= 0 {
+		start := loc.Range.Start + off
+		if start > loc.Range.End || n == 0 {
+			// Nothing to fetch; an empty body keeps the contract simple.
+			return io.NopCloser(bytes.NewReader(nil)), 0, nil
+		}
+		end := start + n - 1
+		if end > loc.Range.End {
+			end = loc.Range.End
+		}
+		loc.Range = blobindex.Range{Start: start, End: end}
+	}
+
 	// The commitment's space scopes the retrieve capability; fall back to
 	// the caller's space if the commitment lacks one (the local locator
 	// always stamps it).
@@ -236,7 +260,7 @@ func (f *Forge) doRetrieve(ctx context.Context, space did.DID, c cid.Cid) (io.Re
 // via a UCAN-authorized /content/retrieve. It buffers the whole block, so it is
 // for small catalog blocks; object-body blobs use the streaming OpenBlob.
 func (f *Forge) GetBlock(ctx context.Context, space did.DID, c cid.Cid) (block.Block, error) {
-	rc, wantLen, err := f.retrieve(ctx, space, c)
+	rc, wantLen, err := f.retrieve(ctx, space, c, 0, -1)
 	if err != nil {
 		return nil, err
 	}
@@ -257,7 +281,15 @@ func (f *Forge) GetBlock(ctx context.Context, space did.DID, c cid.Cid) (block.B
 // straight off the /content/retrieve response; the caller owns the reader and
 // must Close it.
 func (f *Forge) OpenBlob(ctx context.Context, space did.DID, digest mh.Multihash) (io.ReadCloser, error) {
-	rc, _, err := f.retrieve(ctx, space, cid.NewCidV1(cid.Raw, digest))
+	rc, _, err := f.retrieve(ctx, space, cid.NewCidV1(cid.Raw, digest), 0, -1)
+	return rc, err
+}
+
+// OpenBlobRange streams stored bytes [off, off+n) of an object-body blob from
+// piri — a ranged /content/retrieve, so the decrypting read path fetches only
+// the ciphertext chunks a plaintext range needs rather than the whole blob.
+func (f *Forge) OpenBlobRange(ctx context.Context, space did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error) {
+	rc, _, err := f.retrieve(ctx, space, cid.NewCidV1(cid.Raw, digest), off, n)
 	return rc, err
 }
 

blockstore/layered.go

@@ -94,6 +94,26 @@ func (l *Layered) OpenBlob(ctx context.Context, space did.DID, digest mh.Multiha
 	return nil, ErrNotFound
 }
 
+// OpenBlobRange streams stored bytes [off, off+n) of a body blob with the
+// same tiering as OpenBlob: spool first, then the network base. A tier that
+// implements only BlobReader is served through OpenBlob with the prefix
+// discarded (OpenBlobRangeOf).
+func (l *Layered) OpenBlobRange(ctx context.Context, space did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error) {
+	if br, ok := l.spool.(BlobReader); ok {
+		rc, err := OpenBlobRangeOf(ctx, br, space, digest, off, n)
+		if err == nil {
+			return rc, nil
+		}
+		if !errors.Is(err, ErrNotFound) {
+			return nil, err
+		}
+	}
+	if br, ok := l.base.(BlobReader); ok {
+		return OpenBlobRangeOf(ctx, br, space, digest, off, n)
+	}
+	return nil, ErrNotFound
+}
+
 // layeredAsBlockstore lifts Layered into a BaseStore for the
 // CborStore wrapper, with the read's space bound in. Internal-only —
 // exists so the CBOR decoder reuses Layered's fallthrough order (and

blockstore/openblob_test.go

@@ -79,3 +79,79 @@ func TestCached_OpenBlob_BypassesCache(t *testing.T) {
 		t.Fatalf("cached OpenBlob: got %q err %v, want streamed", got, err)
 	}
 }
+
+// TestSpool_OpenBlobRange: the section reader serves exactly [off, off+n) of
+// the spooled file, and a range past the end yields a shorter stream.
+func TestSpool_OpenBlobRange(t *testing.T) {
+	ctx := context.Background()
+	sp, err := NewSpool(t.TempDir())
+	if err != nil {
+		t.Fatalf("NewSpool: %v", err)
+	}
+	data := []byte("0123456789abcdef")
+	digest, _, err := sp.WriteBlob(ctx, bytes.NewReader(data))
+	if err != nil {
+		t.Fatalf("WriteBlob: %v", err)
+	}
+
+	cases := []struct {
+		name   string
+		off, n int64
+		want   string
+	}{
+		{"interior", 4, 6, "456789"},
+		{"prefix", 0, 3, "012"},
+		{"suffix past end clamps", 12, 100, "cdef"},
+		{"empty", 5, 0, ""},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			rc, err := sp.OpenBlobRange(ctx, did.Undef, digest, c.off, c.n)
+			if err != nil {
+				t.Fatalf("OpenBlobRange: %v", err)
+			}
+			defer rc.Close()
+			got, err := io.ReadAll(rc)
+			if err != nil {
+				t.Fatalf("read: %v", err)
+			}
+			if string(got) != c.want {
+				t.Fatalf("range [%d,+%d) = %q, want %q", c.off, c.n, got, c.want)
+			}
+		})
+	}
+
+	if _, err := sp.OpenBlobRange(ctx, did.Undef, mustDigest(t, []byte("absent")), 0, 1); !errors.Is(err, ErrNotFound) {
+		t.Fatalf("missing blob err = %v, want ErrNotFound", err)
+	}
+}
+
+// TestOpenBlobRangeOf_Fallback: a tier without the BlobRangeReader capability
+// is served through OpenBlob with the prefix discarded and the tail limited.
+func TestOpenBlobRangeOf_Fallback(t *testing.T) {
+	ctx := context.Background()
+	tier := fakeBlobTier{data: []byte("0123456789abcdef")}
+
+	rc, err := OpenBlobRangeOf(ctx, tier, did.Undef, nil, 4, 6)
+	if err != nil {
+		t.Fatalf("OpenBlobRangeOf: %v", err)
+	}
+	defer rc.Close()
+	got, err := io.ReadAll(rc)
+	if err != nil {
+		t.Fatalf("read: %v", err)
+	}
+	if string(got) != "456789" {
+		t.Fatalf("fallback range = %q, want %q", got, "456789")
+	}
+}
+
+// mustDigest hashes data into the multihash form blobs are keyed by.
+func mustDigest(t *testing.T, data []byte) mh.Multihash {
+	t.Helper()
+	d, err := mh.Sum(data, mh.SHA2_256, -1)
+	if err != nil {
+		t.Fatalf("multihash: %v", err)
+	}
+	return d
+}

blockstore/spool.go

@@ -116,6 +116,21 @@ func (s *Spool) OpenBlob(_ context.Context, _ did.DID, digest mh.Multihash) (io.
 	return f, nil
 }
 
+// OpenBlobRange returns a reader over stored bytes [off, off+n) of the
+// spooled blob, or ErrNotFound — OpenBlob restricted to a section, for the
+// decrypting read path, which fetches only the ciphertext span a plaintext
+// range needs. A range past the file's end yields a shorter stream.
+func (s *Spool) OpenBlobRange(_ context.Context, _ did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error) {
+	f, err := os.Open(s.Path(digest))
+	if errors.Is(err, os.ErrNotExist) {
+		return nil, ErrNotFound
+	}
+	if err != nil {
+		return nil, fmt.Errorf("blockstore: spool open %s: %w", digest.B58String(), err)
+	}
+	return readerCloser{Reader: io.NewSectionReader(f, off, n), Closer: f}, nil
+}
+
 // GetBlock returns the blob stored under c's multihash, or ErrNotFound. A miss
 // is expected and cheap: it lets the layered read path fall through to the log
 // (for catalog blocks, which are never spooled) or the network tier (for a body
@@ -134,7 +149,8 @@ func (s *Spool) GetBlock(_ context.Context, _ did.DID, c cid.Cid) (block.Block,
 // Compile-time assertions: Spool is a raw-block read tier and the streaming
 // blob tier for object bodies.
 var (
-	_ BlockReader = (*Spool)(nil)
-	_ BlobReader  = (*Spool)(nil)
-	_ BlobWriter  = (*Spool)(nil)
+	_ BlockReader     = (*Spool)(nil)
+	_ BlobReader      = (*Spool)(nil)
+	_ BlobRangeReader = (*Spool)(nil)
+	_ BlobWriter      = (*Spool)(nil)
 )

blockstore/store.go

@@ -43,6 +43,7 @@ package blockstore
 
 import (
 	"context"
+	"fmt"
 	"io"
 
 	"github.com/fil-forge/ucantone/did"
@@ -104,6 +105,50 @@ type BlobReader interface {
 	OpenBlob(ctx context.Context, space did.DID, digest mh.Multihash) (io.ReadCloser, error)
 }
 
+// BlobRangeReader is the optional ranged counterpart of BlobReader: it
+// streams only the stored bytes [off, off+n) of a blob. The decrypting read
+// path uses it to fetch exactly the ciphertext chunks a plaintext range
+// needs instead of the whole blob. Returns ErrNotFound if the blob is not in
+// this tier; a range past the stored bytes yields a shorter stream, which
+// the caller detects. A tier without this capability is served through
+// OpenBlob with the prefix discarded (see OpenBlobRangeOf).
+type BlobRangeReader interface {
+	OpenBlobRange(ctx context.Context, space did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error)
+}
+
+// OpenBlobRangeOf opens stored bytes [off, off+n) of a blob through br,
+// using its BlobRangeReader capability when present and falling back to
+// OpenBlob with the prefix read-and-discarded (or Seek'd, for a spool file)
+// when not. It is how range-consuming callers stay correct over any tier.
+func OpenBlobRangeOf(ctx context.Context, br BlobReader, space did.DID, digest mh.Multihash, off, n int64) (io.ReadCloser, error) {
+	if rr, ok := br.(BlobRangeReader); ok {
+		return rr.OpenBlobRange(ctx, space, digest, off, n)
+	}
+	rc, err := br.OpenBlob(ctx, space, digest)
+	if err != nil {
+		return nil, err
+	}
+	if off > 0 {
+		if seeker, ok := rc.(io.Seeker); ok {
+			if _, err := seeker.Seek(off, io.SeekStart); err != nil {
+				_ = rc.Close()
+				return nil, fmt.Errorf("blockstore: seek blob to %d: %w", off, err)
+			}
+		} else if _, err := io.CopyN(io.Discard, rc, off); err != nil {
+			_ = rc.Close()
+			return nil, fmt.Errorf("blockstore: skip into blob to %d: %w", off, err)
+		}
+	}
+	return readerCloser{Reader: io.LimitReader(rc, n), Closer: rc}, nil
+}
+
+// readerCloser pairs a wrapped reader with the closer releasing its
+// underlying stream.
+type readerCloser struct {
+	io.Reader
+	io.Closer
+}
+
 // BlobWriter streams an object-body blob to local storage, computing its sha256
 // digest as it writes (never buffering the whole blob), and returns that digest
 // and the byte count. It copies r to EOF; an empty r stores nothing and returns