Add WebTorrent module

2023-03-06 17:17:30 +03:00
parent 1e83dc85f7
commit 775b1818d1
10 changed files with 866 additions and 91 deletions
@@ -1,71 +1,19 @@
 package core

 import (
-	"sync"
+	cryptorand "crypto/rand"
 )

-// Waiter support:
-// - autotart on first Wait
-// - block new waiters after last Done
-// - safe Done after finish
-type Waiter struct {
-	sync.WaitGroup
-	mu    sync.Mutex
-	state int // state < 0 means finish
-}
+const digits = "0123456789abcdefghijklmnopqrstuvwxyz"
+const maxSize = byte(len(digits))

-func (w *Waiter) Add(delta int) {
-	w.mu.Lock()
-	if w.state >= 0 {
-		w.state += delta
-		w.WaitGroup.Add(delta)
+func RandString(size byte) string {
+	b := make([]byte, size)
+	if _, err := cryptorand.Read(b); err != nil {
+		panic(err)
 	}
-	w.mu.Unlock()
-}
-
-func (w *Waiter) Wait() {
-	w.mu.Lock()
-	// first wait auto start waiter
-	if w.state == 0 {
-		w.state++
-		w.WaitGroup.Add(1)
+	for i := byte(0); i < size; i++ {
+		b[i] = digits[b[i]%maxSize]
 	}
-	w.mu.Unlock()
-
-	w.WaitGroup.Wait()
-}
-
-func (w *Waiter) Done() {
-	w.mu.Lock()
-
-	// safe run Done only when have tasks
-	if w.state > 0 {
-		w.state--
-		w.WaitGroup.Done()
-	}
-
-	// block waiter for any operations after last done
-	if w.state == 0 {
-		w.state = -1
-	}
-
-	w.mu.Unlock()
-}
-
-func (w *Waiter) WaitChan() <-chan struct{} {
-	var ch chan struct{}
-
-	w.mu.Lock()
-
-	if w.state >= 0 {
-		ch = make(chan struct{})
-		go func() {
-			w.Wait()
-			ch <- struct{}{}
-		}()
-	}
-
-	w.mu.Unlock()
-
-	return ch
+	return string(b)
 }
@@ -0,0 +1,71 @@
+package core
+
+import (
+	"sync"
+)
+
+// Waiter support:
+// - autotart on first Wait
+// - block new waiters after last Done
+// - safe Done after finish
+type Waiter struct {
+	sync.WaitGroup
+	mu    sync.Mutex
+	state int // state < 0 means finish
+}
+
+func (w *Waiter) Add(delta int) {
+	w.mu.Lock()
+	if w.state >= 0 {
+		w.state += delta
+		w.WaitGroup.Add(delta)
+	}
+	w.mu.Unlock()
+}
+
+func (w *Waiter) Wait() {
+	w.mu.Lock()
+	// first wait auto start waiter
+	if w.state == 0 {
+		w.state++
+		w.WaitGroup.Add(1)
+	}
+	w.mu.Unlock()
+
+	w.WaitGroup.Wait()
+}
+
+func (w *Waiter) Done() {
+	w.mu.Lock()
+
+	// safe run Done only when have tasks
+	if w.state > 0 {
+		w.state--
+		w.WaitGroup.Done()
+	}
+
+	// block waiter for any operations after last done
+	if w.state == 0 {
+		w.state = -1
+	}
+
+	w.mu.Unlock()
+}
+
+func (w *Waiter) WaitChan() <-chan struct{} {
+	var ch chan struct{}
+
+	w.mu.Lock()
+
+	if w.state >= 0 {
+		ch = make(chan struct{})
+		go func() {
+			w.Wait()
+			ch <- struct{}{}
+		}()
+	}
+
+	w.mu.Unlock()
+
+	return ch
+}
@@ -0,0 +1,52 @@
+package core
+
+import (
+	"time"
+)
+
+type Worker struct {
+	timer *time.Timer
+	done  chan struct{}
+}
+
+// NewWorker run f after d
+func NewWorker(d time.Duration, f func() time.Duration) *Worker {
+	timer := time.NewTimer(d)
+	done := make(chan struct{})
+
+	go func() {
+		for {
+			select {
+			case <-timer.C:
+				if d = f(); d > 0 {
+					timer.Reset(d)
+					continue
+				}
+			case <-done:
+				timer.Stop()
+			}
+			break
+		}
+	}()
+
+	return &Worker{timer: timer, done: done}
+}
+
+// Do - instant timer run
+func (w *Worker) Do() {
+	if w == nil {
+		return
+	}
+	w.timer.Reset(0)
+}
+
+func (w *Worker) Stop() {
+	if w == nil {
+		return
+	}
+
+	select {
+	case w.done <- struct{}{}:
+	default:
+	}
+}
@@ -0,0 +1,91 @@
+package webtorrent
+
+import (
+	"encoding/base64"
+	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/AlexxIT/go2rtc/pkg/webrtc"
+	"github.com/gorilla/websocket"
+	pion "github.com/pion/webrtc/v3"
+	"strconv"
+	"time"
+)
+
+func NewClient(tracker, share, pwd string, pc *pion.PeerConnection) (*webrtc.Conn, error) {
+	// 1. Create WebRTC producer
+	prod := webrtc.NewConn(pc)
+
+	medias := []*streamer.Media{
+		{Kind: streamer.KindVideo, Direction: streamer.DirectionRecvonly},
+		{Kind: streamer.KindAudio, Direction: streamer.DirectionRecvonly},
+	}
+
+	// 2. Create offer
+	offer, err := prod.CreateCompleteOffer(medias)
+	if err != nil {
+		return nil, err
+	}
+
+	// 3. Encrypt offer
+	nonce := strconv.FormatInt(time.Now().UnixNano(), 36)
+
+	cipher, err := NewCipher(share, pwd, nonce)
+	if err != nil {
+		return nil, err
+	}
+
+	enc := cipher.Encrypt([]byte(offer))
+
+	// 4. Connect to Tracker
+	ws, _, err := websocket.DefaultDialer.Dial(tracker, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	defer ws.Close()
+
+	// 5. Send offer
+	msg := fmt.Sprintf(
+		`{"action":"announce","info_hash":"%s","peer_id":"%s","offers":[{"offer_id":"%s","offer":{"type":"offer","sdp":"%s"}}],"numwant":1}`,
+		InfoHash(share), core.RandString(16), nonce, base64.StdEncoding.EncodeToString(enc),
+	)
+	if err = ws.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
+		return nil, err
+	}
+
+	// wait 30 seconds until full answer
+	if err = ws.SetReadDeadline(time.Now().Add(time.Second * 30)); err != nil {
+		return nil, err
+	}
+
+	for {
+		// 6. Read answer
+		var v Message
+		if err = ws.ReadJSON(&v); err != nil {
+			return nil, err
+		}
+
+		if v.Answer == nil {
+			continue
+		}
+
+		// 7. Decrypt answer
+		enc, err = base64.StdEncoding.DecodeString(v.Answer.SDP)
+		if err != nil {
+			return nil, err
+		}
+
+		answer, err := cipher.Decrypt(enc)
+		if err != nil {
+			return nil, err
+		}
+
+		// 8. Set answer
+		if err = prod.SetAnswer(string(answer)); err != nil {
+			return nil, err
+		}
+
+		return prod, nil
+	}
+}
@@ -0,0 +1,72 @@
+package webtorrent
+
+import (
+	"crypto/aes"
+	"crypto/cipher"
+	"crypto/sha256"
+	"encoding/base64"
+	"fmt"
+	"strconv"
+	"time"
+)
+
+type Cipher struct {
+	gcm   cipher.AEAD
+	iv    []byte
+	nonce []byte
+}
+
+func NewCipher(share, pwd, nonce string) (*Cipher, error) {
+	timestamp, err := strconv.ParseInt(nonce, 36, 0)
+	if err != nil {
+		return nil, err
+	}
+
+	delta := time.Duration(time.Now().UnixNano() - timestamp)
+	if delta < 0 {
+		delta = -delta
+	}
+
+	// protect from replay attack, but respect wrong timezone on server
+	if delta > 12*time.Hour {
+		return nil, fmt.Errorf("wrong timedelta %s", delta)
+	}
+
+	c := &Cipher{}
+
+	hash := sha256.New()
+	hash.Write([]byte(nonce + ":" + pwd))
+	key := hash.Sum(nil)
+
+	hash.Reset()
+	hash.Write([]byte(share + ":" + nonce))
+	c.iv = hash.Sum(nil)[:12]
+
+	block, err := aes.NewCipher(key)
+	if err != nil {
+		return nil, err
+	}
+	c.gcm, err = cipher.NewGCM(block)
+	if err != nil {
+		return nil, err
+	}
+
+	c.nonce = []byte(nonce)
+
+	return c, nil
+}
+
+func (c *Cipher) Decrypt(ciphertext []byte) ([]byte, error) {
+	return c.gcm.Open(nil, c.iv, ciphertext, c.nonce)
+}
+
+func (c *Cipher) Encrypt(plaintext []byte) []byte {
+	return c.gcm.Seal(nil, c.iv, plaintext, c.nonce)
+}
+
+func InfoHash(share string) string {
+	hash := sha256.New()
+	hash.Write([]byte(share))
+	sum := hash.Sum(nil)
+	return base64.StdEncoding.EncodeToString(sum)
+}
@@ -0,0 +1,222 @@
+package webtorrent
+
+import (
+	"encoding/base64"
+	"fmt"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/streamer"
+	"github.com/gorilla/websocket"
+	"sync"
+	"time"
+)
+
+type Server struct {
+	streamer.Element
+
+	URL      string
+	Exchange func(src, offer string) (answer string, err error)
+
+	shares   map[string]*Share
+	mu       sync.Mutex
+	announce *core.Worker
+}
+
+type Share struct {
+	name string
+	pwd  string
+	src  string
+}
+
+func (s *Server) AddShare(name, pwd, src string) {
+	s.mu.Lock()
+
+	if s.shares == nil {
+		s.shares = map[string]*Share{}
+	}
+
+	if len(s.shares) == 0 {
+		go s.Serve()
+	}
+
+	hash := InfoHash(name)
+	s.shares[hash] = &Share{
+		name: name,
+		pwd:  pwd,
+		src:  src,
+	}
+
+	s.announce.Do()
+
+	s.mu.Unlock()
+}
+
+func (s *Server) GetSharePwd(name string) (pwd string) {
+	hash := InfoHash(name)
+	s.mu.Lock()
+	if share, ok := s.shares[hash]; ok {
+		pwd = share.pwd
+	}
+	s.mu.Unlock()
+	return
+}
+
+func (s *Server) RemoveShare(name string) {
+	hash := InfoHash(name)
+	s.mu.Lock()
+	if _, ok := s.shares[hash]; ok {
+		delete(s.shares, hash)
+	}
+	s.mu.Unlock()
+}
+
+// Serve - run reconnection loop, will exit on??
+func (s *Server) Serve() error {
+	for s.HasShares() {
+		s.Fire("connect to tracker: " + s.URL)
+
+		ws, _, err := websocket.DefaultDialer.Dial(s.URL, nil)
+		if err != nil {
+			s.Fire(err)
+			time.Sleep(time.Minute)
+			continue
+		}
+
+		peerID := core.RandString(16)
+
+		// instant run announce worker
+		s.announce = core.NewWorker(0, func() time.Duration {
+			if err = s.writer(ws, peerID); err != nil {
+				return 0
+			}
+			return time.Minute
+		})
+
+		// run reader forewer
+		for {
+			if err = s.reader(ws, peerID); err != nil {
+				break
+			}
+		}
+
+		// stop announcing worker
+		s.announce.Stop()
+
+		// ensure ws is stopped
+		_ = ws.Close()
+
+		time.Sleep(time.Minute)
+	}
+
+	s.Fire("disconnect")
+
+	return nil
+}
+
+// reader - receive offers in the loop, will exit on ws.Close
+func (s *Server) reader(ws *websocket.Conn, peerID string) error {
+	var v Message
+	if err := ws.ReadJSON(&v); err != nil {
+		return err
+	}
+
+	s.Fire(&v)
+
+	s.mu.Lock()
+	share, ok := s.shares[v.InfoHash]
+	s.mu.Unlock()
+
+	// skip any unknown shares
+	if !ok || v.OfferId == "" || v.Offer == nil {
+		return nil
+	}
+
+	cipher, err := NewCipher(share.name, share.pwd, v.OfferId)
+	if err != nil {
+		return nil
+	}
+
+	enc, err := base64.StdEncoding.DecodeString(v.Offer.SDP)
+	if err != nil {
+		return nil
+	}
+
+	offer, err := cipher.Decrypt(enc)
+	if err != nil {
+		return nil
+	}
+
+	answer, err := s.Exchange(share.src, string(offer))
+	if err != nil {
+		return nil
+	}
+
+	enc = cipher.Encrypt([]byte(answer))
+
+	raw := fmt.Sprintf(
+		`{"action":"announce","info_hash":"%s","peer_id":"%s","offer_id":"%s","answer":{"type":"answer","sdp":"%s"},"to_peer_id":"%s"}`,
+		v.InfoHash, peerID, v.OfferId, base64.StdEncoding.EncodeToString(enc), v.PeerId,
+	)
+	return ws.WriteMessage(websocket.TextMessage, []byte(raw))
+}
+
+func (s *Server) writer(ws *websocket.Conn, peerID string) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if len(s.shares) == 0 {
+		return ws.Close()
+	}
+
+	for hash := range s.shares {
+		msg := fmt.Sprintf(
+			`{"action":"announce","info_hash":"%s","peer_id":"%s","offers":[],"numwant":10}`,
+			hash, peerID,
+		)
+		if err := ws.WriteMessage(websocket.TextMessage, []byte(msg)); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (s *Server) HasShares() bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return len(s.shares) > 0
+}
+
+type Message struct {
+	Action   string `json:"action"`
+	InfoHash string `json:"info_hash"`
+
+	// Announce msg
+	Numwant int    `json:"numwant,omitempty"`
+	PeerId  string `json:"peer_id,omitempty"`
+	Offers  []struct {
+		OfferId string `json:"offer_id"`
+		Offer   struct {
+			Type string `json:"type"`
+			SDP  string `json:"sdp"`
+		} `json:"offer"`
+	} `json:"offers,omitempty"`
+
+	// Interval msg
+	Interval   int `json:"interval,omitempty"`
+	Complete   int `json:"complete,omitempty"`
+	Incomplete int `json:"incomplete,omitempty"`
+
+	// Offer msg
+	OfferId string `json:"offer_id,omitempty"`
+	Offer   *struct {
+		Type string `json:"type"`
+		SDP  string `json:"sdp"`
+	} `json:"offer,omitempty"`
+
+	// Answer msg
+	ToPeerId string `json:"to_peer_id,omitempty"`
+	Answer   *struct {
+		Type string `json:"type"`
+		SDP  string `json:"sdp"`
+	} `json:"answer,omitempty"`
+}