cerc-io/lotus
16
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 27 Wiki Activity

Add rate limiting to the lotus gateway

Browse Source
This commit is contained in:
Cory Schwartz 2022-04-19 18:35:19 -07:00
parent a3a3fefb1c
commit fc42d03ea4
2 changed files with 182 additions and 25 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
cmd/lotus-gateway/main.go
View File

@ -133,6 +133,16 @@ var runCmd = &cli.Command{
Usage: "maximum number of blocks to search back through for message inclusion",
Value: int64(gateway.DefaultStateWaitLookbackLimit),
},
&cli.Int64Flag{
Name: "rate-limit",
Usage: "rate-limit API calls. Use 0 to disable",
Value: 0,
},
&cli.DurationFlag{
Name: "rate-limit-timeout",
Usage: "the maximum time to wait for the rate limter before returning an error to clients",
Value: gateway.DefaultRateLimitTimeout,
},
},
Action: func(cctx *cli.Context) error {
log.Info("Starting lotus gateway")
@ -151,9 +161,11 @@ var runCmd = &cli.Command{
defer closer()
var (
lookbackCap = cctx.Duration("api-max-lookback")
address = cctx.String("listen")
waitLookback = abi.ChainEpoch(cctx.Int64("api-wait-lookback-limit"))
lookbackCap = cctx.Duration("api-max-lookback")
address = cctx.String("listen")
waitLookback = abi.ChainEpoch(cctx.Int64("api-wait-lookback-limit"))
rateLimit = cctx.Int64("rate-limit")
rateLimitTimeout = cctx.Duration("rate-limit-timeout")
)
serverOptions := make([]jsonrpc.ServerOption, 0)
@ -173,7 +185,7 @@ var runCmd = &cli.Command{
return xerrors.Errorf("failed to convert endpoint address to multiaddr: %w", err)
}
gwapi := gateway.NewNode(api, lookbackCap, waitLookback)
gwapi := gateway.NewNode(api, lookbackCap, waitLookback, rateLimit, rateLimitTimeout)
h, err := gateway.Handler(gwapi, serverOptions...)
if err != nil {
return xerrors.Errorf("failed to set up gateway HTTP handler")

187
gateway/node.go
View File

@ -6,6 +6,7 @@ import (
"time"
"github.com/ipfs/go-cid"
"golang.org/x/time/rate"
"golang.org/x/xerrors"
"github.com/filecoin-project/go-address"
@ -27,6 +28,11 @@ import (
const (
DefaultLookbackCap = time.Hour * 24
DefaultStateWaitLookbackLimit = abi.ChainEpoch(20)
DefaultRateLimitTimeout = time.Minute * 10
basicRateLimitTokens = 1
walletRateLimitTokens = 1
chainRateLimitTokens = 2
stateRateLimitTokens = 3
)
// TargetAPI defines the API methods that the Node depends on
@ -84,7 +90,10 @@ type Node struct {
target TargetAPI
lookbackCap time.Duration
stateWaitLookbackLimit abi.ChainEpoch
rateLimiter *rate.Limiter
rateLimitTimeout time.Duration
errLookback error
errRateLimit error
}
var (
@ -96,12 +105,21 @@ var (
)
// NewNode creates a new gateway node.
func NewNode(api TargetAPI, lookbackCap time.Duration, stateWaitLookbackLimit abi.ChainEpoch) *Node {
func NewNode(api TargetAPI, lookbackCap time.Duration, stateWaitLookbackLimit abi.ChainEpoch, rateLimit int64, rateLimitTimeout time.Duration) *Node {
var limit rate.Limit
if rateLimit == 0 {
limit = rate.Inf
} else {
limit = rate.Every(time.Second / time.Duration(rateLimit))
}
return &Node{
target: api,
lookbackCap: lookbackCap,
stateWaitLookbackLimit: stateWaitLookbackLimit,
rateLimiter: rate.NewLimiter(limit, stateRateLimitTokens),
rateLimitTimeout: rateLimitTimeout,
errLookback: fmt.Errorf("lookbacks of more than %s are disallowed", lookbackCap),
errRateLimit: fmt.Errorf("this respones has been rate limited"),
}
}
@ -144,53 +162,90 @@ func (gw *Node) checkTimestamp(at time.Time) error {
return nil
}
func (gw *Node) limit(ctx context.Context, tokens int) error {
ctx2, cancel := context.WithTimeout(ctx, gw.rateLimitTimeout)
defer cancel()
err := gw.rateLimiter.WaitN(ctx2, tokens)
if err != nil {
return fmt.Errorf("rate limited: %w", err)
}
return nil
}
func (gw *Node) Version(ctx context.Context) (api.APIVersion, error) {
if err := gw.limit(ctx, basicRateLimitTokens); err != nil {
return api.APIVersion{}, err
}
return gw.target.Version(ctx)
}
func (gw *Node) ChainGetParentMessages(ctx context.Context, c cid.Cid) ([]api.Message, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetParentMessages(ctx, c)
}
func (gw *Node) ChainGetParentReceipts(ctx context.Context, c cid.Cid) ([]*types.MessageReceipt, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetParentReceipts(ctx, c)
}
func (gw *Node) ChainGetBlockMessages(ctx context.Context, c cid.Cid) (*api.BlockMessages, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetBlockMessages(ctx, c)
}
func (gw *Node) ChainHasObj(ctx context.Context, c cid.Cid) (bool, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return false, err
}
return gw.target.ChainHasObj(ctx, c)
}
func (gw *Node) ChainHead(ctx context.Context) (*types.TipSet, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
// TODO: cache and invalidate cache when timestamp is up (or have internal ChainNotify)
return gw.target.ChainHead(ctx)
}
func (gw *Node) ChainGetMessage(ctx context.Context, mc cid.Cid) (*types.Message, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetMessage(ctx, mc)
}
func (gw *Node) ChainGetTipSet(ctx context.Context, tsk types.TipSetKey) (*types.TipSet, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetTipSet(ctx, tsk)
}
func (gw *Node) ChainGetTipSetByHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipSetHeight(ctx, h, tsk); err != nil {
return nil, err
}
return gw.target.ChainGetTipSetByHeight(ctx, h, tsk)
}
func (gw *Node) ChainGetTipSetAfterHeight(ctx context.Context, h abi.ChainEpoch, tsk types.TipSetKey) (*types.TipSet, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipSetHeight(ctx, h, tsk); err != nil {
return nil, err
}
return gw.target.ChainGetTipSetAfterHeight(ctx, h, tsk)
}
@ -224,66 +279,91 @@ func (gw *Node) checkTipSetHeight(ctx context.Context, h abi.ChainEpoch, tsk typ
}
func (gw *Node) ChainGetNode(ctx context.Context, p string) (*api.IpldObject, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetNode(ctx, p)
}
func (gw *Node) ChainNotify(ctx context.Context) (<-chan []*api.HeadChange, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainNotify(ctx)
}
func (gw *Node) ChainGetPath(ctx context.Context, from, to types.TipSetKey) ([]*api.HeadChange, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, from); err != nil {
return nil, xerrors.Errorf("gateway: checking 'from' tipset: %w", err)
}
if err := gw.checkTipsetKey(ctx, to); err != nil {
return nil, xerrors.Errorf("gateway: checking 'to' tipset: %w", err)
}
return gw.target.ChainGetPath(ctx, from, to)
}
func (gw *Node) ChainGetGenesis(ctx context.Context) (*types.TipSet, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainGetGenesis(ctx)
}
func (gw *Node) ChainReadObj(ctx context.Context, c cid.Cid) ([]byte, error) {
if err := gw.limit(ctx, chainRateLimitTokens); err != nil {
return nil, err
}
return gw.target.ChainReadObj(ctx, c)
}
func (gw *Node) GasEstimateMessageGas(ctx context.Context, msg *types.Message, spec *api.MessageSendSpec, tsk types.TipSetKey) (*types.Message, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.GasEstimateMessageGas(ctx, msg, spec, tsk)
}
func (gw *Node) MpoolPush(ctx context.Context, sm *types.SignedMessage) (cid.Cid, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return cid.Cid{}, err
}
// TODO: additional anti-spam checks
return gw.target.MpoolPushUntrusted(ctx, sm)
}
func (gw *Node) MsigGetAvailableBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (types.BigInt, error) {
if err := gw.limit(ctx, walletRateLimitTokens); err != nil {
return types.BigInt{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return types.NewInt(0), err
}
return gw.target.MsigGetAvailableBalance(ctx, addr, tsk)
}
func (gw *Node) MsigGetVested(ctx context.Context, addr address.Address, start types.TipSetKey, end types.TipSetKey) (types.BigInt, error) {
if err := gw.limit(ctx, walletRateLimitTokens); err != nil {
return types.BigInt{}, err
}
if err := gw.checkTipsetKey(ctx, start); err != nil {
return types.NewInt(0), err
}
if err := gw.checkTipsetKey(ctx, end); err != nil {
return types.NewInt(0), err
}
return gw.target.MsigGetVested(ctx, addr, start, end)
}
func (gw *Node) MsigGetVestingSchedule(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MsigVesting, error) {
if err := gw.limit(ctx, walletRateLimitTokens); err != nil {
return api.MsigVesting{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return api.MsigVesting{}, err
}
@ -291,78 +371,99 @@ func (gw *Node) MsigGetVestingSchedule(ctx context.Context, addr address.Address
}
func (gw *Node) MsigGetPending(ctx context.Context, addr address.Address, tsk types.TipSetKey) ([]*api.MsigTransaction, error) {
if err := gw.limit(ctx, walletRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.MsigGetPending(ctx, addr, tsk)
}
func (gw *Node) StateAccountKey(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return address.Address{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return address.Undef, err
}
return gw.target.StateAccountKey(ctx, addr, tsk)
}
func (gw *Node) StateDealProviderCollateralBounds(ctx context.Context, size abi.PaddedPieceSize, verified bool, tsk types.TipSetKey) (api.DealCollateralBounds, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return api.DealCollateralBounds{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return api.DealCollateralBounds{}, err
}
return gw.target.StateDealProviderCollateralBounds(ctx, size, verified, tsk)
}
func (gw *Node) StateGetActor(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*types.Actor, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.StateGetActor(ctx, actor, tsk)
}
func (gw *Node) StateListMiners(ctx context.Context, tsk types.TipSetKey) ([]address.Address, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.StateListMiners(ctx, tsk)
}
func (gw *Node) StateLookupID(ctx context.Context, addr address.Address, tsk types.TipSetKey) (address.Address, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return address.Address{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return address.Undef, err
}
return gw.target.StateLookupID(ctx, addr, tsk)
}
func (gw *Node) StateMarketBalance(ctx context.Context, addr address.Address, tsk types.TipSetKey) (api.MarketBalance, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return api.MarketBalance{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return api.MarketBalance{}, err
}
return gw.target.StateMarketBalance(ctx, addr, tsk)
}
func (gw *Node) StateMarketStorageDeal(ctx context.Context, dealId abi.DealID, tsk types.TipSetKey) (*api.MarketDeal, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
return gw.target.StateMarketStorageDeal(ctx, dealId, tsk)
}
func (gw *Node) StateNetworkVersion(ctx context.Context, tsk types.TipSetKey) (network.Version, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return network.VersionMax, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return network.VersionMax, err
}
return gw.target.StateNetworkVersion(ctx, tsk)
}
func (gw *Node) StateSearchMsg(ctx context.Context, from types.TipSetKey, msg cid.Cid, limit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if limit == api.LookbackNoLimit {
limit = gw.stateWaitLookbackLimit
}
@ -372,22 +473,26 @@ func (gw *Node) StateSearchMsg(ctx context.Context, from types.TipSetKey, msg ci
if err := gw.checkTipsetKey(ctx, from); err != nil {
return nil, err
}
return gw.target.StateSearchMsg(ctx, from, msg, limit, allowReplaced)
}
func (gw *Node) StateWaitMsg(ctx context.Context, msg cid.Cid, confidence uint64, limit abi.ChainEpoch, allowReplaced bool) (*api.MsgLookup, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if limit == api.LookbackNoLimit {
limit = gw.stateWaitLookbackLimit
}
if gw.stateWaitLookbackLimit != api.LookbackNoLimit && limit > gw.stateWaitLookbackLimit {
limit = gw.stateWaitLookbackLimit
}
return gw.target.StateWaitMsg(ctx, msg, confidence, limit, allowReplaced)
}
func (gw *Node) StateReadState(ctx context.Context, actor address.Address, tsk types.TipSetKey) (*api.ActorState, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -395,6 +500,9 @@ func (gw *Node) StateReadState(ctx context.Context, actor address.Address, tsk t
}
func (gw *Node) StateMinerPower(ctx context.Context, m address.Address, tsk types.TipSetKey) (*api.MinerPower, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -402,12 +510,19 @@ func (gw *Node) StateMinerPower(ctx context.Context, m address.Address, tsk type
}
func (gw *Node) StateMinerFaults(ctx context.Context, m address.Address, tsk types.TipSetKey) (bitfield.BitField, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return bitfield.BitField{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return bitfield.BitField{}, err
}
return gw.target.StateMinerFaults(ctx, m, tsk)
}
func (gw *Node) StateMinerRecoveries(ctx context.Context, m address.Address, tsk types.TipSetKey) (bitfield.BitField, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return bitfield.BitField{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return bitfield.BitField{}, err
}
@ -415,6 +530,9 @@ func (gw *Node) StateMinerRecoveries(ctx context.Context, m address.Address, tsk
}
func (gw *Node) StateMinerInfo(ctx context.Context, m address.Address, tsk types.TipSetKey) (miner.MinerInfo, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return miner.MinerInfo{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return miner.MinerInfo{}, err
}
@ -422,6 +540,9 @@ func (gw *Node) StateMinerInfo(ctx context.Context, m address.Address, tsk types
}
func (gw *Node) StateMinerDeadlines(ctx context.Context, m address.Address, tsk types.TipSetKey) ([]api.Deadline, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -429,6 +550,9 @@ func (gw *Node) StateMinerDeadlines(ctx context.Context, m address.Address, tsk
}
func (gw *Node) StateMinerAvailableBalance(ctx context.Context, m address.Address, tsk types.TipSetKey) (types.BigInt, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return types.BigInt{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return types.BigInt{}, err
}
@ -436,6 +560,9 @@ func (gw *Node) StateMinerAvailableBalance(ctx context.Context, m address.Addres
}
func (gw *Node) StateMinerProvingDeadline(ctx context.Context, m address.Address, tsk types.TipSetKey) (*dline.Info, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -443,13 +570,19 @@ func (gw *Node) StateMinerProvingDeadline(ctx context.Context, m address.Address
}
func (gw *Node) StateCirculatingSupply(ctx context.Context, tsk types.TipSetKey) (abi.TokenAmount, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return abi.TokenAmount{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return types.BigInt{}, err
return abi.TokenAmount{}, err
}
return gw.target.StateCirculatingSupply(ctx, tsk)
}
func (gw *Node) StateSectorGetInfo(ctx context.Context, maddr address.Address, n abi.SectorNumber, tsk types.TipSetKey) (*miner.SectorOnChainInfo, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -457,6 +590,9 @@ func (gw *Node) StateSectorGetInfo(ctx context.Context, maddr address.Address, n
}
func (gw *Node) StateVerifiedClientStatus(ctx context.Context, addr address.Address, tsk types.TipSetKey) (*abi.StoragePower, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return nil, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return nil, err
}
@ -464,6 +600,9 @@ func (gw *Node) StateVerifiedClientStatus(ctx context.Context, addr address.Addr
}
func (gw *Node) StateVMCirculatingSupplyInternal(ctx context.Context, tsk types.TipSetKey) (api.CirculatingSupply, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return api.CirculatingSupply{}, err
}
if err := gw.checkTipsetKey(ctx, tsk); err != nil {
return api.CirculatingSupply{}, err
}
@ -471,9 +610,15 @@ func (gw *Node) StateVMCirculatingSupplyInternal(ctx context.Context, tsk types.
}
func (gw *Node) WalletVerify(ctx context.Context, k address.Address, msg []byte, sig *crypto.Signature) (bool, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return false, err
}
return sigs.Verify(sig, k, msg) == nil, nil
}
func (gw *Node) WalletBalance(ctx context.Context, k address.Address) (types.BigInt, error) {
if err := gw.limit(ctx, stateRateLimitTokens); err != nil {
return types.BigInt{}, err
}
return gw.target.WalletBalance(ctx, k)
}