monte

README

monte

The bare minimum for high performance, fully-encrypted RPC over TCP in Go.

Features

1. Send requests, receive responses, or send messages without waiting for a response.
2. Send from 50MiB/s to 1500MiB/s, with zero allocations per sent message or RPC call.
3. Gracefully establish multiple client connections to a single endpoint up to a configurable limit.
4. Set the total number of connections that may concurrently be accepted and handled by a single endpoint.
5. Configure read/write timeouts, dial timeouts, handshake timeouts, or customize the handshaking protocol.
6. All messages, once the handshake protocol is complete, are encrypted and non-distinguishable from each other.
7. Supports graceful shutdowns for both client and server, with extensive tests for highly-concurrent scenarios.

Protocol

Handshake

1. Send X25519 curve point (32 bytes) to peer.
2. Receive X25519 curve point (32 bytes) from our peer.
3. Multiply X25519 curve scalar with X25519 curve point received from our peer.
4. Derive a shared key by using BLAKE-2b as a key derivation function over our scalar point multiplication result.
5. Encrypt further communication with AES 256-bit GCM using our shared key, with a nonce counter increasing for every incoming/outgoing message.

Message Format

1. Encrypted messages are prefixed with an unsigned 32-bit integer denoting the message's length.
2. The decoded message content is prefixed with an unsigned 32-bit integer designating a sequence number.
3. The sequence number is used as an identifier to identify requests/responses from one another.
4. The sequence number 0 is reserved for requests that do not expect a response.

Benchmarks

$ cat /proc/cpuinfo | grep 'model name' | uniq
model name : Intel(R) Core(TM) i7-7700HQ CPU @ 2.80GHz

$ go test -bench=. -benchtime=10s
goos: linux
goarch: amd64
pkg: github.com/lithdew/monte
BenchmarkSend-8                          1814391              6690 ns/op         209.27 MB/s         115 B/op          0 allocs/op
BenchmarkSendNoWait-8                   10638730              1153 ns/op        1214.19 MB/s         141 B/op          0 allocs/op
BenchmarkRequest-8                        438381             28556 ns/op          49.03 MB/s         140 B/op          0 allocs/op
BenchmarkParallelSend-8                  4917001              2876 ns/op         486.70 MB/s         115 B/op          0 allocs/op
BenchmarkParallelSendNoWait-8           10317255              1291 ns/op        1084.78 MB/s         150 B/op          0 allocs/op
BenchmarkParallelRequest-8               1341444              8520 ns/op         164.32 MB/s         140 B/op          0 allocs/op

Documentation

Index

• Variables
• func AcquireTimer(timeout time.Duration) *time.Timer
• func IsEOF(err error) bool
• func Read(dst []byte, r io.Reader) ([]byte, error)
• func ReadSized(dst []byte, r io.Reader, max int) ([]byte, error)
• func ReleaseTimer(t *time.Timer)
• func Write(w io.Writer, buf []byte) error
• func WriteSized(w io.Writer, buf []byte) error
• type BufferedConn
• type Client
  • func (c *Client) Get() (*Conn, error)
  • func (c *Client) NumPendingWrites() int
  • func (c *Client) Request(dst, buf []byte) ([]byte, error)
  • func (c *Client) Send(buf []byte) error
  • func (c *Client) SendNoWait(buf []byte) error
  • func (c *Client) Shutdown()
• type Conn
  • func (c *Conn) Handle(done chan struct{}, conn BufferedConn) error
  • func (c *Conn) NumPendingWrites() int
  • func (c *Conn) Request(dst []byte, payload []byte) ([]byte, error)
  • func (c *Conn) Send(payload []byte) error
  • func (c *Conn) SendNoWait(payload []byte) error
• type ConnState
• type ConnStateHandler
• type ConnStateHandlerFunc
  • func (fn ConnStateHandlerFunc) HandleConnState(conn *Conn, state ConnState)
• type Context
  • func (c *Context) Body() []byte
  • func (c *Context) Conn() *Conn
  • func (c *Context) Reply(buf []byte) error
• type Handler
• type HandlerFunc
  • func (fn HandlerFunc) HandleMessage(ctx *Context) error
• type Handshaker
• type HandshakerFunc
  • func (fn HandshakerFunc) Handshake(conn net.Conn) (BufferedConn, error)
• type Server
  • func (s *Server) Serve(ln net.Listener) error
  • func (s *Server) Shutdown()
