tracker/udp/server/server.go

   1 package udpTrackerServer
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "crypto/rand"
   7         "encoding/binary"
   8         "fmt"
   9         "io"
  10         "net"
  11         "net/netip"
  12
  13         "github.com/anacrolix/dht/v2/krpc"
  14         "github.com/anacrolix/generics"
  15         "github.com/anacrolix/log"
  16         "go.opentelemetry.io/otel"
  17         "go.opentelemetry.io/otel/codes"
  18
  19         "github.com/anacrolix/torrent/tracker"
  20         "github.com/anacrolix/torrent/tracker/udp"
  21 )
  22
  23 type ConnectionTrackerAddr = string
  24
  25 type ConnectionTracker interface {
  26         Add(ctx context.Context, addr ConnectionTrackerAddr, id udp.ConnectionId) error
  27         Check(ctx context.Context, addr ConnectionTrackerAddr, id udp.ConnectionId) (bool, error)
  28 }
  29
  30 type InfoHash = [20]byte
  31
  32 type AnnounceTracker = tracker.AnnounceTracker
  33
  34 type Server struct {
  35         ConnTracker  ConnectionTracker
  36         SendResponse func(data []byte, addr net.Addr) (int, error)
  37         Announce     *tracker.AnnounceHandler
  38 }
  39
  40 type RequestSourceAddr = net.Addr
  41
  42 var tracer = otel.Tracer("torrent.tracker.udp")
  43
  44 func (me *Server) HandleRequest(
  45         ctx context.Context,
  46         family udp.AddrFamily,
  47         source RequestSourceAddr,
  48         body []byte,
  49 ) (err error) {
  50         ctx, span := tracer.Start(ctx, "Server.HandleRequest")
  51         defer span.End()
  52         defer func() {
  53                 if err != nil {
  54                         span.SetStatus(codes.Error, err.Error())
  55                 }
  56         }()
  57         var h udp.RequestHeader
  58         var r bytes.Reader
  59         r.Reset(body)
  60         err = udp.Read(&r, &h)
  61         if err != nil {
  62                 err = fmt.Errorf("reading request header: %w", err)
  63                 return err
  64         }
  65         switch h.Action {
  66         case udp.ActionConnect:
  67                 err = me.handleConnect(ctx, source, h.TransactionId)
  68         case udp.ActionAnnounce:
  69                 err = me.handleAnnounce(ctx, family, source, h.ConnectionId, h.TransactionId, &r)
  70         default:
  71                 err = fmt.Errorf("unimplemented")
  72         }
  73         if err != nil {
  74                 err = fmt.Errorf("handling action %v: %w", h.Action, err)
  75         }
  76         return err
  77 }
  78
  79 func (me *Server) handleAnnounce(
  80         ctx context.Context,
  81         addrFamily udp.AddrFamily,
  82         source RequestSourceAddr,
  83         connId udp.ConnectionId,
  84         tid udp.TransactionId,
  85         r *bytes.Reader,
  86 ) error {
  87         // Should we set a timeout of 10s or something for the entire response, so that we give up if a
  88         // retry is imminent?
  89
  90         ok, err := me.ConnTracker.Check(ctx, source.String(), connId)
  91         if err != nil {
  92                 err = fmt.Errorf("checking conn id: %w", err)
  93                 return err
  94         }
  95         if !ok {
  96                 return fmt.Errorf("incorrect connection id: %x", connId)
  97         }
  98         var req udp.AnnounceRequest
  99         err = udp.Read(r, &req)
 100         if err != nil {
 101                 return err
 102         }
 103         // TODO: This should be done asynchronously to responding to the announce.
 104         announceAddr, err := netip.ParseAddrPort(source.String())
 105         if err != nil {
 106                 err = fmt.Errorf("converting source net.Addr to AnnounceAddr: %w", err)
 107                 return err
 108         }
 109         opts := tracker.GetPeersOpts{MaxCount: generics.Some[uint](50)}
 110         if addrFamily == udp.AddrFamilyIpv4 {
 111                 opts.MaxCount = generics.Some[uint](150)
 112         }
 113         res := me.Announce.Serve(ctx, req, announceAddr, opts)
 114         if res.Err != nil {
 115                 return res.Err
 116         }
 117         nodeAddrs := make([]krpc.NodeAddr, 0, len(res.Peers))
 118         for _, p := range res.Peers {
 119                 var ip net.IP
 120                 switch addrFamily {
 121                 default:
 122                         continue
 123                 case udp.AddrFamilyIpv4:
 124                         if !p.Addr().Unmap().Is4() {
 125                                 continue
 126                         }
 127                         ipBuf := p.Addr().As4()
 128                         ip = ipBuf[:]
 129                 case udp.AddrFamilyIpv6:
 130                         ipBuf := p.Addr().As16()
 131                         ip = ipBuf[:]
 132                 }
 133                 nodeAddrs = append(nodeAddrs, krpc.NodeAddr{
 134                         IP:   ip[:],
 135                         Port: int(p.Port()),
 136                 })
 137         }
 138         var buf bytes.Buffer
 139         err = udp.Write(&buf, udp.ResponseHeader{
 140                 Action:        udp.ActionAnnounce,
 141                 TransactionId: tid,
 142         })
 143         if err != nil {
 144                 return err
 145         }
 146         err = udp.Write(&buf, udp.AnnounceResponseHeader{
 147                 Interval: res.Interval.UnwrapOr(5 * 60),
 148         })
 149         if err != nil {
 150                 return err
 151         }
 152         b, err := udp.GetNodeAddrsCompactMarshaler(nodeAddrs, addrFamily).MarshalBinary()
 153         if err != nil {
 154                 err = fmt.Errorf("marshalling compact node addrs: %w", err)
 155                 return err
 156         }
 157         buf.Write(b)
 158         n, err := me.SendResponse(buf.Bytes(), source)
 159         if err != nil {
 160                 return err
 161         }
 162         if n < buf.Len() {
 163                 err = io.ErrShortWrite
 164         }
 165         return err
 166 }
 167
 168 func (me *Server) handleConnect(ctx context.Context, source RequestSourceAddr, tid udp.TransactionId) error {
 169         connId := randomConnectionId()
 170         err := me.ConnTracker.Add(ctx, source.String(), connId)
 171         if err != nil {
 172                 err = fmt.Errorf("recording conn id: %w", err)
 173                 return err
 174         }
 175         var buf bytes.Buffer
 176         udp.Write(&buf, udp.ResponseHeader{
 177                 Action:        udp.ActionConnect,
 178                 TransactionId: tid,
 179         })
 180         udp.Write(&buf, udp.ConnectionResponse{connId})
 181         n, err := me.SendResponse(buf.Bytes(), source)
 182         if err != nil {
 183                 return err
 184         }
 185         if n < buf.Len() {
 186                 err = io.ErrShortWrite
 187         }
 188         return err
 189 }
 190
 191 func randomConnectionId() udp.ConnectionId {
 192         var b [8]byte
 193         _, err := rand.Read(b[:])
 194         if err != nil {
 195                 panic(err)
 196         }
 197         return binary.BigEndian.Uint64(b[:])
 198 }
 199
 200 func RunSimple(ctx context.Context, s *Server, pc net.PacketConn, family udp.AddrFamily) error {
 201         ctx, cancel := context.WithCancel(ctx)
 202         defer cancel()
 203         for {
 204                 var b [1500]byte
 205                 n, addr, err := pc.ReadFrom(b[:])
 206                 if err != nil {
 207                         return err
 208                 }
 209                 go func() {
 210                         err := s.HandleRequest(ctx, family, addr, b[:n])
 211                         if err != nil {
 212                                 log.Printf("error handling %v byte request from %v: %v", n, addr, err)
 213                         }
 214                 }()
 215         }
 216 }