openvibers/openvibe.link Commits
A11: send cloud browsing through the owner's Link (#143)
Changes
@@@ -5,12 +5,15 @@ import ("flag""fmt""io"++ "net""os""os/signal""strings""syscall""github.com/OpenVibers/OpenVibe.Node/internal/config"++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ linkegress "github.com/OpenVibers/OpenVibe.Node/internal/egress""github.com/OpenVibers/OpenVibe.Node/internal/worker")@@@ -45,6 +48,7 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {diskMB := fs.Int64("disk-mb", 512, "the size of the desk's /tmp, where its profile lives, in MiB")pids := fs.Int64("pids", 256, "the most processes and threads of the desk")egress := fs.String("egress", "", `"public" to reach the public internet (never the host's networks); none by default`)++ linkDesk := fs.String("link-egress-desk", "", "the cloud desk assigned by Run")var binds, env listFlagfs.Var(&binds, "bind", "a read-only path the program needs (repeatable): its install directory, /etc/fonts")fs.Var(&env, "setenv", "NAME=VALUE in the program's environment (repeatable); it gets nothing else")@@@ -75,7 +79,33 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {}ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP)defer stop()-- code, err := worker.RunDesk(ctx, wcfg, worker.DeskSpec{Argv: argv, Env: env, Binds: binds,++ var proxy func(string) (string, func(), error)++ if *linkDesk != "" {++ if !linkegress.IsCloudDesk(*linkDesk) {++ fmt.Fprintln(stderr, "invalid cloud desk")++ return 125++ }++ creds, e := credentials.Load(paths.CredentialFile(), nil)++ if e != nil || !creds.NetworkPaired() {++ fmt.Fprintln(stderr, "computer sign-in unavailable")++ return 125++ }++ plan, e := linkegress.NewClient(creds).Desk(ctx, *linkDesk)++ if e != nil {++ fmt.Fprintln(stderr, "browsing connection unavailable")++ return 125++ }++ proxy = func(gateway string) (string, func(), error) {++ listener, e := net.Listen("tcp", net.JoinHostPort(gateway, "0"))++ if e != nil {++ return "", nil, e++ }++ proxyCtx, cancelProxy := context.WithCancel(ctx)++ server := linkegress.Bridge(proxyCtx, listener, plan)++ return listener.Addr().String(), func() { cancelProxy(); _ = server.Close() }, nil++ }++ }++ code, err := worker.RunDesk(ctx, wcfg, worker.DeskSpec{Argv: argv, Env: env, Binds: binds, Proxy: proxy,MemBytes: *memMB << 20, CPUMillis: *cpu, DiskBytes: *diskMB << 20, Pids: *pids, Egress: *egress,In: in, Out: out, Stderr: os.Stderr})if err != nil && ctx.Err() == nil {nnew file mode 100644@@@ -0,0 +1,31 @@++# Browsing through your own Link++++Link automatically offers an authenticated browsing endpoint for its owner's cloud sessions after pairing.++The switch, offline choice, monthly limit (20 GB by default) and site exceptions are on openvibe.link.++Only public web destinations on TCP ports 80 and 443 are allowed. DNS is resolved and pinned on the Link device;++private, local, metadata and IPv6 translation addresses are refused.++++Connect carries a WebSocket byte stream through its existing share relay. The stream contains HTTP CONNECT,++ordinary HTTP proxy requests, or authenticated SOCKS5 CONNECT. Both Connect and Node ask Network about every++new stream. Network compares the current device owner with the Run session owner, checks the credential hash,++expiry, opt-out and cap. A cookie or an ordinary share API key never grants access to the browsing endpoint.++Node checks active streams again every five seconds, closing them on expiry, revocation, opt-out or a reported++battery/metered state. OS readings are best effort: Linux battery/NetworkManager, macOS battery, Windows battery/network cost.++++The shared desk plugin adds only the desk name to `desk-sandbox`. The Node authenticates to Network, reads the++session assigned to its own machine, and gets a fresh credential kept in its proxy process. Chrome never receives++that credential. Its namespace reaches one gateway proxy socket only; all direct forwarding is refused. Chrome++also disables direct DNS, QUIC and non-proxied WebRTC UDP, with the loopback bypass disabled.++++The device persists bandwidth reservations before forwarding bytes, shared across its sessions and directions.++Reservations may overcount after a failed write or crash; restarting Link does not reopen the monthly allowance.++The allowance is per Link device. A damaged usage file refuses service until repaired by the owner.++++Run must renew authorization within two minutes and revoke it when ending the desk. A missing or unavailable++Network never means direct fallback. The saved offline choice applies when the Link connection cannot be made:++use OpenVibe's address or wait. Explicit site exceptions use OpenVibe's address and public-only DNS there.++++Deployment order: Network's expand migration and routes, Connect's expand migration and route, Node including++the updated computer plugin on cloud desk hosts and home devices, then Run. Existing cloud hosts must be upgraded++before enabling Run's integration. The integration applies to browser desks created through Run; providers must++use that session path to inherit the policy. No blind-fill files are modified.@@@ -71,14 +71,15 @@ var ErrStopped = errors.New("connect: share stopped by the server")// Options tune a client; the zero value is right for production.type Options struct {-- Log *slog.Logger-- Version string-- LocalHost string // default "localhost"-- OnReady func(Ready) // called after each successful hello-- OnRequest func(method, path string, status int, d time.Duration)-- Dialer *websocket.Dialer-- MinWait time.Duration // reconnect backoff floor (default 1 s)-- MaxWait time.Duration // reconnect backoff ceiling (default 30 s)++ Log *slog.Logger++ Version string++ LocalHost string // default "localhost"++ OnDisconnect func()++ OnReady func(Ready) // called after each successful hello++ OnRequest func(method, path string, status int, d time.Duration)++ Dialer *websocket.Dialer++ MinWait 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.link, 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.@@@ -143,6 +144,9 @@ func Run(ctx context.Context, s Share, o Options) error {wait := o.MinWaitfor {err := runOnce(ctx, s, o, func() { wait = o.MinWait })++ if o.OnDisconnect != nil {++ o.OnDisconnect()++ }if ctx.Err() != nil {return ctx.Err()}nnew file mode 100644@@@ -0,0 +1,106 @@++package egress++++import (++ "bufio"++ "context"++ "encoding/json"++ "io"++ "net"++ "net/http"++ "net/http/httptest"++ "path/filepath"++ "strings"++ "sync/atomic"++ "testing"++ "time"++)++++func echoDial(_ context.Context, _ string) (net.Conn, error) {++ a, b := net.Pipe()++ go func() { defer b.Close(); _, _ = io.Copy(b, b) }()++ return a, nil++}++func proxyConnect(t *testing.T, address string) (net.Conn, *bufio.Reader, *http.Response) {++ t.Helper()++ c, err := net.DialTimeout("tcp", address, time.Second)++ if err != nil {++ t.Fatal(err)++ }++ _ = c.SetDeadline(time.Now().Add(3 * time.Second))++ _, _ = io.WriteString(c, "CONNECT web.example:443 HTTP/1.1\r\nHost: web.example:443\r\n\r\n")++ br := bufio.NewReader(c)++ r, err := http.ReadResponse(br, nil)++ if err != nil {++ c.Close()++ t.Fatal(err)++ }++ return c, br, r++}++func TestBridgeThroughNodeAndCapClosesTheTunnel(t *testing.T) {++ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {++ var b map[string]string++ _ = json.NewDecoder(r.Body).Decode(&b)++ if b["credential"] != strings.Repeat("c", 43) {++ w.WriteHeader(403)++ return++ }++ _ = json.NewEncoder(w).Encode(Grant{Session: "s", Node: "n", Enabled: true, Cap: 4, Expires: time.Now().Add(time.Minute).UnixMilli()})++ }))++ defer network.Close()++ budget, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))++ targets := make(chan string, 1)++ endpoint := httptest.NewServer(&Endpoint{Client: &Client{Base: network.URL}, Node: "n", Budget: budget, Pause: func() bool { return false }, Dial: func(ctx context.Context, s string) (net.Conn, error) { targets <- s; return echoDial(ctx, s) }})++ defer endpoint.Close()++ listener, err := net.Listen("tcp", "127.0.0.1:0")++ if err != nil {++ t.Fatal(err)++ }++ ctx, cancel := context.WithCancel(context.Background())++ defer cancel()++ server := Bridge(ctx, listener, Plan{Mode: "link", Relay: "ws" + strings.TrimPrefix(endpoint.URL, "http") + "/__ov_egress", Session: "s", Credential: strings.Repeat("c", 43), Offline: "wait"})++ defer server.Close()++ conn, br, response := proxyConnect(t, listener.Addr().String())++ defer conn.Close()++ if response.StatusCode != 200 {++ t.Fatal("bridge failed")++ }++ if <-targets != "web.example:443" {++ t.Fatal("destination was resolved before reaching Node")++ }++ _, _ = conn.Write([]byte("ab"))++ echo := make([]byte, 2)++ if _, err = io.ReadFull(br, echo); err != nil || string(echo) != "ab" {++ t.Fatal("bridge did not carry bytes")++ }++ // Two bytes each way consumed the whole allowance; no fifth byte may leave the device.++ _, _ = conn.Write([]byte("x"))++ if _, err = br.ReadByte(); err == nil {++ t.Fatal("spent bandwidth did not close tunnel")++ }++}++func TestBridgeOfflineChoicesAndAuthorizationFailures(t *testing.T) {++ for _, tc := range []struct {++ name, choice string++ status, want int++ direct bool++ }{{"offline fallback", "openvibe", 503, 200, true}, {"offline wait", "wait", 503, 503, false}, {"authorization refused", "openvibe", 403, 503, false}} {++ t.Run(tc.name, func(t *testing.T) {++ remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(tc.status) }))++ defer remote.Close()++ ctx, cancel := context.WithCancel(context.Background())++ defer cancel()++ listener, err := net.Listen("tcp", "127.0.0.1:0")++ if err != nil {++ t.Fatal(err)++ }++ var used atomic.Bool++ server := Bridge(ctx, listener, Plan{Mode: "link", Relay: "ws" + strings.TrimPrefix(remote.URL, "http"), Offline: tc.choice}, func(ctx context.Context, s string) (net.Conn, error) { used.Store(true); return echoDial(ctx, s) })++ defer server.Close()++ conn, _, response := proxyConnect(t, listener.Addr().String())++ conn.Close()++ if response.StatusCode != tc.want || used.Load() != tc.direct {++ t.Fatalf("status %d, direct %v", response.StatusCode, used.Load())++ }++ })++ }++}nnew file mode 100644@@@ -0,0 +1,81 @@++package egress++++import (++ "encoding/json"++ "errors"++ "os"++ "path/filepath"++ "runtime"++ "sync"++ "time"++)++++// Reserve before forwarding bytes. A crash can overcount, never reopen spent bandwidth.++// One Budget belongs to the running Link, shared by all its sessions and both directions.++type Budget struct {++ mu sync.Mutex++ path string++ Month string `json:"month"`++ Used int64 `json:"used"`++ Now func() time.Time `json:"-"`++}++++func NewBudget(path string) (*Budget, error) {++ b := &Budget{path: path, Now: time.Now}++ raw, err := os.ReadFile(path)++ if err != nil && !os.IsNotExist(err) {++ return nil, err++ }++ if err == nil {++ if err = json.Unmarshal(raw, b); err != nil || b.Used < 0 {++ return nil, errors.New("browsing usage could not be read")++ }++ }++ return b, nil++}++func (b *Budget) Reserve(n, cap int64) error {++ b.mu.Lock()++ defer b.mu.Unlock()++ month := b.Now().UTC().Format("2006-01")++ if month > b.Month {++ b.Month, b.Used = month, 0++ } // a clock moving back cannot reset the cap++ if n < 0 || cap < 0 || n > cap-b.Used || (n == 0 && b.Used >= cap) {++ return errors.New("monthly browsing limit reached")++ }++ b.Used += n++ raw, err := json.Marshal(b)++ if err != nil {++ return err++ }++ if err = os.MkdirAll(filepath.Dir(b.path), 0700); err != nil {++ return err++ }++ f, err := os.OpenFile(b.path+".tmp", os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)++ if err != nil {++ return err++ }++ _, err = f.Write(raw)++ if err == nil {++ err = f.Sync()++ }++ closeErr := f.Close()++ if err != nil {++ return err++ }++ if closeErr != nil {++ return closeErr++ }++ if err = os.Rename(b.path+".tmp", b.path); err != nil {++ return err++ }++ if runtime.GOOS != "windows" {++ dir, err := os.Open(filepath.Dir(b.path))++ if err != nil {++ return err++ }++ defer dir.Close()++ return dir.Sync()++ }++ return nil++}nnew file mode 100644@@@ -0,0 +1,113 @@++// Package egress carries one person's cloud browsing through their own paired Link.++package egress++++import (++ "bytes"++ "context"++ "encoding/json"++ "errors"++ "io"++ "net/http"++ "net/url"++ "strings"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/link"++)++++type Tokens interface {++ Token(context.Context) (credentials.Secret, error)++}++type Client struct {++ Base string++ Tokens Tokens++ HTTP *http.Client++}++type Plan struct {++ Mode string `json:"mode"`++ Session string `json:"session_id"`++ Credential string `json:"credential"`++ Device string `json:"device"`++ DeviceName string `json:"device_name"`++ Relay string `json:"relay_url"`++ Enabled bool `json:"egress_via_link"`++ Offline string `json:"offline_choice"`++ Cap int64 `json:"monthly_bytes"`++ Excluded []string `json:"excluded_sites"`++}++type Grant struct {++ Session string `json:"session_id"`++ Node string `json:"node_id"`++ Expires int64 `json:"expires_at"`++ Cap int64 `json:"monthly_bytes"`++ Enabled bool `json:"egress_via_link"`++}++++func NewClient(c *credentials.Credentials) *Client {++ return &Client{Base: strings.TrimRight(c.Network, "/"), Tokens: link.NewTokenSourceFor(c, nil, "openvibe-link/egress", "openvibe.network")}++}++func (c *Client) Call(ctx context.Context, method, path string, body, out any, authenticated bool) error {++ var data io.Reader++ if body != nil {++ b, err := json.Marshal(body)++ if err != nil {++ return err++ }++ data = bytes.NewReader(b)++ }++ req, err := http.NewRequestWithContext(ctx, method, c.Base+path, data)++ if err != nil {++ return errors.New("browsing connection unavailable")++ }++ req.Header.Set("Content-Type", "application/json")++ if authenticated {++ token, err := c.Tokens.Token(ctx)++ if err != nil {++ return errors.New("computer sign-in unavailable")++ }++ req.Header.Set("Authorization", "Bearer "+token.Reveal())++ }++ client := c.HTTP++ if client == nil {++ client = &http.Client{Timeout: 10 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}++ }++ resp, err := client.Do(req)++ if err != nil {++ return errors.New("browsing connection unavailable")++ }++ defer resp.Body.Close()++ if resp.StatusCode/100 != 2 {++ return errors.New("browsing connection refused")++ }++ if out == nil {++ return nil++ }++ return json.NewDecoder(io.LimitReader(resp.Body, 32<<10)).Decode(out)++}++func (c *Client) Open(ctx context.Context, session, credential, node string) (Grant, error) {++ var grant Grant++ err := c.Call(ctx, "POST", "/api/v1/link-egress/open", map[string]string{"session_id": session, "credential": credential, "node_id": node}, &grant, false)++ if err == nil && (grant.Session != session || grant.Node != node || !grant.Enabled || grant.Expires <= time.Now().UnixMilli()) {++ err = errors.New("browsing connection refused")++ }++ return grant, err++}++func (c *Client) Desk(ctx context.Context, desk string) (Plan, error) {++ var p Plan++ err := c.Call(ctx, "GET", "/api/v1/node/self/egress/desks/"+url.PathEscape(desk), nil, &p, true)++ if err != nil {++ return p, err++ }++ if p.Mode == "wait" {++ return p, errors.New("waiting for your computer")++ }++ if p.Mode == "direct" {++ return p, nil++ }++ u, err := url.Parse(p.Relay)++ if err != nil || p.Mode != "link" || u.Scheme != "wss" || !strings.HasSuffix(u.Hostname(), ".openvibe.link") || u.Port() != "" || u.User != nil || u.Path != "/__ov_egress" || u.RawQuery != "" || len(p.Credential) != 43 {++ return p, errors.New("invalid browsing connection")++ }++ return p, nil++}nnew file mode 100644@@@ -0,0 +1,58 @@++package egress++++import (++ "context"++ "errors"++ "net"++ "net/netip"++ "strconv"++ "time"++)++++var blocked = []netip.Prefix{++ netip.MustParsePrefix("0.0.0.0/8"), netip.MustParsePrefix("100.64.0.0/10"), netip.MustParsePrefix("192.0.0.0/24"),++ netip.MustParsePrefix("192.0.2.0/24"), netip.MustParsePrefix("192.88.99.0/24"), netip.MustParsePrefix("198.18.0.0/15"),++ netip.MustParsePrefix("198.51.100.0/24"), netip.MustParsePrefix("203.0.113.0/24"), netip.MustParsePrefix("240.0.0.0/4"),++ netip.MustParsePrefix("64:ff9b::/96"), netip.MustParsePrefix("64:ff9b:1::/48"), netip.MustParsePrefix("2001::/32"), netip.MustParsePrefix("2002::/16"), netip.MustParsePrefix("2001:db8::/32"),++}++++func Public(ip netip.Addr) bool {++ ip = ip.Unmap()++ if !ip.IsValid() || !ip.IsGlobalUnicast() || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.Zone() != "" {++ return false++ }++ for _, p := range blocked {++ if p.Contains(ip) {++ return false++ }++ }++ return true++}++++type DialFunc func(context.Context, string) (net.Conn, error)++++func PublicDial(ctx context.Context, address string) (net.Conn, error) {++ return checkedDial(ctx, address, net.DefaultResolver.LookupIPAddr, (&net.Dialer{Timeout: 10 * time.Second}).DialContext)++}++func checkedDial(ctx context.Context, address string, resolve func(context.Context, string) ([]net.IPAddr, error), dial func(context.Context, string, string) (net.Conn, error)) (net.Conn, error) {++ host, port, err := net.SplitHostPort(address)++ if err != nil {++ return nil, err++ }++ p, err := strconv.Atoi(port)++ if err != nil || (p != 80 && p != 443) {++ return nil, errors.New("only web ports are allowed")++ }++ // Resolve on the person's Node and pin the checked address for the actual dial.++ ips, err := resolve(ctx, host)++ if err != nil || len(ips) == 0 {++ return nil, errors.New("site address unavailable")++ }++ for _, ip := range ips {++ a, ok := netip.AddrFromSlice(ip.IP)++ if !ok || !Public(a) {++ return nil, errors.New("private site address refused")++ }++ }++ return dial(ctx, "tcp", net.JoinHostPort(ips[0].IP.String(), port))++}nnew file mode 100644@@@ -0,0 +1,201 @@++package egress++++import (++ "bufio"++ "context"++ "encoding/base64"++ "encoding/json"++ "io"++ "net"++ "net/http"++ "net/http/httptest"++ "net/netip"++ "path/filepath"++ "strings"++ "sync"++ "sync/atomic"++ "testing"++ "time"++++ "github.com/gorilla/websocket"++)++++func TestBudgetPersistsAndSharesCap(t *testing.T) {++ path := filepath.Join(t.TempDir(), "usage.json")++ b, err := NewBudget(path)++ if err != nil {++ t.Fatal(err)++ }++ var ok atomic.Int64++ var wg sync.WaitGroup++ for i := 0; i < 20; i++ {++ wg.Add(1)++ go func() {++ defer wg.Done()++ if b.Reserve(10, 100) == nil {++ ok.Add(1)++ }++ }()++ }++ wg.Wait()++ if ok.Load() != 10 {++ t.Fatalf("cap allowed %d chunks", ok.Load())++ }++ b, err = NewBudget(path)++ if err != nil {++ t.Fatal(err)++ }++ if b.Reserve(1, 100) == nil {++ t.Fatal("restart reset the cap")++ }++ b.Now = func() time.Time { return time.Now().AddDate(0, 1, 0) }++ if b.Reserve(10, 100) != nil {++ t.Fatal("month did not reset")++ }++ b.Now = time.Now++ if b.Reserve(91, 100) == nil {++ t.Fatal("clock reversal reset the cap")++ }++}++func TestDNSRejectsMixedAnswersAndTranslationRanges(t *testing.T) {++ for _, s := range []string{"127.0.0.1", "10.0.0.1", "169.254.169.254", "::1", "64:ff9b::7f00:1", "2002:7f00:1::", "2001::1", "::ffff:192.168.1.1"} {++ if Public(netip.MustParseAddr(s)) {++ t.Fatalf("allowed %s", s)++ }++ }++ called := false++ _, err := checkedDial(context.Background(), "site.example:443", func(context.Context, string) ([]net.IPAddr, error) {++ return []net.IPAddr{{IP: net.ParseIP("8.8.8.8")}, {IP: net.ParseIP("10.1.1.1")}}, nil++ }, func(context.Context, string, string) (net.Conn, error) { called = true; return nil, nil })++ if err == nil || called {++ t.Fatal("mixed DNS answer was dialed")++ }++ var address string++ _, err = checkedDial(context.Background(), "site.example:443", func(context.Context, string) ([]net.IPAddr, error) {++ return []net.IPAddr{{IP: net.ParseIP("8.8.8.8")}}, nil++ }, func(_ context.Context, _, a string) (net.Conn, error) { address = a; return nil, nil })++ if err != nil || address != "8.8.8.8:443" {++ t.Fatal("DNS result not pinned")++ }++}++func TestMandatoryBrowserFlags(t *testing.T) {++ flags := strings.Join(BrowserArgs("169.254.240.1:12345"), " ")++ for _, s := range []string{"--proxy-server=http://169.254.240.1:12345", "--proxy-bypass-list=<-loopback>", "MAP * ~NOTFOUND", "disable_non_proxied_udp", "--disable-quic", "DnsOverHttps"} {++ if !strings.Contains(flags, s) {++ t.Fatalf("missing %s", s)++ }++ }++}++++func TestAuthenticatedCONNECTAndSOCKSWithRemoteDNS(t *testing.T) {++ const session = "ses_test"++ credential := strings.Repeat("a", 43)++ var checks atomic.Int64++ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {++ var b map[string]string++ _ = json.NewDecoder(r.Body).Decode(&b)++ if b["credential"] != credential {++ w.WriteHeader(403)++ return++ }++ checks.Add(1)++ _ = json.NewEncoder(w).Encode(Grant{Session: session, Node: "nod_test", Enabled: true, Cap: 100000, Expires: time.Now().Add(time.Minute).UnixMilli()})++ }))++ defer network.Close()++ budget, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))++ targets := make(chan string, 4)++ endpoint := &Endpoint{Client: &Client{Base: network.URL}, Node: "nod_test", Budget: budget, Pause: func() bool { return false }, Dial: func(_ context.Context, address string) (net.Conn, error) {++ targets <- address++ a, b := net.Pipe()++ go func() { defer b.Close(); _, _ = io.Copy(b, b) }()++ return a, nil++ }}++ server := httptest.NewServer(endpoint)++ defer server.Close()++ connect := func(pass string) (*wsConn, *http.Response, error) {++ ws, r, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http")+"/__ov_egress", http.Header{"X-OV-Session": {session}, "X-OV-Egress": {pass}})++ if err != nil {++ return nil, r, err++ }++ return &wsConn{Conn: ws}, r, nil++ }++ if _, r, err := connect("wrong"); err == nil || r.StatusCode != 403 {++ t.Fatal("wrong session credential accepted")++ }++ c, _, err := connect(credential)++ if err != nil {++ t.Fatal(err)++ }++ _, _ = io.WriteString(c, "CONNECT site.example:443 HTTP/1.1\r\nHost: site.example:443\r\nProxy-Authorization: Basic "+base64.StdEncoding.EncodeToString([]byte(session+":"+credential))+"\r\n\r\n")++ br := bufio.NewReader(c)++ resp, err := http.ReadResponse(br, nil)++ if err != nil || resp.StatusCode != 200 {++ t.Fatalf("CONNECT failed: %v", err)++ }++ _, _ = c.Write([]byte("hello"))++ b := make([]byte, 5)++ if _, err = io.ReadFull(br, b); err != nil || string(b) != "hello" {++ t.Fatal("CONNECT bytes lost")++ }++ c.Close()++ if <-targets != "site.example:443" {++ t.Fatal("resolved before reaching Node")++ }++ c, _, err = connect(credential)++ if err != nil {++ t.Fatal(err)++ }++ _, _ = c.Write([]byte{5, 1, 2})++ answer := make([]byte, 2)++ _, _ = io.ReadFull(c, answer)++ if answer[1] != 2 {++ t.Fatal("SOCKS auth missing")++ }++ auth := append([]byte{1, byte(len(session))}, []byte(session)...)++ auth = append(auth, byte(len(credential)))++ auth = append(auth, []byte(credential)...)++ _, _ = c.Write(auth)++ _, _ = io.ReadFull(c, answer)++ if answer[1] != 0 {++ t.Fatal("SOCKS auth refused")++ }++ host := "site.example"++ request := append([]byte{5, 1, 0, 3, byte(len(host))}, []byte(host)...)++ request = append(request, 1, 187)++ _, _ = c.Write(request)++ reply := make([]byte, 10)++ if _, err = io.ReadFull(c, reply); err != nil || reply[1] != 0 {++ t.Fatal("SOCKS refused")++ }++ c.Close()++ if <-targets != "site.example:443" {++ t.Fatal("SOCKS DNS not left to Node")++ }++ if checks.Load() != 2 {++ t.Fatalf("every open must check Network, got %d", checks.Load())++ }++}++++func TestCapAndPowerRefuseTunnel(t *testing.T) {++ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {++ _ = json.NewEncoder(w).Encode(Grant{Session: "s", Node: "n", Enabled: true, Cap: 10, Expires: time.Now().Add(time.Minute).UnixMilli()})++ }))++ defer network.Close()++ b, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))++ if err := b.Reserve(10, 10); err != nil {++ t.Fatal(err)++ }++ // Spent cap closes as soon as data would be forwarded; a cap lowered below existing usage refuses even an open.++ if b.Reserve(1, 10) == nil {++ t.Fatal("cap did not stop bytes")++ }++ e := &Endpoint{Client: &Client{Base: network.URL}, Node: "n", Budget: b, Pause: func() bool { return true }}++ r := httptest.NewRequest("GET", "http://local/__ov_egress", nil)++ r.Header.Set("X-OV-Session", "s")++ r.Header.Set("X-OV-Egress", "test")++ w := httptest.NewRecorder()++ e.ServeHTTP(w, r)++ if w.Code != 403 {++ t.Fatal("battery or metered state did not pause")++ }++}nnew file mode 100644@@@ -0,0 +1,17 @@++package egress++++import (++ "os/exec"++ "testing"++)++++func TestCloudPluginStartup(t *testing.T) {++ python, err := exec.LookPath("python3")++ if err != nil {++ t.Skip("Python not installed")++ }++ out, err := exec.Command(python, "../../plugins/computer/tests/test_link_egress.py").CombinedOutput()++ if err != nil {++ t.Fatalf("cloud plugin: %v\n%s", err, out)++ }++}nnew file mode 100644@@@ -0,0 +1,44 @@++package egress++++import (++ "context"++ "os"++ "os/exec"++ "path/filepath"++ "runtime"++ "strings"++ "time"++)++++// Unknown OS readings do not invent a metered/battery state. Known readings pause immediately.++func Paused() bool {++ if runtime.GOOS == "linux" {++ paths, _ := filepath.Glob("/sys/class/power_supply/*/status")++ for _, p := range paths {++ b, _ := os.ReadFile(p)++ if strings.TrimSpace(string(b)) == "Discharging" {++ return true++ }++ }++ out := probe("nmcli", "-t", "-f", "GENERAL.METERED", "device", "show")++ for _, line := range strings.Split(out, "\n") {++ _, v, _ := strings.Cut(line, ":")++ if strings.HasPrefix(strings.TrimSpace(v), "yes") {++ return true++ }++ }++ }++ if runtime.GOOS == "darwin" {++ return strings.Contains(probe("pmset", "-g", "batt"), "'Battery Power'")++ }++ if runtime.GOOS == "windows" {++ return strings.TrimSpace(probe("powershell.exe", "-NoProfile", "-NonInteractive", "-Command", `Add-Type -AssemblyName System.Windows.Forms; if ([System.Windows.Forms.SystemInformation]::PowerStatus.PowerLineStatus -eq 'Offline') {'pause'; exit}; $p = [Windows.Networking.Connectivity.NetworkInformation, Windows.Networking.Connectivity, ContentType=WindowsRuntime]::GetInternetConnectionProfile(); if ($p) {$c=$p.GetConnectionCost(); if ($c.NetworkCostType -eq 'Fixed' -or $c.NetworkCostType -eq 'Variable' -or $c.Roaming -or $c.OverDataLimit) {'pause'}}`)) == "pause"++ }++ return false++}++func probe(name string, args ...string) string {++ ctx, cancel := context.WithTimeout(context.Background(), time.Second)++ defer cancel()++ b, _ := exec.CommandContext(ctx, name, args...).Output()++ return string(b)++}nnew file mode 100644@@@ -0,0 +1,419 @@++package egress++++import (++ "bufio"++ "bytes"++ "context"++ "crypto/subtle"++ "encoding/base64"++ "encoding/binary"++ "errors"++ "io"++ "net"++ "net/http"++ "strconv"++ "strings"++ "sync"++ "sync/atomic"++ "time"++++ "github.com/gorilla/websocket"++)++++// The WebSocket is the byte pipe already supported by Connect; no destination or credential is in its URL.++type wsConn struct {++ *websocket.Conn++ reader io.Reader++ mu sync.Mutex++}++++func (c *wsConn) Read(b []byte) (int, error) {++ for {++ if c.reader == nil {++ kind, r, err := c.NextReader()++ if err != nil {++ return 0, err++ }++ if kind != websocket.BinaryMessage {++ return 0, errors.New("binary stream required")++ }++ c.reader = r++ }++ n, err := c.reader.Read(b)++ if err == io.EOF {++ c.reader = nil++ if n > 0 {++ return n, nil++ }++ continue++ }++ return n, err++ }++}++func (c *wsConn) Write(b []byte) (int, error) {++ c.mu.Lock()++ defer c.mu.Unlock()++ if err := c.WriteMessage(websocket.BinaryMessage, b); err != nil {++ return 0, err++ }++ return len(b), nil++}++func (c *wsConn) LocalAddr() net.Addr { return c.UnderlyingConn().LocalAddr() }++func (c *wsConn) RemoteAddr() net.Addr { return c.UnderlyingConn().RemoteAddr() }++func (c *wsConn) SetDeadline(t time.Time) error {++ _ = c.SetReadDeadline(t)++ return c.SetWriteDeadline(t)++}++++type Endpoint struct {++ Client *Client++ Node string++ Budget *Budget++ Dial DialFunc++ Pause func() bool++}++++func (e *Endpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {++ if r.URL.Path != "/__ov_egress" || r.Method != "GET" {++ http.Error(w, "refused", 403)++ return++ }++ session, credential := r.Header.Get("X-OV-Session"), r.Header.Get("X-OV-Egress")++ grant, err := e.Client.Open(r.Context(), session, credential, e.Node)++ if err != nil || e.Pause() || e.Budget.Reserve(0, grant.Cap) != nil {++ http.Error(w, "refused", 403)++ return++ }++ ws, err := (&websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return r.Header.Get("Origin") == "" }}).Upgrade(w, r, nil)++ if err != nil {++ return++ }++ ws.SetReadLimit(64 << 10)++ conn := &wsConn{Conn: ws}++ defer conn.Close()++ ctx, cancel := context.WithCancel(r.Context())++ var currentCap atomic.Int64++ currentCap.Store(grant.Cap)++ defer cancel()++ // Revocation, a stop, opt-out and a newly reported battery state close existing streams too.++ go func() {++ ticker := time.NewTicker(5 * time.Second)++ defer ticker.Stop()++ for {++ select {++ case <-ctx.Done():++ return++ case <-ticker.C:++ g, err := e.Client.Open(ctx, session, credential, e.Node)++ currentCap.Store(g.Cap)++ if err != nil || e.Pause() || e.Budget.Reserve(0, g.Cap) != nil {++ cancel()++ _ = conn.Close()++ return++ }++ }++ }++ }()++ _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))++ br := bufio.NewReader(conn)++ first, err := br.Peek(1)++ if err != nil {++ return++ }++ dial := e.Dial++ if dial == nil {++ dial = PublicDial++ }++ var upstream net.Conn++ if first[0] == 5 {++ upstream, err = e.socks(ctx, conn, br, session, credential, dial)++ } else {++ upstream, br, err = e.http(ctx, conn, br, session, credential, dial, grant.Cap)++ }++ if err != nil {++ return++ }++ defer upstream.Close()++ go func() { <-ctx.Done(); _ = upstream.Close() }()++ _ = conn.SetReadDeadline(time.Time{})++ copyBudget := func(dst net.Conn, src io.Reader) {++ buf := make([]byte, 32<<10)++ for {++ n, err := src.Read(buf)++ if n > 0 {++ if e.Budget.Reserve(int64(n), currentCap.Load()) != nil {++ return++ }++ if _, werr := dst.Write(buf[:n]); werr != nil {++ return++ }++ }++ if err != nil {++ return++ }++ }++ }++ done := make(chan struct{}, 1)++ go func() { copyBudget(upstream, br); _ = upstream.Close(); _ = conn.Close(); done <- struct{}{} }()++ copyBudget(conn, upstream)++ _ = conn.Close()++ _ = upstream.Close()++ <-done++}++func (e *Endpoint) http(ctx context.Context, c net.Conn, br *bufio.Reader, session, credential string, dial DialFunc, cap int64) (net.Conn, *bufio.Reader, error) {++ r, next, err := readRequest(br)++ br = next++ if err != nil {++ return nil, br, err++ }++ expected := "Basic " + base64.StdEncoding.EncodeToString([]byte(session+":"+credential))++ if subtle.ConstantTimeCompare([]byte(r.Header.Get("Proxy-Authorization")), []byte(expected)) != 1 {++ _, _ = io.WriteString(c, "HTTP/1.1 407 Proxy Authentication Required\r\nConnection: close\r\n\r\n")++ return nil, br, errors.New("refused")++ }++ address := r.Host++ if r.Method != "CONNECT" {++ if r.URL.Scheme != "http" || r.URL.User != nil {++ return nil, br, errors.New("HTTP proxy required")++ }++ address = r.URL.Host++ if r.URL.Port() == "" {++ address = net.JoinHostPort(r.URL.Hostname(), "80")++ }++ }++ upstream, err := dial(ctx, address)++ if err != nil {++ _, _ = io.WriteString(c, "HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n")++ return nil, br, err++ }++ if r.Method == "CONNECT" {++ _, err = io.WriteString(c, "HTTP/1.1 200 Connection Established\r\n\r\n")++ } else {++ r.Header.Del("Proxy-Authorization")++ r.Header.Del("Proxy-Connection")++ r.Close = true++ // Count request headers and bodies as well as the response. Reserve in the writer before any bytes leave.++ err = r.Write(&budgetWriter{Writer: upstream, budget: e.Budget, cap: cap})++ }++ if err != nil {++ _ = upstream.Close()++ return nil, br, err++ }++ return upstream, br, nil++}++++type budgetWriter struct {++ io.Writer++ budget *Budget++ cap int64++}++++func (w *budgetWriter) Write(b []byte) (int, error) {++ if err := w.budget.Reserve(int64(len(b)), w.cap); err != nil {++ return 0, err++ }++ return w.Writer.Write(b)++}++func (e *Endpoint) socks(ctx context.Context, c net.Conn, br *bufio.Reader, session, credential string, dial DialFunc) (net.Conn, error) {++ head := make([]byte, 2)++ if _, err := io.ReadFull(br, head); err != nil {++ return nil, err++ }++ methods := make([]byte, int(head[1]))++ if _, err := io.ReadFull(br, methods); err != nil {++ return nil, err++ }++ found := false++ for _, m := range methods {++ if m == 2 {++ found = true++ }++ }++ if !found {++ _, _ = c.Write([]byte{5, 255})++ return nil, errors.New("authentication required")++ }++ _, _ = c.Write([]byte{5, 2})++ if _, err := io.ReadFull(br, head); err != nil || head[0] != 1 {++ return nil, errors.New("authentication required")++ }++ user := make([]byte, int(head[1]))++ if _, err := io.ReadFull(br, user); err != nil {++ return nil, err++ }++ n, err := br.ReadByte()++ if err != nil {++ return nil, err++ }++ pass := make([]byte, int(n))++ if _, err = io.ReadFull(br, pass); err != nil {++ return nil, err++ }++ if subtle.ConstantTimeCompare(user, []byte(session)) != 1 || subtle.ConstantTimeCompare(pass, []byte(credential)) != 1 {++ _, _ = c.Write([]byte{1, 1})++ return nil, errors.New("refused")++ }++ _, _ = c.Write([]byte{1, 0})++ req := make([]byte, 4)++ if _, err = io.ReadFull(br, req); err != nil {++ return nil, err++ }++ if req[0] != 5 || req[1] != 1 || req[2] != 0 {++ _, _ = c.Write([]byte{5, 7, 0, 1, 0, 0, 0, 0, 0, 0})++ return nil, errors.New("CONNECT only")++ }++ var host string++ switch req[3] {++ case 3:++ n, err = br.ReadByte()++ if err != nil || n == 0 {++ return nil, errors.New("invalid host")++ }++ b := make([]byte, int(n))++ if _, err = io.ReadFull(br, b); err != nil {++ return nil, err++ }++ host = string(b)++ case 1, 4:++ size := 4++ if req[3] == 4 {++ size = 16++ }++ b := make([]byte, size)++ if _, err = io.ReadFull(br, b); err != nil {++ return nil, err++ }++ host = net.IP(b).String()++ default:++ return nil, errors.New("invalid address")++ }++ if _, err = io.ReadFull(br, head); err != nil {++ return nil, err++ }++ up, err := dial(ctx, net.JoinHostPort(host, strconv.Itoa(int(binary.BigEndian.Uint16(head)))))++ if err != nil {++ _, _ = c.Write([]byte{5, 2, 0, 1, 0, 0, 0, 0, 0, 0})++ return nil, err++ }++ _, err = c.Write([]byte{5, 0, 0, 1, 0, 0, 0, 0, 0, 0})++ if err != nil {++ up.Close()++ return nil, err++ }++ return up, nil++}++++// Bridge is bound to one isolated desk. It speaks an ordinary HTTP proxy to Chrome and the authenticated++// WebSocket byte pipe to Link. It never changes to a direct route after Link accepted a connection.++func Bridge(ctx context.Context, listener net.Listener, p Plan, testDial ...DialFunc) *http.Server {++ directDial := PublicDial++ if len(testDial) > 0 {++ directDial = testDial[0]++ }++ handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {++ var remote net.Conn++ host := r.URL.Hostname()++ if r.Method == "CONNECT" {++ host, _, _ = net.SplitHostPort(r.Host)++ }++ direct := p.Mode == "direct"++ for _, s := range p.Excluded {++ if strings.EqualFold(host, s) || strings.HasSuffix(strings.ToLower(host), "."+s) {++ direct = true++ }++ }++ if !direct {++ ws, resp, err := (&websocket.Dialer{HandshakeTimeout: 10 * time.Second}).DialContext(r.Context(), p.Relay, http.Header{"X-OV-Session": {p.Session}, "X-OV-Egress": {p.Credential}})++ if err == nil {++ remote = &wsConn{Conn: ws}++ } else if p.Offline == "openvibe" && (resp == nil || resp.StatusCode == 503) {++ direct = true++ } else {++ http.Error(w, "Your computer is unavailable.", 503)++ return++ }++ }++ if direct {++ address := r.Host++ if r.Method != "CONNECT" && r.URL.Port() == "" {++ address = net.JoinHostPort(r.URL.Hostname(), "80")++ }++ var err error++ remote, err = directDial(r.Context(), address)++ if err != nil {++ http.Error(w, "Site unavailable", 403)++ return++ }++ }++ defer remote.Close()++ if !direct {++ r.Header.Set("Proxy-Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(p.Session+":"+p.Credential)))++ if err := r.WriteProxy(remote); err != nil {++ return++ }++ } else if r.Method != "CONNECT" {++ r.Header.Del("Proxy-Authorization")++ r.Close = true++ if err := r.Write(remote); err != nil {++ return++ }++ }++ if r.Method != "CONNECT" {++ response, err := http.ReadResponse(bufio.NewReader(remote), r)++ if err != nil {++ http.Error(w, "Site unavailable", 502)++ return++ }++ defer response.Body.Close()++ for k, v := range response.Header {++ w.Header()[k] = v++ }++ w.WriteHeader(response.StatusCode)++ _, _ = io.Copy(w, response.Body)++ return++ }++ client, buffer, err := w.(http.Hijacker).Hijack()++ if err != nil {++ return++ }++ defer client.Close()++ if direct {++ _, _ = io.WriteString(client, "HTTP/1.1 200 Connection Established\r\n\r\n")++ }++ done := make(chan struct{}, 1)++ go func() { _, _ = io.Copy(remote, buffer); _ = remote.Close(); _ = client.Close(); done <- struct{}{} }()++ _, _ = io.Copy(client, remote)++ _ = remote.Close()++ _ = client.Close()++ <-done++ })++ server := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second, MaxHeaderBytes: 32 << 10, BaseContext: func(net.Listener) context.Context { return ctx }}++ go server.Serve(listener)++ go func() { <-ctx.Done(); _ = server.Close() }()++ return server++}++++func BrowserArgs(address string) []string {++ host, _, _ := net.SplitHostPort(address)++ return []string{"--proxy-server=http://" + address, "--proxy-bypass-list=<-loopback>", "--host-resolver-rules=MAP * ~NOTFOUND, EXCLUDE " + host, "--force-webrtc-ip-handling-policy=disable_non_proxied_udp", "--disable-quic", "--disable-features=DnsOverHttps,UseDnsHttpsSvcbAlpn"}++}++++func readRequest(br *bufio.Reader) (*http.Request, *bufio.Reader, error) {++ var header []byte++ for {++ line, err := br.ReadSlice('\n')++ if err != nil {++ return nil, br, err++ }++ header = append(header, line...)++ if len(header) > 32<<10 {++ return nil, br, errors.New("headers too large")++ }++ if string(line) == "\r\n" {++ break++ }++ }++ next := bufio.NewReader(io.MultiReader(bytes.NewReader(header), br))++ r, err := http.ReadRequest(next)++ return r, next, err++}nnew file mode 100644@@@ -0,0 +1,107 @@++package egress++++import (++ "bytes"++ "context"++ "encoding/json"++ "errors"++ "io"++ "net"++ "net/http"++ "path/filepath"++ "strings"++ "sync/atomic"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/connect"++ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"++ "github.com/OpenVibers/OpenVibe.Node/internal/link"++)++++// Serve exposes no public listening port. Connect owns the authenticated incoming byte pipe.++func Serve(ctx context.Context, creds *credentials.Credentials, stateDir string) error {++ if creds == nil || !creds.NetworkPaired() {++ return nil++ }++ budget, err := NewBudget(filepath.Join(stateDir, "link-egress-usage.json"))++ if err != nil {++ return err++ }++ listener, err := net.Listen("tcp", "127.0.0.1:0")++ if err != nil {++ return err++ }++ defer listener.Close()++ client := NewClient(creds)++ server := &http.Server{Handler: &Endpoint{Client: client, Node: creds.Principal, Budget: budget, Pause: Paused}, ReadHeaderTimeout: 10 * time.Second, MaxHeaderBytes: 8 << 10}++ go server.Serve(listener)++ defer server.Close()++ networkToken, err := client.Tokens.Token(ctx)++ if err != nil {++ return errors.New("computer sign-in unavailable")++ }++ connectTokens := link.NewTokenSourceFor(creds, nil, "openvibe-link/egress", link.ConnectAudience)++ token, err := connectTokens.Token(ctx)++ if err != nil {++ return errors.New("computer sign-in unavailable")++ }++ data, _ := json.Marshal(map[string]any{"port": listener.Addr().(*net.TCPAddr).Port, "network_token": networkToken.Reveal()})++ req, _ := http.NewRequestWithContext(ctx, "POST", "https://connect.openvibe.link/api/v1/node/egress-share", bytes.NewReader(data))++ req.Header.Set("Authorization", "Bearer "+token.Reveal())++ req.Header.Set("Content-Type", "application/json")++ resp, err := (&http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}).Do(req)++ if err != nil {++ return errors.New("Link connection unavailable")++ }++ defer resp.Body.Close()++ if resp.StatusCode != 200 {++ return errors.New("Link connection refused")++ }++ var result struct {++ Share connect.NodeShare `json:"share"`++ }++ if err = json.NewDecoder(io.LimitReader(resp.Body, 16<<10)).Decode(&result); err != nil {++ return errors.New("invalid Link connection")++ }++ var online atomic.Bool++ heartbeat := func() {++ if !online.Load() {++ return++ }++ relay := "wss://" + result.Share.Host + "/__ov_egress"++ _ = client.Call(ctx, "PUT", "/api/v1/node/self/egress/heartbeat", map[string]string{"relay_url": relay}, nil, true)++ }++ go func() {++ ticker := time.NewTicker(15 * time.Second)++ defer ticker.Stop()++ for {++ select {++ case <-ctx.Done():++ return++ case <-ticker.C:++ heartbeat()++ }++ }++ }()++ return connect.Run(ctx, connect.Share{ID: result.Share.ID, Port: listener.Addr().(*net.TCPAddr).Port}, connect.Options{++ LocalHost: "127.0.0.1", NodeToken: func(ctx context.Context) (string, error) { t, e := connectTokens.Token(ctx); return t.Reveal(), e },++ OnReady: func(connect.Ready) { online.Store(true); go heartbeat() }, OnDisconnect: func() { online.Store(false) },++ })++}++++// A transient service failure retries without taking down the person's Node. Errors contain no response bodies.++func Run(ctx context.Context, creds *credentials.Credentials, stateDir string) {++ for {++ _ = Serve(ctx, creds, stateDir)++ if creds == nil || !creds.NetworkPaired() {++ return++ }++ select {++ case <-ctx.Done():++ return++ case <-time.After(15 * time.Second):++ }++ }++}++++func IsCloudDesk(name string) bool { return strings.HasPrefix(name, "desk-s") && len(name) == 22 }@@@ -9,6 +9,7 @@ import ("encoding/json""errors""fmt"++ "github.com/OpenVibers/OpenVibe.Node/internal/egress""os""path/filepath""runtime"@@@ -189,6 +190,9 @@ func savePanel(dir string, off map[string]bool, first *panelFirstRun) error {// startPanel serves Link's window until ctx ends. A failure leaves the Node running without it.func (n *Node) startPanel(ctx context.Context) {++ if !n.opt.NoShares {++ go egress.Run(ctx, n.opt.Creds, n.opt.Paths.StateDir)++ }token, err := panel.NewToken()if err != nil {n.log.Warn("Link's window off: no token", "err", err)@@@ -13,6 +13,7 @@ import ("syscall""github.com/OpenVibers/OpenVibe.Node/internal/config"++ "github.com/OpenVibers/OpenVibe.Node/internal/egress""github.com/OpenVibers/OpenVibe.Node/internal/protocol")@@@ -21,8 +22,10 @@ import (// namespaces, as worker.run_as, inside its own cgroup with its own memory, CPU and process limits, seeing only the// system paths, its own /tmp (its profile) and the read-only Binds it names, and reaching only the public internet.type DeskSpec struct {-- Argv []string-- Env []string++ // Proxy starts the authenticated bridge on this desk's isolated gateway.++ Proxy func(string) (string, func(), error)++ Argv []string++ Env []string// Binds are host paths the program needs beyond the system paths: the browser's install directory, /etc/fonts.// Read-only, and only under deskBindRoots (CheckDeskBinds).Binds []string@@@ -123,10 +126,32 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, errsb.goW, sb.child = gw, append(sb.child, gr)// A browser keeps hundreds of files open and maps tens of GB of address space it never uses: no RLIMIT_AS (the// cgroup's memory.max is the limit) and no RLIMIT_CPU (cpu.max is: a desk lives for hours).++ if d.Egress != "" {++ if sb.egress, err = newEgress(cfg); err != nil {++ return 0, fmt.Errorf("desk egress: %w", err)++ }++ }++ if d.Proxy != nil {++ if sb.egress == nil {++ return 0, errors.New("a Link desk requires its private connection")++ }++ address, closeProxy, err := d.Proxy(sb.egress.host.String())++ if err != nil {++ return 0, err++ }++ defer closeProxy()++ closeRules, err := sb.egress.allowProxy(address)++ if err != nil {++ return 0, err++ }++ defer closeRules()++ // These mandatory flags are last: no caller-supplied flag can override the proxy or leak protection.++ d.Argv = append(d.Argv, egress.BrowserArgs(address)...)++ }spec := sandboxSpec{Root: root, Work: "/tmp/desk", Disk: d.DiskBytes, Links: links, Argv: d.Argv, Env: d.Env, Pipe: true,Limits: []rlimit{{rlimitCORE, 0, 0}, {rlimitNOFILE, 8192, 8192}, {rlimitNPROC, uint64(d.Pids), uint64(d.Pids)},{rlimitFSIZE, uint64(d.DiskBytes), uint64(d.DiskBytes)}}}-- if d.Egress != "" {++ if d.Egress != "" && d.Proxy == nil {spec.Resolvers = cfg.DNS}for _, b := range binds {@@@ -140,11 +165,6 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, errreturn 0, err}sb.child = append(sb.child, sf)-- if d.Egress != "" {-- if sb.egress, err = newEgress(cfg); err != nil {-- return 0, fmt.Errorf("desk egress: %w", err)-- }-- }if sb.cg, err = newJobCgroup(caps, protocol.JobLimits{MemBytes: d.MemBytes}); err != nil {// Its cgroup is made beside the Node's own leaf, which the Node sets up when its worker is on (the probe);// a desk started by a plugin cannot rearrange the Node's cgroup itself.@@@ -11,6 +11,7 @@ import (// DeskSpec is one desk's program as RunDesk runs it; off Linux a desk never runs sandboxed (desk_linux.go).type DeskSpec struct {++ Proxy func(string) (string, func(), error)Argv []stringEnv []stringBinds []string@@@ -7,6 +7,7 @@ import ("encoding/binary""errors""fmt"++ "net""net/netip""os""os/exec"@@@ -114,7 +115,10 @@ func newEgress(cfg config.WorkerConfig) (*egressNet, error) {e := &egressNet{tools: t, n: -1}vethMu.Lock()for i := range vethMax {-- if n := (vethNext + i) % vethMax; !vethUsed[n] {++ if n := (vethNext + i + os.Getpid()) % vethMax; !vethUsed[n] {++ if _, e := net.InterfaceByName(vethPrefix + strconv.Itoa(n)); e == nil {++ continue++ }e.n, vethUsed[n], vethNext = n, true, n+1break}@@@ -264,6 +268,7 @@ func ruleset(cfg config.WorkerConfig, uplinks []string) string {w("\t}")w("\tchain input {")w("\t\ttype filter hook input priority -10; policy accept;")++ w("\t\tmeta mark 0x4f564547 accept")w("\t\tiifname \"%s*\" goto refuse", vethPrefix)w("\t}")w("\tchain postrouting {")nnew file mode 100644@@@ -0,0 +1,43 @@++//go:build linux++++package worker++++import (++ "fmt"++ "net"++ "strconv"++)++++// Each desk has one host-side proxy socket. Every other packet from its veth is refused,++// including DNS, QUIC, IPv6 and WebRTC UDP. The proxy alone can dial the relay or an explicitly excluded site.++// A separate table survives another worker's refresh of openvibe_worker.++func (e *egressNet) proxyRules(port int) string {++ name := "ov_link_" + strconv.Itoa(e.n)++ return fmt.Sprintf(`table inet %s++delete table inet %s++table inet %s {++ chain mark_proxy {++ type filter hook prerouting priority -100; policy accept;++ iifname "%s" ip saddr %s ip daddr %s tcp dport %d meta mark set 0x4f564547++ }++ chain no_forward {++ type filter hook forward priority -20; policy accept;++ iifname "%s" reject with icmpx admin-prohibited++ }++}++`, name, name, name, e.hostIf(), e.job, e.host, port, e.hostIf())++}++func (e *egressNet) allowProxy(address string) (func(), error) {++ host, p, err := net.SplitHostPort(address)++ if err != nil || host != e.host.String() {++ return nil, fmt.Errorf("desk proxy must bind its own gateway")++ }++ port, err := strconv.Atoi(p)++ if err != nil || port < 1 || port > 65535 {++ return nil, fmt.Errorf("invalid desk proxy port")++ }++ if err = runTool(e.proxyRules(port), e.tools.nft, "-f", "-"); err != nil {++ return nil, err++ }++ return func() { _ = runTool("", e.tools.nft, "delete", "table", "inet", "ov_link_"+strconv.Itoa(e.n)) }, nil++}nnew file mode 100644@@@ -0,0 +1,31 @@++//go:build linux++++package worker++++import (++ "net/netip"++ "strings"++ "testing"++++ "github.com/OpenVibers/OpenVibe.Node/internal/config"++)++++func TestLinkProxyFirewallOnlyAllowsItsOwnSocket(t *testing.T) {++ e := &egressNet{n: 7, host: netip.MustParseAddr("169.254.240.29"), job: netip.MustParseAddr("169.254.240.30")}++ rules := e.proxyRules(32123)++ for _, required := range []string{`iifname "ovw7" ip saddr 169.254.240.30 ip daddr 169.254.240.29 tcp dport 32123 meta mark set 0x4f564547`, `iifname "ovw7" reject with icmpx admin-prohibited`, `hook forward priority -20`} {++ if !strings.Contains(rules, required) {++ t.Fatalf("missing restriction: %s", required)++ }++ }++ base := ruleset(config.WorkerConfig{Egress: config.EgressPublic}, []string{"eth0"})++ if !strings.Contains(base, "meta mark 0x4f564547 accept") {++ t.Fatal("desk cannot reach its proxy")++ }++ if strings.Contains(rules, "udp dport") || strings.Contains(rules, "ip6 daddr") {++ t.Fatal("direct DNS/WebRTC/IPv6 allowed")++ }++ if _, err := e.allowProxy("127.0.0.1:32123"); err == nil {++ t.Fatal("another gateway accepted")++ }++}@@@ -1,6 +1,6 @@import sys--from openvibe_computer import Computer++from openvibe_computer.link_egress import LinkComputer as Computerfrom openvibe_plugin import runsys.exit(run(Computer()))nnew file mode 100644@@@ -0,0 +1,16 @@++"""Cloud desk startup integration, kept separate from the browser's blind-fill code.++++Only a desk name crosses the process boundary. The Node reads its assigned session from Network;++credentials stay in the Node's proxy process and never enter Chrome's arguments or the model's tools.++"""++from openvibe_computer import Computer++++++class LinkComputer(Computer):++ def _desk_browser(self, desk, binary=None, env=None):++ browser = super()._desk_browser(desk, binary=binary, env=env)++ if self._shared() and desk.startswith("desk-s") and len(desk) == 22:++ if browser.sandbox and "--link-egress-desk" not in browser.sandbox:++ separator = browser.sandbox.index("--")++ browser.sandbox[separator:separator] = ["--link-egress-desk", desk]++ return browsernnew file mode 100644@@@ -0,0 +1,35 @@++import os++import sys++import unittest++from types import SimpleNamespace++from unittest import mock++++HERE = os.path.dirname(os.path.abspath(__file__))++sys.path.insert(0, os.path.dirname(HERE))++sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))++from openvibe_computer import Computer++from openvibe_computer.link_egress import LinkComputer++++++class LinkEgressTest(unittest.TestCase):++ def test_cloud_browser_gets_its_own_desk_before_start(self):++ plugin = LinkComputer()++ browser = SimpleNamespace(sandbox=["node", "desk-sandbox", "--egress", "public", "--"])++ desk = "desk-s0123456789abcdef"++ with mock.patch.object(Computer, "_desk_browser", return_value=browser), mock.patch.object(plugin, "_shared", return_value=True):++ self.assertIs(plugin._desk_browser(desk), browser)++ self.assertEqual(browser.sandbox[-3:], ["--link-egress-desk", desk, "--"])++ plugin._desk_browser(desk)++ self.assertEqual(browser.sandbox.count("--link-egress-desk"), 1)++ self.assertFalse(any("credential" in arg for arg in browser.sandbox))++++ def test_personal_desks_keep_their_normal_start(self):++ plugin = LinkComputer()++ browser = SimpleNamespace(sandbox=None)++ with mock.patch.object(Computer, "_desk_browser", return_value=browser), mock.patch.object(plugin, "_shared", return_value=False):++ self.assertIs(plugin._desk_browser("desk-personal"), browser)++ self.assertIsNone(browser.sandbox)++++++if __name__ == "__main__":++ unittest.main()
@@ -5,12 +5,15 @@ import (
@@ -5,12 +5,15 @@ import (
"flag"
"flag"
"fmt"
"fmt"
"io"
"io"
+ "net"
"os"
"os"
"os/signal"
"os/signal"
"strings"
"strings"
"syscall"
"syscall"
"github.com/OpenVibers/OpenVibe.Node/internal/config"
"github.com/OpenVibers/OpenVibe.Node/internal/config"
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ linkegress "github.com/OpenVibers/OpenVibe.Node/internal/egress"
"github.com/OpenVibers/OpenVibe.Node/internal/worker"
"github.com/OpenVibers/OpenVibe.Node/internal/worker"
)
)
@@ -45,6 +48,7 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {
@@ -45,6 +48,7 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {
diskMB := fs.Int64("disk-mb", 512, "the size of the desk's /tmp, where its profile lives, in MiB")
diskMB := fs.Int64("disk-mb", 512, "the size of the desk's /tmp, where its profile lives, in MiB")
pids := fs.Int64("pids", 256, "the most processes and threads of the desk")
pids := fs.Int64("pids", 256, "the most processes and threads of the desk")
egress := fs.String("egress", "", `"public" to reach the public internet (never the host's networks); none by default`)
egress := fs.String("egress", "", `"public" to reach the public internet (never the host's networks); none by default`)
+ linkDesk := fs.String("link-egress-desk", "", "the cloud desk assigned by Run")
var binds, env listFlag
var binds, env listFlag
fs.Var(&binds, "bind", "a read-only path the program needs (repeatable): its install directory, /etc/fonts")
fs.Var(&binds, "bind", "a read-only path the program needs (repeatable): its install directory, /etc/fonts")
fs.Var(&env, "setenv", "NAME=VALUE in the program's environment (repeatable); it gets nothing else")
fs.Var(&env, "setenv", "NAME=VALUE in the program's environment (repeatable); it gets nothing else")
@@ -75,7 +79,33 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {
@@ -75,7 +79,33 @@ func cmdDeskSandbox(args []string, stderr io.Writer) int {
}
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP)
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT, syscall.SIGHUP)
defer stop()
defer stop()
- code, err := worker.RunDesk(ctx, wcfg, worker.DeskSpec{Argv: argv, Env: env, Binds: binds,
+ var proxy func(string) (string, func(), error)
+ if *linkDesk != "" {
+ if !linkegress.IsCloudDesk(*linkDesk) {
+ fmt.Fprintln(stderr, "invalid cloud desk")
+ return 125
+ }
+ creds, e := credentials.Load(paths.CredentialFile(), nil)
+ if e != nil || !creds.NetworkPaired() {
+ fmt.Fprintln(stderr, "computer sign-in unavailable")
+ return 125
+ }
+ plan, e := linkegress.NewClient(creds).Desk(ctx, *linkDesk)
+ if e != nil {
+ fmt.Fprintln(stderr, "browsing connection unavailable")
+ return 125
+ }
+ proxy = func(gateway string) (string, func(), error) {
+ listener, e := net.Listen("tcp", net.JoinHostPort(gateway, "0"))
+ if e != nil {
+ return "", nil, e
+ }
+ proxyCtx, cancelProxy := context.WithCancel(ctx)
+ server := linkegress.Bridge(proxyCtx, listener, plan)
+ return listener.Addr().String(), func() { cancelProxy(); _ = server.Close() }, nil
+ }
+ }
+ code, err := worker.RunDesk(ctx, wcfg, worker.DeskSpec{Argv: argv, Env: env, Binds: binds, Proxy: proxy,
MemBytes: *memMB << 20, CPUMillis: *cpu, DiskBytes: *diskMB << 20, Pids: *pids, Egress: *egress,
MemBytes: *memMB << 20, CPUMillis: *cpu, DiskBytes: *diskMB << 20, Pids: *pids, Egress: *egress,
In: in, Out: out, Stderr: os.Stderr})
In: in, Out: out, Stderr: os.Stderr})
if err != nil && ctx.Err() == nil {
if err != nil && ctx.Err() == nil {
new file mode 100644
new file mode 100644
@@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
+# Browsing through your own Link
+
+Link automatically offers an authenticated browsing endpoint for its owner's cloud sessions after pairing.
+The switch, offline choice, monthly limit (20 GB by default) and site exceptions are on openvibe.link.
+Only public web destinations on TCP ports 80 and 443 are allowed. DNS is resolved and pinned on the Link device;
+private, local, metadata and IPv6 translation addresses are refused.
+
+Connect carries a WebSocket byte stream through its existing share relay. The stream contains HTTP CONNECT,
+ordinary HTTP proxy requests, or authenticated SOCKS5 CONNECT. Both Connect and Node ask Network about every
+new stream. Network compares the current device owner with the Run session owner, checks the credential hash,
+expiry, opt-out and cap. A cookie or an ordinary share API key never grants access to the browsing endpoint.
+Node checks active streams again every five seconds, closing them on expiry, revocation, opt-out or a reported
+battery/metered state. OS readings are best effort: Linux battery/NetworkManager, macOS battery, Windows battery/network cost.
+
+The shared desk plugin adds only the desk name to `desk-sandbox`. The Node authenticates to Network, reads the
+session assigned to its own machine, and gets a fresh credential kept in its proxy process. Chrome never receives
+that credential. Its namespace reaches one gateway proxy socket only; all direct forwarding is refused. Chrome
+also disables direct DNS, QUIC and non-proxied WebRTC UDP, with the loopback bypass disabled.
+
+The device persists bandwidth reservations before forwarding bytes, shared across its sessions and directions.
+Reservations may overcount after a failed write or crash; restarting Link does not reopen the monthly allowance.
+The allowance is per Link device. A damaged usage file refuses service until repaired by the owner.
+
+Run must renew authorization within two minutes and revoke it when ending the desk. A missing or unavailable
+Network never means direct fallback. The saved offline choice applies when the Link connection cannot be made:
+use OpenVibe's address or wait. Explicit site exceptions use OpenVibe's address and public-only DNS there.
+
+Deployment order: Network's expand migration and routes, Connect's expand migration and route, Node including
+the updated computer plugin on cloud desk hosts and home devices, then Run. Existing cloud hosts must be upgraded
+before enabling Run's integration. The integration applies to browser desks created through Run; providers must
+use that session path to inherit the policy. No blind-fill files are modified.
@@ -71,14 +71,15 @@ var ErrStopped = errors.New("connect: share stopped by the server")
@@ -71,14 +71,15 @@ var ErrStopped = errors.New("connect: share stopped by the server")
// Options tune a client; the zero value is right for production.
// Options tune a client; the zero value is right for production.
type Options struct {
type Options struct {
- Log *slog.Logger
- Version string
- LocalHost string // default "localhost"
- OnReady func(Ready) // called after each successful hello
- OnRequest func(method, path string, status int, d time.Duration)
- Dialer *websocket.Dialer
- MinWait time.Duration // reconnect backoff floor (default 1 s)
- MaxWait time.Duration // reconnect backoff ceiling (default 30 s)
+ Log *slog.Logger
+ Version string
+ LocalHost string // default "localhost"
+ OnDisconnect func()
+ OnReady func(Ready) // called after each successful hello
+ OnRequest func(method, path string, status int, d time.Duration)
+ Dialer *websocket.Dialer
+ MinWait 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.link, for shares it serves by id (node.go). An error
// NodeToken buys this computer's node token for openvibe.link, 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.
// wrapping ErrStopped (the machine was removed on OpenVibe) stops the share; any other is tried again.
@@ -143,6 +144,9 @@ func Run(ctx context.Context, s Share, o Options) error {
@@ -143,6 +144,9 @@ func Run(ctx context.Context, s Share, o Options) error {
wait := o.MinWait
wait := o.MinWait
for {
for {
err := runOnce(ctx, s, o, func() { wait = o.MinWait })
err := runOnce(ctx, s, o, func() { wait = o.MinWait })
+ if o.OnDisconnect != nil {
+ o.OnDisconnect()
+ }
if ctx.Err() != nil {
if ctx.Err() != nil {
return ctx.Err()
return ctx.Err()
}
}
new file mode 100644
new file mode 100644
@@ -0,0 +1,106 @@
@@ -0,0 +1,106 @@
+package egress
+
+import (
+ "bufio"
+ "context"
+ "encoding/json"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "testing"
+ "time"
+)
+
+func echoDial(_ context.Context, _ string) (net.Conn, error) {
+ a, b := net.Pipe()
+ go func() { defer b.Close(); _, _ = io.Copy(b, b) }()
+ return a, nil
+}
+func proxyConnect(t *testing.T, address string) (net.Conn, *bufio.Reader, *http.Response) {
+ t.Helper()
+ c, err := net.DialTimeout("tcp", address, time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _ = c.SetDeadline(time.Now().Add(3 * time.Second))
+ _, _ = io.WriteString(c, "CONNECT web.example:443 HTTP/1.1\r\nHost: web.example:443\r\n\r\n")
+ br := bufio.NewReader(c)
+ r, err := http.ReadResponse(br, nil)
+ if err != nil {
+ c.Close()
+ t.Fatal(err)
+ }
+ return c, br, r
+}
+func TestBridgeThroughNodeAndCapClosesTheTunnel(t *testing.T) {
+ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ var b map[string]string
+ _ = json.NewDecoder(r.Body).Decode(&b)
+ if b["credential"] != strings.Repeat("c", 43) {
+ w.WriteHeader(403)
+ return
+ }
+ _ = json.NewEncoder(w).Encode(Grant{Session: "s", Node: "n", Enabled: true, Cap: 4, Expires: time.Now().Add(time.Minute).UnixMilli()})
+ }))
+ defer network.Close()
+ budget, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))
+ targets := make(chan string, 1)
+ endpoint := httptest.NewServer(&Endpoint{Client: &Client{Base: network.URL}, Node: "n", Budget: budget, Pause: func() bool { return false }, Dial: func(ctx context.Context, s string) (net.Conn, error) { targets <- s; return echoDial(ctx, s) }})
+ defer endpoint.Close()
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ server := Bridge(ctx, listener, Plan{Mode: "link", Relay: "ws" + strings.TrimPrefix(endpoint.URL, "http") + "/__ov_egress", Session: "s", Credential: strings.Repeat("c", 43), Offline: "wait"})
+ defer server.Close()
+ conn, br, response := proxyConnect(t, listener.Addr().String())
+ defer conn.Close()
+ if response.StatusCode != 200 {
+ t.Fatal("bridge failed")
+ }
+ if <-targets != "web.example:443" {
+ t.Fatal("destination was resolved before reaching Node")
+ }
+ _, _ = conn.Write([]byte("ab"))
+ echo := make([]byte, 2)
+ if _, err = io.ReadFull(br, echo); err != nil || string(echo) != "ab" {
+ t.Fatal("bridge did not carry bytes")
+ }
+ // Two bytes each way consumed the whole allowance; no fifth byte may leave the device.
+ _, _ = conn.Write([]byte("x"))
+ if _, err = br.ReadByte(); err == nil {
+ t.Fatal("spent bandwidth did not close tunnel")
+ }
+}
+func TestBridgeOfflineChoicesAndAuthorizationFailures(t *testing.T) {
+ for _, tc := range []struct {
+ name, choice string
+ status, want int
+ direct bool
+ }{{"offline fallback", "openvibe", 503, 200, true}, {"offline wait", "wait", 503, 503, false}, {"authorization refused", "openvibe", 403, 503, false}} {
+ t.Run(tc.name, func(t *testing.T) {
+ remote := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(tc.status) }))
+ defer remote.Close()
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatal(err)
+ }
+ var used atomic.Bool
+ server := Bridge(ctx, listener, Plan{Mode: "link", Relay: "ws" + strings.TrimPrefix(remote.URL, "http"), Offline: tc.choice}, func(ctx context.Context, s string) (net.Conn, error) { used.Store(true); return echoDial(ctx, s) })
+ defer server.Close()
+ conn, _, response := proxyConnect(t, listener.Addr().String())
+ conn.Close()
+ if response.StatusCode != tc.want || used.Load() != tc.direct {
+ t.Fatalf("status %d, direct %v", response.StatusCode, used.Load())
+ }
+ })
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,81 @@
@@ -0,0 +1,81 @@
+package egress
+
+import (
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "runtime"
+ "sync"
+ "time"
+)
+
+// Reserve before forwarding bytes. A crash can overcount, never reopen spent bandwidth.
+// One Budget belongs to the running Link, shared by all its sessions and both directions.
+type Budget struct {
+ mu sync.Mutex
+ path string
+ Month string `json:"month"`
+ Used int64 `json:"used"`
+ Now func() time.Time `json:"-"`
+}
+
+func NewBudget(path string) (*Budget, error) {
+ b := &Budget{path: path, Now: time.Now}
+ raw, err := os.ReadFile(path)
+ if err != nil && !os.IsNotExist(err) {
+ return nil, err
+ }
+ if err == nil {
+ if err = json.Unmarshal(raw, b); err != nil || b.Used < 0 {
+ return nil, errors.New("browsing usage could not be read")
+ }
+ }
+ return b, nil
+}
+func (b *Budget) Reserve(n, cap int64) error {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ month := b.Now().UTC().Format("2006-01")
+ if month > b.Month {
+ b.Month, b.Used = month, 0
+ } // a clock moving back cannot reset the cap
+ if n < 0 || cap < 0 || n > cap-b.Used || (n == 0 && b.Used >= cap) {
+ return errors.New("monthly browsing limit reached")
+ }
+ b.Used += n
+ raw, err := json.Marshal(b)
+ if err != nil {
+ return err
+ }
+ if err = os.MkdirAll(filepath.Dir(b.path), 0700); err != nil {
+ return err
+ }
+ f, err := os.OpenFile(b.path+".tmp", os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
+ if err != nil {
+ return err
+ }
+ _, err = f.Write(raw)
+ if err == nil {
+ err = f.Sync()
+ }
+ closeErr := f.Close()
+ if err != nil {
+ return err
+ }
+ if closeErr != nil {
+ return closeErr
+ }
+ if err = os.Rename(b.path+".tmp", b.path); err != nil {
+ return err
+ }
+ if runtime.GOOS != "windows" {
+ dir, err := os.Open(filepath.Dir(b.path))
+ if err != nil {
+ return err
+ }
+ defer dir.Close()
+ return dir.Sync()
+ }
+ return nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,113 @@
@@ -0,0 +1,113 @@
+// Package egress carries one person's cloud browsing through their own paired Link.
+package egress
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/http"
+ "net/url"
+ "strings"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/link"
+)
+
+type Tokens interface {
+ Token(context.Context) (credentials.Secret, error)
+}
+type Client struct {
+ Base string
+ Tokens Tokens
+ HTTP *http.Client
+}
+type Plan struct {
+ Mode string `json:"mode"`
+ Session string `json:"session_id"`
+ Credential string `json:"credential"`
+ Device string `json:"device"`
+ DeviceName string `json:"device_name"`
+ Relay string `json:"relay_url"`
+ Enabled bool `json:"egress_via_link"`
+ Offline string `json:"offline_choice"`
+ Cap int64 `json:"monthly_bytes"`
+ Excluded []string `json:"excluded_sites"`
+}
+type Grant struct {
+ Session string `json:"session_id"`
+ Node string `json:"node_id"`
+ Expires int64 `json:"expires_at"`
+ Cap int64 `json:"monthly_bytes"`
+ Enabled bool `json:"egress_via_link"`
+}
+
+func NewClient(c *credentials.Credentials) *Client {
+ return &Client{Base: strings.TrimRight(c.Network, "/"), Tokens: link.NewTokenSourceFor(c, nil, "openvibe-link/egress", "openvibe.network")}
+}
+func (c *Client) Call(ctx context.Context, method, path string, body, out any, authenticated bool) error {
+ var data io.Reader
+ if body != nil {
+ b, err := json.Marshal(body)
+ if err != nil {
+ return err
+ }
+ data = bytes.NewReader(b)
+ }
+ req, err := http.NewRequestWithContext(ctx, method, c.Base+path, data)
+ if err != nil {
+ return errors.New("browsing connection unavailable")
+ }
+ req.Header.Set("Content-Type", "application/json")
+ if authenticated {
+ token, err := c.Tokens.Token(ctx)
+ if err != nil {
+ return errors.New("computer sign-in unavailable")
+ }
+ req.Header.Set("Authorization", "Bearer "+token.Reveal())
+ }
+ client := c.HTTP
+ if client == nil {
+ client = &http.Client{Timeout: 10 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+ }
+ resp, err := client.Do(req)
+ if err != nil {
+ return errors.New("browsing connection unavailable")
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode/100 != 2 {
+ return errors.New("browsing connection refused")
+ }
+ if out == nil {
+ return nil
+ }
+ return json.NewDecoder(io.LimitReader(resp.Body, 32<<10)).Decode(out)
+}
+func (c *Client) Open(ctx context.Context, session, credential, node string) (Grant, error) {
+ var grant Grant
+ err := c.Call(ctx, "POST", "/api/v1/link-egress/open", map[string]string{"session_id": session, "credential": credential, "node_id": node}, &grant, false)
+ if err == nil && (grant.Session != session || grant.Node != node || !grant.Enabled || grant.Expires <= time.Now().UnixMilli()) {
+ err = errors.New("browsing connection refused")
+ }
+ return grant, err
+}
+func (c *Client) Desk(ctx context.Context, desk string) (Plan, error) {
+ var p Plan
+ err := c.Call(ctx, "GET", "/api/v1/node/self/egress/desks/"+url.PathEscape(desk), nil, &p, true)
+ if err != nil {
+ return p, err
+ }
+ if p.Mode == "wait" {
+ return p, errors.New("waiting for your computer")
+ }
+ if p.Mode == "direct" {
+ return p, nil
+ }
+ u, err := url.Parse(p.Relay)
+ if err != nil || p.Mode != "link" || u.Scheme != "wss" || !strings.HasSuffix(u.Hostname(), ".openvibe.link") || u.Port() != "" || u.User != nil || u.Path != "/__ov_egress" || u.RawQuery != "" || len(p.Credential) != 43 {
+ return p, errors.New("invalid browsing connection")
+ }
+ return p, nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,58 @@
@@ -0,0 +1,58 @@
+package egress
+
+import (
+ "context"
+ "errors"
+ "net"
+ "net/netip"
+ "strconv"
+ "time"
+)
+
+var blocked = []netip.Prefix{
+ netip.MustParsePrefix("0.0.0.0/8"), netip.MustParsePrefix("100.64.0.0/10"), netip.MustParsePrefix("192.0.0.0/24"),
+ netip.MustParsePrefix("192.0.2.0/24"), netip.MustParsePrefix("192.88.99.0/24"), netip.MustParsePrefix("198.18.0.0/15"),
+ netip.MustParsePrefix("198.51.100.0/24"), netip.MustParsePrefix("203.0.113.0/24"), netip.MustParsePrefix("240.0.0.0/4"),
+ netip.MustParsePrefix("64:ff9b::/96"), netip.MustParsePrefix("64:ff9b:1::/48"), netip.MustParsePrefix("2001::/32"), netip.MustParsePrefix("2002::/16"), netip.MustParsePrefix("2001:db8::/32"),
+}
+
+func Public(ip netip.Addr) bool {
+ ip = ip.Unmap()
+ if !ip.IsValid() || !ip.IsGlobalUnicast() || ip.IsPrivate() || ip.IsLoopback() || ip.IsLinkLocalUnicast() || ip.Zone() != "" {
+ return false
+ }
+ for _, p := range blocked {
+ if p.Contains(ip) {
+ return false
+ }
+ }
+ return true
+}
+
+type DialFunc func(context.Context, string) (net.Conn, error)
+
+func PublicDial(ctx context.Context, address string) (net.Conn, error) {
+ return checkedDial(ctx, address, net.DefaultResolver.LookupIPAddr, (&net.Dialer{Timeout: 10 * time.Second}).DialContext)
+}
+func checkedDial(ctx context.Context, address string, resolve func(context.Context, string) ([]net.IPAddr, error), dial func(context.Context, string, string) (net.Conn, error)) (net.Conn, error) {
+ host, port, err := net.SplitHostPort(address)
+ if err != nil {
+ return nil, err
+ }
+ p, err := strconv.Atoi(port)
+ if err != nil || (p != 80 && p != 443) {
+ return nil, errors.New("only web ports are allowed")
+ }
+ // Resolve on the person's Node and pin the checked address for the actual dial.
+ ips, err := resolve(ctx, host)
+ if err != nil || len(ips) == 0 {
+ return nil, errors.New("site address unavailable")
+ }
+ for _, ip := range ips {
+ a, ok := netip.AddrFromSlice(ip.IP)
+ if !ok || !Public(a) {
+ return nil, errors.New("private site address refused")
+ }
+ }
+ return dial(ctx, "tcp", net.JoinHostPort(ips[0].IP.String(), port))
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,201 @@
@@ -0,0 +1,201 @@
+package egress
+
+import (
+ "bufio"
+ "context"
+ "encoding/base64"
+ "encoding/json"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/netip"
+ "path/filepath"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/gorilla/websocket"
+)
+
+func TestBudgetPersistsAndSharesCap(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "usage.json")
+ b, err := NewBudget(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var ok atomic.Int64
+ var wg sync.WaitGroup
+ for i := 0; i < 20; i++ {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ if b.Reserve(10, 100) == nil {
+ ok.Add(1)
+ }
+ }()
+ }
+ wg.Wait()
+ if ok.Load() != 10 {
+ t.Fatalf("cap allowed %d chunks", ok.Load())
+ }
+ b, err = NewBudget(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if b.Reserve(1, 100) == nil {
+ t.Fatal("restart reset the cap")
+ }
+ b.Now = func() time.Time { return time.Now().AddDate(0, 1, 0) }
+ if b.Reserve(10, 100) != nil {
+ t.Fatal("month did not reset")
+ }
+ b.Now = time.Now
+ if b.Reserve(91, 100) == nil {
+ t.Fatal("clock reversal reset the cap")
+ }
+}
+func TestDNSRejectsMixedAnswersAndTranslationRanges(t *testing.T) {
+ for _, s := range []string{"127.0.0.1", "10.0.0.1", "169.254.169.254", "::1", "64:ff9b::7f00:1", "2002:7f00:1::", "2001::1", "::ffff:192.168.1.1"} {
+ if Public(netip.MustParseAddr(s)) {
+ t.Fatalf("allowed %s", s)
+ }
+ }
+ called := false
+ _, err := checkedDial(context.Background(), "site.example:443", func(context.Context, string) ([]net.IPAddr, error) {
+ return []net.IPAddr{{IP: net.ParseIP("8.8.8.8")}, {IP: net.ParseIP("10.1.1.1")}}, nil
+ }, func(context.Context, string, string) (net.Conn, error) { called = true; return nil, nil })
+ if err == nil || called {
+ t.Fatal("mixed DNS answer was dialed")
+ }
+ var address string
+ _, err = checkedDial(context.Background(), "site.example:443", func(context.Context, string) ([]net.IPAddr, error) {
+ return []net.IPAddr{{IP: net.ParseIP("8.8.8.8")}}, nil
+ }, func(_ context.Context, _, a string) (net.Conn, error) { address = a; return nil, nil })
+ if err != nil || address != "8.8.8.8:443" {
+ t.Fatal("DNS result not pinned")
+ }
+}
+func TestMandatoryBrowserFlags(t *testing.T) {
+ flags := strings.Join(BrowserArgs("169.254.240.1:12345"), " ")
+ for _, s := range []string{"--proxy-server=http://169.254.240.1:12345", "--proxy-bypass-list=<-loopback>", "MAP * ~NOTFOUND", "disable_non_proxied_udp", "--disable-quic", "DnsOverHttps"} {
+ if !strings.Contains(flags, s) {
+ t.Fatalf("missing %s", s)
+ }
+ }
+}
+
+func TestAuthenticatedCONNECTAndSOCKSWithRemoteDNS(t *testing.T) {
+ const session = "ses_test"
+ credential := strings.Repeat("a", 43)
+ var checks atomic.Int64
+ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ var b map[string]string
+ _ = json.NewDecoder(r.Body).Decode(&b)
+ if b["credential"] != credential {
+ w.WriteHeader(403)
+ return
+ }
+ checks.Add(1)
+ _ = json.NewEncoder(w).Encode(Grant{Session: session, Node: "nod_test", Enabled: true, Cap: 100000, Expires: time.Now().Add(time.Minute).UnixMilli()})
+ }))
+ defer network.Close()
+ budget, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))
+ targets := make(chan string, 4)
+ endpoint := &Endpoint{Client: &Client{Base: network.URL}, Node: "nod_test", Budget: budget, Pause: func() bool { return false }, Dial: func(_ context.Context, address string) (net.Conn, error) {
+ targets <- address
+ a, b := net.Pipe()
+ go func() { defer b.Close(); _, _ = io.Copy(b, b) }()
+ return a, nil
+ }}
+ server := httptest.NewServer(endpoint)
+ defer server.Close()
+ connect := func(pass string) (*wsConn, *http.Response, error) {
+ ws, r, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(server.URL, "http")+"/__ov_egress", http.Header{"X-OV-Session": {session}, "X-OV-Egress": {pass}})
+ if err != nil {
+ return nil, r, err
+ }
+ return &wsConn{Conn: ws}, r, nil
+ }
+ if _, r, err := connect("wrong"); err == nil || r.StatusCode != 403 {
+ t.Fatal("wrong session credential accepted")
+ }
+ c, _, err := connect(credential)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, _ = io.WriteString(c, "CONNECT site.example:443 HTTP/1.1\r\nHost: site.example:443\r\nProxy-Authorization: Basic "+base64.StdEncoding.EncodeToString([]byte(session+":"+credential))+"\r\n\r\n")
+ br := bufio.NewReader(c)
+ resp, err := http.ReadResponse(br, nil)
+ if err != nil || resp.StatusCode != 200 {
+ t.Fatalf("CONNECT failed: %v", err)
+ }
+ _, _ = c.Write([]byte("hello"))
+ b := make([]byte, 5)
+ if _, err = io.ReadFull(br, b); err != nil || string(b) != "hello" {
+ t.Fatal("CONNECT bytes lost")
+ }
+ c.Close()
+ if <-targets != "site.example:443" {
+ t.Fatal("resolved before reaching Node")
+ }
+ c, _, err = connect(credential)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, _ = c.Write([]byte{5, 1, 2})
+ answer := make([]byte, 2)
+ _, _ = io.ReadFull(c, answer)
+ if answer[1] != 2 {
+ t.Fatal("SOCKS auth missing")
+ }
+ auth := append([]byte{1, byte(len(session))}, []byte(session)...)
+ auth = append(auth, byte(len(credential)))
+ auth = append(auth, []byte(credential)...)
+ _, _ = c.Write(auth)
+ _, _ = io.ReadFull(c, answer)
+ if answer[1] != 0 {
+ t.Fatal("SOCKS auth refused")
+ }
+ host := "site.example"
+ request := append([]byte{5, 1, 0, 3, byte(len(host))}, []byte(host)...)
+ request = append(request, 1, 187)
+ _, _ = c.Write(request)
+ reply := make([]byte, 10)
+ if _, err = io.ReadFull(c, reply); err != nil || reply[1] != 0 {
+ t.Fatal("SOCKS refused")
+ }
+ c.Close()
+ if <-targets != "site.example:443" {
+ t.Fatal("SOCKS DNS not left to Node")
+ }
+ if checks.Load() != 2 {
+ t.Fatalf("every open must check Network, got %d", checks.Load())
+ }
+}
+
+func TestCapAndPowerRefuseTunnel(t *testing.T) {
+ network := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ _ = json.NewEncoder(w).Encode(Grant{Session: "s", Node: "n", Enabled: true, Cap: 10, Expires: time.Now().Add(time.Minute).UnixMilli()})
+ }))
+ defer network.Close()
+ b, _ := NewBudget(filepath.Join(t.TempDir(), "usage"))
+ if err := b.Reserve(10, 10); err != nil {
+ t.Fatal(err)
+ }
+ // Spent cap closes as soon as data would be forwarded; a cap lowered below existing usage refuses even an open.
+ if b.Reserve(1, 10) == nil {
+ t.Fatal("cap did not stop bytes")
+ }
+ e := &Endpoint{Client: &Client{Base: network.URL}, Node: "n", Budget: b, Pause: func() bool { return true }}
+ r := httptest.NewRequest("GET", "http://local/__ov_egress", nil)
+ r.Header.Set("X-OV-Session", "s")
+ r.Header.Set("X-OV-Egress", "test")
+ w := httptest.NewRecorder()
+ e.ServeHTTP(w, r)
+ if w.Code != 403 {
+ t.Fatal("battery or metered state did not pause")
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,17 @@
@@ -0,0 +1,17 @@
+package egress
+
+import (
+ "os/exec"
+ "testing"
+)
+
+func TestCloudPluginStartup(t *testing.T) {
+ python, err := exec.LookPath("python3")
+ if err != nil {
+ t.Skip("Python not installed")
+ }
+ out, err := exec.Command(python, "../../plugins/computer/tests/test_link_egress.py").CombinedOutput()
+ if err != nil {
+ t.Fatalf("cloud plugin: %v\n%s", err, out)
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,44 @@
@@ -0,0 +1,44 @@
+package egress
+
+import (
+ "context"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "strings"
+ "time"
+)
+
+// Unknown OS readings do not invent a metered/battery state. Known readings pause immediately.
+func Paused() bool {
+ if runtime.GOOS == "linux" {
+ paths, _ := filepath.Glob("/sys/class/power_supply/*/status")
+ for _, p := range paths {
+ b, _ := os.ReadFile(p)
+ if strings.TrimSpace(string(b)) == "Discharging" {
+ return true
+ }
+ }
+ out := probe("nmcli", "-t", "-f", "GENERAL.METERED", "device", "show")
+ for _, line := range strings.Split(out, "\n") {
+ _, v, _ := strings.Cut(line, ":")
+ if strings.HasPrefix(strings.TrimSpace(v), "yes") {
+ return true
+ }
+ }
+ }
+ if runtime.GOOS == "darwin" {
+ return strings.Contains(probe("pmset", "-g", "batt"), "'Battery Power'")
+ }
+ if runtime.GOOS == "windows" {
+ return strings.TrimSpace(probe("powershell.exe", "-NoProfile", "-NonInteractive", "-Command", `Add-Type -AssemblyName System.Windows.Forms; if ([System.Windows.Forms.SystemInformation]::PowerStatus.PowerLineStatus -eq 'Offline') {'pause'; exit}; $p = [Windows.Networking.Connectivity.NetworkInformation, Windows.Networking.Connectivity, ContentType=WindowsRuntime]::GetInternetConnectionProfile(); if ($p) {$c=$p.GetConnectionCost(); if ($c.NetworkCostType -eq 'Fixed' -or $c.NetworkCostType -eq 'Variable' -or $c.Roaming -or $c.OverDataLimit) {'pause'}}`)) == "pause"
+ }
+ return false
+}
+func probe(name string, args ...string) string {
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ b, _ := exec.CommandContext(ctx, name, args...).Output()
+ return string(b)
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,419 @@
@@ -0,0 +1,419 @@
+package egress
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "crypto/subtle"
+ "encoding/base64"
+ "encoding/binary"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "strconv"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/gorilla/websocket"
+)
+
+// The WebSocket is the byte pipe already supported by Connect; no destination or credential is in its URL.
+type wsConn struct {
+ *websocket.Conn
+ reader io.Reader
+ mu sync.Mutex
+}
+
+func (c *wsConn) Read(b []byte) (int, error) {
+ for {
+ if c.reader == nil {
+ kind, r, err := c.NextReader()
+ if err != nil {
+ return 0, err
+ }
+ if kind != websocket.BinaryMessage {
+ return 0, errors.New("binary stream required")
+ }
+ c.reader = r
+ }
+ n, err := c.reader.Read(b)
+ if err == io.EOF {
+ c.reader = nil
+ if n > 0 {
+ return n, nil
+ }
+ continue
+ }
+ return n, err
+ }
+}
+func (c *wsConn) Write(b []byte) (int, error) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ if err := c.WriteMessage(websocket.BinaryMessage, b); err != nil {
+ return 0, err
+ }
+ return len(b), nil
+}
+func (c *wsConn) LocalAddr() net.Addr { return c.UnderlyingConn().LocalAddr() }
+func (c *wsConn) RemoteAddr() net.Addr { return c.UnderlyingConn().RemoteAddr() }
+func (c *wsConn) SetDeadline(t time.Time) error {
+ _ = c.SetReadDeadline(t)
+ return c.SetWriteDeadline(t)
+}
+
+type Endpoint struct {
+ Client *Client
+ Node string
+ Budget *Budget
+ Dial DialFunc
+ Pause func() bool
+}
+
+func (e *Endpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path != "/__ov_egress" || r.Method != "GET" {
+ http.Error(w, "refused", 403)
+ return
+ }
+ session, credential := r.Header.Get("X-OV-Session"), r.Header.Get("X-OV-Egress")
+ grant, err := e.Client.Open(r.Context(), session, credential, e.Node)
+ if err != nil || e.Pause() || e.Budget.Reserve(0, grant.Cap) != nil {
+ http.Error(w, "refused", 403)
+ return
+ }
+ ws, err := (&websocket.Upgrader{CheckOrigin: func(r *http.Request) bool { return r.Header.Get("Origin") == "" }}).Upgrade(w, r, nil)
+ if err != nil {
+ return
+ }
+ ws.SetReadLimit(64 << 10)
+ conn := &wsConn{Conn: ws}
+ defer conn.Close()
+ ctx, cancel := context.WithCancel(r.Context())
+ var currentCap atomic.Int64
+ currentCap.Store(grant.Cap)
+ defer cancel()
+ // Revocation, a stop, opt-out and a newly reported battery state close existing streams too.
+ go func() {
+ ticker := time.NewTicker(5 * time.Second)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ g, err := e.Client.Open(ctx, session, credential, e.Node)
+ currentCap.Store(g.Cap)
+ if err != nil || e.Pause() || e.Budget.Reserve(0, g.Cap) != nil {
+ cancel()
+ _ = conn.Close()
+ return
+ }
+ }
+ }
+ }()
+ _ = conn.SetReadDeadline(time.Now().Add(10 * time.Second))
+ br := bufio.NewReader(conn)
+ first, err := br.Peek(1)
+ if err != nil {
+ return
+ }
+ dial := e.Dial
+ if dial == nil {
+ dial = PublicDial
+ }
+ var upstream net.Conn
+ if first[0] == 5 {
+ upstream, err = e.socks(ctx, conn, br, session, credential, dial)
+ } else {
+ upstream, br, err = e.http(ctx, conn, br, session, credential, dial, grant.Cap)
+ }
+ if err != nil {
+ return
+ }
+ defer upstream.Close()
+ go func() { <-ctx.Done(); _ = upstream.Close() }()
+ _ = conn.SetReadDeadline(time.Time{})
+ copyBudget := func(dst net.Conn, src io.Reader) {
+ buf := make([]byte, 32<<10)
+ for {
+ n, err := src.Read(buf)
+ if n > 0 {
+ if e.Budget.Reserve(int64(n), currentCap.Load()) != nil {
+ return
+ }
+ if _, werr := dst.Write(buf[:n]); werr != nil {
+ return
+ }
+ }
+ if err != nil {
+ return
+ }
+ }
+ }
+ done := make(chan struct{}, 1)
+ go func() { copyBudget(upstream, br); _ = upstream.Close(); _ = conn.Close(); done <- struct{}{} }()
+ copyBudget(conn, upstream)
+ _ = conn.Close()
+ _ = upstream.Close()
+ <-done
+}
+func (e *Endpoint) http(ctx context.Context, c net.Conn, br *bufio.Reader, session, credential string, dial DialFunc, cap int64) (net.Conn, *bufio.Reader, error) {
+ r, next, err := readRequest(br)
+ br = next
+ if err != nil {
+ return nil, br, err
+ }
+ expected := "Basic " + base64.StdEncoding.EncodeToString([]byte(session+":"+credential))
+ if subtle.ConstantTimeCompare([]byte(r.Header.Get("Proxy-Authorization")), []byte(expected)) != 1 {
+ _, _ = io.WriteString(c, "HTTP/1.1 407 Proxy Authentication Required\r\nConnection: close\r\n\r\n")
+ return nil, br, errors.New("refused")
+ }
+ address := r.Host
+ if r.Method != "CONNECT" {
+ if r.URL.Scheme != "http" || r.URL.User != nil {
+ return nil, br, errors.New("HTTP proxy required")
+ }
+ address = r.URL.Host
+ if r.URL.Port() == "" {
+ address = net.JoinHostPort(r.URL.Hostname(), "80")
+ }
+ }
+ upstream, err := dial(ctx, address)
+ if err != nil {
+ _, _ = io.WriteString(c, "HTTP/1.1 403 Forbidden\r\nConnection: close\r\n\r\n")
+ return nil, br, err
+ }
+ if r.Method == "CONNECT" {
+ _, err = io.WriteString(c, "HTTP/1.1 200 Connection Established\r\n\r\n")
+ } else {
+ r.Header.Del("Proxy-Authorization")
+ r.Header.Del("Proxy-Connection")
+ r.Close = true
+ // Count request headers and bodies as well as the response. Reserve in the writer before any bytes leave.
+ err = r.Write(&budgetWriter{Writer: upstream, budget: e.Budget, cap: cap})
+ }
+ if err != nil {
+ _ = upstream.Close()
+ return nil, br, err
+ }
+ return upstream, br, nil
+}
+
+type budgetWriter struct {
+ io.Writer
+ budget *Budget
+ cap int64
+}
+
+func (w *budgetWriter) Write(b []byte) (int, error) {
+ if err := w.budget.Reserve(int64(len(b)), w.cap); err != nil {
+ return 0, err
+ }
+ return w.Writer.Write(b)
+}
+func (e *Endpoint) socks(ctx context.Context, c net.Conn, br *bufio.Reader, session, credential string, dial DialFunc) (net.Conn, error) {
+ head := make([]byte, 2)
+ if _, err := io.ReadFull(br, head); err != nil {
+ return nil, err
+ }
+ methods := make([]byte, int(head[1]))
+ if _, err := io.ReadFull(br, methods); err != nil {
+ return nil, err
+ }
+ found := false
+ for _, m := range methods {
+ if m == 2 {
+ found = true
+ }
+ }
+ if !found {
+ _, _ = c.Write([]byte{5, 255})
+ return nil, errors.New("authentication required")
+ }
+ _, _ = c.Write([]byte{5, 2})
+ if _, err := io.ReadFull(br, head); err != nil || head[0] != 1 {
+ return nil, errors.New("authentication required")
+ }
+ user := make([]byte, int(head[1]))
+ if _, err := io.ReadFull(br, user); err != nil {
+ return nil, err
+ }
+ n, err := br.ReadByte()
+ if err != nil {
+ return nil, err
+ }
+ pass := make([]byte, int(n))
+ if _, err = io.ReadFull(br, pass); err != nil {
+ return nil, err
+ }
+ if subtle.ConstantTimeCompare(user, []byte(session)) != 1 || subtle.ConstantTimeCompare(pass, []byte(credential)) != 1 {
+ _, _ = c.Write([]byte{1, 1})
+ return nil, errors.New("refused")
+ }
+ _, _ = c.Write([]byte{1, 0})
+ req := make([]byte, 4)
+ if _, err = io.ReadFull(br, req); err != nil {
+ return nil, err
+ }
+ if req[0] != 5 || req[1] != 1 || req[2] != 0 {
+ _, _ = c.Write([]byte{5, 7, 0, 1, 0, 0, 0, 0, 0, 0})
+ return nil, errors.New("CONNECT only")
+ }
+ var host string
+ switch req[3] {
+ case 3:
+ n, err = br.ReadByte()
+ if err != nil || n == 0 {
+ return nil, errors.New("invalid host")
+ }
+ b := make([]byte, int(n))
+ if _, err = io.ReadFull(br, b); err != nil {
+ return nil, err
+ }
+ host = string(b)
+ case 1, 4:
+ size := 4
+ if req[3] == 4 {
+ size = 16
+ }
+ b := make([]byte, size)
+ if _, err = io.ReadFull(br, b); err != nil {
+ return nil, err
+ }
+ host = net.IP(b).String()
+ default:
+ return nil, errors.New("invalid address")
+ }
+ if _, err = io.ReadFull(br, head); err != nil {
+ return nil, err
+ }
+ up, err := dial(ctx, net.JoinHostPort(host, strconv.Itoa(int(binary.BigEndian.Uint16(head)))))
+ if err != nil {
+ _, _ = c.Write([]byte{5, 2, 0, 1, 0, 0, 0, 0, 0, 0})
+ return nil, err
+ }
+ _, err = c.Write([]byte{5, 0, 0, 1, 0, 0, 0, 0, 0, 0})
+ if err != nil {
+ up.Close()
+ return nil, err
+ }
+ return up, nil
+}
+
+// Bridge is bound to one isolated desk. It speaks an ordinary HTTP proxy to Chrome and the authenticated
+// WebSocket byte pipe to Link. It never changes to a direct route after Link accepted a connection.
+func Bridge(ctx context.Context, listener net.Listener, p Plan, testDial ...DialFunc) *http.Server {
+ directDial := PublicDial
+ if len(testDial) > 0 {
+ directDial = testDial[0]
+ }
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ var remote net.Conn
+ host := r.URL.Hostname()
+ if r.Method == "CONNECT" {
+ host, _, _ = net.SplitHostPort(r.Host)
+ }
+ direct := p.Mode == "direct"
+ for _, s := range p.Excluded {
+ if strings.EqualFold(host, s) || strings.HasSuffix(strings.ToLower(host), "."+s) {
+ direct = true
+ }
+ }
+ if !direct {
+ ws, resp, err := (&websocket.Dialer{HandshakeTimeout: 10 * time.Second}).DialContext(r.Context(), p.Relay, http.Header{"X-OV-Session": {p.Session}, "X-OV-Egress": {p.Credential}})
+ if err == nil {
+ remote = &wsConn{Conn: ws}
+ } else if p.Offline == "openvibe" && (resp == nil || resp.StatusCode == 503) {
+ direct = true
+ } else {
+ http.Error(w, "Your computer is unavailable.", 503)
+ return
+ }
+ }
+ if direct {
+ address := r.Host
+ if r.Method != "CONNECT" && r.URL.Port() == "" {
+ address = net.JoinHostPort(r.URL.Hostname(), "80")
+ }
+ var err error
+ remote, err = directDial(r.Context(), address)
+ if err != nil {
+ http.Error(w, "Site unavailable", 403)
+ return
+ }
+ }
+ defer remote.Close()
+ if !direct {
+ r.Header.Set("Proxy-Authorization", "Basic "+base64.StdEncoding.EncodeToString([]byte(p.Session+":"+p.Credential)))
+ if err := r.WriteProxy(remote); err != nil {
+ return
+ }
+ } else if r.Method != "CONNECT" {
+ r.Header.Del("Proxy-Authorization")
+ r.Close = true
+ if err := r.Write(remote); err != nil {
+ return
+ }
+ }
+ if r.Method != "CONNECT" {
+ response, err := http.ReadResponse(bufio.NewReader(remote), r)
+ if err != nil {
+ http.Error(w, "Site unavailable", 502)
+ return
+ }
+ defer response.Body.Close()
+ for k, v := range response.Header {
+ w.Header()[k] = v
+ }
+ w.WriteHeader(response.StatusCode)
+ _, _ = io.Copy(w, response.Body)
+ return
+ }
+ client, buffer, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ return
+ }
+ defer client.Close()
+ if direct {
+ _, _ = io.WriteString(client, "HTTP/1.1 200 Connection Established\r\n\r\n")
+ }
+ done := make(chan struct{}, 1)
+ go func() { _, _ = io.Copy(remote, buffer); _ = remote.Close(); _ = client.Close(); done <- struct{}{} }()
+ _, _ = io.Copy(client, remote)
+ _ = remote.Close()
+ _ = client.Close()
+ <-done
+ })
+ server := &http.Server{Handler: handler, ReadHeaderTimeout: 10 * time.Second, MaxHeaderBytes: 32 << 10, BaseContext: func(net.Listener) context.Context { return ctx }}
+ go server.Serve(listener)
+ go func() { <-ctx.Done(); _ = server.Close() }()
+ return server
+}
+
+func BrowserArgs(address string) []string {
+ host, _, _ := net.SplitHostPort(address)
+ return []string{"--proxy-server=http://" + address, "--proxy-bypass-list=<-loopback>", "--host-resolver-rules=MAP * ~NOTFOUND, EXCLUDE " + host, "--force-webrtc-ip-handling-policy=disable_non_proxied_udp", "--disable-quic", "--disable-features=DnsOverHttps,UseDnsHttpsSvcbAlpn"}
+}
+
+func readRequest(br *bufio.Reader) (*http.Request, *bufio.Reader, error) {
+ var header []byte
+ for {
+ line, err := br.ReadSlice('\n')
+ if err != nil {
+ return nil, br, err
+ }
+ header = append(header, line...)
+ if len(header) > 32<<10 {
+ return nil, br, errors.New("headers too large")
+ }
+ if string(line) == "\r\n" {
+ break
+ }
+ }
+ next := bufio.NewReader(io.MultiReader(bytes.NewReader(header), br))
+ r, err := http.ReadRequest(next)
+ return r, next, err
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,107 @@
@@ -0,0 +1,107 @@
+package egress
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "path/filepath"
+ "strings"
+ "sync/atomic"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/connect"
+ "github.com/OpenVibers/OpenVibe.Node/internal/credentials"
+ "github.com/OpenVibers/OpenVibe.Node/internal/link"
+)
+
+// Serve exposes no public listening port. Connect owns the authenticated incoming byte pipe.
+func Serve(ctx context.Context, creds *credentials.Credentials, stateDir string) error {
+ if creds == nil || !creds.NetworkPaired() {
+ return nil
+ }
+ budget, err := NewBudget(filepath.Join(stateDir, "link-egress-usage.json"))
+ if err != nil {
+ return err
+ }
+ listener, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ return err
+ }
+ defer listener.Close()
+ client := NewClient(creds)
+ server := &http.Server{Handler: &Endpoint{Client: client, Node: creds.Principal, Budget: budget, Pause: Paused}, ReadHeaderTimeout: 10 * time.Second, MaxHeaderBytes: 8 << 10}
+ go server.Serve(listener)
+ defer server.Close()
+ networkToken, err := client.Tokens.Token(ctx)
+ if err != nil {
+ return errors.New("computer sign-in unavailable")
+ }
+ connectTokens := link.NewTokenSourceFor(creds, nil, "openvibe-link/egress", link.ConnectAudience)
+ token, err := connectTokens.Token(ctx)
+ if err != nil {
+ return errors.New("computer sign-in unavailable")
+ }
+ data, _ := json.Marshal(map[string]any{"port": listener.Addr().(*net.TCPAddr).Port, "network_token": networkToken.Reveal()})
+ req, _ := http.NewRequestWithContext(ctx, "POST", "https://connect.openvibe.link/api/v1/node/egress-share", bytes.NewReader(data))
+ req.Header.Set("Authorization", "Bearer "+token.Reveal())
+ req.Header.Set("Content-Type", "application/json")
+ resp, err := (&http.Client{Timeout: 15 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}).Do(req)
+ if err != nil {
+ return errors.New("Link connection unavailable")
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != 200 {
+ return errors.New("Link connection refused")
+ }
+ var result struct {
+ Share connect.NodeShare `json:"share"`
+ }
+ if err = json.NewDecoder(io.LimitReader(resp.Body, 16<<10)).Decode(&result); err != nil {
+ return errors.New("invalid Link connection")
+ }
+ var online atomic.Bool
+ heartbeat := func() {
+ if !online.Load() {
+ return
+ }
+ relay := "wss://" + result.Share.Host + "/__ov_egress"
+ _ = client.Call(ctx, "PUT", "/api/v1/node/self/egress/heartbeat", map[string]string{"relay_url": relay}, nil, true)
+ }
+ go func() {
+ ticker := time.NewTicker(15 * time.Second)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return
+ case <-ticker.C:
+ heartbeat()
+ }
+ }
+ }()
+ return connect.Run(ctx, connect.Share{ID: result.Share.ID, Port: listener.Addr().(*net.TCPAddr).Port}, connect.Options{
+ LocalHost: "127.0.0.1", NodeToken: func(ctx context.Context) (string, error) { t, e := connectTokens.Token(ctx); return t.Reveal(), e },
+ OnReady: func(connect.Ready) { online.Store(true); go heartbeat() }, OnDisconnect: func() { online.Store(false) },
+ })
+}
+
+// A transient service failure retries without taking down the person's Node. Errors contain no response bodies.
+func Run(ctx context.Context, creds *credentials.Credentials, stateDir string) {
+ for {
+ _ = Serve(ctx, creds, stateDir)
+ if creds == nil || !creds.NetworkPaired() {
+ return
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case <-time.After(15 * time.Second):
+ }
+ }
+}
+
+func IsCloudDesk(name string) bool { return strings.HasPrefix(name, "desk-s") && len(name) == 22 }
@@ -9,6 +9,7 @@ import (
@@ -9,6 +9,7 @@ import (
"encoding/json"
"encoding/json"
"errors"
"errors"
"fmt"
"fmt"
+ "github.com/OpenVibers/OpenVibe.Node/internal/egress"
"os"
"os"
"path/filepath"
"path/filepath"
"runtime"
"runtime"
@@ -189,6 +190,9 @@ func savePanel(dir string, off map[string]bool, first *panelFirstRun) error {
@@ -189,6 +190,9 @@ func savePanel(dir string, off map[string]bool, first *panelFirstRun) error {
// startPanel serves Link's window until ctx ends. A failure leaves the Node running without it.
// startPanel serves Link's window until ctx ends. A failure leaves the Node running without it.
func (n *Node) startPanel(ctx context.Context) {
func (n *Node) startPanel(ctx context.Context) {
+ if !n.opt.NoShares {
+ go egress.Run(ctx, n.opt.Creds, n.opt.Paths.StateDir)
+ }
token, err := panel.NewToken()
token, err := panel.NewToken()
if err != nil {
if err != nil {
n.log.Warn("Link's window off: no token", "err", err)
n.log.Warn("Link's window off: no token", "err", err)
@@ -13,6 +13,7 @@ import (
@@ -13,6 +13,7 @@ import (
"syscall"
"syscall"
"github.com/OpenVibers/OpenVibe.Node/internal/config"
"github.com/OpenVibers/OpenVibe.Node/internal/config"
+ "github.com/OpenVibers/OpenVibe.Node/internal/egress"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
"github.com/OpenVibers/OpenVibe.Node/internal/protocol"
)
)
@@ -21,8 +22,10 @@ import (
@@ -21,8 +22,10 @@ import (
// namespaces, as worker.run_as, inside its own cgroup with its own memory, CPU and process limits, seeing only the
// namespaces, as worker.run_as, inside its own cgroup with its own memory, CPU and process limits, seeing only the
// system paths, its own /tmp (its profile) and the read-only Binds it names, and reaching only the public internet.
// system paths, its own /tmp (its profile) and the read-only Binds it names, and reaching only the public internet.
type DeskSpec struct {
type DeskSpec struct {
- Argv []string
- Env []string
+ // Proxy starts the authenticated bridge on this desk's isolated gateway.
+ Proxy func(string) (string, func(), error)
+ Argv []string
+ Env []string
// Binds are host paths the program needs beyond the system paths: the browser's install directory, /etc/fonts.
// Binds are host paths the program needs beyond the system paths: the browser's install directory, /etc/fonts.
// Read-only, and only under deskBindRoots (CheckDeskBinds).
// Read-only, and only under deskBindRoots (CheckDeskBinds).
Binds []string
Binds []string
@@ -123,10 +126,32 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, err
@@ -123,10 +126,32 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, err
sb.goW, sb.child = gw, append(sb.child, gr)
sb.goW, sb.child = gw, append(sb.child, gr)
// A browser keeps hundreds of files open and maps tens of GB of address space it never uses: no RLIMIT_AS (the
// A browser keeps hundreds of files open and maps tens of GB of address space it never uses: no RLIMIT_AS (the
// cgroup's memory.max is the limit) and no RLIMIT_CPU (cpu.max is: a desk lives for hours).
// cgroup's memory.max is the limit) and no RLIMIT_CPU (cpu.max is: a desk lives for hours).
+ if d.Egress != "" {
+ if sb.egress, err = newEgress(cfg); err != nil {
+ return 0, fmt.Errorf("desk egress: %w", err)
+ }
+ }
+ if d.Proxy != nil {
+ if sb.egress == nil {
+ return 0, errors.New("a Link desk requires its private connection")
+ }
+ address, closeProxy, err := d.Proxy(sb.egress.host.String())
+ if err != nil {
+ return 0, err
+ }
+ defer closeProxy()
+ closeRules, err := sb.egress.allowProxy(address)
+ if err != nil {
+ return 0, err
+ }
+ defer closeRules()
+ // These mandatory flags are last: no caller-supplied flag can override the proxy or leak protection.
+ d.Argv = append(d.Argv, egress.BrowserArgs(address)...)
+ }
spec := sandboxSpec{Root: root, Work: "/tmp/desk", Disk: d.DiskBytes, Links: links, Argv: d.Argv, Env: d.Env, Pipe: true,
spec := sandboxSpec{Root: root, Work: "/tmp/desk", Disk: d.DiskBytes, Links: links, Argv: d.Argv, Env: d.Env, Pipe: true,
Limits: []rlimit{{rlimitCORE, 0, 0}, {rlimitNOFILE, 8192, 8192}, {rlimitNPROC, uint64(d.Pids), uint64(d.Pids)},
Limits: []rlimit{{rlimitCORE, 0, 0}, {rlimitNOFILE, 8192, 8192}, {rlimitNPROC, uint64(d.Pids), uint64(d.Pids)},
{rlimitFSIZE, uint64(d.DiskBytes), uint64(d.DiskBytes)}}}
{rlimitFSIZE, uint64(d.DiskBytes), uint64(d.DiskBytes)}}}
- if d.Egress != "" {
+ if d.Egress != "" && d.Proxy == nil {
spec.Resolvers = cfg.DNS
spec.Resolvers = cfg.DNS
}
}
for _, b := range binds {
for _, b := range binds {
@@ -140,11 +165,6 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, err
@@ -140,11 +165,6 @@ func RunDesk(ctx context.Context, cfg config.WorkerConfig, d DeskSpec) (int, err
return 0, err
return 0, err
}
}
sb.child = append(sb.child, sf)
sb.child = append(sb.child, sf)
- if d.Egress != "" {
- if sb.egress, err = newEgress(cfg); err != nil {
- return 0, fmt.Errorf("desk egress: %w", err)
- }
- }
if sb.cg, err = newJobCgroup(caps, protocol.JobLimits{MemBytes: d.MemBytes}); err != nil {
if sb.cg, err = newJobCgroup(caps, protocol.JobLimits{MemBytes: d.MemBytes}); err != nil {
// Its cgroup is made beside the Node's own leaf, which the Node sets up when its worker is on (the probe);
// Its cgroup is made beside the Node's own leaf, which the Node sets up when its worker is on (the probe);
// a desk started by a plugin cannot rearrange the Node's cgroup itself.
// a desk started by a plugin cannot rearrange the Node's cgroup itself.
@@ -11,6 +11,7 @@ import (
@@ -11,6 +11,7 @@ import (
// DeskSpec is one desk's program as RunDesk runs it; off Linux a desk never runs sandboxed (desk_linux.go).
// DeskSpec is one desk's program as RunDesk runs it; off Linux a desk never runs sandboxed (desk_linux.go).
type DeskSpec struct {
type DeskSpec struct {
+ Proxy func(string) (string, func(), error)
Argv []string
Argv []string
Env []string
Env []string
Binds []string
Binds []string
@@ -7,6 +7,7 @@ import (
@@ -7,6 +7,7 @@ import (
"encoding/binary"
"encoding/binary"
"errors"
"errors"
"fmt"
"fmt"
+ "net"
"net/netip"
"net/netip"
"os"
"os"
"os/exec"
"os/exec"
@@ -114,7 +115,10 @@ func newEgress(cfg config.WorkerConfig) (*egressNet, error) {
@@ -114,7 +115,10 @@ func newEgress(cfg config.WorkerConfig) (*egressNet, error) {
e := &egressNet{tools: t, n: -1}
e := &egressNet{tools: t, n: -1}
vethMu.Lock()
vethMu.Lock()
for i := range vethMax {
for i := range vethMax {
- if n := (vethNext + i) % vethMax; !vethUsed[n] {
+ if n := (vethNext + i + os.Getpid()) % vethMax; !vethUsed[n] {
+ if _, e := net.InterfaceByName(vethPrefix + strconv.Itoa(n)); e == nil {
+ continue
+ }
e.n, vethUsed[n], vethNext = n, true, n+1
e.n, vethUsed[n], vethNext = n, true, n+1
break
break
}
}
@@ -264,6 +268,7 @@ func ruleset(cfg config.WorkerConfig, uplinks []string) string {
@@ -264,6 +268,7 @@ func ruleset(cfg config.WorkerConfig, uplinks []string) string {
w("\t}")
w("\t}")
w("\tchain input {")
w("\tchain input {")
w("\t\ttype filter hook input priority -10; policy accept;")
w("\t\ttype filter hook input priority -10; policy accept;")
+ w("\t\tmeta mark 0x4f564547 accept")
w("\t\tiifname \"%s*\" goto refuse", vethPrefix)
w("\t\tiifname \"%s*\" goto refuse", vethPrefix)
w("\t}")
w("\t}")
w("\tchain postrouting {")
w("\tchain postrouting {")
new file mode 100644
new file mode 100644
@@ -0,0 +1,43 @@
@@ -0,0 +1,43 @@
+//go:build linux
+
+package worker
+
+import (
+ "fmt"
+ "net"
+ "strconv"
+)
+
+// Each desk has one host-side proxy socket. Every other packet from its veth is refused,
+// including DNS, QUIC, IPv6 and WebRTC UDP. The proxy alone can dial the relay or an explicitly excluded site.
+// A separate table survives another worker's refresh of openvibe_worker.
+func (e *egressNet) proxyRules(port int) string {
+ name := "ov_link_" + strconv.Itoa(e.n)
+ return fmt.Sprintf(`table inet %s
+delete table inet %s
+table inet %s {
+ chain mark_proxy {
+ type filter hook prerouting priority -100; policy accept;
+ iifname "%s" ip saddr %s ip daddr %s tcp dport %d meta mark set 0x4f564547
+ }
+ chain no_forward {
+ type filter hook forward priority -20; policy accept;
+ iifname "%s" reject with icmpx admin-prohibited
+ }
+}
+`, name, name, name, e.hostIf(), e.job, e.host, port, e.hostIf())
+}
+func (e *egressNet) allowProxy(address string) (func(), error) {
+ host, p, err := net.SplitHostPort(address)
+ if err != nil || host != e.host.String() {
+ return nil, fmt.Errorf("desk proxy must bind its own gateway")
+ }
+ port, err := strconv.Atoi(p)
+ if err != nil || port < 1 || port > 65535 {
+ return nil, fmt.Errorf("invalid desk proxy port")
+ }
+ if err = runTool(e.proxyRules(port), e.tools.nft, "-f", "-"); err != nil {
+ return nil, err
+ }
+ return func() { _ = runTool("", e.tools.nft, "delete", "table", "inet", "ov_link_"+strconv.Itoa(e.n)) }, nil
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,31 @@
@@ -0,0 +1,31 @@
+//go:build linux
+
+package worker
+
+import (
+ "net/netip"
+ "strings"
+ "testing"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/config"
+)
+
+func TestLinkProxyFirewallOnlyAllowsItsOwnSocket(t *testing.T) {
+ e := &egressNet{n: 7, host: netip.MustParseAddr("169.254.240.29"), job: netip.MustParseAddr("169.254.240.30")}
+ rules := e.proxyRules(32123)
+ for _, required := range []string{`iifname "ovw7" ip saddr 169.254.240.30 ip daddr 169.254.240.29 tcp dport 32123 meta mark set 0x4f564547`, `iifname "ovw7" reject with icmpx admin-prohibited`, `hook forward priority -20`} {
+ if !strings.Contains(rules, required) {
+ t.Fatalf("missing restriction: %s", required)
+ }
+ }
+ base := ruleset(config.WorkerConfig{Egress: config.EgressPublic}, []string{"eth0"})
+ if !strings.Contains(base, "meta mark 0x4f564547 accept") {
+ t.Fatal("desk cannot reach its proxy")
+ }
+ if strings.Contains(rules, "udp dport") || strings.Contains(rules, "ip6 daddr") {
+ t.Fatal("direct DNS/WebRTC/IPv6 allowed")
+ }
+ if _, err := e.allowProxy("127.0.0.1:32123"); err == nil {
+ t.Fatal("another gateway accepted")
+ }
+}
@@ -1,6 +1,6 @@
@@ -1,6 +1,6 @@
import sys
import sys
-from openvibe_computer import Computer
+from openvibe_computer.link_egress import LinkComputer as Computer
from openvibe_plugin import run
from openvibe_plugin import run
sys.exit(run(Computer()))
sys.exit(run(Computer()))
new file mode 100644
new file mode 100644
@@ -0,0 +1,16 @@
@@ -0,0 +1,16 @@
+"""Cloud desk startup integration, kept separate from the browser's blind-fill code.
+
+Only a desk name crosses the process boundary. The Node reads its assigned session from Network;
+credentials stay in the Node's proxy process and never enter Chrome's arguments or the model's tools.
+"""
+from openvibe_computer import Computer
+
+
+class LinkComputer(Computer):
+ def _desk_browser(self, desk, binary=None, env=None):
+ browser = super()._desk_browser(desk, binary=binary, env=env)
+ if self._shared() and desk.startswith("desk-s") and len(desk) == 22:
+ if browser.sandbox and "--link-egress-desk" not in browser.sandbox:
+ separator = browser.sandbox.index("--")
+ browser.sandbox[separator:separator] = ["--link-egress-desk", desk]
+ return browser
new file mode 100644
new file mode 100644
@@ -0,0 +1,35 @@
@@ -0,0 +1,35 @@
+import os
+import sys
+import unittest
+from types import SimpleNamespace
+from unittest import mock
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+sys.path.insert(0, os.path.dirname(HERE))
+sys.path.insert(0, os.path.join(os.path.dirname(os.path.dirname(HERE)), "sdk"))
+from openvibe_computer import Computer
+from openvibe_computer.link_egress import LinkComputer
+
+
+class LinkEgressTest(unittest.TestCase):
+ def test_cloud_browser_gets_its_own_desk_before_start(self):
+ plugin = LinkComputer()
+ browser = SimpleNamespace(sandbox=["node", "desk-sandbox", "--egress", "public", "--"])
+ desk = "desk-s0123456789abcdef"
+ with mock.patch.object(Computer, "_desk_browser", return_value=browser), mock.patch.object(plugin, "_shared", return_value=True):
+ self.assertIs(plugin._desk_browser(desk), browser)
+ self.assertEqual(browser.sandbox[-3:], ["--link-egress-desk", desk, "--"])
+ plugin._desk_browser(desk)
+ self.assertEqual(browser.sandbox.count("--link-egress-desk"), 1)
+ self.assertFalse(any("credential" in arg for arg in browser.sandbox))
+
+ def test_personal_desks_keep_their_normal_start(self):
+ plugin = LinkComputer()
+ browser = SimpleNamespace(sandbox=None)
+ with mock.patch.object(Computer, "_desk_browser", return_value=browser), mock.patch.object(plugin, "_shared", return_value=False):
+ self.assertIs(plugin._desk_browser("desk-personal"), browser)
+ self.assertIsNone(browser.sandbox)
+
+
+if __name__ == "__main__":
+ unittest.main()