Add darksidewalletd mode for reorg testing.

cmd/server/main.go

@@ -0,0 +1,266 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"net"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/sirupsen/logrus"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/peer"
+	"google.golang.org/grpc/reflection"
+
+	"github.com/zcash/lightwalletd/common"
+	"github.com/zcash/lightwalletd/frontend"
+	"github.com/zcash/lightwalletd/walletrpc"
+)
+
+var logger = logrus.New()
+
+func init() {
+	logger.SetFormatter(&logrus.TextFormatter{
+		//DisableColors:          true,
+		FullTimestamp:          true,
+		DisableLevelTruncation: true,
+	})
+
+	onexit := func() {
+		fmt.Println("Lightwalletd died with a Fatal error. Check logfile for details.")
+	}
+
+	common.Log = logger.WithFields(logrus.Fields{
+		"app": "frontend-grpc",
+	})
+
+	logrus.RegisterExitHandler(onexit)
+}
+
+// TODO stream logging
+
+func LoggingInterceptor() grpc.ServerOption {
+	return grpc.UnaryInterceptor(logInterceptor)
+}
+
+func logInterceptor(ctx context.Context, req interface{}, info *grpc.UnaryServerInfo, handler grpc.UnaryHandler) (interface{}, error) {
+	reqLog := loggerFromContext(ctx)
+	start := time.Now()
+
+	resp, err := handler(ctx, req)
+
+	entry := reqLog.WithFields(logrus.Fields{
+		"method":   info.FullMethod,
+		"duration": time.Since(start),
+		"error":    err,
+	})
+
+	if err != nil {
+		entry.Error("call failed")
+	} else {
+		entry.Info("method called")
+	}
+
+	return resp, err
+}
+
+func loggerFromContext(ctx context.Context) *logrus.Entry {
+	// TODO: anonymize the addresses. cryptopan?
+	if peerInfo, ok := peer.FromContext(ctx); ok {
+		return common.Log.WithFields(logrus.Fields{"peer_addr": peerInfo.Addr})
+	}
+	return common.Log.WithFields(logrus.Fields{"peer_addr": "unknown"})
+}
+
+type Options struct {
+	bindAddr      string `json:"bind_address,omitempty"`
+	tlsCertPath   string `json:"tls_cert_path,omitempty"`
+	tlsKeyPath    string `json:"tls_cert_key,omitempty"`
+	logLevel      uint64 `json:"log_level,omitempty"`
+	logPath       string `json:"log_file,omitempty"`
+	zcashConfPath string `json:"zcash_conf,omitempty"`
+	veryInsecure  bool   `json:"very_insecure,omitempty"`
+	darkSide      bool   `json:"very_insecure,omitempty"`
+	cacheSize     int    `json:"cache_size,omitempty"`
+	wantVersion   bool
+}
+
+func fileExists(filename string) bool {
+	info, err := os.Stat(filename)
+	if os.IsNotExist(err) {
+		return false
+	}
+	return !info.IsDir()
+}
+
+func main() {
+	opts := &Options{}
+	flag.StringVar(&opts.bindAddr, "bind-addr", "127.0.0.1:9067", "the address to listen on")
+	flag.StringVar(&opts.tlsCertPath, "tls-cert", "", "the path to a TLS certificate")
+	flag.StringVar(&opts.tlsKeyPath, "tls-key", "", "the path to a TLS key file")
+	flag.Uint64Var(&opts.logLevel, "log-level", uint64(logrus.InfoLevel), "log level (logrus 1-7)")
+	flag.StringVar(&opts.logPath, "log-file", "./server.log", "log file to write to")
+	flag.StringVar(&opts.zcashConfPath, "conf-file", "./zcash.conf", "conf file to pull RPC creds from")
+	flag.BoolVar(&opts.veryInsecure, "no-tls-very-insecure", false, "run without the required TLS certificate, only for debugging, DO NOT use in production")
+	flag.BoolVar(&opts.darkSide, "darkside-very-insecure", false, "run with GRPC-controllable mock zcashd for integration testing (shuts down after 30 minutes)")
+	flag.BoolVar(&opts.wantVersion, "version", false, "version (major.minor.patch)")
+	flag.IntVar(&opts.cacheSize, "cache-size", 80000, "number of blocks to hold in the cache")
+
+	// TODO prod metrics
+	// TODO support config from file and env vars
+	flag.Parse()
+
+	if opts.wantVersion {
+		fmt.Println("lightwalletd version v0.3.0")
+		return
+	}
+
+	// production (unlike unit tests) use the real sleep function
+	common.Sleep = time.Sleep
+
+	if opts.logPath != "" {
+		// instead write parsable logs for logstash/splunk/etc
+		output, err := os.OpenFile(opts.logPath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
+		if err != nil {
+			os.Stderr.WriteString(fmt.Sprintf("Cannot open log file %s: %v\n",
+				opts.logPath, err))
+			os.Exit(1)
+		}
+		defer output.Close()
+		logger.SetOutput(output)
+		logger.SetFormatter(&logrus.JSONFormatter{})
+	}
+	logger.SetLevel(logrus.Level(opts.logLevel))
+
+	filesThatShouldExist := []string{ }
+
+    if !opts.darkSide {
+		filesThatShouldExist = append(filesThatShouldExist, opts.zcashConfPath)
+    }
+	if opts.tlsCertPath != "" {
+		filesThatShouldExist = append(filesThatShouldExist, opts.tlsCertPath)
+	}
+	if opts.tlsKeyPath != "" {
+		filesThatShouldExist = append(filesThatShouldExist, opts.tlsKeyPath)
+	}
+
+	for _, filename := range filesThatShouldExist {
+		if !fileExists(filename) {
+			common.Log.WithFields(logrus.Fields{
+				"filename": filename,
+			}).Error("cannot open required file")
+			os.Stderr.WriteString("Cannot open required file: " + filename + "\n")
+			os.Exit(1)
+		}
+	}
+
+	// gRPC initialization
+	var server *grpc.Server
+	var err error
+
+	if opts.veryInsecure {
+		server = grpc.NewServer(LoggingInterceptor())
+	} else {
+		var transportCreds credentials.TransportCredentials
+		if (opts.tlsCertPath == "") && (opts.tlsKeyPath == "") {
+			common.Log.Warning("Certificate and key not provided, generating self signed values")
+			tlsCert := common.GenerateCerts()
+			transportCreds = credentials.NewServerTLSFromCert(tlsCert)
+		} else {
+			transportCreds, err = credentials.NewServerTLSFromFile(opts.tlsCertPath, opts.tlsKeyPath)
+			if err != nil {
+				common.Log.WithFields(logrus.Fields{
+					"cert_file": opts.tlsCertPath,
+					"key_path":  opts.tlsKeyPath,
+					"error":     err,
+				}).Fatal("couldn't load TLS credentials")
+			}
+		}
+		server = grpc.NewServer(grpc.Creds(transportCreds), LoggingInterceptor())
+	}
+
+	// Enable reflection for debugging
+	if opts.logLevel >= uint64(logrus.WarnLevel) {
+		reflection.Register(server)
+	}
+
+    if opts.darkSide {
+        common.RawRequest = common.DarkSideRawRequest
+    } else {
+        // Initialize Zcash RPC client. Right now (Jan 2018) this is only for
+        // sending transactions, but in the future it could back a different type
+        // of block streamer.
+
+        rpcClient, err := frontend.NewZRPCFromConf(opts.zcashConfPath)
+        if err != nil {
+            common.Log.WithFields(logrus.Fields{
+                "error": err,
+            }).Fatal("setting up RPC connection to zcashd")
+        }
+
+        // indirect function for test mocking (so unit tests can talk to stub functions)
+	    common.RawRequest = rpcClient.RawRequest
+    }
+
+	// Get the sapling activation height from the RPC
+	// (this first RPC also verifies that we can communicate with zcashd)
+	saplingHeight, blockHeight, chainName, branchID := common.GetSaplingInfo()
+	common.Log.Info("Got sapling height ", saplingHeight, " chain ", chainName, " branchID ", branchID)
+
+	// Initialize the cache
+	cache := common.NewBlockCache(opts.cacheSize)
+
+	// Start the block cache importer at cacheSize blocks before current height
+	cacheStart := blockHeight - opts.cacheSize
+	if cacheStart < saplingHeight {
+		cacheStart = saplingHeight
+	}
+
+	// The last argument, repetition count, is only nonzero for testing
+	go common.BlockIngestor(cache, cacheStart, 0)
+
+	// Compact transaction service initialization
+	service, err := frontend.NewLwdStreamer(cache)
+	if err != nil {
+		common.Log.WithFields(logrus.Fields{
+			"error": err,
+		}).Fatal("couldn't create backend")
+	}
+
+	// Register service
+	walletrpc.RegisterCompactTxStreamerServer(server, service)
+
+	// Start listening
+	listener, err := net.Listen("tcp", opts.bindAddr)
+	if err != nil {
+		common.Log.WithFields(logrus.Fields{
+			"bind_addr": opts.bindAddr,
+			"error":     err,
+		}).Fatal("couldn't create listener")
+	}
+
+	// Signal handler for graceful stops
+	signals := make(chan os.Signal, 1)
+	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)
+	go func() {
+		s := <-signals
+		common.Log.WithFields(logrus.Fields{
+			"signal": s.String(),
+		}).Info("caught signal, stopping gRPC server")
+		os.Stderr.WriteString("Caught signal: " + s.String() + "\n")
+		os.Exit(1)
+	}()
+
+	common.Log.Infof("Starting gRPC server on %s", opts.bindAddr)
+
+	err = server.Serve(listener)
+	if err != nil {
+		common.Log.WithFields(logrus.Fields{
+			"error": err,
+		}).Fatal("gRPC server exited")
+	}
+}

