eth,server,cmd: Add LIP-118 reward caller support

CHANGELOG_PENDING.md

@@ -12,6 +12,8 @@
 
 #### Orchestrator
 
+- [#4011](https://github.com/livepeer/go-livepeer/pull/4011) Add LIP-118 reward caller support, so a low-privilege wallet can call `reward()` on behalf of an orchestrator whose own wallet stays offline. Set one with the new "Set reward caller" action in `livepeer_cli`, then run the node with that wallet and `-ethOrchAddr <orchestrator>` (@rickstaa)
+
 #### Transcoder
 
 ### Bug Fixes 🐞

cmd/livepeer/starter/starter.go

@@ -1262,27 +1262,64 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 			glog.Infof("Redeemer started on %v", *cfg.HttpAddr)
 		}
 
+		// Reward is called for recipientAddr, which is the node's own account unless
+		// -ethOrchAddr points at a separately registered orchestrator. Authorization is
+		// only ever checked when reward is actually going to run: -ethOrchAddr is also
+		// used for ticket recipients by gateways, redeemers and orchestrators that never
+		// call reward, and those must not pay for an extra lookup or inherit a new
+		// failure mode.
 		var reward bool
-		if cfg.Reward == nil {
-			// If the node address is an on-chain registered address, start the reward service
-			t, err := n.Eth.GetTranscoder(n.Eth.Account().Address)
-			if err != nil {
-				glog.Error(err)
-				return
+		if cfg.Reward == nil || *cfg.Reward {
+			explicit := cfg.Reward != nil
+
+			// A node running on the orchestrator's own wallet calls reward directly.
+			// Otherwise it must be the reward caller the orchestrator authorized
+			// on-chain, or every reward transaction would revert. Check up front so the
+			// operator finds out at startup rather than a round later.
+			authorized := true
+			if recipientAddr != n.Eth.Account().Address {
+				rewardCaller, err := n.Eth.GetRewardCaller(recipientAddr)
+				if err != nil {
+					glog.Errorf("Could not look up reward caller for orchestrator %v err=%q", recipientAddr.Hex(), err)
+					return
+				}
+				authorized = rewardCaller == n.Eth.Account().Address
+				if !authorized {
+					msg := fmt.Sprintf(
+						"node account %v is not the reward caller for orchestrator %v (on-chain reward caller is %v); "+
+							"set it with livepeer_cli from the orchestrator's wallet",
+						n.Eth.Account().Address.Hex(), recipientAddr.Hex(), rewardCaller.Hex(),
+					)
+					if explicit {
+						exit("-reward was set but %s", msg)
+					}
+					glog.Warningf("Not starting reward service; %s", msg)
+				}
 			}
-			if t.Status == "Registered" {
-				reward = true
-			} else {
+
+			switch {
+			case !authorized:
 				reward = false
+			case explicit:
+				// Deliberately not gated on registration status. The orchestrator may
+				// still be registering, possibly via livepeer_cli against this very
+				// node, and the reward service is a no-op until it is active.
+				reward = true
+			default:
+				// Auto-enable only when there is a registered orchestrator to reward.
+				t, err := n.Eth.GetTranscoder(recipientAddr)
+				if err != nil {
+					glog.Error(err)
+					return
+				}
+				reward = t.Status == "Registered"
 			}
-		} else {
-			reward = *cfg.Reward
 		}
 
 		if reward {
 			// Start reward service
 			// The node will only call reward if it is active in the current round
-			rs := eth.NewRewardService(n.Eth, timeWatcher)
+			rs := eth.NewRewardService(n.Eth, timeWatcher, recipientAddr)
 			go func() {
 				if err := rs.Start(ctx); err != nil {
 					serviceErr <- err
@@ -1792,7 +1829,12 @@ func StartLivepeer(ctx context.Context, cfg LivepeerConfig) {
 	} else if n.NodeType == core.OrchestratorNode {
 		*cfg.CliAddr = defaultAddr(*cfg.CliAddr, "127.0.0.1", OrchestratorCliPort)
 
-		suri, err := getServiceURI(n, *cfg.ServiceAddr)
+		var ethOrchAddr ethcommon.Address
+		if *cfg.EthOrchAddr != "" {
+			ethOrchAddr = ethcommon.HexToAddress(*cfg.EthOrchAddr)
+		}
+
+		suri, err := getServiceURI(n, *cfg.ServiceAddr, ethOrchAddr)
 		if err != nil {
 			glog.Exit("Error getting service URI: ", err)
 		}
@@ -2261,7 +2303,9 @@ func isLocalURL(u string) (bool, error) {
 // Else: get on-chain sURI
 // If on-chain sURI mismatches inferred address: print warning
 // Return on-chain sURI
-func getServiceURI(n *core.LivepeerNode, serviceAddr string) (*url.URL, error) {
+// ethOrchAddr is the on-chain registered orchestrator whose service URI should be looked
+// up. It is the zero address when the node's own account is the orchestrator.
+func getServiceURI(n *core.LivepeerNode, serviceAddr string, ethOrchAddr ethcommon.Address) (*url.URL, error) {
 	// Passed in via CLI
 	if serviceAddr != "" {
 		if serviceAddr == "none" {
@@ -2298,8 +2342,14 @@ func getServiceURI(n *core.LivepeerNode, serviceAddr string) (*url.URL, error) {
 		return inferredUri, err
 	}
 
-	// On-chain lookup and matching with inferred public address
-	addr, err = n.Eth.GetServiceURI(n.Eth.Account().Address)
+	// On-chain lookup and matching with inferred public address.
+	// The service URI is registered against the orchestrator, so a node running on a
+	// reward caller wallet must not look it up under its own account.
+	uriAddr := n.Eth.Account().Address
+	if ethOrchAddr != (ethcommon.Address{}) {
+		uriAddr = ethOrchAddr
+	}
+	addr, err = n.Eth.GetServiceURI(uriAddr)
 	if err != nil {
 		glog.Errorf("Could not get service URI; orchestrator may be unreachable err=%q", err)
 		return nil, err

cmd/livepeer/starter/starter_test.go

@@ -91,23 +91,23 @@ func TestIsLocalURL(t *testing.T) {
 }
 
 func TestGetServiceURIServiceAddrScheme(t *testing.T) {
-	uri, err := getServiceURI(nil, "127.0.0.1:8935")
+	uri, err := getServiceURI(nil, "127.0.0.1:8935", ethcommon.Address{})
 	require.NoError(t, err)
 	require.Equal(t, "https://127.0.0.1:8935", uri.String())
 
-	uri, err = getServiceURI(nil, "http://127.0.0.1:8935")
+	uri, err = getServiceURI(nil, "http://127.0.0.1:8935", ethcommon.Address{})
 	require.NoError(t, err)
 	require.Equal(t, "http://127.0.0.1:8935", uri.String())
 
-	uri, err = getServiceURI(nil, "https://orch.example.com:443")
+	uri, err = getServiceURI(nil, "https://orch.example.com:443", ethcommon.Address{})
 	require.NoError(t, err)
 	require.Equal(t, "https://orch.example.com:443", uri.String())
 
-	uri, err = getServiceURI(nil, "gopher://orch.example.com:443")
+	uri, err = getServiceURI(nil, "gopher://orch.example.com:443", ethcommon.Address{})
 	require.NoError(t, err)
 	require.Equal(t, "https://gopher://orch.example.com:443", uri.String())
 
-	uri, err = getServiceURI(nil, "none")
+	uri, err = getServiceURI(nil, "none", ethcommon.Address{})
 	require.NoError(t, err)
 	require.Equal(t, "", uri.String())
 }

cmd/livepeer_cli/livepeer_cli.go

@@ -97,6 +97,7 @@ func (w *wizard) initializeOptions() []wizardOpt {
 		{desc: "Invoke \"reward\"", invoke: w.callReward, orchestrator: true},
 		{desc: "Invoke multi-step \"become an orchestrator\"", invoke: w.activateOrchestrator, orchestrator: true},
 		{desc: "Set orchestrator config", invoke: w.setOrchestratorConfig, orchestrator: true},
+		{desc: "Set reward caller", invoke: w.setRewardCaller, orchestrator: true},
 		{desc: "Invoke \"deposit broadcasting funds\" (ETH)", invoke: w.deposit, notOrchestrator: true},
 		{desc: "Invoke \"unlock broadcasting funds\"", invoke: w.unlock, notOrchestrator: true},
 		{desc: "Invoke \"cancel unlock of broadcasting funds\"", invoke: w.cancelUnlock, notOrchestrator: true},

cmd/livepeer_cli/wizard_transcoder.go

@@ -228,6 +228,67 @@ func (w *wizard) callReward() {
 	httpGet(fmt.Sprintf("http://%v:%v/reward", w.host, w.httpPort))
 }
 
+// setRewardCaller authorizes another address to call reward on this orchestrator's behalf
+// (LIP-118), so the orchestrator's own wallet can be kept offline. The authorization is
+// keyed on msg.sender, so this must be run against a node holding the orchestrator's key.
+func (w *wizard) setRewardCaller() {
+	if w.offchain {
+		fmt.Println("Cannot set a reward caller in off-chain mode")
+		return
+	}
+
+	current := strings.TrimSpace(httpGet(fmt.Sprintf("http://%v:%v/rewardCaller", w.host, w.httpPort)))
+
+	var rewardCaller string
+	if current != "" && current != "0x0000000000000000000000000000000000000000" {
+		fmt.Printf("Current reward caller: %v\n", current)
+		fmt.Print("Unset the current reward caller? (y/n) - ")
+		if w.readStringYesOrNo() == "y" {
+			fmt.Printf("Unsetting the reward caller. %v will no longer be able to call reward.\n", current)
+		} else {
+			rewardCaller = w.promptRewardCaller()
+		}
+	} else {
+		fmt.Println("No reward caller is currently set")
+		rewardCaller = w.promptRewardCaller()
+	}
+
+	val := url.Values{"rewardCaller": {rewardCaller}}
+	result, ok := httpPostWithParams(fmt.Sprintf("http://%v:%v/setRewardCaller", w.host, w.httpPort), val)
+	if !ok {
+		fmt.Printf("Error setting reward caller: %s\n", result)
+		return
+	}
+
+	if rewardCaller == "" {
+		fmt.Println("\nTransaction sent. Once confirmed, no address can call reward on your behalf.")
+		return
+	}
+	fmt.Printf("\nTransaction sent. Once confirmed, run the node with %v's keystore and -ethOrchAddr %v to call reward from it.\n",
+		rewardCaller, w.getOrchestratorAddr())
+}
+
+func (w *wizard) promptRewardCaller() string {
+	fmt.Print("Enter the address that should be allowed to call reward - ")
+	return w.readStringAndValidate(func(in string) (string, error) {
+		if !ethcommon.IsHexAddress(in) {
+			return "", fmt.Errorf("invalid hex address address=%v", in)
+		}
+		if ethcommon.HexToAddress(in) == (ethcommon.Address{}) {
+			return "", fmt.Errorf("cannot set the zero address; answer y to the unset prompt instead")
+		}
+		return in, nil
+	})
+}
+
+func (w *wizard) getOrchestratorAddr() string {
+	t, _, err := w.getOrchestratorInfo()
+	if err != nil || t == nil {
+		return "<your orchestrator address>"
+	}
+	return t.Address.Hex()
+}
+
 func (w *wizard) vote() {
 	if w.offchain {
 		glog.Error("Can not vote in 'offchain' mode")

eth/client.go

@@ -70,6 +70,12 @@ type LivepeerEthClient interface {
 	// Staking
 	Transcoder(blockRewardCut, feeShare *big.Int) (*types.Transaction, error)
 	Reward() (*types.Transaction, error)
+	// Reward caller (LIP-118). RewardForTranscoder is called by an address the
+	// transcoder authorized via SetRewardCaller, so the transcoder's own wallet
+	// can stay offline. Passing the zero address to SetRewardCaller unsets it.
+	SetRewardCaller(rewardCaller ethcommon.Address) (*types.Transaction, error)
+	GetRewardCaller(transcoder ethcommon.Address) (ethcommon.Address, error)
+	RewardForTranscoder(transcoder ethcommon.Address) (*types.Transaction, error)
 	Bond(amount *big.Int, toAddr ethcommon.Address) (*types.Transaction, error)
 	Rebond(unbondingLockID *big.Int) (*types.Transaction, error)
 	RebondFromUnbonded(toAddr ethcommon.Address, unbondingLockID *big.Int) (*types.Transaction, error)
@@ -1021,7 +1027,23 @@ func (c *client) ProposalVoteWithReason(proposalId *big.Int, support uint8, reas
 }
 
 func (c *client) Reward() (*types.Transaction, error) {
-	addr := c.accountManager.Account().Address
+	return c.rewardFor(c.accountManager.Account().Address, false)
+}
+
+// RewardForTranscoder calls reward on behalf of `transcoder`. The node's account must
+// have been authorized by `transcoder` via SetRewardCaller, otherwise the transaction
+// reverts with "caller must be a reward caller set by the transcoder".
+func (c *client) RewardForTranscoder(transcoder ethcommon.Address) (*types.Transaction, error) {
+	return c.rewardFor(transcoder, true)
+}
+
+// rewardFor computes the transcoder pool position hints for `transcoder` and submits the
+// reward transaction. The hint math is identical for both paths and must always be based
+// on the transcoder's stake, never the caller's. When `delegated` is set the transaction
+// is submitted through rewardForTranscoderWithHint, which resolves the authorized caller
+// on-chain; otherwise the caller is the transcoder itself.
+func (c *client) rewardFor(transcoder ethcommon.Address, delegated bool) (*types.Transaction, error) {
+	addr := transcoder
 
 	tr, err := c.GetTranscoder(addr)
 	if err != nil {
@@ -1065,9 +1087,26 @@ func (c *client) Reward() (*types.Transaction, error) {
 
 	hints := simulateTranscoderPoolUpdate(addr, reward.Add(reward, tr.DelegatedStake), transcoders, len(transcoders) == int(maxSize.Int64()))
 
+	if delegated {
+		return c.bondingManager.RewardForTranscoderWithHint(c.transactOpts(), addr, hints.PosPrev, hints.PosNext)
+	}
+
 	return c.bondingManager.RewardWithHint(c.transactOpts(), hints.PosPrev, hints.PosNext)
 }
 
+// SetRewardCaller authorizes `rewardCaller` to call reward on behalf of the node's
+// account. Passing the zero address unsets any existing authorization. This must be sent
+// from the transcoder's own wallet; the contract keys the mapping on msg.sender.
+func (c *client) SetRewardCaller(rewardCaller ethcommon.Address) (*types.Transaction, error) {
+	return c.bondingManager.SetRewardCaller(c.transactOpts(), rewardCaller)
+}
+
+// GetRewardCaller returns the address `transcoder` authorized to call reward on its
+// behalf, or the zero address if none is set.
+func (c *client) GetRewardCaller(transcoder ethcommon.Address) (ethcommon.Address, error) {
+	return c.bondingManager.TranscoderToRewardCaller(c.callOpts(), transcoder)
+}
+
 func (c *client) WithdrawFees(addr ethcommon.Address, amount *big.Int) (*types.Transaction, error) {
 	return c.bondingManager.WithdrawFees(c.transactOpts(), addr, amount)
 }