Skip to content

Commit e361c80

Browse files
AkramBitarAkramBitar
committed
feat(storage): enforce API payload size limits and query result caps (#1630)
Add upper bounds at the SQL transaction-store chokepoint so a single oversized write or unbounded scan cannot monopolise DB resources: - Write cap: AddTokenRequest/AddTransaction/AddMovement reject payloads over a max size (DefaultMaxPayloadSize 4 MiB; 0 disables). - Read caps: QueryTransactions requires bounded pagination (rejects nil/None); QueryTokenRequests is hard-LIMITed to the max page size (DefaultMaxPageSize 1000). QueryMovements is intentionally excluded. - Limits default via consts and are overridable from config (token.storage.maxPayloadSize / maxPageSize) or WithMax* options. - Migrate unbounded callers (checks.go, history views) to page loops. Signed-off-by: AkramBitar <akram@il.ibm.com>
1 parent 5b5e165 commit e361c80

17 files changed

Lines changed: 637 additions & 91 deletions

File tree

‎docs/configuration.md‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -587,6 +587,39 @@ to `<params>_<short_code>` (params still apply when provided; short-code overrid
587587

588588
---
589589

590+
### Optional: token.storage.maxPayloadSize
591+
592+
Maximum serialised size, in bytes, accepted by a single transaction-store write
593+
(`AddTokenRequest`, `AddTransaction`, `AddMovement`). Writes whose payload exceeds
594+
this size are rejected before reaching the database. A value of `0` disables the
595+
check. When the key is absent, the store default (4 MiB) applies.
596+
597+
```yaml
598+
token:
599+
storage:
600+
maxPayloadSize: 4194304 # 4 MiB
601+
```
602+
603+
---
604+
605+
### Optional: token.storage.maxPageSize
606+
607+
Maximum number of rows a single transaction-store read may return. It bounds the
608+
pages accepted by transaction queries and caps token-request queries so an
609+
unbounded scan cannot monopolise database resources. When the key is absent, the
610+
store default (1000) applies.
611+
612+
```yaml
613+
token:
614+
storage:
615+
maxPageSize: 1000
616+
```
617+
618+
See [Transaction Store API Bounds](services/storage/api-bounds.md) for details,
619+
including why movement/balance queries are intentionally not row-capped.
620+
621+
---
622+
590623
### Optional: token.tms.<name>.services.storage.cleanup
591624

592625
If not specified, the default configuration is:

‎docs/services/storage.md‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -314,3 +314,10 @@ The cleanup service supports both PostgreSQL and SQLite backends, with different
314314
Cleanup behavior is controlled by the configuration section. See the [Configuration Guide](../configuration.md) for detailed parameter descriptions and tuning recommendations.
315315

316316
See the [Configuration Guide](../configuration.md), Section `Optional: token.tms.<name>.services.network.fabric.recovery`, for detailed parameter descriptions and tuning recommendations.
317+
318+
## Transaction Store API Bounds
319+
320+
The transaction store enforces upper bounds on its write and read paths (maximum
321+
serialised write payload, mandatory bounded pagination on transaction queries,
322+
and a hard cap on token-request queries) so a single oversized write or unbounded
323+
scan cannot monopolise database resources. See [**Transaction Store API Bounds**](storage/api-bounds.md).

‎docs/services/storage/api-bounds.md‎

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
# Transaction Store API Bounds
2+
3+
The transaction store (shared by **TTXDB** and **AuditDB**) enforces upper bounds
4+
on its write and read paths so that a single oversized write or an unbounded
5+
table scan cannot monopolise database resources. The checks live at the single
6+
SQL implementation in `token/services/storage/db/sql/common/transactions.go`,
7+
which is the only implementor of the write/read interfaces, so both the owner
8+
(TTXDB) and auditor (AuditDB) stores are covered.
9+
10+
## Write payload limit
11+
12+
`AddTokenRequest`, `AddTransaction`, and `AddMovement` reject a call whose
13+
serialised payload exceeds a maximum size (in bytes):
14+
15+
- **`AddTokenRequest`** — measures the raw token request plus the marshalled
16+
application and public metadata (`len(tr) + len(applicationMetadata) + len(publicMetadata)`).
17+
- **`AddTransaction`** / **`AddMovement`** — measure the summed byte length of each
18+
record's fields (transaction/movement IDs, enrollment/sender/recipient IDs,
19+
token type, and the decimal amount).
20+
21+
The limit defaults to `DefaultMaxPayloadSize` (4 MiB). A value of `0` disables the
22+
check. It can be set from the node configuration under `token.storage.maxPayloadSize`
23+
(bytes), or programmatically with the `WithMaxPayloadSize` store option (same
24+
pattern as `auditdb.WithLocker`):
25+
26+
```yaml
27+
token:
28+
storage:
29+
maxPayloadSize: 4194304 # 4 MiB; 0 disables the check
30+
```
31+
32+
```go
33+
store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi,
34+
common.WithMaxPayloadSize(8<<20)) // 8 MiB
35+
```
36+
37+
## Read result caps
38+
39+
- **`QueryTransactions`** requires a bounded pagination argument. A `nil`
40+
pagination and the unbounded `pagination.None()` are rejected; an offset page
41+
whose size exceeds the store maximum is also rejected. Callers that need the
42+
full result set page through it (see `pagination.Offset` and
43+
`PageIterator.Pagination.Next`).
44+
- **`QueryTokenRequests`** applies a hard `LIMIT` equal to the store maximum page
45+
size, bounding the scan without changing its signature.
46+
- **`QueryMovements`** is intentionally **not** capped. It feeds balance
47+
aggregation (`HoldingsFilter.Sum`, `PaymentsFilter.Sum`), where truncating rows
48+
would silently corrupt balances. It is already constrained by its required
49+
filter predicates (enrollment IDs, token types, statuses).
50+
51+
The read cap defaults to `DefaultMaxPageSize` (1000 rows). It can be set from the
52+
node configuration under `token.storage.maxPageSize`, or programmatically with
53+
the `WithMaxPageSize` store option:
54+
55+
```yaml
56+
token:
57+
storage:
58+
maxPageSize: 1000
59+
```
60+
61+
```go
62+
store, _ := common.NewOwnerTransactionStore(readDB, writeDB, tables, ci, pi,
63+
common.WithMaxPageSize(500))
64+
```
65+
66+
## Notes
67+
68+
- When a configuration key is absent, the store default applies. Both keys are
69+
read once at driver construction (`LoadStorageConfig`) and applied to the owner
70+
(TTXDB) and auditor (AuditDB) transaction stores in the SQLite and PostgreSQL
71+
drivers.
72+
- The write payload measurement for transaction/movement records uses summed
73+
field lengths rather than a full serialisation, keeping the write hot path
74+
cheap; those records are internally derived, so the meaningful oversized-write
75+
vector is the raw token request handled by `AddTokenRequest`.

‎integration/token/fungible/views/history.go‎

Lines changed: 57 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -17,11 +17,44 @@ import (
1717
"github.com/LFDT-Panurus/panurus/token/services/ttx"
1818
token2 "github.com/LFDT-Panurus/panurus/token/token"
1919
"github.com/hyperledger-labs/fabric-smart-client/pkg/utils/errors"
20+
driver2 "github.com/hyperledger-labs/fabric-smart-client/platform/common/driver"
2021
"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/assert"
2122
"github.com/hyperledger-labs/fabric-smart-client/platform/common/utils/collections/iterators"
2223
"github.com/hyperledger-labs/fabric-smart-client/platform/view/view"
2324
)
2425

26+
// historyPageSize bounds each page when listing all transactions; the storage
27+
// layer rejects unbounded queries, so these views page through the full set.
28+
const historyPageSize = 100
29+
30+
// collectAllTransactions pages through queryPage (one page per call) and returns
31+
// every record. queryPage runs the underlying query for the given pagination.
32+
func collectAllTransactions(queryPage func(driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error)) ([]*ttxdb.TransactionRecord, error) {
33+
var page driver2.Pagination
34+
page, err := pagination.Offset(0, historyPageSize)
35+
if err != nil {
36+
return nil, errors.Wrapf(err, "failed to create pagination")
37+
}
38+
var all []*ttxdb.TransactionRecord
39+
for {
40+
items, err := queryPage(page)
41+
if err != nil {
42+
return nil, errors.Wrapf(err, "failed querying transactions")
43+
}
44+
records, err := iterators.ReadAllPointers(items)
45+
if err != nil {
46+
return nil, errors.Wrapf(err, "failed reading transactions")
47+
}
48+
all = append(all, records...)
49+
if len(records) < historyPageSize {
50+
return all, nil
51+
}
52+
if page, err = page.Next(); err != nil {
53+
return nil, errors.Wrapf(err, "failed advancing pagination")
54+
}
55+
}
56+
}
57+
2558
// ListIssuedTokens contains the input to query the list of issued tokens
2659
type ListIssuedTokens struct {
2760
// Wallet whose identities own the token
@@ -143,12 +176,14 @@ func (p *ListAuditedTransactionsView) Call(context view.Context) (any, error) {
143176
return nil, errors.Wrapf(err, "failed to get auditor instance")
144177
}
145178

146-
it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, pagination.None())
147-
if err != nil {
148-
return nil, errors.Wrapf(err, "failed querying transactions")
149-
}
179+
return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) {
180+
it, err := auditor.Transactions(context.Context(), ttxdb.QueryTransactionsParams{From: p.From, To: p.To, SearchDirection: p.SearchDirection}, page)
181+
if err != nil {
182+
return nil, err
183+
}
150184

151-
return iterators.ReadAllPointers(it.Items)
185+
return it.Items, nil
186+
})
152187
}
153188

