derohe-miniblock-mod/p2p/controller.go

// Copyright 2017-2021 DERO Project. All rights reserved.
// Use of this source code in any form is governed by RESEARCH license.
// license can be found in the LICENSE file.
// GPG: 0F39 E425 8C65 3947 702A  8234 08B2 0360 A03A 9DE8
//
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY
// EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
// MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL
// THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
// PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
// INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT,
// STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF
// THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

package p2p

//import "os"
import "fmt"
import "net"
import "net/rpc"
import "time"
import "sort"
import "strings"
import "math/big"
import "strconv"

//import "crypto/rsa"
import "crypto/ecdsa"
import "crypto/elliptic"

import "crypto/tls"
import "crypto/rand"
import "crypto/x509"
import "encoding/pem"
import "sync/atomic"
import "runtime/debug"

import "github.com/romana/rlog"
import log "github.com/sirupsen/logrus"

import "github.com/deroproject/derohe/config"
import "github.com/deroproject/derohe/globals"
import "github.com/deroproject/derohe/blockchain"
import "github.com/deroproject/derohe/metrics"

var chain *blockchain.Blockchain // external reference to chain

var P2P_Port int // this will be exported while doing handshake

var Exit_Event = make(chan bool) // causes all threads to exit
var Exit_In_Progress bool        // marks we are doing exit
var logger *log.Entry            // global logger, every logger in this package is a child of this
var sync_node bool               // whether sync mode is activated

var nonbanlist []string // any ips in this list will never be banned
// the list will include seed nodes, any nodes provided at command prompt

// Initialize P2P subsystem
func P2P_Init(params map[string]interface{}) error {
	logger = globals.Logger.WithFields(log.Fields{"com": "P2P"}) // all components must use this logger

	// register_handlers()

	GetPeerID() // Initialize peer id once

	// parse node tag if availble
	if _, ok := globals.Arguments["--node-tag"]; ok {
		if globals.Arguments["--node-tag"] != nil {

			node_tag = globals.Arguments["--node-tag"].(string)
		}
	}

	// permanently unban any seed nodes
	if globals.IsMainnet() {
		for i := range config.Mainnet_seed_nodes {
			nonbanlist = append(nonbanlist, strings.ToLower(config.Mainnet_seed_nodes[i]))
		}
	} else { // initial bootstrap
		for i := range config.Testnet_seed_nodes {
			nonbanlist = append(nonbanlist, strings.ToLower(config.Testnet_seed_nodes[i]))
		}
	}

	chain = params["chain"].(*blockchain.Blockchain)
	load_ban_list()  // load ban list
	load_peer_list() // load old list if availble

	// if user provided a sync node, connect with it
	if _, ok := globals.Arguments["--sync-node"]; ok { // check if parameter is supported
		if globals.Arguments["--sync-node"].(bool) {
			sync_node = true
			// disable p2p port
			globals.Arguments["--p2p-bind"] = ":0"

			// disable all connections except seed nodes
			globals.Arguments["--add-exclusive-node"] = []string{"0.0.0.0:0"}
			globals.Arguments["--add-priority-node"] = []string{"0.0.0.0:0"}

			go maintain_seed_node_connection()

			logger.Warnf("Sync mode is enabled. Please remove this option after chain syncs successfully")
		}
	}

	// register the metrics with the metrics registry
	metrics.Registry.MustRegister(block_propagation)
	metrics.Registry.MustRegister(transaction_propagation)

	go P2P_Server_v2()        // start accepting connections
	go P2P_engine()           // start outgoing engine
	go syncroniser()          // start sync engine
	go clean_up_propagation() // clean up propagation map
	go ping_loop()            // keep pinging
	logger.Infof("P2P started")
	atomic.AddUint32(&globals.Subsystem_Active, 1) // increment subsystem
	return nil
}

