client.go | 5 ++++- global.go | 1 + peer.go | 3 ++- peer_protocol/ut-holepunch/ut-holepunch.go | 92 +++++++++++++++++++++++++++++++++++++++++++++++++++++ peerconn.go | 18 ++++++++++++++++++ torrent.go | 74 ++++++++++++++++++++++++++++++++++++++++++++++++++++- diff --git a/client.go b/client.go index 1ba3787117c9652af2e9e08cfe424cd9c4612365..75960923d7e02b35a917348d375c4abe9c29de91 100644 --- a/client.go +++ b/client.go @@ -19,6 +19,8 @@ "strconv" "strings" "time" + utHolepunch "github.com/anacrolix/torrent/peer_protocol/ut-holepunch" + "github.com/anacrolix/chansync" "github.com/anacrolix/chansync/events" "github.com/anacrolix/dht/v2" @@ -1045,7 +1047,8 @@ ExtendedID: pp.HandshakeExtendedID, ExtendedPayload: func() []byte { msg := pp.ExtendedHandshakeMessage{ M: map[pp.ExtensionName]pp.ExtensionNumber{ - pp.ExtensionNameMetadata: metadataExtendedId, + pp.ExtensionNameMetadata: metadataExtendedId, + utHolepunch.ExtensionName: utHolepunchExtendedId, }, V: cl.config.ExtendedHandshakeClientVersion, Reqq: localClientReqq, diff --git a/global.go b/global.go index 988d434a28783f6f6b07c5aeb9a8a6c89d120546..dfb6bd4cedaab757b578d58a715ffe36569953cc 100644 --- a/global.go +++ b/global.go @@ -24,6 +24,7 @@ // select which extension a message is intended for. const ( metadataExtendedId = iota + 1 // 0 is reserved for deleting keys pexExtendedId + utHolepunchExtendedId ) func defaultPeerExtensionBytes() PeerExtensionBits { diff --git a/peer.go b/peer.go index e88485b296e5b93c44c8d91597c1f413192200f8..4e2a24f60b5de6e2feb0ee981beecfbfc9806fd2 100644 --- a/peer.go +++ b/peer.go @@ -21,7 +21,7 @@ "github.com/anacrolix/torrent/internal/alloclim" "github.com/anacrolix/torrent/mse" pp "github.com/anacrolix/torrent/peer_protocol" request_strategy "github.com/anacrolix/torrent/request-strategy" - "github.com/anacrolix/torrent/typed-roaring" + typedRoaring "github.com/anacrolix/torrent/typed-roaring" ) type ( @@ -117,6 +117,7 @@ peerRequests = orderedBitmap[RequestIndex] ) const ( + PeerSourceUtHolepunch = "C" PeerSourceTracker = "Tr" PeerSourceIncoming = "I" PeerSourceDhtGetPeers = "Hg" // Peers we found by searching a DHT. diff --git a/peer_protocol/ut-holepunch/ut-holepunch.go b/peer_protocol/ut-holepunch/ut-holepunch.go new file mode 100644 index 0000000000000000000000000000000000000000..f3ff0c1908dd8cbf082266cdedccb9c2b4588b37 --- /dev/null +++ b/peer_protocol/ut-holepunch/ut-holepunch.go @@ -0,0 +1,92 @@ +package utHolepunch + +import ( + "bytes" + "encoding/binary" + "fmt" + "net/netip" +) + +const ExtensionName = "ut_holepunch" + +type ( + Msg struct { + MsgType MsgType + AddrPort netip.AddrPort + ErrCode ErrCode + } + MsgType byte + AddrType byte + ErrCode uint32 +) + +const ( + Rendezvous MsgType = iota + Connect + Error +) + +const ( + Ipv4 AddrType = iota + Ipv6 AddrType = iota +) + +const ( + NoSuchPeer ErrCode = iota + 1 + NotConnected + NoSupport + NoSelf +) + +func (m *Msg) UnmarshalBinary(b []byte) error { + if len(b) < 12 { + return fmt.Errorf("buffer too small to be valid") + } + m.MsgType = MsgType(b[0]) + b = b[1:] + addrType := AddrType(b[0]) + b = b[1:] + var addr netip.Addr + switch addrType { + case Ipv4: + addr = netip.AddrFrom4([4]byte(b[:4])) + b = b[4:] + case Ipv6: + if len(b) < 22 { + return fmt.Errorf("not enough bytes") + } + addr = netip.AddrFrom16([16]byte(b[:16])) + b = b[16:] + default: + return fmt.Errorf("unhandled addr type value %v", addrType) + } + port := binary.BigEndian.Uint16(b[:]) + b = b[2:] + m.AddrPort = netip.AddrPortFrom(addr, port) + m.ErrCode = ErrCode(binary.BigEndian.Uint32(b[:])) + b = b[4:] + if len(b) != 0 { + return fmt.Errorf("%v trailing unused bytes", len(b)) + } + return nil +} + +func (m *Msg) MarshalBinary() (_ []byte, err error) { + var buf bytes.Buffer + buf.Grow(24) + buf.WriteByte(byte(m.MsgType)) + addr := m.AddrPort.Addr() + switch { + case addr.Is4(): + buf.WriteByte(byte(Ipv4)) + case addr.Is6(): + buf.WriteByte(byte(Ipv6)) + default: + err = fmt.Errorf("unhandled addr type: %v", addr) + return + } + buf.Write(addr.AsSlice()) + binary.Write(&buf, binary.BigEndian, m.AddrPort.Port()) + binary.Write(&buf, binary.BigEndian, m.ErrCode) + return buf.Bytes(), nil +} diff --git a/peerconn.go b/peerconn.go index 1163ad7740d6143278fe8cc02681222e71925a90..c45d87ff6d06e527f7f82a7c006014352c497d30 100644 --- a/peerconn.go +++ b/peerconn.go @@ -9,9 +9,12 @@ "fmt" "io" "math/rand" "net" + "net/netip" "strconv" "strings" "time" + + utHolepunch "github.com/anacrolix/torrent/peer_protocol/ut-holepunch" "github.com/RoaringBitmap/roaring" . "github.com/anacrolix/generics" @@ -59,6 +62,8 @@ // we may not even know the number of pieces in the torrent yet. peerSentHaveAll bool peerRequestDataAllocLimiter alloclim.Limiter + + outstandingHolepunchingRendezvous map[netip.AddrPort]struct{} } func (cn *PeerConn) peerImplStatusLines() []string { @@ -879,6 +884,13 @@ if err != nil { err = fmt.Errorf("receiving pex message: %w", err) } return + case utHolepunchExtendedId: + var msg utHolepunch.Msg + err = msg.UnmarshalBinary(payload) + if err != nil { + err = fmt.Errorf("unmarshalling ut_holepunch message: %w", err) + return + } default: return fmt.Errorf("unexpected extended message ID: %v", id) } @@ -1050,3 +1062,9 @@ func (pc *PeerConn) remoteIsTransmission() bool { return bytes.HasPrefix(pc.PeerID[:], []byte("-TR")) && pc.PeerID[7] == '-' } + +func (pc *PeerConn) remoteAddrPort() Option[netip.AddrPort] { + return Some(pc.conn.RemoteAddr().(interface { + AddrPort() netip.AddrPort + }).AddrPort()) +} diff --git a/torrent.go b/torrent.go index b0820e5b3b0e97f142ba71d1b7d7ac1fde10356c..285f5cc804cb07e159aed181c42aee2eff97c4e6 100644 --- a/torrent.go +++ b/torrent.go @@ -17,6 +17,8 @@ "text/tabwriter" "time" "unsafe" + utHolepunch "github.com/anacrolix/torrent/peer_protocol/ut-holepunch" + "github.com/RoaringBitmap/roaring" "github.com/anacrolix/chansync" "github.com/anacrolix/chansync/events" @@ -2357,7 +2359,8 @@ t.Piece(i).VerifyData() } } -// Start the process of connecting to the given peer for the given torrent if appropriate. +// Start the process of connecting to the given peer for the given torrent if appropriate. I'm not +// sure all the PeerInfo fields are being used. func (t *Torrent) initiateConn(peer PeerInfo) { if peer.Id == t.cl.peerID { return @@ -2664,3 +2667,72 @@ // should be considerable checks elsewhere for this case due to the network overhead. We should // catch most of the overflow manipulation stuff by checking index and begin above. return nil } + +func (t *Torrent) peerConnsWithRemoteAddrPort(addrPort netip.AddrPort) (ret []*PeerConn) { + for pc := range t.conns { + addr := pc.remoteAddrPort() + if !(addr.Ok && addr.Value == addrPort) { + continue + } + ret = append(ret, pc) + } + return +} + +func makeUtHolepunchMsgForPeerConn( + recipient *PeerConn, + msgType utHolepunch.MsgType, + addrPort netip.AddrPort, + errCode utHolepunch.ErrCode, +) pp.Message { + utHolepunchMsg := utHolepunch.Msg{ + MsgType: msgType, + AddrPort: addrPort, + ErrCode: errCode, + } + extendedPayload, err := utHolepunchMsg.MarshalBinary() + if err != nil { + panic(err) + } + return pp.Message{ + Type: pp.Extended, + ExtendedID: MapMustGet(recipient.PeerExtensionIDs, utHolepunch.ExtensionName), + ExtendedPayload: extendedPayload, + } +} + +func (t *Torrent) handleReceivedUtHolepunchMsg(msg utHolepunch.Msg, sender *PeerConn) error { + switch msg.MsgType { + case utHolepunch.Rendezvous: + sendMsg := func( + pc *PeerConn, + msgType utHolepunch.MsgType, + addrPort netip.AddrPort, + errCode utHolepunch.ErrCode, + ) { + pc.write(makeUtHolepunchMsgForPeerConn(pc, msgType, addrPort, errCode)) + } + targets := t.peerConnsWithRemoteAddrPort(msg.AddrPort) + if len(targets) == 0 { + sendMsg(sender, utHolepunch.Error, msg.AddrPort, utHolepunch.NotConnected) + break + } + for _, pc := range targets { + if !pc.supportsExtension(utHolepunch.ExtensionName) { + sendMsg(sender, utHolepunch.Error, msg.AddrPort, utHolepunch.NoSupport) + continue + } + sendMsg(sender, utHolepunch.Connect, msg.AddrPort, 0) + sendMsg(pc, utHolepunch.Connect, sender.remoteAddrPort().Unwrap(), 0) + } + case utHolepunch.Connect: + t.initiateConn(PeerInfo{ + Addr: msg.AddrPort, + Source: PeerSourceUtHolepunch, + }) + case utHolepunch.Error: + + default: + return fmt.Errorf("unhandled msg type %v", msg.MsgType) + } +}