openvibers/openvibe.link Commits
Core: control link, pairing, and a fake Bot server for tests
Changes
@@@ -1,3 +1,5 @@module github.com/OpenVibers/OpenVibe.Nodego 1.24.7++++require github.com/gorilla/websocket v1.5.3nnew file mode 100644@@@ -0,0 +1,2 @@++github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=++github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=nnew file mode 100644@@@ -0,0 +1,307 @@++// Package fakebot is a stand-in for OpenVibe.Bot used by tests: it redeems pairing codes, accepts device WebSockets++// (credential in the Authorization header only), speaks the ADR-043 message set, and can mount a WHIP endpoint. Tests++// drive it to send commands, e-stops and silence, and read back what the device sent.++package fakebot++++import (++ "crypto/rand"++ "encoding/hex"++ "encoding/json"++ "fmt"++ "net/http"++ "net/http/httptest"++ "strings"++ "sync"++ "sync/atomic"++ "time"++++ "github.com/gorilla/websocket"++++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++// Server is the fake service.++type Server struct {++ HTTP *httptest.Server++++ // Config is sent to every device right after hello.++ Config protocol.Config++ // WHIP, when set, serves POST/DELETE under /whip/.++ WHIP http.Handler++++ mu sync.Mutex++ codes map[string]bool++ creds map[string]string // credential → device id++ pairRequests []protocol.PairRequest++ credInURL bool++++ conns chan *Conn++ count atomic.Int64++}++++// New starts the fake server.++func New() *Server {++ s := &Server{++ codes: map[string]bool{}, creds: map[string]string{}, conns: make(chan *Conn, 16),++ Config: protocol.Config{HeartbeatMS: 1000, Limits: protocol.Limits{MaxCommandMS: 1000}},++ }++ mux := http.NewServeMux()++ mux.HandleFunc(protocol.PairPath, s.pair)++ mux.HandleFunc(protocol.DevicePath, s.device)++ mux.HandleFunc("/whip/", func(w http.ResponseWriter, r *http.Request) {++ s.mu.Lock()++ h := s.WHIP++ s.mu.Unlock()++ if h == nil {++ http.NotFound(w, r)++ return++ }++ h.ServeHTTP(w, r)++ })++ s.HTTP = httptest.NewServer(mux)++ return s++}++++func (s *Server) Close() { s.HTTP.Close() }++++// URL is the origin, e.g. http://127.0.0.1:1234.++func (s *Server) URL() string { return s.HTTP.URL }++++// SetWHIP mounts a WHIP handler.++func (s *Server) SetWHIP(h http.Handler) {++ s.mu.Lock()++ s.WHIP = h++ s.mu.Unlock()++}++++// AddCode makes a one-time pairing code valid.++func (s *Server) AddCode(code string) {++ s.mu.Lock()++ s.codes[code] = true++ s.mu.Unlock()++}++++// AddCredential registers a credential directly (skipping pairing).++func (s *Server) AddCredential(cred, deviceID string) {++ s.mu.Lock()++ s.creds[cred] = deviceID++ s.mu.Unlock()++}++++// Revoke invalidates a credential for future connections.++func (s *Server) Revoke(cred string) {++ s.mu.Lock()++ delete(s.creds, cred)++ s.mu.Unlock()++}++++// PairRequests returns what devices sent to /api/v1/pair.++func (s *Server) PairRequests() []protocol.PairRequest {++ s.mu.Lock()++ defer s.mu.Unlock()++ return append([]protocol.PairRequest(nil), s.pairRequests...)++}++++// CredentialSeenInURL reports whether any request carried the credential in its URL.++func (s *Server) CredentialSeenInURL() bool {++ s.mu.Lock()++ defer s.mu.Unlock()++ return s.credInURL++}++++// Connections counts accepted device connections.++func (s *Server) Connections() int { return int(s.count.Load()) }++++func token() string {++ b := make([]byte, 32)++ _, _ = rand.Read(b)++ return hex.EncodeToString(b)++}++++func (s *Server) pair(w http.ResponseWriter, r *http.Request) {++ if r.Method != http.MethodPost {++ http.Error(w, "method", http.StatusMethodNotAllowed)++ return++ }++ var req protocol.PairRequest++ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {++ w.WriteHeader(http.StatusBadRequest)++ return++ }++ s.mu.Lock()++ s.pairRequests = append(s.pairRequests, req)++ ok := s.codes[req.Code]++ delete(s.codes, req.Code)++ var cred, dev string++ if ok {++ cred, dev = token(), "dev_"+token()[:12]++ s.creds[cred] = dev++ }++ s.mu.Unlock()++ w.Header().Set("Content-Type", "application/json")++ if !ok {++ w.WriteHeader(http.StatusNotFound)++ _ = json.NewEncoder(w).Encode(protocol.PairError{Error: "invalid_code"})++ return++ }++ _ = json.NewEncoder(w).Encode(protocol.PairResponse{++ DeviceID: dev, Credential: cred, PublishKey: "pk_" + token()[:16],++ WHIPURL: s.HTTP.URL + "/whip/" + dev, Profile: json.RawMessage(`{"profile":"sim.rover","capabilities":["drive"]}`),++ })++}++++var upgrader = websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}++++func (s *Server) device(w http.ResponseWriter, r *http.Request) {++ s.mu.Lock()++ for cred := range s.creds {++ if strings.Contains(r.URL.RawQuery, cred) || strings.Contains(r.URL.Path, cred) {++ s.credInURL = true++ }++ }++ auth := r.Header.Get("Authorization")++ dev, ok := s.creds[strings.TrimPrefix(auth, "Bearer ")]++ s.mu.Unlock()++ if !strings.HasPrefix(auth, "Bearer ") || !ok {++ http.Error(w, "unauthorized", http.StatusUnauthorized)++ return++ }++ ws, err := upgrader.Upgrade(w, r, nil)++ if err != nil {++ return++ }++ s.count.Add(1)++ c := &Conn{ws: ws, DeviceID: dev, recv: make(chan protocol.Frame, 1024), done: make(chan struct{})}++ c.autoAck.Store(true)++ _ = c.Send(protocol.Hello{DeviceID: dev, Session: token()[:8], ServerTime: time.Now().UnixMilli()})++ _ = c.Send(s.Config)++ go c.read()++ select {++ case s.conns <- c:++ default:++ }++}++++// NextConn waits for the next device connection.++func (s *Server) NextConn(timeout time.Duration) (*Conn, error) {++ select {++ case c := <-s.conns:++ return c, nil++ case <-time.After(timeout):++ return nil, fmt.Errorf("no device connected within %s", timeout)++ }++}++++// Conn is one device connection as the server sees it.++type Conn struct {++ DeviceID string++ ws *websocket.Conn++ wmu sync.Mutex++ seq uint64++ recv chan protocol.Frame++ done chan struct{}++ autoAck atomic.Bool++ muted atomic.Bool++ frames []protocol.Frame++ fmu sync.Mutex++}++++// Send writes one frame to the device.++func (c *Conn) Send(m protocol.Message) error {++ c.wmu.Lock()++ defer c.wmu.Unlock()++ c.seq++++ b, err := protocol.Encode(c.seq, time.Now().UnixMilli(), m)++ if err != nil {++ return err++ }++ return c.ws.WriteMessage(websocket.TextMessage, b)++}++++// SendRaw writes arbitrary bytes (for malformed-frame tests).++func (c *Conn) SendRaw(b []byte) error {++ c.wmu.Lock()++ defer c.wmu.Unlock()++ return c.ws.WriteMessage(websocket.TextMessage, b)++}++++// Mute stops answering heartbeats and sending anything, as a hung server would.++func (c *Conn) Mute() { c.muted.Store(true); c.autoAck.Store(false) }++++// Close drops the connection.++func (c *Conn) Close() { _ = c.ws.Close() }++++// Done is closed when the connection ends.++func (c *Conn) Done() <-chan struct{} { return c.done }++++func (c *Conn) read() {++ defer close(c.done)++ for {++ _, data, err := c.ws.ReadMessage()++ if err != nil {++ return++ }++ f, err := protocol.Decode(data)++ if err != nil {++ continue++ }++ if hb, ok := f.Msg.(protocol.Heartbeat); ok && c.autoAck.Load() && !c.muted.Load() {++ _ = c.Send(protocol.HeartbeatAck{T: hb.T})++ }++ c.fmu.Lock()++ c.frames = append(c.frames, f)++ c.fmu.Unlock()++ select {++ case c.recv <- f:++ default:++ }++ }++}++++// Frames returns everything received so far.++func (c *Conn) Frames() []protocol.Frame {++ c.fmu.Lock()++ defer c.fmu.Unlock()++ return append([]protocol.Frame(nil), c.frames...)++}++++// Expect waits for the next frame of the given type whose message satisfies match (nil matches all), skipping others.++func (c *Conn) Expect(typ string, timeout time.Duration, match func(protocol.Message) bool) (protocol.Frame, error) {++ t := time.After(timeout)++ for {++ select {++ case f := <-c.recv:++ if f.Type == typ && (match == nil || match(f.Msg)) {++ return f, nil++ }++ case <-t:++ return protocol.Frame{}, fmt.Errorf("no %s frame within %s", typ, timeout)++ case <-c.done:++ return protocol.Frame{}, fmt.Errorf("connection closed while waiting for %s", typ)++ }++ }++}++++// Reply waits for the ack or nack of command id.++func (c *Conn) Reply(id string, timeout time.Duration) (protocol.Message, error) {++ t := time.After(timeout)++ for {++ select {++ case f := <-c.recv:++ switch m := f.Msg.(type) {++ case protocol.Ack:++ if m.ID == id {++ return m, nil++ }++ case protocol.Nack:++ if m.ID == id {++ return m, nil++ }++ }++ case <-t:++ return nil, fmt.Errorf("no reply to %s within %s", id, timeout)++ case <-c.done:++ return nil, fmt.Errorf("connection closed while waiting for the reply to %s", id)++ }++ }++}nnew file mode 100644@@@ -0,0 +1,362 @@++// Package link keeps the Node's one outbound control connection to OpenVibe.Bot: a WebSocket to wss://<origin>/device++// authenticated with the device credential in the Authorization header of the upgrade (never in the URL). It sends a++// heartbeat every interval, declares the link lost when nothing has arrived for the deadman interval (two heartbeats++// by default), and reconnects with exponential backoff and full jitter. Nothing is queued while offline: Send fails++// and the caller decides.++package link++++import (++ "context"++ "errors"++ "fmt"++ "log/slog"++ "math/rand"++ "net/http"++ "net/url"++ "strings"++ "sync"++ "sync/atomic"++ "time"++++ "github.com/gorilla/websocket"++++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++// ErrOffline is returned by Send while there is no connection.++var ErrOffline = errors.New("link offline")++++// Handler receives link events. Calls are made from the link's goroutines, one at a time per connection.++type Handler interface {++ // Connected runs after the upgrade, before any frame is read. The handler sends status and estop_state here.++ Connected()++ // Frame is every decoded server frame except heartbeat_ack.++ Frame(f protocol.Frame)++ // Disconnected runs when a connection ends for any reason. The handler must stop every actuator.++ Disconnected(err error)++}++++// Options configure a Link.++type Options struct {++ URL string // wss://openvibe.bot/device++ Credential credentials.Secret++ UserAgent string++ Log *slog.Logger++ Dialer *websocket.Dialer++++ HeartbeatInterval time.Duration // default 1 s++ Deadman time.Duration // default 2 × heartbeat++ BackoffMin time.Duration // default 500 ms++ BackoffMax time.Duration // default 30 s++}++++// Link is the reconnecting control connection.++type Link struct {++ opt Options++ handler Handler++ log *slog.Logger++++ mu sync.Mutex++ out chan []byte++ seq uint64++ heartbeat time.Duration++ deadman time.Duration++++ connected atomic.Bool++ lastRTT atomic.Int64++ reconnects atomic.Int64++ lastErr atomic.Value // string++ connectedAt atomic.Int64++}++++func New(opt Options, h Handler) *Link {++ if opt.HeartbeatInterval <= 0 {++ opt.HeartbeatInterval = time.Duration(protocol.DefaultHeartbeatMS) * time.Millisecond++ }++ if opt.Deadman <= 0 {++ opt.Deadman = 2 * opt.HeartbeatInterval++ }++ if opt.BackoffMin <= 0 {++ opt.BackoffMin = 500 * time.Millisecond++ }++ if opt.BackoffMax <= 0 {++ opt.BackoffMax = 30 * time.Second++ }++ if opt.Log == nil {++ opt.Log = slog.Default()++ }++ if opt.Dialer == nil {++ opt.Dialer = &websocket.Dialer{Proxy: http.ProxyFromEnvironment, HandshakeTimeout: 15 * time.Second}++ }++ l := &Link{opt: opt, handler: h, log: opt.Log.With("component", "link"), heartbeat: opt.HeartbeatInterval, deadman: opt.Deadman}++ l.lastErr.Store("")++ return l++}++++// SetTiming applies the server's heartbeat_ms / deadman_ms (from config) to the current and later connections.++func (l *Link) SetTiming(heartbeatMS, deadmanMS int) {++ l.mu.Lock()++ defer l.mu.Unlock()++ if heartbeatMS > 0 {++ l.heartbeat = time.Duration(heartbeatMS) * time.Millisecond++ l.deadman = 2 * l.heartbeat++ }++ if deadmanMS > 0 {++ l.deadman = time.Duration(deadmanMS) * time.Millisecond++ }++}++++func (l *Link) timing() (time.Duration, time.Duration) {++ l.mu.Lock()++ defer l.mu.Unlock()++ return l.heartbeat, l.deadman++}++++// Connected reports whether a connection is up.++func (l *Link) Connected() bool { return l.connected.Load() }++++// Stats for status.++type Stats struct {++ Connected bool++ Since time.Time++ RTT time.Duration++ Reconnects int64++ LastError string++}++++func (l *Link) Stats() Stats {++ s := Stats{Connected: l.connected.Load(), RTT: time.Duration(l.lastRTT.Load()), Reconnects: l.reconnects.Load(),++ LastError: l.lastErr.Load().(string)}++ if at := l.connectedAt.Load(); at > 0 && s.Connected {++ s.Since = time.UnixMilli(at)++ }++ return s++}++++// Send queues a frame on the current connection. It never waits for a connection: offline means ErrOffline.++func (l *Link) Send(m protocol.Message) error {++ l.mu.Lock()++ defer l.mu.Unlock()++ if l.out == nil {++ return ErrOffline++ }++ l.seq++++ b, err := protocol.Encode(l.seq, time.Now().UnixMilli(), m)++ if err != nil {++ return err++ }++ select {++ case l.out <- b:++ return nil++ default:++ return errors.New("link send queue full")++ }++}++++// Run connects and reconnects until ctx ends.++func (l *Link) Run(ctx context.Context) {++ attempt := 0++ for ctx.Err() == nil {++ start := time.Now()++ err := l.session(ctx)++ if ctx.Err() != nil {++ return++ }++ msg := errMsg(err)++ l.lastErr.Store(msg)++ if time.Since(start) > 30*time.Second {++ attempt = 0++ }++ max := l.opt.BackoffMin << attempt++ if max > l.opt.BackoffMax || max <= 0 {++ max = l.opt.BackoffMax++ }++ if errors.Is(err, errUnauthorized) && max < 10*time.Second {++ max = 10 * time.Second++ }++ // Full jitter, with a floor so a flapping server is not hammered.++ wait := l.opt.BackoffMin/2 + time.Duration(rand.Int63n(int64(max)))++ attempt++++ l.reconnects.Add(1)++ l.log.Warn("link down; reconnecting", "err", msg, "in", wait.Round(time.Millisecond))++ select {++ case <-ctx.Done():++ return++ case <-time.After(wait):++ }++ }++}++++var errUnauthorized = errors.New("the server refused the device credential (revoked or rotated?); pair again with `openvibe-node pair <CODE>`")++++func errMsg(err error) string {++ if err == nil {++ return "closed"++ }++ return err.Error()++}++++func (l *Link) session(ctx context.Context) error {++ h := http.Header{}++ h.Set("Authorization", "Bearer "+l.opt.Credential.Reveal())++ if l.opt.UserAgent != "" {++ h.Set("User-Agent", l.opt.UserAgent)++ }++ conn, resp, err := l.opt.Dialer.DialContext(ctx, l.opt.URL, h)++ if err != nil {++ if resp != nil && (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) {++ return errUnauthorized++ }++ if resp != nil {++ return fmt.Errorf("connect %s: HTTP %d", redactURL(l.opt.URL), resp.StatusCode)++ }++ return fmt.Errorf("connect %s: %s", redactURL(l.opt.URL), scrub(err.Error(), l.opt.Credential))++ }++ defer conn.Close()++ conn.SetReadLimit(1 << 20)++++ out := make(chan []byte, 256)++ l.mu.Lock()++ l.out, l.seq = out, 0++ l.mu.Unlock()++ l.connected.Store(true)++ l.connectedAt.Store(time.Now().UnixMilli())++ l.log.Info("link up", "url", redactURL(l.opt.URL))++++ sessCtx, cancel := context.WithCancel(ctx)++ var endErr error++ var endOnce sync.Once++ end := func(err error) {++ endOnce.Do(func() {++ endErr = err++ cancel()++ _ = conn.Close()++ })++ }++ defer func() {++ l.mu.Lock()++ l.out = nil++ l.mu.Unlock()++ l.connected.Store(false)++ end(nil)++ l.handler.Disconnected(endErr)++ }()++++ // Ending the session (or the Node) closes the socket, which ends the read loop below.++ go func() {++ <-sessCtx.Done()++ end(ctx.Err())++ }()++++ var lastRx atomic.Int64++ lastRx.Store(time.Now().UnixNano())++++ // Writer.++ go func() {++ for {++ select {++ case <-sessCtx.Done():++ return++ case b := <-out:++ _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))++ if err := conn.WriteMessage(websocket.TextMessage, b); err != nil {++ end(fmt.Errorf("write: %w", err))++ return++ }++ }++ }++ }()++++ // Heartbeat sender and deadman watchdog.++ go func() {++ hb, _ := l.timing()++ tick := time.NewTicker(50 * time.Millisecond)++ defer tick.Stop()++ next := time.Now()++ for {++ select {++ case <-sessCtx.Done():++ return++ case now := <-tick.C:++ hbNow, deadman := l.timing()++ if hbNow != hb {++ hb, next = hbNow, now++ }++ if !now.Before(next) {++ next = now.Add(hb)++ _ = l.Send(protocol.Heartbeat{T: now.UnixMilli()})++ }++ if now.Sub(time.Unix(0, lastRx.Load())) > deadman {++ end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))++ return++ }++ }++ }++ }()++++ l.handler.Connected()++++ for {++ _, data, err := conn.ReadMessage()++ if err != nil {++ if sessCtx.Err() == nil {++ end(fmt.Errorf("read: %s", scrub(err.Error(), l.opt.Credential)))++ }++ return endErr++ }++ lastRx.Store(time.Now().UnixNano())++ f, err := protocol.Decode(data)++ if err != nil {++ if errors.Is(err, protocol.ErrUnknownType) {++ l.log.Debug("ignoring unknown frame", "type", f.Type)++ } else {++ l.log.Warn("bad frame", "err", err)++ }++ continue++ }++ if ack, ok := f.Msg.(protocol.HeartbeatAck); ok {++ if ack.T > 0 {++ l.lastRTT.Store(int64(time.Since(time.UnixMilli(ack.T))))++ }++ continue++ }++ l.handler.Frame(f)++ }++}++++// DeviceURL derives wss://host/device from the server origin.++func DeviceURL(server string) (string, error) {++ u, err := url.Parse(server)++ if err != nil {++ return "", err++ }++ switch u.Scheme {++ case "https":++ u.Scheme = "wss"++ case "http":++ u.Scheme = "ws"++ case "ws", "wss":++ default:++ return "", fmt.Errorf("server %q must be https://", server)++ }++ u.Path = strings.TrimSuffix(u.Path, "/") + protocol.DevicePath++ u.RawQuery, u.Fragment = "", ""++ return u.String(), nil++}++++func redactURL(s string) string {++ u, err := url.Parse(s)++ if err != nil {++ return "(invalid url)"++ }++ u.User, u.RawQuery = nil, ""++ return u.String()++}++++func scrub(s string, secret credentials.Secret) string {++ if v := secret.Reveal(); v != "" {++ s = strings.ReplaceAll(s, v, "[redacted]")++ }++ return s++}nnew file mode 100644@@@ -0,0 +1,262 @@++package link++++import (++ "bytes"++ "context"++ "errors"++ "log/slog"++ "strings"++ "sync"++ "testing"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++type recorder struct {++ mu sync.Mutex++ frames []protocol.Frame++ connected int++ discon []error++ onConnect func()++ gotFrame chan protocol.Frame++ gotDown chan error++}++++func newRecorder() *recorder {++ return &recorder{gotFrame: make(chan protocol.Frame, 100), gotDown: make(chan error, 100)}++}++++func (r *recorder) Connected() {++ r.mu.Lock()++ r.connected++++ f := r.onConnect++ r.mu.Unlock()++ if f != nil {++ f()++ }++}++func (r *recorder) Frame(f protocol.Frame) {++ r.mu.Lock()++ r.frames = append(r.frames, f)++ r.mu.Unlock()++ r.gotFrame <- f++}++func (r *recorder) Disconnected(err error) {++ r.mu.Lock()++ r.discon = append(r.discon, err)++ r.mu.Unlock()++ r.gotDown <- err++}++++func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {++ t.Helper()++ srv := fakebot.New()++ t.Cleanup(srv.Close)++ cred := "cred-0123456789abcdef"++ srv.AddCredential(cred, "dev_test")++ u, err := DeviceURL(srv.URL())++ if err != nil {++ t.Fatal(err)++ }++ logs := &lockedWriter{&bytes.Buffer{}, &sync.Mutex{}}++ rec := newRecorder()++ l := New(Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,++ BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}, rec)++ ctx, cancel := context.WithCancel(context.Background())++ done := make(chan struct{})++ go func() { l.Run(ctx); close(done) }()++ t.Cleanup(func() { cancel(); <-done })++ return srv, l, rec, logs, cancel++}++++type lockedWriter struct {++ b *bytes.Buffer++ mu *sync.Mutex++}++++func (w *lockedWriter) String() string {++ w.mu.Lock()++ defer w.mu.Unlock()++ return w.b.String()++}++++func (w *lockedWriter) Write(p []byte) (int, error) {++ w.mu.Lock()++ defer w.mu.Unlock()++ return w.b.Write(p)++}++++func waitFrame(t *testing.T, r *recorder, typ string) protocol.Frame {++ t.Helper()++ timeout := time.After(3 * time.Second)++ for {++ select {++ case f := <-r.gotFrame:++ if f.Type == typ {++ return f++ }++ case <-timeout:++ t.Fatalf("no %s frame", typ)++ }++ }++}++++func TestConnectHelloConfigHeartbeat(t *testing.T) {++ srv, l, rec, _, _ := setup(t, 50*time.Millisecond)++ c, err := srv.NextConn(3 * time.Second)++ if err != nil {++ t.Fatal(err)++ }++ waitFrame(t, rec, protocol.TypeHello)++ waitFrame(t, rec, protocol.TypeConfig)++ f, err := c.Expect(protocol.TypeHeartbeat, time.Second, nil)++ if err != nil {++ t.Fatal(err)++ }++ if f.V != 1 || f.Seq == 0 || f.TS == 0 {++ t.Fatalf("envelope %+v", f.Envelope)++ }++ if srv.CredentialSeenInURL() {++ t.Fatal("credential in URL")++ }++ // Heartbeats keep the link up well past the deadman.++ time.Sleep(300 * time.Millisecond)++ if !l.Connected() {++ t.Fatal("dropped while heartbeats were acked")++ }++ if l.Stats().RTT <= 0 {++ t.Fatal("no RTT measured")++ }++ // seq increases per direction.++ var last uint64++ for _, f := range c.Frames() {++ if f.Seq <= last {++ t.Fatalf("seq not increasing: %d after %d", f.Seq, last)++ }++ last = f.Seq++ }++}++++func TestDeadmanOnSilentServer(t *testing.T) {++ srv, _, rec, _, _ := setup(t, 50*time.Millisecond)++ c, _ := srv.NextConn(3 * time.Second)++ waitFrame(t, rec, protocol.TypeConfig)++ start := time.Now()++ c.Mute()++ select {++ case err := <-rec.gotDown:++ if err == nil || !strings.Contains(err.Error(), "heartbeat lost") {++ t.Fatalf("err %v", err)++ }++ if d := time.Since(start); d > 400*time.Millisecond {++ t.Fatalf("deadman took %s", d)++ }++ case <-time.After(2 * time.Second):++ t.Fatal("silent server not detected")++ }++ // And it reconnects.++ if _, err := srv.NextConn(3 * time.Second); err != nil {++ t.Fatal(err)++ }++}++++func TestReconnectAfterDrop(t *testing.T) {++ srv, l, rec, _, _ := setup(t, 100*time.Millisecond)++ c, _ := srv.NextConn(3 * time.Second)++ waitFrame(t, rec, protocol.TypeConfig)++ c.Close()++ <-rec.gotDown++ if err := l.Send(protocol.Heartbeat{T: 1}); !errors.Is(err, ErrOffline) && l.Connected() == false {++ t.Fatalf("send while offline: %v", err)++ }++ if _, err := srv.NextConn(3 * time.Second); err != nil {++ t.Fatal(err)++ }++ if srv.Connections() != 2 {++ t.Fatalf("connections %d", srv.Connections())++ }++}++++func TestSendOffline(t *testing.T) {++ l := New(Options{URL: "ws://127.0.0.1:1/device"}, newRecorder())++ if err := l.Send(protocol.Heartbeat{}); !errors.Is(err, ErrOffline) {++ t.Fatalf("%v", err)++ }++}++++func TestUnauthorizedNeverLogsCredential(t *testing.T) {++ srv, l, rec, logs, cancel := setup(t, 100*time.Millisecond)++ c, _ := srv.NextConn(3 * time.Second)++ waitFrame(t, rec, protocol.TypeConfig)++ srv.Revoke("cred-0123456789abcdef")++ c.Close()++ <-rec.gotDown++ deadline := time.Now().Add(3 * time.Second)++ for !strings.Contains(l.Stats().LastError, "pair again") {++ if time.Now().After(deadline) {++ t.Fatalf("last error %q", l.Stats().LastError)++ }++ time.Sleep(10 * time.Millisecond)++ }++ cancel()++ if strings.Contains(logs.String(), "cred-0123456789abcdef") {++ t.Fatalf("credential in logs: %s", logs.String())++ }++}++++func TestDeviceURL(t *testing.T) {++ for in, want := range map[string]string{++ "https://openvibe.bot": "wss://openvibe.bot/device",++ "https://openvibe.bot/": "wss://openvibe.bot/device",++ "http://127.0.0.1:8080": "ws://127.0.0.1:8080/device",++ "https://x.example/base?q=1#f": "wss://x.example/base/device",++ } {++ got, err := DeviceURL(in)++ if err != nil || got != want {++ t.Errorf("%s → %s (%v), want %s", in, got, err, want)++ }++ }++ if _, err := DeviceURL("ftp://x"); err == nil {++ t.Error("ftp accepted")++ }++}++++func TestPair(t *testing.T) {++ srv := fakebot.New()++ defer srv.Close()++ srv.AddCode("ABCD1234")++ req := protocol.PairRequest{Code: "ABCD1234", AgentVersion: "t", DeviceKind: "onboard", Drivers: []string{"dryrun"}}++ c, err := Pair(context.Background(), nil, srv.URL(), req)++ if err != nil {++ t.Fatal(err)++ }++ if c.DeviceID == "" || c.Credential.IsZero() || c.PublishKey.IsZero() || !strings.HasSuffix(c.WHIPURL, c.DeviceID) {++ t.Fatalf("%+v", c)++ }++ if _, err := Pair(context.Background(), nil, srv.URL(), req); err == nil || !strings.Contains(err.Error(), "wrong, used or expired") {++ t.Fatalf("code reused: %v", err)++ }++ if got := srv.PairRequests(); len(got) != 2 || got[0].Drivers[0] != "dryrun" || got[0].DeviceKind != "onboard" {++ t.Fatalf("%+v", got)++ }++ if _, err := Pair(context.Background(), nil, "http://openvibe.example", req); err == nil {++ t.Fatal("plain http to a remote host accepted")++ }++}++++func TestNormalizeCode(t *testing.T) {++ for in, want := range map[string]string{"abcd-1234": "ABCD1234", " WXYZ 7890 ": "WXYZ7890", "OOII-LLOO": "00111100"} {++ got, err := NormalizeCode(in)++ if err != nil || got != want {++ t.Errorf("%q → %q %v", in, got, err)++ }++ }++ for _, bad := range []string{"", "ABC", "ABCD-12345", "ABCU1234"} {++ if _, err := NormalizeCode(bad); err == nil {++ t.Errorf("%q accepted", bad)++ }++ }++}nnew file mode 100644@@@ -0,0 +1,96 @@++package link++++import (++ "bytes"++ "context"++ "encoding/json"++ "errors"++ "fmt"++ "io"++ "net/http"++ "net/url"++ "regexp"++ "strings"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++// Pairing codes are 8 Crockford base32 characters, shown as XXXX-XXXX (ADR-043 decision 2).++var codeRe = regexp.MustCompile(`^[0-9A-HJKMNP-TV-Z]{8}$`)++++// NormalizeCode upper-cases the code, drops the dash and spaces, and maps the Crockford look-alikes (I, L → 1; O → 0).++func NormalizeCode(code string) (string, error) {++ c := strings.ToUpper(strings.NewReplacer("-", "", " ", "").Replace(strings.TrimSpace(code)))++ c = strings.NewReplacer("I", "1", "L", "1", "O", "0").Replace(c)++ if !codeRe.MatchString(c) {++ return "", errors.New("a pairing code is 8 letters and digits, like ABCD-1234")++ }++ return c, nil++}++++// Pair redeems a one-time code at <server>/api/v1/pair.++func Pair(ctx context.Context, client *http.Client, server string, req protocol.PairRequest) (*credentials.Credentials, error) {++ u, err := url.Parse(server)++ if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {++ return nil, fmt.Errorf("server %q is not an https:// origin", server)++ }++ if u.Scheme == "http" && !isLoopback(u.Hostname()) {++ return nil, fmt.Errorf("refusing to pair over plain http with %s", u.Host)++ }++ body, err := json.Marshal(req)++ if err != nil {++ return nil, err++ }++ endpoint := strings.TrimSuffix(server, "/") + protocol.PairPath++ hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))++ if err != nil {++ return nil, err++ }++ hreq.Header.Set("Content-Type", "application/json")++ hreq.Header.Set("User-Agent", "openvibe-node/"+req.AgentVersion)++ if client == nil {++ client = &http.Client{Timeout: 20 * time.Second}++ }++ resp, err := client.Do(hreq)++ if err != nil {++ return nil, fmt.Errorf("pairing request failed: %w", err)++ }++ defer resp.Body.Close()++ raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))++ if err != nil {++ return nil, fmt.Errorf("pairing response: %w", err)++ }++ if resp.StatusCode/100 != 2 {++ var pe protocol.PairError++ _ = json.Unmarshal(raw, &pe)++ switch {++ case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone || pe.Error == "invalid_code" || pe.Error == "expired":++ return nil, errors.New("the pairing code is wrong, used or expired; ask for a new one on openvibe.bot")++ case resp.StatusCode == http.StatusTooManyRequests:++ return nil, errors.New("too many wrong tries for this code; ask for a new one on openvibe.bot")++ }++ msg := pe.Message++ if msg == "" {++ msg = pe.Error++ }++ return nil, fmt.Errorf("pairing refused: HTTP %d %s", resp.StatusCode, msg)++ }++ var pr protocol.PairResponse++ if err := json.Unmarshal(raw, &pr); err != nil {++ return nil, errors.New("pairing response is not valid JSON")++ }++ if pr.DeviceID == "" || pr.Credential == "" {++ return nil, errors.New("pairing response has no device id or credential")++ }++ return &credentials.Credentials{++ DeviceID: pr.DeviceID, Credential: credentials.NewSecret(pr.Credential), PublishKey: credentials.NewSecret(pr.PublishKey),++ Server: strings.TrimSuffix(server, "/"), DeviceURL: pr.DeviceURL, WHIPURL: pr.WHIPURL, ICEServers: pr.ICEServers,++ Profile: pr.Profile, PairedAt: time.Now().UTC(),++ }, nil++}++++func isLoopback(host string) bool {++ return host == "localhost" || host == "127.0.0.1" || host == "::1"++}
@@ -1,3 +1,5 @@
@@ -1,3 +1,5 @@
module github.com/OpenVibers/OpenVibe.Node
module github.com/OpenVibers/OpenVibe.Node
go 1.24.7
go 1.24.7
+
+require github.com/gorilla/websocket v1.5.3
new file mode 100644
new file mode 100644
@@ -0,0 +1,2 @@
@@ -0,0 +1,2 @@
+github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
+github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
new file mode 100644
new file mode 100644
@@ -0,0 +1,307 @@
@@ -0,0 +1,307 @@
+// Package fakebot is a stand-in for OpenVibe.Bot used by tests: it redeems pairing codes, accepts device WebSockets
+// (credential in the Authorization header only), speaks the ADR-043 message set, and can mount a WHIP endpoint. Tests
+// drive it to send commands, e-stops and silence, and read back what the device sent.
+package fakebot
+
+import (
+ "crypto/rand"
+ "encoding/hex"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/gorilla/websocket"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+// Server is the fake service.
+type Server struct {
+ HTTP *httptest.Server
+
+ // Config is sent to every device right after hello.
+ Config protocol.Config
+ // WHIP, when set, serves POST/DELETE under /whip/.
+ WHIP http.Handler
+
+ mu sync.Mutex
+ codes map[string]bool
+ creds map[string]string // credential → device id
+ pairRequests []protocol.PairRequest
+ credInURL bool
+
+ conns chan *Conn
+ count atomic.Int64
+}
+
+// New starts the fake server.
+func New() *Server {
+ s := &Server{
+ codes: map[string]bool{}, creds: map[string]string{}, conns: make(chan *Conn, 16),
+ Config: protocol.Config{HeartbeatMS: 1000, Limits: protocol.Limits{MaxCommandMS: 1000}},
+ }
+ mux := http.NewServeMux()
+ mux.HandleFunc(protocol.PairPath, s.pair)
+ mux.HandleFunc(protocol.DevicePath, s.device)
+ mux.HandleFunc("/whip/", func(w http.ResponseWriter, r *http.Request) {
+ s.mu.Lock()
+ h := s.WHIP
+ s.mu.Unlock()
+ if h == nil {
+ http.NotFound(w, r)
+ return
+ }
+ h.ServeHTTP(w, r)
+ })
+ s.HTTP = httptest.NewServer(mux)
+ return s
+}
+
+func (s *Server) Close() { s.HTTP.Close() }
+
+// URL is the origin, e.g. http://127.0.0.1:1234.
+func (s *Server) URL() string { return s.HTTP.URL }
+
+// SetWHIP mounts a WHIP handler.
+func (s *Server) SetWHIP(h http.Handler) {
+ s.mu.Lock()
+ s.WHIP = h
+ s.mu.Unlock()
+}
+
+// AddCode makes a one-time pairing code valid.
+func (s *Server) AddCode(code string) {
+ s.mu.Lock()
+ s.codes[code] = true
+ s.mu.Unlock()
+}
+
+// AddCredential registers a credential directly (skipping pairing).
+func (s *Server) AddCredential(cred, deviceID string) {
+ s.mu.Lock()
+ s.creds[cred] = deviceID
+ s.mu.Unlock()
+}
+
+// Revoke invalidates a credential for future connections.
+func (s *Server) Revoke(cred string) {
+ s.mu.Lock()
+ delete(s.creds, cred)
+ s.mu.Unlock()
+}
+
+// PairRequests returns what devices sent to /api/v1/pair.
+func (s *Server) PairRequests() []protocol.PairRequest {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return append([]protocol.PairRequest(nil), s.pairRequests...)
+}
+
+// CredentialSeenInURL reports whether any request carried the credential in its URL.
+func (s *Server) CredentialSeenInURL() bool {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.credInURL
+}
+
+// Connections counts accepted device connections.
+func (s *Server) Connections() int { return int(s.count.Load()) }
+
+func token() string {
+ b := make([]byte, 32)
+ _, _ = rand.Read(b)
+ return hex.EncodeToString(b)
+}
+
+func (s *Server) pair(w http.ResponseWriter, r *http.Request) {
+ if r.Method != http.MethodPost {
+ http.Error(w, "method", http.StatusMethodNotAllowed)
+ return
+ }
+ var req protocol.PairRequest
+ if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+ w.WriteHeader(http.StatusBadRequest)
+ return
+ }
+ s.mu.Lock()
+ s.pairRequests = append(s.pairRequests, req)
+ ok := s.codes[req.Code]
+ delete(s.codes, req.Code)
+ var cred, dev string
+ if ok {
+ cred, dev = token(), "dev_"+token()[:12]
+ s.creds[cred] = dev
+ }
+ s.mu.Unlock()
+ w.Header().Set("Content-Type", "application/json")
+ if !ok {
+ w.WriteHeader(http.StatusNotFound)
+ _ = json.NewEncoder(w).Encode(protocol.PairError{Error: "invalid_code"})
+ return
+ }
+ _ = json.NewEncoder(w).Encode(protocol.PairResponse{
+ DeviceID: dev, Credential: cred, PublishKey: "pk_" + token()[:16],
+ WHIPURL: s.HTTP.URL + "/whip/" + dev, Profile: json.RawMessage(`{"profile":"sim.rover","capabilities":["drive"]}`),
+ })
+}
+
+var upgrader = websocket.Upgrader{CheckOrigin: func(*http.Request) bool { return true }}
+
+func (s *Server) device(w http.ResponseWriter, r *http.Request) {
+ s.mu.Lock()
+ for cred := range s.creds {
+ if strings.Contains(r.URL.RawQuery, cred) || strings.Contains(r.URL.Path, cred) {
+ s.credInURL = true
+ }
+ }
+ auth := r.Header.Get("Authorization")
+ dev, ok := s.creds[strings.TrimPrefix(auth, "Bearer ")]
+ s.mu.Unlock()
+ if !strings.HasPrefix(auth, "Bearer ") || !ok {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ ws, err := upgrader.Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+ s.count.Add(1)
+ c := &Conn{ws: ws, DeviceID: dev, recv: make(chan protocol.Frame, 1024), done: make(chan struct{})}
+ c.autoAck.Store(true)
+ _ = c.Send(protocol.Hello{DeviceID: dev, Session: token()[:8], ServerTime: time.Now().UnixMilli()})
+ _ = c.Send(s.Config)
+ go c.read()
+ select {
+ case s.conns <- c:
+ default:
+ }
+}
+
+// NextConn waits for the next device connection.
+func (s *Server) NextConn(timeout time.Duration) (*Conn, error) {
+ select {
+ case c := <-s.conns:
+ return c, nil
+ case <-time.After(timeout):
+ return nil, fmt.Errorf("no device connected within %s", timeout)
+ }
+}
+
+// Conn is one device connection as the server sees it.
+type Conn struct {
+ DeviceID string
+ ws *websocket.Conn
+ wmu sync.Mutex
+ seq uint64
+ recv chan protocol.Frame
+ done chan struct{}
+ autoAck atomic.Bool
+ muted atomic.Bool
+ frames []protocol.Frame
+ fmu sync.Mutex
+}
+
+// Send writes one frame to the device.
+func (c *Conn) Send(m protocol.Message) error {
+ c.wmu.Lock()
+ defer c.wmu.Unlock()
+ c.seq++
+ b, err := protocol.Encode(c.seq, time.Now().UnixMilli(), m)
+ if err != nil {
+ return err
+ }
+ return c.ws.WriteMessage(websocket.TextMessage, b)
+}
+
+// SendRaw writes arbitrary bytes (for malformed-frame tests).
+func (c *Conn) SendRaw(b []byte) error {
+ c.wmu.Lock()
+ defer c.wmu.Unlock()
+ return c.ws.WriteMessage(websocket.TextMessage, b)
+}
+
+// Mute stops answering heartbeats and sending anything, as a hung server would.
+func (c *Conn) Mute() { c.muted.Store(true); c.autoAck.Store(false) }
+
+// Close drops the connection.
+func (c *Conn) Close() { _ = c.ws.Close() }
+
+// Done is closed when the connection ends.
+func (c *Conn) Done() <-chan struct{} { return c.done }
+
+func (c *Conn) read() {
+ defer close(c.done)
+ for {
+ _, data, err := c.ws.ReadMessage()
+ if err != nil {
+ return
+ }
+ f, err := protocol.Decode(data)
+ if err != nil {
+ continue
+ }
+ if hb, ok := f.Msg.(protocol.Heartbeat); ok && c.autoAck.Load() && !c.muted.Load() {
+ _ = c.Send(protocol.HeartbeatAck{T: hb.T})
+ }
+ c.fmu.Lock()
+ c.frames = append(c.frames, f)
+ c.fmu.Unlock()
+ select {
+ case c.recv <- f:
+ default:
+ }
+ }
+}
+
+// Frames returns everything received so far.
+func (c *Conn) Frames() []protocol.Frame {
+ c.fmu.Lock()
+ defer c.fmu.Unlock()
+ return append([]protocol.Frame(nil), c.frames...)
+}
+
+// Expect waits for the next frame of the given type whose message satisfies match (nil matches all), skipping others.
+func (c *Conn) Expect(typ string, timeout time.Duration, match func(protocol.Message) bool) (protocol.Frame, error) {
+ t := time.After(timeout)
+ for {
+ select {
+ case f := <-c.recv:
+ if f.Type == typ && (match == nil || match(f.Msg)) {
+ return f, nil
+ }
+ case <-t:
+ return protocol.Frame{}, fmt.Errorf("no %s frame within %s", typ, timeout)
+ case <-c.done:
+ return protocol.Frame{}, fmt.Errorf("connection closed while waiting for %s", typ)
+ }
+ }
+}
+
+// Reply waits for the ack or nack of command id.
+func (c *Conn) Reply(id string, timeout time.Duration) (protocol.Message, error) {
+ t := time.After(timeout)
+ for {
+ select {
+ case f := <-c.recv:
+ switch m := f.Msg.(type) {
+ case protocol.Ack:
+ if m.ID == id {
+ return m, nil
+ }
+ case protocol.Nack:
+ if m.ID == id {
+ return m, nil
+ }
+ }
+ case <-t:
+ return nil, fmt.Errorf("no reply to %s within %s", id, timeout)
+ case <-c.done:
+ return nil, fmt.Errorf("connection closed while waiting for the reply to %s", id)
+ }
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,362 @@
@@ -0,0 +1,362 @@
+// Package link keeps the Node's one outbound control connection to OpenVibe.Bot: a WebSocket to wss://<origin>/device
+// authenticated with the device credential in the Authorization header of the upgrade (never in the URL). It sends a
+// heartbeat every interval, declares the link lost when nothing has arrived for the deadman interval (two heartbeats
+// by default), and reconnects with exponential backoff and full jitter. Nothing is queued while offline: Send fails
+// and the caller decides.
+package link
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log/slog"
+ "math/rand"
+ "net/http"
+ "net/url"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/gorilla/websocket"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+// ErrOffline is returned by Send while there is no connection.
+var ErrOffline = errors.New("link offline")
+
+// Handler receives link events. Calls are made from the link's goroutines, one at a time per connection.
+type Handler interface {
+ // Connected runs after the upgrade, before any frame is read. The handler sends status and estop_state here.
+ Connected()
+ // Frame is every decoded server frame except heartbeat_ack.
+ Frame(f protocol.Frame)
+ // Disconnected runs when a connection ends for any reason. The handler must stop every actuator.
+ Disconnected(err error)
+}
+
+// Options configure a Link.
+type Options struct {
+ URL string // wss://openvibe.bot/device
+ Credential credentials.Secret
+ UserAgent string
+ Log *slog.Logger
+ Dialer *websocket.Dialer
+
+ HeartbeatInterval time.Duration // default 1 s
+ Deadman time.Duration // default 2 × heartbeat
+ BackoffMin time.Duration // default 500 ms
+ BackoffMax time.Duration // default 30 s
+}
+
+// Link is the reconnecting control connection.
+type Link struct {
+ opt Options
+ handler Handler
+ log *slog.Logger
+
+ mu sync.Mutex
+ out chan []byte
+ seq uint64
+ heartbeat time.Duration
+ deadman time.Duration
+
+ connected atomic.Bool
+ lastRTT atomic.Int64
+ reconnects atomic.Int64
+ lastErr atomic.Value // string
+ connectedAt atomic.Int64
+}
+
+func New(opt Options, h Handler) *Link {
+ if opt.HeartbeatInterval <= 0 {
+ opt.HeartbeatInterval = time.Duration(protocol.DefaultHeartbeatMS) * time.Millisecond
+ }
+ if opt.Deadman <= 0 {
+ opt.Deadman = 2 * opt.HeartbeatInterval
+ }
+ if opt.BackoffMin <= 0 {
+ opt.BackoffMin = 500 * time.Millisecond
+ }
+ if opt.BackoffMax <= 0 {
+ opt.BackoffMax = 30 * time.Second
+ }
+ if opt.Log == nil {
+ opt.Log = slog.Default()
+ }
+ if opt.Dialer == nil {
+ opt.Dialer = &websocket.Dialer{Proxy: http.ProxyFromEnvironment, HandshakeTimeout: 15 * time.Second}
+ }
+ l := &Link{opt: opt, handler: h, log: opt.Log.With("component", "link"), heartbeat: opt.HeartbeatInterval, deadman: opt.Deadman}
+ l.lastErr.Store("")
+ return l
+}
+
+// SetTiming applies the server's heartbeat_ms / deadman_ms (from config) to the current and later connections.
+func (l *Link) SetTiming(heartbeatMS, deadmanMS int) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ if heartbeatMS > 0 {
+ l.heartbeat = time.Duration(heartbeatMS) * time.Millisecond
+ l.deadman = 2 * l.heartbeat
+ }
+ if deadmanMS > 0 {
+ l.deadman = time.Duration(deadmanMS) * time.Millisecond
+ }
+}
+
+func (l *Link) timing() (time.Duration, time.Duration) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ return l.heartbeat, l.deadman
+}
+
+// Connected reports whether a connection is up.
+func (l *Link) Connected() bool { return l.connected.Load() }
+
+// Stats for status.
+type Stats struct {
+ Connected bool
+ Since time.Time
+ RTT time.Duration
+ Reconnects int64
+ LastError string
+}
+
+func (l *Link) Stats() Stats {
+ s := Stats{Connected: l.connected.Load(), RTT: time.Duration(l.lastRTT.Load()), Reconnects: l.reconnects.Load(),
+ LastError: l.lastErr.Load().(string)}
+ if at := l.connectedAt.Load(); at > 0 && s.Connected {
+ s.Since = time.UnixMilli(at)
+ }
+ return s
+}
+
+// Send queues a frame on the current connection. It never waits for a connection: offline means ErrOffline.
+func (l *Link) Send(m protocol.Message) error {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ if l.out == nil {
+ return ErrOffline
+ }
+ l.seq++
+ b, err := protocol.Encode(l.seq, time.Now().UnixMilli(), m)
+ if err != nil {
+ return err
+ }
+ select {
+ case l.out <- b:
+ return nil
+ default:
+ return errors.New("link send queue full")
+ }
+}
+
+// Run connects and reconnects until ctx ends.
+func (l *Link) Run(ctx context.Context) {
+ attempt := 0
+ for ctx.Err() == nil {
+ start := time.Now()
+ err := l.session(ctx)
+ if ctx.Err() != nil {
+ return
+ }
+ msg := errMsg(err)
+ l.lastErr.Store(msg)
+ if time.Since(start) > 30*time.Second {
+ attempt = 0
+ }
+ max := l.opt.BackoffMin << attempt
+ if max > l.opt.BackoffMax || max <= 0 {
+ max = l.opt.BackoffMax
+ }
+ if errors.Is(err, errUnauthorized) && max < 10*time.Second {
+ max = 10 * time.Second
+ }
+ // Full jitter, with a floor so a flapping server is not hammered.
+ wait := l.opt.BackoffMin/2 + time.Duration(rand.Int63n(int64(max)))
+ attempt++
+ l.reconnects.Add(1)
+ l.log.Warn("link down; reconnecting", "err", msg, "in", wait.Round(time.Millisecond))
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(wait):
+ }
+ }
+}
+
+var errUnauthorized = errors.New("the server refused the device credential (revoked or rotated?); pair again with `openvibe-node pair <CODE>`")
+
+func errMsg(err error) string {
+ if err == nil {
+ return "closed"
+ }
+ return err.Error()
+}
+
+func (l *Link) session(ctx context.Context) error {
+ h := http.Header{}
+ h.Set("Authorization", "Bearer "+l.opt.Credential.Reveal())
+ if l.opt.UserAgent != "" {
+ h.Set("User-Agent", l.opt.UserAgent)
+ }
+ conn, resp, err := l.opt.Dialer.DialContext(ctx, l.opt.URL, h)
+ if err != nil {
+ if resp != nil && (resp.StatusCode == http.StatusUnauthorized || resp.StatusCode == http.StatusForbidden) {
+ return errUnauthorized
+ }
+ if resp != nil {
+ return fmt.Errorf("connect %s: HTTP %d", redactURL(l.opt.URL), resp.StatusCode)
+ }
+ return fmt.Errorf("connect %s: %s", redactURL(l.opt.URL), scrub(err.Error(), l.opt.Credential))
+ }
+ defer conn.Close()
+ conn.SetReadLimit(1 << 20)
+
+ out := make(chan []byte, 256)
+ l.mu.Lock()
+ l.out, l.seq = out, 0
+ l.mu.Unlock()
+ l.connected.Store(true)
+ l.connectedAt.Store(time.Now().UnixMilli())
+ l.log.Info("link up", "url", redactURL(l.opt.URL))
+
+ sessCtx, cancel := context.WithCancel(ctx)
+ var endErr error
+ var endOnce sync.Once
+ end := func(err error) {
+ endOnce.Do(func() {
+ endErr = err
+ cancel()
+ _ = conn.Close()
+ })
+ }
+ defer func() {
+ l.mu.Lock()
+ l.out = nil
+ l.mu.Unlock()
+ l.connected.Store(false)
+ end(nil)
+ l.handler.Disconnected(endErr)
+ }()
+
+ // Ending the session (or the Node) closes the socket, which ends the read loop below.
+ go func() {
+ <-sessCtx.Done()
+ end(ctx.Err())
+ }()
+
+ var lastRx atomic.Int64
+ lastRx.Store(time.Now().UnixNano())
+
+ // Writer.
+ go func() {
+ for {
+ select {
+ case <-sessCtx.Done():
+ return
+ case b := <-out:
+ _ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
+ if err := conn.WriteMessage(websocket.TextMessage, b); err != nil {
+ end(fmt.Errorf("write: %w", err))
+ return
+ }
+ }
+ }
+ }()
+
+ // Heartbeat sender and deadman watchdog.
+ go func() {
+ hb, _ := l.timing()
+ tick := time.NewTicker(50 * time.Millisecond)
+ defer tick.Stop()
+ next := time.Now()
+ for {
+ select {
+ case <-sessCtx.Done():
+ return
+ case now := <-tick.C:
+ hbNow, deadman := l.timing()
+ if hbNow != hb {
+ hb, next = hbNow, now
+ }
+ if !now.Before(next) {
+ next = now.Add(hb)
+ _ = l.Send(protocol.Heartbeat{T: now.UnixMilli()})
+ }
+ if now.Sub(time.Unix(0, lastRx.Load())) > deadman {
+ end(fmt.Errorf("heartbeat lost: nothing from the server for %s", deadman))
+ return
+ }
+ }
+ }
+ }()
+
+ l.handler.Connected()
+
+ for {
+ _, data, err := conn.ReadMessage()
+ if err != nil {
+ if sessCtx.Err() == nil {
+ end(fmt.Errorf("read: %s", scrub(err.Error(), l.opt.Credential)))
+ }
+ return endErr
+ }
+ lastRx.Store(time.Now().UnixNano())
+ f, err := protocol.Decode(data)
+ if err != nil {
+ if errors.Is(err, protocol.ErrUnknownType) {
+ l.log.Debug("ignoring unknown frame", "type", f.Type)
+ } else {
+ l.log.Warn("bad frame", "err", err)
+ }
+ continue
+ }
+ if ack, ok := f.Msg.(protocol.HeartbeatAck); ok {
+ if ack.T > 0 {
+ l.lastRTT.Store(int64(time.Since(time.UnixMilli(ack.T))))
+ }
+ continue
+ }
+ l.handler.Frame(f)
+ }
+}
+
+// DeviceURL derives wss://host/device from the server origin.
+func DeviceURL(server string) (string, error) {
+ u, err := url.Parse(server)
+ if err != nil {
+ return "", err
+ }
+ switch u.Scheme {
+ case "https":
+ u.Scheme = "wss"
+ case "http":
+ u.Scheme = "ws"
+ case "ws", "wss":
+ default:
+ return "", fmt.Errorf("server %q must be https://", server)
+ }
+ u.Path = strings.TrimSuffix(u.Path, "/") + protocol.DevicePath
+ u.RawQuery, u.Fragment = "", ""
+ return u.String(), nil
+}
+
+func redactURL(s string) string {
+ u, err := url.Parse(s)
+ if err != nil {
+ return "(invalid url)"
+ }
+ u.User, u.RawQuery = nil, ""
+ return u.String()
+}
+
+func scrub(s string, secret credentials.Secret) string {
+ if v := secret.Reveal(); v != "" {
+ s = strings.ReplaceAll(s, v, "[redacted]")
+ }
+ return s
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,262 @@
@@ -0,0 +1,262 @@
+package link
+
+import (
+ "bytes"
+ "context"
+ "errors"
+ "log/slog"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/fakebot"
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+type recorder struct {
+ mu sync.Mutex
+ frames []protocol.Frame
+ connected int
+ discon []error
+ onConnect func()
+ gotFrame chan protocol.Frame
+ gotDown chan error
+}
+
+func newRecorder() *recorder {
+ return &recorder{gotFrame: make(chan protocol.Frame, 100), gotDown: make(chan error, 100)}
+}
+
+func (r *recorder) Connected() {
+ r.mu.Lock()
+ r.connected++
+ f := r.onConnect
+ r.mu.Unlock()
+ if f != nil {
+ f()
+ }
+}
+func (r *recorder) Frame(f protocol.Frame) {
+ r.mu.Lock()
+ r.frames = append(r.frames, f)
+ r.mu.Unlock()
+ r.gotFrame <- f
+}
+func (r *recorder) Disconnected(err error) {
+ r.mu.Lock()
+ r.discon = append(r.discon, err)
+ r.mu.Unlock()
+ r.gotDown <- err
+}
+
+func setup(t *testing.T, hb time.Duration) (*fakebot.Server, *Link, *recorder, *lockedWriter, context.CancelFunc) {
+ t.Helper()
+ srv := fakebot.New()
+ t.Cleanup(srv.Close)
+ cred := "cred-0123456789abcdef"
+ srv.AddCredential(cred, "dev_test")
+ u, err := DeviceURL(srv.URL())
+ if err != nil {
+ t.Fatal(err)
+ }
+ logs := &lockedWriter{&bytes.Buffer{}, &sync.Mutex{}}
+ rec := newRecorder()
+ l := New(Options{URL: u, Credential: credentials.NewSecret(cred), HeartbeatInterval: hb, BackoffMin: 20 * time.Millisecond,
+ BackoffMax: 100 * time.Millisecond, Log: slog.New(slog.NewTextHandler(logs, nil))}, rec)
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() { l.Run(ctx); close(done) }()
+ t.Cleanup(func() { cancel(); <-done })
+ return srv, l, rec, logs, cancel
+}
+
+type lockedWriter struct {
+ b *bytes.Buffer
+ mu *sync.Mutex
+}
+
+func (w *lockedWriter) String() string {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.b.String()
+}
+
+func (w *lockedWriter) Write(p []byte) (int, error) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.b.Write(p)
+}
+
+func waitFrame(t *testing.T, r *recorder, typ string) protocol.Frame {
+ t.Helper()
+ timeout := time.After(3 * time.Second)
+ for {
+ select {
+ case f := <-r.gotFrame:
+ if f.Type == typ {
+ return f
+ }
+ case <-timeout:
+ t.Fatalf("no %s frame", typ)
+ }
+ }
+}
+
+func TestConnectHelloConfigHeartbeat(t *testing.T) {
+ srv, l, rec, _, _ := setup(t, 50*time.Millisecond)
+ c, err := srv.NextConn(3 * time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitFrame(t, rec, protocol.TypeHello)
+ waitFrame(t, rec, protocol.TypeConfig)
+ f, err := c.Expect(protocol.TypeHeartbeat, time.Second, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if f.V != 1 || f.Seq == 0 || f.TS == 0 {
+ t.Fatalf("envelope %+v", f.Envelope)
+ }
+ if srv.CredentialSeenInURL() {
+ t.Fatal("credential in URL")
+ }
+ // Heartbeats keep the link up well past the deadman.
+ time.Sleep(300 * time.Millisecond)
+ if !l.Connected() {
+ t.Fatal("dropped while heartbeats were acked")
+ }
+ if l.Stats().RTT <= 0 {
+ t.Fatal("no RTT measured")
+ }
+ // seq increases per direction.
+ var last uint64
+ for _, f := range c.Frames() {
+ if f.Seq <= last {
+ t.Fatalf("seq not increasing: %d after %d", f.Seq, last)
+ }
+ last = f.Seq
+ }
+}
+
+func TestDeadmanOnSilentServer(t *testing.T) {
+ srv, _, rec, _, _ := setup(t, 50*time.Millisecond)
+ c, _ := srv.NextConn(3 * time.Second)
+ waitFrame(t, rec, protocol.TypeConfig)
+ start := time.Now()
+ c.Mute()
+ select {
+ case err := <-rec.gotDown:
+ if err == nil || !strings.Contains(err.Error(), "heartbeat lost") {
+ t.Fatalf("err %v", err)
+ }
+ if d := time.Since(start); d > 400*time.Millisecond {
+ t.Fatalf("deadman took %s", d)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("silent server not detected")
+ }
+ // And it reconnects.
+ if _, err := srv.NextConn(3 * time.Second); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func TestReconnectAfterDrop(t *testing.T) {
+ srv, l, rec, _, _ := setup(t, 100*time.Millisecond)
+ c, _ := srv.NextConn(3 * time.Second)
+ waitFrame(t, rec, protocol.TypeConfig)
+ c.Close()
+ <-rec.gotDown
+ if err := l.Send(protocol.Heartbeat{T: 1}); !errors.Is(err, ErrOffline) && l.Connected() == false {
+ t.Fatalf("send while offline: %v", err)
+ }
+ if _, err := srv.NextConn(3 * time.Second); err != nil {
+ t.Fatal(err)
+ }
+ if srv.Connections() != 2 {
+ t.Fatalf("connections %d", srv.Connections())
+ }
+}
+
+func TestSendOffline(t *testing.T) {
+ l := New(Options{URL: "ws://127.0.0.1:1/device"}, newRecorder())
+ if err := l.Send(protocol.Heartbeat{}); !errors.Is(err, ErrOffline) {
+ t.Fatalf("%v", err)
+ }
+}
+
+func TestUnauthorizedNeverLogsCredential(t *testing.T) {
+ srv, l, rec, logs, cancel := setup(t, 100*time.Millisecond)
+ c, _ := srv.NextConn(3 * time.Second)
+ waitFrame(t, rec, protocol.TypeConfig)
+ srv.Revoke("cred-0123456789abcdef")
+ c.Close()
+ <-rec.gotDown
+ deadline := time.Now().Add(3 * time.Second)
+ for !strings.Contains(l.Stats().LastError, "pair again") {
+ if time.Now().After(deadline) {
+ t.Fatalf("last error %q", l.Stats().LastError)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ cancel()
+ if strings.Contains(logs.String(), "cred-0123456789abcdef") {
+ t.Fatalf("credential in logs: %s", logs.String())
+ }
+}
+
+func TestDeviceURL(t *testing.T) {
+ for in, want := range map[string]string{
+ "https://openvibe.bot": "wss://openvibe.bot/device",
+ "https://openvibe.bot/": "wss://openvibe.bot/device",
+ "http://127.0.0.1:8080": "ws://127.0.0.1:8080/device",
+ "https://x.example/base?q=1#f": "wss://x.example/base/device",
+ } {
+ got, err := DeviceURL(in)
+ if err != nil || got != want {
+ t.Errorf("%s → %s (%v), want %s", in, got, err, want)
+ }
+ }
+ if _, err := DeviceURL("ftp://x"); err == nil {
+ t.Error("ftp accepted")
+ }
+}
+
+func TestPair(t *testing.T) {
+ srv := fakebot.New()
+ defer srv.Close()
+ srv.AddCode("ABCD1234")
+ req := protocol.PairRequest{Code: "ABCD1234", AgentVersion: "t", DeviceKind: "onboard", Drivers: []string{"dryrun"}}
+ c, err := Pair(context.Background(), nil, srv.URL(), req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if c.DeviceID == "" || c.Credential.IsZero() || c.PublishKey.IsZero() || !strings.HasSuffix(c.WHIPURL, c.DeviceID) {
+ t.Fatalf("%+v", c)
+ }
+ if _, err := Pair(context.Background(), nil, srv.URL(), req); err == nil || !strings.Contains(err.Error(), "wrong, used or expired") {
+ t.Fatalf("code reused: %v", err)
+ }
+ if got := srv.PairRequests(); len(got) != 2 || got[0].Drivers[0] != "dryrun" || got[0].DeviceKind != "onboard" {
+ t.Fatalf("%+v", got)
+ }
+ if _, err := Pair(context.Background(), nil, "http://openvibe.example", req); err == nil {
+ t.Fatal("plain http to a remote host accepted")
+ }
+}
+
+func TestNormalizeCode(t *testing.T) {
+ for in, want := range map[string]string{"abcd-1234": "ABCD1234", " WXYZ 7890 ": "WXYZ7890", "OOII-LLOO": "00111100"} {
+ got, err := NormalizeCode(in)
+ if err != nil || got != want {
+ t.Errorf("%q → %q %v", in, got, err)
+ }
+ }
+ for _, bad := range []string{"", "ABC", "ABCD-12345", "ABCU1234"} {
+ if _, err := NormalizeCode(bad); err == nil {
+ t.Errorf("%q accepted", bad)
+ }
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,96 @@
@@ -0,0 +1,96 @@
+package link
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "regexp"
+ "strings"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+// Pairing codes are 8 Crockford base32 characters, shown as XXXX-XXXX (ADR-043 decision 2).
+var codeRe = regexp.MustCompile(`^[0-9A-HJKMNP-TV-Z]{8}$`)
+
+// NormalizeCode upper-cases the code, drops the dash and spaces, and maps the Crockford look-alikes (I, L → 1; O → 0).
+func NormalizeCode(code string) (string, error) {
+ c := strings.ToUpper(strings.NewReplacer("-", "", " ", "").Replace(strings.TrimSpace(code)))
+ c = strings.NewReplacer("I", "1", "L", "1", "O", "0").Replace(c)
+ if !codeRe.MatchString(c) {
+ return "", errors.New("a pairing code is 8 letters and digits, like ABCD-1234")
+ }
+ return c, nil
+}
+
+// Pair redeems a one-time code at <server>/api/v1/pair.
+func Pair(ctx context.Context, client *http.Client, server string, req protocol.PairRequest) (*credentials.Credentials, error) {
+ u, err := url.Parse(server)
+ if err != nil || (u.Scheme != "https" && u.Scheme != "http") || u.Host == "" {
+ return nil, fmt.Errorf("server %q is not an https:// origin", server)
+ }
+ if u.Scheme == "http" && !isLoopback(u.Hostname()) {
+ return nil, fmt.Errorf("refusing to pair over plain http with %s", u.Host)
+ }
+ body, err := json.Marshal(req)
+ if err != nil {
+ return nil, err
+ }
+ endpoint := strings.TrimSuffix(server, "/") + protocol.PairPath
+ hreq, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewReader(body))
+ if err != nil {
+ return nil, err
+ }
+ hreq.Header.Set("Content-Type", "application/json")
+ hreq.Header.Set("User-Agent", "openvibe-node/"+req.AgentVersion)
+ if client == nil {
+ client = &http.Client{Timeout: 20 * time.Second}
+ }
+ resp, err := client.Do(hreq)
+ if err != nil {
+ return nil, fmt.Errorf("pairing request failed: %w", err)
+ }
+ defer resp.Body.Close()
+ raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
+ if err != nil {
+ return nil, fmt.Errorf("pairing response: %w", err)
+ }
+ if resp.StatusCode/100 != 2 {
+ var pe protocol.PairError
+ _ = json.Unmarshal(raw, &pe)
+ switch {
+ case resp.StatusCode == http.StatusNotFound || resp.StatusCode == http.StatusGone || pe.Error == "invalid_code" || pe.Error == "expired":
+ return nil, errors.New("the pairing code is wrong, used or expired; ask for a new one on openvibe.bot")
+ case resp.StatusCode == http.StatusTooManyRequests:
+ return nil, errors.New("too many wrong tries for this code; ask for a new one on openvibe.bot")
+ }
+ msg := pe.Message
+ if msg == "" {
+ msg = pe.Error
+ }
+ return nil, fmt.Errorf("pairing refused: HTTP %d %s", resp.StatusCode, msg)
+ }
+ var pr protocol.PairResponse
+ if err := json.Unmarshal(raw, &pr); err != nil {
+ return nil, errors.New("pairing response is not valid JSON")
+ }
+ if pr.DeviceID == "" || pr.Credential == "" {
+ return nil, errors.New("pairing response has no device id or credential")
+ }
+ return &credentials.Credentials{
+ DeviceID: pr.DeviceID, Credential: credentials.NewSecret(pr.Credential), PublishKey: credentials.NewSecret(pr.PublishKey),
+ Server: strings.TrimSuffix(server, "/"), DeviceURL: pr.DeviceURL, WHIPURL: pr.WHIPURL, ICEServers: pr.ICEServers,
+ Profile: pr.Profile, PairedAt: time.Now().UTC(),
+ }, nil
+}
+
+func isLoopback(host string) bool {
+ return host == "localhost" || host == "127.0.0.1" || host == "::1"
+}