mirror of
https://github.com/tailscale/tailscale.git
synced 2024-11-26 03:25:35 +00:00
394 lines
8.6 KiB
Go
394 lines
8.6 KiB
Go
// Copyright (c) 2020 Tailscale Inc & AUTHORS All rights reserved.
|
|
// Use of this source code is governed by a BSD-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package derp
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
crand "crypto/rand"
|
|
"errors"
|
|
"expvar"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"testing"
|
|
"time"
|
|
|
|
"tailscale.com/net/nettest"
|
|
"tailscale.com/types/key"
|
|
)
|
|
|
|
func newPrivateKey(t *testing.T) (k key.Private) {
|
|
t.Helper()
|
|
if _, err := crand.Read(k[:]); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return
|
|
}
|
|
|
|
func TestSendRecv(t *testing.T) {
|
|
serverPrivateKey := newPrivateKey(t)
|
|
s := NewServer(serverPrivateKey, t.Logf)
|
|
defer s.Close()
|
|
|
|
const numClients = 3
|
|
var clientPrivateKeys []key.Private
|
|
var clientKeys []key.Public
|
|
for i := 0; i < numClients; i++ {
|
|
priv := newPrivateKey(t)
|
|
clientPrivateKeys = append(clientPrivateKeys, priv)
|
|
clientKeys = append(clientKeys, priv.Public())
|
|
}
|
|
|
|
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer ln.Close()
|
|
|
|
var clients []*Client
|
|
var connsOut []Conn
|
|
var recvChs []chan []byte
|
|
errCh := make(chan error, 3)
|
|
|
|
for i := 0; i < numClients; i++ {
|
|
t.Logf("Connecting client %d ...", i)
|
|
cout, err := net.Dial("tcp", ln.Addr().String())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer cout.Close()
|
|
connsOut = append(connsOut, cout)
|
|
|
|
cin, err := ln.Accept()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer cin.Close()
|
|
brwServer := bufio.NewReadWriter(bufio.NewReader(cin), bufio.NewWriter(cin))
|
|
go s.Accept(cin, brwServer, fmt.Sprintf("test-client-%d", i))
|
|
|
|
key := clientPrivateKeys[i]
|
|
brw := bufio.NewReadWriter(bufio.NewReader(cout), bufio.NewWriter(cout))
|
|
c, err := NewClient(key, cout, brw, t.Logf)
|
|
if err != nil {
|
|
t.Fatalf("client %d: %v", i, err)
|
|
}
|
|
clients = append(clients, c)
|
|
recvChs = append(recvChs, make(chan []byte))
|
|
t.Logf("Connected client %d.", i)
|
|
}
|
|
|
|
var peerGoneCount expvar.Int
|
|
|
|
t.Logf("Starting read loops")
|
|
for i := 0; i < numClients; i++ {
|
|
go func(i int) {
|
|
for {
|
|
b := make([]byte, 1<<16)
|
|
m, err := clients[i].Recv(b)
|
|
if err != nil {
|
|
errCh <- err
|
|
return
|
|
}
|
|
switch m := m.(type) {
|
|
default:
|
|
t.Errorf("unexpected message type %T", m)
|
|
continue
|
|
case PeerGoneMessage:
|
|
peerGoneCount.Add(1)
|
|
case ReceivedPacket:
|
|
if m.Source.IsZero() {
|
|
t.Errorf("zero Source address in ReceivedPacket")
|
|
}
|
|
recvChs[i] <- m.Data
|
|
}
|
|
}
|
|
}(i)
|
|
}
|
|
|
|
recv := func(i int, want string) {
|
|
t.Helper()
|
|
select {
|
|
case b := <-recvChs[i]:
|
|
if got := string(b); got != want {
|
|
t.Errorf("client1.Recv=%q, want %q", got, want)
|
|
}
|
|
case <-time.After(1 * time.Second):
|
|
t.Errorf("client%d.Recv, got nothing, want %q", i, want)
|
|
}
|
|
}
|
|
recvNothing := func(i int) {
|
|
t.Helper()
|
|
select {
|
|
case b := <-recvChs[0]:
|
|
t.Errorf("client%d.Recv=%q, want nothing", i, string(b))
|
|
default:
|
|
}
|
|
}
|
|
|
|
wantActive := func(total, home int64) {
|
|
t.Helper()
|
|
dl := time.Now().Add(5 * time.Second)
|
|
var gotTotal, gotHome int64
|
|
for time.Now().Before(dl) {
|
|
gotTotal, gotHome = s.curClients.Value(), s.curHomeClients.Value()
|
|
if gotTotal == total && gotHome == home {
|
|
return
|
|
}
|
|
time.Sleep(10 * time.Millisecond)
|
|
}
|
|
t.Errorf("total/home=%v/%v; want %v/%v", gotTotal, gotHome, total, home)
|
|
}
|
|
|
|
wantClosedPeers := func(want int64) {
|
|
t.Helper()
|
|
var got int64
|
|
dl := time.Now().Add(5 * time.Second)
|
|
for time.Now().Before(dl) {
|
|
if got = peerGoneCount.Value(); got == want {
|
|
return
|
|
}
|
|
}
|
|
t.Errorf("peer gone count = %v; want %v", got, want)
|
|
}
|
|
|
|
msg1 := []byte("hello 0->1\n")
|
|
if err := clients[0].Send(clientKeys[1], msg1); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
recv(1, string(msg1))
|
|
recvNothing(0)
|
|
recvNothing(2)
|
|
|
|
msg2 := []byte("hello 1->2\n")
|
|
if err := clients[1].Send(clientKeys[2], msg2); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
recv(2, string(msg2))
|
|
recvNothing(0)
|
|
recvNothing(1)
|
|
|
|
wantActive(3, 0)
|
|
clients[0].NotePreferred(true)
|
|
wantActive(3, 1)
|
|
clients[0].NotePreferred(true)
|
|
wantActive(3, 1)
|
|
clients[0].NotePreferred(false)
|
|
wantActive(3, 0)
|
|
clients[0].NotePreferred(false)
|
|
wantActive(3, 0)
|
|
clients[1].NotePreferred(true)
|
|
wantActive(3, 1)
|
|
connsOut[1].Close()
|
|
wantActive(2, 0)
|
|
wantClosedPeers(1)
|
|
clients[2].NotePreferred(true)
|
|
wantActive(2, 1)
|
|
clients[2].NotePreferred(false)
|
|
wantActive(2, 0)
|
|
connsOut[2].Close()
|
|
wantActive(1, 0)
|
|
wantClosedPeers(1)
|
|
|
|
t.Logf("passed")
|
|
s.Close()
|
|
|
|
}
|
|
|
|
func TestSendFreeze(t *testing.T) {
|
|
serverPrivateKey := newPrivateKey(t)
|
|
s := NewServer(serverPrivateKey, t.Logf)
|
|
defer s.Close()
|
|
s.WriteTimeout = 100 * time.Millisecond
|
|
|
|
// We send two streams of messages:
|
|
//
|
|
// alice --> bob
|
|
// alice --> cathy
|
|
//
|
|
// Then cathy stops processing messsages.
|
|
// That should not interfere with alice talking to bob.
|
|
|
|
newClient := func(name string, k key.Private) (c *Client, clientConn nettest.Conn) {
|
|
t.Helper()
|
|
c1, c2 := nettest.NewConn(name, 1024)
|
|
go s.Accept(c1, bufio.NewReadWriter(bufio.NewReader(c1), bufio.NewWriter(c1)), name)
|
|
|
|
brw := bufio.NewReadWriter(bufio.NewReader(c2), bufio.NewWriter(c2))
|
|
c, err := NewClient(k, c2, brw, t.Logf)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return c, c2
|
|
}
|
|
|
|
aliceKey := newPrivateKey(t)
|
|
aliceClient, aliceConn := newClient("alice", aliceKey)
|
|
|
|
bobKey := newPrivateKey(t)
|
|
bobClient, bobConn := newClient("bob", bobKey)
|
|
|
|
cathyKey := newPrivateKey(t)
|
|
cathyClient, cathyConn := newClient("cathy", cathyKey)
|
|
|
|
var (
|
|
aliceCh = make(chan struct{}, 32)
|
|
bobCh = make(chan struct{}, 32)
|
|
cathyCh = make(chan struct{}, 32)
|
|
)
|
|
chs := func(name string) chan struct{} {
|
|
switch name {
|
|
case "alice":
|
|
return aliceCh
|
|
case "bob":
|
|
return bobCh
|
|
case "cathy":
|
|
return cathyCh
|
|
default:
|
|
panic("unknown ch: " + name)
|
|
}
|
|
}
|
|
|
|
errCh := make(chan error, 4)
|
|
recv := func(name string, client *Client) {
|
|
ch := chs(name)
|
|
for {
|
|
b := make([]byte, 1<<9)
|
|
m, err := client.Recv(b)
|
|
if err != nil {
|
|
errCh <- fmt.Errorf("%s: %w", name, err)
|
|
return
|
|
}
|
|
switch m := m.(type) {
|
|
default:
|
|
errCh <- fmt.Errorf("%s: unexpected message type %T", name, m)
|
|
return
|
|
case ReceivedPacket:
|
|
if m.Source.IsZero() {
|
|
errCh <- fmt.Errorf("%s: zero Source address in ReceivedPacket", name)
|
|
return
|
|
}
|
|
select {
|
|
case ch <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
}
|
|
}
|
|
go recv("alice", aliceClient)
|
|
go recv("bob", bobClient)
|
|
go recv("cathy", cathyClient)
|
|
|
|
var cancel func()
|
|
go func() {
|
|
t := time.NewTicker(2 * time.Millisecond)
|
|
defer t.Stop()
|
|
var ctx context.Context
|
|
ctx, cancel = context.WithCancel(context.Background())
|
|
for {
|
|
select {
|
|
case <-t.C:
|
|
case <-ctx.Done():
|
|
errCh <- nil
|
|
return
|
|
}
|
|
|
|
msg1 := []byte("hello alice->bob\n")
|
|
if err := aliceClient.Send(bobKey.Public(), msg1); err != nil {
|
|
errCh <- fmt.Errorf("alice send to bob: %w", err)
|
|
return
|
|
}
|
|
msg2 := []byte("hello alice->cathy\n")
|
|
|
|
// TODO: an error is expected here.
|
|
// We ignore it, maybe we should log it somehow?
|
|
aliceClient.Send(cathyKey.Public(), msg2)
|
|
}
|
|
}()
|
|
|
|
drainAny := func(ch chan struct{}) {
|
|
// We are draining potentially infinite sources,
|
|
// so place some reasonable upper limit.
|
|
//
|
|
// The important thing here is to make sure that
|
|
// if any tokens remain in the channel, they
|
|
// must have been generated after drainAny was
|
|
// called.
|
|
for i := 0; i < cap(ch); i++ {
|
|
select {
|
|
case <-ch:
|
|
default:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
drain := func(t *testing.T, name string) bool {
|
|
t.Helper()
|
|
timer := time.NewTimer(1 * time.Second)
|
|
defer timer.Stop()
|
|
|
|
// Ensure ch has at least one element.
|
|
ch := chs(name)
|
|
select {
|
|
case <-ch:
|
|
case <-timer.C:
|
|
t.Errorf("no packet received by %s", name)
|
|
return false
|
|
}
|
|
// Drain remaining.
|
|
drainAny(ch)
|
|
return true
|
|
}
|
|
isEmpty := func(t *testing.T, name string) {
|
|
t.Helper()
|
|
select {
|
|
case <-chs(name):
|
|
t.Errorf("packet received by %s, want none", name)
|
|
default:
|
|
}
|
|
}
|
|
|
|
t.Run("initial send", func(t *testing.T) {
|
|
drain(t, "bob")
|
|
drain(t, "cathy")
|
|
isEmpty(t, "alice")
|
|
})
|
|
|
|
t.Run("block cathy", func(t *testing.T) {
|
|
// Block cathy. Now the cathyConn buffer will fill up quickly,
|
|
// and the derp server will back up.
|
|
cathyConn.SetReadBlock(true)
|
|
time.Sleep(2 * s.WriteTimeout)
|
|
|
|
drain(t, "bob")
|
|
drainAny(chs("cathy"))
|
|
isEmpty(t, "alice")
|
|
|
|
// Now wait a little longer, and ensure packets still flow to bob
|
|
if !drain(t, "bob") {
|
|
t.Errorf("connection alice->bob frozen by alice->cathy")
|
|
}
|
|
})
|
|
|
|
// Cleanup, make sure we process all errors.
|
|
t.Logf("TEST COMPLETE, cancelling sender")
|
|
cancel()
|
|
t.Logf("closing connections")
|
|
aliceConn.Close()
|
|
bobConn.Close()
|
|
cathyConn.Close()
|
|
|
|
for i := 0; i < cap(errCh); i++ {
|
|
err := <-errCh
|
|
if err != nil {
|
|
if errors.Is(err, io.EOF) {
|
|
continue
|
|
}
|
|
t.Error(err)
|
|
}
|
|
}
|
|
}
|