Enforce API payload size limits and query result caps (#1630)

docs/configuration.md

@@ -587,6 +587,39 @@ to `<params>_<short_code>` (params still apply when provided; short-code overrid
 
 ---
 
+### Optional: token.storage.maxPayloadSize
+
+Maximum serialised size, in bytes, accepted by a single transaction-store write
+(`AddTokenRequest`, `AddTransaction`, `AddMovement`). Writes whose payload exceeds
+this size are rejected before reaching the database. A value of `0` disables the
+check. When the key is absent, the store default (4 MiB) applies.
+
+```yaml
+token:
+  storage:
+    maxPayloadSize: 4194304   # 4 MiB
+```
+
+---
+
+### Optional: token.storage.maxPageSize
+
+Maximum number of rows a single transaction-store read may return. It bounds the
+pages accepted by transaction queries and caps token-request queries so an
+unbounded scan cannot monopolise database resources. When the key is absent, the
+store default (1000) applies.
+
+```yaml
+token:
+  storage:
+    maxPageSize: 1000
+```
+
+See [Transaction Store API Bounds](services/storage/api-bounds.md) for details,
+including why movement/balance queries are intentionally not row-capped.
+
+---
+
 ### Optional: token.tms.<name>.services.storage.cleanup
 
 If not specified, the default configuration is:

docs/services/storage.md

@@ -314,3 +314,10 @@ The cleanup service supports both PostgreSQL and SQLite backends, with different
 Cleanup behavior is controlled by the configuration section. See the [Configuration Guide](../configuration.md) for detailed parameter descriptions and tuning recommendations.
 
 See the [Configuration Guide](../configuration.md), Section `Optional: token.tms.<name>.services.network.fabric.recovery`, for detailed parameter descriptions and tuning recommendations.
+
+## Transaction Store API Bounds
+
+The transaction store enforces upper bounds on its write and read paths (maximum
+serialised write payload, mandatory bounded pagination on transaction queries,
+and a hard cap on token-request queries) so a single oversized write or unbounded
+scan cannot monopolise database resources. See [**Transaction Store API Bounds**](storage/api-bounds.md).

docs/services/storage/api-bounds.md

@@ -0,0 +1,75 @@
+# Transaction Store API Bounds
+
+The transaction store (shared by **TTXDB** and **AuditDB**) enforces upper bounds
+on its write and read paths so that a single oversized write or an unbounded
+table scan cannot monopolise database resources. The checks live at the single
+SQL implementation in `token/services/storage/db/sql/common/transactions.go`,
+which is the only implementor of the write/read interfaces, so both the owner
+(TTXDB) and auditor (AuditDB) stores are covered.
+
+## Write payload limit
+
+`AddTokenRequest`, `AddTransaction`, and `AddMovement` reject a call whose
+serialised payload exceeds a maximum size (in bytes):
+
+- **`AddTokenRequest`** — measures the raw token request plus the marshalled
+  application and public metadata (`len(tr) + len(applicationMetadata) + len(publicMetadata)`).
+- **`AddTransaction`** / **`AddMovement`** — measure the summed byte length of each
+  record's fields (transaction/movement IDs, enrollment/sender/recipient IDs,
+  token type, and the decimal amount).
+
+The limit defaults to `DefaultMaxPayloadSize` (4 MiB). A value of `0` disables the
+check. It can be set from the node configuration under `token.storage.maxPayloadSize`
+(bytes), or programmatically with the `WithMaxPayloadSize` store option (same
+pattern as `auditdb.WithLocker`):
+
+```yaml
+token:
+  storage:
+    maxPayloadSize: 4194304   # 4 MiB; 0 disables the check
+```
+
+```go
+store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi,
+    common.WithMaxPayloadSize(8<<20)) // 8 MiB
+```
+
+## Read result caps
+
+- **`QueryTransactions`** requires a bounded pagination argument. A `nil`
+  pagination and the unbounded `pagination.None()` are rejected; an offset page
+  whose size exceeds the store maximum is also rejected. Callers that need the
+  full result set page through it (see `pagination.Offset` and
+  `PageIterator.Pagination.Next`).
+- **`QueryTokenRequests`** applies a hard `LIMIT` equal to the store maximum page
+  size, bounding the scan without changing its signature.
+- **`QueryMovements`** is intentionally **not** capped. It feeds balance
+  aggregation (`HoldingsFilter.Sum`, `PaymentsFilter.Sum`), where truncating rows
+  would silently corrupt balances. It is already constrained by its required
+  filter predicates (enrollment IDs, token types, statuses).
+
+The read cap defaults to `DefaultMaxPageSize` (1000 rows). It can be set from the
+node configuration under `token.storage.maxPageSize`, or programmatically with
+the `WithMaxPageSize` store option:
+
+```yaml
+token:
+  storage:
+    maxPageSize: 1000
+```
+
+```go
+store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi,
+    common.WithMaxPageSize(500))
+```
+
+## Notes
+
+- When a configuration key is absent, the store default applies. Both keys are
+  read once at driver construction (`LoadStorageConfig`) and applied to the owner
+  (TTXDB) and auditor (AuditDB) transaction stores in the SQLite and PostgreSQL
+  drivers.
+- The write payload measurement for transaction/movement records uses summed
+  field lengths rather than a full serialisation, keeping the write hot path
+  cheap; those records are internally derived, so the meaningful oversized-write
+  vector is the raw token request handled by `AddTokenRequest`.

integration/token/fungible/views/history.go

@@ -17,11 +17,44 @@ import (
 	"github.com/LFDT-Panurus/panurus/token/services/ttx"
 	token2 "github.com/LFDT-Panurus/panurus/token/token"
 	"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
+	driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/assert"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators"
 	"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
 )
 
+// historyPageSize bounds each page when listing all transactions; the storage
+// layer rejects unbounded queries, so these views page through the full set.
+const historyPageSize = 100
+
+// collectAllTransactions pages through queryPage (one page per call) and returns
+// every record. queryPage runs the underlying query for the given pagination.
+func collectAllTransactions(queryPage func(driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error)) ([]*ttxdb.TransactionRecord, error) {
+	var page driver2.Pagination
+	page, err := pagination.Offset(0, historyPageSize)
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to create pagination")
+	}
+	var all []*ttxdb.TransactionRecord
+	for {
+		items, err := queryPage(page)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed querying transactions")
+		}
+		records, err := iterators.ReadAllPointers(items)
+		if err != nil {
+			return nil, errors.Wrapf(err, "failed reading transactions")
+		}
+		all = append(all, records...)
+		if len(records) < historyPageSize {
+			return all, nil
+		}
+		if page, err = page.Next(); err != nil {
+			return nil, errors.Wrapf(err, "failed advancing pagination")
+		}
+	}
+}
+
 // ListIssuedTokens contains the input to query the list of issued tokens
 type ListIssuedTokens struct {
 	// Wallet whose identities own the token
@@ -143,12 +176,14 @@ func (p *ListAuditedTransactionsView) Call(context view.Context) (any, error) {
 		return nil, errors.Wrapf(err, "failed to get auditor instance")
 	}
 
-	it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, pagination.None())
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed querying transactions")
-	}
+	return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) {
+		it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, page)
+		if err != nil {
+			return nil, err
+		}
 
