diff --git a/internal/ipc/client.go b/internal/ipc/client.go index ed4a3d1..5d3bd38 100644 --- a/internal/ipc/client.go +++ b/internal/ipc/client.go @@ -8,13 +8,15 @@ import ( "net" "sync" "time" + + "github.com/kami/maven/internal/netaddr" ) -// Client — the module side of the boundary. Wraps a unix-socket connection -// and satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is -// agnostic to whether it's been wired in-process (tests / daemon-embedded) -// or over this socket (full topology). The swappability is the seam auth -// will insert into without touching module code. +// Client — the module side of the boundary. Wraps a connection to core and +// satisfies CoreAPI, so a module imports ipc, holds a CoreAPI, and is +// agnostic to whether it's been wired in-process (tests / daemon-embedded), +// over a local unix socket, or over tcp to another host. The swappability is +// the seam auth will insert into without touching module code. // // One Client ⇒ one conn ⇒ one concurrent request at a time. A module that // wants parallel requests opens one Client per goroutine; the store is the @@ -22,7 +24,8 @@ import ( // per-Client lock keeps frame interleaving impossible by construction. type Client struct { conn net.Conn - path string // kept so a dropped conn can be re-dialed (core restart) + path string // the address as configured, kept for errors and logs + addr netaddr.Addr // parsed, so a dropped conn can be re-dialed (core restart) mu sync.Mutex } @@ -81,14 +84,22 @@ var readOnlyMethods = map[Method]bool{ MethodPing: true, } -// Dial connects to a core socket at path and returns a Client. The module -// owns its Client lifecycle; Close on shutdown. +// Dial connects to core at path and returns a Client. The module owns its +// Client lifecycle; Close on shutdown. +// +// path is a netaddr seam address: a bare path is the unix socket it has +// always been, and "tcp://host:port?token=..." reaches a core on another +// host. See internal/netaddr. func Dial(path string) (*Client, error) { - c, err := net.Dial("unix", path) + addr, err := netaddr.Parse(path) if err != nil { - return nil, fmt.Errorf("ipc: dial %s: %w", path, err) + return nil, err } - return &Client{conn: c, path: path}, nil + c, err := netaddr.Dial(addr) + if err != nil { + return nil, fmt.Errorf("ipc: dial %s: %w", addr, err) + } + return &Client{conn: c, path: path, addr: addr}, nil } func (c *Client) Close() error { @@ -189,9 +200,9 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { // re-dials clean. Caller holds c.mu. func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error { if c.conn == nil { - conn, err := net.Dial("unix", c.path) + conn, err := netaddr.Dial(c.addr) if err != nil { - return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.path, err) + return fmt.Errorf("%w: dial %s: %v", errWriteLost, c.addr, err) } c.conn = conn } diff --git a/internal/ipc/server.go b/internal/ipc/server.go index a7ed978..8e991e4 100644 --- a/internal/ipc/server.go +++ b/internal/ipc/server.go @@ -8,11 +8,11 @@ import ( "fmt" "log" "net" - "os" "sync" "sync/atomic" "time" + "github.com/kami/maven/internal/netaddr" "github.com/kami/maven/internal/store" "golang.org/x/sys/unix" ) @@ -446,6 +446,7 @@ func mapErr(err error) error { type Server struct { api atomic.Value // stores CoreAPI path string + addr netaddr.Addr ln net.Listener wg sync.WaitGroup @@ -610,31 +611,29 @@ type CheckFunc func(ctx context.Context, m Method, params json.RawMessage) error // MethodAssertStepUp dispatch calls this instead of going through CoreAPI. type StepUpFunc func(ctx context.Context) error -// Listen creates a Server bound to path. path's parent dir must exist and be -// 0700 (we chmod it if we own it); the socket file itself is created 0600 so -// only the same unix user can connect — the current "auth floor", same radius -// as wg at the network boundary. Removing a stale socket at path first lets -// the daemon restart cleanly. +// Listen creates a Server bound to path. +// +// A bare path is a unix socket, unchanged: its parent dir is 0700 and the +// socket file itself is 0600, so only the same unix user can connect — the +// current "auth floor", same radius as wg at the network boundary. A stale +// socket is removed first so the daemon restarts cleanly. +// +// A "tcp://host:port?token=..." address binds a network listener instead, for +// a module that lives on another host. There is no filesystem there to be the +// auth floor, so netaddr checks the shared token before this package sees the +// connection and a token is mandatory. See internal/netaddr. func Listen(path string, api CoreAPI) (*Server, error) { - _ = os.Remove(path) // stale socket from a crashed daemon; ignore missing - if err := os.MkdirAll(parentDir(path), 0o700); err != nil { - return nil, fmt.Errorf("ipc: mkdir socket dir: %w", err) - } - // umask could widen the perms on socket creation; tighten then chmod to - // be explicit. 0600 ⇒ read+write by owner only. - oldMask := unix.Umask(0o077) - ln, err := net.Listen("unix", path) - unix.Umask(oldMask) + addr, err := netaddr.Parse(path) if err != nil { - return nil, fmt.Errorf("ipc: listen %s: %w", path, err) + return nil, err } - if err := os.Chmod(path, 0o600); err != nil { - _ = ln.Close() - _ = os.Remove(path) - return nil, fmt.Errorf("ipc: chmod socket: %w", err) + ln, err := netaddr.Listen(addr) + if err != nil { + return nil, err } s := &Server{ path: path, + addr: addr, ln: ln, done: make(chan struct{}), } @@ -1268,7 +1267,7 @@ func (s *Server) Close() error { // missing the seal costs every write since the last clean shutdown. log.Printf("ipc: %d connection(s) still busy after %s, closing anyway", s.liveConns(), closeGrace) } - _ = os.Remove(s.path) + netaddr.Cleanup(s.addr) return err } @@ -1338,25 +1337,6 @@ func (s *Server) Path() string { return s.path } // while the server is serving (dispatch loads api once per request via atomic). func (s *Server) SetAPI(api CoreAPI) { s.api.Store(api) } -func parentDir(p string) string { - if i := lastIndexByte(p, '/'); i >= 0 { - if i == 0 { - return "/" - } - return p[:i] - } - return "." -} - -func lastIndexByte(s string, b byte) int { - for i := len(s) - 1; i >= 0; i-- { - if s[i] == b { - return i - } - } - return -1 -} - // peerCaller — read SO_PEERCRED off a unix conn to identify the connecting // process. Returns ok=false on a non-unix conn or a platform without // SO_PEERCRED; the caller then proceeds without a Caller (the socket perms diff --git a/internal/worker/client.go b/internal/worker/client.go index 73b62ce..44302ab 100644 --- a/internal/worker/client.go +++ b/internal/worker/client.go @@ -24,6 +24,8 @@ import ( "net" "sync" "time" + + "github.com/kami/maven/internal/netaddr" ) // Client — one connection to one worker module. NOT goroutine-safe for @@ -38,15 +40,25 @@ type Client struct { dial func() (net.Conn, error) } -// Dial opens a Client to the worker socket at path. The first call lazily +// Dial opens a Client to the worker module at path. The first call lazily // dials; subsequent calls reuse the conn (a fresh dial happens on next call // after a teardown). Lazy dial keeps a worker that's restarting from // blocking core's startup; core attempts the dial on first use. +// +// path is a netaddr seam address. A bare path is the unix socket it has +// always been; "tcp://workstation:9310?token=..." reaches a module on another +// host, which is how stt and tts move to the machine with the GPU and the +// microphone. A bad address surfaces on the first call, not here, because +// Dial does not fail — see internal/netaddr. func Dial(path string) *Client { + addr, err := netaddr.Parse(path) return &Client{ path: path, dial: func() (net.Conn, error) { - return net.Dial("unix", path) + if err != nil { + return nil, err + } + return netaddr.Dial(addr) }, } } diff --git a/internal/worker/server.go b/internal/worker/server.go index b8f8f42..9033b5a 100644 --- a/internal/worker/server.go +++ b/internal/worker/server.go @@ -7,10 +7,11 @@ // which module to dial; mixing the two is a config error caught cleanly by // the wire, not a runtime goroutine panic). One Server per module process. // -// Socket perms mirror ipc.Server: dir 0700, socket 0600 ⇒ same unix user. -// The module has no key, so the floor is "same user"; the wg/mTLS layers -// are out of scope here (this socket never crosses the network radius — -// it's local-only, point-to-point between two processes on the box). +// The seam address decides the transport. On the default unix socket the +// perms mirror ipc.Server — dir 0700, socket 0600 ⇒ same unix user — and that +// is the whole auth floor, because the seam never leaves the box. A tcp +// address moves the module to another host and takes that floor away, so +// netaddr checks a shared token before the first frame. See internal/netaddr. package worker import ( @@ -18,11 +19,10 @@ import ( "encoding/json" "fmt" "net" - "os" "sync" "sync/atomic" - "golang.org/x/sys/unix" + "github.com/kami/maven/internal/netaddr" ) // Server — a worker module process's listener. Wires either a Transcriber, @@ -34,6 +34,7 @@ type Server struct { s Synthesizer path string + addr netaddr.Addr ln net.Listener wg sync.WaitGroup @@ -61,25 +62,24 @@ func NewSynthesizerServer(path string, s Synthesizer) *Server { // two separate processes per the restart-free / fail-independent invariant). func (srv *Server) SetSynthesizer(s Synthesizer) { srv.s = s } -// Listen binds the unix socket with 0700 dir + 0600 socket perms (same floor -// as internal/ipc). A stale socket at path is removed first so the worker -// process restarts cleanly after a crash, no manual cleanup needed. +// Listen binds the seam address the Server was built with. +// +// A bare path is a unix socket with 0700 dir + 0600 socket perms, the same +// floor as internal/ipc, and a stale socket is removed first so the worker +// process restarts cleanly after a crash. A "tcp://host:port?token=..." +// address binds a network listener instead, so this module can run on the +// workstation while core stays on homesrv; the token is mandatory there, +// because there is no filesystem to be the auth floor. See internal/netaddr. func (srv *Server) Listen() error { - _ = os.Remove(srv.path) - if err := os.MkdirAll(parentDir(srv.path), 0o700); err != nil { - return fmt.Errorf("worker: mkdir socket dir: %w", err) - } - oldMask := unix.Umask(0o077) - ln, err := net.Listen("unix", srv.path) - unix.Umask(oldMask) + addr, err := netaddr.Parse(srv.path) if err != nil { - return fmt.Errorf("worker: listen %s: %w", srv.path, err) + return err } - if err := os.Chmod(srv.path, 0o600); err != nil { - _ = ln.Close() - _ = os.Remove(srv.path) - return fmt.Errorf("worker: chmod socket: %w", err) + ln, err := netaddr.Listen(addr) + if err != nil { + return err } + srv.addr = addr srv.ln = ln return nil } @@ -193,7 +193,7 @@ func (srv *Server) Close() error { } err := srv.ln.Close() srv.wg.Wait() - _ = os.Remove(srv.path) + netaddr.Cleanup(srv.addr) return err } @@ -214,15 +214,3 @@ func marshalResult(v any) json.RawMessage { b, _ := json.Marshal(v) return b } - -func parentDir(p string) string { - for i := len(p) - 1; i >= 0; i-- { - if p[i] == '/' { - if i == 0 { - return "/" - } - return p[:i] - } - } - return "." -}