154189
type ListAuditedTransactionsViewFactory struct{}
@@ -184,21 +219,24 @@ func (p *ListAcceptedTransactionsView) Call(context view.Context) (any, error) {
184219
tms, err := token.GetManagementService(context, ServiceOpts(p.TMSID)...)
185220
assert.NoError(err, "failed getting management service")
186221
owner := ttx.NewOwner(context, tms)
187-
it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{
188-
SenderWallet: p.SenderWallet,
189-
RecipientWallet: p.RecipientWallet,
190-
From: p.From,
191-
To: p.To,
192-
ActionTypes: p.ActionTypes,
193-
Statuses: p.Statuses,
194-
IDs: p.IDs,
195-
SearchDirection: p.SearchDirection,
196-
}, pagination.None())
197-
if err != nil {
198-
return nil, errors.Wrapf(err, "failed querying transactions")
199-
}
200222

201-
return iterators.ReadAllPointers(it.Items)
223+
return collectAllTransactions(func(page driver2.Pagination) (iterators.Iterator[*ttxdb.TransactionRecord], error) {
224+
it, err := owner.Transactions(context.Context(), ttxdb.QueryTransactionsParams{
225+
SenderWallet: p.SenderWallet,
226+
RecipientWallet: p.RecipientWallet,
227+
From: p.From,
228+
To: p.To,
229+
ActionTypes: p.ActionTypes,
230+
Statuses: p.Statuses,
231+
IDs: p.IDs,
232+
SearchDirection: p.SearchDirection,
233+
}, page)
234+
if err != nil {
235+
return nil, err
236+
}
237+
238+
return it.Items, nil
239+
})
202240
}
203241