// TODO we need to make sure that exclusive/priority nodes are never banned
func P2P_engine() {

	var end_point_list []string
	if _, ok := globals.Arguments["--add-exclusive-node"]; ok { // check if parameter is supported
		if globals.Arguments["--add-exclusive-node"] != nil {
			tmp_list := globals.Arguments["--add-exclusive-node"].([]string)
			for i := range tmp_list {
				end_point_list = append(end_point_list, tmp_list[i])
				nonbanlist = append(nonbanlist, tmp_list[i])
			}
		}
	}

	// all prority nodes will be always connected
	if _, ok := globals.Arguments["--add-priority-node"]; ok { // check if parameter is supported
		if globals.Arguments["--add-priority-node"] != nil {
			tmp_list := globals.Arguments["--add-priority-node"].([]string)
			for i := range tmp_list {
				end_point_list = append(end_point_list, tmp_list[i])
				nonbanlist = append(nonbanlist, tmp_list[i])
			}
		}
	}

	{ // remove duplicates if any
		sort.Strings(end_point_list)
	start_again: // this list is expected to be less  than 100
		for i := range end_point_list {
			if i > 0 && end_point_list[i-1] == end_point_list[i] {
				end_point_list = append(end_point_list[:i-1], end_point_list[i:]...)
				goto start_again
			}
		}
	}

	logger.Debugf("Priority list %+v", end_point_list)

	// maintain connection to exclusive/priority nodes
	for i := range end_point_list {
		go maintain_outgoing_priority_connection(end_point_list[i], false)
	}

	// do not create connections to peers , if requested
	if _, ok := globals.Arguments["--add-exclusive-node"]; ok && len(globals.Arguments["--add-exclusive-node"].([]string)) == 0 { // check if parameter is supported
		go maintain_connection_to_peers()  // maintain certain number of  connections for peer to peers
		go maintain_seed_node_connection() // maintain connection with atleast 1 seed node

		// this code only triggers when we do not have peer list
		if find_peer_to_connect(1) == nil { // either we donot have a peer list or everyone is banned
			// trigger connection to all seed nodes hoping some will be up
			if globals.IsMainnet() { // initial boot strap should be quick
				for i := range config.Mainnet_seed_nodes {
					go connect_with_endpoint(config.Mainnet_seed_nodes[i], true)
				}
			} else { // initial bootstrap
				for i := range config.Testnet_seed_nodes {
					go connect_with_endpoint(config.Testnet_seed_nodes[i], true)
				}
			}

		}

	}

}

// will try to connect with given endpoint
// will block until the connection dies or is killed
func connect_with_endpoint(endpoint string, sync_node bool) {

	defer func() {
		if r := recover(); r != nil {
			rlog.Warnf("Recovered while handling connection, Stack trace below", r)
			rlog.Warnf("Stack trace  \n%s", debug.Stack())
		}
	}()

	remote_ip, err := net.ResolveTCPAddr("tcp", endpoint)
	if err != nil {
		rlog.Warnf("Resolve address failed:", err.Error())
		return
	}

	// check whether are already connected to this address if yes, return
	if IsAddressConnected(remote_ip.String()) {
		return //nil, fmt.Errorf("Already connected")
	}

	// since we may be connecting through socks, grab the remote ip for our purpose rightnow
	conn, err := globals.Dialer.Dial("tcp", remote_ip.String())

	//conn, err := tls.DialWithDialer(&globals.Dialer, "tcp", remote_ip.String(),&tls.Config{InsecureSkipVerify: true})
	//conn, err := tls.Dial("tcp", remote_ip.String(),&tls.Config{InsecureSkipVerify: true})
	if err != nil {
		rlog.Warnf("Dial failed err %s", err.Error())
		Peer_SetFail(remote_ip.String()) // update peer list as we see
		return                           //nil, fmt.Errorf("Dial failed err %s", err.Error())
	}

	tcpc := conn.(*net.TCPConn)
	// detection time: tcp_keepalive_time + tcp_keepalive_probes + tcp_keepalive_intvl
	// default on linux:  30 + 8 * 30
	// default on osx:    30 + 8 * 75
	tcpc.SetKeepAlive(true)
	tcpc.SetKeepAlivePeriod(8 * time.Second)
	tcpc.SetLinger(0) // discard any pending data

	//conn.SetKeepAlive(true) // set keep alive true
	//conn.SetKeepAlivePeriod(10*time.Second) // keep alive every 10 secs

	// upgrade connection TO TLS ( tls.Dial does NOT support proxy)
	// TODO we need to choose fastest cipher here ( so both clients/servers are not loaded)
	conn = tls.Client(conn, &tls.Config{InsecureSkipVerify: true})

	process_connection(conn, remote_ip, false, sync_node)

	//Handle_Connection(conn, remote_ip, false, sync_node) // handle  connection
}

// maintains a persistant connection to endpoint
// if connection drops, tries again after 4 secs
func maintain_outgoing_priority_connection(endpoint string, sync_node bool) {
	for {
		select {
		case <-Exit_Event:
			return
		case <-time.After(4 * time.Second):
		}
		connect_with_endpoint(endpoint, sync_node)
	}
}

// this will maintain connection to 1 seed node randomly
func maintain_seed_node_connection() {
	for {

		select {
		case <-Exit_Event:
			return
		case <-time.After(2 * time.Second):
		}
		endpoint := ""
		if globals.IsMainnet() { // choose mainnet seed node
			r, _ := rand.Int(rand.Reader, big.NewInt(10240))
			endpoint = config.Mainnet_seed_nodes[r.Int64()%int64(len(config.Mainnet_seed_nodes))]
		} else { // choose testnet peer node
			r, _ := rand.Int(rand.Reader, big.NewInt(10240))
			endpoint = config.Testnet_seed_nodes[r.Int64()%int64(len(config.Testnet_seed_nodes))]
		}
		if endpoint != "" {
			//connect_with_endpoint(endpoint, sync_node)
			connect_with_endpoint(endpoint, true) // seed nodes always have sync mode
		}
	}
}

