openvibers/openvibe.link Commits
v0.38.0: a share this computer serves with its own sign-in, no key (CON-4) (#100)
Changes
@@@ -5,7 +5,7 @@"wave": "T15","plan": "Plan track T15 (Bot complete) and ADR-043 (Bot: devices, pairing, control and safety); the computer plugin and the job worker serve OpenVibe.Actor and the Codes job runner (tracks T14/T17)","kind": "client",-- "version": "0.37.2",++ "version": "0.38.0","release": "v0.36.0: the ai plugin (0.2.0) passes tools to the machine's own model server and hands back the calls the model asked for, never running one, so an agent on OpenVibe (Actor) can work on a model this machine serves; its capabilities say tools: true (contracts 0.277.0). v0.35.0: the Node applies the roles the network's machines hub gives it at each check-in (codes.worker starts the job worker it was installed with, desks.host and ai.host are checked against what the machine was installed for, an unknown role is ignored, and the local configuration always wins), and its status reports them back; a desk's first browser start in a cold sandbox gets 90 s. v0.34.0: on Windows a killed plugin takes every process it started with it (a Job Object with kill-on-close), as the process group does on Linux and macOS, so a bridge helper cannot keep a robot's link open after the Node stops or restarts its plugin; the Node reads what the coding tools remember (read-only, local, redacted) and `openvibe-node memory` uploads it to your own store; the heartbeat carries the machine's fleet of coding agents; the Codes job runner pin is v0.5.2","runtime": "go 1.24 (one static binary per platform); plugins in Python 3; an ESP32 Arduino library","database": "none",@@@ -140,6 +140,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) intshKey = fs.String("key", "", "the share's ovc_… connect key from connect.openvibe.website (share)")shPort = fs.Int("port", 0, "the local port to share (share)")shSrv = fs.String("connect-server", "", "OpenVibe Connect's tunnel URL; default wss://connect.openvibe.website/tunnel (share)")++ shID = fs.String("share", "", "a share this computer serves itself, as OpenVibe Node: its shr_… id, no key (share)"))// Accept flags anywhere: split positionals out.var positional []string@@@ -230,7 +231,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int}err = cmdProfileVerify(g, positional[1], *server, *keyID, stdout)case "share":-- err = cmdShare(g, positional, shareArgs{key: *shKey, port: *shPort, server: *shSrv}, stdout, stderr)++ err = cmdShare(g, positional, shareArgs{key: *shKey, port: *shPort, server: *shSrv, share: *shID}, stdout, stderr)if errors.Is(err, errUsage) {return 2}@@@ -598,7 +599,12 @@ func cmdRun(g *globals, dryRun bool) error {return service.Run(serviceOptions(g, ""), func(ctx context.Context) error {if !dryRun {// OpenVibe Connect: every saved share stays up while the Node runs (openvibe-node share add).-- go connect.RunAll(ctx, g.paths.ConfigDir, connect.Options{Log: g.log, Version: version})++ // A share bound to this computer (no key) signs in with the machine's own node token (CON-4).++ co := connect.Options{Log: g.log, Version: version}++ if creds.NetworkPaired() {++ co.NodeToken = connectNodeToken(creds)++ }++ go connect.RunAll(ctx, g.paths.ConfigDir, co)}if err := n.Run(ctx); !errors.Is(err, node.ErrQuit) {return err@@@ -12,6 +12,8 @@ import ("time""github.com/OpenVibers/OpenVibe.Node/internal/connect"++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/link")// shareArgs is `openvibe-node share …` (OpenVibe Connect, Website Connect v0).@@@ -19,16 +21,71 @@ type shareArgs struct {key stringport intserver string++ share string // shr_…: this computer serves it as OpenVibe Node, no key (CON-4)}const shareUsage = `usage:-- openvibe-node share --key ovc_… --port N serve the share now, until Ctrl-C-- openvibe-node share add --key ovc_… --port N keep it up from the Node service (restart the service to start it)-- openvibe-node share list the shares the service keeps up-- openvibe-node share remove <PORT|KEY|ADDRESS> stop keeping one up++ openvibe-node share --share shr_… [--port N] serve a share this computer was picked for, until Ctrl-C (no key)++ openvibe-node share add --share shr_… [--port N] keep it up from the Node service++ openvibe-node share mine the shares this computer was picked for++ openvibe-node share --key ovc_… --port N serve a share with its connect key, until Ctrl-C++ openvibe-node share add --key ovc_… --port N keep it up from the Node service (restart the service to start it)++ openvibe-node share list the shares the service keeps up++ openvibe-node share remove <PORT|KEY|ID|ADDRESS> stop keeping one up--Make a share and its key at https://connect.openvibe.website. The Node only ever connects to localhost on the--share's port; every visitor signs in with OpenVibe first.`++Make a share at https://connect.openvibe.website. On its page, pick this computer under "Or serve it from one of your++computers" to serve it with --share and no key (the computer must be added on openvibe.bot/machines). Without --port,++the Node serves the port the share's page names now, and keeps that one. The Node only ever connects to localhost on++the share's port; every visitor signs in with OpenVibe first.`++++// connectNodeToken buys this machine's node token for OpenVibe Connect (audience openvibe.website). A machine++// removed on OpenVibe stops its shares (connect.ErrStopped) instead of trying forever.++func connectNodeToken(creds *credentials.Credentials) func(context.Context) (string, error) {++ ts := link.NewTokenSourceFor(creds, nil, "openvibe-node/"+version, link.ConnectAudience)++ return func(ctx context.Context) (string, error) {++ tok, err := ts.Token(ctx)++ if errors.Is(err, link.ErrNodeRevoked) {++ return "", fmt.Errorf("%w: %v", connect.ErrStopped, err)++ }++ if err != nil {++ return "", err++ }++ return tok.Reveal(), nil++ }++}++++// nodeShare is a share this computer serves by id: the machine's token, and the port (the one given, else the++// share's page's now).++func nodeShare(ctx context.Context, g *globals, a shareArgs) (connect.Share, func(context.Context) (string, error), error) {++ if !connect.ShareIDRe.MatchString(a.share) {++ return connect.Share{}, nil, errors.New("--share is the share's shr_… id, from its page on connect.openvibe.website")++ }++ creds, err := credentials.Load(g.paths.CredentialFile(), nil)++ if err != nil || !creds.NetworkPaired() {++ return connect.Share{}, nil, errors.New("this computer is not added to OpenVibe yet: add it on openvibe.bot/machines, then pick it on the share's page")++ }++ tokens := connectNodeToken(creds)++ s := connect.Share{ID: a.share, Port: a.port, Server: a.server}++ if s.Port == 0 {++ tok, err := tokens(ctx)++ if err != nil {++ return connect.Share{}, nil, err++ }++ mine, err := connect.NodeShares(ctx, nil, a.server, tok, "openvibe-node/"+version)++ if err != nil {++ return connect.Share{}, nil, err++ }++ for _, m := range mine {++ if m.ID == a.share {++ s.Port, s.Name = m.LocalPort, m.Host++ }++ }++ if s.Port == 0 {++ return connect.Share{}, nil, fmt.Errorf("this computer does not serve %s: pick it on the share's page on connect.openvibe.website", a.share)++ }++ }++ return s, tokens, nil++}func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Writer) error {sub := ""@@@ -40,12 +97,51 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr}switch sub {case "", "run":++ if a.share != "" {++ return shareNow(g, a, stdout)++ }if a.key == "" || a.port == 0 {fmt.Fprintln(stderr, shareUsage)return errUsage}return shareNow(g, a, stdout)++ case "mine":++ creds, err := credentials.Load(g.paths.CredentialFile(), nil)++ if err != nil || !creds.NetworkPaired() {++ return errors.New("this computer is not added to OpenVibe yet: add it on openvibe.bot/machines")++ }++ ctx := context.Background()++ tok, err := connectNodeToken(creds)(ctx)++ if err != nil {++ return err++ }++ mine, err := connect.NodeShares(ctx, nil, a.server, tok, "openvibe-node/"+version)++ if err != nil {++ return err++ }++ if len(mine) == 0 {++ fmt.Fprintln(stdout, "No shares picked this computer yet. On a share's page on connect.openvibe.website, pick it under \"Or serve it from one of your computers\".")++ return nil++ }++ for _, m := range mine {++ fmt.Fprintf(stdout, "%s %s → localhost:%d\n", m.ID, m.URL, m.LocalPort)++ }++ return nilcase "add":++ if a.share != "" {++ if err := g.paths.Ensure(); err != nil {++ return err++ }++ s, _, err := nodeShare(context.Background(), g, a)++ if err != nil {++ return err++ }++ if err := connect.Add(g.paths.ConfigDir, s); err != nil {++ return err++ }++ fmt.Fprintf(stdout, "Saved: %s on localhost:%d, served by this computer (no key). Restart the Node service to start it.\n", s.ID, s.Port)++ return nil++ }if a.key == "" || a.port == 0 {fmt.Fprintln(stderr, shareUsage)return errUsage@@@ -75,6 +171,10 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wrif name == "" {name = "(address shown once it connects)"}++ if s.ID != "" {++ fmt.Fprintf(stdout, "localhost:%d %s share %s (this computer, no key)\n", s.Port, name, s.ID)++ continue++ }fmt.Fprintf(stdout, "localhost:%d %s key %s\n", s.Port, name, connect.Redact(s.Key))}return nil@@@ -102,9 +202,19 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wrfunc shareNow(g *globals, a shareArgs, stdout io.Writer) error {ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)defer stop()-- err := connect.Run(ctx, connect.Share{Key: a.key, Port: a.port, Server: a.server}, connect.Options{-- Log: g.log,-- Version: version,++ share := connect.Share{Key: a.key, Port: a.port, Server: a.server}++ var nodeToken func(context.Context) (string, error)++ if a.share != "" {++ s, tokens, err := nodeShare(ctx, g, a)++ if err != nil {++ return err++ }++ share, nodeToken, a.port = s, tokens, s.Port++ }++ err := connect.Run(ctx, share, connect.Options{++ Log: g.log,++ Version: version,++ NodeToken: nodeToken,OnReady: func(r connect.Ready) {fmt.Fprintf(stdout, "Connected. %s → localhost:%d (only the people you let in can open it). Ctrl-C to stop.\n", r.URL, a.port)},@@@ -52,3 +52,23 @@ func TestShareCommands(t *testing.T) {t.Fatalf("a bad key: %d %q", code, errb.String())}}++++// A share by id needs this computer added to OpenVibe (a Network pairing); without it the commands say where to do that.++func TestShareByIDNeedsThePairing(t *testing.T) {++ home := t.TempDir()++ var out, errb bytes.Buffer++ for _, args := range [][]string{++ {"share", "--share", "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ", "--home", home},++ {"share", "add", "--share", "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ", "--port", "8790", "--home", home},++ {"share", "mine", "--home", home},++ } {++ errb.Reset()++ if code := run(args, &out, &errb); code != 1 || !strings.Contains(errb.String(), "openvibe.bot/machines") {++ t.Fatalf("%v: %d %q", args, code, errb.String())++ }++ }++ errb.Reset()++ if code := run([]string{"share", "--share", "shr_nope", "--home", home}, &out, &errb); code != 1 || !strings.Contains(errb.String(), "shr_") {++ t.Fatalf("a bad id: %d %q", code, errb.String())++ }++}@@@ -5,6 +5,27 @@The Node speaks Connect's protocol itself (`internal/connect`, `ovconnect-ws/1`), so a share can stay up as part ofthe Node service, across reboots, without the one-file client.++## A share this computer serves, with no key++++A computer added on [openvibe.bot/machines](https://openvibe.bot/machines) has its own OpenVibe sign-in (its node++principal). On a share's page on connect.openvibe.website, pick it under "Or serve it from one of your computers",++then on the computer:++++```sh++openvibe-node share mine # the shares this computer was picked for++openvibe-node share --share shr_… # serve one now, until Ctrl-C++sudo openvibe-node share add --share shr_… # or keep it up with the Node service++```++++There is no key to copy or leak. Before every connection the Node buys its node token for OpenVibe Connect (audience++`openvibe.website`, 5 minutes) and asks Connect for a tunnel token for that one share (60 seconds, one use), and says++hello with both (ADR-057 decision 4, CON-4). While connected it sends a fresh node token every 45 seconds. A computer++removed on OpenVibe stops serving within about five minutes, and the Node stops that share instead of retrying.++++Without `--port`, the Node serves the port the share's page names when you run or add it, and keeps that port: if the++page names another one later, the Node keeps serving the port it was given and says so in the log. Connect can never++point it at a different local service. Pass `--port N` to choose the port yourself.++## Make a shareSign in at connect.openvibe.website, give the share a name and the port, and pick who can open it. The page shows its@@@ -5,6 +5,7 @@// the same protocol as the one-file client (OpenVibe.Connect public/ov-connect.js)://// → {"t":"hello","v":1,"key":"ovc_…","client":"openvibe-node/…","port":8790}++// or, served by this computer as OpenVibe Node (node.go, CON-4): {"t":"hello","v":1,"share":"shr_…","node_token":…,"tunnel":"ovt_…",…}// ← {"t":"ready","v":1,"share":{"id","host","url","mode"}} or {"t":"error",…} and a close// ← {"t":"req","id":N,"method","path","headers":[[k,v]…]} an HTTP request for the local port// ← {"t":"raw","id":N,"head":"GET /ws HTTP/1.1\r\n…"} a WebSocket upgrade: a byte pipe@@@ -47,9 +48,11 @@ const (chunkSize = 32 * 1024)--// Share is one share the Node keeps up: its connect key and the local port it serves.++// Share is one share the Node keeps up: its connect key (or, served by this computer as OpenVibe Node, its id) and++// the local port it serves.type Share struct {-- Key string `json:"key"`++ Key string `json:"key,omitempty"`++ ID string `json:"share,omitempty"` // shr_…: this computer serves it with its own node token, no key (CON-4)Port int `json:"port"`Server string `json:"server,omitempty"` // default DefaultServerName string `json:"name,omitempty"` // the share's address, learnt from "ready" (shown by `share list`)@@@ -76,6 +79,20 @@ type Options struct {Dialer *websocket.DialerMinWait time.Duration // reconnect backoff floor (default 1 s)MaxWait time.Duration // reconnect backoff ceiling (default 30 s)++++ // NodeToken buys this computer's node token for openvibe.website, for shares it serves by id (node.go). An error++ // wrapping ErrStopped (the machine was removed on OpenVibe) stops the share; any other is tried again.++ NodeToken func(ctx context.Context) (string, error)++ HTTPClient *http.Client // Connect's API, for tunnel tokens (default: 20 s timeout)++ ReauthEvery time.Duration // default DefaultReauthEvery++}++++// Label names a share in logs and lists without its key: the id, or the key's first characters.++func (s Share) Label() string {++ if s.ID != "" {++ return s.ID++ }++ return Redact(s.Key)}// Run keeps one share connected until ctx ends or the server stops it (ErrStopped).@@@ -83,9 +100,20 @@ func Run(ctx context.Context, s Share, o Options) error {if s.Port < 1 || s.Port > 65535 {return fmt.Errorf("connect: port %d is not a port", s.Port)}-- if !strings.HasPrefix(s.Key, "ovc_") {++ switch {++ case s.ID != "":++ if !ShareIDRe.MatchString(s.ID) {++ return errors.New("connect: the share is its shr_… id from connect.openvibe.website")++ }++ if o.NodeToken == nil {++ return errors.New("connect: this share is served by the computer itself; pair the Node with OpenVibe first (openvibe.bot/machines)")++ }++ case !strings.HasPrefix(s.Key, "ovc_"):return errors.New("connect: the key is the share's ovc_… connect key")}++ if o.ReauthEvery == 0 {++ o.ReauthEvery = DefaultReauthEvery++ }if o.Log == nil {o.Log = slog.New(slog.NewTextHandler(io.Discard, nil))}@@@ -185,6 +213,24 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {if server == "" {server = DefaultServer}++ hello := map[string]any{"t": "hello", "v": protocolVersion, "client": "openvibe-node/" + o.Version, "port": s.Port}++ if s.ID != "" {++ // Served by this computer (node.go): a fresh node token and a tunnel token for this one hello.++ tok, err := o.NodeToken(ctx)++ if err != nil {++ return err++ }++ ticket, info, err := tunnelToken(ctx, o, server, s.ID, tok)++ if err != nil {++ return err++ }++ if info.LocalPort != 0 && info.LocalPort != s.Port {++ o.Log.Warn("connect: the share's page names another port; this computer serves the one it was given", "share", s.ID, "page", info.LocalPort, "serving", s.Port)++ }++ hello["share"], hello["node_token"], hello["tunnel"] = s.ID, tok, ticket++ } else {++ hello["key"] = s.Key++ }dialer := o.Dialerif dialer == nil {dialer = &websocket.Dialer{HandshakeTimeout: 15 * time.Second, Proxy: http.ProxyFromEnvironment}@@@ -199,9 +245,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {defer t.closeAll()go func() { <-ctx.Done(); _ = ws.Close() }()// The server pings every 25 s; answering is gorilla's default ping handler. A dead link shows as a read error.-- if err := t.sendJSON(map[string]any{"t": "hello", "v": protocolVersion, "key": s.Key, "client": "openvibe-node/" + o.Version, "port": s.Port}); err != nil {++ if err := t.sendJSON(hello); err != nil {return err}++ reauthing := falsefor {mt, data, err := ws.ReadMessage()if err != nil {@@@ -232,6 +279,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {switch m.T {case "ready":onReady()++ if s.ID != "" && !reauthing {++ reauthing = true++ go t.reauthLoop(ctx)++ }o.Log.Info("connect: connected", "url", m.Share.URL, "port", s.Port)if o.OnReady != nil {o.OnReady(m.Share)@@@ -252,6 +303,28 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {}}++// reauthLoop sends a fresh node token every ReauthEvery while the tunnel is up (a share served by this computer). A++// token it cannot buy is skipped: Connect closes the tunnel once the last one runs out, and Run reconnects or stops.++func (t *tunnel) reauthLoop(ctx context.Context) {++ tick := time.NewTicker(t.o.ReauthEvery)++ defer tick.Stop()++ for {++ select {++ case <-ctx.Done():++ return++ case <-tick.C:++ }++ tok, err := t.o.NodeToken(ctx)++ if err != nil {++ t.o.Log.Warn("connect: no fresh node token", "share", t.share.ID, "err", err)++ continue++ }++ if err := t.sendJSON(map[string]any{"t": "reauth", "node_token": tok}); err != nil {++ return++ }++ }++}++func (t *tunnel) get(id uint32) *stream {t.mu.Lock()defer t.mu.Unlock()nnew file mode 100644@@@ -0,0 +1,142 @@++package connect++++// A share served by this computer itself, as OpenVibe Node (CON-4, ADR-057 decision 4). The share's owner picks the++// computer on the share's page; there is no connect key. Before every connection the Node buys its node token (subject++// node:nod_…, audience openvibe.website) and asks Connect for a tunnel token for that one share:++//++// POST https://connect.openvibe.website/api/v1/node/shares/<shr_…>/tunnel-token Authorization: Bearer <node token>++// → {"tunnel_token":"ovt_…","expires_in":60,"share":{"id","host","url","local_port","mode"}}++//++// then says hello with both, {"t":"hello","v":1,"share":"shr_…","node_token":…,"tunnel":"ovt_…",…}, and sends++// {"t":"reauth","node_token":…} every ReauthEvery so the tunnel outlives its 5-minute token. Neither token is logged.++//++// The port stays this computer's: the Node serves the port it was given (or the one the share had when it was added)++// even if the share's page says another later, so Connect can never point it at a different local service.++++import (++ "context"++ "encoding/json"++ "errors"++ "fmt"++ "io"++ "net"++ "net/http"++ "net/url"++ "regexp"++ "strings"++ "time"++)++++// ShareIDRe matches a share id on OpenVibe Connect (shr_ + a ULID).++var ShareIDRe = regexp.MustCompile(`^shr_[0-9A-HJKMNP-TV-Z]{26}$`)++++// DefaultReauthEvery is how often a computer's tunnel sends a fresh node token (they last 300 s; Connect allows 30 s++// past the expiry, and the Node's token cache replaces a token 30 s before it).++const DefaultReauthEvery = 45 * time.Second++++// NodeShare is one share this computer serves, as Connect lists it.++type NodeShare struct {++ ID string `json:"id"`++ Host string `json:"host"`++ URL string `json:"url"`++ LocalPort int `json:"local_port"`++ Mode string `json:"mode"`++}++++type problem struct {++ Code string `json:"code"`++ Detail string `json:"detail"`++}++++// apiURL turns the tunnel URL (wss://host/tunnel) into Connect's API URL for path; plain ws only to a loopback host.++func apiURL(server, path string) (string, error) {++ if server == "" {++ server = DefaultServer++ }++ u, err := url.Parse(server)++ if err != nil || u.Host == "" {++ return "", fmt.Errorf("connect: %q is not a tunnel URL", server)++ }++ switch u.Scheme {++ case "wss":++ u.Scheme = "https"++ case "ws":++ h := u.Hostname()++ ip := net.ParseIP(h)++ if h != "localhost" && (ip == nil || !ip.IsLoopback()) {++ return "", fmt.Errorf("connect: refusing to talk to %s over plain http", u.Host)++ }++ u.Scheme = "http"++ default:++ return "", fmt.Errorf("connect: %q is not a ws:// or wss:// URL", server)++ }++ u.Path, u.RawQuery = path, ""++ return u.String(), nil++}++++func nodeCall(ctx context.Context, client *http.Client, method, server, path, nodeToken, userAgent string, out any) (int, error) {++ if client == nil {++ client = &http.Client{Timeout: 20 * time.Second}++ }++ target, err := apiURL(server, path)++ if err != nil {++ return 0, err++ }++ req, err := http.NewRequestWithContext(ctx, method, target, nil)++ if err != nil {++ return 0, err++ }++ req.Header.Set("Authorization", "Bearer "+nodeToken)++ req.Header.Set("Accept", "application/json")++ req.Header.Set("User-Agent", userAgent)++ resp, err := client.Do(req)++ if err != nil {++ return 0, fmt.Errorf("connect: %s: %w", path, err)++ }++ defer resp.Body.Close()++ raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))++ if err != nil {++ return resp.StatusCode, err++ }++ if resp.StatusCode/100 != 2 {++ var p problem++ _ = json.Unmarshal(raw, &p)++ msg := p.Detail++ if msg == "" {++ msg = http.StatusText(resp.StatusCode)++ }++ return resp.StatusCode, fmt.Errorf("connect: HTTP %d %s: %s", resp.StatusCode, p.Code, msg)++ }++ return resp.StatusCode, json.Unmarshal(raw, out)++}++++// NodeShares lists the open shares bound to this computer (GET /api/v1/node/shares).++func NodeShares(ctx context.Context, client *http.Client, server, nodeToken, userAgent string) ([]NodeShare, error) {++ var out struct {++ Shares []NodeShare `json:"shares"`++ }++ if _, err := nodeCall(ctx, client, http.MethodGet, server, "/api/v1/node/shares", nodeToken, userAgent, &out); err != nil {++ return nil, err++ }++ return out.Shares, nil++}++++// tunnelToken asks for one hello's tunnel token. A share this computer does not serve (404) or a closed one (410)++// stops it; anything else is tried again.++func tunnelToken(ctx context.Context, o Options, server, shareID, nodeToken string) (string, NodeShare, error) {++ var out struct {++ Token string `json:"tunnel_token"`++ Share NodeShare `json:"share"`++ }++ status, err := nodeCall(ctx, o.HTTPClient, http.MethodPost, server, "/api/v1/node/shares/"+url.PathEscape(shareID)+"/tunnel-token", nodeToken, "openvibe-node/"+o.Version, &out)++ if err != nil {++ if status == http.StatusNotFound || status == http.StatusGone {++ return "", NodeShare{}, fmt.Errorf("%w: %v", ErrStopped, err)++ }++ return "", NodeShare{}, err++ }++ if !strings.HasPrefix(out.Token, "ovt_") {++ return "", NodeShare{}, errors.New("connect: no tunnel token in the answer")++ }++ return out.Token, out.Share, nil++}nnew file mode 100644@@@ -0,0 +1,214 @@++package connect++++import (++ "context"++ "encoding/json"++ "errors"++ "fmt"++ "net/http"++ "net/http/httptest"++ "strings"++ "sync"++ "sync/atomic"++ "testing"++ "time"++++ "github.com/gorilla/websocket"++)++++const testShareID = "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ"++++// nodeServer plays OpenVibe Connect for a share served by this computer (CON-4): the node API (a tunnel token per++// hello, the computer's shares) and /tunnel, which records the hello and every reauth.++type nodeServer struct {++ srv *httptest.Server++ hellos chan map[string]any++ reauths chan string++ tickets atomic.Int32++ mu sync.Mutex++ bearers []string++ notMine bool // the tunnel-token call answers 404++ pagePort int++}++++func newNodeServer(t *testing.T) *nodeServer {++ n := &nodeServer{hellos: make(chan map[string]any, 8), reauths: make(chan string, 16), pagePort: 8790}++ up := websocket.Upgrader{}++ mux := http.NewServeMux()++ mux.HandleFunc("/api/v1/node/shares", func(w http.ResponseWriter, r *http.Request) {++ n.mu.Lock()++ n.bearers = append(n.bearers, r.Header.Get("Authorization"))++ n.mu.Unlock()++ _ = json.NewEncoder(w).Encode(map[string]any{"shares": []map[string]any{{"id": testShareID, "host": "portal--alex.openvibe.website", "url": "https://portal--alex.openvibe.website", "local_port": n.pagePort, "mode": "owner"}}})++ })++ mux.HandleFunc("/api/v1/node/shares/"+testShareID+"/tunnel-token", func(w http.ResponseWriter, r *http.Request) {++ n.mu.Lock()++ n.bearers = append(n.bearers, r.Header.Get("Authorization"))++ n.mu.Unlock()++ if r.Method != http.MethodPost {++ w.WriteHeader(405)++ return++ }++ if n.notMine {++ w.Header().Set("Content-Type", "application/problem+json")++ w.WriteHeader(404)++ _, _ = w.Write([]byte(`{"code":"share.not_found","detail":"This computer does not serve that share."}`))++ return++ }++ i := n.tickets.Add(1)++ _ = json.NewEncoder(w).Encode(map[string]any{"tunnel_token": fmt.Sprintf("ovt_%043d", i), "expires_in": 60, "share": map[string]any{"id": testShareID, "local_port": n.pagePort}})++ })++ mux.HandleFunc("/tunnel", func(w http.ResponseWriter, r *http.Request) {++ ws, err := up.Upgrade(w, r, nil)++ if err != nil {++ return++ }++ defer ws.Close()++ _, data, err := ws.ReadMessage()++ if err != nil {++ return++ }++ var hello map[string]any++ _ = json.Unmarshal(data, &hello)++ n.hellos <- hello++ _ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"ready","v":1,"node":true,"share":{"id":"`+testShareID+`","host":"portal--alex.openvibe.website","url":"https://portal--alex.openvibe.website","mode":"owner","local_port":8790}}`))++ for {++ _, data, err := ws.ReadMessage()++ if err != nil {++ return++ }++ var m map[string]any++ _ = json.Unmarshal(data, &m)++ if m["t"] == "reauth" {++ tok, _ := m["node_token"].(string)++ n.reauths <- tok++ _ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"reauthed","until":"2026-11-03T00:00:00Z"}`))++ }++ }++ })++ n.srv = httptest.NewServer(mux)++ t.Cleanup(n.srv.Close)++ return n++}++++func (n *nodeServer) url() string { return "ws" + strings.TrimPrefix(n.srv.URL, "http") + "/tunnel" }++++// countingTokens hands out node-token-1, node-token-2, … (a fresh one each call).++func countingTokens() (func(context.Context) (string, error), *atomic.Int32) {++ var n atomic.Int32++ return func(context.Context) (string, error) { return fmt.Sprintf("node-token-%d", n.Add(1)), nil }, &n++}++++func TestAShareServedByThisComputer(t *testing.T) {++ n := newNodeServer(t)++ tokens, _ := countingTokens()++ ctx, cancel := context.WithCancel(context.Background())++ defer cancel()++ ready := make(chan Ready, 2)++ done := make(chan error, 1)++ go func() {++ done <- Run(ctx, Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: tokens, ReauthEvery: 40 * time.Millisecond, OnReady: func(r Ready) { ready <- r }, MinWait: 20 * time.Millisecond})++ }()++ var hello map[string]any++ select {++ case hello = <-n.hellos:++ case err := <-done:++ t.Fatalf("Run ended: %v", err)++ case <-time.After(5 * time.Second):++ t.Fatal("no hello")++ }++ if hello["share"] != testShareID || hello["node_token"] != "node-token-1" || hello["tunnel"] != fmt.Sprintf("ovt_%043d", 1) || hello["port"].(float64) != 9100 {++ t.Fatalf("hello %v", hello)++ }++ if _, has := hello["key"]; has {++ t.Fatal("no key in a computer's hello")++ }++ if r := <-ready; r.Host != "portal--alex.openvibe.website" {++ t.Fatalf("ready %v", r)++ }++ // It keeps the tunnel with fresh node tokens.++ first, second := <-n.reauths, <-n.reauths++ if first == second || !strings.HasPrefix(first, "node-token-") {++ t.Fatalf("reauths %q %q", first, second)++ }++ n.mu.Lock()++ bearer := n.bearers[0]++ n.mu.Unlock()++ if bearer != "Bearer node-token-1" {++ t.Fatalf("the tunnel-token call carried %q", bearer)++ }++}++++func TestAShareThisComputerDoesNotServeStops(t *testing.T) {++ n := newNodeServer(t)++ n.notMine = true++ tokens, _ := countingTokens()++ err := Run(context.Background(), Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: tokens, MinWait: 10 * time.Millisecond})++ if !errors.Is(err, ErrStopped) {++ t.Fatalf("want ErrStopped, got %v", err)++ }++}++++func TestARemovedMachineStops(t *testing.T) {++ n := newNodeServer(t)++ revoked := func(context.Context) (string, error) { return "", fmt.Errorf("%w: refused", ErrStopped) }++ err := Run(context.Background(), Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: revoked, MinWait: 10 * time.Millisecond})++ if !errors.Is(err, ErrStopped) {++ t.Fatalf("want ErrStopped, got %v", err)++ }++ if n.tickets.Load() != 0 {++ t.Fatal("no tunnel token without a node token")++ }++}++++func TestAShareByIDNeedsThePairing(t *testing.T) {++ if err := Run(context.Background(), Share{ID: testShareID, Port: 9100}, Options{}); err == nil || !strings.Contains(err.Error(), "openvibe.bot/machines") {++ t.Fatalf("got %v", err)++ }++ if err := Run(context.Background(), Share{ID: "shr_nope", Port: 9100}, Options{NodeToken: func(context.Context) (string, error) { return "x", nil }}); err == nil {++ t.Fatal("a bad id")++ }++}++++func TestNodeSharesAndTheAPIURL(t *testing.T) {++ n := newNodeServer(t)++ got, err := NodeShares(context.Background(), nil, n.url(), "node-token-9", "test")++ if err != nil || len(got) != 1 || got[0].ID != testShareID || got[0].LocalPort != 8790 {++ t.Fatalf("%v %v", got, err)++ }++ for in, want := range map[string]string{++ "wss://connect.openvibe.website/tunnel": "https://connect.openvibe.website/x",++ "": "https://connect.openvibe.website/x",++ "ws://127.0.0.1:5240/tunnel": "http://127.0.0.1:5240/x",++ } {++ if u, err := apiURL(in, "/x"); err != nil || u != want {++ t.Fatalf("%q → %q %v", in, u, err)++ }++ }++ if _, err := apiURL("ws://evil.example/tunnel", "/x"); err == nil {++ t.Fatal("plain http to another host")++ }++}++++func TestSavedSharesByID(t *testing.T) {++ dir := t.TempDir()++ if err := Add(dir, Share{ID: testShareID, Port: 8790}); err != nil {++ t.Fatal(err)++ }++ if err := Add(dir, Share{Key: "ovc_" + strings.Repeat("K", 43), Port: 8791}); err != nil {++ t.Fatal(err)++ }++ if err := Add(dir, Share{ID: testShareID, Port: 8792}); err != nil { // the same id again replaces it++ t.Fatal(err)++ }++ got, _ := Load(dir)++ if len(got) != 2 {++ t.Fatalf("%v", got)++ }++ if k, _ := Remove(dir, testShareID); k != 1 {++ t.Fatalf("removed %d", k)++ }++ if got, _ := Load(dir); len(got) != 1 || got[0].Port != 8791 {++ t.Fatalf("%v", got)++ }++}@@@ -53,7 +53,7 @@ func save(dir string, shares []Share) error {return os.Rename(tmp, filepath.Join(dir, File))}--// Add saves a share (the same key again replaces its port and name).++// Add saves a share (the same key, or the same share id, again replaces its port and name).func Add(dir string, s Share) error {shares, err := Load(dir)if err != nil {@@@ -61,14 +61,14 @@ func Add(dir string, s Share) error {}out := shares[:0]for _, x := range shares {-- if x.Key != s.Key {++ if (s.Key == "" || x.Key != s.Key) && (s.ID == "" || x.ID != s.ID) {out = append(out, x)}}return save(dir, append(out, s))}--// Remove forgets the share with this key, or every share on this port; it reports how many it removed.++// Remove forgets the share with this key or id, or every share on this port; it reports how many it removed.func Remove(dir string, keyOrPort string) (int, error) {shares, err := Load(dir)if err != nil {@@@ -77,7 +77,7 @@ func Remove(dir string, keyOrPort string) (int, error) {out := shares[:0]n := 0for _, x := range shares {-- if x.Key == keyOrPort || fmt.Sprint(x.Port) == keyOrPort || (x.Name != "" && x.Name == keyOrPort) {++ if (x.Key != "" && x.Key == keyOrPort) || (x.ID != "" && x.ID == keyOrPort) || fmt.Sprint(x.Port) == keyOrPort || (x.Name != "" && x.Name == keyOrPort) {n++continue}@@@ -126,7 +126,7 @@ func RunAll(ctx context.Context, dir string, o Options) {defer wg.Done()err := Run(ctx, s, so)if errors.Is(err, ErrStopped) && o.Log != nil {-- o.Log.Warn("connect: share stopped; remove it with `openvibe-node share remove`", "port", s.Port, "key", Redact(s.Key))++ o.Log.Warn("connect: share stopped; remove it with `openvibe-node share remove`", "port", s.Port, "share", s.Label())}}()}@@@ -151,8 +151,17 @@ type TokenSource struct {// NewTokenSource returns the token source of a Network-paired machine.func NewTokenSource(c *credentials.Credentials, client *http.Client, userAgent string) *TokenSource {++ return NewTokenSourceFor(c, client, userAgent, Audience(c))++}++++// ConnectAudience is OpenVibe Connect's: any person's machine may buy a node token for it, whichever service paired++// the machine (Network, CON-4), to serve the shares bound to it.++const ConnectAudience = "openvibe.website"++++// NewTokenSourceFor returns a token source for another audience the machine may speak to (ConnectAudience).++func NewTokenSourceFor(c *credentials.Credentials, client *http.Client, userAgent, audience string) *TokenSource {return &TokenSource{network: strings.TrimSuffix(c.Network, "/"), principal: c.Principal, credential: c.NodeCredential,-- audience: Audience(c), client: defaultClient(client), userAgent: userAgent, now: time.Now}++ audience: audience, client: defaultClient(client), userAgent: userAgent, now: time.Now}}// Invalidate drops the cached token, so the next Token fetches a fresh one (after a 401 from Bot, before a reauth).
@@ -5,7 +5,7 @@
@@ -5,7 +5,7 @@
"wave": "T15",
"wave": "T15",
"plan": "Plan track T15 (Bot complete) and ADR-043 (Bot: devices, pairing, control and safety); the computer plugin and the job worker serve OpenVibe.Actor and the Codes job runner (tracks T14/T17)",
"plan": "Plan track T15 (Bot complete) and ADR-043 (Bot: devices, pairing, control and safety); the computer plugin and the job worker serve OpenVibe.Actor and the Codes job runner (tracks T14/T17)",
"kind": "client",
"kind": "client",
- "version": "0.37.2",
+ "version": "0.38.0",
"release": "v0.36.0: the ai plugin (0.2.0) passes tools to the machine's own model server and hands back the calls the model asked for, never running one, so an agent on OpenVibe (Actor) can work on a model this machine serves; its capabilities say tools: true (contracts 0.277.0). v0.35.0: the Node applies the roles the network's machines hub gives it at each check-in (codes.worker starts the job worker it was installed with, desks.host and ai.host are checked against what the machine was installed for, an unknown role is ignored, and the local configuration always wins), and its status reports them back; a desk's first browser start in a cold sandbox gets 90 s. v0.34.0: on Windows a killed plugin takes every process it started with it (a Job Object with kill-on-close), as the process group does on Linux and macOS, so a bridge helper cannot keep a robot's link open after the Node stops or restarts its plugin; the Node reads what the coding tools remember (read-only, local, redacted) and `openvibe-node memory` uploads it to your own store; the heartbeat carries the machine's fleet of coding agents; the Codes job runner pin is v0.5.2",
"release": "v0.36.0: the ai plugin (0.2.0) passes tools to the machine's own model server and hands back the calls the model asked for, never running one, so an agent on OpenVibe (Actor) can work on a model this machine serves; its capabilities say tools: true (contracts 0.277.0). v0.35.0: the Node applies the roles the network's machines hub gives it at each check-in (codes.worker starts the job worker it was installed with, desks.host and ai.host are checked against what the machine was installed for, an unknown role is ignored, and the local configuration always wins), and its status reports them back; a desk's first browser start in a cold sandbox gets 90 s. v0.34.0: on Windows a killed plugin takes every process it started with it (a Job Object with kill-on-close), as the process group does on Linux and macOS, so a bridge helper cannot keep a robot's link open after the Node stops or restarts its plugin; the Node reads what the coding tools remember (read-only, local, redacted) and `openvibe-node memory` uploads it to your own store; the heartbeat carries the machine's fleet of coding agents; the Codes job runner pin is v0.5.2",
"runtime": "go 1.24 (one static binary per platform); plugins in Python 3; an ESP32 Arduino library",
"runtime": "go 1.24 (one static binary per platform); plugins in Python 3; an ESP32 Arduino library",
"database": "none",
"database": "none",
@@ -140,6 +140,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
@@ -140,6 +140,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
shKey = fs.String("key", "", "the share's ovc_… connect key from connect.openvibe.website (share)")
shKey = fs.String("key", "", "the share's ovc_… connect key from connect.openvibe.website (share)")
shPort = fs.Int("port", 0, "the local port to share (share)")
shPort = fs.Int("port", 0, "the local port to share (share)")
shSrv = fs.String("connect-server", "", "OpenVibe Connect's tunnel URL; default wss://connect.openvibe.website/tunnel (share)")
shSrv = fs.String("connect-server", "", "OpenVibe Connect's tunnel URL; default wss://connect.openvibe.website/tunnel (share)")
+ shID = fs.String("share", "", "a share this computer serves itself, as OpenVibe Node: its shr_… id, no key (share)")
)
)
// Accept flags anywhere: split positionals out.
// Accept flags anywhere: split positionals out.
var positional []string
var positional []string
@@ -230,7 +231,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
@@ -230,7 +231,7 @@ func runWithStdin(args []string, stdin io.Reader, stdout, stderr io.Writer) int
}
}
err = cmdProfileVerify(g, positional[1], *server, *keyID, stdout)
err = cmdProfileVerify(g, positional[1], *server, *keyID, stdout)
case "share":
case "share":
- err = cmdShare(g, positional, shareArgs{key: *shKey, port: *shPort, server: *shSrv}, stdout, stderr)
+ err = cmdShare(g, positional, shareArgs{key: *shKey, port: *shPort, server: *shSrv, share: *shID}, stdout, stderr)
if errors.Is(err, errUsage) {
if errors.Is(err, errUsage) {
return 2
return 2
}
}
@@ -598,7 +599,12 @@ func cmdRun(g *globals, dryRun bool) error {
@@ -598,7 +599,12 @@ func cmdRun(g *globals, dryRun bool) error {
return service.Run(serviceOptions(g, ""), func(ctx context.Context) error {
return service.Run(serviceOptions(g, ""), func(ctx context.Context) error {
if !dryRun {
if !dryRun {
// OpenVibe Connect: every saved share stays up while the Node runs (openvibe-node share add).
// OpenVibe Connect: every saved share stays up while the Node runs (openvibe-node share add).
- go connect.RunAll(ctx, g.paths.ConfigDir, connect.Options{Log: g.log, Version: version})
+ // A share bound to this computer (no key) signs in with the machine's own node token (CON-4).
+ co := connect.Options{Log: g.log, Version: version}
+ if creds.NetworkPaired() {
+ co.NodeToken = connectNodeToken(creds)
+ }
+ go connect.RunAll(ctx, g.paths.ConfigDir, co)
}
}
if err := n.Run(ctx); !errors.Is(err, node.ErrQuit) {
if err := n.Run(ctx); !errors.Is(err, node.ErrQuit) {
return err
return err
@@ -12,6 +12,8 @@ import (
@@ -12,6 +12,8 @@ import (
"time"
"time"
"github.com/OpenVibers/OpenVibe.Node/internal/connect"
"github.com/OpenVibers/OpenVibe.Node/internal/connect"
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/link"
)
)
// shareArgs is `openvibe-node share …` (OpenVibe Connect, Website Connect v0).
// shareArgs is `openvibe-node share …` (OpenVibe Connect, Website Connect v0).
@@ -19,16 +21,71 @@ type shareArgs struct {
@@ -19,16 +21,71 @@ type shareArgs struct {
key string
key string
port int
port int
server string
server string
+ share string // shr_…: this computer serves it as OpenVibe Node, no key (CON-4)
}
}
const shareUsage = `usage:
const shareUsage = `usage:
- openvibe-node share --key ovc_… --port N serve the share now, until Ctrl-C
- openvibe-node share add --key ovc_… --port N keep it up from the Node service (restart the service to start it)
- openvibe-node share list the shares the service keeps up
- openvibe-node share remove <PORT|KEY|ADDRESS> stop keeping one up
+ openvibe-node share --share shr_… [--port N] serve a share this computer was picked for, until Ctrl-C (no key)
+ openvibe-node share add --share shr_… [--port N] keep it up from the Node service
+ openvibe-node share mine the shares this computer was picked for
+ openvibe-node share --key ovc_… --port N serve a share with its connect key, until Ctrl-C
+ openvibe-node share add --key ovc_… --port N keep it up from the Node service (restart the service to start it)
+ openvibe-node share list the shares the service keeps up
+ openvibe-node share remove <PORT|KEY|ID|ADDRESS> stop keeping one up
-Make a share and its key at https://connect.openvibe.website. The Node only ever connects to localhost on the
-share's port; every visitor signs in with OpenVibe first.`
+Make a share at https://connect.openvibe.website. On its page, pick this computer under "Or serve it from one of your
+computers" to serve it with --share and no key (the computer must be added on openvibe.bot/machines). Without --port,
+the Node serves the port the share's page names now, and keeps that one. The Node only ever connects to localhost on
+the share's port; every visitor signs in with OpenVibe first.`
+
+// connectNodeToken buys this machine's node token for OpenVibe Connect (audience openvibe.website). A machine
+// removed on OpenVibe stops its shares (connect.ErrStopped) instead of trying forever.
+func connectNodeToken(creds *credentials.Credentials) func(context.Context) (string, error) {
+ ts := link.NewTokenSourceFor(creds, nil, "openvibe-node/"+version, link.ConnectAudience)
+ return func(ctx context.Context) (string, error) {
+ tok, err := ts.Token(ctx)
+ if errors.Is(err, link.ErrNodeRevoked) {
+ return "", fmt.Errorf("%w: %v", connect.ErrStopped, err)
+ }
+ if err != nil {
+ return "", err
+ }
+ return tok.Reveal(), nil
+ }
+}
+
+// nodeShare is a share this computer serves by id: the machine's token, and the port (the one given, else the
+// share's page's now).
+func nodeShare(ctx context.Context, g *globals, a shareArgs) (connect.Share, func(context.Context) (string, error), error) {
+ if !connect.ShareIDRe.MatchString(a.share) {
+ return connect.Share{}, nil, errors.New("--share is the share's shr_… id, from its page on connect.openvibe.website")
+ }
+ creds, err := credentials.Load(g.paths.CredentialFile(), nil)
+ if err != nil || !creds.NetworkPaired() {
+ return connect.Share{}, nil, errors.New("this computer is not added to OpenVibe yet: add it on openvibe.bot/machines, then pick it on the share's page")
+ }
+ tokens := connectNodeToken(creds)
+ s := connect.Share{ID: a.share, Port: a.port, Server: a.server}
+ if s.Port == 0 {
+ tok, err := tokens(ctx)
+ if err != nil {
+ return connect.Share{}, nil, err
+ }
+ mine, err := connect.NodeShares(ctx, nil, a.server, tok, "openvibe-node/"+version)
+ if err != nil {
+ return connect.Share{}, nil, err
+ }
+ for _, m := range mine {
+ if m.ID == a.share {
+ s.Port, s.Name = m.LocalPort, m.Host
+ }
+ }
+ if s.Port == 0 {
+ return connect.Share{}, nil, fmt.Errorf("this computer does not serve %s: pick it on the share's page on connect.openvibe.website", a.share)
+ }
+ }
+ return s, tokens, nil
+}
func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Writer) error {
func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Writer) error {
sub := ""
sub := ""
@@ -40,12 +97,51 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
@@ -40,12 +97,51 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
}
}
switch sub {
switch sub {
case "", "run":
case "", "run":
+ if a.share != "" {
+ return shareNow(g, a, stdout)
+ }
if a.key == "" || a.port == 0 {
if a.key == "" || a.port == 0 {
fmt.Fprintln(stderr, shareUsage)
fmt.Fprintln(stderr, shareUsage)
return errUsage
return errUsage
}
}
return shareNow(g, a, stdout)
return shareNow(g, a, stdout)
+ case "mine":
+ creds, err := credentials.Load(g.paths.CredentialFile(), nil)
+ if err != nil || !creds.NetworkPaired() {
+ return errors.New("this computer is not added to OpenVibe yet: add it on openvibe.bot/machines")
+ }
+ ctx := context.Background()
+ tok, err := connectNodeToken(creds)(ctx)
+ if err != nil {
+ return err
+ }
+ mine, err := connect.NodeShares(ctx, nil, a.server, tok, "openvibe-node/"+version)
+ if err != nil {
+ return err
+ }
+ if len(mine) == 0 {
+ fmt.Fprintln(stdout, "No shares picked this computer yet. On a share's page on connect.openvibe.website, pick it under \"Or serve it from one of your computers\".")
+ return nil
+ }
+ for _, m := range mine {
+ fmt.Fprintf(stdout, "%s %s → localhost:%d\n", m.ID, m.URL, m.LocalPort)
+ }
+ return nil
case "add":
case "add":
+ if a.share != "" {
+ if err := g.paths.Ensure(); err != nil {
+ return err
+ }
+ s, _, err := nodeShare(context.Background(), g, a)
+ if err != nil {
+ return err
+ }
+ if err := connect.Add(g.paths.ConfigDir, s); err != nil {
+ return err
+ }
+ fmt.Fprintf(stdout, "Saved: %s on localhost:%d, served by this computer (no key). Restart the Node service to start it.\n", s.ID, s.Port)
+ return nil
+ }
if a.key == "" || a.port == 0 {
if a.key == "" || a.port == 0 {
fmt.Fprintln(stderr, shareUsage)
fmt.Fprintln(stderr, shareUsage)
return errUsage
return errUsage
@@ -75,6 +171,10 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
@@ -75,6 +171,10 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
if name == "" {
if name == "" {
name = "(address shown once it connects)"
name = "(address shown once it connects)"
}
}
+ if s.ID != "" {
+ fmt.Fprintf(stdout, "localhost:%d %s share %s (this computer, no key)\n", s.Port, name, s.ID)
+ continue
+ }
fmt.Fprintf(stdout, "localhost:%d %s key %s\n", s.Port, name, connect.Redact(s.Key))
fmt.Fprintf(stdout, "localhost:%d %s key %s\n", s.Port, name, connect.Redact(s.Key))
}
}
return nil
return nil
@@ -102,9 +202,19 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
@@ -102,9 +202,19 @@ func cmdShare(g *globals, positional []string, a shareArgs, stdout, stderr io.Wr
func shareNow(g *globals, a shareArgs, stdout io.Writer) error {
func shareNow(g *globals, a shareArgs, stdout io.Writer) error {
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
defer stop()
defer stop()
- err := connect.Run(ctx, connect.Share{Key: a.key, Port: a.port, Server: a.server}, connect.Options{
- Log: g.log,
- Version: version,
+ share := connect.Share{Key: a.key, Port: a.port, Server: a.server}
+ var nodeToken func(context.Context) (string, error)
+ if a.share != "" {
+ s, tokens, err := nodeShare(ctx, g, a)
+ if err != nil {
+ return err
+ }
+ share, nodeToken, a.port = s, tokens, s.Port
+ }
+ err := connect.Run(ctx, share, connect.Options{
+ Log: g.log,
+ Version: version,
+ NodeToken: nodeToken,
OnReady: func(r connect.Ready) {
OnReady: func(r connect.Ready) {
fmt.Fprintf(stdout, "Connected. %s → localhost:%d (only the people you let in can open it). Ctrl-C to stop.\n", r.URL, a.port)
fmt.Fprintf(stdout, "Connected. %s → localhost:%d (only the people you let in can open it). Ctrl-C to stop.\n", r.URL, a.port)
},
},
@@ -52,3 +52,23 @@ func TestShareCommands(t *testing.T) {
@@ -52,3 +52,23 @@ func TestShareCommands(t *testing.T) {
t.Fatalf("a bad key: %d %q", code, errb.String())
t.Fatalf("a bad key: %d %q", code, errb.String())
}
}
}
}
+
+// A share by id needs this computer added to OpenVibe (a Network pairing); without it the commands say where to do that.
+func TestShareByIDNeedsThePairing(t *testing.T) {
+ home := t.TempDir()
+ var out, errb bytes.Buffer
+ for _, args := range [][]string{
+ {"share", "--share", "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ", "--home", home},
+ {"share", "add", "--share", "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ", "--port", "8790", "--home", home},
+ {"share", "mine", "--home", home},
+ } {
+ errb.Reset()
+ if code := run(args, &out, &errb); code != 1 || !strings.Contains(errb.String(), "openvibe.bot/machines") {
+ t.Fatalf("%v: %d %q", args, code, errb.String())
+ }
+ }
+ errb.Reset()
+ if code := run([]string{"share", "--share", "shr_nope", "--home", home}, &out, &errb); code != 1 || !strings.Contains(errb.String(), "shr_") {
+ t.Fatalf("a bad id: %d %q", code, errb.String())
+ }
+}
@@ -5,6 +5,27 @@
@@ -5,6 +5,27 @@
The Node speaks Connect's protocol itself (`internal/connect`, `ovconnect-ws/1`), so a share can stay up as part of
The Node speaks Connect's protocol itself (`internal/connect`, `ovconnect-ws/1`), so a share can stay up as part of
the Node service, across reboots, without the one-file client.
the Node service, across reboots, without the one-file client.
+## A share this computer serves, with no key
+
+A computer added on [openvibe.bot/machines](https://openvibe.bot/machines) has its own OpenVibe sign-in (its node
+principal). On a share's page on connect.openvibe.website, pick it under "Or serve it from one of your computers",
+then on the computer:
+
+```sh
+openvibe-node share mine # the shares this computer was picked for
+openvibe-node share --share shr_… # serve one now, until Ctrl-C
+sudo openvibe-node share add --share shr_… # or keep it up with the Node service
+```
+
+There is no key to copy or leak. Before every connection the Node buys its node token for OpenVibe Connect (audience
+`openvibe.website`, 5 minutes) and asks Connect for a tunnel token for that one share (60 seconds, one use), and says
+hello with both (ADR-057 decision 4, CON-4). While connected it sends a fresh node token every 45 seconds. A computer
+removed on OpenVibe stops serving within about five minutes, and the Node stops that share instead of retrying.
+
+Without `--port`, the Node serves the port the share's page names when you run or add it, and keeps that port: if the
+page names another one later, the Node keeps serving the port it was given and says so in the log. Connect can never
+point it at a different local service. Pass `--port N` to choose the port yourself.
+
## Make a share
## Make a share
Sign in at connect.openvibe.website, give the share a name and the port, and pick who can open it. The page shows its
Sign in at connect.openvibe.website, give the share a name and the port, and pick who can open it. The page shows its
@@ -5,6 +5,7 @@
@@ -5,6 +5,7 @@
// the same protocol as the one-file client (OpenVibe.Connect public/ov-connect.js):
// the same protocol as the one-file client (OpenVibe.Connect public/ov-connect.js):
//
//
// → {"t":"hello","v":1,"key":"ovc_…","client":"openvibe-node/…","port":8790}
// → {"t":"hello","v":1,"key":"ovc_…","client":"openvibe-node/…","port":8790}
+// or, served by this computer as OpenVibe Node (node.go, CON-4): {"t":"hello","v":1,"share":"shr_…","node_token":…,"tunnel":"ovt_…",…}
// ← {"t":"ready","v":1,"share":{"id","host","url","mode"}} or {"t":"error",…} and a close
// ← {"t":"ready","v":1,"share":{"id","host","url","mode"}} or {"t":"error",…} and a close
// ← {"t":"req","id":N,"method","path","headers":[[k,v]…]} an HTTP request for the local port
// ← {"t":"req","id":N,"method","path","headers":[[k,v]…]} an HTTP request for the local port
// ← {"t":"raw","id":N,"head":"GET /ws HTTP/1.1\r\n…"} a WebSocket upgrade: a byte pipe
// ← {"t":"raw","id":N,"head":"GET /ws HTTP/1.1\r\n…"} a WebSocket upgrade: a byte pipe
@@ -47,9 +48,11 @@ const (
@@ -47,9 +48,11 @@ const (
chunkSize = 32 * 1024
chunkSize = 32 * 1024
)
)
-// Share is one share the Node keeps up: its connect key and the local port it serves.
+// Share is one share the Node keeps up: its connect key (or, served by this computer as OpenVibe Node, its id) and
+// the local port it serves.
type Share struct {
type Share struct {
- Key string `json:"key"`
+ Key string `json:"key,omitempty"`
+ ID string `json:"share,omitempty"` // shr_…: this computer serves it with its own node token, no key (CON-4)
Port int `json:"port"`
Port int `json:"port"`
Server string `json:"server,omitempty"` // default DefaultServer
Server string `json:"server,omitempty"` // default DefaultServer
Name string `json:"name,omitempty"` // the share's address, learnt from "ready" (shown by `share list`)
Name string `json:"name,omitempty"` // the share's address, learnt from "ready" (shown by `share list`)
@@ -76,6 +79,20 @@ type Options struct {
@@ -76,6 +79,20 @@ type Options struct {
Dialer *websocket.Dialer
Dialer *websocket.Dialer
MinWait time.Duration // reconnect backoff floor (default 1 s)
MinWait time.Duration // reconnect backoff floor (default 1 s)
MaxWait time.Duration // reconnect backoff ceiling (default 30 s)
MaxWait time.Duration // reconnect backoff ceiling (default 30 s)
+
+ // NodeToken buys this computer's node token for openvibe.website, for shares it serves by id (node.go). An error
+ // wrapping ErrStopped (the machine was removed on OpenVibe) stops the share; any other is tried again.
+ NodeToken func(ctx context.Context) (string, error)
+ HTTPClient *http.Client // Connect's API, for tunnel tokens (default: 20 s timeout)
+ ReauthEvery time.Duration // default DefaultReauthEvery
+}
+
+// Label names a share in logs and lists without its key: the id, or the key's first characters.
+func (s Share) Label() string {
+ if s.ID != "" {
+ return s.ID
+ }
+ return Redact(s.Key)
}
}
// Run keeps one share connected until ctx ends or the server stops it (ErrStopped).
// Run keeps one share connected until ctx ends or the server stops it (ErrStopped).
@@ -83,9 +100,20 @@ func Run(ctx context.Context, s Share, o Options) error {
@@ -83,9 +100,20 @@ func Run(ctx context.Context, s Share, o Options) error {
if s.Port < 1 || s.Port > 65535 {
if s.Port < 1 || s.Port > 65535 {
return fmt.Errorf("connect: port %d is not a port", s.Port)
return fmt.Errorf("connect: port %d is not a port", s.Port)
}
}
- if !strings.HasPrefix(s.Key, "ovc_") {
+ switch {
+ case s.ID != "":
+ if !ShareIDRe.MatchString(s.ID) {
+ return errors.New("connect: the share is its shr_… id from connect.openvibe.website")
+ }
+ if o.NodeToken == nil {
+ return errors.New("connect: this share is served by the computer itself; pair the Node with OpenVibe first (openvibe.bot/machines)")
+ }
+ case !strings.HasPrefix(s.Key, "ovc_"):
return errors.New("connect: the key is the share's ovc_… connect key")
return errors.New("connect: the key is the share's ovc_… connect key")
}
}
+ if o.ReauthEvery == 0 {
+ o.ReauthEvery = DefaultReauthEvery
+ }
if o.Log == nil {
if o.Log == nil {
o.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
o.Log = slog.New(slog.NewTextHandler(io.Discard, nil))
}
}
@@ -185,6 +213,24 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
@@ -185,6 +213,24 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
if server == "" {
if server == "" {
server = DefaultServer
server = DefaultServer
}
}
+ hello := map[string]any{"t": "hello", "v": protocolVersion, "client": "openvibe-node/" + o.Version, "port": s.Port}
+ if s.ID != "" {
+ // Served by this computer (node.go): a fresh node token and a tunnel token for this one hello.
+ tok, err := o.NodeToken(ctx)
+ if err != nil {
+ return err
+ }
+ ticket, info, err := tunnelToken(ctx, o, server, s.ID, tok)
+ if err != nil {
+ return err
+ }
+ if info.LocalPort != 0 && info.LocalPort != s.Port {
+ o.Log.Warn("connect: the share's page names another port; this computer serves the one it was given", "share", s.ID, "page", info.LocalPort, "serving", s.Port)
+ }
+ hello["share"], hello["node_token"], hello["tunnel"] = s.ID, tok, ticket
+ } else {
+ hello["key"] = s.Key
+ }
dialer := o.Dialer
dialer := o.Dialer
if dialer == nil {
if dialer == nil {
dialer = &websocket.Dialer{HandshakeTimeout: 15 * time.Second, Proxy: http.ProxyFromEnvironment}
dialer = &websocket.Dialer{HandshakeTimeout: 15 * time.Second, Proxy: http.ProxyFromEnvironment}
@@ -199,9 +245,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
@@ -199,9 +245,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
defer t.closeAll()
defer t.closeAll()
go func() { <-ctx.Done(); _ = ws.Close() }()
go func() { <-ctx.Done(); _ = ws.Close() }()
// The server pings every 25 s; answering is gorilla's default ping handler. A dead link shows as a read error.
// The server pings every 25 s; answering is gorilla's default ping handler. A dead link shows as a read error.
- if err := t.sendJSON(map[string]any{"t": "hello", "v": protocolVersion, "key": s.Key, "client": "openvibe-node/" + o.Version, "port": s.Port}); err != nil {
+ if err := t.sendJSON(hello); err != nil {
return err
return err
}
}
+ reauthing := false
for {
for {
mt, data, err := ws.ReadMessage()
mt, data, err := ws.ReadMessage()
if err != nil {
if err != nil {
@@ -232,6 +279,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
@@ -232,6 +279,10 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
switch m.T {
switch m.T {
case "ready":
case "ready":
onReady()
onReady()
+ if s.ID != "" && !reauthing {
+ reauthing = true
+ go t.reauthLoop(ctx)
+ }
o.Log.Info("connect: connected", "url", m.Share.URL, "port", s.Port)
o.Log.Info("connect: connected", "url", m.Share.URL, "port", s.Port)
if o.OnReady != nil {
if o.OnReady != nil {
o.OnReady(m.Share)
o.OnReady(m.Share)
@@ -252,6 +303,28 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
@@ -252,6 +303,28 @@ func runOnce(ctx context.Context, s Share, o Options, onReady func()) error {
}
}
}
}
+// reauthLoop sends a fresh node token every ReauthEvery while the tunnel is up (a share served by this computer). A
+// token it cannot buy is skipped: Connect closes the tunnel once the last one runs out, and Run reconnects or stops.
+func (t *tunnel) reauthLoop(ctx context.Context) {
+ tick := time.NewTicker(t.o.ReauthEvery)
+ defer tick.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-tick.C:
+ }
+ tok, err := t.o.NodeToken(ctx)
+ if err != nil {
+ t.o.Log.Warn("connect: no fresh node token", "share", t.share.ID, "err", err)
+ continue
+ }
+ if err := t.sendJSON(map[string]any{"t": "reauth", "node_token": tok}); err != nil {
+ return
+ }
+ }
+}
+
func (t *tunnel) get(id uint32) *stream {
func (t *tunnel) get(id uint32) *stream {
t.mu.Lock()
t.mu.Lock()
defer t.mu.Unlock()
defer t.mu.Unlock()
new file mode 100644
new file mode 100644
@@ -0,0 +1,142 @@
@@ -0,0 +1,142 @@
+package connect
+
+// A share served by this computer itself, as OpenVibe Node (CON-4, ADR-057 decision 4). The share's owner picks the
+// computer on the share's page; there is no connect key. Before every connection the Node buys its node token (subject
+// node:nod_…, audience openvibe.website) and asks Connect for a tunnel token for that one share:
+//
+// POST https://connect.openvibe.website/api/v1/node/shares/<shr_…>/tunnel-token Authorization: Bearer <node token>
+// → {"tunnel_token":"ovt_…","expires_in":60,"share":{"id","host","url","local_port","mode"}}
+//
+// then says hello with both, {"t":"hello","v":1,"share":"shr_…","node_token":…,"tunnel":"ovt_…",…}, and sends
+// {"t":"reauth","node_token":…} every ReauthEvery so the tunnel outlives its 5-minute token. Neither token is logged.
+//
+// The port stays this computer's: the Node serves the port it was given (or the one the share had when it was added)
+// even if the share's page says another later, so Connect can never point it at a different local service.
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/url"
+ "regexp"
+ "strings"
+ "time"
+)
+
+// ShareIDRe matches a share id on OpenVibe Connect (shr_ + a ULID).
+var ShareIDRe = regexp.MustCompile(`^shr_[0-9A-HJKMNP-TV-Z]{26}$`)
+
+// DefaultReauthEvery is how often a computer's tunnel sends a fresh node token (they last 300 s; Connect allows 30 s
+// past the expiry, and the Node's token cache replaces a token 30 s before it).
+const DefaultReauthEvery = 45 * time.Second
+
+// NodeShare is one share this computer serves, as Connect lists it.
+type NodeShare struct {
+ ID string `json:"id"`
+ Host string `json:"host"`
+ URL string `json:"url"`
+ LocalPort int `json:"local_port"`
+ Mode string `json:"mode"`
+}
+
+type problem struct {
+ Code string `json:"code"`
+ Detail string `json:"detail"`
+}
+
+// apiURL turns the tunnel URL (wss://host/tunnel) into Connect's API URL for path; plain ws only to a loopback host.
+func apiURL(server, path string) (string, error) {
+ if server == "" {
+ server = DefaultServer
+ }
+ u, err := url.Parse(server)
+ if err != nil || u.Host == "" {
+ return "", fmt.Errorf("connect: %q is not a tunnel URL", server)
+ }
+ switch u.Scheme {
+ case "wss":
+ u.Scheme = "https"
+ case "ws":
+ h := u.Hostname()
+ ip := net.ParseIP(h)
+ if h != "localhost" && (ip == nil || !ip.IsLoopback()) {
+ return "", fmt.Errorf("connect: refusing to talk to %s over plain http", u.Host)
+ }
+ u.Scheme = "http"
+ default:
+ return "", fmt.Errorf("connect: %q is not a ws:// or wss:// URL", server)
+ }
+ u.Path, u.RawQuery = path, ""
+ return u.String(), nil
+}
+
+func nodeCall(ctx context.Context, client *http.Client, method, server, path, nodeToken, userAgent string, out any) (int, error) {
+ if client == nil {
+ client = &http.Client{Timeout: 20 * time.Second}
+ }
+ target, err := apiURL(server, path)
+ if err != nil {
+ return 0, err
+ }
+ req, err := http.NewRequestWithContext(ctx, method, target, nil)
+ if err != nil {
+ return 0, err
+ }
+ req.Header.Set("Authorization", "Bearer "+nodeToken)
+ req.Header.Set("Accept", "application/json")
+ req.Header.Set("User-Agent", userAgent)
+ resp, err := client.Do(req)
+ if err != nil {
+ return 0, fmt.Errorf("connect: %s: %w", path, err)
+ }
+ defer resp.Body.Close()
+ raw, err := io.ReadAll(io.LimitReader(resp.Body, 64<<10))
+ if err != nil {
+ return resp.StatusCode, err
+ }
+ if resp.StatusCode/100 != 2 {
+ var p problem
+ _ = json.Unmarshal(raw, &p)
+ msg := p.Detail
+ if msg == "" {
+ msg = http.StatusText(resp.StatusCode)
+ }
+ return resp.StatusCode, fmt.Errorf("connect: HTTP %d %s: %s", resp.StatusCode, p.Code, msg)
+ }
+ return resp.StatusCode, json.Unmarshal(raw, out)
+}
+
+// NodeShares lists the open shares bound to this computer (GET /api/v1/node/shares).
+func NodeShares(ctx context.Context, client *http.Client, server, nodeToken, userAgent string) ([]NodeShare, error) {
+ var out struct {
+ Shares []NodeShare `json:"shares"`
+ }
+ if _, err := nodeCall(ctx, client, http.MethodGet, server, "/api/v1/node/shares", nodeToken, userAgent, &out); err != nil {
+ return nil, err
+ }
+ return out.Shares, nil
+}
+
+// tunnelToken asks for one hello's tunnel token. A share this computer does not serve (404) or a closed one (410)
+// stops it; anything else is tried again.
+func tunnelToken(ctx context.Context, o Options, server, shareID, nodeToken string) (string, NodeShare, error) {
+ var out struct {
+ Token string `json:"tunnel_token"`
+ Share NodeShare `json:"share"`
+ }
+ status, err := nodeCall(ctx, o.HTTPClient, http.MethodPost, server, "/api/v1/node/shares/"+url.PathEscape(shareID)+"/tunnel-token", nodeToken, "openvibe-node/"+o.Version, &out)
+ if err != nil {
+ if status == http.StatusNotFound || status == http.StatusGone {
+ return "", NodeShare{}, fmt.Errorf("%w: %v", ErrStopped, err)
+ }
+ return "", NodeShare{}, err
+ }
+ if !strings.HasPrefix(out.Token, "ovt_") {
+ return "", NodeShare{}, errors.New("connect: no tunnel token in the answer")
+ }
+ return out.Token, out.Share, nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,214 @@
@@ -0,0 +1,214 @@
+package connect
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/gorilla/websocket"
+)
+
+const testShareID = "shr_01K9ZZZZZZZZZZZZZZZZZZZZZZ"
+
+// nodeServer plays OpenVibe Connect for a share served by this computer (CON-4): the node API (a tunnel token per
+// hello, the computer's shares) and /tunnel, which records the hello and every reauth.
+type nodeServer struct {
+ srv *httptest.Server
+ hellos chan map[string]any
+ reauths chan string
+ tickets atomic.Int32
+ mu sync.Mutex
+ bearers []string
+ notMine bool // the tunnel-token call answers 404
+ pagePort int
+}
+
+func newNodeServer(t *testing.T) *nodeServer {
+ n := &nodeServer{hellos: make(chan map[string]any, 8), reauths: make(chan string, 16), pagePort: 8790}
+ up := websocket.Upgrader{}
+ mux := http.NewServeMux()
+ mux.HandleFunc("/api/v1/node/shares", func(w http.ResponseWriter, r *http.Request) {
+ n.mu.Lock()
+ n.bearers = append(n.bearers, r.Header.Get("Authorization"))
+ n.mu.Unlock()
+ _ = json.NewEncoder(w).Encode(map[string]any{"shares": []map[string]any{{"id": testShareID, "host": "portal--alex.openvibe.website", "url": "https://portal--alex.openvibe.website", "local_port": n.pagePort, "mode": "owner"}}})
+ })
+ mux.HandleFunc("/api/v1/node/shares/"+testShareID+"/tunnel-token", func(w http.ResponseWriter, r *http.Request) {
+ n.mu.Lock()
+ n.bearers = append(n.bearers, r.Header.Get("Authorization"))
+ n.mu.Unlock()
+ if r.Method != http.MethodPost {
+ w.WriteHeader(405)
+ return
+ }
+ if n.notMine {
+ w.Header().Set("Content-Type", "application/problem+json")
+ w.WriteHeader(404)
+ _, _ = w.Write([]byte(`{"code":"share.not_found","detail":"This computer does not serve that share."}`))
+ return
+ }
+ i := n.tickets.Add(1)
+ _ = json.NewEncoder(w).Encode(map[string]any{"tunnel_token": fmt.Sprintf("ovt_%043d", i), "expires_in": 60, "share": map[string]any{"id": testShareID, "local_port": n.pagePort}})
+ })
+ mux.HandleFunc("/tunnel", func(w http.ResponseWriter, r *http.Request) {
+ ws, err := up.Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+ defer ws.Close()
+ _, data, err := ws.ReadMessage()
+ if err != nil {
+ return
+ }
+ var hello map[string]any
+ _ = json.Unmarshal(data, &hello)
+ n.hellos <- hello
+ _ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"ready","v":1,"node":true,"share":{"id":"`+testShareID+`","host":"portal--alex.openvibe.website","url":"https://portal--alex.openvibe.website","mode":"owner","local_port":8790}}`))
+ for {
+ _, data, err := ws.ReadMessage()
+ if err != nil {
+ return
+ }
+ var m map[string]any
+ _ = json.Unmarshal(data, &m)
+ if m["t"] == "reauth" {
+ tok, _ := m["node_token"].(string)
+ n.reauths <- tok
+ _ = ws.WriteMessage(websocket.TextMessage, []byte(`{"t":"reauthed","until":"2026-11-03T00:00:00Z"}`))
+ }
+ }
+ })
+ n.srv = httptest.NewServer(mux)
+ t.Cleanup(n.srv.Close)
+ return n
+}
+
+func (n *nodeServer) url() string { return "ws" + strings.TrimPrefix(n.srv.URL, "http") + "/tunnel" }
+
+// countingTokens hands out node-token-1, node-token-2, … (a fresh one each call).
+func countingTokens() (func(context.Context) (string, error), *atomic.Int32) {
+ var n atomic.Int32
+ return func(context.Context) (string, error) { return fmt.Sprintf("node-token-%d", n.Add(1)), nil }, &n
+}
+
+func TestAShareServedByThisComputer(t *testing.T) {
+ n := newNodeServer(t)
+ tokens, _ := countingTokens()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ ready := make(chan Ready, 2)
+ done := make(chan error, 1)
+ go func() {
+ done <- Run(ctx, Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: tokens, ReauthEvery: 40 * time.Millisecond, OnReady: func(r Ready) { ready <- r }, MinWait: 20 * time.Millisecond})
+ }()
+ var hello map[string]any
+ select {
+ case hello = <-n.hellos:
+ case err := <-done:
+ t.Fatalf("Run ended: %v", err)
+ case <-time.After(5 * time.Second):
+ t.Fatal("no hello")
+ }
+ if hello["share"] != testShareID || hello["node_token"] != "node-token-1" || hello["tunnel"] != fmt.Sprintf("ovt_%043d", 1) || hello["port"].(float64) != 9100 {
+ t.Fatalf("hello %v", hello)
+ }
+ if _, has := hello["key"]; has {
+ t.Fatal("no key in a computer's hello")
+ }
+ if r := <-ready; r.Host != "portal--alex.openvibe.website" {
+ t.Fatalf("ready %v", r)
+ }
+ // It keeps the tunnel with fresh node tokens.
+ first, second := <-n.reauths, <-n.reauths
+ if first == second || !strings.HasPrefix(first, "node-token-") {
+ t.Fatalf("reauths %q %q", first, second)
+ }
+ n.mu.Lock()
+ bearer := n.bearers[0]
+ n.mu.Unlock()
+ if bearer != "Bearer node-token-1" {
+ t.Fatalf("the tunnel-token call carried %q", bearer)
+ }
+}
+
+func TestAShareThisComputerDoesNotServeStops(t *testing.T) {
+ n := newNodeServer(t)
+ n.notMine = true
+ tokens, _ := countingTokens()
+ err := Run(context.Background(), Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: tokens, MinWait: 10 * time.Millisecond})
+ if !errors.Is(err, ErrStopped) {
+ t.Fatalf("want ErrStopped, got %v", err)
+ }
+}
+
+func TestARemovedMachineStops(t *testing.T) {
+ n := newNodeServer(t)
+ revoked := func(context.Context) (string, error) { return "", fmt.Errorf("%w: refused", ErrStopped) }
+ err := Run(context.Background(), Share{ID: testShareID, Port: 9100, Server: n.url()}, Options{Version: "test", NodeToken: revoked, MinWait: 10 * time.Millisecond})
+ if !errors.Is(err, ErrStopped) {
+ t.Fatalf("want ErrStopped, got %v", err)
+ }
+ if n.tickets.Load() != 0 {
+ t.Fatal("no tunnel token without a node token")
+ }
+}
+
+func TestAShareByIDNeedsThePairing(t *testing.T) {
+ if err := Run(context.Background(), Share{ID: testShareID, Port: 9100}, Options{}); err == nil || !strings.Contains(err.Error(), "openvibe.bot/machines") {
+ t.Fatalf("got %v", err)
+ }
+ if err := Run(context.Background(), Share{ID: "shr_nope", Port: 9100}, Options{NodeToken: func(context.Context) (string, error) { return "x", nil }}); err == nil {
+ t.Fatal("a bad id")
+ }
+}
+
+func TestNodeSharesAndTheAPIURL(t *testing.T) {
+ n := newNodeServer(t)
+ got, err := NodeShares(context.Background(), nil, n.url(), "node-token-9", "test")
+ if err != nil || len(got) != 1 || got[0].ID != testShareID || got[0].LocalPort != 8790 {
+ t.Fatalf("%v %v", got, err)
+ }
+ for in, want := range map[string]string{
+ "wss://connect.openvibe.website/tunnel": "https://connect.openvibe.website/x",
+ "": "https://connect.openvibe.website/x",
+ "ws://127.0.0.1:5240/tunnel": "http://127.0.0.1:5240/x",
+ } {
+ if u, err := apiURL(in, "/x"); err != nil || u != want {
+ t.Fatalf("%q → %q %v", in, u, err)
+ }
+ }
+ if _, err := apiURL("ws://evil.example/tunnel", "/x"); err == nil {
+ t.Fatal("plain http to another host")
+ }
+}
+
+func TestSavedSharesByID(t *testing.T) {
+ dir := t.TempDir()
+ if err := Add(dir, Share{ID: testShareID, Port: 8790}); err != nil {
+ t.Fatal(err)
+ }
+ if err := Add(dir, Share{Key: "ovc_" + strings.Repeat("K", 43), Port: 8791}); err != nil {
+ t.Fatal(err)
+ }
+ if err := Add(dir, Share{ID: testShareID, Port: 8792}); err != nil { // the same id again replaces it
+ t.Fatal(err)
+ }
+ got, _ := Load(dir)
+ if len(got) != 2 {
+ t.Fatalf("%v", got)
+ }
+ if k, _ := Remove(dir, testShareID); k != 1 {
+ t.Fatalf("removed %d", k)
+ }
+ if got, _ := Load(dir); len(got) != 1 || got[0].Port != 8791 {
+ t.Fatalf("%v", got)
+ }
+}
@@ -53,7 +53,7 @@ func save(dir string, shares []Share) error {
@@ -53,7 +53,7 @@ func save(dir string, shares []Share) error {
return os.Rename(tmp, filepath.Join(dir, File))
return os.Rename(tmp, filepath.Join(dir, File))
}
}
-// Add saves a share (the same key again replaces its port and name).
+// Add saves a share (the same key, or the same share id, again replaces its port and name).
func Add(dir string, s Share) error {
func Add(dir string, s Share) error {
shares, err := Load(dir)
shares, err := Load(dir)
if err != nil {
if err != nil {
@@ -61,14 +61,14 @@ func Add(dir string, s Share) error {
@@ -61,14 +61,14 @@ func Add(dir string, s Share) error {
}
}
out := shares[:0]
out := shares[:0]
for _, x := range shares {
for _, x := range shares {
- if x.Key != s.Key {
+ if (s.Key == "" || x.Key != s.Key) && (s.ID == "" || x.ID != s.ID) {
out = append(out, x)
out = append(out, x)
}
}
}
}
return save(dir, append(out, s))
return save(dir, append(out, s))
}
}
-// Remove forgets the share with this key, or every share on this port; it reports how many it removed.
+// Remove forgets the share with this key or id, or every share on this port; it reports how many it removed.
func Remove(dir string, keyOrPort string) (int, error) {
func Remove(dir string, keyOrPort string) (int, error) {
shares, err := Load(dir)
shares, err := Load(dir)
if err != nil {
if err != nil {
@@ -77,7 +77,7 @@ func Remove(dir string, keyOrPort string) (int, error) {
@@ -77,7 +77,7 @@ func Remove(dir string, keyOrPort string) (int, error) {
out := shares[:0]
out := shares[:0]
n := 0
n := 0
for _, x := range shares {
for _, x := range shares {
- if x.Key == keyOrPort || fmt.Sprint(x.Port) == keyOrPort || (x.Name != "" && x.Name == keyOrPort) {
+ if (x.Key != "" && x.Key == keyOrPort) || (x.ID != "" && x.ID == keyOrPort) || fmt.Sprint(x.Port) == keyOrPort || (x.Name != "" && x.Name == keyOrPort) {
n++
n++
continue
continue
}
}
@@ -126,7 +126,7 @@ func RunAll(ctx context.Context, dir string, o Options) {
@@ -126,7 +126,7 @@ func RunAll(ctx context.Context, dir string, o Options) {
defer wg.Done()
defer wg.Done()
err := Run(ctx, s, so)
err := Run(ctx, s, so)
if errors.Is(err, ErrStopped) && o.Log != nil {
if errors.Is(err, ErrStopped) && o.Log != nil {
- o.Log.Warn("connect: share stopped; remove it with `openvibe-node share remove`", "port", s.Port, "key", Redact(s.Key))
+ o.Log.Warn("connect: share stopped; remove it with `openvibe-node share remove`", "port", s.Port, "share", s.Label())
}
}
}()
}()
}
}
@@ -151,8 +151,17 @@ type TokenSource struct {
@@ -151,8 +151,17 @@ type TokenSource struct {
// NewTokenSource returns the token source of a Network-paired machine.
// NewTokenSource returns the token source of a Network-paired machine.
func NewTokenSource(c *credentials.Credentials, client *http.Client, userAgent string) *TokenSource {
func NewTokenSource(c *credentials.Credentials, client *http.Client, userAgent string) *TokenSource {
+ return NewTokenSourceFor(c, client, userAgent, Audience(c))
+}
+
+// ConnectAudience is OpenVibe Connect's: any person's machine may buy a node token for it, whichever service paired
+// the machine (Network, CON-4), to serve the shares bound to it.
+const ConnectAudience = "openvibe.website"
+
+// NewTokenSourceFor returns a token source for another audience the machine may speak to (ConnectAudience).
+func NewTokenSourceFor(c *credentials.Credentials, client *http.Client, userAgent, audience string) *TokenSource {
return &TokenSource{network: strings.TrimSuffix(c.Network, "/"), principal: c.Principal, credential: c.NodeCredential,
return &TokenSource{network: strings.TrimSuffix(c.Network, "/"), principal: c.Principal, credential: c.NodeCredential,
- audience: Audience(c), client: defaultClient(client), userAgent: userAgent, now: time.Now}
+ audience: audience, client: defaultClient(client), userAgent: userAgent, now: time.Now}
}
}
// Invalidate drops the cached token, so the next Token fetches a fresh one (after a 401 from Bot, before a reauth).
// Invalidate drops the cached token, so the next Token fetches a fresh one (after a 401 from Bot, before a reauth).