mirror of
https://github.com/sot-tech/mochi.git
synced 2026-07-27 17:48:11 -07:00
middleware: add sanitization hook
This commit is contained in:
@@ -23,7 +23,6 @@ type Config struct {
|
||||
GarbageCollectionInterval time.Duration `yaml:"gc_interval"`
|
||||
PeerLifetime time.Duration `yaml:"peer_lifetime"`
|
||||
ShardCount int `yaml:"shard_count"`
|
||||
MaxNumWant int `yaml:"max_numwant"`
|
||||
}
|
||||
|
||||
// New creates a new PeerStore backed by memory.
|
||||
@@ -38,9 +37,8 @@ func New(cfg Config) (storage.PeerStore, error) {
|
||||
}
|
||||
|
||||
ps := &peerStore{
|
||||
shards: make([]*peerShard, shardCount*2),
|
||||
closed: make(chan struct{}),
|
||||
maxNumWant: cfg.MaxNumWant,
|
||||
shards: make([]*peerShard, shardCount*2),
|
||||
closed: make(chan struct{}),
|
||||
}
|
||||
|
||||
for i := 0; i < shardCount*2; i++ {
|
||||
@@ -77,39 +75,48 @@ type swarm struct {
|
||||
}
|
||||
|
||||
type peerStore struct {
|
||||
shards []*peerShard
|
||||
closed chan struct{}
|
||||
maxNumWant int
|
||||
shards []*peerShard
|
||||
closed chan struct{}
|
||||
}
|
||||
|
||||
var _ storage.PeerStore = &peerStore{}
|
||||
|
||||
func (s *peerStore) shardIndex(infoHash bittorrent.InfoHash, v6 bool) uint32 {
|
||||
func (s *peerStore) shardIndex(infoHash bittorrent.InfoHash, af bittorrent.AddressFamily) uint32 {
|
||||
// There are twice the amount of shards specified by the user, the first
|
||||
// half is dedicated to IPv4 swarms and the second half is dedicated to
|
||||
// IPv6 swarms.
|
||||
idx := binary.BigEndian.Uint32(infoHash[:4]) % (uint32(len(s.shards)) / 2)
|
||||
if v6 {
|
||||
if af == bittorrent.IPv6 {
|
||||
idx += uint32(len(s.shards) / 2)
|
||||
}
|
||||
return idx
|
||||
}
|
||||
|
||||
func newPeerKey(p bittorrent.Peer) serializedPeer {
|
||||
b := make([]byte, 20+2+len(p.IP))
|
||||
b := make([]byte, 20+2+len(p.IP.IP))
|
||||
copy(b[:20], p.ID[:])
|
||||
binary.BigEndian.PutUint16(b[20:22], p.Port)
|
||||
copy(b[22:], p.IP)
|
||||
copy(b[22:], p.IP.IP)
|
||||
|
||||
return serializedPeer(b)
|
||||
}
|
||||
|
||||
func decodePeerKey(pk serializedPeer) bittorrent.Peer {
|
||||
return bittorrent.Peer{
|
||||
peer := bittorrent.Peer{
|
||||
ID: bittorrent.PeerIDFromString(string(pk[:20])),
|
||||
Port: binary.BigEndian.Uint16([]byte(pk[20:22])),
|
||||
IP: net.IP(pk[22:]),
|
||||
IP: bittorrent.IP{IP: net.IP(pk[22:])}}
|
||||
|
||||
if ip := peer.IP.To4(); ip != nil {
|
||||
peer.IP.IP = ip
|
||||
peer.IP.AddressFamily = bittorrent.IPv4
|
||||
} else if len(peer.IP.IP) == net.IPv6len { // implies toReturn.IP.To4() == nil
|
||||
peer.IP.AddressFamily = bittorrent.IPv6
|
||||
} else {
|
||||
panic("IP is neither v4 nor v6")
|
||||
}
|
||||
|
||||
return peer
|
||||
}
|
||||
|
||||
func (s *peerStore) PutSeeder(ih bittorrent.InfoHash, p bittorrent.Peer) error {
|
||||
@@ -121,7 +128,7 @@ func (s *peerStore) PutSeeder(ih bittorrent.InfoHash, p bittorrent.Peer) error {
|
||||
|
||||
pk := newPeerKey(p)
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(p.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, p.IP.AddressFamily)]
|
||||
shard.Lock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -146,7 +153,7 @@ func (s *peerStore) DeleteSeeder(ih bittorrent.InfoHash, p bittorrent.Peer) erro
|
||||
|
||||
pk := newPeerKey(p)
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(p.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, p.IP.AddressFamily)]
|
||||
shard.Lock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -178,7 +185,7 @@ func (s *peerStore) PutLeecher(ih bittorrent.InfoHash, p bittorrent.Peer) error
|
||||
|
||||
pk := newPeerKey(p)
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(p.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, p.IP.AddressFamily)]
|
||||
shard.Lock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -203,7 +210,7 @@ func (s *peerStore) DeleteLeecher(ih bittorrent.InfoHash, p bittorrent.Peer) err
|
||||
|
||||
pk := newPeerKey(p)
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(p.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, p.IP.AddressFamily)]
|
||||
shard.Lock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -235,7 +242,7 @@ func (s *peerStore) GraduateLeecher(ih bittorrent.InfoHash, p bittorrent.Peer) e
|
||||
|
||||
pk := newPeerKey(p)
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(p.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, p.IP.AddressFamily)]
|
||||
shard.Lock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -260,11 +267,7 @@ func (s *peerStore) AnnouncePeers(ih bittorrent.InfoHash, seeder bool, numWant i
|
||||
default:
|
||||
}
|
||||
|
||||
if numWant > s.maxNumWant {
|
||||
numWant = s.maxNumWant
|
||||
}
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, len(announcer.IP) == net.IPv6len)]
|
||||
shard := s.shards[s.shardIndex(ih, announcer.IP.AddressFamily)]
|
||||
shard.RLock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
@@ -319,14 +322,14 @@ func (s *peerStore) AnnouncePeers(ih bittorrent.InfoHash, seeder bool, numWant i
|
||||
return
|
||||
}
|
||||
|
||||
func (s *peerStore) ScrapeSwarm(ih bittorrent.InfoHash, v6 bool) (resp bittorrent.Scrape) {
|
||||
func (s *peerStore) ScrapeSwarm(ih bittorrent.InfoHash, addressFamily bittorrent.AddressFamily) (resp bittorrent.Scrape) {
|
||||
select {
|
||||
case <-s.closed:
|
||||
panic("attempted to interact with stopped memory store")
|
||||
default:
|
||||
}
|
||||
|
||||
shard := s.shards[s.shardIndex(ih, v6)]
|
||||
shard := s.shards[s.shardIndex(ih, addressFamily)]
|
||||
shard.RLock()
|
||||
|
||||
if _, ok := shard.swarms[ih]; !ok {
|
||||
|
||||
@@ -9,7 +9,7 @@ import (
|
||||
)
|
||||
|
||||
func createNew() s.PeerStore {
|
||||
ps, err := New(Config{ShardCount: 1024, GarbageCollectionInterval: 10 * time.Minute, MaxNumWant: 50})
|
||||
ps, err := New(Config{ShardCount: 1024, GarbageCollectionInterval: 10 * time.Minute})
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
|
||||
+2
-2
@@ -57,13 +57,13 @@ type PeerStore interface {
|
||||
|
||||
// ScrapeSwarm returns information required to answer a scrape request
|
||||
// about a swarm identified by the given infohash.
|
||||
// The v6 flag indicates whether or not the IPv6 swarm should be
|
||||
// The AddressFamily indicates whether or not the IPv6 swarm should be
|
||||
// scraped.
|
||||
// The Complete and Incomplete fields of the Scrape must be filled,
|
||||
// filling the Snatches field is optional.
|
||||
// If the infohash is unknown to the PeerStore, an empty Scrape is
|
||||
// returned.
|
||||
ScrapeSwarm(infoHash bittorrent.InfoHash, v6 bool) bittorrent.Scrape
|
||||
ScrapeSwarm(infoHash bittorrent.InfoHash, addressFamily bittorrent.AddressFamily) bittorrent.Scrape
|
||||
|
||||
// Stopper is an interface that expects a Stop method to stop the
|
||||
// PeerStore.
|
||||
|
||||
@@ -45,7 +45,7 @@ func generatePeers() (a [1000]bittorrent.Peer) {
|
||||
port := uint16(r.Uint32())
|
||||
a[i] = bittorrent.Peer{
|
||||
ID: bittorrent.PeerID(id),
|
||||
IP: net.IP(ip),
|
||||
IP: bittorrent.IP{IP: net.IP(ip), AddressFamily: bittorrent.IPv4},
|
||||
Port: port,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,20 +20,20 @@ func TestPeerStore(t *testing.T, p PeerStore) {
|
||||
}{
|
||||
{
|
||||
bittorrent.InfoHashFromString("00000000000000000001"),
|
||||
bittorrent.Peer{ID: bittorrent.PeerIDFromString("00000000000000000001"), Port: 1, IP: net.ParseIP("1.1.1.1").To4()},
|
||||
bittorrent.Peer{ID: bittorrent.PeerIDFromString("00000000000000000001"), Port: 1, IP: bittorrent.IP{IP: net.ParseIP("1.1.1.1").To4(), AddressFamily: bittorrent.IPv4}},
|
||||
},
|
||||
{
|
||||
bittorrent.InfoHashFromString("00000000000000000002"),
|
||||
bittorrent.Peer{ID: bittorrent.PeerIDFromString("00000000000000000002"), Port: 2, IP: net.ParseIP("abab::0001")},
|
||||
bittorrent.Peer{ID: bittorrent.PeerIDFromString("00000000000000000002"), Port: 2, IP: bittorrent.IP{IP: net.ParseIP("abab::0001"), AddressFamily: bittorrent.IPv6}},
|
||||
},
|
||||
}
|
||||
|
||||
v4Peer := bittorrent.Peer{ID: bittorrent.PeerIDFromString("99999999999999999994"), IP: net.ParseIP("99.99.99.99").To4(), Port: 9994}
|
||||
v6Peer := bittorrent.Peer{ID: bittorrent.PeerIDFromString("99999999999999999996"), IP: net.ParseIP("fc00::0001"), Port: 9996}
|
||||
v4Peer := bittorrent.Peer{ID: bittorrent.PeerIDFromString("99999999999999999994"), IP: bittorrent.IP{IP: net.ParseIP("99.99.99.99").To4(), AddressFamily: bittorrent.IPv4}, Port: 9994}
|
||||
v6Peer := bittorrent.Peer{ID: bittorrent.PeerIDFromString("99999999999999999996"), IP: bittorrent.IP{IP: net.ParseIP("fc00::0001"), AddressFamily: bittorrent.IPv6}, Port: 9996}
|
||||
|
||||
for _, c := range testData {
|
||||
peer := v4Peer
|
||||
if len(c.peer.IP) == net.IPv6len {
|
||||
if c.peer.IP.AddressFamily == bittorrent.IPv6 {
|
||||
peer = v6Peer
|
||||
}
|
||||
|
||||
@@ -48,7 +48,7 @@ func TestPeerStore(t *testing.T, p PeerStore) {
|
||||
require.Equal(t, ErrResourceDoesNotExist, err)
|
||||
|
||||
// Test empty scrape response for non-existent swarms.
|
||||
scrape := p.ScrapeSwarm(c.ih, len(c.peer.IP) == net.IPv6len)
|
||||
scrape := p.ScrapeSwarm(c.ih, c.peer.IP.AddressFamily)
|
||||
require.Equal(t, uint32(0), scrape.Complete)
|
||||
require.Equal(t, uint32(0), scrape.Incomplete)
|
||||
require.Equal(t, uint32(0), scrape.Snatches)
|
||||
@@ -76,7 +76,7 @@ func TestPeerStore(t *testing.T, p PeerStore) {
|
||||
require.Nil(t, err)
|
||||
require.True(t, containsPeer(peers, c.peer))
|
||||
|
||||
scrape = p.ScrapeSwarm(c.ih, len(c.peer.IP) == net.IPv6len)
|
||||
scrape = p.ScrapeSwarm(c.ih, c.peer.IP.AddressFamily)
|
||||
require.Equal(t, uint32(2), scrape.Incomplete)
|
||||
require.Equal(t, uint32(0), scrape.Complete)
|
||||
|
||||
@@ -97,7 +97,7 @@ func TestPeerStore(t *testing.T, p PeerStore) {
|
||||
require.Nil(t, err)
|
||||
require.True(t, containsPeer(peers, c.peer))
|
||||
|
||||
scrape = p.ScrapeSwarm(c.ih, len(c.peer.IP) == net.IPv6len)
|
||||
scrape = p.ScrapeSwarm(c.ih, c.peer.IP.AddressFamily)
|
||||
require.Equal(t, uint32(1), scrape.Incomplete)
|
||||
require.Equal(t, uint32(1), scrape.Complete)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user