// Package anchor implements the YAW/2 anchor client. // It connects to the anchor WebSocket, handles challenge/join, and routes // sealed signaling payloads. It manages PeerConnection lifecycle and delegates // DataChannel handling to internal/mesh. package anchor import ( "context" "crypto/sha256" "encoding/hex" "encoding/json" "fmt" "log" "strings" "sync" "time" "filippo.io/edwards25519" "github.com/pion/webrtc/v3" "nhooyr.io/websocket" "nhooyr.io/websocket/wsjson" "github.com/waste-go/internal/crypto" "github.com/waste-go/internal/mesh" "github.com/waste-go/internal/proto" ) // Run connects to anchorURL, joins networkName, and blocks handling signaling. // Reconnects automatically on disconnect. Cancel ctx to stop. func Run(ctx context.Context, anchorURL, networkName string, id *crypto.Identity, m *mesh.Mesh) { netHash := hashNetName(networkName) for { if err := runOnce(ctx, anchorURL, netHash, id, m); err != nil { if ctx.Err() != nil { return } log.Printf("anchor: %v — reconnecting in 5s", err) select { case <-ctx.Done(): return case <-time.After(5 * time.Second): } } } } func runOnce(ctx context.Context, anchorURL, netHash string, id *crypto.Identity, m *mesh.Mesh) error { conn, _, err := websocket.Dial(ctx, anchorURL, nil) if err != nil { return fmt.Errorf("dial: %w", err) } defer conn.Close(websocket.StatusNormalClosure, "bye") log.Printf("anchor: connected to %s", anchorURL) sendCh := make(chan proto.AnchorMessage, 64) go func() { for msg := range sendCh { if err := wsjson.Write(ctx, conn, msg); err != nil { return } } }() var ( mu sync.RWMutex pcs = make(map[proto.PeerID]*webrtc.PeerConnection) ) sender := &sender{id: id, sendCh: sendCh} for { var msg proto.AnchorMessage if err := wsjson.Read(ctx, conn, &msg); err != nil { return fmt.Errorf("read: %w", err) } switch msg.Type { case proto.AnchorChallenge: nonceBytes, err := hex.DecodeString(msg.Nonce) if err != nil { return fmt.Errorf("bad challenge nonce: %w", err) } // §5.1: sig covers nonce_raw || net_ascii (64-char hex string as UTF-8) sig := id.Sign(append(nonceBytes, []byte(netHash)...)) sendCh <- proto.AnchorMessage{ Type: proto.AnchorJoin, ID: string(id.PeerID()), Net: netHash, Sig: sig, } case proto.AnchorJoined: log.Printf("anchor: joined network, %d peer(s) present", len(msg.Peers)) for _, peerHex := range msg.Peers { pid := proto.PeerID(peerHex) if strings.Compare(string(id.PeerID()), peerHex) > 0 { go func(pid proto.PeerID) { pc, err := offer(pid, id, m, sender) if err != nil { log.Printf("anchor: offer to %s: %v", pid.Short(), err) return } mu.Lock() pcs[pid] = pc mu.Unlock() }(pid) } } case proto.AnchorPeerJoin: pid := proto.PeerID(msg.ID) log.Printf("anchor: peer joined: %s", pid.Short()) if strings.Compare(string(id.PeerID()), msg.ID) > 0 { go func(pid proto.PeerID) { pc, err := offer(pid, id, m, sender) if err != nil { log.Printf("anchor: offer to %s: %v", pid.Short(), err) return } mu.Lock() pcs[pid] = pc mu.Unlock() }(pid) } case proto.AnchorPeerLeave: pid := proto.PeerID(msg.ID) mu.Lock() if pc, ok := pcs[pid]; ok { pc.Close() delete(pcs, pid) } mu.Unlock() log.Printf("anchor: peer left: %s", pid.Short()) case proto.AnchorFrom: fromID := proto.PeerID(msg.From) payload, err := openBox(msg.Box, fromID, id) if err != nil { log.Printf("anchor: open box from %s: %v", fromID.Short(), err) continue } dispatchSignaling(ctx, payload, fromID, id, m, sender, pcs, &mu) case proto.AnchorNoPeer: log.Printf("anchor: no such peer: %s", proto.PeerID(msg.ID).Short()) } } } // ── signaling dispatch ──────────────────────────────────────────────────────── func dispatchSignaling( ctx context.Context, payload proto.SignalingPayload, fromID proto.PeerID, id *crypto.Identity, m *mesh.Mesh, s *sender, pcs map[proto.PeerID]*webrtc.PeerConnection, mu *sync.RWMutex, ) { switch payload.Kind { case proto.SigOffer: go func() { pc, err := answer(payload, fromID, id, m, s) if err != nil { log.Printf("anchor: answer to %s: %v", fromID.Short(), err) return } mu.Lock() pcs[fromID] = pc mu.Unlock() }() case proto.SigAnswer: mu.RLock() pc, ok := pcs[fromID] mu.RUnlock() if !ok { log.Printf("anchor: answer from %s but no PeerConnection", fromID.Short()) return } if err := pc.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeAnswer, SDP: payload.SDP, }); err != nil { log.Printf("anchor: set remote answer from %s: %v", fromID.Short(), err) } case proto.SigCandidate: mu.RLock() pc, ok := pcs[fromID] mu.RUnlock() if !ok { return } if err := pc.AddICECandidate(webrtc.ICECandidateInit{ Candidate: payload.Cand, SDPMid: strPtr(payload.Mid), SDPMLineIndex: uint16Ptr(uint16(payload.MLine)), }); err != nil { log.Printf("anchor: add candidate from %s: %v", fromID.Short(), err) } case proto.SigBye: mu.Lock() if pc, ok := pcs[fromID]; ok { pc.Close() delete(pcs, fromID) } mu.Unlock() } } // ── offer / answer helpers ──────────────────────────────────────────────────── func offer(peerID proto.PeerID, id *crypto.Identity, m *mesh.Mesh, s *sender) (*webrtc.PeerConnection, error) { pc, err := newPC() if err != nil { return nil, err } dc, err := pc.CreateDataChannel("yaw", &webrtc.DataChannelInit{Ordered: boolPtr(true)}) if err != nil { pc.Close() return nil, err } mesh.WireDataChannel(dc, pc, peerID, id, m) mesh.WireCandidateTrickle(pc, peerID, s) sdpOffer, err := pc.CreateOffer(nil) if err != nil { pc.Close() return nil, err } if err := pc.SetLocalDescription(sdpOffer); err != nil { pc.Close() return nil, err } return pc, s.SendTo(peerID, proto.SignalingPayload{Kind: proto.SigOffer, SDP: sdpOffer.SDP}) } func answer(payload proto.SignalingPayload, fromID proto.PeerID, id *crypto.Identity, m *mesh.Mesh, s *sender) (*webrtc.PeerConnection, error) { pc, err := newPC() if err != nil { return nil, err } pc.OnDataChannel(func(dc *webrtc.DataChannel) { if dc.Label() == "yaw" { mesh.WireDataChannel(dc, pc, fromID, id, m) } }) mesh.WireCandidateTrickle(pc, fromID, s) if err := pc.SetRemoteDescription(webrtc.SessionDescription{ Type: webrtc.SDPTypeOffer, SDP: payload.SDP, }); err != nil { pc.Close() return nil, err } sdpAnswer, err := pc.CreateAnswer(nil) if err != nil { pc.Close() return nil, err } if err := pc.SetLocalDescription(sdpAnswer); err != nil { pc.Close() return nil, err } return pc, s.SendTo(fromID, proto.SignalingPayload{Kind: proto.SigAnswer, SDP: sdpAnswer.SDP}) } // ── sender implements mesh.Anchor ──────────────────────────────────────────── type sender struct { id *crypto.Identity sendCh chan proto.AnchorMessage } func (s *sender) SendTo(peerID proto.PeerID, payload proto.SignalingPayload) error { plaintext, err := json.Marshal(payload) if err != nil { return err } recipientCurve, err := curveFromPeerID(peerID) if err != nil { return fmt.Errorf("derive curve key for %s: %w", peerID.Short(), err) } sealed := crypto.SignalingBox(plaintext, recipientCurve, s.id.CurvePrivateKey()) select { case s.sendCh <- proto.AnchorMessage{Type: proto.AnchorTo, To: string(peerID), Box: sealed}: return nil default: return fmt.Errorf("send queue full") } } func (s *sender) LocalID() proto.PeerID { return s.id.PeerID() } // ── helpers ─────────────────────────────────────────────────────────────────── func openBox(b64box string, fromID proto.PeerID, localID *crypto.Identity) (proto.SignalingPayload, error) { senderCurve, err := curveFromPeerID(fromID) if err != nil { return proto.SignalingPayload{}, err } plaintext, err := crypto.SignalingOpen(b64box, senderCurve, localID.CurvePrivateKey()) if err != nil { return proto.SignalingPayload{}, err } var p proto.SignalingPayload return p, json.Unmarshal(plaintext, &p) } // curveFromPeerID derives an X25519 public key from a hex Ed25519 peer id // using the Montgomery conversion, identical to crypto.Identity.CurvePublicKey(). func curveFromPeerID(id proto.PeerID) (*[32]byte, error) { pubBytes, err := hex.DecodeString(string(id)) if err != nil || len(pubBytes) != 32 { return nil, fmt.Errorf("invalid peer id %q", id) } edPoint, err := new(edwards25519.Point).SetBytes(pubBytes) if err != nil { return nil, fmt.Errorf("ed25519 point: %w", err) } mont := edPoint.BytesMontgomery() var out [32]byte copy(out[:], mont) return &out, nil } func hashNetName(name string) string { h := sha256.Sum256([]byte("yaw2-net:" + name)) return hex.EncodeToString(h[:]) } func newPC() (*webrtc.PeerConnection, error) { return webrtc.NewPeerConnection(webrtc.Configuration{ ICEServers: []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302"}}}, }) } func boolPtr(b bool) *bool { return &b } func strPtr(s string) *string { return &s } func uint16Ptr(v uint16) *uint16 { return &v }