common/darkside.go

@@ -0,0 +1,157 @@
+package common
+
+import (
+    "errors"
+    "encoding/json"
+    "os"
+    "bufio"
+    "strconv"
+    "time"
+    "encoding/hex"
+)
+
+type DarksideZcashdState struct {
+    start_height int
+    sapling_activation int
+    branch_id string
+    chain_name string
+    // Should always be nonempty. Index 0 is the block at height start_height.
+    blocks []string
+    incoming_transactions [][]byte
+    server_start time.Time
+}
+
+var state *DarksideZcashdState = nil
+
+func DarkSideRawRequest(method string, params []json.RawMessage) (json.RawMessage, error) {
+
+    if state == nil {
+        state = &DarksideZcashdState{
+            start_height: 1000,
+            sapling_activation: 1000,
+            branch_id: "2bb40e60",
+            chain_name: "main",
+            blocks: make([]string, 0),
+            incoming_transactions: make([][]byte, 0),
+            server_start: time.Now(),
+        }
+
+        testBlocks, err := os.Open("./testdata/blocks")
+        if err != nil {
+            Log.Fatal("Error loading testdata blocks")
+        }
+        scan := bufio.NewScanner(testBlocks)
+        for scan.Scan() { // each line (block)
+            block := scan.Bytes()
+            state.blocks = append(state.blocks, string(block))
+        }
+    }
+
+    if time.Now().Sub(state.server_start).Minutes() >= 30 {
+        Log.Fatal("Shutting down darksidewalletd to prevent accidental deployment in production.")
+    }
+
+    switch method {
+    case "getblockchaininfo":
+        // TODO: there has got to be a better way to construct this!
+        data := make(map[string]interface{})
+        data["chain"] = state.chain_name
+        data["upgrades"] = make(map[string]interface{})
+        data["upgrades"].(map[string]interface{})["76b809bb"] = make(map[string]interface{})
+        data["upgrades"].(map[string]interface{})["76b809bb"].(map[string]interface{})["activationheight"] = state.sapling_activation
+        data["headers"] = state.start_height + len(state.blocks) - 1
+        data["consensus"] = make(map[string]interface{})
+        data["consensus"].(map[string]interface{})["nextblock"] = state.branch_id
+
+        return json.Marshal(data)
+
+    case "getblock":
+        var height string
+        err := json.Unmarshal(params[0], &height)
+        if err != nil {
+		    return nil, errors.New("Failed to parse getblock request.")
+        }
+
+        height_i, err := strconv.Atoi(height)
+        if err != nil {
+            return nil, errors.New("Error parsing height as integer.")
+        }
+        index := height_i - state.start_height
+
+        if index == len(state.blocks) {
+            // The current ingestor keeps going until it sees this error, 
+            // meaning it's up to the latest height.
+            return nil, errors.New("-8:")
+        }
+
+        if index < 0 || index > len(state.blocks) {
+            // If an integration test can reach this, it could be a bug, so generate an error.
+            Log.Errorf("getblock request made for out-of-range height %d (have %d to %d)", height_i, state.start_height, state.start_height + len(state.blocks) - 1)
+            return nil, errors.New("-8:")
+        }
+
+        return []byte("\"" + state.blocks[index] + "\""), nil
+
+    case "getaddresstxids":
+        // Not required for minimal reorg testing.
+		return nil, errors.New("Not implemented yet.")
+
+    case "getrawtransaction":
+        // Not required for minimal reorg testing.
+		return nil, errors.New("Not implemented yet.")
+
+    case "sendrawtransaction":
+        var rawtx string
+        err := json.Unmarshal(params[0], &rawtx)
+        if err != nil {
+		    return nil, errors.New("Failed to parse sendrawtransaction JSON.")
+        }
+        txbytes, err := hex.DecodeString(rawtx)
+        if err != nil {
+            return nil, errors.New("Failed to parse sendrawtransaction value as a hex string.")
+        }
+        state.incoming_transactions = append(state.incoming_transactions, txbytes)
+        return nil, nil
+
+    case "x_setstate":
+        var new_state map[string]interface{}
+
+        err := json.Unmarshal(params[0], &new_state)
+        if err != nil {
+            Log.Fatal("Could not unmarshal the provided state.")
+        }
+
+        block_strings := make([]string, 0)
+        for _, block_str := range new_state["blocks"].([]interface{}) {
+            block_strings = append(block_strings, block_str.(string))
+        }
+
+        state = &DarksideZcashdState{
+            start_height: int(new_state["start_height"].(float64)),
+            sapling_activation: int(new_state["sapling_activation"].(float64)),
+            branch_id: new_state["branch_id"].(string),
+            chain_name: new_state["chain_name"].(string),
+            blocks: block_strings,
+            incoming_transactions: state.incoming_transactions,
+            server_start: state.server_start,
+        }
+
+        return nil, nil
+
+    case "x_getincomingtransactions":
+        txlist := "["
+        for i, tx := range state.incoming_transactions {
+            txlist += "\"" + hex.EncodeToString(tx) + "\""
+            // add commas after all but the last
+            if i < len(state.incoming_transactions) - 1 {
+                txlist += ", "
+            }
+        }
+        txlist += "]"
+
+        return []byte(txlist), nil
+
+    default:
+        return nil, errors.New("There was an attempt to call an unsupported RPC.")
+    }
+}

