session.go

package tunnel

import (
	"crypto/cipher"
	"encoding/base32"
	"io"
	"sync"
	"time"

	"github.com/itzmanish/go-micro/v2/logger"
	"github.com/itzmanish/go-micro/v2/transport"
)

// session is our pseudo session for transport.Socket
type session struct {
	// the tunnel id
	tunnel string
	// the channel name
	channel string
	// the session id based on Micro.Tunnel-Session
	session string
	// token is the session token
	token string
	// closed
	closed chan bool
	// remote addr
	remote string
	// local addr
	local string
	// send chan
	send chan *message
	// recv chan
	recv chan *message
	// if the discovery worked
	discovered bool
	// if the session was accepted
	accepted bool
	// outbound marks the session as outbound dialled connection
	outbound bool
	// lookback marks the session as a loopback on the inbound
	loopback bool
	// mode of the connection
	mode Mode
	// the dial timeout
	dialTimeout time.Duration
	// the read timeout
	readTimeout time.Duration
	// the link on which this message was received
	link string
	// the error response
	errChan chan error
	// key for session encryption
	key []byte
	// cipher for session
	gcm cipher.AEAD
	sync.RWMutex
}

// message is sent over the send channel
type message struct {
	// type of message
	typ string
	// tunnel id
	tunnel string
	// channel name
	channel string
	// the session id
	session string
	// outbound marks the message as outbound
	outbound bool
	// loopback marks the message intended for loopback
	loopback bool
	// mode of the connection
	mode Mode
	// the link to send the message on
	link string
	// transport data
	data *transport.Message
	// the error channel
	errChan chan error
}

func (s *session) Remote() string {
	return s.remote
}

func (s *session) Local() string {
	return s.local
}

func (s *session) Link() string {
	return s.link
}

func (s *session) Id() string {
	return s.session
}

func (s *session) Channel() string {
	return s.channel
}

// newMessage creates a new message based on the session
func (s *session) newMessage(typ string) *message {
	return &message{
		typ:      typ,
		tunnel:   s.tunnel,
		channel:  s.channel,
		session:  s.session,
		outbound: s.outbound,
		loopback: s.loopback,
		mode:     s.mode,
		link:     s.link,
		errChan:  s.errChan,
	}
}

func (s *session) sendMsg(msg *message) error {
	select {
	case <-s.closed:
		return io.EOF
	case s.send <- msg:
		return nil
	}
}

func (s *session) wait(msg *message) error {
	// wait for an error response
	select {
	case err := <-msg.errChan:
		if err != nil {
			return err
		}
	case <-s.closed:
		return io.EOF
	}

	return nil
}

// waitFor waits for the message type required until the timeout specified
func (s *session) waitFor(msgType string, timeout time.Duration) (*message, error) {
	now := time.Now()

	after := func(timeout time.Duration) <-chan time.Time {
		if timeout < time.Duration(0) {
			return nil
		}

		// get the delta
		d := time.Since(now)

		// dial timeout minus time since
		wait := timeout - d

		if wait < time.Duration(0) {
			wait = time.Duration(0)
		}

		return time.After(wait)
	}

	// wait for the message type
	for {
		select {
		case msg := <-s.recv:
			// there may be no message type
			if len(msgType) == 0 {
				return msg, nil
			}

			// ignore what we don't want
			if msg.typ != msgType {
				if logger.V(logger.DebugLevel, log) {
					log.Debugf("Tunnel received non %s message in waiting for %s", msg.typ, msgType)
				}
				continue
			}

			// got the message
			return msg, nil
		case <-after(timeout):
			return nil, ErrReadTimeout
		case <-s.closed:
			// check pending message queue
			select {
			case msg := <-s.recv:
				// there may be no message type
				if len(msgType) == 0 {
					return msg, nil
				}

				// ignore what we don't want
				if msg.typ != msgType {
					if logger.V(logger.DebugLevel, log) {
						log.Debugf("Tunnel received non %s message in waiting for %s", msg.typ, msgType)
					}
					continue
				}

				// got the message
				return msg, nil
			default:
				// non blocking
			}
			return nil, io.EOF
		}
	}
}

// Discover attempts to discover the link for a specific channel.
// This is only used by the tunnel.Dial when first connecting.
func (s *session) Discover() error {
	// create a new discovery message for this channel
	msg := s.newMessage("discover")
	// broadcast the message to all links
	msg.mode = Broadcast
	// its an outbound connection since we're dialling
	msg.outbound = true
	// don't set the link since we don't know where it is
	msg.link = ""

	// if multicast then set that as session
	if s.mode == Multicast {
		msg.session = "multicast"
	}

	// send discover message
	if err := s.sendMsg(msg); err != nil {
		return err
	}

	// set time now
	now := time.Now()

	// after strips down the dial timeout
	after := func() time.Duration {
		d := time.Since(now)
		// dial timeout minus time since
		wait := s.dialTimeout - d
		// make sure its always > 0
		if wait < time.Duration(0) {
			return time.Duration(0)
		}
		return wait
	}

	// the discover message is sent out, now
	// wait to hear back about the sent message
	select {
	case <-time.After(after()):
		return ErrDialTimeout
	case err := <-s.errChan:
		if err != nil {
			return err
		}
	}

	// bail early if its not unicast
	// we don't need to wait for the announce
	if s.mode != Unicast {
		s.discovered = true
		s.accepted = true
		return nil
	}

	// wait for announce
	_, err := s.waitFor("announce", after())
	if err != nil {
		return err
	}

	// set discovered
	s.discovered = true

	return nil
}

