node: add Discord notifications for missing signatures

Change-Id: If09643c2e02c4c166577082cd9be9124d2e775d4

node/cmd/guardiand/node.go

@@ -3,6 +3,7 @@ package guardiand
 import (
 	"context"
 	"fmt"
+	"github.com/certusone/wormhole/node/pkg/notify/discord"
 	"log"
 	"net/http"
 	_ "net/http/pprof"
@@ -85,6 +86,9 @@ var (
 
 	disableHeartbeatVerify *bool
 
+	discordToken   *string
+	discordChannel *string
+
 	bigTablePersistenceEnabled *bool
 	bigTableGCPProject         *string
 	bigTableInstanceName       *string
@@ -137,6 +141,9 @@ func init() {
 	disableHeartbeatVerify = NodeCmd.Flags().Bool("disableHeartbeatVerify", false,
 		"Disable heartbeat signature verification (useful during network startup)")
 
+	discordToken = NodeCmd.Flags().String("discordToken", "", "Discord bot token (optional)")
+	discordChannel = NodeCmd.Flags().String("discordChannel", "", "Discord channel name (optional)")
+
 	bigTablePersistenceEnabled = NodeCmd.Flags().Bool("bigTablePersistenceEnabled", false, "Turn on forwarding events to BigTable")
 	bigTableGCPProject = NodeCmd.Flags().String("bigTableGCPProject", "", "Google Cloud project ID for storing events")
 	bigTableInstanceName = NodeCmd.Flags().String("bigTableInstanceName", "", "BigTable instance name for storing events")
@@ -415,6 +422,14 @@ func runNode(cmd *cobra.Command, args []string) {
 	// Guardian set state managed by processor
 	gst := common.NewGuardianSetState()
 
+	var notifier *discord.DiscordNotifier
+	if *discordToken != "" {
+		notifier, err = discord.NewDiscordNotifier(*discordToken, *discordChannel, logger)
+		if err != nil {
+			logger.Error("failed to initialize Discord bot", zap.Error(err))
+		}
+	}
+
 	// Load p2p private key
 	var priv crypto.PrivKey
 	if *unsafeDevMode {
@@ -501,6 +516,7 @@ func runNode(cmd *cobra.Command, args []string) {
 			*terraLCD,
 			*terraContract,
 			attestationEvents,
+			notifier,
 		)
 		if err := supervisor.Run(ctx, "processor", p.Run); err != nil {
 			return err

node/go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/davecgh/go-spew v1.1.1
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
 	github.com/dgraph-io/badger/v3 v3.2103.1
+	github.com/diamondburned/arikawa/v3 v3.0.0-rc.2
 	github.com/ethereum/go-ethereum v1.10.6
 	github.com/gagliardetto/solana-go v0.3.5-0.20210727215348-0cf016734976
 	github.com/gorilla/mux v1.7.4
@@ -101,6 +102,7 @@ require (
 	github.com/google/gopacket v1.1.19 // indirect
 	github.com/google/uuid v1.2.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.0.5 // indirect
+	github.com/gorilla/schema v1.2.0 // indirect
 	github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect
 	github.com/gtank/merlin v0.1.1 // indirect
 	github.com/gtank/ristretto255 v0.1.2 // indirect

node/go.sum

@@ -267,6 +267,8 @@ github.com/dgryski/go-farm v0.0.0-20190104051053-3adb47b1fb0f/go.mod h1:SqUrOPUn
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2 h1:tdlZCpZ/P9DhczCTSixgIKmwPv6+wP5DGjqLYw5SUiA=
 github.com/dgryski/go-farm v0.0.0-20190423205320-6a90982ecee2/go.mod h1:SqUrOPUnsFjfmXRMNPybcSiG0BgUW2AuFH8PAnS2iTw=
 github.com/dgryski/go-sip13 v0.0.0-20181026042036-e10d5fee7954/go.mod h1:vAd38F8PWV+bWy6jNmig1y/TA+kYO4g3RSRF0IAv0no=
+github.com/diamondburned/arikawa/v3 v3.0.0-rc.2 h1:KP0c+FPykYQFjPwY0ezqx/kPgMZz6oXBXrADeMHnLpw=
+github.com/diamondburned/arikawa/v3 v3.0.0-rc.2/go.mod h1:sNqM/iGXuH87wEH1rpQBEY1PR0AAkRKJuUhJGOdo7To=
 github.com/dlclark/regexp2 v1.2.0/go.mod h1:2pZnwuY/m+8K6iRw6wQdMtk+rH5tNGR1i55kozfMjCc=
 github.com/docker/docker v1.4.2-0.20180625184442-8e610b2b55bf/go.mod h1:eEKB0N0r5NX/I1kEveEz05bcu8tLC/8azJZsviup8Sk=
 github.com/dop251/goja v0.0.0-20200721192441-a695b0cdd498/go.mod h1:Mw6PkjjMXWbTj+nnj4s3QPXq1jaT0s5pC0iFD4+BOAA=
@@ -484,6 +486,8 @@ github.com/gorilla/mux v1.7.3/go.mod h1:1lud6UwP+6orDFRuTfBEV8e9/aOM/c4fVVCaMa2z
 github.com/gorilla/mux v1.7.4 h1:VuZ8uybHlWmqV03+zRzdwKL4tUnIp1MAQtp1mIFE1bc=
 github.com/gorilla/mux v1.7.4/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
 github.com/gorilla/rpc v1.2.0/go.mod h1:V4h9r+4sF5HnzqbwIez0fKSpANP0zlYd3qR7p36jkTQ=
+github.com/gorilla/schema v1.2.0 h1:YufUaxZYCKGFuAq3c96BOhjgd5nmXiOY9NGzF247Tsc=
+github.com/gorilla/schema v1.2.0/go.mod h1:kgLaKoK1FELgZqMAVxx/5cbj0kT+57qxUrAlIO2eleU=
 github.com/gorilla/websocket v0.0.0-20170926233335-4201258b820c/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.0/go.mod h1:E7qHFY5m1UJ88s3WnNqhKjPHQ0heANvMoAMk2YaljkQ=
 github.com/gorilla/websocket v1.4.1/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
@@ -1464,6 +1468,7 @@ golang.org/x/crypto v0.0.0-20200602180216-279210d13fed/go.mod h1:LzIPMQfyMNhhGPh
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200728195943-123391ffb6de/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20200820211705-5c72a883971a/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.0.0-20201016220609-9e8e0b390897/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20210220033148-5ea612d1eb83/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20210322153248-0c34fe9e7dc2/go.mod h1:T9bdIzuCu7OtxOm1hfPfRQxPLYneinmdGuTeoZ9dtd4=
 golang.org/x/crypto v0.0.0-20210506145944-38f3c27a63bf/go.mod h1:P+XmwS30IXTQdn5tA2iutPOUgjI07+tq3H3K9MVA1s8=
@@ -1661,6 +1666,7 @@ golang.org/x/sys v0.0.0-20200803210538-64077c9b5642/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20200814200057-3d37ad5750ed/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200905004654-be1d3432aa8f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
+golang.org/x/sys v0.0.0-20201018230417-eeed37f84f13/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201201145000-ef89a241ccb3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20201214210602-f9fddec55a1e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
@@ -1704,6 +1710,7 @@ golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxb
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
+golang.org/x/time v0.0.0-20200630173020-3af7569d3a1e/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324 h1:Hir2P/De0WpUhtrKGGjvSb2YxUgyZ7EFOSLIcSSpiwE=
 golang.org/x/time v0.0.0-20201208040808-7e3f01d25324/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/tools v0.0.0-20180221164845-07fd8470d635/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

node/pkg/notify/discord/notify.go

@@ -0,0 +1,99 @@
+package discord
+
+import (
+	"bytes"
+	"fmt"
+	"github.com/certusone/wormhole/node/pkg/vaa"
+	"github.com/diamondburned/arikawa/v3/api"
+	"github.com/diamondburned/arikawa/v3/discord"
+	"go.uber.org/zap"
+	"strings"
+)
+
+type DiscordNotifier struct {
+	c      *api.Client
+	chans  []discord.Channel
+	logger *zap.Logger
+}
+
+// NewDiscordNotifier returns and initializes a new Discord notifier.
+//
+// During initialization, a list of all guilds and channels is fetched.
+// Newly added guilds and channels won't be detected at runtime.
+func NewDiscordNotifier(botToken string, channelName string, logger *zap.Logger) (*DiscordNotifier, error) {
+	c := api.NewClient("Bot " + botToken)
+	chans := make([]discord.Channel, 0)
+
+	guilds, err := c.Guilds(0)
+	if err != nil {
+		return nil, fmt.Errorf("failed to retrieve guilds: %w", err)
+	}
+
+	for _, guild := range guilds {
+		gcn, err := c.Channels(guild.ID)
+		if err != nil {
+			return nil, fmt.Errorf("failed to retrieve channels for %s: %w", guild.ID, err)
+		}
+
+		for _, cn := range gcn {
+			if cn.Name == channelName {
+				chans = append(chans, cn)
+			}
+		}
+	}
+
+	logger.Info("notification channels", zap.Any("channels", chans))
+
+	return &DiscordNotifier{
+		c:      c,
+		chans:  chans,
+		logger: logger,
+	}, nil
+}
+
+func wrapCode(in string) string {
+	return fmt.Sprintf("`%s`", in)
+}
+
+func (d DiscordNotifier) MissingSignaturesOnTransaction(v *vaa.VAA, hasSigs, wantSigs int, quorum bool, missing []string) error {
+	if len(missing) == 0 {
+		panic("no missing nodes specified")
+	}
+	var quorumText string
+	if quorum {
+		quorumText = fmt.Sprintf("✔️ yes (%d/%d)", hasSigs, wantSigs)
+	} else {
+		quorumText = fmt.Sprintf("🚨️ **NO** (%d/%d)", hasSigs, wantSigs)
+	}
+
+	var messageText string
+	if !quorum {
+		messageText = "**NO QUORUM** - Wormhole likely failed to achieve consensus on this message @here"
+	}
+
+	missingText := &bytes.Buffer{}
+	for _, m := range missing {
+		if _, err := fmt.Fprintf(missingText, "- %s\n", m); err != nil {
+			panic(err)
+		}
+	}
+
+	for _, cn := range d.chans {
+		if _, err := d.c.SendMessage(cn.ID, messageText,
+			discord.Embed{
+				Title: "Message with missing signatures",
+				Fields: []discord.EmbedField{
+					{Name: "Message ID", Value: wrapCode(v.MessageID()), Inline: true},
+					{Name: "Digest", Value: wrapCode(v.HexDigest()), Inline: true},
+					{Name: "Quorum", Value: quorumText, Inline: true},
+					{Name: "Source Chain", Value: strings.Title(v.EmitterChain.String()), Inline: false},
+					{Name: "Missing Guardians", Value: missingText.String(), Inline: false},
+				},
+			},
+		); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

node/pkg/processor/cleanup.go

@@ -2,6 +2,7 @@ package processor
 
 import (
 	"context"
+	"encoding/hex"
 	"github.com/certusone/wormhole/node/pkg/common"
 	"github.com/certusone/wormhole/node/pkg/vaa"
 	"github.com/prometheus/client_golang/prometheus"
@@ -69,10 +70,39 @@ func (p *Processor) handleCleanup(ctx context.Context) {
 
 			hasSigs := len(s.signatures)
 			wantSigs := CalculateQuorum(len(gs.Keys))
+			quorum := hasSigs >= wantSigs
 
 			var chain vaa.ChainID
 			if s.ourVAA != nil {
 				chain = s.ourVAA.EmitterChain
+
+				// If a notifier is configured, send a notification for any missing signatures.
+				//
+				// Only send a notification if we have a VAA. Otherwise, bogus observations
+				// could cause invalid alerts.
+				if p.notifier != nil && hasSigs < len(gs.Keys) {
+					p.logger.Info("sending miss notification", zap.String("digest", hash))
+					// Find names of missing validators
+					missing := make([]string, 0, len(gs.Keys))
+					for _, k := range gs.Keys {
+						if s.signatures[k] == nil {
+							name := hex.EncodeToString(k.Bytes())
+							h := p.gst.LastHeartbeat(k)
+							// Pick first node if there are multiple peers.
+							for _, hb := range h {
+								name = hb.NodeName
+								break
+							}
+							missing = append(missing, name)
+						}
+					}
+
+					go func(v *vaa.VAA, hasSigs, wantSigs int, quorum bool, missing []string) {
+						if err := p.notifier.MissingSignaturesOnTransaction(v, hasSigs, wantSigs, quorum, missing); err != nil {
+							p.logger.Error("failed to send notification", zap.Error(err))
+						}
+					}(s.ourVAA, hasSigs, wantSigs, quorum, missing)
+				}
 			}
 
 			p.logger.Info("VAA considered settled",
@@ -80,7 +110,7 @@ func (p *Processor) handleCleanup(ctx context.Context) {
 				zap.Duration("delta", delta),
 				zap.Int("have_sigs", hasSigs),
 				zap.Int("required_sigs", wantSigs),
-				zap.Bool("quorum", hasSigs >= wantSigs),
+				zap.Bool("quorum", quorum),
 				zap.Stringer("emitter_chain", chain),
 			)
 

node/pkg/processor/processor.go

@@ -3,6 +3,7 @@ package processor
 import (
 	"context"
 	"crypto/ecdsa"
+	"github.com/certusone/wormhole/node/pkg/notify/discord"
 	"time"
 
 	"github.com/certusone/wormhole/node/pkg/db"
@@ -97,6 +98,8 @@ type Processor struct {
 	ourAddr ethcommon.Address
 	// cleanup triggers periodic state cleanup
 	cleanup *time.Ticker
+
+	notifier *discord.DiscordNotifier
 }
 
 func NewProcessor(
@@ -116,6 +119,7 @@ func NewProcessor(
 	terraLCD string,
 	terraContract string,
 	attestationEvents *reporter.AttestationEventReporter,
+	notifier *discord.DiscordNotifier,
 ) *Processor {
 
 	return &Processor{
@@ -137,6 +141,8 @@ func NewProcessor(
 
 		attestationEvents: attestationEvents,
 
+		notifier: notifier,
+
 		logger:  supervisor.Logger(ctx),
 		state:   &aggregationState{vaaMap{}},
 		ourAddr: crypto.PubkeyToAddress(gk.PublicKey),

node/pkg/vaa/structs.go

@@ -268,6 +268,15 @@ func (v *VAA) MessageID() string {
 	return fmt.Sprintf("%d/%s/%d", v.EmitterChain, v.EmitterAddress, v.Sequence)
 }
 
+// HexDigest returns the hex-encoded digest.
+func (v *VAA) HexDigest() string {
+	b, err := v.SigningMsg()
+	if err != nil {
+		panic(err)
+	}
+	return hex.EncodeToString(b.Bytes())
+}
+
 func (v *VAA) serializeBody() ([]byte, error) {
 	buf := new(bytes.Buffer)
 	MustWrite(buf, binary.BigEndian, uint32(v.Timestamp.Unix()))