ceremonyclient/go-libp2p/p2p/protocol/circuitv2/relay/relay.go
2023-08-20 23:07:43 -05:00

688 lines
16 KiB
Go

package relay
import (
"context"
"errors"
"fmt"
"io"
"sync"
"sync/atomic"
"time"
"github.com/libp2p/go-libp2p/core/host"
"github.com/libp2p/go-libp2p/core/network"
"github.com/libp2p/go-libp2p/core/peer"
"github.com/libp2p/go-libp2p/core/record"
pbv2 "github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/pb"
"github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/proto"
"github.com/libp2p/go-libp2p/p2p/protocol/circuitv2/util"
logging "github.com/ipfs/go-log/v2"
pool "github.com/libp2p/go-buffer-pool"
asnutil "github.com/libp2p/go-libp2p-asn-util"
ma "github.com/multiformats/go-multiaddr"
manet "github.com/multiformats/go-multiaddr/net"
)
const (
ServiceName = "libp2p.relay/v2"
ReservationTagWeight = 10
StreamTimeout = time.Minute
ConnectTimeout = 30 * time.Second
HandshakeTimeout = time.Minute
relayHopTag = "relay-v2-hop"
relayHopTagValue = 2
maxMessageSize = 4096
)
var log = logging.Logger("relay")
// Relay is the (limited) relay service object.
type Relay struct {
ctx context.Context
cancel func()
host host.Host
rc Resources
acl ACLFilter
constraints *constraints
scope network.ResourceScopeSpan
notifiee network.Notifiee
mx sync.Mutex
rsvp map[peer.ID]time.Time
conns map[peer.ID]int
closed bool
selfAddr ma.Multiaddr
metricsTracer MetricsTracer
}
// New constructs a new limited relay that can provide relay services in the given host.
func New(h host.Host, opts ...Option) (*Relay, error) {
ctx, cancel := context.WithCancel(context.Background())
r := &Relay{
ctx: ctx,
cancel: cancel,
host: h,
rc: DefaultResources(),
acl: nil,
rsvp: make(map[peer.ID]time.Time),
conns: make(map[peer.ID]int),
}
for _, opt := range opts {
err := opt(r)
if err != nil {
return nil, fmt.Errorf("error applying relay option: %w", err)
}
}
// get a scope for memory reservations at service level
err := h.Network().ResourceManager().ViewService(ServiceName,
func(s network.ServiceScope) error {
var err error
r.scope, err = s.BeginSpan()
return err
})
if err != nil {
return nil, err
}
r.constraints = newConstraints(&r.rc)
r.selfAddr = ma.StringCast(fmt.Sprintf("/p2p/%s", h.ID()))
h.SetStreamHandler(proto.ProtoIDv2Hop, r.handleStream)
r.notifiee = &network.NotifyBundle{DisconnectedF: r.disconnected}
h.Network().Notify(r.notifiee)
if r.metricsTracer != nil {
r.metricsTracer.RelayStatus(true)
}
go r.background()
return r, nil
}
func (r *Relay) Close() error {
r.mx.Lock()
if !r.closed {
r.closed = true
r.mx.Unlock()
r.host.RemoveStreamHandler(proto.ProtoIDv2Hop)
r.host.Network().StopNotify(r.notifiee)
r.scope.Done()
r.cancel()
r.gc()
if r.metricsTracer != nil {
r.metricsTracer.RelayStatus(false)
}
return nil
}
r.mx.Unlock()
return nil
}
func (r *Relay) handleStream(s network.Stream) {
log.Infof("new relay stream from: %s", s.Conn().RemotePeer())
if err := s.Scope().SetService(ServiceName); err != nil {
log.Debugf("error attaching stream to relay service: %s", err)
s.Reset()
return
}
if err := s.Scope().ReserveMemory(maxMessageSize, network.ReservationPriorityAlways); err != nil {
log.Debugf("error reserving memory for stream: %s", err)
s.Reset()
return
}
defer s.Scope().ReleaseMemory(maxMessageSize)
rd := util.NewDelimitedReader(s, maxMessageSize)
defer rd.Close()
s.SetReadDeadline(time.Now().Add(StreamTimeout))
var msg pbv2.HopMessage
err := rd.ReadMsg(&msg)
if err != nil {
r.handleError(s, pbv2.Status_MALFORMED_MESSAGE)
return
}
// reset stream deadline as message has been read
s.SetReadDeadline(time.Time{})
switch msg.GetType() {
case pbv2.HopMessage_RESERVE:
status := r.handleReserve(s)
if r.metricsTracer != nil {
r.metricsTracer.ReservationRequestHandled(status)
}
case pbv2.HopMessage_CONNECT:
status := r.handleConnect(s, &msg)
if r.metricsTracer != nil {
r.metricsTracer.ConnectionRequestHandled(status)
}
default:
r.handleError(s, pbv2.Status_MALFORMED_MESSAGE)
}
}
func (r *Relay) handleReserve(s network.Stream) pbv2.Status {
defer s.Close()
p := s.Conn().RemotePeer()
a := s.Conn().RemoteMultiaddr()
if isRelayAddr(a) {
log.Debugf("refusing relay reservation for %s; reservation attempt over relay connection")
r.handleError(s, pbv2.Status_PERMISSION_DENIED)
return pbv2.Status_PERMISSION_DENIED
}
if r.acl != nil && !r.acl.AllowReserve(p, a) {
log.Debugf("refusing relay reservation for %s; permission denied", p)
r.handleError(s, pbv2.Status_PERMISSION_DENIED)
return pbv2.Status_PERMISSION_DENIED
}
r.mx.Lock()
// Check if relay is still active. Otherwise ConnManager.UnTagPeer will not be called if this block runs after
// Close() call
if r.closed {
r.mx.Unlock()
log.Debugf("refusing relay reservation for %s; relay closed", p)
r.handleError(s, pbv2.Status_PERMISSION_DENIED)
return pbv2.Status_PERMISSION_DENIED
}
now := time.Now()
_, exists := r.rsvp[p]
if !exists {
if err := r.constraints.AddReservation(p, a); err != nil {
r.mx.Unlock()
log.Debugf("refusing relay reservation for %s; IP constraint violation: %s", p, err)
r.handleError(s, pbv2.Status_RESERVATION_REFUSED)
return pbv2.Status_RESERVATION_REFUSED
}
}
expire := now.Add(r.rc.ReservationTTL)
r.rsvp[p] = expire
r.host.ConnManager().TagPeer(p, "relay-reservation", ReservationTagWeight)
r.mx.Unlock()
if r.metricsTracer != nil {
r.metricsTracer.ReservationAllowed(exists)
}
log.Debugf("reserving relay slot for %s", p)
// Delivery of the reservation might fail for a number of reasons.
// For example, the stream might be reset or the connection might be closed before the reservation is received.
// In that case, the reservation will just be garbage collected later.
if err := r.writeResponse(s, pbv2.Status_OK, r.makeReservationMsg(p, expire), r.makeLimitMsg(p)); err != nil {
log.Debugf("error writing reservation response; retracting reservation for %s", p)
s.Reset()
return pbv2.Status_CONNECTION_FAILED
}
return pbv2.Status_OK
}
func (r *Relay) handleConnect(s network.Stream, msg *pbv2.HopMessage) pbv2.Status {
src := s.Conn().RemotePeer()
a := s.Conn().RemoteMultiaddr()
span, err := r.scope.BeginSpan()
if err != nil {
log.Debugf("failed to begin relay transaction: %s", err)
r.handleError(s, pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
fail := func(status pbv2.Status) {
span.Done()
r.handleError(s, status)
}
// reserve buffers for the relay
if err := span.ReserveMemory(2*r.rc.BufferSize, network.ReservationPriorityHigh); err != nil {
log.Debugf("error reserving memory for relay: %s", err)
fail(pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
if isRelayAddr(a) {
log.Debugf("refusing connection from %s; connection attempt over relay connection")
fail(pbv2.Status_PERMISSION_DENIED)
return pbv2.Status_PERMISSION_DENIED
}
dest, err := util.PeerToPeerInfoV2(msg.GetPeer())
if err != nil {
fail(pbv2.Status_MALFORMED_MESSAGE)
return pbv2.Status_MALFORMED_MESSAGE
}
if r.acl != nil && !r.acl.AllowConnect(src, s.Conn().RemoteMultiaddr(), dest.ID) {
log.Debugf("refusing connection from %s to %s; permission denied", src, dest.ID)
fail(pbv2.Status_PERMISSION_DENIED)
return pbv2.Status_PERMISSION_DENIED
}
r.mx.Lock()
_, rsvp := r.rsvp[dest.ID]
if !rsvp {
r.mx.Unlock()
log.Debugf("refusing connection from %s to %s; no reservation", src, dest.ID)
fail(pbv2.Status_NO_RESERVATION)
return pbv2.Status_NO_RESERVATION
}
srcConns := r.conns[src]
if srcConns >= r.rc.MaxCircuits {
r.mx.Unlock()
log.Debugf("refusing connection from %s to %s; too many connections from %s", src, dest.ID, src)
fail(pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
destConns := r.conns[dest.ID]
if destConns >= r.rc.MaxCircuits {
r.mx.Unlock()
log.Debugf("refusing connection from %s to %s; too many connecitons to %s", src, dest.ID, dest.ID)
fail(pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
r.addConn(src)
r.addConn(dest.ID)
r.mx.Unlock()
if r.metricsTracer != nil {
r.metricsTracer.ConnectionOpened()
}
connStTime := time.Now()
cleanup := func() {
span.Done()
r.mx.Lock()
r.rmConn(src)
r.rmConn(dest.ID)
r.mx.Unlock()
if r.metricsTracer != nil {
r.metricsTracer.ConnectionClosed(time.Since(connStTime))
}
}
ctx, cancel := context.WithTimeout(r.ctx, ConnectTimeout)
defer cancel()
ctx = network.WithNoDial(ctx, "relay connect")
bs, err := r.host.NewStream(ctx, dest.ID, proto.ProtoIDv2Stop)
if err != nil {
log.Debugf("error opening relay stream to %s: %s", dest.ID, err)
cleanup()
r.handleError(s, pbv2.Status_CONNECTION_FAILED)
return pbv2.Status_CONNECTION_FAILED
}
fail = func(status pbv2.Status) {
bs.Reset()
cleanup()
r.handleError(s, status)
}
if err := bs.Scope().SetService(ServiceName); err != nil {
log.Debugf("error attaching stream to relay service: %s", err)
fail(pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
// handshake
if err := bs.Scope().ReserveMemory(maxMessageSize, network.ReservationPriorityAlways); err != nil {
log.Debugf("error reserving memory for stream: %s", err)
fail(pbv2.Status_RESOURCE_LIMIT_EXCEEDED)
return pbv2.Status_RESOURCE_LIMIT_EXCEEDED
}
defer bs.Scope().ReleaseMemory(maxMessageSize)
rd := util.NewDelimitedReader(bs, maxMessageSize)
wr := util.NewDelimitedWriter(bs)
defer rd.Close()
var stopmsg pbv2.StopMessage
stopmsg.Type = pbv2.StopMessage_CONNECT.Enum()
stopmsg.Peer = util.PeerInfoToPeerV2(peer.AddrInfo{ID: src})
stopmsg.Limit = r.makeLimitMsg(dest.ID)
bs.SetDeadline(time.Now().Add(HandshakeTimeout))
err = wr.WriteMsg(&stopmsg)
if err != nil {
log.Debugf("error writing stop handshake")
fail(pbv2.Status_CONNECTION_FAILED)
return pbv2.Status_CONNECTION_FAILED
}
stopmsg.Reset()
err = rd.ReadMsg(&stopmsg)
if err != nil {
log.Debugf("error reading stop response: %s", err.Error())
fail(pbv2.Status_CONNECTION_FAILED)
return pbv2.Status_CONNECTION_FAILED
}
if t := stopmsg.GetType(); t != pbv2.StopMessage_STATUS {
log.Debugf("unexpected stop response; not a status message (%d)", t)
fail(pbv2.Status_CONNECTION_FAILED)
return pbv2.Status_CONNECTION_FAILED
}
if status := stopmsg.GetStatus(); status != pbv2.Status_OK {
log.Debugf("relay stop failure: %d", status)
fail(pbv2.Status_CONNECTION_FAILED)
return pbv2.Status_CONNECTION_FAILED
}
var response pbv2.HopMessage
response.Type = pbv2.HopMessage_STATUS.Enum()
response.Status = pbv2.Status_OK.Enum()
response.Limit = r.makeLimitMsg(dest.ID)
wr = util.NewDelimitedWriter(s)
err = wr.WriteMsg(&response)
if err != nil {
log.Debugf("error writing relay response: %s", err)
bs.Reset()
s.Reset()
cleanup()
return pbv2.Status_CONNECTION_FAILED
}
// reset deadline
bs.SetDeadline(time.Time{})
log.Infof("relaying connection from %s to %s", src, dest.ID)
var goroutines atomic.Int32
goroutines.Store(2)
done := func() {
if goroutines.Add(-1) == 0 {
s.Close()
bs.Close()
cleanup()
}
}
if r.rc.Limit != nil {
deadline := time.Now().Add(r.rc.Limit.Duration)
s.SetDeadline(deadline)
bs.SetDeadline(deadline)
go r.relayLimited(s, bs, src, dest.ID, r.rc.Limit.Data, done)
go r.relayLimited(bs, s, dest.ID, src, r.rc.Limit.Data, done)
} else {
go r.relayUnlimited(s, bs, src, dest.ID, done)
go r.relayUnlimited(bs, s, dest.ID, src, done)
}
return pbv2.Status_OK
}
func (r *Relay) addConn(p peer.ID) {
conns := r.conns[p]
conns++
r.conns[p] = conns
if conns == 1 {
r.host.ConnManager().TagPeer(p, relayHopTag, relayHopTagValue)
}
}
func (r *Relay) rmConn(p peer.ID) {
conns := r.conns[p]
conns--
if conns > 0 {
r.conns[p] = conns
} else {
delete(r.conns, p)
r.host.ConnManager().UntagPeer(p, relayHopTag)
}
}
func (r *Relay) relayLimited(src, dest network.Stream, srcID, destID peer.ID, limit int64, done func()) {
defer done()
buf := pool.Get(r.rc.BufferSize)
defer pool.Put(buf)
limitedSrc := io.LimitReader(src, limit)
count, err := r.copyWithBuffer(dest, limitedSrc, buf)
if err != nil {
log.Debugf("relay copy error: %s", err)
// Reset both.
src.Reset()
dest.Reset()
} else {
// propagate the close
dest.CloseWrite()
if count == limit {
// we've reached the limit, discard further input
src.CloseRead()
}
}
log.Debugf("relayed %d bytes from %s to %s", count, srcID, destID)
}
func (r *Relay) relayUnlimited(src, dest network.Stream, srcID, destID peer.ID, done func()) {
defer done()
buf := pool.Get(r.rc.BufferSize)
defer pool.Put(buf)
count, err := r.copyWithBuffer(dest, src, buf)
if err != nil {
log.Debugf("relay copy error: %s", err)
// Reset both.
src.Reset()
dest.Reset()
} else {
// propagate the close
dest.CloseWrite()
}
log.Debugf("relayed %d bytes from %s to %s", count, srcID, destID)
}
// errInvalidWrite means that a write returned an impossible count.
// copied from io.errInvalidWrite
var errInvalidWrite = errors.New("invalid write result")
// copyWithBuffer copies from src to dst using the provided buf until either EOF is reached
// on src or an error occurs. It reports the number of bytes transferred to metricsTracer.
// The implementation is a modified form of io.CopyBuffer to support metrics tracking.
func (r *Relay) copyWithBuffer(dst io.Writer, src io.Reader, buf []byte) (written int64, err error) {
for {
nr, er := src.Read(buf)
if nr > 0 {
nw, ew := dst.Write(buf[0:nr])
if nw < 0 || nr < nw {
nw = 0
if ew == nil {
ew = errInvalidWrite
}
}
written += int64(nw)
if ew != nil {
err = ew
break
}
if nr != nw {
err = io.ErrShortWrite
break
}
if r.metricsTracer != nil {
r.metricsTracer.BytesTransferred(nw)
}
}
if er != nil {
if er != io.EOF {
err = er
}
break
}
}
return written, err
}
func (r *Relay) handleError(s network.Stream, status pbv2.Status) {
log.Debugf("relay error: %s (%d)", pbv2.Status_name[int32(status)], status)
err := r.writeResponse(s, status, nil, nil)
if err != nil {
s.Reset()
log.Debugf("error writing relay response: %s", err.Error())
} else {
s.Close()
}
}
func (r *Relay) writeResponse(s network.Stream, status pbv2.Status, rsvp *pbv2.Reservation, limit *pbv2.Limit) error {
wr := util.NewDelimitedWriter(s)
var msg pbv2.HopMessage
msg.Type = pbv2.HopMessage_STATUS.Enum()
msg.Status = status.Enum()
msg.Reservation = rsvp
msg.Limit = limit
return wr.WriteMsg(&msg)
}
func (r *Relay) makeReservationMsg(p peer.ID, expire time.Time) *pbv2.Reservation {
expireUnix := uint64(expire.Unix())
var addrBytes [][]byte
for _, addr := range r.host.Addrs() {
if !manet.IsPublicAddr(addr) {
continue
}
addr = addr.Encapsulate(r.selfAddr)
addrBytes = append(addrBytes, addr.Bytes())
}
rsvp := &pbv2.Reservation{
Expire: &expireUnix,
Addrs: addrBytes,
}
voucher := &proto.ReservationVoucher{
Relay: r.host.ID(),
Peer: p,
Expiration: expire,
}
envelope, err := record.Seal(voucher, r.host.Peerstore().PrivKey(r.host.ID()))
if err != nil {
log.Errorf("error sealing voucher for %s: %s", p, err)
return rsvp
}
blob, err := envelope.Marshal()
if err != nil {
log.Errorf("error marshalling voucher for %s: %s", p, err)
return rsvp
}
rsvp.Voucher = blob
return rsvp
}
func (r *Relay) makeLimitMsg(p peer.ID) *pbv2.Limit {
if r.rc.Limit == nil {
return nil
}
duration := uint32(r.rc.Limit.Duration / time.Second)
data := uint64(r.rc.Limit.Data)
return &pbv2.Limit{
Duration: &duration,
Data: &data,
}
}
func (r *Relay) background() {
asnutil.Store.Init()
ticker := time.NewTicker(time.Minute)
defer ticker.Stop()
for {
select {
case <-ticker.C:
r.gc()
case <-r.ctx.Done():
return
}
}
}
func (r *Relay) gc() {
r.mx.Lock()
defer r.mx.Unlock()
now := time.Now()
cnt := 0
for p, expire := range r.rsvp {
if r.closed || expire.Before(now) {
delete(r.rsvp, p)
r.host.ConnManager().UntagPeer(p, "relay-reservation")
cnt++
}
}
if r.metricsTracer != nil {
r.metricsTracer.ReservationClosed(cnt)
}
for p, count := range r.conns {
if count == 0 {
delete(r.conns, p)
}
}
}
func (r *Relay) disconnected(n network.Network, c network.Conn) {
p := c.RemotePeer()
if n.Connectedness(p) == network.Connected {
return
}
r.mx.Lock()
_, ok := r.rsvp[p]
if ok {
delete(r.rsvp, p)
}
r.mx.Unlock()
if ok && r.metricsTracer != nil {
r.metricsTracer.ReservationClosed(1)
}
}
func isRelayAddr(a ma.Multiaddr) bool {
_, err := a.ValueForProtocol(ma.P_CIRCUIT)
return err == nil
}