FedP2P/tracker/http/http.go

package http

import (
	"bytes"
	"context"
	"crypto/tls"
	"expvar"
	"fmt"
	"io"
	"math"
	"net"
	"net/http"
	"net/url"
	"strconv"

	"github.com/anacrolix/dht/v2/krpc"
	"github.com/anacrolix/missinggo/httptoo"
	"github.com/anacrolix/torrent/bencode"
	"github.com/anacrolix/torrent/tracker/shared"
	"github.com/anacrolix/torrent/tracker/udp"
)

var vars = expvar.NewMap("tracker/http")

type Client struct {
	hc *http.Client
}

type NewClientOpts struct {
	Proxy      func(*http.Request) (*url.URL, error)
	ServerName string
}

func NewClient(opts NewClientOpts) Client {
	return Client{
		hc: &http.Client{
			Transport: &http.Transport{
				Proxy: opts.Proxy,
				TLSClientConfig: &tls.Config{
					InsecureSkipVerify: true,
					ServerName:         opts.ServerName,
				},
				// This is for S3 trackers that hold connections open.
				DisableKeepAlives: true,
			},
		},
	}
}

type HttpResponse struct {
	FailureReason string `bencode:"failure reason"`
	Interval      int32  `bencode:"interval"`
	TrackerId     string `bencode:"tracker id"`
	Complete      int32  `bencode:"complete"`
	Incomplete    int32  `bencode:"incomplete"`
	Peers         Peers  `bencode:"peers"`
	// BEP 7
	Peers6 krpc.CompactIPv6NodeAddrs `bencode:"peers6"`
}

type Peers []Peer

func (me *Peers) UnmarshalBencode(b []byte) (err error) {
	var _v interface{}
	err = bencode.Unmarshal(b, &_v)
	if err != nil {
		return
	}
	switch v := _v.(type) {
	case string:
		vars.Add("http responses with string peers", 1)
		var cnas krpc.CompactIPv4NodeAddrs
		err = cnas.UnmarshalBinary([]byte(v))
		if err != nil {
			return
		}
		for _, cp := range cnas {
			*me = append(*me, Peer{
				IP:   cp.IP[:],
				Port: int(cp.Port),
			})
		}
		return
	case []interface{}:
		vars.Add("http responses with list peers", 1)
		for _, i := range v {
			var p Peer
			p.FromDictInterface(i.(map[string]interface{}))
			*me = append(*me, p)
		}
		return
	default:
		vars.Add("http responses with unhandled peers type", 1)
		err = fmt.Errorf("unsupported type: %T", _v)
		return
	}
}

func setAnnounceParams(_url *url.URL, ar *AnnounceRequest, opts AnnounceOpt) {
	q := _url.Query()

	q.Set("key", strconv.FormatInt(int64(ar.Key), 10))
	q.Set("info_hash", string(ar.InfoHash[:]))
	q.Set("peer_id", string(ar.PeerId[:]))
	// AFAICT, port is mandatory, and there's no implied port key.
	q.Set("port", fmt.Sprintf("%d", ar.Port))
	q.Set("uploaded", strconv.FormatInt(ar.Uploaded, 10))
	q.Set("downloaded", strconv.FormatInt(ar.Downloaded, 10))

	// The AWS S3 tracker returns "400 Bad Request: left(-1) was not in the valid range 0 -
	// 9223372036854775807" if left is out of range, or "500 Internal Server Error: Internal Server
	// Error" if omitted entirely.
	left := ar.Left
	if left < 0 {
		left = math.MaxInt64
	}
	q.Set("left", strconv.FormatInt(left, 10))

	if ar.Event != shared.None {
		q.Set("event", ar.Event.String())
	}
	// http://stackoverflow.com/questions/17418004/why-does-tracker-server-not-understand-my-request-bittorrent-protocol
	q.Set("compact", "1")
	// According to https://wiki.vuze.com/w/Message_Stream_Encryption. TODO:
	// Take EncryptionPolicy or something like it as a parameter.
	q.Set("supportcrypto", "1")
	doIp := func(versionKey string, ip net.IP) {
		if ip == nil {
			return
		}
		ipString := ip.String()
		q.Set(versionKey, ipString)
		// Let's try listing them. BEP 3 mentions having an "ip" param, and BEP 7 says we can list
		// addresses for other address-families, although it's not encouraged.
		q.Add("ip", ipString)
	}
	doIp("ipv4", opts.ClientIp4)
	doIp("ipv6", opts.ClientIp6)
	_url.RawQuery = q.Encode()
}

type AnnounceOpt struct {
	UserAgent  string
	HostHeader string
	ClientIp4  net.IP
	ClientIp6  net.IP
}

type AnnounceRequest = udp.AnnounceRequest

func (cl Client) Announce(ctx context.Context, ar AnnounceRequest, opt AnnounceOpt, _url *url.URL) (ret AnnounceResponse, err error) {
	_url = httptoo.CopyURL(_url)
	setAnnounceParams(_url, &ar, opt)
	req, err := http.NewRequestWithContext(ctx, http.MethodGet, _url.String(), nil)
	req.Header.Set("User-Agent", opt.UserAgent)
	req.Host = opt.HostHeader
	resp, err := cl.hc.Do(req)
	if err != nil {
		return
	}
	defer resp.Body.Close()
	var buf bytes.Buffer
	io.Copy(&buf, resp.Body)
	if resp.StatusCode != 200 {
		err = fmt.Errorf("response from tracker: %s: %s", resp.Status, buf.String())
		return
	}
	var trackerResponse HttpResponse
	err = bencode.Unmarshal(buf.Bytes(), &trackerResponse)
	if _, ok := err.(bencode.ErrUnusedTrailingBytes); ok {
		err = nil
	} else if err != nil {
		err = fmt.Errorf("error decoding %q: %s", buf.Bytes(), err)
		return
	}
	if trackerResponse.FailureReason != "" {
		err = fmt.Errorf("tracker gave failure reason: %q", trackerResponse.FailureReason)
		return
	}
	vars.Add("successful http announces", 1)
	ret.Interval = trackerResponse.Interval
	ret.Leechers = trackerResponse.Incomplete
	ret.Seeders = trackerResponse.Complete
	if len(trackerResponse.Peers) != 0 {
		vars.Add("http responses with nonempty peers key", 1)
	}
	ret.Peers = trackerResponse.Peers
	if len(trackerResponse.Peers6) != 0 {
		vars.Add("http responses with nonempty peers6 key", 1)
	}
	for _, na := range trackerResponse.Peers6 {
		ret.Peers = append(ret.Peers, Peer{
			IP:   na.IP,
			Port: na.Port,
		})
	}
	return
}

type AnnounceResponse struct {
	Interval int32 // Minimum seconds the local peer should wait before next announce.
	Leechers int32
	Seeders  int32
	Peers    []Peer
}