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