17132206455

Committed 21 Aug 2025 03:56PM UTC coverage: 54.685% (-2.6%) from 57.321%

Build # 17132206455

Build Type

Pull #10167

github

Committed by

web-flow

Commit Message

Merge 5dd2ed093 into 0c2f045f5

Pull Request Pull Request #10167: multi: bump Go to 1.24.6

Run Details

4 of 31 new or added lines in 10 files covered. (12.9%)

23854 existing lines in 284 files now uncovered.

108937 of 199210 relevant lines covered (54.68%)

22026.48 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/lnrpc/wtclientrpc/wtclient.go

package wtclientrpc

import (
        "bytes"
        "context"
        "encoding/binary"
        "errors"
        "fmt"
        "net"
        "sort"
        "strconv"

        "github.com/btcsuite/btcd/btcec/v2"
        "github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
        "github.com/lightningnetwork/lnd/lncfg"
        "github.com/lightningnetwork/lnd/lnrpc"
        "github.com/lightningnetwork/lnd/lnwire"
        "github.com/lightningnetwork/lnd/watchtower"
        "github.com/lightningnetwork/lnd/watchtower/blob"
        "github.com/lightningnetwork/lnd/watchtower/wtclient"
        "github.com/lightningnetwork/lnd/watchtower/wtdb"
        "google.golang.org/grpc"
        "gopkg.in/macaroon-bakery.v2/bakery"
)

const (
        // subServerName is the name of the sub rpc server. We'll use this name
        // to register ourselves, and we also require that the main
        // SubServerConfigDispatcher instance recognizes it as the name of our
        // RPC service.
        subServerName = "WatchtowerClientRPC"
)

var (
        // macPermissions maps RPC calls to the permissions they require.
        //
        // TODO(wilmer): create tower macaroon?
        macPermissions = map[string][]bakery.Op{
                "/wtclientrpc.WatchtowerClient/AddTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/RemoveTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/DeactivateTower": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/TerminateSession": {{
                        Entity: "offchain",
                        Action: "write",
                }},
                "/wtclientrpc.WatchtowerClient/ListTowers": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/GetTowerInfo": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/Stats": {{
                        Entity: "offchain",
                        Action: "read",
                }},
                "/wtclientrpc.WatchtowerClient/Policy": {{
                        Entity: "offchain",
                        Action: "read",
                }},
        }

        // ErrWtclientNotActive signals that RPC calls cannot be processed
        // because the watchtower client is not active.
        ErrWtclientNotActive = errors.New("watchtower client not active")
)

// ServerShell is a shell struct holding a reference to the actual sub-server.
// It is used to register the gRPC sub-server with the root server before we
// have the necessary dependencies to populate the actual sub-server.
type ServerShell struct {
        WatchtowerClientServer
}

// WatchtowerClient is the RPC server we'll use to interact with the backing
// active watchtower client.
//
// TODO(wilmer): better name?
type WatchtowerClient struct {
        // Required by the grpc-gateway/v2 library for forward compatibility.
        UnimplementedWatchtowerClientServer

        cfg Config
}

// A compile time check to ensure that WatchtowerClient fully implements the
// WatchtowerClientWatchtowerClient gRPC service.
var _ WatchtowerClientServer = (*WatchtowerClient)(nil)

// New returns a new instance of the wtclientrpc WatchtowerClient sub-server.
// We also return the set of permissions for the macaroons that we may create
// within this method. If the macaroons we need aren't found in the filepath,
// then we'll create them on start up. If we're unable to locate, or create the
// macaroons we need, then we'll return with an error.
func New(cfg *Config) (*WatchtowerClient, lnrpc.MacaroonPerms, error) {
        return &WatchtowerClient{cfg: *cfg}, macPermissions, nil
}

// Start launches any helper goroutines required for the WatchtowerClient to
// function.
//
// NOTE: This is part of the lnrpc.SubWatchtowerClient interface.
func (c *WatchtowerClient) Start() error {
        return nil
}

// Stop signals any active goroutines for a graceful closure.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (c *WatchtowerClient) Stop() error {
        return nil
}

// Name returns a unique string representation of the sub-server. This can be
// used to identify the sub-server and also de-duplicate them.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (c *WatchtowerClient) Name() string {
        return subServerName
}

// RegisterWithRootServer will be called by the root gRPC server to direct a sub
// RPC server to register itself with the main gRPC root server. Until this is
// called, each sub-server won't be able to have requests routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) RegisterWithRootServer(grpcServer *grpc.Server) error {
        // We make sure that we register it with the main gRPC server to ensure
        // all our methods are routed properly.
        RegisterWatchtowerClientServer(grpcServer, r)

        return nil
}

// RegisterWithRestServer will be called by the root REST mux to direct a sub
// RPC server to register itself with the main REST mux server. Until this is
// called, each sub-server won't be able to have requests routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) RegisterWithRestServer(ctx context.Context,
        mux *runtime.ServeMux, dest string, opts []grpc.DialOption) error {

        // We make sure that we register it with the main REST server to ensure
        // all our methods are routed properly.
        err := RegisterWatchtowerClientHandlerFromEndpoint(ctx, mux, dest, opts)
        if err != nil {
                return err
        }

        return nil
}

// CreateSubServer populates the subserver's dependencies using the passed
// SubServerConfigDispatcher. This method should fully initialize the
// sub-server instance, making it ready for action. It returns the macaroon
// permissions that the sub-server wishes to pass on to the root server for all
// methods routed towards it.
//
// NOTE: This is part of the lnrpc.GrpcHandler interface.
func (r *ServerShell) CreateSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (
        lnrpc.SubServer, lnrpc.MacaroonPerms, error) {

        subServer, macPermissions, err := createNewSubServer(configRegistry)
        if err != nil {
                return nil, nil, err
        }

        r.WatchtowerClientServer = subServer
        return subServer, macPermissions, nil
}