• type Session
  • func (s *Session) DoClient(conn net.Conn) error
  • func (s *Session) DoServer(conn net.Conn) error
  • func (s *Session) Establish(ourPriv []byte) error
  • func (s *Session) GenerateEphemeralKeys() ([]byte, []byte, error)
  • func (s *Session) Read(conn net.Conn) error
  • func (s *Session) SharedKey() []byte
  • func (s *Session) Suite() cipher.AEAD
  • func (s *Session) Write(conn net.Conn, ourPub []byte) error
• type SessionConn
  • func NewSessionConn(suite cipher.AEAD, conn net.Conn) *SessionConn
  • func (s *SessionConn) Close() error
  • func (s *SessionConn) Flush() error
  • func (s *SessionConn) LocalAddr() net.Addr
  • func (s *SessionConn) Read(b []byte) (int, error)
  • func (s *SessionConn) RemoteAddr() net.Addr
  • func (s *SessionConn) SetDeadline(t time.Time) error
  • func (s *SessionConn) SetReadDeadline(t time.Time) error
  • func (s *SessionConn) SetWriteDeadline(t time.Time) error
  • func (s *SessionConn) Write(b []byte) (int, error)

Variables

var DefaultClientSeqDelta uint32 = 2

var DefaultClientSeqOffset uint32 = 1

var DefaultDialTimeout = 3 * time.Second

var DefaultHandshakeTimeout = 3 * time.Second

var DefaultMaxClientConns = 4

var DefaultMaxConnWaitTimeout = 3 * time.Second

var DefaultMaxServerConns = 1024

var DefaultNumDialAttempts = 1

var DefaultReadBufferSize = 4096

var DefaultReadTimeout = 3 * time.Second

var DefaultSeqDelta uint32 = 2

var DefaultSeqOffset uint32 = 1

var DefaultServerSeqDelta uint32 = 2

var DefaultServerSeqOffset uint32 = 2

var DefaultWriteBufferSize = 4096

var DefaultWriteTimeout = 3 * time.Second

Functions

func AcquireTimer

func AcquireTimer(timeout time.Duration) *time.Timer

func IsEOF

func IsEOF(err error) bool

func Read

func Read(dst []byte, r io.Reader) ([]byte, error)

func ReadSized

func ReadSized(dst []byte, r io.Reader, max int) ([]byte, error)

func ReleaseTimer

func ReleaseTimer(t *time.Timer)

func Write

func Write(w io.Writer, buf []byte) error

func WriteSized

func WriteSized(w io.Writer, buf []byte) error

Types

type BufferedConn

type BufferedConn interface {
	net.Conn
	Flush() error
}

type Client

type Client struct {
	Addr string

	Handler   Handler
	ConnState ConnStateHandler

	Handshaker       Handshaker
	HandshakeTimeout time.Duration

	MaxConns        int
	NumDialAttempts int

	ReadBufferSize  int
	WriteBufferSize int

	DialTimeout  time.Duration
	ReadTimeout  time.Duration
	WriteTimeout time.Duration

	SeqOffset uint32
	SeqDelta  uint32
	// contains filtered or unexported fields
}

func (*Client) Get

func (c *Client) Get() (*Conn, error)

func (*Client) NumPendingWrites

func (c *Client) NumPendingWrites() int

func (*Client) Request

func (c *Client) Request(dst, buf []byte) ([]byte, error)

func (*Client) Send

func (c *Client) Send(buf []byte) error

func (*Client) SendNoWait

func (c *Client) SendNoWait(buf []byte) error

func (*Client) Shutdown

func (c *Client) Shutdown()

type Conn

type Conn struct {
	Handler Handler

	ReadBufferSize  int
	WriteBufferSize int

	ReadTimeout  time.Duration
	WriteTimeout time.Duration

	SeqOffset uint32
	SeqDelta  uint32
	// contains filtered or unexported fields
}

func (*Conn) Handle

func (c *Conn) Handle(done chan struct{}, conn BufferedConn) error

func (*Conn) NumPendingWrites

func (c *Conn) NumPendingWrites() int

func (*Conn) Request

func (c *Conn) Request(dst []byte, payload []byte) ([]byte, error)

func (*Conn) Send

func (c *Conn) Send(payload []byte) error

func (*Conn) SendNoWait

func (c *Conn) SendNoWait(payload []byte) error

type ConnState

type ConnState int

const (
	StateNew ConnState = iota
	StateClosed
)

type ConnStateHandler

type ConnStateHandler interface {
	HandleConnState(conn *Conn, state ConnState)
}

type ConnStateHandlerFunc

type ConnStateHandlerFunc func(conn *Conn, state ConnState)

