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

Merge pull request #360 from neilalexander/multilink

Browse source 
Support for multiple listeners
This commit is contained in:
Arceliar 2019-03-08 18:56:47 -06:00 • committed by GitHub
commit a0e6edd219
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
17 changed files with 359 additions and 203 deletions
Download patch file Download diff file Expand all files Collapse all files

21
src/yggdrasil/admin.go
View file

@ -591,18 +591,9 @@ func (a *admin) printInfos(infos []admin_nodeInfo) string {
// addPeer triggers a connection attempt to a node.
func (a *admin) addPeer(addr string, sintf string) error {
u, err := url.Parse(addr)
if err == nil {
switch strings.ToLower(u.Scheme) {
case "tcp":
a.core.tcp.connect(u.Host, sintf)
case "socks":
a.core.tcp.connectSOCKS(u.Host, u.Path[1:])
default:
return errors.New("invalid peer: " + addr)
}
} else {
return errors.New("invalid peer: " + addr)
err := a.core.link.call(addr, sintf)
if err != nil {
return err
}
return nil
}
@ -684,7 +675,8 @@ func (a *admin) getData_getPeers() []admin_nodeInfo {
{"uptime", int(time.Since(p.firstSeen).Seconds())},
{"bytes_sent", atomic.LoadUint64(&p.bytesSent)},
{"bytes_recvd", atomic.LoadUint64(&p.bytesRecvd)},
{"endpoint", p.endpoint},
{"proto", p.intf.info.linkType},
{"endpoint", p.intf.name},
{"box_pub_key", hex.EncodeToString(p.box[:])},
}
peerInfos = append(peerInfos, info)
@ -710,7 +702,8 @@ func (a *admin) getData_getSwitchPeers() []admin_nodeInfo {
{"port", elem.port},
{"bytes_sent", atomic.LoadUint64(&peer.bytesSent)},
{"bytes_recvd", atomic.LoadUint64(&peer.bytesRecvd)},
{"endpoint", peer.endpoint},
{"proto", peer.intf.info.linkType},
{"endpoint", peer.intf.info.remote},
{"box_pub_key", hex.EncodeToString(peer.box[:])},
}
peerInfos = append(peerInfos, info)

14
src/yggdrasil/awdl.go
View file

@ -7,9 +7,10 @@ import (
)
type awdl struct {
link *link
mutex sync.RWMutex // protects interfaces below
interfaces map[string]*awdlInterface
link *link
reconfigure chan chan error
mutex sync.RWMutex // protects interfaces below
interfaces map[string]*awdlInterface
}
type awdlInterface struct {
@ -49,8 +50,15 @@ func (a *awdl) init(l *link) error {
a.link = l
a.mutex.Lock()
a.interfaces = make(map[string]*awdlInterface)
a.reconfigure = make(chan chan error, 1)
a.mutex.Unlock()
go func() {
for e := range a.reconfigure {
e <- nil
}
}()
return nil
}

8
src/yggdrasil/core.go
View file

@ -44,7 +44,6 @@ type Core struct {
admin admin
searches searches
multicast multicast
tcp tcpInterface
link link
log *log.Logger
}
@ -144,7 +143,7 @@ func (c *Core) UpdateConfig(config *config.NodeConfig) {
c.router.tun.reconfigure,
c.router.cryptokey.reconfigure,
c.switchTable.reconfigure,
c.tcp.reconfigure,
c.link.reconfigure,
c.multicast.reconfigure,
}
@ -205,11 +204,6 @@ func (c *Core) Start(nc *config.NodeConfig, log *log.Logger) error {
c.init()
if err := c.tcp.init(c); err != nil {
c.log.Errorln("Failed to start TCP interface")
return err
}
if err := c.link.init(c); err != nil {
c.log.Errorln("Failed to start link interfaces")
return err

20
src/yggdrasil/debug.go
View file

@ -97,7 +97,15 @@ func (c *Core) DEBUG_getPeers() *peers {
}
func (ps *peers) DEBUG_newPeer(box crypto.BoxPubKey, sig crypto.SigPubKey, link crypto.BoxSharedKey) *peer {
return ps.newPeer(&box, &sig, &link, "(simulator)", nil)
sim := linkInterface{
name: "(simulator)",
info: linkInfo{
local: "(simulator)",
remote: "(simulator)",
linkType: "sim",
},
}
return ps.newPeer(&box, &sig, &link, &sim, nil)
}
/*
@ -449,19 +457,19 @@ func (c *Core) DEBUG_addSOCKSConn(socksaddr, peeraddr string) {
//*
func (c *Core) DEBUG_setupAndStartGlobalTCPInterface(addrport string) {
c.config.Listen = addrport
if err := c.tcp.init(c /*, addrport, 0*/); err != nil {
c.log.Println("Failed to start TCP interface:", err)
c.config.Listen = []string{addrport}
if err := c.link.init(c); err != nil {
c.log.Println("Failed to start interfaces:", err)
panic(err)
}
}
func (c *Core) DEBUG_getGlobalTCPAddr() *net.TCPAddr {
return c.tcp.serv.Addr().(*net.TCPAddr)
return c.link.tcp.getAddr()
}
func (c *Core) DEBUG_addTCPConn(saddr string) {
c.tcp.call(saddr, nil, "")
c.link.tcp.call(saddr, nil, "")
}
//*/

82
src/yggdrasil/link.go
View file

@ -6,8 +6,10 @@ import (
"fmt"
"io"
"net"
"net/url"
"strings"
"sync"
//"sync/atomic"
"time"
@ -17,10 +19,12 @@ import (
)
type link struct {
core *Core
mutex sync.RWMutex // protects interfaces below
interfaces map[linkInfo]*linkInterface
awdl awdl // AWDL interface support
core *Core
reconfigure chan chan error
mutex sync.RWMutex // protects interfaces below
interfaces map[linkInfo]*linkInterface
awdl awdl // AWDL interface support
tcp tcp // TCP interface support
// TODO timeout (to remove from switch), read from config.ReadTimeout
}
@ -56,16 +60,72 @@ func (l *link) init(c *Core) error {
l.core = c
l.mutex.Lock()
l.interfaces = make(map[linkInfo]*linkInterface)
l.reconfigure = make(chan chan error)
l.mutex.Unlock()
if err := l.awdl.init(l); err != nil {
l.core.log.Errorln("Failed to start AWDL interface")
if err := l.tcp.init(l); err != nil {
c.log.Errorln("Failed to start TCP interface")
return err
}
if err := l.awdl.init(l); err != nil {
c.log.Errorln("Failed to start AWDL interface")
return err
}
go func() {
for {
e := <-l.reconfigure
tcpresponse := make(chan error)
awdlresponse := make(chan error)
l.tcp.reconfigure <- tcpresponse
if err := <-tcpresponse; err != nil {
e <- err
continue
}
l.awdl.reconfigure <- awdlresponse
if err := <-awdlresponse; err != nil {
e <- err
continue
}
e <- nil
}
}()
return nil
}
func (l *link) call(uri string, sintf string) error {
u, err := url.Parse(uri)
if err != nil {
return err
}
pathtokens := strings.Split(strings.Trim(u.Path, "/"), "/")
switch u.Scheme {
case "tcp":
l.tcp.call(u.Host, nil, sintf)
case "socks":
l.tcp.call(pathtokens[0], u.Host, sintf)
default:
return errors.New("unknown call scheme: " + u.Scheme)
}
return nil
}
func (l *link) listen(uri string) error {
u, err := url.Parse(uri)
if err != nil {
return err
}
switch u.Scheme {
case "tcp":
_, err := l.tcp.listen(u.Host)
return err
default:
return errors.New("unknown listen scheme: " + u.Scheme)
}
}
func (l *link) create(msgIO linkInterfaceMsgIO, name, linkType, local, remote string, incoming, force bool) (*linkInterface, error) {
// Technically anything unique would work for names, but lets pick something human readable, just for debugging
intf := linkInterface{
@ -147,7 +207,7 @@ func (intf *linkInterface) handler() error {
intf.link.mutex.Unlock()
// Create peer
shared := crypto.GetSharedKey(myLinkPriv, &meta.link)
intf.peer = intf.link.core.peers.newPeer(&meta.box, &meta.sig, shared, intf.name, func() { intf.msgIO.close() })
intf.peer = intf.link.core.peers.newPeer(&meta.box, &meta.sig, shared, intf, func() { intf.msgIO.close() })
if intf.peer == nil {
return errors.New("failed to create peer")
}
@ -201,11 +261,11 @@ func (intf *linkInterface) handler() error {
// Now block until something is ready or the timer triggers keepalive traffic
select {
case <-tcpTimer.C:
intf.link.core.log.Debugf("Sending (legacy) keep-alive to %s: %s, source %s",
intf.link.core.log.Tracef("Sending (legacy) keep-alive to %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
send(nil)
case <-sendAck:
intf.link.core.log.Debugf("Sending ack to %s: %s, source %s",
intf.link.core.log.Tracef("Sending ack to %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
send(nil)
case msg := <-intf.peer.linkOut:
@ -220,7 +280,7 @@ func (intf *linkInterface) handler() error {
case signalReady <- struct{}{}:
default:
}
//intf.link.core.log.Debugf("Sending packet to %s: %s, source %s",
//intf.link.core.log.Tracef("Sending packet to %s: %s, source %s",
// strings.ToUpper(intf.info.linkType), themString, intf.info.local)
}
}
@ -271,7 +331,7 @@ func (intf *linkInterface) handler() error {
sendTimerRunning = true
}
if !gotMsg {
intf.link.core.log.Debugf("Received ack from %s: %s, source %s",
intf.link.core.log.Tracef("Received ack from %s: %s, source %s",
strings.ToUpper(intf.info.linkType), themString, intf.info.local)
}
case sentMsg, ok := <-signalSent:

81
src/yggdrasil/multicast.go
View file

@ -5,7 +5,6 @@ import (
"fmt"
"net"
"regexp"
"sync"
"time"
"golang.org/x/net/ipv6"
@ -16,19 +15,16 @@ type multicast struct {
reconfigure chan chan error
sock *ipv6.PacketConn
groupAddr string
myAddr *net.TCPAddr
myAddrMutex sync.RWMutex
listeners map[string]*tcpListener
}
func (m *multicast) init(core *Core) {
m.core = core
m.reconfigure = make(chan chan error, 1)
m.listeners = make(map[string]*tcpListener)
go func() {
for {
e := <-m.reconfigure
m.myAddrMutex.Lock()
m.myAddr = m.core.tcp.getAddr()
m.myAddrMutex.Unlock()
e <- nil
}
}()
@ -68,13 +64,13 @@ func (m *multicast) start() error {
return nil
}
func (m *multicast) interfaces() []net.Interface {
func (m *multicast) interfaces() map[string]net.Interface {
// Get interface expressions from config
m.core.configMutex.RLock()
exprs := m.core.config.MulticastInterfaces
m.core.configMutex.RUnlock()
// Ask the system for network interfaces
var interfaces []net.Interface
interfaces := make(map[string]net.Interface)
allifaces, err := net.Interfaces()
if err != nil {
panic(err)
@ -94,12 +90,14 @@ func (m *multicast) interfaces() []net.Interface {
continue
}
for _, expr := range exprs {
// Compile each regular expression
e, err := regexp.Compile(expr)
if err != nil {
panic(err)
}
// Does the interface match the regular expression? Store it if so
if e.MatchString(iface.Name) {
interfaces = append(interfaces, iface)
interfaces[iface.Name] = iface
}
}
}
@ -107,10 +105,6 @@ func (m *multicast) interfaces() []net.Interface {
}
func (m *multicast) announce() {
var anAddr net.TCPAddr
m.myAddrMutex.Lock()
m.myAddr = m.core.tcp.getAddr()
m.myAddrMutex.Unlock()
groupAddr, err := net.ResolveUDPAddr("udp6", m.groupAddr)
if err != nil {
panic(err)
@ -120,33 +114,66 @@ func (m *multicast) announce() {
panic(err)
}
for {
for _, iface := range m.interfaces() {
m.sock.JoinGroup(&iface, groupAddr)
interfaces := m.interfaces()
// There might be interfaces that we configured listeners for but are no
// longer up - if that's the case then we should stop the listeners
for name, listener := range m.listeners {
if _, ok := interfaces[name]; !ok {
listener.stop <- true
delete(m.listeners, name)
m.core.log.Debugln("No longer multicasting on", name)
}
}
// Now that we have a list of valid interfaces from the operating system,
// we can start checking if we can send multicasts on them
for _, iface := range interfaces {
// Find interface addresses
addrs, err := iface.Addrs()
if err != nil {
panic(err)
}
m.myAddrMutex.RLock()
anAddr.Port = m.myAddr.Port
m.myAddrMutex.RUnlock()
for _, addr := range addrs {
addrIP, _, _ := net.ParseCIDR(addr.String())
// Ignore IPv4 addresses
if addrIP.To4() != nil {
continue
} // IPv6 only
}
// Ignore non-link-local addresses
if !addrIP.IsLinkLocalUnicast() {
continue
}
anAddr.IP = addrIP
anAddr.Zone = iface.Name
destAddr.Zone = iface.Name
msg := []byte(anAddr.String())
m.sock.WriteTo(msg, nil, destAddr)
// Join the multicast group
m.sock.JoinGroup(&iface, groupAddr)
// Try and see if we already have a TCP listener for this interface
var listener *tcpListener
if l, ok := m.listeners[iface.Name]; !ok || l.listener == nil {
// No listener was found - let's create one
listenaddr := fmt.Sprintf("[%s%%%s]:0", addrIP, iface.Name)
if li, err := m.core.link.tcp.listen(listenaddr); err == nil {
m.core.log.Debugln("Started multicasting on", iface.Name)
// Store the listener so that we can stop it later if needed
m.listeners[iface.Name] = li
listener = li
}
} else {
// An existing listener was found
listener = m.listeners[iface.Name]
}
// Make sure nothing above failed for some reason
if listener == nil {
continue
}
// Get the listener details and construct the multicast beacon
lladdr := listener.listener.Addr().String()
if a, err := net.ResolveTCPAddr("tcp6", lladdr); err == nil {
destAddr.Zone = iface.Name
msg := []byte(a.String())
m.sock.WriteTo(msg, nil, destAddr)
}
break
}
time.Sleep(time.Second)
}
time.Sleep(time.Second)
time.Sleep(time.Second * 15)
}
}
@ -183,6 +210,6 @@ func (m *multicast) listen() {
}
addr.Zone = from.Zone
saddr := addr.String()
m.core.tcp.connect(saddr, addr.Zone)
m.core.link.call("tcp://"+saddr, addr.Zone)
}
}

8
src/yggdrasil/peer.go
View file

@ -98,6 +98,7 @@ type peer struct {
bytesRecvd uint64 // To track bandwidth usage for getPeers
// BUG: sync/atomic, 32 bit platforms need the above to be the first element
core *Core
intf *linkInterface
port switchPort
box crypto.BoxPubKey
sig crypto.SigPubKey
@ -113,18 +114,19 @@ type peer struct {
}
// Creates a new peer with the specified box, sig, and linkShared keys, using the lowest unoccupied port number.
func (ps *peers) newPeer(box *crypto.BoxPubKey, sig *crypto.SigPubKey, linkShared *crypto.BoxSharedKey, endpoint string, closer func()) *peer {
func (ps *peers) newPeer(box *crypto.BoxPubKey, sig *crypto.SigPubKey, linkShared *crypto.BoxSharedKey, intf *linkInterface, closer func()) *peer {
now := time.Now()
p := peer{box: *box,
sig: *sig,
shared: *crypto.GetSharedKey(&ps.core.boxPriv, box),
linkShared: *linkShared,
endpoint: endpoint,
firstSeen: now,
doSend: make(chan struct{}, 1),
dinfo: make(chan *dhtInfo, 1),
close: closer,
core: ps.core}
core: ps.core,
intf: intf,
}
ps.mutex.Lock()
defer ps.mutex.Unlock()
oldPorts := ps.getPorts()

10
src/yggdrasil/router.go
View file

@ -67,7 +67,15 @@ func (r *router) init(core *Core) {
r.addr = *address.AddrForNodeID(&r.core.dht.nodeID)
r.subnet = *address.SubnetForNodeID(&r.core.dht.nodeID)
in := make(chan []byte, 1) // TODO something better than this...
p := r.core.peers.newPeer(&r.core.boxPub, &r.core.sigPub, &crypto.BoxSharedKey{}, "(self)", nil)
self := linkInterface{
name: "(self)",
info: linkInfo{
local: "(self)",
remote: "(self)",
linkType: "self",
},
}
p := r.core.peers.newPeer(&r.core.boxPub, &r.core.sigPub, &crypto.BoxSharedKey{}, &self, nil)
p.out = func(packet []byte) { in <- packet }
r.in = in
out := make(chan []byte, 32)

254
src/yggdrasil/tcp.go
View file

@ -24,35 +24,29 @@ import (
"golang.org/x/net/proxy"
"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
"github.com/yggdrasil-network/yggdrasil-go/src/util"
)
const default_timeout = 6 * time.Second
const tcp_ping_interval = (default_timeout * 2 / 3)
// The TCP listener and information about active TCP connections, to avoid duplication.
type tcpInterface struct {
core *Core
type tcp struct {
link *link
reconfigure chan chan error
serv net.Listener
stop chan bool
addr string
mutex sync.Mutex // Protecting the below
listeners map[string]*tcpListener
calls map[string]struct{}
conns map[tcpInfo](chan struct{})
conns map[linkInfo](chan struct{})
}
// This is used as the key to a map that tracks existing connections, to prevent multiple connections to the same keys and local/remote address pair from occuring.
// Different address combinations are allowed, so multi-homing is still technically possible (but not necessarily advisable).
type tcpInfo struct {
box crypto.BoxPubKey
sig crypto.SigPubKey
localAddr string
remoteAddr string
type tcpListener struct {
listener net.Listener
stop chan bool
}
// Wrapper function to set additional options for specific connection types.
func (iface *tcpInterface) setExtraOptions(c net.Conn) {
func (t *tcp) setExtraOptions(c net.Conn) {
switch sock := c.(type) {
case *net.TCPConn:
sock.SetNoDelay(true)
@ -62,104 +56,152 @@ func (iface *tcpInterface) setExtraOptions(c net.Conn) {
}
// Returns the address of the listener.
func (iface *tcpInterface) getAddr() *net.TCPAddr {
return iface.serv.Addr().(*net.TCPAddr)
}
// Attempts to initiate a connection to the provided address.
func (iface *tcpInterface) connect(addr string, intf string) {
iface.call(addr, nil, intf)
}
// Attempst to initiate a connection to the provided address, viathe provided socks proxy address.
func (iface *tcpInterface) connectSOCKS(socksaddr, peeraddr string) {
iface.call(peeraddr, &socksaddr, "")
func (t *tcp) getAddr() *net.TCPAddr {
// TODO: Fix this, because this will currently only give a single address
// to multicast.go, which obviously is not great, but right now multicast.go
// doesn't have the ability to send more than one address in a packet either
t.mutex.Lock()
defer t.mutex.Unlock()
for _, l := range t.listeners {
return l.listener.Addr().(*net.TCPAddr)
}
return nil
}
// Initializes the struct.
func (iface *tcpInterface) init(core *Core) (err error) {
iface.core = core
iface.stop = make(chan bool, 1)
iface.reconfigure = make(chan chan error, 1)
func (t *tcp) init(l *link) error {
t.link = l
t.reconfigure = make(chan chan error, 1)
t.mutex.Lock()
t.calls = make(map[string]struct{})
t.conns = make(map[linkInfo](chan struct{}))
t.listeners = make(map[string]*tcpListener)
t.mutex.Unlock()
go func() {
for {
e := <-iface.reconfigure
iface.core.configMutex.RLock()
updated := iface.core.config.Listen != iface.core.configOld.Listen
iface.core.configMutex.RUnlock()
if updated {
iface.stop <- true
iface.serv.Close()
e <- iface.listen()
e := <-t.reconfigure
t.link.core.configMutex.RLock()
added := util.Difference(t.link.core.config.Listen, t.link.core.configOld.Listen)
deleted := util.Difference(t.link.core.configOld.Listen, t.link.core.config.Listen)
t.link.core.configMutex.RUnlock()
if len(added) > 0 || len(deleted) > 0 {
for _, a := range added {
if a[:6] != "tcp://" {
continue
}
if _, err := t.listen(a[6:]); err != nil {
e <- err
continue
}
}
for _, d := range deleted {
if d[:6] != "tcp://" {
continue
}
t.mutex.Lock()
if listener, ok := t.listeners[d[6:]]; ok {
t.mutex.Unlock()
listener.stop <- true
} else {
t.mutex.Unlock()
}
}
e <- nil
} else {
e <- nil
}
}
}()
return iface.listen()
t.link.core.configMutex.RLock()
defer t.link.core.configMutex.RUnlock()
for _, listenaddr := range t.link.core.config.Listen {
if listenaddr[:6] != "tcp://" {
continue
}
if _, err := t.listen(listenaddr[6:]); err != nil {
return err
}
}
return nil
}
func (iface *tcpInterface) listen() error {
func (t *tcp) listen(listenaddr string) (*tcpListener, error) {
var err error
iface.core.configMutex.RLock()
iface.addr = iface.core.config.Listen
iface.core.configMutex.RUnlock()
ctx := context.Background()
lc := net.ListenConfig{
Control: iface.tcpContext,
Control: t.tcpContext,
}
iface.serv, err = lc.Listen(ctx, "tcp", iface.addr)
listener, err := lc.Listen(ctx, "tcp", listenaddr)
if err == nil {
iface.mutex.Lock()
iface.calls = make(map[string]struct{})
iface.conns = make(map[tcpInfo](chan struct{}))
iface.mutex.Unlock()
go iface.listener()
return nil
l := tcpListener{
listener: listener,
stop: make(chan bool),
}
go t.listener(&l, listenaddr)
return &l, nil
}
return err
return nil, err
}
// Runs the listener, which spawns off goroutines for incoming connections.
func (iface *tcpInterface) listener() {
defer iface.serv.Close()
iface.core.log.Infoln("Listening for TCP on:", iface.serv.Addr().String())
func (t *tcp) listener(l *tcpListener, listenaddr string) {
if l == nil {
return
}
// Track the listener so that we can find it again in future
t.mutex.Lock()
if _, isIn := t.listeners[listenaddr]; isIn {
t.mutex.Unlock()
l.listener.Close()
return
} else {
t.listeners[listenaddr] = l
t.mutex.Unlock()
}
// And here we go!
accepted := make(chan bool)
defer func() {
t.link.core.log.Infoln("Stopping TCP listener on:", l.listener.Addr().String())
l.listener.Close()
t.mutex.Lock()
delete(t.listeners, listenaddr)
t.mutex.Unlock()
}()
t.link.core.log.Infoln("Listening for TCP on:", l.listener.Addr().String())
for {
sock, err := iface.serv.Accept()
if err != nil {
iface.core.log.Errorln("Failed to accept connection:", err)
return
}
var sock net.Conn
var err error
// Listen in a separate goroutine, as that way it does not block us from
// receiving "stop" events
go func() {
sock, err = l.listener.Accept()
accepted <- true
}()
// Wait for either an accepted connection, or a message telling us to stop
// the TCP listener
select {
case <-iface.stop:
iface.core.log.Errorln("Stopping listener")
return
default:
case <-accepted:
if err != nil {
panic(err)
t.link.core.log.Errorln("Failed to accept connection:", err)
return
}
go iface.handler(sock, true)
go t.handler(sock, true, nil)
case <-l.stop:
return
}
}
}
// Checks if we already have a connection to this node
func (iface *tcpInterface) isAlreadyConnected(info tcpInfo) bool {
iface.mutex.Lock()
defer iface.mutex.Unlock()
_, isIn := iface.conns[info]
return isIn
}
// Checks if we already are calling this address
func (iface *tcpInterface) isAlreadyCalling(saddr string) bool {
iface.mutex.Lock()
defer iface.mutex.Unlock()
_, isIn := iface.calls[saddr]
func (t *tcp) isAlreadyCalling(saddr string) bool {
t.mutex.Lock()
defer t.mutex.Unlock()
_, isIn := t.calls[saddr]
return isIn
}
@ -168,34 +210,39 @@ func (iface *tcpInterface) isAlreadyCalling(saddr string) bool {
// If the dial is successful, it launches the handler.
// When finished, it removes the outgoing call, so reconnection attempts can be made later.
// This all happens in a separate goroutine that it spawns.
func (iface *tcpInterface) call(saddr string, socksaddr *string, sintf string) {
func (t *tcp) call(saddr string, options interface{}, sintf string) {
go func() {
callname := saddr
if sintf != "" {
callname = fmt.Sprintf("%s/%s", saddr, sintf)
}
if iface.isAlreadyCalling(callname) {
if t.isAlreadyCalling(callname) {
return
}
iface.mutex.Lock()
iface.calls[callname] = struct{}{}
iface.mutex.Unlock()
t.mutex.Lock()
t.calls[callname] = struct{}{}
t.mutex.Unlock()
defer func() {
// Block new calls for a little while, to mitigate livelock scenarios
time.Sleep(default_timeout)
time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
iface.mutex.Lock()
delete(iface.calls, callname)
iface.mutex.Unlock()
t.mutex.Lock()
delete(t.calls, callname)
t.mutex.Unlock()
}()
var conn net.Conn
var err error
if socksaddr != nil {
socksaddr, issocks := options.(string)
if issocks {
if sintf != "" {
return
}
dialerdst, er := net.ResolveTCPAddr("tcp", socksaddr)
if er != nil {
return
}
var dialer proxy.Dialer
dialer, err = proxy.SOCKS5("tcp", *socksaddr, nil, proxy.Direct)
dialer, err = proxy.SOCKS5("tcp", dialerdst.String(), nil, proxy.Direct)
if err != nil {
return
}
@ -210,9 +257,10 @@ func (iface *tcpInterface) call(saddr string, socksaddr *string, sintf string) {
addr: saddr,
},
}
t.handler(conn, false, dialerdst.String())
} else {
dialer := net.Dialer{
Control: iface.tcpContext,
Control: t.tcpContext,
}
if sintf != "" {
ief, err := net.InterfaceByName(sintf)
@ -266,26 +314,34 @@ func (iface *tcpInterface) call(saddr string, socksaddr *string, sintf string) {
if err != nil {
return
}
t.handler(conn, false, nil)
}
iface.handler(conn, false)
}()
}
func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
func (t *tcp) handler(sock net.Conn, incoming bool, options interface{}) {
defer sock.Close()
iface.setExtraOptions(sock)
t.setExtraOptions(sock)
stream := stream{}
stream.init(sock)
local, _, _ := net.SplitHostPort(sock.LocalAddr().String())
remote, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
remotelinklocal := net.ParseIP(remote).IsLinkLocalUnicast()
name := "tcp://" + sock.RemoteAddr().String()
link, err := iface.core.link.create(&stream, name, "tcp", local, remote, incoming, remotelinklocal)
var name string
var proto string
if socksaddr, issocks := options.(string); issocks {
name = "socks://" + socksaddr + "/" + sock.RemoteAddr().String()
proto = "socks"
} else {
name = "tcp://" + sock.RemoteAddr().String()
proto = "tcp"
}
link, err := t.link.core.link.create(&stream, name, proto, local, remote, incoming, remotelinklocal)
if err != nil {
iface.core.log.Println(err)
t.link.core.log.Println(err)
panic(err)
}
iface.core.log.Debugln("DEBUG: starting handler for", name)
t.link.core.log.Debugln("DEBUG: starting handler for", name)
err = link.handler()
iface.core.log.Debugln("DEBUG: stopped handler for", name, err)
t.link.core.log.Debugln("DEBUG: stopped handler for", name, err)
}

2
src/yggdrasil/tcp_darwin.go
View file

@ -10,7 +10,7 @@ import (
// WARNING: This context is used both by net.Dialer and net.Listen in tcp.go
func (iface *tcpInterface) tcpContext(network, address string, c syscall.RawConn) error {
func (t *tcp) tcpContext(network, address string, c syscall.RawConn) error {
var control error
var recvanyif error

2
src/yggdrasil/tcp_other.go
View file

@ -8,6 +8,6 @@ import (
// WARNING: This context is used both by net.Dialer and net.Listen in tcp.go
func (iface *tcpInterface) tcpContext(network, address string, c syscall.RawConn) error {
func (t *tcp) tcpContext(network, address string, c syscall.RawConn) error {
return nil
}