fix: validate spaces are provisioned by known providers during receipt validation

Author: volmedo

cmd/etracker/start.go

@@ -118,7 +118,15 @@ func init() {
 
 	cobra.CheckErr(viper.BindEnv("consumer_table_name", "CONSUMER_TABLE_NAME"))
 	cobra.CheckErr(viper.BindEnv("consumer_table_region", "CONSUMER_TABLE_REGION"))
+	cobra.CheckErr(viper.BindEnv("consumer_consumer_index_name", "CONSUMER_CONSUMER_INDEX_NAME"))
 	cobra.CheckErr(viper.BindEnv("consumer_customer_index_name", "CONSUMER_CUSTOMER_INDEX_NAME"))
+
+	startCmd.Flags().StringSlice(
+		"known-providers",
+		presets.KnownProviders,
+		"List of known provider DIDs (defaults to presets if not specified)",
+	)
+	cobra.CheckErr(viper.BindPFlag("known_providers", startCmd.Flags().Lookup("known-providers")))
 }
 
 func startService(cmd *cobra.Command, args []string) error {
@@ -164,7 +172,7 @@ func startService(cmd *cobra.Command, args []string) error {
 
 	consumerCfg := cfg.AWSConfig.Copy()
 	consumerCfg.Region = cfg.ConsumerTableRegion
-	consumerTable := consumer.NewDynamoConsumerTable(dynamodb.NewFromConfig(consumerCfg), cfg.ConsumerTableName, cfg.ConsumerCustomerIndexName)
+	consumerTable := consumer.NewDynamoConsumerTable(dynamodb.NewFromConfig(consumerCfg), cfg.ConsumerTableName, cfg.ConsumerConsumerIndexName, cfg.ConsumerCustomerIndexName)
 
 	// Create service
 	svc, err := service.New(
@@ -190,7 +198,7 @@ func startService(cmd *cobra.Command, args []string) error {
 	interval := time.Duration(cfg.ConsolidationInterval) * time.Second
 	batchSize := cfg.ConsolidationBatchSize
 
-	cons, err := consolidator.New(id, egressTable, consolidatedTable, spaceStatsTable, interval, batchSize, presolver)
+	cons, err := consolidator.New(id, egressTable, consolidatedTable, spaceStatsTable, consumerTable, cfg.KnownProviders, interval, batchSize, presolver)
 	if err != nil {
 		return fmt.Errorf("creating consolidator: %w", err)
 	}

deploy/.env.production.local.tpl

@@ -8,6 +8,7 @@ if [ "$TF_WORKSPACE" == "prod" ]; then
 
   CONSUMER_TABLE_NAME="prod-upload-api-consumer"
   CONSUMER_TABLE_REGION="us-west-2"
+  CONSUMER_CONSUMER_INDEX_NAME="consumer"
   CONSUMER_CUSTOMER_INDEX_NAME="customer"
 else
   STORAGE_PROVIDER_TABLE_NAME="staging-warm-upload-api-storage-provider"
@@ -18,6 +19,7 @@ else
 
   CONSUMER_TABLE_NAME="staging-warm-upload-api-consumer"
   CONSUMER_TABLE_REGION="us-east-2"
+  CONSUMER_CONSUMER_INDEX_NAME="consumer"
   CONSUMER_CUSTOMER_INDEX_NAME="customer"
 fi
 %>
@@ -30,4 +32,5 @@ CUSTOMER_TABLE_REGION=<%= $CUSTOMER_TABLE_REGION %>
 
 CONSUMER_TABLE_NAME=<%= $CONSUMER_TABLE_NAME %>
 CONSUMER_TABLE_REGION=<%= $CONSUMER_TABLE_REGION %>
+CONSUMER_CONSUMER_INDEX_NAME=<%= $CONSUMER_CONSUMER_INDEX_NAME %>
 CONSUMER_CUSTOMER_INDEX_NAME=<%= $CONSUMER_CUSTOMER_INDEX_NAME %>

deploy/app/external.tf

@@ -49,6 +49,7 @@ data "aws_iam_policy_document" "task_external_dynamodb_scan_query_document" {
       data.aws_dynamodb_table.storage_provider_table.arn,
       data.aws_dynamodb_table.customer_table.arn,
       data.aws_dynamodb_table.consumer_table.arn,
+      "${data.aws_dynamodb_table.consumer_table.arn}/index/consumer",
       "${data.aws_dynamodb_table.consumer_table.arn}/index/customer",
     ]
   }

internal/config/config.go

@@ -30,7 +30,9 @@ type Config struct {
 	CustomerTableRegion            string     `mapstructure:"customer_table_region" validate:"required"`
 	ConsumerTableName              string     `mapstructure:"consumer_table_name" validate:"required"`
 	ConsumerTableRegion            string     `mapstructure:"consumer_table_region" validate:"required"`
+	ConsumerConsumerIndexName      string     `mapstructure:"consumer_consumer_index_name" validate:"required"`
 	ConsumerCustomerIndexName      string     `mapstructure:"consumer_customer_index_name" validate:"required"`
+	KnownProviders                 []string   `mapstructure:"known_providers"`
 }
 
 func Load(ctx context.Context) (*Config, error) {

internal/consolidator/consolidator.go

@@ -6,6 +6,7 @@ import (
 	"iter"
 	"net/http"
 	"net/url"
+	"slices"
 	"strings"
 	"time"
 
@@ -32,6 +33,7 @@ import (
 	"go.opentelemetry.io/otel/metric"
 
 	"github.com/storacha/etracker/internal/db/consolidated"
+	"github.com/storacha/etracker/internal/db/consumer"
 	"github.com/storacha/etracker/internal/db/egress"
 	"github.com/storacha/etracker/internal/db/spacestats"
 	"github.com/storacha/etracker/internal/metrics"
@@ -46,6 +48,8 @@ type Consolidator struct {
 	egressTable           egress.EgressTable
 	consolidatedTable     consolidated.ConsolidatedTable
 	spaceStatsTable       spacestats.SpaceStatsTable
+	consumerTable         consumer.ConsumerTable
+	knownProviders        []string
 	ucantoSrv             ucanto.ServerView[ucanto.Service]
 	retrieveValidationCtx validator.ValidationContext[content.RetrieveCaveats]
 	httpClient            *http.Client
@@ -59,6 +63,8 @@ func New(
 	egressTable egress.EgressTable,
 	consolidatedTable consolidated.ConsolidatedTable,
 	spaceStatsTable spacestats.SpaceStatsTable,
+	consumerTable consumer.ConsumerTable,
+	knownProviders []string,
 	interval time.Duration,
 	batchSize int,
 	presolver validator.PrincipalResolver,
@@ -84,6 +90,8 @@ func New(
 		egressTable:           egressTable,
 		consolidatedTable:     consolidatedTable,
 		spaceStatsTable:       spaceStatsTable,
+		consumerTable:         consumerTable,
+		knownProviders:        knownProviders,
 		retrieveValidationCtx: retrieveValidationCtx,
 		httpClient:            &http.Client{Timeout: 30 * time.Second},
 		interval:              interval,
@@ -318,7 +326,7 @@ func (c *Consolidator) ucanConsolidateHandler(
 			continue
 		}
 
-		cap, err := validateRetrievalReceipt(ctx, requesterNode, rcpt, c.retrieveValidationCtx)
+		cap, err := validateRetrievalReceipt(ctx, requesterNode, rcpt, c.retrieveValidationCtx, c.consumerTable, c.knownProviders)
 		if err != nil {
 			log.Warnf("Invalid receipt: %v", err)
 			continue
@@ -415,6 +423,8 @@ func validateRetrievalReceipt(
 	requesterNode did.DID,
 	rcpt receipt.AnyReceipt,
 	validationCtx validator.ValidationContext[content.RetrieveCaveats],
+	consumerTable consumer.ConsumerTable,
+	knownProviders []string,
 ) (ucan.Capability[content.RetrieveCaveats], error) {
 	// Confirm the receipt is not a failure receipt
 	_, x := result.Unwrap(rcpt.Out())
@@ -461,6 +471,16 @@ func validateRetrievalReceipt(
 		return nil, fmt.Errorf("original invocation is not a %s invocation, but a %s one", content.RetrieveAbility, cap.Can())
 	}
 
+	// Check the space has been provisioned by the upload service
+	space := cap.With()
+	consumer, err := consumerTable.Get(ctx, space)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get consumer: %w", err)
+	}
+	if !slices.Contains(knownProviders, consumer.Provider.String()) {
+		return nil, fmt.Errorf("unknown space provider %s", consumer.Provider)
+	}
+
 	// Verify the delegation chain
 	auth, verr := validator.Access(ctx, inv, validationCtx)
 	if verr != nil {

internal/consolidator/consolidator_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/ipfs/go-cid"
 	"github.com/ipld/go-ipld-prime"
+	"github.com/storacha/etracker/internal/db/consumer"
 	"github.com/storacha/go-libstoracha/capabilities/space/content"
 	"github.com/storacha/go-libstoracha/testutil"
 	"github.com/storacha/go-ucanto/core/dag/blockstore"
@@ -21,13 +22,38 @@ import (
 	"github.com/storacha/go-ucanto/core/receipt/ran"
 	"github.com/storacha/go-ucanto/core/result"
 	"github.com/storacha/go-ucanto/core/result/failure"
+	"github.com/storacha/go-ucanto/did"
 	"github.com/storacha/go-ucanto/principal/ed25519/verifier"
 	"github.com/storacha/go-ucanto/ucan"
 	"github.com/storacha/go-ucanto/validator"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
 
+var _ consumer.ConsumerTable = (*mockConsumerTable)(nil)
+
+type mockConsumerTable struct {
+	t        *testing.T
+	provider did.DID
+}
+
+func (m *mockConsumerTable) Get(ctx context.Context, space string) (consumer.Consumer, error) {
+	s, err := did.Parse(space)
+	if err != nil {
+		return consumer.Consumer{}, err
+	}
+
+	return consumer.Consumer{
+		ID:           s,
+		Provider:     m.provider,
+		Subscription: testutil.RandomCID(m.t).String(),
+	}, nil
+}
+
+func (m *mockConsumerTable) ListByCustomer(ctx context.Context, customer did.DID) ([]did.DID, error) {
+	return []did.DID{}, nil
+}
+
 func TestValidateRetrievalReceipt(t *testing.T) {
 	vCtx := validator.NewValidationContext(
 		testutil.Service.Verifier(),
@@ -45,6 +71,11 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		},
 	)
 
+	knownProvider, err := did.Parse("did:web:up.test.storacha.network")
+	require.NoError(t, err)
+
+	consumerTable := &mockConsumerTable{t: t, provider: knownProvider}
+
 	space := testutil.RandomSigner(t)
 	randBytes := testutil.RandomBytes(t, 256)
 	blob := struct {
@@ -95,7 +126,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		cap, err := validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		cap, err := validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		require.NoError(t, err)
 		assert.Equal(t, content.RetrieveAbility, cap.Can())
 	})
@@ -108,7 +139,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		assert.ErrorContains(t, err, "receipt is a failure receipt")
 	})
 
@@ -121,7 +152,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		assert.ErrorContains(t, err, "receipt is not issued by the requester node")
 	})
 
@@ -137,7 +168,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		// Tamper with the receipt to change its result
 		tamperReceiptResult(t, rcpt)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		assert.ErrorContains(t, err, "receipt signature is invalid")
 	})
 
@@ -149,7 +180,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		assert.ErrorContains(t, err, "original retrieve invocation must be attached to the receipt")
 	})
 
@@ -173,11 +204,28 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		expectedErr := "original invocation is not a " + content.RetrieveAbility + " invocation, but a other/ability one"
 		assert.ErrorContains(t, err, expectedErr)
 	})
 
+	t.Run("wrong space provider", func(t *testing.T) {
+		rcpt, err := receipt.Issue(
+			storageNode,
+			result.Ok[content.RetrieveOk, failure.IPLDBuilderFailure](content.RetrieveOk{}),
+			ran.FromInvocation(inv),
+		)
+		require.NoError(t, err)
+
+		otherProvider, err := did.Parse("did:web:up.other.net")
+		require.NoError(t, err)
+
+		consumerTable := &mockConsumerTable{t: t, provider: otherProvider}
+
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
+		assert.ErrorContains(t, err, "unknown space provider")
+	})
+
 	t.Run("invalid delegation chain", func(t *testing.T) {
 		// Bob invokes on the space, but the proof is from the space to Alice
 		bobInvokes, err := invocation.Invoke(
@@ -201,7 +249,7 @@ func TestValidateRetrievalReceipt(t *testing.T) {
 		)
 		require.NoError(t, err)
 
-		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx)
+		_, err = validateRetrievalReceipt(context.Background(), storageNode.DID(), rcpt, vCtx, consumerTable, []string{knownProvider.String()})
 		assert.ErrorContains(t, err, "invalid delegation chain")
 	})
 }

