098cbc9c67
all of the packages and tests work with both grpc and drpc. we'll probably need to do some jenkins pipelines to run the tests with drpc as well. most of the changes are really due to a bit of cleanup of the pkg/transport.Client api into an rpc.Dialer in the spirit of a net.Dialer. now that we don't need observers, we can pass around stateless configuration to everything rather than stateful things that issue observations. it also adds a DialAddressID for the case where we don't have a pb.Node, but we do have an address and want to assert some ID. this happened pretty frequently, and now there's no more weird contortions creating custom tls options, etc. a lot of the other changes are being consistent/using the abstractions in the rpc package to do rpc style things like finding peer information, or checking status codes. Change-Id: Ief62875e21d80a21b3c56a5a37f45887679f9412
133 lines
3.4 KiB
Go
133 lines
3.4 KiB
Go
// Copyright (C) 2019 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package trust
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
|
|
"github.com/zeebo/errs"
|
|
monkit "gopkg.in/spacemonkeygo/monkit.v2"
|
|
|
|
"storj.io/storj/pkg/identity"
|
|
"storj.io/storj/pkg/rpc"
|
|
"storj.io/storj/pkg/signing"
|
|
"storj.io/storj/pkg/storj"
|
|
)
|
|
|
|
// Error is the default error class
|
|
var Error = errs.Class("trust:")
|
|
|
|
var mon = monkit.Package()
|
|
|
|
// Pool implements different peer verifications.
|
|
//
|
|
// architecture: Service
|
|
type Pool struct {
|
|
mu sync.RWMutex
|
|
dialer rpc.Dialer
|
|
|
|
trustedSatellites map[storj.NodeID]*satelliteInfoCache
|
|
}
|
|
|
|
// satelliteInfoCache caches identity information about a satellite
|
|
type satelliteInfoCache struct {
|
|
mu sync.Mutex
|
|
url storj.NodeURL
|
|
identity *identity.PeerIdentity
|
|
}
|
|
|
|
// NewPool creates a new trust pool of the specified list of trusted satellites.
|
|
func NewPool(dialer rpc.Dialer, trustedSatellites storj.NodeURLs) (*Pool, error) {
|
|
// TODO: preload all satellite peer identities
|
|
|
|
// parse the comma separated list of approved satellite IDs into an array of storj.NodeIDs
|
|
trusted := make(map[storj.NodeID]*satelliteInfoCache)
|
|
|
|
for _, node := range trustedSatellites {
|
|
trusted[node.ID] = &satelliteInfoCache{url: node}
|
|
}
|
|
|
|
return &Pool{
|
|
dialer: dialer,
|
|
trustedSatellites: trusted,
|
|
}, nil
|
|
}
|
|
|
|
// VerifySatelliteID checks whether id corresponds to a trusted satellite.
|
|
func (pool *Pool) VerifySatelliteID(ctx context.Context, id storj.NodeID) (err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
_, ok := pool.trustedSatellites[id]
|
|
if !ok {
|
|
return Error.New("satellite %q is untrusted", id)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// GetSignee gets the corresponding signee for verifying signatures.
|
|
// It ignores passed in ctx cancellation to avoid miscaching between concurrent requests.
|
|
func (pool *Pool) GetSignee(ctx context.Context, id storj.NodeID) (_ signing.Signee, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
// lookup peer identity with id
|
|
pool.mu.RLock()
|
|
info, ok := pool.trustedSatellites[id]
|
|
pool.mu.RUnlock()
|
|
|
|
if !ok {
|
|
return nil, Error.New("signee %q is untrusted", id)
|
|
}
|
|
|
|
info.mu.Lock()
|
|
defer info.mu.Unlock()
|
|
|
|
if info.identity == nil {
|
|
identity, err := pool.FetchPeerIdentity(ctx, info.url)
|
|
if err != nil {
|
|
return nil, Error.Wrap(err)
|
|
}
|
|
info.identity = identity
|
|
}
|
|
|
|
return signing.SigneeFromPeerIdentity(info.identity), nil
|
|
}
|
|
|
|
// FetchPeerIdentity dials the url and fetches the identity.
|
|
func (pool *Pool) FetchPeerIdentity(ctx context.Context, url storj.NodeURL) (_ *identity.PeerIdentity, err error) {
|
|
conn, err := pool.dialer.DialAddressID(ctx, url.Address, url.ID)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer func() { err = errs.Combine(err, conn.Close()) }()
|
|
|
|
return conn.PeerIdentity()
|
|
}
|
|
|
|
// GetSatellites returns a slice containing all trusted satellites
|
|
func (pool *Pool) GetSatellites(ctx context.Context) (satellites []storj.NodeID) {
|
|
defer mon.Task()(&ctx)(nil)
|
|
for sat := range pool.trustedSatellites {
|
|
satellites = append(satellites, sat)
|
|
}
|
|
return satellites
|
|
}
|
|
|
|
// GetAddress returns the address of a satellite in the trusted list
|
|
func (pool *Pool) GetAddress(ctx context.Context, id storj.NodeID) (_ string, err error) {
|
|
defer mon.Task()(&ctx)(&err)
|
|
|
|
pool.mu.RLock()
|
|
defer pool.mu.RUnlock()
|
|
|
|
info, ok := pool.trustedSatellites[id]
|
|
if !ok {
|
|
return "", Error.New("ID %v not found in trusted list", id)
|
|
}
|
|
return info.url.Address, nil
|
|
}
|