tracker/udp.go

   1 package tracker
   2
   3 import (
   4         "bytes"
   5         "encoding/binary"
   6         "errors"
   7         "fmt"
   8         "io"
   9         "math/rand"
  10         "net"
  11         "net/url"
  12         "time"
  13
  14         "github.com/anacrolix/missinggo"
  15         "github.com/anacrolix/missinggo/pproffd"
  16
  17         "github.com/anacrolix/torrent/util"
  18 )
  19
  20 type Action int32
  21
  22 const (
  23         ActionConnect Action = iota
  24         ActionAnnounce
  25         ActionScrape
  26         ActionError
  27
  28         connectRequestConnectionId = 0x41727101980
  29
  30         // BEP 41
  31         optionTypeEndOfOptions = 0
  32         optionTypeNOP          = 1
  33         optionTypeURLData      = 2
  34 )
  35
  36 type ConnectionRequest struct {
  37         ConnectionId int64
  38         Action       int32
  39         TransctionId int32
  40 }
  41
  42 type ConnectionResponse struct {
  43         ConnectionId int64
  44 }
  45
  46 type ResponseHeader struct {
  47         Action        Action
  48         TransactionId int32
  49 }
  50
  51 type RequestHeader struct {
  52         ConnectionId  int64
  53         Action        Action
  54         TransactionId int32
  55 } // 16 bytes
  56
  57 type AnnounceResponseHeader struct {
  58         Interval int32
  59         Leechers int32
  60         Seeders  int32
  61 }
  62
  63 func init() {
  64         RegisterClientScheme("udp", newClient)
  65 }
  66
  67 func newClient(url *url.URL) Client {
  68         return &udpClient{
  69                 url: *url,
  70         }
  71 }
  72
  73 func newTransactionId() int32 {
  74         return int32(rand.Uint32())
  75 }
  76
  77 func timeout(contiguousTimeouts int) (d time.Duration) {
  78         if contiguousTimeouts > 8 {
  79                 contiguousTimeouts = 8
  80         }
  81         d = 15 * time.Second
  82         for ; contiguousTimeouts > 0; contiguousTimeouts-- {
  83                 d *= 2
  84         }
  85         return
  86 }
  87
  88 type udpClient struct {
  89         contiguousTimeouts   int
  90         connectionIdReceived time.Time
  91         connectionId         int64
  92         socket               net.Conn
  93         url                  url.URL
  94 }
  95
  96 func (c *udpClient) URL() string {
  97         return c.url.String()
  98 }
  99
 100 func (c *udpClient) String() string {
 101         return c.URL()
 102 }
 103
 104 func (c *udpClient) Announce(req *AnnounceRequest) (res AnnounceResponse, err error) {
 105         if !c.connected() {
 106                 err = ErrNotConnected
 107                 return
 108         }
 109         reqURI := c.url.RequestURI()
 110         // Clearly this limits the request URI to 255 bytes. BEP 41 supports
 111         // longer but I'm not fussed.
 112         options := append([]byte{optionTypeURLData, byte(len(reqURI))}, []byte(reqURI)...)
 113         b, err := c.request(ActionAnnounce, req, options)
 114         if err != nil {
 115                 return
 116         }
 117         var h AnnounceResponseHeader
 118         err = readBody(b, &h)
 119         if err != nil {
 120                 if err == io.EOF {
 121                         err = io.ErrUnexpectedEOF
 122                 }
 123                 err = fmt.Errorf("error parsing announce response: %s", err)
 124                 return
 125         }
 126         res.Interval = h.Interval
 127         res.Leechers = h.Leechers
 128         res.Seeders = h.Seeders
 129         cps, err := util.UnmarshalIPv4CompactPeers(b.Bytes())
 130         if err != nil {
 131                 return
 132         }
 133         for _, cp := range cps {
 134                 res.Peers = append(res.Peers, Peer{
 135                         IP:   cp.IP[:],
 136                         Port: int(cp.Port),
 137                 })
 138         }
 139         return
 140 }
 141
 142 // body is the binary serializable request body. trailer is optional data
 143 // following it, such as for BEP 41.
 144 func (c *udpClient) write(h *RequestHeader, body interface{}, trailer []byte) (err error) {
 145         var buf bytes.Buffer
 146         err = binary.Write(&buf, binary.BigEndian, h)
 147         if err != nil {
 148                 panic(err)
 149         }
 150         if body != nil {
 151                 err = binary.Write(&buf, binary.BigEndian, body)
 152                 if err != nil {
 153                         panic(err)
 154                 }
 155         }
 156         _, err = buf.Write(trailer)
 157         if err != nil {
 158                 return
 159         }
 160         n, err := c.socket.Write(buf.Bytes())
 161         if err != nil {
 162                 return
 163         }
 164         if n != buf.Len() {
 165                 panic("write should send all or error")
 166         }
 167         return
 168 }
 169
 170 func read(r io.Reader, data interface{}) error {
 171         return binary.Read(r, binary.BigEndian, data)
 172 }
 173
 174 func write(w io.Writer, data interface{}) error {
 175         return binary.Write(w, binary.BigEndian, data)
 176 }
 177
 178 // args is the binary serializable request body. trailer is optional data
 179 // following it, such as for BEP 41.
 180 func (c *udpClient) request(action Action, args interface{}, options []byte) (responseBody *bytes.Buffer, err error) {
 181         tid := newTransactionId()
 182         err = c.write(&RequestHeader{
 183                 ConnectionId:  c.connectionId,
 184                 Action:        action,
 185                 TransactionId: tid,
 186         }, args, options)
 187         if err != nil {
 188                 return
 189         }
 190         c.socket.SetReadDeadline(time.Now().Add(timeout(c.contiguousTimeouts)))
 191         b := make([]byte, 0x800) // 2KiB
 192         for {
 193                 var n int
 194                 n, err = c.socket.Read(b)
 195                 if opE, ok := err.(*net.OpError); ok {
 196                         if opE.Timeout() {
 197                                 c.contiguousTimeouts++
 198                                 return
 199                         }
 200                 }
 201                 if err != nil {
 202                         return
 203                 }
 204                 buf := bytes.NewBuffer(b[:n])
 205                 var h ResponseHeader
 206                 err = binary.Read(buf, binary.BigEndian, &h)
 207                 switch err {
 208                 case io.ErrUnexpectedEOF:
 209                         continue
 210                 case nil:
 211                 default:
 212                         return
 213                 }
 214                 if h.TransactionId != tid {
 215                         continue
 216                 }
 217                 c.contiguousTimeouts = 0
 218                 if h.Action == ActionError {
 219                         err = errors.New(buf.String())
 220                 }
 221                 responseBody = buf
 222                 return
 223         }
 224 }
 225
 226 func readBody(r io.Reader, data ...interface{}) (err error) {
 227         for _, datum := range data {
 228                 err = binary.Read(r, binary.BigEndian, datum)
 229                 if err != nil {
 230                         break
 231                 }
 232         }
 233         return
 234 }
 235
 236 func (c *udpClient) connected() bool {
 237         return !c.connectionIdReceived.IsZero() && time.Now().Before(c.connectionIdReceived.Add(time.Minute))
 238 }
 239
 240 func (c *udpClient) Connect() (err error) {
 241         if c.connected() {
 242                 return nil
 243         }
 244         c.connectionId = connectRequestConnectionId
 245         if c.socket == nil {
 246                 hmp := missinggo.SplitHostPort(c.url.Host)
 247                 if hmp.NoPort {
 248                         hmp.NoPort = false
 249                         hmp.Port = 80
 250                 }
 251                 c.socket, err = net.Dial("udp", hmp.String())
 252                 if err != nil {
 253                         return
 254                 }
 255                 c.socket = pproffd.WrapNetConn(c.socket)
 256         }
 257         b, err := c.request(ActionConnect, nil, nil)
 258         if err != nil {
 259                 return
 260         }
 261         var res ConnectionResponse
 262         err = readBody(b, &res)
 263         if err != nil {
 264                 return
 265         }
 266         c.connectionId = res.ConnectionId
 267         c.connectionIdReceived = time.Now()
 268         return
 269 }