diff --git a/inmem/store.go b/inmem/store.go index 9264963..0568ef8 100644 --- a/inmem/store.go +++ b/inmem/store.go @@ -59,6 +59,7 @@ type MemStore struct { blobRefs map[claimKey]registry.BlobClaim intents map[string]registry.UploadIntent // keyed by string(digest) locations map[locKey]registry.BlobLocation // keyed by (space, digest) + encParams map[locKey]registry.BlobEncryptionParams // keyed by (space, digest) inclusions map[locKey]registry.BlobInclusion // keyed by (space, digest) parks map[string]registry.BlobPark // keyed by string(digest) sessions map[string]registry.MultipartSession // keyed by uploadID @@ -67,8 +68,9 @@ type MemStore struct { revCursor *registry.RevocationCursor // the single revocation_cursor row } -// claimKey / locKey are the composite map keys for the blob_refs and -// blob_locations tables (digest bytes carried as a string for comparability). +// claimKey / locKey are the composite map keys for the blob_refs and the +// (space, digest)-keyed tables — blob_locations, blob_encryption_params and +// shard_inclusions (digest bytes carried as a string for comparability). type claimKey struct { digest, bucket, objectKey, versionID string } @@ -87,6 +89,7 @@ func NewMemStore() *MemStore { blobRefs: map[claimKey]registry.BlobClaim{}, intents: map[string]registry.UploadIntent{}, locations: map[locKey]registry.BlobLocation{}, + encParams: map[locKey]registry.BlobEncryptionParams{}, inclusions: map[locKey]registry.BlobInclusion{}, parks: map[string]registry.BlobPark{}, sessions: map[string]registry.MultipartSession{}, diff --git a/inmem/stores.go b/inmem/stores.go index 9450651..51f13ff 100644 --- a/inmem/stores.go +++ b/inmem/stores.go @@ -11,35 +11,29 @@ import ( ) // In-memory implementations of the architecture's relational stores -// (registry.BlobRefStore / IntentStore / LocationStore / MultipartStore / -// GCStore), mirroring the Postgres tables so the in-process suite and -// standalone mode exercise the same write/read/delete code paths. +// (registry.BlobRefStore / IntentStore / LocationStore / EncryptionParamsStore / +// MultipartStore / GCStore), mirroring the Postgres tables so the in-process +// suite and standalone mode exercise the same write/read/delete code paths. // Compile-time assertions: *MemStore satisfies every store interface. var ( _ registry.BlobRefStore = (*MemStore)(nil) _ registry.IntentStore = (*MemStore)(nil) _ registry.LocationStore = (*MemStore)(nil) + _ registry.EncryptionParamsStore = (*MemStore)(nil) _ registry.InclusionStore = (*MemStore)(nil) _ registry.MultipartStore = (*MemStore)(nil) _ registry.GCStore = (*MemStore)(nil) _ registry.RevocationCursorStore = (*MemStore)(nil) ) -func cloneBytes(b []byte) []byte { - if b == nil { - return nil - } - return append([]byte(nil), b...) -} - // BlobRefStore =============================================================== func (m *MemStore) AddBlobClaim(_ context.Context, c registry.BlobClaim) error { m.mu.Lock() defer m.mu.Unlock() k := claimKey{string(c.Digest), c.Bucket, c.ObjectKey, c.VersionID} cp := c - cp.Digest = cloneBytes(c.Digest) + cp.Digest = bytes.Clone(c.Digest) m.blobRefs[k] = cp return nil } @@ -70,7 +64,7 @@ func (m *MemStore) PutIntent(_ context.Context, in registry.UploadIntent) error m.mu.Lock() defer m.mu.Unlock() cp := in - cp.Digest = cloneBytes(in.Digest) + cp.Digest = bytes.Clone(in.Digest) m.intents[string(in.Digest)] = cp return nil } @@ -95,7 +89,7 @@ func (m *MemStore) GetIntent(_ context.Context, digest []byte) (*registry.Upload return nil, registry.ErrNotFound } cp := in - cp.Digest = cloneBytes(in.Digest) + cp.Digest = bytes.Clone(in.Digest) return &cp, nil } @@ -108,7 +102,7 @@ func (m *MemStore) ListIntentsByState(_ context.Context, state string) ([]regist continue } cp := in - cp.Digest = cloneBytes(in.Digest) + cp.Digest = bytes.Clone(in.Digest) out = append(out, cp) } return out, nil @@ -126,9 +120,7 @@ func (m *MemStore) DeleteIntent(_ context.Context, digest []byte) error { func (m *MemStore) PutLocation(_ context.Context, loc registry.BlobLocation) error { m.mu.Lock() defer m.mu.Unlock() - cp := loc - cp.Digest = cloneBytes(loc.Digest) - m.locations[locKey{loc.Space, string(loc.Digest)}] = cp + m.locations[locKey{loc.Space, string(loc.Digest)}] = cloneLocation(loc) return nil } @@ -139,8 +131,7 @@ func (m *MemStore) GetLocation(_ context.Context, space did.DID, digest []byte) if !ok { return nil, registry.ErrNotFound } - cp := loc - cp.Digest = cloneBytes(loc.Digest) + cp := cloneLocation(loc) return &cp, nil } @@ -151,16 +142,43 @@ func (m *MemStore) DeleteLocation(_ context.Context, space did.DID, digest []byt return nil } +// EncryptionParamsStore ====================================================== + +func (m *MemStore) PutEncryptionParams(_ context.Context, params registry.BlobEncryptionParams) error { + m.mu.Lock() + defer m.mu.Unlock() + m.encParams[locKey{params.Space, string(params.Digest)}] = cloneEncryptionParams(params) + return nil +} + +func (m *MemStore) GetEncryptionParams(_ context.Context, space did.DID, digest []byte) (*registry.BlobEncryptionParams, error) { + m.mu.Lock() + defer m.mu.Unlock() + params, ok := m.encParams[locKey{space, string(digest)}] + if !ok { + return nil, registry.ErrNotFound + } + cp := cloneEncryptionParams(params) + return &cp, nil +} + +func (m *MemStore) DeleteEncryptionParams(_ context.Context, space did.DID, digest []byte) error { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.encParams, locKey{space, string(digest)}) + return nil +} + // ParkStore ================================================================== func (m *MemStore) PutPark(_ context.Context, p registry.BlobPark) error { m.mu.Lock() defer m.mu.Unlock() cp := p - cp.Digest = cloneBytes(p.Digest) - cp.AddTask = cloneBytes(p.AddTask) - cp.AcceptTask = cloneBytes(p.AcceptTask) - cp.PutInvocation = cloneBytes(p.PutInvocation) + cp.Digest = bytes.Clone(p.Digest) + cp.AddTask = bytes.Clone(p.AddTask) + cp.AcceptTask = bytes.Clone(p.AcceptTask) + cp.PutInvocation = bytes.Clone(p.PutInvocation) m.parks[string(p.Digest)] = cp return nil } @@ -173,10 +191,10 @@ func (m *MemStore) GetPark(_ context.Context, digest []byte) (*registry.BlobPark return nil, registry.ErrNotFound } cp := park - cp.Digest = cloneBytes(park.Digest) - cp.AddTask = cloneBytes(park.AddTask) - cp.AcceptTask = cloneBytes(park.AcceptTask) - cp.PutInvocation = cloneBytes(park.PutInvocation) + cp.Digest = bytes.Clone(park.Digest) + cp.AddTask = bytes.Clone(park.AddTask) + cp.AcceptTask = bytes.Clone(park.AcceptTask) + cp.PutInvocation = bytes.Clone(park.PutInvocation) return &cp, nil } @@ -194,8 +212,8 @@ func (m *MemStore) PutInclusions(_ context.Context, incs []registry.BlobInclusio defer m.mu.Unlock() for _, inc := range incs { cp := inc - cp.Digest = cloneBytes(inc.Digest) - cp.ShardDigest = cloneBytes(inc.ShardDigest) + cp.Digest = bytes.Clone(inc.Digest) + cp.ShardDigest = bytes.Clone(inc.ShardDigest) m.inclusions[locKey{inc.Space, string(inc.Digest)}] = cp } return nil @@ -209,8 +227,8 @@ func (m *MemStore) GetInclusion(_ context.Context, space did.DID, digest []byte) return nil, registry.ErrNotFound } cp := inc - cp.Digest = cloneBytes(inc.Digest) - cp.ShardDigest = cloneBytes(inc.ShardDigest) + cp.Digest = bytes.Clone(inc.Digest) + cp.ShardDigest = bytes.Clone(inc.ShardDigest) return &cp, nil } @@ -406,12 +424,28 @@ func cloneSession(s registry.MultipartSession) registry.MultipartSession { return s } +// cloneLocation deep-copies a BlobLocation's digest so the stored copy and any +// returned copy never alias the caller's slice. +func cloneLocation(loc registry.BlobLocation) registry.BlobLocation { + loc.Digest = bytes.Clone(loc.Digest) + return loc +} + +// cloneEncryptionParams deep-copies a BlobEncryptionParams' byte-slice fields +// so the stored copy and any returned copy never alias the caller's slices. +func cloneEncryptionParams(p registry.BlobEncryptionParams) registry.BlobEncryptionParams { + p.Digest = bytes.Clone(p.Digest) + p.BaseNonce = bytes.Clone(p.BaseNonce) + p.AAD = bytes.Clone(p.AAD) + return p +} + func clonePart(p registry.MultipartPart) registry.MultipartPart { - p.ETagMD5 = cloneBytes(p.ETagMD5) + p.ETagMD5 = bytes.Clone(p.ETagMD5) if p.BlobDigests != nil { ds := make([][]byte, len(p.BlobDigests)) for i, d := range p.BlobDigests { - ds[i] = cloneBytes(d) + ds[i] = bytes.Clone(d) } p.BlobDigests = ds } diff --git a/inmem/stores_test.go b/inmem/stores_test.go index 6a565c0..4f281b0 100644 --- a/inmem/stores_test.go +++ b/inmem/stores_test.go @@ -2,6 +2,8 @@ package inmem import ( "context" + "errors" + "reflect" "sync" "sync/atomic" "testing" @@ -254,6 +256,130 @@ func TestLocations_RoundTrip(t *testing.T) { } } +// feeParams returns a complete parameter set for space/digest, the shape a FEE +// envelope's row takes. +func feeParams(space did.DID, digest []byte) registry.BlobEncryptionParams { + return registry.BlobEncryptionParams{ + Space: space, + Digest: digest, + HeaderLen: 212, + BaseNonce: []byte("nonce07"), + ChunkSize: 65536, + AAD: []byte("cose-enc-structure"), + } +} + +func TestEncryptionParams_RoundTrip(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + space := testutil.RandomDID(t) + digest := []byte("enc-digest") + want := feeParams(space, digest) + + if err := m.PutEncryptionParams(ctx, want); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + got, err := m.GetEncryptionParams(ctx, space, digest) + if err != nil { + t.Fatalf("GetEncryptionParams: %v", err) + } + if !reflect.DeepEqual(*got, want) { + t.Fatalf("GetEncryptionParams = %+v, want %+v", *got, want) + } +} + +// A blob with no row is a plaintext blob, which is how the read path learns not +// to decrypt. +func TestEncryptionParams_MissingIsNotFound(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + + _, err := m.GetEncryptionParams(ctx, testutil.RandomDID(t), []byte("absent")) + if !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("GetEncryptionParams err = %v, want ErrNotFound", err) + } +} + +func TestEncryptionParams_DeleteRemovesRow(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + space := testutil.RandomDID(t) + digest := []byte("enc-digest") + if err := m.PutEncryptionParams(ctx, feeParams(space, digest)); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + + if err := m.DeleteEncryptionParams(ctx, space, digest); err != nil { + t.Fatalf("DeleteEncryptionParams: %v", err) + } + if _, err := m.GetEncryptionParams(ctx, space, digest); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("GetEncryptionParams after delete err = %v, want ErrNotFound", err) + } +} + +func TestEncryptionParams_DeleteIsIdempotent(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + + if err := m.DeleteEncryptionParams(ctx, testutil.RandomDID(t), []byte("absent")); err != nil { + t.Fatalf("DeleteEncryptionParams(absent): %v", err) + } +} + +// The store must not alias the caller's byte slices, nor let a caller reach +// back into it through a returned copy. +func TestEncryptionParams_NoSliceAliasing(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + space := testutil.RandomDID(t) + digest := []byte("enc-digest") + params := feeParams(space, digest) + + if err := m.PutEncryptionParams(ctx, params); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + params.BaseNonce[0] = 'X' + params.AAD[0] = 'X' + + got, err := m.GetEncryptionParams(ctx, space, digest) + if err != nil { + t.Fatalf("GetEncryptionParams: %v", err) + } + got.BaseNonce[0] = 'Y' + + again, err := m.GetEncryptionParams(ctx, space, digest) + if err != nil { + t.Fatalf("GetEncryptionParams (again): %v", err) + } + if !reflect.DeepEqual(*again, feeParams(space, digest)) { + t.Fatalf("stored params were aliased: %+v", *again) + } +} + +// The two tables have independent lifecycles and no cascade between them: +// deleting a location leaves the encryption parameters in place, which is why a +// caller removing a blob must delete both. +func TestEncryptionParams_IndependentOfLocation(t *testing.T) { + ctx := context.Background() + m := NewMemStore() + space := testutil.RandomDID(t) + digest := []byte("enc-digest") + if err := m.PutLocation(ctx, registry.BlobLocation{Space: space, Digest: digest, Provider: "did:piri:1", URL: "http://piri/enc", Size: 4096}); err != nil { + t.Fatalf("PutLocation: %v", err) + } + if err := m.PutEncryptionParams(ctx, feeParams(space, digest)); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + + if err := m.DeleteLocation(ctx, space, digest); err != nil { + t.Fatalf("DeleteLocation: %v", err) + } + + if _, err := m.GetEncryptionParams(ctx, space, digest); err != nil { + t.Fatalf("DeleteLocation shredded the encryption params: %v", err) + } +} + func TestRevocationCursor_UpsertRoundTrip(t *testing.T) { ctx := context.Background() m := NewMemStore() diff --git a/migrations/sql/00014_blob_encryption.sql b/migrations/sql/00014_blob_encryption.sql new file mode 100644 index 0000000..9cdda4b --- /dev/null +++ b/migrations/sql/00014_blob_encryption.sql @@ -0,0 +1,51 @@ +-- +goose Up +-- FEE (Filecoin Encryption Envelope) per-blob encryption parameters. When Ingot +-- encrypts an object's body, each body blob is stored as an independent +-- COSE/STREAM ciphertext envelope; a range GET must be able to decrypt any byte +-- span of that envelope WITHOUT first fetching and parsing its header. A row +-- here caches exactly the inputs the read path's decryptor needs, so a read +-- unwraps the CEK (held in OpenBao, under the region KEK) and goes straight to +-- a body-range fetch — no envelope-header round-trip. +-- +-- The existence of a row is what marks a blob as encrypted, so every column is +-- NOT NULL: there is no such thing as a half-populated parameter set the +-- decrypt path could not use. An unencrypted blob simply has no row. +-- +-- Deliberately a separate table from ingot.blob_locations, and deliberately +-- WITHOUT a foreign key to it. blob_locations is a reconstructible cache of the +-- indexing-service contract — every row can be re-derived from the indexer or +-- the accept receipt, and the table goes away when the topology moves to a real +-- indexer. A row here is instead the marker that a blob is encrypted, so the +-- two have independent lifecycles, and an FK would additionally force +-- location-before-parameters write ordering. +-- +-- aad holds the whole COSE Enc_structure rather than just the protected header, +-- because the structure's context string differs between a COSE_Encrypt and a +-- COSE_Encrypt0 and a bare row cannot record which form was used. The protected +-- header stays recoverable from it as element 1. +-- +-- Only what the region-KEK read path needs is cached. The COSE recipients, +-- including the tenant wrap key the insurance-recovery unwrap uses, stay in the +-- envelope header: that path is rare and out-of-band, and it reads the header +-- anyway. No key material is stored here either — the region-KEK-wrapped CEK +-- and its key version live in OpenBao. Per-blob crypto-shred is deleting the +-- key there; deleting this row only drops the cached decrypt parameters. +-- Because there is no cascade, a caller removing a blob must delete here as +-- well as from blob_locations. +CREATE TABLE ingot.blob_encryption_params ( + space text NOT NULL, + digest bytea NOT NULL, -- ciphertext blob multihash, as in blob_locations + header_len bigint NOT NULL -- encoded envelope length; the ciphertext starts at this offset + CHECK (header_len > 0), + base_nonce bytea NOT NULL -- COSE iv: the STREAM nonce seed for this blob's ciphertext + CHECK (octet_length(base_nonce) > 0), + chunk_size bigint NOT NULL -- FEE chunk size from the COSE protected header + CHECK (chunk_size > 0), + aad bytea NOT NULL -- COSE Enc_structure, bound into every chunk's GCM tag + CHECK (octet_length(aad) > 0), + created_at timestamptz NOT NULL DEFAULT now(), + PRIMARY KEY (space, digest) +); + +-- +goose Down +DROP TABLE ingot.blob_encryption_params; diff --git a/migrations/up_live_test.go b/migrations/up_live_test.go index 5ccbdec..6f2e08f 100644 --- a/migrations/up_live_test.go +++ b/migrations/up_live_test.go @@ -42,7 +42,7 @@ func TestUp_Live(t *testing.T) { want := []string{ "buckets", "segments", "segment_op_roots", - "blob_refs", "upload_intents", "blob_locations", + "blob_refs", "upload_intents", "blob_locations", "blob_encryption_params", "multipart_sessions", "multipart_parts", "gc_candidates", } for _, tbl := range want { @@ -71,4 +71,24 @@ func TestUp_Live(t *testing.T) { t.Errorf("column ingot.buckets.%s does not exist after migration", col) } } + + // blob_encryption_params (00014): every column exists and is NOT NULL — the + // presence of a row is what marks a blob as encrypted, so there is no such + // thing as a half-populated parameter set. + for _, col := range []string{ + "space", "digest", "header_len", "base_nonce", "chunk_size", "aad", + "created_at", + } { + var nullable string + err := pool.QueryRow(ctx, + `SELECT is_nullable FROM information_schema.columns + WHERE table_schema = 'ingot' AND table_name = 'blob_encryption_params' AND column_name = $1`, col).Scan(&nullable) + if err != nil { + t.Errorf("column ingot.blob_encryption_params.%s missing after migration: %v", col, err) + continue + } + if nullable != "NO" { + t.Errorf("column ingot.blob_encryption_params.%s is_nullable = %q, want NO", col, nullable) + } + } } diff --git a/registry/postgres_live_test.go b/registry/postgres_live_test.go index 97eca2d..6ea2fbf 100644 --- a/registry/postgres_live_test.go +++ b/registry/postgres_live_test.go @@ -2,7 +2,9 @@ package registry_test import ( "context" + "errors" "os" + "reflect" "testing" "time" @@ -50,8 +52,8 @@ func TestPostgresStores_Live(t *testing.T) { } if _, err := pool.Exec(ctx, `TRUNCATE ingot.blob_refs, ingot.upload_intents, ingot.blob_locations, - ingot.multipart_sessions, ingot.multipart_parts, ingot.gc_candidates, ingot.buckets, - ingot.revocation_cursor CASCADE`); err != nil { + ingot.blob_encryption_params, ingot.multipart_sessions, ingot.multipart_parts, + ingot.gc_candidates, ingot.buckets, ingot.revocation_cursor CASCADE`); err != nil { t.Fatalf("truncate: %v", err) } @@ -156,6 +158,83 @@ func TestPostgresStores_Live(t *testing.T) { } }) + // liveFEEParams is a complete parameter set whose bytea values carry + // embedded NULs and high bytes, to exercise real byte round-trips. + liveFEEParams := func(space did.DID, d []byte) registry.BlobEncryptionParams { + return registry.BlobEncryptionParams{ + Space: space, + Digest: d, + HeaderLen: 212, + BaseNonce: []byte{0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07}, + ChunkSize: 65536, + AAD: []byte{0xa1, 0x00, 0x18, 0x20}, + } + } + + t.Run("encryption params round trip", func(t *testing.T) { + space := testutil.RandomDID(t) + encDigest := []byte{0x00, 0x01, 0x02, 0xff} + want := liveFEEParams(space, encDigest) + if err := r.PutEncryptionParams(ctx, want); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + got, err := r.GetEncryptionParams(ctx, space, encDigest) + if err != nil { + t.Fatalf("GetEncryptionParams: %v", err) + } + if !reflect.DeepEqual(*got, want) { + t.Fatalf("GetEncryptionParams = %+v, want %+v", *got, want) + } + }) + + t.Run("encryption params delete removes the row", func(t *testing.T) { + space := testutil.RandomDID(t) + encDigest := []byte{0x05, 0x06} + if err := r.PutEncryptionParams(ctx, liveFEEParams(space, encDigest)); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + if err := r.DeleteEncryptionParams(ctx, space, encDigest); err != nil { + t.Fatalf("DeleteEncryptionParams: %v", err) + } + if _, err := r.GetEncryptionParams(ctx, space, encDigest); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("GetEncryptionParams after delete = %v, want ErrNotFound", err) + } + }) + + t.Run("incomplete encryption params rejected", func(t *testing.T) { + // The column constraints are the invariant: a half-populated set never + // reaches the table. + space := testutil.RandomDID(t) + d := []byte{0x77} + partial := liveFEEParams(space, d) + partial.AAD = nil + if err := r.PutEncryptionParams(ctx, partial); err == nil { + t.Fatal("PutEncryptionParams(partial) = nil, want a constraint error") + } + if _, err := r.GetEncryptionParams(ctx, space, d); !errors.Is(err, registry.ErrNotFound) { + t.Fatalf("incomplete params leaked a row: %v", err) + } + }) + + t.Run("encryption params independent of location", func(t *testing.T) { + // No foreign key and no cascade: the params outlive their location row, + // so a caller removing a blob must delete from both tables. + space := testutil.RandomDID(t) + encDigest := []byte{0x08, 0x09} + if err := r.PutEncryptionParams(ctx, liveFEEParams(space, encDigest)); err != nil { + t.Fatalf("PutEncryptionParams: %v", err) + } + if err := r.PutLocation(ctx, registry.BlobLocation{Space: space, Digest: encDigest, Provider: "did:piri", URL: "http://piri/enc", Size: 4096}); err != nil { + t.Fatalf("PutLocation: %v", err) + } + if err := r.DeleteLocation(ctx, space, encDigest); err != nil { + t.Fatalf("DeleteLocation: %v", err) + } + if _, err := r.GetEncryptionParams(ctx, space, encDigest); err != nil { + t.Fatalf("DeleteLocation shredded the encryption params: %v", err) + } + }) + t.Run("park round trip", func(t *testing.T) { park := registry.BlobPark{ Digest: digest, diff --git a/registry/stores.go b/registry/stores.go index e606f84..c7db0f4 100644 --- a/registry/stores.go +++ b/registry/stores.go @@ -12,6 +12,7 @@ import ( // architecture relies on (docs/architecture.md §5–§7, Appendix C): // the reverse reference index (blob_refs), the local-store index // (upload_intents), the local blob-location table (blob_locations), the +// per-blob FEE encryption parameters (blob_encryption_params), the // multipart session/part tables, and the gc_candidates log. The stores // are split into focused interfaces so a caller can depend on only what // it needs; *Postgres and *inmem.MemStore satisfy all of them. @@ -78,6 +79,45 @@ type BlobLocation struct { Size int64 } +// BlobEncryptionParams is one row of ingot.blob_encryption_params: the FEE +// (Filecoin Encryption Envelope) parameters a read needs to decrypt an +// encrypted blob, keyed by (Space, Digest) like the blob's location. +// +// An encrypted body blob is an independent COSE/STREAM ciphertext envelope. +// These are the cached inputs a (range) GET's decryptor needs, so a read +// unwraps the CEK and goes straight to a body-range fetch with no COSE +// envelope-header round-trip. Existence of the row is what marks a blob as +// encrypted: every field is required, and an unencrypted blob simply has no row +// (see EncryptionParamsStore). A fresh CEK per encryption event makes every +// ciphertext digest unique to one encryption, so these parameters are a 1:1 +// fact about the blob. +// +// Only what the region-KEK read path needs is cached. The COSE recipients, +// including the tenant wrap key the insurance-recovery unwrap uses, stay in the +// envelope header: that path is rare and out-of-band, and it reads the header +// anyway. No key material is stored here either — the wrapped CEK and its +// region-KEK version live in OpenBao. See docs/architecture.md §8. +type BlobEncryptionParams struct { + Space did.DID + Digest []byte + + // HeaderLen is the encoded length of the blob's COSE envelope, and so the + // offset at which its ciphertext begins. Without it a read could not locate + // byte 0 of the ciphertext without decoding the header. + HeaderLen int64 + // BaseNonce is the COSE iv: the STREAM nonce seed for this blob's ciphertext. + BaseNonce []byte + // ChunkSize is the FEE chunk size written into the COSE protected header, + // cached so the read path need not fetch and decode the envelope header. + ChunkSize int64 + // AAD is the envelope's COSE Enc_structure — the additional authenticated + // data bound into every chunk's GCM tag. Stored whole rather than as the + // protected header because the Enc_structure's context string differs + // between a COSE_Encrypt and a COSE_Encrypt0, which a bare row cannot + // record. The protected header remains recoverable from it as element 1. + AAD []byte +} + // BlobInclusion is one row of ingot.shard_inclusions: block Digest lives at // the inclusive byte range [RangeStart, RangeEnd] inside the shard CAR whose // own location is the (Space, ShardDigest) row of blob_locations. It is the @@ -179,6 +219,27 @@ type LocationStore interface { DeleteLocation(ctx context.Context, space did.DID, digest []byte) error } +// EncryptionParamsStore is the per-blob FEE encryption-parameter table +// (blob_encryption_params): what a read needs to decrypt an encrypted blob +// without fetching its envelope header. A blob has a row only when it is +// encrypted, so GetEncryptionParams returning ErrNotFound is the answer "this +// blob is stored as plaintext". +// +// It is deliberately separate from LocationStore, with no foreign key between +// the tables: the row's existence is what marks a blob as encrypted, so its +// lifecycle is independent of the reconstructible location cache. No key +// material lives here — the wrapped CEK sits in OpenBao, so per-blob +// crypto-shred means deleting the key there; Delete only drops the cached +// decrypt parameters. Put is an upsert, so re-encrypting a blob replaces its +// whole parameter set. Delete is idempotent — and because nothing cascades, +// DeleteLocation does NOT touch this table: a caller removing a blob must +// call both. +type EncryptionParamsStore interface { + PutEncryptionParams(ctx context.Context, params BlobEncryptionParams) error + GetEncryptionParams(ctx context.Context, space did.DID, digest []byte) (*BlobEncryptionParams, error) + DeleteEncryptionParams(ctx context.Context, space did.DID, digest []byte) error +} + // BlobPark is one row of ingot.blob_parks: the persistable state of a blob // that is durable on its provider but not yet accepted (multipart's deferred // conclude, §7.2). AddTask/AcceptTask are the /blob/add and diff --git a/registry/stores_postgres.go b/registry/stores_postgres.go index 49f2968..e3b7634 100644 --- a/registry/stores_postgres.go +++ b/registry/stores_postgres.go @@ -18,6 +18,7 @@ var ( _ BlobRefStore = (*Postgres)(nil) _ IntentStore = (*Postgres)(nil) _ LocationStore = (*Postgres)(nil) + _ EncryptionParamsStore = (*Postgres)(nil) _ InclusionStore = (*Postgres)(nil) _ MultipartStore = (*Postgres)(nil) _ GCStore = (*Postgres)(nil) @@ -163,7 +164,8 @@ func (r *Postgres) PutLocation(ctx context.Context, loc BlobLocation) error { func (r *Postgres) GetLocation(ctx context.Context, space did.DID, digest []byte) (*BlobLocation, error) { loc := &BlobLocation{Space: space, Digest: digest} err := r.pool.QueryRow(ctx, - `SELECT provider, url, size FROM ingot.blob_locations WHERE space = $1 AND digest = $2`, + `SELECT provider, url, size + FROM ingot.blob_locations WHERE space = $1 AND digest = $2`, space, digest).Scan(&loc.Provider, &loc.URL, &loc.Size) if errors.Is(err, pgx.ErrNoRows) { return nil, ErrNotFound @@ -183,6 +185,54 @@ func (r *Postgres) DeleteLocation(ctx context.Context, space did.DID, digest []b return nil } +// EncryptionParamsStore ====================================================== + +func (r *Postgres) PutEncryptionParams(ctx context.Context, params BlobEncryptionParams) error { + // Upsert: a re-encryption replaces the parameter set for a blob already + // stored. Every column is NOT NULL with a CHECK, so an incomplete set is + // rejected by the constraint rather than by a second check here. + _, err := r.pool.Exec(ctx, + `INSERT INTO ingot.blob_encryption_params + (space, digest, header_len, base_nonce, chunk_size, aad) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (space, digest) DO UPDATE + SET header_len = EXCLUDED.header_len, + base_nonce = EXCLUDED.base_nonce, + chunk_size = EXCLUDED.chunk_size, + aad = EXCLUDED.aad`, + params.Space, params.Digest, + params.HeaderLen, params.BaseNonce, params.ChunkSize, params.AAD) + if err != nil { + return fmt.Errorf("registry: put encryption params: %w", err) + } + return nil +} + +func (r *Postgres) GetEncryptionParams(ctx context.Context, space did.DID, digest []byte) (*BlobEncryptionParams, error) { + params := &BlobEncryptionParams{Space: space, Digest: digest} + err := r.pool.QueryRow(ctx, + `SELECT header_len, base_nonce, chunk_size, aad + FROM ingot.blob_encryption_params WHERE space = $1 AND digest = $2`, + space, digest).Scan(¶ms.HeaderLen, ¶ms.BaseNonce, + ¶ms.ChunkSize, ¶ms.AAD) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrNotFound + } + if err != nil { + return nil, fmt.Errorf("registry: get encryption params: %w", err) + } + return params, nil +} + +func (r *Postgres) DeleteEncryptionParams(ctx context.Context, space did.DID, digest []byte) error { + _, err := r.pool.Exec(ctx, + `DELETE FROM ingot.blob_encryption_params WHERE space = $1 AND digest = $2`, space, digest) + if err != nil { + return fmt.Errorf("registry: delete encryption params: %w", err) + } + return nil +} + // ParkStore ================================================================== func (r *Postgres) PutPark(ctx context.Context, p BlobPark) error {