mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-09-19 06:31:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Fix conflicts with upsteam changes

Browse Source
This commit is contained in:
Zaki Manian
2019-07-05 20:23:23 -07:00
parent ce9cde0888 94e0176ac2
commit 165a3d15fb
521 changed files with 41981 additions and 9903 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/proxy/proxy.go
+102 -21
View File
@@ -1,15 +1,19 @@
package proxy
import (
"context"
"net/http"
amino "github.com/tendermint/go-amino"
"github.com/tendermint/tendermint/libs/log"
cmn "github.com/tendermint/tendermint/libs/common"
"github.com/tendermint/tendermint/libs/log"
rpcclient "github.com/tendermint/tendermint/rpc/client"
"github.com/tendermint/tendermint/rpc/core"
ctypes "github.com/tendermint/tendermint/rpc/core/types"
rpcserver "github.com/tendermint/tendermint/rpc/lib/server"
rpctypes "github.com/tendermint/tendermint/rpc/lib/types"
"github.com/tendermint/tendermint/types"
)
const (
@@ -34,16 +38,23 @@ func StartProxy(c rpcclient.Client, listenAddr string, logger log.Logger, maxOpe
mux := http.NewServeMux()
rpcserver.RegisterRPCFuncs(mux, r, cdc, logger)
wm := rpcserver.NewWebsocketManager(r, cdc, rpcserver.EventSubscriber(c))
unsubscribeFromAllEvents := func(remoteAddr string) {
if err := c.UnsubscribeAll(context.Background(), remoteAddr); err != nil {
logger.Error("Failed to unsubscribe from events", "err", err)
}
}
wm := rpcserver.NewWebsocketManager(r, cdc, rpcserver.OnDisconnect(unsubscribeFromAllEvents))
wm.SetLogger(logger)
core.SetLogger(logger)
mux.HandleFunc(wsEndpoint, wm.WebsocketHandler)
l, err := rpcserver.Listen(listenAddr, rpcserver.Config{MaxOpenConnections: maxOpenConnections})
config := rpcserver.DefaultConfig()
config.MaxOpenConnections = maxOpenConnections
l, err := rpcserver.Listen(listenAddr, config)
if err != nil {
return err
}
return rpcserver.StartHTTPServer(l, mux, logger)
return rpcserver.StartHTTPServer(l, mux, logger, config)
}
// RPCRoutes just routes everything to the given client, as if it were
@@ -51,30 +62,100 @@ func StartProxy(c rpcclient.Client, listenAddr string, logger log.Logger, maxOpe
//
// if we want security, the client must implement it as a secure client
func RPCRoutes(c rpcclient.Client) map[string]*rpcserver.RPCFunc {
return map[string]*rpcserver.RPCFunc{
// Subscribe/unsubscribe are reserved for websocket events.
// We can just use the core tendermint impl, which uses the
// EventSwitch we registered in NewWebsocketManager above
"subscribe": rpcserver.NewWSRPCFunc(core.Subscribe, "query"),
"unsubscribe": rpcserver.NewWSRPCFunc(core.Unsubscribe, "query"),
"subscribe": rpcserver.NewWSRPCFunc(c.(Wrapper).SubscribeWS, "query"),
"unsubscribe": rpcserver.NewWSRPCFunc(c.(Wrapper).UnsubscribeWS, "query"),
"unsubscribe_all": rpcserver.NewWSRPCFunc(c.(Wrapper).UnsubscribeAllWS, ""),
// info API
"status": rpcserver.NewRPCFunc(c.Status, ""),
"blockchain": rpcserver.NewRPCFunc(c.BlockchainInfo, "minHeight,maxHeight"),
"genesis": rpcserver.NewRPCFunc(c.Genesis, ""),
"block": rpcserver.NewRPCFunc(c.Block, "height"),
"commit": rpcserver.NewRPCFunc(c.Commit, "height"),
"tx": rpcserver.NewRPCFunc(c.Tx, "hash,prove"),
"validators": rpcserver.NewRPCFunc(c.Validators, ""),
"status": rpcserver.NewRPCFunc(makeStatusFunc(c), ""),
"blockchain": rpcserver.NewRPCFunc(makeBlockchainInfoFunc(c), "minHeight,maxHeight"),
"genesis": rpcserver.NewRPCFunc(makeGenesisFunc(c), ""),
"block": rpcserver.NewRPCFunc(makeBlockFunc(c), "height"),
"commit": rpcserver.NewRPCFunc(makeCommitFunc(c), "height"),
"tx": rpcserver.NewRPCFunc(makeTxFunc(c), "hash,prove"),
"validators": rpcserver.NewRPCFunc(makeValidatorsFunc(c), "height"),
// broadcast API
"broadcast_tx_commit": rpcserver.NewRPCFunc(c.BroadcastTxCommit, "tx"),
"broadcast_tx_sync": rpcserver.NewRPCFunc(c.BroadcastTxSync, "tx"),
"broadcast_tx_async": rpcserver.NewRPCFunc(c.BroadcastTxAsync, "tx"),
"broadcast_tx_commit": rpcserver.NewRPCFunc(makeBroadcastTxCommitFunc(c), "tx"),
"broadcast_tx_sync": rpcserver.NewRPCFunc(makeBroadcastTxSyncFunc(c), "tx"),
"broadcast_tx_async": rpcserver.NewRPCFunc(makeBroadcastTxAsyncFunc(c), "tx"),
// abci API
"abci_query": rpcserver.NewRPCFunc(c.ABCIQuery, "path,data,prove"),
"abci_info": rpcserver.NewRPCFunc(c.ABCIInfo, ""),
"abci_query": rpcserver.NewRPCFunc(makeABCIQueryFunc(c), "path,data"),
"abci_info": rpcserver.NewRPCFunc(makeABCIInfoFunc(c), ""),
}
}
func makeStatusFunc(c rpcclient.Client) func(ctx *rpctypes.Context) (*ctypes.ResultStatus, error) {
return func(ctx *rpctypes.Context) (*ctypes.ResultStatus, error) {
return c.Status()
}
}
func makeBlockchainInfoFunc(c rpcclient.Client) func(ctx *rpctypes.Context, minHeight, maxHeight int64) (*ctypes.ResultBlockchainInfo, error) {
return func(ctx *rpctypes.Context, minHeight, maxHeight int64) (*ctypes.ResultBlockchainInfo, error) {
return c.BlockchainInfo(minHeight, maxHeight)
}
}
func makeGenesisFunc(c rpcclient.Client) func(ctx *rpctypes.Context) (*ctypes.ResultGenesis, error) {
return func(ctx *rpctypes.Context) (*ctypes.ResultGenesis, error) {
return c.Genesis()
}
}
func makeBlockFunc(c rpcclient.Client) func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultBlock, error) {
return func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultBlock, error) {
return c.Block(height)
}
}
func makeCommitFunc(c rpcclient.Client) func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultCommit, error) {
return func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultCommit, error) {
return c.Commit(height)
}
}
func makeTxFunc(c rpcclient.Client) func(ctx *rpctypes.Context, hash []byte, prove bool) (*ctypes.ResultTx, error) {
return func(ctx *rpctypes.Context, hash []byte, prove bool) (*ctypes.ResultTx, error) {
return c.Tx(hash, prove)
}
}
func makeValidatorsFunc(c rpcclient.Client) func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultValidators, error) {
return func(ctx *rpctypes.Context, height *int64) (*ctypes.ResultValidators, error) {
return c.Validators(height)
}
}
func makeBroadcastTxCommitFunc(c rpcclient.Client) func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
return func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTxCommit, error) {
return c.BroadcastTxCommit(tx)
}
}
func makeBroadcastTxSyncFunc(c rpcclient.Client) func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
return func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
return c.BroadcastTxSync(tx)
}
}
func makeBroadcastTxAsyncFunc(c rpcclient.Client) func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
return func(ctx *rpctypes.Context, tx types.Tx) (*ctypes.ResultBroadcastTx, error) {
return c.BroadcastTxAsync(tx)
}
}
func makeABCIQueryFunc(c rpcclient.Client) func(ctx *rpctypes.Context, path string, data cmn.HexBytes) (*ctypes.ResultABCIQuery, error) {
return func(ctx *rpctypes.Context, path string, data cmn.HexBytes) (*ctypes.ResultABCIQuery, error) {
return c.ABCIQuery(path, data)
}
}
func makeABCIInfoFunc(c rpcclient.Client) func(ctx *rpctypes.Context) (*ctypes.ResultABCIInfo, error) {
return func(ctx *rpctypes.Context) (*ctypes.ResultABCIInfo, error) {
return c.ABCIInfo()
}
}
lite/proxy/query.go
+28 -6
View File
@@ -2,6 +2,7 @@ package proxy
import (
"fmt"
"strings"
cmn "github.com/tendermint/tendermint/libs/common"
@@ -43,11 +44,7 @@ func GetWithProof(prt *merkle.ProofRuntime, key []byte, reqHeight int64, node rp
func GetWithProofOptions(prt *merkle.ProofRuntime, path string, key []byte, opts rpcclient.ABCIQueryOptions,
node rpcclient.Client, cert lite.Verifier) (
*ctypes.ResultABCIQuery, error) {
if !opts.Prove {
return nil, cmn.NewError("require ABCIQueryOptions.Prove to be true")
}
opts.Prove = true
res, err := node.ABCIQueryWithOptions(path, key, opts)
if err != nil {
return nil, err
@@ -77,7 +74,14 @@ func GetWithProofOptions(prt *merkle.ProofRuntime, path string, key []byte, opts
if resp.Value != nil {
// Value exists
// XXX How do we encode the key into a string...
err = prt.VerifyValue(resp.Proof, signedHeader.AppHash, string(resp.Key), resp.Value)
storeName, err := parseQueryStorePath(path)
if err != nil {
return nil, err
}
kp := merkle.KeyPath{}
kp = kp.AppendKey([]byte(storeName), merkle.KeyEncodingURL)
kp = kp.AppendKey(resp.Key, merkle.KeyEncodingURL)
err = prt.VerifyValue(resp.Proof, signedHeader.AppHash, kp.String(), resp.Value)
if err != nil {
return nil, cmn.ErrorWrap(err, "Couldn't verify value proof")
}
@@ -94,6 +98,24 @@ func GetWithProofOptions(prt *merkle.ProofRuntime, path string, key []byte, opts
}
}
func parseQueryStorePath(path string) (storeName string, err error) {
if !strings.HasPrefix(path, "/") {
return "", fmt.Errorf("expected path to start with /")
}
paths := strings.SplitN(path[1:], "/", 3)
switch {
case len(paths) != 3:
return "", fmt.Errorf("expected format like /store/<storeName>/key")
case paths[0] != "store":
return "", fmt.Errorf("expected format like /store/<storeName>/key")
case paths[2] != "key":
return "", fmt.Errorf("expected format like /store/<storeName>/key")
}
return paths[1], nil
}
// GetCertifiedCommit gets the signed header for a given height and certifies
// it. Returns error if unable to get a proven header.
func GetCertifiedCommit(h int64, client rpcclient.Client, cert lite.Verifier) (types.SignedHeader, error) {
lite/proxy/query_test.go
+5 -2
View File
@@ -21,6 +21,7 @@ import (
var node *nm.Node
var chainID = "tendermint_test" // TODO use from config.
//nolint:unused
var waitForEventTimeout = 5 * time.Second
// TODO fix tests!!
@@ -31,8 +32,7 @@ func TestMain(m *testing.M) {
code := m.Run()
node.Stop()
node.Wait()
rpctest.StopTendermint(node)
os.Exit(code)
}
@@ -42,6 +42,7 @@ func kvstoreTx(k, v []byte) []byte {
// TODO: enable it after general proof format has been adapted
// in abci/examples/kvstore.go
//nolint:unused,deadcode
func _TestAppProofs(t *testing.T) {
assert, require := assert.New(t), require.New(t)
@@ -92,6 +93,8 @@ func _TestAppProofs(t *testing.T) {
// verify a query before the tx block has no data (and valid non-exist proof)
bs, height, proof, err := GetWithProof(prt, k, brh-1, cl, cert)
require.NoError(err, "%#v", err)
require.NotNil(proof)
require.Equal(height, brh-1)
// require.NotNil(proof)
// TODO: Ensure that *some* keys will be there, ensuring that proof is nil,
// (currently there's a race condition)
lite/proxy/validate_test.go
+4 -4
View File
@@ -70,7 +70,7 @@ func TestValidateBlock(t *testing.T) {
},
signedHeader: types.SignedHeader{
Header: &types.Header{Height: 11},
Commit: &types.Commit{BlockID: types.BlockID{Hash: []byte("0xDEADBEEF")}},
Commit: types.NewCommit(types.BlockID{Hash: []byte("0xDEADBEEF")}, nil),
},
wantErr: "Data hash doesn't match header",
},
@@ -81,7 +81,7 @@ func TestValidateBlock(t *testing.T) {
},
signedHeader: types.SignedHeader{
Header: &types.Header{Height: 11},
Commit: &types.Commit{BlockID: types.BlockID{Hash: []byte("DEADBEEF")}},
Commit: types.NewCommit(types.BlockID{Hash: []byte("DEADBEEF")}, nil),
},
},
// End Header.Data hash mismatch test
@@ -169,7 +169,7 @@ func TestValidateBlockMeta(t *testing.T) {
ValidatorsHash: []byte("Tendermint"),
Time: testTime2,
},
Commit: &types.Commit{BlockID: types.BlockID{Hash: []byte("DEADBEEF")}},
Commit: types.NewCommit(types.BlockID{Hash: []byte("DEADBEEF")}, nil),
},
wantErr: "Headers don't match",
},
@@ -188,7 +188,7 @@ func TestValidateBlockMeta(t *testing.T) {
ValidatorsHash: []byte("Tendermint-x"),
Time: testTime2,
},
Commit: &types.Commit{BlockID: types.BlockID{Hash: []byte("DEADBEEF")}},
Commit: types.NewCommit(types.BlockID{Hash: []byte("DEADBEEF")}, nil),
},
wantErr: "Headers don't match",
},
lite/proxy/wrapper.go
+57
View File
@@ -1,12 +1,16 @@
package proxy
import (
"context"
"fmt"
cmn "github.com/tendermint/tendermint/libs/common"
"github.com/tendermint/tendermint/crypto/merkle"
"github.com/tendermint/tendermint/lite/verifying"
rpcclient "github.com/tendermint/tendermint/rpc/client"
ctypes "github.com/tendermint/tendermint/rpc/core/types"
rpctypes "github.com/tendermint/tendermint/rpc/lib/types"
)
var _ rpcclient.Client = Wrapper{}
@@ -145,6 +149,59 @@ func (w Wrapper) Commit(height *int64) (*ctypes.ResultCommit, error) {
return res, err
}
func (w Wrapper) RegisterOpDecoder(typ string, dec merkle.OpDecoder) {
w.prt.RegisterOpDecoder(typ, dec)
}
// SubscribeWS subscribes for events using the given query and remote address as
// a subscriber, but does not verify responses (UNSAFE)!
func (w Wrapper) SubscribeWS(ctx *rpctypes.Context, query string) (*ctypes.ResultSubscribe, error) {
out, err := w.Client.Subscribe(context.Background(), ctx.RemoteAddr(), query)
if err != nil {
return nil, err
}
go func() {
for {
select {
case resultEvent := <-out:
// XXX(melekes) We should have a switch here that performs a validation
// depending on the event's type.
ctx.WSConn.TryWriteRPCResponse(
rpctypes.NewRPCSuccessResponse(
ctx.WSConn.Codec(),
rpctypes.JSONRPCStringID(fmt.Sprintf("%v#event", ctx.JSONReq.ID)),
resultEvent,
))
case <-w.Client.Quit():
return
}
}
}()
return &ctypes.ResultSubscribe{}, nil
}
// UnsubscribeWS calls original client's Unsubscribe using remote address as a
// subscriber.
func (w Wrapper) UnsubscribeWS(ctx *rpctypes.Context, query string) (*ctypes.ResultUnsubscribe, error) {
err := w.Client.Unsubscribe(context.Background(), ctx.RemoteAddr(), query)
if err != nil {
return nil, err
}
return &ctypes.ResultUnsubscribe{}, nil
}
// UnsubscribeAllWS calls original client's UnsubscribeAll using remote address
// as a subscriber.
func (w Wrapper) UnsubscribeAllWS(ctx *rpctypes.Context) (*ctypes.ResultUnsubscribe, error) {
err := w.Client.UnsubscribeAll(context.Background(), ctx.RemoteAddr())
if err != nil {
return nil, err
}
return &ctypes.ResultUnsubscribe{}, nil
}
// // WrappedSwitch creates a websocket connection that auto-verifies any info
// // coming through before passing it along.
// //