package frontend import ( "context" "encoding/hex" "encoding/json" "errors" "regexp" "strconv" "strings" "time" "github.com/btcsuite/btcd/rpcclient" "github.com/sirupsen/logrus" //"git.hush.is/hush/lightwalletd/common" "git.hush.is/duke/lightwalletd/common" // go should be able to import branches! // duke master = mempool branch "git.hush.is/duke/lightwalletd/walletrpc" //"git.hush.is/hush/lightwalletd/walletrpc" "git.hush.is/duke/lightwalletd/parser" ) type lwdStreamer struct { cache *common.BlockCache chainName string pingEnable bool walletrpc.UnimplementedCompactTxStreamerServer } // NewLwdStreamer constructs a gRPC context. func NewLwdStreamer(cache *common.BlockCache, chainName string, enablePing bool) (walletrpc.CompactTxStreamerServer, error) { return &lwdStreamer{cache: cache, chainName: chainName, pingEnable: enablePing}, nil } var ( ErrUnspecified = errors.New("request for unspecified identifier") ) // the service type type SqlStreamer struct { cache *common.BlockCache client *rpcclient.Client log *logrus.Entry } func (s *SqlStreamer) GracefulStop() error { return nil } func (s *SqlStreamer) GetCache() *common.BlockCache { return s.cache } // GetLatestBlock returns the height of the best chain according to hushd func (s *lwdStreamer) GetLatestBlock(ctx context.Context, placeholder *walletrpc.ChainSpec) (*walletrpc.BlockID, error) { latestBlock := s.cache.GetLatestHeight() latestHash := s.cache.GetLatestHash() if latestBlock == -1 { return nil, errors.New("Cache is empty. Server is probably not yet ready") } return &walletrpc.BlockID{Height: uint64(latestBlock), Hash: latestHash}, nil } func (s *SqlStreamer) GetAddressTxids(addressBlockFilter *walletrpc.TransparentAddressBlockFilter, resp walletrpc.CompactTxStreamer_GetAddressTxidsServer) error { var err error var errCode int64 // Test to make sure Address is a single t address match, err := regexp.Match("^R[a-zA-Z0-9]{33}$", []byte(addressBlockFilter.Address)) if err != nil || !match { s.log.Errorf("Unrecognized address: %s", addressBlockFilter.Address) return nil } params := make([]json.RawMessage, 1) st := "{\"addresses\": [\"" + addressBlockFilter.Address + "\"]," + "\"start\": " + strconv.FormatUint(addressBlockFilter.Range.Start.Height, 10) + ", \"end\": " + strconv.FormatUint(addressBlockFilter.Range.End.Height, 10) + "}" params[0] = json.RawMessage(st) result, rpcErr := s.client.RawRequest("getaddresstxids", params) // For some reason, the error responses are not JSON if rpcErr != nil { s.log.Errorf("Got error: %s", rpcErr.Error()) errParts := strings.SplitN(rpcErr.Error(), ":", 2) errCode, err = strconv.ParseInt(errParts[0], 10, 32) //Check to see if we are requesting a height the hushd doesn't have yet if err == nil && errCode == -8 { return nil } return nil } var txids []string err = json.Unmarshal(result, &txids) if err != nil { s.log.Errorf("Got error: %s", err.Error()) return nil } timeout, cancel := context.WithTimeout(resp.Context(), 30*time.Second) defer cancel() for _, txidstr := range txids { txid, _ := hex.DecodeString(txidstr) // Txid is read as a string, which is in big-endian order. But when converting // to bytes, it should be little-endian for left, right := 0, len(txid)-1; left < right; left, right = left+1, right-1 { txid[left], txid[right] = txid[right], txid[left] } tx, err := s.GetTransaction(timeout, &walletrpc.TxFilter{Hash: txid}) if err != nil { s.log.Errorf("Got error: %s", err.Error()) return nil } resp.Send(tx) } return nil } func (s *SqlStreamer) GetBlock(ctx context.Context, id *walletrpc.BlockID) (*walletrpc.CompactBlock, error) { if id.Height == 0 && id.Hash == nil { return nil, ErrUnspecified } // Precedence: a hash is more specific than a height. If we have it, use it first. if id.Hash != nil { // TODO: Get block by hash return nil, errors.New("GetBlock by Hash is not yet implemented") } else { cBlock, err := common.GetBlock(s.client, s.cache, int(id.Height)) if err != nil { return nil, err } return cBlock, err } } func (s *SqlStreamer) GetBlockRange(span *walletrpc.BlockRange, resp walletrpc.CompactTxStreamer_GetBlockRangeServer) error { blockChan := make(chan walletrpc.CompactBlock) errChan := make(chan error) go common.GetBlockRange(s.client, s.cache, blockChan, errChan, int(span.Start.Height), int(span.End.Height)) for { select { case err := <-errChan: // this will also catch context.DeadlineExceeded from the timeout return err case cBlock := <-blockChan: err := resp.Send(&cBlock) if err != nil { return err } } } return nil } func (s *SqlStreamer) GetTransaction(ctx context.Context, txf *walletrpc.TxFilter) (*walletrpc.RawTransaction, error) { var txBytes []byte var txHeight float64 if txf.Hash != nil { txid := txf.Hash for left, right := 0, len(txid)-1; left < right; left, right = left+1, right-1 { txid[left], txid[right] = txid[right], txid[left] } leHashString := hex.EncodeToString(txid) // First call to get the raw transaction bytes params := make([]json.RawMessage, 1) params[0] = json.RawMessage("\"" + leHashString + "\"") result, rpcErr := s.client.RawRequest("getrawtransaction", params) var err error var errCode int64 // For some reason, the error responses are not JSON if rpcErr != nil { s.log.Errorf("Got error: %s", rpcErr.Error()) errParts := strings.SplitN(rpcErr.Error(), ":", 2) errCode, err = strconv.ParseInt(errParts[0], 10, 32) //Check to see if we are requesting a height the hushd doesn't have yet if err == nil && errCode == -8 { return nil, err } return nil, err } var txhex string err = json.Unmarshal(result, &txhex) if err != nil { return nil, err } txBytes, err = hex.DecodeString(txhex) if err != nil { return nil, err } // Second call to get height params = make([]json.RawMessage, 2) params[0] = json.RawMessage("\"" + leHashString + "\"") params[1] = json.RawMessage("1") result, rpcErr = s.client.RawRequest("getrawtransaction", params) // For some reason, the error responses are not JSON if rpcErr != nil { s.log.Errorf("Got error: %s", rpcErr.Error()) errParts := strings.SplitN(rpcErr.Error(), ":", 2) errCode, err = strconv.ParseInt(errParts[0], 10, 32) //Check to see if we are requesting a height the hushd doesn't have yet if err == nil && errCode == -8 { return nil, err } return nil, err } var txinfo interface{} err = json.Unmarshal(result, &txinfo) if err != nil { return nil, err } txHeight = txinfo.(map[string]interface{})["height"].(float64) return &walletrpc.RawTransaction{Data: txBytes, Height: uint64(txHeight)}, nil } if txf.Block.Hash != nil { s.log.Error("Can't GetTransaction with a blockhash+num. Please call GetTransaction with txid") return nil, errors.New("Can't GetTransaction with a blockhash+num. Please call GetTransaction with txid") } return &walletrpc.RawTransaction{Data: txBytes, Height: uint64(txHeight)}, nil } // GetLightdInfo gets the LightWalletD (this server) info func (s *SqlStreamer) GetLightdInfo(ctx context.Context, in *walletrpc.Empty) (*walletrpc.LightdInfo, error) { saplingHeight, blockHeight, chainName, consensusBranchId, difficulty, longestchain, notarized, err := common.GetSaplingInfo(s.client) if err != nil { s.log.WithFields(logrus.Fields{ "error": err, }).Warn("Unable to get sapling activation height") return nil, err } // TODO these are called Error but they aren't at the moment. // A success will return code 0 and message txhash. return &walletrpc.LightdInfo{ Version: "0.1.1-hushlightd", Vendor: "Silentdragonlite LightWalletD", TaddrSupport: true, ChainName: chainName, SaplingActivationHeight: uint64(saplingHeight), ConsensusBranchId: consensusBranchId, BlockHeight: uint64(blockHeight), Difficulty: uint64(difficulty), Longestchain: uint64(longestchain), Notarized: uint64(notarized), }, nil } // GetCoinsupply gets the Coinsupply info func (s *SqlStreamer) GetCoinsupply(ctx context.Context, in *walletrpc.Empty) (*walletrpc.Coinsupply, error) { result, coin, height, supply, zfunds, total, err := common.GetCoinsupply(s.client) if err != nil { s.log.WithFields(logrus.Fields{ "error": err, }).Warn("Unable to get Coinsupply") return nil, err } // TODO these are called Error but they aren't at the moment. // A success will return code 0 and message txhash. return &walletrpc.Coinsupply{ Result: result, Coin: coin, Height: uint64(height), Supply: uint64(supply), Zfunds: uint64(zfunds), Total: uint64(total), }, nil } // SendTransaction forwards raw transaction bytes to a hushd instance over JSON-RPC func (s *SqlStreamer) SendTransaction(ctx context.Context, rawtx *walletrpc.RawTransaction) (*walletrpc.SendResponse, error) { // sendrawtransaction "hexstring" ( allowhighfees ) // // Submits raw transaction (serialized, hex-encoded) to local node and network. // // Also see createrawtransaction and signrawtransaction calls. // // Arguments: // 1. "hexstring" (string, required) The hex string of the raw transaction) // 2. allowhighfees (boolean, optional, default=false) Allow high fees // // Result: // "hex" (string) The transaction hash in hex // Construct raw JSON-RPC params params := make([]json.RawMessage, 1) txHexString := hex.EncodeToString(rawtx.Data) params[0] = json.RawMessage("\"" + txHexString + "\"") result, rpcErr := s.client.RawRequest("sendrawtransaction", params) var err error var errCode int64 var errMsg string // For some reason, the error responses are not JSON if rpcErr != nil { errParts := strings.SplitN(rpcErr.Error(), ":", 2) errMsg = strings.TrimSpace(errParts[1]) errCode, err = strconv.ParseInt(errParts[0], 10, 32) if err != nil { // This should never happen. We can't panic here, but it's that class of error. // This is why we need integration testing to work better than regtest currently does. TODO. return nil, errors.New("SendTransaction couldn't parse error code") } } else { errMsg = string(result) } // TODO these are called Error but they aren't at the moment. // A success will return code 0 and message txhash. return &walletrpc.SendResponse{ ErrorCode: int32(errCode), ErrorMessage: errMsg, }, nil } func (s *lwdStreamer) GetMempoolStream(_empty *walletrpc.Empty, resp walletrpc.CompactTxStreamer_GetMempoolStreamServer) error { ch := make(chan *walletrpc.RawTransaction, 200) go common.AddNewClient(ch) for { select { case rtx, more := <-ch: if !more || rtx == nil { return nil } if resp.Send(rtx) != nil { return nil } // Timeout after 5 mins case <-time.After(5 * time.Minute): return nil } } } // Key is 32-byte txid (as a 64-character string), data is pointer to compact tx. var mempoolMap *map[string]*walletrpc.CompactTx var mempoolList []string // Last time we pulled a copy of the mempool from zcashd. var lastMempool time.Time func (s *lwdStreamer) GetMempoolTx(exclude *walletrpc.Exclude, resp walletrpc.CompactTxStreamer_GetMempoolTxServer) error { if time.Now().Sub(lastMempool).Seconds() >= 2 { lastMempool = time.Now() // Refresh our copy of the mempool. params := make([]json.RawMessage, 0) result, rpcErr := common.RawRequest("getrawmempool", params) if rpcErr != nil { return rpcErr } err := json.Unmarshal(result, &mempoolList) if err != nil { return err } newmempoolMap := make(map[string]*walletrpc.CompactTx) if mempoolMap == nil { mempoolMap = &newmempoolMap } for _, txidstr := range mempoolList { if ctx, ok := (*mempoolMap)[txidstr]; ok { // This ctx has already been fetched, copy pointer to it. newmempoolMap[txidstr] = ctx continue } txidJSON, err := json.Marshal(txidstr) if err != nil { return err } // The "0" is because we only need the raw hex, which is returned as // just a hex string, and not even a json string (with quotes). params := []json.RawMessage{txidJSON, json.RawMessage("0")} result, rpcErr := common.RawRequest("getrawtransaction", params) if rpcErr != nil { // Not an error; mempool transactions can disappear continue } // strip the quotes var txStr string err = json.Unmarshal(result, &txStr) if err != nil { return err } // conver to binary txBytes, err := hex.DecodeString(txStr) if err != nil { return err } tx := parser.NewTransaction() txdata, err := tx.ParseFromSlice(txBytes) if len(txdata) > 0 { return errors.New("extra data deserializing transaction") } newmempoolMap[txidstr] = &walletrpc.CompactTx{} if tx.HasSaplingElements() { newmempoolMap[txidstr] = tx.ToCompact( /* height */ 0) } } mempoolMap = &newmempoolMap } excludeHex := make([]string, len(exclude.Txid)) for i := 0; i < len(exclude.Txid); i++ { excludeHex[i] = hex.EncodeToString(parser.Reverse(exclude.Txid[i])) } for _, txid := range MempoolFilter(mempoolList, excludeHex) { tx := (*mempoolMap)[txid] if len(tx.Hash) > 0 { err := resp.Send(tx) if err != nil { return err } } } return nil } // Return the subset of items that aren't excluded, but // if more than one item matches an exclude entry, return // all those items. func MempoolFilter(items, exclude []string) []string { sort.Slice(items, func(i, j int) bool { return items[i] < items[j] }) sort.Slice(exclude, func(i, j int) bool { return exclude[i] < exclude[j] }) // Determine how many items match each exclude item. nmatches := make([]int, len(exclude)) // is the exclude string less than the item string? lessthan := func(e, i string) bool { l := len(e) if l > len(i) { l = len(i) } return e < i[0:l] } ei := 0 for _, item := range items { for ei < len(exclude) && lessthan(exclude[ei], item) { ei++ } match := ei < len(exclude) && strings.HasPrefix(item, exclude[ei]) if match { nmatches[ei]++ } } // Add each item that isn't uniquely excluded to the results. tosend := make([]string, 0) ei = 0 for _, item := range items { for ei < len(exclude) && lessthan(exclude[ei], item) { ei++ } match := ei < len(exclude) && strings.HasPrefix(item, exclude[ei]) if !match || nmatches[ei] > 1 { tosend = append(tosend, item) } } return tosend }