diff --git a/go.mod b/go.mod index c6c6d05d..fce20a32 100644 --- a/go.mod +++ b/go.mod @@ -13,7 +13,7 @@ require ( github.com/exaring/otelpgx v0.12.0 github.com/fil-forge/libforge v0.0.0-20260928151559-99540866a323 github.com/fil-forge/smelt v0.0.0-20260914083257-f594cf02655c - github.com/fil-forge/swarf v0.0.1-0.20260821142121-d5d1a0a56f00 + github.com/fil-forge/swarf v0.0.1-0.20260922133653-0b2e5b2e1c6c github.com/fil-forge/ucantone v0.0.0-20260924160040-c31dec73d9b3 github.com/ipfs/go-cid v0.6.2 github.com/jackc/pgerrcode v0.0.0-20250907135507-afb5586c32a6 diff --git a/go.sum b/go.sum index ed8ce8f2..232b4e3f 100644 --- a/go.sum +++ b/go.sum @@ -166,8 +166,8 @@ github.com/fil-forge/libforge v0.0.0-20260928151559-99540866a323 h1:2hZKA/dxRN7+ github.com/fil-forge/libforge v0.0.0-20260928151559-99540866a323/go.mod h1:sKml7ScjN2ov8F7jTuBIv823EgyWGTJG6cvA8ronPk4= github.com/fil-forge/smelt v0.0.0-20260914083257-f594cf02655c h1:AEUEUEs/mDNksJIXdMgNMGJ03SHk6NlKjtDH58d1T04= github.com/fil-forge/smelt v0.0.0-20260914083257-f594cf02655c/go.mod h1:OUO2GxgUihmYfywNIfvWTQocg6jqaNiTd04cG5BbdII= -github.com/fil-forge/swarf v0.0.1-0.20260821142121-d5d1a0a56f00 h1:uSJ2YCxOJQA7PDnDiBJAOstoJcw2IPnbpssKYCihsxg= -github.com/fil-forge/swarf v0.0.1-0.20260821142121-d5d1a0a56f00/go.mod h1:k1gay5Byjloml4k1SylxhRff1rMH4LcGKpbMEDLZR1o= +github.com/fil-forge/swarf v0.0.1-0.20260922133653-0b2e5b2e1c6c h1:uzLK4I8rD2SXBTeb/qOMbnfD/oAr2RoFVP730abzzyo= +github.com/fil-forge/swarf v0.0.1-0.20260922133653-0b2e5b2e1c6c/go.mod h1:sSe3eJWknrWxnldIpU46aWI2Rvq9PwiA3nUiSFn9Rxk= github.com/fil-forge/ucantone v0.0.0-20260924160040-c31dec73d9b3 h1:Ta5ODMWR9mM4zMfRp/K1unAKvnt7f5dhehRA+YHiv7w= github.com/fil-forge/ucantone v0.0.0-20260924160040-c31dec73d9b3/go.mod h1:aX35jlUs3hnWAhigjO8z9kMPiYnkW3XOW8iVoJyz7cg= github.com/filecoin-project/go-data-segment v0.0.1 h1:1wmDxOG4ubWQm3ZC1XI5nCon5qgSq7Ra3Rb6Dbu10Gs= diff --git a/internal/testutil/swarf.go b/internal/testutil/swarf.go new file mode 100644 index 00000000..a49720fe --- /dev/null +++ b/internal/testutil/swarf.go @@ -0,0 +1,80 @@ +package testutil + +import ( + "context" + "sync" + + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/ucantone/ucan" + "github.com/ipfs/go-cid" +) + +// Revocation is one delegation a [FakeSwarf] was asked to revoke. +type Revocation struct { + Revoker did.DID + Revoked cid.Cid + Delegation ucan.Delegation +} + +// FakeSwarf stands in for the revocation service. It records every +// PublishBatch call, fails each one with Err while it is set, and runs +// OnPublish after each successful call. +type FakeSwarf struct { + Err error + OnPublish func() + + mu sync.Mutex + batches [][]Revocation +} + +func (f *FakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { + if f.Err != nil { + return f.Err + } + b := make([]Revocation, 0, len(revoked)) + for _, d := range revoked { + b = append(b, Revocation{Revoker: revoker.DID(), Revoked: d.Link(), Delegation: d}) + } + f.mu.Lock() + f.batches = append(f.batches, b) + f.mu.Unlock() + if f.OnPublish != nil { + f.OnPublish() + } + return nil +} + +// Reset forgets every recorded call. +func (f *FakeSwarf) Reset() { + f.mu.Lock() + defer f.mu.Unlock() + f.batches = nil +} + +// Calls is the number of PublishBatch calls that succeeded. +func (f *FakeSwarf) Calls() int { return len(f.Batches()) } + +// Batches returns the revocations of each successful call, in call order. +func (f *FakeSwarf) Batches() [][]Revocation { + f.mu.Lock() + defer f.mu.Unlock() + return append([][]Revocation(nil), f.batches...) +} + +// Revocations returns every revocation across all calls. +func (f *FakeSwarf) Revocations() []Revocation { + var out []Revocation + for _, b := range f.Batches() { + out = append(out, b...) + } + return out +} + +// Revoked returns the CID of every revoked delegation across all calls. +func (f *FakeSwarf) Revoked() []cid.Cid { + var out []cid.Cid + for _, r := range f.Revocations() { + out = append(out, r.Revoked) + } + return out +} diff --git a/pkg/api/access_keys_test.go b/pkg/api/access_keys_test.go index 4571b3da..6a0aab58 100644 --- a/pkg/api/access_keys_test.go +++ b/pkg/api/access_keys_test.go @@ -22,7 +22,6 @@ import ( tenantmemory "github.com/fil-forge/hilt/pkg/store/tenant/memory" "github.com/fil-forge/hilt/pkg/vault" vaultmemory "github.com/fil-forge/hilt/pkg/vault/memory" - swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/did/plc" "github.com/fil-forge/ucantone/multikey/secp256k1" @@ -36,7 +35,7 @@ import ( // HTTP layer, not revocation (see the accesskey service tests for that). type noopRevocations struct{} -func (noopRevocations) Publish(context.Context, ucan.Issuer, ucan.Delegation, ...swarfclient.PublishOption) error { +func (noopRevocations) PublishBatch(context.Context, ucan.Issuer, []ucan.Delegation) error { return nil } diff --git a/pkg/api/service/accesskey/service.go b/pkg/api/service/accesskey/service.go index 3c68925f..c851f442 100644 --- a/pkg/api/service/accesskey/service.go +++ b/pkg/api/service/accesskey/service.go @@ -26,29 +26,15 @@ import ( "github.com/fil-forge/hilt/pkg/store/principal" "github.com/fil-forge/hilt/pkg/store/tenant" "github.com/fil-forge/hilt/pkg/vault" - swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" - "github.com/fil-forge/ucantone/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" - "github.com/fil-forge/ucantone/multikey/secp256k1" "github.com/fil-forge/ucantone/ucan" - "github.com/fil-forge/ucantone/validator" "github.com/multiformats/go-multibase" "go.uber.org/zap" ) const maxNameLength = 64 -// RevocationPublisher is the subset of the revocation service (Swarf) that access -// key deletion needs. It is satisfied by [*swarfclient.Client]; the interface -// lets the logic be unit tested without a live revocation service. -type RevocationPublisher interface { - // Publish submits a /ucan/revoke invocation self-signed by revoker for the - // revoked delegation, which revoker must have issued unless a witness path is - // supplied with [swarfclient.WithWitnessPath]. - Publish(ctx context.Context, revoker ucan.Issuer, revoked ucan.Delegation, opts ...swarfclient.PublishOption) error -} - // Service implements S3 access-key operations shared by the REST handlers. type Service struct { logger *zap.Logger @@ -59,7 +45,7 @@ type Service struct { policies bucketpolicystore.Store delegations delegationstore.Store secrets vault.Vault - revocations RevocationPublisher + revocations grant.RevocationPublisher } // New constructs the access-key service. @@ -72,7 +58,7 @@ func New( policies bucketpolicystore.Store, delegations delegationstore.Store, secrets vault.Vault, - revocations RevocationPublisher, + revocations grant.RevocationPublisher, ) *Service { return &Service{ logger: logger, @@ -144,7 +130,7 @@ func (s *Service) Create(ctx context.Context, externalID, name string, permissio // Load the tenant signer up front: it is required to issue the key's // delegations and its absence is unrecoverable, so fail before creating any // state. - issuer, err := s.tenantIssuer(ctx, tenantRec.ID) + issuer, err := vault.TenantIssuer(ctx, s.secrets, tenantRec.ID) if err != nil { return accesskeystore.Record{}, "", err } @@ -276,7 +262,19 @@ func (s *Service) Create(ctx context.Context, externalID, name string, permissio } } if len(dels) > 0 { - if err := s.delegations.PutBatch(ctx, dels); err != nil { + // Store the delegations under the access key's delegation lock rather + // than with PutBatch: a policy write rotating the principal's keys + // between the key record's commit and this write has already stored + // the key's delegations from the new policy, and the ones issued above + // may predate it. The access-key DID is freshly generated, so a key + // holding anything was rotated meanwhile, and the creation fails and + // rolls back rather than store stale grants beside it. + if err := s.delegations.Replace(ctx, []did.DID{accessKeyID}, func(_ context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + if len(current[accessKeyID]) > 0 { + return nil, fmt.Errorf("access key already holds delegations: %w", store.ErrInvalidArgument) + } + return map[did.DID][]ucan.Delegation{accessKeyID: dels}, nil + }); err != nil { rollback() return accesskeystore.Record{}, "", fmt.Errorf("storing delegations: %w", err) } @@ -349,11 +347,14 @@ func (s *Service) Get(ctx context.Context, externalID, accessKeyID string) (acce } // Delete revokes an access key belonging to the tenant: publishing UCAN -// revocations for its delegations, then removing those delegations, its vault -// key, and its record. Revocations are published first so that a revocation -// service failure leaves the key intact and the call cleanly retryable — -// otherwise the delegations would live on with nothing for a verifier to check. -// A principal-bound key's marker is revoked the same way. +// revocations for its delegations, then removing its record and those +// delegations under the key's delegation lock, and its vault key. Revocations +// are published first so that a revocation service failure leaves the key +// intact and the call cleanly retryable — otherwise the delegations would live +// on with nothing for a verifier to check. The record goes under the lock so a +// policy rotation that listed the key and is waiting on the lock finds it gone +// rather than issuing it fresh delegations. Both kinds of key are deleted the +// same way. func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) error { tenantRec, err := s.tenants.GetByExternalID(ctx, externalID) if errors.Is(err, store.ErrRecordNotFound) { @@ -373,79 +374,37 @@ func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) er return fmt.Errorf("looking up access key: %w", err) } - if err := s.revokeDelegations(ctx, tenantRec.ID, id); err != nil { - return err - } - - if err := s.delegations.DeleteByAudience(ctx, id); err != nil { - return fmt.Errorf("deleting access key delegations: %w", err) - } - if err := s.secrets.Delete(ctx, vault.AccessKeyPath(tenantRec.ID, id)); err != nil { - s.logger.Warn("removing access key from vault", zap.Error(err)) - } - if err := s.accessKeys.Delete(ctx, id); err != nil { - return fmt.Errorf("deleting access key: %w", err) - } - return nil -} - -// revokeDelegations publishes a UCAN revocation for every delegation issued to -// the access key, signed by the tenant that issued them. -// -// No witness path accompanies them: the revocation service only requires one to -// prove authority over a delegation the revoker did not issue, and the tenant -// issues every delegation its access keys hold. -func (s *Service) revokeDelegations(ctx context.Context, tenantID, accessKeyID did.DID) error { - dels, err := store.Collect(ctx, func(ctx context.Context, opts store.PaginationConfig) (store.Page[ucan.Delegation], error) { - var listOpts []store.PaginationOption - if opts.Cursor != nil { - listOpts = append(listOpts, store.WithCursor(*opts.Cursor)) + // The key's delegations are revoked and removed, and its record deleted, + // under the key's delegation lock and, on Postgres, in its transaction. A + // policy write rotating the key meanwhile serializes with it, and a failure + // after the row's delete leaves the record, its delegations and its secret + // in place for a retry. The tenant key is read only when there is + // something to revoke: a key holding no delegations is deleted without it. + log := s.logger.With(zap.Stringer("tenant", tenantRec.ID), zap.Stringer("access_key", id)) + err = s.delegations.Replace(ctx, []did.DID{id}, func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + if len(current[id]) > 0 { + issuer, err := vault.TenantIssuer(ctx, s.secrets, tenantRec.ID) + if err != nil { + return nil, err + } + if err := grant.PublishRevocations(ctx, log, s.revocations, issuer, current[id]); err != nil { + return nil, err + } } - return s.delegations.ListByAudience(ctx, accessKeyID, listOpts...) + if err := s.accessKeys.Delete(ctx, id); err != nil { + return nil, fmt.Errorf("deleting access key: %w", err) + } + return nil, nil }) if err != nil { - return fmt.Errorf("listing access key delegations: %w", err) - } - if len(dels) == 0 { - return nil - } - - issuer, err := s.tenantIssuer(ctx, tenantID) - if err != nil { - return err + return fmt.Errorf("revoking access key: %w", err) } - log := s.logger.With(zap.Stringer("tenant", tenantID), zap.Stringer("access_key", accessKeyID)) - - now := ucan.UnixTimestamp(time.Now().Unix()) - for _, d := range dels { - // An expired delegation is rejected by the revocation service, and is - // unusable regardless, so revoking it is moot. - if err := validator.ValidateNotExpired(d, now); err != nil { - log.Info("skipping revocation of expired delegation", zap.Stringer("delegation", d.Link())) - continue - } - if err := s.revocations.Publish(ctx, issuer, d); err != nil { - return fmt.Errorf("publishing revocation for %s: %w", d.Link(), err) - } - log.Info("published revocation", zap.Stringer("delegation", d.Link())) + if err := s.secrets.Delete(ctx, vault.AccessKeyPath(tenantRec.ID, id)); err != nil { + s.logger.Warn("removing access key from vault", zap.Error(err)) } return nil } -// tenantIssuer loads the tenant's secp256k1 signing key from the vault and -// returns an issuer that signs as the tenant. -func (s *Service) tenantIssuer(ctx context.Context, tenantID did.DID) (ucan.Issuer, error) { - keyBytes, err := s.secrets.Read(ctx, vault.TenantKeyPath(tenantID)) - if err != nil { - return nil, fmt.Errorf("reading tenant key: %w", err) - } - signer, err := secp256k1.Decode(keyBytes) - if err != nil { - return nil, fmt.Errorf("decoding tenant key: %w", err) - } - return multikey.NewIssuer(tenantID, signer), nil -} - // bucketNamesByID returns a DID→name map for the given bucket IDs owned by the // tenant. IDs that don't resolve (e.g. a deleted bucket) are simply absent. func (s *Service) bucketNamesByID(ctx context.Context, tenantID did.DID, ids []did.DID) (map[did.DID]string, error) { diff --git a/pkg/api/service/accesskey/service_test.go b/pkg/api/service/accesskey/service_test.go index aa9699db..33cc25b5 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -13,17 +13,25 @@ import ( "github.com/fil-forge/hilt/pkg/store" accesskeystore "github.com/fil-forge/hilt/pkg/store/accesskey" accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory" + accesskeypostgres "github.com/fil-forge/hilt/pkg/store/accesskey/postgres" + "github.com/fil-forge/hilt/pkg/store/bucket" bucketmemory "github.com/fil-forge/hilt/pkg/store/bucket/memory" + bucketpostgres "github.com/fil-forge/hilt/pkg/store/bucket/postgres" bucketpolicystore "github.com/fil-forge/hilt/pkg/store/bucketpolicy" bucketpolicymemory "github.com/fil-forge/hilt/pkg/store/bucketpolicy/memory" + bucketpolicypostgres "github.com/fil-forge/hilt/pkg/store/bucketpolicy/postgres" + delegationstore "github.com/fil-forge/hilt/pkg/store/delegation" delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory" + delegationpostgres "github.com/fil-forge/hilt/pkg/store/delegation/postgres" "github.com/fil-forge/hilt/pkg/store/principal" principalmemory "github.com/fil-forge/hilt/pkg/store/principal/memory" + principalpostgres "github.com/fil-forge/hilt/pkg/store/principal/postgres" + providerpostgres "github.com/fil-forge/hilt/pkg/store/provider/postgres" "github.com/fil-forge/hilt/pkg/store/tenant" tenantmemory "github.com/fil-forge/hilt/pkg/store/tenant/memory" + tenantpostgres "github.com/fil-forge/hilt/pkg/store/tenant/postgres" "github.com/fil-forge/hilt/pkg/vault" vaultmemory "github.com/fil-forge/hilt/pkg/vault/memory" - swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" @@ -36,30 +44,7 @@ import ( "go.uber.org/zap" ) -// revocation records one published revocation. options counts the [PublishOption]s -// it was published with: Swarf's publishConfig is unexported, so the count is how -// a witness path being sent is detected. -type revocation struct { - revoker did.DID - revoked cid.Cid - options int -} - -// fakeSwarf is a stub of the revocation service, recording what it was asked to -// publish. -type fakeSwarf struct { - err error - revocations []revocation -} - -func (f *fakeSwarf) Publish(_ context.Context, revoker ucan.Issuer, revoked ucan.Delegation, opts ...swarfclient.PublishOption) error { - if f.err != nil { - return f.err - } - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: revoked.Link(), options: len(opts)}) - return nil -} - +// revocation records one published revocation. // failReadBack wraps an access-key store and fails its first Get, standing in // for a read-back that cannot see the row the call just wrote. It records the ID // it was asked for, so the test can check what the rollback cleaned up. @@ -81,31 +66,67 @@ type deps struct { svc *accesskeysvc.Service accessKeys accesskeystore.Store principals principal.Store - delegations *delegationmemory.Store - buckets *bucketmemory.Store - secrets *vaultmemory.Store - swarf *fakeSwarf + delegations delegationstore.Store + buckets bucket.Store + secrets vault.Vault + swarf *testutil.FakeSwarf tenantID did.DID bucketID did.DID bucketRoot ucan.Delegation } -// setup wires the service over memory stores with one tenant ("tenant-1") whose -// secp256k1 key is in the vault, owning one bucket ("bucket-a") that has issued -// the tenant top authority over itself — the root of every proof chain through -// the bucket, as [bucket.Service.Create] would have stored it. Its audience is the -// tenant, not an access key, so it must never be revoked along with one. The -// tenant has one principal, "alice". -// wrapAccessKeys, when given, wraps the memory access-key store the service is -// built over, so a test can make one of its methods fail. +// setup is setupWith over the memory stores. func setup(t *testing.T, wrapAccessKeys ...func(accesskeystore.Store) accesskeystore.Store) deps { + t.Helper() + return setupWith(t, backend{ + tenants: tenantmemory.New(), accessKeys: accesskeymemory.New(), principals: principalmemory.New(), + buckets: bucketmemory.New(), delegations: delegationmemory.New(), policies: bucketpolicymemory.New(), + providerID: testutil.RandomDID(t), + }, wrapAccessKeys...) +} + +// setupPostgres is setup over the Postgres stores, for what only a real +// transaction shows. It skips without Docker. +func setupPostgres(t *testing.T, wrapAccessKeys ...func(accesskeystore.Store) accesskeystore.Store) deps { + t.Helper() + pool := testutil.PostgresOrSkip(t) + providerID := testutil.RandomDID(t) + require.NoError(t, providerpostgres.New(pool).Add(t.Context(), providerID, "region-1", nil)) + return setupWith(t, backend{ + tenants: tenantpostgres.New(pool), accessKeys: accesskeypostgres.New(pool), principals: principalpostgres.New(pool), + buckets: bucketpostgres.New(pool), delegations: delegationpostgres.New(pool), policies: bucketpolicypostgres.New(pool), + providerID: providerID, + }, wrapAccessKeys...) +} + +// backend is the stores setupWith builds the service over, and the provider +// the tenant is placed with. +type backend struct { + tenants tenant.Store + accessKeys accesskeystore.Store + principals principal.Store + buckets bucket.Store + delegations delegationstore.Store + policies bucketpolicystore.Store + providerID did.DID +} + +// setupWith wires the service over b with one tenant ("tenant-1") whose +// secp256k1 key is in the vault, owning one bucket ("bucket-a") that has issued +// the tenant top authority over itself: the root of every proof chain through +// the bucket, as [bucket.Service.Create] would have stored it. Its audience is +// the tenant, not an access key, so it must never be revoked along with one. +// The tenant has one principal, "alice". wrapAccessKeys, when given, wraps the +// access-key store the service is built over, so a test can make one of its +// methods fail. +func setupWith(t *testing.T, b backend, wrapAccessKeys ...func(accesskeystore.Store) accesskeystore.Store) deps { t.Helper() ctx := t.Context() - tenants, accessKeys, principals := tenantmemory.New(), accesskeymemory.New(), principalmemory.New() - buckets, delegations, secrets := bucketmemory.New(), delegationmemory.New(), vaultmemory.New() - policies := bucketpolicymemory.New() + tenants, accessKeys, principals := b.tenants, b.accessKeys, b.principals + buckets, delegations, policies := b.buckets, b.delegations, b.policies + secrets := vaultmemory.New() - var keys accesskeystore.Store = accessKeys + keys := accessKeys for _, wrap := range wrapAccessKeys { keys = wrap(keys) } @@ -113,7 +134,7 @@ func setup(t *testing.T, wrapAccessKeys ...func(accesskeystore.Store) accesskeys signer, err := secp256k1.Generate() require.NoError(t, err) tenantID := signer.KeyDID() - require.NoError(t, tenants.Add(ctx, tenantID, "tenant-1", testutil.RandomDID(t), tenant.Active)) + require.NoError(t, tenants.Add(ctx, tenantID, "tenant-1", b.providerID, tenant.Active)) require.NoError(t, secrets.Write(ctx, vault.TenantKeyPath(tenantID), signer.Bytes())) require.NoError(t, principals.Add(ctx, tenantID, "alice")) @@ -132,7 +153,7 @@ func setup(t *testing.T, wrapAccessKeys ...func(accesskeystore.Store) accesskeys }}, nil) require.NoError(t, err) - swarf := &fakeSwarf{} + swarf := &testutil.FakeSwarf{} return deps{ svc: accesskeysvc.New(zap.NewNop(), tenants, keys, principals, buckets, policies, delegations, secrets, swarf), accessKeys: accessKeys, @@ -157,6 +178,13 @@ func TestCreate(t *testing.T) { require.NotEmpty(t, secret) require.Equal(t, "k1", rec.Name) require.Equal(t, []did.DID{d.bucketID}, rec.Buckets) + + issued, err := d.delegations.ListByAudience(ctx, rec.ID) + require.NoError(t, err) + require.NotEmpty(t, issued.Results, "the key's delegations are stored") + for _, dlg := range issued.Results { + require.Equal(t, d.bucketID, dlg.Subject()) + } }) t.Run("rejects an empty name", func(t *testing.T) { @@ -323,9 +351,9 @@ func TestCreatePrincipalBound(t *testing.T) { require.NoError(t, d.svc.Delete(ctx, "tenant-1", rec.ID.Identifier())) // The marker is the one delegation a revocation can name for the key. - require.Len(t, d.swarf.revocations, 1) - require.Equal(t, d.tenantID, d.swarf.revocations[0].revoker) - require.Equal(t, issued.Results[0].Link(), d.swarf.revocations[0].revoked) + require.Len(t, d.swarf.Revocations(), 1) + require.Equal(t, d.tenantID, d.swarf.Revocations()[0].Revoker) + require.Equal(t, issued.Results[0].Link(), d.swarf.Revocations()[0].Revoked) remaining, err := d.delegations.ListByAudience(ctx, rec.ID) require.NoError(t, err) require.Empty(t, remaining.Results) @@ -412,7 +440,7 @@ func TestListGetDelete(t *testing.T) { func TestDeleteRevokes(t *testing.T) { ctx := t.Context() - t.Run("revokes every bucket-scoped delegation, with no witness path", func(t *testing.T) { + t.Run("revokes every bucket-scoped delegation in one request, with no witness path", func(t *testing.T) { d := setup(t) // s3:PutObject maps to several commands, so the key gets several delegations. created, _, err := d.svc.Create(ctx, "tenant-1", "k1", []string{"s3:PutObject"}, []string{"bucket-a"}, nil, nil) @@ -423,14 +451,14 @@ func TestDeleteRevokes(t *testing.T) { require.NoError(t, d.svc.Delete(ctx, "tenant-1", created.ID.Identifier())) - require.Len(t, d.swarf.revocations, len(issued.Results)) + require.Len(t, d.swarf.Revocations(), len(issued.Results)) + require.Equal(t, 1, d.swarf.Calls(), "every revocation goes in one request") revoked := map[cid.Cid]bool{} - for _, r := range d.swarf.revocations { + for _, r := range d.swarf.Revocations() { // The tenant issued the delegations, so the tenant revokes them directly: // no witness path is needed to prove its authority over them. - require.Equal(t, d.tenantID, r.revoker) - require.Zero(t, r.options) - revoked[r.revoked] = true + require.Equal(t, d.tenantID, r.Revoker) + revoked[r.Revoked] = true } for _, dlg := range issued.Results { require.True(t, revoked[dlg.Link()], "delegation %s was not revoked", dlg.Link()) @@ -450,11 +478,10 @@ func TestDeleteRevokes(t *testing.T) { require.NoError(t, d.svc.Delete(ctx, "tenant-1", created.ID.Identifier())) - require.Len(t, d.swarf.revocations, 1) - r := d.swarf.revocations[0] - require.Equal(t, d.tenantID, r.revoker) - require.Equal(t, issued.Results[0].Link(), r.revoked) - require.Zero(t, r.options) + require.Len(t, d.swarf.Revocations(), 1) + r := d.swarf.Revocations()[0] + require.Equal(t, d.tenantID, r.Revoker) + require.Equal(t, issued.Results[0].Link(), r.Revoked) }) t.Run("revokes a powerline delegation when the tenant owns no bucket", func(t *testing.T) { @@ -466,7 +493,7 @@ func TestDeleteRevokes(t *testing.T) { require.NoError(t, d.buckets.Delete(ctx, d.bucketID)) require.NoError(t, d.svc.Delete(ctx, "tenant-1", created.ID.Identifier())) - require.Len(t, d.swarf.revocations, 1) + require.Len(t, d.swarf.Revocations(), 1) _, _, err = d.svc.Get(ctx, "tenant-1", created.ID.Identifier()) require.ErrorIs(t, err, accesskeysvc.ErrAccessKeyNotFound) }) @@ -480,7 +507,21 @@ func TestDeleteRevokes(t *testing.T) { // The revocation service rejects expired delegations, and they are unusable // anyway — so the key is still deleted, just with nothing published. require.NoError(t, d.svc.Delete(ctx, "tenant-1", created.ID.Identifier())) - require.Empty(t, d.swarf.revocations) + require.Empty(t, d.swarf.Revocations()) + _, _, err = d.svc.Get(ctx, "tenant-1", created.ID.Identifier()) + require.ErrorIs(t, err, accesskeysvc.ErrAccessKeyNotFound) + }) + + t.Run("deletes a key holding no delegations without the tenant key", func(t *testing.T) { + d := setup(t) + require.NoError(t, d.principals.Add(ctx, d.tenantID, "bob")) + created, _, err := d.svc.Create(ctx, "tenant-1", "laptop", nil, nil, new("bob"), nil) + require.NoError(t, err) + // With nothing to revoke, the tenant's signing key is never needed. + require.NoError(t, d.secrets.Delete(ctx, vault.TenantKeyPath(d.tenantID))) + + require.NoError(t, d.svc.Delete(ctx, "tenant-1", created.ID.Identifier())) + require.Empty(t, d.swarf.Revocations()) _, _, err = d.svc.Get(ctx, "tenant-1", created.ID.Identifier()) require.ErrorIs(t, err, accesskeysvc.ErrAccessKeyNotFound) }) @@ -489,7 +530,7 @@ func TestDeleteRevokes(t *testing.T) { d := setup(t) created, _, err := d.svc.Create(ctx, "tenant-1", "k1", []string{"s3:GetObject"}, []string{"bucket-a"}, nil, nil) require.NoError(t, err) - d.swarf.err = errors.New("swarf is down") + d.swarf.Err = errors.New("swarf is down") err = d.svc.Delete(ctx, "tenant-1", created.ID.Identifier()) require.ErrorContains(t, err, "publishing revocation") @@ -504,3 +545,51 @@ func TestDeleteRevokes(t *testing.T) { require.NoError(t, err) }) } + +// cancelAfterDelete cancels the request's ctx once the key row's delete has +// run, so the delegation write that follows it fails and the transaction rolls +// back. +type cancelAfterDelete struct { + accesskeystore.Store + cancel context.CancelFunc +} + +func (c *cancelAfterDelete) Delete(ctx context.Context, id did.DID) error { + err := c.Store.Delete(ctx, id) + c.cancel() + return err +} + +// A deletion that fails after the key row's delete leaves the row, its +// delegations and its secret for a retry: the row's delete commits with the +// delegation write or not at all. +func TestDeletePostgresRollsBackAfterRowDelete(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + wrap := &cancelAfterDelete{cancel: cancel} + d := setupPostgres(t, func(s accesskeystore.Store) accesskeystore.Store { wrap.Store = s; return wrap }) + rec, _, err := d.svc.Create(t.Context(), "tenant-1", "k1", nil, nil, new("alice"), nil) + require.NoError(t, err) + page, err := d.delegations.ListByAudience(t.Context(), rec.ID) + require.NoError(t, err) + require.NotEmpty(t, page.Results, "the key holds alice's delegations") + + require.ErrorIs(t, d.svc.Delete(ctx, "tenant-1", rec.ID.Identifier()), context.Canceled) + + _, err = d.accessKeys.Get(t.Context(), rec.ID) + require.NoError(t, err, "the row survives the failed deletion") + page, err = d.delegations.ListByAudience(t.Context(), rec.ID) + require.NoError(t, err) + require.NotEmpty(t, page.Results, "so do the delegations") + _, err = d.secrets.Read(t.Context(), vault.AccessKeyPath(d.tenantID, rec.ID)) + require.NoError(t, err, "and the secret") + + require.NoError(t, d.svc.Delete(t.Context(), "tenant-1", rec.ID.Identifier())) + _, err = d.accessKeys.Get(t.Context(), rec.ID) + require.ErrorIs(t, err, store.ErrRecordNotFound) + page, err = d.delegations.ListByAudience(t.Context(), rec.ID) + require.NoError(t, err) + require.Empty(t, page.Results) + _, err = d.secrets.Read(t.Context(), vault.AccessKeyPath(d.tenantID, rec.ID)) + require.Error(t, err) +} diff --git a/pkg/fx/revocation.go b/pkg/fx/revocation.go index 874d5435..a845196d 100644 --- a/pkg/fx/revocation.go +++ b/pkg/fx/revocation.go @@ -4,33 +4,31 @@ import ( "fmt" "net/url" - accesskeysvc "github.com/fil-forge/hilt/pkg/api/service/accesskey" "github.com/fil-forge/hilt/pkg/config" - bucketsvc "github.com/fil-forge/hilt/pkg/rpc/service/bucket" + "github.com/fil-forge/hilt/pkg/grant" "github.com/fil-forge/hilt/pkg/tracing" swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" "go.uber.org/fx" ) -// RevocationModule provides the Swarf revocation-service client, published as the -// narrow interface each consumer declares as well as the concrete type. Both the -// REST access-key service and the UCAN bucket service revoke delegations, so the -// client is shared rather than owned by either module. +// RevocationModule provides the Swarf revocation-service client, published as +// [grant.RevocationPublisher] as well as the concrete type. The REST access-key +// service and the UCAN bucket service both revoke delegations through it, so +// the client is shared rather than owned by either module. var RevocationModule = fx.Module("revocation", fx.Provide( fx.Annotate( NewRevocationClient, fx.As(fx.Self()), - fx.As(new(accesskeysvc.RevocationPublisher)), - fx.As(new(bucketsvc.RevocationPublisher)), + fx.As(new(grant.RevocationPublisher)), ), ), ) // NewRevocationClient builds the Swarf revocation-service client from // configuration. It takes no issuer: revocations are signed by the tenant that -// issued the revoked delegation, which is passed per Publish call. +// issued the revoked delegations, which is passed per publish call. func NewRevocationClient(cfg config.RevocationConfig) (*swarfclient.Client, error) { serviceID, err := did.Parse(cfg.ServiceID) if err != nil { diff --git a/pkg/grant/rotator.go b/pkg/grant/rotator.go new file mode 100644 index 00000000..86ea29bb --- /dev/null +++ b/pkg/grant/rotator.go @@ -0,0 +1,208 @@ +package grant + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + "time" + + "github.com/fil-forge/hilt/pkg/store" + accesskeystore "github.com/fil-forge/hilt/pkg/store/accesskey" + delegationstore "github.com/fil-forge/hilt/pkg/store/delegation" + "github.com/fil-forge/hilt/pkg/vault" + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/ucantone/ucan" + "github.com/fil-forge/ucantone/validator" + "go.uber.org/zap" +) + +// BatchTimeout caps one Rotate or Revoke call, which covers every key the +// write touches. A policy write rotates the keys of each principal whose +// actions changed, the wildcard fanning out to all of the tenant's, and a +// principal removal revokes each of its keys' delegations; both hold a row +// lock throughout, while a share-locked reader of that row gives up after +// [store.LockTimeout]. Each call shortens its ctx to this deadline, below the +// reader's bound (an earlier caller deadline still wins), so a large batch or +// a stalled publish cannot lock the data path out. [PublishRevocations] applies +// the same bound, so every publish made under a delegation lock, a key or +// bucket deletion's and a tenant removal's included, has it. +const BatchTimeout = 8 * time.Second + +// The batch bound must stay below the reader's; a negative difference fails +// to compile. +const _ = uint(store.LockTimeout - BatchTimeout) + +// RevocationPublisher is the subset of the revocation service (Swarf) Hilt +// needs. It is satisfied by [*swarfclient.Client]; the interface lets the logic +// be unit tested without a live revocation service. +type RevocationPublisher interface { + // PublishBatch submits one /ucan/revoke invocation per revoked delegation, + // self-signed by revoker, which must have issued every one of them, in a + // single request. A repeat for a delegation already revoked is a duplicate + // the service records once. + PublishBatch(ctx context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error +} + +// Rotator rewrites the stored delegations of principal-bound access keys as +// the bucket policies change. The delegations a change replaces are revoked +// through the revocation service, which makes the gateway drop what it cached +// for the key, and the whole rewrite runs under the keys' delegation locks so +// a concurrent rotation or deletion of one of the keys serializes behind it. +type Rotator struct { + logger *zap.Logger + delegations delegationstore.Store + accessKeys accesskeystore.Store + secrets vault.Vault + revocations RevocationPublisher +} + +// NewRotator constructs a Rotator signing revocations and delegations as the +// tenant, whose key it reads from secrets. +func NewRotator( + logger *zap.Logger, + delegations delegationstore.Store, + accessKeys accesskeystore.Store, + secrets vault.Vault, + revocations RevocationPublisher, +) *Rotator { + return &Rotator{ + logger: logger, + delegations: delegations, + accessKeys: accessKeys, + secrets: secrets, + revocations: revocations, + } +} + +// Rotate sets, for every key of each principal in actions, the key's +// delegations over bucket to those the principal's actions map to, and leaves +// its delegations over other buckets alone. The delegations replaced are +// revoked in one request before anything is stored, so a publish failure +// leaves every key unchanged; a key that held nothing over the bucket costs no +// revocation. An empty action set leaves the key with nothing over the bucket. +func (r *Rotator) Rotate(ctx context.Context, tenant, bucket did.DID, actions map[string][]string) error { + ctx, cancel := context.WithTimeout(ctx, BatchTimeout) + defer cancel() + keys, issuer, err := r.load(ctx, tenant, slices.Sorted(maps.Keys(actions))...) + if err != nil || len(keys) == 0 { + return err + } + principalOf := make(map[did.DID]string, len(keys)) + for _, key := range keys { + principalOf[key.ID] = *key.Principal + } + log := r.logger.With(zap.Stringer("tenant", tenant), zap.Stringer("bucket", bucket)) + + return r.delegations.Replace(ctx, audiences(keys), func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + var revoked []ucan.Delegation + next := make(map[did.DID][]ucan.Delegation, len(keys)) + for _, key := range keys { + // The listing above is a snapshot: a key deleted while this waited + // on the lock is left with nothing rather than issued fresh grants. + if _, err := r.accessKeys.Get(ctx, key.ID); errors.Is(err, store.ErrRecordNotFound) { + continue + } else if err != nil { + return nil, fmt.Errorf("looking up key %s: %w", key.ID, err) + } + for _, d := range current[key.ID] { + if d.Subject() == bucket { + revoked = append(revoked, d) + } else { + next[key.ID] = append(next[key.ID], d) + } + } + fresh, err := Issue(issuer, key.ID, []did.DID{bucket}, actions[principalOf[key.ID]], key.ExpiresAt) + if err != nil { + return nil, fmt.Errorf("issuing delegations of %s: %w", key.ID, err) + } + next[key.ID] = append(next[key.ID], fresh...) + } + if err := PublishRevocations(ctx, log, r.revocations, issuer, revoked); err != nil { + return nil, err + } + return next, nil + }) +} + +// Revoke revokes every delegation of every key bound to the principal, in one +// request, and leaves the keys with none, for a principal being removed. +func (r *Rotator) Revoke(ctx context.Context, tenant did.DID, principal string) error { + ctx, cancel := context.WithTimeout(ctx, BatchTimeout) + defer cancel() + keys, issuer, err := r.load(ctx, tenant, principal) + if err != nil || len(keys) == 0 { + return err + } + log := r.logger.With(zap.Stringer("tenant", tenant), zap.String("principal", principal)) + + return r.delegations.Replace(ctx, audiences(keys), func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + var all []ucan.Delegation + for _, key := range keys { + all = append(all, current[key.ID]...) + } + return nil, PublishRevocations(ctx, log, r.revocations, issuer, all) + }) +} + +// load lists the keys bound to the principals, in principal order, and, when +// there are any, the tenant's issuer. +func (r *Rotator) load(ctx context.Context, tenant did.DID, principals ...string) ([]accesskeystore.Record, ucan.Issuer, error) { + var keys []accesskeystore.Record + for _, p := range principals { + recs, err := r.accessKeys.ListByTenant(ctx, tenant, accesskeystore.WithPrincipal(p)) + if err != nil { + return nil, nil, fmt.Errorf("listing keys of principal %q: %w", p, err) + } + keys = append(keys, recs...) + } + if len(keys) == 0 { + return nil, nil, nil + } + issuer, err := vault.TenantIssuer(ctx, r.secrets, tenant) + if err != nil { + return nil, nil, err + } + return keys, issuer, nil +} + +func audiences(keys []accesskeystore.Record) []did.DID { + ids := make([]did.DID, len(keys)) + for i, key := range keys { + ids[i] = key.ID + } + return ids +} + +// PublishRevocations publishes a revocation for each unexpired delegation in +// one request, signed by the tenant that issued them. It is the callback body +// of a [delegationstore.Store.Replace] that removes or replaces what keys hold, +// so it runs while their delegations are locked, and it shortens its ctx to +// [BatchTimeout] itself: whoever calls it, a stalled publish releases the locks +// at the bound. An earlier caller deadline still wins. +func PublishRevocations(ctx context.Context, log *zap.Logger, publisher RevocationPublisher, issuer ucan.Issuer, dels []ucan.Delegation) error { + ctx, cancel := context.WithTimeout(ctx, BatchTimeout) + defer cancel() + now := ucan.UnixTimestamp(time.Now().Unix()) + live := make([]ucan.Delegation, 0, len(dels)) + for _, d := range dels { + // An expired delegation is rejected by the revocation service, and is + // unusable regardless, so revoking it is moot. + if err := validator.ValidateNotExpired(d, now); err != nil { + log.Info("skipping revocation of expired delegation", zap.Stringer("delegation", d.Link())) + continue + } + live = append(live, d) + } + if len(live) == 0 { + return nil + } + if err := publisher.PublishBatch(ctx, issuer, live); err != nil { + return fmt.Errorf("publishing revocations for %d delegations: %w", len(live), err) + } + for _, d := range live { + log.Info("published revocation", zap.Stringer("delegation", d.Link())) + } + return nil +} diff --git a/pkg/grant/rotator_test.go b/pkg/grant/rotator_test.go new file mode 100644 index 00000000..b1ae3d09 --- /dev/null +++ b/pkg/grant/rotator_test.go @@ -0,0 +1,315 @@ +package grant_test + +import ( + "context" + "errors" + htestutil "github.com/fil-forge/hilt/internal/testutil" + "github.com/fil-forge/libforge/testutil" + "testing" + "time" + + "github.com/fil-forge/hilt/pkg/grant" + "github.com/fil-forge/hilt/pkg/s3perm" + accesskeystore "github.com/fil-forge/hilt/pkg/store/accesskey" + accesskeymemory "github.com/fil-forge/hilt/pkg/store/accesskey/memory" + delegationstore "github.com/fil-forge/hilt/pkg/store/delegation" + delegationmemory "github.com/fil-forge/hilt/pkg/store/delegation/memory" + "github.com/fil-forge/hilt/pkg/vault" + vaultmemory "github.com/fil-forge/hilt/pkg/vault/memory" + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/ucantone/multikey" + "github.com/fil-forge/ucantone/multikey/secp256k1" + "github.com/fil-forge/ucantone/ucan" + "github.com/ipfs/go-cid" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +type deps struct { + rotator *grant.Rotator + delegations *delegationmemory.Store + accessKeys *accesskeymemory.Store + secrets *vaultmemory.Store + swarf *htestutil.FakeSwarf + tenant ucan.Issuer + // photos and backups are the tenant's buckets. + photos, backups did.DID + // alice holds two keys, bob one. Every key holds s3:GetObject over photos + // and s3:GetObject over backups at setup. + alice, bob []did.DID +} + +func setup(t *testing.T, expiresAt *time.Time) deps { + t.Helper() + ctx := t.Context() + accessKeys, delegations, secrets := accesskeymemory.New(), delegationmemory.New(), vaultmemory.New() + + signer, err := secp256k1.Generate() + require.NoError(t, err) + tenantID := signer.KeyDID() + tenant := multikey.NewIssuer(tenantID, signer) + require.NoError(t, secrets.Write(ctx, vault.TenantKeyPath(tenantID), signer.Bytes())) + + d := deps{ + delegations: delegations, + accessKeys: accessKeys, + secrets: secrets, + swarf: &htestutil.FakeSwarf{}, + tenant: tenant, + photos: testutil.RandomDID(t), + backups: testutil.RandomDID(t), + } + addKey := func(principal, name string) did.DID { + id := testutil.RandomDID(t) + require.NoError(t, accessKeys.Add(ctx, accesskeystore.Input{ + ID: id, Tenant: tenantID, Name: name, Principal: &principal, ExpiresAt: expiresAt, + })) + dels, err := grant.Issue(tenant, id, []did.DID{d.photos, d.backups}, []string{"s3:GetObject"}, expiresAt) + require.NoError(t, err) + require.NoError(t, delegations.PutBatch(ctx, dels)) + return id + } + d.alice = []did.DID{addKey("alice", "laptop"), addKey("alice", "phone")} + d.bob = []did.DID{addKey("bob", "laptop")} + d.rotator = grant.NewRotator(zap.NewNop(), delegations, accessKeys, secrets, d.swarf) + return d +} + +// deadlinePublisher records how long each publish had left on its deadline, +// zero when it had none. +type deadlinePublisher struct { + *htestutil.FakeSwarf + remaining []time.Duration +} + +func (p *deadlinePublisher) PublishBatch(ctx context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { + var left time.Duration + if deadline, ok := ctx.Deadline(); ok { + left = time.Until(deadline) + } + p.remaining = append(p.remaining, left) + return p.FakeSwarf.PublishBatch(ctx, revoker, revoked) +} + +// deleteOnReplace runs del once the audiences are locked, before the callback. +type deleteOnReplace struct { + delegationstore.Store + del func(context.Context) error +} + +func (s *deleteOnReplace) Replace(ctx context.Context, audiences []did.DID, next func(context.Context, map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error)) error { + return s.Store.Replace(ctx, audiences, func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + if err := s.del(ctx); err != nil { + return nil, err + } + return next(ctx, current) + }) +} + +// held returns the key's delegations over the bucket. +func (d deps) held(t *testing.T, key, bucket did.DID) []ucan.Delegation { + t.Helper() + page, err := d.delegations.ListByAudience(t.Context(), key) + require.NoError(t, err) + var out []ucan.Delegation + for _, dlg := range page.Results { + if dlg.Subject() == bucket { + out = append(out, dlg) + } + } + return out +} + +func commands(dels []ucan.Delegation) []string { + var out []string + for _, d := range dels { + out = append(out, d.Command().String()) + } + return out +} + +func commandsFor(actions ...string) []string { + var out []string + for _, c := range s3perm.CommandsFor(actions...) { + out = append(out, c.String()) + } + return out +} + +func links(dels []ucan.Delegation) []cid.Cid { + var out []cid.Cid + for _, d := range dels { + out = append(out, d.Link()) + } + return out +} + +func TestRotate(t *testing.T) { + ctx := t.Context() + + t.Run("rewrites each key's delegations over the bucket and revokes the old ones in one request", func(t *testing.T) { + d := setup(t, nil) + var old []ucan.Delegation + for _, key := range d.alice { + old = append(old, d.held(t, key, d.photos)...) + } + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"alice": {"s3:PutObject"}})) + + require.Len(t, d.swarf.Batches(), 1, "every revocation of the write goes in one request") + require.Equal(t, d.tenant.DID(), d.swarf.Batches()[0][0].Revoker) + require.ElementsMatch(t, links(old), d.swarf.Revoked()) + for _, key := range d.alice { + held := d.held(t, key, d.photos) + require.ElementsMatch(t, commandsFor("s3:PutObject"), commands(held)) + for _, m := range held { + require.Equal(t, d.tenant.DID(), m.Issuer()) + require.Equal(t, key, m.Audience()) + require.Nil(t, m.Expiration()) + } + require.ElementsMatch(t, commandsFor("s3:GetObject"), commands(d.held(t, key, d.backups)), "other buckets are untouched") + } + require.ElementsMatch(t, commandsFor("s3:GetObject"), commands(d.held(t, d.bob[0], d.photos)), "bob's key is untouched") + }) + + t.Run("the fresh delegations expire with the key", func(t *testing.T) { + exp := time.Now().Add(time.Hour).UTC().Truncate(time.Second) + d := setup(t, &exp) + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"bob": {"s3:PutObject"}})) + + held := d.held(t, d.bob[0], d.photos) + require.NotEmpty(t, held) + for _, m := range held { + require.NotNil(t, m.Expiration()) + require.Equal(t, ucan.UnixTimestamp(exp.Unix()), *m.Expiration()) + } + }) + + t.Run("a first grant over a bucket publishes nothing", func(t *testing.T) { + d := setup(t, nil) + videos := testutil.RandomDID(t) + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), videos, map[string][]string{"alice": {"s3:GetObject"}, "bob": {"s3:GetObject"}})) + + require.Empty(t, d.swarf.Batches()) + for _, key := range append(d.alice, d.bob...) { + require.ElementsMatch(t, commandsFor("s3:GetObject"), commands(d.held(t, key, videos))) + } + }) + + t.Run("an empty action set leaves the key with nothing over the bucket", func(t *testing.T) { + d := setup(t, nil) + old := d.held(t, d.bob[0], d.photos) + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"bob": nil})) + + require.ElementsMatch(t, links(old), d.swarf.Revoked()) + require.Empty(t, d.held(t, d.bob[0], d.photos)) + require.NotEmpty(t, d.held(t, d.bob[0], d.backups)) + }) + + t.Run("a publish failure leaves every key unchanged", func(t *testing.T) { + d := setup(t, nil) + d.swarf.Err = errors.New("swarf is down") + before := map[did.DID][]cid.Cid{} + for _, key := range append(d.alice, d.bob...) { + before[key] = links(d.held(t, key, d.photos)) + } + + err := d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"alice": {"s3:PutObject"}, "bob": nil}) + require.ErrorContains(t, err, "swarf is down") + for key, want := range before { + require.Equal(t, want, links(d.held(t, key, d.photos))) + } + }) + + t.Run("an expired delegation is replaced without a revocation", func(t *testing.T) { + exp := time.Now().Add(-time.Hour) + d := setup(t, &exp) + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"bob": {"s3:PutObject"}})) + + require.Empty(t, d.swarf.Batches()) + require.ElementsMatch(t, commandsFor("s3:PutObject"), commands(d.held(t, d.bob[0], d.photos))) + }) + + t.Run("a key deleted while the rotation waited on the lock gets nothing", func(t *testing.T) { + d := setup(t, nil) + gone, kept := d.alice[0], d.alice[1] + // The key goes between the rotation's listing and its lock, as a + // concurrent removal holding the lock first would have it. + dels := &deleteOnReplace{Store: d.delegations, del: func(ctx context.Context) error { return d.accessKeys.Delete(ctx, gone) }} + rotator := grant.NewRotator(zap.NewNop(), dels, d.accessKeys, d.secrets, d.swarf) + require.NoError(t, rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"alice": {"s3:PutObject"}})) + + require.Empty(t, d.held(t, gone, d.photos)) + require.ElementsMatch(t, commandsFor("s3:PutObject"), commands(d.held(t, kept, d.photos))) + }) + + t.Run("a principal without keys is a no-op", func(t *testing.T) { + d := setup(t, nil) + require.NoError(t, d.rotator.Rotate(ctx, d.tenant.DID(), d.photos, map[string][]string{"ghost": {"s3:GetObject"}})) + require.Empty(t, d.swarf.Batches()) + }) +} + +func TestRevoke(t *testing.T) { + ctx := t.Context() + + t.Run("revokes every delegation of each key in one request and leaves the keys with none", func(t *testing.T) { + d := setup(t, nil) + var old []ucan.Delegation + for _, key := range d.alice { + page, err := d.delegations.ListByAudience(ctx, key) + require.NoError(t, err) + old = append(old, page.Results...) + } + require.NoError(t, d.rotator.Revoke(ctx, d.tenant.DID(), "alice")) + + require.Len(t, d.swarf.Batches(), 1) + require.Equal(t, d.tenant.DID(), d.swarf.Batches()[0][0].Revoker) + require.ElementsMatch(t, links(old), d.swarf.Revoked()) + for _, key := range d.alice { + page, err := d.delegations.ListByAudience(ctx, key) + require.NoError(t, err) + require.Empty(t, page.Results) + } + require.NotEmpty(t, d.held(t, d.bob[0], d.photos)) + }) + + t.Run("a publish failure leaves the delegations in place", func(t *testing.T) { + d := setup(t, nil) + d.swarf.Err = errors.New("swarf is down") + + require.ErrorContains(t, d.rotator.Revoke(ctx, d.tenant.DID(), "alice"), "swarf is down") + for _, key := range d.alice { + require.NotEmpty(t, d.held(t, key, d.photos)) + } + }) +} + +// The rotator bounds each rotation and revocation at BatchTimeout itself, so a +// caller without a deadline cannot hold the delegation locks behind a stalled +// publish. +func TestRotatorPublishesUnderBatchTimeout(t *testing.T) { + d := setup(t, nil) + pub := &deadlinePublisher{FakeSwarf: d.swarf} + rotator := grant.NewRotator(zap.NewNop(), d.delegations, d.accessKeys, d.secrets, pub) + + require.NoError(t, rotator.Rotate(context.Background(), d.tenant.DID(), d.photos, map[string][]string{"alice": {"s3:ListBucket"}})) + require.NoError(t, rotator.Revoke(context.Background(), d.tenant.DID(), "bob")) + require.Len(t, pub.remaining, 2) + for _, left := range pub.remaining { + require.Greater(t, left, time.Duration(0)) + require.LessOrEqual(t, left, grant.BatchTimeout) + } +} + +// PublishRevocations bounds its own publish, so a caller that holds a +// delegation lock with no deadline of its own still releases it at BatchTimeout. +func TestPublishRevocationsUnderBatchTimeout(t *testing.T) { + d := setup(t, nil) + pub := &deadlinePublisher{FakeSwarf: d.swarf} + held := d.held(t, d.alice[0], d.photos) + require.NotEmpty(t, held) + require.NoError(t, grant.PublishRevocations(context.Background(), zap.NewNop(), pub, d.tenant, held)) + require.Len(t, pub.remaining, 1) + require.Greater(t, pub.remaining[0], time.Duration(0)) + require.LessOrEqual(t, pub.remaining[0], grant.BatchTimeout) +} diff --git a/pkg/rpc/service/auth/auth.go b/pkg/rpc/service/auth/auth.go index 15a687be..ba916621 100644 --- a/pkg/rpc/service/auth/auth.go +++ b/pkg/rpc/service/auth/auth.go @@ -22,7 +22,6 @@ import ( "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" - "github.com/fil-forge/ucantone/multikey/secp256k1" "github.com/fil-forge/ucantone/ucan" "github.com/multiformats/go-multibase" "go.uber.org/zap" @@ -270,15 +269,7 @@ func (a *Authorizer) resolveBucket(ctx context.Context, log *zap.Logger, akRec a // returns an issuer that signs as the tenant — used to act on the tenant's // behalf (e.g. provisioning a bucket's space with Sprue). func (a *Authorizer) TenantIssuer(ctx context.Context, tenantID did.DID) (ucan.Issuer, error) { - keyBytes, err := a.secrets.Read(ctx, vault.TenantKeyPath(tenantID)) - if err != nil { - return nil, fmt.Errorf("reading tenant key: %w", err) - } - signer, err := secp256k1.Decode(keyBytes) - if err != nil { - return nil, fmt.Errorf("decoding tenant key: %w", err) - } - return multikey.NewIssuer(tenantID, signer), nil + return vault.TenantIssuer(ctx, a.secrets, tenantID) } // AccessKeySigner reads the access key's ed25519 private key from the vault. diff --git a/pkg/rpc/service/bucket/service.go b/pkg/rpc/service/bucket/service.go index a543271f..9add5cad 100644 --- a/pkg/rpc/service/bucket/service.go +++ b/pkg/rpc/service/bucket/service.go @@ -16,6 +16,7 @@ import ( "time" "github.com/fil-forge/hilt/pkg/client/upload" + "github.com/fil-forge/hilt/pkg/grant" "github.com/fil-forge/hilt/pkg/rpc/service/auth" "github.com/fil-forge/hilt/pkg/sigv4" "github.com/fil-forge/hilt/pkg/store" @@ -27,14 +28,12 @@ import ( s3 "github.com/fil-forge/libforge/commands/s3" s3bkt "github.com/fil-forge/libforge/commands/s3/bucket" s3req "github.com/fil-forge/libforge/commands/s3/request" - swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" "github.com/fil-forge/ucantone/ucan" "github.com/fil-forge/ucantone/ucan/command" "github.com/fil-forge/ucantone/ucan/delegation" - "github.com/fil-forge/ucantone/validator" "github.com/ipfs/go-cid" "go.uber.org/zap" ) @@ -52,16 +51,6 @@ type UploadClient interface { UseRoutingPolicy(ctx context.Context, space did.DID, policy *did.DID, opts ...upload.MethodOption) error } -// RevocationPublisher is the subset of the revocation service (Swarf) the bucket -// operations need. It is satisfied by [*swarfclient.Client]; the interface lets the -// logic be unit tested without a live revocation service. -type RevocationPublisher interface { - // Publish submits a /ucan/revoke invocation self-signed by revoker for the - // revoked delegation, which revoker must have issued unless a witness path is - // supplied with [swarfclient.WithWitnessPath]. - Publish(ctx context.Context, revoker ucan.Issuer, revoked ucan.Delegation, opts ...swarfclient.PublishOption) error -} - // Service implements the S3 bucket operations shared by the UCAN command handlers. type Service struct { logger *zap.Logger @@ -72,7 +61,7 @@ type Service struct { tenants tenantstore.Store policies bucketpolicystore.Store uploads UploadClient - revocations RevocationPublisher + revocations grant.RevocationPublisher } // New constructs the bucket service. @@ -85,7 +74,7 @@ func New( tenants tenantstore.Store, policies bucketpolicystore.Store, uploads UploadClient, - revocations RevocationPublisher, + revocations grant.RevocationPublisher, ) *Service { return &Service{ logger: logger, @@ -333,10 +322,11 @@ func (s *Service) Delete(ctx context.Context, issuer did.DID, args *s3bkt.Delete return &s3bkt.DeleteOK{}, nil } -// revokeDelegations publishes a UCAN revocation for the delegations over the -// bucket that the tenant issued — the grants held by its access keys — signed by -// the tenant. No witness path accompanies them: the revocation service only -// requires one to prove authority over a delegation the revoker did not issue. +// revokeDelegations publishes, in one request, a UCAN revocation for every +// delegation over the bucket that the tenant issued — the grants held by its +// access keys, service and principal-bound alike — signed by the tenant. No +// witness path accompanies them: the revocation service only requires one to +// prove authority over a delegation the revoker did not issue. // // Powerline delegations are not affected: they carry no subject because they grant // access to every bucket the tenant owns, so deleting one bucket must not revoke @@ -354,7 +344,7 @@ func (s *Service) revokeDelegations(ctx context.Context, tenantID, bucketID did. } log := s.logger.With(zap.Stringer("tenant", tenantID), zap.Stringer("bucket", bucketID)) - now := ucan.UnixTimestamp(time.Now().Unix()) + issued := make([]ucan.Delegation, 0, len(dels)) for _, d := range dels { // The bucket→tenant root is signed by the bucket's own key, which was // discarded once it had issued the root, and a root is the start of its own @@ -365,18 +355,9 @@ func (s *Service) revokeDelegations(ctx context.Context, tenantID, bucketID did. zap.Stringer("delegation", d.Link()), zap.Stringer("issuer", d.Issuer())) continue } - // An expired delegation is rejected by the revocation service, and is - // unusable regardless, so revoking it is moot. - if err := validator.ValidateNotExpired(d, now); err != nil { - log.Info("skipping revocation of expired delegation", zap.Stringer("delegation", d.Link())) - continue - } - if err := s.revocations.Publish(ctx, revoker, d); err != nil { - return fmt.Errorf("publishing revocation for %s: %w", d.Link(), err) - } - log.Info("published revocation", zap.Stringer("delegation", d.Link())) + issued = append(issued, d) } - return nil + return grant.PublishRevocations(ctx, log, s.revocations, revoker, issued) } // List authorizes the request (which also verifies the access key holds the diff --git a/pkg/rpc/service/bucket/service_test.go b/pkg/rpc/service/bucket/service_test.go index 94d77a5e..0b55a463 100644 --- a/pkg/rpc/service/bucket/service_test.go +++ b/pkg/rpc/service/bucket/service_test.go @@ -30,7 +30,6 @@ import ( s3 "github.com/fil-forge/libforge/commands/s3" s3bkt "github.com/fil-forge/libforge/commands/s3/bucket" "github.com/fil-forge/libforge/testutil" - swarfclient "github.com/fil-forge/swarf/pkg/client" "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" @@ -38,7 +37,6 @@ import ( "github.com/fil-forge/ucantone/ucan" "github.com/fil-forge/ucantone/ucan/command" "github.com/fil-forge/ucantone/ucan/delegation" - "github.com/ipfs/go-cid" "github.com/multiformats/go-multibase" "github.com/stretchr/testify/require" "go.uber.org/zap" @@ -126,7 +124,7 @@ func TestCreate(t *testing.T) { require.NoError(t, err) require.NoError(t, delegations.PutBatch(ctx, []ucan.Delegation{powerline})) az := auth.NewAuthorizer(zap.NewNop(), accessKeys, tenants, providers, buckets, secrets) - return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, sprue, &fakeSwarf{}), buckets + return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, sprue, &htestutil.FakeSwarf{}), buckets } args := func() *s3bkt.CreateArguments { @@ -234,43 +232,14 @@ func (f failingListDelegations) ListByAudience(context.Context, did.DID, ...stor return store.Page[ucan.Delegation]{}, f.err } -// revocation records one published revocation. options counts the [PublishOption]s -// it was published with: Swarf's publishConfig is unexported, so the count is how -// a witness path being sent is detected. -type revocation struct { - revoker did.DID - revoked cid.Cid - options int -} - -// fakeSwarf is a stub of the revocation service, recording what it was asked to -// publish. -type fakeSwarf struct { - err error - revocations []revocation - // onPublish, when set, runs once before the first publish is recorded. - onPublish func() -} - -func (f *fakeSwarf) Publish(_ context.Context, revoker ucan.Issuer, revoked ucan.Delegation, opts ...swarfclient.PublishOption) error { - if f.err != nil { - return f.err - } - if f.onPublish != nil { - f.onPublish() - f.onPublish = nil - } - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: revoked.Link(), options: len(opts)}) - return nil -} - +// revocation records one published revocation. // deleteDeps is the world a Delete subtest operates on. type deleteDeps struct { svc *bucketsvc.Service buckets *bucketmemory.Store delegations *delegationmemory.Store policies bucketpolicystore.Store - swarf *fakeSwarf + swarf *htestutil.FakeSwarf tenantID did.DID bucketID did.DID root ucan.Delegation // bucket→tenant, signed by the bucket's discarded key @@ -317,7 +286,7 @@ func TestDelete(t *testing.T) { require.NoError(t, err) require.NoError(t, delegations.PutBatch(ctx, []ucan.Delegation{root, grant})) az := auth.NewAuthorizer(zap.NewNop(), accessKeys, tenants, providers, buckets, secrets) - swarf := &fakeSwarf{} + swarf := &htestutil.FakeSwarf{} policies := bucketpolicymemory.New(bucketpolicymemory.WithBuckets(buckets.Has)) return deleteDeps{ svc: bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, sprue, swarf), @@ -354,7 +323,7 @@ func TestDelete(t *testing.T) { d := setup(t, []string{"s3:DeleteBucket"}, &fakeSprue{empty: true}) deleted, written := htestutil.RequireWaitsForWriter(t, func(entered chan<- struct{}, release <-chan struct{}) error { - d.swarf.onPublish = func() { close(entered); <-release } + d.swarf.OnPublish = func() { close(entered); <-release } _, err := d.svc.Delete(ctx, providerID, del(bucketName)) return err }, @@ -387,12 +356,12 @@ func TestDelete(t *testing.T) { _, err := d.svc.Delete(ctx, providerID, del(bucketName)) require.NoError(t, err) - require.Len(t, d.swarf.revocations, 1) - r := d.swarf.revocations[0] + require.Len(t, d.swarf.Revocations(), 1) + require.Equal(t, 1, d.swarf.Calls()) + r := d.swarf.Revocations()[0] // The tenant issued the grant, so the tenant revokes it directly. - require.Equal(t, tenantID, r.revoker) - require.Equal(t, d.grant.Link(), r.revoked) - require.Zero(t, r.options) + require.Equal(t, tenantID, r.Revoker) + require.Equal(t, d.grant.Link(), r.Revoked) }) t.Run("does not revoke the bucket root", func(t *testing.T) { @@ -402,8 +371,8 @@ func TestDelete(t *testing.T) { // The root was signed by the bucket's discarded key: the tenant is only its // audience, so no revocation it could sign would be accepted. - for _, r := range d.swarf.revocations { - require.NotEqual(t, d.root.Link(), r.revoked) + for _, r := range d.swarf.Revocations() { + require.NotEqual(t, d.root.Link(), r.Revoked) } }) @@ -415,14 +384,14 @@ func TestDelete(t *testing.T) { // anyway — so the bucket is still deleted, just with nothing published. _, err := d.svc.Delete(ctx, providerID, del(bucketName)) require.NoError(t, err) - require.Empty(t, d.swarf.revocations) + require.Empty(t, d.swarf.Revocations()) _, err = d.buckets.GetByName(ctx, bucketName) require.ErrorIs(t, err, store.ErrRecordNotFound) }) t.Run("a revocation failure leaves the bucket intact", func(t *testing.T) { d := setup(t, []string{"s3:DeleteBucket"}, &fakeSprue{empty: true}) - d.swarf.err = errors.New("swarf is down") + d.swarf.Err = errors.New("swarf is down") _, err := d.svc.Delete(ctx, providerID, del(bucketName)) require.ErrorContains(t, err, "publishing revocation") @@ -453,7 +422,7 @@ func TestDelete(t *testing.T) { require.ErrorIs(t, err, bucketsvc.ErrBucketNotEmpty) _, err = d.buckets.GetByName(ctx, bucketName) require.NoError(t, err) - require.Empty(t, d.swarf.revocations, "nothing is revoked when the delete is refused") + require.Empty(t, d.swarf.Revocations(), "nothing is revoked when the delete is refused") }) t.Run("propagates a SpaceEmpty error", func(t *testing.T) { @@ -483,7 +452,7 @@ func TestList(t *testing.T) { require.NoError(t, accessKeys.Add(ctx, accesskey.Input{ID: akDID, Tenant: tenantID, Name: "k1", Permissions: perms})) require.NoError(t, secrets.Write(ctx, vault.AccessKeyPath(tenantID, akDID), signer.Bytes())) az := auth.NewAuthorizer(zap.NewNop(), accessKeys, tenants, providers, buckets, secrets) - return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, &fakeSprue{}, &fakeSwarf{}), buckets, tenantID + return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, &fakeSprue{}, &htestutil.FakeSwarf{}), buckets, tenantID } // listArgs presigns a ListBuckets request; extra ListBuckets query params @@ -601,7 +570,7 @@ func TestInfo(t *testing.T) { require.NoError(t, delegations.PutBatch(ctx, []ucan.Delegation{root, grant})) // Info does not use the authorizer; a minimal one over the same stores suffices. az := auth.NewAuthorizer(zap.NewNop(), accessKeys, tenants, providermemory.New(), buckets, vaultmemory.New()) - return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, &fakeSprue{}, &fakeSwarf{}), accessKeys + return bucketsvc.New(zap.NewNop(), az, buckets, delegations, accessKeys, tenants, policies, &fakeSprue{}, &htestutil.FakeSwarf{}), accessKeys } t.Run("returns the bucket, permissions, and delegation chain", func(t *testing.T) { diff --git a/pkg/store/delegation/delegation.go b/pkg/store/delegation/delegation.go index 4efe8d60..d3e332c4 100644 --- a/pkg/store/delegation/delegation.go +++ b/pkg/store/delegation/delegation.go @@ -2,6 +2,7 @@ package delegation import ( "context" + "fmt" "github.com/fil-forge/hilt/pkg/store" "github.com/fil-forge/ucantone/did" @@ -10,7 +11,8 @@ import ( ) type Store interface { - // DeleteByAudience removes all delegation records for a given audience. + // DeleteByAudience removes all delegation records for a given audience, + // under that audience's lock. DeleteByAudience(ctx context.Context, audience did.DID) error // DeleteBySubject removes all delegation records for a given subject. // The undefined DID (powerline) is not removed by this method: it returns @@ -31,7 +33,40 @@ type Store interface { // of the next Delegation. It returns [store.ErrInvalidArgument] for an undef // subject. ProofChain(ctx context.Context, aud did.DID, cmd ucan.Command, sub did.DID) ([]ucan.Delegation, []cid.Cid, error) - // PutBatch stores a batch of delegation records. It returns - // [store.ErrInvalidArgument] if the batch contains a nil delegation. + // PutBatch stores a batch of delegation records. It takes the lock of + // every audience the batch touches, so it does not interleave with a + // Replace of any of them. It returns [store.ErrInvalidArgument] if the + // batch contains a nil delegation. PutBatch(ctx context.Context, delegation []ucan.Delegation) error + // Replace atomically swaps the delegations of every listed audience in one + // transaction: it locks the audiences, hands next the current set of each, + // deletes them, stores what next returns, and commits. An audience missing + // from next's result, or mapped to nil, is left with none; an error from + // next rolls everything back and is returned. An audience holding nothing + // is handed a nil set, so Replace also stores a first delegation. next runs + // while the audiences are locked and must not call back into the store. It + // returns [store.ErrInvalidArgument] if next returns a nil delegation or + // one whose audience is not the audience it is filed under; entries for + // audiences not listed are ignored. On Postgres next's + // ctx carries the transaction, so a store that opens with pglock.Begin + // writes inside it. The lock wait is bounded at [store.LockTimeout] and + // returns [store.ErrLockTimeout] when it runs out. + Replace(ctx context.Context, audiences []did.DID, next func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error)) error +} + +// CheckReplacement refuses a replacement that holds a nil delegation or one +// filed under an audience other than its own, which would be written under an +// audience the call did not lock. Both backends call it after next returns. +func CheckReplacement(audiences []did.DID, replacement map[did.DID][]ucan.Delegation) error { + for _, aud := range audiences { + for _, d := range replacement[aud] { + if d == nil { + return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) + } + if d.Audience() != aud { + return fmt.Errorf("delegation %s is for %s, filed under %s: %w", d.Link(), d.Audience(), aud, store.ErrInvalidArgument) + } + } + } + return nil } diff --git a/pkg/store/delegation/delegation_test.go b/pkg/store/delegation/delegation_test.go index dfd8c90f..8521b41a 100644 --- a/pkg/store/delegation/delegation_test.go +++ b/pkg/store/delegation/delegation_test.go @@ -2,8 +2,10 @@ package delegation_test import ( "context" + "errors" "runtime" "testing" + "time" htestutil "github.com/fil-forge/hilt/internal/testutil" "github.com/fil-forge/hilt/pkg/store" @@ -113,6 +115,177 @@ func TestDelegationStore(t *testing.T) { require.Empty(t, page.Results) }) + t.Run("Replace hands next the current set and stores its result", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + cmd := command.MustParse("/test/run") + old := makeDelegation(t, issuer, audience, issuer.DID(), cmd) + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{old})) + fresh := makeDelegation(t, issuer, audience, issuer.DID(), cmd) + + var seen []ucan.Delegation + require.NoError(t, replaceOne(t.Context(), s, audience, func(_ context.Context, current []ucan.Delegation) ([]ucan.Delegation, error) { + seen = current + return []ucan.Delegation{fresh}, nil + })) + + require.Len(t, seen, 1) + require.Equal(t, old.Link(), seen[0].Link()) + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Len(t, page.Results, 1) + require.Equal(t, fresh.Link(), page.Results[0].Link()) + }) + + t.Run("Replace with next returning nil leaves the audience with none", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + cmd := command.MustParse("/test/run") + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{ + makeDelegation(t, issuer, audience, issuer.DID(), cmd), + makeDelegation(t, issuer, audience, issuer.DID(), cmd), + })) + + require.NoError(t, replaceOne(t.Context(), s, audience, func(_ context.Context, current []ucan.Delegation) ([]ucan.Delegation, error) { + require.Len(t, current, 2) + return nil, nil + })) + + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Empty(t, page.Results) + }) + + t.Run("Replace rolls back when next errors", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + old := makeDelegation(t, issuer, audience, issuer.DID(), command.MustParse("/test/run")) + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{old})) + boom := errors.New("publish failed") + + err := replaceOne(t.Context(), s, audience, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + return nil, boom + }) + require.ErrorIs(t, err, boom) + + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Len(t, page.Results, 1) + require.Equal(t, old.Link(), page.Results[0].Link()) + }) + + t.Run("Replace stores next's result for an audience holding nothing", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + fresh := makeDelegation(t, issuer, audience, issuer.DID(), command.MustParse("/test/run")) + + called := false + require.NoError(t, replaceOne(t.Context(), s, audience, func(_ context.Context, current []ucan.Delegation) ([]ucan.Delegation, error) { + called = true + require.Empty(t, current) + return []ucan.Delegation{fresh}, nil + })) + + require.True(t, called, "next is told the audience holds nothing") + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Len(t, page.Results, 1) + require.Equal(t, fresh.Link(), page.Results[0].Link()) + }) + + t.Run("PutBatch and DeleteByAudience wait for an in-flight Replace", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + cmd := command.MustParse("/test/run") + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{ + makeDelegation(t, issuer, audience, issuer.DID(), cmd), + })) + fresh := makeDelegation(t, issuer, audience, issuer.DID(), cmd) + other := makeDelegation(t, issuer, audience, issuer.DID(), cmd) + + entered, release := make(chan struct{}), make(chan struct{}) + replaced := make(chan error, 1) + go func() { + replaced <- replaceOne(t.Context(), s, audience, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + close(entered) + <-release + return []ucan.Delegation{fresh}, nil + }) + }() + <-entered + + put, deleted := make(chan error, 1), make(chan error, 1) + go func() { put <- s.PutBatch(t.Context(), []ucan.Delegation{other}) }() + go func() { deleted <- s.DeleteByAudience(t.Context(), audience) }() + + select { + case err := <-put: + t.Fatalf("PutBatch completed while Replace held the audience: %v", err) + case err := <-deleted: + t.Fatalf("DeleteByAudience completed while Replace held the audience: %v", err) + case <-time.After(200 * time.Millisecond): + } + + close(release) + require.NoError(t, <-replaced) + require.NoError(t, <-put) + require.NoError(t, <-deleted) + }) + + t.Run("Replace swaps several audiences in one call", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + a, b := testutil.RandomDID(t), testutil.RandomDID(t) + cmd := command.MustParse("/test/run") + oldA := makeDelegation(t, issuer, a, issuer.DID(), cmd) + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{oldA})) + freshA, freshB := makeDelegation(t, issuer, a, issuer.DID(), cmd), makeDelegation(t, issuer, b, issuer.DID(), cmd) + + require.NoError(t, s.Replace(t.Context(), []did.DID{a, b}, func(_ context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + require.Len(t, current[a], 1) + require.Equal(t, oldA.Link(), current[a][0].Link()) + require.Empty(t, current[b]) + return map[did.DID][]ucan.Delegation{a: {freshA}, b: {freshB}}, nil + })) + + for aud, want := range map[did.DID]ucan.Delegation{a: freshA, b: freshB} { + page, err := s.ListByAudience(t.Context(), aud) + require.NoError(t, err) + require.Len(t, page.Results, 1) + require.Equal(t, want.Link(), page.Results[0].Link()) + } + }) + + t.Run("Replace returns ErrInvalidArgument for a nil delegation from next", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + old := makeDelegation(t, issuer, audience, issuer.DID(), command.MustParse("/test/run")) + require.NoError(t, s.PutBatch(t.Context(), []ucan.Delegation{old})) + + err := replaceOne(t.Context(), s, audience, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + return []ucan.Delegation{nil}, nil + }) + require.ErrorIs(t, err, store.ErrInvalidArgument) + + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Len(t, page.Results, 1, "the current set is kept") + }) + + t.Run("Replace returns ErrInvalidArgument for a delegation filed under another audience", func(t *testing.T) { + issuer := testutil.RandomIssuer(t) + locked, other := testutil.RandomDID(t), testutil.RandomDID(t) + stray := makeDelegation(t, issuer, other, issuer.DID(), command.MustParse("/test/run")) + + err := replaceOne(t.Context(), s, locked, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + return []ucan.Delegation{stray}, nil + }) + require.ErrorIs(t, err, store.ErrInvalidArgument) + + page, err := s.ListByAudience(t.Context(), other) + require.NoError(t, err) + require.Empty(t, page.Results, "nothing is written under an audience the call did not lock") + }) + t.Run("DeleteBySubject removes only that subject's delegations", func(t *testing.T) { subjectA, subjectB := testutil.RandomDID(t), testutil.RandomDID(t) audA1, audA2 := testutil.RandomDID(t), testutil.RandomDID(t) @@ -333,3 +506,59 @@ func TestDelegationStore(t *testing.T) { }) } } + +// TestDelegationStoreMemoryLockTimeout: a write that waits out a Replace in +// flight gives up with [store.ErrLockTimeout] instead of wedging against it, +// as lock_timeout bounds the advisory lock wait on Postgres. +func TestDelegationStoreMemoryLockTimeout(t *testing.T) { + delegationmemory.LockWait = 100 * time.Millisecond + t.Cleanup(func() { delegationmemory.LockWait = store.LockTimeout }) + + s := delegationmemory.New() + issuer := testutil.RandomIssuer(t) + audience := testutil.RandomDID(t) + fresh := makeDelegation(t, issuer, audience, issuer.DID(), command.MustParse("/test/run")) + + entered, release := make(chan struct{}), make(chan struct{}) + replaced := make(chan error, 1) + go func() { + replaced <- replaceOne(context.Background(), s, audience, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + close(entered) + <-release + return []ucan.Delegation{fresh}, nil + }) + }() + <-entered + + // The second Replace waits on the audience the first holds and gives up. + waited := make(chan error, 1) + go func() { + waited <- replaceOne(context.Background(), s, audience, func(context.Context, []ucan.Delegation) ([]ucan.Delegation, error) { + return nil, errors.New("next must not run while the first Replace holds the audience") + }) + }() + select { + case err := <-waited: + require.ErrorIs(t, err, store.ErrLockTimeout) + case <-time.After(htestutil.WaitTimeout): + t.Fatal("the second Replace waited on the first without a bound") + } + + close(release) + require.NoError(t, <-replaced) + page, err := s.ListByAudience(t.Context(), audience) + require.NoError(t, err) + require.Len(t, page.Results, 1) + require.Equal(t, fresh.Link(), page.Results[0].Link()) +} + +// replaceOne calls Replace over one audience, the shape most tests need. +func replaceOne(ctx context.Context, s dlgstore.Store, audience did.DID, next func(ctx context.Context, current []ucan.Delegation) ([]ucan.Delegation, error)) error { + return s.Replace(ctx, []did.DID{audience}, func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error) { + out, err := next(ctx, current[audience]) + if err != nil { + return nil, err + } + return map[did.DID][]ucan.Delegation{audience: out}, nil + }) +} diff --git a/pkg/store/delegation/memory/store.go b/pkg/store/delegation/memory/store.go index fbd44e88..d7d0c37d 100644 --- a/pkg/store/delegation/memory/store.go +++ b/pkg/store/delegation/memory/store.go @@ -7,6 +7,7 @@ import ( "slices" "strings" "sync" + "time" "github.com/fil-forge/hilt/pkg/store" dlgstore "github.com/fil-forge/hilt/pkg/store/delegation" @@ -18,27 +19,101 @@ import ( const defaultListLimit = 1000 +// Store holds two locks. mutex guards the map and is held for one read or one +// write at a time, never across Replace's callback. writes serializes the +// writers, as the per-audience advisory locks do on Postgres, and is the lock +// Replace holds while next runs; it is a one-slot channel rather than a mutex +// so [Store.hold] can bound its wait. type Store struct { - mutex sync.RWMutex + mutex sync.RWMutex + writes chan struct{} // audience DID -> delegations (sorted by link string) byAudience map[did.DID][]ucan.Delegation } +// LockWait bounds how long a writer waits for a write in flight before giving +// up with [store.ErrLockTimeout], as lock_timeout bounds the wait a Postgres +// advisory lock gives. It is a variable so a test can shorten it. +var LockWait = store.LockTimeout + var _ dlgstore.Store = (*Store)(nil) func New() *Store { - return &Store{byAudience: map[did.DID][]ucan.Delegation{}} + return &Store{writes: make(chan struct{}, 1), byAudience: map[did.DID][]ucan.Delegation{}} +} + +// hold takes the write lock, waiting at most [LockWait] or until ctx is done, +// and release gives it back. +func (s *Store) hold(ctx context.Context) error { + timer := time.NewTimer(LockWait) + defer timer.Stop() + select { + case s.writes <- struct{}{}: + return nil + case <-timer.C: + return fmt.Errorf("waited %s for an audience a write holds: %w", LockWait, store.ErrLockTimeout) + case <-ctx.Done(): + return ctx.Err() + } } +func (s *Store) release() { <-s.writes } + func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) error { // Validate the whole batch before storing anything. if slices.Contains(delegations, nil) { return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) } + if err := s.hold(ctx); err != nil { + return err + } + defer s.release() + s.mutex.Lock() + defer s.mutex.Unlock() + + s.put(delegations) + return nil +} + +// Replace swaps the audiences' delegations holding the write lock for the +// whole call, which serializes it against every other write: next sees the +// settled current sets and its result is in place before the lock is +// released. Another writer waits at most [LockWait] for it. Reads take the +// map alone and are answered while next runs, as on Postgres. +func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error)) error { + if err := s.hold(ctx); err != nil { + return err + } + defer s.release() + + s.mutex.RLock() + current := make(map[did.DID][]ucan.Delegation, len(audiences)) + for _, aud := range audiences { + current[aud] = slices.Clone(s.byAudience[aud]) + } + s.mutex.RUnlock() + + replacement, err := next(ctx, current) + if err != nil { + return err + } + if err := dlgstore.CheckReplacement(audiences, replacement); err != nil { + return err + } + s.mutex.Lock() defer s.mutex.Unlock() + for _, aud := range audiences { + delete(s.byAudience, aud) + s.put(replacement[aud]) + } + return nil +} +// put stores the delegations, skipping any already held. The caller holds +// both locks. +func (s *Store) put(delegations []ucan.Delegation) { for _, d := range delegations { aud := d.Audience() existing := s.byAudience[aud] @@ -53,7 +128,6 @@ func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) err }) s.byAudience[aud] = existing } - return nil } func (s *Store) ListByAudience(ctx context.Context, audience did.DID, opts ...store.PaginationOption) (store.Page[ucan.Delegation], error) { @@ -126,6 +200,10 @@ func page(dlgs []ucan.Delegation, opts []store.PaginationOption) store.Page[ucan } func (s *Store) DeleteByAudience(ctx context.Context, audience did.DID) error { + if err := s.hold(ctx); err != nil { + return err + } + defer s.release() s.mutex.Lock() defer s.mutex.Unlock() @@ -138,6 +216,10 @@ func (s *Store) DeleteBySubject(ctx context.Context, subject did.DID) error { return fmt.Errorf("cannot delete powerline delegations: %w", store.ErrInvalidArgument) } + if err := s.hold(ctx); err != nil { + return err + } + defer s.release() s.mutex.Lock() defer s.mutex.Unlock() diff --git a/pkg/store/delegation/postgres/store.go b/pkg/store/delegation/postgres/store.go index bdf5193b..53072362 100644 --- a/pkg/store/delegation/postgres/store.go +++ b/pkg/store/delegation/postgres/store.go @@ -1,6 +1,8 @@ // Package postgres provides a PostgreSQL-backed implementation of // delegation.Store. Encoded delegation payloads are stored directly in the -// delegation table's data column. +// delegation table's data column. Every write that mutates an audience's set +// (PutBatch, Replace, DeleteByAudience) runs in a transaction holding that +// audience's advisory lock, so the writes of one audience serialize. package postgres import ( @@ -12,6 +14,7 @@ import ( "github.com/fil-forge/hilt/pkg/store" dlgstore "github.com/fil-forge/hilt/pkg/store/delegation" + "github.com/fil-forge/hilt/pkg/store/pglock" "github.com/fil-forge/ucantone/did" "github.com/fil-forge/ucantone/ucan" "github.com/fil-forge/ucantone/ucan/delegation" @@ -35,41 +38,158 @@ func New(pool *pgxpool.Pool) *Store { // Initialize is a no-op. Schema is managed by the shared goose migrations. func (s *Store) Initialize(ctx context.Context) error { return nil } -func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) error { +// PutBatch stores the batch in one transaction holding the advisory lock of +// every audience it touches, so a concurrent Replace of any of them sees the +// whole batch or none of it. +func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) (err error) { // Validate the whole batch before storing anything. if slices.Contains(delegations, nil) { return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) } - tx, err := s.pool.Begin(ctx) + defer func() { err = pglock.MapError(err) }() + tx, err := pglock.Begin(ctx, s.pool) if err != nil { return fmt.Errorf("beginning transaction: %w", err) } defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + audiences := make([]string, 0, len(delegations)) for _, d := range delegations { - data, err := delegation.Encode(d) - if err != nil { - return fmt.Errorf("encoding delegation %s: %w", d.Link(), err) + audiences = append(audiences, d.Audience().String()) + } + if err := lockAudiences(ctx, tx, audiences...); err != nil { + return err + } + + for _, d := range delegations { + if err := insert(ctx, tx, d); err != nil { + return err } + } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } + return nil +} + +// insert stores one delegation inside tx, leaving an already-stored one alone. +func insert(ctx context.Context, tx pgx.Tx, d ucan.Delegation) error { + data, err := delegation.Encode(d) + if err != nil { + return fmt.Errorf("encoding delegation %s: %w", d.Link(), err) + } + + var subject *string + if d.Subject().Defined() { + str := d.Subject().String() + subject = &str + } + + var expiresAt *time.Time + if exp := d.Expiration(); exp != nil { + t := time.Unix(int64(*exp), 0).UTC() + expiresAt = &t + } - var subject *string - if d.Subject().Defined() { - str := d.Subject().String() - subject = &str + if _, err := tx.Exec(ctx, ` + INSERT INTO delegation (id, issuer, audience, subject, command, data, expires_at) + VALUES ($1, $2, $3, $4, $5, $6, $7) + ON CONFLICT (id) DO NOTHING + `, d.Link().String(), d.Issuer().String(), d.Audience().String(), subject, d.Command().String(), data, expiresAt); err != nil { + return fmt.Errorf("storing delegation %s: %w", d.Link(), err) + } + return nil +} + +// lockNamespace is the first key of the advisory lock every audience-mutating +// write takes per audience. Postgres identifies an advisory lock by its key +// pair and nothing else, so the fixed first key keeps this store's locks apart +// from any other advisory lock taken on the same database; the second key is +// hashtext(audience DID). +const lockNamespace int32 = 0x44454c47 // "DELG" + +// lockAudiences bounds tx's lock waits at [store.LockTimeout] and takes the +// exclusive advisory lock of each distinct audience, in sorted order. The +// order is what keeps two transactions locking overlapping audiences from +// deadlocking: they queue for the shared audiences in the same sequence. The +// locks are released when tx commits or rolls back. +func lockAudiences(ctx context.Context, tx pgx.Tx, audiences ...string) error { + if err := pglock.SetTimeout(ctx, tx); err != nil { + return err + } + slices.Sort(audiences) + for _, aud := range slices.Compact(audiences) { + if err := pglock.Advisory(ctx, tx, lockNamespace, aud, false); err != nil { + return fmt.Errorf("locking delegation audience: %w", err) } + } + return nil +} - var expiresAt *time.Time - if exp := d.Expiration(); exp != nil { - t := time.Unix(int64(*exp), 0).UTC() - expiresAt = &t +// Replace runs in one transaction that takes the audiences' advisory locks +// (released when the transaction ends), reads the current sets, calls next, +// deletes the sets and inserts next's result. Another write of any of the +// audiences waits on the lock until the first commits or rolls back, so it +// sees the settled state; the holder runs next while it holds the locks, so +// the wait is bounded at [store.LockTimeout] and a longer one returns +// [store.ErrLockTimeout]. +func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx context.Context, current map[did.DID][]ucan.Delegation) (map[did.DID][]ucan.Delegation, error)) (err error) { + defer func() { err = pglock.MapError(err) }() + tx, err := pglock.Begin(ctx, s.pool) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + + strs := make([]string, len(audiences)) + for i, aud := range audiences { + strs[i] = aud.String() + } + if err := lockAudiences(ctx, tx, strs...); err != nil { + return err + } + + current := make(map[did.DID][]ucan.Delegation, len(audiences)) + rows, err := tx.Query(ctx, `SELECT id, data FROM delegation WHERE audience = ANY($1) ORDER BY id ASC`, strs) + if err != nil { + return fmt.Errorf("querying delegations by audience: %w", err) + } + for rows.Next() { + var id string + var data []byte + if err := rows.Scan(&id, &data); err != nil { + rows.Close() + return fmt.Errorf("scanning delegation: %w", err) + } + dlg, err := delegation.Decode(data) + if err != nil { + rows.Close() + return fmt.Errorf("decoding delegation %s: %w", id, err) } + current[dlg.Audience()] = append(current[dlg.Audience()], dlg) + } + rows.Close() + if err := rows.Err(); err != nil { + return fmt.Errorf("iterating delegations: %w", err) + } - if _, err := tx.Exec(ctx, ` - INSERT INTO delegation (id, issuer, audience, subject, command, data, expires_at) - VALUES ($1, $2, $3, $4, $5, $6, $7) - ON CONFLICT (id) DO NOTHING - `, d.Link().String(), d.Issuer().String(), d.Audience().String(), subject, d.Command().String(), data, expiresAt); err != nil { - return fmt.Errorf("storing delegation %s: %w", d.Link(), err) + replacement, err := next(pglock.WithTx(ctx, tx), current) + if err != nil { + return err + } + if err := dlgstore.CheckReplacement(audiences, replacement); err != nil { + return err + } + + if _, err := tx.Exec(ctx, `DELETE FROM delegation WHERE audience = ANY($1)`, strs); err != nil { + return fmt.Errorf("deleting delegations by audience: %w", err) + } + for _, aud := range audiences { + for _, d := range replacement[aud] { + if err := insert(ctx, tx, d); err != nil { + return err + } } } @@ -166,20 +286,61 @@ func (s *Store) listBy(ctx context.Context, column listColumn, value string, opt return store.Page[ucan.Delegation]{Cursor: cursor, Results: results}, nil } -func (s *Store) DeleteByAudience(ctx context.Context, audience did.DID) error { - if _, err := s.pool.Exec(ctx, `DELETE FROM delegation WHERE audience = $1`, audience.String()); err != nil { +// DeleteByAudience removes the audience's delegations in one transaction +// holding its advisory lock, so a concurrent Replace of the audience does not +// reinsert what this call deleted. +func (s *Store) DeleteByAudience(ctx context.Context, audience did.DID) (err error) { + defer func() { err = pglock.MapError(err) }() + tx, err := pglock.Begin(ctx, s.pool) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + + if err := lockAudiences(ctx, tx, audience.String()); err != nil { + return err + } + if _, err := tx.Exec(ctx, `DELETE FROM delegation WHERE audience = $1`, audience.String()); err != nil { return fmt.Errorf("deleting delegations by audience: %w", err) } + + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } return nil } -func (s *Store) DeleteBySubject(ctx context.Context, subject did.DID) error { +// DeleteBySubject locks the audiences holding delegations over the subject +// before deleting them, so it does not interleave with a Replace of one of +// those audiences. +func (s *Store) DeleteBySubject(ctx context.Context, subject did.DID) (err error) { if !subject.Defined() { return fmt.Errorf("cannot delete powerline delegations: %w", store.ErrInvalidArgument) } - if _, err := s.pool.Exec(ctx, `DELETE FROM delegation WHERE subject = $1`, subject.String()); err != nil { + defer func() { err = pglock.MapError(err) }() + tx, err := pglock.Begin(ctx, s.pool) + if err != nil { + return fmt.Errorf("beginning transaction: %w", err) + } + defer tx.Rollback(ctx) // no-op once committed; rolls back on any early return + + rows, err := tx.Query(ctx, `SELECT DISTINCT audience FROM delegation WHERE subject = $1`, subject.String()) + if err != nil { + return fmt.Errorf("querying delegation audiences by subject: %w", err) + } + audiences, err := pgx.CollectRows(rows, pgx.RowTo[string]) + if err != nil { + return fmt.Errorf("scanning delegation audiences: %w", err) + } + if err := lockAudiences(ctx, tx, audiences...); err != nil { + return err + } + if _, err := tx.Exec(ctx, `DELETE FROM delegation WHERE subject = $1`, subject.String()); err != nil { return fmt.Errorf("deleting delegations by subject: %w", err) } + if err := tx.Commit(ctx); err != nil { + return fmt.Errorf("committing transaction: %w", err) + } return nil } diff --git a/pkg/vault/issuer.go b/pkg/vault/issuer.go new file mode 100644 index 00000000..6dffe066 --- /dev/null +++ b/pkg/vault/issuer.go @@ -0,0 +1,25 @@ +package vault + +import ( + "context" + "fmt" + + "github.com/fil-forge/ucantone/did" + "github.com/fil-forge/ucantone/multikey" + "github.com/fil-forge/ucantone/multikey/secp256k1" + "github.com/fil-forge/ucantone/ucan" +) + +// TenantIssuer loads the tenant's secp256k1 signing key from v and returns an +// issuer that signs as the tenant. +func TenantIssuer(ctx context.Context, v Vault, tenantID did.DID) (ucan.Issuer, error) { + keyBytes, err := v.Read(ctx, TenantKeyPath(tenantID)) + if err != nil { + return nil, fmt.Errorf("reading tenant key: %w", err) + } + signer, err := secp256k1.Decode(keyBytes) + if err != nil { + return nil, fmt.Errorf("decoding tenant key: %w", err) + } + return multikey.NewIssuer(tenantID, signer), nil +}