// Open will fire the open message for the session. This is called by the dialler.
// This is to indicate that we want to create a new session.
func (s *session) Open() error {
	// create a new message
	msg := s.newMessage("open")

	// send open message
	if err := s.sendMsg(msg); err != nil {
		return err
	}

	// wait for an error response for send
	if err := s.wait(msg); err != nil {
		return err
	}

	// now wait for the accept message to be returned
	msg, err := s.waitFor("accept", s.dialTimeout)
	if err != nil {
		return err
	}

	// set to accepted
	s.accepted = true
	// set link
	s.link = msg.link

	return nil
}

// Accept sends the accept response to an open message from a dialled connection
func (s *session) Accept() error {
	msg := s.newMessage("accept")

	// send the accept message
	if err := s.sendMsg(msg); err != nil {
		return err
	}

	// wait for send response
	return s.wait(msg)
}

// Announce sends an announcement to notify that this session exists.
// This is primarily used by the listener.
func (s *session) Announce() error {
	msg := s.newMessage("announce")
	// we don't need an error back
	msg.errChan = nil
	// announce to all
	msg.mode = Broadcast
	// we don't need the link
	msg.link = ""

	// send announce message
	return s.sendMsg(msg)
}

// Send is used to send a message
func (s *session) Send(m *transport.Message) error {
	var err error

	s.RLock()
	gcm := s.gcm
	s.RUnlock()

	if gcm == nil {
		gcm, err = newCipher(s.key)
		if err != nil {
			return err
		}
		s.Lock()
		s.gcm = gcm
		s.Unlock()
	}
	// encrypt the transport message payload
	body, err := Encrypt(gcm, m.Body)
	if err != nil {
		log.Debugf("failed to encrypt message body: %v", err)
		return err
	}

	// make copy, without rehash and realloc
	data := &transport.Message{
		Header: make(map[string]string, len(m.Header)),
		Body:   body,
	}

	// encrypt all the headers
	for k, v := range m.Header {
		// encrypt the transport message payload
		val, err := Encrypt(s.gcm, []byte(v))
		if err != nil {
			log.Debugf("failed to encrypt message header %s: %v", k, err)
			return err
		}
		// add the encrypted header value
		data.Header[k] = base32.StdEncoding.EncodeToString(val)
	}

	// create a new message
	msg := s.newMessage("session")
	// set the data
	msg.data = data

	// if multicast don't set the link
	if s.mode != Unicast {
		msg.link = ""
	}

	if logger.V(logger.TraceLevel, log) {
		log.Tracef("Appending to send backlog: %v", msg)
	}
	// send the actual message
	if err := s.sendMsg(msg); err != nil {
		return err
	}

	// wait for an error response
	return s.wait(msg)
}

// Recv is used to receive a message
func (s *session) Recv(m *transport.Message) error {
	var msg *message

	msg, err := s.waitFor("", s.readTimeout)
	if err != nil {
		return err
	}

	// check the error if one exists
	select {
	case err := <-msg.errChan:
		return err
	default:
	}

	if logger.V(logger.TraceLevel, log) {
		log.Tracef("Received from recv backlog: %v", msg)
	}

	gcm, err := newCipher([]byte(s.token + s.channel + msg.session))
	if err != nil {
		if logger.V(logger.ErrorLevel, log) {
			log.Errorf("unable to create cipher: %v", err)
		}
		return err
	}

	// decrypt the received payload using the token
	// we have to used msg.session because multicast has a shared
	// session id of "multicast" in this session struct on
	// the listener side
	msg.data.Body, err = Decrypt(gcm, msg.data.Body)
	if err != nil {
		if logger.V(logger.DebugLevel, log) {
			log.Debugf("failed to decrypt message body: %v", err)
		}
		return err
	}

	// dencrypt all the headers
	for k, v := range msg.data.Header {
		// decode the header values
		h, err := base32.StdEncoding.DecodeString(v)
		if err != nil {
			if logger.V(logger.DebugLevel, log) {
				log.Debugf("failed to decode message header %s: %v", k, err)
			}
			return err
		}

		// dencrypt the transport message payload
		val, err := Decrypt(gcm, h)
		if err != nil {
			if logger.V(logger.DebugLevel, log) {
				log.Debugf("failed to decrypt message header %s: %v", k, err)
			}
			return err
		}
		// add decrypted header value
		msg.data.Header[k] = string(val)
	}

	// set the link
	// TODO: decruft, this is only for multicast
	// since the session is now a single session
	// likely provide as part of message.Link()
	msg.data.Header["Micro-Link"] = msg.link

	// set message
	*m = *msg.data
	// return nil
	return nil
}

// Close closes the session by sending a close message
func (s *session) Close() error {
	select {
	case <-s.closed:
		// no op
	default:
		close(s.closed)

		// don't send close on multicast or broadcast
		if s.mode != Unicast {
			return nil
		}

		// append to backlog
		msg := s.newMessage("close")
		// no error response on close
		msg.errChan = nil

		// send the close message
		select {
		case s.send <- msg:
		case <-time.After(time.Millisecond * 10):
		}
	}

	return nil
}