edgevpn/pkg/node/connection.go

/*
Copyright © 2021-2022 Ettore Di Giacinto <[email protected]>
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
    http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/

package node

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"io"
	mrand "math/rand"
	"net"

	internalCrypto "github.com/mudler/edgevpn/pkg/crypto"

	"github.com/libp2p/go-libp2p"
	"github.com/libp2p/go-libp2p/core/crypto"
	"github.com/libp2p/go-libp2p/core/host"
	"github.com/libp2p/go-libp2p/core/peer"
	conngater "github.com/libp2p/go-libp2p/p2p/net/conngater"
	hub "github.com/mudler/edgevpn/pkg/hub"
	multiaddr "github.com/multiformats/go-multiaddr"
)

// Host returns the libp2p peer host
func (e *Node) Host() host.Host {
	return e.host
}

// ConnectionGater returns the underlying libp2p conngater
func (e *Node) ConnectionGater() *conngater.BasicConnectionGater {
	return e.cg
}

// BlockSubnet blocks the CIDR subnet from connections
func (e *Node) BlockSubnet(cidr string) error {
	// Avoid to loopback traffic by trying to connect to nodes in via VPN
	_, n, err := net.ParseCIDR(cidr)
	if err != nil {
		return err
	}

	return e.ConnectionGater().BlockSubnet(n)
}

func GenPrivKey(seed int64) (crypto.PrivKey, error) {
	var r io.Reader
	if seed == 0 {
		r = rand.Reader
	} else {
		r = mrand.New(mrand.NewSource(seed))
	}
	prvKey, _, err := crypto.GenerateKeyPairWithReader(crypto.Ed25519, 4096, r)
	return prvKey, err
}

func (e *Node) genHost(ctx context.Context) (host.Host, error) {
	var prvKey crypto.PrivKey

	opts := e.config.Options

	cg, err := conngater.NewBasicConnectionGater(nil)
	if err != nil {
		return nil, err
	}

	e.cg = cg

	if e.config.InterfaceAddress != "" {
		e.BlockSubnet(e.config.InterfaceAddress)
	}

	for _, b := range e.config.Blacklist {
		_, net, err := net.ParseCIDR(b)
		if err != nil {
			// Assume it's a peerID
			cg.BlockPeer(peer.ID(b))
		}
		if net != nil {
			cg.BlockSubnet(net)
		}
	}

	// generate privkey if not specified
	if len(e.config.PrivateKey) > 0 {
		prvKey, err = crypto.UnmarshalPrivateKey(e.config.PrivateKey)
	} else {
		prvKey, err = GenPrivKey(e.seed)
	}

	if err != nil {
		return nil, err
	}

	opts = append(opts, libp2p.ConnectionGater(cg), libp2p.Identity(prvKey))

	addrs := []multiaddr.Multiaddr{}
	for _, l := range e.config.ListenAddresses {
		addrs = append(addrs, []multiaddr.Multiaddr(l)...)
	}
	opts = append(opts, libp2p.ListenAddrs(addrs...))

	for _, d := range e.config.ServiceDiscovery {
		opts = append(opts, d.Option(ctx))
	}

	opts = append(opts, e.config.AdditionalOptions...)

	if e.config.Insecure {
		e.config.Logger.Info("Disabling Security transport layer")
		opts = append(opts, libp2p.NoSecurity)
	}

	opts = append(opts, FallbackDefaults)

	return libp2p.New(opts...)
}

// FallbackDefaults applies default options to the libp2p node if and only if no
// other relevant options have been applied. will be appended to the options
// passed into New.
var FallbackDefaults libp2p.Option = func(cfg *libp2p.Config) error {
	for _, def := range defaults {
		if !def.fallback(cfg) {
			continue
		}
		if err := cfg.Apply(def.opt); err != nil {
			return err
		}
	}
	return nil
}

// Complete list of default options and when to fallback on them.
//
// Please *DON'T* specify default options any other way. Putting this all here
// makes tracking defaults *much* easier.
var defaults = []struct {
	fallback func(cfg *libp2p.Config) bool
	opt      libp2p.Option
}{
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.Transports == nil && cfg.ListenAddrs == nil },
		opt:      libp2p.DefaultListenAddrs,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.Transports == nil && cfg.PSK == nil },
		opt:      libp2p.DefaultTransports,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.Transports == nil && cfg.PSK != nil },
		opt:      libp2p.DefaultPrivateTransports,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.Muxers == nil },
		opt:      libp2p.DefaultMuxers,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return !cfg.Insecure && cfg.SecurityTransports == nil },
		opt:      libp2p.DefaultSecurity,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.PeerKey == nil },
		opt:      libp2p.RandomIdentity,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.Peerstore == nil },
		opt:      libp2p.DefaultPeerstore,
	},
	{
		fallback: func(cfg *libp2p.Config) bool { return !cfg.RelayCustom },
		opt:      libp2p.DefaultEnableRelay,
	},
	//{
	//	fallback: func(cfg *libp2p.Config) bool { return cfg.ResourceManager == nil },
	//	opt:      libp2p.DefaultResourceManager,
	//},
	//{
	//	fallback: func(cfg *libp2p.Config) bool { return cfg.ConnManager == nil },
	//	opt:      libp2p.DefaultConnectionManager,
	//},
	{
		fallback: func(cfg *libp2p.Config) bool { return cfg.MultiaddrResolver == nil },
		opt:      libp2p.DefaultMultiaddrResolver,
	},
	//{
	//	fallback: func(cfg *libp2p.Config) bool { return !cfg.DisableMetrics && cfg.PrometheusRegisterer == nil },
	//	opt:      libp2p.DefaultPrometheusRegisterer,
	//},
}

func (e *Node) sealkey() string {
	return internalCrypto.MD5(internalCrypto.TOTP(sha256.New, e.config.SealKeyLength, e.config.SealKeyInterval, e.config.ExchangeKey))
}

func (e *Node) handleEvents(ctx context.Context, inputChannel chan *hub.Message, roomMessages chan *hub.Message, pub func(*hub.Message) error, handlers []Handler, peerGater bool) {
	for {
		select {
		case m := <-inputChannel:
			if m == nil {
				continue
			}
			c := m.Copy()
			str, err := e.config.Sealer.Seal(c.Message, e.sealkey())
			if err != nil {
				e.config.Logger.Warnf("%w from %s", err.Error(), c.SenderID)
			}
			c.Message = str

			if err := pub(c); err != nil {
				e.config.Logger.Warnf("publish error: %s", err)
			}

		case m := <-roomMessages:
			if m == nil {
				continue
			}

			if peerGater {
				if e.config.PeerGater != nil && e.config.PeerGater.Gate(e, peer.ID(m.SenderID)) {
					e.config.Logger.Warnf("gated message from %s", m.SenderID)
					continue
				}
			}
			if len(e.config.PeerTable) > 0 {
				found := false
				for _, p := range e.config.PeerTable {
					if p.String() == peer.ID(m.SenderID).String() {
						found = true
					}
				}
				if !found {
					e.config.Logger.Warnf("gated message from %s - not present in peertable", m.SenderID)
					continue
				}
			}

			c := m.Copy()
			str, err := e.config.Sealer.Unseal(c.Message, e.sealkey())
			if err != nil {
				e.config.Logger.Warnf("%w from %s", err.Error(), c.SenderID)
			}
			c.Message = str
			e.handleReceivedMessage(c, handlers, inputChannel)
		case <-ctx.Done():
			return
		}
	}
}

func (e *Node) handleReceivedMessage(m *hub.Message, handlers []Handler, c chan *hub.Message) {
	for _, h := range handlers {
		if err := h(e.ledger, m, c); err != nil {
			e.config.Logger.Warnf("handler error: %s", err)
		}
	}
}