From d24922dc090bbac095640ea891fb1a3006fe7c19 Mon Sep 17 00:00:00 2001 From: Matt Joiner Date: Thu, 20 Feb 2020 16:47:37 +1100 Subject: [PATCH] Add support for non-IP-based networks Includes a test with unix sockets. Exposes AddDialer, AddListener, and reworks Peer. --- Peer.go | 8 +-- Peers.go | 3 +- client.go | 135 +++++++++++++++++++++++--------------- client_test.go | 2 +- cmd/torrent-pick/main.go | 3 +- cmd/torrent/main.go | 3 +- cmd/torrentfs/main.go | 3 +- config.go | 5 +- connection.go | 19 ++++-- connection_test.go | 4 +- dialer.go | 45 +++++++++++++ go.mod | 3 +- ipport.go | 67 +++++++++++++++++++ misc.go | 2 +- prioritized_peers_test.go | 8 +-- socket.go | 38 +++-------- test/init_test.go | 9 +++ test/transfer_test.go | 19 ++++++ test/unix_test.go | 41 ++++++++++++ torrent.go | 26 ++++---- 20 files changed, 313 insertions(+), 130 deletions(-) create mode 100644 dialer.go create mode 100644 ipport.go create mode 100644 test/init_test.go create mode 100644 test/unix_test.go diff --git a/Peer.go b/Peer.go index 02b2c31a..c71aff21 100644 --- a/Peer.go +++ b/Peer.go @@ -11,8 +11,7 @@ import ( // Peer connection info, handed about publicly. type Peer struct { Id [20]byte - IP net.IP - Port int + Addr net.Addr Source peerSource // Peer is known to support encryption. SupportsEncryption bool @@ -23,8 +22,7 @@ type Peer struct { // FromPex generate Peer from peer exchange func (me *Peer) FromPex(na krpc.NodeAddr, fs peer_protocol.PexPeerFlags) { - me.IP = append([]byte(nil), na.IP...) - me.Port = na.Port + me.Addr = ipPortAddr{append([]byte(nil), na.IP...), na.Port} me.Source = peerSourcePex // If they prefer encryption, they must support it. if fs.Get(peer_protocol.PexPrefersEncryption) { @@ -34,5 +32,5 @@ func (me *Peer) FromPex(na krpc.NodeAddr, fs peer_protocol.PexPeerFlags) { } func (me Peer) addr() IpPort { - return IpPort{IP: me.IP, Port: uint16(me.Port)} + return IpPort{IP: addrIpOrNil(me.Addr), Port: uint16(addrPortOrZero(me.Addr))} } diff --git a/Peers.go b/Peers.go index be702032..0c2d726f 100644 --- a/Peers.go +++ b/Peers.go @@ -24,8 +24,7 @@ func (me *Peers) AppendFromPex(nas []krpc.NodeAddr, fs []peer_protocol.PexPeerFl func (ret Peers) AppendFromTracker(ps []tracker.Peer) Peers { for _, p := range ps { _p := Peer{ - IP: p.IP, - Port: p.Port, + Addr: ipPortAddr{p.IP, p.Port}, Source: peerSourceTracker, } copy(_p.Id[:], p.ID) diff --git a/client.go b/client.go index 60bd27e8..451788c5 100644 --- a/client.go +++ b/client.go @@ -58,8 +58,8 @@ type Client struct { peerID PeerID defaultStorage *storage.Client onClose []func() - dialers []dialer - listeners []listener + dialers []Dialer + listeners []Listener dhtServers []*dht.Server ipBlockList iplist.Ranger // Our BitTorrent protocol extension bytes, sent in our BT handshakes. @@ -92,18 +92,13 @@ func (cl *Client) PeerID() PeerID { return cl.peerID } +// Returns the port number for the first listener that has one. No longer assumes that all port +// numbers are the same, due to support for custom listeners. Returns zero if no port number is +// found. func (cl *Client) LocalPort() (port int) { - cl.eachListener(func(l listener) bool { - _port := missinggo.AddrPort(l.Addr()) - if _port == 0 { - panic(l) - } - if port == 0 { - port = _port - } else if port != _port { - panic("mismatched ports") - } - return true + cl.eachListener(func(l Listener) bool { + port = addrPortOrZero(l.Addr()) + return port == 0 }) return } @@ -263,6 +258,19 @@ func NewClient(cfg *ClientConfig) (cl *Client, err error) { return } +// Adds a Dialer for outgoing connections. All Dialers are used when attempting to connect to a +// given address for any Torrent. +func (cl *Client) AddDialer(d Dialer) { + cl.dialers = append(cl.dialers, d) +} + +// Registers a Listener, and starts Accepting on it. You must Close Listeners provided this way +// yourself. +func (cl *Client) AddListener(l Listener) { + cl.listeners = append(cl.listeners, l) + go cl.acceptConnections(l) +} + func (cl *Client) firewallCallback(net.Addr) bool { cl.rLock() block := !cl.wantConns() @@ -389,24 +397,26 @@ func (cl *Client) waitAccept() { } } +// TODO: Apply filters for non-standard networks, particularly rate-limiting. func (cl *Client) rejectAccepted(conn net.Conn) error { ra := conn.RemoteAddr() - rip := missinggo.AddrIP(ra) - if cl.config.DisableIPv4Peers && rip.To4() != nil { - return errors.New("ipv4 peers disabled") - } - if cl.config.DisableIPv4 && len(rip) == net.IPv4len { - return errors.New("ipv4 disabled") + if rip := addrIpOrNil(ra); rip != nil { + if cl.config.DisableIPv4Peers && rip.To4() != nil { + return errors.New("ipv4 peers disabled") + } + if cl.config.DisableIPv4 && len(rip) == net.IPv4len { + return errors.New("ipv4 disabled") - } - if cl.config.DisableIPv6 && len(rip) == net.IPv6len && rip.To4() == nil { - return errors.New("ipv6 disabled") - } - if cl.rateLimitAccept(rip) { - return errors.New("source IP accepted rate limited") - } - if cl.badPeerIPPort(rip, missinggo.AddrPort(ra)) { - return errors.New("bad source addr") + } + if cl.config.DisableIPv6 && len(rip) == net.IPv6len && rip.To4() == nil { + return errors.New("ipv6 disabled") + } + if cl.rateLimitAccept(rip) { + return errors.New("source IP accepted rate limited") + } + if cl.badPeerIPPort(rip, missinggo.AddrPort(ra)) { + return errors.New("bad source addr") + } } return nil } @@ -441,8 +451,12 @@ func (cl *Client) acceptConnections(l net.Listener) { } else { go cl.incomingConnection(conn) } - log.Fmsg("accepted %s connection from %s", conn.RemoteAddr().Network(), conn.RemoteAddr()).AddValue(debugLogValue).Log(cl.logger) - torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(missinggo.AddrIP(conn.RemoteAddr()))), 1) + log.Fmsg("accepted %q connection at %q from %q", + l.Addr().Network(), + conn.LocalAddr(), + conn.RemoteAddr(), + ).AddValue(debugLogValue).Log(cl.logger) + torrent.Add(fmt.Sprintf("accepted conn remote IP len=%d", len(addrIpOrNil(conn.RemoteAddr()))), 1) torrent.Add(fmt.Sprintf("accepted conn network=%s", conn.RemoteAddr().Network()), 1) torrent.Add(fmt.Sprintf("accepted on %s listener", l.Addr().Network()), 1) }() @@ -454,7 +468,7 @@ func (cl *Client) incomingConnection(nc net.Conn) { if tc, ok := nc.(*net.TCPConn); ok { tc.SetLinger(0) } - c := cl.newConnection(nc, false, missinggo.IpPortFromNetAddr(nc.RemoteAddr()), nc.RemoteAddr().Network()) + c := cl.newConnection(nc, false, nc.RemoteAddr(), nc.RemoteAddr().Network()) c.Discovery = peerSourceIncoming cl.runReceivedConn(c) } @@ -518,7 +532,7 @@ func (cl *Client) dialFirst(ctx context.Context, addr string) (res dialResult) { func() { cl.lock() defer cl.unlock() - cl.eachDialer(func(s dialer) bool { + cl.eachDialer(func(s Dialer) bool { func() { left++ //cl.logger.Printf("dialing %s on %s/%s", addr, s.Addr().Network(), s.Addr()) @@ -559,7 +573,7 @@ func (cl *Client) dialFirst(ctx context.Context, addr string) (res dialResult) { return res } -func (cl *Client) dialFromSocket(ctx context.Context, s dialer, addr string) net.Conn { +func (cl *Client) dialFromSocket(ctx context.Context, s Dialer, addr string) net.Conn { network := s.LocalAddr().Network() cte := cl.config.ConnTracker.Wait( ctx, @@ -575,7 +589,7 @@ func (cl *Client) dialFromSocket(ctx context.Context, s dialer, addr string) net } return nil } - c, err := s.dial(ctx, addr) + c, err := s.Dial(ctx, addr) // This is a bit optimistic, but it looks non-trivial to thread this through the proxy code. Set // it now in case we close the connection forthwith. if tc, ok := c.(*net.TCPConn); ok { @@ -611,7 +625,7 @@ func (cl *Client) noLongerHalfOpen(t *Torrent, addr string) { // Performs initiator handshakes and returns a connection. Returns nil // *connection if no connection for valid reasons. -func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr IpPort, network string) (c *connection, err error) { +func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torrent, encryptHeader bool, remoteAddr net.Addr, network string) (c *connection, err error) { c = cl.newConnection(nc, true, remoteAddr, network) c.headerEncrypted = encryptHeader ctx, cancel := context.WithTimeout(ctx, cl.config.HandshakesTimeout) @@ -630,7 +644,7 @@ func (cl *Client) handshakesConnection(ctx context.Context, nc net.Conn, t *Torr // Returns nil connection and nil error if no connection could be established // for valid reasons. -func (cl *Client) establishOutgoingConnEx(t *Torrent, addr IpPort, obfuscatedHeader bool) (*connection, error) { +func (cl *Client) establishOutgoingConnEx(t *Torrent, addr net.Addr, obfuscatedHeader bool) (*connection, error) { dialCtx, cancel := context.WithTimeout(context.Background(), func() time.Duration { cl.rLock() defer cl.rUnlock() @@ -654,7 +668,7 @@ func (cl *Client) establishOutgoingConnEx(t *Torrent, addr IpPort, obfuscatedHea // Returns nil connection and nil error if no connection could be established // for valid reasons. -func (cl *Client) establishOutgoingConn(t *Torrent, addr IpPort) (c *connection, err error) { +func (cl *Client) establishOutgoingConn(t *Torrent, addr net.Addr) (c *connection, err error) { torrent.Add("establish outgoing connection", 1) obfuscatedHeaderFirst := cl.config.HeaderObfuscationPolicy.Preferred c, err = cl.establishOutgoingConnEx(t, addr, obfuscatedHeaderFirst) @@ -679,7 +693,7 @@ func (cl *Client) establishOutgoingConn(t *Torrent, addr IpPort) (c *connection, // Called to dial out and run a connection. The addr we're given is already // considered half-open. -func (cl *Client) outgoingConnection(t *Torrent, addr IpPort, ps peerSource, trusted bool) { +func (cl *Client) outgoingConnection(t *Torrent, addr net.Addr, ps peerSource, trusted bool) { cl.dialRateLimiter.Wait(context.Background()) c, err := cl.establishOutgoingConn(t, addr) cl.lock() @@ -699,8 +713,7 @@ func (cl *Client) outgoingConnection(t *Torrent, addr IpPort, ps peerSource, tru cl.runHandshookConn(c, t) } -// The port number for incoming peer connections. 0 if the client isn't -// listening. +// The port number for incoming peer connections. 0 if the client isn't listening. func (cl *Client) incomingPeerPort() int { return cl.LocalPort() } @@ -885,7 +898,7 @@ func (cl *Client) sendInitialMessages(conn *connection, torrent *Torrent) { }, V: cl.config.ExtendedHandshakeClientVersion, Reqq: 64, // TODO: Really? - YourIp: pp.CompactIp(conn.remoteAddr.IP), + YourIp: pp.CompactIp(addrIpOrNil(conn.remoteAddr)), Encryption: cl.config.HeaderObfuscationPolicy.Preferred || !cl.config.HeaderObfuscationPolicy.RequirePreferred, Port: cl.incomingPeerPort(), MetadataSize: torrent.metadataSize(), @@ -980,6 +993,13 @@ func (cl *Client) gotMetadataExtensionMsg(payload []byte, t *Torrent, c *connect } } +func (cl *Client) badPeerAddr(addr net.Addr) bool { + if ipa, ok := tryIpPortFromNetAddr(addr); ok { + return cl.badPeerIPPort(ipa.IP, ipa.Port) + } + return false +} + func (cl *Client) badPeerIPPort(ip net.IP, port int) bool { if port == 0 { return true @@ -1010,7 +1030,7 @@ func (cl *Client) newTorrent(ih metainfo.Hash, specStorage storage.ClientImpl) ( peers: prioritizedPeers{ om: btree.New(32), getPrio: func(p Peer) peerPriority { - return bep40PriorityIgnoreError(cl.publicAddr(p.IP), p.addr()) + return bep40PriorityIgnoreError(cl.publicAddr(addrIpOrNil(p.Addr)), p.addr()) }, }, conns: make(map[*connection]struct{}, 2*cl.config.EstablishedConnsPerTorrent), @@ -1208,7 +1228,7 @@ func (cl *Client) banPeerIP(ip net.IP) { cl.badPeerIPs[ip.String()] = struct{}{} } -func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr IpPort, network string) (c *connection) { +func (cl *Client) newConnection(nc net.Conn, outgoing bool, remoteAddr net.Addr, network string) (c *connection) { c = &connection{ conn: nc, outgoing: outgoing, @@ -1242,8 +1262,7 @@ func (cl *Client) onDHTAnnouncePeer(ih metainfo.Hash, ip net.IP, port int, portO return } t.addPeers([]Peer{{ - IP: ip, - Port: port, + Addr: ipPortAddr{ip, port}, Source: peerSourceDhtAnnouncePeer, }}) } @@ -1257,7 +1276,7 @@ func firstNotNil(ips ...net.IP) net.IP { return nil } -func (cl *Client) eachDialer(f func(dialer) bool) { +func (cl *Client) eachDialer(f func(Dialer) bool) { for _, s := range cl.dialers { if !f(s) { break @@ -1265,7 +1284,7 @@ func (cl *Client) eachDialer(f func(dialer) bool) { } } -func (cl *Client) eachListener(f func(listener) bool) { +func (cl *Client) eachListener(f func(Listener) bool) { for _, s := range cl.listeners { if !f(s) { break @@ -1274,7 +1293,7 @@ func (cl *Client) eachListener(f func(listener) bool) { } func (cl *Client) findListener(f func(net.Listener) bool) (ret net.Listener) { - cl.eachListener(func(l listener) bool { + cl.eachListener(func(l Listener) bool { ret = l return !f(l) }) @@ -1297,9 +1316,13 @@ func (cl *Client) publicIp(peer net.IP) net.IP { } func (cl *Client) findListenerIp(f func(net.IP) bool) net.IP { - return missinggo.AddrIP(cl.findListener(func(l net.Listener) bool { - return f(missinggo.AddrIP(l.Addr())) - }).Addr()) + return addrIpOrNil( + cl.findListener( + func(l net.Listener) bool { + return f(addrIpOrNil(l.Addr())) + }, + ).Addr(), + ) } // Our IP as a peer should see it. @@ -1311,15 +1334,19 @@ func (cl *Client) publicAddr(peer net.IP) IpPort { func (cl *Client) ListenAddrs() (ret []net.Addr) { cl.lock() defer cl.unlock() - cl.eachListener(func(l listener) bool { + cl.eachListener(func(l Listener) bool { ret = append(ret, l.Addr()) return true }) return } -func (cl *Client) onBadAccept(addr IpPort) { - ip := maskIpForAcceptLimiting(addr.IP) +func (cl *Client) onBadAccept(addr net.Addr) { + ipa, ok := tryIpPortFromNetAddr(addr) + if !ok { + return + } + ip := maskIpForAcceptLimiting(ipa.IP) if cl.acceptLimiter == nil { cl.acceptLimiter = make(map[ipStr]int) } diff --git a/client_test.go b/client_test.go index 42c10b57..09d9637a 100644 --- a/client_test.go +++ b/client_test.go @@ -574,7 +574,7 @@ func TestClientDynamicListenPortAllProtocols(t *testing.T) { defer cl.Close() port := cl.LocalPort() assert.NotEqual(t, 0, port) - cl.eachListener(func(s listener) bool { + cl.eachListener(func(s Listener) bool { assert.Equal(t, port, missinggo.AddrPort(s.Addr())) return true }) diff --git a/cmd/torrent-pick/main.go b/cmd/torrent-pick/main.go index eada5687..4c01baee 100644 --- a/cmd/torrent-pick/main.go +++ b/cmd/torrent-pick/main.go @@ -32,8 +32,7 @@ func resolvedPeerAddrs(ss []string) (ret []torrent.Peer, err error) { return } ret = append(ret, torrent.Peer{ - IP: addr.IP, - Port: addr.Port, + Addr: addr, }) } return diff --git a/cmd/torrent/main.go b/cmd/torrent/main.go index e21c3d2b..1e7feba0 100644 --- a/cmd/torrent/main.go +++ b/cmd/torrent/main.go @@ -114,8 +114,7 @@ func addTorrents(client *torrent.Client) error { t.AddPeers(func() (ret []torrent.Peer) { for _, ta := range flags.TestPeer { ret = append(ret, torrent.Peer{ - IP: ta.IP, - Port: ta.Port, + Addr: ta, }) } return diff --git a/cmd/torrentfs/main.go b/cmd/torrentfs/main.go index b2d34efe..172f6102 100644 --- a/cmd/torrentfs/main.go +++ b/cmd/torrentfs/main.go @@ -62,8 +62,7 @@ func exitSignalHandlers(fs *torrentfs.TorrentFS) { func addTestPeer(client *torrent.Client) { for _, t := range client.Torrents() { t.AddPeers([]torrent.Peer{{ - IP: args.TestPeer.IP, - Port: args.TestPeer.Port, + Addr: args.TestPeer, }}) } } diff --git a/config.go b/config.go index 569b4b2a..692b0751 100644 --- a/config.go +++ b/config.go @@ -26,9 +26,8 @@ type ClientConfig struct { // Store torrent file data in this directory unless .DefaultStorage is // specified. DataDir string `long:"data-dir" description:"directory to store downloaded torrent data"` - // The address to listen for new uTP and TCP bittorrent protocol - // connections. DHT shares a UDP socket with uTP unless configured - // otherwise. + // The address to listen for new uTP and TCP BitTorrent protocol connections. DHT shares a UDP + // socket with uTP unless configured otherwise. ListenHost func(network string) string ListenPort int NoDefaultPortForwarding bool diff --git a/connection.go b/connection.go index f5a58627..a7f8edf2 100644 --- a/connection.go +++ b/connection.go @@ -46,7 +46,7 @@ type connection struct { conn net.Conn outgoing bool network string - remoteAddr IpPort + remoteAddr net.Addr // The Reader and Writer for this Conn, with hooks installed for stats, // limiting, deadlines etc. w io.Writer @@ -139,14 +139,14 @@ func (cn *connection) expectingChunks() bool { // Returns true if the connection is over IPv6. func (cn *connection) ipv6() bool { - ip := cn.remoteAddr.IP + ip := addrIpOrNil(cn.remoteAddr) if ip.To4() != nil { return false } return len(ip) == net.IPv6len } -// Returns true the dialer has the lower client peer ID. TODO: Find the +// Returns true the if the dialer/initiator has the lower client peer ID. TODO: Find the // specification for this. func (cn *connection) isPreferredDirection() bool { return bytes.Compare(cn.t.cl.peerID[:], cn.PeerID[:]) < 0 == cn.outgoing @@ -1049,9 +1049,13 @@ func (c *connection) mainReadLoop() (err error) { req := newRequestFromMessage(&msg) c.onPeerSentCancel(req) case pp.Port: + ipa, ok := tryIpPortFromNetAddr(c.remoteAddr) + if !ok { + break + } pingAddr := net.UDPAddr{ - IP: c.remoteAddr.IP, - Port: int(c.remoteAddr.Port), + IP: ipa.IP, + Port: ipa.Port, } if msg.Port != 0 { pingAddr.Port = int(msg.Port) @@ -1458,11 +1462,12 @@ func (c *connection) peerPriority() peerPriority { } func (c *connection) remoteIp() net.IP { - return c.remoteAddr.IP + return addrIpOrNil(c.remoteAddr) } func (c *connection) remoteIpPort() IpPort { - return c.remoteAddr + ipa, _ := tryIpPortFromNetAddr(c.remoteAddr) + return IpPort{ipa.IP, uint16(ipa.Port)} } func (c *connection) String() string { diff --git a/connection_test.go b/connection_test.go index 58153fa5..2f11fc2b 100644 --- a/connection_test.go +++ b/connection_test.go @@ -23,7 +23,7 @@ func TestSendBitfieldThenHave(t *testing.T) { config: TestingConfig(), } cl.initLogger() - c := cl.newConnection(nil, false, IpPort{}, "") + c := cl.newConnection(nil, false, nil, "") c.setTorrent(cl.newTorrent(metainfo.Hash{}, nil)) c.t.setInfo(&metainfo.Info{ Pieces: make([]byte, metainfo.HashSize*3), @@ -107,7 +107,7 @@ func BenchmarkConnectionMainReadLoop(b *testing.B) { t.setChunkSize(defaultChunkSize) t._pendingPieces.Set(0, PiecePriorityNormal.BitmapPriority()) r, w := net.Pipe() - cn := cl.newConnection(r, true, IpPort{}, "") + cn := cl.newConnection(r, true, nil, "") cn.setTorrent(t) mrlErr := make(chan error) msg := pp.Message{ diff --git a/dialer.go b/dialer.go new file mode 100644 index 00000000..32ab91f7 --- /dev/null +++ b/dialer.go @@ -0,0 +1,45 @@ +package torrent + +import ( + "context" + "net" + + "github.com/anacrolix/missinggo/perf" +) + +type Dialer interface { + // The network is implied by the instance. + Dial(_ context.Context, addr string) (net.Conn, error) + // This is required for registering with the connection tracker (router connection table + // emulating rate-limiter) before dialing. TODO: What about connections that wouldn't infringe + // on routers, like localhost or unix sockets. + LocalAddr() net.Addr +} + +type NetDialer struct { + Network string + Dialer net.Dialer +} + +func (me NetDialer) Dial(ctx context.Context, addr string) (_ net.Conn, err error) { + defer perf.ScopeTimerErr(&err)() + return me.Dialer.DialContext(ctx, me.Network, addr) +} + +func (me NetDialer) LocalAddr() net.Addr { + return netDialerLocalAddr{me.Network, me.Dialer.LocalAddr} +} + +type netDialerLocalAddr struct { + network string + addr net.Addr +} + +func (me netDialerLocalAddr) Network() string { return me.network } + +func (me netDialerLocalAddr) String() string { + if me.addr == nil { + return "" + } + return me.addr.String() +} diff --git a/go.mod b/go.mod index 42a71643..612cff5f 100644 --- a/go.mod +++ b/go.mod @@ -11,7 +11,6 @@ require ( github.com/anacrolix/missinggo/perf v1.0.0 github.com/anacrolix/missinggo/v2 v2.3.2-0.20200110051601-fc3212fb3984 github.com/anacrolix/multiless v0.0.0-20191223025854-070b7994e841 - github.com/anacrolix/stm v0.2.0 github.com/anacrolix/sync v0.2.0 github.com/anacrolix/tagflag v1.0.1 github.com/anacrolix/upnp v0.1.1 @@ -29,7 +28,7 @@ require ( github.com/pkg/errors v0.9.1 github.com/stretchr/testify v1.4.0 github.com/tinylib/msgp v1.1.1 // indirect - golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa + golang.org/x/net v0.0.0-20200114155413-6afb5195e5aa // indirect golang.org/x/time v0.0.0-20191024005414-555d28b269f0 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 ) diff --git a/ipport.go b/ipport.go new file mode 100644 index 00000000..d65624c3 --- /dev/null +++ b/ipport.go @@ -0,0 +1,67 @@ +package torrent + +import ( + "net" + "strconv" +) + +// Extracts the port as an integer from an address string. +func addrPortOrZero(addr net.Addr) int { + switch raw := addr.(type) { + case *net.UDPAddr: + return raw.Port + case *net.TCPAddr: + return raw.Port + default: + _, port, err := net.SplitHostPort(addr.String()) + if err != nil { + return 0 + } + i64, err := strconv.ParseInt(port, 0, 0) + if err != nil { + panic(err) + } + return int(i64) + } +} + +func addrIpOrNil(addr net.Addr) net.IP { + if addr == nil { + return nil + } + switch raw := addr.(type) { + case *net.UDPAddr: + return raw.IP + case *net.TCPAddr: + return raw.IP + default: + host, _, err := net.SplitHostPort(addr.String()) + if err != nil { + return nil + } + return net.ParseIP(host) + } +} + +type ipPortAddr struct { + IP net.IP + Port int +} + +func (ipPortAddr) Network() string { + return "" +} + +func (me ipPortAddr) String() string { + return net.JoinHostPort(me.IP.String(), strconv.FormatInt(int64(me.Port), 10)) +} + +func tryIpPortFromNetAddr(na net.Addr) (ret ipPortAddr, ok bool) { + ret.IP = addrIpOrNil(na) + if ret.IP == nil { + return + } + ret.Port = addrPortOrZero(na) + ok = true + return +} diff --git a/misc.go b/misc.go index 9e1628a9..d1f179a3 100644 --- a/misc.go +++ b/misc.go @@ -114,7 +114,7 @@ func connIsIpv6(nc interface { LocalAddr() net.Addr }) bool { ra := nc.LocalAddr() - rip := missinggo.AddrIP(ra) + rip := addrIpOrNil(ra) return rip.To4() == nil && rip.To16() != nil } diff --git a/prioritized_peers_test.go b/prioritized_peers_test.go index 364178c1..c08c5a33 100644 --- a/prioritized_peers_test.go +++ b/prioritized_peers_test.go @@ -19,10 +19,10 @@ func TestPrioritizedPeers(t *testing.T) { assert.Panics(t, func() { pp.PopMax() }) assert.False(t, ok) ps := []Peer{ - {IP: net.ParseIP("1.2.3.4")}, - {IP: net.ParseIP("1::2")}, - {IP: net.ParseIP("")}, - {IP: net.ParseIP(""), Trusted: true}, + {Addr: ipPortAddr{IP: net.ParseIP("1.2.3.4")}}, + {Addr: ipPortAddr{IP: net.ParseIP("1::2")}}, + {Addr: ipPortAddr{IP: net.ParseIP("")}}, + {Addr: ipPortAddr{IP: net.ParseIP("")}, Trusted: true}, } for i, p := range ps { t.Logf("peer %d priority: %08x trusted: %t\n", i, pp.getPrio(p), p.Trusted) diff --git a/socket.go b/socket.go index c5f7dbcd..1ae4b42f 100644 --- a/socket.go +++ b/socket.go @@ -10,18 +10,13 @@ import ( "github.com/pkg/errors" ) -type dialer interface { - dial(_ context.Context, addr string) (net.Conn, error) - LocalAddr() net.Addr -} - -type listener interface { +type Listener interface { net.Listener } type socket interface { - listener - dialer + Listener + Dialer } func listen(n network, addr string, f firewallCallback) (socket, error) { @@ -39,34 +34,17 @@ func listenTcp(network, address string) (s socket, err error) { l, err := net.Listen(network, address) return tcpSocket{ Listener: l, - network: network, + NetDialer: NetDialer{ + Network: network, + }, }, err } type tcpSocket struct { net.Listener - network string - dialer net.Dialer + NetDialer } -func (me tcpSocket) dial(ctx context.Context, addr string) (_ net.Conn, err error) { - defer perf.ScopeTimerErr(&err)() - return me.dialer.DialContext(ctx, me.network, addr) -} - -func (me tcpSocket) LocalAddr() net.Addr { - return tcpSocketLocalAddr{me.network, me.Listener.Addr().String()} -} - -type tcpSocketLocalAddr struct { - network string - s string -} - -func (me tcpSocketLocalAddr) Network() string { return me.network } - -func (me tcpSocketLocalAddr) String() string { return "" } - func listenAll(networks []network, getHost func(string) string, port int, f firewallCallback) ([]socket, error) { if len(networks) == 0 { return nil, nil @@ -128,7 +106,7 @@ type utpSocketSocket struct { network string } -func (me utpSocketSocket) dial(ctx context.Context, addr string) (conn net.Conn, err error) { +func (me utpSocketSocket) Dial(ctx context.Context, addr string) (conn net.Conn, err error) { defer perf.ScopeTimerErr(&err)() return me.utpSocket.DialContext(ctx, me.network, addr) } diff --git a/test/init_test.go b/test/init_test.go new file mode 100644 index 00000000..3aa40695 --- /dev/null +++ b/test/init_test.go @@ -0,0 +1,9 @@ +package test + +import ( + "github.com/anacrolix/torrent" +) + +func init() { + torrent.TestingTempDir.Init("torrent-test.test") +} diff --git a/test/transfer_test.go b/test/transfer_test.go index c36ccc98..8bd604c3 100644 --- a/test/transfer_test.go +++ b/test/transfer_test.go @@ -28,6 +28,8 @@ type testClientTransferParams struct { SeederStorage func(string) storage.ClientImpl SeederUploadRateLimiter *rate.Limiter LeecherDownloadRateLimiter *rate.Limiter + ConfigureSeeder ConfigureClient + ConfigureLeecher ConfigureClient } func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) { @@ -57,8 +59,14 @@ func testClientTransfer(t *testing.T, ps testClientTransferParams) { } else { cfg.DataDir = greetingTempDir } + if ps.ConfigureSeeder.Config != nil { + ps.ConfigureSeeder.Config(cfg) + } seeder, err := torrent.NewClient(cfg) require.NoError(t, err) + if ps.ConfigureSeeder.Client != nil { + ps.ConfigureSeeder.Client(seeder) + } if ps.ExportClientStatus { defer testutil.ExportStatusWriter(seeder, "s")() } @@ -83,9 +91,15 @@ func testClientTransfer(t *testing.T, ps testClientTransferParams) { } cfg.Seed = false //cfg.Debug = true + if ps.ConfigureLeecher.Config != nil { + ps.ConfigureLeecher.Config(cfg) + } leecher, err := torrent.NewClient(cfg) require.NoError(t, err) defer leecher.Close() + if ps.ConfigureLeecher.Client != nil { + ps.ConfigureLeecher.Client(leecher) + } if ps.ExportClientStatus { defer testutil.ExportStatusWriter(leecher, "l")() } @@ -335,3 +349,8 @@ func TestSeedAfterDownloading(t *testing.T) { }() wg.Wait() } + +type ConfigureClient struct { + Config func(*torrent.ClientConfig) + Client func(*torrent.Client) +} diff --git a/test/unix_test.go b/test/unix_test.go new file mode 100644 index 00000000..d021b985 --- /dev/null +++ b/test/unix_test.go @@ -0,0 +1,41 @@ +package test + +import ( + "io" + "log" + "net" + "path/filepath" + "testing" + + "github.com/anacrolix/torrent" +) + +func TestUnixConns(t *testing.T) { + var closers []io.Closer + defer func() { + for _, c := range closers { + c.Close() + } + }() + configure := ConfigureClient{ + Config: func(cfg *torrent.ClientConfig) { + cfg.DisableUTP = true + cfg.DisableTCP = true + cfg.Debug = true + }, + Client: func(cl *torrent.Client) { + cl.AddDialer(torrent.NetDialer{Network: "unix"}) + l, err := net.Listen("unix", filepath.Join(torrent.TestingTempDir.NewSub(), "socket")) + if err != nil { + panic(err) + } + log.Printf("created listener %q", l) + closers = append(closers, l) + cl.AddListener(l) + }, + } + testClientTransfer(t, testClientTransferParams{ + ConfigureSeeder: configure, + ConfigureLeecher: configure, + }) +} diff --git a/torrent.go b/torrent.go index bb10264f..474acfd1 100644 --- a/torrent.go +++ b/torrent.go @@ -189,8 +189,7 @@ func (t *Torrent) KnownSwarm() (ks []Peer) { ks = append(ks, Peer{ Id: conn.PeerID, - IP: conn.remoteAddr.IP, - Port: int(conn.remoteAddr.Port), + Addr: conn.remoteAddr, Source: conn.Discovery, // > If the connection is encrypted, that's certainly enough to set SupportsEncryption. // > But if we're not connected to them with an encrypted connection, I couldn't say @@ -253,9 +252,11 @@ func (t *Torrent) addPeer(p Peer) { if t.closed.IsSet() { return } - if cl.badPeerIPPort(p.IP, p.Port) { - torrent.Add("peers not added because of bad addr", 1) - return + if ipAddr, ok := tryIpPortFromNetAddr(p.Addr); ok { + if cl.badPeerIPPort(ipAddr.IP, ipAddr.Port) { + torrent.Add("peers not added because of bad addr", 1) + return + } } if t.peers.Add(p) { torrent.Add("peers replaced", 1) @@ -1350,8 +1351,7 @@ func (t *Torrent) consumeDhtAnnouncePeers(pvs <-chan dht.PeersValues) { continue } t.addPeer(Peer{ - IP: cp.IP[:], - Port: cp.Port, + Addr: ipPortAddr{cp.IP, cp.Port}, Source: peerSourceDhtGetPeers, }) } @@ -1433,7 +1433,7 @@ func (t *Torrent) numTotalPeers() int { peers[addr] = struct{}{} } t.peers.Each(func(peer Peer) { - peers[fmt.Sprintf("%s:%d", peer.IP, peer.Port)] = struct{}{} + peers[peer.Addr.String()] = struct{}{} }) return len(peers) } @@ -1592,7 +1592,7 @@ func (t *Torrent) pieceHashed(piece pieceIndex, passed bool, hashIoErr error) { if len(bannableTouchers) >= 1 { c := bannableTouchers[0] - t.cl.banPeerIP(c.remoteAddr.IP) + t.cl.banPeerIP(c.remoteIp()) c.Drop() } } @@ -1738,10 +1738,11 @@ func (t *Torrent) initiateConn(peer Peer) { if peer.Id == t.cl.peerID { return } - if t.cl.badPeerIPPort(peer.IP, peer.Port) && !peer.Trusted { + + if t.cl.badPeerAddr(peer.Addr) && !peer.Trusted { return } - addr := IpPort{peer.IP, uint16(peer.Port)} + addr := peer.Addr if t.addrActive(addr.String()) { return } @@ -1754,8 +1755,7 @@ func (t *Torrent) AddClientPeer(cl *Client) { t.AddPeers(func() (ps []Peer) { for _, la := range cl.ListenAddrs() { ps = append(ps, Peer{ - IP: missinggo.AddrIP(la), - Port: missinggo.AddrPort(la), + Addr: la, Trusted: true, }) } -- 2.51.0