frontend/service.go

@@ -255,3 +255,67 @@ func (s *LwdStreamer) Ping(ctx context.Context, in *walletrpc.Duration) (*wallet
 	response.Exit = atomic.AddInt64(&concurrent, -1)
 	return &response, nil
 }
+
+// Evil
+func (s *LwdStreamer) EvilGetIncomingTransactions(in *walletrpc.Empty, resp walletrpc.CompactTxStreamer_EvilGetIncomingTransactionsServer) error {
+	// Get all of the new incoming transactions evil zcashd has accepted.
+	result, rpcErr := common.RawRequest("x_getincomingtransactions", nil)
+
+	var new_txs []string
+	if rpcErr != nil {
+		return rpcErr
+	}
+	err := json.Unmarshal(result, &new_txs)
+
+	if err != nil {
+		return err
+	}
+
+	for _, tx_str := range new_txs {
+		tx_bytes, err := hex.DecodeString(tx_str)
+		if err != nil {
+			return err
+		}
+		err = resp.Send(&walletrpc.RawTransaction{Data: tx_bytes, Height: 0})
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *LwdStreamer) EvilSetState(ctx context.Context, state *walletrpc.EvilLightwalletdState) (*walletrpc.Empty, error) {
+	match, err := regexp.Match("\\A[a-zA-Z0-9]+\\z", []byte(state.BranchID))
+	if err != nil || !match {
+		return nil, errors.New("Invalid branch ID")
+	}
+
+	match, err = regexp.Match("\\A[a-zA-Z0-9]+\\z", []byte(state.ChainName))
+	if err != nil || !match {
+		return nil, errors.New("Invalid chain name")
+	}
+
+	st := "{" +
+		"\"start_height\": " + strconv.Itoa(int(state.StartHeight)) +
+		", \"sapling_activation\": " + strconv.Itoa(int(state.SaplingActivation)) +
+		", \"branch_id\": \"" + state.BranchID + "\"" +
+		", \"chain_name\": \"" + state.ChainName + "\"" +
+		", \"blocks\": ["
+
+	for i, block := range state.Blocks {
+		st += "\"" + block + "\""
+		if i < len(state.Blocks) - 1 {
+			st += ", "
+		}
+	}
+
+	st += "]}"
+
+	params := make([]json.RawMessage, 1)
+	params[0] = json.RawMessage(st)
+
+	_, rpcErr := common.RawRequest("x_setstate", params)
+
+	return &walletrpc.Empty{}, rpcErr
+}

go.sum

@@ -245,6 +245,8 @@ google.golang.org/grpc v1.27.0/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8
 google.golang.org/grpc v1.28.1 h1:C1QC6KzgSiLyBabDi87BbjaGreoRgGUF5nOyvfrAZ1k=
 google.golang.org/grpc v1.28.1/go.mod h1:rpkK4SK4GF4Ach/+MFLZUBavHOvF2JJB5uozKKal+60=
 gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw=
+google.golang.org/grpc v1.27.1 h1:zvIju4sqAGvwKspUQOhwnpcqSbzi7/H6QomNNjTL4sk=
+google.golang.org/grpc v1.27.1/go.mod h1:qbnxyOmOxrQa7FizSgH+ReBfzJrCY1pSN7KXBS8abTk=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=

walletrpc/service.pb.go

@@ -486,6 +486,12 @@ func (m *TransparentAddressBlockFilter) GetRange() *BlockRange {
 // are microseconds.
 type Duration struct {
 	IntervalUs           int64    `protobuf:"varint,1,opt,name=intervalUs,proto3" json:"intervalUs,omitempty"`
+type EvilLightwalletdState struct {
+	StartHeight          int32    `protobuf:"varint,1,opt,name=startHeight,proto3" json:"startHeight,omitempty"`
+	SaplingActivation    int32    `protobuf:"varint,2,opt,name=saplingActivation,proto3" json:"saplingActivation,omitempty"`
+	BranchID             string   `protobuf:"bytes,3,opt,name=branchID,proto3" json:"branchID,omitempty"`
+	ChainName            string   `protobuf:"bytes,4,opt,name=chainName,proto3" json:"chainName,omitempty"`
+	Blocks               []string `protobuf:"bytes,5,rep,name=blocks,proto3" json:"blocks,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -496,29 +502,47 @@ func (m *Duration) String() string { return proto.CompactTextString(m) }
 func (*Duration) ProtoMessage()    {}
 func (*Duration) Descriptor() ([]byte, []int) {
 	return fileDescriptor_a0b84a42fa06f626, []int{9}
+func (m *EvilLightwalletdState) Reset()         { *m = EvilLightwalletdState{} }
+func (m *EvilLightwalletdState) String() string { return proto.CompactTextString(m) }
+func (*EvilLightwalletdState) ProtoMessage()    {}
+func (*EvilLightwalletdState) Descriptor() ([]byte, []int) {
+	return fileDescriptor_a0b84a42fa06f626, []int{10}
 }
 
 func (m *Duration) XXX_Unmarshal(b []byte) error {
 	return xxx_messageInfo_Duration.Unmarshal(m, b)
+func (m *EvilLightwalletdState) XXX_Unmarshal(b []byte) error {
+	return xxx_messageInfo_EvilLightwalletdState.Unmarshal(m, b)
 }
 func (m *Duration) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
 	return xxx_messageInfo_Duration.Marshal(b, m, deterministic)
+func (m *EvilLightwalletdState) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
+	return xxx_messageInfo_EvilLightwalletdState.Marshal(b, m, deterministic)
 }
 func (m *Duration) XXX_Merge(src proto.Message) {
 	xxx_messageInfo_Duration.Merge(m, src)
+func (m *EvilLightwalletdState) XXX_Merge(src proto.Message) {
+	xxx_messageInfo_EvilLightwalletdState.Merge(m, src)
 }
 func (m *Duration) XXX_Size() int {
 	return xxx_messageInfo_Duration.Size(m)
+func (m *EvilLightwalletdState) XXX_Size() int {
+	return xxx_messageInfo_EvilLightwalletdState.Size(m)
 }
 func (m *Duration) XXX_DiscardUnknown() {
 	xxx_messageInfo_Duration.DiscardUnknown(m)
+func (m *EvilLightwalletdState) XXX_DiscardUnknown() {
+	xxx_messageInfo_EvilLightwalletdState.DiscardUnknown(m)
 }
 
 var xxx_messageInfo_Duration proto.InternalMessageInfo
+var xxx_messageInfo_EvilLightwalletdState proto.InternalMessageInfo
 
 func (m *Duration) GetIntervalUs() int64 {
+func (m *EvilLightwalletdState) GetStartHeight() int32 {
 	if m != nil {
 		return m.IntervalUs
+		return m.StartHeight
 	}
 	return 0
 }
@@ -532,6 +556,11 @@ type PingResponse struct {
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
+func (m *EvilLightwalletdState) GetSaplingActivation() int32 {
+	if m != nil {
+		return m.SaplingActivation
+	}
+	return 0
 }
 
 func (m *PingResponse) Reset()         { *m = PingResponse{} }
@@ -539,6 +568,11 @@ func (m *PingResponse) String() string { return proto.CompactTextString(m) }
 func (*PingResponse) ProtoMessage()    {}
 func (*PingResponse) Descriptor() ([]byte, []int) {
 	return fileDescriptor_a0b84a42fa06f626, []int{10}
+func (m *EvilLightwalletdState) GetBranchID() string {
+	if m != nil {
+		return m.BranchID
+	}
+	return ""
 }
 
 func (m *PingResponse) XXX_Unmarshal(b []byte) error {
@@ -560,17 +594,23 @@ func (m *PingResponse) XXX_DiscardUnknown() {
 var xxx_messageInfo_PingResponse proto.InternalMessageInfo
 
 func (m *PingResponse) GetEntry() int64 {
+func (m *EvilLightwalletdState) GetChainName() string {
 	if m != nil {
 		return m.Entry
+		return m.ChainName
 	}
 	return 0
+	return ""
 }
 
 func (m *PingResponse) GetExit() int64 {
+func (m *EvilLightwalletdState) GetBlocks() []string {
 	if m != nil {
 		return m.Exit
+		return m.Blocks
 	}
 	return 0
+	return nil
 }
 
 func init() {
@@ -585,6 +625,7 @@ func init() {
 	proto.RegisterType((*TransparentAddressBlockFilter)(nil), "cash.z.wallet.sdk.rpc.TransparentAddressBlockFilter")
 	proto.RegisterType((*Duration)(nil), "cash.z.wallet.sdk.rpc.Duration")
 	proto.RegisterType((*PingResponse)(nil), "cash.z.wallet.sdk.rpc.PingResponse")
+	proto.RegisterType((*EvilLightwalletdState)(nil), "cash.z.wallet.sdk.rpc.EvilLightwalletdState")
 }
 
 func init() {
@@ -638,6 +679,55 @@ var fileDescriptor_a0b84a42fa06f626 = []byte{
 	0x72, 0x79, 0xc9, 0x0e, 0x06, 0x2d, 0xec, 0x2d, 0x9e, 0xd5, 0x5d, 0x80, 0x4c, 0xfc, 0xdf, 0xa5,
 	0xc2, 0xc5, 0x9c, 0xfd, 0x5b, 0xbd, 0xfd, 0x1b, 0x00, 0x00, 0xff, 0xff, 0x3f, 0xd2, 0xc0, 0xf7,
 	0xec, 0x06, 0x00, 0x00,
+	// 756 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x94, 0x55, 0xdd, 0x4e, 0xdb, 0x4a,
+	0x10, 0x4e, 0x42, 0x9c, 0x9f, 0x21, 0x80, 0x58, 0x1d, 0x38, 0x51, 0x0e, 0xa7, 0x4d, 0xb7, 0xaa,
+	0xc4, 0x05, 0xb2, 0x10, 0xa5, 0x6a, 0x2f, 0x7a, 0xc3, 0x5f, 0x43, 0x24, 0x5a, 0xb5, 0x9b, 0x5c,
+	0xd1, 0x4a, 0x68, 0xb1, 0x97, 0xc4, 0x25, 0x59, 0x5b, 0xbb, 0x4b, 0x48, 0xfb, 0x08, 0x7d, 0xa3,
+	0xf6, 0x79, 0xfa, 0x20, 0xd5, 0x8e, 0x1d, 0x62, 0x14, 0x4c, 0xd2, 0x3b, 0xcf, 0xee, 0xcc, 0xf7,
+	0xcd, 0x7e, 0xf3, 0x63, 0x58, 0xd1, 0x42, 0x8d, 0x02, 0x4f, 0xb8, 0x91, 0x0a, 0x4d, 0x48, 0x36,
+	0x3c, 0xae, 0xfb, 0xee, 0x77, 0xf7, 0x96, 0x0f, 0x06, 0xc2, 0xb8, 0xda, 0xbf, 0x76, 0x55, 0xe4,
+	0x35, 0x36, 0xbc, 0x70, 0x18, 0x71, 0xcf, 0x5c, 0x5c, 0x85, 0x6a, 0xc8, 0x8d, 0x8e, 0xbd, 0xe9,
+	0x2b, 0x28, 0x1f, 0x0e, 0x42, 0xef, 0xba, 0x7d, 0x4c, 0x36, 0xa1, 0xd4, 0x17, 0x41, 0xaf, 0x6f,
+	0xea, 0xf9, 0x66, 0x7e, 0xbb, 0xc8, 0x12, 0x8b, 0x10, 0x28, 0xf6, 0xb9, 0xee, 0xd7, 0x0b, 0xcd,
+	0xfc, 0x76, 0x8d, 0xe1, 0x37, 0x35, 0x00, 0x18, 0xc6, 0xb8, 0xec, 0x09, 0xb2, 0x0f, 0x8e, 0x36,
+	0x5c, 0xc5, 0x81, 0xcb, 0x7b, 0x4f, 0xdc, 0x07, 0x53, 0x70, 0x13, 0x22, 0x16, 0x3b, 0x93, 0x5d,
+	0x58, 0x12, 0xd2, 0x47, 0xd8, 0xf9, 0x31, 0xd6, 0x95, 0x7e, 0x85, 0x4a, 0x77, 0xfc, 0x2e, 0x18,
+	0x18, 0xa1, 0x2c, 0xe7, 0xa5, 0xbd, 0x5b, 0x94, 0x13, 0x9d, 0xc9, 0x3f, 0xe0, 0x04, 0xd2, 0x17,
+	0x63, 0x64, 0x2d, 0xb2, 0xd8, 0xb8, 0x7b, 0xe1, 0x52, 0xea, 0x85, 0x6f, 0x61, 0x95, 0xf1, 0xdb,
+	0xae, 0xe2, 0x52, 0x73, 0xcf, 0x04, 0xa1, 0xb4, 0x5e, 0x3e, 0x37, 0x1c, 0x09, 0x6b, 0x0c, 0xbf,
+	0x53, 0x9a, 0x15, 0xd2, 0x9a, 0xd1, 0x8f, 0x50, 0xeb, 0x08, 0xe9, 0x33, 0xa1, 0xa3, 0x50, 0x6a,
+	0x41, 0xb6, 0xa0, 0x2a, 0x94, 0x0a, 0xd5, 0x51, 0xe8, 0x0b, 0x04, 0x70, 0xd8, 0xf4, 0x80, 0x50,
+	0xa8, 0xa1, 0xf1, 0x5e, 0x68, 0xcd, 0x7b, 0x02, 0xb1, 0xaa, 0xec, 0xde, 0x19, 0x5d, 0x86, 0xea,
+	0x51, 0x9f, 0x07, 0xb2, 0x13, 0x09, 0x8f, 0x96, 0xc1, 0x39, 0x19, 0x46, 0xe6, 0x1b, 0xfd, 0x51,
+	0x00, 0x38, 0xb3, 0x8c, 0x7e, 0x5b, 0x5e, 0x85, 0xa4, 0x0e, 0xe5, 0x91, 0x50, 0x3a, 0x08, 0x25,
+	0x92, 0x54, 0xd9, 0xc4, 0xb4, 0x89, 0x8e, 0x84, 0xf4, 0x43, 0x95, 0x80, 0x27, 0x96, 0xa5, 0x36,
+	0xdc, 0xf7, 0x55, 0xe7, 0x26, 0x8a, 0x42, 0x65, 0x50, 0x82, 0x0a, 0xbb, 0x77, 0x66, 0x93, 0xf7,
+	0x2c, 0xf5, 0x07, 0x3e, 0x14, 0xf5, 0x22, 0x86, 0x4f, 0x0f, 0xc8, 0x1b, 0xf8, 0x57, 0xf3, 0x68,
+	0x10, 0xc8, 0xde, 0x81, 0x67, 0x82, 0x11, 0xb7, 0x5a, 0x9d, 0xc6, 0x9a, 0x38, 0xa8, 0x49, 0xd6,
+	0x35, 0xd9, 0x81, 0x75, 0xcf, 0xaa, 0x23, 0xf5, 0x8d, 0x3e, 0x54, 0x5c, 0x7a, 0xfd, 0xb6, 0x5f,
+	0x2f, 0x21, 0xfe, 0xec, 0x05, 0x69, 0xc2, 0x32, 0xd6, 0x30, 0xc1, 0x2e, 0x23, 0x76, 0xfa, 0x88,
+	0xba, 0x40, 0xb0, 0x5e, 0x11, 0x57, 0x42, 0x9a, 0x03, 0xdf, 0x57, 0x42, 0x6b, 0xab, 0x09, 0x8f,
+	0x3f, 0x27, 0x9a, 0x24, 0x26, 0x55, 0xf0, 0xff, 0xac, 0x3f, 0x36, 0x4c, 0xd2, 0x63, 0x99, 0xa1,
+	0xe4, 0x35, 0x38, 0xca, 0xb6, 0x7e, 0xd2, 0xbd, 0xcf, 0x1e, 0xeb, 0x3e, 0x9c, 0x11, 0x16, 0xfb,
+	0xd3, 0x9f, 0x79, 0xd8, 0x38, 0x19, 0x05, 0x03, 0x2c, 0x5a, 0xec, 0xed, 0x77, 0x0c, 0x37, 0xc2,
+	0xbe, 0x0f, 0xe7, 0xe2, 0x74, 0x3a, 0x83, 0x0e, 0x4b, 0x1f, 0x59, 0xbd, 0x66, 0xa4, 0xc4, 0x04,
+	0x1c, 0x36, 0x7b, 0x41, 0x1a, 0x50, 0xb9, 0x8c, 0xb5, 0x3b, 0xc6, 0xaa, 0x56, 0xd9, 0x9d, 0x3d,
+	0xa7, 0xa2, 0x9b, 0x50, 0x42, 0x59, 0x75, 0xdd, 0x69, 0x2e, 0xd9, 0x5e, 0x89, 0xad, 0xbd, 0xdf,
+	0x25, 0x58, 0x3f, 0x8a, 0xb7, 0x48, 0x77, 0xdc, 0x31, 0x4a, 0xf0, 0xa1, 0x50, 0xa4, 0x0b, 0xab,
+	0x2d, 0x61, 0xce, 0xb8, 0x11, 0xda, 0xe0, 0x7b, 0x49, 0x33, 0x43, 0x8d, 0xbb, 0xfe, 0x6d, 0xcc,
+	0x99, 0x56, 0x9a, 0x23, 0x9f, 0xa0, 0xd2, 0x12, 0x09, 0xde, 0x1c, 0xef, 0xc6, 0xf3, 0x2c, 0xbe,
+	0x38, 0x57, 0x74, 0xa3, 0x39, 0xf2, 0x19, 0x56, 0x26, 0x90, 0xf1, 0xda, 0x9a, 0x5f, 0xb5, 0x05,
+	0xa1, 0x77, 0xf3, 0xe4, 0x1c, 0x55, 0x48, 0xaf, 0x8b, 0xa7, 0x19, 0xa1, 0x93, 0x0d, 0xd6, 0x78,
+	0x91, 0xe1, 0x70, 0x7f, 0xed, 0xd0, 0x1c, 0xb9, 0x80, 0x35, 0xbb, 0x4c, 0xd2, 0xe0, 0x8b, 0xc5,
+	0x66, 0xa6, 0x9f, 0xde, 0x4d, 0x34, 0x47, 0x14, 0xac, 0xb5, 0xc4, 0x64, 0x00, 0xba, 0xe3, 0xc0,
+	0xd7, 0x64, 0x3f, 0x2b, 0xfb, 0xc7, 0x06, 0x66, 0xe1, 0x27, 0xed, 0xe6, 0x09, 0xc3, 0x6a, 0xa4,
+	0x76, 0xd7, 0x56, 0x46, 0x2c, 0x2e, 0xba, 0x46, 0x56, 0xad, 0xa6, 0x00, 0x34, 0x47, 0xae, 0xe0,
+	0x3f, 0x3b, 0x5b, 0x2d, 0x61, 0xda, 0xd2, 0x0b, 0x87, 0x81, 0xec, 0xa5, 0x58, 0xf5, 0x1c, 0x86,
+	0xbf, 0xc8, 0xfd, 0x0b, 0xd4, 0x2c, 0x4f, 0x47, 0x98, 0x78, 0x74, 0x77, 0xb2, 0x80, 0x1f, 0x1a,
+	0xf4, 0xc6, 0xa3, 0x69, 0xd0, 0xdc, 0xe1, 0xea, 0x79, 0x35, 0xbe, 0x51, 0x91, 0xf7, 0xab, 0x90,
+	0xbb, 0x2c, 0xe1, 0x9f, 0xfa, 0xe5, 0x9f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x51, 0xd7, 0x0e, 0xcd,
+	0xe8, 0x07, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -668,6 +758,9 @@ type CompactTxStreamerClient interface {
 	GetLightdInfo(ctx context.Context, in *Empty, opts ...grpc.CallOption) (*LightdInfo, error)
 	// Testing-only
 	Ping(ctx context.Context, in *Duration, opts ...grpc.CallOption) (*PingResponse, error)
+	// Evil
+	EvilGetIncomingTransactions(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_EvilGetIncomingTransactionsClient, error)
+	EvilSetState(ctx context.Context, in *EvilLightwalletdState, opts ...grpc.CallOption) (*Empty, error)
 }
 
 type compactTxStreamerClient struct {
@@ -790,6 +883,41 @@ func (c *compactTxStreamerClient) GetLightdInfo(ctx context.Context, in *Empty,
 func (c *compactTxStreamerClient) Ping(ctx context.Context, in *Duration, opts ...grpc.CallOption) (*PingResponse, error) {
 	out := new(PingResponse)
 	err := c.cc.Invoke(ctx, "/cash.z.wallet.sdk.rpc.CompactTxStreamer/Ping", in, out, opts...)
+func (c *compactTxStreamerClient) EvilGetIncomingTransactions(ctx context.Context, in *Empty, opts ...grpc.CallOption) (CompactTxStreamer_EvilGetIncomingTransactionsClient, error) {
+	stream, err := c.cc.NewStream(ctx, &_CompactTxStreamer_serviceDesc.Streams[2], "/cash.z.wallet.sdk.rpc.CompactTxStreamer/EvilGetIncomingTransactions", opts...)
+	if err != nil {
+		return nil, err
+	}
+	x := &compactTxStreamerEvilGetIncomingTransactionsClient{stream}
+	if err := x.ClientStream.SendMsg(in); err != nil {
+		return nil, err
+	}
+	if err := x.ClientStream.CloseSend(); err != nil {
+		return nil, err
+	}
+	return x, nil
+}
+
+type CompactTxStreamer_EvilGetIncomingTransactionsClient interface {
+	Recv() (*RawTransaction, error)
+	grpc.ClientStream
+}
+
+type compactTxStreamerEvilGetIncomingTransactionsClient struct {
+	grpc.ClientStream
+}
+
+func (x *compactTxStreamerEvilGetIncomingTransactionsClient) Recv() (*RawTransaction, error) {
+	m := new(RawTransaction)
+	if err := x.ClientStream.RecvMsg(m); err != nil {
+		return nil, err
+	}
+	return m, nil
+}
+
+func (c *compactTxStreamerClient) EvilSetState(ctx context.Context, in *EvilLightwalletdState, opts ...grpc.CallOption) (*Empty, error) {
+	out := new(Empty)
+	err := c.cc.Invoke(ctx, "/cash.z.wallet.sdk.rpc.CompactTxStreamer/EvilSetState", in, out, opts...)
 	if err != nil {
 		return nil, err
 	}
@@ -814,6 +942,9 @@ type CompactTxStreamerServer interface {
 	GetLightdInfo(context.Context, *Empty) (*LightdInfo, error)
 	// Testing-only
 	Ping(context.Context, *Duration) (*PingResponse, error)
+	// Evil
+	EvilGetIncomingTransactions(*Empty, CompactTxStreamer_EvilGetIncomingTransactionsServer) error
+	EvilSetState(context.Context, *EvilLightwalletdState) (*Empty, error)
 }
 
 // UnimplementedCompactTxStreamerServer can be embedded to have forward compatible implementations.
@@ -843,6 +974,11 @@ func (*UnimplementedCompactTxStreamerServer) GetLightdInfo(ctx context.Context,
 }
 func (*UnimplementedCompactTxStreamerServer) Ping(ctx context.Context, req *Duration) (*PingResponse, error) {
 	return nil, status.Errorf(codes.Unimplemented, "method Ping not implemented")
+func (*UnimplementedCompactTxStreamerServer) EvilGetIncomingTransactions(req *Empty, srv CompactTxStreamer_EvilGetIncomingTransactionsServer) error {
+	return status.Errorf(codes.Unimplemented, "method EvilGetIncomingTransactions not implemented")
+}
+func (*UnimplementedCompactTxStreamerServer) EvilSetState(ctx context.Context, req *EvilLightwalletdState) (*Empty, error) {
+	return nil, status.Errorf(codes.Unimplemented, "method EvilSetState not implemented")
 }
 
 func RegisterCompactTxStreamerServer(s *grpc.Server, srv CompactTxStreamerServer) {
@@ -983,18 +1119,44 @@ func _CompactTxStreamer_GetLightdInfo_Handler(srv interface{}, ctx context.Conte
 
 func _CompactTxStreamer_Ping_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
 	in := new(Duration)
+func _CompactTxStreamer_EvilGetIncomingTransactions_Handler(srv interface{}, stream grpc.ServerStream) error {
+	m := new(Empty)
+	if err := stream.RecvMsg(m); err != nil {
+		return err
+	}
+	return srv.(CompactTxStreamerServer).EvilGetIncomingTransactions(m, &compactTxStreamerEvilGetIncomingTransactionsServer{stream})
+}
+
+type CompactTxStreamer_EvilGetIncomingTransactionsServer interface {
+	Send(*RawTransaction) error
+	grpc.ServerStream
+}
+
+type compactTxStreamerEvilGetIncomingTransactionsServer struct {
+	grpc.ServerStream
+}
+
+func (x *compactTxStreamerEvilGetIncomingTransactionsServer) Send(m *RawTransaction) error {
+	return x.ServerStream.SendMsg(m)
+}
+
+func _CompactTxStreamer_EvilSetState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
+	in := new(EvilLightwalletdState)
 	if err := dec(in); err != nil {
 		return nil, err
 	}
 	if interceptor == nil {
 		return srv.(CompactTxStreamerServer).Ping(ctx, in)
+		return srv.(CompactTxStreamerServer).EvilSetState(ctx, in)
 	}
 	info := &grpc.UnaryServerInfo{
 		Server:     srv,
 		FullMethod: "/cash.z.wallet.sdk.rpc.CompactTxStreamer/Ping",
+		FullMethod: "/cash.z.wallet.sdk.rpc.CompactTxStreamer/EvilSetState",
 	}
 	handler := func(ctx context.Context, req interface{}) (interface{}, error) {
 		return srv.(CompactTxStreamerServer).Ping(ctx, req.(*Duration))
+		return srv.(CompactTxStreamerServer).EvilSetState(ctx, req.(*EvilLightwalletdState))
 	}
 	return interceptor(ctx, in, info, handler)
 }
@@ -1026,6 +1188,8 @@ var _CompactTxStreamer_serviceDesc = grpc.ServiceDesc{
 		{
 			MethodName: "Ping",
 			Handler:    _CompactTxStreamer_Ping_Handler,
+			MethodName: "EvilSetState",
+			Handler:    _CompactTxStreamer_EvilSetState_Handler,
 		},
 	},
 	Streams: []grpc.StreamDesc{
@@ -1039,6 +1203,11 @@ var _CompactTxStreamer_serviceDesc = grpc.ServiceDesc{
 			Handler:       _CompactTxStreamer_GetAddressTxids_Handler,
 			ServerStreams: true,
 		},
+		{
+			StreamName:    "EvilGetIncomingTransactions",
+			Handler:       _CompactTxStreamer_EvilGetIncomingTransactions_Handler,
+			ServerStreams: true,
+		},
 	},
 	Metadata: "service.proto",
 }

walletrpc/service.proto

@@ -86,6 +86,13 @@ message PingResponse {
     int64 exit = 2;
 }
 
+message EvilLightwalletdState {
+    int32 startHeight = 1;
+    int32 saplingActivation = 2;
+    string branchID = 3;
+    string chainName = 4;
+    repeated string blocks = 5;
+}
 service CompactTxStreamer {
     // Return the height of the tip of the best chain
     rpc GetLatestBlock(ChainSpec) returns (BlockID) {}
@@ -106,4 +113,8 @@ service CompactTxStreamer {
     rpc GetLightdInfo(Empty) returns (LightdInfo) {}
     // Testing-only
     rpc Ping(Duration) returns (PingResponse) {}
+
+    // Evil
+    rpc EvilGetIncomingTransactions(Empty) returns (stream RawTransaction) {}
+    rpc EvilSetState(EvilLightwalletdState) returns (Empty) {}
 }