micro/tunnel/session.go

503 lines
10 KiB
Go
Raw Normal View History

2019-08-06 11:45:25 +01:00
package tunnel
2019-08-07 18:44:33 +01:00
import (
"crypto/cipher"
"encoding/base32"
2019-08-30 20:05:00 +01:00
"io"
"sync"
"time"
2019-08-06 11:45:25 +01:00
"github.com/micro/go-micro/v3/logger"
"github.com/micro/go-micro/v3/transport"
2019-08-07 18:44:33 +01:00
)
2019-08-06 11:45:25 +01:00
2019-08-30 20:05:00 +01:00
// session is our pseudo session for transport.Socket
type session struct {
2019-09-03 15:56:37 +01:00
// the tunnel id
tunnel string
2019-08-30 20:05:00 +01:00
// the channel name
channel string
2019-08-07 18:44:33 +01:00
// the session id based on Micro.Tunnel-Session
session string
// token is the session token
token string
2019-08-07 18:44:33 +01:00
// 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
2019-08-30 20:05:00 +01:00
// outbound marks the session as outbound dialled connection
2019-08-14 17:14:39 +01:00
outbound bool
2019-08-30 20:05:00 +01:00
// lookback marks the session as a loopback on the inbound
loopback bool
2019-10-15 15:40:04 +01:00
// mode of the connection
mode Mode
2019-12-07 23:28:39 +00:00
// the dial timeout
dialTimeout time.Duration
// the read timeout
readTimeout time.Duration
2019-08-29 12:42:27 +01:00
// the link on which this message was received
link string
2019-08-30 20:05:00 +01:00
// the error response
errChan chan error
// key for session encryption
key []byte
// cipher for session
gcm cipher.AEAD
sync.RWMutex
2019-08-06 11:45:25 +01:00
}
2019-08-07 18:44:33 +01:00
// message is sent over the send channel
type message struct {
2019-08-29 12:42:27 +01:00
// type of message
typ string
2019-08-07 18:44:33 +01:00
// tunnel id
2019-09-03 15:56:37 +01:00
tunnel string
2019-08-30 20:05:00 +01:00
// channel name
channel string
2019-08-07 18:44:33 +01:00
// the session id
session string
2019-08-14 17:14:39 +01:00
// outbound marks the message as outbound
outbound bool
// loopback marks the message intended for loopback
loopback bool
2019-10-15 15:40:04 +01:00
// mode of the connection
mode Mode
2019-08-29 12:42:27 +01:00
// the link to send the message on
link string
2019-08-07 18:44:33 +01:00
// transport data
data *transport.Message
2019-08-30 20:05:00 +01:00
// the error channel
errChan chan error
2019-08-07 18:44:33 +01:00
}
2019-08-30 20:05:00 +01:00
func (s *session) Remote() string {
2019-08-07 18:44:33 +01:00
return s.remote
}
2019-08-30 20:05:00 +01:00
func (s *session) Local() string {
2019-08-07 18:44:33 +01:00
return s.local
2019-08-06 11:45:25 +01:00
}
2019-09-11 12:07:43 -07:00
func (s *session) Link() string {
return s.link
}
2019-08-30 20:05:00 +01:00
func (s *session) Id() string {
return s.session
2019-08-07 18:44:33 +01:00
}
2019-08-30 20:05:00 +01:00
func (s *session) Channel() string {
return s.channel
2019-08-07 18:44:33 +01:00
}
2019-09-04 12:18:37 +01:00
// newMessage creates a new message based on the session
2019-09-04 12:16:31 +01:00
func (s *session) newMessage(typ string) *message {
return &message{
2019-10-15 15:40:04 +01:00
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,
}
2019-09-04 12:16:31 +01:00
}
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()
2019-12-07 23:28:39 +00:00
after := func(timeout time.Duration) <-chan time.Time {
if timeout < time.Duration(0) {
return nil
}
// get the delta
d := time.Since(now)
2019-12-07 23:28:39 +00:00
// dial timeout minus time since
wait := timeout - d
if wait < time.Duration(0) {
2019-12-07 23:28:39 +00:00
wait = time.Duration(0)
}
2019-12-07 23:28:39 +00:00
return time.After(wait)
}
// wait for the message type
for {
select {
case msg := <-s.recv:
2019-12-07 23:28:39 +00:00
// 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
}
2019-12-07 23:28:39 +00:00
// got the message
return msg, nil
2019-12-07 23:28:39 +00:00
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
2019-12-07 23:28:39 +00:00
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
}
2019-09-04 12:16:31 +01:00
// 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.
2019-09-04 12:16:31 +01:00
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
2019-12-07 23:28:39 +00:00
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
}
2019-09-04 11:58:03 +01:00
// Accept sends the accept response to an open message from a dialled connection
func (s *session) Accept() error {
2019-09-04 12:16:31 +01:00
msg := s.newMessage("accept")
// send the accept message
if err := s.sendMsg(msg); err != nil {
return err
2019-10-15 15:40:04 +01:00
}
// 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.
2019-09-04 11:58:03 +01:00
func (s *session) Announce() error {
2019-09-04 12:16:31 +01:00
msg := s.newMessage("announce")
// we don't need an error back
msg.errChan = nil
// announce to all
2019-10-15 15:40:04 +01:00
msg.mode = Broadcast
2019-09-04 12:16:31 +01:00
// we don't need the link
msg.link = ""
2019-09-04 11:58:03 +01:00
// send announce message
return s.sendMsg(msg)
2019-09-04 11:58:03 +01:00
}
// Send is used to send a message
2019-08-30 20:05:00 +01:00
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()
}
2019-11-25 15:34:41 +00:00
// encrypt the transport message payload
body, err := Encrypt(gcm, m.Body)
2019-11-25 15:34:41 +00:00
if err != nil {
log.Debugf("failed to encrypt message body: %v", err)
2019-11-25 15:34:41 +00:00
return err
}
// make copy, without rehash and realloc
2019-08-11 18:11:33 +01:00
data := &transport.Message{
Header: make(map[string]string, len(m.Header)),
2019-11-25 15:34:41 +00:00
Body: body,
2019-08-11 18:11:33 +01:00
}
// encrypt all the headers
2019-08-11 18:11:33 +01:00
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)
2019-08-11 18:11:33 +01:00
}
2019-09-04 12:16:31 +01:00
// create a new message
msg := s.newMessage("session")
// set the data
msg.data = data
2019-09-04 12:16:31 +01:00
// if multicast don't set the link
if s.mode != Unicast {
2019-09-04 12:16:31 +01:00
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 {
2019-08-30 20:05:00 +01:00
return err
}
// wait for an error response
return s.wait(msg)
2019-08-06 11:45:25 +01:00
}
2019-09-04 11:58:03 +01:00
// Recv is used to receive a message
2019-08-30 20:05:00 +01:00
func (s *session) Recv(m *transport.Message) error {
2019-12-03 08:11:36 +00:00
var msg *message
2019-12-07 23:28:39 +00:00
msg, err := s.waitFor("", s.readTimeout)
if err != nil {
return err
2019-12-03 08:11:36 +00:00
}
2019-08-30 20:05:00 +01:00
// 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
}
2019-11-25 15:34:41 +00:00
// decrypt the received payload using the token
2019-12-06 00:18:40 +00:00
// we have to used msg.session because multicast has a shared
// session id of "multicast" in this session struct on
2019-12-06 00:18:40 +00:00
// the listener side
msg.data.Body, err = Decrypt(gcm, msg.data.Body)
2019-11-25 15:34:41 +00:00
if err != nil {
if logger.V(logger.DebugLevel, log) {
log.Debugf("failed to decrypt message body: %v", err)
}
2019-11-25 15:34:41 +00:00
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)
}
2019-12-08 00:53:55 +00:00
// 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
2019-08-07 18:44:33 +01:00
// set message
*m = *msg.data
// return nil
return nil
2019-08-06 11:45:25 +01:00
}
2019-09-04 11:58:03 +01:00
// Close closes the session by sending a close message
2019-08-30 20:05:00 +01:00
func (s *session) Close() error {
2019-08-07 18:44:33 +01:00
select {
case <-s.closed:
// no op
default:
close(s.closed)
2019-12-08 12:12:20 +00:00
// don't send close on multicast or broadcast
if s.mode != Unicast {
return nil
}
// append to backlog
2019-09-04 12:16:31 +01:00
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):
}
2019-08-07 18:44:33 +01:00
}
2019-08-07 18:44:33 +01:00
return nil
2019-08-06 11:45:25 +01:00
}