From f773335d7ba1ca0929df04b77c583764e5084fc7 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 10 Sep 2026 23:20:52 +0200 Subject: [PATCH 1/9] feat(grant): rotate and revoke principal-bound key delegations A policy change has to make the gateway drop what it cached for the keys of the principals it affects, and a principal's removal has to do the same for every key the principal holds. A principal-bound key holds the delegations its principal's effective actions map to, one per bucket and Forge command, so the record that clears the gateway is the ordinary revocation of those delegations through Swarf's /ucan/revoke. pkg/grant gains Rotator. Rotate sets every key's delegations over one bucket to the commands the principal's new actions map to and revokes the ones it replaces; a key that held nothing over the bucket costs no revocation, and an empty action set leaves it with nothing over the bucket. Revoke revokes every delegation of the principal's keys and leaves them empty. Both run in one delegation-store transaction over all the keys, and every revocation of one write goes to Swarf in a single request through the client's new PublishBatch, so the write waits on one round trip however many keys it touches. No revocation carries a nonce: a retry after a rolled-back write republishes the same invocations, which Swarf records once. The delegation store's Replace takes several audiences and swaps their sets in one transaction under their sorted advisory locks; DeleteBySubject locks the audiences it touches for the same reason. One RevocationPublisher interface, in pkg/grant, replaces the copies the access-key and bucket services declared, and both services publish through PublishBatch: a key deletion revokes its delegations under the key's lock, and a bucket deletion revokes every grant over the bucket in one request. vault.TenantIssuer is shared by the access-key service and the rotator. Swarf is pinned to the commit that adds PublishBatch, and ucantone to the one that exposes batch execution. Co-Authored-By: Claude Fable 5.1 --- go.mod | 2 +- go.sum | 4 +- pkg/api/access_keys_test.go | 3 +- pkg/api/service/accesskey/service.go | 107 +++------ pkg/api/service/accesskey/service_test.go | 26 ++- pkg/fx/revocation.go | 17 +- pkg/fx/revocation_test.go | 17 ++ pkg/grant/rotator.go | 171 ++++++++++++++ pkg/grant/rotator_test.go | 273 ++++++++++++++++++++++ pkg/rpc/service/bucket/service.go | 41 +--- pkg/rpc/service/bucket/service_test.go | 16 +- pkg/store/delegation/delegation.go | 18 +- pkg/store/delegation/delegation_test.go | 170 ++++++++++++++ pkg/store/delegation/memory/store.go | 35 ++- pkg/store/delegation/postgres/store.go | 215 +++++++++++++++-- pkg/vault/issuer.go | 25 ++ 16 files changed, 974 insertions(+), 166 deletions(-) create mode 100644 pkg/grant/rotator.go create mode 100644 pkg/grant/rotator_test.go create mode 100644 pkg/vault/issuer.go 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/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..2fb56369 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) } @@ -353,7 +351,7 @@ func (s *Service) Get(ctx context.Context, externalID, accessKeyID string) (acce // 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. +// 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,12 +371,18 @@ 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 { + // The key's delegations are revoked and removed under the key's delegation + // lock, so a policy write rotating the key meanwhile serializes with it. + issuer, err := vault.TenantIssuer(ctx, s.secrets, tenantRec.ID) + if err != nil { return err } - - if err := s.delegations.DeleteByAudience(ctx, id); err != nil { - return fmt.Errorf("deleting access key delegations: %w", err) + 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) { + return nil, grant.PublishRevocations(ctx, log, s.revocations, issuer, current[id]) + }) + if err != nil { + return fmt.Errorf("revoking 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)) @@ -389,63 +393,6 @@ func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) er 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)) - } - return s.delegations.ListByAudience(ctx, accessKeyID, listOpts...) - }) - 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 - } - 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())) - } - 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..6ee1a929 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -23,7 +23,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/multikey" "github.com/fil-forge/ucantone/multikey/ed25519" @@ -36,27 +35,28 @@ 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. +// revocation records one published revocation. 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 + calls int revocations []revocation } -func (f *fakeSwarf) Publish(_ context.Context, revoker ucan.Issuer, revoked ucan.Delegation, opts ...swarfclient.PublishOption) error { +func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { if f.err != nil { return f.err } - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: revoked.Link(), options: len(opts)}) + f.calls++ + for _, d := range revoked { + f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: d.Link()}) + } return nil } @@ -157,6 +157,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) { @@ -412,7 +419,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) @@ -424,12 +431,12 @@ 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.Equal(t, 1, d.swarf.calls, "every revocation goes in one request") revoked := map[cid.Cid]bool{} 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 } for _, dlg := range issued.Results { @@ -454,7 +461,6 @@ func TestDeleteRevokes(t *testing.T) { 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) }) t.Run("revokes a powerline delegation when the tenant owns no bucket", func(t *testing.T) { diff --git a/pkg/fx/revocation.go b/pkg/fx/revocation.go index 874d5435..844d6f1c 100644 --- a/pkg/fx/revocation.go +++ b/pkg/fx/revocation.go @@ -4,33 +4,32 @@ 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, the UCAN bucket service and the grant rotator all revoke delegations +// through it, so the client is shared rather than owned by any of their modules. 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)), ), + grant.NewRotator, ), ) // 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/fx/revocation_test.go b/pkg/fx/revocation_test.go index 9e240589..1d835f0c 100644 --- a/pkg/fx/revocation_test.go +++ b/pkg/fx/revocation_test.go @@ -5,8 +5,10 @@ import ( "github.com/fil-forge/hilt/pkg/config" appfx "github.com/fil-forge/hilt/pkg/fx" + "github.com/fil-forge/hilt/pkg/grant" "github.com/fil-forge/libforge/testutil" "github.com/stretchr/testify/require" + "go.uber.org/fx" ) func TestNewRevocationClient(t *testing.T) { @@ -37,3 +39,18 @@ func TestNewRevocationClient(t *testing.T) { require.ErrorContains(t, err, "revocation.service_url") }) } + +// TestRotatorResolves pins that the app graph can build the grant rotator. It +// is provided from the revocation module, so nothing else in the graph would +// report a broken wiring until a consumer asks for it. +func TestRotatorResolves(t *testing.T) { + cfg := &config.Config{ + Storage: config.StorageConfig{Type: config.StorageTypeMemory}, + Vault: config.VaultConfig{Type: config.VaultTypeMemory}, + } + require.NoError(t, fx.ValidateApp( + appfx.AppModule(cfg), + fx.NopLogger, + fx.Invoke(func(*grant.Rotator) {}), + )) +} diff --git a/pkg/grant/rotator.go b/pkg/grant/rotator.go new file mode 100644 index 00000000..c75e7b37 --- /dev/null +++ b/pkg/grant/rotator.go @@ -0,0 +1,171 @@ +package grant + +import ( + "context" + "fmt" + "maps" + "slices" + "time" + + 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" +) + +// 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 { + principalOf := map[did.DID]string{} + var keys []accesskeystore.Record + for _, p := range slices.Sorted(maps.Keys(actions)) { + recs, err := r.accessKeys.ListByTenant(ctx, tenant, accesskeystore.WithPrincipal(p)) + if err != nil { + return fmt.Errorf("listing keys of principal %q: %w", p, err) + } + for _, rec := range recs { + principalOf[rec.ID] = p + } + keys = append(keys, recs...) + } + if len(keys) == 0 { + return nil + } + issuer, err := vault.TenantIssuer(ctx, r.secrets, tenant) + if err != nil { + return err + } + 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 { + 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 { + keys, err := r.accessKeys.ListByTenant(ctx, tenant, accesskeystore.WithPrincipal(principal)) + if err != nil { + return fmt.Errorf("listing keys of principal %q: %w", principal, err) + } + if len(keys) == 0 { + return nil + } + issuer, err := vault.TenantIssuer(ctx, r.secrets, tenant) + if err != nil { + 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) + }) +} + +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. +func PublishRevocations(ctx context.Context, log *zap.Logger, publisher RevocationPublisher, issuer ucan.Issuer, dels []ucan.Delegation) error { + 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..4a1e50a9 --- /dev/null +++ b/pkg/grant/rotator_test.go @@ -0,0 +1,273 @@ +package grant_test + +import ( + "context" + "errors" + "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" + 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/ed25519" + "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" +) + +// batch records one PublishBatch call. +type batch struct { + revoker did.DID + revoked []cid.Cid +} + +type fakeSwarf struct { + err error + batches []batch +} + +func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { + if f.err != nil { + return f.err + } + b := batch{revoker: revoker.DID()} + for _, d := range revoked { + b.revoked = append(b.revoked, d.Link()) + } + f.batches = append(f.batches, b) + return nil +} + +// revoked returns every CID revoked across all batches. +func (f *fakeSwarf) revoked() []cid.Cid { + var out []cid.Cid + for _, b := range f.batches { + out = append(out, b.revoked...) + } + return out +} + +type deps struct { + rotator *grant.Rotator + delegations *delegationmemory.Store + swarf *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, + swarf: &fakeSwarf{}, + tenant: tenant, + photos: randomKey(t), + backups: randomKey(t), + } + addKey := func(principal, name string) did.DID { + id := randomKey(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 +} + +func randomKey(t *testing.T) did.DID { + signer, err := ed25519.Generate() + require.NoError(t, err) + return signer.KeyDID() +} + +// 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].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 := randomKey(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 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].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)) + } + }) +} 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..f6e58b52 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" @@ -234,25 +233,23 @@ 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. +// revocation records one published revocation. 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 + calls int 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 { +func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { if f.err != nil { return f.err } @@ -260,7 +257,10 @@ func (f *fakeSwarf) Publish(_ context.Context, revoker ucan.Issuer, revoked ucan f.onPublish() f.onPublish = nil } - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: revoked.Link(), options: len(opts)}) + f.calls++ + for _, d := range revoked { + f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: d.Link()}) + } return nil } @@ -388,11 +388,11 @@ func TestDelete(t *testing.T) { require.NoError(t, err) 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) }) t.Run("does not revoke the bucket root", func(t *testing.T) { diff --git a/pkg/store/delegation/delegation.go b/pkg/store/delegation/delegation.go index 4efe8d60..37d1746d 100644 --- a/pkg/store/delegation/delegation.go +++ b/pkg/store/delegation/delegation.go @@ -10,7 +10,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 +32,18 @@ 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. + 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 } diff --git a/pkg/store/delegation/delegation_test.go b/pkg/store/delegation/delegation_test.go index dfd8c90f..fc576ebe 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,163 @@ 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, 2) + 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("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 +492,14 @@ func TestDelegationStore(t *testing.T) { }) } } + +// 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..a0ddf582 100644 --- a/pkg/store/delegation/memory/store.go +++ b/pkg/store/delegation/memory/store.go @@ -39,6 +39,40 @@ func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) err s.mutex.Lock() defer s.mutex.Unlock() + s.put(delegations) + return nil +} + +// Replace swaps the audiences' delegations under the store's lock, which +// serializes it against every other write: next sees the settled current sets +// and its result is in place before the lock is released. +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 { + s.mutex.Lock() + defer s.mutex.Unlock() + + current := make(map[did.DID][]ucan.Delegation, len(audiences)) + for _, aud := range audiences { + current[aud] = slices.Clone(s.byAudience[aud]) + } + replacement, err := next(ctx, current) + if err != nil { + return err + } + for _, aud := range audiences { + if slices.Contains(replacement[aud], nil) { + return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) + } + } + 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 the +// lock. +func (s *Store) put(delegations []ucan.Delegation) { for _, d := range delegations { aud := d.Audience() existing := s.byAudience[aud] @@ -53,7 +87,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) { diff --git a/pkg/store/delegation/postgres/store.go b/pkg/store/delegation/postgres/store.go index bdf5193b..4945b174 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,6 +38,9 @@ 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 } +// 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) error { // Validate the whole batch before storing anything. if slices.Contains(delegations, nil) { @@ -46,30 +52,156 @@ func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) 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 + } + + 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" + +// lockTimeout bounds how long an audience-mutating write waits on another +// holding the same audience's lock. Replace runs next while it holds the lock, +// so a hung next must fail the waiter rather than pin a pool connection. +const lockTimeout = "10s" + +// lockAudiences sets tx's lock timeout 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 := tx.Exec(ctx, `SET LOCAL lock_timeout = '`+lockTimeout+`'`); err != nil { + return fmt.Errorf("setting lock timeout: %w", 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 +} + +// 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)) error { + return s.replace(ctx, audiences, next) +} + +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 { + tx, err := s.pool.Begin(ctx) + 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 + } - var subject *string - if d.Subject().Defined() { - str := d.Subject().String() - subject = &str + current := make(map[did.DID][]ucan.Delegation, len(audiences)) + for _, aud := range audiences { + current[aud] = nil + } + 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) + } - var expiresAt *time.Time - if exp := d.Expiration(); exp != nil { - t := time.Unix(int64(*exp), 0).UTC() - expiresAt = &t + replacement, err := next(ctx, current) + if err != nil { + return err + } + for _, aud := range audiences { + if slices.Contains(replacement[aud], nil) { + return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) } + } - 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) + 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 +298,63 @@ func (s *Store) listBy(ctx context.Context, column listColumn, value string, opt return store.Page[ucan.Delegation]{Cursor: cursor, Results: results}, 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) error { - if _, err := s.pool.Exec(ctx, `DELETE FROM delegation WHERE audience = $1`, audience.String()); err != nil { + tx, err := s.pool.Begin(ctx) + 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 } +// 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) 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 { + return s.deleteBySubject(ctx, subject) +} + +func (s *Store) deleteBySubject(ctx context.Context, subject did.DID) error { + tx, err := s.pool.Begin(ctx) + 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 +} From d98c63e0472d89306ec29b00286df3db088f31e5 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Wed, 23 Sep 2026 14:36:03 +0200 Subject: [PATCH 2/9] fix(grant): keep a rotation from re-granting a key deleted under its lock A rotation lists a principal's keys and then waits on their delegation locks; a removal of one of those keys holding the lock first published its revocations and cleared its delegations, and the rotation went on to issue it fresh ones that outlived the record. The access-key service now deletes the record inside the same Replace as the revocation, and Rotate re-reads each key once it holds the lock and leaves a removed one alone. Co-Authored-By: Claude Fable 5.1 --- internal/testutil/swarf.go | 80 ++++++++++++++++ pkg/api/service/accesskey/service.go | 31 ++++--- pkg/api/service/accesskey/service_test.go | 58 ++++-------- pkg/fx/revocation.go | 5 +- pkg/fx/revocation_test.go | 17 ---- pkg/grant/rotator.go | 77 ++++++++++------ pkg/grant/rotator_test.go | 106 +++++++++++----------- pkg/rpc/service/auth/auth.go | 11 +-- pkg/rpc/service/bucket/service_test.go | 63 ++++--------- pkg/store/delegation/delegation.go | 4 +- pkg/store/delegation/delegation_test.go | 1 - pkg/store/delegation/postgres/store.go | 42 +++------ 12 files changed, 255 insertions(+), 240 deletions(-) create mode 100644 internal/testutil/swarf.go 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/service/accesskey/service.go b/pkg/api/service/accesskey/service.go index 2fb56369..4eaf8258 100644 --- a/pkg/api/service/accesskey/service.go +++ b/pkg/api/service/accesskey/service.go @@ -347,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. -// Both kinds of key are deleted 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) { @@ -371,25 +374,29 @@ func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) er return fmt.Errorf("looking up access key: %w", err) } - // The key's delegations are revoked and removed under the key's delegation - // lock, so a policy write rotating the key meanwhile serializes with it. + // The key's delegations are revoked and removed, and its record deleted, + // under the key's delegation lock, so a policy write rotating the key + // meanwhile serializes with it. issuer, err := vault.TenantIssuer(ctx, s.secrets, tenantRec.ID) if err != nil { return err } 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) { - return nil, grant.PublishRevocations(ctx, log, s.revocations, issuer, current[id]) + if err := grant.PublishRevocations(ctx, log, s.revocations, issuer, current[id]); err != nil { + return nil, err + } + 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("revoking access key delegations: %w", err) + return fmt.Errorf("revoking access key: %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 } diff --git a/pkg/api/service/accesskey/service_test.go b/pkg/api/service/accesskey/service_test.go index 6ee1a929..fb48dd0b 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -36,30 +36,6 @@ import ( ) // revocation records one published revocation. -type revocation struct { - revoker did.DID - revoked cid.Cid -} - -// fakeSwarf is a stub of the revocation service, recording what it was asked to -// publish. -type fakeSwarf struct { - err error - calls int - revocations []revocation -} - -func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { - if f.err != nil { - return f.err - } - f.calls++ - for _, d := range revoked { - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: d.Link()}) - } - return nil -} - // 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. @@ -84,7 +60,7 @@ type deps struct { delegations *delegationmemory.Store buckets *bucketmemory.Store secrets *vaultmemory.Store - swarf *fakeSwarf + swarf *testutil.FakeSwarf tenantID did.DID bucketID did.DID bucketRoot ucan.Delegation @@ -132,7 +108,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, @@ -330,9 +306,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) @@ -430,14 +406,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.Equal(t, 1, d.swarf.calls, "every revocation goes in one request") + 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) - 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()) @@ -457,10 +433,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.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) { @@ -472,7 +448,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) }) @@ -486,7 +462,7 @@ 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) }) @@ -495,7 +471,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") diff --git a/pkg/fx/revocation.go b/pkg/fx/revocation.go index 844d6f1c..a845196d 100644 --- a/pkg/fx/revocation.go +++ b/pkg/fx/revocation.go @@ -14,8 +14,8 @@ import ( // RevocationModule provides the Swarf revocation-service client, published as // [grant.RevocationPublisher] as well as the concrete type. The REST access-key -// service, the UCAN bucket service and the grant rotator all revoke delegations -// through it, so the client is shared rather than owned by any of their modules. +// 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( @@ -23,7 +23,6 @@ var RevocationModule = fx.Module("revocation", fx.As(fx.Self()), fx.As(new(grant.RevocationPublisher)), ), - grant.NewRotator, ), ) diff --git a/pkg/fx/revocation_test.go b/pkg/fx/revocation_test.go index 1d835f0c..9e240589 100644 --- a/pkg/fx/revocation_test.go +++ b/pkg/fx/revocation_test.go @@ -5,10 +5,8 @@ import ( "github.com/fil-forge/hilt/pkg/config" appfx "github.com/fil-forge/hilt/pkg/fx" - "github.com/fil-forge/hilt/pkg/grant" "github.com/fil-forge/libforge/testutil" "github.com/stretchr/testify/require" - "go.uber.org/fx" ) func TestNewRevocationClient(t *testing.T) { @@ -39,18 +37,3 @@ func TestNewRevocationClient(t *testing.T) { require.ErrorContains(t, err, "revocation.service_url") }) } - -// TestRotatorResolves pins that the app graph can build the grant rotator. It -// is provided from the revocation module, so nothing else in the graph would -// report a broken wiring until a consumer asks for it. -func TestRotatorResolves(t *testing.T) { - cfg := &config.Config{ - Storage: config.StorageConfig{Type: config.StorageTypeMemory}, - Vault: config.VaultConfig{Type: config.VaultTypeMemory}, - } - require.NoError(t, fx.ValidateApp( - appfx.AppModule(cfg), - fx.NopLogger, - fx.Invoke(func(*grant.Rotator) {}), - )) -} diff --git a/pkg/grant/rotator.go b/pkg/grant/rotator.go index c75e7b37..6df2bba4 100644 --- a/pkg/grant/rotator.go +++ b/pkg/grant/rotator.go @@ -2,11 +2,13 @@ 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" @@ -16,6 +18,19 @@ import ( "go.uber.org/zap" ) +// BatchTimeout bounds every rotation of one write together. 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]. The batch as a whole +// gets this deadline, below the reader's bound, so a large batch cannot lock +// the data path out. +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. @@ -65,31 +80,27 @@ func NewRotator( // 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 { - principalOf := map[did.DID]string{} - var keys []accesskeystore.Record - for _, p := range slices.Sorted(maps.Keys(actions)) { - recs, err := r.accessKeys.ListByTenant(ctx, tenant, accesskeystore.WithPrincipal(p)) - if err != nil { - return fmt.Errorf("listing keys of principal %q: %w", p, err) - } - for _, rec := range recs { - principalOf[rec.ID] = p - } - keys = append(keys, recs...) - } - if len(keys) == 0 { - return nil - } - issuer, err := vault.TenantIssuer(ctx, r.secrets, tenant) - if err != nil { + 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) @@ -113,15 +124,8 @@ func (r *Rotator) Rotate(ctx context.Context, tenant, bucket did.DID, actions ma // 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 { - keys, err := r.accessKeys.ListByTenant(ctx, tenant, accesskeystore.WithPrincipal(principal)) - if err != nil { - return fmt.Errorf("listing keys of principal %q: %w", principal, err) - } - if len(keys) == 0 { - return nil - } - issuer, err := vault.TenantIssuer(ctx, r.secrets, tenant) - if err != nil { + 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)) @@ -135,6 +139,27 @@ func (r *Rotator) Revoke(ctx context.Context, tenant did.DID, principal string) }) } +// 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 { diff --git a/pkg/grant/rotator_test.go b/pkg/grant/rotator_test.go index 4a1e50a9..ce5a2861 100644 --- a/pkg/grant/rotator_test.go +++ b/pkg/grant/rotator_test.go @@ -3,6 +3,8 @@ package grant_test import ( "context" "errors" + htestutil "github.com/fil-forge/hilt/internal/testutil" + "github.com/fil-forge/libforge/testutil" "testing" "time" @@ -10,12 +12,12 @@ import ( "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/ed25519" "github.com/fil-forge/ucantone/multikey/secp256k1" "github.com/fil-forge/ucantone/ucan" "github.com/ipfs/go-cid" @@ -23,42 +25,12 @@ import ( "go.uber.org/zap" ) -// batch records one PublishBatch call. -type batch struct { - revoker did.DID - revoked []cid.Cid -} - -type fakeSwarf struct { - err error - batches []batch -} - -func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { - if f.err != nil { - return f.err - } - b := batch{revoker: revoker.DID()} - for _, d := range revoked { - b.revoked = append(b.revoked, d.Link()) - } - f.batches = append(f.batches, b) - return nil -} - -// revoked returns every CID revoked across all batches. -func (f *fakeSwarf) revoked() []cid.Cid { - var out []cid.Cid - for _, b := range f.batches { - out = append(out, b.revoked...) - } - return out -} - type deps struct { rotator *grant.Rotator delegations *delegationmemory.Store - swarf *fakeSwarf + accessKeys *accesskeymemory.Store + secrets *vaultmemory.Store + swarf *htestutil.FakeSwarf tenant ucan.Issuer // photos and backups are the tenant's buckets. photos, backups did.DID @@ -80,13 +52,15 @@ func setup(t *testing.T, expiresAt *time.Time) deps { d := deps{ delegations: delegations, - swarf: &fakeSwarf{}, + accessKeys: accessKeys, + secrets: secrets, + swarf: &htestutil.FakeSwarf{}, tenant: tenant, - photos: randomKey(t), - backups: randomKey(t), + photos: testutil.RandomDID(t), + backups: testutil.RandomDID(t), } addKey := func(principal, name string) did.DID { - id := randomKey(t) + id := testutil.RandomDID(t) require.NoError(t, accessKeys.Add(ctx, accesskeystore.Input{ ID: id, Tenant: tenantID, Name: name, Principal: &principal, ExpiresAt: expiresAt, })) @@ -101,10 +75,19 @@ func setup(t *testing.T, expiresAt *time.Time) deps { return d } -func randomKey(t *testing.T) did.DID { - signer, err := ed25519.Generate() - require.NoError(t, err) - return signer.KeyDID() +// 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. @@ -156,9 +139,9 @@ func TestRotate(t *testing.T) { } 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].revoker) - require.ElementsMatch(t, links(old), d.swarf.revoked()) + 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)) @@ -187,10 +170,10 @@ func TestRotate(t *testing.T) { t.Run("a first grant over a bucket publishes nothing", func(t *testing.T) { d := setup(t, nil) - videos := randomKey(t) + 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) + 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))) } @@ -201,14 +184,14 @@ func TestRotate(t *testing.T) { 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.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") + 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)) @@ -226,14 +209,27 @@ func TestRotate(t *testing.T) { 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.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) + require.Empty(t, d.swarf.Batches()) }) } @@ -250,9 +246,9 @@ func TestRevoke(t *testing.T) { } 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].revoker) - require.ElementsMatch(t, links(old), d.swarf.revoked()) + 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) @@ -263,7 +259,7 @@ func TestRevoke(t *testing.T) { 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") + 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 { 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_test.go b/pkg/rpc/service/bucket/service_test.go index f6e58b52..0b55a463 100644 --- a/pkg/rpc/service/bucket/service_test.go +++ b/pkg/rpc/service/bucket/service_test.go @@ -37,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" @@ -125,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 +233,13 @@ func (f failingListDelegations) ListByAudience(context.Context, did.DID, ...stor } // revocation records one published revocation. -type revocation struct { - revoker did.DID - revoked cid.Cid -} - -// fakeSwarf is a stub of the revocation service, recording what it was asked to -// publish. -type fakeSwarf struct { - err error - calls int - revocations []revocation - // onPublish, when set, runs once before the first publish is recorded. - onPublish func() -} - -func (f *fakeSwarf) PublishBatch(_ context.Context, revoker ucan.Issuer, revoked []ucan.Delegation) error { - if f.err != nil { - return f.err - } - if f.onPublish != nil { - f.onPublish() - f.onPublish = nil - } - f.calls++ - for _, d := range revoked { - f.revocations = append(f.revocations, revocation{revoker: revoker.DID(), revoked: d.Link()}) - } - return nil -} - // 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) - require.Equal(t, 1, d.swarf.calls) - 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.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 37d1746d..6450f782 100644 --- a/pkg/store/delegation/delegation.go +++ b/pkg/store/delegation/delegation.go @@ -44,6 +44,8 @@ type Store interface { // 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. + // returns [store.ErrInvalidArgument] if next returns a nil delegation. 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 } diff --git a/pkg/store/delegation/delegation_test.go b/pkg/store/delegation/delegation_test.go index fc576ebe..e79216cf 100644 --- a/pkg/store/delegation/delegation_test.go +++ b/pkg/store/delegation/delegation_test.go @@ -241,7 +241,6 @@ func TestDelegationStore(t *testing.T) { 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, 2) require.Len(t, current[a], 1) require.Equal(t, oldA.Link(), current[a][0].Link()) require.Empty(t, current[b]) diff --git a/pkg/store/delegation/postgres/store.go b/pkg/store/delegation/postgres/store.go index 4945b174..346b6f3d 100644 --- a/pkg/store/delegation/postgres/store.go +++ b/pkg/store/delegation/postgres/store.go @@ -41,11 +41,12 @@ func (s *Store) Initialize(ctx context.Context) error { return nil } // 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) error { +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) } + defer func() { err = pglock.MapError(err) }() tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("beginning transaction: %w", err) @@ -108,19 +109,14 @@ func insert(ctx context.Context, tx pgx.Tx, d ucan.Delegation) error { // hashtext(audience DID). const lockNamespace int32 = 0x44454c47 // "DELG" -// lockTimeout bounds how long an audience-mutating write waits on another -// holding the same audience's lock. Replace runs next while it holds the lock, -// so a hung next must fail the waiter rather than pin a pool connection. -const lockTimeout = "10s" - -// lockAudiences sets tx's lock timeout 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. +// 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 := tx.Exec(ctx, `SET LOCAL lock_timeout = '`+lockTimeout+`'`); err != nil { - return fmt.Errorf("setting lock timeout: %w", err) + if err := pglock.SetTimeout(ctx, tx); err != nil { + return err } slices.Sort(audiences) for _, aud := range slices.Compact(audiences) { @@ -138,11 +134,8 @@ func lockAudiences(ctx context.Context, tx pgx.Tx, audiences ...string) error { // 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)) error { - return s.replace(ctx, audiences, next) -} - -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 { +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 := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("beginning transaction: %w", err) @@ -158,9 +151,6 @@ func (s *Store) replace(ctx context.Context, audiences []did.DID, next func(ctx } current := make(map[did.DID][]ucan.Delegation, len(audiences)) - for _, aud := range audiences { - current[aud] = nil - } 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) @@ -301,7 +291,8 @@ func (s *Store) listBy(ctx context.Context, column listColumn, value string, opt // 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) error { +func (s *Store) DeleteByAudience(ctx context.Context, audience did.DID) (err error) { + defer func() { err = pglock.MapError(err) }() tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("beginning transaction: %w", err) @@ -324,14 +315,11 @@ func (s *Store) DeleteByAudience(ctx context.Context, audience 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) error { +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) } - return s.deleteBySubject(ctx, subject) -} - -func (s *Store) deleteBySubject(ctx context.Context, subject did.DID) error { + defer func() { err = pglock.MapError(err) }() tx, err := s.pool.Begin(ctx) if err != nil { return fmt.Errorf("beginning transaction: %w", err) From b4addbbcdb5b31073dbc0d9e5d694c8458cd85ad Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 1 Oct 2026 12:53:36 +0200 Subject: [PATCH 3/9] fix(store): bound the memory delegation writers' lock wait The memory delegation store took its RWMutex across Replace's callback, so a second writer waited for as long as the callback ran, ignoring its ctx and the bound the interface promises. The writers now take a one-slot channel for the whole call, waiting at most LockWait (store.LockTimeout) or until ctx is done, and give up with store.ErrLockTimeout the way the Postgres advisory lock does under lock_timeout. The RWMutex now guards only the map, so reads are answered while next runs, as on Postgres. Co-Authored-By: Claude Fable 5.1 --- pkg/store/delegation/delegation_test.go | 45 ++++++++++++++++ pkg/store/delegation/memory/store.go | 69 +++++++++++++++++++++---- 2 files changed, 105 insertions(+), 9 deletions(-) diff --git a/pkg/store/delegation/delegation_test.go b/pkg/store/delegation/delegation_test.go index e79216cf..2a02f15d 100644 --- a/pkg/store/delegation/delegation_test.go +++ b/pkg/store/delegation/delegation_test.go @@ -492,6 +492,51 @@ 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) { diff --git a/pkg/store/delegation/memory/store.go b/pkg/store/delegation/memory/store.go index a0ddf582..9dc99354 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,24 +19,56 @@ 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() @@ -43,17 +76,24 @@ func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) err return nil } -// Replace swaps the audiences' delegations under the store's lock, which -// serializes it against every other write: next sees the settled current sets -// and its result is in place before the lock is released. +// 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 { - s.mutex.Lock() - defer s.mutex.Unlock() + 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 @@ -63,6 +103,9 @@ func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) } } + + s.mutex.Lock() + defer s.mutex.Unlock() for _, aud := range audiences { delete(s.byAudience, aud) s.put(replacement[aud]) @@ -70,8 +113,8 @@ func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx return nil } -// put stores the delegations, skipping any already held. The caller holds the -// lock. +// 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() @@ -159,6 +202,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() @@ -171,6 +218,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() From 92bdf4f805a5667d124d74f4a1fe7b5113847f34 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 1 Oct 2026 15:50:14 +0200 Subject: [PATCH 4/9] fix(api): read the tenant key only when a deleted access key has delegations Delete loaded the tenant issuer before its Replace, so a vault read failure blocked deleting a key that held nothing to revoke. A principal-bound key whose principal reaches no bucket is that key, and it is the common case. The issuer is now resolved inside the callback, only when the key's current delegations are non-empty. Co-Authored-By: Claude Fable 5.1 --- pkg/api/service/accesskey/service.go | 17 ++++++++++------- pkg/api/service/accesskey/service_test.go | 14 ++++++++++++++ 2 files changed, 24 insertions(+), 7 deletions(-) diff --git a/pkg/api/service/accesskey/service.go b/pkg/api/service/accesskey/service.go index 4eaf8258..6a770218 100644 --- a/pkg/api/service/accesskey/service.go +++ b/pkg/api/service/accesskey/service.go @@ -376,15 +376,18 @@ func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) er // The key's delegations are revoked and removed, and its record deleted, // under the key's delegation lock, so a policy write rotating the key - // meanwhile serializes with it. - issuer, err := vault.TenantIssuer(ctx, s.secrets, tenantRec.ID) - if err != nil { - return err - } + // meanwhile serializes with it. 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 err := grant.PublishRevocations(ctx, log, s.revocations, issuer, current[id]); err != nil { - return nil, err + 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 + } } if err := s.accessKeys.Delete(ctx, id); err != nil { return nil, fmt.Errorf("deleting access key: %w", err) diff --git a/pkg/api/service/accesskey/service_test.go b/pkg/api/service/accesskey/service_test.go index fb48dd0b..e387e4a1 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -467,6 +467,20 @@ func TestDeleteRevokes(t *testing.T) { 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, "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) + }) + t.Run("a revocation failure leaves the key intact", func(t *testing.T) { d := setup(t) created, _, err := d.svc.Create(ctx, "tenant-1", "k1", []string{"s3:GetObject"}, []string{"bucket-a"}, nil, nil) From 063445e198089c79fe8ba4202431bf93e2345cb9 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Fri, 2 Oct 2026 17:19:56 +0200 Subject: [PATCH 5/9] fix(grant): bound each rotation and revocation at BatchTimeout Rotate and Revoke relied on their callers for the deadline BatchTimeout documents, and a Revoke from a request context with no deadline held the keys' delegation locks for as long as a stalled publish took. Each call now shortens its ctx to BatchTimeout itself; an earlier caller deadline still wins. Test: both operations publish with at most BatchTimeout left. Co-Authored-By: Claude Fable 5.1 --- pkg/grant/rotator.go | 19 ++++++++++++------- pkg/grant/rotator_test.go | 33 +++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+), 7 deletions(-) diff --git a/pkg/grant/rotator.go b/pkg/grant/rotator.go index 6df2bba4..0af4824e 100644 --- a/pkg/grant/rotator.go +++ b/pkg/grant/rotator.go @@ -18,13 +18,14 @@ import ( "go.uber.org/zap" ) -// BatchTimeout bounds every rotation of one write together. 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]. The batch as a whole -// gets this deadline, below the reader's bound, so a large batch cannot lock -// the data path out. +// 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. const BatchTimeout = 8 * time.Second // The batch bound must stay below the reader's; a negative difference fails @@ -80,6 +81,8 @@ func NewRotator( // 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 @@ -124,6 +127,8 @@ func (r *Rotator) Rotate(ctx context.Context, tenant, bucket did.DID, actions ma // 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 diff --git a/pkg/grant/rotator_test.go b/pkg/grant/rotator_test.go index ce5a2861..0b34a0a7 100644 --- a/pkg/grant/rotator_test.go +++ b/pkg/grant/rotator_test.go @@ -75,6 +75,22 @@ func setup(t *testing.T, expiresAt *time.Time) deps { 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 @@ -267,3 +283,20 @@ func TestRevoke(t *testing.T) { } }) } + +// 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) + } +} From f71c07681254d03356996675e6315a05884ef2e3 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Fri, 2 Oct 2026 17:19:56 +0200 Subject: [PATCH 6/9] fix(store): refuse a replacement filed under another audience and join the caller's transaction Replace checked only for nil, so a delegation returned under a map key other than its own audience was written under an audience the call had not locked. Both backends now refuse it with ErrInvalidArgument through the shared CheckReplacement. The Postgres store opens its transactions through pglock.Begin and hands next a ctx carrying the transaction, so a store write the callback makes commits with the replacement or not at all. Suite test on both backends: a delegation to one audience filed under another is refused and nothing lands under the other. Co-Authored-By: Claude Fable 5.1 --- pkg/store/delegation/delegation.go | 27 ++++++++++++++++++++++--- pkg/store/delegation/delegation_test.go | 15 ++++++++++++++ pkg/store/delegation/memory/store.go | 6 ++---- pkg/store/delegation/postgres/store.go | 16 +++++++-------- 4 files changed, 48 insertions(+), 16 deletions(-) diff --git a/pkg/store/delegation/delegation.go b/pkg/store/delegation/delegation.go index 6450f782..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" @@ -44,8 +45,28 @@ type Store interface { // 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. The - // lock wait is bounded at [store.LockTimeout] and returns - // [store.ErrLockTimeout] when it runs out. + // 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 2a02f15d..8521b41a 100644 --- a/pkg/store/delegation/delegation_test.go +++ b/pkg/store/delegation/delegation_test.go @@ -271,6 +271,21 @@ func TestDelegationStore(t *testing.T) { 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) diff --git a/pkg/store/delegation/memory/store.go b/pkg/store/delegation/memory/store.go index 9dc99354..d7d0c37d 100644 --- a/pkg/store/delegation/memory/store.go +++ b/pkg/store/delegation/memory/store.go @@ -98,10 +98,8 @@ func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx if err != nil { return err } - for _, aud := range audiences { - if slices.Contains(replacement[aud], nil) { - return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) - } + if err := dlgstore.CheckReplacement(audiences, replacement); err != nil { + return err } s.mutex.Lock() diff --git a/pkg/store/delegation/postgres/store.go b/pkg/store/delegation/postgres/store.go index 346b6f3d..53072362 100644 --- a/pkg/store/delegation/postgres/store.go +++ b/pkg/store/delegation/postgres/store.go @@ -47,7 +47,7 @@ func (s *Store) PutBatch(ctx context.Context, delegations []ucan.Delegation) (er return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) } defer func() { err = pglock.MapError(err) }() - tx, err := s.pool.Begin(ctx) + tx, err := pglock.Begin(ctx, s.pool) if err != nil { return fmt.Errorf("beginning transaction: %w", err) } @@ -136,7 +136,7 @@ func lockAudiences(ctx context.Context, tx pgx.Tx, audiences ...string) error { // [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 := s.pool.Begin(ctx) + tx, err := pglock.Begin(ctx, s.pool) if err != nil { return fmt.Errorf("beginning transaction: %w", err) } @@ -174,14 +174,12 @@ func (s *Store) Replace(ctx context.Context, audiences []did.DID, next func(ctx return fmt.Errorf("iterating delegations: %w", err) } - replacement, err := next(ctx, current) + replacement, err := next(pglock.WithTx(ctx, tx), current) if err != nil { return err } - for _, aud := range audiences { - if slices.Contains(replacement[aud], nil) { - return fmt.Errorf("delegations must not be nil: %w", store.ErrInvalidArgument) - } + 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 { @@ -293,7 +291,7 @@ func (s *Store) listBy(ctx context.Context, column listColumn, value string, opt // 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 := s.pool.Begin(ctx) + tx, err := pglock.Begin(ctx, s.pool) if err != nil { return fmt.Errorf("beginning transaction: %w", err) } @@ -320,7 +318,7 @@ func (s *Store) DeleteBySubject(ctx context.Context, subject did.DID) (err error return fmt.Errorf("cannot delete powerline delegations: %w", store.ErrInvalidArgument) } defer func() { err = pglock.MapError(err) }() - tx, err := s.pool.Begin(ctx) + tx, err := pglock.Begin(ctx, s.pool) if err != nil { return fmt.Errorf("beginning transaction: %w", err) } From e388d8e1bcad55bb25e45a10f2f591ec54af5295 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Fri, 2 Oct 2026 17:19:56 +0200 Subject: [PATCH 7/9] test(api): a key deletion that fails after the row delete leaves the key for a retry On Postgres the access-key row's delete inside the deletion's Replace callback committed on its own, so a failure after it left the delegation rows and the vault secret behind with no row, and a retry answered not-found. The row delete now joins the delegation transaction (through the stores' pglock.Begin adoption below this PR), so the Postgres test here cancels the request right after the row delete and checks the row, the delegations and the secret survive and a retry removes all three. The access-key service test gains a Postgres fixture for it. Co-Authored-By: Claude Fable 5.1 --- pkg/api/service/accesskey/service.go | 6 +- pkg/api/service/accesskey/service_test.go | 125 +++++++++++++++++++--- 2 files changed, 113 insertions(+), 18 deletions(-) diff --git a/pkg/api/service/accesskey/service.go b/pkg/api/service/accesskey/service.go index 6a770218..c851f442 100644 --- a/pkg/api/service/accesskey/service.go +++ b/pkg/api/service/accesskey/service.go @@ -375,8 +375,10 @@ func (s *Service) Delete(ctx context.Context, externalID, accessKeyID string) er } // The key's delegations are revoked and removed, and its record deleted, - // under the key's delegation lock, so a policy write rotating the key - // meanwhile serializes with it. The tenant key is read only when there is + // 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) { diff --git a/pkg/api/service/accesskey/service_test.go b/pkg/api/service/accesskey/service_test.go index e387e4a1..0dfdd993 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -13,14 +13,23 @@ 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" "github.com/fil-forge/ucantone/did" @@ -57,31 +66,67 @@ type deps struct { svc *accesskeysvc.Service accessKeys accesskeystore.Store principals principal.Store - delegations *delegationmemory.Store - buckets *bucketmemory.Store - secrets *vaultmemory.Store + 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) } @@ -89,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")) @@ -500,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, "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) +} From 65fe54b2465b2ff37948a0c1ae263ec52e904d9a Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Fri, 2 Oct 2026 18:00:08 +0200 Subject: [PATCH 8/9] fix(grant): bound every revocation publish made under a delegation lock PublishRevocations is the callback body of every Replace that revokes what keys hold: the rotator's, a key deletion's, a bucket deletion's and a tenant removal's. Only the rotator applied BatchTimeout, so a key deletion from a request with no deadline held the key's lock behind a stalled publish. The publish now shortens its ctx to BatchTimeout itself, whoever calls it. Test: a publish from a caller with no deadline has at most BatchTimeout left. Co-Authored-By: Claude Fable 5.1 --- pkg/grant/rotator.go | 11 +++++++++-- pkg/grant/rotator_test.go | 13 +++++++++++++ 2 files changed, 22 insertions(+), 2 deletions(-) diff --git a/pkg/grant/rotator.go b/pkg/grant/rotator.go index 0af4824e..86ea29bb 100644 --- a/pkg/grant/rotator.go +++ b/pkg/grant/rotator.go @@ -25,7 +25,9 @@ import ( // 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. +// 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 @@ -175,8 +177,13 @@ func audiences(keys []accesskeystore.Record) []did.DID { // 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. +// 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 { diff --git a/pkg/grant/rotator_test.go b/pkg/grant/rotator_test.go index 0b34a0a7..b1ae3d09 100644 --- a/pkg/grant/rotator_test.go +++ b/pkg/grant/rotator_test.go @@ -300,3 +300,16 @@ func TestRotatorPublishesUnderBatchTimeout(t *testing.T) { 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) +} From 44fb918b775b33fcfcaf7d88f78bd5c6183d4440 Mon Sep 17 00:00:00 2001 From: Srdjan S Date: Thu, 8 Oct 2026 19:08:54 +0200 Subject: [PATCH 9/9] refactor(api): follow the access key principalId pointer Co-Authored-By: Claude Opus 5.5 --- pkg/api/service/accesskey/service_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/api/service/accesskey/service_test.go b/pkg/api/service/accesskey/service_test.go index 0dfdd993..33cc25b5 100644 --- a/pkg/api/service/accesskey/service_test.go +++ b/pkg/api/service/accesskey/service_test.go @@ -515,7 +515,7 @@ func TestDeleteRevokes(t *testing.T) { 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, "bob", nil) + 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))) @@ -568,7 +568,7 @@ func TestDeletePostgresRollsBackAfterRowDelete(t *testing.T) { 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, "alice", nil) + 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)