I am working in a Noise-based P2P lib that has only basic TCP networking implemented so far. I am looking for anyone who get interested in this project that want to helps with reviews to the code and give some feedbacks about design, good practices, potential improvements, fixes, etc.
In general, it is a library that aims to serve as a tool to create secure P2P networks based on the Noise Framework.
"Noise is a framework for building crypto protocols. Noise protocols support mutual and optional authentication, identity hiding, forward secrecy, zero round-trip encryption, and other advanced features."
Specifically, the code below tries to symmetrically implement the basic methods for establishing the connection between one or more peers. It uses pubsub as a pattern for internal notifications and so far a mesh routing (unstructured network).
//Copyright (c) 2022, Geolffrey Mena <gmjun2000@gmail.com>
//P2P Noise Secure handshake.
//
//See also: http://www.noiseprotocol.org/noise.html#introduction
package noise
import (
"context"
"io"
"log"
"net"
"github.com/geolffreym/p2p-noise/errors"
)
// Default protocol
const PROTOCOL = "tcp"
type Node struct {
sentinel chan bool // Channel flag waiting for signal to close connection.
router *Router // Routing hash table eg. {Socket: Conn interface}.
events *Events // Pubsub notifications.
}
func NewNode() *Node {
return &Node{
router: newRouter(),
events: newEvents(),
sentinel: make(chan bool),
}
}
// Events proxy channels event to subscriber
// The listening routine should be stopped using context param.
func (n *Node) Events(ctx context.Context) <-chan Message {
ch := make(chan Message)
go n.events.Subscriber().Listen(ctx, ch)
return ch // read only channel <-chan
}
// watch keep running waiting for incoming messages.
// After every new message the connection is verified, if local connection is closed or remote peer is disconnected the watch routine is stopped.
// Incoming message monitor is suggested to be processed in go routines.
func (n *Node) watch(peer *Peer) {
buf := make([]byte, 1024)
KEEPALIVE:
for {
// Sync buffer reading
_, err := peer.Receive(buf)
// If connection is closed
if n.Closed() {
// stop routines watching for peers
return
}
if err != nil {
// net: don't return io.EOF from zero byte reads
// if err == io.EOF then peer connection is closed
_, isNetError := err.(*net.OpError)
if err == io.EOF || isNetError {
// Close disconnected peer
if err := peer.Close(); err != nil {
log.Fatal(errors.Closing(err).Error())
}
// Notify about the remote peer state
n.events.PeerDisconnected([]byte(peer.Socket()))
// Remove peer from router table
n.router.Remove(peer)
return
}
// Keep alive always that zero bytes are not received
break KEEPALIVE
}
// Emit new incoming message notification
n.events.NewMessage(buf)
}
}
// routing initialize route in routing table from connection interface.
// If TCP protocol is used connection is enforced to keep alive.
// It return new peer added to table.
func (n *Node) routing(conn net.Conn) *Peer {
// Assertion for tcp connection to keep alive
connection, ok := conn.(*net.TCPConn)
if ok {
// If tcp enforce keep alive connection
// SetKeepAlive sets whether the operating system should send keep-alive messages on the connection.
connection.SetKeepAlive(true)
}
// Routing connections
remote := connection.RemoteAddr().String()
// eg. 192.168.1.1:8080
socket := Socket(remote)
// We need to know how interact with peer based on socket and connection
peer := newPeer(socket, connection)
n.router.Add(peer)
return peer
}
// Listen start listening on the given address and wait for new connection.
// Return error if error occurred while listening.
func (n *Node) Listen(addr string) error {
listener, err := net.Listen(PROTOCOL, addr)
if err != nil {
return err
}
// Dispatch event on start listening
n.events.Listening([]byte(addr))
//wait until sentinel channel is closed to close listener
go func(listener net.Listener) {
<-n.sentinel
err := listener.Close()
if err != nil {
log.Fatal(errors.Closing(err).Error())
}
}(listener)
for {
// Block/Hold while waiting for new incoming connection
// Synchronized incoming connections
conn, err := listener.Accept()
// If connection is closed
if n.Closed() {
// Graceful stop listening
return nil
}
if err != nil {
log.Fatal(errors.Binding(err).Error())
return err
}
peer := n.routing(conn) // Routing for connection
go n.watch(peer) // Wait for incoming messages
// Dispatch event for new peer connected
payload := []byte(peer.Socket())
n.events.PeerConnected(payload)
}
}
// Table return current routing table
func (n *Node) Table() Table {
return n.router.Table()
}
// Closed check connection state.
// Return true for connection open else false
func (n *Node) Closed() bool {
select {
case <-n.sentinel:
return true
default:
return false
}
}
// Close all peers connections and stop listening
func (n *Node) Close() {
for _, peer := range n.router.Table() {
go func(p *Peer) {
if err := p.Close(); err != nil {
log.Fatal(errors.Closing(err).Error())
}
}(peer)
}
// Dispatch event on node get closed
n.events.ClosedConnection()
// If channel get closed then all routines waiting for connections
// or waiting for incoming messages get closed too.
close(n.sentinel)
}
// Dial attempt to connect to remote node and add connected peer to routing table.
// Return error if error occurred while dialing node.
func (n *Node) Dial(addr string) error {
conn, err := net.Dial(PROTOCOL, addr)
if err != nil {
return errors.Dialing(err, addr)
}
peer := n.routing(conn) // Routing for connection
go n.watch(peer) // Wait for incoming messages
// Dispatch event for new peer connected
n.events.PeerConnected([]byte(peer.Socket()))
return nil
}
Example of usage:
Note: Its a work in progress so many things can change about the design or usage.
//Copyright (c) 2022, Geolffrey Mena <gmjun2000@gmail.com>
//P2P Noise Secure handshake.
//
//See also: http://www.noiseprotocol.org/noise.html#introduction
package main
import (
"context"
"log"
noise "github.com/geolffreym/p2p-noise"
)
func main() {
node := noise.NewNode()
// Network events channel
ctx, cancel := context.WithCancel(context.Background())
events := node.Events(ctx)
go func() {
for msg := range events {
log.Printf("Listening on: %s \n", msg.Payload())
cancel() // stop listening for events
}
}()
// ... some code here
// node.Dial("192.168.1.1:4008")
// node.Close()
// ... more code here
node.Listen("127.0.0.1:4008")
}