-	return iterators.ReadAllPointers(it.Items)
+		return it.Items, nil
+	})
 }
 
 type ListAuditedTransactionsViewFactory struct{}
@@ -184,21 +219,24 @@ func (p *ListAcceptedTransactionsView) Call(context view.Context) (any, error) {
 	tms, err := token.GetManagementService(context, ServiceOpts(p.TMSID)...)
 	assert.NoError(err, "failed getting management service")
 	owner := ttx.NewOwner(context, tms)
-	it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{
-		SenderWallet:    p.SenderWallet,
-		RecipientWallet: p.RecipientWallet,
-		From:            p.From,
-		To:              p.To,
-		ActionTypes:     p.ActionTypes,
-		Statuses:        p.Statuses,
-		IDs:             p.IDs,
-		SearchDirection: p.SearchDirection,
-	}, pagination.None())
-	if err != nil {
-		return nil, errors.Wrapf(err, "failed querying transactions")
-	}
 
-	return iterators.ReadAllPointers(it.Items)
+	return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) {
+		it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{
+			SenderWallet:    p.SenderWallet,
+			RecipientWallet: p.RecipientWallet,
+			From:            p.From,
+			To:              p.To,
+			ActionTypes:     p.ActionTypes,
+			Statuses:        p.Statuses,
+			IDs:             p.IDs,
+			SearchDirection: p.SearchDirection,
+		}, page)
+		if err != nil {
+			return nil, err
+		}
+
+		return it.Items, nil
+	})
 }
 
 type ListAcceptedTransactionsViewFactory struct{}

token/services/storage/db/common/checks.go