204242
type ListAcceptedTransactionsViewFactory struct{}

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

Lines changed: 68 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"github.com/LFDT-Panurus/panurus/token/services/logging"
1616
"github.com/LFDT-Panurus/panurus/token/services/network"
1717
"github.com/LFDT-Panurus/panurus/token/services/storage/db/driver"
18+
"github.com/LFDT-Panurus/panurus/token/services/storage/db/sql/query/pagination"
1819
"github.com/LFDT-Panurus/panurus/token/services/tokens"
1920
"github.com/LFDT-Panurus/panurus/token/services/utils"
2021
token2 "github.com/LFDT-Panurus/panurus/token/token"
@@ -27,6 +28,10 @@ var (
2728
logger = logging.MustGetLogger()
2829
)
2930

31+
// transactionsCheckPageSize is the page size used to iterate all transactions
32+
// during integrity checks; kept below the storage layer's max page size.
33+
const transactionsCheckPageSize = 100
34+
3035
type TokenTransactionDB interface {
3136
GetTokenRequest(ctx context.Context, txID string) ([]byte, error)
3237
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
115120
return nil, errors.WithMessagef(err, "failed to get ledger [%s]", tms.ID())
116121
}
117122

118-
it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, nil)
123+
// Iterate over transactions one bounded page at a time. A single unbounded
124+
// query is rejected by the storage layer, so we page through the whole set
125+
// using offset pagination and stop once a page comes back short.
126+
var page driver2.Pagination
127+
page, err = pagination.Offset(0, transactionsCheckPageSize)
119128
if err != nil {
120-
return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
129+
return nil, errors.WithMessagef(err, "failed to create pagination [%s]", tms.ID())
121130
}
122-
defer it.Items.Close()
123131
for {
124-
transactionRecord, err := it.Items.Next()
125-
if err != nil {
126-
return nil, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
127-
}
128-
if transactionRecord == nil {
129-
break
130-
}
131-
132-
tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID)
133-
if err != nil {
134-
return nil, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID)
135-
}
136-
if tokenRequest == nil {
137-
return nil, errors.Errorf("token request [%s] is nil", transactionRecord.TxID)
138-
}
139-
140-
// check the ledger
141-
lVC, _, err := l.Status(transactionRecord.TxID)
142-
if err != nil {
143-
lVC = network.Unknown
144-
}
145-
switch {
146-
case transactionRecord.Status == driver.Confirmed && lVC != network.Valid:
132+
count, err := func() (int, error) {
133+
it, err := a.db.Transactions(ctx, driver.QueryTransactionsParams{}, page)
147134
if err != nil {
148-
errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
135+
return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
149136
}
150-
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
151-
case transactionRecord.Status == driver.Deleted && lVC != network.Invalid:
152-
if lVC != network.Unknown || transactionRecord.Status != driver.Deleted {
137+
defer it.Items.Close()
138+
count := 0
139+
for {
140+
transactionRecord, err := it.Items.Next()
141+
if err != nil {
142+
return 0, errors.WithMessagef(err, "failed querying transactions [%s]", tms.ID())
143+
}
144+
if transactionRecord == nil {
145+
break
146+
}
147+
count++
148+
149+
tokenRequest, err := a.db.GetTokenRequest(ctx, transactionRecord.TxID)
150+
if err != nil {
151+
return 0, errors.WithMessagef(err, "failed getting token request [%s]", transactionRecord.TxID)
152+
}
153+
if tokenRequest == nil {
154+
return 0, errors.Errorf("token request [%s] is nil", transactionRecord.TxID)
155+
}
156+
157+
// check the ledger
158+
lVC, _, err := l.Status(transactionRecord.TxID)
153159
if err != nil {
154-
errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
160+
lVC = network.Unknown
161+
}
162+
switch {
163+
case transactionRecord.Status == driver.Confirmed && lVC != network.Valid:
164+
if err != nil {
165+
errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
166+
}
167+
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is valid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
168+
case transactionRecord.Status == driver.Deleted && lVC != network.Invalid:
169+
if lVC != network.Unknown || transactionRecord.Status != driver.Deleted {
170+
if err != nil {
171+
errorMessages = append(errorMessages, fmt.Sprintf("failed to get ledger transaction status for [%s]: [%s]", transactionRecord.TxID, err))
172+
}
173+
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
174+
}
175+
case transactionRecord.Status == driver.Unknown && lVC != network.Unknown:
176+
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
177+
case transactionRecord.Status == driver.Pending && lVC == network.Busy:
178+
// this is fine, let's continue
179+
case transactionRecord.Status == driver.Pending && lVC != network.Unknown:
180+
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
155181
}
156-
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is invalid for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
157182
}
158-
case transactionRecord.Status == driver.Unknown && lVC != network.Unknown:
159-
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is unknown for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
160-
case transactionRecord.Status == driver.Pending && lVC == network.Busy:
161-
// this is fine, let's continue
162-
case transactionRecord.Status == driver.Pending && lVC != network.Unknown:
163-
errorMessages = append(errorMessages, fmt.Sprintf("transaction record [%s] is busy for vault but not for the ledger [%d]", transactionRecord.TxID, lVC))
183+
184+
return count, nil
185+
}()
186+
if err != nil {
187+
return nil, err
188+
}
189+
if count < transactionsCheckPageSize {
190+
break
191+
}
192+
if page, err = page.Next(); err != nil {
193+
return nil, errors.WithMessagef(err, "failed advancing pagination [%s]", tms.ID())
164194
}
165195
}
166196

0 commit comments

Comments
 (0)