Implement validation for Stellar deposits (#86)

backend/main.go

@@ -129,7 +129,7 @@ func (w *Worker) processStellarWithdrawalRequest(sr store.SignatureRequest) erro
 
 	// Load source account sequence
 	sourceAccount, err := w.StellarClient.AccountDetail(horizonclient.AccountRequest{
-		AccountID: deposit.Destination,
+		AccountID: details.Recipient,
 	})
 	if err != nil {
 		return errors.Wrap(err, "error getting account details")
@@ -157,8 +157,8 @@ func (w *Worker) processStellarWithdrawalRequest(sr store.SignatureRequest) erro
 	}
 
 	tx, err := w.StellarBuilder.BuildTransaction(
-		deposit.Destination,
-		deposit.Destination,
+		details.Recipient,
+		details.Recipient,
 		amountRat.FloatString(7),
 		sourceAccount.Sequence+1,
 		// TODO: ensure using WithdrawExpiration without any time buffer is safe
@@ -184,10 +184,11 @@ func (w *Worker) processStellarWithdrawalRequest(sr store.SignatureRequest) erro
 	}
 
 	outgoingTx := store.OutgoingStellarTransaction{
-		Envelope:  txBase64,
-		Action:    sr.Action,
-		DepositID: sr.DepositID,
-		Sequence:  tx.SeqNum(),
+		Envelope:      txBase64,
+		Action:        sr.Action,
+		DepositID:     sr.DepositID,
+		SourceAccount: details.Recipient,
+		Sequence:      tx.SeqNum(),
 	}
 	err = w.Store.UpsertOutgoingStellarTransaction(context.TODO(), outgoingTx)
 	if err != nil {

backend/stellar_withdrawal_validator.go

@@ -7,9 +7,10 @@ import (
 	"net/http"
 	"time"
 
-	"github.com/stellar/go/support/db"
-
 	"github.com/ethereum/go-ethereum/common"
+
+	"github.com/stellar/go/strkey"
+	"github.com/stellar/go/support/db"
 	"github.com/stellar/go/support/errors"
 	"github.com/stellar/go/support/render/problem"
 	"github.com/stellar/starbridge/store"
@@ -37,11 +38,17 @@ var (
 	}
 	WithdrawalAmountInvalid = problem.P{
 		Type:   "withdrawal_amount_invalid",
-		Title:  "Withdrawal Amouont Invalid",
+		Title:  "Withdrawal Amount Invalid",
 		Status: http.StatusBadRequest,
 		Detail: "Withdrawing the requested amount is not supported by the bridge." +
 			"Refund the deposit once the withdrawal period has expired.",
 	}
+	InvalidStellarRecipient = problem.P{
+		Type:   "invalid_stellar_recipient",
+		Title:  "Invalid Stellar Recipient",
+		Status: http.StatusBadRequest,
+		Detail: "The recipient of the deposit is not a valid Stellar address.",
+	}
 
 	ethereumTokenAddress = common.Address{}
 )
@@ -60,6 +67,9 @@ type StellarWithdrawalDetails struct {
 	// Deadline is the deadline for executing the withdrawal
 	// transaction on Stellar.
 	Deadline time.Time
+	// Recipient is the Stellar account which should receive the
+	// withdrawal.
+	Recipient string
 	// LedgerSequence is the sequence number of the Stellar ledger
 	// for which the validation result is accurate.
 	LedgerSequence uint32
@@ -79,8 +89,20 @@ func (s StellarWithdrawalValidator) CanWithdraw(ctx context.Context, deposit sto
 		return StellarWithdrawalDetails{}, WithdrawalAmountInvalid
 	}
 
+	destination, ok := new(big.Int).SetString(deposit.Destination, 10)
+	if !ok {
+		return StellarWithdrawalDetails{}, InvalidStellarRecipient
+	}
+	destinationAccountID, err := strkey.Encode(
+		strkey.VersionByteAccountID,
+		destination.Bytes(),
+	)
+	if err != nil {
+		return StellarWithdrawalDetails{}, InvalidStellarRecipient
+	}
+
 	dbStore := store.DB{Session: s.Session.Clone()}
-	err := dbStore.Session.BeginTx(&sql.TxOptions{
+	err = dbStore.Session.BeginTx(&sql.TxOptions{
 		Isolation: sql.LevelRepeatableRead,
 		ReadOnly:  true,
 	})
@@ -117,6 +139,7 @@ func (s StellarWithdrawalValidator) CanWithdraw(ctx context.Context, deposit sto
 
 	return StellarWithdrawalDetails{
 		Deadline:       withdrawalDeadline,
+		Recipient:      destinationAccountID,
 		LedgerSequence: lastLedgerSequence,
 	}, nil
 }

stellar/controllers/ethereum_deposit.go

@@ -6,7 +6,6 @@ import (
 	"regexp"
 	"strconv"
 
-	"github.com/stellar/go/strkey"
 	"github.com/stellar/go/support/render/problem"
 	"github.com/stellar/starbridge/ethereum"
 	"github.com/stellar/starbridge/store"
@@ -43,12 +42,6 @@ var (
 		Status: http.StatusUnprocessableEntity,
 		Detail: "Retry later once the transaction has more confirmations.",
 	}
-	InvalidStellarRecipient = problem.P{
-		Type:   "invalid_stellar_recipient",
-		Title:  "Invalid Stellar Recipient",
-		Status: http.StatusUnprocessableEntity,
-		Detail: "The recipient of the deposit is not a valid Stellar address.",
-	}
 
 	validTxHash = regexp.MustCompile("^(0x)?([A-Fa-f0-9]{64})$")
 )
@@ -89,18 +82,11 @@ func getEthereumDeposit(observer ethereum.Observer, depositStore *store.DB, fina
 		return store.EthereumDeposit{}, EthereumTxRequiresMoreConfirmations
 	}
 
-	destinationAccountID, err := strkey.Encode(
-		strkey.VersionByteAccountID,
-		deposit.Destination.Bytes(),
-	)
-	if err != nil {
-		return store.EthereumDeposit{}, InvalidStellarRecipient
-	}
 	storeDeposit = store.EthereumDeposit{
 		ID:          depositID,
 		Token:       deposit.Token.String(),
 		Sender:      deposit.Sender.String(),
-		Destination: destinationAccountID,
+		Destination: deposit.Destination.String(),
 		Amount:      deposit.Amount.String(),
 		Hash:        deposit.TxHash.String(),
 		LogIndex:    deposit.LogIndex,

stellar/controllers/stellar_deposit.go

@@ -0,0 +1,46 @@
+package controllers
+
+import (
+	"database/sql"
+	"net/http"
+	"strings"
+
+	"github.com/stellar/go/support/render/problem"
+	"github.com/stellar/starbridge/store"
+)
+
+var (
+	InvalidStellarTxHash = problem.P{
+		Type:   "invalid_stellar_tx_hash",
+		Title:  "Invalid Stellar Transaction Hash",
+		Status: http.StatusBadRequest,
+		Detail: "The transaction hash of the Stellar transaction is invalid.",
+	}
+	StellarTxHashNotFound = problem.P{
+		Type:   "stellar_tx_hash_not_found",
+		Title:  "Stellar Transaction Hash Not Found",
+		Status: http.StatusNotFound,
+		Detail: "The stellar transaction cannot be found.",
+	}
+	InvalidEthereumRecipient = problem.P{
+		Type:   "invalid_ethereum_recipient",
+		Title:  "Invalid Ethereum Recipient",
+		Status: http.StatusUnprocessableEntity,
+		Detail: "The recipient of the deposit is not a valid Ethereum address.",
+	}
+	// TODO: remove this once getStellarDeposit is used
+	_ = getStellarDeposit
+)
+
+func getStellarDeposit(depositStore *store.DB, r *http.Request) (store.StellarDeposit, error) {
+	txHash := strings.TrimPrefix(r.PostFormValue("transaction_hash"), "0x")
+	if !validTxHash.MatchString(txHash) {
+		return store.StellarDeposit{}, InvalidStellarTxHash
+	}
+
+	deposit, err := depositStore.GetStellarDeposit(r.Context(), txHash)
+	if err == sql.ErrNoRows {
+		return store.StellarDeposit{}, StellarTxHashNotFound
+	}
+	return deposit, err
+}

stellar/controllers/stellar_withdrawal_handler.go

@@ -34,7 +34,7 @@ func (c *StellarWithdrawalHandler) ServeHTTP(w http.ResponseWriter, r *http.Requ
 	}
 	if err == nil {
 		sourceAccount, err := c.StellarClient.AccountDetail(horizonclient.AccountRequest{
-			AccountID: deposit.Destination,
+			AccountID: outgoingTransaction.SourceAccount,
 		})
 		if err != nil {
 			problem.Render(r.Context(), w, err)

stellar/txobserver/main.go

@@ -80,7 +80,7 @@ func (o *Observer) ProcessNewLedgers() {
 			} else {
 				o.log.WithField("sequence", o.ledgerSequence).Info("Processing ledger...")
 
-				err = o.processSingleLedger(ledger)
+				err = o.ingestLedger(ledger)
 				if err != nil {
 					o.log.WithFields(slog.F{"error": err, "sequence": o.ledgerSequence}).Error("Error processing a single ledger details")
 				} else {
@@ -114,15 +114,15 @@ func (o *Observer) catchupLedgers() error {
 
 	// Process past bridge account payments
 	cursor := toid.AfterLedger(ledgerSeq).String()
-	previousHash := ""
 	var lastOp operations.Operation
 	for o.ctx.Err() == nil {
 		ops, err := o.client.Payments(horizonclient.OperationRequest{
-			ForAccount: o.bridgeAccount,
-			Cursor:     cursor,
-			Order:      horizonclient.OrderDesc,
-			Limit:      200,
-			Join:       "transactions",
+			ForAccount:    o.bridgeAccount,
+			Cursor:        cursor,
+			Order:         horizonclient.OrderDesc,
+			Limit:         200,
+			IncludeFailed: false,
+			Join:          "transactions",
 		})
 		if err != nil {
 			return errors.Wrap(err, "error getting operations")
@@ -132,14 +132,13 @@ func (o *Observer) catchupLedgers() error {
 			break
 		}
 
-		err = o.processOpsSinglePage(ops.Embedded.Records, previousHash)
+		err = o.ingestPage(ops.Embedded.Records)
 		if err != nil {
 			return err
 		}
 
 		lastOp = ops.Embedded.Records[len(ops.Embedded.Records)-1]
 		cursor = lastOp.PagingToken()
-		previousHash = lastOp.GetBase().TransactionHash
 	}
 
 	if o.ctx.Err() != nil {
@@ -169,7 +168,7 @@ func (o *Observer) catchupLedgers() error {
 	return nil
 }
 
-func (o *Observer) processSingleLedger(ledger horizon.Ledger) error {
+func (o *Observer) ingestLedger(ledger horizon.Ledger) error {
 	err := o.store.Session.Begin()
 	if err != nil {
 		return errors.Wrap(err, "error starting a transaction")
@@ -180,13 +179,12 @@ func (o *Observer) processSingleLedger(ledger horizon.Ledger) error {
 	}()
 	// Process operations
 	cursor := ""
-	previousHash := ""
 	for {
-		ops, err := o.client.Operations(horizonclient.OperationRequest{
+		ops, err := o.client.Payments(horizonclient.OperationRequest{
 			ForLedger:     uint(ledger.Sequence),
 			Cursor:        cursor,
 			Limit:         200,
-			IncludeFailed: true,
+			IncludeFailed: false,
 			Join:          "transactions",
 		})
 		if err != nil {
@@ -197,14 +195,13 @@ func (o *Observer) processSingleLedger(ledger horizon.Ledger) error {
 			break
 		}
 
-		err = o.processOpsSinglePage(ops.Embedded.Records, previousHash)
+		err = o.ingestPage(ops.Embedded.Records)
 		if err != nil {
 			return err
 		}
 
 		lastOp := ops.Embedded.Records[len(ops.Embedded.Records)-1].GetBase()
 		cursor = lastOp.PagingToken()
-		previousHash = lastOp.TransactionHash
 	}
 
 	err = o.store.UpdateLastLedgerSequence(context.TODO(), uint32(ledger.Sequence))
@@ -226,37 +223,81 @@ func (o *Observer) processSingleLedger(ledger horizon.Ledger) error {
 	return nil
 }
 
-func (o *Observer) processOpsSinglePage(ops []operations.Operation, previousHash string) error {
-	for _, op := range ops {
-		baseOp := op.GetBase()
+func validTransaction(horizonTx *horizon.Transaction) bool {
+	// ignore failed transactions
+	return horizonTx.Successful &&
+		// Skip inserting transactions with multiple ops. Currently Starbridge
+		// does not create such transactions but it can change in the future.
+		horizonTx.OperationCount == 1
+}
 
-		// Ignore ops not coming from bridge account
-		if baseOp.SourceAccount != o.bridgeAccount {
+func (o *Observer) ingestPage(ops []operations.Operation) error {
+	for _, op := range ops {
+		payment, ok := op.(operations.Payment)
+		// only consider payment operations
+		if !ok {
 			continue
 		}
 
-		tx := baseOp.Transaction
-		if tx.MemoType != "hash" || tx.Memo == "" || !tx.Successful ||
-			// Skip inserting transactions with multiple ops. Currently Starbridge
-			// does not create such transactions but it can change in the future.
-			previousHash == baseOp.TransactionHash {
+		tx := payment.Transaction
+		if !validTransaction(tx) {
 			continue
 		}
 
-		memoBytes, err := base64.StdEncoding.DecodeString(tx.Memo)
-		if err != nil {
-			return errors.Wrapf(err, "error decoding memo: %s", tx.Memo)
+		if payment.From == o.bridgeAccount {
+			if err := o.ingestOutgoingPayment(payment); err != nil {
+				return err
+			}
+		} else if payment.To == o.bridgeAccount {
+			if err := o.ingestIncomingPayment(payment); err != nil {
+				return err
+			}
 		}
+	}
 
-		err = o.store.InsertHistoryStellarTransaction(context.TODO(), store.HistoryStellarTransaction{
-			Hash:     tx.Hash,
-			Envelope: tx.EnvelopeXdr,
-			MemoHash: hex.EncodeToString(memoBytes),
-		})
-		if err != nil {
-			return errors.Wrapf(err, "error inserting history transaction: %s", tx.Hash)
-		}
-		previousHash = baseOp.TransactionHash
+	return nil
+}
+
+func (o *Observer) ingestOutgoingPayment(payment operations.Payment) error {
+	if payment.Transaction.MemoType != "hash" || payment.Transaction.Memo == "" {
+		return nil
+	}
+
+	memoBytes, err := base64.StdEncoding.DecodeString(payment.Transaction.Memo)
+	if err != nil {
+		return errors.Wrapf(err, "error decoding memo: %s", payment.Transaction.Memo)
+	}
+
+	err = o.store.InsertHistoryStellarTransaction(context.TODO(), store.HistoryStellarTransaction{
+		Hash:     payment.Transaction.Hash,
+		Envelope: payment.Transaction.EnvelopeXdr,
+		MemoHash: hex.EncodeToString(memoBytes),
+	})
+	if err != nil {
+		return errors.Wrapf(err, "error inserting history transaction: %s", payment.Transaction.Hash)
+	}
+
+	return nil
+}
+
+func (o *Observer) ingestIncomingPayment(payment operations.Payment) error {
+	var assetString string
+	if payment.Asset.Type == "native" {
+		assetString = "native"
+	} else {
+		assetString = payment.Asset.Code + ":" + payment.Asset.Issuer
+	}
+
+	deposit := store.StellarDeposit{
+		ID:          payment.Transaction.Hash,
+		Asset:       assetString,
+		LedgerTime:  payment.LedgerCloseTime.Unix(),
+		Sender:      payment.From,
+		Destination: payment.Transaction.Memo,
+		Amount:      payment.Amount,
+	}
+	if err := o.store.InsertStellarDeposit(context.TODO(), deposit); err != nil {
+		return errors.Wrapf(err, "error inserting stellar deposit: %s", payment.Transaction.Hash)
 	}
 
 	return nil