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", newUDPClient)
  65 }
  66
  67 func newUDPClient(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) Close() error {
  97         if c.socket != nil {
  98                 return c.socket.Close()
  99         }
 100         return nil
 101 }
 102
 103 func (c *udpClient) URL() string {
 104         return c.url.String()
 105 }
 106
 107 func (c *udpClient) String() string {
 108         return c.URL()
 109 }
 110
 111 func (c *udpClient) Announce(req *AnnounceRequest) (res AnnounceResponse, err error) {
 112         if !c.connected() {
 113                 err = ErrNotConnected
 114                 return
 115         }
 116         reqURI := c.url.RequestURI()
 117         // Clearly this limits the request URI to 255 bytes. BEP 41 supports
 118         // longer but I'm not fussed.
 119         options := append([]byte{optionTypeURLData, byte(len(reqURI))}, []byte(reqURI)...)
 120         b, err := c.request(ActionAnnounce, req, options)
 121         if err != nil {
 122                 return
 123         }
 124         var h AnnounceResponseHeader
 125         err = readBody(b, &h)
 126         if err != nil {
 127                 if err == io.EOF {
 128                         err = io.ErrUnexpectedEOF
 129                 }
 130                 err = fmt.Errorf("error parsing announce response: %s", err)
 131                 return
 132         }
 133         res.Interval = h.Interval
 134         res.Leechers = h.Leechers
 135         res.Seeders = h.Seeders
 136         cps, err := util.UnmarshalIPv4CompactPeers(b.Bytes())
 137         if err != nil {
 138                 return
 139         }
 140         for _, cp := range cps {
 141                 res.Peers = append(res.Peers, Peer{
 142                         IP:   cp.IP[:],
 143                         Port: int(cp.Port),
 144                 })
 145         }
 146         return
 147 }
 148
 149 // body is the binary serializable request body. trailer is optional data
 150 // following it, such as for BEP 41.
 151 func (c *udpClient) write(h *RequestHeader, body interface{}, trailer []byte) (err error) {
 152         var buf bytes.Buffer
 153         err = binary.Write(&buf, binary.BigEndian, h)
 154         if err != nil {
 155                 panic(err)
 156         }
 157         if body != nil {
 158                 err = binary.Write(&buf, binary.BigEndian, body)
 159                 if err != nil {
 160                         panic(err)
 161                 }
 162         }
 163         _, err = buf.Write(trailer)
 164         if err != nil {
 165                 return
 166         }
 167         n, err := c.socket.Write(buf.Bytes())
 168         if err != nil {
 169                 return
 170         }
 171         if n != buf.Len() {
 172                 panic("write should send all or error")
 173         }
 174         return
 175 }
 176
 177 func read(r io.Reader, data interface{}) error {
 178         return binary.Read(r, binary.BigEndian, data)
 179 }
 180
 181 func write(w io.Writer, data interface{}) error {
 182         return binary.Write(w, binary.BigEndian, data)
 183 }
 184
 185 // args is the binary serializable request body. trailer is optional data
 186 // following it, such as for BEP 41.
 187 func (c *udpClient) request(action Action, args interface{}, options []byte) (responseBody *bytes.Buffer, err error) {
 188         tid := newTransactionId()
 189         err = c.write(&RequestHeader{
 190                 ConnectionId:  c.connectionId,
 191                 Action:        action,
 192                 TransactionId: tid,
 193         }, args, options)
 194         if err != nil {
 195                 return
 196         }
 197         c.socket.SetReadDeadline(time.Now().Add(timeout(c.contiguousTimeouts)))
 198         b := make([]byte, 0x800) // 2KiB
 199         for {
 200                 var n int
 201                 n, err = c.socket.Read(b)
 202                 if opE, ok := err.(*net.OpError); ok {
 203                         if opE.Timeout() {
 204                                 c.contiguousTimeouts++
 205                                 return
 206                         }
 207                 }
 208                 if err != nil {
 209                         return
 210                 }
 211                 buf := bytes.NewBuffer(b[:n])
 212                 var h ResponseHeader
 213                 err = binary.Read(buf, binary.BigEndian, &h)
 214                 switch err {
 215                 case io.ErrUnexpectedEOF:
 216                         continue
 217                 case nil:
 218                 default:
 219                         return
 220                 }
 221                 if h.TransactionId != tid {
 222                         continue
 223                 }
 224                 c.contiguousTimeouts = 0
 225                 if h.Action == ActionError {
 226                         err = errors.New(buf.String())
 227                 }
 228                 responseBody = buf
 229                 return
 230         }
 231 }
 232
 233 func readBody(r io.Reader, data ...interface{}) (err error) {
 234         for _, datum := range data {
 235                 err = binary.Read(r, binary.BigEndian, datum)
 236                 if err != nil {
 237                         break
 238                 }
 239         }
 240         return
 241 }
 242
 243 func (c *udpClient) connected() bool {
 244         return !c.connectionIdReceived.IsZero() && time.Now().Before(c.connectionIdReceived.Add(time.Minute))
 245 }
 246
 247 func (c *udpClient) Connect() (err error) {
 248         if c.connected() {
 249                 return nil
 250         }
 251         c.connectionId = connectRequestConnectionId
 252         if c.socket == nil {
 253                 hmp := missinggo.SplitHostMaybePort(c.url.Host)
 254                 if hmp.NoPort {
 255                         hmp.NoPort = false
 256                         hmp.Port = 80
 257                 }
 258                 c.socket, err = net.Dial("udp", hmp.String())
 259                 if err != nil {
 260                         return
 261                 }
 262                 c.socket = pproffd.WrapNetConn(c.socket)
 263         }
 264         b, err := c.request(ActionConnect, nil, nil)
 265         if err != nil {
 266                 return
 267         }
 268         var res ConnectionResponse
 269         err = readBody(b, &res)
 270         if err != nil {
 271                 return
 272         }
 273         c.connectionId = res.ConnectionId
 274         c.connectionIdReceived = time.Now()
 275         return
 276 }