// keep building connections to network, we are talking outgoing connections
func maintain_connection_to_peers() {

	Min_Peers := int64(13) // we need to expose this to be modifieable at runtime without taking daemon offline
	// check how many connections are active
	if _, ok := globals.Arguments["--min-peers"]; ok && globals.Arguments["--min-peers"] != nil { // user specified a limit, use it if possible
		i, err := strconv.ParseInt(globals.Arguments["--min-peers"].(string), 10, 64)
		if err != nil {
			logger.Warnf("Error Parsing --max-peers err %s", err)
		} else {
			if i <= 1 {
				logger.Warnf("--min-peers should be positive and more than 1")
			} else {
				Min_Peers = i
			}
		}
		logger.Infof("Min outgoing peers limit %d", Min_Peers)
	}

	for {
		select {
		case <-Exit_Event:
			return
		case <-time.After(1000 * time.Millisecond):
		}

		// check number of connections, if limit is reached, trigger new connections if we have peers
		// if we have more do nothing
		_, out := Peer_Direction_Count()
		if out >= uint64(Min_Peers) { // we already have required number of peers, donot connect to more peers
			continue
		}

		peer := find_peer_to_connect(1)
		if peer != nil {
			go connect_with_endpoint(peer.Address, false)
		}
	}
}

func P2P_Server_v2() {

	default_address := "0.0.0.0:" + fmt.Sprintf("%d", config.Mainnet.P2P_Default_Port)
	P2P_Port = config.Mainnet.P2P_Default_Port
	if !globals.IsMainnet() {
		default_address = "0.0.0.0:" + fmt.Sprintf("%d", config.Testnet.P2P_Default_Port)
		P2P_Port = config.Testnet.P2P_Default_Port
	}

	if _, ok := globals.Arguments["--p2p-bind"]; ok && globals.Arguments["--p2p-bind"] != nil {
		addr, err := net.ResolveTCPAddr("tcp", globals.Arguments["--p2p-bind"].(string))
		if err != nil {
			logger.Warnf("--p2p-bind address is invalid, err = %s", err)
		} else {
			if addr.Port == 0 {
				logger.Infof("P2P server is disabled, No ports will be opened for P2P activity")
				return
			} else {
				default_address = addr.String()
				P2P_Port = addr.Port
			}
		}
	}

	logger.Infof("P2P  will listen on %s", default_address)
	tlsconfig := &tls.Config{Certificates: []tls.Certificate{generate_random_tls_cert()}}
	//l, err := tls.Listen("tcp", default_address, tlsconfig) // listen as TLS server

	// listen to incoming tcp connections tls style
	l, err := net.Listen("tcp", default_address) // listen as simple TCP server
	if err != nil {
		logger.Fatalf("Could not listen on %s, err %s", default_address, err)
	}
	defer l.Close()

	// p2p is shutting down, close the listening socket
	go func() { <-Exit_Event; l.Close() }()

	// A common pattern is to start a loop to continously accept connections
	for {
		conn, err := l.Accept() //accept connections using Listener.Accept()
		if err != nil {
			select {
			case <-Exit_Event:
				return
			default:
			}
			logger.Warnf("Err while accepting incoming connection errr %s", err)
			continue
		}
		raddr := conn.RemoteAddr().(*net.TCPAddr)

		//if incoming IP is banned, disconnect now
		if IsAddressInBanList(raddr.IP.String()) {
			rlog.Tracef(1, "Incoming IP %s is banned, disconnecting now", raddr.IP.String())
			conn.Close()
		} else {

			tcpc := conn.(*net.TCPConn)
			// detection time: tcp_keepalive_time + tcp_keepalive_probes + tcp_keepalive_intvl
			// default on linux:  30 + 8 * 30
			// default on osx:    30 + 8 * 75
			tcpc.SetKeepAlive(true)
			tcpc.SetKeepAlivePeriod(8 * time.Second)
			tcpc.SetLinger(0) // discard any pending data

			tlsconn := tls.Server(conn, tlsconfig)
			go process_connection(tlsconn, raddr, true, false) // handle connection in a different go routine
		}
	}

}