// isActive returns nil if the watchtower client is initialized so that we can
// process RPC requests.
func (c *WatchtowerClient) isActive() error {
        if c.cfg.Active {
                return nil
        }
        return ErrWtclientNotActive
}

// AddTower adds a new watchtower reachable at the given address and considers
// it for new sessions. If the watchtower already exists, then any new addresses
// included will be considered when dialing it for session negotiations and
// backups.
func (c *WatchtowerClient) AddTower(ctx context.Context,
        req *AddTowerRequest) (*AddTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }
        addr, err := lncfg.ParseAddressString(
                req.Address, strconv.Itoa(watchtower.DefaultPeerPort),
                c.cfg.Resolver,
        )
        if err != nil {
                return nil, fmt.Errorf("invalid address %v: %w", req.Address,
                        err)
        }

        towerAddr := &lnwire.NetAddress{
                IdentityKey: pubKey,
                Address:     addr,
        }

        if err := c.cfg.ClientMgr.AddTower(towerAddr); err != nil {
                return nil, err
        }

        return &AddTowerResponse{}, nil
}

// RemoveTower removes a watchtower from being considered for future session
// negotiations and from being used for any subsequent backups until it's added
// again. If an address is provided, then this RPC only serves as a way of
// removing the address from the watchtower instead.
func (c *WatchtowerClient) RemoveTower(ctx context.Context,
        req *RemoveTowerRequest) (*RemoveTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        var addr net.Addr
        if req.Address != "" {
                addr, err = lncfg.ParseAddressString(
                        req.Address, strconv.Itoa(watchtower.DefaultPeerPort),
                        c.cfg.Resolver,
                )
                if err != nil {
                        return nil, fmt.Errorf("unable to parse tower "+
                                "address %v: %v", req.Address, err)
                }
        }

        err = c.cfg.ClientMgr.RemoveTower(pubKey, addr)
        if err != nil {
                return nil, err
        }

        return &RemoveTowerResponse{}, nil
}

// DeactivateTower sets the given tower's status to inactive so that it is not
// considered for session negotiation. Its sessions will also not be used while
// the tower is inactive.
func (c *WatchtowerClient) DeactivateTower(_ context.Context,
        req *DeactivateTowerRequest) (*DeactivateTowerResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        err = c.cfg.ClientMgr.DeactivateTower(pubKey)
        if err != nil {
                return nil, err
        }

        return &DeactivateTowerResponse{
                Status: fmt.Sprintf("Successful deactivation of tower: %x",
                        req.Pubkey),
        }, nil
}

// TerminateSession terminates the given session and marks it as terminal so
// that it is never used again.
func (c *WatchtowerClient) TerminateSession(_ context.Context,
        req *TerminateSessionRequest) (*TerminateSessionResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.SessionId)
        if err != nil {
                return nil, err
        }

        sessionID := wtdb.NewSessionIDFromPubKey(pubKey)

        err = c.cfg.ClientMgr.TerminateSession(sessionID)
        if err != nil {
                return nil, err
        }

        return &TerminateSessionResponse{
                Status: fmt.Sprintf("Successful termination of session: %s",
                        sessionID),
        }, nil
}

// ListTowers returns the list of watchtowers registered with the client.
func (c *WatchtowerClient) ListTowers(ctx context.Context,
        req *ListTowersRequest) (*ListTowersResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(
                req.IncludeSessions, req.ExcludeExhaustedSessions,
        )

        towersPerBlobType, err := c.cfg.ClientMgr.RegisteredTowers(opts...)
        if err != nil {
                return nil, err
        }

        // Collect all the legacy client towers. If it has any of the same
        // towers that the anchors client has, then just add the session info
        // for the legacy client to the existing tower.
        rpcTowers := make(map[wtdb.TowerID]*Tower)
        for blobType, towers := range towersPerBlobType {
                policyType, err := blobTypeToPolicyType(blobType)
                if err != nil {
                        return nil, err
                }

                for _, tower := range towers {
                        rpcTower := marshallTower(
                                tower, policyType, req.IncludeSessions,
                                ackCounts, committedUpdateCounts,
                        )

                        t, ok := rpcTowers[tower.ID]
                        if !ok {
                                rpcTowers[tower.ID] = rpcTower
                                continue
                        }

                        t.SessionInfo = append(
                                t.SessionInfo, rpcTower.SessionInfo...,
                        )
                        t.Sessions = append(
                                t.Sessions, rpcTower.Sessions...,
                        )
                }
        }

        towers := make([]*Tower, 0, len(rpcTowers))
        for _, tower := range rpcTowers {
                towers = append(towers, tower)
        }

        return &ListTowersResponse{Towers: towers}, nil
}

// GetTowerInfo retrieves information for a registered watchtower.
func (c *WatchtowerClient) GetTowerInfo(ctx context.Context,
        req *GetTowerInfoRequest) (*Tower, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        pubKey, err := btcec.ParsePubKey(req.Pubkey)
        if err != nil {
                return nil, err
        }

        opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(
                req.IncludeSessions, req.ExcludeExhaustedSessions,
        )

        towersPerBlobType, err := c.cfg.ClientMgr.LookupTower(pubKey, opts...)
        if err != nil {
                return nil, err
        }

        var resTower *Tower
        for blobType, tower := range towersPerBlobType {
                policyType, err := blobTypeToPolicyType(blobType)
                if err != nil {
                        return nil, err
                }

                rpcTower := marshallTower(
                        tower, policyType, req.IncludeSessions,
                        ackCounts, committedUpdateCounts,
                )

                if resTower == nil {
                        resTower = rpcTower
                        continue
                }

                if !bytes.Equal(rpcTower.Pubkey, resTower.Pubkey) {
                        return nil, fmt.Errorf("tower clients returned " +
                                "inconsistent results for the given tower")
                }

                resTower.SessionInfo = append(
                        resTower.SessionInfo, rpcTower.SessionInfo...,
                )
                resTower.Sessions = append(
                        resTower.Sessions, rpcTower.Sessions...,
                )
        }

        return resTower, nil
}

