diff --git a/derp/derp.go b/derp/derp.go index 5c4eee226..87d714a6c 100644 --- a/derp/derp.go +++ b/derp/derp.go @@ -67,12 +67,13 @@ Steady state: * server then sends frameRecvPacket to recipient */ const ( - frameServerKey = frameType(0x01) // 8B magic + 32B public key + (0+ bytes future use) - frameClientInfo = frameType(0x02) // 32B pub key + 24B nonce + naclbox(json) - frameServerInfo = frameType(0x03) // 24B nonce + naclbox(json) - frameSendPacket = frameType(0x04) // 32B dest pub key + packet bytes - frameRecvPacket = frameType(0x05) // v0/1: packet bytes, v2: 32B src pub key + packet bytes - frameKeepAlive = frameType(0x06) // no payload, no-op (to be replaced with ping/pong) + frameServerKey = frameType(0x01) // 8B magic + 32B public key + (0+ bytes future use) + frameClientInfo = frameType(0x02) // 32B pub key + 24B nonce + naclbox(json) + frameServerInfo = frameType(0x03) // 24B nonce + naclbox(json) + frameSendPacket = frameType(0x04) // 32B dest pub key + packet bytes + frameRecvPacket = frameType(0x05) // v0/1: packet bytes, v2: 32B src pub key + packet bytes + frameKeepAlive = frameType(0x06) // no payload, no-op (to be replaced with ping/pong) + frameNotePreferred = frameType(0x07) // 1 byte payload: 0x01 or 0x00 for whether this is client's home node ) var bin = binary.BigEndian diff --git a/derp/derp_client.go b/derp/derp_client.go index 5779eb8c8..79f713210 100644 --- a/derp/derp_client.go +++ b/derp/derp_client.go @@ -12,6 +12,7 @@ import ( "fmt" "io" "net" + "sync" "time" "golang.org/x/crypto/nacl/box" @@ -27,8 +28,10 @@ type Client struct { logf logger.Logf nc net.Conn br *bufio.Reader - bw *bufio.Writer - readErr error // sticky read error + + wmu sync.Mutex // hold while writing to bw + bw *bufio.Writer + readErr error // sticky read error } func NewClient(privateKey key.Private, nc net.Conn, brw *bufio.ReadWriter, logf logger.Logf) (*Client, error) { @@ -143,6 +146,9 @@ func (c *Client) send(dstKey key.Public, pkt []byte) (ret error) { return fmt.Errorf("packet too big: %d", len(pkt)) } + c.wmu.Lock() + defer c.wmu.Unlock() + if err := writeFrameHeader(c.bw, frameSendPacket, uint32(len(dstKey)+len(pkt))); err != nil { return err } @@ -155,6 +161,32 @@ func (c *Client) send(dstKey key.Public, pkt []byte) (ret error) { return c.bw.Flush() } +// NotePreferred sends a packet that tells the server whether this +// client is the user's preferred server. This is only used in the +// server for stats. +func (c *Client) NotePreferred(preferred bool) (err error) { + defer func() { + if err != nil { + err = fmt.Errorf("derp.NotePreferred: %v", err) + } + }() + + c.wmu.Lock() + defer c.wmu.Unlock() + + if err := writeFrameHeader(c.bw, frameNotePreferred, 1); err != nil { + return err + } + var b byte = 0x00 + if preferred { + b = 0x01 + } + if err := c.bw.WriteByte(b); err != nil { + return err + } + return c.bw.Flush() +} + // ReceivedMessage represents a type returned by Client.Recv. Unless // otherwise documented, the returned message aliases the byte slice // provided to Recv and thus the message is only as good as that diff --git a/derp/derp_server.go b/derp/derp_server.go index 17d431ad4..a30c66979 100644 --- a/derp/derp_server.go +++ b/derp/derp_server.go @@ -16,6 +16,7 @@ import ( "expvar" "fmt" "io" + "io/ioutil" "math/big" "net" "os" @@ -47,6 +48,9 @@ type Server struct { packetsRecv, bytesRecv expvar.Int packetsDropped expvar.Int accepts expvar.Int + curClients expvar.Int + curHomeClients expvar.Int // ones with preferred + unknownFrames expvar.Int mu sync.Mutex closed bool @@ -137,13 +141,14 @@ func (s *Server) registerClient(c *sclient) { defer s.mu.Unlock() old := s.clients[c.key] if old == nil { - s.logf("derp: %s: client %x: adding connection", c.nc.RemoteAddr(), c.key) + c.logf("adding connection") } else { old.nc.Close() - s.logf("derp: %s: client %x: adding connection, replacing %s", c.nc.RemoteAddr(), c.key, old.nc.RemoteAddr()) + c.logf("adding connection, replacing %s", old.nc.RemoteAddr()) } s.clients[c.key] = c s.clientsEver[c.key] = true + s.curClients.Add(1) } // unregisterClient removes a client from the server. @@ -152,9 +157,14 @@ func (s *Server) unregisterClient(c *sclient) { defer s.mu.Unlock() cur := s.clients[c.key] if cur == c { - s.logf("derp: %s: client %x: removing connection", c.nc.RemoteAddr(), c.key) + c.logf("removing connection") delete(s.clients, c.key) } + + s.curClients.Add(-1) + if c.preferred { + s.curHomeClients.Add(-1) + } } func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error { @@ -172,14 +182,24 @@ func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error { return fmt.Errorf("client %x rejected: %v", clientKey, err) } + lim := rate.Inf + if s.BytesPerSecond != 0 { + lim = rate.Limit(s.BytesPerSecond) + } + const burstBytes = 1 << 20 // generous bandwidth delay product? must be over 64k max packet size. + limiter := rate.NewLimiter(lim, burstBytes) + // At this point we trust the client so we don't time out. nc.SetDeadline(time.Time{}) c := &sclient{ - key: clientKey, - nc: nc, - br: br, - bw: bw, + s: s, + key: clientKey, + nc: nc, + br: br, + bw: bw, + limiter: limiter, + logf: logger.WithPrefix(s.logf, fmt.Sprintf("derp client %v/%x: ", nc.RemoteAddr(), clientKey)), } if clientInfo != nil { c.info = *clientInfo @@ -199,64 +219,96 @@ func (s *Server) accept(nc net.Conn, brw *bufio.ReadWriter) error { } defer s.unregisterClient(c) + return c.run() +} + +func (c *sclient) run() error { + s := c.s ctx, cancel := context.WithCancel(context.Background()) defer cancel() - go s.sendClientKeepAlives(ctx, c) - lim := rate.Inf - if s.BytesPerSecond != 0 { - lim = rate.Limit(s.BytesPerSecond) - } - const burstBytes = 1 << 20 // generous bandwidth delay product? must be over 64k max packet size. - limiter := rate.NewLimiter(lim, burstBytes) + go s.sendClientKeepAlives(ctx, c) for { ft, fl, err := readFrameHeader(c.br) if err != nil { return fmt.Errorf("client %x: readFrameHeader: %v", c.key, err) } - if ft != frameSendPacket { - // TODO: nothing else yet supported - return fmt.Errorf("client %x: unsupported frame %v", c.key, ft) + switch ft { + case frameNotePreferred: + err = c.handleFrameNotePreferred(ft, fl) + case frameSendPacket: + err = c.handleFrameSendPacket(ctx, ft, fl) + default: + err = c.handleUnknownFrame(ctx, ft, fl) } - dstKey, contents, err := s.recvPacket(ctx, c.br, fl, limiter) if err != nil { - return fmt.Errorf("client %x: recvPacket: %v", c.key, err) + return err } + } +} - s.mu.Lock() - dst := s.clients[dstKey] - s.mu.Unlock() +func (c *sclient) handleUnknownFrame(ctx context.Context, ft frameType, fl uint32) error { + if err := c.limiter.WaitN(ctx, int(fl)); err != nil { + return fmt.Errorf("rate limit: %v", err) + } + _, err := io.CopyN(ioutil.Discard, c.br, int64(fl)) + return err +} - if dst == nil { - s.packetsDropped.Add(1) - if debug { - s.logf("derp: %s: client %x: dropping packet for unknown %x", nc.RemoteAddr(), c.key, dstKey) - } - continue +func (c *sclient) handleFrameNotePreferred(ft frameType, fl uint32) error { + if fl != 1 { + return fmt.Errorf("frameNotePreferred wrong size") + } + v, err := c.br.ReadByte() + if err != nil { + return fmt.Errorf("frameNotePreferred ReadByte: %v", err) + } + c.setPreferred(v != 0) + return nil +} + +func (c *sclient) handleFrameSendPacket(ctx context.Context, ft frameType, fl uint32) error { + s := c.s + + dstKey, contents, err := s.recvPacket(ctx, c.br, fl, c.limiter) + if err != nil { + return fmt.Errorf("client %x: recvPacket: %v", c.key, err) + } + + s.mu.Lock() + dst := s.clients[dstKey] + s.mu.Unlock() + + if dst == nil { + s.packetsDropped.Add(1) + if debug { + c.logf("dropping packet for unknown %x", dstKey) } + return nil + } - dst.mu.Lock() - err = s.sendPacket(dst.bw, &dst.info, c.key, contents) - dst.mu.Unlock() + dst.mu.Lock() + err = s.sendPacket(dst.bw, &dst.info, c.key, contents) + dst.mu.Unlock() - if err != nil { - s.logf("derp: %s: client %x: dropping packet for %x: %v", nc.RemoteAddr(), c.key, dstKey, err) + if err != nil { + c.logf("write error sending packet to %x: %v", dstKey, err) - // If we cannot send to a destination, shut it down. - // Let its receive loop do the cleanup. - s.mu.Lock() - if s.clients[dstKey] == dst { - s.clients[dstKey].nc.Close() - } - s.mu.Unlock() + // If we cannot send to a destination, shut it down. + // Let its receive loop do the cleanup. + s.mu.Lock() + if s.clients[dstKey] == dst { + s.clients[dstKey].nc.Close() } + s.mu.Unlock() } + return err } func (s *Server) sendClientKeepAlives(ctx context.Context, c *sclient) { if err := c.keepAliveLoop(ctx); err != nil { - s.logf("derp: %s: client %x: keep alive failed: %v", c.nc.RemoteAddr(), c.key, err) + c.logf("keep alive failed: %v", err) } } @@ -392,18 +444,34 @@ func (s *Server) recvPacket(ctx context.Context, br *bufio.Reader, frameLen uint // // (The "s" prefix is to more explicitly distinguish it from Client in derp_client.go) type sclient struct { - nc net.Conn - key key.Public - info clientInfo + s *Server + nc net.Conn + key key.Public + info clientInfo + logf logger.Logf + limiter *rate.Limiter keepAliveTimer *time.Timer keepAliveReset chan struct{} + preferred bool + mu sync.Mutex // mu guards writing to bw br *bufio.Reader bw *bufio.Writer } +func (c *sclient) setPreferred(v bool) { + if c.preferred == v { + return + } + if v { + c.s.curHomeClients.Add(1) + } else { + c.s.curHomeClients.Add(-1) + } +} + func (c *sclient) keepAliveLoop(ctx context.Context) error { jitterMs, err := crand.Int(crand.Reader, big.NewInt(5000)) if err != nil { @@ -449,13 +517,15 @@ func (s *Server) expVarFunc(f func() interface{}) expvar.Func { // ExpVar returns an expvar variable suitable for registering with expvar.Publish. func (s *Server) ExpVar() expvar.Var { m := new(metrics.Set) - m.Set("gauge_current_connnections", s.expVarFunc(func() interface{} { return len(s.netConns) })) m.Set("counter_unique_clients_ever", s.expVarFunc(func() interface{} { return len(s.clientsEver) })) + m.Set("gauge_current_connnections", &s.curClients) + m.Set("gauge_current_home_connnections", &s.curHomeClients) m.Set("accepts", &s.accepts) m.Set("bytes_received", &s.bytesRecv) m.Set("bytes_sent", &s.bytesSent) m.Set("packets_dropped", &s.packetsDropped) m.Set("packets_sent", &s.packetsSent) m.Set("packets_received", &s.packetsRecv) + m.Set("unknown_frames", &s.unknownFrames) return m } diff --git a/derp/derphttp/derphttp_client.go b/derp/derphttp/derphttp_client.go index 08c53ee90..7765a8e1f 100644 --- a/derp/derphttp/derphttp_client.go +++ b/derp/derphttp/derphttp_client.go @@ -48,10 +48,11 @@ type Client struct { ctx context.Context // closed via cancelCtx in Client.Close cancelCtx context.CancelFunc - mu sync.Mutex - closed bool - netConn io.Closer - client *derp.Client + mu sync.Mutex + preferred bool + closed bool + netConn io.Closer + client *derp.Client } // NewClient returns a new DERP-over-HTTP client. It connects lazily. @@ -223,6 +224,12 @@ func (c *Client) connect(ctx context.Context, caller string) (client *derp.Clien if err != nil { return nil, err } + if c.preferred { + if err := derpClient.NotePreferred(true); err != nil { + go httpConn.Close() + return nil, err + } + } c.client = derpClient c.netConn = tcpConn @@ -240,6 +247,25 @@ func (c *Client) Send(dstKey key.Public, b []byte) error { return err } +// NotePreferred notes whether this Client is the caller's preferred +// (home) DERP node. It's only used for stats. +func (c *Client) NotePreferred(v bool) { + c.mu.Lock() + if c.preferred == v { + c.mu.Unlock() + return + } + c.preferred = v + client := c.client + c.mu.Unlock() + + if client != nil { + if err := client.NotePreferred(v); err != nil { + c.closeForReconnect() + } + } +} + func (c *Client) Recv(b []byte) (derp.ReceivedMessage, error) { client, err := c.connect(context.TODO(), "derphttp.Client.Recv") if err != nil { diff --git a/wgengine/magicsock/magicsock.go b/wgengine/magicsock/magicsock.go index a8160f38b..140a515e0 100644 --- a/wgengine/magicsock/magicsock.go +++ b/wgengine/magicsock/magicsock.go @@ -348,11 +348,18 @@ func (c *Conn) setNearestDERP(derpNum int) (wantDERP bool) { c.myDerp = 0 return false } + if derpNum == c.myDerp { + // No change. + return true + } + c.myDerp = derpNum + for i, ad := range c.activeDerp { + go ad.c.NotePreferred(i == c.myDerp) + } if derpNum != 0 && derpNum != c.myDerp { // On change, start connecting to it: go c.derpWriteChanOfAddr(&net.UDPAddr{IP: derpMagicIP, Port: derpNum}) } - c.myDerp = derpNum return true } @@ -641,6 +648,7 @@ func (c *Conn) derpWriteChanOfAddr(addr *net.UDPAddr) chan<- derpWriteRequest { c.logf("derphttp.NewClient: port %d, host %q invalid? err: %v", addr.Port, host, err) return nil } + dc.NotePreferred(c.myDerp == addr.Port) dc.DNSCache = dnscache.Get() dc.TLSConfig = c.derpTLSConfig