2920 lines
78 KiB
Go
2920 lines
78 KiB
Go
package dht
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"crypto/sha1"
|
|
"database/sql"
|
|
"encoding/binary"
|
|
"encoding/hex"
|
|
"fmt"
|
|
"log"
|
|
"net"
|
|
"sync"
|
|
"time"
|
|
|
|
"torrentGateway/internal/config"
|
|
"github.com/anacrolix/torrent/bencode"
|
|
)
|
|
|
|
const (
|
|
// Node ID length in bytes (160 bits for SHA-1)
|
|
NodeIDLength = 20
|
|
// K-bucket size
|
|
BucketSize = 8
|
|
// Number of buckets (160 bits = 160 buckets max)
|
|
NumBuckets = 160
|
|
// DHT protocol constants
|
|
Alpha = 3 // Parallelism parameter
|
|
|
|
// BEP-5 DHT protocol constants
|
|
TransactionIDLength = 2
|
|
TokenLength = 8
|
|
MaxNodesReturn = 8
|
|
|
|
// Message types
|
|
QueryMessage = "q"
|
|
ResponseMessage = "r"
|
|
ErrorMessage = "e"
|
|
|
|
// Storage limits
|
|
MaxPeersPerTorrent = 500
|
|
PeerExpiryMinutes = 30
|
|
TokenRotationMinutes = 5
|
|
MaxResponsePeers = 50
|
|
MaxResponseNodes = 8
|
|
|
|
// Rate limiting
|
|
MaxAnnouncesPerIPPerHour = 100
|
|
MaxStoragePerIP = 50
|
|
BlocklistDurationMinutes = 60
|
|
|
|
// Peer verification
|
|
PeerVerificationTimeout = 5 * time.Second
|
|
MinReliabilityScore = 20
|
|
MaxReliabilityScore = 100
|
|
)
|
|
|
|
// NodeID represents a 160-bit node identifier
|
|
type NodeID [NodeIDLength]byte
|
|
|
|
// NodeHealth represents the health state of a node
|
|
type NodeHealth int
|
|
|
|
const (
|
|
NodeGood NodeHealth = iota // Responded recently
|
|
NodeQuestionable // Old, needs verification
|
|
NodeBad // Failed to respond, should be removed
|
|
)
|
|
|
|
// Node represents a DHT node
|
|
type Node struct {
|
|
ID NodeID
|
|
Addr *net.UDPAddr
|
|
LastSeen time.Time
|
|
LastQuery time.Time // When we last queried this node
|
|
Health NodeHealth
|
|
FailedQueries int // Count of consecutive failed queries
|
|
RTT time.Duration // Round trip time for last successful query
|
|
}
|
|
|
|
// Bucket represents a k-bucket in the routing table
|
|
type Bucket struct {
|
|
mu sync.RWMutex
|
|
nodes []*Node
|
|
lastChanged time.Time // When bucket was last modified
|
|
lastRefresh time.Time // When bucket was last refreshed
|
|
}
|
|
|
|
// RoutingTable implements Kademlia routing table
|
|
type RoutingTable struct {
|
|
mu sync.RWMutex
|
|
selfID NodeID
|
|
buckets [NumBuckets]*Bucket
|
|
}
|
|
|
|
// PendingQuery tracks outgoing queries
|
|
type PendingQuery struct {
|
|
TransactionID string
|
|
QueryType string
|
|
Target *net.UDPAddr
|
|
SentTime time.Time
|
|
Retries int
|
|
ResponseChan chan *DHTMessage
|
|
}
|
|
|
|
// RateLimiter tracks announce rates per IP
|
|
type RateLimiter struct {
|
|
mu sync.RWMutex
|
|
announces map[string][]time.Time // IP -> announce times
|
|
storage map[string]int // IP -> peer count
|
|
blocklist map[string]time.Time // IP -> unblock time
|
|
}
|
|
|
|
// DHT represents a DHT node with enhanced peer storage
|
|
type DHT struct {
|
|
config *config.DHTConfig
|
|
nodeID NodeID
|
|
routingTable *RoutingTable
|
|
conn *net.UDPConn
|
|
storage map[string][]byte // Simple in-memory storage for demo
|
|
storageMu sync.RWMutex
|
|
peerStorage *PeerStorage // Enhanced peer storage with database backing
|
|
|
|
// Channels for communication
|
|
stopCh chan struct{}
|
|
announceQueue chan AnnounceRequest
|
|
verificationQueue chan *PeerVerificationRequest
|
|
|
|
// Transaction tracking
|
|
pendingQueries map[string]*PendingQuery
|
|
pendingMu sync.RWMutex
|
|
transactionCounter uint16
|
|
|
|
// Token generation with rotation
|
|
tokenSecret []byte
|
|
previousSecret []byte
|
|
lastSecretRotation time.Time
|
|
tokenMu sync.RWMutex
|
|
|
|
// Rate limiting and security
|
|
rateLimiter *RateLimiter
|
|
peerVerification bool // Whether to verify peers
|
|
|
|
// Statistics
|
|
stats Stats
|
|
statsMu sync.RWMutex
|
|
}
|
|
|
|
// PeerVerificationRequest represents a peer verification request
|
|
type PeerVerificationRequest struct {
|
|
InfoHash string
|
|
Peer *Peer
|
|
Retries int
|
|
}
|
|
|
|
// Stats tracks DHT statistics
|
|
type Stats struct {
|
|
PacketsSent int64
|
|
PacketsReceived int64
|
|
NodesInTable int
|
|
StoredItems int
|
|
}
|
|
|
|
// AnnounceRequest represents a request to announce a torrent
|
|
type AnnounceRequest struct {
|
|
InfoHash string
|
|
Port int
|
|
}
|
|
|
|
// Peer represents a BitTorrent peer with enhanced metadata
|
|
type Peer struct {
|
|
IP net.IP
|
|
Port int
|
|
PeerID string // Peer ID if available
|
|
Added time.Time // When this peer was added
|
|
LastAnnounced time.Time // When this peer last announced
|
|
Source string // How we got this peer ("announce", "dht", etc.)
|
|
ReliabilityScore int // Reliability score (0-100)
|
|
AnnounceCount int // Number of successful announces
|
|
Verified bool // Whether peer has been verified
|
|
LastVerified time.Time // When peer was last verified
|
|
}
|
|
|
|
// PeerList holds peers for a specific infohash
|
|
type PeerList struct {
|
|
mu sync.RWMutex
|
|
peers map[string]*Peer // Key is "IP:Port"
|
|
}
|
|
|
|
// DatabaseInterface defines the database operations for peer storage
|
|
type DatabaseInterface interface {
|
|
StorePeer(infohash, ip string, port int, peerID string, lastAnnounced time.Time) error
|
|
GetPeersForTorrent(infohash string) ([]*StoredPeer, error)
|
|
CleanExpiredPeers() error
|
|
GetPeerCount(infohash string) (int, error)
|
|
UpdatePeerReliability(infohash, ip string, port int, reliabilityScore int, verified bool, lastVerified time.Time) error
|
|
}
|
|
|
|
// StoredPeer represents a peer stored in the database
|
|
type StoredPeer struct {
|
|
InfoHash string
|
|
IP string
|
|
Port int
|
|
PeerID string
|
|
LastAnnounced time.Time
|
|
ReliabilityScore int
|
|
AnnounceCount int
|
|
Verified bool
|
|
LastVerified time.Time
|
|
}
|
|
|
|
// PeerStorage manages peer storage with expiration and database backing
|
|
type PeerStorage struct {
|
|
mu sync.RWMutex
|
|
infohashes map[string]*PeerList // Key is infohash (in-memory cache)
|
|
db DatabaseInterface // Database backing
|
|
stats PeerStorageStats
|
|
}
|
|
|
|
// PeerStorageStats tracks storage statistics
|
|
type PeerStorageStats struct {
|
|
TotalPeers int64
|
|
TotalTorrents int64
|
|
CacheHits int64
|
|
CacheMisses int64
|
|
DatabaseWrites int64
|
|
DatabaseReads int64
|
|
}
|
|
|
|
// DHT message types
|
|
|
|
// DHTMessage represents the base structure of a DHT message
|
|
type DHTMessage struct {
|
|
TransactionID string `bencode:"t"`
|
|
MessageType string `bencode:"y"`
|
|
QueryType string `bencode:"q,omitempty"`
|
|
Response map[string]interface{} `bencode:"r,omitempty"`
|
|
Arguments map[string]interface{} `bencode:"a,omitempty"`
|
|
Error []interface{} `bencode:"e,omitempty"`
|
|
}
|
|
|
|
// PingQuery represents a ping query
|
|
type PingQuery struct {
|
|
ID string `bencode:"id"`
|
|
}
|
|
|
|
// PingResponse represents a ping response
|
|
type PingResponse struct {
|
|
ID string `bencode:"id"`
|
|
}
|
|
|
|
// FindNodeQuery represents a find_node query
|
|
type FindNodeQuery struct {
|
|
ID string `bencode:"id"`
|
|
Target string `bencode:"target"`
|
|
}
|
|
|
|
// FindNodeResponse represents a find_node response
|
|
type FindNodeResponse struct {
|
|
ID string `bencode:"id"`
|
|
Nodes string `bencode:"nodes"`
|
|
}
|
|
|
|
// GetPeersQuery represents a get_peers query
|
|
type GetPeersQuery struct {
|
|
ID string `bencode:"id"`
|
|
InfoHash string `bencode:"info_hash"`
|
|
}
|
|
|
|
// GetPeersResponse represents a get_peers response
|
|
type GetPeersResponse struct {
|
|
ID string `bencode:"id"`
|
|
Token string `bencode:"token"`
|
|
Values []string `bencode:"values,omitempty"`
|
|
Nodes string `bencode:"nodes,omitempty"`
|
|
}
|
|
|
|
// AnnouncePeerQuery represents an announce_peer query
|
|
type AnnouncePeerQuery struct {
|
|
ID string `bencode:"id"`
|
|
InfoHash string `bencode:"info_hash"`
|
|
Port int `bencode:"port"`
|
|
Token string `bencode:"token"`
|
|
}
|
|
|
|
// AnnouncePeerResponse represents an announce_peer response
|
|
type AnnouncePeerResponse struct {
|
|
ID string `bencode:"id"`
|
|
}
|
|
const (
|
|
MsgQuery = "q"
|
|
MsgResponse = "r"
|
|
MsgError = "e"
|
|
)
|
|
|
|
// NewDHT creates a new DHT node
|
|
func NewDHT(config *config.DHTConfig) (*DHT, error) {
|
|
// Generate random node ID
|
|
var nodeID NodeID
|
|
if _, err := rand.Read(nodeID[:]); err != nil {
|
|
return nil, fmt.Errorf("failed to generate node ID: %w", err)
|
|
}
|
|
|
|
// Generate token secret
|
|
tokenSecret := make([]byte, 32)
|
|
if _, err := rand.Read(tokenSecret); err != nil {
|
|
return nil, fmt.Errorf("failed to generate token secret: %w", err)
|
|
}
|
|
|
|
dht := &DHT{
|
|
config: config,
|
|
nodeID: nodeID,
|
|
routingTable: NewRoutingTable(nodeID),
|
|
storage: make(map[string][]byte),
|
|
stopCh: make(chan struct{}),
|
|
announceQueue: make(chan AnnounceRequest, 100),
|
|
verificationQueue: make(chan *PeerVerificationRequest, 100),
|
|
pendingQueries: make(map[string]*PendingQuery),
|
|
tokenSecret: tokenSecret,
|
|
previousSecret: make([]byte, 32), // Initialize empty previous secret
|
|
lastSecretRotation: time.Now(),
|
|
peerStorage: NewPeerStorage(nil), // Database will be set later
|
|
rateLimiter: NewRateLimiter(),
|
|
peerVerification: true, // Enable peer verification by default
|
|
}
|
|
|
|
return dht, nil
|
|
}
|
|
|
|
// NewRateLimiter creates a new rate limiter
|
|
func NewRateLimiter() *RateLimiter {
|
|
return &RateLimiter{
|
|
announces: make(map[string][]time.Time),
|
|
storage: make(map[string]int),
|
|
blocklist: make(map[string]time.Time),
|
|
}
|
|
}
|
|
|
|
// IsBlocked checks if an IP is currently blocked
|
|
func (rl *RateLimiter) IsBlocked(ip string) bool {
|
|
rl.mu.RLock()
|
|
defer rl.mu.RUnlock()
|
|
|
|
if unblockTime, exists := rl.blocklist[ip]; exists {
|
|
if time.Now().Before(unblockTime) {
|
|
return true
|
|
}
|
|
// Unblock expired entries
|
|
rl.mu.RUnlock()
|
|
rl.mu.Lock()
|
|
delete(rl.blocklist, ip)
|
|
rl.mu.Unlock()
|
|
rl.mu.RLock()
|
|
}
|
|
return false
|
|
}
|
|
|
|
// CheckAnnounceRate checks if an IP can announce (not rate limited)
|
|
func (rl *RateLimiter) CheckAnnounceRate(ip string) bool {
|
|
rl.mu.Lock()
|
|
defer rl.mu.Unlock()
|
|
|
|
now := time.Now()
|
|
cutoff := now.Add(-time.Hour)
|
|
|
|
// Clean old announces
|
|
if announces, exists := rl.announces[ip]; exists {
|
|
var recent []time.Time
|
|
for _, t := range announces {
|
|
if t.After(cutoff) {
|
|
recent = append(recent, t)
|
|
}
|
|
}
|
|
rl.announces[ip] = recent
|
|
|
|
// Check rate limit
|
|
if len(recent) >= MaxAnnouncesPerIPPerHour {
|
|
// Block this IP
|
|
rl.blocklist[ip] = now.Add(BlocklistDurationMinutes * time.Minute)
|
|
log.Printf("DHT: Rate limited IP %s (%d announces in last hour)", ip, len(recent))
|
|
return false
|
|
}
|
|
}
|
|
|
|
// Record this announce
|
|
rl.announces[ip] = append(rl.announces[ip], now)
|
|
return true
|
|
}
|
|
|
|
// CheckStorageLimit checks if an IP can store more peers
|
|
func (rl *RateLimiter) CheckStorageLimit(ip string) bool {
|
|
rl.mu.RLock()
|
|
defer rl.mu.RUnlock()
|
|
|
|
if count, exists := rl.storage[ip]; exists {
|
|
return count < MaxStoragePerIP
|
|
}
|
|
return true
|
|
}
|
|
|
|
// IncrementStorage increments the storage count for an IP
|
|
func (rl *RateLimiter) IncrementStorage(ip string) {
|
|
rl.mu.Lock()
|
|
defer rl.mu.Unlock()
|
|
rl.storage[ip]++
|
|
}
|
|
|
|
// NewPeerStorage creates a new peer storage with optional database backing
|
|
func NewPeerStorage(db DatabaseInterface) *PeerStorage {
|
|
return &PeerStorage{
|
|
infohashes: make(map[string]*PeerList),
|
|
db: db,
|
|
stats: PeerStorageStats{},
|
|
}
|
|
}
|
|
|
|
// validateTokenForTimeWindow validates a token for a specific time window and secret
|
|
func (d *DHT) validateTokenForTimeWindow(token string, addr *net.UDPAddr, timeWindow int64, secret []byte) bool {
|
|
h := sha1.New()
|
|
h.Write(secret)
|
|
h.Write(addr.IP)
|
|
binary.Write(h, binary.BigEndian, uint16(addr.Port))
|
|
binary.Write(h, binary.BigEndian, timeWindow)
|
|
expected := fmt.Sprintf("%x", h.Sum(nil)[:TokenLength])
|
|
return token == expected
|
|
}
|
|
|
|
// rotateTokenSecretIfNeeded rotates the token secret every 5 minutes
|
|
func (d *DHT) rotateTokenSecretIfNeeded() {
|
|
if time.Since(d.lastSecretRotation) >= TokenRotationMinutes*time.Minute {
|
|
// Move current secret to previous
|
|
copy(d.previousSecret, d.tokenSecret)
|
|
|
|
// Generate new secret
|
|
if _, err := rand.Read(d.tokenSecret); err != nil {
|
|
log.Printf("DHT: Failed to rotate token secret: %v", err)
|
|
return
|
|
}
|
|
|
|
d.lastSecretRotation = time.Now()
|
|
log.Printf("DHT: Token secret rotated")
|
|
}
|
|
}
|
|
|
|
// SetDatabase sets the database interface for peer storage
|
|
func (d *DHT) SetDatabase(db DatabaseInterface) {
|
|
d.peerStorage.db = db
|
|
log.Printf("DHT: Database interface configured")
|
|
}
|
|
|
|
// AddPeer adds a peer to the storage with database persistence
|
|
func (ps *PeerStorage) AddPeer(infohash string, peer *Peer) error {
|
|
ps.mu.Lock()
|
|
defer ps.mu.Unlock()
|
|
|
|
// Check if we're at the limit for this torrent
|
|
if ps.db != nil {
|
|
count, err := ps.db.GetPeerCount(infohash)
|
|
if err == nil && count >= MaxPeersPerTorrent {
|
|
log.Printf("DHT: Max peers reached for torrent %s (%d peers)", infohash[:8], count)
|
|
return fmt.Errorf("max peers reached for torrent")
|
|
}
|
|
}
|
|
|
|
// Update in-memory cache
|
|
peerList, exists := ps.infohashes[infohash]
|
|
if !exists {
|
|
peerList = &PeerList{
|
|
peers: make(map[string]*Peer),
|
|
}
|
|
ps.infohashes[infohash] = peerList
|
|
}
|
|
|
|
peerList.mu.Lock()
|
|
key := fmt.Sprintf("%s:%d", peer.IP.String(), peer.Port)
|
|
now := time.Now()
|
|
peer.Added = now
|
|
peer.LastAnnounced = now
|
|
peerList.peers[key] = peer
|
|
peerList.mu.Unlock()
|
|
|
|
// Store in database if available
|
|
if ps.db != nil {
|
|
err := ps.db.StorePeer(infohash, peer.IP.String(), peer.Port, peer.PeerID, peer.LastAnnounced)
|
|
if err != nil {
|
|
log.Printf("DHT: Failed to store peer in database: %v", err)
|
|
return err
|
|
}
|
|
ps.stats.DatabaseWrites++
|
|
}
|
|
|
|
ps.stats.TotalPeers++
|
|
return nil
|
|
}
|
|
|
|
// GetPeers gets peers for an infohash with database fallback
|
|
func (ps *PeerStorage) GetPeers(infohash string) []*Peer {
|
|
ps.mu.RLock()
|
|
peerList, exists := ps.infohashes[infohash]
|
|
ps.mu.RUnlock()
|
|
|
|
// Try cache first
|
|
if exists {
|
|
peerList.mu.RLock()
|
|
if len(peerList.peers) > 0 {
|
|
var peers []*Peer
|
|
for _, peer := range peerList.peers {
|
|
// Filter expired peers
|
|
if time.Since(peer.LastAnnounced) <= PeerExpiryMinutes*time.Minute {
|
|
peers = append(peers, peer)
|
|
}
|
|
}
|
|
peerList.mu.RUnlock()
|
|
ps.stats.CacheHits++
|
|
return peers
|
|
}
|
|
peerList.mu.RUnlock()
|
|
}
|
|
|
|
// Cache miss - try database
|
|
if ps.db != nil {
|
|
storedPeers, err := ps.db.GetPeersForTorrent(infohash)
|
|
if err != nil {
|
|
log.Printf("DHT: Failed to get peers from database: %v", err)
|
|
return nil
|
|
}
|
|
|
|
// Convert stored peers to Peer objects
|
|
var peers []*Peer
|
|
for _, sp := range storedPeers {
|
|
peer := &Peer{
|
|
IP: net.ParseIP(sp.IP),
|
|
Port: sp.Port,
|
|
PeerID: sp.PeerID,
|
|
Added: sp.LastAnnounced,
|
|
LastAnnounced: sp.LastAnnounced,
|
|
Source: "database",
|
|
}
|
|
peers = append(peers, peer)
|
|
}
|
|
|
|
ps.stats.DatabaseReads++
|
|
ps.stats.CacheMisses++
|
|
return peers
|
|
}
|
|
|
|
ps.stats.CacheMisses++
|
|
return nil
|
|
}
|
|
|
|
// CleanExpiredPeers removes peers older than configured expiry time
|
|
func (ps *PeerStorage) CleanExpiredPeers() {
|
|
cutoff := time.Now().Add(-PeerExpiryMinutes * time.Minute)
|
|
|
|
// Clean database first if available
|
|
if ps.db != nil {
|
|
if err := ps.db.CleanExpiredPeers(); err != nil {
|
|
log.Printf("DHT: Failed to clean expired peers from database: %v", err)
|
|
}
|
|
}
|
|
|
|
// Clean in-memory cache
|
|
ps.mu.Lock()
|
|
defer ps.mu.Unlock()
|
|
|
|
expiredCount := 0
|
|
for infohash, peerList := range ps.infohashes {
|
|
peerList.mu.Lock()
|
|
|
|
activePeers := make(map[string]*Peer)
|
|
for key, peer := range peerList.peers {
|
|
if peer.LastAnnounced.After(cutoff) {
|
|
activePeers[key] = peer
|
|
} else {
|
|
expiredCount++
|
|
}
|
|
}
|
|
|
|
peerList.peers = activePeers
|
|
peerList.mu.Unlock()
|
|
|
|
// Remove empty peer lists
|
|
if len(activePeers) == 0 {
|
|
delete(ps.infohashes, infohash)
|
|
}
|
|
}
|
|
|
|
if expiredCount > 0 {
|
|
log.Printf("DHT: Cleaned %d expired peers from cache", expiredCount)
|
|
ps.stats.TotalPeers -= int64(expiredCount)
|
|
}
|
|
}
|
|
|
|
// NewRoutingTable creates a new routing table
|
|
func NewRoutingTable(selfID NodeID) *RoutingTable {
|
|
rt := &RoutingTable{
|
|
selfID: selfID,
|
|
}
|
|
|
|
// Initialize buckets
|
|
now := time.Now()
|
|
for i := range rt.buckets {
|
|
rt.buckets[i] = &Bucket{
|
|
nodes: make([]*Node, 0, BucketSize),
|
|
lastChanged: now,
|
|
lastRefresh: now,
|
|
}
|
|
}
|
|
|
|
return rt
|
|
}
|
|
|
|
// Start starts the DHT node
|
|
func (d *DHT) Start() error {
|
|
// Listen on UDP port
|
|
addr, err := net.ResolveUDPAddr("udp", fmt.Sprintf(":%d", d.config.Port))
|
|
if err != nil {
|
|
return fmt.Errorf("failed to resolve UDP address: %w", err)
|
|
}
|
|
|
|
conn, err := net.ListenUDP("udp", addr)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to listen on UDP: %w", err)
|
|
}
|
|
|
|
d.conn = conn
|
|
log.Printf("DHT node started on port %d with ID %x", d.config.Port, d.nodeID)
|
|
|
|
// Start goroutines
|
|
go d.handlePackets()
|
|
go d.bootstrap()
|
|
go d.maintenance()
|
|
go d.handleAnnouncements()
|
|
go d.handlePeerVerification()
|
|
|
|
return nil
|
|
}
|
|
|
|
// Stop stops the DHT node
|
|
func (d *DHT) Stop() error {
|
|
close(d.stopCh)
|
|
if d.conn != nil {
|
|
return d.conn.Close()
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// handlePackets handles incoming UDP packets
|
|
func (d *DHT) handlePackets() {
|
|
buffer := make([]byte, 2048)
|
|
|
|
for {
|
|
select {
|
|
case <-d.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
|
|
d.conn.SetReadDeadline(time.Now().Add(1 * time.Second))
|
|
n, addr, err := d.conn.ReadFromUDP(buffer)
|
|
if err != nil {
|
|
if netErr, ok := err.(net.Error); ok && netErr.Timeout() {
|
|
continue
|
|
}
|
|
log.Printf("Error reading UDP packet: %v", err)
|
|
continue
|
|
}
|
|
|
|
d.statsMu.Lock()
|
|
d.stats.PacketsReceived++
|
|
d.statsMu.Unlock()
|
|
|
|
// Parse and handle DHT message
|
|
go d.handlePacket(buffer[:n], addr)
|
|
}
|
|
}
|
|
|
|
// handlePacket processes a single packet
|
|
func (d *DHT) handlePacket(data []byte, addr *net.UDPAddr) {
|
|
var msg DHTMessage
|
|
if err := bencode.Unmarshal(data, &msg); err != nil {
|
|
log.Printf("Failed to unmarshal DHT message from %s: %v", addr, err)
|
|
return
|
|
}
|
|
|
|
switch msg.MessageType {
|
|
case QueryMessage:
|
|
d.handleQuery(&msg, addr)
|
|
case ResponseMessage:
|
|
d.handleResponse(&msg, addr)
|
|
case ErrorMessage:
|
|
d.handleError(&msg, addr)
|
|
default:
|
|
log.Printf("Unknown message type '%s' from %s", msg.MessageType, addr)
|
|
}
|
|
}
|
|
|
|
// handleQuery handles incoming query messages
|
|
func (d *DHT) handleQuery(msg *DHTMessage, addr *net.UDPAddr) {
|
|
switch msg.QueryType {
|
|
case "ping":
|
|
d.handlePingQuery(msg, addr)
|
|
case "find_node":
|
|
d.handleFindNodeQuery(msg, addr)
|
|
case "get_peers":
|
|
d.handleGetPeersQuery(msg, addr)
|
|
case "announce_peer":
|
|
d.handleAnnouncePeerQuery(msg, addr)
|
|
default:
|
|
d.sendError(msg.TransactionID, addr, 204, "Method Unknown")
|
|
}
|
|
}
|
|
|
|
// handleResponse handles incoming response messages
|
|
func (d *DHT) handleResponse(msg *DHTMessage, addr *net.UDPAddr) {
|
|
d.pendingMu.RLock()
|
|
pendingQuery, exists := d.pendingQueries[msg.TransactionID]
|
|
d.pendingMu.RUnlock()
|
|
|
|
if !exists {
|
|
log.Printf("Received response for unknown transaction %s from %s", msg.TransactionID, addr)
|
|
return
|
|
}
|
|
|
|
// Send response to waiting goroutine
|
|
select {
|
|
case pendingQuery.ResponseChan <- msg:
|
|
default:
|
|
// Channel might be closed
|
|
}
|
|
|
|
// Update routing table with responding node
|
|
if nodeIDBytes, ok := msg.Response["id"].(string); ok {
|
|
if len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
}
|
|
}
|
|
|
|
// handleError handles incoming error messages
|
|
func (d *DHT) handleError(msg *DHTMessage, addr *net.UDPAddr) {
|
|
log.Printf("Received error from %s: %v", addr, msg.Error)
|
|
|
|
d.pendingMu.RLock()
|
|
pendingQuery, exists := d.pendingQueries[msg.TransactionID]
|
|
d.pendingMu.RUnlock()
|
|
|
|
if exists {
|
|
select {
|
|
case pendingQuery.ResponseChan <- msg:
|
|
default:
|
|
// Channel might be closed
|
|
}
|
|
}
|
|
}
|
|
|
|
// bootstrap connects to bootstrap nodes and builds routing table
|
|
func (d *DHT) bootstrap() {
|
|
log.Printf("Starting DHT bootstrap with %d nodes", len(d.config.BootstrapNodes))
|
|
|
|
// First ping all bootstrap nodes to establish initial connections
|
|
var activeBootstrapNodes []*net.UDPAddr
|
|
for _, bootstrapAddr := range d.config.BootstrapNodes {
|
|
addr, err := net.ResolveUDPAddr("udp", bootstrapAddr)
|
|
if err != nil {
|
|
log.Printf("Failed to resolve bootstrap node %s: %v", bootstrapAddr, err)
|
|
continue
|
|
}
|
|
|
|
// Ping bootstrap node
|
|
if _, err := d.Ping(addr); err == nil {
|
|
activeBootstrapNodes = append(activeBootstrapNodes, addr)
|
|
log.Printf("Connected to bootstrap node: %s", addr)
|
|
} else {
|
|
log.Printf("Failed to ping bootstrap node %s: %v", addr, err)
|
|
}
|
|
time.Sleep(500 * time.Millisecond)
|
|
}
|
|
|
|
if len(activeBootstrapNodes) == 0 {
|
|
log.Printf("Warning: No bootstrap nodes responded")
|
|
return
|
|
}
|
|
|
|
// Perform iterative node lookup for our own ID to find neighbors
|
|
log.Printf("Performing bootstrap lookup for own node ID")
|
|
closestNodes := d.iterativeFindNode(d.nodeID, activeBootstrapNodes)
|
|
|
|
// Add all discovered nodes to routing table
|
|
for _, node := range closestNodes {
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
log.Printf("Bootstrap completed: discovered %d nodes", len(closestNodes))
|
|
}
|
|
|
|
// generateTransactionID generates a transaction ID for DHT messages
|
|
func (d *DHT) generateTransactionID() string {
|
|
d.transactionCounter++
|
|
tid := make([]byte, TransactionIDLength)
|
|
binary.BigEndian.PutUint16(tid, d.transactionCounter)
|
|
return string(tid)
|
|
}
|
|
|
|
// sendQuery sends a DHT query and tracks the transaction
|
|
func (d *DHT) sendQuery(queryType string, args map[string]interface{}, target *net.UDPAddr) (*DHTMessage, error) {
|
|
tid := d.generateTransactionID()
|
|
|
|
msg := map[string]interface{}{
|
|
"t": tid,
|
|
"y": QueryMessage,
|
|
"q": queryType,
|
|
"a": args,
|
|
}
|
|
|
|
data, err := bencode.Marshal(msg)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to marshal query: %w", err)
|
|
}
|
|
|
|
// Create pending query
|
|
responseChan := make(chan *DHTMessage, 1)
|
|
pendingQuery := &PendingQuery{
|
|
TransactionID: tid,
|
|
QueryType: queryType,
|
|
Target: target,
|
|
SentTime: time.Now(),
|
|
Retries: 0,
|
|
ResponseChan: responseChan,
|
|
}
|
|
|
|
d.pendingMu.Lock()
|
|
d.pendingQueries[tid] = pendingQuery
|
|
d.pendingMu.Unlock()
|
|
|
|
// Send the query
|
|
_, err = d.conn.WriteToUDP(data, target)
|
|
if err != nil {
|
|
d.pendingMu.Lock()
|
|
delete(d.pendingQueries, tid)
|
|
d.pendingMu.Unlock()
|
|
return nil, fmt.Errorf("failed to send query: %w", err)
|
|
}
|
|
|
|
d.statsMu.Lock()
|
|
d.stats.PacketsSent++
|
|
d.statsMu.Unlock()
|
|
|
|
// Wait for response with timeout
|
|
timer := time.NewTimer(5 * time.Second)
|
|
defer timer.Stop()
|
|
|
|
select {
|
|
case response := <-responseChan:
|
|
d.pendingMu.Lock()
|
|
delete(d.pendingQueries, tid)
|
|
d.pendingMu.Unlock()
|
|
return response, nil
|
|
case <-timer.C:
|
|
d.pendingMu.Lock()
|
|
delete(d.pendingQueries, tid)
|
|
d.pendingMu.Unlock()
|
|
return nil, fmt.Errorf("query timeout")
|
|
case <-d.stopCh:
|
|
d.pendingMu.Lock()
|
|
delete(d.pendingQueries, tid)
|
|
d.pendingMu.Unlock()
|
|
return nil, fmt.Errorf("DHT stopping")
|
|
}
|
|
}
|
|
|
|
// sendError sends an error response
|
|
func (d *DHT) sendError(transactionID string, addr *net.UDPAddr, code int, message string) {
|
|
errorMsg := map[string]interface{}{
|
|
"t": transactionID,
|
|
"y": ErrorMessage,
|
|
"e": []interface{}{code, message},
|
|
}
|
|
|
|
data, err := bencode.Marshal(errorMsg)
|
|
if err != nil {
|
|
log.Printf("Failed to marshal error message: %v", err)
|
|
return
|
|
}
|
|
|
|
d.conn.WriteToUDP(data, addr)
|
|
}
|
|
|
|
// sendResponse sends a response message
|
|
func (d *DHT) sendResponse(transactionID string, addr *net.UDPAddr, response map[string]interface{}) {
|
|
response["id"] = string(d.nodeID[:])
|
|
|
|
respMsg := map[string]interface{}{
|
|
"t": transactionID,
|
|
"y": ResponseMessage,
|
|
"r": response,
|
|
}
|
|
|
|
data, err := bencode.Marshal(respMsg)
|
|
if err != nil {
|
|
log.Printf("Failed to marshal response message: %v", err)
|
|
return
|
|
}
|
|
|
|
d.conn.WriteToUDP(data, addr)
|
|
d.statsMu.Lock()
|
|
d.stats.PacketsSent++
|
|
d.statsMu.Unlock()
|
|
}
|
|
|
|
// generateToken generates a secure token with rotation for announce_peer
|
|
func (d *DHT) generateToken(addr *net.UDPAddr) string {
|
|
d.tokenMu.RLock()
|
|
defer d.tokenMu.RUnlock()
|
|
|
|
// Rotate secret if needed
|
|
d.rotateTokenSecretIfNeeded()
|
|
|
|
// Get current time window (5-minute windows)
|
|
timeWindow := time.Now().Unix() / (TokenRotationMinutes * 60)
|
|
|
|
h := sha1.New()
|
|
h.Write(d.tokenSecret)
|
|
h.Write(addr.IP)
|
|
binary.Write(h, binary.BigEndian, uint16(addr.Port))
|
|
binary.Write(h, binary.BigEndian, timeWindow)
|
|
return fmt.Sprintf("%x", h.Sum(nil)[:TokenLength])
|
|
}
|
|
|
|
// validateToken validates a token for announce_peer (accepts current and previous time window)
|
|
func (d *DHT) validateToken(token string, addr *net.UDPAddr) bool {
|
|
d.tokenMu.RLock()
|
|
defer d.tokenMu.RUnlock()
|
|
|
|
currentTime := time.Now().Unix() / (TokenRotationMinutes * 60)
|
|
|
|
// Check current time window
|
|
if d.validateTokenForTimeWindow(token, addr, currentTime, d.tokenSecret) {
|
|
return true
|
|
}
|
|
|
|
// Check previous time window with current secret
|
|
if d.validateTokenForTimeWindow(token, addr, currentTime-1, d.tokenSecret) {
|
|
return true
|
|
}
|
|
|
|
// Check current time window with previous secret (if available)
|
|
if len(d.previousSecret) > 0 {
|
|
if d.validateTokenForTimeWindow(token, addr, currentTime, d.previousSecret) {
|
|
return true
|
|
}
|
|
|
|
// Check previous time window with previous secret
|
|
if d.validateTokenForTimeWindow(token, addr, currentTime-1, d.previousSecret) {
|
|
return true
|
|
}
|
|
}
|
|
|
|
return false
|
|
}
|
|
|
|
// handlePingQuery handles ping queries
|
|
func (d *DHT) handlePingQuery(msg *DHTMessage, addr *net.UDPAddr) {
|
|
// Add querying node to routing table
|
|
if args := msg.Arguments; args != nil {
|
|
if nodeIDBytes, ok := args["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
}
|
|
|
|
// Send pong response
|
|
response := map[string]interface{}{}
|
|
d.sendResponse(msg.TransactionID, addr, response)
|
|
}
|
|
|
|
// handleFindNodeQuery handles find_node queries
|
|
func (d *DHT) handleFindNodeQuery(msg *DHTMessage, addr *net.UDPAddr) {
|
|
// Add querying node to routing table
|
|
if args := msg.Arguments; args != nil {
|
|
if nodeIDBytes, ok := args["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
// Get target from arguments
|
|
if targetBytes, ok := args["target"].(string); ok && len(targetBytes) == NodeIDLength {
|
|
var targetID NodeID
|
|
copy(targetID[:], targetBytes)
|
|
|
|
// Find closest nodes
|
|
closestNodes := d.findClosestNodes(targetID, MaxNodesReturn)
|
|
nodesBytes := d.encodeNodes(closestNodes)
|
|
|
|
response := map[string]interface{}{
|
|
"nodes": nodesBytes,
|
|
}
|
|
d.sendResponse(msg.TransactionID, addr, response)
|
|
return
|
|
}
|
|
}
|
|
|
|
d.sendError(msg.TransactionID, addr, 203, "Protocol Error")
|
|
}
|
|
|
|
// handleGetPeersQuery handles get_peers queries with enhanced peer selection and node inclusion
|
|
func (d *DHT) handleGetPeersQuery(msg *DHTMessage, addr *net.UDPAddr) {
|
|
// Add querying node to routing table
|
|
if args := msg.Arguments; args != nil {
|
|
if nodeIDBytes, ok := args["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
// Get info_hash from arguments
|
|
if infoHashBytes, ok := args["info_hash"].(string); ok {
|
|
infoHash := fmt.Sprintf("%x", infoHashBytes)
|
|
|
|
// Generate token for potential announce_peer
|
|
token := d.generateToken(addr)
|
|
|
|
// Get all peers for this torrent
|
|
allPeers := d.peerStorage.GetPeers(infoHash)
|
|
|
|
// Always find closest nodes for recursive search
|
|
var targetID NodeID
|
|
copy(targetID[:], infoHashBytes)
|
|
closestNodes := d.findClosestNodes(targetID, MaxResponseNodes)
|
|
nodesBytes := d.encodeNodes(closestNodes)
|
|
|
|
// Prepare response
|
|
response := map[string]interface{}{
|
|
"token": token,
|
|
}
|
|
|
|
if len(allPeers) > 0 {
|
|
// Select reliable peers preferentially
|
|
reliablePeers := d.selectReliablePeers(allPeers, MaxResponsePeers)
|
|
|
|
// Convert peers to compact format
|
|
var values []string
|
|
for _, peer := range reliablePeers {
|
|
// Handle both IPv4 and IPv6
|
|
if ip4 := peer.IP.To4(); ip4 != nil {
|
|
// IPv4 format: 6 bytes (4 IP + 2 port)
|
|
peerBytes := make([]byte, 6)
|
|
copy(peerBytes[:4], ip4)
|
|
binary.BigEndian.PutUint16(peerBytes[4:], uint16(peer.Port))
|
|
values = append(values, string(peerBytes))
|
|
} else if len(peer.IP) == 16 {
|
|
// IPv6 format: 18 bytes (16 IP + 2 port)
|
|
peerBytes := make([]byte, 18)
|
|
copy(peerBytes[:16], peer.IP)
|
|
binary.BigEndian.PutUint16(peerBytes[16:], uint16(peer.Port))
|
|
values = append(values, string(peerBytes))
|
|
}
|
|
}
|
|
|
|
if len(values) > 0 {
|
|
response["values"] = values
|
|
log.Printf("DHT: Returned %d peers for %s to %s", len(values), infoHash[:8], addr)
|
|
}
|
|
}
|
|
|
|
// Always include closest nodes to help with recursive search
|
|
if len(closestNodes) > 0 {
|
|
response["nodes"] = nodesBytes
|
|
log.Printf("DHT: Also returned %d nodes for %s to %s", len(closestNodes), infoHash[:8], addr)
|
|
}
|
|
|
|
// Send response (will contain peers and/or nodes plus token)
|
|
d.sendResponse(msg.TransactionID, addr, response)
|
|
return
|
|
}
|
|
}
|
|
|
|
d.sendError(msg.TransactionID, addr, 203, "Protocol Error")
|
|
}
|
|
|
|
// min returns the smaller of two integers
|
|
func min(a, b int) int {
|
|
if a < b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
// handleAnnouncePeerQuery handles announce_peer queries with enhanced validation and storage
|
|
func (d *DHT) handleAnnouncePeerQuery(msg *DHTMessage, addr *net.UDPAddr) {
|
|
ipStr := addr.IP.String()
|
|
|
|
// Check if IP is blocked
|
|
if d.rateLimiter.IsBlocked(ipStr) {
|
|
log.Printf("DHT: Blocked IP %s attempted announce", ipStr)
|
|
d.sendError(msg.TransactionID, addr, 201, "Generic Error")
|
|
return
|
|
}
|
|
|
|
// Check announce rate limit
|
|
if !d.rateLimiter.CheckAnnounceRate(ipStr) {
|
|
log.Printf("DHT: Rate limited IP %s", ipStr)
|
|
d.sendError(msg.TransactionID, addr, 201, "Generic Error")
|
|
return
|
|
}
|
|
|
|
if args := msg.Arguments; args != nil {
|
|
// Validate token (must be from our get_peers response)
|
|
if token, ok := args["token"].(string); ok {
|
|
if !d.validateToken(token, addr) {
|
|
log.Printf("DHT: Invalid token from %s (token: %s)", addr, token[:min(len(token), 16)])
|
|
d.sendError(msg.TransactionID, addr, 203, "Protocol Error")
|
|
return
|
|
}
|
|
} else {
|
|
log.Printf("DHT: Missing token from %s", addr)
|
|
d.sendError(msg.TransactionID, addr, 203, "Protocol Error")
|
|
return
|
|
}
|
|
|
|
// Extract peer ID if available
|
|
var peerID string
|
|
if nodeIDBytes, ok := args["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
peerID = hex.EncodeToString([]byte(nodeIDBytes))
|
|
|
|
// Add querying node to routing table
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
// Store peer info
|
|
if infoHashBytes, ok := args["info_hash"].(string); ok {
|
|
if port, ok := args["port"].(int); ok {
|
|
infoHash := fmt.Sprintf("%x", infoHashBytes)
|
|
|
|
// Check storage limit for this IP
|
|
if !d.rateLimiter.CheckStorageLimit(ipStr) {
|
|
log.Printf("DHT: Storage limit exceeded for IP %s", ipStr)
|
|
d.sendError(msg.TransactionID, addr, 201, "Generic Error")
|
|
return
|
|
}
|
|
|
|
// Create enhanced peer with metadata
|
|
now := time.Now()
|
|
peer := &Peer{
|
|
IP: addr.IP,
|
|
Port: port,
|
|
PeerID: peerID,
|
|
Added: now,
|
|
LastAnnounced: now,
|
|
Source: "announce",
|
|
ReliabilityScore: 50, // Start with neutral score
|
|
AnnounceCount: 1,
|
|
Verified: false,
|
|
}
|
|
|
|
// Store peer with error handling
|
|
if err := d.peerStorage.AddPeer(infoHash, peer); err != nil {
|
|
log.Printf("DHT: Failed to store peer %s:%d for %s: %v", addr.IP, port, infoHash[:8], err)
|
|
// Still send success to avoid revealing internal errors
|
|
} else {
|
|
// Increment storage count for rate limiting
|
|
d.rateLimiter.IncrementStorage(ipStr)
|
|
|
|
// Queue peer for verification if enabled
|
|
if d.peerVerification {
|
|
select {
|
|
case d.verificationQueue <- &PeerVerificationRequest{
|
|
InfoHash: infoHash,
|
|
Peer: peer,
|
|
Retries: 0,
|
|
}:
|
|
default:
|
|
// Verification queue full, skip verification
|
|
}
|
|
}
|
|
}
|
|
|
|
log.Printf("DHT: Successfully stored peer %s:%d for infohash %s", addr.IP, port, infoHash[:8])
|
|
|
|
// Send success response
|
|
response := map[string]interface{}{}
|
|
d.sendResponse(msg.TransactionID, addr, response)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
d.sendError(msg.TransactionID, addr, 203, "Protocol Error")
|
|
}
|
|
|
|
// findClosestNodes finds the closest nodes to a target ID (wrapper for RoutingTable method)
|
|
func (d *DHT) findClosestNodes(targetID NodeID, count int) []*Node {
|
|
return d.routingTable.FindClosestNodes(targetID, count)
|
|
}
|
|
|
|
// compareNodeIDs compares two node IDs, returns -1 if a < b, 0 if a == b, 1 if a > b
|
|
func compareNodeIDs(a, b NodeID) int {
|
|
for i := 0; i < NodeIDLength; i++ {
|
|
if a[i] < b[i] {
|
|
return -1
|
|
}
|
|
if a[i] > b[i] {
|
|
return 1
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// encodeNodes encodes nodes in compact format
|
|
func (d *DHT) encodeNodes(nodes []*Node) string {
|
|
var result []byte
|
|
|
|
for _, node := range nodes {
|
|
// Each node: 20 bytes node ID + 4 bytes IP + 2 bytes port
|
|
result = append(result, node.ID[:]...)
|
|
result = append(result, node.Addr.IP.To4()...)
|
|
|
|
portBytes := make([]byte, 2)
|
|
binary.BigEndian.PutUint16(portBytes, uint16(node.Addr.Port))
|
|
result = append(result, portBytes...)
|
|
}
|
|
|
|
return string(result)
|
|
}
|
|
|
|
// sendPing sends a ping message to a node (wrapper for Ping operation)
|
|
func (d *DHT) sendPing(addr *net.UDPAddr) error {
|
|
_, err := d.Ping(addr)
|
|
return err
|
|
}
|
|
|
|
// maintenance performs periodic maintenance tasks
|
|
func (d *DHT) maintenance() {
|
|
ticker := time.NewTicker(5 * time.Minute)
|
|
defer ticker.Stop()
|
|
|
|
for {
|
|
select {
|
|
case <-d.stopCh:
|
|
return
|
|
case <-ticker.C:
|
|
d.performMaintenance()
|
|
}
|
|
}
|
|
}
|
|
|
|
// performMaintenance cleans up old nodes and refreshes buckets
|
|
func (d *DHT) performMaintenance() {
|
|
// Rotate token secret if needed
|
|
d.tokenMu.Lock()
|
|
d.rotateTokenSecretIfNeeded()
|
|
d.tokenMu.Unlock()
|
|
|
|
// Perform health checks on all nodes
|
|
d.routingTable.PerformHealthCheck(d)
|
|
|
|
// Clean up nodes marked as bad
|
|
d.routingTable.CleanupBadNodes()
|
|
|
|
// Clean up expired peers
|
|
d.peerStorage.CleanExpiredPeers()
|
|
|
|
// Clean up unreliable peers
|
|
d.CleanUnreliablePeers()
|
|
|
|
// Refresh stale buckets
|
|
d.routingTable.RefreshBuckets(d)
|
|
|
|
// Update statistics
|
|
totalNodes := 0
|
|
goodNodes := 0
|
|
questionableNodes := 0
|
|
badNodes := 0
|
|
|
|
d.routingTable.mu.RLock()
|
|
for _, bucket := range d.routingTable.buckets {
|
|
if bucket == nil {
|
|
continue
|
|
}
|
|
|
|
bucket.mu.RLock()
|
|
for _, node := range bucket.nodes {
|
|
totalNodes++
|
|
switch node.Health {
|
|
case NodeGood:
|
|
goodNodes++
|
|
case NodeQuestionable:
|
|
questionableNodes++
|
|
case NodeBad:
|
|
badNodes++
|
|
}
|
|
}
|
|
bucket.mu.RUnlock()
|
|
}
|
|
d.routingTable.mu.RUnlock()
|
|
|
|
// Get peer storage stats
|
|
peerStats := d.peerStorage.GetStats()
|
|
torrentCount := d.peerStorage.GetTorrentCount()
|
|
verificationStats := d.GetVerificationStats()
|
|
|
|
d.statsMu.Lock()
|
|
d.stats.NodesInTable = totalNodes
|
|
d.stats.StoredItems = int(peerStats.TotalPeers)
|
|
d.statsMu.Unlock()
|
|
|
|
log.Printf("DHT maintenance: %d nodes (%d good, %d questionable, %d bad), %d peers across %d torrents",
|
|
totalNodes, goodNodes, questionableNodes, badNodes, peerStats.TotalPeers, torrentCount)
|
|
log.Printf("DHT peer cache: %d hits, %d misses, %d DB writes, %d DB reads",
|
|
peerStats.CacheHits, peerStats.CacheMisses, peerStats.DatabaseWrites, peerStats.DatabaseReads)
|
|
log.Printf("DHT verification: %.1f%% verified (%d/%d), %d high reliability, %d low reliability",
|
|
verificationStats["verification_rate"], verificationStats["verified_peers"],
|
|
verificationStats["total_peers"], verificationStats["high_reliability_peers"],
|
|
verificationStats["low_reliability_peers"])
|
|
}
|
|
|
|
// handleAnnouncements processes torrent announcements
|
|
func (d *DHT) handleAnnouncements() {
|
|
for {
|
|
select {
|
|
case <-d.stopCh:
|
|
return
|
|
case req := <-d.announceQueue:
|
|
d.processAnnounce(req)
|
|
}
|
|
}
|
|
}
|
|
|
|
// processAnnounce processes a torrent announce request
|
|
func (d *DHT) processAnnounce(req AnnounceRequest) {
|
|
// Store our own peer info for this torrent
|
|
peerInfo := make([]byte, 6) // 4 bytes IP + 2 bytes port
|
|
|
|
// Get our external IP (simplified - would need proper detection)
|
|
ip := net.ParseIP("127.0.0.1").To4()
|
|
copy(peerInfo[:4], ip)
|
|
binary.BigEndian.PutUint16(peerInfo[4:], uint16(req.Port))
|
|
|
|
d.storageMu.Lock()
|
|
d.storage[req.InfoHash] = peerInfo
|
|
d.storageMu.Unlock()
|
|
|
|
log.Printf("Announced torrent %s on port %d", req.InfoHash, req.Port)
|
|
}
|
|
|
|
// Announce announces a torrent to the DHT
|
|
func (d *DHT) Announce(infoHash string, port int) {
|
|
select {
|
|
case d.announceQueue <- AnnounceRequest{InfoHash: infoHash, Port: port}:
|
|
log.Printf("Queued announce for torrent %s", infoHash)
|
|
default:
|
|
log.Printf("Announce queue full, dropping announce for %s", infoHash)
|
|
}
|
|
}
|
|
|
|
// FindPeers searches for peers for a given info hash using new peer storage
|
|
func (d *DHT) FindPeers(infoHash string) ([]*Peer, error) {
|
|
peers := d.peerStorage.GetPeers(infoHash)
|
|
return peers, nil
|
|
}
|
|
|
|
// FindPeersFromNetwork actively searches the network for peers
|
|
func (d *DHT) FindPeersFromNetwork(infohash []byte) ([]*Peer, error) {
|
|
var allPeers []*Peer
|
|
|
|
// Get closest nodes to the infohash
|
|
var targetID NodeID
|
|
copy(targetID[:], infohash)
|
|
closestNodes := d.routingTable.FindClosestNodes(targetID, Alpha)
|
|
|
|
// Query each node for peers
|
|
for _, node := range closestNodes {
|
|
peers, nodes, err := d.GetPeers(infohash, node.Addr)
|
|
if err != nil {
|
|
log.Printf("GetPeers failed for node %s: %v", node.Addr, err)
|
|
continue
|
|
}
|
|
|
|
if peers != nil {
|
|
// Found peers
|
|
allPeers = append(allPeers, peers...)
|
|
} else if nodes != nil {
|
|
// Got more nodes to query
|
|
for _, newNode := range nodes {
|
|
morePeers, _, err := d.GetPeers(infohash, newNode.Addr)
|
|
if err == nil && morePeers != nil {
|
|
allPeers = append(allPeers, morePeers...)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
return allPeers, nil
|
|
}
|
|
|
|
// GetPeerStorageStats returns peer storage statistics
|
|
func (ps *PeerStorage) GetStats() PeerStorageStats {
|
|
ps.mu.RLock()
|
|
defer ps.mu.RUnlock()
|
|
return ps.stats
|
|
}
|
|
|
|
// GetTorrentCount returns the number of torrents with stored peers
|
|
func (ps *PeerStorage) GetTorrentCount() int {
|
|
ps.mu.RLock()
|
|
defer ps.mu.RUnlock()
|
|
return len(ps.infohashes)
|
|
}
|
|
|
|
// GetPeerCountForTorrent returns the number of peers for a specific torrent
|
|
func (ps *PeerStorage) GetPeerCountForTorrent(infohash string) int {
|
|
ps.mu.RLock()
|
|
peerList, exists := ps.infohashes[infohash]
|
|
ps.mu.RUnlock()
|
|
|
|
if !exists {
|
|
// Try database
|
|
if ps.db != nil {
|
|
count, err := ps.db.GetPeerCount(infohash)
|
|
if err == nil {
|
|
return count
|
|
}
|
|
}
|
|
return 0
|
|
}
|
|
|
|
peerList.mu.RLock()
|
|
defer peerList.mu.RUnlock()
|
|
return len(peerList.peers)
|
|
}
|
|
|
|
// GetStats returns current DHT statistics including peer storage
|
|
func (d *DHT) GetStats() Stats {
|
|
d.statsMu.Lock()
|
|
defer d.statsMu.Unlock()
|
|
|
|
// Update peer storage stats
|
|
peerStats := d.peerStorage.GetStats()
|
|
d.stats.StoredItems = int(peerStats.TotalPeers)
|
|
|
|
return d.stats
|
|
}
|
|
|
|
// AddNode adds a node to the routing table following Kademlia rules
|
|
func (rt *RoutingTable) AddNode(node *Node) {
|
|
// Don't add ourselves
|
|
if node.ID == rt.selfID {
|
|
return
|
|
}
|
|
|
|
distance := xor(rt.selfID, node.ID)
|
|
bucketIndex := leadingZeros(distance)
|
|
|
|
if bucketIndex >= NumBuckets {
|
|
bucketIndex = NumBuckets - 1
|
|
}
|
|
|
|
bucket := rt.buckets[bucketIndex]
|
|
bucket.mu.Lock()
|
|
defer bucket.mu.Unlock()
|
|
|
|
// Check if node already exists
|
|
for i, existingNode := range bucket.nodes {
|
|
if existingNode.ID == node.ID {
|
|
// Update existing node - move to tail (most recently seen)
|
|
updated := *node
|
|
updated.Health = NodeGood
|
|
updated.FailedQueries = 0
|
|
|
|
// Remove from current position
|
|
bucket.nodes = append(bucket.nodes[:i], bucket.nodes[i+1:]...)
|
|
// Add to tail
|
|
bucket.nodes = append(bucket.nodes, &updated)
|
|
bucket.lastChanged = time.Now()
|
|
return
|
|
}
|
|
}
|
|
|
|
// Add new node if bucket not full
|
|
if len(bucket.nodes) < BucketSize {
|
|
newNode := *node
|
|
newNode.Health = NodeGood
|
|
bucket.nodes = append(bucket.nodes, &newNode)
|
|
bucket.lastChanged = time.Now()
|
|
return
|
|
}
|
|
|
|
// Bucket is full - apply Kademlia replacement rules
|
|
// Look for bad nodes to replace first
|
|
for i, n := range bucket.nodes {
|
|
if n.Health == NodeBad {
|
|
newNode := *node
|
|
newNode.Health = NodeGood
|
|
bucket.nodes[i] = &newNode
|
|
bucket.lastChanged = time.Now()
|
|
return
|
|
}
|
|
}
|
|
|
|
// No bad nodes - look for questionable nodes
|
|
for i, n := range bucket.nodes {
|
|
if n.Health == NodeQuestionable {
|
|
// Ping the questionable node to verify it's still alive
|
|
// For now, just replace it (in a full implementation, we'd ping first)
|
|
newNode := *node
|
|
newNode.Health = NodeGood
|
|
bucket.nodes[i] = &newNode
|
|
bucket.lastChanged = time.Now()
|
|
return
|
|
}
|
|
}
|
|
|
|
// All nodes are good - check if the head node is stale
|
|
// Kademlia: if all nodes are good, ping the head (least recently seen)
|
|
// If it responds, move to tail and discard new node
|
|
// If it doesn't respond, replace it with new node
|
|
head := bucket.nodes[0]
|
|
if time.Since(head.LastSeen) > 15*time.Minute {
|
|
head.Health = NodeQuestionable
|
|
// In a full implementation, we'd ping here and wait for response
|
|
// For now, just replace stale nodes
|
|
newNode := *node
|
|
newNode.Health = NodeGood
|
|
bucket.nodes[0] = &newNode
|
|
bucket.lastChanged = time.Now()
|
|
}
|
|
// If head is fresh, discard the new node (Kademlia rule)
|
|
}
|
|
|
|
// FindClosestNodes returns the K closest nodes to a target ID using proper Kademlia algorithm
|
|
func (rt *RoutingTable) FindClosestNodes(targetID NodeID, count int) []*Node {
|
|
distance := xor(rt.selfID, targetID)
|
|
startBucket := leadingZeros(distance)
|
|
|
|
if startBucket >= NumBuckets {
|
|
startBucket = NumBuckets - 1
|
|
}
|
|
|
|
var candidates []*Node
|
|
|
|
// Start from the closest bucket and expand outward
|
|
for offset := 0; offset < NumBuckets; offset++ {
|
|
// Check bucket at startBucket + offset
|
|
if startBucket+offset < NumBuckets {
|
|
bucket := rt.buckets[startBucket+offset]
|
|
bucket.mu.RLock()
|
|
for _, node := range bucket.nodes {
|
|
if node.Health != NodeBad {
|
|
candidates = append(candidates, node)
|
|
}
|
|
}
|
|
bucket.mu.RUnlock()
|
|
}
|
|
|
|
// Check bucket at startBucket - offset (if different)
|
|
if offset > 0 && startBucket-offset >= 0 {
|
|
bucket := rt.buckets[startBucket-offset]
|
|
bucket.mu.RLock()
|
|
for _, node := range bucket.nodes {
|
|
if node.Health != NodeBad {
|
|
candidates = append(candidates, node)
|
|
}
|
|
}
|
|
bucket.mu.RUnlock()
|
|
}
|
|
|
|
// If we have enough candidates, stop searching
|
|
if len(candidates) >= count*2 {
|
|
break
|
|
}
|
|
}
|
|
|
|
// Sort candidates by XOR distance
|
|
type nodeDistance struct {
|
|
node *Node
|
|
distance NodeID
|
|
}
|
|
|
|
var distances []nodeDistance
|
|
for _, node := range candidates {
|
|
dist := xor(node.ID, targetID)
|
|
distances = append(distances, nodeDistance{node: node, distance: dist})
|
|
}
|
|
|
|
// Sort using more efficient sorting
|
|
for i := 0; i < len(distances)-1; i++ {
|
|
minIdx := i
|
|
for j := i + 1; j < len(distances); j++ {
|
|
if compareNodeIDs(distances[j].distance, distances[minIdx].distance) < 0 {
|
|
minIdx = j
|
|
}
|
|
}
|
|
if minIdx != i {
|
|
distances[i], distances[minIdx] = distances[minIdx], distances[i]
|
|
}
|
|
}
|
|
|
|
// Return up to count closest nodes
|
|
var result []*Node
|
|
for i := 0; i < len(distances) && i < count; i++ {
|
|
result = append(result, distances[i].node)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// UpdateNodeHealth updates the health status of a node
|
|
func (rt *RoutingTable) UpdateNodeHealth(nodeID NodeID, health NodeHealth) {
|
|
distance := xor(rt.selfID, nodeID)
|
|
bucketIndex := leadingZeros(distance)
|
|
|
|
if bucketIndex >= NumBuckets {
|
|
bucketIndex = NumBuckets - 1
|
|
}
|
|
|
|
bucket := rt.buckets[bucketIndex]
|
|
bucket.mu.Lock()
|
|
defer bucket.mu.Unlock()
|
|
|
|
for _, node := range bucket.nodes {
|
|
if node.ID == nodeID {
|
|
node.Health = health
|
|
if health == NodeBad {
|
|
node.FailedQueries++
|
|
} else if health == NodeGood {
|
|
node.FailedQueries = 0
|
|
node.LastSeen = time.Now()
|
|
}
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// RemoveNode removes a node from the routing table
|
|
func (rt *RoutingTable) RemoveNode(nodeID NodeID) {
|
|
distance := xor(rt.selfID, nodeID)
|
|
bucketIndex := leadingZeros(distance)
|
|
|
|
if bucketIndex >= NumBuckets {
|
|
bucketIndex = NumBuckets - 1
|
|
}
|
|
|
|
bucket := rt.buckets[bucketIndex]
|
|
bucket.mu.Lock()
|
|
defer bucket.mu.Unlock()
|
|
|
|
for i, node := range bucket.nodes {
|
|
if node.ID == nodeID {
|
|
// Remove node from slice
|
|
bucket.nodes = append(bucket.nodes[:i], bucket.nodes[i+1:]...)
|
|
bucket.lastChanged = time.Now()
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// GetBucketInfo returns information about a bucket for debugging
|
|
func (rt *RoutingTable) GetBucketInfo(bucketIndex int) (int, time.Time) {
|
|
if bucketIndex >= NumBuckets || bucketIndex < 0 {
|
|
return 0, time.Time{}
|
|
}
|
|
|
|
bucket := rt.buckets[bucketIndex]
|
|
bucket.mu.RLock()
|
|
defer bucket.mu.RUnlock()
|
|
|
|
return len(bucket.nodes), bucket.lastChanged
|
|
}
|
|
|
|
// RefreshBuckets performs bucket refresh according to Kademlia
|
|
func (rt *RoutingTable) RefreshBuckets(dht *DHT) {
|
|
now := time.Now()
|
|
refreshInterval := 15 * time.Minute
|
|
|
|
for i, bucket := range rt.buckets {
|
|
bucket.mu.Lock()
|
|
needsRefresh := now.Sub(bucket.lastRefresh) > refreshInterval
|
|
|
|
if needsRefresh {
|
|
// Generate a random ID in this bucket's range
|
|
targetID := rt.generateRandomIDForBucket(i)
|
|
bucket.lastRefresh = now
|
|
bucket.mu.Unlock()
|
|
|
|
// Perform lookup to refresh this bucket
|
|
go dht.performLookup(targetID)
|
|
} else {
|
|
bucket.mu.Unlock()
|
|
}
|
|
}
|
|
}
|
|
|
|
// generateRandomIDForBucket generates a random node ID that would fall in the specified bucket
|
|
func (rt *RoutingTable) generateRandomIDForBucket(bucketIndex int) NodeID {
|
|
var targetID NodeID
|
|
copy(targetID[:], rt.selfID[:])
|
|
|
|
if bucketIndex == 0 {
|
|
// For bucket 0, flip a random bit in the last few bits
|
|
randomByte := byte(time.Now().UnixNano() % 256)
|
|
targetID[NodeIDLength-1] ^= randomByte
|
|
} else if bucketIndex < 160 {
|
|
// Calculate which bit to flip based on bucket index
|
|
byteIndex := (159 - bucketIndex) / 8
|
|
bitIndex := (159 - bucketIndex) % 8
|
|
|
|
// Flip the bit at the calculated position
|
|
targetID[byteIndex] ^= (1 << bitIndex)
|
|
|
|
// Add some randomness to the remaining bits
|
|
for i := byteIndex + 1; i < NodeIDLength; i++ {
|
|
targetID[i] ^= byte(time.Now().UnixNano() % 256)
|
|
}
|
|
}
|
|
|
|
return targetID
|
|
}
|
|
|
|
// PerformHealthCheck checks the health of nodes and marks questionable ones
|
|
func (rt *RoutingTable) PerformHealthCheck(dht *DHT) {
|
|
now := time.Now()
|
|
questionableThreshold := 15 * time.Minute
|
|
badThreshold := 30 * time.Minute
|
|
|
|
for _, bucket := range rt.buckets {
|
|
bucket.mu.Lock()
|
|
|
|
for _, node := range bucket.nodes {
|
|
timeSinceLastSeen := now.Sub(node.LastSeen)
|
|
|
|
// Mark nodes as questionable if not seen recently
|
|
if node.Health == NodeGood && timeSinceLastSeen > questionableThreshold {
|
|
node.Health = NodeQuestionable
|
|
// Ping questionable nodes to verify they're still alive
|
|
go dht.pingNode(node)
|
|
} else if node.Health == NodeQuestionable && timeSinceLastSeen > badThreshold {
|
|
// Mark as bad if no response for a long time
|
|
node.Health = NodeBad
|
|
}
|
|
}
|
|
|
|
bucket.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// CleanupBadNodes removes nodes marked as bad from the routing table
|
|
func (rt *RoutingTable) CleanupBadNodes() {
|
|
for _, bucket := range rt.buckets {
|
|
bucket.mu.Lock()
|
|
|
|
// Filter out bad nodes
|
|
goodNodes := make([]*Node, 0, len(bucket.nodes))
|
|
for _, node := range bucket.nodes {
|
|
if node.Health != NodeBad || node.FailedQueries < 3 {
|
|
goodNodes = append(goodNodes, node)
|
|
}
|
|
}
|
|
|
|
if len(goodNodes) != len(bucket.nodes) {
|
|
bucket.nodes = goodNodes
|
|
bucket.lastChanged = time.Now()
|
|
}
|
|
|
|
bucket.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// pingNode pings a node to check if it's still alive
|
|
func (d *DHT) pingNode(node *Node) {
|
|
start := time.Now()
|
|
err := d.sendPing(node.Addr)
|
|
|
|
if err == nil {
|
|
// Node responded - mark as good and update RTT
|
|
node.Health = NodeGood
|
|
node.FailedQueries = 0
|
|
node.LastSeen = time.Now()
|
|
node.RTT = time.Since(start)
|
|
} else {
|
|
// Node didn't respond - increment failed queries
|
|
node.FailedQueries++
|
|
if node.FailedQueries >= 3 {
|
|
node.Health = NodeBad
|
|
}
|
|
}
|
|
}
|
|
|
|
// performLookup performs a node lookup for bucket refresh
|
|
func (d *DHT) performLookup(targetID NodeID) {
|
|
// This is a simplified lookup - in a full implementation,
|
|
// this would be an iterative process finding closer and closer nodes
|
|
closestNodes := d.routingTable.FindClosestNodes(targetID, Alpha)
|
|
|
|
for _, node := range closestNodes {
|
|
// Send find_node query to each close node
|
|
args := map[string]interface{}{
|
|
"id": string(d.nodeID[:]),
|
|
"target": string(targetID[:]),
|
|
}
|
|
|
|
go func(n *Node) {
|
|
response, err := d.sendQuery("find_node", args, n.Addr)
|
|
if err == nil && response.Response != nil {
|
|
if nodesBytes, ok := response.Response["nodes"].(string); ok {
|
|
// Parse and add returned nodes to routing table
|
|
d.parseAndAddNodes([]byte(nodesBytes))
|
|
}
|
|
}
|
|
}(node)
|
|
}
|
|
}
|
|
|
|
// ============ CORE DHT OPERATIONS ============
|
|
|
|
// Ping sends a ping query to a node and updates routing table on response
|
|
func (d *DHT) Ping(addr *net.UDPAddr) (*Node, error) {
|
|
args := map[string]interface{}{
|
|
"id": string(d.nodeID[:]),
|
|
}
|
|
|
|
response, err := d.sendQuery("ping", args, addr)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("ping failed: %w", err)
|
|
}
|
|
|
|
// Parse response and extract node ID
|
|
if response.Response == nil {
|
|
return nil, fmt.Errorf("invalid ping response")
|
|
}
|
|
|
|
nodeIDBytes, ok := response.Response["id"].(string)
|
|
if !ok || len(nodeIDBytes) != NodeIDLength {
|
|
return nil, fmt.Errorf("invalid node ID in ping response")
|
|
}
|
|
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
|
|
// Create and add node to routing table
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
RTT: time.Since(time.Now()), // This would be calculated properly in sendQuery
|
|
}
|
|
|
|
d.routingTable.AddNode(node)
|
|
log.Printf("Ping successful: node %x at %s", nodeID[:8], addr)
|
|
|
|
return node, nil
|
|
}
|
|
|
|
// FindNode queries for nodes close to a target ID
|
|
func (d *DHT) FindNode(targetID NodeID, addr *net.UDPAddr) ([]*Node, error) {
|
|
args := map[string]interface{}{
|
|
"id": string(d.nodeID[:]),
|
|
"target": string(targetID[:]),
|
|
}
|
|
|
|
response, err := d.sendQuery("find_node", args, addr)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("find_node failed: %w", err)
|
|
}
|
|
|
|
if response.Response == nil {
|
|
return nil, fmt.Errorf("invalid find_node response")
|
|
}
|
|
|
|
// Parse responding node ID and add to routing table
|
|
if nodeIDBytes, ok := response.Response["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
respondingNode := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(respondingNode)
|
|
}
|
|
|
|
// Parse nodes from response
|
|
nodesBytes, ok := response.Response["nodes"].(string)
|
|
if !ok {
|
|
return nil, fmt.Errorf("no nodes in find_node response")
|
|
}
|
|
|
|
nodes := d.parseCompactNodes([]byte(nodesBytes))
|
|
|
|
// Add discovered nodes to routing table
|
|
for _, node := range nodes {
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
log.Printf("FindNode successful: discovered %d nodes from %s", len(nodes), addr)
|
|
return nodes, nil
|
|
}
|
|
|
|
// GetPeers searches for peers of a torrent by infohash
|
|
func (d *DHT) GetPeers(infohash []byte, addr *net.UDPAddr) ([]*Peer, []*Node, error) {
|
|
args := map[string]interface{}{
|
|
"id": string(d.nodeID[:]),
|
|
"info_hash": string(infohash),
|
|
}
|
|
|
|
response, err := d.sendQuery("get_peers", args, addr)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("get_peers failed: %w", err)
|
|
}
|
|
|
|
if response.Response == nil {
|
|
return nil, nil, fmt.Errorf("invalid get_peers response")
|
|
}
|
|
|
|
// Parse responding node ID and add to routing table
|
|
if nodeIDBytes, ok := response.Response["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
respondingNode := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(respondingNode)
|
|
}
|
|
|
|
// Check if response contains peers or nodes
|
|
if values, ok := response.Response["values"].([]interface{}); ok {
|
|
// Response contains peers
|
|
var peers []*Peer
|
|
for _, value := range values {
|
|
if peerBytes, ok := value.(string); ok && len(peerBytes) == 6 {
|
|
ip := net.IP(peerBytes[:4])
|
|
port := int(binary.BigEndian.Uint16([]byte(peerBytes[4:])))
|
|
peers = append(peers, &Peer{
|
|
IP: ip,
|
|
Port: port,
|
|
Added: time.Now(),
|
|
})
|
|
}
|
|
}
|
|
log.Printf("GetPeers successful: found %d peers from %s", len(peers), addr)
|
|
return peers, nil, nil
|
|
}
|
|
|
|
// Response contains nodes
|
|
nodesBytes, ok := response.Response["nodes"].(string)
|
|
if !ok {
|
|
return nil, nil, fmt.Errorf("no peers or nodes in get_peers response")
|
|
}
|
|
|
|
nodes := d.parseCompactNodes([]byte(nodesBytes))
|
|
|
|
// Add discovered nodes to routing table
|
|
for _, node := range nodes {
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
|
|
log.Printf("GetPeers successful: found %d nodes from %s", len(nodes), addr)
|
|
return nil, nodes, nil
|
|
}
|
|
|
|
// AnnouncePeer stores peer info for an infohash after validating token
|
|
func (d *DHT) AnnouncePeer(infohash []byte, port int, token string, addr *net.UDPAddr) error {
|
|
args := map[string]interface{}{
|
|
"id": string(d.nodeID[:]),
|
|
"info_hash": string(infohash),
|
|
"port": port,
|
|
"token": token,
|
|
}
|
|
|
|
response, err := d.sendQuery("announce_peer", args, addr)
|
|
if err != nil {
|
|
return fmt.Errorf("announce_peer failed: %w", err)
|
|
}
|
|
|
|
if response.Response == nil {
|
|
return fmt.Errorf("invalid announce_peer response")
|
|
}
|
|
|
|
// Parse responding node ID and add to routing table
|
|
if nodeIDBytes, ok := response.Response["id"].(string); ok && len(nodeIDBytes) == NodeIDLength {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodeIDBytes)
|
|
respondingNode := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
d.routingTable.AddNode(respondingNode)
|
|
}
|
|
|
|
// Store peer info in our storage
|
|
peer := &Peer{
|
|
IP: addr.IP,
|
|
Port: port,
|
|
Added: time.Now(),
|
|
}
|
|
|
|
infohashHex := fmt.Sprintf("%x", infohash)
|
|
d.peerStorage.AddPeer(infohashHex, peer)
|
|
|
|
log.Printf("AnnouncePeer successful: stored peer %s:%d for %s", addr.IP, port, infohashHex[:8])
|
|
return nil
|
|
}
|
|
|
|
// ============ HELPER FUNCTIONS ============
|
|
|
|
// parseCompactNodes parses compact node format (26 bytes per node)
|
|
func (d *DHT) parseCompactNodes(nodesBytes []byte) []*Node {
|
|
var nodes []*Node
|
|
nodeSize := 26 // 20-byte ID + 4-byte IP + 2-byte port
|
|
|
|
for i := 0; i+nodeSize <= len(nodesBytes); i += nodeSize {
|
|
var nodeID NodeID
|
|
copy(nodeID[:], nodesBytes[i:i+20])
|
|
|
|
ip := net.IP(nodesBytes[i+20 : i+24])
|
|
port := binary.BigEndian.Uint16(nodesBytes[i+24 : i+26])
|
|
|
|
addr := &net.UDPAddr{
|
|
IP: ip,
|
|
Port: int(port),
|
|
}
|
|
|
|
node := &Node{
|
|
ID: nodeID,
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
|
|
nodes = append(nodes, node)
|
|
}
|
|
|
|
return nodes
|
|
}
|
|
|
|
// parseAndAddNodes parses compact node format and adds nodes to routing table
|
|
func (d *DHT) parseAndAddNodes(nodesBytes []byte) {
|
|
nodes := d.parseCompactNodes(nodesBytes)
|
|
for _, node := range nodes {
|
|
d.routingTable.AddNode(node)
|
|
}
|
|
}
|
|
|
|
// ============ ITERATIVE LOOKUP FUNCTIONS ============
|
|
|
|
// iterativeFindNode performs an iterative node lookup using Kademlia algorithm
|
|
func (d *DHT) iterativeFindNode(targetID NodeID, initialNodes []*net.UDPAddr) []*Node {
|
|
return d.iterativeLookup(targetID, initialNodes, "find_node")
|
|
}
|
|
|
|
// iterativeFindPeers performs an iterative peer lookup
|
|
func (d *DHT) iterativeFindPeers(targetID NodeID, initialNodes []*net.UDPAddr) ([]*Peer, []*Node) {
|
|
nodes := d.iterativeLookup(targetID, initialNodes, "get_peers")
|
|
// For now, return nodes only. Peers would be collected during the lookup process
|
|
return nil, nodes
|
|
}
|
|
|
|
// iterativeLookup performs the core iterative lookup algorithm
|
|
func (d *DHT) iterativeLookup(targetID NodeID, initialNodes []*net.UDPAddr, queryType string) []*Node {
|
|
// Shortlist of closest nodes found so far
|
|
shortlist := make(map[string]*Node)
|
|
queried := make(map[string]bool)
|
|
|
|
// Convert initial nodes to Node objects
|
|
for _, addr := range initialNodes {
|
|
// Create temporary node (we'll get real node ID when we query)
|
|
tempNode := &Node{
|
|
Addr: addr,
|
|
LastSeen: time.Now(),
|
|
Health: NodeGood,
|
|
}
|
|
key := fmt.Sprintf("%s:%d", addr.IP, addr.Port)
|
|
shortlist[key] = tempNode
|
|
}
|
|
|
|
// Add nodes from our routing table
|
|
routingTableNodes := d.routingTable.FindClosestNodes(targetID, BucketSize*2)
|
|
for _, node := range routingTableNodes {
|
|
key := fmt.Sprintf("%s:%d", node.Addr.IP, node.Addr.Port)
|
|
shortlist[key] = node
|
|
}
|
|
|
|
log.Printf("Starting iterative lookup for target %x with %d initial nodes", targetID[:8], len(shortlist))
|
|
|
|
// Iterative process
|
|
for iteration := 0; iteration < 10; iteration++ { // Max 10 iterations
|
|
// Find Alpha closest unqueried nodes
|
|
candidates := d.selectCandidates(shortlist, queried, targetID, Alpha)
|
|
if len(candidates) == 0 {
|
|
log.Printf("No more candidates, stopping after %d iterations", iteration)
|
|
break
|
|
}
|
|
|
|
log.Printf("Iteration %d: querying %d nodes", iteration+1, len(candidates))
|
|
|
|
// Query candidates in parallel
|
|
newNodes := d.queryNodesParallel(candidates, targetID, queryType, queried)
|
|
|
|
// Add new nodes to shortlist
|
|
addedNew := false
|
|
for _, node := range newNodes {
|
|
key := fmt.Sprintf("%s:%d", node.Addr.IP, node.Addr.Port)
|
|
if _, exists := shortlist[key]; !exists {
|
|
shortlist[key] = node
|
|
addedNew = true
|
|
}
|
|
}
|
|
|
|
// If no new nodes found, stop
|
|
if !addedNew {
|
|
log.Printf("No new nodes found, stopping after %d iterations", iteration+1)
|
|
break
|
|
}
|
|
}
|
|
|
|
// Return the closest nodes found
|
|
result := d.getClosestFromShortlist(shortlist, targetID, BucketSize)
|
|
log.Printf("Iterative lookup completed: found %d nodes", len(result))
|
|
return result
|
|
}
|
|
|
|
// selectCandidates selects the Alpha closest unqueried nodes
|
|
func (d *DHT) selectCandidates(shortlist map[string]*Node, queried map[string]bool, targetID NodeID, count int) []*Node {
|
|
type nodeDistance struct {
|
|
node *Node
|
|
distance NodeID
|
|
key string
|
|
}
|
|
|
|
var candidates []nodeDistance
|
|
for key, node := range shortlist {
|
|
if !queried[key] && node.Health != NodeBad {
|
|
// For initial nodes without real node IDs, use a hash of their address
|
|
var nodeIDForDistance NodeID
|
|
if node.ID == (NodeID{}) {
|
|
// Create pseudo node ID from address for distance calculation
|
|
hash := sha1.Sum([]byte(fmt.Sprintf("%s:%d", node.Addr.IP, node.Addr.Port)))
|
|
copy(nodeIDForDistance[:], hash[:])
|
|
} else {
|
|
nodeIDForDistance = node.ID
|
|
}
|
|
|
|
dist := xor(nodeIDForDistance, targetID)
|
|
candidates = append(candidates, nodeDistance{
|
|
node: node,
|
|
distance: dist,
|
|
key: key,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Sort by distance
|
|
for i := 0; i < len(candidates)-1; i++ {
|
|
for j := i + 1; j < len(candidates); j++ {
|
|
if compareNodeIDs(candidates[i].distance, candidates[j].distance) > 0 {
|
|
candidates[i], candidates[j] = candidates[j], candidates[i]
|
|
}
|
|
}
|
|
}
|
|
|
|
// Return up to count closest candidates
|
|
var result []*Node
|
|
for i := 0; i < len(candidates) && i < count; i++ {
|
|
result = append(result, candidates[i].node)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// queryNodesParallel queries multiple nodes in parallel
|
|
func (d *DHT) queryNodesParallel(candidates []*Node, targetID NodeID, queryType string, queried map[string]bool) []*Node {
|
|
type queryResult struct {
|
|
nodes []*Node
|
|
node *Node
|
|
}
|
|
|
|
resultChan := make(chan queryResult, len(candidates))
|
|
var wg sync.WaitGroup
|
|
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
|
defer cancel()
|
|
|
|
// Launch parallel queries
|
|
for _, candidate := range candidates {
|
|
wg.Add(1)
|
|
go func(node *Node) {
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
log.Printf("DHT query panic recovered: %v", r)
|
|
}
|
|
wg.Done()
|
|
}()
|
|
|
|
key := fmt.Sprintf("%s:%d", node.Addr.IP, node.Addr.Port)
|
|
queried[key] = true
|
|
|
|
var discoveredNodes []*Node
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
default:
|
|
if queryType == "find_node" {
|
|
nodes, err := d.FindNode(targetID, node.Addr)
|
|
if err == nil {
|
|
discoveredNodes = nodes
|
|
}
|
|
} else if queryType == "get_peers" {
|
|
_, nodes, err := d.GetPeers(targetID[:], node.Addr)
|
|
if err == nil && nodes != nil {
|
|
discoveredNodes = nodes
|
|
}
|
|
}
|
|
|
|
select {
|
|
case resultChan <- queryResult{nodes: discoveredNodes, node: node}:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
}(candidate)
|
|
}
|
|
|
|
// Close channel when all queries complete or timeout
|
|
go func() {
|
|
done := make(chan struct{})
|
|
go func() {
|
|
defer close(done)
|
|
wg.Wait()
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-ctx.Done():
|
|
log.Printf("DHT parallel query timeout after 30s")
|
|
}
|
|
close(resultChan)
|
|
}()
|
|
|
|
// Collect results with timeout
|
|
var allNodes []*Node
|
|
for result := range resultChan {
|
|
if len(result.nodes) > 0 {
|
|
allNodes = append(allNodes, result.nodes...)
|
|
}
|
|
}
|
|
|
|
return allNodes
|
|
}
|
|
|
|
// getClosestFromShortlist returns the closest nodes from the shortlist
|
|
func (d *DHT) getClosestFromShortlist(shortlist map[string]*Node, targetID NodeID, count int) []*Node {
|
|
type nodeDistance struct {
|
|
node *Node
|
|
distance NodeID
|
|
}
|
|
|
|
var candidates []nodeDistance
|
|
for _, node := range shortlist {
|
|
if node.Health != NodeBad {
|
|
// Use actual node ID if available, otherwise create from address
|
|
var nodeIDForDistance NodeID
|
|
if node.ID != (NodeID{}) {
|
|
nodeIDForDistance = node.ID
|
|
} else {
|
|
hash := sha1.Sum([]byte(fmt.Sprintf("%s:%d", node.Addr.IP, node.Addr.Port)))
|
|
copy(nodeIDForDistance[:], hash[:])
|
|
}
|
|
|
|
dist := xor(nodeIDForDistance, targetID)
|
|
candidates = append(candidates, nodeDistance{
|
|
node: node,
|
|
distance: dist,
|
|
})
|
|
}
|
|
}
|
|
|
|
// Sort by distance
|
|
for i := 0; i < len(candidates)-1; i++ {
|
|
for j := i + 1; j < len(candidates); j++ {
|
|
if compareNodeIDs(candidates[i].distance, candidates[j].distance) > 0 {
|
|
candidates[i], candidates[j] = candidates[j], candidates[i]
|
|
}
|
|
}
|
|
}
|
|
|
|
// Return up to count closest nodes
|
|
var result []*Node
|
|
for i := 0; i < len(candidates) && i < count; i++ {
|
|
result = append(result, candidates[i].node)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// xor calculates XOR distance between two node IDs
|
|
func xor(a, b NodeID) NodeID {
|
|
var result NodeID
|
|
for i := 0; i < NodeIDLength; i++ {
|
|
result[i] = a[i] ^ b[i]
|
|
}
|
|
return result
|
|
}
|
|
|
|
// leadingZeros counts leading zero bits
|
|
func leadingZeros(id NodeID) int {
|
|
for i, b := range id {
|
|
if b != 0 {
|
|
// Count zeros in this byte
|
|
zeros := 0
|
|
for bit := 7; bit >= 0; bit-- {
|
|
if (b>>bit)&1 == 0 {
|
|
zeros++
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
return i*8 + zeros
|
|
}
|
|
}
|
|
return NodeIDLength * 8
|
|
}
|
|
|
|
// SimpleDatabaseImpl provides a simple database implementation
|
|
type SimpleDatabaseImpl struct {
|
|
db *sql.DB
|
|
}
|
|
|
|
// NewSimpleDatabase creates a new simple database implementation
|
|
func NewSimpleDatabase(db *sql.DB) *SimpleDatabaseImpl {
|
|
return &SimpleDatabaseImpl{db: db}
|
|
}
|
|
|
|
// CreateDHTTables creates the necessary DHT tables with reliability tracking
|
|
func (sdb *SimpleDatabaseImpl) CreateDHTTables() error {
|
|
query := `
|
|
CREATE TABLE IF NOT EXISTS dht_stored_peers (
|
|
infohash TEXT NOT NULL,
|
|
ip TEXT NOT NULL,
|
|
port INTEGER NOT NULL,
|
|
peer_id TEXT,
|
|
last_announced DATETIME DEFAULT CURRENT_TIMESTAMP,
|
|
reliability_score INTEGER DEFAULT 50,
|
|
announce_count INTEGER DEFAULT 1,
|
|
verified BOOLEAN DEFAULT FALSE,
|
|
last_verified DATETIME,
|
|
PRIMARY KEY (infohash, ip, port)
|
|
);
|
|
|
|
CREATE INDEX IF NOT EXISTS idx_dht_peers_infohash ON dht_stored_peers(infohash);
|
|
CREATE INDEX IF NOT EXISTS idx_dht_peers_last_announced ON dht_stored_peers(last_announced);
|
|
CREATE INDEX IF NOT EXISTS idx_dht_peers_reliability ON dht_stored_peers(reliability_score);
|
|
`
|
|
_, err := sdb.db.Exec(query)
|
|
return err
|
|
}
|
|
|
|
// StorePeer stores a peer in the database with reliability tracking
|
|
func (sdb *SimpleDatabaseImpl) StorePeer(infohash, ip string, port int, peerID string, lastAnnounced time.Time) error {
|
|
// Check if peer exists to update announce count
|
|
var existingCount int
|
|
existQuery := `SELECT announce_count FROM dht_stored_peers WHERE infohash = ? AND ip = ? AND port = ?`
|
|
err := sdb.db.QueryRow(existQuery, infohash, ip, port).Scan(&existingCount)
|
|
|
|
if err == sql.ErrNoRows {
|
|
// New peer
|
|
query := `INSERT INTO dht_stored_peers (infohash, ip, port, peer_id, last_announced, reliability_score, announce_count)
|
|
VALUES (?, ?, ?, ?, ?, 50, 1)`
|
|
_, err := sdb.db.Exec(query, infohash, ip, port, peerID, lastAnnounced)
|
|
return err
|
|
} else if err != nil {
|
|
return err
|
|
} else {
|
|
// Existing peer - update
|
|
query := `UPDATE dht_stored_peers
|
|
SET peer_id = ?, last_announced = ?, announce_count = announce_count + 1,
|
|
reliability_score = MIN(100, reliability_score + 5)
|
|
WHERE infohash = ? AND ip = ? AND port = ?`
|
|
_, err := sdb.db.Exec(query, peerID, lastAnnounced, infohash, ip, port)
|
|
return err
|
|
}
|
|
}
|
|
|
|
// UpdatePeerReliability updates peer reliability information
|
|
func (sdb *SimpleDatabaseImpl) UpdatePeerReliability(infohash, ip string, port int, reliabilityScore int, verified bool, lastVerified time.Time) error {
|
|
query := `UPDATE dht_stored_peers
|
|
SET reliability_score = ?, verified = ?, last_verified = ?
|
|
WHERE infohash = ? AND ip = ? AND port = ?`
|
|
_, err := sdb.db.Exec(query, reliabilityScore, verified, lastVerified, infohash, ip, port)
|
|
return err
|
|
}
|
|
|
|
// GetPeersForTorrent retrieves peers for a specific torrent ordered by reliability
|
|
func (sdb *SimpleDatabaseImpl) GetPeersForTorrent(infohash string) ([]*StoredPeer, error) {
|
|
query := `SELECT infohash, ip, port, peer_id, last_announced, reliability_score, announce_count, verified, last_verified
|
|
FROM dht_stored_peers
|
|
WHERE infohash = ? AND last_announced > datetime('now', '-30 minutes')
|
|
ORDER BY reliability_score DESC, last_announced DESC
|
|
LIMIT ?`
|
|
|
|
rows, err := sdb.db.Query(query, infohash, MaxPeersPerTorrent)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
var peers []*StoredPeer
|
|
for rows.Next() {
|
|
var peer StoredPeer
|
|
var lastVerified sql.NullTime
|
|
err := rows.Scan(&peer.InfoHash, &peer.IP, &peer.Port, &peer.PeerID,
|
|
&peer.LastAnnounced, &peer.ReliabilityScore, &peer.AnnounceCount,
|
|
&peer.Verified, &lastVerified)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if lastVerified.Valid {
|
|
peer.LastVerified = lastVerified.Time
|
|
}
|
|
|
|
peers = append(peers, &peer)
|
|
}
|
|
|
|
return peers, rows.Err()
|
|
}
|
|
|
|
// CleanExpiredPeers removes peers older than 30 minutes
|
|
func (sdb *SimpleDatabaseImpl) CleanExpiredPeers() error {
|
|
query := `DELETE FROM dht_stored_peers WHERE last_announced < datetime('now', '-30 minutes')`
|
|
result, err := sdb.db.Exec(query)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rowsAffected, _ := result.RowsAffected()
|
|
if rowsAffected > 0 {
|
|
log.Printf("DHT: Cleaned %d expired peers from database", rowsAffected)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// GetPeerCount returns the number of peers for a specific torrent
|
|
func (sdb *SimpleDatabaseImpl) GetPeerCount(infohash string) (int, error) {
|
|
query := `SELECT COUNT(*) FROM dht_stored_peers
|
|
WHERE infohash = ? AND last_announced > datetime('now', '-30 minutes')`
|
|
|
|
var count int
|
|
err := sdb.db.QueryRow(query, infohash).Scan(&count)
|
|
return count, err
|
|
}
|
|
|
|
// GetStoredPeersWithNodes returns both stored peers and closest nodes for better queryability
|
|
func (d *DHT) GetStoredPeersWithNodes(infoHash string) (peers []*Peer, nodes []*Node, err error) {
|
|
// Get stored peers first
|
|
peers = d.peerStorage.GetPeers(infoHash)
|
|
|
|
// If we have peers, return them
|
|
if len(peers) > 0 {
|
|
return peers, nil, nil
|
|
}
|
|
|
|
// Otherwise, find closest nodes to the infohash
|
|
var targetID NodeID
|
|
if hashBytes, err := hex.DecodeString(infoHash); err == nil && len(hashBytes) == 20 {
|
|
copy(targetID[:], hashBytes)
|
|
} else {
|
|
// Generate target from infohash string
|
|
hash := sha1.Sum([]byte(infoHash))
|
|
copy(targetID[:], hash[:])
|
|
}
|
|
|
|
nodes = d.routingTable.FindClosestNodes(targetID, MaxResponseNodes)
|
|
return nil, nodes, nil
|
|
}
|
|
|
|
// GetAllStoredTorrents returns a list of all torrents with stored peers
|
|
func (ps *PeerStorage) GetAllStoredTorrents() []string {
|
|
ps.mu.RLock()
|
|
defer ps.mu.RUnlock()
|
|
|
|
var torrents []string
|
|
for infohash := range ps.infohashes {
|
|
torrents = append(torrents, infohash)
|
|
}
|
|
|
|
return torrents
|
|
}
|
|
|
|
// GetStorageInfo returns comprehensive storage information
|
|
func (d *DHT) GetStorageInfo() map[string]interface{} {
|
|
peerStats := d.peerStorage.GetStats()
|
|
torrentCount := d.peerStorage.GetTorrentCount()
|
|
allTorrents := d.peerStorage.GetAllStoredTorrents()
|
|
|
|
return map[string]interface{}{
|
|
"total_peers": peerStats.TotalPeers,
|
|
"total_torrents": torrentCount,
|
|
"cache_hits": peerStats.CacheHits,
|
|
"cache_misses": peerStats.CacheMisses,
|
|
"database_writes": peerStats.DatabaseWrites,
|
|
"database_reads": peerStats.DatabaseReads,
|
|
"stored_torrents": allTorrents,
|
|
"max_peers_limit": MaxPeersPerTorrent,
|
|
"peer_expiry_mins": PeerExpiryMinutes,
|
|
"token_rotation_mins": TokenRotationMinutes,
|
|
}
|
|
}
|
|
|
|
// FindPeersWithFallback provides enhanced peer discovery with multiple fallbacks
|
|
func (d *DHT) FindPeersWithFallback(infoHash string) ([]*Peer, []*Node, error) {
|
|
// Try storage first
|
|
peers, nodes, err := d.GetStoredPeersWithNodes(infoHash)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
|
|
// If we have peers, return them
|
|
if len(peers) > 0 {
|
|
log.Printf("DHT: Found %d stored peers for %s", len(peers), infoHash[:8])
|
|
return peers, nil, nil
|
|
}
|
|
|
|
// If we have nodes, we can use them for further queries
|
|
if len(nodes) > 0 {
|
|
log.Printf("DHT: Found %d closest nodes for %s", len(nodes), infoHash[:8])
|
|
return nil, nodes, nil
|
|
}
|
|
|
|
// Last resort: try network lookup if we have routing table nodes
|
|
if infoHashBytes, err := hex.DecodeString(infoHash); err == nil && len(infoHashBytes) == 20 {
|
|
networkPeers, err := d.FindPeersFromNetwork(infoHashBytes)
|
|
if err == nil && len(networkPeers) > 0 {
|
|
log.Printf("DHT: Found %d network peers for %s", len(networkPeers), infoHash[:8])
|
|
return networkPeers, nil, nil
|
|
}
|
|
}
|
|
|
|
log.Printf("DHT: No peers found for %s", infoHash[:8])
|
|
return nil, nil, fmt.Errorf("no peers found")
|
|
}
|
|
|
|
// selectReliablePeers selects the most reliable peers up to the limit
|
|
func (d *DHT) selectReliablePeers(peers []*Peer, limit int) []*Peer {
|
|
if len(peers) <= limit {
|
|
return peers
|
|
}
|
|
|
|
// Sort peers by reliability score (descending)
|
|
sortedPeers := make([]*Peer, len(peers))
|
|
copy(sortedPeers, peers)
|
|
|
|
// Simple bubble sort by reliability score
|
|
for i := 0; i < len(sortedPeers)-1; i++ {
|
|
for j := 0; j < len(sortedPeers)-1-i; j++ {
|
|
if sortedPeers[j].ReliabilityScore < sortedPeers[j+1].ReliabilityScore {
|
|
sortedPeers[j], sortedPeers[j+1] = sortedPeers[j+1], sortedPeers[j]
|
|
}
|
|
}
|
|
}
|
|
|
|
// Filter out peers with very low reliability scores
|
|
var reliablePeers []*Peer
|
|
for _, peer := range sortedPeers {
|
|
if peer.ReliabilityScore >= MinReliabilityScore && len(reliablePeers) < limit {
|
|
reliablePeers = append(reliablePeers, peer)
|
|
}
|
|
}
|
|
|
|
// If we don't have enough reliable peers, fill with less reliable ones
|
|
if len(reliablePeers) < limit {
|
|
for _, peer := range sortedPeers {
|
|
if peer.ReliabilityScore < MinReliabilityScore && len(reliablePeers) < limit {
|
|
// Include less reliable peers but flag them
|
|
reliablePeers = append(reliablePeers, peer)
|
|
}
|
|
}
|
|
}
|
|
|
|
return reliablePeers[:min(len(reliablePeers), limit)]
|
|
}
|
|
|
|
// handlePeerVerification processes peer verification requests
|
|
func (d *DHT) handlePeerVerification() {
|
|
for {
|
|
select {
|
|
case <-d.stopCh:
|
|
return
|
|
case req := <-d.verificationQueue:
|
|
d.verifyPeer(req)
|
|
}
|
|
}
|
|
}
|
|
|
|
// verifyPeer attempts to verify a peer by connecting to it
|
|
func (d *DHT) verifyPeer(req *PeerVerificationRequest) {
|
|
peerAddr := fmt.Sprintf("%s:%d", req.Peer.IP.String(), req.Peer.Port)
|
|
|
|
// Attempt to connect with timeout
|
|
conn, err := net.DialTimeout("tcp", peerAddr, PeerVerificationTimeout)
|
|
if err != nil {
|
|
// Peer verification failed
|
|
d.handleVerificationFailure(req)
|
|
return
|
|
}
|
|
defer conn.Close()
|
|
|
|
// Peer responded - update reliability score
|
|
d.updatePeerReliability(req.InfoHash, req.Peer, true)
|
|
log.Printf("DHT: Verified peer %s for %s", peerAddr, req.InfoHash[:8])
|
|
}
|
|
|
|
// handleVerificationFailure handles when peer verification fails
|
|
func (d *DHT) handleVerificationFailure(req *PeerVerificationRequest) {
|
|
peerAddr := fmt.Sprintf("%s:%d", req.Peer.IP.String(), req.Peer.Port)
|
|
|
|
// Retry up to 2 times with exponential backoff
|
|
if req.Retries < 2 {
|
|
req.Retries++
|
|
|
|
// Exponential backoff: 30s, 60s
|
|
delay := time.Duration(30*req.Retries) * time.Second
|
|
|
|
go func() {
|
|
time.Sleep(delay)
|
|
select {
|
|
case d.verificationQueue <- req:
|
|
log.Printf("DHT: Retrying verification for peer %s (attempt %d)", peerAddr, req.Retries+1)
|
|
default:
|
|
// Queue full, skip retry
|
|
log.Printf("DHT: Failed to queue verification retry for peer %s", peerAddr)
|
|
}
|
|
}()
|
|
return
|
|
}
|
|
|
|
// All retries failed - mark peer as unreliable
|
|
d.updatePeerReliability(req.InfoHash, req.Peer, false)
|
|
log.Printf("DHT: Peer verification failed for %s after %d attempts", peerAddr, req.Retries+1)
|
|
}
|
|
|
|
// updatePeerReliability updates the reliability score of a peer
|
|
func (d *DHT) updatePeerReliability(infoHash string, peer *Peer, verified bool) {
|
|
d.peerStorage.mu.Lock()
|
|
defer d.peerStorage.mu.Unlock()
|
|
|
|
peerList, exists := d.peerStorage.infohashes[infoHash]
|
|
if !exists {
|
|
return
|
|
}
|
|
|
|
peerList.mu.Lock()
|
|
defer peerList.mu.Unlock()
|
|
|
|
key := fmt.Sprintf("%s:%d", peer.IP.String(), peer.Port)
|
|
if storedPeer, exists := peerList.peers[key]; exists {
|
|
now := time.Now()
|
|
|
|
if verified {
|
|
// Increase reliability score
|
|
storedPeer.ReliabilityScore = min(MaxReliabilityScore, storedPeer.ReliabilityScore+20)
|
|
storedPeer.Verified = true
|
|
storedPeer.LastVerified = now
|
|
} else {
|
|
// Decrease reliability score
|
|
storedPeer.ReliabilityScore = max(0, storedPeer.ReliabilityScore-30)
|
|
// Don't mark as verified=false immediately, might be temporary network issue
|
|
}
|
|
|
|
// Update database if available
|
|
if d.peerStorage.db != nil {
|
|
err := d.updatePeerInDatabase(infoHash, storedPeer)
|
|
if err != nil {
|
|
log.Printf("DHT: Failed to update peer reliability in database: %v", err)
|
|
}
|
|
}
|
|
|
|
log.Printf("DHT: Updated peer %s reliability score to %d (verified: %v)",
|
|
key, storedPeer.ReliabilityScore, storedPeer.Verified)
|
|
}
|
|
}
|
|
|
|
// max returns the larger of two integers
|
|
func max(a, b int) int {
|
|
if a > b {
|
|
return a
|
|
}
|
|
return b
|
|
}
|
|
|
|
// updatePeerInDatabase updates peer information in the database
|
|
func (d *DHT) updatePeerInDatabase(infoHash string, peer *Peer) error {
|
|
if d.peerStorage.db == nil {
|
|
return nil
|
|
}
|
|
|
|
// Update peer reliability information
|
|
return d.peerStorage.db.UpdatePeerReliability(
|
|
infoHash,
|
|
peer.IP.String(),
|
|
peer.Port,
|
|
peer.ReliabilityScore,
|
|
peer.Verified,
|
|
peer.LastVerified,
|
|
)
|
|
}
|
|
|
|
// SetPeerVerification enables or disables peer verification
|
|
func (d *DHT) SetPeerVerification(enabled bool) {
|
|
d.peerVerification = enabled
|
|
log.Printf("DHT: Peer verification %s", map[bool]string{true: "enabled", false: "disabled"}[enabled])
|
|
}
|
|
|
|
// GetVerificationStats returns peer verification statistics
|
|
func (d *DHT) GetVerificationStats() map[string]interface{} {
|
|
stats := make(map[string]interface{})
|
|
|
|
// Count peers by verification status
|
|
totalPeers := 0
|
|
verifiedPeers := 0
|
|
highReliabilityPeers := 0
|
|
lowReliabilityPeers := 0
|
|
|
|
d.peerStorage.mu.RLock()
|
|
for _, peerList := range d.peerStorage.infohashes {
|
|
peerList.mu.RLock()
|
|
for _, peer := range peerList.peers {
|
|
totalPeers++
|
|
if peer.Verified {
|
|
verifiedPeers++
|
|
}
|
|
if peer.ReliabilityScore >= 80 {
|
|
highReliabilityPeers++
|
|
} else if peer.ReliabilityScore <= MinReliabilityScore {
|
|
lowReliabilityPeers++
|
|
}
|
|
}
|
|
peerList.mu.RUnlock()
|
|
}
|
|
d.peerStorage.mu.RUnlock()
|
|
|
|
stats["total_peers"] = totalPeers
|
|
stats["verified_peers"] = verifiedPeers
|
|
stats["high_reliability_peers"] = highReliabilityPeers
|
|
stats["low_reliability_peers"] = lowReliabilityPeers
|
|
stats["verification_enabled"] = d.peerVerification
|
|
stats["verification_queue_length"] = len(d.verificationQueue)
|
|
|
|
if totalPeers > 0 {
|
|
stats["verification_rate"] = float64(verifiedPeers) / float64(totalPeers) * 100
|
|
} else {
|
|
stats["verification_rate"] = 0.0
|
|
}
|
|
|
|
return stats
|
|
}
|
|
|
|
// CleanUnreliablePeers removes peers with very low reliability scores
|
|
func (d *DHT) CleanUnreliablePeers() {
|
|
d.peerStorage.mu.Lock()
|
|
defer d.peerStorage.mu.Unlock()
|
|
|
|
cleanedCount := 0
|
|
|
|
for infoHash, peerList := range d.peerStorage.infohashes {
|
|
peerList.mu.Lock()
|
|
|
|
reliablePeers := make(map[string]*Peer)
|
|
for key, peer := range peerList.peers {
|
|
// Keep peers with decent reliability or recent announces
|
|
if peer.ReliabilityScore >= 10 || time.Since(peer.LastAnnounced) < 10*time.Minute {
|
|
reliablePeers[key] = peer
|
|
} else {
|
|
cleanedCount++
|
|
log.Printf("DHT: Removed unreliable peer %s (score: %d) for %s",
|
|
key, peer.ReliabilityScore, infoHash[:8])
|
|
}
|
|
}
|
|
|
|
peerList.peers = reliablePeers
|
|
peerList.mu.Unlock()
|
|
|
|
// Remove empty peer lists
|
|
if len(reliablePeers) == 0 {
|
|
delete(d.peerStorage.infohashes, infoHash)
|
|
}
|
|
}
|
|
|
|
if cleanedCount > 0 {
|
|
log.Printf("DHT: Cleaned %d unreliable peers", cleanedCount)
|
|
}
|
|
}
|
|
|
|
// GetRateLimitStats returns rate limiting statistics
|
|
func (d *DHT) GetRateLimitStats() map[string]interface{} {
|
|
d.rateLimiter.mu.RLock()
|
|
defer d.rateLimiter.mu.RUnlock()
|
|
|
|
blockedIPs := 0
|
|
activeIPs := len(d.rateLimiter.announces)
|
|
|
|
now := time.Now()
|
|
for _, unblockTime := range d.rateLimiter.blocklist {
|
|
if now.Before(unblockTime) {
|
|
blockedIPs++
|
|
}
|
|
}
|
|
|
|
return map[string]interface{}{
|
|
"active_ips": activeIPs,
|
|
"blocked_ips": blockedIPs,
|
|
"max_announces_per_hour": MaxAnnouncesPerIPPerHour,
|
|
"max_storage_per_ip": MaxStoragePerIP,
|
|
"blocklist_duration_mins": BlocklistDurationMinutes,
|
|
}
|
|
}
|
|
|
|
// GetComprehensiveStats returns all DHT statistics
|
|
func (d *DHT) GetComprehensiveStats() map[string]interface{} {
|
|
baseStats := d.GetStats()
|
|
storageInfo := d.GetStorageInfo()
|
|
verificationStats := d.GetVerificationStats()
|
|
rateLimitStats := d.GetRateLimitStats()
|
|
|
|
return map[string]interface{}{
|
|
"node_stats": baseStats,
|
|
"storage_stats": storageInfo,
|
|
"verification_stats": verificationStats,
|
|
"rate_limit_stats": rateLimitStats,
|
|
"node_id": hex.EncodeToString(d.nodeID[:]),
|
|
"port": d.config.Port,
|
|
"routing_table_buckets": NumBuckets,
|
|
}
|
|
}
|
|
|
|
// ForceVerifyPeer manually triggers verification for a specific peer
|
|
func (d *DHT) ForceVerifyPeer(infoHash string, ip string, port int) error {
|
|
peers := d.peerStorage.GetPeers(infoHash)
|
|
|
|
for _, peer := range peers {
|
|
if peer.IP.String() == ip && peer.Port == port {
|
|
req := &PeerVerificationRequest{
|
|
InfoHash: infoHash,
|
|
Peer: peer,
|
|
Retries: 0,
|
|
}
|
|
|
|
select {
|
|
case d.verificationQueue <- req:
|
|
log.Printf("DHT: Queued forced verification for peer %s:%d", ip, port)
|
|
return nil
|
|
default:
|
|
return fmt.Errorf("verification queue full")
|
|
}
|
|
}
|
|
}
|
|
|
|
return fmt.Errorf("peer not found")
|
|
}
|
|
|
|
// GetTokenStats returns token generation statistics
|
|
func (d *DHT) GetTokenStats() map[string]interface{} {
|
|
d.tokenMu.RLock()
|
|
defer d.tokenMu.RUnlock()
|
|
|
|
return map[string]interface{}{
|
|
"secret_rotated_at": d.lastSecretRotation,
|
|
"next_rotation": d.lastSecretRotation.Add(TokenRotationMinutes * time.Minute),
|
|
"rotation_interval_mins": TokenRotationMinutes,
|
|
"token_length_bytes": TokenLength,
|
|
"has_previous_secret": len(d.previousSecret) > 0,
|
|
}
|
|
}
|
|
|
|
// GenerateInfoHash generates an info hash for a torrent name (for testing)
|
|
func GenerateInfoHash(name string) string {
|
|
hash := sha1.Sum([]byte(name))
|
|
return fmt.Sprintf("%x", hash)
|
|
} |