From 7bd50cd25171d17f6b4d08a69fb6616d803ee31f Mon Sep 17 00:00:00 2001
From: Asim Aslam <asim@aslam.me>
Date: Thu, 12 Dec 2019 17:10:32 +0000
Subject: [PATCH] fix more broken cruft
---
network/default.go | 6 ++++--
tunnel/crypto.go | 5 +++++
tunnel/default.go | 14 +++-----------
tunnel/listener.go | 19 +++++++++++++++----
tunnel/tunnel.go | 2 ++
5 files changed, 29 insertions(+), 17 deletions(-)
diff --git a/network/default.go b/network/default.go
index a96efad2..b7a61ec1 100644
--- a/network/default.go
+++ b/network/default.go
@@ -331,7 +331,9 @@ func (n *network) advertise(advertChan <-chan *router.Advert) {
// someone requested the route
n.sendTo("advert", ControlChannel, peer, msg)
default:
- // no one to send to
+ if err := n.sendMsg("advert", ControlChannel, msg); err != nil {
+ log.Debugf("Network failed to advertise routes: %v", err)
+ }
}
case <-n.closed:
return
@@ -958,7 +960,7 @@ func (n *network) manage() {
return
case <-announce.C:
// jitter
- j := rand.Int63n(int64(AnnounceTime / 2))
+ j := rand.Int63n(int64(AnnounceTime.Seconds() / 2.0))
time.Sleep(time.Duration(j) * time.Second)
// TODO: intermittently flip between peer selection
diff --git a/tunnel/crypto.go b/tunnel/crypto.go
index ba0d5057..9f9f45af 100644
--- a/tunnel/crypto.go
+++ b/tunnel/crypto.go
@@ -60,6 +60,11 @@ func Decrypt(data []byte, key string) ([]byte, error) {
}
nonceSize := gcm.NonceSize()
+
+ if len(data) < nonceSize {
+ return nil, ErrDecryptingData
+ }
+
// NOTE: we need to parse out nonce from the payload
// we prepend the nonce to every encrypted payload
nonce, ciphertext := data[:nonceSize], data[nonceSize:]
diff --git a/tunnel/default.go b/tunnel/default.go
index 0b47007e..36781b53 100644
--- a/tunnel/default.go
+++ b/tunnel/default.go
@@ -229,7 +229,7 @@ func (t *tun) manageLink(link *link) {
wait := func(d time.Duration) {
// jitter
- j := rand.Int63n(int64(d / 2))
+ j := rand.Int63n(int64(d.Seconds() / 2.0))
time.Sleep(time.Duration(j) * time.Second)
}
@@ -387,15 +387,6 @@ func (t *tun) process() {
for {
select {
case msg := <-t.send:
- // no links yet
- if len(links) == 0 {
- // TODO: should we block here rather than throwing away messages...
- // Or should we return an error?
- log.Debugf("No links to send message type: %s channel: %s", msg.typ, msg.channel)
- time.Sleep(time.Millisecond * 100)
- continue
- }
-
// build a list of links to send to
var sendTo []*link
var err error
@@ -456,7 +447,7 @@ func (t *tun) process() {
// no links to send to
if len(sendTo) == 0 {
- log.Log("no links")
+ log.Debugf("No links to send message type: %s channel: %s", msg.typ, msg.channel)
t.respond(msg, err)
continue
}
@@ -532,6 +523,7 @@ func (t *tun) sendTo(links []*link, msg *message) error {
// make a copy
m := &transport.Message{
Header: make(map[string]string),
+ Body: make([]byte, len(newMsg.Body)),
}
copy(m.Body, newMsg.Body)
for k, v := range newMsg.Header {
diff --git a/tunnel/listener.go b/tunnel/listener.go
index 16f37e35..052955cb 100644
--- a/tunnel/listener.go
+++ b/tunnel/listener.go
@@ -2,6 +2,7 @@ package tunnel
import (
"io"
+ "sync"
"github.com/micro/go-micro/util/log"
)
@@ -13,14 +14,16 @@ type tunListener struct {
token string
// the accept channel
accept chan *session
- // the channel to close
- closed chan bool
// the tunnel closed channel
tunClosed chan bool
// the listener session
session *session
// del func to kill listener
delFunc func()
+
+ sync.RWMutex
+ // the channel to close
+ closed chan bool
}
func (t *tunListener) process() {
@@ -49,7 +52,7 @@ func (t *tunListener) process() {
var sessionId string
var linkId string
- switch m.mode {
+ switch t.session.mode {
case Multicast:
sessionId = "multicast"
linkId = "multicast"
@@ -87,7 +90,7 @@ func (t *tunListener) process() {
// the link the message was received on
link: linkId,
// set the connection mode
- mode: m.mode,
+ mode: t.session.mode,
// close chan
closed: make(chan bool),
// recv called by the acceptor
@@ -115,6 +118,11 @@ func (t *tunListener) process() {
switch m.typ {
case "close":
+ // don't close multicast sessions
+ if sess.mode > Unicast {
+ continue
+ }
+
// received a close message
select {
// check if the session is closed
@@ -154,6 +162,9 @@ func (t *tunListener) Channel() string {
// Close closes tunnel listener
func (t *tunListener) Close() error {
+ t.Lock()
+ defer t.Unlock()
+
select {
case <-t.closed:
return nil
diff --git a/tunnel/tunnel.go b/tunnel/tunnel.go
index 15ff6078..b59eb31d 100644
--- a/tunnel/tunnel.go
+++ b/tunnel/tunnel.go
@@ -28,6 +28,8 @@ var (
ErrLinkNotFound = errors.New("link not found")
// ErrReadTimeout is a timeout on session.Recv
ErrReadTimeout = errors.New("read timeout")
+ // ErrDecryptingData is for when theres a nonce error
+ ErrDecryptingData = errors.New("error decrypting data")
)
// Mode of the session