11954082915

Committed 21 Nov 2024 01:20PM UTC coverage: 59.327% (+0.6%) from 58.776%

Build # 11954082915

Build Type

Pull #8754

github

Committed by

ViktorTigerstrom

Commit Message

itest: wrap deriveCustomScopeAccounts at 80 chars

This commit fixes that word wrapping for the deriveCustomScopeAccounts
function docs, and ensures that it wraps at 80 characters or less.

Pull Request Pull Request #8754: Add `Outbound` Remote Signer implementation

Run Details

1940 of 2984 new or added lines in 44 files covered. (65.01%)

226 existing lines in 37 files now uncovered.

135234 of 227947 relevant lines covered (59.33%)

19316.75 hits per line

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

63.16

/lnrpc/wtclientrpc/wtclient.go

package wtclientrpc

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

        "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 {
        injected int32 // To be used atomically.

        // 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() (*WatchtowerClient, lnrpc.MacaroonPerms, error) {
        return &WatchtowerClient{cfg: Config{}}, macPermissions, 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
}

// InjectDependencies populates the sub-server's dependencies. If the
// finalizeDependencies boolean is true, then the sub-server will finalize its
// dependencies and return an error if any required dependencies are missing.
//
// NOTE: This is part of the lnrpc.SubServer interface.
func (c *WatchtowerClient) InjectDependencies(
        configRegistry lnrpc.SubServerConfigDispatcher,
        finalizeDependencies bool) error {

        if finalizeDependencies && atomic.AddInt32(&c.injected, 1) != 1 {
                return lnrpc.ErrDependenciesFinalized
        }

        cfg, err := getConfig(configRegistry, finalizeDependencies)
        if err != nil {
                return err
        }

        c.cfg = *cfg

        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 creates an instance of the sub-server, and 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() (
        lnrpc.SubServer, lnrpc.MacaroonPerms, error) {

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

lightningnetwork / lnd / 11954082915

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