// constructFunctionalOptions is a helper function that constructs a list of
// functional options to be used when fetching a tower from the DB. It also
// returns a map of acked-update counts and one for un-acked-update counts that
// will be populated once the db call has been made.
func constructFunctionalOptions(includeSessions,
        excludeExhaustedSessions bool) ([]wtdb.ClientSessionListOption,
        map[wtdb.SessionID]uint16, map[wtdb.SessionID]uint16) {

        var (
                opts                  []wtdb.ClientSessionListOption
                committedUpdateCounts = make(map[wtdb.SessionID]uint16)
                ackCounts             = make(map[wtdb.SessionID]uint16)
        )
        if !includeSessions {
                return opts, ackCounts, committedUpdateCounts
        }

        perNumRogueUpdates := func(s *wtdb.ClientSession, numUpdates uint16) {
                ackCounts[s.ID] += numUpdates
        }

        perNumAckedUpdates := func(s *wtdb.ClientSession, id lnwire.ChannelID,
                numUpdates uint16) {

                ackCounts[s.ID] += numUpdates
        }

        perCommittedUpdate := func(s *wtdb.ClientSession,
                u *wtdb.CommittedUpdate) {

                committedUpdateCounts[s.ID]++
        }

        opts = []wtdb.ClientSessionListOption{
                wtdb.WithPerNumAckedUpdates(perNumAckedUpdates),
                wtdb.WithPerCommittedUpdate(perCommittedUpdate),
                wtdb.WithPerRogueUpdateCount(perNumRogueUpdates),
        }

        if excludeExhaustedSessions {
                opts = append(opts, wtdb.WithPostEvalFilterFn(
                        wtclient.ExhaustedSessionFilter(),
                ))
        }

        return opts, ackCounts, committedUpdateCounts
}

// Stats returns the in-memory statistics of the client since startup.
func (c *WatchtowerClient) Stats(_ context.Context,
        _ *StatsRequest) (*StatsResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        stats := c.cfg.ClientMgr.Stats()

        return &StatsResponse{
                NumBackups:           uint32(stats.NumTasksAccepted),
                NumFailedBackups:     uint32(stats.NumTasksIneligible),
                NumPendingBackups:    uint32(stats.NumTasksPending),
                NumSessionsAcquired:  uint32(stats.NumSessionsAcquired),
                NumSessionsExhausted: uint32(stats.NumSessionsExhausted),
        }, nil
}

// Policy returns the active watchtower client policy configuration.
func (c *WatchtowerClient) Policy(ctx context.Context,
        req *PolicyRequest) (*PolicyResponse, error) {

        if err := c.isActive(); err != nil {
                return nil, err
        }

        blobType, err := policyTypeToBlobType(req.PolicyType)
        if err != nil {
                return nil, err
        }

        policy, err := c.cfg.ClientMgr.Policy(blobType)
        if err != nil {
                return nil, err
        }

        return &PolicyResponse{
                MaxUpdates:       uint32(policy.MaxUpdates),
                SweepSatPerVbyte: uint32(policy.SweepFeeRate.FeePerVByte()),

                // Deprecated field.
                SweepSatPerByte: uint32(policy.SweepFeeRate.FeePerVByte()),
        }, nil
}

// marshallTower converts a client registered watchtower into its corresponding
// RPC type.
func marshallTower(tower *wtclient.RegisteredTower, policyType PolicyType,
        includeSessions bool, ackCounts map[wtdb.SessionID]uint16,
        pendingCounts map[wtdb.SessionID]uint16) *Tower {

        rpcAddrs := make([]string, 0, len(tower.Addresses))
        for _, addr := range tower.Addresses {
                rpcAddrs = append(rpcAddrs, addr.String())
        }

        var rpcSessions []*TowerSession
        if includeSessions {
                // To ensure that the output order is deterministic for a given
                // set of sessions, we put the sessions into a slice and order
                // them based on session ID.
                sessions := make([]*wtdb.ClientSession, 0, len(tower.Sessions))
                for _, session := range tower.Sessions {
                        sessions = append(sessions, session)
                }

                sort.Slice(sessions, func(i, j int) bool {
                        id1 := sessions[i].ID
                        id2 := sessions[j].ID

                        return binary.BigEndian.Uint64(id1[:]) <
                                binary.BigEndian.Uint64(id2[:])
                })

                rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))
                for _, session := range sessions {
                        satPerVByte := session.Policy.SweepFeeRate.FeePerVByte()
                        rpcSessions = append(rpcSessions, &TowerSession{
                                Id:                session.ID[:],
                                NumBackups:        uint32(ackCounts[session.ID]),
                                NumPendingBackups: uint32(pendingCounts[session.ID]),
                                MaxBackups:        uint32(session.Policy.MaxUpdates),
                                SweepSatPerVbyte:  uint32(satPerVByte),

                                // Deprecated field.
                                SweepSatPerByte: uint32(satPerVByte),
                        })
                }
        }

        rpcTower := &Tower{
                Pubkey:    tower.IdentityKey.SerializeCompressed(),
                Addresses: rpcAddrs,
                SessionInfo: []*TowerSessionInfo{{
                        PolicyType:             policyType,
                        ActiveSessionCandidate: tower.ActiveSessionCandidate,
                        NumSessions:            uint32(len(tower.Sessions)),
                        Sessions:               rpcSessions,
                }},
                // The below fields are populated for backwards compatibility
                // but will be removed in a future commit when the proto fields
                // are removed.
                ActiveSessionCandidate: tower.ActiveSessionCandidate,
                NumSessions:            uint32(len(tower.Sessions)),
                Sessions:               rpcSessions,
        }

        return rpcTower
}

