yggdrasil-go/src/yggdrasil/tcp.go

297 lines
7.7 KiB
Go
Raw Normal View History

2017-12-29 07:16:20 +03:00
package yggdrasil
// This sends packets to peers using TCP as a transport
// It's generally better tested than the UDP implementation
// Using it regularly is insane, but I find TCP easier to test/debug with it
// Updating and optimizing the UDP version is a higher priority
// TODO:
// Something needs to make sure we're getting *valid* packets
// Could be used to DoS (connect, give someone else's keys, spew garbage)
// I guess the "peer" part should watch for link packets, disconnect?
// TCP connections start with a metadata exchange.
// It involves exchanging version numbers and crypto keys
// See version.go for version metadata format
2018-06-13 01:50:08 +03:00
import (
2019-01-13 21:08:41 +03:00
"context"
2018-06-13 01:50:08 +03:00
"fmt"
2018-06-21 18:39:43 +03:00
"math/rand"
2018-06-13 01:50:08 +03:00
"net"
"sync"
"time"
"golang.org/x/net/proxy"
"github.com/yggdrasil-network/yggdrasil-go/src/crypto"
2018-06-13 01:50:08 +03:00
)
2017-12-29 07:16:20 +03:00
const default_timeout = 6 * time.Second
const tcp_ping_interval = (default_timeout * 2 / 3)
2017-12-29 07:16:20 +03:00
// The TCP listener and information about active TCP connections, to avoid duplication.
2017-12-29 07:16:20 +03:00
type tcpInterface struct {
core *Core
2018-12-30 18:21:09 +03:00
reconfigure chan chan error
serv net.Listener
stop chan bool
timeout time.Duration
addr string
mutex sync.Mutex // Protecting the below
calls map[string]struct{}
conns map[tcpInfo](chan struct{})
2017-12-29 07:16:20 +03:00
}
// 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).
2018-02-18 05:44:23 +03:00
type tcpInfo struct {
box crypto.BoxPubKey
sig crypto.SigPubKey
localAddr string
2018-04-20 15:41:09 +03:00
remoteAddr string
2017-12-29 07:16:20 +03:00
}
// Wrapper function to set additional options for specific connection types.
func (iface *tcpInterface) setExtraOptions(c net.Conn) {
switch sock := c.(type) {
case *net.TCPConn:
sock.SetNoDelay(true)
// TODO something for socks5
default:
}
}
// Returns the address of the listener.
2018-05-28 00:13:37 +03:00
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)
2018-05-28 00:13:37 +03:00
}
// Attempst to initiate a connection to the provided address, viathe provided socks proxy address.
2018-05-28 00:13:37 +03:00
func (iface *tcpInterface) connectSOCKS(socksaddr, peeraddr string) {
iface.call(peeraddr, &socksaddr, "")
2018-05-28 00:13:37 +03:00
}
// Initializes the struct.
2018-12-29 22:14:26 +03:00
func (iface *tcpInterface) init(core *Core) (err error) {
2018-01-05 01:37:51 +03:00
iface.core = core
iface.stop = make(chan bool, 1)
2018-12-30 18:21:09 +03:00
iface.reconfigure = make(chan chan error, 1)
go func() {
for {
2019-01-15 11:51:19 +03:00
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
2019-01-15 11:51:19 +03:00
iface.serv.Close()
e <- iface.listen()
} else {
e <- nil
2018-12-30 18:21:09 +03:00
}
}
}()
return iface.listen()
}
func (iface *tcpInterface) listen() error {
var err error
2018-12-29 22:53:31 +03:00
iface.core.configMutex.RLock()
iface.addr = iface.core.config.Listen
iface.timeout = time.Duration(iface.core.config.ReadTimeout) * time.Millisecond
2018-12-29 22:53:31 +03:00
iface.core.configMutex.RUnlock()
2018-12-30 18:21:09 +03:00
if iface.timeout >= 0 && iface.timeout < default_timeout {
iface.timeout = default_timeout
}
2019-01-13 21:08:41 +03:00
ctx := context.Background()
lc := net.ListenConfig{
Control: iface.tcpContext,
}
iface.serv, err = lc.Listen(ctx, "tcp", iface.addr)
2018-04-19 17:30:40 +03:00
if err == nil {
2019-01-16 16:20:12 +03:00
iface.mutex.Lock()
2018-04-19 17:30:40 +03:00
iface.calls = make(map[string]struct{})
iface.conns = make(map[tcpInfo](chan struct{}))
2019-01-16 16:20:12 +03:00
iface.mutex.Unlock()
2018-04-19 17:30:40 +03:00
go iface.listener()
2018-12-30 18:21:09 +03:00
return nil
2018-01-05 01:37:51 +03:00
}
2018-05-28 00:13:37 +03:00
return err
2017-12-29 07:16:20 +03:00
}
// Runs the listener, which spawns off goroutines for incoming connections.
2017-12-29 07:16:20 +03:00
func (iface *tcpInterface) listener() {
2018-01-05 01:37:51 +03:00
defer iface.serv.Close()
iface.core.log.Println("Listening for TCP on:", iface.serv.Addr().String())
2018-01-05 01:37:51 +03:00
for {
2018-04-19 17:30:40 +03:00
sock, err := iface.serv.Accept()
if err != nil {
iface.core.log.Println("Failed to accept connection:", err)
return
}
2018-12-30 18:21:09 +03:00
select {
case <-iface.stop:
2018-12-30 18:21:09 +03:00
iface.core.log.Println("Stopping listener")
return
default:
if err != nil {
panic(err)
}
go iface.handler(sock, true)
2018-01-05 01:37:51 +03:00
}
}
2017-12-29 07:16:20 +03:00
}
// 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]
return isIn
}
// Checks if a connection already exists.
// If not, it adds it to the list of active outgoing calls (to block future attempts) and dials the address.
// 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) {
2018-01-05 01:37:51 +03:00
go func() {
callname := saddr
if sintf != "" {
callname = fmt.Sprintf("%s/%s", saddr, sintf)
}
if iface.isAlreadyCalling(callname) {
2018-06-14 17:11:34 +03:00
return
}
2019-01-16 16:20:12 +03:00
iface.mutex.Lock()
iface.calls[callname] = struct{}{}
2019-01-16 16:20:12 +03:00
iface.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()
}()
2018-06-14 17:11:34 +03:00
var conn net.Conn
var err error
if socksaddr != nil {
2018-09-25 21:46:06 +03:00
if sintf != "" {
return
}
var dialer proxy.Dialer
dialer, err = proxy.SOCKS5("tcp", *socksaddr, nil, proxy.Direct)
if err != nil {
return
}
2018-06-14 17:11:34 +03:00
conn, err = dialer.Dial("tcp", saddr)
if err != nil {
return
}
conn = &wrappedConn{
c: conn,
raddr: &wrappedAddr{
network: "tcp",
addr: saddr,
},
}
} else {
2019-01-13 21:08:41 +03:00
dialer := net.Dialer{
Control: iface.tcpContext,
}
if sintf != "" {
ief, err := net.InterfaceByName(sintf)
if err != nil {
return
}
if ief.Flags&net.FlagUp == 0 {
return
}
addrs, err := ief.Addrs()
if err == nil {
dst, err := net.ResolveTCPAddr("tcp", saddr)
if err != nil {
return
2018-09-25 21:46:06 +03:00
}
for addrindex, addr := range addrs {
src, _, err := net.ParseCIDR(addr.String())
if err != nil {
continue
}
2019-01-18 02:06:59 +03:00
if src.Equal(dst.IP) {
continue
}
if !src.IsGlobalUnicast() && !src.IsLinkLocalUnicast() {
continue
}
bothglobal := src.IsGlobalUnicast() == dst.IP.IsGlobalUnicast()
bothlinklocal := src.IsLinkLocalUnicast() == dst.IP.IsLinkLocalUnicast()
if !bothglobal && !bothlinklocal {
continue
}
if (src.To4() != nil) != (dst.IP.To4() != nil) {
continue
}
if bothglobal || bothlinklocal || addrindex == len(addrs)-1 {
dialer.LocalAddr = &net.TCPAddr{
IP: src,
Port: 0,
Zone: sintf,
}
2019-01-18 02:06:59 +03:00
break
}
}
if dialer.LocalAddr == nil {
return
}
}
}
conn, err = dialer.Dial("tcp", saddr)
2018-01-05 01:37:51 +03:00
if err != nil {
return
}
}
2018-06-14 17:11:34 +03:00
iface.handler(conn, false)
2018-01-05 01:37:51 +03:00
}()
2017-12-29 07:16:20 +03:00
}
func (iface *tcpInterface) handler(sock net.Conn, incoming bool) {
defer sock.Close()
iface.setExtraOptions(sock)
stream := stream{}
2019-01-23 18:16:22 +03:00
stream.init(sock)
local, _, _ := net.SplitHostPort(sock.LocalAddr().String())
remote, _, _ := net.SplitHostPort(sock.RemoteAddr().String())
name := "tcp://" + sock.RemoteAddr().String()
link, err := iface.core.link.create(&stream, name, "tcp", local, remote)
if err != nil {
iface.core.log.Println(err)
panic(err)
}
iface.core.log.Println("DEBUG: starting handler for", name)
2019-01-23 06:48:43 +03:00
err = link.handler()
iface.core.log.Println("DEBUG: stopped handler for", name, err)
}