var DefaultConnStateHandler ConnStateHandlerFunc = func(conn *Conn, state ConnState) {}

func (ConnStateHandlerFunc) HandleConnState

func (fn ConnStateHandlerFunc) HandleConnState(conn *Conn, state ConnState)

type Context

type Context struct {
	// contains filtered or unexported fields
}

func (*Context) Body

func (c *Context) Body() []byte

func (*Context) Conn

func (c *Context) Conn() *Conn

func (*Context) Reply

func (c *Context) Reply(buf []byte) error

type Handler

type Handler interface {
	HandleMessage(ctx *Context) error
}

type HandlerFunc

type HandlerFunc func(ctx *Context) error

var DefaultHandler HandlerFunc = func(ctx *Context) error { return nil }

func (HandlerFunc) HandleMessage

func (fn HandlerFunc) HandleMessage(ctx *Context) error

type Handshaker

type Handshaker interface {
	Handshake(conn net.Conn) (BufferedConn, error)
}

type HandshakerFunc

type HandshakerFunc func(conn net.Conn) (BufferedConn, error)

var DefaultClientHandshaker HandshakerFunc = func(conn net.Conn) (BufferedConn, error) {
	var session Session
	err := session.DoClient(conn)
	if err != nil {
		return nil, err
	}
	return NewSessionConn(session.Suite(), conn), nil
}

var DefaultServerHandshaker HandshakerFunc = func(conn net.Conn) (BufferedConn, error) {
	var session Session
	err := session.DoServer(conn)
	if err != nil {
		return nil, err
	}
	return NewSessionConn(session.Suite(), conn), nil
}

func (HandshakerFunc) Handshake

func (fn HandshakerFunc) Handshake(conn net.Conn) (BufferedConn, error)

type Server

type Server struct {
	Handler   Handler
	ConnState ConnStateHandler

	Handshaker       Handshaker
	HandshakeTimeout time.Duration

	MaxConns           int
	MaxConnWaitTimeout time.Duration

	ReadBufferSize  int
	WriteBufferSize int

	ReadTimeout  time.Duration
	WriteTimeout time.Duration

	SeqOffset uint32
	SeqDelta  uint32
	// contains filtered or unexported fields
}

func (*Server) Serve

func (s *Server) Serve(ln net.Listener) error

func (*Server) Shutdown

func (s *Server) Shutdown()

type Session

type Session struct {
	// contains filtered or unexported fields
}

Session is not safe for concurrent use.

func (*Session) DoClient

func (s *Session) DoClient(conn net.Conn) error

func (*Session) DoServer

func (s *Session) DoServer(conn net.Conn) error

func (*Session) Establish

func (s *Session) Establish(ourPriv []byte) error

func (*Session) GenerateEphemeralKeys

func (s *Session) GenerateEphemeralKeys() ([]byte, []byte, error)

func (*Session) Read

func (s *Session) Read(conn net.Conn) error

func (*Session) SharedKey

func (s *Session) SharedKey() []byte

func (*Session) Suite

func (s *Session) Suite() cipher.AEAD

func (*Session) Write

func (s *Session) Write(conn net.Conn, ourPub []byte) error

type SessionConn

type SessionConn struct {
	// contains filtered or unexported fields
}

SessionConn is not safe for concurrent use. It decrypts on reads and encrypts on writes via a provided cipher.AEAD suite for a given conn that implements net.Conn. It assumes all packets sent/received are to be prefixed with a 32-bit unsigned integer that designates the length of each individual packet.

The same cipher.AEAD suite must not be used for multiple SessionConn instances. Doing so will cause for plaintext data to be leaked.

func NewSessionConn

func NewSessionConn(suite cipher.AEAD, conn net.Conn) *SessionConn

func (*SessionConn) Close

func (s *SessionConn) Close() error

func (*SessionConn) Flush

func (s *SessionConn) Flush() error

func (*SessionConn) LocalAddr

func (s *SessionConn) LocalAddr() net.Addr

func (*SessionConn) Read

func (s *SessionConn) Read(b []byte) (int, error)

func (*SessionConn) RemoteAddr

func (s *SessionConn) RemoteAddr() net.Addr

func (*SessionConn) SetDeadline

func (s *SessionConn) SetDeadline(t time.Time) error

func (*SessionConn) SetReadDeadline

func (s *SessionConn) SetReadDeadline(t time.Time) error

func (*SessionConn) SetWriteDeadline

func (s *SessionConn) SetWriteDeadline(t time.Time) error

func (*SessionConn) Write

func (s *SessionConn) Write(b []byte) (int, error)