internal/db/consumer/consumer.go

@@ -6,6 +6,13 @@ import (
 	"github.com/storacha/go-ucanto/did"
 )
 
+type Consumer struct {
+	ID           did.DID
+	Provider     did.DID
+	Subscription string
+}
+
 type ConsumerTable interface {
+	Get(ctx context.Context, consumerID string) (Consumer, error)
 	ListByCustomer(ctx context.Context, customerID did.DID) ([]did.DID, error)
 }

internal/db/consumer/dynamodb.go

@@ -16,11 +16,38 @@ var _ ConsumerTable = (*DynamoConsumerTable)(nil)
 type DynamoConsumerTable struct {
 	client            *dynamodb.Client
 	tableName         string
+	consumerIndexName string
 	customerIndexName string
 }
 
-func NewDynamoConsumerTable(client *dynamodb.Client, tableName string, customerIndexName string) *DynamoConsumerTable {
-	return &DynamoConsumerTable{client, tableName, customerIndexName}
+func NewDynamoConsumerTable(client *dynamodb.Client, tableName, consumerIndexName, customerIndexName string) *DynamoConsumerTable {
+	return &DynamoConsumerTable{client, tableName, consumerIndexName, customerIndexName}
+}
+
+func (d *DynamoConsumerTable) Get(ctx context.Context, consumerID string) (Consumer, error) {
+	// Query the consumer index to get the item by consumer ID
+	result, err := d.client.Query(ctx, &dynamodb.QueryInput{
+		TableName:              aws.String(d.tableName),
+		IndexName:              aws.String(d.consumerIndexName),
+		KeyConditionExpression: aws.String("consumer = :consumer"),
+		ExpressionAttributeValues: map[string]types.AttributeValue{
+			":consumer": &types.AttributeValueMemberS{Value: consumerID},
+		},
+	})
+	if err != nil {
+		return Consumer{}, fmt.Errorf("querying consumer by ID: %w", err)
+	}
+
+	if len(result.Items) == 0 {
+		return Consumer{}, fmt.Errorf("consumer not found: %s", consumerID)
+	}
+
+	consumer, err := d.unmarshalConsumer(result.Items[0])
+	if err != nil {
+		return Consumer{}, fmt.Errorf("unmarshaling consumer: %w", err)
+	}
+
+	return *consumer, nil
 }
 
 func (d *DynamoConsumerTable) ListByCustomer(ctx context.Context, customerID did.DID) ([]did.DID, error) {
@@ -55,7 +82,7 @@ func (d *DynamoConsumerTable) ListByCustomer(ctx context.Context, customerID did
 			if err != nil {
 				return nil, err
 			}
-			consumers = append(consumers, consumer)
+			consumers = append(consumers, consumer.ID)
 		}
 
 		// Check if there are more results to fetch
@@ -70,19 +97,33 @@ func (d *DynamoConsumerTable) ListByCustomer(ctx context.Context, customerID did
 
 // consumerRecord is the internal struct for unmarshaling from DynamoDB
 type consumerRecord struct {
-	Consumer string `dynamodbav:"consumer"`
+	Consumer     string `dynamodbav:"consumer"`
+	Provider     string `dynamodbav:"provider,omitempty"`
+	Subscription string `dynamodbav:"subscription,omitempty"`
 }
 
-func (d *DynamoConsumerTable) unmarshalConsumer(item map[string]types.AttributeValue) (did.DID, error) {
+func (d *DynamoConsumerTable) unmarshalConsumer(item map[string]types.AttributeValue) (*Consumer, error) {
 	var record consumerRecord
 	if err := attributevalue.UnmarshalMap(item, &record); err != nil {
-		return did.DID{}, fmt.Errorf("unmarshaling consumer record: %w", err)
+		return nil, fmt.Errorf("unmarshaling consumer record: %w", err)
 	}
 
 	consumerDID, err := did.Parse(record.Consumer)
 	if err != nil {
-		return did.DID{}, fmt.Errorf("parsing consumer DID: %w", err)
+		return nil, fmt.Errorf("parsing consumer DID: %w", err)
+	}
+
+	providerDID := did.Undef
+	if record.Provider != "" {
+		providerDID, err = did.Parse(record.Provider)
+		if err != nil {
+			return nil, fmt.Errorf("parsing provider DID: %w", err)
+		}
 	}
 
-	return consumerDID, nil
+	return &Consumer{
+		ID:           consumerDID,
+		Provider:     providerDID,
+		Subscription: record.Subscription,
+	}, nil
 }