func blobTypeToPolicyType(t blob.Type) (PolicyType, error) {
        switch t {
        case blob.TypeAltruistTaprootCommit:
                return PolicyType_TAPROOT, nil

        case blob.TypeAltruistAnchorCommit:
                return PolicyType_ANCHOR, nil

        case blob.TypeAltruistCommit:
                return PolicyType_LEGACY, nil

        default:
                return 0, fmt.Errorf("unknown blob type: %s", t)
        }
}

func policyTypeToBlobType(t PolicyType) (blob.Type, error) {
        switch t {
        case PolicyType_TAPROOT:
                return blob.TypeAltruistTaprootCommit, nil

        case PolicyType_ANCHOR:
                return blob.TypeAltruistAnchorCommit, nil

        case PolicyType_LEGACY:
                return blob.TypeAltruistCommit, nil

        default:
                return 0, fmt.Errorf("unknown policy type: %s", t)
        }
}

1	package wtclientrpc
2
3	import (
4	"bytes"
5	"context"
6	"encoding/binary"
7	"errors"
8	"fmt"
9	"net"
10	"sort"
11	"strconv"
12
13	"github.com/btcsuite/btcd/btcec/v2"
14	"github.com/grpc-ecosystem/grpc-gateway/v2/runtime"
15	"github.com/lightningnetwork/lnd/lncfg"
16	"github.com/lightningnetwork/lnd/lnrpc"
17	"github.com/lightningnetwork/lnd/lnwire"
18	"github.com/lightningnetwork/lnd/watchtower"
19	"github.com/lightningnetwork/lnd/watchtower/blob"
20	"github.com/lightningnetwork/lnd/watchtower/wtclient"
21	"github.com/lightningnetwork/lnd/watchtower/wtdb"
22	"google.golang.org/grpc"
23	"gopkg.in/macaroon-bakery.v2/bakery"
24	)
25
26	const (
27	// subServerName is the name of the sub rpc server. We'll use this name
28	// to register ourselves, and we also require that the main
29	// SubServerConfigDispatcher instance recognizes it as the name of our
30	// RPC service.
31	subServerName = "WatchtowerClientRPC"
32	)
33
34	var (
35	// macPermissions maps RPC calls to the permissions they require.
36	//
37	// TODO(wilmer): create tower macaroon?
38	macPermissions = map[string][]bakery.Op{
39	"/wtclientrpc.WatchtowerClient/AddTower": {{
40	Entity: "offchain",
41	Action: "write",
42	}},
43	"/wtclientrpc.WatchtowerClient/RemoveTower": {{
44	Entity: "offchain",
45	Action: "write",
46	}},
47	"/wtclientrpc.WatchtowerClient/DeactivateTower": {{
48	Entity: "offchain",
49	Action: "write",
50	}},
51	"/wtclientrpc.WatchtowerClient/TerminateSession": {{
52	Entity: "offchain",
53	Action: "write",
54	}},
55	"/wtclientrpc.WatchtowerClient/ListTowers": {{
56	Entity: "offchain",
57	Action: "read",
58	}},
59	"/wtclientrpc.WatchtowerClient/GetTowerInfo": {{
60	Entity: "offchain",
61	Action: "read",
62	}},
63	"/wtclientrpc.WatchtowerClient/Stats": {{
64	Entity: "offchain",
65	Action: "read",
66	}},
67	"/wtclientrpc.WatchtowerClient/Policy": {{
68	Entity: "offchain",
69	Action: "read",
70	}},
71	}
72
73	// ErrWtclientNotActive signals that RPC calls cannot be processed
74	// because the watchtower client is not active.
75	ErrWtclientNotActive = errors.New("watchtower client not active")
76	)
77
78	// ServerShell is a shell struct holding a reference to the actual sub-server.
79	// It is used to register the gRPC sub-server with the root server before we
80	// have the necessary dependencies to populate the actual sub-server.
81	type ServerShell struct {
82	WatchtowerClientServer
83	}
84
85	// WatchtowerClient is the RPC server we'll use to interact with the backing
86	// active watchtower client.
87	//
88	// TODO(wilmer): better name?
89	type WatchtowerClient struct {
90	// Required by the grpc-gateway/v2 library for forward compatibility.
91	UnimplementedWatchtowerClientServer
92
93	cfg Config
94	}
95
96	// A compile time check to ensure that WatchtowerClient fully implements the
97	// WatchtowerClientWatchtowerClient gRPC service.
98	var _ WatchtowerClientServer = (*WatchtowerClient)(nil)
99
100	// New returns a new instance of the wtclientrpc WatchtowerClient sub-server.
101	// We also return the set of permissions for the macaroons that we may create
102	// within this method. If the macaroons we need aren't found in the filepath,
103	// then we'll create them on start up. If we're unable to locate, or create the
104	// macaroons we need, then we'll return with an error.
UNCOV 105	func New(cfg Config) (WatchtowerClient, lnrpc.MacaroonPerms, error) {	×
UNCOV 106	return &WatchtowerClient{cfg: *cfg}, macPermissions, nil	×
UNCOV 107	}	×
108
109	// Start launches any helper goroutines required for the WatchtowerClient to
110	// function.
111	//
112	// NOTE: This is part of the lnrpc.SubWatchtowerClient interface.
UNCOV 113	func (c *WatchtowerClient) Start() error {	×
UNCOV 114	return nil	×
UNCOV 115	}	×
116
117	// Stop signals any active goroutines for a graceful closure.
118	//
119	// NOTE: This is part of the lnrpc.SubServer interface.
UNCOV 120	func (c *WatchtowerClient) Stop() error {	×
UNCOV 121	return nil	×
UNCOV 122	}	×
123
124	// Name returns a unique string representation of the sub-server. This can be
125	// used to identify the sub-server and also de-duplicate them.
126	//
127	// NOTE: This is part of the lnrpc.SubServer interface.
UNCOV 128	func (c *WatchtowerClient) Name() string {	×
UNCOV 129	return subServerName	×
UNCOV 130	}	×
131
132	// RegisterWithRootServer will be called by the root gRPC server to direct a sub
133	// RPC server to register itself with the main gRPC root server. Until this is
134	// called, each sub-server won't be able to have requests routed towards it.
135	//
136	// NOTE: This is part of the lnrpc.GrpcHandler interface.
UNCOV 137	func (r ServerShell) RegisterWithRootServer(grpcServer grpc.Server) error {	×
UNCOV 138	// We make sure that we register it with the main gRPC server to ensure	×
UNCOV 139	// all our methods are routed properly.	×
UNCOV 140	RegisterWatchtowerClientServer(grpcServer, r)	×
UNCOV 141		×
UNCOV 142	return nil	×
UNCOV 143	}	×
144
145	// RegisterWithRestServer will be called by the root REST mux to direct a sub
146	// RPC server to register itself with the main REST mux server. Until this is
147	// called, each sub-server won't be able to have requests routed towards it.
148	//
149	// NOTE: This is part of the lnrpc.GrpcHandler interface.
150	func (r *ServerShell) RegisterWithRestServer(ctx context.Context,
UNCOV 151	mux *runtime.ServeMux, dest string, opts []grpc.DialOption) error {	×
UNCOV 152		×
UNCOV 153	// We make sure that we register it with the main REST server to ensure	×
UNCOV 154	// all our methods are routed properly.	×
UNCOV 155	err := RegisterWatchtowerClientHandlerFromEndpoint(ctx, mux, dest, opts)	×
UNCOV 156	if err != nil {	×
157	return err	×
158	}	×
159
UNCOV 160	return nil	×
161	}
162
163	// CreateSubServer populates the subserver's dependencies using the passed
164	// SubServerConfigDispatcher. This method should fully initialize the
165	// sub-server instance, making it ready for action. It returns the macaroon
166	// permissions that the sub-server wishes to pass on to the root server for all
167	// methods routed towards it.
168	//
169	// NOTE: This is part of the lnrpc.GrpcHandler interface.
170	func (r *ServerShell) CreateSubServer(configRegistry lnrpc.SubServerConfigDispatcher) (
UNCOV 171	lnrpc.SubServer, lnrpc.MacaroonPerms, error) {	×
UNCOV 172		×
UNCOV 173	subServer, macPermissions, err := createNewSubServer(configRegistry)	×
UNCOV 174	if err != nil {	×
175	return nil, nil, err	×
176	}	×
177
UNCOV 178	r.WatchtowerClientServer = subServer	×
UNCOV 179	return subServer, macPermissions, nil	×
180	}
181
182	// isActive returns nil if the watchtower client is initialized so that we can
183	// process RPC requests.
UNCOV 184	func (c *WatchtowerClient) isActive() error {	×
UNCOV 185	if c.cfg.Active {	×
UNCOV 186	return nil	×
UNCOV 187	}	×
188	return ErrWtclientNotActive	×
189	}
190
191	// AddTower adds a new watchtower reachable at the given address and considers
192	// it for new sessions. If the watchtower already exists, then any new addresses
193	// included will be considered when dialing it for session negotiations and
194	// backups.
195	func (c *WatchtowerClient) AddTower(ctx context.Context,
UNCOV 196	req AddTowerRequest) (AddTowerResponse, error) {	×
UNCOV 197		×
UNCOV 198	if err := c.isActive(); err != nil {	×
199	return nil, err	×
200	}	×
201
UNCOV 202	pubKey, err := btcec.ParsePubKey(req.Pubkey)	×
UNCOV 203	if err != nil {	×
204	return nil, err	×
205	}	×
UNCOV 206	addr, err := lncfg.ParseAddressString(	×
UNCOV 207	req.Address, strconv.Itoa(watchtower.DefaultPeerPort),	×
UNCOV 208	c.cfg.Resolver,	×
UNCOV 209	)	×
UNCOV 210	if err != nil {	×
211	return nil, fmt.Errorf("invalid address %v: %w", req.Address,	×
212	err)	×
213	}	×
214
UNCOV 215	towerAddr := &lnwire.NetAddress{	×
UNCOV 216	IdentityKey: pubKey,	×
UNCOV 217	Address: addr,	×
UNCOV 218	}	×
UNCOV 219		×
UNCOV 220	if err := c.cfg.ClientMgr.AddTower(towerAddr); err != nil {	×
221	return nil, err	×
222	}	×
223
UNCOV 224	return &AddTowerResponse{}, nil	×
225	}
226
227	// RemoveTower removes a watchtower from being considered for future session
228	// negotiations and from being used for any subsequent backups until it's added
229	// again. If an address is provided, then this RPC only serves as a way of
230	// removing the address from the watchtower instead.
231	func (c *WatchtowerClient) RemoveTower(ctx context.Context,
UNCOV 232	req RemoveTowerRequest) (RemoveTowerResponse, error) {	×
UNCOV 233		×
UNCOV 234	if err := c.isActive(); err != nil {	×
235	return nil, err	×
236	}	×
237
UNCOV 238	pubKey, err := btcec.ParsePubKey(req.Pubkey)	×
UNCOV 239	if err != nil {	×
240	return nil, err	×
241	}	×
242
UNCOV 243	var addr net.Addr	×
UNCOV 244	if req.Address != "" {	×
245	addr, err = lncfg.ParseAddressString(	×
246	req.Address, strconv.Itoa(watchtower.DefaultPeerPort),	×
247	c.cfg.Resolver,	×
248	)	×
249	if err != nil {	×
250	return nil, fmt.Errorf("unable to parse tower "+	×
251	"address %v: %v", req.Address, err)	×
252	}	×
253	}
254
UNCOV 255	err = c.cfg.ClientMgr.RemoveTower(pubKey, addr)	×
UNCOV 256	if err != nil {	×
257	return nil, err	×
258	}	×
259
UNCOV 260	return &RemoveTowerResponse{}, nil	×
261	}
262
263	// DeactivateTower sets the given tower's status to inactive so that it is not
264	// considered for session negotiation. Its sessions will also not be used while
265	// the tower is inactive.
266	func (c *WatchtowerClient) DeactivateTower(_ context.Context,
UNCOV 267	req DeactivateTowerRequest) (DeactivateTowerResponse, error) {	×
UNCOV 268		×
UNCOV 269	if err := c.isActive(); err != nil {	×
270	return nil, err	×
271	}	×
272
UNCOV 273	pubKey, err := btcec.ParsePubKey(req.Pubkey)	×
UNCOV 274	if err != nil {	×
275	return nil, err	×
276	}	×
277
UNCOV 278	err = c.cfg.ClientMgr.DeactivateTower(pubKey)	×
UNCOV 279	if err != nil {	×
280	return nil, err	×
281	}	×
282
UNCOV 283	return &DeactivateTowerResponse{	×
UNCOV 284	Status: fmt.Sprintf("Successful deactivation of tower: %x",	×
UNCOV 285	req.Pubkey),	×
UNCOV 286	}, nil	×
287	}
288
289	// TerminateSession terminates the given session and marks it as terminal so
290	// that it is never used again.
291	func (c *WatchtowerClient) TerminateSession(_ context.Context,
UNCOV 292	req TerminateSessionRequest) (TerminateSessionResponse, error) {	×
UNCOV 293		×
UNCOV 294	if err := c.isActive(); err != nil {	×
295	return nil, err	×
296	}	×
297
UNCOV 298	pubKey, err := btcec.ParsePubKey(req.SessionId)	×
UNCOV 299	if err != nil {	×
300	return nil, err	×
301	}	×
302
UNCOV 303	sessionID := wtdb.NewSessionIDFromPubKey(pubKey)	×
UNCOV 304		×
UNCOV 305	err = c.cfg.ClientMgr.TerminateSession(sessionID)	×
UNCOV 306	if err != nil {	×
307	return nil, err	×
308	}	×
309
UNCOV 310	return &TerminateSessionResponse{	×
UNCOV 311	Status: fmt.Sprintf("Successful termination of session: %s",	×
UNCOV 312	sessionID),	×
UNCOV 313	}, nil	×
314	}
315
316	// ListTowers returns the list of watchtowers registered with the client.
317	func (c *WatchtowerClient) ListTowers(ctx context.Context,
318	req ListTowersRequest) (ListTowersResponse, error) {	×
319		×
320	if err := c.isActive(); err != nil {	×
321	return nil, err	×
322	}	×
323
324	opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(	×
325	req.IncludeSessions, req.ExcludeExhaustedSessions,	×
326	)	×
327		×
328	towersPerBlobType, err := c.cfg.ClientMgr.RegisteredTowers(opts...)	×
329	if err != nil {	×
330	return nil, err	×
331	}	×
332
333	// Collect all the legacy client towers. If it has any of the same
334	// towers that the anchors client has, then just add the session info
335	// for the legacy client to the existing tower.
336	rpcTowers := make(map[wtdb.TowerID]*Tower)	×
337	for blobType, towers := range towersPerBlobType {	×
338	policyType, err := blobTypeToPolicyType(blobType)	×
339	if err != nil {	×
340	return nil, err	×
341	}	×
342
343	for _, tower := range towers {	×
344	rpcTower := marshallTower(	×
345	tower, policyType, req.IncludeSessions,	×
346	ackCounts, committedUpdateCounts,	×
347	)	×
348		×
349	t, ok := rpcTowers[tower.ID]	×
350	if !ok {	×
351	rpcTowers[tower.ID] = rpcTower	×
352	continue	×
353	}
354
355	t.SessionInfo = append(	×
356	t.SessionInfo, rpcTower.SessionInfo...,	×
357	)	×
358	t.Sessions = append(	×
359	t.Sessions, rpcTower.Sessions...,	×
360	)	×
361	}
362	}
363
364	towers := make([]*Tower, 0, len(rpcTowers))	×
365	for _, tower := range rpcTowers {	×
366	towers = append(towers, tower)	×
367	}	×
368
369	return &ListTowersResponse{Towers: towers}, nil	×
370	}
371
372	// GetTowerInfo retrieves information for a registered watchtower.
373	func (c *WatchtowerClient) GetTowerInfo(ctx context.Context,
UNCOV 374	req GetTowerInfoRequest) (Tower, error) {	×
UNCOV 375		×
UNCOV 376	if err := c.isActive(); err != nil {	×
377	return nil, err	×
378	}	×
379
UNCOV 380	pubKey, err := btcec.ParsePubKey(req.Pubkey)	×
UNCOV 381	if err != nil {	×
382	return nil, err	×
383	}	×
384
UNCOV 385	opts, ackCounts, committedUpdateCounts := constructFunctionalOptions(	×
UNCOV 386	req.IncludeSessions, req.ExcludeExhaustedSessions,	×
UNCOV 387	)	×
UNCOV 388		×
UNCOV 389	towersPerBlobType, err := c.cfg.ClientMgr.LookupTower(pubKey, opts...)	×
UNCOV 390	if err != nil {	×
391	return nil, err	×
392	}	×
393
UNCOV 394	var resTower *Tower	×
UNCOV 395	for blobType, tower := range towersPerBlobType {	×
UNCOV 396	policyType, err := blobTypeToPolicyType(blobType)	×
UNCOV 397	if err != nil {	×
398	return nil, err	×
399	}	×
400
UNCOV 401	rpcTower := marshallTower(	×
UNCOV 402	tower, policyType, req.IncludeSessions,	×
UNCOV 403	ackCounts, committedUpdateCounts,	×
UNCOV 404	)	×
UNCOV 405		×
UNCOV 406	if resTower == nil {	×
UNCOV 407	resTower = rpcTower	×
UNCOV 408	continue	×
409	}
410
UNCOV 411	if !bytes.Equal(rpcTower.Pubkey, resTower.Pubkey) {	×
412	return nil, fmt.Errorf("tower clients returned " +	×
413	"inconsistent results for the given tower")	×
414	}	×
415
UNCOV 416	resTower.SessionInfo = append(	×
UNCOV 417	resTower.SessionInfo, rpcTower.SessionInfo...,	×
UNCOV 418	)	×
UNCOV 419	resTower.Sessions = append(	×
UNCOV 420	resTower.Sessions, rpcTower.Sessions...,	×
UNCOV 421	)	×
422	}
423
UNCOV 424	return resTower, nil	×
425	}
426
427	// constructFunctionalOptions is a helper function that constructs a list of
428	// functional options to be used when fetching a tower from the DB. It also
429	// returns a map of acked-update counts and one for un-acked-update counts that
430	// will be populated once the db call has been made.
431	func constructFunctionalOptions(includeSessions,
432	excludeExhaustedSessions bool) ([]wtdb.ClientSessionListOption,
UNCOV 433	map[wtdb.SessionID]uint16, map[wtdb.SessionID]uint16) {	×
UNCOV 434		×
UNCOV 435	var (	×
UNCOV 436	opts []wtdb.ClientSessionListOption	×
UNCOV 437	committedUpdateCounts = make(map[wtdb.SessionID]uint16)	×
UNCOV 438	ackCounts = make(map[wtdb.SessionID]uint16)	×
UNCOV 439	)	×
UNCOV 440	if !includeSessions {	×
UNCOV 441	return opts, ackCounts, committedUpdateCounts	×
UNCOV 442	}	×
443
UNCOV 444	perNumRogueUpdates := func(s *wtdb.ClientSession, numUpdates uint16) {	×
UNCOV 445	ackCounts[s.ID] += numUpdates	×
UNCOV 446	}	×
447
UNCOV 448	perNumAckedUpdates := func(s *wtdb.ClientSession, id lnwire.ChannelID,	×
UNCOV 449	numUpdates uint16) {	×
UNCOV 450		×
UNCOV 451	ackCounts[s.ID] += numUpdates	×
UNCOV 452	}	×
453
UNCOV 454	perCommittedUpdate := func(s *wtdb.ClientSession,	×
UNCOV 455	u *wtdb.CommittedUpdate) {	×
UNCOV 456		×
UNCOV 457	committedUpdateCounts[s.ID]++	×
UNCOV 458	}	×
459
UNCOV 460	opts = []wtdb.ClientSessionListOption{	×
UNCOV 461	wtdb.WithPerNumAckedUpdates(perNumAckedUpdates),	×
UNCOV 462	wtdb.WithPerCommittedUpdate(perCommittedUpdate),	×
UNCOV 463	wtdb.WithPerRogueUpdateCount(perNumRogueUpdates),	×
UNCOV 464	}	×
UNCOV 465		×
UNCOV 466	if excludeExhaustedSessions {	×
467	opts = append(opts, wtdb.WithPostEvalFilterFn(	×
468	wtclient.ExhaustedSessionFilter(),	×
469	))	×
470	}	×
471
UNCOV 472	return opts, ackCounts, committedUpdateCounts	×
473	}
474
475	// Stats returns the in-memory statistics of the client since startup.
476	func (c *WatchtowerClient) Stats(_ context.Context,
UNCOV 477	_ StatsRequest) (StatsResponse, error) {	×
UNCOV 478		×
UNCOV 479	if err := c.isActive(); err != nil {	×
480	return nil, err	×
481	}	×
482
UNCOV 483	stats := c.cfg.ClientMgr.Stats()	×
UNCOV 484		×
UNCOV 485	return &StatsResponse{	×
UNCOV 486	NumBackups: uint32(stats.NumTasksAccepted),	×
UNCOV 487	NumFailedBackups: uint32(stats.NumTasksIneligible),	×
UNCOV 488	NumPendingBackups: uint32(stats.NumTasksPending),	×
UNCOV 489	NumSessionsAcquired: uint32(stats.NumSessionsAcquired),	×
UNCOV 490	NumSessionsExhausted: uint32(stats.NumSessionsExhausted),	×
UNCOV 491	}, nil	×
492	}
493
494	// Policy returns the active watchtower client policy configuration.
495	func (c *WatchtowerClient) Policy(ctx context.Context,
496	req PolicyRequest) (PolicyResponse, error) {	×
497		×
498	if err := c.isActive(); err != nil {	×
499	return nil, err	×
500	}	×
501
502	blobType, err := policyTypeToBlobType(req.PolicyType)	×
503	if err != nil {	×
504	return nil, err	×
505	}	×
506
507	policy, err := c.cfg.ClientMgr.Policy(blobType)	×
508	if err != nil {	×
509	return nil, err	×
510	}	×
511
512	return &PolicyResponse{	×
513	MaxUpdates: uint32(policy.MaxUpdates),	×
514	SweepSatPerVbyte: uint32(policy.SweepFeeRate.FeePerVByte()),	×
515		×
516	// Deprecated field.	×
517	SweepSatPerByte: uint32(policy.SweepFeeRate.FeePerVByte()),	×
518	}, nil	×
519	}
520
521	// marshallTower converts a client registered watchtower into its corresponding
522	// RPC type.
523	func marshallTower(tower *wtclient.RegisteredTower, policyType PolicyType,
524	includeSessions bool, ackCounts map[wtdb.SessionID]uint16,
UNCOV 525	pendingCounts map[wtdb.SessionID]uint16) *Tower {	×
UNCOV 526		×
UNCOV 527	rpcAddrs := make([]string, 0, len(tower.Addresses))	×
UNCOV 528	for _, addr := range tower.Addresses {	×
UNCOV 529	rpcAddrs = append(rpcAddrs, addr.String())	×
UNCOV 530	}	×
531
UNCOV 532	var rpcSessions []*TowerSession	×
UNCOV 533	if includeSessions {	×
UNCOV 534	// To ensure that the output order is deterministic for a given	×
UNCOV 535	// set of sessions, we put the sessions into a slice and order	×
UNCOV 536	// them based on session ID.	×
UNCOV 537	sessions := make([]*wtdb.ClientSession, 0, len(tower.Sessions))	×
UNCOV 538	for _, session := range tower.Sessions {	×
UNCOV 539	sessions = append(sessions, session)	×
UNCOV 540	}	×
541
UNCOV 542	sort.Slice(sessions, func(i, j int) bool {	×
UNCOV 543	id1 := sessions[i].ID	×
UNCOV 544	id2 := sessions[j].ID	×
UNCOV 545		×
UNCOV 546	return binary.BigEndian.Uint64(id1[:]) <	×
UNCOV 547	binary.BigEndian.Uint64(id2[:])	×
UNCOV 548	})	×
549
UNCOV 550	rpcSessions = make([]*TowerSession, 0, len(tower.Sessions))	×
UNCOV 551	for _, session := range sessions {	×
UNCOV 552	satPerVByte := session.Policy.SweepFeeRate.FeePerVByte()	×
UNCOV 553	rpcSessions = append(rpcSessions, &TowerSession{	×
UNCOV 554	Id: session.ID[:],	×
UNCOV 555	NumBackups: uint32(ackCounts[session.ID]),	×
UNCOV 556	NumPendingBackups: uint32(pendingCounts[session.ID]),	×
UNCOV 557	MaxBackups: uint32(session.Policy.MaxUpdates),	×
UNCOV 558	SweepSatPerVbyte: uint32(satPerVByte),	×
UNCOV 559		×
UNCOV 560	// Deprecated field.	×
UNCOV 561	SweepSatPerByte: uint32(satPerVByte),	×
UNCOV 562	})	×
UNCOV 563	}	×
564	}
565
UNCOV 566	rpcTower := &Tower{	×
UNCOV 567	Pubkey: tower.IdentityKey.SerializeCompressed(),	×
UNCOV 568	Addresses: rpcAddrs,	×
UNCOV 569	SessionInfo: []*TowerSessionInfo{{	×
UNCOV 570	PolicyType: policyType,	×
UNCOV 571	ActiveSessionCandidate: tower.ActiveSessionCandidate,	×
UNCOV 572	NumSessions: uint32(len(tower.Sessions)),	×
UNCOV 573	Sessions: rpcSessions,	×
UNCOV 574	}},	×
UNCOV 575	// The below fields are populated for backwards compatibility	×
UNCOV 576	// but will be removed in a future commit when the proto fields	×
UNCOV 577	// are removed.	×
UNCOV 578	ActiveSessionCandidate: tower.ActiveSessionCandidate,	×
UNCOV 579	NumSessions: uint32(len(tower.Sessions)),	×
UNCOV 580	Sessions: rpcSessions,	×
UNCOV 581	}	×
UNCOV 582		×
UNCOV 583	return rpcTower	×
584	}
585
UNCOV 586	func blobTypeToPolicyType(t blob.Type) (PolicyType, error) {	×
UNCOV 587	switch t {	×
UNCOV 588	case blob.TypeAltruistTaprootCommit:	×
UNCOV 589	return PolicyType_TAPROOT, nil	×
590
UNCOV 591	case blob.TypeAltruistAnchorCommit:	×
UNCOV 592	return PolicyType_ANCHOR, nil	×
593
UNCOV 594	case blob.TypeAltruistCommit:	×
UNCOV 595	return PolicyType_LEGACY, nil	×
596
597	default:	×
598	return 0, fmt.Errorf("unknown blob type: %s", t)	×
599	}
600	}
601
602	func policyTypeToBlobType(t PolicyType) (blob.Type, error) {	×
603	switch t {	×
604	case PolicyType_TAPROOT:	×
605	return blob.TypeAltruistTaprootCommit, nil	×
606
607	case PolicyType_ANCHOR:	×
608	return blob.TypeAltruistAnchorCommit, nil	×
609
610	case PolicyType_LEGACY:	×
611	return blob.TypeAltruistCommit, nil	×
612
613	default:	×
614	return 0, fmt.Errorf("unknown policy type: %s", t)	×
615	}
616	}

lightningnetwork / lnd / 17132206455

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous