From f990a56046646ef97322c910b8cd057e99ad035d Mon Sep 17 00:00:00 2001 From: Arceliar Date: Mon, 5 Jul 2021 13:14:12 -0500 Subject: [PATCH 1/5] have the core wrap and export the underlying PacketConn, move IPv6 ReadWriteCloser wrapper logic to a separate package --- cmd/yggdrasil/main.go | 4 +- src/core/api.go | 42 +------ src/core/core.go | 62 ++++++++- src/core/link.go | 2 +- src/core/nodeinfo.go | 4 +- src/core/proto.go | 9 +- src/core/types.go | 7 -- src/{core => ipv6rwc}/icmpv6.go | 2 +- src/{core/keystore.go => ipv6rwc/ipv6rwc.go} | 125 +++++++++++++------ src/tuntap/iface.go | 4 +- src/tuntap/tun.go | 19 ++- 11 files changed, 170 insertions(+), 110 deletions(-) rename src/{core => ipv6rwc}/icmpv6.go (99%) rename src/{core/keystore.go => ipv6rwc/ipv6rwc.go} (77%) diff --git a/cmd/yggdrasil/main.go b/cmd/yggdrasil/main.go index d6d0d1a..95d4015 100644 --- a/cmd/yggdrasil/main.go +++ b/cmd/yggdrasil/main.go @@ -29,6 +29,7 @@ import ( "github.com/yggdrasil-network/yggdrasil-go/src/defaults" "github.com/yggdrasil-network/yggdrasil-go/src/core" + "github.com/yggdrasil-network/yggdrasil-go/src/ipv6rwc" "github.com/yggdrasil-network/yggdrasil-go/src/multicast" "github.com/yggdrasil-network/yggdrasil-go/src/tuntap" "github.com/yggdrasil-network/yggdrasil-go/src/version" @@ -353,7 +354,8 @@ func run(args yggArgs, ctx context.Context, done chan struct{}) { } n.multicast.SetupAdminHandlers(n.admin) // Start the TUN/TAP interface - if err := n.tuntap.Init(&n.core, cfg, logger, nil); err != nil { + rwc := ipv6rwc.NewReadWriteCloser(&n.core) + if err := n.tuntap.Init(rwc, cfg, logger, nil); err != nil { logger.Errorln("An error occurred initialising TUN/TAP:", err) } else if err := n.tuntap.Start(); err != nil { logger.Errorln("An error occurred starting TUN/TAP:", err) diff --git a/src/core/api.go b/src/core/api.go index 05d9f36..6b21934 100644 --- a/src/core/api.go +++ b/src/core/api.go @@ -48,7 +48,7 @@ type Session struct { func (c *Core) GetSelf() Self { var self Self - s := c.pc.PacketConn.Debug.GetSelf() + s := c.PacketConn.PacketConn.Debug.GetSelf() self.Key = s.Key self.Root = s.Root self.Coords = s.Coords @@ -63,7 +63,7 @@ func (c *Core) GetPeers() []Peer { names[info.conn] = info.lname } c.links.mutex.Unlock() - ps := c.pc.PacketConn.Debug.GetPeers() + ps := c.PacketConn.PacketConn.Debug.GetPeers() for _, p := range ps { var info Peer info.Key = p.Key @@ -81,7 +81,7 @@ func (c *Core) GetPeers() []Peer { func (c *Core) GetDHT() []DHTEntry { var dhts []DHTEntry - ds := c.pc.PacketConn.Debug.GetDHT() + ds := c.PacketConn.PacketConn.Debug.GetDHT() for _, d := range ds { var info DHTEntry info.Key = d.Key @@ -94,7 +94,7 @@ func (c *Core) GetDHT() []DHTEntry { func (c *Core) GetPaths() []PathEntry { var paths []PathEntry - ps := c.pc.PacketConn.Debug.GetPaths() + ps := c.PacketConn.PacketConn.Debug.GetPaths() for _, p := range ps { var info PathEntry info.Key = p.Key @@ -106,7 +106,7 @@ func (c *Core) GetPaths() []PathEntry { func (c *Core) GetSessions() []Session { var sessions []Session - ss := c.pc.Debug.GetSessions() + ss := c.PacketConn.Debug.GetSessions() for _, s := range ss { var info Session info.Key = s.Key @@ -239,38 +239,6 @@ func (c *Core) PublicKey() ed25519.PublicKey { return c.public } -func (c *Core) MaxMTU() uint64 { - return c.store.maxSessionMTU() -} - -func (c *Core) SetMTU(mtu uint64) { - if mtu < 1280 { - mtu = 1280 - } - c.store.mutex.Lock() - c.store.mtu = mtu - c.store.mutex.Unlock() -} - -func (c *Core) MTU() uint64 { - c.store.mutex.Lock() - mtu := c.store.mtu - c.store.mutex.Unlock() - return mtu -} - -// Implement io.ReadWriteCloser - -func (c *Core) Read(p []byte) (n int, err error) { - n, err = c.store.readPC(p) - return -} - -func (c *Core) Write(p []byte) (n int, err error) { - n, err = c.store.writePC(p) - return -} - func (c *Core) Close() error { c.Stop() return nil diff --git a/src/core/core.go b/src/core/core.go index 89d4917..ac0ea1f 100644 --- a/src/core/core.go +++ b/src/core/core.go @@ -7,10 +7,12 @@ import ( "errors" "fmt" "io/ioutil" + "net" "net/url" "time" - iw "github.com/Arceliar/ironwood/encrypted" + iwe "github.com/Arceliar/ironwood/encrypted" + iwt "github.com/Arceliar/ironwood/types" "github.com/Arceliar/phony" "github.com/gologme/log" @@ -26,13 +28,12 @@ type Core struct { // We're going to keep our own copy of the provided config - that way we can // guarantee that it will be covered by the mutex phony.Inbox - pc *iw.PacketConn + *iwe.PacketConn config *config.NodeConfig // Config secret ed25519.PrivateKey public ed25519.PublicKey links links proto protoHandler - store keyStore log *log.Logger addPeerTimer *time.Timer ctx context.Context @@ -62,9 +63,8 @@ func (c *Core) _init() error { c.public = c.secret.Public().(ed25519.PublicKey) // TODO check public against current.PublicKey, error if they don't match - c.pc, err = iw.NewPacketConn(c.secret) + c.PacketConn, err = iwe.NewPacketConn(c.secret) c.ctx, c.ctxCancel = context.WithCancel(context.Background()) - c.store.init(c) c.proto.init(c) if err := c.proto.nodeinfo.setNodeInfo(c.config.NodeInfo, c.config.NodeInfoPrivacy); err != nil { return fmt.Errorf("setNodeInfo: %w", err) @@ -168,7 +168,7 @@ func (c *Core) Stop() { func (c *Core) _stop() { c.log.Infoln("Stopping...") c.ctxCancel() - c.pc.Close() + c.PacketConn.Close() // TODO make c.Close() do the right thing (act like c.Stop()) if c.addPeerTimer != nil { c.addPeerTimer.Stop() c.addPeerTimer = nil @@ -181,3 +181,53 @@ func (c *Core) _stop() { */ c.log.Infoln("Stopped") } + +func (c *Core) MTU() uint64 { + const sessionTypeOverhead = 1 + return c.PacketConn.MTU() - sessionTypeOverhead +} + +func (c *Core) ReadFrom(p []byte) (n int, from net.Addr, err error) { + buf := make([]byte, c.PacketConn.MTU(), 65535) + for { + bs := buf + n, from, err = c.PacketConn.ReadFrom(bs) + if err != nil { + return 0, from, err + } + if n == 0 { + continue + } + switch bs[0] { + case typeSessionTraffic: + // This is what we want to handle here + case typeSessionProto: + var key keyArray + copy(key[:], from.(iwt.Addr)) + data := append([]byte(nil), bs[1:n]...) + c.proto.handleProto(nil, key, data) + continue + default: + continue + } + bs = bs[1:n] + copy(p, bs) + if len(p) < len(bs) { + n = len(p) + } else { + n = len(bs) + } + return + } +} + +func (c *Core) WriteTo(p []byte, addr net.Addr) (n int, err error) { + buf := make([]byte, 0, 65535) + buf = append(buf, typeSessionTraffic) + buf = append(buf, p...) + n, err = c.PacketConn.WriteTo(buf, addr) + if n > 0 { + n -= 1 + } + return +} diff --git a/src/core/link.go b/src/core/link.go index 165b18b..ccab921 100644 --- a/src/core/link.go +++ b/src/core/link.go @@ -230,7 +230,7 @@ func (intf *link) handler() (chan struct{}, error) { intf.links.core.log.Infof("Connected %s: %s, source %s", strings.ToUpper(intf.info.linkType), themString, intf.info.local) // Run the handler - err = intf.links.core.pc.HandleConn(ed25519.PublicKey(intf.info.key[:]), intf.conn) + err = intf.links.core.HandleConn(ed25519.PublicKey(intf.info.key[:]), intf.conn) // TODO don't report an error if it's just a 'use of closed network connection' if err != nil { intf.links.core.log.Infof("Disconnected %s: %s, source %s; error: %s", diff --git a/src/core/nodeinfo.go b/src/core/nodeinfo.go index 3064471..9015311 100644 --- a/src/core/nodeinfo.go +++ b/src/core/nodeinfo.go @@ -129,7 +129,7 @@ func (m *nodeinfo) _sendReq(key keyArray, callback func(nodeinfo NodeInfoPayload if callback != nil { m._addCallback(key, callback) } - _, _ = m.proto.core.pc.WriteTo([]byte{typeSessionProto, typeProtoNodeInfoRequest}, iwt.Addr(key[:])) + _, _ = m.proto.core.WriteTo([]byte{typeSessionProto, typeProtoNodeInfoRequest}, iwt.Addr(key[:])) } func (m *nodeinfo) handleReq(from phony.Actor, key keyArray) { @@ -146,7 +146,7 @@ func (m *nodeinfo) handleRes(from phony.Actor, key keyArray, info NodeInfoPayloa func (m *nodeinfo) _sendRes(key keyArray) { bs := append([]byte{typeSessionProto, typeProtoNodeInfoResponse}, m._getNodeInfo()...) - _, _ = m.proto.core.pc.WriteTo(bs, iwt.Addr(key[:])) + _, _ = m.proto.core.WriteTo(bs, iwt.Addr(key[:])) } // Admin socket stuff diff --git a/src/core/proto.go b/src/core/proto.go index 557ac1d..f9d12c8 100644 --- a/src/core/proto.go +++ b/src/core/proto.go @@ -1,6 +1,7 @@ package core import ( + "crypto/ed25519" "encoding/hex" "encoding/json" "errors" @@ -29,6 +30,8 @@ type reqInfo struct { timer *time.Timer // time.AfterFunc cleanup } +type keyArray [ed25519.PublicKeySize]byte + type protoHandler struct { phony.Inbox core *Core @@ -149,7 +152,7 @@ func (p *protoHandler) _handleGetPeersRequest(key keyArray) { for _, pinfo := range peers { tmp := append(bs, pinfo.Key[:]...) const responseOverhead = 2 // 1 debug type, 1 getpeers type - if uint64(len(tmp))+responseOverhead > p.core.store.maxSessionMTU() { + if uint64(len(tmp))+responseOverhead > p.core.MTU() { break } bs = tmp @@ -191,7 +194,7 @@ func (p *protoHandler) _handleGetDHTRequest(key keyArray) { for _, dinfo := range dinfos { tmp := append(bs, dinfo.Key[:]...) const responseOverhead = 2 // 1 debug type, 1 getdht type - if uint64(len(tmp))+responseOverhead > p.core.store.maxSessionMTU() { + if uint64(len(tmp))+responseOverhead > p.core.MTU() { break } bs = tmp @@ -209,7 +212,7 @@ func (p *protoHandler) _handleGetDHTResponse(key keyArray, bs []byte) { func (p *protoHandler) _sendDebug(key keyArray, dType uint8, data []byte) { bs := append([]byte{typeSessionProto, typeProtoDebug, dType}, data...) - _, _ = p.core.pc.WriteTo(bs, iwt.Addr(key[:])) + _, _ = p.core.WriteTo(bs, iwt.Addr(key[:])) } // Admin socket stuff diff --git a/src/core/types.go b/src/core/types.go index e325b55..258563a 100644 --- a/src/core/types.go +++ b/src/core/types.go @@ -1,12 +1,5 @@ package core -// Out-of-band packet types -const ( - typeKeyDummy = iota // nolint:deadcode,varcheck - typeKeyLookup - typeKeyResponse -) - // In-band packet types const ( typeSessionDummy = iota // nolint:deadcode,varcheck diff --git a/src/core/icmpv6.go b/src/ipv6rwc/icmpv6.go similarity index 99% rename from src/core/icmpv6.go rename to src/ipv6rwc/icmpv6.go index d15fbbc..8faf1d5 100644 --- a/src/core/icmpv6.go +++ b/src/ipv6rwc/icmpv6.go @@ -1,4 +1,4 @@ -package core +package ipv6rwc // The ICMPv6 module implements functions to easily create ICMPv6 // packets. These functions, when mixed with the built-in Go IPv6 diff --git a/src/core/keystore.go b/src/ipv6rwc/ipv6rwc.go similarity index 77% rename from src/core/keystore.go rename to src/ipv6rwc/ipv6rwc.go index 21fb845..1c715f0 100644 --- a/src/core/keystore.go +++ b/src/ipv6rwc/ipv6rwc.go @@ -1,4 +1,4 @@ -package core +package ipv6rwc import ( "crypto/ed25519" @@ -14,14 +14,22 @@ import ( iwt "github.com/Arceliar/ironwood/types" "github.com/yggdrasil-network/yggdrasil-go/src/address" + "github.com/yggdrasil-network/yggdrasil-go/src/core" ) const keyStoreTimeout = 2 * time.Minute +// Out-of-band packet types +const ( + typeKeyDummy = iota // nolint:deadcode,varcheck + typeKeyLookup + typeKeyResponse +) + type keyArray [ed25519.PublicKeySize]byte type keyStore struct { - core *Core + core *core.Core address address.Address subnet address.Subnet mutex sync.Mutex @@ -45,11 +53,11 @@ type buffer struct { timeout *time.Timer } -func (k *keyStore) init(core *Core) { - k.core = core - k.address = *address.AddrForKey(k.core.public) - k.subnet = *address.SubnetForKey(k.core.public) - if err := k.core.pc.SetOutOfBandHandler(k.oobHandler); err != nil { +func (k *keyStore) init(c *core.Core) { + k.core = c + k.address = *address.AddrForKey(k.core.PublicKey()) + k.subnet = *address.SubnetForKey(k.core.PublicKey()) + if err := k.core.SetOutOfBandHandler(k.oobHandler); err != nil { err = fmt.Errorf("tun.core.SetOutOfBandHander: %w", err) panic(err) } @@ -66,7 +74,7 @@ func (k *keyStore) sendToAddress(addr address.Address, bs []byte) { if info := k.addrToInfo[addr]; info != nil { k.resetTimeout(info) k.mutex.Unlock() - _, _ = k.core.pc.WriteTo(bs, iwt.Addr(info.key[:])) + _, _ = k.core.WriteTo(bs, iwt.Addr(info.key[:])) } else { var buf *buffer if buf = k.addrBuffer[addr]; buf == nil { @@ -95,7 +103,7 @@ func (k *keyStore) sendToSubnet(subnet address.Subnet, bs []byte) { if info := k.subnetToInfo[subnet]; info != nil { k.resetTimeout(info) k.mutex.Unlock() - _, _ = k.core.pc.WriteTo(bs, iwt.Addr(info.key[:])) + _, _ = k.core.WriteTo(bs, iwt.Addr(info.key[:])) } else { var buf *buffer if buf = k.subnetBuffer[subnet]; buf == nil { @@ -135,11 +143,11 @@ func (k *keyStore) update(key ed25519.PublicKey) *keyInfo { k.resetTimeout(info) k.mutex.Unlock() if buf := k.addrBuffer[info.address]; buf != nil { - k.core.pc.WriteTo(buf.packet, iwt.Addr(info.key[:])) + k.core.WriteTo(buf.packet, iwt.Addr(info.key[:])) delete(k.addrBuffer, info.address) } if buf := k.subnetBuffer[info.subnet]; buf != nil { - k.core.pc.WriteTo(buf.packet, iwt.Addr(info.key[:])) + k.core.WriteTo(buf.packet, iwt.Addr(info.key[:])) delete(k.subnetBuffer, info.subnet) } } else { @@ -191,46 +199,29 @@ func (k *keyStore) oobHandler(fromKey, toKey ed25519.PublicKey, data []byte) { } func (k *keyStore) sendKeyLookup(partial ed25519.PublicKey) { - sig := ed25519.Sign(k.core.secret, partial[:]) + sig := ed25519.Sign(k.core.PrivateKey(), partial[:]) bs := append([]byte{typeKeyLookup}, sig...) - _ = k.core.pc.SendOutOfBand(partial, bs) + _ = k.core.SendOutOfBand(partial, bs) } func (k *keyStore) sendKeyResponse(dest ed25519.PublicKey) { - sig := ed25519.Sign(k.core.secret, dest[:]) + sig := ed25519.Sign(k.core.PrivateKey(), dest[:]) bs := append([]byte{typeKeyResponse}, sig...) - _ = k.core.pc.SendOutOfBand(dest, bs) -} - -func (k *keyStore) maxSessionMTU() uint64 { - const sessionTypeOverhead = 1 - return k.core.pc.MTU() - sessionTypeOverhead + _ = k.core.SendOutOfBand(dest, bs) } func (k *keyStore) readPC(p []byte) (int, error) { - buf := make([]byte, k.core.pc.MTU(), 65535) + buf := make([]byte, k.core.MTU(), 65535) for { bs := buf - n, from, err := k.core.pc.ReadFrom(bs) + n, from, err := k.core.ReadFrom(bs) if err != nil { return n, err } if n == 0 { continue } - switch bs[0] { - case typeSessionTraffic: - // This is what we want to handle here - case typeSessionProto: - var key keyArray - copy(key[:], from.(iwt.Addr)) - data := append([]byte(nil), bs[1:n]...) - k.core.proto.handleProto(nil, key, data) - continue - default: - continue - } - bs = bs[1:n] + bs = bs[:n] if len(bs) == 0 { continue } @@ -294,15 +285,69 @@ func (k *keyStore) writePC(bs []byte) (int, error) { strErr := fmt.Sprint("incorrect source address: ", net.IP(srcAddr[:]).String()) return 0, errors.New(strErr) } - buf := make([]byte, 1+len(bs), 65535) - buf[0] = typeSessionTraffic - copy(buf[1:], bs) if dstAddr.IsValid() { - k.sendToAddress(dstAddr, buf) + k.sendToAddress(dstAddr, bs) } else if dstSubnet.IsValid() { - k.sendToSubnet(dstSubnet, buf) + k.sendToSubnet(dstSubnet, bs) } else { return 0, errors.New("invalid destination address") } return len(bs), nil } + +// Exported API + +func (k *keyStore) MaxMTU() uint64 { + return k.core.MTU() +} + +func (k *keyStore) SetMTU(mtu uint64) { + if mtu > k.MaxMTU() { + mtu = k.MaxMTU() + } + if mtu < 1280 { + mtu = 1280 + } + k.mutex.Lock() + k.mtu = mtu + k.mutex.Unlock() +} + +func (k *keyStore) MTU() uint64 { + k.mutex.Lock() + mtu := k.mtu + k.mutex.Unlock() + return mtu +} + +type ReadWriteCloser struct { + keyStore +} + +func NewReadWriteCloser(c *core.Core) *ReadWriteCloser { + rwc := new(ReadWriteCloser) + rwc.init(c) + return rwc +} + +func (rwc *ReadWriteCloser) Address() address.Address { + return rwc.address +} + +func (rwc *ReadWriteCloser) Subnet() address.Subnet { + return rwc.subnet +} + +func (rwc *ReadWriteCloser) Read(p []byte) (n int, err error) { + return rwc.readPC(p) +} + +func (rwc *ReadWriteCloser) Write(p []byte) (n int, err error) { + return rwc.writePC(p) +} + +func (rwc *ReadWriteCloser) Close() error { + err := rwc.core.Close() + rwc.core.Stop() + return err +} diff --git a/src/tuntap/iface.go b/src/tuntap/iface.go index e72b091..f629399 100644 --- a/src/tuntap/iface.go +++ b/src/tuntap/iface.go @@ -17,7 +17,7 @@ func (tun *TunAdapter) read() { begin := TUN_OFFSET_BYTES end := begin + n bs := buf[begin:end] - if _, err := tun.core.Write(bs); err != nil { + if _, err := tun.rwc.Write(bs); err != nil { tun.log.Debugln("Unable to send packet:", err) } } @@ -27,7 +27,7 @@ func (tun *TunAdapter) write() { var buf [TUN_OFFSET_BYTES + 65535]byte for { bs := buf[TUN_OFFSET_BYTES:] - n, err := tun.core.Read(bs) + n, err := tun.rwc.Read(bs) if err != nil { tun.log.Errorln("Exiting tun writer due to core read error:", err) return diff --git a/src/tuntap/tun.go b/src/tuntap/tun.go index dbba018..4caefe4 100644 --- a/src/tuntap/tun.go +++ b/src/tuntap/tun.go @@ -21,8 +21,8 @@ import ( "github.com/yggdrasil-network/yggdrasil-go/src/address" "github.com/yggdrasil-network/yggdrasil-go/src/config" - "github.com/yggdrasil-network/yggdrasil-go/src/core" "github.com/yggdrasil-network/yggdrasil-go/src/defaults" + "github.com/yggdrasil-network/yggdrasil-go/src/ipv6rwc" ) type MTU uint16 @@ -32,7 +32,7 @@ type MTU uint16 // should pass this object to the yggdrasil.SetRouterAdapter() function before // calling yggdrasil.Start(). type TunAdapter struct { - core *core.Core + rwc *ipv6rwc.ReadWriteCloser config *config.NodeConfig log *log.Logger addr address.Address @@ -93,8 +93,8 @@ func MaximumMTU() uint64 { // Init initialises the TUN module. You must have acquired a Listener from // the Yggdrasil core before this point and it must not be in use elsewhere. -func (tun *TunAdapter) Init(core *core.Core, config *config.NodeConfig, log *log.Logger, options interface{}) error { - tun.core = core +func (tun *TunAdapter) Init(rwc *ipv6rwc.ReadWriteCloser, config *config.NodeConfig, log *log.Logger, options interface{}) error { + tun.rwc = rwc tun.config = config tun.log = log return nil @@ -119,9 +119,8 @@ func (tun *TunAdapter) _start() error { if tun.config == nil { return errors.New("no configuration available to TUN") } - pk := tun.core.PublicKey() - tun.addr = *address.AddrForKey(pk) - tun.subnet = *address.SubnetForKey(pk) + tun.addr = tun.rwc.Address() + tun.subnet = tun.rwc.Subnet() addr := fmt.Sprintf("%s/%d", net.IP(tun.addr[:]).String(), 8*len(address.GetPrefix())-1) if tun.config.IfName == "none" || tun.config.IfName == "dummy" { tun.log.Debugln("Not starting TUN as ifname is none or dummy") @@ -130,8 +129,8 @@ func (tun *TunAdapter) _start() error { return nil } mtu := tun.config.IfMTU - if tun.core.MaxMTU() < mtu { - mtu = tun.core.MaxMTU() + if tun.rwc.MaxMTU() < mtu { + mtu = tun.rwc.MaxMTU() } if err := tun.setup(tun.config.IfName, addr, mtu); err != nil { return err @@ -139,7 +138,7 @@ func (tun *TunAdapter) _start() error { if tun.MTU() != mtu { tun.log.Warnf("Warning: Interface MTU %d automatically adjusted to %d (supported range is 1280-%d)", tun.config.IfMTU, tun.MTU(), MaximumMTU()) } - tun.core.SetMTU(tun.MTU()) + tun.rwc.SetMTU(tun.MTU()) tun.isOpen = true tun.isEnabled = true go tun.read() From e4ce2c79a9a6430f7145798a9a3e93c544bd869c Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 5 Jul 2021 22:26:09 +0100 Subject: [PATCH 2/5] Add LocalAddr to complete net.PacketConn interface --- src/core/core.go | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/src/core/core.go b/src/core/core.go index ac0ea1f..8c816d2 100644 --- a/src/core/core.go +++ b/src/core/core.go @@ -231,3 +231,7 @@ func (c *Core) WriteTo(p []byte, addr net.Addr) (n int, err error) { } return } + +func (c *Core) LocalAddr() net.Addr { + return iwt.Addr(c.public) +} From e224c02d6d88cfd36a0f8598dccfd4cbd4e1f297 Mon Sep 17 00:00:00 2001 From: Neil Alexander Date: Mon, 5 Jul 2021 22:35:46 +0100 Subject: [PATCH 3/5] Revert "Add LocalAddr to complete net.PacketConn interface" This reverts commit e4ce2c79a9a6430f7145798a9a3e93c544bd869c. --- src/core/core.go | 4 ---- 1 file changed, 4 deletions(-) diff --git a/src/core/core.go b/src/core/core.go index 8c816d2..ac0ea1f 100644 --- a/src/core/core.go +++ b/src/core/core.go @@ -231,7 +231,3 @@ func (c *Core) WriteTo(p []byte, addr net.Addr) (n int, err error) { } return } - -func (c *Core) LocalAddr() net.Addr { - return iwt.Addr(c.public) -} From 3704ebf4cbeab2b5818ca545e0bebfce2c7bd10a Mon Sep 17 00:00:00 2001 From: Arceliar Date: Tue, 6 Jul 2021 19:45:12 -0500 Subject: [PATCH 4/5] fix debug rpcs and cleanup core.Close/core.Stop --- src/core/api.go | 5 ----- src/core/core.go | 26 ++++++++++++++++---------- src/core/nodeinfo.go | 4 ++-- src/core/proto.go | 2 +- 4 files changed, 19 insertions(+), 18 deletions(-) diff --git a/src/core/api.go b/src/core/api.go index 6b21934..c312923 100644 --- a/src/core/api.go +++ b/src/core/api.go @@ -239,11 +239,6 @@ func (c *Core) PublicKey() ed25519.PublicKey { return c.public } -func (c *Core) Close() error { - c.Stop() - return nil -} - // Hack to get the admin stuff working, TODO something cleaner type AddHandler interface { diff --git a/src/core/core.go b/src/core/core.go index ac0ea1f..0332980 100644 --- a/src/core/core.go +++ b/src/core/core.go @@ -161,25 +161,31 @@ func (c *Core) _start(nc *config.NodeConfig, log *log.Logger) error { // Stop shuts down the Yggdrasil node. func (c *Core) Stop() { - phony.Block(c, c._stop) + phony.Block(c, func() { + c.log.Infoln("Stopping...") + c._close() + c.log.Infoln("Stopped") + }) +} + +func (c *Core) Close() error { + var err error + phony.Block(c, func() { + err = c._close() + }) + return err } // This function is unsafe and should only be ran by the core actor. -func (c *Core) _stop() { - c.log.Infoln("Stopping...") +func (c *Core) _close() error { c.ctxCancel() - c.PacketConn.Close() // TODO make c.Close() do the right thing (act like c.Stop()) + err := c.PacketConn.Close() if c.addPeerTimer != nil { c.addPeerTimer.Stop() c.addPeerTimer = nil } _ = c.links.stop() - /* FIXME this deadlocks, need a waitgroup or something to coordinate shutdown - for _, peer := range c.GetPeers() { - c.DisconnectPeer(peer.Port) - } - */ - c.log.Infoln("Stopped") + return err } func (c *Core) MTU() uint64 { diff --git a/src/core/nodeinfo.go b/src/core/nodeinfo.go index 9015311..4ca21d7 100644 --- a/src/core/nodeinfo.go +++ b/src/core/nodeinfo.go @@ -129,7 +129,7 @@ func (m *nodeinfo) _sendReq(key keyArray, callback func(nodeinfo NodeInfoPayload if callback != nil { m._addCallback(key, callback) } - _, _ = m.proto.core.WriteTo([]byte{typeSessionProto, typeProtoNodeInfoRequest}, iwt.Addr(key[:])) + _, _ = m.proto.core.PacketConn.WriteTo([]byte{typeSessionProto, typeProtoNodeInfoRequest}, iwt.Addr(key[:])) } func (m *nodeinfo) handleReq(from phony.Actor, key keyArray) { @@ -146,7 +146,7 @@ func (m *nodeinfo) handleRes(from phony.Actor, key keyArray, info NodeInfoPayloa func (m *nodeinfo) _sendRes(key keyArray) { bs := append([]byte{typeSessionProto, typeProtoNodeInfoResponse}, m._getNodeInfo()...) - _, _ = m.proto.core.WriteTo(bs, iwt.Addr(key[:])) + _, _ = m.proto.core.PacketConn.WriteTo(bs, iwt.Addr(key[:])) } // Admin socket stuff diff --git a/src/core/proto.go b/src/core/proto.go index f9d12c8..e60caef 100644 --- a/src/core/proto.go +++ b/src/core/proto.go @@ -212,7 +212,7 @@ func (p *protoHandler) _handleGetDHTResponse(key keyArray, bs []byte) { func (p *protoHandler) _sendDebug(key keyArray, dType uint8, data []byte) { bs := append([]byte{typeSessionProto, typeProtoDebug, dType}, data...) - _, _ = p.core.WriteTo(bs, iwt.Addr(key[:])) + _, _ = p.core.PacketConn.WriteTo(bs, iwt.Addr(key[:])) } // Admin socket stuff From cd5383f7b799932b2536e4ee82e768558a073ec5 Mon Sep 17 00:00:00 2001 From: Arceliar Date: Wed, 7 Jul 2021 18:36:51 -0500 Subject: [PATCH 5/5] fix core tests --- src/core/core_test.go | 15 +++++++-------- 1 file changed, 7 insertions(+), 8 deletions(-) diff --git a/src/core/core_test.go b/src/core/core_test.go index dd60af2..fcfe2e3 100644 --- a/src/core/core_test.go +++ b/src/core/core_test.go @@ -44,13 +44,11 @@ func CreateAndConnectTwo(t testing.TB, verbose bool) (nodeA *Core, nodeB *Core) if err := nodeA.Start(GenerateConfig(), GetLoggerWithPrefix("A: ", verbose)); err != nil { t.Fatal(err) } - nodeA.SetMTU(1500) nodeB = new(Core) if err := nodeB.Start(GenerateConfig(), GetLoggerWithPrefix("B: ", verbose)); err != nil { t.Fatal(err) } - nodeB.SetMTU(1500) u, err := url.Parse("tcp://" + nodeA.links.tcp.getAddr().String()) if err != nil { @@ -94,7 +92,7 @@ func CreateEchoListener(t testing.TB, nodeA *Core, bufLen int, repeats int) chan buf := make([]byte, bufLen) res := make([]byte, bufLen) for i := 0; i < repeats; i++ { - n, err := nodeA.Read(buf) + n, from, err := nodeA.ReadFrom(buf) if err != nil { t.Error(err) return @@ -106,7 +104,7 @@ func CreateEchoListener(t testing.TB, nodeA *Core, bufLen int, repeats int) chan copy(res, buf) copy(res[8:24], buf[24:40]) copy(res[24:40], buf[8:24]) - _, err = nodeA.Write(res) + _, err = nodeA.WriteTo(res, from) if err != nil { t.Error(err) } @@ -141,12 +139,12 @@ func TestCore_Start_Transfer(t *testing.T) { msg[0] = 0x60 copy(msg[8:24], nodeB.Address()) copy(msg[24:40], nodeA.Address()) - _, err := nodeB.Write(msg) + _, err := nodeB.WriteTo(msg, nodeA.LocalAddr()) if err != nil { t.Fatal(err) } buf := make([]byte, msgLen) - _, err = nodeB.Read(buf) + _, _, err = nodeB.ReadFrom(buf) if err != nil { t.Fatal(err) } @@ -179,12 +177,13 @@ func BenchmarkCore_Start_Transfer(b *testing.B) { b.SetBytes(int64(msgLen)) b.ResetTimer() + addr := nodeA.LocalAddr() for i := 0; i < b.N; i++ { - _, err := nodeB.Write(msg) + _, err := nodeB.WriteTo(msg, addr) if err != nil { b.Fatal(err) } - _, err = nodeB.Read(buf) + _, _, err = nodeB.ReadFrom(buf) if err != nil { b.Fatal(err) }