yggdrasil-network/yggdrasil-go
1
0
Fork 0
mirror of https://github.com/yggdrasil-network/yggdrasil-go.git synced 2025-04-28 22:25:07 +03:00
Code Issues Projects Releases Packages Wiki Activity Actions

(broken state) WIP rewriting core to use ironwood

Browse source
This commit is contained in:
Arceliar 2021-05-08 08:35:58 -05:00
parent ace7b43b6d
commit f1c37f8440
24 changed files with 162 additions and 4723 deletions
Download patch file Download diff file Expand all files Collapse all files

354
src/yggdrasil/link.go
View file

@ -1,6 +1,7 @@
package yggdrasil
import (
"crypto/ed25519"
"encoding/hex"
"errors"
"fmt"
@ -17,8 +18,7 @@ import (
"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
"github.com/yggdrasil-network/yggdrasil-go/src/util"
"golang.org/x/net/proxy"
"github.com/Arceliar/phony"
//"github.com/Arceliar/phony" // TODO? use instead of mutexes
)
type links struct {
@ -30,48 +30,27 @@ type links struct {
// TODO timeout (to remove from switch), read from config.ReadTimeout
}
// linkInfo is used as a map key
type linkInfo struct {
box crypto.BoxPubKey // Their encryption key
sig crypto.SigPubKey // Their signing key
linkType string // Type of link, e.g. TCP, AWDL
local string // Local name or address
remote string // Remote name or address
}
type linkMsgIO interface {
readMsg() ([]byte, error)
writeMsgs([][]byte) (int, error)
close() error
// These are temporary workarounds to stream semantics
_sendMetaBytes([]byte) error
_recvMetaBytes() ([]byte, error)
key crypto.SigPubKey
linkType string // Type of link, e.g. TCP, AWDL
local string // Local name or address
remote string // Remote name or address
}
type link struct {
lname string
links *links
peer *peer
options linkOptions
msgIO linkMsgIO
info linkInfo
incoming bool
force bool
closed chan struct{}
reader linkReader // Reads packets, notifies this link, passes packets to switch
writer linkWriter // Writes packets, notifies this link
phony.Inbox // Protects the below
sendTimer *time.Timer // Fires to signal that sending is blocked
keepAliveTimer *time.Timer // Fires to send keep-alive traffic
stallTimer *time.Timer // Fires to signal that no incoming traffic (including keep-alive) has been seen
closeTimer *time.Timer // Fires when the link has been idle so long we need to close it
readUnblocked bool // True if we've sent a read message unblocking this peer in the switch
writeUnblocked bool // True if we've sent a write message unblocking this peer in the swithc
shutdown bool // True if we're shutting down, avoids sending some messages that could race with new peers being crated in the same port
lname string
links *links
conn net.Conn
options linkOptions
info linkInfo
incoming bool
force bool
closed chan struct{}
}
type linkOptions struct {
pinnedCurve25519Keys map[crypto.BoxPubKey]struct{}
pinnedEd25519Keys map[crypto.SigPubKey]struct{}
pinnedEd25519Keys map[crypto.SigPubKey]struct{}
}
func (l *links) init(c *Core) error {
@ -100,16 +79,6 @@ func (l *links) call(uri string, sintf string) error {
}
pathtokens := strings.Split(strings.Trim(u.Path, "/"), "/")
tcpOpts := tcpOptions{}
if pubkeys, ok := u.Query()["curve25519"]; ok && len(pubkeys) > 0 {
tcpOpts.pinnedCurve25519Keys = make(map[crypto.BoxPubKey]struct{})
for _, pubkey := range pubkeys {
if boxPub, err := hex.DecodeString(pubkey); err == nil {
var boxPubKey crypto.BoxPubKey
copy(boxPubKey[:], boxPub)
tcpOpts.pinnedCurve25519Keys[boxPubKey] = struct{}{}
}
}
}
if pubkeys, ok := u.Query()["ed25519"]; ok && len(pubkeys) > 0 {
tcpOpts.pinnedEd25519Keys = make(map[crypto.SigPubKey]struct{})
for _, pubkey := range pubkeys {
@ -157,13 +126,13 @@ func (l *links) listen(uri string) error {
}
}
func (l *links) create(msgIO linkMsgIO, name, linkType, local, remote string, incoming, force bool, options linkOptions) (*link, error) {
func (l *links) create(conn net.Conn, name, linkType, local, remote string, incoming, force bool, options linkOptions) (*link, error) {
// Technically anything unique would work for names, but let's pick something human readable, just for debugging
intf := link{
conn: conn,
lname: name,
links: l,
options: options,
msgIO: msgIO,
info: linkInfo{
linkType: linkType,
local: local,
@ -172,10 +141,6 @@ func (l *links) create(msgIO linkMsgIO, name, linkType, local, remote string, in
incoming: incoming,
force: force,
}
intf.writer.intf = &intf
intf.writer.worker = make(chan [][]byte, 1)
intf.reader.intf = &intf
intf.reader.err = make(chan error)
return &intf, nil
}
@ -189,30 +154,31 @@ func (l *links) stop() error {
func (intf *link) handler() (chan struct{}, error) {
// TODO split some of this into shorter functions, so it's easier to read, and for the FIXME duplicate peer issue mentioned later
go func() {
for bss := range intf.writer.worker {
intf.msgIO.writeMsgs(bss)
}
}()
defer intf.writer.Act(nil, func() {
intf.writer.closed = true
close(intf.writer.worker)
})
myLinkPub, myLinkPriv := crypto.NewBoxKeys()
defer intf.conn.Close()
meta := version_getBaseMetadata()
meta.box = intf.links.core.boxPub
meta.sig = intf.links.core.sigPub
meta.link = *myLinkPub
meta.key = intf.links.core.public
metaBytes := meta.encode()
// TODO timeouts on send/recv (goroutine for send/recv, channel select w/ timer)
var err error
if !util.FuncTimeout(func() { err = intf.msgIO._sendMetaBytes(metaBytes) }, 30*time.Second) {
if !util.FuncTimeout(30*time.Second, func() {
var n int
n, err = intf.conn.Write(metaBytes)
if err == nil && n != len(metaBytes) {
err = errors.New("incomplete metadata send")
}
}) {
return nil, errors.New("timeout on metadata send")
}
if err != nil {
return nil, err
}
if !util.FuncTimeout(func() { metaBytes, err = intf.msgIO._recvMetaBytes() }, 30*time.Second) {
if !util.FuncTimeout(30*time.Second, func() {
var n int
n, err = io.ReadFull(intf.conn, metaBytes)
if err == nil && n != len(metaBytes) {
err = errors.New("incomplete metadata recv")
}
}) {
return nil, errors.New("timeout on metadata recv")
}
if err != nil {
@ -229,35 +195,31 @@ func (intf *link) handler() (chan struct{}, error) {
}
// Check if the remote side matches the keys we expected. This is a bit of a weak
// check - in future versions we really should check a signature or something like that.
if pinned := intf.options.pinnedCurve25519Keys; pinned != nil {
if _, allowed := pinned[meta.box]; !allowed {
intf.links.core.log.Errorf("Failed to connect to node: %q sent curve25519 key that does not match pinned keys", intf.name)
return nil, fmt.Errorf("failed to connect: host sent curve25519 key that does not match pinned keys")
}
}
if pinned := intf.options.pinnedEd25519Keys; pinned != nil {
if _, allowed := pinned[meta.sig]; !allowed {
var key crypto.SigPubKey
copy(key[:], meta.key)
if _, allowed := pinned[key]; !allowed {
intf.links.core.log.Errorf("Failed to connect to node: %q sent ed25519 key that does not match pinned keys", intf.name)
return nil, fmt.Errorf("failed to connect: host sent ed25519 key that does not match pinned keys")
}
}
// Check if we're authorized to connect to this key / IP
/* TODO check allowed public keys
if intf.incoming && !intf.force && !intf.links.core.peers.isAllowedEncryptionPublicKey(&meta.box) {
intf.links.core.log.Warnf("%s connection from %s forbidden: AllowedEncryptionPublicKeys does not contain key %s",
strings.ToUpper(intf.info.linkType), intf.info.remote, hex.EncodeToString(meta.box[:]))
intf.msgIO.close()
return nil, nil
}
*/
// Check if we already have a link to this node
intf.info.box = meta.box
intf.info.sig = meta.sig
copy(intf.info.key[:], meta.key)
intf.links.mutex.Lock()
if oldIntf, isIn := intf.links.links[intf.info]; isIn {
intf.links.mutex.Unlock()
// FIXME we should really return an error and let the caller block instead
// That lets them do things like close connections on its own, avoid printing a connection message in the first place, etc.
intf.links.core.log.Debugln("DEBUG: found existing interface for", intf.name)
intf.msgIO.close()
return oldIntf.closed, nil
} else {
intf.closed = make(chan struct{})
@ -271,43 +233,13 @@ func (intf *link) handler() (chan struct{}, error) {
intf.links.core.log.Debugln("DEBUG: registered interface for", intf.name)
}
intf.links.mutex.Unlock()
// Create peer
shared := crypto.GetSharedKey(myLinkPriv, &meta.link)
phony.Block(&intf.links.core.peers, func() {
// FIXME don't use phony.Block, it's bad practice, even if it's safe here
intf.peer = intf.links.core.peers._newPeer(&meta.box, &meta.sig, shared, intf)
})
if intf.peer == nil {
return nil, errors.New("failed to create peer")
}
defer func() {
// More cleanup can go here
intf.Act(nil, func() {
intf.shutdown = true
intf.peer.Act(intf, intf.peer._removeSelf)
})
}()
themAddr := make([]byte, 16) // TODO address.AddrForNodeID(crypto.GetNodeID(&intf.info.box))
themAddrString := net.IP(themAddr[:]).String()
themString := fmt.Sprintf("%s@%s", themAddrString, intf.info.remote)
intf.links.core.log.Infof("Connected %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
// Start things
go intf.peer.start()
intf.Act(nil, intf._notifyIdle)
intf.reader.Act(nil, intf.reader._read)
// Wait for the reader to finish
// TODO find a way to do this without keeping live goroutines around
done := make(chan struct{})
defer close(done)
go func() {
select {
case <-intf.links.stopped:
intf.msgIO.close()
case <-done:
}
}()
err = <-intf.reader.err
// Run the handler
err = intf.links.core.PacketConn.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",
@ -319,43 +251,8 @@ func (intf *link) handler() (chan struct{}, error) {
return nil, err
}
////////////////////////////////////////////////////////////////////////////////
// link needs to match the linkInterface type needed by the peers
type linkInterface interface {
out([][]byte)
linkOut([]byte)
close()
// These next ones are only used by the API
name() string
local() string
remote() string
interfaceType() string
}
func (intf *link) out(bss [][]byte) {
intf.Act(nil, func() {
// nil to prevent it from blocking if the link is somehow frozen
// this is safe because another packet won't be sent until the link notifies
// the peer that it's ready for one
intf.writer.sendFrom(nil, bss)
})
}
func (intf *link) linkOut(bs []byte) {
intf.Act(nil, func() {
// nil to prevent it from blocking if the link is somehow frozen
// FIXME this is hypothetically not safe, the peer shouldn't be sending
// additional packets until this one finishes, otherwise this could leak
// memory if writing happens slower than link packets are generated...
// that seems unlikely, so it's a lesser evil than deadlocking for now
intf.writer.sendFrom(nil, [][]byte{bs})
})
}
func (intf *link) close() {
intf.Act(nil, func() { intf.msgIO.close() })
intf.conn.Close()
}
func (intf *link) name() string {
@ -373,168 +270,3 @@ func (intf *link) remote() string {
func (intf *link) interfaceType() string {
return intf.info.linkType
}
////////////////////////////////////////////////////////////////////////////////
const (
sendTime = 1 * time.Second // How long to wait before deciding a send is blocked
keepAliveTime = 2 * time.Second // How long to wait before sending a keep-alive response if we have no real traffic to send
stallTime = 6 * time.Second // How long to wait for response traffic before deciding the connection has stalled
closeTime = 2 * switch_timeout // How long to wait before closing the link
)
// notify the intf that we're currently sending
func (intf *link) notifySending(size int) {
intf.Act(&intf.writer, func() {
intf.sendTimer = time.AfterFunc(sendTime, intf.notifyBlockedSend)
if intf.keepAliveTimer != nil {
intf.keepAliveTimer.Stop()
intf.keepAliveTimer = nil
}
intf.peer.notifyBlocked(intf)
})
}
// This gets called from a time.AfterFunc, and notifies the switch that we appear
// to have gotten blocked on a write, so the switch should start routing traffic
// through other links, if alternatives exist
func (intf *link) notifyBlockedSend() {
intf.Act(nil, func() {
if intf.sendTimer != nil {
//As far as we know, we're still trying to send, and the timer fired.
intf.sendTimer.Stop()
intf.sendTimer = nil
if !intf.shutdown && intf.writeUnblocked {
intf.writeUnblocked = false
intf.links.core.switchTable.blockPeer(intf, intf.peer.port, true)
}
}
})
}
// notify the intf that we've finished sending, returning the peer to the switch
func (intf *link) notifySent(size int) {
intf.Act(&intf.writer, func() {
if intf.sendTimer != nil {
intf.sendTimer.Stop()
intf.sendTimer = nil
}
if intf.keepAliveTimer != nil {
// TODO? unset this when we start sending, not when we finish...
intf.keepAliveTimer.Stop()
intf.keepAliveTimer = nil
}
intf._notifyIdle()
if size > 0 && intf.stallTimer == nil {
intf.stallTimer = time.AfterFunc(stallTime, intf.notifyStalled)
}
if !intf.shutdown && !intf.writeUnblocked {
intf.writeUnblocked = true
intf.links.core.switchTable.unblockPeer(intf, intf.peer.port, true)
}
})
}
// Notify the peer that we're ready for more traffic
func (intf *link) _notifyIdle() {
intf.peer.Act(intf, intf.peer._handleIdle)
}
// Set the peer as stalled, to prevent them from returning to the switch until a read succeeds
func (intf *link) notifyStalled() {
intf.Act(nil, func() { // Sent from a time.AfterFunc
if intf.stallTimer != nil {
intf.stallTimer.Stop()
intf.stallTimer = nil
if !intf.shutdown && intf.readUnblocked {
intf.readUnblocked = false
intf.links.core.switchTable.blockPeer(intf, intf.peer.port, false)
}
}
})
}
// reset the close timer
func (intf *link) notifyReading() {
intf.Act(&intf.reader, func() {
intf.closeTimer = time.AfterFunc(closeTime, func() { intf.msgIO.close() })
})
}
// wake up the link if it was stalled, and (if size > 0) prepare to send keep-alive traffic
func (intf *link) notifyRead(size int) {
intf.Act(&intf.reader, func() {
intf.closeTimer.Stop()
if intf.stallTimer != nil {
intf.stallTimer.Stop()
intf.stallTimer = nil
}
if size > 0 && intf.keepAliveTimer == nil {
intf.keepAliveTimer = time.AfterFunc(keepAliveTime, intf.notifyDoKeepAlive)
}
if !intf.shutdown && !intf.readUnblocked {
intf.readUnblocked = true
intf.links.core.switchTable.unblockPeer(intf, intf.peer.port, false)
}
})
}
// We need to send keep-alive traffic now
func (intf *link) notifyDoKeepAlive() {
intf.Act(nil, func() { // Sent from a time.AfterFunc
if intf.keepAliveTimer != nil {
intf.keepAliveTimer.Stop()
intf.keepAliveTimer = nil
intf.writer.sendFrom(nil, [][]byte{nil}) // Empty keep-alive traffic
}
})
}
////////////////////////////////////////////////////////////////////////////////
type linkWriter struct {
phony.Inbox
intf *link
worker chan [][]byte
closed bool
}
func (w *linkWriter) sendFrom(from phony.Actor, bss [][]byte) {
w.Act(from, func() {
if w.closed {
return
}
var size int
for _, bs := range bss {
size += len(bs)
}
w.intf.notifySending(size)
w.worker <- bss
w.intf.notifySent(size)
})
}
////////////////////////////////////////////////////////////////////////////////
type linkReader struct {
phony.Inbox
intf *link
err chan error
}
func (r *linkReader) _read() {
r.intf.notifyReading()
msg, err := r.intf.msgIO.readMsg()
r.intf.notifyRead(len(msg))
if len(msg) > 0 {
r.intf.peer.handlePacketFrom(r, msg)
}
if err != nil {
if err != io.EOF {
r.err <- err
}
close(r.err)
return
}
// Now try to read again
r.Act(nil, r._read)
}