func process_connection(conn net.Conn, remote_addr *net.TCPAddr, incoming, sync_node bool) {
	defer globals.Recover()

	var rconn *RPC_Connection
	var err error
	if incoming {
		rconn, err = wait_stream_creation_server_side(conn) // do server side processing
	} else {
		rconn, err = stream_creation_client_side(conn) // do client side processing
	}
	if err == nil {

		var RPCSERVER = rpc.NewServer()
		c := &Connection{RConn: rconn, Addr: remote_addr, State: HANDSHAKE_PENDING, Incoming: incoming, SyncNode: sync_node}
		RPCSERVER.RegisterName("Peer", c) // register the handlers

		if incoming {
			c.logger = logger.WithFields(log.Fields{"RIP": remote_addr.String(), "DIR": "INC"})
		} else {
			c.logger = logger.WithFields(log.Fields{"RIP": remote_addr.String(), "DIR": "OUT"})
		}
		go func() {
			//defer globals.Recover()
		defer func() {
		if r := recover(); r != nil {
			rlog.Warnf("Recovered while handling connection, Stack trace below", r)
			rlog.Warnf("Stack trace  \n%s", debug.Stack())
		}
		}()
			//RPCSERVER.ServeConn(rconn.ServerConn)                      // start single threaded rpc server with GOB encoding
			RPCSERVER.ServeCodec(NewCBORServerCodec(rconn.ServerConn)) // use CBOR encoding on rpc
		}()

		c.dispatch_test_handshake()

		<-rconn.Session.CloseChan()
		Connection_Delete(c)
		//fmt.Printf("closing connection status err: %s\n",err)
	}
	conn.Close()

}

// shutdown the p2p component
func P2P_Shutdown() {
	close(Exit_Event) // send signal to all connections to exit
	save_peer_list()  // save peer list
	save_ban_list()   // save ban list

	// TODO we  must wait for connections to kill themselves
	time.Sleep(1 * time.Second)
	logger.Infof("P2P Shutdown")
	atomic.AddUint32(&globals.Subsystem_Active, ^uint32(0)) // this decrement 1 fom subsystem

}

// generate default tls cert to encrypt everything
// NOTE: this does NOT protect from individual active man-in-the-middle attacks
func generate_random_tls_cert() tls.Certificate {

	/* RSA can do only 500 exchange per second, we need to be faster
	     * reference https://github.com/golang/go/issues/20058
	    key, err := rsa.GenerateKey(rand.Reader, 512) // current using minimum size
	if err != nil {
	    log.Fatal("Private key cannot be created.", err.Error())
	}

	// Generate a pem block with the private key
	keyPem := pem.EncodeToMemory(&pem.Block{
	    Type:  "RSA PRIVATE KEY",
	    Bytes: x509.MarshalPKCS1PrivateKey(key),
	})
	*/
	// EC256 does roughly 20000 exchanges per second
	key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
	b, err := x509.MarshalECPrivateKey(key)
	if err != nil {
		log.Fatalf("Unable to marshal ECDSA private key: %v", err)
	}
	// Generate a pem block with the private key
	keyPem := pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: b})

	tml := x509.Certificate{
		SerialNumber: big.NewInt(int64(GetPeerID()) ^ int64(time.Now().UnixNano())),

		// TODO do we need to add more parameters to make our certificate more authentic
		// and thwart traffic identification as a mass scale
		/*
		   // you can add any attr that you need
		   NotBefore:    time.Now(),
		   NotAfter:     time.Now().AddDate(5, 0, 0),
		   // you have to generate a different serial number each execution

		   Subject: pkix.Name{
		       CommonName:   "New Name",
		       Organization: []string{"New Org."},
		   },
		   BasicConstraintsValid: true,   // even basic constraints are not required
		*/
	}
	cert, err := x509.CreateCertificate(rand.Reader, &tml, &tml, &key.PublicKey, key)
	if err != nil {
		log.Fatal("Certificate cannot be created.", err.Error())
	}

	// Generate a pem block with the certificate
	certPem := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cert})
	tlsCert, err := tls.X509KeyPair(certPem, keyPem)
	if err != nil {
		log.Fatal("Cannot be loaded the certificate.", err.Error())
	}
	return tlsCert
}

/*
// register all the handlers
func register_handlers(){
    arpc.DefaultHandler.Handle("/handshake",Handshake_Handler)
    arpc.DefaultHandler.Handle("/active",func (ctx *arpc.Context) { // set the connection active
    if c,ok := ctx.Client.Get("connection");ok {
        connection := c.(*Connection)
        atomic.StoreUint32(&connection.State, ACTIVE)
       }} )

    arpc.DefaultHandler.HandleConnected(OnConnected_Handler) // all incoming connections will first processed here
arpc.DefaultHandler.HandleDisconnected(OnDisconnected_Handler) // all disconnected
}



// triggers when new clients connect and
func OnConnected_Handler(c *arpc.Client){
    dispatch_test_handshake(c, c.Conn.RemoteAddr().(*net.TCPAddr) ,true,false) // client connected we must handshake
}

func OnDisconnected_Handler(c *arpc.Client){
    c.Stop()
}
*/