"os/exec"
"os/user"
"path/filepath"
"runtime"
"sort"
"slices"
"strings"
"sync"
"time"
"github.com/klauspost/compress/zstd"
"github.com/samber/lo"
"tobi/internal/session"
"tobi/internal/target"
"tobi/internal/version"
)
//go:embed payloads/*
var embeddedPayloads embed.FS
// hosts the user has addressed with an explicit user@ prefix are remembered,
// so bare peer-name probes (picker listing, attach by peer) can authenticate
// without a User line in ssh_config
// so bare peer-name probes can authenticate without a User line in ssh_config
var hostMu sync.Mutex
func hostMapPath() string {
base, err := os.UserConfigDir()
if err != nil || base == "" {
return ""
if base, err := os.UserConfigDir(); err == nil && base != "" {
return filepath.Join(base, "tobi", "hosts.json")
}
return filepath.Join(base, "tobi", "hosts.json")
return ""
}
// loadHosts reads the remembered destinations per bare host, accepting the
// older single-string form
// older all-strings single form
func loadHosts() map[string][]string {
p := hostMapPath()
if p == "" {
return map[string][]string{}
m := map[string][]string{}
data, err := os.ReadFile(hostMapPath())
if err != nil || json.Unmarshal(data, &m) == nil {
return m
}
data, err := os.ReadFile(p)
if err != nil {
var old map[string]string
if json.Unmarshal(data, &old) != nil {
return map[string][]string{}
}
var raw map[string]json.RawMessage
if json.Unmarshal(data, &raw) != nil {
return map[string][]string{}
for host, dest := range old {
if dest != "" {
m[host] = []string{dest}
}
}
return lo.MapEntries(raw, func(host string, v json.RawMessage) (string, []string) {
return host, parseDests(v)
})
return m
}
// parseDests accepts the older single-string form too
func parseDests(v json.RawMessage) []string {
var one string
if json.Unmarshal(v, &one) == nil && one != "" {
return []string{one}
}
var many []string
if json.Unmarshal(v, &many) == nil {
return many
}
return nil
// remembered is the user@host list stored for a bare host
func remembered(host string) []string {
hostMu.Lock()
defer hostMu.Unlock()
return loadHosts()[host]
}
func saveHosts(m map[string][]string) {
p := hostMapPath()
if p == "" {
return
}
if err := os.MkdirAll(filepath.Dir(p), 0700); err == nil {
if data, err := json.Marshal(m); err == nil {
_ = os.WriteFile(p, data, 0600)
}
}
}
// peerIdent is this machine as a ssh destination: user@hostname
func peerIdent() string {
// peerEnv is the TOBI_PEER=... exec prefix that introduces this machine to
// remote RPCs as a ssh destination, so they can remember how to reach us
// back; "" when this machine is unidentifiable
func peerEnv() string {
host := session.ShortHostname()
if host == "" || host == "localhost" {
return ""
}
u := os.Getenv("USER")
if u == "" {
if cur, err := user.Current(); err == nil {
u = cur.Username
}
}
if u == "" {
if host == "" || host == "localhost" || u == "" {
return ""
}
return u + "@" + host
return "TOBI_PEER=" + u + "@" + host + " "
}
// peerEnv prefixes remote RPC execs with TOBI_PEER so the remote tobi can
// remember how to reach us back (the reverse of RememberHost)
func peerEnv() string {
if p := peerIdent(); p != "" {
return "TOBI_PEER=" + p + " "
}
return ""
}
// RememberHost records a user@host destination for its bare host, most
// recent first. a host may serve several users, each with its own sessions
func RememberHost(dest string) {
_, bare, ok := strings.Cut(dest, "@")
if !ok || bare == "" {
t := target.Parse(dest)
if t.User == "" || t.Host == "" {
return
}
hostMu.Lock()
defer hostMu.Unlock()
m := loadHosts()
if len(m[bare]) > 0 && m[bare][0] == dest {
if len(m[t.Host]) > 0 && m[t.Host][0] == dest {
return
}
rest := lo.Filter(m[bare], func(d string, _ int) bool { return d != dest })
m[bare] = append([]string{dest}, rest...)
if len(m[bare]) > 8 {
m[bare] = m[bare][:8]
m[t.Host] = append([]string{dest}, slices.DeleteFunc(m[t.Host], func(d string) bool { return d == dest })...)
if len(m[t.Host]) > 8 {
m[t.Host] = m[t.Host][:8]
}
saveHosts(m)
}
func saveHosts(m map[string][]string) {
p := hostMapPath()
if p == "" {
return
}
if data, err := json.Marshal(m); err == nil {
if os.MkdirAll(filepath.Dir(p), 0700) == nil {
_ = os.WriteFile(p, data, 0600)
}
}
}
// hostDest resolves a host to its ssh destination: the most recently used
// user@host for that bare host
func hostDest(host string) string {
if strings.Contains(host, "@") {
if t := target.Parse(host); t.User != "" {
RememberHost(host)
return host
}
hostMu.Lock()
defer hostMu.Unlock()
if list := loadHosts()[host]; len(list) > 0 && list[0] != "" {
if list := remembered(host); len(list) > 0 && list[0] != "" {
return list[0]
}
return host
}
// KnownDests lists every destination worth probing for a host: itself plus
// each remembered user@host, most recent first
func KnownDests(host string) []string {
hostMu.Lock()
defer hostMu.Unlock()
known := loadHosts()[host]
return lo.Uniq(append([]string{host}, lo.Filter(known, func(d string, _ int) bool {
return d != "" && d != host
})...))
known := slices.DeleteFunc(remembered(host), func(d string) bool { return d == "" || d == host })
return lo.Uniq(append([]string{host}, known...))
}
// ProbeDests is the list of destinations worth probing for a host: every
// remembered user when one exists, the bare host otherwise
func ProbeDests(host string) []string {
dests := KnownDests(host)
if len(dests) > 1 {
if dests := KnownDests(host); len(dests) > 1 {
return dests[1:]
}
return dests
return []string{host}
}
// QueryHost lists a host's sessions across every probed destination,
// deduplicated by bare host and id
func QueryHost(ctx context.Context, host string) []session.Session {
bare := host
if _, b, ok := strings.Cut(host, "@"); ok {
bare = b
}
dests := ProbeDests(bare)
t := target.Parse(host)
dests := ProbeDests(t.Host)
// an explicitly addressed user is probed even when never remembered
if host != bare {
if t.User != "" {
dests = lo.Uniq(append([]string{host}, dests...))
}
sessions := lo.FlatMap(dests, func(dest string, _ int) []session.Session {
list, err := QueryRemote(ctx, dest)
if err != nil {
return nil
var found []session.Session
for _, dest := range dests {
if list, err := QueryRemote(ctx, dest); err == nil {
found = append(found, list...)
}
return list
}
return lo.UniqBy(found, func(s session.Session) string {
return strings.ToLower(target.Parse(s.Host).Host + "." + s.ID)
})
return lo.UniqBy(sessions, func(s session.Session) string {
bare, _, _ := strings.Cut(s.Host, "@")
return strings.ToLower(bare + "." + s.ID)
})
}
const remotePathEnv = `PATH="$HOME/.local/bin:$PATH:$HOME/.nix-profile/bin:/opt/homebrew/bin:/usr/local/bin:$HOME/go/bin"`
// tssh exits 0 regardless of remote command failure, so remote steps confirm
// themselves with stdout markers instead of relying on exit codes.
// themselves with stdout markers instead of relying on exit codes
const (
haveMarker = "__TOBI_HAVE__"
uploadMarker = "__TOBI_UPLOADED__"
spawnMarker = "__TOBI_SPAWNED__"
)
const remotePathEnv = `PATH="$HOME/.local/bin:$PATH:$HOME/.nix-profile/bin:/opt/homebrew/bin:/usr/local/bin:$HOME/go/bin"`
// rpcCommand is the remote exec prefix for a tobi RPC: PATH fixup, peer
// identity, the RPC verb and optional session id
func rpcCommand(op, id string) string {
cmd := remotePathEnv + " " + peerEnv() + "TOBI_RPC=" + op
if id != "" {
cmd += " TOBI_ID=" + id
}
return cmd + " tobi"
}
// archAlias maps uname -m names onto GOARCH names; uname -s is already a
// GOOS name modulo case
var archAlias = map[string]string{"x86_64": "amd64", "aarch64": "arm64"}
var remoteInstalled sync.Map
func sshCommand(ctx context.Context, host, remoteCmd string, batch bool, connectTimeout string, master bool) *exec.Cmd {
// prefer tssh when installed so ssh_config options like UdpMode keep working
bin := "ssh"
if p, err := exec.LookPath("tssh"); err == nil {
bin = p
// sshOpts is the shape of a remote ssh exec: batch mode, master creation,
// connect timeout, and the remote command
type sshOpts struct {
cmd string
batch bool
master bool
timeout string
}
func sshCommand(ctx context.Context, dest string, opts sshOpts) *exec.Cmd {
// prefer tssh when installed so ssh_config options like UdpMode work
bin, err := exec.LookPath("tssh")
if err != nil {
bin = "ssh"
}
args := []string{"-T", "-o", "ConnectTimeout=" + connectTimeout}
if batch {
args := []string{"-T", "-o", "ConnectTimeout=" + opts.timeout}
if opts.batch {
args = append(args, "-o", "BatchMode=yes")
}
if home, err := os.UserHomeDir(); err == nil && home != "" {
// every call reuses an existing master; only user-initiated flows
// (attach, spawn) create one, so peer probes never litter masters
mode := lo.Ternary(master, "auto", "no")
ctl := filepath.Join(home, ".ssh", "tobi-control-%r@%h:%p")
args = append(args, "-o", "ControlMaster="+mode, "-o", "ControlPath="+ctl, "-o", "ControlPersist=60")
args = append(args, "-o", "ControlMaster="+lo.Ternary(opts.master, "auto", "no"),
"-o", "ControlPath="+ctl, "-o", "ControlPersist=60")
}
args = append(args, host, remoteCmd)
c := exec.CommandContext(ctx, bin, args...)
// a killed ssh can leave children holding the stdout pipe; bound how
// long Wait waits on them so probes settle at timeout + 500ms
c := exec.CommandContext(ctx, bin, append(args, dest, opts.cmd)...)
// a killed ssh can leave children holding the stdout pipe; bound the
// wait so probes settle at timeout + 500ms
c.WaitDelay = 500 * time.Millisecond
return c
}
type Stream struct {
io.Reader
io.WriteCloser
cmd *exec.Cmd
once sync.Once
}
func (s *Stream) Close() error {
s.once.Do(func() {
_ = s.WriteCloser.Close()
_ = s.cmd.Wait()
})
return nil
}
var DiscoverPeersFunc = DiscoverPeers
type tsPeer struct {
Online bool
OS string
DNSName string
HostName string
}
// Peer is a tailnet machine that could run tobi, with its mesh presence.
type Peer struct {
Name string
Online bool
}
// DiscoverPeers asks the tailscale CLI for linux/mac peers with their mesh
// presence, sorted by name; the CLI
// is exec'd so ssh_config options like UdpMode keep working
func DiscoverPeers(ctx context.Context) []Peer {
c := exec.CommandContext(ctx, "tailscale", "status", "--json")
c.WaitDelay = 500 * time.Millisecond
out, err := c.Output()
if err != nil {
return nil
}
var st struct{ Peer map[string]tsPeer }
if json.Unmarshal(out, &st) != nil {
return nil
}
self := strings.ToLower(session.ShortHostname())
peers := lo.FilterMap(lo.Values(st.Peer), func(p tsPeer, _ int) (Peer, bool) {
if p.OS != "linux" && p.OS != "macOS" && p.OS != "darwin" {
return Peer{}, false
}
name, _, _ := strings.Cut(p.DNSName, ".")
if name == "" {
name = p.HostName
}
return Peer{Name: name, Online: p.Online}, name != "" && !strings.EqualFold(name, self)
})
sort.Slice(peers, func(i, j int) bool { return peers[i].Name < peers[j].Name })
return peers
}
// ErrUnreachable means the host could not be reached over ssh at all
// (connect timeout, refused, auth): the difference between "machine is
// away" and "machine answered but has nothing"
// (connect timeout, refused, auth): the difference between "machine is away"
// and "machine answered but has nothing"
var ErrUnreachable = errors.New("unreachable")
func sshRun(ctx context.Context, host, cmd, timeout string, in io.Reader, master bool) (string, error) {
c := sshCommand(ctx, host, cmd, true, timeout, master)
func sshRun(ctx context.Context, dest string, in io.Reader, opts sshOpts) (string, error) {
c := sshCommand(ctx, dest, opts)
c.Stdin = in
var out, errBuf bytes.Buffer
c.Stdout = &out
c.Stderr = &errBuf
c.Stdout, c.Stderr = &out, &errBuf
if err := c.Run(); err != nil {
// ssh reports transport-level failures (connect timeout, refused,
// auth) as exit 255; remote command failures pass their own code
// through, and a killed probe reports a signal
// exit 255 is a transport failure (timeout, refused, auth); remote
// command failures pass their own code through
var ee *exec.ExitError
if (errors.As(err, &ee) && ee.ExitCode() == 255) || errors.Is(err, context.DeadlineExceeded) || ctx.Err() != nil {
err = fmt.Errorf("%w: %w", ErrUnreachable, err)
}
}
return out.String(), nil
}
func getBinaryPayload(targetOS, targetArch string) ([]byte, error) {
base := fmt.Sprintf("payloads/tobi-%s-%s", targetOS, targetArch)
if data, err := embeddedPayloads.ReadFile(base + ".zst"); err == nil && len(data) > 0 {
func getBinaryPayload(goos, arch string) ([]byte, error) {
if data, err := embeddedPayloads.ReadFile(fmt.Sprintf("payloads/tobi-%s-%s.zst", goos, arch)); err == nil && len(data) > 0 {
if zr, err := zstd.NewReader(bytes.NewReader(data)); err == nil {
defer zr.Close()
return io.ReadAll(zr)
}
}
if targetOS == runtime.GOOS && targetArch == runtime.GOARCH {
if goos == runtime.GOOS && arch == runtime.GOARCH {
self, err := os.Executable()
if err != nil {
self = os.Args[0]
}
return os.ReadFile(self)
}
return nil, fmt.Errorf("no binary available for %s/%s (build the carrier slices: nix build or scripts/build-universal.sh)", targetOS, targetArch)
return nil, fmt.Errorf("no binary available for %s/%s (build the carrier slices: nix build or scripts/build-universal.sh)", goos, arch)
}
const uploadCmd = "mkdir -p ~/.local/bin && cat > ~/.local/bin/tobi.tmp && chmod 755 ~/.local/bin/tobi.tmp && mv -f ~/.local/bin/tobi.tmp ~/.local/bin/tobi"
func EnsureRemote(ctx context.Context, host string) error {
host = hostDest(host)
if _, ok := remoteInstalled.Load(host); ok {
dest := hostDest(host)
if _, ok := remoteInstalled.Load(dest); ok {
return nil
}
// skip the upload only when the host already runs THIS build; an older
// (or newer, or foreign) install gets upgraded in place
checkCmd := fmt.Sprintf(`%s %scommand -v tobi >/dev/null 2>&1 && TOBI_RPC=version tobi 2>/dev/null`, remotePathEnv, peerEnv())
if out, err := sshRun(ctx, host, checkCmd, "3", nil, true); err == nil && strings.TrimSpace(out) == version.Version {
remoteInstalled.Store(host, true)
check := remotePathEnv + " " + peerEnv() + `command -v tobi >/dev/null 2>&1 && TOBI_RPC=version tobi`
if out, err := sshRun(ctx, dest, nil, sshOpts{cmd: check, batch: true, master: true, timeout: "3"}); err == nil && strings.TrimSpace(out) == version.Version {
remoteInstalled.Store(dest, true)
return nil
}
out, err := sshRun(ctx, host, "uname -s && uname -m", "3", nil, false)
out, err := sshRun(ctx, dest, nil, sshOpts{cmd: "uname -s && uname -m", batch: true, timeout: "3"})
if err != nil {
return fmt.Errorf("detect %s: %w", host, err)
return fmt.Errorf("detect %s: %w", dest, err)
}
fields := strings.Fields(out)
if len(fields) < 2 {
return fmt.Errorf("unexpected uname from %s: %q", host, out)
osName, arch, ok := strings.Cut(strings.TrimSpace(out), "\n")
if !ok {
return fmt.Errorf("unexpected uname from %s: %q", dest, out)
}
targetOS := strings.ToLower(fields[0])
targetArch := strings.ToLower(fields[1])
if a := map[string]string{"x86_64": "amd64", "aarch64": "arm64"}[targetArch]; a != "" {
targetArch = a
osName, arch = strings.ToLower(osName), strings.ToLower(arch)
if a := archAlias[arch]; a != "" {
arch = a
}
binData, err := getBinaryPayload(targetOS, targetArch)
bin, err := getBinaryPayload(osName, arch)
if err != nil {
return fmt.Errorf("prepare tobi for %s (%s/%s): %w", host, targetOS, targetArch, err)
return fmt.Errorf("prepare tobi for %s (%s/%s): %w", dest, osName, arch, err)
}
out, err = sshRun(ctx, host, uploadCmd+" && echo "+uploadMarker, "5", bytes.NewReader(binData), false)
out, err = sshRun(ctx, dest, bytes.NewReader(bin), sshOpts{cmd: uploadCmd + " && echo " + uploadMarker, batch: true, timeout: "5"})
if err != nil {
return fmt.Errorf("upload tobi to %s: %w", host, err)
return fmt.Errorf("upload tobi to %s: %w", dest, err)
}
if !strings.Contains(out, uploadMarker) {
return fmt.Errorf("upload tobi to %s: upload did not complete", host)
return fmt.Errorf("upload tobi to %s: upload did not complete", dest)
}
remoteInstalled.Store(host, true)
remoteInstalled.Store(dest, true)
return nil
}
// QueryRemote lists a host's sessions without touching its tobi install;
// a missing or failing tobi is just "no sessions", not an error
func QueryRemote(ctx context.Context, host string) ([]session.Session, error) {
host = hostDest(host)
out, err := sshRun(ctx, host, fmt.Sprintf(`%s %sTOBI_RPC=list tobi`, remotePathEnv, peerEnv()), "10", nil, false)
dest := hostDest(host)
out, err := sshRun(ctx, dest, nil, sshOpts{cmd: rpcCommand("list", ""), batch: true, timeout: "10"})
if err != nil {
return nil, err
}
var list []session.Session
if err := json.Unmarshal([]byte(strings.TrimSpace(out)), &list); err != nil {
return nil, err
}
for i := range list {
list[i].Host = host
list[i].Host = dest
}
return list, nil
}
// AttachRemote installs tobi on the host if needed, then opens an attach
// session as a raw byte stream
func AttachRemote(host, id string) (io.ReadWriteCloser, error) {
ctx := context.Background()
host = hostDest(host)
if err := EnsureRemote(ctx, host); err != nil {
dest := hostDest(host)
if err := EnsureRemote(ctx, dest); err != nil {
return nil, err
}
c := sshCommand(ctx, host, fmt.Sprintf(`%s %sTOBI_RPC=attach TOBI_ID=%s tobi`, remotePathEnv, peerEnv(), id), true, "2", false)
c := sshCommand(ctx, dest, sshOpts{cmd: rpcCommand("attach", id), batch: true, timeout: "2"})
in, err := c.StdinPipe()
if err != nil {
return nil, err
}
out, err := c.StdoutPipe()
if err != nil {
return nil, err
}
// best-effort channel: unreachable peers are expected and their client
// warnings are noise
c.Stderr = nil
c.Stderr = nil // unreachable peers are expected; their stderr is noise
if err := c.Start(); err != nil {
return nil, err
}
return &Stream{Reader: out, WriteCloser: in, cmd: c}, nil
}
// Stream is a live remote attach channel
type Stream struct {
io.Reader
io.WriteCloser
cmd *exec.Cmd
once sync.Once
}
func (s *Stream) Close() error {
s.once.Do(func() {
_ = s.WriteCloser.Close()
_ = s.cmd.Wait()
})
return nil
}
var DiscoverPeersFunc = DiscoverPeers
type tsPeer struct {
Online bool
OS string
DNSName string
HostName string
}
// Peer is a tailnet machine that could run tobi, with its mesh presence.
type Peer struct {
Name string
Online bool
}
// DiscoverPeers asks the tailscale CLI for linux/mac peers with their mesh
// presence, sorted by name; the CLI is exec'd so ssh_config options like
// UdpMode keep working
func DiscoverPeers(ctx context.Context) []Peer {
c := exec.CommandContext(ctx, "tailscale", "status", "--json")
c.WaitDelay = 500 * time.Millisecond
out, err := c.Output()
if err != nil {
return nil
}
var st struct{ Peer map[string]tsPeer }
if json.Unmarshal(out, &st) != nil {
return nil
}
self := strings.ToLower(session.ShortHostname())
peers := lo.FilterMap(lo.Values(st.Peer), func(p tsPeer, _ int) (Peer, bool) {
if p.OS != "linux" && p.OS != "macOS" && p.OS != "darwin" {
return Peer{}, false
}
name, _, _ := strings.Cut(p.DNSName, ".")
if name == "" {
name = p.HostName
}
return Peer{Name: name, Online: p.Online}, name != "" && !strings.EqualFold(name, self)
})
slices.SortFunc(peers, func(a, b Peer) int { return strings.Compare(a.Name, b.Name) })
return peers
}
func RunRPC(op, id string) error {
// the calling tobi introduced itself: remember how to reach it back,
// so its sessions show up in our picker without manual user@ targets
if peer := os.Getenv("TOBI_PEER"); strings.Contains(peer, "@") {
if _, bare, _ := strings.Cut(peer, "@"); bare != "" && !strings.EqualFold(bare, session.ShortHostname()) {
RememberHost(peer)
}
if t := target.Parse(os.Getenv("TOBI_PEER")); t.User != "" && t.Host != "" && !strings.EqualFold(t.Host, session.ShortHostname()) {
RememberHost(t.Dest())
}
switch op {
case "list":
return json.NewEncoder(os.Stdout).Encode(session.ListLocal())
case "spawn":
case "version":
fmt.Println(version.Version)
return nil
}
conn, err := net.Dial("unix", session.SocketPath(id))
if err != nil {
return err
}
defer conn.Close()
done := make(chan struct{}, 2)
pump := func(dst io.Writer, src io.Reader) {
_, _ = io.Copy(dst, src)
_ = conn.Close()
<-done
return nil
}
type spawnRequest struct {
ID string `json:"id"`
Argv []string `json:"argv"`
}
func validateSpawnRequest(req spawnRequest) error {
if !session.ValidateID(req.ID) {
return fmt.Errorf("invalid session id %q", req.ID)
}
}
return session.EnsureOwner(req.ID, req.Argv)
}
type spawnRequest struct {
ID string `json:"id"`
Argv []string `json:"argv"`
}
func SpawnRemote(host, id string, cmdArgs []string) error {
if !session.ValidateID(id) {
return fmt.Errorf("invalid session id: %q", id)
}
if len(cmdArgs) == 0 {
return errors.New("empty command")
}
host = hostDest(host)
if err := EnsureRemote(context.Background(), host); err != nil {
dest := hostDest(host)
if err := EnsureRemote(context.Background(), dest); err != nil {
return err
}
req, err := json.Marshal(spawnRequest{ID: id, Argv: cmdArgs})
if err != nil {
return err
}
c := sshCommand(context.Background(), host, fmt.Sprintf(`%s %sTOBI_RPC=spawn tobi && echo %s`, remotePathEnv, peerEnv(), spawnMarker), false, "5", true)
c.Stdin = bytes.NewReader(append(req, '\n'))
var out, errBuf bytes.Buffer
c.Stdout, c.Stderr = &out, &errBuf
if err := c.Run(); err != nil {
if msg := strings.TrimSpace(errBuf.String()); msg != "" {
return fmt.Errorf("spawn on %s: %s", host, msg)
}
return fmt.Errorf("spawn on %s: %w", host, err)
out, err := sshRun(context.Background(), dest, bytes.NewReader(append(req, '\n')), sshOpts{
cmd: rpcCommand("spawn", "") + " && echo " + spawnMarker, timeout: "5", master: true})
if err == nil && !strings.Contains(out, spawnMarker) {
err = fmt.Errorf("did not confirm: %s", strings.TrimSpace(out))
}
if !strings.Contains(out.String(), spawnMarker) {
return fmt.Errorf("spawn on %s did not confirm: %s", host, strings.TrimSpace(errBuf.String()))
if err != nil {
return fmt.Errorf("spawn on %s: %w", dest, err)
}
return nil
}