tracker/udp.go

   1 package tracker
   2
   3 import (
   4         "bytes"
   5         "encoding"
   6         "encoding/binary"
   7         "errors"
   8         "fmt"
   9         "io"
  10         "math/rand"
  11         "net"
  12         "net/url"
  13         "time"
  14
  15         "github.com/anacrolix/dht/krpc"
  16         "github.com/anacrolix/missinggo"
  17         "github.com/anacrolix/missinggo/pproffd"
  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 newTransactionId() int32 {
  64         return int32(rand.Uint32())
  65 }
  66
  67 func timeout(contiguousTimeouts int) (d time.Duration) {
  68         if contiguousTimeouts > 8 {
  69                 contiguousTimeouts = 8
  70         }
  71         d = 15 * time.Second
  72         for ; contiguousTimeouts > 0; contiguousTimeouts-- {
  73                 d *= 2
  74         }
  75         return
  76 }
  77
  78 type udpAnnounce struct {
  79         contiguousTimeouts   int
  80         connectionIdReceived time.Time
  81         connectionId         int64
  82         socket               net.Conn
  83         url                  url.URL
  84         a                    *Announce
  85 }
  86
  87 func (c *udpAnnounce) Close() error {
  88         if c.socket != nil {
  89                 return c.socket.Close()
  90         }
  91         return nil
  92 }
  93
  94 func (c *udpAnnounce) ipv6() bool {
  95         if c.a.UdpNetwork == "udp6" {
  96                 return true
  97         }
  98         rip := missinggo.AddrIP(c.socket.RemoteAddr())
  99         return rip.To16() != nil && rip.To4() == nil
 100 }
 101
 102 func (c *udpAnnounce) Do(req *AnnounceRequest) (res AnnounceResponse, err error) {
 103         err = c.connect()
 104         if err != nil {
 105                 return
 106         }
 107         reqURI := c.url.RequestURI()
 108         // Clearly this limits the request URI to 255 bytes. BEP 41 supports
 109         // longer but I'm not fussed.
 110         options := append([]byte{optionTypeURLData, byte(len(reqURI))}, []byte(reqURI)...)
 111         b, err := c.request(ActionAnnounce, req, options)
 112         if err != nil {
 113                 return
 114         }
 115         var h AnnounceResponseHeader
 116         err = readBody(b, &h)
 117         if err != nil {
 118                 if err == io.EOF {
 119                         err = io.ErrUnexpectedEOF
 120                 }
 121                 err = fmt.Errorf("error parsing announce response: %s", err)
 122                 return
 123         }
 124         res.Interval = h.Interval
 125         res.Leechers = h.Leechers
 126         res.Seeders = h.Seeders
 127         nas := func() interface {
 128                 encoding.BinaryUnmarshaler
 129                 NodeAddrs() []krpc.NodeAddr
 130         } {
 131                 if c.ipv6() {
 132                         return &krpc.CompactIPv6NodeAddrs{}
 133                 } else {
 134                         return &krpc.CompactIPv4NodeAddrs{}
 135                 }
 136         }()
 137         err = nas.UnmarshalBinary(b.Bytes())
 138         if err != nil {
 139                 return
 140         }
 141         for _, cp := range nas.NodeAddrs() {
 142                 res.Peers = append(res.Peers, Peer{}.FromNodeAddr(cp))
 143         }
 144         return
 145 }
 146
 147 // body is the binary serializable request body. trailer is optional data
 148 // following it, such as for BEP 41.
 149 func (c *udpAnnounce) write(h *RequestHeader, body interface{}, trailer []byte) (err error) {
 150         var buf bytes.Buffer
 151         err = binary.Write(&buf, binary.BigEndian, h)
 152         if err != nil {
 153                 panic(err)
 154         }
 155         if body != nil {
 156                 err = binary.Write(&buf, binary.BigEndian, body)
 157                 if err != nil {
 158                         panic(err)
 159                 }
 160         }
 161         _, err = buf.Write(trailer)
 162         if err != nil {
 163                 return
 164         }
 165         n, err := c.socket.Write(buf.Bytes())
 166         if err != nil {
 167                 return
 168         }
 169         if n != buf.Len() {
 170                 panic("write should send all or error")
 171         }
 172         return
 173 }
 174
 175 func read(r io.Reader, data interface{}) error {
 176         return binary.Read(r, binary.BigEndian, data)
 177 }
 178
 179 func write(w io.Writer, data interface{}) error {
 180         return binary.Write(w, binary.BigEndian, data)
 181 }
 182
 183 // args is the binary serializable request body. trailer is optional data
 184 // following it, such as for BEP 41.
 185 func (c *udpAnnounce) request(action Action, args interface{}, options []byte) (responseBody *bytes.Buffer, err error) {
 186         tid := newTransactionId()
 187         err = c.write(&RequestHeader{
 188                 ConnectionId:  c.connectionId,
 189                 Action:        action,
 190                 TransactionId: tid,
 191         }, args, options)
 192         if err != nil {
 193                 return
 194         }
 195         c.socket.SetReadDeadline(time.Now().Add(timeout(c.contiguousTimeouts)))
 196         b := make([]byte, 0x800) // 2KiB
 197         for {
 198                 var n int
 199                 n, err = c.socket.Read(b)
 200                 if opE, ok := err.(*net.OpError); ok {
 201                         if opE.Timeout() {
 202                                 c.contiguousTimeouts++
 203                                 return
 204                         }
 205                 }
 206                 if err != nil {
 207                         return
 208                 }
 209                 buf := bytes.NewBuffer(b[:n])
 210                 var h ResponseHeader
 211                 err = binary.Read(buf, binary.BigEndian, &h)
 212                 switch err {
 213                 case io.ErrUnexpectedEOF:
 214                         continue
 215                 case nil:
 216                 default:
 217                         return
 218                 }
 219                 if h.TransactionId != tid {
 220                         continue
 221                 }
 222                 c.contiguousTimeouts = 0
 223                 if h.Action == ActionError {
 224                         err = errors.New(buf.String())
 225                 }
 226                 responseBody = buf
 227                 return
 228         }
 229 }
 230
 231 func readBody(r io.Reader, data ...interface{}) (err error) {
 232         for _, datum := range data {
 233                 err = binary.Read(r, binary.BigEndian, datum)
 234                 if err != nil {
 235                         break
 236                 }
 237         }
 238         return
 239 }
 240
 241 func (c *udpAnnounce) connected() bool {
 242         return !c.connectionIdReceived.IsZero() && time.Now().Before(c.connectionIdReceived.Add(time.Minute))
 243 }
 244
 245 func (c *udpAnnounce) dialNetwork() string {
 246         if c.a.UdpNetwork != "" {
 247                 return c.a.UdpNetwork
 248         }
 249         return "udp"
 250 }
 251
 252 func (c *udpAnnounce) connect() (err error) {
 253         if c.connected() {
 254                 return nil
 255         }
 256         c.connectionId = connectRequestConnectionId
 257         if c.socket == nil {
 258                 hmp := missinggo.SplitHostMaybePort(c.url.Host)
 259                 if hmp.NoPort {
 260                         hmp.NoPort = false
 261                         hmp.Port = 80
 262                 }
 263                 c.socket, err = net.Dial(c.dialNetwork(), hmp.String())
 264                 if err != nil {
 265                         return
 266                 }
 267                 c.socket = pproffd.WrapNetConn(c.socket)
 268         }
 269         b, err := c.request(ActionConnect, nil, nil)
 270         if err != nil {
 271                 return
 272         }
 273         var res ConnectionResponse
 274         err = readBody(b, &res)
 275         if err != nil {
 276                 return
 277         }
 278         c.connectionId = res.ConnectionId
 279         c.connectionIdReceived = time.Now()
 280         return
 281 }
 282
 283 // TODO: Split on IPv6, as BEP 15 says response peer decoding depends on
 284 // network in use.
 285 func announceUDP(opt Announce, _url *url.URL) (AnnounceResponse, error) {
 286         ua := udpAnnounce{
 287                 url: *_url,
 288                 a:   &opt,
 289         }
 290         defer ua.Close()
 291         return ua.Do(&opt.Request)
 292 }