@@ -15,6 +15,7 @@ import (
 	"github.com/LFDT-Panurus/panurus/token/services/logging"
 	"github.com/LFDT-Panurus/panurus/token/services/network"
 	"github.com/LFDT-Panurus/panurus/token/services/storage/db/driver"
+	"github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination"
 	"github.com/LFDT-Panurus/panurus/token/services/tokens"
 	"github.com/LFDT-Panurus/panurus/token/services/utils"
 	token2 "github.com/LFDT-Panurus/panurus/token/token"
@@ -27,6 +28,10 @@ var (
 	logger = logging.MustGetLogger()
 )
 
+// transactionsCheckPageSize is the page size used to iterate all transactions
+// during integrity checks; kept below the storage layer's max page size.
+const transactionsCheckPageSize = 100
+
 type TokenTransactionDB interface {
 	GetTokenRequest(ctx context.Context, txID string) ([]byte, error)
 	Transactions(ctx context.Context, params driver.QueryTransactionsParams, pagination driver2.Pagination) (*driver2.PageIterator[*driver.TransactionRecord], error)
@@ -115,52 +120,77 @@ func (a *DefaultCheckers) CheckTransactions(ctx context.Context) ([]string, erro
 		return nil, errors.WithMessagef(err, "failed to get ledger [%s]", tms.ID())
 	}
 
-	it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, nil)
+	// Iterate over transactions one bounded page at a time. A single unbounded
+	// query is rejected by the storage layer, so we page through the whole set
+	// using offset pagination and stop once a page comes back short.
+	var page driver2.Pagination
+	page, err = pagination.Offset(0, transactionsCheckPageSize)
 	if err != nil {
-		return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
+		return nil, errors.WithMessagef(err, "failed to create pagination [%s]", tms.ID())
 	}
-	defer it.Items.Close()
 	for {
-		transactionRecord, err := it.Items.Next()
-		if err != nil {
-			return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
-		}
-		if transactionRecord == nil {
-			break
-		}
-
-		tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID)
-		if err != nil {
-			return nil, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID)
-		}
-		if tokenRequest == nil {
-			return nil, errors.Errorf("token request [%s] is nil", transactionRecord.TxID)
-		}
-
-		// check the ledger
-		lVC, _, err := l.Status(transactionRecord.TxID)
-		if err != nil {
-			lVC = network.Unknown
-		}
-		switch {
-		case transactionRecord.Status == driver.Confirmed && lVC != network.Valid:
+		count, err := func() (int, error) {
+			it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, page)
 			if err != nil {
-				errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
+				return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
 			}
-			errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
-		case transactionRecord.Status == driver.Deleted && lVC != network.Invalid:
-			if lVC != network.Unknown || transactionRecord.Status != driver.Deleted {
+			defer it.Items.Close()
+			count := 0
+			for {
+				transactionRecord, err := it.Items.Next()
+				if err != nil {
+					return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
+				}
+				if transactionRecord == nil {
+					break
+				}
+				count++
+
+				tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID)
+				if err != nil {
+					return 0, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID)
+				}
+				if tokenRequest == nil {
+					return 0, errors.Errorf("token request [%s] is nil", transactionRecord.TxID)
+				}
+
+				// check the ledger
+				lVC, _, err := l.Status(transactionRecord.TxID)
 				if err != nil {
-					errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
+					lVC = network.Unknown
+				}
+				switch {
+				case transactionRecord.Status == driver.Confirmed && lVC != network.Valid:
+					if err != nil {
+						errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
+					}
+					errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
+				case transactionRecord.Status == driver.Deleted && lVC != network.Invalid:
+					if lVC != network.Unknown || transactionRecord.Status != driver.Deleted {
+						if err != nil {
+							errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
+						}
+						errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
+					}
+				case transactionRecord.Status == driver.Unknown && lVC != network.Unknown:
+					errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
+				case transactionRecord.Status == driver.Pending && lVC == network.Busy:
+					// this is fine, let's continue
+				case transactionRecord.Status == driver.Pending && lVC != network.Unknown:
+					errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
 				}
-				errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
 			}
-		case transactionRecord.Status == driver.Unknown && lVC != network.Unknown:
-			errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
-		case transactionRecord.Status == driver.Pending && lVC == network.Busy:
-			// this is fine, let's continue
-		case transactionRecord.Status == driver.Pending && lVC != network.Unknown:
-			errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
+
+			return count, nil
+		}()
+		if err != nil {
+			return nil, err
+		}
+		if count < transactionsCheckPageSize {
+			break
+		}
+		if page, err = page.Next(); err != nil {
+			return nil, errors.WithMessagef(err, "failed advancing pagination [%s]", tms.ID())
 		}
 	}